Support PG types in CellMaker (#9856)

3 files changed, 26 insertions, 2 deletions

diff --git a/ydb/core/io_formats/cell_maker/cell_maker.cpp b/ydb/core/io_formats/cell_maker/cell_maker.cpp
index 2cb4441bc7a..4e7a40e6548 100644
--- a/ydb/core/io_formats/cell_maker/cell_maker.cpp
+++ b/ydb/core/io_formats/cell_maker/cell_maker.cpp
@@ -393,6 +393,13 @@ bool MakeCell(TCell& cell, const NJson::TJsonValue& value, const NScheme::TTypeI
             return TCellMaker<TMaybe<TString>, TStringBuf>::Make(cell, value.GetStringSafe(), pool, err, &DyNumberToStringBuf);
         case NScheme::NTypeIds::Decimal:
             return TCellMaker<NYql::NDecimal::TInt128, std::pair<ui64, ui64>>::Make(cell, value.GetStringSafe(), pool, err, &Int128ToPair);
+        case NScheme::NTypeIds::Pg:
+            if (auto result = NPg::PgNativeBinaryFromNativeText(value.GetStringSafe(), typeInfo.GetPgTypeDesc()); result.Error) {
+                err = *result.Error;
+                return false;
+            } else {
+                return TCellMaker<NPg::TConvertResult, TStringBuf>::MakeDirect(cell, result, pool, err, &PgToStringBuf);
+            }
         case NScheme::NTypeIds::Uuid:
             return TCellMaker<TUuidHolder, TStringBuf>::Make(cell, value.GetStringSafe(), pool, err, &UuidToStringBuf);
         default:
diff --git a/ydb/core/tx/replication/service/table_writer_ut.cpp b/ydb/core/tx/replication/service/table_writer_ut.cpp
index 5b1b3de7b16..2ec78aaa6b2 100644
--- a/ydb/core/tx/replication/service/table_writer_ut.cpp
+++ b/ydb/core/tx/replication/service/table_writer_ut.cpp
@@ -39,7 +39,10 @@ Y_UNIT_TEST_SUITE(LocalTableWriter) {
     }
 
     Y_UNIT_TEST(SupportedTypes) {
-        TEnv env(TFeatureFlags().SetEnableTableDatetime64(true));
+        TEnv env(TFeatureFlags()
+            .SetEnableTableDatetime64(true)
+            .SetEnableTablePgTypes(true)
+            .SetEnablePgSyntax(true));
         env.GetRuntime().SetLogPriority(NKikimrServices::REPLICATION_SERVICE, NLog::PRI_DEBUG);
 
         env.CreateTable("/Root", *MakeTableDescription(TTestTableDescription{
@@ -70,6 +73,13 @@ Y_UNIT_TEST_SUITE(LocalTableWriter) {
                 {.Name = "datetime64_value", .Type = "Datetime64"},
                 {.Name = "timestamp64_value", .Type = "Timestamp64"},
                 {.Name = "interval64_value", .Type = "Interval64"},
+                {.Name = "pgint2_value", .Type = "pgint2"},
+                {.Name = "pgint4_value", .Type = "pgint4"},
+                {.Name = "pgint8_value", .Type = "pgint8"},
+                {.Name = "pgfloat4_value", .Type = "pgfloat4"},
+                {.Name = "pgfloat8_value", .Type = "pgfloat8"},
+                {.Name = "pgbytea_value", .Type = "pgbytea"},
+                {.Name = "pgtext_value", .Type = "pgtext"},
             },
         }));
 
@@ -101,6 +111,13 @@ Y_UNIT_TEST_SUITE(LocalTableWriter) {
             TRecord(21, R"({"key":[21], "update":{"datetime64_value":1597235696}})"),
             TRecord(22, R"({"key":[22], "update":{"timestamp64_value":1597235696123456}})"),
             TRecord(23, R"({"key":[23], "update":{"interval64_value":-300500}})"),
+            TRecord(24, R"({"key":[24], "update":{"pgint2_value":"-42"}})"),
+            TRecord(25, R"({"key":[25], "update":{"pgint4_value":"-420"}})"),
+            TRecord(26, R"({"key":[26], "update":{"pgint8_value":"-4200"}})"),
+            TRecord(27, R"({"key":[27], "update":{"pgfloat4_value":"3.1415"}})"),
+            TRecord(28, R"({"key":[28], "update":{"pgfloat8_value":"2.718"}})"),
+            TRecord(29, R"({"key":[29], "update":{"pgbytea_value":"\\x6c6f72656d2022697073756d22"}})"),
+            TRecord(30, R"({"key":[30], "update":{"pgtext_value":"lorem \"ipsum\""}})"),
         }));
     }
 }
diff --git a/ydb/core/tx/replication/ut_helpers/ya.make b/ydb/core/tx/replication/ut_helpers/ya.make
index dfe6913524a..d7c9cc6af3a 100644
--- a/ydb/core/tx/replication/ut_helpers/ya.make
+++ b/ydb/core/tx/replication/ut_helpers/ya.make
@@ -3,7 +3,7 @@ LIBRARY()
 PEERDIR(
     ydb/core/base
     ydb/core/protos
-    ydb/core/testlib/default
+    ydb/core/testlib/pg
     ydb/core/tx/replication/ydb_proxy
     ydb/public/sdk/cpp/client/ydb_topic
     library/cpp/testing/unittest