Stable 25 3 critical fixes (#26185)

Author: dorooleg

ydb/core/formats/arrow/program/filter.cpp

@@ -104,8 +104,8 @@ TConclusion<IResourceProcessor::EExecutionResult> TFilterProcessor::DoExecute(
     AFL_VERIFY(filter.GetRecordsCountVerified() == inputColumns.front()->GetRecordsCount())("filter", filter.GetRecordsCountVerified())(
                                                      "input", inputColumns.front()->GetRecordsCount());
     if (context.GetLimit()) {
-        context.MutableResources().AddFilter(
-            filter.Cut(context.GetResources().GetRecordsCountRobustVerified(), *context.GetLimit(), context.GetReverse()));
+        context.MutableResources().AddFilter(filter);
+        context.MutableResources().CutFilter(context.GetResources().GetRecordsCountRobustVerified(), *context.GetLimit(), context.GetReverse());
     } else {
         context.MutableResources().AddFilter(filter);
     }

ydb/core/kqp/ut/olap/kqp_olap_ut.cpp

@@ -4703,5 +4703,46 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
             UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
         }
     }
+
+    Y_UNIT_TEST(PredicateWithLimitLostRecords) {
+        auto settings = TKikimrSettings().SetWithSampleTables(false);
+        settings.AppConfig.MutableTableServiceConfig()->SetEnableOlapSink(true);
+        TKikimrRunner kikimr(settings);
+
+        auto queryClient = kikimr.GetQueryClient();
+        {
+            auto status = queryClient.ExecuteQuery(R"(
+                    CREATE TABLE IF NOT EXISTS `olap_table` (
+                        id Uint64 NOT NULL,
+                        message Utf8,
+                        PRIMARY KEY (id)
+                    )
+                    WITH (
+                        STORE = COLUMN,
+                        PARTITION_COUNT = 1
+                    );
+                )", NYdb::NQuery::TTxControl::NoTx()).GetValueSync();
+            UNIT_ASSERT_C(status.IsSuccess(), status.GetIssues().ToString());
+        }
+
+        {
+            TString query = "UPSERT INTO `olap_table` (id, message) VALUES (1, '2'), (2, '2');";
+            auto status = queryClient.ExecuteQuery(query, NYdb::NQuery::TTxControl::BeginTx().CommitTx()).GetValueSync();
+            UNIT_ASSERT_C(status.IsSuccess(), status.GetIssues().ToString());
+        }
+
+        {
+            auto status = queryClient.ExecuteQuery(R"(
+                SELECT id FROM `olap_table`
+                WHERE id = 2 AND message = '2'
+                LIMIT 1;
+            )", NYdb::NQuery::TTxControl::BeginTx().CommitTx()).GetValueSync();
+
+            UNIT_ASSERT_C(status.IsSuccess(), status.GetIssues().ToString());
+            TString result = FormatResultSetYson(status.GetResultSet(0));
+
+            CompareYson(result, R"([[2u]])");
+        }
+    }
 }
 }

ydb/core/kqp/ut/service/kqp_qs_queries_ut.cpp

@@ -3660,6 +3660,48 @@ Y_UNIT_TEST_SUITE(KqpQueryService) {
         }
     }
 
