PG comparators for BlockItem

18 files changed, 248 insertions, 86 deletions

diff --git a/ydb/library/yql/minikql/comp_nodes/mkql_block_agg_minmax.cpp b/ydb/library/yql/minikql/comp_nodes/mkql_block_agg_minmax.cpp
index 9f19ee56071..bbfddbacac9 100644
--- a/ydb/library/yql/minikql/comp_nodes/mkql_block_agg_minmax.cpp
+++ b/ydb/library/yql/minikql/comp_nodes/mkql_block_agg_minmax.cpp
@@ -208,7 +208,7 @@ public:
         , ReaderOne_(MakeBlockReader(TTypeInfoHelper(), type))
         , ReaderTwo_(MakeBlockReader(TTypeInfoHelper(), type))
         , Converter_(MakeBlockItemConverter(TTypeInfoHelper(), type, ctx.Builder->GetPgBuilder()))
-        , Compare_(NYql::NUdf::MakeBlockItemComparator(TTypeInfoHelper(), type))
+        , Compare_(TBlockTypeHelper().MakeComparator(type))
     {
     }
 
@@ -280,7 +280,7 @@ private:
     const std::unique_ptr<IBlockReader> ReaderOne_;
     const std::unique_ptr<IBlockReader> ReaderTwo_;
     const std::unique_ptr<IBlockItemConverter> Converter_;
-    const std::unique_ptr<NYql::NUdf::IBlockItemComparator> Compare_;
+    const NYql::NUdf::IBlockItemComparator::TPtr Compare_;
 };
 
 template<bool IsMin>
@@ -294,7 +294,7 @@ public:
         , Type_(type)
         , Reader_(MakeBlockReader(TTypeInfoHelper(), type))
         , Converter_(MakeBlockItemConverter(TTypeInfoHelper(), type, ctx.Builder->GetPgBuilder()))
-        , Compare_(NYql::NUdf::MakeBlockItemComparator(TTypeInfoHelper(), type))
+        , Compare_(TBlockTypeHelper().MakeComparator(type))
     {
     }
 
@@ -324,7 +324,7 @@ private:
     TType* const Type_;
     const std::unique_ptr<IBlockReader> Reader_;
     const std::unique_ptr<IBlockItemConverter> Converter_;
-    const std::unique_ptr<NYql::NUdf::IBlockItemComparator> Compare_;
+    const NYql::NUdf::IBlockItemComparator::TPtr Compare_;
 };
 
 template<bool IsMin>
@@ -338,8 +338,7 @@ public:
         , Type_(type)
         , Reader_(MakeBlockReader(TTypeInfoHelper(), type))
         , Converter_(MakeBlockItemConverter(TTypeInfoHelper(), type, ctx.Builder->GetPgBuilder()))
-        , Compare_(NYql::NUdf::MakeBlockItemComparator(TTypeInfoHelper(), type))
-
+        , Compare_(TBlockTypeHelper().MakeComparator(type))
     {
     }
 
@@ -369,7 +368,7 @@ private:
     TType* const Type_;
     const std::unique_ptr<IBlockReader> Reader_;
     const std::unique_ptr<IBlockItemConverter> Converter_;
-    const std::unique_ptr<NYql::NUdf::IBlockItemComparator> Compare_;
+    const NYql::NUdf::IBlockItemComparator::TPtr Compare_;
 };
 
 template <typename TStringType, bool IsMin>
diff --git a/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.cpp b/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.cpp
index 581bc112eb5..5fdcf2a7b9f 100644
--- a/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.cpp
+++ b/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.cpp
@@ -174,9 +174,25 @@ struct TConverterTraits {
     using TTuple = TTupleBlockItemConverter<Nullable>;
     template <typename T, bool Nullable>
     using TFixedSize = TFixedSizeBlockItemConverter<T, Nullable>;
-    template <typename TStringType, bool Nullable, NUdf::EPgStringType PgString>
+    template <typename TStringType, bool Nullable, NUdf::EPgStringType PgString = NUdf::EPgStringType::None>
     using TStrings = TStringBlockItemConverter<TStringType, Nullable, PgString>;
     using TExtOptional = TExternalOptionalBlockItemConverter;
+
+    static std::unique_ptr<TResult> MakePg(const NUdf::TPgTypeDescription& desc, const NUdf::IPgBuilder* pgBuilder) {
+        if (desc.PassByValue) {
+            return std::make_unique<TFixedSize<ui64, true>>();
+        } else {
+            if (desc.Typelen == -1) {
+                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EPgStringType::Text>>();
+                ret->SetPgBuilder(pgBuilder, desc.Typelen, desc.TypeId);
+                return ret;
+            } else {
+                auto ret = std::make_unique<TStrings<arrow::BinaryType, true, NUdf::EPgStringType::CString>>();
+                ret->SetPgBuilder(pgBuilder, desc.Typelen, desc.TypeId);
+                return ret;
+            }
+        }        
+    }
 };
 
 } // namespace
