YQL-9517: RPC Arrow reader YT column converters (#599)

* YQL-9517: RPC Arrow reader YT column converters

* .make + bool

* wideflow => stream

* yson fix

* fix

* fix

* remove unused

Author: MrLolthe1st

ydb/library/yql/minikql/computation/mkql_block_reader.cpp

@@ -184,7 +184,7 @@ struct TConverterTraits {
     using TTuple = TTupleBlockItemConverter<Nullable>;
     template <typename T, bool Nullable>
     using TFixedSize = TFixedSizeBlockItemConverter<T, Nullable>;
-    template <typename TStringType, bool Nullable, NUdf::EPgStringType PgString = NUdf::EPgStringType::None>
+    template <typename TStringType, bool Nullable, NUdf::EDataSlot TOriginal = NUdf::EDataSlot::String, NUdf::EPgStringType PgString = NUdf::EPgStringType::None>
     using TStrings = TStringBlockItemConverter<TStringType, Nullable, PgString>;
     using TExtOptional = TExternalOptionalBlockItemConverter;
 
@@ -193,15 +193,15 @@ struct TConverterTraits {
             return std::make_unique<TFixedSize<ui64, true>>();
         } else {
             if (desc.Typelen == -1) {
-                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EPgStringType::Text>>();
+                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EDataSlot::String, NUdf::EPgStringType::Text>>();
                 ret->SetPgBuilder(pgBuilder, desc.TypeId, desc.Typelen);
                 return ret;
             } else if (desc.Typelen == -2) {
-                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EPgStringType::CString>>();
+                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EDataSlot::String, NUdf::EPgStringType::CString>>();
                 ret->SetPgBuilder(pgBuilder, desc.TypeId, desc.Typelen);
                 return ret;
             } else {
-                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EPgStringType::Fixed>>();
+                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EDataSlot::String, NUdf::EPgStringType::Fixed>>();
                 ret->SetPgBuilder(pgBuilder, desc.TypeId, desc.Typelen);
                 return ret;
             }

ydb/library/yql/minikql/computation/mkql_block_transport.cpp

@@ -500,7 +500,7 @@ struct TSerializerTraits {
     using TTuple = TTupleBlockSerializer<Nullable>;
     template <typename T, bool Nullable>
     using TFixedSize = TFixedSizeBlockSerializer<sizeof(T), Nullable>;
-    template <typename TStringType, bool Nullable>
+    template <typename TStringType, bool Nullable, NUdf::EDataSlot TOriginal = NUdf::EDataSlot::String>
     using TStrings = TStringBlockSerializer<TStringType, Nullable>;
     using TExtOptional = TExtOptionalBlockSerializer;
 
@@ -519,7 +519,7 @@ struct TDeserializerTraits {
     using TTuple = TTupleBlockDeserializer<Nullable>;
     template <typename T, bool Nullable>
     using TFixedSize = TFixedSizeBlockDeserializer<sizeof(T), Nullable>;
-    template <typename TStringType, bool Nullable>
+    template <typename TStringType, bool Nullable, NUdf::EDataSlot TOriginal = NUdf::EDataSlot::String>
     using TStrings = TStringBlockDeserializer<TStringType, Nullable>;
     using TExtOptional = TExtOptionalBlockDeserializer;
 

ydb/library/yql/minikql/mkql_type_builder.cpp

@@ -2425,7 +2425,7 @@ struct TComparatorTraits {
     using TTuple = NUdf::TTupleBlockItemComparator<Nullable>;
     template <typename T, bool Nullable>
     using TFixedSize = NUdf::TFixedSizeBlockItemComparator<T, Nullable>;
-    template <typename TStringType, bool Nullable>
+    template <typename TStringType, bool Nullable, NUdf::EDataSlot TOriginal = NUdf::EDataSlot::String>
     using TStrings = NUdf::TStringBlockItemComparator<TStringType, Nullable>;
     using TExtOptional = NUdf::TExternalOptionalBlockItemComparator;
 
@@ -2441,7 +2441,7 @@ struct THasherTraits {
     using TTuple = NUdf::TTupleBlockItemHasher<Nullable>;
     template <typename T, bool Nullable>
     using TFixedSize = NUdf::TFixedSizeBlockItemHasher<T, Nullable>;
-    template <typename TStringType, bool Nullable>
+    template <typename TStringType, bool Nullable, NUdf::EDataSlot TOriginal = NUdf::EDataSlot::String>
     using TStrings = NUdf::TStringBlockItemHasher<TStringType, Nullable>;
     using TExtOptional = NUdf::TExternalOptionalBlockItemHasher;
 

ydb/library/yql/providers/common/dq/yql_dq_integration_impl.cpp

@@ -22,20 +22,6 @@ TMaybe<ui64> TDqIntegrationBase::EstimateReadSize(ui64, ui32, const TVector<cons
     return Nothing();
 }
 
-bool TDqIntegrationBase::CanBlockReadTypes(const TStructExprType* node) {
-    for (const auto& e: node->GetItems()) {
-        // Check type
-        auto type = e->GetItemType();
-        while (ETypeAnnotationKind::Optional == type->GetKind()) {
-            type = type->Cast<TOptionalExprType>()->GetItemType();
-        }
-        if (ETypeAnnotationKind::Data != type->GetKind()) {
-            return false;
-        }
-    }
-    return true;
-}
-
 TExprNode::TPtr TDqIntegrationBase::WrapRead(const TDqSettings&, const TExprNode::TPtr& read, TExprContext&) {
     return read;
 }

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

@@ -10,6 +10,7 @@
 #include <ydb/library/yql/core/type_ann/type_ann_core.h>
 #include <ydb/library/yql/core/yql_expr_type_annotation.h>
 #include <ydb/library/yql/core/yql_type_annotation.h>
+#include <ydb/library/yql/core/yql_opt_utils.h>
 
 #include <ydb/library/yql/dq/opt/dq_opt.h>
 #include <ydb/library/yql/dq/opt/dq_opt_phy.h>
@@ -92,12 +93,13 @@ namespace NYql::NDqs {
                     }
 
                     YQL_CLOG(INFO, ProviderDq) << "DqsRewritePhyBlockReadOnDqIntegration";
-
                     return Build<TCoWideFromBlocks>(ctx, node->Pos())
-                            .Input(Build<TDqReadBlockWideWrap>(ctx, node->Pos())
-                                    .Input(readWideWrap.Input())
-                                    .Flags(readWideWrap.Flags())
-                                    .Token(readWideWrap.Token())
+                            .Input(Build<TCoToFlow>(ctx, node->Pos())
+                                .Input(Build<TDqReadBlockWideWrap>(ctx, node->Pos())
+                                        .Input(readWideWrap.Input())
+                                        .Flags(readWideWrap.Flags())
+                                        .Token(readWideWrap.Token())
+                                    .Done())
                                 .Done())
                             .Done().Ptr();
                 }, ctx, optSettings);

ydb/library/yql/providers/dq/provider/yql_dq_datasource_type_ann.cpp

@@ -155,6 +155,8 @@ class TDqsDataSourceTypeAnnotationTransformer : public TVisitorTransformerBase {
             }
 
             types.push_back(ctx.MakeType<TScalarExprType>(ctx.MakeType<TDataExprType>(EDataSlot::Uint64)));
+            input->SetTypeAnn(ctx.MakeType<TStreamExprType>(ctx.MakeType<TMultiExprType>(types)));
+            return TStatus::Ok;
         }
 
         input->SetTypeAnn(ctx.MakeType<TFlowExprType>(ctx.MakeType<TMultiExprType>(types)));