+     Y_UNIT_TEST_TWIN(CTAS_BadKey, IsOlap) {
+        auto setting = NKikimrKqp::TKqpSetting();
+        auto serverSettings = TKikimrSettings()
+            .SetKqpSettings({setting})
+            .SetWithSampleTables(false)
+            .SetEnableTempTables(true);
+        serverSettings.AppConfig.MutableTableServiceConfig()->SetEnableOlapSink(true);
+        serverSettings.AppConfig.MutableTableServiceConfig()->SetEnableCreateTableAs(true);
+        serverSettings.AppConfig.MutableTableServiceConfig()->SetEnableDataShardCreateTableAs(true);
+
+        TKikimrRunner kikimr(serverSettings);
+        auto db = kikimr.GetQueryClient();
+
+        {
+            auto result = db.ExecuteQuery(Sprintf(R"(
+                CREATE TABLE Table (
+                    PRIMARY KEY (Key2, Key)
+                ) WITH (STORE=%s)
+                AS SELECT 1u AS Key, "1" AS Value1, "1" AS Value2;
+                )", IsOlap ? "COLUMN" : "ROW"), TTxControl::NoTx(), TExecuteQuerySettings()).ExtractValueSync();
+
+            UNIT_ASSERT_C(!result.IsSuccess(), result.GetIssues().ToString());
+            UNIT_ASSERT_C(
+                result.GetIssues().ToString().contains("Unknown column 'Key2' specified in key column list"),
+                result.GetIssues().ToString());
+        }
+
+        {
+            auto result = db.ExecuteQuery(Sprintf(R"(
+                CREATE TABLE Table (
+                    PRIMARY KEY (Key, Key2)
+                ) WITH (STORE=%s)
+                AS SELECT 1u AS Key, "1" AS Value1, "1" AS Value2;
+                )", IsOlap ? "COLUMN" : "ROW"), TTxControl::NoTx(), TExecuteQuerySettings()).ExtractValueSync();
+
+            UNIT_ASSERT_C(!result.IsSuccess(), result.GetIssues().ToString());
+            UNIT_ASSERT_C(
+                result.GetIssues().ToString().contains("Unknown column 'Key2' specified in key column list"),
+                result.GetIssues().ToString());
+        }
+     }
+
     Y_UNIT_TEST(CheckIsolationLevelFroPerStatementMode) {
         auto setting = NKikimrKqp::TKqpSetting();
         auto serverSettings = TKikimrSettings().SetKqpSettings({setting});

ydb/core/tx/schemeshard/olap/columns/schema.cpp

@@ -84,7 +84,16 @@ bool TOlapColumnsDescription::ApplyUpdate(
         auto it = orderedKeyColumnIds.begin();
         for (ui32 i = 0; i < orderedKeyColumnIds.size(); ++i, ++it) {
             KeyColumnIds.emplace_back(it->second);
-            Y_ABORT_UNLESS(i == it->first);
+            if (i != it->first) {
+                const TString missedColumnName = i < schemaUpdate.GetPrimaryKeyColumnNames().size() ? schemaUpdate.GetPrimaryKeyColumnNames()[i] : "@unknown";
+                errors.AddError(NKikimrScheme::StatusSchemeError, Sprintf("Unknown column '%s' specified in key column list", missedColumnName.data()));
+                return false;
+            }
+        }
+        if (orderedKeyColumnIds.size() < schemaUpdate.GetPrimaryKeyColumnNames().size()) {
+            const TString missedColumnName = schemaUpdate.GetPrimaryKeyColumnNames()[orderedKeyColumnIds.size()];
+            errors.AddError(NKikimrScheme::StatusSchemeError, Sprintf("Unknown column '%s' specified in key column list", missedColumnName.data()));
+            return false;
         }
         if (KeyColumnIds.empty()) {
             errors.AddError(NKikimrScheme::StatusSchemeError, "No primary key specified");

ydb/core/tx/schemeshard/olap/columns/update.cpp

@@ -405,6 +405,7 @@ void TOlapColumnAdd::ParseFromLocalDB(const NKikimrSchemeOp::TOlapColumnDescript
                 errors.AddError(NKikimrScheme::StatusSchemeError, TStringBuilder() << "Duplicate key column '" << pkKey << "'");
                 return false;
             }
+            PrimaryKeyColumnNames.emplace_back(pkKey);
         }
 
         TSet<TString> columnNames;

ydb/core/tx/schemeshard/olap/columns/update.h

@@ -80,6 +80,7 @@ class TOlapColumnsUpdate {
     YDB_READONLY_DEF(TVector<TOlapColumnAdd>, AddColumns);
     YDB_READONLY_DEF(TSet<TString>, DropColumns);
     YDB_READONLY_DEF(TVector<TOlapColumnDiff>, AlterColumns);
+    YDB_READONLY_DEF(TVector<TString>, PrimaryKeyColumnNames);
 public:
     bool Parse(const NKikimrSchemeOp::TColumnTableSchema& tableSchema, IErrorCollector& errors, bool allowNullKeys = false);
     bool Parse(const NKikimrSchemeOp::TAlterColumnTableSchema& alterRequest, IErrorCollector& errors);