Nullable parameter for readers, converters from UV to BlockItem, comparators over BlockItem

8 files changed, 427 insertions, 45 deletions

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 a73c7cabf25..fb6c555b324 100644
--- a/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.cpp
+++ b/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.cpp
@@ -14,36 +14,71 @@ namespace NMiniKQL {
 
 namespace {
 
-template <typename T>
+template <typename T, bool Nullable>
 class TFixedSizeBlockItemConverter : public IBlockItemConverter {
 public:
     NUdf::TUnboxedValuePod MakeValue(TBlockItem item, const THolderFactory& holderFactory) const final {
         Y_UNUSED(holderFactory);
-        return item ? NUdf::TUnboxedValuePod(item.As<T>()) : NUdf::TUnboxedValuePod{};
+        if constexpr (Nullable) {
+            if (!item) {
+                return {};
+            }
+        }
+
+        return NUdf::TUnboxedValuePod(item.As<T>());
+    }
+
+    TBlockItem MakeItem(NUdf::TUnboxedValuePod value) const final {
+        if constexpr (Nullable) {
+            if (!value) {
+                return {};
+            }
+        }
+
+        return TBlockItem(value.Get<T>());
     }
 };
 
-template<typename TStringType>
+template<typename TStringType, bool Nullable>
 class TStringBlockItemConverter : public IBlockItemConverter {
 public:
     NUdf::TUnboxedValuePod MakeValue(TBlockItem item, const THolderFactory& holderFactory) const final {
         Y_UNUSED(holderFactory);
-        if (!item) {
-            return {};
+        if constexpr (Nullable) {
+            if (!item) {
+                return {};
+            }
         }
+
         return MakeString(item.AsStringRef());
     }
+
+    TBlockItem MakeItem(NUdf::TUnboxedValuePod value) const final {
+        if constexpr (Nullable) {
+            if (!value) {
+                return {};
+            }
+        }
+
+        return TBlockItem(value.AsStringRef());
+    }
 };
 
+template <bool Nullable>
 class TTupleBlockItemConverter : public IBlockItemConverter {
 public:
     TTupleBlockItemConverter(TVector<std::unique_ptr<IBlockItemConverter>>&& children)
         : Children(std::move(children))
-    {}
+    {
+        Items.resize(Children.size());
+        Unboxed.resize(Children.size());
+    }
 
     NUdf::TUnboxedValuePod MakeValue(TBlockItem item, const THolderFactory& holderFactory) const final {
-        if (!item) {
-            return {};
+        if constexpr (Nullable) {
+            if (!item) {
+                return {};
+            }
         }
 
         NUdf::TUnboxedValue* values;
@@ -56,9 +91,34 @@ public:
         return result;
     }
 
+    TBlockItem MakeItem(NUdf::TUnboxedValuePod value) const final {
+        if constexpr (Nullable) {
+            if (!value) {
+                return {};
+            }
+        }
+
+        auto elements = value.GetElements();
+        if (!elements) {
+            for (ui32 i = 0; i < Children.size(); ++i) {
+                Unboxed[i] = value.GetElement(i);
+            }
+
+            elements = Unboxed.data();
+        }
+
+        for (ui32 i = 0; i < Children.size(); ++i) {
+            Items[i] = Children[i]->MakeItem(elements[i]);
+        }
+
+        return TBlockItem{ Items.data() };
+    }
+
 private:
     const TVector<std::unique_ptr<IBlockItemConverter>> Children;
     mutable TPlainContainerCache Cache;
+    mutable TVector<NUdf::TUnboxedValue> Unboxed;
+    mutable TVector<TBlockItem> Items;
 };
 
 class TExternalOptionalBlockItemConverter : public IBlockItemConverter {
@@ -74,17 +134,26 @@ public:
         return Inner->MakeValue(item.GetOptionalValue(), holderFactory).MakeOptional();
     }
 
+    TBlockItem MakeItem(NUdf::TUnboxedValuePod value) const final {
+        if (!value) {
+            return {};
+        }
+
+        return Inner->MakeItem(value.GetOptionalValue()).MakeOptional();
+    }
+
 private:
     const std::unique_ptr<IBlockItemConverter> Inner;
 };
 
 struct TConverterTraits {
     using TResult = IBlockItemConverter;
-    using TTuple = TTupleBlockItemConverter;
-    template <typename T>
-    using TFixedSize = TFixedSizeBlockItemConverter<T>;
-    template <typename TStringType>
-    using TStrings = TStringBlockItemConverter<TStringType>;
+    template <bool Nullable>
+    using TTuple = TTupleBlockItemConverter<Nullable>;
+    template <typename T, bool Nullable>
+    using TFixedSize = TFixedSizeBlockItemConverter<T, Nullable>;
+    template <typename TStringType, bool Nullable>
+    using TStrings = TStringBlockItemConverter<TStringType, Nullable>;
     using TExtOptional = TExternalOptionalBlockItemConverter;
 };
 
diff --git a/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.h b/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.h
index 3e51769e7f7..50cf5d51277 100644
--- a/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.h
+++ b/ydb/library/yql/minikql/comp_nodes/mkql_block_reader.h
@@ -19,6 +19,7 @@ public:
     virtual ~IBlockItemConverter() = default;
 
     virtual NUdf::TUnboxedValuePod MakeValue(TBlockItem item, const THolderFactory& holderFactory) const = 0;
+    virtual TBlockItem MakeItem(NUdf::TUnboxedValuePod value) const = 0;
 };
 
 using NYql::NUdf::MakeBlockReader;
diff --git a/ydb/library/yql/public/udf/arrow/CMakeLists.darwin.txt b/ydb/library/yql/public/udf/arrow/CMakeLists.darwin.txt
index c928bed6eaf..1b41c0f06e9 100644
--- a/ydb/library/yql/public/udf/arrow/CMakeLists.darwin.txt
+++ b/ydb/library/yql/public/udf/arrow/CMakeLists.darwin.txt
@@ -24,4 +24,5 @@ target_sources(public-udf-arrow PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/util.cpp
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_reader.cpp
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_item.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_item_comparator.cpp
 )
diff --git a/ydb/library/yql/public/udf/arrow/CMakeLists.linux-aarch64.txt b/ydb/library/yql/public/udf/arrow/CMakeLists.linux-aarch64.txt
index 4c3155d270b..59ba7f29325 100644
--- a/ydb/library/yql/public/udf/arrow/CMakeLists.linux-aarch64.txt
+++ b/ydb/library/yql/public/udf/arrow/CMakeLists.linux-aarch64.txt
@@ -25,4 +25,5 @@ target_sources(public-udf-arrow PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/util.cpp
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_reader.cpp
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_item.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_item_comparator.cpp
 )
diff --git a/ydb/library/yql/public/udf/arrow/CMakeLists.linux.txt b/ydb/library/yql/public/udf/arrow/CMakeLists.linux.txt
index 4c3155d270b..59ba7f29325 100644
--- a/ydb/library/yql/public/udf/arrow/CMakeLists.linux.txt
+++ b/ydb/library/yql/public/udf/arrow/CMakeLists.linux.txt
@@ -25,4 +25,5 @@ target_sources(public-udf-arrow PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/util.cpp
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_reader.cpp
   ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_item.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/library/yql/public/udf/arrow/block_item_comparator.cpp
 )
diff --git a/ydb/library/yql/public/udf/arrow/block_item_comparator.cpp b/ydb/library/yql/public/udf/arrow/block_item_comparator.cpp
new file mode 100644
index 00000000000..34965bb05c2
--- /dev/null
+++ b/ydb/library/yql/public/udf/arrow/block_item_comparator.cpp
@@ -0,0 +1 @@
+#include "block_item_comparator.h"
diff --git a/ydb/library/yql/public/udf/arrow/block_item_comparator.h b/ydb/library/yql/public/udf/arrow/block_item_comparator.h
new file mode 100644
index 00000000000..93764dc6bad
--- /dev/null
+++ b/ydb/library/yql/public/udf/arrow/block_item_comparator.h
@@ -0,0 +1,265 @@
+#pragma once
+
+#include "block_item.h"
+#include "block_reader.h"
+
+#include <ydb/library/yql/public/udf/udf_type_inspection.h>
+
+namespace NYql::NUdf {
+
+class IBlockItemComparator {
+public:
+    virtual ~IBlockItemComparator() = default;
+
+    virtual i64 Compare(TBlockItem lhs, TBlockItem rhs) const = 0;
+    virtual bool Equals(TBlockItem lhs, TBlockItem rhs) const = 0;
+    virtual bool Less(TBlockItem lhs, TBlockItem rhs) const = 0;
+    virtual bool Greater(TBlockItem lhs, TBlockItem rhs) const = 0;
+};
+
+
+template <typename TDerived, bool Nullable>
+class TBlockItemComparatorBase : public IBlockItemComparator {
+public:
+    const TDerived* Derived() const {
+        return static_cast<const TDerived*>(this);
+    }
+
+    // returns <0 if lhs < rhs
+    i64 Compare(TBlockItem lhs, TBlockItem rhs) const final {
+        if constexpr (Nullable) {
+            if (lhs) {
+                if (rhs) {
+                    return Derived()->DoCompare(lhs, rhs);
+                } else {
+                    return +1;
+                }
+            } else {
+                if (rhs) {
+                    return -1;
+                } else {
+                    return 0;
+                }
+            }
+        } else {
+            return Derived()->DoCompare(lhs, rhs);
+        }
+    }
+
+    bool Equals(TBlockItem lhs, TBlockItem rhs) const final {
+        if constexpr (Nullable) {
+            if (lhs) {
+                if (rhs) {
+                    return Derived()->DoEquals(lhs, rhs);
+                } else {
+                    return false;
+                }
+            } else {
+                if (rhs) {
+                    return false;
+                } else {
+                    return true;
+                }
+            }
+        } else {
+            return Derived()->DoEquals(lhs, rhs);
+        }
+    }
+
+    bool Less(TBlockItem lhs, TBlockItem rhs) const final {
+        if constexpr (Nullable) {
+            if (lhs) {
+                if (rhs) {
+                    return Derived()->DoLess(lhs, rhs);
+                }
+                else {
+                    return false;
+                }
+            } else {
+                if (rhs) {
+                    return true;
+                } else {
+                    return false;
+                }
+            }
+        } else {
+            return Derived()->DoLess(lhs, rhs);
+        }
+    }
+
+    bool Greater(TBlockItem lhs, TBlockItem rhs) const final {
+        if constexpr (Nullable) {
+            if (lhs) {
+                if (rhs) {
+                    return Derived()->DoGreater(lhs, rhs);
+                } else {
+                    return true;
+                }
+            } else {
+                if (rhs) {
+                    return false;
+                } else {
+                    return false;
+                }
+            }
+        } else {
+            return Derived()->DoGreater(lhs, rhs);
+        }
+    }
+};
+
+template <typename T, bool Nullable>
+class TFixedSizeBlockItemComparator : public TBlockItemComparatorBase<TFixedSizeBlockItemComparator<T, Nullable>, Nullable> {
+public:
+    i64 DoCompare(TBlockItem lhs, TBlockItem rhs) const {
+        if constexpr (std::is_integral<T>::value && sizeof(T) < sizeof(i64)) {
+            return i64(lhs.As<T>()) - i64(rhs.As<T>());
+        } else {
+            if constexpr (std::is_floating_point<T>::value) {
+                if (std::isunordered(lhs.As<T>(), rhs.As<T>())) {
+                    return i64(std::isnan(lhs.As<T>())) - i64(std::isnan(rhs.As<T>()));
+                }
+            }
+
+            return (lhs.As<T>() > rhs.As<T>()) - (lhs.As<T>() < rhs.As<T>());
+        }
+    }
+
+    bool DoEquals(TBlockItem lhs, TBlockItem rhs) const {
+        return lhs.As<T>() == rhs.As<T>();
+    }
+
+    bool DoLess(TBlockItem lhs, TBlockItem rhs) const {
+        return lhs.As<T>() < rhs.As<T>();
+    }
+
+    bool DoGreater(TBlockItem lhs, TBlockItem rhs) const {
+        return lhs.As<T>() > rhs.As<T>();
+    }
+};
+
+template <typename TStringType, bool Nullable>
+class TStringBlockItemComparator : public TBlockItemComparatorBase<TStringBlockItemComparator<TStringType, Nullable>, Nullable> {
+public:
+    i64 DoCompare(TBlockItem lhs, TBlockItem rhs) const {
+        return lhs.AsStringRef().Compare(rhs.AsStringRef());
+    }
+
+    bool DoEquals(TBlockItem lhs, TBlockItem rhs) const {
+        return lhs.AsStringRef() == rhs.AsStringRef();
+    }
+
+    bool DoLess(TBlockItem lhs, TBlockItem rhs) const {
+        return lhs.AsStringRef() < rhs.AsStringRef();
+    }
+
+    bool DoGreater(TBlockItem lhs, TBlockItem rhs) const {
+        return lhs.AsStringRef() > rhs.AsStringRef();
+    }
+};
+
+template <bool Nullable>
+class TTupleBlockItemComparator : public TBlockItemComparatorBase<TTupleBlockItemComparator<Nullable>, Nullable> {
+public:
+    TTupleBlockItemComparator(TVector<std::unique_ptr<IBlockItemComparator>>&& children)
+        : Children_(std::move(children))
+    {}
+
+public:
+    i64 DoCompare(TBlockItem lhs, TBlockItem rhs) const {
+        for (ui32 i = 0; i < Children_.size(); ++i) {
+            auto res = Children_[i]->Compare(lhs.AsTuple()[i], rhs.AsTuple()[i]);
+            if (res != 0) {
+                return res;
+            }
+        }
+
+        return 0;
+    }
+
+    bool DoEquals(TBlockItem lhs, TBlockItem rhs) const {
+        for (ui32 i = 0; i < Children_.size(); ++i) {
+            if (!Children_[i]->Equals(lhs.AsTuple()[i], rhs.AsTuple()[i])) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    bool DoLess(TBlockItem lhs, TBlockItem rhs) const {
+        for (ui32 i = 0; i < Children_.size(); ++i) {
+            auto res = Children_[i]->Compare(lhs.AsTuple()[i], rhs.AsTuple()[i]);
+            if (res < 0) {
+                return true;
+            }
+
+            if (res > 0) {
+                return false;
+            }
+        }
+
+        return false;
+    }
+
+    bool DoGreater(TBlockItem lhs, TBlockItem rhs) const {
+        for (ui32 i = 0; i < Children_.size(); ++i) {
+            auto res = Children_[i]->Compare(lhs.AsTuple()[i], rhs.AsTuple()[i]);
+            if (res > 0) {
+                return true;
+            }
+
+            if (res < 0) {
+                return false;
+            }
+        }
+
+        return false;
+    }
+
+private:
+    const TVector<std::unique_ptr<IBlockItemComparator>> Children_;
+};
+
+class TExternalOptionalBlockItemComparator : public TBlockItemComparatorBase<TExternalOptionalBlockItemComparator, true> {
+public:
+    TExternalOptionalBlockItemComparator(std::unique_ptr<IBlockItemComparator> inner)
+        : Inner_(std::move(inner))
+    {}
+
+    i64 DoCompare(TBlockItem lhs, TBlockItem rhs) const {
+        return Inner_->Compare(lhs.GetOptionalValue(), rhs.GetOptionalValue());
+    }
+
+    bool DoEquals(TBlockItem lhs, TBlockItem rhs) const {
+        return Inner_->Equals(lhs.GetOptionalValue(), rhs.GetOptionalValue());
+    }
+
+    bool DoLess(TBlockItem lhs, TBlockItem rhs) const {
+        return Inner_->Less(lhs.GetOptionalValue(), rhs.GetOptionalValue());
+    }
+
+    bool DoGreater(TBlockItem lhs, TBlockItem rhs) const {
+        return Inner_->Greater(lhs.GetOptionalValue(), rhs.GetOptionalValue());
+    }
+
+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>
+    using TStrings = TStringBlockItemComparator<TStringType, Nullable>;
+    using TExtOptional = TExternalOptionalBlockItemComparator;
+};
+
+inline std::unique_ptr<IBlockItemComparator> MakeBlockItemComparator(const ITypeInfoHelper& typeInfoHelper, const TType* type) {
+    return MakeBlockReaderImpl<TComparatorTraits>(typeInfoHelper, type);
+}
+
+}
diff --git a/ydb/library/yql/public/udf/arrow/block_reader.h b/ydb/library/yql/public/udf/arrow/block_reader.h
index f998b9c357d..def4dadd4c2 100644
--- a/ydb/library/yql/public/udf/arrow/block_reader.h
+++ b/ydb/library/yql/public/udf/arrow/block_reader.h
@@ -17,35 +17,41 @@ public:
     virtual TBlockItem GetScalarItem(const arrow::Scalar& scalar) = 0;
 };
 
-template <typename T>
+template <typename T, bool Nullable>
 class TFixedSizeBlockReader : public IBlockReader {
 public:
     TBlockItem GetItem(const arrow::ArrayData& data, size_t index) final {
-        if (IsNull(data, index)) {
-            return {};
+        if constexpr (Nullable) {
+            if (IsNull(data, index)) {
+                return {};
+            }
         }
 
         return TBlockItem(data.GetValues<T>(1)[index]);
     }
 
     TBlockItem GetScalarItem(const arrow::Scalar& scalar) final {
-        if (!scalar.is_valid) {
-            return {};
+        if constexpr (Nullable) {
+            if (!scalar.is_valid) {
+                return {};
+            }
         }
 
         return TBlockItem(*static_cast<const T*>(arrow::internal::checked_cast<const arrow::internal::PrimitiveScalarBase&>(scalar).data()));
     }
 };
 
-template<typename TStringType>
+template<typename TStringType, bool Nullable>
 class TStringBlockReader : public IBlockReader {
 public:
     using TOffset = typename TStringType::offset_type;
 
     TBlockItem GetItem(const arrow::ArrayData& data, size_t index) final {
         Y_VERIFY_DEBUG(data.buffers.size() == 3);
-        if (IsNull(data, index)) {
-            return {};
+        if constexpr (Nullable) {
+            if (IsNull(data, index)) {
+                return {};
+            }
         }
 
         const TOffset* offsets = data.GetValues<TOffset>(1);
@@ -56,8 +62,10 @@ public:
     }
 
     TBlockItem GetScalarItem(const arrow::Scalar& scalar) final {
-        if (!scalar.is_valid) {
-            return {};
+        if constexpr (Nullable) {
+            if (!scalar.is_valid) {
+                return {};
+            }
         }
 
         auto buffer = arrow::internal::checked_cast<const arrow::BaseBinaryScalar&>(scalar).value;
@@ -66,6 +74,7 @@ public:
     }
 };
 
+template <bool Nullable>
 class TTupleBlockReader : public IBlockReader {
 public:
     TTupleBlockReader(TVector<std::unique_ptr<IBlockReader>>&& children)
@@ -74,8 +83,10 @@ public:
     {}
 
     TBlockItem GetItem(const arrow::ArrayData& data, size_t index) final {
-        if (IsNull(data, index)) {
-            return {};
+        if constexpr (Nullable) {
+            if (IsNull(data, index)) {
+                return {};
+            }
         }
 
         for (ui32 i = 0; i < Children.size(); ++i) {
@@ -86,8 +97,10 @@ public:
     }
 
     TBlockItem GetScalarItem(const arrow::Scalar& scalar) final {
-        if (!scalar.is_valid) {
-            return {};
+        if constexpr (Nullable) {
+            if (!scalar.is_valid) {
+                return {};
+            }
         }
 
         const auto& structScalar = arrow::internal::checked_cast<const arrow::StructScalar&>(scalar);
@@ -133,20 +146,50 @@ private:
 
 struct TReaderTraits {
     using TResult = IBlockReader;
-    using TTuple = TTupleBlockReader;
-    template <typename T>
-    using TFixedSize = TFixedSizeBlockReader<T>;
-    template <typename TStringType>
-    using TStrings = TStringBlockReader<TStringType>;
+    template <bool Nullable>
+    using TTuple = TTupleBlockReader<Nullable>;
+    template <typename T, bool Nullable>
+    using TFixedSize = TFixedSizeBlockReader<T, Nullable>;
+    template <typename TStringType, bool Nullable>
+    using TStrings = TStringBlockReader<TStringType, Nullable>;
     using TExtOptional = TExternalOptionalBlockReader;
 };
 
 template <typename TTraits>
+std::unique_ptr<typename TTraits::TResult> MakeTupleBlockReaderImpl(bool isOptional, TVector<std::unique_ptr<typename TTraits::TResult>>&& children) {
+    if (isOptional) {
+        return std::make_unique<typename TTraits::template TTuple<true>>(std::move(children));
+    } else {
+        return std::make_unique<typename TTraits::template TTuple<false>>(std::move(children));
+    }
+}
+
+template <typename TTraits, typename T>
+std::unique_ptr<typename TTraits::TResult> MakeFixedSizeBlockReaderImpl(bool isOptional) {
+    if (isOptional) {
+        return std::make_unique<typename TTraits::template TFixedSize<T, true>>();
+    } else {
+        return std::make_unique<typename TTraits::template TFixedSize<T, false>>();
+    }
+}
+
+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>>();
+    } else {
+        return std::make_unique<typename TTraits::template TStrings<T, false>>();
+    }
+}
+
+template <typename TTraits>
 std::unique_ptr<typename TTraits::TResult> MakeBlockReaderImpl(const ITypeInfoHelper& typeInfoHelper, const TType* type) {
     const TType* unpacked = type;
     TOptionalTypeInspector typeOpt(typeInfoHelper, type);
+    bool isOptional = false;
     if (typeOpt) {
         unpacked = typeOpt.GetItemType();
+        isOptional = true;
     }
 
     TOptionalTypeInspector unpackedOpt(typeInfoHelper, unpacked);
@@ -184,7 +227,7 @@ std::unique_ptr<typename TTraits::TResult> MakeBlockReaderImpl(const ITypeInfoHe
             children.emplace_back(MakeBlockReaderImpl<TTraits>(typeInfoHelper, typeTuple.GetElementType(i)));
         }
 
-        return std::make_unique<typename TTraits::TTuple>(std::move(children));
+        return MakeTupleBlockReaderImpl<TTraits>(isOptional, std::move(children));
     }
 
     TDataTypeInspector typeData(typeInfoHelper, type);
@@ -192,34 +235,34 @@ std::unique_ptr<typename TTraits::TResult> MakeBlockReaderImpl(const ITypeInfoHe
         auto typeId = typeData.GetTypeId();
         switch (GetDataSlot(typeId)) {
         case NUdf::EDataSlot::Int8:
-            return std::make_unique<typename TTraits::template TFixedSize<i8>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, i8>(isOptional);
         case NUdf::EDataSlot::Bool:
         case NUdf::EDataSlot::Uint8:
-            return std::make_unique<typename TTraits::template TFixedSize<ui8>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, ui8>(isOptional);
         case NUdf::EDataSlot::Int16:
-            return std::make_unique<typename TTraits::template TFixedSize<i16>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, i16>(isOptional);
         case NUdf::EDataSlot::Uint16:
         case NUdf::EDataSlot::Date:
-            return std::make_unique<typename TTraits::template TFixedSize<ui16>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, ui16>(isOptional);
         case NUdf::EDataSlot::Int32:
-            return std::make_unique<typename TTraits::template TFixedSize<i32>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, i32>(isOptional);
         case NUdf::EDataSlot::Uint32:
         case NUdf::EDataSlot::Datetime:
-            return std::make_unique<typename TTraits::template TFixedSize<ui32>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, ui32>(isOptional);
         case NUdf::EDataSlot::Int64:
         case NUdf::EDataSlot::Interval:
-            return std::make_unique<typename TTraits::template TFixedSize<i64>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, i64>(isOptional);
         case NUdf::EDataSlot::Uint64:
         case NUdf::EDataSlot::Timestamp:
-            return std::make_unique<typename TTraits::template TFixedSize<ui64>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, ui64>(isOptional);
         case NUdf::EDataSlot::Float:
-            return std::make_unique<typename TTraits::template TFixedSize<float>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, float>(isOptional);
         case NUdf::EDataSlot::Double:
-            return std::make_unique<typename TTraits::template TFixedSize<double>>();
+            return MakeFixedSizeBlockReaderImpl<TTraits, double>(isOptional);
         case NUdf::EDataSlot::String:
-            return std::make_unique<typename TTraits::template TStrings<arrow::BinaryType>>();
+            return MakeStringBlockReaderImpl<TTraits, arrow::BinaryType>(isOptional);
         case NUdf::EDataSlot::Utf8:
-            return std::make_unique<typename TTraits::template TStrings<arrow::StringType>>();
+            return MakeStringBlockReaderImpl<TTraits, arrow::StringType>(isOptional);
         default:
             Y_ENSURE(false, "Unsupported data slot");
         }