YQL-10211: into_table_stmt: Pass arbitrary write hints into AST

YQL-10211: into_table_stmt: Pass arbitrary write hints into AST
Example:
USE plato;

INSERT INTO Output WITH (TRUNCATE, COMPRESSION_ALGO="zstd", COMPRESSION_LVL="8", TRUNCATE)
SELECT
key as key,
"" as subkey,
"value:" || value as value
FROM Input WITH (TRUNCATE, LOG)
WHERE key < "100"
ORDER BY key;

ref:ccceedbf2611c4ca5a9fce3afb205082cdc1499f

ydb/library/yql/sql/v1/SQLv1.g.in

@@ -575,7 +575,7 @@ table_key: id_table_or_type (VIEW an_id)?;
 table_arg: AT? named_expr (VIEW an_id)?;
 table_hints: WITH (table_hint | LPAREN table_hint (COMMA table_hint)* RPAREN);
 table_hint:
-      an_id_hint
+      an_id_hint (EQUALS type_name_tag)?
     | (SCHEMA | COLUMNS) type_name_or_bind
     | SCHEMA LPAREN (struct_arg_as (COMMA struct_arg_as)*)? COMMA? RPAREN
 ;

ydb/library/yql/sql/v1/node.h

@@ -1312,6 +1312,7 @@ namespace NSQLTranslationV1 {
 
     EWriteColumnMode ToWriteColumnsMode(ESQLWriteColumnMode sqlWriteColumnMode);
     TNodePtr BuildEraseColumns(TPosition pos, const TVector<TString>& columns);
+    TNodePtr BuildIntoTableOptions(TPosition pos, const TVector<TString>& eraseColumns, const TTableHints& hints);
     TNodePtr BuildWriteColumns(TPosition pos, TScopedStatePtr scoped, const TTableRef& table, EWriteColumnMode mode, TSourcePtr values, TNodePtr options = nullptr);
     TNodePtr BuildUpdateColumns(TPosition pos, TScopedStatePtr scoped, const TTableRef& table, TSourcePtr values, TSourcePtr source);
     TNodePtr BuildDelete(TPosition pos, TScopedStatePtr scoped, const TTableRef& table, TSourcePtr source);

ydb/library/yql/sql/v1/query.cpp

@@ -505,6 +505,61 @@ TNodePtr BuildInputOptions(TPosition pos, const TTableHints& hints) {
     return new TInputOptions(pos, hints);
 }
 
+class TIntoTableOptions: public TAstListNode {
+public:
+    TIntoTableOptions(TPosition pos, const TVector<TString>& columns, const TTableHints& hints)
+        : TAstListNode(pos)
+        , Columns(columns)
+        , Hints(hints)
+    {
+    }
+
+    bool DoInit(TContext& ctx, ISource* src) override {
+        Y_UNUSED(ctx);
+        Y_UNUSED(src);
+
+        TNodePtr options = Y();
+        for (const auto& column: Columns) {
+            options->Add(Q(column));
+        }
+        if (Columns) {
+            Add(Q(Y(Q("erase_columns"), Q(options))));
+        }
+
+        for (const auto& hint : Hints) {
+            TString hintName = hint.first;
+            TMaybe<TIssue> normalizeError = NormalizeName(Pos, hintName);
+            if (!normalizeError.Empty()) {
+                ctx.Error() << normalizeError->Message;
+                ctx.IncrementMonCounter("sql_errors", "NormalizeHintError");
+                return false;
+            }
+            TNodePtr option = Y(BuildQuotedAtom(Pos, hintName));
+            for (auto& x : hint.second) {
+                if (!x->Init(ctx, src)) {
+                    return false;
+                }
+                option = L(option, x);
+            }
+            Add(Q(option));
+        }
+
+        return true;
+    }
+
+    TNodePtr DoClone() const final {
+        return new TIntoTableOptions(GetPos(), Columns, Hints);
+    }
+
+private:
+    TVector<TString> Columns;
+    TTableHints Hints;
+};
+
+TNodePtr BuildIntoTableOptions(TPosition pos, const TVector<TString>& eraseColumns, const TTableHints& hints) {
+    return new TIntoTableOptions(pos, eraseColumns, hints);
+}
+
 class TInputTablesNode final: public TAstListNode {
 public:
     TInputTablesNode(TPosition pos, const TTableList& tables, bool inSubquery, TScopedStatePtr scoped)

ydb/library/yql/sql/v1/sql.cpp

@@ -2945,18 +2945,23 @@ TNodePtr TSqlTranslation::StructLiteral(const TRule_struct_literal& node) {
 
 bool TSqlTranslation::TableHintImpl(const TRule_table_hint& rule, TTableHints& hints) {
     // table_hint:
-    //      an_id_hint
+    //      an_id_hint (EQUALS type_name_tag)?
     //    | (SCHEMA | COLUMNS) type_name_or_bind
     //    | SCHEMA LPAREN (struct_arg_as (COMMA struct_arg_as)*)? COMMA? RPAREN
     switch (rule.Alt_case()) {
     case TRule_table_hint::kAltTableHint1: {
-        const TString id = Id(rule.GetAlt_table_hint1().GetRule_an_id_hint1(), *this);
+        const auto& alt = rule.GetAlt_table_hint1();
+        const TString id = Id(alt.GetRule_an_id_hint1(), *this);
         const auto idLower = to_lower(id);
         if (idLower == "schema" || idLower == "columns") {
             Error() << "Expected type after " << to_upper(id);
             return false;
         }
-        hints[id] = {};
+        TVector<TNodePtr> hint_val;
+        if (alt.HasBlock2()) {
+            hint_val.push_back(TypeNameTag(alt.GetBlock2().GetRule_type_name_tag2()));
+        }
+        hints[id] = hint_val;
         break;
     }
 
@@ -8023,21 +8028,20 @@ TNodePtr TSqlIntoTable::Build(const TRule_into_table_stmt& node) {
     }
 
     bool withTruncate = false;
+    TTableHints tableHints;
     if (tableRef.HasBlock2()) {
         auto hints = TableHintsImpl(tableRef.GetBlock2().GetRule_table_hints1());
         if (!hints) {
             Ctx.Error() << "Failed to parse table hints";
             return nullptr;
         }
-
         for (const auto& hint : *hints) {
             if (to_upper(hint.first) == "TRUNCATE") {
                 withTruncate = true;
-            } else {
-                Ctx.Error() << "Unsupported hint: " << hint.first;
-                return nullptr;
             }
         }
+        std::erase_if(*hints, [](const auto &hint) { return to_upper(hint.first) == "TRUNCATE"; });
+        tableHints = std::move(*hints);
     }
 
     TVector<TString> eraseColumns;
@@ -8068,6 +8072,7 @@ TNodePtr TSqlIntoTable::Build(const TRule_into_table_stmt& node) {
     TNodePtr tableKey = BuildTableKey(pos, service, cluster, nameOrAt.second, nameOrAt.first ? "@" : "");
 
     TTableRef table(Ctx.MakeName("table"), service, cluster, tableKey);
+
     Ctx.IncrementMonCounter("sql_insert_clusters", table.Cluster.GetLiteral() ? *table.Cluster.GetLiteral() : "unknown");
 
     auto values = TSqlIntoValues(Ctx, Mode).Build(node.GetRule_into_values_source4(), SqlIntoUserModeStr);
@@ -8079,12 +8084,9 @@ TNodePtr TSqlIntoTable::Build(const TRule_into_table_stmt& node) {
     }
     Ctx.IncrementMonCounter("sql_features", SqlIntoModeStr);
 
-    TNodePtr options;
-    if (eraseColumns) {
-        options = BuildEraseColumns(pos, std::move(eraseColumns));
-    }
-
-    return BuildWriteColumns(pos, Ctx.Scoped, table, ToWriteColumnsMode(SqlIntoMode), std::move(values), std::move(options));
+    return BuildWriteColumns(pos, Ctx.Scoped, table,
+                             ToWriteColumnsMode(SqlIntoMode), std::move(values),
+                             BuildIntoTableOptions(pos, eraseColumns, tableHints));
 }
 
 bool TSqlIntoTable::ValidateServiceName(const TRule_into_table_stmt& node, const TTableRef& table,