change LB metadata syntax

7 files changed, 41 insertions, 49 deletions

diff --git a/ydb/library/yql/core/expr_nodes/yql_expr_nodes.json b/ydb/library/yql/core/expr_nodes/yql_expr_nodes.json
index 6057f723c4d..342a1f2e704 100644
--- a/ydb/library/yql/core/expr_nodes/yql_expr_nodes.json
+++ b/ydb/library/yql/core/expr_nodes/yql_expr_nodes.json
@@ -1777,14 +1777,9 @@
             "Match": {"Type": "Callable", "Name": "TableRecord"}
         },
         {
-            "Name": "TCoWriteTime",
+            "Name": "TCoSystemMetadata",
             "Base": "TCoTablePropBase",
-            "Match": {"Type": "Callable", "Name": "WriteTime"}
-        },
-        {
-            "Name": "TCoOffset",
-            "Base": "TCoTablePropBase",
-            "Match": {"Type": "Callable", "Name": "Offset"}
+            "Match": {"Type": "Callable", "Name": "SystemMetadata"}
         },
         {
             "Name": "TCoIsKeySwitch",
diff --git a/ydb/library/yql/providers/pq/common/pq_meta_fields.cpp b/ydb/library/yql/providers/pq/common/pq_meta_fields.cpp
index c0f613f593d..2c9e94faacb 100644
--- a/ydb/library/yql/providers/pq/common/pq_meta_fields.cpp
+++ b/ydb/library/yql/providers/pq/common/pq_meta_fields.cpp
@@ -5,22 +5,22 @@
 
 namespace {
     const std::vector<NYql::TMetaFieldDescriptor> PqMetaFields = {
-        NYql::TMetaFieldDescriptor("CreateTime", "_yql_sys_create_time", NYql::NUdf::EDataSlot::Timestamp),
-        NYql::TMetaFieldDescriptor("WriteTime", "_yql_sys_write_time", NYql::NUdf::EDataSlot::Timestamp),
-        NYql::TMetaFieldDescriptor("PartitionId", "_yql_sys_partition_id", NYql::NUdf::EDataSlot::Uint64),
-        NYql::TMetaFieldDescriptor("Offset", "_yql_sys_offset", NYql::NUdf::EDataSlot::Uint64),
-        NYql::TMetaFieldDescriptor("MessageGroupId", "_yql_sys_message_group_id", NYql::NUdf::EDataSlot::String),
-        NYql::TMetaFieldDescriptor("SeqNo", "_yql_sys_seq_no", NYql::NUdf::EDataSlot::Uint64),
+        NYql::TMetaFieldDescriptor("create_time", "_yql_sys_create_time", NYql::NUdf::EDataSlot::Timestamp),
+        NYql::TMetaFieldDescriptor("write_time", "_yql_sys_write_time", NYql::NUdf::EDataSlot::Timestamp),
+        NYql::TMetaFieldDescriptor("partition_id", "_yql_sys_partition_id", NYql::NUdf::EDataSlot::Uint64),
+        NYql::TMetaFieldDescriptor("offset", "_yql_sys_offset", NYql::NUdf::EDataSlot::Uint64),
+        NYql::TMetaFieldDescriptor("message_group_id", "_yql_sys_message_group_id", NYql::NUdf::EDataSlot::String),
+        NYql::TMetaFieldDescriptor("seq_no", "_yql_sys_seq_no", NYql::NUdf::EDataSlot::Uint64),
     };
 }
 
 namespace NYql {
 
-const TMetaFieldDescriptor* FindPqMetaFieldDescriptorByCallable(const TString& callableName) {
+const TMetaFieldDescriptor* FindPqMetaFieldDescriptorByKey(const TString& key) {
     const auto iter = std::find_if(
         PqMetaFields.begin(),
         PqMetaFields.end(),
-        [&](const NYql::TMetaFieldDescriptor& item){ return item.CallableName == callableName; });
+        [&](const NYql::TMetaFieldDescriptor& item){ return item.Key == key; });
     if (iter != PqMetaFields.end()) {
         return iter;
     }
@@ -51,15 +51,4 @@ std::vector<TString> AllowedPqMetaSysColumns() {
     return res;
 }
 
-std::vector<TString> AllowedPqMetaCallables() {
-    std::vector<TString> res;
-    res.reserve(PqMetaFields.size());
-
-    for (const auto& descriptor : PqMetaFields) {
-        res.emplace_back(descriptor.CallableName);
-    }
-
-    return res;
-}
-
 }
diff --git a/ydb/library/yql/providers/pq/common/pq_meta_fields.h b/ydb/library/yql/providers/pq/common/pq_meta_fields.h
index ce73ca5ccce..1a0daa51699 100644
--- a/ydb/library/yql/providers/pq/common/pq_meta_fields.h
+++ b/ydb/library/yql/providers/pq/common/pq_meta_fields.h
@@ -13,24 +13,22 @@ namespace NYql {
 
 struct TMetaFieldDescriptor {
 public:
-    TMetaFieldDescriptor(TString callableName, TString sysColumn, NUdf::EDataSlot type)
-        : CallableName(callableName)
+    TMetaFieldDescriptor(TString key, TString sysColumn, NUdf::EDataSlot type)
+        : Key(key)
         , SysColumn(sysColumn)
         , Type(type)
     { }
 
 public:
-    const TString CallableName;
+    const TString Key;
     const TString SysColumn;
     const NUdf::EDataSlot Type;
 };
 
-const TMetaFieldDescriptor* FindPqMetaFieldDescriptorByCallable(const TString& callableName);
+const TMetaFieldDescriptor* FindPqMetaFieldDescriptorByKey(const TString& key);
 
 const TMetaFieldDescriptor* FindPqMetaFieldDescriptorBySysColumn(const TString& sysColumn);
 
 std::vector<TString> AllowedPqMetaSysColumns();
 
-std::vector<TString> AllowedPqMetaCallables();
-
 }
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp b/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp
index e9f4178cd8c..f5f075f5d9e 100644
--- a/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp
+++ b/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp
@@ -27,7 +27,7 @@ public:
         AddHandler({TPqReadTopic::CallableName()}, Hndl(&TSelf::HandleReadTopic));
         AddHandler({TPqTopic::CallableName()}, Hndl(&TSelf::HandleTopic));
         AddHandler({TDqPqTopicSource::CallableName()}, Hndl(&TSelf::HandleDqTopicSource));
-        AddHandler({TCoWriteTime::CallableName(), TCoOffset::CallableName()}, Hndl(&TSelf::HandleMetadata));
+        AddHandler({TCoSystemMetadata::CallableName()}, Hndl(&TSelf::HandleMetadata));
     }
 
     TStatus HandleConfigure(const TExprNode::TPtr& input, TExprContext& ctx) {
@@ -191,33 +191,44 @@ public:
     }
 
     TStatus HandleMetadata(const TExprNode::TPtr& input, TExprNode::TPtr& output, TExprContext& ctx) {
-        const auto* descriptor = FindPqMetaFieldDescriptorByCallable(TString(input->Content()));
+        const auto key = input->ChildPtr(0);
+        if (!EnsureCallable(*key, ctx)) {
+            return IGraphTransformer::TStatus::Error;
+        }
+
+        const auto metadataKey = TString(key->TailPtr()->Content());
+        const auto* descriptor = FindPqMetaFieldDescriptorByKey(metadataKey);
+        if (!descriptor) {
+            ctx.AddError(TIssue(ctx.GetPosition(input->Pos()), TStringBuilder() << "Metadata key " << metadataKey << " wasn't found"));
+            return IGraphTransformer::TStatus::Error;
+        }
 
-        if (!EnsureDependsOn(input->Head(), ctx)) {
+        const auto dependsOn = input->Child(1);
+        if (!EnsureDependsOn(*dependsOn, ctx)) {
             return IGraphTransformer::TStatus::Error;
         }
 
-        auto depOn = input->Head().HeadPtr();
-        if (!EnsureStructType(*depOn, ctx)) {
+        const auto row = dependsOn->TailPtr();
+        if (!EnsureStructType(*row, ctx)) {
             return IGraphTransformer::TStatus::Error;
         }
 
-        if (depOn->GetTypeAnn()->GetKind() == ETypeAnnotationKind::Struct) {
-            auto structType = depOn->GetTypeAnn()->Cast<TStructExprType>();
+        if (row->GetTypeAnn()->GetKind() == ETypeAnnotationKind::Struct) {
+            auto structType = row->GetTypeAnn()->Cast<TStructExprType>();
             if (auto pos = structType->FindItem(descriptor->SysColumn)) {
                 bool isOptional = false;
                 const TDataExprType* dataType = nullptr;
-                if (!EnsureDataOrOptionalOfData(depOn->Pos(), structType->GetItems()[*pos]->GetItemType(), isOptional, dataType, ctx)) {
+                if (!EnsureDataOrOptionalOfData(row->Pos(), structType->GetItems()[*pos]->GetItemType(), isOptional, dataType, ctx)) {
                     return IGraphTransformer::TStatus::Error;
                 }
 
-                if (!EnsureSpecificDataType(depOn->Pos(), *dataType, descriptor->Type, ctx)) {
+                if (!EnsureSpecificDataType(row->Pos(), *dataType, descriptor->Type, ctx)) {
                     return IGraphTransformer::TStatus::Error;
                 }
 
                 output = ctx.Builder(input->Pos())
                     .Callable("Member")
-                        .Add(0, depOn)
+                        .Add(0, row)
                         .Atom(1, descriptor->SysColumn, TNodeFlags::Default)
                     .Seal()
                     .Build();
diff --git a/ydb/library/yql/sql/v1/builtin.cpp b/ydb/library/yql/sql/v1/builtin.cpp
index 77006be36a8..27e5448af6c 100644
--- a/ydb/library/yql/sql/v1/builtin.cpp
+++ b/ydb/library/yql/sql/v1/builtin.cpp
@@ -2912,13 +2912,7 @@ struct TBuiltinFuncData {
             {"tablerows", BuildSimpleBuiltinFactoryCallback<TTableRows>() },
             {"weakfield", BuildSimpleBuiltinFactoryCallback<TWeakFieldOp>()},
 
-            // meta fields
-            {"writetime", BuildNamedBuiltinFactoryCallback<TCallDirectRow>("WriteTime")},
-            {"offset", BuildNamedBuiltinFactoryCallback<TCallDirectRow>("Offset")},
-            // {"createtime", BuildNamedBuiltinFactoryCallback<TCallDirectRow>("CreateTime")},
-            // {"partitionid", BuildNamedBuiltinFactoryCallback<TCallDirectRow>("PartitionId")},
-            // {"messagegroupid", BuildNamedBuiltinFactoryCallback<TCallDirectRow>("MessageGroupId")},
-            // {"seqno", BuildNamedBuiltinFactoryCallback<TCallDirectRow>("SeqNo")},
+            {"systemmetadata", BuildNamedArgcBuiltinFactoryCallback<TCallDirectRow>("SystemMetadata", 1, -1)},
 
             // Hint builtins
             {"grouping", BuildSimpleBuiltinFactoryCallback<TGroupingNode>()},
diff --git a/ydb/library/yql/sql/v1/node.cpp b/ydb/library/yql/sql/v1/node.cpp
index 734c9d6b381..d69193ec018 100644
--- a/ydb/library/yql/sql/v1/node.cpp
+++ b/ydb/library/yql/sql/v1/node.cpp
@@ -672,6 +672,10 @@ TCallDirectRow::TPtr TCallDirectRow::DoClone() const {
     return new TCallDirectRow(Pos, OpName, CloneContainer(Args));
 }
 
+TCallDirectRow::TCallDirectRow(TPosition pos, const TString& opName, i32 minArgs, i32 maxArgs, const TVector<TNodePtr>& args)
+    : TCallNode(pos, opName, minArgs, maxArgs, args)
+{}
+
 TCallDirectRow::TCallDirectRow(TPosition pos, const TString& opName, const TVector<TNodePtr>& args)
     : TCallNode(pos, opName, 0, 0, args)
 {}
diff --git a/ydb/library/yql/sql/v1/node.h b/ydb/library/yql/sql/v1/node.h
index 585c21ee1c4..84be0f9be12 100644
--- a/ydb/library/yql/sql/v1/node.h
+++ b/ydb/library/yql/sql/v1/node.h
@@ -391,6 +391,7 @@ namespace NSQLTranslationV1 {
     class TCallDirectRow final : public TCallNode {
         TPtr DoClone() const final;
     public:
+        TCallDirectRow(TPosition pos, const TString& opName, i32 minArgs, i32 maxArgs, const TVector<TNodePtr>& args);
         TCallDirectRow(TPosition pos, const TString& opName, const TVector<TNodePtr>& args);
     protected:
         bool DoInit(TContext& ctx, ISource* src) override;