22-2: Don't allow users to configure shards with more than 2GB of data, KIKIMR-14850

Merge from trunk: r9424352

REVIEW: 2524848
x-ydb-stable-ref: a6c50a6d3539aee29bf4d516187c06da9bb95fc0

4 files changed, 67 insertions, 20 deletions

diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_split_merge.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_split_merge.cpp
index fde555580f..9a4bfc7daf 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_split_merge.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_split_merge.cpp
@@ -649,7 +649,11 @@ public:
             return false;
         }
 
-        if (tableInfo->GetExpectedPartitionCount() + count - 1 > tableInfo->GetMaxPartitionsCount()) {
+        auto srcShardIdx = tableInfo->GetPartitions()[srcPartitionIdx].ShardIdx;
+
+        if (tableInfo->GetExpectedPartitionCount() + count - 1 > tableInfo->GetMaxPartitionsCount() &&
+            !tableInfo->IsForceSplitBySizeShardIdx(srcShardIdx))
+        {
             errStr = "Reached MaxPartitionsCount limit: " + ToString(tableInfo->GetMaxPartitionsCount());
             return false;
         }
@@ -677,7 +681,6 @@ public:
 
         op.SplitDescription = std::make_shared<NKikimrTxDataShard::TSplitMergeDescription>();
         auto* srcRange = op.SplitDescription->AddSourceRanges();
-        auto srcShardIdx = tableInfo->GetPartitions()[srcPartitionIdx].ShardIdx;
         srcRange->SetShardIdx(ui64(srcShardIdx.GetLocalId()));
         srcRange->SetTabletID(ui64(context.SS->ShardInfos[srcShardIdx].TabletID));
         srcRange->SetKeyRangeEnd(tableInfo->GetPartitions()[srcPartitionIdx].EndOfRange);
diff --git a/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp b/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp
index 21007ff77d..a464fc39f9 100644
--- a/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp
@@ -287,17 +287,18 @@ bool TTxStorePartitionStats::Execute(TTransactionContext& txc, const TActorConte
     ui64 dataSizeResolution = 0; // Datashard will use default resolution
     ui64 rowCountResolution = 0; // Datashard will use default resolution
     bool collectKeySample = false;
-    if (table->CheckFastSplitForPartition(Self->SplitSettings, shardIdx, dataSize, rowCount)) {
+    if (table->ShouldSplitBySize(dataSize)) {
+        // We would like to split by size and do this no matter how many partitions there are
+    } else if (table->GetPartitions().size() >= table->GetMaxPartitionsCount()) {
+        // We cannot split as there are max partitions already
+        return true;
+    } else if (table->CheckFastSplitForPartition(Self->SplitSettings, shardIdx, dataSize, rowCount)) {
         dataSizeResolution = Max<ui64>(dataSize / 100, 100*1024);
         rowCountResolution = Max<ui64>(rowCount / 100, 1000);
         collectKeySample = true;
     } else if (table->CheckSplitByLoad(Self->SplitSettings, shardIdx, dataSize, rowCount)) {
         collectKeySample = true;
-    } else if (dataSize < table->GetShardSizeToSplit()) {
-        return true;
-    }
-
-    if (table->GetPartitions().size() >= table->GetMaxPartitionsCount()) {
+    } else {
         return true;
     }
 
diff --git a/ydb/core/tx/schemeshard/schemeshard__table_stats_histogram.cpp b/ydb/core/tx/schemeshard/schemeshard__table_stats_histogram.cpp
index a5fb3bdac4..4f15001413 100644
--- a/ydb/core/tx/schemeshard/schemeshard__table_stats_histogram.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__table_stats_histogram.cpp
@@ -310,17 +310,17 @@ bool TTxPartitionHistogram::Execute(TTransactionContext& txc, const TActorContex
     auto shardIdx = Self->TabletIdToShardIdx[datashardId];
 
     ESplitReason splitReason = ESplitReason::NO_SPLIT;
-    if (table->CheckFastSplitForPartition(Self->SplitSettings, shardIdx, dataSize, rowCount)) {
+    if (table->ShouldSplitBySize(dataSize)) {
+        splitReason = ESplitReason::SPLIT_BY_SIZE;
+    }
+
+    if (splitReason == ESplitReason::NO_SPLIT && table->CheckFastSplitForPartition(Self->SplitSettings, shardIdx, dataSize, rowCount)) {
         const TTableInfo* parentTable = Self->GetMainTableForIndex(tableId);
         if (parentTable && table->GetPartitions().size() < parentTable->GetPartitions().size()) {
             splitReason = ESplitReason::FAST_SPLIT_INDEX;
         }
     }
 
-    if (splitReason == ESplitReason::NO_SPLIT && dataSize >= table->GetShardSizeToSplit()) {
-        splitReason = ESplitReason::SPLIT_BY_SIZE;
-    }
-
     if (splitReason == ESplitReason::NO_SPLIT && table->CheckSplitByLoad(Self->SplitSettings, shardIdx, dataSize, rowCount)) {
         splitReason = ESplitReason::SPLIT_BY_LOAD;
     }
@@ -329,8 +329,7 @@ bool TTxPartitionHistogram::Execute(TTransactionContext& txc, const TActorContex
         return true;
     }
 
-
-    if (table->GetPartitions().size() >= table->GetMaxPartitionsCount()) {
+    if (splitReason != ESplitReason::SPLIT_BY_SIZE && table->GetPartitions().size() >= table->GetMaxPartitionsCount()) {
         return true;
     }
 
diff --git a/ydb/core/tx/schemeshard/schemeshard_info_types.h b/ydb/core/tx/schemeshard/schemeshard_info_types.h
index e241dd170b..6147a18fd9 100644
--- a/ydb/core/tx/schemeshard/schemeshard_info_types.h
+++ b/ydb/core/tx/schemeshard/schemeshard_info_types.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include "schemeshard.h"
 #include "schemeshard_types.h"
 #include "schemeshard_tx_infly.h"
 #include "schemeshard_path_element.h"
@@ -565,11 +566,26 @@ public:
     bool CheckSplitByLoad(const TSplitSettings& splitSettings, TShardIdx shardIdx, ui64 dataSize, ui64 rowCount) const;
 
     bool IsSplitBySizeEnabled() const {
-        return PartitionConfig().GetPartitioningPolicy().GetSizeToSplit() != 0;
+        // Auto split is always enabled, unless table is using external blobs
+        return !PartitionConfigHasExternalBlobsEnabled(PartitionConfig());
     }
 
     bool IsMergeBySizeEnabled() const {
-        return IsSplitBySizeEnabled() && PartitionConfig().GetPartitioningPolicy().GetMinPartitionsCount() != 0;
+        // Auto merge is only enabled when auto split is also enabled
+        if (!IsSplitBySizeEnabled()) {
+            return false;
+        }
+        // We want auto merge enabled when user has explicitly specified the
+        // size to split and the minimum partitions count.
+        if (PartitionConfig().GetPartitioningPolicy().GetSizeToSplit() > 0 &&
+            PartitionConfig().GetPartitioningPolicy().GetMinPartitionsCount() != 0)
+        {
+            return true;
+        }
+        // We also want auto merge enabled when table has more shards than the
+        // specified maximum number of partitions. This way when something
+        // splits by size over the limit we merge some smaller partitions.
+        return Partitions.size() > GetMaxPartitionsCount();
     }
 
     bool IsSplitByLoadEnabled() const {
@@ -580,11 +596,19 @@ public:
         return IsSplitByLoadEnabled();
     }
 
+    static ui64 GetShardForceSizeToSplit() {
+        return 2ULL * 1024 * 1024 * 1024; // 2GiB
+    }
+
     ui64 GetShardSizeToSplit() const {
+        if (!IsSplitBySizeEnabled()) {
+            return Max<ui64>();
+        }
         ui64 threshold = PartitionConfig().GetPartitioningPolicy().GetSizeToSplit();
-        return threshold == 0 ?
-            Max<ui64>() :   // Autosplit is OFF if theshold is not specified
-            threshold;
+        if (threshold == 0 || threshold >= GetShardForceSizeToSplit()) {
+            return GetShardForceSizeToSplit();
+        }
+        return threshold;
     }
 
     ui64 GetSizeToMerge() const {
@@ -606,6 +630,26 @@ public:
         return val == 0 ? 32*1024 : val;
     }
 
+    bool IsForceSplitBySizeShardIdx(TShardIdx shardIdx) const {
+        if (!Stats.PartitionStats.contains(shardIdx)) {
+            return false;
+        }
+        const auto& stats = Stats.PartitionStats.at(shardIdx);
+        return stats.DataSize >= GetShardForceSizeToSplit();
+    }
+
+    bool ShouldSplitBySize(ui64 dataSize) const {
+        if (!IsSplitBySizeEnabled()) {
+            return false;
+        }
+        // When shard is over the maximum size we split even when over max partitions
+        if (dataSize >= GetShardForceSizeToSplit()) {
+            return true;
+        }
+        // Otherwise we split when we may add one more partition
+        return Partitions.size() < GetMaxPartitionsCount() && dataSize >= GetShardSizeToSplit();
+    }
+
     bool NeedRecreateParts() const {
         if (!AlterData) {
             return false;