From 8303fa39376d3ba5f9ad723ea8106c3d6bb2d822 Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Tue, 26 Dec 2023 14:40:09 +0100
Subject: [PATCH 01/11] Implement double/int -> pg numeric conversion YQL-16767

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp | 83 +++++++++++++++++++++
 1 file changed, 83 insertions(+)

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index 3c33570d8142..2bc82487f8fa 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -13,6 +13,7 @@
 extern "C" {
 #include "utils/date.h"
 #include "utils/timestamp.h"
+#include "utils/numeric.h"
 }
 
 namespace NYql {
@@ -151,6 +152,85 @@ std::shared_ptr<arrow::Array> PgConvertString(const std::shared_ptr<arrow::Array
     return ret;
 }
 
+Numeric PgFloatToNumeric(double item, ui64 scale, int digits) {
+    double int_part, frac_part;
+    bool error;
+    frac_part = modf(item, &int_part);
+    i64 frac_int = round(frac_part * scale);
+
+    // scale compaction: represent 711.56000 as 711.56
+    while (frac_int && frac_int % 10 == 0) {
+        frac_int /= 10;
+        digits -= 1;
+    }
+
+    return numeric_add_opt_error(
+        int64_to_numeric(int_part),
+        int64_div_fast_to_numeric(frac_int, digits),
+        &error);
+}
+
+template<typename T>
+std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Array>& value) {
+    TArenaMemoryContext arena;
+    const auto& data = value->data();
+    size_t length = data->length;
+    arrow::BinaryBuilder builder;
+    std::vector<char> tmp;
+    auto input = data->GetValues<T>(1);
+    for (size_t i = 0; i < length; ++i) {
+        if (value->IsNull(i)) {
+            builder.AppendNull();
+            continue;
+        }
+        T item = input[i];
+        Numeric v;
+        if constexpr(std::is_same_v<T,double>) {
+            v = PgFloatToNumeric(item, 1000000000000LL, 12);
+        } else if constexpr(std::is_same_v<T,float>) {
+            v = PgFloatToNumeric(item, 1000000LL, 6);
+        } else {
+            v = int64_to_numeric(item);
+        }
+        auto datum = NumericGetDatum(v);
+        auto ptr = (char*)datum;
+        auto len = GetFullVarSize((const text*)datum);
+        NUdf::ZeroMemoryContext(ptr);
+        ARROW_OK(builder.Append(ptr - sizeof(void*), len + sizeof(void*)));
+    }
+
+    std::shared_ptr<arrow::BinaryArray> ret;
+    ARROW_OK(builder.Finish(&ret));
+    return ret;
+}
+
+TColumnConverter BuildPgNumericColumnConverter(const std::shared_ptr<arrow::DataType>& originalType) {
+    switch (originalType->id()) {
+    case arrow::Type::INT16:
+        return [](const std::shared_ptr<arrow::Array>& value) {
+            return PgConvertNumeric<i16>(value);
+        };
+    case arrow::Type::INT32:
+        return [](const std::shared_ptr<arrow::Array>& value) {
+            return PgConvertNumeric<i32>(value);
+        };
+    case arrow::Type::INT64:
+        return [](const std::shared_ptr<arrow::Array>& value) {
+            return PgConvertNumeric<i64>(value);
+        };
+    case arrow::Type::FLOAT:
+        return [](const std::shared_ptr<arrow::Array>& value) {
+            return PgConvertNumeric<float>(value);
+        };
+    case arrow::Type::DOUBLE:
+        return [](const std::shared_ptr<arrow::Array>& value) {
+            return PgConvertNumeric<double>(value);
+        };
+    default:
+        return {};
+    }
+}
+
 template <typename T, typename F>
 TColumnConverter BuildPgFixedColumnConverter(const std::shared_ptr<arrow::DataType>& originalType, const F& f) {
     auto primaryType = NKikimr::NMiniKQL::GetPrimitiveDataType<T>();
@@ -199,6 +279,9 @@ TColumnConverter BuildPgColumnConverter(const std::shared_ptr<arrow::DataType>&
     }
     case FLOAT8OID: {
         return BuildPgFixedColumnConverter<double>(originalType, [](auto value){ return Float8GetDatum(value); });
+    }
+	case NUMERICOID: {
+        return BuildPgNumericColumnConverter(originalType);
     }
     case BYTEAOID:
     case VARCHAROID:

From 461f6966960f80438e9d47f0e41bedcdb689cede Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Tue, 26 Dec 2023 14:41:09 +0100
Subject: [PATCH 02/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index 2bc82487f8fa..8a95de51c686 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -280,7 +280,7 @@ TColumnConverter BuildPgColumnConverter(const std::shared_ptr<arrow::DataType>&
     case FLOAT8OID: {
         return BuildPgFixedColumnConverter<double>(originalType, [](auto value){ return Float8GetDatum(value); });
     }
-	case NUMERICOID: {
+    case NUMERICOID: {
         return BuildPgNumericColumnConverter(originalType);
     }
     case BYTEAOID:

From 6c5c7d47ccaf406670ee82a95f45f4ca4cd8066b Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Tue, 26 Dec 2023 14:45:46 +0100
Subject: [PATCH 03/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp | 1 -
 1 file changed, 1 deletion(-)

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index 8a95de51c686..5f5ca9cb9663 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -176,7 +176,6 @@ std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Arra
     const auto& data = value->data();
     size_t length = data->length;
     arrow::BinaryBuilder builder;
-    std::vector<char> tmp;
     auto input = data->GetValues<T>(1);
     for (size_t i = 0; i < length; ++i) {
         if (value->IsNull(i)) {

From 8d67e089782d13098270b1e37c85b576cc25765b Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Tue, 26 Dec 2023 14:49:40 +0100
Subject: [PATCH 04/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index 5f5ca9cb9663..1474e566c434 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -153,20 +153,20 @@ std::shared_ptr<arrow::Array> PgConvertString(const std::shared_ptr<arrow::Array
 }
 
 Numeric PgFloatToNumeric(double item, ui64 scale, int digits) {
-    double int_part, frac_part;
+    double intPart, fracPart;
     bool error;
-    frac_part = modf(item, &int_part);
-    i64 frac_int = round(frac_part * scale);
+    fracPart = modf(item, &intPart);
+    i64 fracInt = round(fracPart * scale);
 
     // scale compaction: represent 711.56000 as 711.56
-    while (frac_int && frac_int % 10 == 0) {
-        frac_int /= 10;
+    while (fracInt && fracInt % 10 == 0) {
+        fracInt /= 10;
         digits -= 1;
     }
 
     return numeric_add_opt_error(
-        int64_to_numeric(int_part),
-        int64_div_fast_to_numeric(frac_int, digits),
+        int64_to_numeric(intPart),
+        int64_div_fast_to_numeric(fracInt, digits),
         &error);
 }
 

From 39fc63dc356f696b3220cd6536652225e8dd646c Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Wed, 27 Dec 2023 12:06:03 +0100
Subject: [PATCH 05/11] Add tests

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp   |  14 ++-
 .../yql/parser/pg_wrapper/ut/arrow_ut.cpp     | 116 ++++++++++++++++++
 ydb/library/yql/parser/pg_wrapper/ut/ya.make  |   1 +
 3 files changed, 126 insertions(+), 5 deletions(-)
 create mode 100644 ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index 1474e566c434..899bb3794c34 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -159,15 +159,19 @@ Numeric PgFloatToNumeric(double item, ui64 scale, int digits) {
     i64 fracInt = round(fracPart * scale);
 
     // scale compaction: represent 711.56000 as 711.56
-    while (fracInt && fracInt % 10 == 0) {
+    while (digits > 0 && fracInt % 10 == 0) {
         fracInt /= 10;
         digits -= 1;
     }
 
-    return numeric_add_opt_error(
-        int64_to_numeric(intPart),
-        int64_div_fast_to_numeric(fracInt, digits),
-        &error);
+    if (digits == 0) {
+        return int64_to_numeric(int_part);
+    } else {
+        return numeric_add_opt_error(
+            int64_to_numeric(intPart),
+            int64_div_fast_to_numeric(fracInt, digits),
+            &error);
+    }
 }
 
 template<typename T>
diff --git a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
new file mode 100644
index 000000000000..9e5cfeecef8c
--- /dev/null
+++ b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
@@ -0,0 +1,116 @@
+#include <arrow/api.h>
+#include <arrow/array.h>
+
+#include <library/cpp/testing/unittest/registar.h>
+
+#include "arrow.h"
+
+extern "C" {
+#include "utils/numeric.h"
+
+Datum numeric_out(PG_FUNCTION_ARGS);
+}
+
+namespace NYql {
+
+Numeric PgFloatToNumeric(double item, ui64 scale, int digits);
+
+template<typename T>
+std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Array>& value);
+
+Y_UNIT_TEST_SUITE(TArrowUtilsTests) {
+
+Y_UNIT_TEST(TestPgFloatToNumeric) {
+    TArenaMemoryContext arena;
+    auto n = PgFloatToNumeric(711.56, 1000000000000LL, 12);
+    auto value = TString(DatumGetCString(DirectFunctionCall1(numeric_out, NumericGetDatum(n))));
+    UNIT_ASSERT_VALUES_EQUAL(value, "711.56");
+
+    n = PgFloatToNumeric(-711.56, 1000000000000LL, 12);
+    value = TString(DatumGetCString(DirectFunctionCall1(numeric_out, NumericGetDatum(n))));
+    UNIT_ASSERT_VALUES_EQUAL(value, "-711.56");
+
+    n = PgFloatToNumeric(711.56f, 100000LL, 5);
+    value = TString(DatumGetCString(DirectFunctionCall1(numeric_out, NumericGetDatum(n))));
+    UNIT_ASSERT_VALUES_EQUAL(value, "711.56");
+
+    n = PgFloatToNumeric(-711.56f, 100000LL, 5);
+    value = TString(DatumGetCString(DirectFunctionCall1(numeric_out, NumericGetDatum(n))));
+    UNIT_ASSERT_VALUES_EQUAL(value, "-711.56");
+}
+
+
+Y_UNIT_TEST(PgConvertNumericDouble) {
+    TArenaMemoryContext arena;
+ 
+    arrow::DoubleBuilder builder;
+    builder.Append(1.1);
+    builder.Append(31.37);
+    builder.AppendNull();
+    builder.Append(-1.337);
+    builder.Append(0.0);
+
+    std::shared_ptr<arrow::Array> array;
+    builder.Finish(&array);
+
+    auto result = PgConvertNumeric<double>(array);
+    const auto& data = result->data();
+
+    const char* expected[] = {
+        "1.1", "31.37", nullptr, "-1.337", "0"
+    };
+    
+    NUdf::TStringBlockReader<arrow::BinaryType, true> reader;
+    for (int i = 0; i < 5; i++) {
+        auto item = reader.GetItem(*data, i);
+        if (!item) {
+            UNIT_ASSERT(expected[i] == nullptr);
+        } else {
+            const char* addr = item.AsStringRef().Data() + sizeof(void*);
+            UNIT_ASSERT_VALUES_EQUAL(
+                TString(DatumGetCString(DirectFunctionCall1(numeric_out, (Datum)addr))),
+                expected[i]
+            );
+        }
+    }
+}
+
+Y_UNIT_TEST(PgConvertNumericInt) {
+    TArenaMemoryContext arena;
+ 
+    arrow::Int64Builder builder;
+    builder.Append(11);
+    builder.Append(3137);
+    builder.AppendNull();
+    builder.Append(-1337);
+    builder.Append(0);
+
+    std::shared_ptr<arrow::Array> array;
+    builder.Finish(&array);
+
+    auto result = PgConvertNumeric<i64>(array);
+    const auto& data = result->data();
+
+    const char* expected[] = {
+        "11", "3137", nullptr, "-1337", "0"
+    };
+    
+    NUdf::TStringBlockReader<arrow::BinaryType, true> reader;
+    for (int i = 0; i < 5; i++) {
+        auto item = reader.GetItem(*data, i);
+        if (!item) {
+            UNIT_ASSERT(expected[i] == nullptr);
+        } else {
+            const char* addr = item.AsStringRef().Data() + sizeof(void*);
+            UNIT_ASSERT_VALUES_EQUAL(
+                TString(DatumGetCString(DirectFunctionCall1(numeric_out, (Datum)addr))),
+                expected[i]
+            );
+        }
+    }
+}
+
+} // Y_UNIT_TEST_SUITE(TArrowUtilsTests)
+
+} // namespace NYql
+
diff --git a/ydb/library/yql/parser/pg_wrapper/ut/ya.make b/ydb/library/yql/parser/pg_wrapper/ut/ya.make
index c4a443707204..848e33acfdf0 100644
--- a/ydb/library/yql/parser/pg_wrapper/ut/ya.make
+++ b/ydb/library/yql/parser/pg_wrapper/ut/ya.make
@@ -8,6 +8,7 @@ NO_COMPILER_WARNINGS()
 INCLUDE(../cflags.inc)
 
 SRCS(
+    arrow_ut.cpp
     codegen_ut.cpp
     error_ut.cpp
     parser_ut.cpp

From 8898b5660daaa6026a55319ce189f8458b2d7542 Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Wed, 27 Dec 2023 16:06:48 +0100
Subject: [PATCH 06/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index 899bb3794c34..3737a5b47a20 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -165,7 +165,7 @@ Numeric PgFloatToNumeric(double item, ui64 scale, int digits) {
     }
 
     if (digits == 0) {
-        return int64_to_numeric(int_part);
+        return int64_to_numeric(intPart);
     } else {
         return numeric_add_opt_error(
             int64_to_numeric(intPart),

From 8d7b46da55f655230978b21a3a97f5b2d9dc9048 Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Wed, 27 Dec 2023 16:10:52 +0100
Subject: [PATCH 07/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/arrow.h         | 6 ++++++
 ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp | 5 -----
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.h b/ydb/library/yql/parser/pg_wrapper/arrow.h
index 1ef0ead7c554..6736dd8c6dca 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.h
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.h
@@ -17,6 +17,7 @@ extern "C" {
 #include "fmgr.h"
 #include "catalog/pg_type_d.h"
 #include "catalog/pg_collation_d.h"
+#include "utils/numeric.h"
 }
 
 #include "utils.h"
@@ -1298,4 +1299,9 @@ TExecFunc FindExec(Oid oid);
 
 const NPg::TAggregateDesc& ResolveAggregation(const TString& name, NKikimr::NMiniKQL::TTupleType* tupleType, const std::vector<ui32>& argsColumns, NKikimr::NMiniKQL::TType* returnType);
 
+Numeric PgFloatToNumeric(double item, ui64 scale, int digits);
+
+template<typename T>
+std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Array>& value);
+
 }
diff --git a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
index 9e5cfeecef8c..bed40954bccc 100644
--- a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
@@ -13,11 +13,6 @@ Datum numeric_out(PG_FUNCTION_ARGS);
 
 namespace NYql {
 
-Numeric PgFloatToNumeric(double item, ui64 scale, int digits);
-
-template<typename T>
-std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Array>& value);
-
 Y_UNIT_TEST_SUITE(TArrowUtilsTests) {
 
 Y_UNIT_TEST(TestPgFloatToNumeric) {

From 3c859b931f8ca3288609c30764c7cc4136d3c6b8 Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Wed, 27 Dec 2023 16:13:39 +0100
Subject: [PATCH 08/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
index bed40954bccc..ce33968e5b7b 100644
--- a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
@@ -7,8 +7,7 @@
 
 extern "C" {
 #include "utils/numeric.h"
-
-Datum numeric_out(PG_FUNCTION_ARGS);
+#include "utils/fmgrprotos.h"
 }
 
 namespace NYql {

From fd9800cf7b54eeb15a5926212ecf46d160d7fa8c Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Wed, 27 Dec 2023 16:59:57 +0100
Subject: [PATCH 09/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp   | 33 --------------
 ydb/library/yql/parser/pg_wrapper/arrow.h     |  6 +--
 .../yql/parser/pg_wrapper/arrow_impl.h        | 44 +++++++++++++++++++
 3 files changed, 46 insertions(+), 37 deletions(-)
 create mode 100644 ydb/library/yql/parser/pg_wrapper/arrow_impl.h

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index 3737a5b47a20..e7ed9a8d8170 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -174,39 +174,6 @@ Numeric PgFloatToNumeric(double item, ui64 scale, int digits) {
     }
 }
 
-template<typename T>
-std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Array>& value) {
-    TArenaMemoryContext arena;
-    const auto& data = value->data();
-    size_t length = data->length;
-    arrow::BinaryBuilder builder;
-    auto input = data->GetValues<T>(1);
-    for (size_t i = 0; i < length; ++i) {
-        if (value->IsNull(i)) {
-            builder.AppendNull();
-            continue;
-        }
-        T item = input[i];
-        Numeric v;
-        if constexpr(std::is_same_v<T,double>) {
-            v = PgFloatToNumeric(item, 1000000000000LL, 12);
-        } else if constexpr(std::is_same_v<T,float>) {
-            v = PgFloatToNumeric(item, 1000000LL, 6);
-        } else {
-            v = int64_to_numeric(item);
-        }
-        auto datum = NumericGetDatum(v);
-        auto ptr = (char*)datum;
-        auto len = GetFullVarSize((const text*)datum);
-        NUdf::ZeroMemoryContext(ptr);
-        ARROW_OK(builder.Append(ptr - sizeof(void*), len + sizeof(void*)));
-    }
-
-    std::shared_ptr<arrow::BinaryArray> ret;
-    ARROW_OK(builder.Finish(&ret));
-    return ret;
-}
-
 TColumnConverter BuildPgNumericColumnConverter(const std::shared_ptr<arrow::DataType>& originalType) {
     switch (originalType->id()) {
     case arrow::Type::INT16:
diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.h b/ydb/library/yql/parser/pg_wrapper/arrow.h
index 6736dd8c6dca..17b51cfafb12 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.h
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.h
@@ -1299,9 +1299,7 @@ TExecFunc FindExec(Oid oid);
 
 const NPg::TAggregateDesc& ResolveAggregation(const TString& name, NKikimr::NMiniKQL::TTupleType* tupleType, const std::vector<ui32>& argsColumns, NKikimr::NMiniKQL::TType* returnType);
 
-Numeric PgFloatToNumeric(double item, ui64 scale, int digits);
+}
 
-template<typename T>
-std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Array>& value);
+#include "arrow_impl.h"
 
-}
diff --git a/ydb/library/yql/parser/pg_wrapper/arrow_impl.h b/ydb/library/yql/parser/pg_wrapper/arrow_impl.h
new file mode 100644
index 000000000000..c052c2267290
--- /dev/null
+++ b/ydb/library/yql/parser/pg_wrapper/arrow_impl.h
@@ -0,0 +1,44 @@
+#pragma once
+
+#include <arrow/array.h>
+#include <arrow/array/builder_binary.h>
+
+namespace NYql {
+
+Numeric PgFloatToNumeric(double item, ui64 scale, int digits);
+
+template<typename T>
+std::shared_ptr<arrow::Array> PgConvertNumeric(const std::shared_ptr<arrow::Array>& value) {
+    TArenaMemoryContext arena;
+    const auto& data = value->data();
+    size_t length = data->length;
+    arrow::BinaryBuilder builder;
+    auto input = data->GetValues<T>(1);
+    for (size_t i = 0; i < length; ++i) {
+        if (value->IsNull(i)) {
+            builder.AppendNull();
+            continue;
+        }
+        T item = input[i];
+        Numeric v;
+        if constexpr(std::is_same_v<T,double>) {
+            v = PgFloatToNumeric(item, 1000000000000LL, 12);
+        } else if constexpr(std::is_same_v<T,float>) {
+            v = PgFloatToNumeric(item, 1000000LL, 6);
+        } else {
+            v = int64_to_numeric(item);
+        }
+        auto datum = NumericGetDatum(v);
+        auto ptr = (char*)datum;
+        auto len = GetFullVarSize((const text*)datum);
+        NUdf::ZeroMemoryContext(ptr);
+        ARROW_OK(builder.Append(ptr - sizeof(void*), len + sizeof(void*)));
+    }
+
+    std::shared_ptr<arrow::BinaryArray> ret;
+    ARROW_OK(builder.Finish(&ret));
+    return ret;
+}
+
+}
+

From b33422ffc35342c609d01a9661bac8cc6654d8dc Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Wed, 27 Dec 2023 17:00:50 +0100
Subject: [PATCH 10/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
index ce33968e5b7b..59cb4af5bf1b 100644
--- a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
@@ -61,6 +61,7 @@ Y_UNIT_TEST(PgConvertNumericDouble) {
             UNIT_ASSERT(expected[i] == nullptr);
         } else {
             const char* addr = item.AsStringRef().Data() + sizeof(void*);
+            UNIT_ASSERT(expected[i] != nullptr);
             UNIT_ASSERT_VALUES_EQUAL(
                 TString(DatumGetCString(DirectFunctionCall1(numeric_out, (Datum)addr))),
                 expected[i]
@@ -96,6 +97,7 @@ Y_UNIT_TEST(PgConvertNumericInt) {
             UNIT_ASSERT(expected[i] == nullptr);
         } else {
             const char* addr = item.AsStringRef().Data() + sizeof(void*);
+            UNIT_ASSERT(expected[i] != nullptr);
             UNIT_ASSERT_VALUES_EQUAL(
                 TString(DatumGetCString(DirectFunctionCall1(numeric_out, (Datum)addr))),
                 expected[i]

From 692dccd290b688fc77c74e11d7c3b62e639a9258 Mon Sep 17 00:00:00 2001
From: Alexey Ozeritskiy <aozeritsky@ydb.tech>
Date: Thu, 28 Dec 2023 11:57:04 +0100
Subject: [PATCH 11/11] Polish

---
 ydb/library/yql/parser/pg_wrapper/arrow.cpp       | 2 +-
 ydb/library/yql/parser/pg_wrapper/arrow.h         | 3 ---
 ydb/library/yql/parser/pg_wrapper/arrow_impl.h    | 4 ++++
 ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp | 2 +-
 4 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.cpp b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
index e7ed9a8d8170..5a8559ff8f2d 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.cpp
@@ -1,4 +1,5 @@
 #include "arrow.h"
+#include "arrow_impl.h"
 #include <ydb/library/yql/parser/pg_wrapper/interface/arrow.h>
 #include <ydb/library/yql/parser/pg_wrapper/interface/utils.h>
 #include <ydb/library/yql/minikql/mkql_node_cast.h>
@@ -13,7 +14,6 @@
 extern "C" {
 #include "utils/date.h"
 #include "utils/timestamp.h"
-#include "utils/numeric.h"
 }
 
 namespace NYql {
diff --git a/ydb/library/yql/parser/pg_wrapper/arrow.h b/ydb/library/yql/parser/pg_wrapper/arrow.h
index 17b51cfafb12..7bd0cfcdd98d 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow.h
+++ b/ydb/library/yql/parser/pg_wrapper/arrow.h
@@ -17,7 +17,6 @@ extern "C" {
 #include "fmgr.h"
 #include "catalog/pg_type_d.h"
 #include "catalog/pg_collation_d.h"
-#include "utils/numeric.h"
 }
 
 #include "utils.h"
@@ -1301,5 +1300,3 @@ const NPg::TAggregateDesc& ResolveAggregation(const TString& name, NKikimr::NMin
 
 }
 
-#include "arrow_impl.h"
-
diff --git a/ydb/library/yql/parser/pg_wrapper/arrow_impl.h b/ydb/library/yql/parser/pg_wrapper/arrow_impl.h
index c052c2267290..7574aada429f 100644
--- a/ydb/library/yql/parser/pg_wrapper/arrow_impl.h
+++ b/ydb/library/yql/parser/pg_wrapper/arrow_impl.h
@@ -3,6 +3,10 @@
 #include <arrow/array.h>
 #include <arrow/array/builder_binary.h>
 
+extern "C" {
+#include "utils/numeric.h"
+}
+
 namespace NYql {
 
 Numeric PgFloatToNumeric(double item, ui64 scale, int digits);
diff --git a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
index 59cb4af5bf1b..c09359f40523 100644
--- a/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/ut/arrow_ut.cpp
@@ -4,9 +4,9 @@
 #include <library/cpp/testing/unittest/registar.h>
 
 #include "arrow.h"
+#include "arrow_impl.h"
 
 extern "C" {
-#include "utils/numeric.h"
 #include "utils/fmgrprotos.h"
 }