diff --git a/ydb/library/yql/minikql/comp_nodes/mkql_block_top.cpp b/ydb/library/yql/minikql/comp_nodes/mkql_block_top.cpp
index f51fd0e7890..6891e5d930f 100644
--- a/ydb/library/yql/minikql/comp_nodes/mkql_block_top.cpp
+++ b/ydb/library/yql/minikql/comp_nodes/mkql_block_top.cpp
@@ -199,7 +199,7 @@ private:
         TVector<std::unique_ptr<IArrayBuilder>> Builders_;
         ui64 BuilderMaxLength_ = 0;
         ui64 BuilderLength_ = 0;
-        TVector<std::unique_ptr<NUdf::IBlockItemComparator>> Comparators_; // by key columns only
+        TVector<NUdf::IBlockItemComparator::TPtr> Comparators_; // by key columns only
 
         TVector<NUdf::TUnboxedValue> Values_;
         TVector<NUdf::TUnboxedValue*> ValuePointers_;
@@ -228,7 +228,7 @@ private:
 
             Comparators_.resize(keyIndicies.size());
             for (ui32 k = 0; k < keyIndicies.size(); ++k) {
-                Comparators_[k] = NUdf::MakeBlockItemComparator(TTypeInfoHelper(), columns[keyIndicies[k]]->GetItemType());
+                Comparators_[k] = TBlockTypeHelper().MakeComparator(columns[keyIndicies[k]]->GetItemType());
             }
 
             SortInput_.resize(columns.size());
diff --git a/ydb/library/yql/minikql/mkql_type_builder.cpp b/ydb/library/yql/minikql/mkql_type_builder.cpp
index cf028560347..a723a5b4208 100644
--- a/ydb/library/yql/minikql/mkql_type_builder.cpp
+++ b/ydb/library/yql/minikql/mkql_type_builder.cpp
@@ -4,11 +4,13 @@
 #include "mkql_alloc.h"
 
 #include <ydb/library/yql/public/udf/udf_type_ops.h>
+#include <ydb/library/yql/public/udf/arrow/block_item_comparator.h>
 
 #include <library/cpp/containers/stack_vector/stack_vec.h>
 #include <ydb/library/yql/minikql/computation/mkql_computation_node_impl.h>
 #include <ydb/library/yql/minikql/mkql_node_printer.h>
 #include <ydb/library/yql/parser/pg_catalog/catalog.h>
+#include <ydb/library/yql/parser/pg_wrapper/interface/compare.h>
 #include <array>
 
 #include <arrow/c/bridge.h>
@@ -1656,6 +1658,10 @@ NUdf::IFunctionTypeInfoBuilder15& TFunctionTypeInfoBuilder::IsStrict() {
     return *this;
 }
 
+const NUdf::IBlockTypeHelper& TFunctionTypeInfoBuilder::IBlockTypeHelper() const {
+    return BlockTypeHelper;
+}
+
 bool TFunctionTypeInfoBuilder::GetSecureParam(NUdf::TStringRef key, NUdf::TStringRef& value) const {
     if (SecureParamsProvider_)
         return SecureParamsProvider_->GetSecureParam(key, value);
@@ -2392,5 +2398,25 @@ size_t CalcMaxBlockItemSize(const TType* type) {
     MKQL_ENSURE(false, "Unsupported type");
 }
 
+struct TComparatorTraits {
+    using TResult = NUdf::IBlockItemComparator;
+    template <bool Nullable>
+    using TTuple = NUdf::TTupleBlockItemComparator<Nullable>;
+    template <typename T, bool Nullable>
+    using TFixedSize = NUdf::TFixedSizeBlockItemComparator<T, Nullable>;
+    template <typename TStringType, bool Nullable>
+    using TStrings = NUdf::TStringBlockItemComparator<TStringType, Nullable>;
+    using TExtOptional = NUdf::TExternalOptionalBlockItemComparator;
+
+    static std::unique_ptr<TResult> MakePg(const NUdf::TPgTypeDescription& desc, const NUdf::IPgBuilder* pgBuilder) {
+        Y_UNUSED(pgBuilder);
+        return std::unique_ptr<TResult>(MakePgItemComparator(desc.TypeId).Release());
+    }
+};
+
+NUdf::IBlockItemComparator::TPtr TBlockTypeHelper::MakeComparator(NUdf::TType* type) const {
+    return NUdf::MakeBlockReaderImpl<TComparatorTraits>(TTypeInfoHelper(), type, nullptr).release();
+}
+
 } // namespace NMiniKQL
 } // namespace Nkikimr
diff --git a/ydb/library/yql/minikql/mkql_type_builder.h b/ydb/library/yql/minikql/mkql_type_builder.h
index 580f3e4732d..2043c5b3fcb 100644
--- a/ydb/library/yql/minikql/mkql_type_builder.h
+++ b/ydb/library/yql/minikql/mkql_type_builder.h
@@ -3,6 +3,7 @@
 #include "mkql_node.h"
 
 #include <ydb/library/yql/public/udf/udf_type_builder.h>
