Merge 8222d36 into dc9f405

Author: GrigoriyPA

ydb/core/fq/libs/row_dispatcher/json_filter.cpp

@@ -53,7 +53,7 @@ class TFilterInputSpec : public NYql::NPureCalc::TInputSpecBase {
     TVector<NYT::TNode> Schemas;
 };
 
-class TFilterInputConsumer : public NYql::NPureCalc::IConsumer<std::pair<ui64, TList<TString>>> {
+class TFilterInputConsumer : public NYql::NPureCalc::IConsumer<std::pair<ui64, TVector<TVector<std::string_view>>>> {
 public:
     TFilterInputConsumer(
         const TFilterInputSpec& spec,
@@ -91,28 +91,32 @@ class TFilterInputConsumer : public NYql::NPureCalc::IConsumer<std::pair<ui64, T
         }
     }
 
-    void OnObject(std::pair<ui64, TList<TString>> value) override {
+    void OnObject(std::pair<ui64, TVector<TVector<std::string_view>>> values) override {
+        Y_ENSURE(FieldsPositions.size() == values.second.size());
+
         NKikimr::NMiniKQL::TThrowingBindTerminator bind;
-        
         with_lock (Worker->GetScopedAlloc()) {
             auto& holderFactory = Worker->GetGraph().GetHolderFactory();
-            NYql::NUdf::TUnboxedValue* items = nullptr;
 
-            NYql::NUdf::TUnboxedValue result = Cache.NewArray(
-                holderFactory,
-                static_cast<ui32>(value.second.size() + 1),
-                items);
-    
-            items[OffsetPosition] = NYql::NUdf::TUnboxedValuePod(value.first);
+            // TODO: use blocks here
+            for (size_t rowId = 0; rowId < values.second.front().size(); ++rowId) {
+                NYql::NUdf::TUnboxedValue* items = nullptr;
+
+                NYql::NUdf::TUnboxedValue result = Cache.NewArray(
+                    holderFactory,
+                    static_cast<ui32>(values.second.size() + 1),
+                    items);
 
-            Y_ENSURE(FieldsPositions.size() == value.second.size());
+                items[OffsetPosition] = NYql::NUdf::TUnboxedValuePod(values.first++);
 
-            size_t i = 0;
-            for (const auto& v : value.second) {
-                NYql::NUdf::TStringValue str(v);
-                items[FieldsPositions[i++]] = NYql::NUdf::TUnboxedValuePod(std::move(str));
+                size_t fieldId = 0;
+                for (const auto& column : values.second) {
+                    NYql::NUdf::TStringValue str(column[rowId]);
+                    items[FieldsPositions[fieldId++]] = NYql::NUdf::TUnboxedValuePod(std::move(str));
+                }
+
+                Worker->Push(std::move(result));
             }
-            Worker->Push(std::move(result));
         }
     }
 
@@ -196,7 +200,7 @@ struct NYql::NPureCalc::TInputSpecTraits<TFilterInputSpec> {
     static constexpr bool IsPartial = false;
     static constexpr bool SupportPushStreamMode = true;
 
-    using TConsumerType = THolder<NYql::NPureCalc::IConsumer<std::pair<ui64, TList<TString>>>>;
+    using TConsumerType = THolder<NYql::NPureCalc::IConsumer<std::pair<ui64, TVector<TVector<std::string_view>>>>>;
 
     static TConsumerType MakeConsumer(
         const TFilterInputSpec& spec,
@@ -238,8 +242,9 @@ class TJsonFilter::TImpl {
         LOG_ROW_DISPATCHER_DEBUG("Program created");
     }
 
-    void Push(ui64 offset, const TList<TString>& value) {
-        InputConsumer->OnObject(std::make_pair(offset, value));
+    void Push(ui64 offset, const TVector<TVector<std::string_view>>& values) {
+        Y_ENSURE(values, "Expected non empty schema");
+        InputConsumer->OnObject(std::make_pair(offset, values));
     }
 
     TString GetSql() const {
@@ -266,7 +271,7 @@ class TJsonFilter::TImpl {
 
 private:
     THolder<NYql::NPureCalc::TPushStreamProgram<TFilterInputSpec, TFilterOutputSpec>> Program;
-    THolder<NYql::NPureCalc::IConsumer<std::pair<ui64, TList<TString>>>> InputConsumer;
+    THolder<NYql::NPureCalc::IConsumer<std::pair<ui64, TVector<TVector<std::string_view>>>>> InputConsumer;
     const TString Sql;
 };
 
@@ -280,9 +285,9 @@ TJsonFilter::TJsonFilter(
 
 TJsonFilter::~TJsonFilter() {
 }
-    
-void TJsonFilter::Push(ui64 offset, const TList<TString>& value) {
-     Impl->Push(offset, value);
+
+void TJsonFilter::Push(ui64 offset, const TVector<TVector<std::string_view>>& values) {
+    Impl->Push(offset, values);
 }
 
 TString TJsonFilter::GetSql() {

ydb/core/fq/libs/row_dispatcher/json_filter.h

@@ -1,23 +1,24 @@
-
 #pragma once
 
-namespace NFq {
-
 #include <ydb/library/yql/public/udf/udf_data_type.h>
 #include <ydb/library/yql/public/udf/udf_value.h>
 
+namespace NFq {
+
 class TJsonFilter {
 public:
     using TCallback = std::function<void(ui64, const TString&)>;
-    
+
 public:
     TJsonFilter(
-        const TVector<TString>& columns, 
+        const TVector<TString>& columns,
         const TVector<TString>& types,
         const TString& whereFilter,
         TCallback callback);
+
     ~TJsonFilter();
-    void Push(ui64 offset, const TList<TString>& value);
+
+    void Push(ui64 offset, const TVector<TVector<std::string_view>>& values);
     TString GetSql();
 
 private: