pg syntax support

10 files changed, 121 insertions, 8 deletions

diff --git a/ydb/core/kqp/ut/pg/kqp_pg_ut.cpp b/ydb/core/kqp/ut/pg/kqp_pg_ut.cpp
index 4b26725f6d4..a2a1dd6d31a 100644
--- a/ydb/core/kqp/ut/pg/kqp_pg_ut.cpp
+++ b/ydb/core/kqp/ut/pg/kqp_pg_ut.cpp
@@ -83,7 +83,7 @@ Y_UNIT_TEST_SUITE(KqpPg) {
         }
         rows.EndList();
 
-        result = db.BulkUpsert("/Root/Pg", rows.Build()).GetValueSync();;
+        result = db.BulkUpsert("/Root/Pg", rows.Build()).GetValueSync();
         UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
 
         session.Close().GetValueSync();
@@ -156,6 +156,40 @@ Y_UNIT_TEST_SUITE(KqpPg) {
             }
         }
    }
+
+   Y_UNIT_TEST(EmptyQuery) {
+        auto kikimr = DefaultKikimrRunner();
+        NYdb::NScripting::TScriptingClient client(kikimr.GetDriver());
+
+        auto result = client.ExecuteYqlScript(R"(
+            --!syntax_pg
+        )").GetValueSync();
+
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        Y_ENSURE(result.GetResultSets().empty());
+    }
+
+    Y_UNIT_TEST(NoTableQuery) {
+        auto kikimr = DefaultKikimrRunner();
+        NYdb::NScripting::TScriptingClient client(kikimr.GetDriver());
+
+        auto result = client.ExecuteYqlScript(R"(
+            --!syntax_pg
+            SELECT * FROM (VALUES
+                (1, 'one'),
+                (2, 'two'),
+                (3, 'three')
+            ) AS t (int8, varchar);
+        )").GetValueSync();
+
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+        CompareYson(R"([
+            ["1";"one"];
+            ["2";"two"];
+            ["3";"three"]
+        ])", FormatResultSetYson(result.GetResultSet(0)));
+    }
 }
 
 } // namespace NKqp
diff --git a/ydb/core/ydb_convert/ydb_convert.cpp b/ydb/core/ydb_convert/ydb_convert.cpp
index ccd6828e9cd..1e2b0904d76 100644
--- a/ydb/core/ydb_convert/ydb_convert.cpp
+++ b/ydb/core/ydb_convert/ydb_convert.cpp
@@ -123,6 +123,12 @@ void ConvertMiniKQLTypeToYdbType(const NKikimrMiniKQL::TType& input, Ydb::Type&
             }
             break;
         }
+        case NKikimrMiniKQL::ETypeKind::Pg: {
+            const NKikimrMiniKQL::TPgType& pgType = input.GetPg();
+            auto pgOut = output.mutable_pg_type();
+            pgOut->Setoid(pgType.Getoid());
+            break;
+        }
         default: {
             Y_FAIL("Unknown protobuf type: %s", input.DebugString().c_str());
         }
@@ -559,6 +565,11 @@ void ConvertMiniKQLValueToYdbValue(const NKikimrMiniKQL::TType& inputType,
             }
             break;
         }
+        case NKikimrMiniKQL::ETypeKind::Pg: {
+            const auto& stringRef = inputValue.GetText();
+            output.set_text_value(stringRef.data(), stringRef.size());
+            break;
+        }
         default: {
             ythrow yexception() << "Unknown protobuf type: "
                                 << inputType.DebugString();
diff --git a/ydb/core/ydb_convert/ydb_convert_ut.cpp b/ydb/core/ydb_convert/ydb_convert_ut.cpp
index 27fdcf2fdac..911e972735f 100644
--- a/ydb/core/ydb_convert/ydb_convert_ut.cpp
+++ b/ydb/core/ydb_convert/ydb_convert_ut.cpp
@@ -364,6 +364,18 @@ Variant {
         TestConvertTypeFromYdb(expected, input);
     }
 
+    Y_UNIT_TEST(PgType) {
+        const TString input =
+            "Kind: Pg\n"
+            "Pg {\n"
+            "  oid: 16\n"
+            "}\n";
+        const TString expected =
+            "pg_type {\n"
+            "  oid: 16\n"
+            "}\n";
+        TestConvertTypeToYdb(input, expected);
+    }
 } // ConvertMiniKQLTypeToYdbTypeTest
 
 Y_UNIT_TEST_SUITE(ConvertMiniKQLValueToYdbValueTest) {
@@ -1017,6 +1029,19 @@ variant_index: 3435973836
         UNIT_ASSERT_EXCEPTION(TestConvertValueFromYdb(inputType, inputValue, expected), yexception);
     }
 
