[25.2] Impl methods to detect BulkUpsert with DEFAULT columns and a flag to disable (#27596) (#28871)

Author: dahbka-lis

ydb/core/kqp/ut/scheme/kqp_constraints_ut.cpp

@@ -782,12 +782,15 @@ Y_UNIT_TEST_SUITE(KqpConstraints) {
         TKikimrRunner kikimr(TKikimrSettings()
             .SetUseRealThreads(false)
             .SetEnableAddColumsWithDefaults(true)
+            .SetDisableMissingDefaultColumnsInBulkUpsert(true)
             .SetWithSampleTables(false));
 
         auto db = kikimr.RunCall([&] { return kikimr.GetQueryClient(); } );
         auto session = kikimr.RunCall([&] { return db.GetSession().GetValueSync().GetSession(); } );
         auto querySession = kikimr.RunCall([&] { return db.GetSession().GetValueSync().GetSession(); } );
 
+        auto tableClient = kikimr.RunCall([&] { return kikimr.GetTableClient(); } );
+
         auto& runtime = *kikimr.GetTestServer().GetRuntime();
 
         {
@@ -860,7 +863,7 @@ Y_UNIT_TEST_SUITE(KqpConstraints) {
 
         auto alterQuery = R"(
             ALTER TABLE `/Root/AddNonColumnDoesnotReturnInternalError`
-            ADD COLUMN Value3 Int32 NOT NULL DEFAULT 7;
+            ADD COLUMN Value3 Int32 DEFAULT 7;
         )";
 
         auto alterFuture = kikimr.RunInThreadPool([&] { return session.ExecuteQuery(alterQuery, TTxControl::NoTx()).GetValueSync(); });
@@ -924,8 +927,8 @@ Y_UNIT_TEST_SUITE(KqpConstraints) {
 
         auto result = runtime.WaitFuture(alterFuture);
         fCompareTable(R"([
-            [1u;"Changed";"Updated";7];
-            [2u;"New";"text";7]
+            [1u;"Changed";"Updated";[7]];
+            [2u;"New";"text";[7]]
         ])");
     }
 

ydb/core/protos/feature_flags.proto

@@ -213,4 +213,5 @@ message TFeatureFlags {
     optional bool EnableTopicCompactificationByKey = 196 [default = true];
     optional bool EnableCompactionOverloadDetection = 197 [default = true];
     reserved 198; // DisableColumnShardBulkUpsertRequireAllColumns
+    optional bool DisableMissingDefaultColumnsInBulkUpsert = 215 [default = false];
 }

ydb/core/testlib/basics/feature_flags.h

@@ -78,6 +78,7 @@ class TTestFeatureFlagsHolder {
     FEATURE_FLAG_SETTER(EnablePermissionsExport)
     FEATURE_FLAG_SETTER(EnableLocalDBBtreeIndex)
     FEATURE_FLAG_SETTER(EnableSharedMetadataAccessorCache)
+    FEATURE_FLAG_SETTER(DisableMissingDefaultColumnsInBulkUpsert)
 
     #undef FEATURE_FLAG_SETTER
 };

ydb/core/tx/tx_proxy/upload_rows_common_impl.h

@@ -407,6 +407,7 @@ class TUploadRowsBase : public TActorBootstrapped<TUploadRowsBase<DerivedActivit
         THashMap<TString, ui32> columnByName;
         THashSet<TString> keyColumnsLeft;
         THashSet<TString> notNullColumnsLeft = entry.NotNullColumns;
+        THashSet<TString> defaultColumnsLeft;
         SrcColumns.reserve(entry.Columns.size());
         THashSet<TString> HasInternalConversion;
 
@@ -426,6 +427,10 @@ class TUploadRowsBase : public TActorBootstrapped<TUploadRowsBase<DerivedActivit
                 keyColumnIds[keyOrder] = id;
                 keyColumnsLeft.insert(name);
             }
+
+            if (colInfo.IsDefaultFromLiteral()) {
+                defaultColumnsLeft.insert(name);
+            }
         }
 
         if (entry.ColumnTableInfo) {
@@ -524,6 +529,10 @@ class TUploadRowsBase : public TActorBootstrapped<TUploadRowsBase<DerivedActivit
                 NotNullColumns.emplace(ci.Name);
             }
 