+#include <ydb/library/yql/public/udf/arrow/block_type_helper.h>
 #include <ydb/library/yql/parser/pg_wrapper/interface/compare.h>
 
 #include <util/generic/size_literals.h>
@@ -12,6 +13,11 @@
 namespace NKikimr {
 namespace NMiniKQL {
 
+class TBlockTypeHelper : public NUdf::IBlockTypeHelper {
+public:
+    NUdf::IBlockItemComparator::TPtr MakeComparator(NUdf::TType* type) const final;
+};
+
 constexpr size_t MaxBlockSizeInBytes = 1_MB;
 static_assert(MaxBlockSizeInBytes < (size_t)std::numeric_limits<i32>::max());
 
@@ -157,6 +163,7 @@ public:
 
     NUdf::IFunctionTypeInfoBuilder15& SupportsBlocks() override;
     NUdf::IFunctionTypeInfoBuilder15& IsStrict() override;
+    const NUdf::IBlockTypeHelper& IBlockTypeHelper() const override;
 
     bool GetSecureParam(NUdf::TStringRef key, NUdf::TStringRef& value) const override;
 
@@ -171,6 +178,7 @@ private:
     ui32 OptionalArgs_ = 0;
     TString Payload_;
     NUdf::ITypeInfoHelper::TPtr TypeInfoHelper_;
+    TBlockTypeHelper BlockTypeHelper;
     TStringBuf ModuleName_;
     NUdf::ICountersProvider* CountersProvider_;
     NUdf::TSourcePosition Pos_;
diff --git a/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp b/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
index 606e149a8b7..efdf042b959 100644
--- a/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
@@ -124,6 +124,10 @@ Datum ScalarDatumFromPod(const NUdf::TUnboxedValuePod& value) {
     return (Datum)value.Get<ui64>();
 }
 
+Datum ScalarDatumFromItem(const NUdf::TBlockItem& value) {
+    return (Datum)value.As<ui64>();
+}
+
 class TBoxedValueWithFree : public NUdf::TBoxedValueBase {
 public:
     void operator delete(void *mem) noexcept {
@@ -184,6 +188,10 @@ Datum PointerDatumFromPod(const NUdf::TUnboxedValuePod& value) {
     return (Datum)(((const char*)value.AsBoxed().Get()) + PallocHdrSize);
 }
 
+Datum PointerDatumFromItem(const NUdf::TBlockItem& value) {
+    return (Datum)value.AsStringRef().Data();
+}
+
 NUdf::TUnboxedValue CreatePgString(i32 typeLen, ui32 targetTypeId, TStringBuf data) {
     // typname => 'cstring', typlen => '-2', the only type with typlen == -2
     // typname = > 'text', typlen = > '-1'
@@ -2619,12 +2627,13 @@ void PgDestroyContext(const std::string_view& contextType, void* ctx) {
     }
 }
 
+template <bool PassByValue>
 class TPgHash : public NUdf::IHash {
 public:
-    TPgHash(const NMiniKQL::TPgType* type)
-        : Type(type)
-        , TypeDesc(NYql::NPg::LookupType(type->GetTypeId()))
+    TPgHash(const NYql::NPg::TTypeDesc& typeDesc)
+        : TypeDesc(typeDesc)
     {
+        Y_ENSURE(PassByValue == TypeDesc.PassByValue);
         Y_ENSURE(TypeDesc.HashProcId);
 
         Zero(FInfoHash);
@@ -2645,7 +2654,7 @@ public:
             return 0;
         }
 
-        callInfo->args[0] = { TypeDesc.PassByValue ?
+        callInfo->args[0] = { PassByValue ?
             ScalarDatumFromPod(lhs) :
             PointerDatumFromPod(lhs), false };
 
@@ -2655,24 +2664,30 @@ public:
     }
 
 private:
-    const NMiniKQL::TPgType* Type;
     const NYql::NPg::TTypeDesc TypeDesc;
 
     FmgrInfo FInfoHash;
 };
 
 NUdf::IHash::TPtr MakePgHash(const NMiniKQL::TPgType* type) {
-    return new TPgHash(type);
+    const auto& typeDesc = NYql::NPg::LookupType(type->GetTypeId());
+    if (typeDesc.PassByValue) {
+        return new TPgHash<true>(typeDesc);
+    } else {
+        return new TPgHash<false>(typeDesc);
+    }
 }
 
-class TPgCompare : public NUdf::ICompare {
+template <bool PassByValue>
+class TPgCompare : public NUdf::ICompare, public NUdf::TBlockItemComparatorBase<TPgCompare<PassByValue>, true> {
 public:
-    TPgCompare(const NMiniKQL::TPgType* type)
-        : Type(type)
-        , TypeDesc(NYql::NPg::LookupType(type->GetTypeId()))
+    TPgCompare(const NYql::NPg::TTypeDesc& typeDesc)
+        : TypeDesc(typeDesc)
     {
+        Y_ENSURE(PassByValue == TypeDesc.PassByValue);
         Y_ENSURE(TypeDesc.LessProcId);
         Y_ENSURE(TypeDesc.CompareProcId);
+        Y_ENSURE(TypeDesc.EqualProcId);
 
         Zero(FInfoLess);
         fmgr_info(TypeDesc.LessProcId, &FInfoLess);
@@ -2685,6 +2700,12 @@ public:
         Y_ENSURE(!FInfoCompare.fn_retset);
         Y_ENSURE(FInfoCompare.fn_addr);
         Y_ENSURE(FInfoCompare.fn_nargs == 2);
+
+        Zero(FInfoEquals);
+        fmgr_info(TypeDesc.EqualProcId, &FInfoEquals);
+        Y_ENSURE(!FInfoEquals.fn_retset);
+        Y_ENSURE(FInfoEquals.fn_addr);
+        Y_ENSURE(FInfoEquals.fn_nargs == 2);        
     }
 
     bool Less(NUdf::TUnboxedValuePod lhs, NUdf::TUnboxedValuePod rhs) const override {
@@ -2706,10 +2727,10 @@ public:
             return false;
         }
 
-        callInfo->args[0] = { TypeDesc.PassByValue ?
+        callInfo->args[0] = { PassByValue ?
             ScalarDatumFromPod(lhs) :
             PointerDatumFromPod(lhs), false };
-        callInfo->args[1] = { TypeDesc.PassByValue ?
+        callInfo->args[1] = { PassByValue ?
             ScalarDatumFromPod(rhs) :
             PointerDatumFromPod(rhs), false };
 
@@ -2737,10 +2758,10 @@ public:
             return 1;
         }
 
-        callInfo->args[0] = { TypeDesc.PassByValue ?
+        callInfo->args[0] = { PassByValue ?
             ScalarDatumFromPod(lhs) :
             PointerDatumFromPod(lhs), false };
-        callInfo->args[1] = { TypeDesc.PassByValue ?
+        callInfo->args[1] = { PassByValue ?
             ScalarDatumFromPod(rhs) :
             PointerDatumFromPod(rhs), false };
 
@@ -2749,23 +2770,94 @@ public:
         return DatumGetInt32(x);
     }
 
+    i64 DoCompare(NUdf::TBlockItem lhs, NUdf::TBlockItem rhs) const {
+        LOCAL_FCINFO(callInfo, 2);
+        Zero(*callInfo);
+        callInfo->flinfo = const_cast<FmgrInfo*>(&FInfoCompare); // don't copy becase of ICompare isn't threadsafe
+        callInfo->nargs = 2;
+        callInfo->fncollation = DEFAULT_COLLATION_OID;
+        callInfo->isnull = false;
+        callInfo->args[0] = { PassByValue ?
+            ScalarDatumFromItem(lhs) :
+            PointerDatumFromItem(lhs), false };
+        callInfo->args[1] = { PassByValue ?
+            ScalarDatumFromItem(rhs) :
+            PointerDatumFromItem(rhs), false };
+
+        auto x = FInfoCompare.fn_addr(callInfo);
+        Y_ENSURE(!callInfo->isnull);
+        return DatumGetInt32(x);
+    }
+
+    bool DoEquals(NUdf::TBlockItem lhs, NUdf::TBlockItem rhs) const {
+        LOCAL_FCINFO(callInfo, 2);
+        Zero(*callInfo);
+        callInfo->flinfo = const_cast<FmgrInfo*>(&FInfoEquals); // don't copy becase of ICompare isn't threadsafe
+        callInfo->nargs = 2;
+        callInfo->fncollation = DEFAULT_COLLATION_OID;
+        callInfo->isnull = false;
+        callInfo->args[0] = { PassByValue ?
+            ScalarDatumFromItem(lhs) :
+            PointerDatumFromItem(lhs), false };
+        callInfo->args[1] = { PassByValue ?
+            ScalarDatumFromItem(rhs) :
+            PointerDatumFromItem(rhs), false };
+
+        auto x = FInfoEquals.fn_addr(callInfo);
+        Y_ENSURE(!callInfo->isnull);
+        return DatumGetBool(x);
+    }
+
+    bool DoLess(NUdf::TBlockItem lhs, NUdf::TBlockItem rhs) const {
+        LOCAL_FCINFO(callInfo, 2);
+        Zero(*callInfo);
+        callInfo->flinfo = const_cast<FmgrInfo*>(&FInfoLess); // don't copy becase of ICompare isn't threadsafe
+        callInfo->nargs = 2;
+        callInfo->fncollation = DEFAULT_COLLATION_OID;
+        callInfo->isnull = false;
+        callInfo->args[0] = { PassByValue ?
+            ScalarDatumFromItem(lhs) :
+            PointerDatumFromItem(lhs), false };
+        callInfo->args[1] = { PassByValue ?
+            ScalarDatumFromItem(rhs) :
+            PointerDatumFromItem(rhs), false };
+
+        auto x = FInfoLess.fn_addr(callInfo);
+        Y_ENSURE(!callInfo->isnull);
+        return DatumGetBool(x);
+    }
+
 private:
-    const NMiniKQL::TPgType* Type;
     const NYql::NPg::TTypeDesc TypeDesc;
 
-    FmgrInfo FInfoLess, FInfoCompare;
+    FmgrInfo FInfoLess, FInfoCompare, FInfoEquals;
 };
 
 NUdf::ICompare::TPtr MakePgCompare(const NMiniKQL::TPgType* type) {
-    return new TPgCompare(type);
+    const auto& typeDesc = NYql::NPg::LookupType(type->GetTypeId());
+    if (typeDesc.PassByValue) {
+        return new TPgCompare<true>(typeDesc);
+    } else {
+        return new TPgCompare<false>(typeDesc);
+    }
 }
 
+NUdf::IBlockItemComparator::TPtr MakePgItemComparator(ui32 typeId) {
+    const auto& typeDesc = NYql::NPg::LookupType(typeId);
+    if (typeDesc.PassByValue) {
+        return new TPgCompare<true>(typeDesc);
+    } else {
+        return new TPgCompare<false>(typeDesc);
+    }
+}
+
+template <bool PassByValue>
 class TPgEquate: public NUdf::IEquate {
 public:
-    TPgEquate(const NMiniKQL::TPgType* type)
-        : Type(type)
-        , TypeDesc(NYql::NPg::LookupType(type->GetTypeId()))
+    TPgEquate(const NYql::NPg::TTypeDesc& typeDesc)
+        : TypeDesc(typeDesc)
     {
+        Y_ENSURE(PassByValue == TypeDesc.PassByValue);
         Y_ENSURE(TypeDesc.EqualProcId);
 
         Zero(FInfoEquate);
@@ -2794,10 +2886,10 @@ public:
             return false;
         }
 
-        callInfo->args[0] = { TypeDesc.PassByValue ?
+        callInfo->args[0] = { PassByValue ?
             ScalarDatumFromPod(lhs) :
             PointerDatumFromPod(lhs), false };
-        callInfo->args[1] = { TypeDesc.PassByValue ?
+        callInfo->args[1] = { PassByValue ?
             ScalarDatumFromPod(rhs) :
             PointerDatumFromPod(rhs), false };
 
@@ -2807,14 +2899,18 @@ public:
     }
 
 private:
-    const NMiniKQL::TPgType* Type;
     const NYql::NPg::TTypeDesc TypeDesc;
 
     FmgrInfo FInfoEquate;
 };
 
 NUdf::IEquate::TPtr MakePgEquate(const TPgType* type) {
-    return new TPgEquate(type);
+    const auto& typeDesc = NYql::NPg::LookupType(type->GetTypeId());
+    if (typeDesc.PassByValue) {
+        return new TPgEquate<true>(typeDesc);
+    } else {
+        return new TPgEquate<false>(typeDesc);
+    }
 }
 
 void* PgInitializeMainContext() {
diff --git a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.darwin-x86_64.txt b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.darwin-x86_64.txt
index 39f80d898a9..a1a2fa37d7f 100644
--- a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.darwin-x86_64.txt
+++ b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.darwin-x86_64.txt
@@ -16,6 +16,7 @@ target_link_libraries(parser-pg_wrapper-interface PUBLIC
   yutil
   library-yql-ast
   yql-public-udf
+  public-udf-arrow
 )
 target_sources(parser-pg_wrapper-interface PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/parser/pg_wrapper/interface/interface.cpp
diff --git a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-aarch64.txt b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-aarch64.txt
index 34d54d3014f..6bc55e03a45 100644
--- a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-aarch64.txt
+++ b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-aarch64.txt
@@ -17,6 +17,7 @@ target_link_libraries(parser-pg_wrapper-interface PUBLIC
   yutil
   library-yql-ast
   yql-public-udf
+  public-udf-arrow
 )
 target_sources(parser-pg_wrapper-interface PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/parser/pg_wrapper/interface/interface.cpp
diff --git a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-x86_64.txt b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-x86_64.txt
index 34d54d3014f..6bc55e03a45 100644
--- a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-x86_64.txt
+++ b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.linux-x86_64.txt
@@ -17,6 +17,7 @@ target_link_libraries(parser-pg_wrapper-interface PUBLIC
   yutil
   library-yql-ast
   yql-public-udf
+  public-udf-arrow
 )
 target_sources(parser-pg_wrapper-interface PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/parser/pg_wrapper/interface/interface.cpp
diff --git a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.windows-x86_64.txt b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.windows-x86_64.txt
index 39f80d898a9..a1a2fa37d7f 100644
--- a/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.windows-x86_64.txt
+++ b/ydb/library/yql/parser/pg_wrapper/interface/CMakeLists.windows-x86_64.txt
@@ -16,6 +16,7 @@ target_link_libraries(parser-pg_wrapper-interface PUBLIC
   yutil
   library-yql-ast
   yql-public-udf
+  public-udf-arrow
 )
 target_sources(parser-pg_wrapper-interface PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/parser/pg_wrapper/interface/interface.cpp
diff --git a/ydb/library/yql/parser/pg_wrapper/interface/compare.h b/ydb/library/yql/parser/pg_wrapper/interface/compare.h
index 4b7eb15ef1a..a3870c5fce6 100644
--- a/ydb/library/yql/parser/pg_wrapper/interface/compare.h
+++ b/ydb/library/yql/parser/pg_wrapper/interface/compare.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ydb/library/yql/public/udf/udf_type_builder.h>
+#include <ydb/library/yql/public/udf/arrow/block_item_comparator.h>
 
 namespace NKikimr {
 namespace NMiniKQL {
@@ -10,6 +11,7 @@ class TPgType;
 NUdf::IHash::TPtr MakePgHash(const TPgType* type);
 NUdf::ICompare::TPtr MakePgCompare(const TPgType* type);
 NUdf::IEquate::TPtr MakePgEquate(const TPgType* type);
+NUdf::IBlockItemComparator::TPtr MakePgItemComparator(ui32 typeId);
 
 } // namespace NMiniKQL
 } // namespace NKikimr
diff --git a/ydb/library/yql/public/udf/arrow/block_item.h b/ydb/library/yql/public/udf/arrow/block_item.h
index 727d273d3a2..21042c7d48a 100644
--- a/ydb/library/yql/public/udf/arrow/block_item.h
+++ b/ydb/library/yql/public/udf/arrow/block_item.h
@@ -6,6 +6,7 @@
 
 namespace NYql::NUdf {
 
+// ABI stable
 class TBlockItem {
     enum class EMarkers : ui8 {
         Empty = 0,
diff --git a/ydb/library/yql/public/udf/arrow/block_item_comparator.h b/ydb/library/yql/public/udf/arrow/block_item_comparator.h
index 3044f8d4e3d..98ca57d7f94 100644
--- a/ydb/library/yql/public/udf/arrow/block_item_comparator.h
+++ b/ydb/library/yql/public/udf/arrow/block_item_comparator.h
@@ -3,12 +3,17 @@
 #include "block_item.h"
 #include "block_reader.h"
 
+#include <ydb/library/yql/public/udf/udf_ptr.h>
 #include <ydb/library/yql/public/udf/udf_type_inspection.h>
+#include <ydb/library/yql/public/udf/udf_type_size_check.h>
 
 namespace NYql::NUdf {
 
+// ABI stable
 class IBlockItemComparator {
 public:
+    using TPtr = TUniquePtr<IBlockItemComparator>;
+
     virtual ~IBlockItemComparator() = default;
 
     virtual i64 Compare(TBlockItem lhs, TBlockItem rhs) const = 0;
@@ -19,6 +24,7 @@ public:
     }
 };
 
+UDF_ASSERT_TYPE_SIZE(IBlockItemComparator, 8);
 
 template <typename TDerived, bool Nullable>
 class TBlockItemComparatorBase : public IBlockItemComparator {
@@ -126,15 +132,9 @@ public:
     }
 };
 
-template <typename TStringType, bool Nullable, NUdf::EPgStringType PgString>
-class TStringBlockItemComparator : public TBlockItemComparatorBase<TStringBlockItemComparator<TStringType, Nullable, PgString>, Nullable> {
+template <typename TStringType, bool Nullable>
+class TStringBlockItemComparator : public TBlockItemComparatorBase<TStringBlockItemComparator<TStringType, Nullable>, Nullable> {
 public:
-    void SetPgBuilder(const IPgBuilder* pgBuilder, i32 typeLen, ui32 pgTypeId) {
-        Y_UNUSED(pgBuilder);
-        Y_UNUSED(typeLen);
-        Y_UNUSED(pgTypeId);
-    }
-
     i64 DoCompare(TBlockItem lhs, TBlockItem rhs) const {
         return lhs.AsStringRef().Compare(rhs.AsStringRef());
     }
@@ -218,19 +218,4 @@ private:
     std::unique_ptr<IBlockItemComparator> Inner_;
 };
 
-struct TComparatorTraits {
-    using TResult = IBlockItemComparator;
-    template <bool Nullable>
-    using TTuple = TTupleBlockItemComparator<Nullable>;
-    template <typename T, bool Nullable>
-    using TFixedSize = TFixedSizeBlockItemComparator<T, Nullable>;
-    template <typename TStringType, bool Nullable, NUdf::EPgStringType PgString>
-    using TStrings = TStringBlockItemComparator<TStringType, Nullable, PgString>;
-    using TExtOptional = TExternalOptionalBlockItemComparator;
-};
-
-inline std::unique_ptr<IBlockItemComparator> MakeBlockItemComparator(const ITypeInfoHelper& typeInfoHelper, const TType* type) {
-    return MakeBlockReaderImpl<TComparatorTraits>(typeInfoHelper, type, nullptr);
-}
-
 }
diff --git a/ydb/library/yql/public/udf/arrow/block_reader.h b/ydb/library/yql/public/udf/arrow/block_reader.h
index 55cc8e60d0d..e79f11ee206 100644
--- a/ydb/library/yql/public/udf/arrow/block_reader.h
+++ b/ydb/library/yql/public/udf/arrow/block_reader.h
@@ -74,17 +74,11 @@ public:
     }
 };
 
-template<typename TStringType, bool Nullable, EPgStringType PgString = EPgStringType::None>
+template<typename TStringType, bool Nullable>
 class TStringBlockReader final : public IBlockReader {
 public:
     using TOffset = typename TStringType::offset_type;
 
-    void SetPgBuilder(const IPgBuilder* pgBuilder, i32 typeLen, ui32 pgTypeId) {
-        Y_UNUSED(pgBuilder);
-        Y_UNUSED(typeLen);
-        Y_UNUSED(pgTypeId);
-    }
-
     TBlockItem GetItem(const arrow::ArrayData& data, size_t index) final {
         Y_VERIFY_DEBUG(data.buffers.size() == 3);
         if constexpr (Nullable) {
@@ -265,9 +259,18 @@ struct TReaderTraits {
     using TTuple = TTupleBlockReader<Nullable>;
     template <typename T, bool Nullable>
     using TFixedSize = TFixedSizeBlockReader<T, Nullable>;
-    template <typename TStringType, bool Nullable, EPgStringType PgString>
-    using TStrings = TStringBlockReader<TStringType, Nullable, PgString>;
+    template <typename TStringType, bool Nullable>
+    using TStrings = TStringBlockReader<TStringType, Nullable>;
     using TExtOptional = TExternalOptionalBlockReader;
+
+    static std::unique_ptr<TResult> MakePg(const TPgTypeDescription& desc, const IPgBuilder* pgBuilder) {
+        Y_UNUSED(pgBuilder);
+        if (desc.PassByValue) {
+            return std::make_unique<TFixedSize<ui64, true>>();
+        } else {
+            return std::make_unique<TStrings<arrow::BinaryType, true>>();
+        }        
+    }
 };
 
 template <typename TTraits>
@@ -288,12 +291,12 @@ std::unique_ptr<typename TTraits::TResult> MakeFixedSizeBlockReaderImpl(bool isO
     }
 }
 
-template <typename TTraits, typename T, EPgStringType PgString>
+template <typename TTraits, typename T>
 std::unique_ptr<typename TTraits::TResult> MakeStringBlockReaderImpl(bool isOptional) {
     if (isOptional) {
-        return std::make_unique<typename TTraits::template TStrings<T, true, PgString>>();
+        return std::make_unique<typename TTraits::template TStrings<T, true>>();
     } else {
-        return std::make_unique<typename TTraits::template TStrings<T, false, PgString>>();
+        return std::make_unique<typename TTraits::template TStrings<T, false>>();
     }
 }
 
@@ -376,9 +379,9 @@ std::unique_ptr<typename TTraits::TResult> MakeBlockReaderImpl(const ITypeInfoHe
         case NUdf::EDataSlot::Double:
             return MakeFixedSizeBlockReaderImpl<TTraits, double>(isOptional);
         case NUdf::EDataSlot::String:
-            return MakeStringBlockReaderImpl<TTraits, arrow::BinaryType, EPgStringType::None>(isOptional);
+            return MakeStringBlockReaderImpl<TTraits, arrow::BinaryType>(isOptional);
         case NUdf::EDataSlot::Utf8:
-            return MakeStringBlockReaderImpl<TTraits, arrow::StringType, EPgStringType::None>(isOptional);
+            return MakeStringBlockReaderImpl<TTraits, arrow::StringType>(isOptional);
         default:
             Y_ENSURE(false, "Unsupported data slot");
         }
@@ -387,19 +390,7 @@ std::unique_ptr<typename TTraits::TResult> MakeBlockReaderImpl(const ITypeInfoHe
     TPgTypeInspector typePg(typeInfoHelper, type);
     if (typePg) {
         auto desc = typeInfoHelper.FindPgTypeDescription(typePg.GetTypeId());
-        if (desc->PassByValue) {
-            return MakeFixedSizeBlockReaderImpl<TTraits, ui64>(true);
-        } else {
-            if (desc->Typelen == -1) {
-                auto ret = std::make_unique<typename TTraits::template TStrings<arrow::BinaryType, true, EPgStringType::Text>>();
-                ret->SetPgBuilder(pgBuilder, desc->Typelen, typePg.GetTypeId());
-                return ret;
-            } else {
-                auto ret = std::make_unique<typename TTraits::template TStrings<arrow::BinaryType, true, EPgStringType::CString>>();
-                ret->SetPgBuilder(pgBuilder, desc->Typelen, typePg.GetTypeId());
-                return ret;
-            }
-        }
+        return TTraits::MakePg(*desc, pgBuilder);        
     }
 
     Y_ENSURE(false, "Unsupported type");
diff --git a/ydb/library/yql/public/udf/arrow/block_type_helper.h b/ydb/library/yql/public/udf/arrow/block_type_helper.h
new file mode 100644
index 00000000000..9eb7fc9bf9d
--- /dev/null
+++ b/ydb/library/yql/public/udf/arrow/block_type_helper.h
@@ -0,0 +1,18 @@
+#pragma once
+#include "block_item_comparator.h"
+#include <ydb/library/yql/public/udf/udf_type_size_check.h>
+
+namespace NYql {
+namespace NUdf {
+
+// ABI stable
+class IBlockTypeHelper {
+public:
+    virtual ~IBlockTypeHelper() = default;
+    virtual IBlockItemComparator::TPtr MakeComparator(TType* type) const = 0;
+};
+
+UDF_ASSERT_TYPE_SIZE(IBlockTypeHelper, 8);
+
+}
+}
diff --git a/ydb/library/yql/public/udf/udf_type_builder.h b/ydb/library/yql/public/udf/udf_type_builder.h
index 30597ed836f..ef27298ef42 100644
--- a/ydb/library/yql/public/udf/udf_type_builder.h
+++ b/ydb/library/yql/public/udf/udf_type_builder.h
@@ -644,7 +644,18 @@ public:
 };
 #endif
 
-#if UDF_ABI_COMPATIBILITY_VERSION_CURRENT >= UDF_ABI_COMPATIBILITY_VERSION(2, 28)
+#if UDF_ABI_COMPATIBILITY_VERSION_CURRENT >= UDF_ABI_COMPATIBILITY_VERSION(2, 32)
+class IBlockTypeHelper;
+
+class IFunctionTypeInfoBuilder16: public IFunctionTypeInfoBuilder15 {
+public:
+    virtual const IBlockTypeHelper& IBlockTypeHelper() const = 0;
+};
+#endif
+
+#if UDF_ABI_COMPATIBILITY_VERSION_CURRENT >= UDF_ABI_COMPATIBILITY_VERSION(2, 32)
+using IFunctionTypeInfoBuilderImpl = IFunctionTypeInfoBuilder16;
+#elif UDF_ABI_COMPATIBILITY_VERSION_CURRENT >= UDF_ABI_COMPATIBILITY_VERSION(2, 28)
 using IFunctionTypeInfoBuilderImpl = IFunctionTypeInfoBuilder15;
 #elif UDF_ABI_COMPATIBILITY_VERSION_CURRENT >= UDF_ABI_COMPATIBILITY_VERSION(2, 26)
 using IFunctionTypeInfoBuilderImpl = IFunctionTypeInfoBuilder14;
diff --git a/ydb/library/yql/public/udf/udf_version.h b/ydb/library/yql/public/udf/udf_version.h
index 36023d6ecab..e1d13de0643 100644
--- a/ydb/library/yql/public/udf/udf_version.h
+++ b/ydb/library/yql/public/udf/udf_version.h
@@ -7,7 +7,7 @@ namespace NYql {
 namespace NUdf {
 
 #define CURRENT_UDF_ABI_VERSION_MAJOR 2
-#define CURRENT_UDF_ABI_VERSION_MINOR 31
+#define CURRENT_UDF_ABI_VERSION_MINOR 32
 #define CURRENT_UDF_ABI_VERSION_PATCH 0
 
 #ifdef USE_CURRENT_UDF_ABI_VERSION
diff --git a/ydb/library/yql/sql/pg_dummy/pg_sql_dummy.cpp b/ydb/library/yql/sql/pg_dummy/pg_sql_dummy.cpp
index 66e68d5db1c..9ec3e2427d5 100644
--- a/ydb/library/yql/sql/pg_dummy/pg_sql_dummy.cpp
+++ b/ydb/library/yql/sql/pg_dummy/pg_sql_dummy.cpp
@@ -219,6 +219,11 @@ NUdf::IEquate::TPtr MakePgEquate(const NMiniKQL::TPgType* type) {
     throw yexception() << "PG types are not supported";
 }
 
+NUdf::IBlockItemComparator::TPtr MakePgItemComparator(ui32 typeId) {
+    Y_UNUSED(typeId);
+    throw yexception() << "PG types are not supported";
+}
+
 } // namespace NMiniKQL
 } // namespace NKikimr