streamlookup: pass and use MaxDelayedRows parameter (#10489)

Author: yumkam

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

@@ -42,6 +42,7 @@ class TInputTransformStreamLookupBase
         const NMiniKQL::TStructType* lookupPayloadType,
         const NMiniKQL::TMultiType* outputRowType,
         TOutputRowColumnOrder&& outputRowColumnOrder,
+        size_t maxDelayedRows,
         size_t cacheLimit,
         std::chrono::seconds cacheTtl
     )
@@ -63,6 +64,7 @@ class TInputTransformStreamLookupBase
         , OutputRowColumnOrder(std::move(outputRowColumnOrder))
         , InputFlowFetchStatus(NUdf::EFetchStatus::Yield)
         , LruCache(std::make_unique<NKikimr::NMiniKQL::TUnboxedKeyValueLruCacheWithTtl>(cacheLimit, lookupKeyType))
+        , MaxDelayedRows(maxDelayedRows)
         , CacheTtl(cacheTtl)
         , ReadyQueue(OutputRowType)
     {
@@ -142,8 +144,9 @@ class TInputTransformStreamLookupBase
     }
 
     void Handle(IDqAsyncLookupSource::TEvLookupResult::TPtr ev) {
-        if (!KeysForLookup)
+        if (!KeysForLookup) {
             return;
+        }
         auto guard = BindAllocator();
         const auto now = std::chrono::steady_clock::now();
         auto lookupResult = ev->Get()->Result.lock();
@@ -207,7 +210,10 @@ class TInputTransformStreamLookupBase
              NUdf::TUnboxedValue inputRow = HolderFactory.CreateDirectArrayHolder(InputRowType->GetElementsCount(), inputRowItems);
             const auto now = std::chrono::steady_clock::now();
             LruCache->Prune(now);
+            size_t rowLimit = std::numeric_limits<size_t>::max();
+            size_t row = 0;
             while (
+                row < rowLimit &&
                 (KeysForLookup->size() < MaxKeysInRequest) &&
                 ((InputFlowFetchStatus = FetchWideInputValue(inputRowItems)) == NUdf::EFetchStatus::Ok)) {
                 NUdf::TUnboxedValue* keyItems;
@@ -223,9 +229,14 @@ class TInputTransformStreamLookupBase
                 if (auto lookupPayload = LruCache->Get(key, now)) {
                     AddReadyQueue(key, other, &*lookupPayload);
                 } else {
+                    if (AwaitingQueue.empty()) {
+                        // look ahead at most MaxDelayedRows after first missing
+                        rowLimit = row + MaxDelayedRows;
+                    }
                     AwaitingQueue.emplace_back(key, std::move(other));
                     KeysForLookup->emplace(std::move(key), NUdf::TUnboxedValue{});
                 }
+                ++row;
             }
             if (!KeysForLookup->empty()) {
                 Send(LookupSourceId, new IDqAsyncLookupSource::TEvLookupRequest(KeysForLookup));
@@ -283,6 +294,7 @@ class TInputTransformStreamLookupBase
 
     NUdf::EFetchStatus InputFlowFetchStatus;
     std::unique_ptr<NKikimr::NMiniKQL::TUnboxedKeyValueLruCacheWithTtl> LruCache;
+    size_t MaxDelayedRows;
     std::chrono::seconds CacheTtl;
     using TInputKeyOtherPair = std::pair<NUdf::TUnboxedValue, NUdf::TUnboxedValue>;
     using TAwaitingQueue = std::deque<TInputKeyOtherPair, NKikimr::NMiniKQL::TMKQLAllocator<TInputKeyOtherPair>>; //input row split in two parts: key columns and other columns
@@ -545,6 +557,7 @@ std::pair<IDqComputeActorAsyncInput*, NActors::IActor*> CreateInputTransformStre
             lookupPayloadType,
             outputRowType,
             std::move(outputColumnsOrder),
+            settings.GetMaxDelayedRows(),
             settings.GetCacheLimit(),
             std::chrono::seconds(settings.GetCacheTtlSeconds())
         ) :
@@ -564,6 +577,7 @@ std::pair<IDqComputeActorAsyncInput*, NActors::IActor*> CreateInputTransformStre
             lookupPayloadType,
             outputRowType,
             std::move(outputColumnsOrder),
+            settings.GetMaxDelayedRows(),
             settings.GetCacheLimit(),
             std::chrono::seconds(settings.GetCacheTtlSeconds())
         );

ydb/library/yql/dq/expr_nodes/dq_expr_nodes.json

@@ -193,7 +193,7 @@
             {"Index": 6, "Name": "LeftJoinKeyNames", "Type": "TCoAtomList"},
             {"Index": 7, "Name": "RightJoinKeyNames", "Type": "TCoAtomList"},
             {"Index": 8, "Name": "TTL", "Type": "TCoAtom"},
-            {"Index": 9, "Name": "MaxDelay", "Type": "TCoAtom"},
+            {"Index": 9, "Name": "MaxDelayedRows", "Type": "TCoAtom"},
             {"Index": 10, "Name": "MaxCachedRows", "Type": "TCoAtom"}
         ]
     },

ydb/library/yql/dq/proto/dq_tasks.proto

@@ -186,6 +186,7 @@ message TDqInputTransformLookupSettings {
     bytes NarrowOutputRowType = 8; //Serialized struct type
     uint64 CacheLimit = 9;
     uint64 CacheTtlSeconds = 10;
+    uint64 MaxDelayedRows = 11;
 }
 
 message TDqTask {

ydb/library/yql/providers/dq/opt/physical_optimize.cpp

@@ -263,7 +263,7 @@ class TDqsPhysicalOptProposalTransformer : public TOptimizeTransformerBase {
             .RightJoinKeyNames(join.RightJoinKeyNames())
             .TTL(ctx.NewAtom(pos, 300)) //TODO configure me
             .MaxCachedRows(ctx.NewAtom(pos, 1'000'000)) //TODO configure me
-            .MaxDelay(ctx.NewAtom(pos, 1'000'000)) //Configure me
+            .MaxDelayedRows(ctx.NewAtom(pos, 1'000'000)) //Configure me
         .Done();
 
         auto lambda = Build<TCoLambda>(ctx, pos)