+            if (defaultColumnsLeft.contains(ci.Name)) {
+                defaultColumnsLeft.erase(ci.Name);
+            }
+
             if (ci.KeyOrder != -1) {
                 KeyColumnPositions[ci.KeyOrder] = TFieldDescription{ci.Id, ci.Name, (ui32)pos, ci.PType, pgTypeMod, notNull};
                 keyColumnsLeft.erase(ci.Name);
@@ -608,6 +617,21 @@ class TUploadRowsBase : public TActorBootstrapped<TUploadRowsBase<DerivedActivit
             return TConclusionStatus::Fail(Sprintf("Missing not null columns: %s", JoinSeq(", ", notNullColumnsLeft).c_str()));
         }
 
+        if (!defaultColumnsLeft.empty() && UpsertIfExists) {
+            // some default columns are not specified in the request, but upsert will only update existing rows
+            // and only the columns specified in the request will be updated; unspecified default columns will not be changed.
+            defaultColumnsLeft.clear();
+        }
+
+        if (!defaultColumnsLeft.empty()) {
+            if (AppData(ctx)->FeatureFlags.GetDisableMissingDefaultColumnsInBulkUpsert()) {
+                return TConclusionStatus::Fail(Sprintf("Missing default columns: %s", JoinSeq(", ", defaultColumnsLeft).c_str()));
+            }
+
+            UploadCounters.OnMissingDefaultColumns();
+            LOG_WARN_S(ctx, NKikimrServices::RPC_REQUEST, "Missing default columns: " << JoinSeq(", ", defaultColumnsLeft).c_str());
+        }
+
         TConclusionStatus res = TConclusionStatus::Success();
         if (isColumnTable && HasAppData() && AppDataVerified().ColumnShardConfig.GetBulkUpsertRequireAllColumns()) {
             res = CheckRequiredColumns(entry, *reqColumns);

ydb/core/tx/tx_proxy/upload_rows_counters.cpp

@@ -99,5 +99,6 @@ TUploadCounters::TUploadCounters()
             WrittenBytes = TBase::GetDeriviative("Replies/WrittenBytes");
             FailedBytes = TBase::GetDeriviative("Replies/FailedBytes");
             RequestsBytes = TBase::GetDeriviative("Requests/Bytes");
+            MissingDefaultColumnsCount = TBase::GetDeriviative("MissingDefaultColumns/Count");
 }
 }

ydb/core/tx/tx_proxy/upload_rows_counters.h

@@ -76,6 +76,8 @@ class TUploadCounters: public NColumnShard::TCommonCountersOwner {
     NMonitoring::TDynamicCounters::TCounterPtr FailedBytes;
     NMonitoring::TDynamicCounters::TCounterPtr RequestsBytes;
 
+    NMonitoring::TDynamicCounters::TCounterPtr MissingDefaultColumnsCount;
+
     THashMap<TUploadStatus, NMonitoring::TDynamicCounters::TCounterPtr, TUploadStatus::THasher> CodesCount;
 
     NMonitoring::TDynamicCounters::TCounterPtr GetCodeCounter(const TUploadStatus& status);
@@ -142,6 +144,10 @@ class TUploadCounters: public NColumnShard::TCommonCountersOwner {
         PackageSizeCountByRecords->Collect(rowsCount);
         RequestsBytes->Add(requestBytes);
     }
+
+    void OnMissingDefaultColumns() {
+        MissingDefaultColumnsCount->Inc();
+    }
 };
 
 }   // namespace NKikimr