Merge 9649bd1 into cfbb87c

Author: yumkam

ydb/library/yql/dq/actors/input_transforms/dq_input_transform_lookup.cpp

@@ -101,8 +101,17 @@ class TInputTransformStreamLookupBase
 private: //events
     STRICT_STFUNC(StateFunc,
         hFunc(IDqAsyncLookupSource::TEvLookupResult, Handle);
+        hFunc(IDqComputeActorAsyncInput::TEvAsyncInputError, Handle);
     )
 
+    void Handle(IDqComputeActorAsyncInput::TEvAsyncInputError::TPtr ev) {
+        auto evptr = ev->Get();
+        Send(ComputeActorId, new IDqComputeActorAsyncInput::TEvAsyncInputError(
+                                  InputIndex,
+                                  evptr->Issues,
+                                  evptr->FatalCode));
+    }
+
     void AddReadyQueue(NUdf::TUnboxedValue& lookupKey, NUdf::TUnboxedValue& inputOther, NUdf::TUnboxedValue *lookupPayload) {
             NUdf::TUnboxedValue* outputRowItems;
             NUdf::TUnboxedValue outputRow = HolderFactory.CreateDirectArrayHolder(OutputRowColumnOrder.size(), outputRowItems);

ydb/library/yql/providers/generic/actors/yql_generic_lookup_actor.cpp

@@ -188,8 +188,14 @@ namespace NYql::NDq {
             FinalizeRequest();
         }
 
-        void Handle(TEvError::TPtr) {
-            FinalizeRequest();
+        void Handle(TEvError::TPtr ev) {
+            auto actorSystem = TActivationContext::ActorSystem();
+            auto error = ev->Get()->Error;
+            auto errEv = new IDqComputeActorAsyncInput::TEvAsyncInputError(
+                                  -1,
+                                  NConnector::ErrorToIssues(error),
+                                  NConnector::ErrorToDqStatus(error));
+            actorSystem->Send(new NActors::IEventHandle(ParentId, SelfId(), errEv));
         }
 
         void Handle(NActors::TEvents::TEvPoison::TPtr) {