+    Y_UNIT_TEST(PgValue) {
+        const TString inputType =
+            "Kind: Pg\n"
+            "Pg {\n"
+            "  oid: 16\n"
+            "}\n";
+        const TString inputValue =
+            "Text: \"123\"\n";
+        const TString expectedValue =
+            "text_value: \"123\"\n";
+        TestConvertValueToYdb(inputType, inputValue, expectedValue);
+    }
+
 } // ConvertYdbValueToMiniKQLValueTest
 
 } // namespace NKikimr
diff --git a/ydb/library/mkql_proto/mkql_proto.cpp b/ydb/library/mkql_proto/mkql_proto.cpp
index 96f38a9c9a3..c30a8bc114a 100644
--- a/ydb/library/mkql_proto/mkql_proto.cpp
+++ b/ydb/library/mkql_proto/mkql_proto.cpp
@@ -462,12 +462,12 @@ void ExportValueToProtoImpl(TType* type, const NUdf::TUnboxedValuePod& value, NK
 
         case TType::EKind::Pg: {
             if (!value) {
-                // do not set Bytes field
+                // do not set Text field
                 return;
             }
             auto pgType = static_cast<TPgType*>(type);
-            auto binaryValue = NYql::NCommon::PgValueToNativeBinary(value, pgType->GetTypeId());
-            res.SetBytes(binaryValue);
+            auto textValue = NYql::NCommon::PgValueToString(value, pgType->GetTypeId());
+            res.SetText(textValue);
             break;
         }
 
diff --git a/ydb/library/yql/dq/runtime/CMakeLists.txt b/ydb/library/yql/dq/runtime/CMakeLists.txt
index 6e46f5cf5cf..254660b490a 100644
--- a/ydb/library/yql/dq/runtime/CMakeLists.txt
+++ b/ydb/library/yql/dq/runtime/CMakeLists.txt
@@ -25,6 +25,7 @@ target_link_libraries(yql-dq-runtime PUBLIC
   yql-dq-common
   yql-dq-expr_nodes
   yql-dq-type_ann
+  parser-pg_wrapper-interface
   common-schema-mkql
   tools-enum_parser-enum_serialization_runtime
 )
diff --git a/ydb/library/yql/dq/runtime/dq_compute.cpp b/ydb/library/yql/dq/runtime/dq_compute.cpp
index 0da2ed9630d..181c2d3d1ed 100644
--- a/ydb/library/yql/dq/runtime/dq_compute.cpp
+++ b/ydb/library/yql/dq/runtime/dq_compute.cpp
@@ -2,6 +2,8 @@
 
 #include <ydb/library/yql/minikql/comp_nodes/mkql_factories.h>
 #include <ydb/library/yql/minikql/mkql_node.h>
+#include <ydb/library/yql/parser/pg_wrapper/interface/pack.h>
+#include <ydb/library/yql/parser/pg_wrapper/interface/comp_factory.h>
 #include "ydb/library/yql/utils/yql_panic.h"
 
 namespace NYql::NDq {
@@ -11,7 +13,9 @@ using namespace NMiniKQL;
 
 TComputationNodeFactory GetDqBaseComputeFactory(const TDqComputeContextBase* computeCtx) {
     YQL_ENSURE(computeCtx);
-    auto builtinFactory = GetBuiltinFactory();
+    auto builtinFactory = GetCompositeWithBuiltinFactory({
+        NYql::GetPgFactory()
+    });
 
     return [builtinFactory]
         (TCallable& callable, const TComputationNodeFactoryContext& ctx) -> IComputationNode* {
diff --git a/ydb/library/yql/dq/runtime/dq_transport.cpp b/ydb/library/yql/dq/runtime/dq_transport.cpp
index 7008b1a3695..b15d2403d98 100644
--- a/ydb/library/yql/dq/runtime/dq_transport.cpp
+++ b/ydb/library/yql/dq/runtime/dq_transport.cpp
@@ -4,6 +4,8 @@
 #include <ydb/library/mkql_proto/mkql_proto.h>
 #include <ydb/library/yql/minikql/computation/mkql_computation_node_holders.h>
 #include <ydb/library/yql/minikql/computation/mkql_computation_node_pack.h>
+#include <ydb/library/yql/parser/pg_wrapper/interface/comp_factory.h>
+#include <ydb/library/yql/parser/pg_wrapper/interface/pack.h>
 #include <ydb/library/yql/providers/common/mkql/yql_type_mkql.h>
 #include <ydb/library/yql/utils/yql_panic.h>
 
@@ -380,6 +382,13 @@ ui64 EstimateSizeImpl(const NUdf::TUnboxedValuePod& value, const NKikimr::NMiniK
             return 2 + EstimateSizeImpl(value.GetVariantItem(), innerType, fixed);
         }
 
+        case TType::EKind::Pg: {
+            if (value) {
+                auto pgType = static_cast<const TPgType*>(type);
+                return NKikimr::NMiniKQL::PgValueSize(pgType, value);
+            }
+            return 0;
+        }
         case TType::EKind::Type:
         case TType::EKind::Stream:
         case TType::EKind::Callable:
@@ -389,7 +398,6 @@ ui64 EstimateSizeImpl(const NUdf::TUnboxedValuePod& value, const NKikimr::NMiniK
         case TType::EKind::ReservedKind:
         case TType::EKind::Tagged:
         case TType::EKind::Block:
-        case TType::EKind::Pg:
             THROW yexception() << "Unsupported type: " << type->GetKindAsStr();
     }
 }
diff --git a/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp b/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
index 5d7b386667f..e0b4d7e2efa 100644
--- a/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
+++ b/ydb/library/yql/parser/pg_wrapper/comp_factory.cpp
@@ -2336,6 +2336,22 @@ namespace NMiniKQL {
 
 using namespace NYql;
 
+ui64 PgValueSize(const TPgType* type, const NUdf::TUnboxedValuePod& value) {
+    const auto& typeDesc = NYql::NPg::LookupType(type->GetTypeId());
+    if (typeDesc.TypeLen >= 0) {
+        return typeDesc.TypeLen;
+    }
+    Y_ENSURE(typeDesc.TypeLen == -1 || typeDesc.TypeLen == -2);
+    auto datum = PointerDatumFromPod(value);
+    if (typeDesc.TypeLen == -1) {
+        const auto x = (const text*)PointerDatumFromPod(value);
+        return GetCleanVarSize(x);
+    } else {
+        const auto x = (const char*)PointerDatumFromPod(value);
+        return strlen(x);
+    }
+}
+
 void PGPackImpl(bool stable, const TPgType* type, const NUdf::TUnboxedValuePod& value, TBuffer& buf) {
     switch (type->GetTypeId()) {
     case BOOLOID: {
@@ -3167,4 +3183,3 @@ ui64 PgNativeBinaryHash(const char* data, size_t size, void* typeDesc) {
 }
 
 } // namespace NKikimr::NPg
-
diff --git a/ydb/library/yql/parser/pg_wrapper/interface/pack.h b/ydb/library/yql/parser/pg_wrapper/interface/pack.h
index 0a66c7fbb29..72b598136d6 100644
--- a/ydb/library/yql/parser/pg_wrapper/interface/pack.h
+++ b/ydb/library/yql/parser/pg_wrapper/interface/pack.h
@@ -17,5 +17,7 @@ NUdf::TUnboxedValue PGUnpackImpl(const TPgType* type, TStringBuf& buf);
 void EncodePresortPGValue(TPgType* type, const NUdf::TUnboxedValue& value, TVector<ui8>& output);
 NUdf::TUnboxedValue DecodePresortPGValue(TPgType* type, TStringBuf& input, TVector<ui8>& buffer);
 
+ui64 PgValueSize(const TPgType* type, const NUdf::TUnboxedValuePod& value);
+
 } // namespace NMiniKQL
 } // namespace NKikimr
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 2bf416fbc5f..bc2fedc05a5 100644
--- a/ydb/library/yql/sql/pg_dummy/pg_sql_dummy.cpp
+++ b/ydb/library/yql/sql/pg_dummy/pg_sql_dummy.cpp
@@ -130,6 +130,12 @@ void PgReleaseThreadContext(void* ctx) {
     Y_UNUSED(ctx);
 }
 
+ui64 PgValueSize(const TPgType* type, const NUdf::TUnboxedValuePod& value) {
+    Y_UNUSED(type);
+    Y_UNUSED(value);
+    throw yexception() << "PG types are not supported";
+}
+
 void PGPackImpl(bool stable, const TPgType* type, const NUdf::TUnboxedValuePod& value, TBuffer& buf) {
    Y_UNUSED(stable);
    Y_UNUSED(type);
@@ -249,7 +255,14 @@ std::unique_ptr<NUdf::IPgBuilder> CreatePgBuilder() {
 std::function<NKikimr::NMiniKQL::IComputationNode* (NKikimr::NMiniKQL::TCallable&,
     const NKikimr::NMiniKQL::TComputationNodeFactoryContext&)> GetPgFactory()
 {
-    return {};
+    return [] (
+        NKikimr::NMiniKQL::TCallable& callable,
+        const NKikimr::NMiniKQL::TComputationNodeFactoryContext& ctx
+    ) -> NKikimr::NMiniKQL::IComputationNode* {
+        Y_UNUSED(callable);
+        Y_UNUSED(ctx);
+        return nullptr;
+    };
 }
 
 } // NYql