KIKIMR-25621: Introduce multi-column statistics

#### Introduce multi-column statistics support ✎

- Adds support for defining multi-column statistics in table schemas, allowing users to specify statistics for combinations of columns rather than individual columns
- Introduces new SQL syntax for CREATE TABLE and ALTER TABLE statements to declare statistics with optional type specifications
- Implements validation for statistics definitions, ensuring column references are valid and names are unique
- Extends the query processing pipeline to handle statistics declarations during table creation and modification operations
- Updates syntax highlighting and parsing rules to recognize the new statistics keyword and related constructs
- Adds comprehensive unit tests covering various statistics scenarios including multi-column definitions, type specifications, and validation rules

<a href="https://nda.ya.ru/t/qa0kX64r7DqvtN"><font size="2">Autodescription by Yandex Code Assistant</font></a>
commit_hash:5ca3f8721d3e3660f2bb890c1408e420cf7c3e81

12 files changed, 365 insertions, 18 deletions

diff --git a/yql/essentials/core/expr_nodes/yql_expr_nodes.json b/yql/essentials/core/expr_nodes/yql_expr_nodes.json
index cad088b0152..de5c4add7ca 100644
--- a/yql/essentials/core/expr_nodes/yql_expr_nodes.json
+++ b/yql/essentials/core/expr_nodes/yql_expr_nodes.json
@@ -52,6 +52,20 @@
             "ListBase": "TCoIndex"
         },
         {
+            "Name": "TCoStatistics",
+            "Base": "TExprBase",
+            "Match": {"Type": "Tuple"},
+            "Children": [
+                {"Index": 0, "Name": "Name", "Type": "TCoAtom"},
+                {"Index": 1, "Name": "Columns", "Type": "TCoAtomList"},
+                {"Index": 2, "Name": "Types", "Type": "TCoAtomList"}
+            ]
+        },
+        {
+            "Name": "TCoStatisticsList",
+            "ListBase": "TCoStatistics"
+        },
+        {
             "Name": "TCoChangefeed",
             "Base": "TExprBase",
             "Match": {"Type": "Tuple"},
diff --git a/yql/essentials/providers/common/provider/yql_provider.cpp b/yql/essentials/providers/common/provider/yql_provider.cpp
index e72ce7db8a1..d9c411b5661 100644
--- a/yql/essentials/providers/common/provider/yql_provider.cpp
+++ b/yql/essentials/providers/common/provider/yql_provider.cpp
@@ -228,6 +228,7 @@ TWriteTableSettings ParseWriteTableSettings(TExprList node, TExprContext& ctx) {
     TMaybeNode<TCoLambda> update;
     TVector<TCoNameValueTuple> other;
     TVector<TCoIndex> indexes;
+    TVector<TCoStatistics> statistics;
     TVector<TCoChangefeed> changefeeds;
     TMaybeNode<TExprList> columnFamilies;
     TVector<TCoNameValueTuple> tableSettings;
@@ -300,6 +301,22 @@ TWriteTableSettings ParseWriteTableSettings(TExprList node, TExprContext& ctx) {
                     index.Name(InferIndexName(*columnList, ctx));
                 }
                 indexes.push_back(index.Done());
+            } else if (name == "statistics") {
+                YQL_ENSURE(tuple.Value().Maybe<TCoNameValueTupleList>());
+                auto stat = Build<TCoStatistics>(ctx, node.Pos());
+                for (const auto& item : tuple.Value().Cast<TCoNameValueTupleList>()) {
+                    const auto& statItemName = item.Name().Value();
+                    if (statItemName == "statisticsName") {
+                        stat.Name(item.Value().Cast<TCoAtom>());
+                    } else if (statItemName == "statisticsColumns") {
+                        stat.Columns(item.Value().Cast<TCoAtomList>());
+                    } else if (statItemName == "statisticsTypes") {
+                        stat.Types(item.Value().Cast<TCoAtomList>());
+                    } else {
+                        YQL_ENSURE(false, "unknown statistics item " << statItemName);
+                    }
+                }
+                statistics.push_back(stat.Done());
             } else if (name == "changefeed") {
                 YQL_ENSURE(tuple.Value().Maybe<TCoNameValueTupleList>());
                 auto cf = Build<TCoChangefeed>(ctx, node.Pos());
@@ -357,6 +374,10 @@ TWriteTableSettings ParseWriteTableSettings(TExprList node, TExprContext& ctx) {
                           .Add(indexes)
                           .Done();
 
+    const auto& stats = Build<TCoStatisticsList>(ctx, node.Pos())
+                            .Add(statistics)
+                            .Done();
+
     const auto& cfs = Build<TCoChangefeedList>(ctx, node.Pos())
                           .Add(changefeeds)
                           .Done();
@@ -385,6 +406,7 @@ TWriteTableSettings ParseWriteTableSettings(TExprList node, TExprContext& ctx) {
     ret.Filter = filter;
     ret.Update = update;
     ret.Indexes = idx;
+    ret.Statistics = stats;
     ret.Changefeeds = cfs;
     ret.ColumnFamilies = columnFamilies;
     ret.TableSettings = tableProfileSettings;
diff --git a/yql/essentials/providers/common/provider/yql_provider.h b/yql/essentials/providers/common/provider/yql_provider.h
index 6dec1b9636d..757f05b302d 100644
--- a/yql/essentials/providers/common/provider/yql_provider.h
+++ b/yql/essentials/providers/common/provider/yql_provider.h
@@ -44,6 +44,7 @@ struct TWriteTableSettings {
     NNodes::TMaybeNode<NNodes::TCoLambda> Filter;
     NNodes::TMaybeNode<NNodes::TCoLambda> Update;
     NNodes::TMaybeNode<NNodes::TCoIndexList> Indexes;
+    NNodes::TMaybeNode<NNodes::TCoStatisticsList> Statistics;
     NNodes::TMaybeNode<NNodes::TCoChangefeedList> Changefeeds;
     NNodes::TCoNameValueTupleList Other;
     NNodes::TMaybeNode<NNodes::TExprList> ColumnFamilies;
diff --git a/yql/essentials/sql/v1/SQLv1Antlr4.g.in b/yql/essentials/sql/v1/SQLv1Antlr4.g.in
index 2a6cb35093b..4c396fa2b9d 100644
--- a/yql/essentials/sql/v1/SQLv1Antlr4.g.in
+++ b/yql/essentials/sql/v1/SQLv1Antlr4.g.in
@@ -745,6 +745,7 @@ create_table_entry:
   | family_entry
   | changefeed
   | an_id_schema
+  | table_statistics
 ;
 
 create_backup_collection_stmt: CREATE backup_collection create_backup_collection_entries? WITH LPAREN backup_collection_settings RPAREN;
@@ -823,6 +824,8 @@ alter_table_action:
   | alter_table_alter_column_set_default
   | alter_table_alter_column_drop_default
   | alter_table_alter_column_set_encoding
+  | alter_table_add_statistics
+  | alter_table_drop_statistics
 ;
 
 alter_external_table_stmt: ALTER EXTERNAL TABLE simple_table_ref alter_external_table_action (COMMA alter_external_table_action)*;
@@ -854,6 +857,8 @@ alter_table_set_table_setting_compat: SET LPAREN alter_table_setting_entry (COMM
 alter_table_reset_table_setting: RESET LPAREN an_id (COMMA an_id)* RPAREN;
 alter_table_add_index: ADD table_index;
 alter_table_drop_index: DROP INDEX an_id;
+alter_table_add_statistics: ADD table_statistics;
+alter_table_drop_statistics: DROP STATISTICS an_id;
 alter_table_rename_to: RENAME TO an_id_table;
 alter_table_rename_index_to: RENAME INDEX an_id TO an_id;
 alter_table_add_changefeed: ADD changefeed;
@@ -923,6 +928,13 @@ index_setting_value:
     | real
 ;
 
+table_statistics: STATISTICS an_id
+    ON LPAREN an_id_schema (COMMA an_id_schema)* RPAREN
+    with_statistics_types?
+;
+
+with_statistics_types: WITH LPAREN an_id_or_type (COMMA an_id_or_type)* COMMA? RPAREN;
+
 with_compact_settings: WITH LPAREN compact_setting_entry (COMMA compact_setting_entry)* COMMA? RPAREN;
 compact_setting_entry: an_id EQUALS compact_setting_value;
 compact_setting_value: literal_value;
@@ -1656,6 +1668,7 @@ keyword_as_compat:
   | SEQUENCE
   | SOURCE
   | START
+  | STATISTICS
   | STREAMING
   | SUBQUERY
   | SUBSET
@@ -1896,6 +1909,7 @@ keyword_compat: (
   | SEQUENCE
   | SOURCE
   | START
+  | STATISTICS
   | STREAMING
   | SUBQUERY
   | SUBSET
@@ -2286,6 +2300,7 @@ SECRET: S E C R E T;
 SEQUENCE: S E Q U E N C E;
 SOURCE: S O U R C E;
 START: S T A R T;
+STATISTICS: S T A T I S T I C S;
 STREAM: S T R E A M;
 STREAMING: S T R E A M I N G;
 STRUCT: S T R U C T;
diff --git a/yql/essentials/sql/v1/format/sql_format_ut.h b/yql/essentials/sql/v1/format/sql_format_ut.h
index fefef8c7663..d5c63621229 100644
--- a/yql/essentials/sql/v1/format/sql_format_ut.h
+++ b/yql/essentials/sql/v1/format/sql_format_ut.h
@@ -399,6 +399,10 @@ Y_UNIT_TEST(CreateTable) {
          "CREATE TABLE user (\n\tCHANGEFEED user WITH (user = 'foo')\n);\n"},
         {"create table user(changefeed user with (user='foo',user='bar'))",
          "CREATE TABLE user (\n\tCHANGEFEED user WITH (user = 'foo', user = 'bar')\n);\n"},
+        {"create table user(statistics user on (user) with (count_min_sketch))",
+         "CREATE TABLE user (\n\tSTATISTICS user ON (user) WITH (count_min_sketch)\n);\n"},
+        {"create table user(statistics user on (user,user) with (count_min_sketch,histogram))",
+         "CREATE TABLE user (\n\tSTATISTICS user ON (user, user) WITH (count_min_sketch, histogram)\n);\n"},
         {"create table user(user) AS SELECT 1", "CREATE TABLE user (\n\tuser\n)\nAS\nSELECT\n\t1\n;\n"},
         {"create table user(user) AS VALUES (1), (2)", "CREATE TABLE user (\n\tuser\n)\nAS\nVALUES\n\t(1),\n\t(2)\n;\n"},
         {"create table user(foo int32, bar bool ?) inherits (s3:$cluster.xxx) partition by hash(a,b,hash) with (inherits=interval('PT1D') ON logical_time) tablestore tablestore",
@@ -623,6 +627,12 @@ Y_UNIT_TEST(AlterTable) {
          "ALTER TABLE user\n\tADD INDEX idx GLOBAL USING subtype ON (col) COVER (col) WITH (setting = foo, another_setting = 'bar')\n;\n"},
         {"alter table user drop index user",
          "ALTER TABLE user\n\tDROP INDEX user\n;\n"},
+        {"alter table user add statistics user on (user) with (count_min_sketch)",
+         "ALTER TABLE user\n\tADD STATISTICS user ON (user) WITH (count_min_sketch)\n;\n"},
+        {"alter table user add statistics s on (a,b) with (count_min_sketch), drop statistics t",
+         "ALTER TABLE user\n\tADD STATISTICS s ON (a, b) WITH (count_min_sketch),\n\tDROP STATISTICS t\n;\n"},
+        {"alter table user drop statistics user",
+         "ALTER TABLE user\n\tDROP STATISTICS user\n;\n"},
         {"alter table user rename to user",
          "ALTER TABLE user\n\tRENAME TO user\n;\n"},
         {"alter table user add changefeed user with (user = 'foo')",
diff --git a/yql/essentials/sql/v1/node.h b/yql/essentials/sql/v1/node.h
index b11ccb5f481..c9e8b16a799 100644
--- a/yql/essentials/sql/v1/node.h
+++ b/yql/essentials/sql/v1/node.h
@@ -1343,6 +1343,17 @@ struct TIndexDescription {
     TIndexSettings IndexSettings;
 };
 
+struct TStatisticsDescription {
+    explicit TStatisticsDescription(TIdentifier name)
+        : Name(std::move(name))
+    {
+    }
+
+    TIdentifier Name;
+    TVector<TIdentifier> Columns;
+    TVector<TIdentifier> Types; // raw parsed type identifiers; validated later
+};
+
 struct TChangefeedSettings {
     struct TLocalSinkSettings {
         // no special settings
@@ -1382,6 +1393,7 @@ struct TCreateTableParameters {
     TVector<TIdentifier> PartitionByColumns;
     TVector<std::pair<TIdentifier, bool>> OrderByColumns;
     TVector<TIndexDescription> Indexes;
+    TVector<TStatisticsDescription> Statistics;
     TVector<TFamilyEntry> ColumnFamilies;
     TVector<TChangefeedDescription> Changefeeds;
     TTableSettings TableSettings;
@@ -1418,6 +1430,8 @@ struct TAlterTableParameters {
     TVector<TIndexDescription> AddIndexes;
     TVector<TIndexDescription> AlterIndexes;
     TVector<TIdentifier> DropIndexes;
+    TVector<TStatisticsDescription> AddStatistics;
+    TVector<TIdentifier> DropStatistics;
     TMaybe<std::pair<TIdentifier, TIdentifier>> RenameIndexTo;
     TMaybe<TIdentifier> RenameTo;
     TVector<TChangefeedDescription> AddChangefeeds;
@@ -1436,6 +1450,8 @@ struct TAlterTableParameters {
                AddIndexes.empty() &&
                AlterIndexes.empty() &&
                DropIndexes.empty() &&
+               AddStatistics.empty() &&
+               DropStatistics.empty() &&
                !RenameIndexTo.Defined() &&
                !RenameTo.Defined() &&
                AddChangefeeds.empty() &&
diff --git a/yql/essentials/sql/v1/query.cpp b/yql/essentials/sql/v1/query.cpp
index 22fc6e7b9cf..6ec40c25ce5 100644
--- a/yql/essentials/sql/v1/query.cpp
+++ b/yql/essentials/sql/v1/query.cpp
@@ -27,6 +27,26 @@ bool ValidateView(TPosition pos, TContext& ctx, TStringBuf service, TViewDescrip
     return true;
 }
 
+namespace {
+
+bool ValidateColumnIsDefined(TContext& ctx, const TIdentifier& column, const THashSet<TString>& definedColumns, bool allowUndefinedColumns) {
+    if (!allowUndefinedColumns && !definedColumns.contains(column.Name)) {
+        ctx.Error(column.Pos) << "Undefined column: " << column.Name;
+        return false;
+    }
+    return true;
+}
+
+bool ValidateNameIsUnique(TContext& ctx, THashSet<TString>& seenNames, const TIdentifier& name, TStringBuf entityKind) {
+    if (!seenNames.insert(name.Name).second) {
+        ctx.Error(name.Pos) << entityKind << " " << name.Name << " must be defined once";
+        return false;
+    }
+    return true;
+}
+
+} // anonymous namespace
+
 class TUniqueTableKey: public ITableKeys {
 public:
     TUniqueTableKey(TPosition pos, TString service, TDeferredAtom cluster,
@@ -377,6 +397,23 @@ INode::TPtr CreateAlterIndex(const TIndexDescription& index, const INode& node)
     return alterIndexNode;
 }
 
+INode::TPtr CreateStatisticsDesc(const TStatisticsDescription& statistics, const INode& node) {
+    auto statisticsColumns = node.Y();
+    for (const auto& col : statistics.Columns) {
+        statisticsColumns = node.L(statisticsColumns, BuildQuotedAtom(col.Pos, col.Name));
+    }
+    auto statisticsTypes = node.Y();
+    for (const auto& type : statistics.Types) {
+        statisticsTypes = node.L(statisticsTypes, BuildQuotedAtom(type.Pos, type.Name));
+    }
+    const auto& statisticsName = node.Y(node.Q("statisticsName"), BuildQuotedAtom(statistics.Name.Pos, statistics.Name.Name));
+    auto statisticsNode = node.Y(
+        node.Q(statisticsName),
+        node.Q(node.Y(node.Q("statisticsColumns"), node.Q(statisticsColumns))),
+        node.Q(node.Y(node.Q("statisticsTypes"), node.Q(statisticsTypes))));
+    return statisticsNode;
+}
+
 INode::TPtr CreateChangefeedDesc(const TChangefeedDescription& desc, const INode& node) {
     auto settings = node.Y();
     if (desc.Settings.Mode) {
@@ -1165,7 +1202,7 @@ public:
             return false;
         }
 
-        if (!Params_.PkColumns.empty() || !Params_.PartitionByColumns.empty() || !Params_.OrderByColumns.empty() || !Params_.Indexes.empty() || !Params_.Changefeeds.empty())
+        if (!Params_.PkColumns.empty() || !Params_.PartitionByColumns.empty() || !Params_.OrderByColumns.empty() || !Params_.Indexes.empty() || !Params_.Statistics.empty() || !Params_.Changefeeds.empty())
         {
             THashSet<TString> columnsSet;
             for (auto& col : Params_.Columns) {
@@ -1176,8 +1213,7 @@ public:
 
             THashSet<TString> pkColumns;
             for (auto& keyColumn : Params_.PkColumns) {
-                if (!allowUndefinedColumns && !columnsSet.contains(keyColumn.Name)) {
-                    ctx.Error(keyColumn.Pos) << "Undefined column: " << keyColumn.Name;
+                if (!ValidateColumnIsDefined(ctx, keyColumn, columnsSet, allowUndefinedColumns)) {
                     return false;
                 }
                 if (!pkColumns.insert(keyColumn.Name).second) {
@@ -1186,35 +1222,30 @@ public:
                 }
             }
             for (auto& keyColumn : Params_.PartitionByColumns) {
-                if (!allowUndefinedColumns && !columnsSet.contains(keyColumn.Name)) {
-                    ctx.Error(keyColumn.Pos) << "Undefined column: " << keyColumn.Name;
+                if (!ValidateColumnIsDefined(ctx, keyColumn, columnsSet, allowUndefinedColumns)) {
                     return false;
                 }
             }
             for (auto& keyColumn : Params_.OrderByColumns) {
-                if (!allowUndefinedColumns && !columnsSet.contains(keyColumn.first.Name)) {
-                    ctx.Error(keyColumn.first.Pos) << "Undefined column: " << keyColumn.first.Name;
+                if (!ValidateColumnIsDefined(ctx, keyColumn.first, columnsSet, allowUndefinedColumns)) {
                     return false;
                 }
             }
 
             THashSet<TString> indexNames;
             for (const auto& index : Params_.Indexes) {
-                if (!indexNames.insert(index.Name.Name).second) {
-                    ctx.Error(index.Name.Pos) << "Index " << index.Name.Name << " must be defined once";
+                if (!ValidateNameIsUnique(ctx, indexNames, index.Name, "Index")) {
                     return false;
                 }
 
                 for (const auto& indexColumn : index.IndexColumns) {
-                    if (!allowUndefinedColumns && !columnsSet.contains(indexColumn.Name)) {
-                        ctx.Error(indexColumn.Pos) << "Undefined column: " << indexColumn.Name;
+                    if (!ValidateColumnIsDefined(ctx, indexColumn, columnsSet, allowUndefinedColumns)) {
                         return false;
                     }
                 }
 
                 for (const auto& dataColumn : index.DataColumns) {
-                    if (!allowUndefinedColumns && !columnsSet.contains(dataColumn.Name)) {
-                        ctx.Error(dataColumn.Pos) << "Undefined column: " << dataColumn.Name;
+                    if (!ValidateColumnIsDefined(ctx, dataColumn, columnsSet, allowUndefinedColumns)) {
                         return false;
                     }
                 }
@@ -1222,10 +1253,29 @@ public:
 
             THashSet<TString> cfNames;
             for (const auto& cf : Params_.Changefeeds) {
-                if (!cfNames.insert(cf.Name.Name).second) {
-                    ctx.Error(cf.Name.Pos) << "Changefeed " << cf.Name.Name << " must be defined once";
+                if (!ValidateNameIsUnique(ctx, cfNames, cf.Name, "Changefeed")) {
+                    return false;
+                }
+            }
+
+            THashSet<TString> statisticsNames;
+            for (const auto& statistics : Params_.Statistics) {
+                if (!ValidateNameIsUnique(ctx, statisticsNames, statistics.Name, "Statistics")) {
+                    return false;
+                }
+
+                if (statistics.Columns.empty()) {
+                    ctx.Error(statistics.Name.Pos) << "Statistics " << statistics.Name.Name << " must have at least one column";
                     return false;
                 }
+
+                for (const auto& statisticsColumn : statistics.Columns) {
+                    if (!ValidateColumnIsDefined(ctx, statisticsColumn, columnsSet, allowUndefinedColumns)) {
+                        return false;
+                    }
+                }
+
+                // statistics.Types may be empty: WITH is optional and its omission means "all supported statistic types"
             }
         }
 
@@ -1399,6 +1449,11 @@ public:
             opts = L(opts, Q(Y(Q("index"), Q(desc))));
         }
 
+        for (const auto& statistics : Params_.Statistics) {
+            const auto& desc = CreateStatisticsDesc(statistics, *this);
+            opts = L(opts, Q(Y(Q("statistics"), Q(desc))));
+        }
+
         for (const auto& cf : Params_.Changefeeds) {
             const auto& desc = CreateChangefeedDesc(cf, *this);
             opts = L(opts, Q(Y(Q("changefeed"), Q(desc))));
@@ -1856,6 +1911,30 @@ public:
             actions = L(actions, Q(Y(Q("dropIndex"), indexName)));
         }
 
+        {
+            THashSet<TString> statisticsNames;
+            for (const auto& statistics : Params_.AddStatistics) {
+                if (!ValidateNameIsUnique(ctx, statisticsNames, statistics.Name, "Statistics")) {
+                    return false;
+                }
+
+                if (statistics.Columns.empty()) {
+                    ctx.Error(statistics.Name.Pos) << "Statistics " << statistics.Name.Name << " must have at least one column";
+                    return false;
+                }
+
+                // statistics.Types may be empty: WITH is optional and its omission means "all supported statistic types"
+
+                const auto& desc = CreateStatisticsDesc(statistics, *this);
+                actions = L(actions, Q(Y(Q("addStatistics"), Q(desc))));
+            }
+        }
+
+        for (const auto& id : Params_.DropStatistics) {
+            auto statisticsName = BuildQuotedAtom(id.Pos, id.Name);
+            actions = L(actions, Q(Y(Q("dropStatistics"), statisticsName)));
+        }
+
         if (Params_.RenameIndexTo) {
             auto src = BuildQuotedAtom(Params_.RenameIndexTo->first.Pos, Params_.RenameIndexTo->first.Name);
             auto dst = BuildQuotedAtom(Params_.RenameIndexTo->second.Pos, Params_.RenameIndexTo->second.Name);
diff --git a/yql/essentials/sql/v1/sql_query.cpp b/yql/essentials/sql/v1/sql_query.cpp
index 4983e961f60..45d2b5b79c2 100644
--- a/yql/essentials/sql/v1/sql_query.cpp
+++ b/yql/essentials/sql/v1/sql_query.cpp
@@ -2565,7 +2565,9 @@ bool TSqlQuery::AlterTableAction(const TRule_alter_table_action& node, TAlterTab
         case TRule_alter_table_action::kAltAlterTableAction10: {
             // DROP INDEX
             const auto& dropIndex = node.GetAlt_alter_table_action10().GetRule_alter_table_drop_index1();
-            AlterTableDropIndex(dropIndex, params);
+            if (!AlterTableDropIndex(dropIndex, params)) {
+                return false;
+            }
             break;
         }
         case TRule_alter_table_action::kAltAlterTableAction11: {
@@ -2692,6 +2694,22 @@ bool TSqlQuery::AlterTableAction(const TRule_alter_table_action& node, TAlterTab
 
             break;
         }
+        case TRule_alter_table_action::kAltAlterTableAction24: {
+            // ADD STATISTICS
+            const auto& addStatistics = node.GetAlt_alter_table_action24().GetRule_alter_table_add_statistics1();
+            if (!AlterTableAddStatistics(addStatistics, params)) {
+                return false;
+            }
+            break;
+        }
+        case TRule_alter_table_action::kAltAlterTableAction25: {
+            // DROP STATISTICS
+            const auto& dropStatistics = node.GetAlt_alter_table_action25().GetRule_alter_table_drop_statistics1();
+            if (!AlterTableDropStatistics(dropStatistics, params)) {
+                return false;
+            }
+            break;
+        }
         case TRule_alter_table_action::ALT_NOT_SET:
             YQL_ENSURE(false, "Unreachable");
     }
@@ -3025,8 +3043,18 @@ bool TSqlQuery::AlterTableAddIndex(const TRule_alter_table_add_index& node, TAlt
     return CreateTableIndex(node.GetRule_table_index2(), params.AddIndexes);
 }
 
-void TSqlQuery::AlterTableDropIndex(const TRule_alter_table_drop_index& node, TAlterTableParameters& params) {
+bool TSqlQuery::AlterTableDropIndex(const TRule_alter_table_drop_index& node, TAlterTableParameters& params) {
     params.DropIndexes.emplace_back(IdEx(node.GetRule_an_id3(), *this));
+    return !Ctx_.HasPendingErrors;
+}
+
+bool TSqlQuery::AlterTableAddStatistics(const TRule_alter_table_add_statistics& node, TAlterTableParameters& params) {
+    return CreateTableStatistics(node.GetRule_table_statistics2(), params.AddStatistics);
+}
+
+bool TSqlQuery::AlterTableDropStatistics(const TRule_alter_table_drop_statistics& node, TAlterTableParameters& params) {
+    params.DropStatistics.emplace_back(IdEx(node.GetRule_an_id3(), *this));
+    return !Ctx_.HasPendingErrors;
 }
 
 void TSqlQuery::AlterTableRenameTo(const TRule_alter_table_rename_to& node, TAlterTableParameters& params) {
diff --git a/yql/essentials/sql/v1/sql_query.h b/yql/essentials/sql/v1/sql_query.h
index 741a568ee65..b8daa3ca99e 100644
--- a/yql/essentials/sql/v1/sql_query.h
+++ b/yql/essentials/sql/v1/sql_query.h
@@ -39,7 +39,9 @@ private:
     bool AlterTableSetIndexSetting(const TRule_alter_table_set_table_setting_compat& node, TTableSettings& tableSettings, TIndexDescription::TIndexSettings& indexSettings, ETableType tableType);
     bool AlterTableResetTableSetting(const TRule_alter_table_reset_table_setting& node, TTableSettings& tableSettings, ETableType tableType);
     bool AlterTableAddIndex(const TRule_alter_table_add_index& node, TAlterTableParameters& params);
-    void AlterTableDropIndex(const TRule_alter_table_drop_index& node, TAlterTableParameters& params);
+    bool AlterTableDropIndex(const TRule_alter_table_drop_index& node, TAlterTableParameters& params);
+    bool AlterTableAddStatistics(const TRule_alter_table_add_statistics& node, TAlterTableParameters& params);
+    bool AlterTableDropStatistics(const TRule_alter_table_drop_statistics& node, TAlterTableParameters& params);
     void AlterTableRenameTo(const TRule_alter_table_rename_to& node, TAlterTableParameters& params);
     bool AlterTableAddChangefeed(const TRule_alter_table_add_changefeed& node, TAlterTableParameters& params);
     bool AlterTableAlterChangefeed(const TRule_alter_table_alter_changefeed& node, TAlterTableParameters& params);
diff --git a/yql/essentials/sql/v1/sql_translation.cpp b/yql/essentials/sql/v1/sql_translation.cpp
index b95e1c61b56..17596ab4d9c 100644
--- a/yql/essentials/sql/v1/sql_translation.cpp
+++ b/yql/essentials/sql/v1/sql_translation.cpp
@@ -790,6 +790,28 @@ bool TSqlTranslation::CreateTableIndex(const TRule_table_index& node, TVector<TI
     return true;
 }
 
+bool TSqlTranslation::CreateTableStatistics(const TRule_table_statistics& node, TVector<TStatisticsDescription>& statistics) {
+    statistics.emplace_back(IdEx(node.GetRule_an_id2(), *this));
+    auto& stat = statistics.back();
+
+    // ON (columns)
+    stat.Columns.emplace_back(IdEx(node.GetRule_an_id_schema5(), *this));
+    for (const auto& block : node.GetBlock6()) {
+        stat.Columns.emplace_back(IdEx(block.GetRule_an_id_schema2(), *this));
+    }
+
+    // WITH (types) — optional; omission means "all supported statistic types"
+    if (node.HasBlock8()) {
+        const auto& withTypes = node.GetBlock8().GetRule_with_statistics_types1();
+        stat.Types.emplace_back(IdEx(withTypes.GetRule_an_id_or_type3(), *this));
+        for (const auto& block : withTypes.GetBlock4()) {
+            stat.Types.emplace_back(IdEx(block.GetRule_an_id_or_type2(), *this));
+        }
+    }
+
+    return true;
+}
+
 bool TSqlTranslation::ParseDatabaseSettings(const TRule_database_settings& in, THashMap<TString, TNodePtr>& out) {
     if (!ParseDatabaseSetting(in.GetRule_database_setting1(), out)) {
         return false;
@@ -2156,6 +2178,14 @@ bool TSqlTranslation::CreateTableEntry(const TRule_create_table_entry& node, TCr
             params.Columns.push_back({.Pos = pos, .Name = name, .Nullable = true});
             break;
         }
+        case TRule_create_table_entry::kAltCreateTableEntry7: {
+            // table_statistics
+            auto& table_statistics = node.GetAlt_create_table_entry7().GetRule_table_statistics1();
+            if (!CreateTableStatistics(table_statistics, params.Statistics)) {
+                return false;
+            }
+            break;
+        }
         case NSQLv1Generated::TRule_create_table_entry::ALT_NOT_SET:
             YQL_ENSURE(false, "Unreachable");
     }
diff --git a/yql/essentials/sql/v1/sql_translation.h b/yql/essentials/sql/v1/sql_translation.h
index e48063b5f3b..d3dd7d3e311 100644
--- a/yql/essentials/sql/v1/sql_translation.h
+++ b/yql/essentials/sql/v1/sql_translation.h
@@ -236,6 +236,7 @@ protected:
     bool ResetTableSettingsEntry(const TIdentifier& id, TTableSettings& settings, ETableType tableType);
 
     bool CreateTableIndex(const TRule_table_index& node, TVector<TIndexDescription>& indexes);
+    bool CreateTableStatistics(const TRule_table_statistics& node, TVector<TStatisticsDescription>& statistics);
     bool FillIndexSettings(const TRule_with_index_settings& settingsNode, TIndexDescription::TIndexSettings& indexSettings);
     bool AddIndexSetting(const TIdentifier& id, const TRule_index_setting_value& value, TIndexDescription::TIndexSettings& indexSettings);
     bool AddCompactSetting(const TIdentifier& id, const TRule_compact_setting_value& value, TCompactEntry& compactEntry);
diff --git a/yql/essentials/sql/v1/sql_ut_common.h b/yql/essentials/sql/v1/sql_ut_common.h
index f919798f838..05e7d38b87f 100644
--- a/yql/essentials/sql/v1/sql_ut_common.h
+++ b/yql/essentials/sql/v1/sql_ut_common.h
@@ -4294,6 +4294,135 @@ Y_UNIT_TEST(AlterTableDropIndexIsCorrect) {
     UNIT_ASSERT_C(result.IsOk(), result.Issues.ToString());
 }
 
+Y_UNIT_TEST(CreateTableWithStatisticsIsSupported) {
+    auto res = SqlToYql(R"sql(
+        USE ydb;
+        CREATE TABLE table (
+            k Uint64 NOT NULL,
+            a Uint64,
+            b Utf8,
+            PRIMARY KEY (k),
+            STATISTICS s ON (a, b) WITH (COUNT_MIN_SKETCH)
+        );
+    )sql");
+    UNIT_ASSERT_C(res.IsOk(), Err2Str(res));
+
+    TVerifyLineFunc verifyLine = [](const TString& word, const TString& line) {
+        if (word == "Write") {
+            UNIT_ASSERT_STRING_CONTAINS(line, "statisticsName");
+            UNIT_ASSERT_STRING_CONTAINS(line, "statisticsColumns");
+            UNIT_ASSERT_STRING_CONTAINS(line, "statisticsTypes");
+            UNIT_ASSERT_STRING_CONTAINS(line, "COUNT_MIN_SKETCH");
+        }
+    };
+    TWordCountHive elementStat = {"Write"};
+    VerifyProgram(res, elementStat, verifyLine);
+    UNIT_ASSERT_VALUES_EQUAL(1, elementStat["Write"]);
+}
+
+Y_UNIT_TEST(CreateTableWithStatisticsOnUnknownColumnFails) {
+    ExpectFailWithError(R"sql(
+        USE ydb;
+        CREATE TABLE table (
+            k Uint64 NOT NULL,
+            a Uint64,
+            PRIMARY KEY (k),
+            STATISTICS s ON (missing) WITH (COUNT_MIN_SKETCH)
+        );
+    )sql", "<main>:7:30: Error: Undefined column: missing\n");
+}
+
+Y_UNIT_TEST(CreateTableWithStatisticsWithoutWithIsSupported) {
+    auto res = SqlToYql(R"sql(
+        USE ydb;
+        CREATE TABLE table (
+            k Uint64 NOT NULL,
+            a Uint64,
+            b Utf8,
+            PRIMARY KEY (k),
+            STATISTICS s ON (a, b)
+        );
+    )sql");
+    UNIT_ASSERT_C(res.IsOk(), Err2Str(res));
+}
+
+Y_UNIT_TEST(CreateTableWithStatisticsEmptyWithFails) {
+    // Empty WITH () is rejected by the grammar: the statistic type list must be non-empty.
+    ExpectFailWithFuzzyError(R"sql(
+        USE ydb;
+        CREATE TABLE table (
+            k Uint64 NOT NULL,
+            a Uint64,
+            PRIMARY KEY (k),
+            STATISTICS s ON (a) WITH ()
+        );
+    )sql", "<main>:7:38: Error: mismatched input");
+}
+
+Y_UNIT_TEST(AlterTableAddStatisticsIsSupported) {
+    auto res = SqlToYql(R"sql(
+        USE ydb;
+        ALTER TABLE table
+            ADD STATISTICS s1 ON (a, b) WITH (COUNT_MIN_SKETCH),
+            ADD STATISTICS s2 ON (a) WITH (COUNT_MIN_SKETCH);
+    )sql");
+    UNIT_ASSERT_C(res.IsOk(), Err2Str(res));
+
+    TVerifyLineFunc verifyLine = [](const TString& word, const TString& line) {
+        if (word == "Write") {
+            UNIT_ASSERT_STRING_CONTAINS(line, "addStatistics");
+        }
+    };
+    TWordCountHive elementStat = {"Write"};
+    VerifyProgram(res, elementStat, verifyLine);
+}
+
+Y_UNIT_TEST(AlterTableAddStatisticsDuplicateFails) {
+    ExpectFailWithError(R"sql(
+        USE ydb;
+        ALTER TABLE table
+            ADD STATISTICS s1 ON (a) WITH (COUNT_MIN_SKETCH),
+            ADD STATISTICS s1 ON (b) WITH (COUNT_MIN_SKETCH);
+    )sql", "<main>:5:28: Error: Statistics s1 must be defined once\n");
+}
+
+Y_UNIT_TEST(AlterTableAddStatisticsWithoutWithIsSupported) {
+    auto res = SqlToYql(R"sql(
+        USE ydb;
+        ALTER TABLE table
+            ADD STATISTICS s1 ON (a, b);
+    )sql");
+    UNIT_ASSERT_C(res.IsOk(), Err2Str(res));
+}
+
+Y_UNIT_TEST(AlterTableAddStatisticsEmptyWithFails) {
+    // Empty WITH () is rejected by the grammar: the statistic type list must be non-empty.
+    ExpectFailWithFuzzyError(R"sql(
+        USE ydb;
+        ALTER TABLE table
+            ADD STATISTICS s1 ON (a) WITH ();
+    )sql", "<main>:4:43: Error: mismatched input");
+}
+
+Y_UNIT_TEST(AlterTableDropStatisticsIsSupported) {
+    auto res = SqlToYql("USE ydb; ALTER TABLE table DROP STATISTICS s");
+    UNIT_ASSERT_C(res.IsOk(), Err2Str(res));
+
+    TVerifyLineFunc verifyLine = [](const TString& word, const TString& line) {
+        if (word == "Write") {
+            UNIT_ASSERT_STRING_CONTAINS(line, "dropStatistics");
+        }
+    };
+    TWordCountHive elementStat = {"Write"};
+    VerifyProgram(res, elementStat, verifyLine);
+}
+
+Y_UNIT_TEST(StatisticsKeywordIsNotReserved) {
+    // STATISTICS is a non-reserved keyword and must remain usable as an identifier.
+    UNIT_ASSERT(SqlToYql("SELECT 1 AS statistics;").IsOk());
+    UNIT_ASSERT(SqlToYql("USE ydb; SELECT statistics FROM plato.Input;").IsOk());
+}
+
 Y_UNIT_TEST(CreateTableWithLocalBloomFilterAndDropIndexIsCorrect) {
     const auto result = SqlToYql(R"sql(
         USE ydb;