This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new a433d82b316 branch-4.0: [fix](cloud) Fix disable_auto_compaction
property not taking effect in cloud mode #60544 (#60771)
a433d82b316 is described below
commit a433d82b31616f0225b9808945274d9d775e2ab0
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Feb 16 01:59:39 2026 +0800
branch-4.0: [fix](cloud) Fix disable_auto_compaction property not taking
effect in cloud mode #60544 (#60771)
Cherry-picked from #60544
Co-authored-by: Jimmy <[email protected]>
---
be/src/cloud/cloud_tablet.cpp | 6 +
be/test/cloud/cloud_tablet_test.cpp | 267 +++++++++++++++++++++
cloud/src/meta-service/meta_service.cpp | 33 ++-
cloud/test/schema_kv_test.cpp | 167 ++++++++++++-
...test_cloud_alter_disable_auto_compaction.groovy | 193 +++++++++++++++
5 files changed, 656 insertions(+), 10 deletions(-)
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
index 020c197dfe6..237b5187abf 100644
--- a/be/src/cloud/cloud_tablet.cpp
+++ b/be/src/cloud/cloud_tablet.cpp
@@ -1590,6 +1590,12 @@ Status CloudTablet::sync_meta() {
_tablet_meta->set_time_series_compaction_level_threshold(
new_time_series_compaction_level_threshold);
}
+ // Sync disable_auto_compaction (stored in tablet_schema)
+ auto new_disable_auto_compaction =
tablet_meta->tablet_schema()->disable_auto_compaction();
+ if (_tablet_meta->tablet_schema()->disable_auto_compaction() !=
new_disable_auto_compaction) {
+ _tablet_meta->mutable_tablet_schema()->set_disable_auto_compaction(
+ new_disable_auto_compaction);
+ }
return Status::OK();
}
diff --git a/be/test/cloud/cloud_tablet_test.cpp
b/be/test/cloud/cloud_tablet_test.cpp
index 0e9c39144c6..2c375dc1bca 100644
--- a/be/test/cloud/cloud_tablet_test.cpp
+++ b/be/test/cloud/cloud_tablet_test.cpp
@@ -28,6 +28,8 @@
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_warm_up_manager.h"
+#include "common/config.h"
+#include "cpp/sync_point.h"
#include "olap/rowset/rowset.h"
#include "olap/rowset/rowset_factory.h"
#include "olap/rowset/rowset_meta.h"
@@ -673,4 +675,269 @@ TEST_F(CloudTabletWarmUpStateTest,
TestJobCanOverrideDoneStateFromJob) {
expected_state = WarmUpState {WarmUpTriggerSource::JOB,
WarmUpProgress::DOING};
EXPECT_EQ(state, expected_state);
}
+
+// Test class for sync_meta functionality
+class CloudTabletSyncMetaTest : public testing::Test {
+public:
+ CloudTabletSyncMetaTest() : _engine(CloudStorageEngine(EngineOptions {}))
{}
+
+ void SetUp() override {
+ config::enable_file_cache = true;
+
+ // Use incrementing tablet_id to create unique schema cache keys for
each test
+ // This avoids cache pollution between tests
+ int64_t unique_tablet_id = 15673 + _test_counter++;
+
+ // Create tablet meta with a schema that has disable_auto_compaction =
false
+ TTabletSchema tablet_schema;
+ tablet_schema.__set_disable_auto_compaction(false);
+ // Add a unique column to ensure unique cache key
+ TColumn col;
+ col.__set_column_name("test_col_" + std::to_string(unique_tablet_id));
+ col.__set_column_type(TColumnType());
+ col.column_type.__set_type(TPrimitiveType::INT);
+ col.__set_is_key(true);
+ col.__set_aggregation_type(TAggregationType::NONE);
+ col.__set_col_unique_id(0);
+ tablet_schema.__set_columns({col});
+ tablet_schema.__set_keys_type(TKeysType::DUP_KEYS);
+
+ // Use column ordinal 0 -> unique_id 0 mapping
+ _tablet_meta.reset(new TabletMeta(1, 2, unique_tablet_id, 15674, 4, 5,
tablet_schema, 1,
+ {{0, 0}}, UniqueId(9, 10),
TTabletType::TABLET_TYPE_DISK,
+ TCompressionType::LZ4F));
+
+ _tablet =
+ std::make_shared<CloudTablet>(_engine,
std::make_shared<TabletMeta>(*_tablet_meta));
+
+ _current_tablet_id = unique_tablet_id;
+ }
+
+ void TearDown() override { config::enable_file_cache = false; }
+
+protected:
+ // Helper to create a unique TabletMeta for mock responses
+ TabletMetaSharedPtr createMockTabletMeta(bool disable_auto_compaction,
+ const std::string&
compaction_policy = "size_based") {
+ TTabletSchema new_schema;
+ new_schema.__set_disable_auto_compaction(disable_auto_compaction);
+ TColumn col;
+ col.__set_column_name("test_col_" +
std::to_string(_current_tablet_id));
+ col.__set_column_type(TColumnType());
+ col.column_type.__set_type(TPrimitiveType::INT);
+ col.__set_is_key(true);
+ col.__set_aggregation_type(TAggregationType::NONE);
+ col.__set_col_unique_id(0);
+ new_schema.__set_columns({col});
+ new_schema.__set_keys_type(TKeysType::DUP_KEYS);
+
+ TabletMetaSharedPtr meta;
+ meta.reset(new TabletMeta(1, 2, _current_tablet_id, 15674, 4, 5,
new_schema, 1, {{0, 0}},
+ UniqueId(9, 10),
TTabletType::TABLET_TYPE_DISK,
+ TCompressionType::LZ4F, 0, false,
std::nullopt,
+ compaction_policy));
+ return meta;
+ }
+
+ TabletMetaSharedPtr _tablet_meta;
+ std::shared_ptr<CloudTablet> _tablet;
+ CloudStorageEngine _engine;
+ int64_t _current_tablet_id;
+ static inline int _test_counter = 0;
+};
+
+// Test sync_meta syncs disable_auto_compaction from false to true
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaDisableAutoCompactionFalseToTrue) {
+ // Verify initial state: disable_auto_compaction = false
+
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ // Create mock tablet meta with disable_auto_compaction = true
+ auto mock_tablet_meta = createMockTabletMeta(true);
+
+ // Mock get_tablet_meta to return tablet_meta with disable_auto_compaction
= true
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+
+ // Tell the sync point to return with Status::OK()
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+
+ // Call sync_meta
+ Status st = _tablet->sync_meta();
+ EXPECT_TRUE(st.ok());
+
+ // Verify disable_auto_compaction has been synced to true
+
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
+// Test sync_meta syncs disable_auto_compaction from true to false
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaDisableAutoCompactionTrueToFalse) {
+ // Create a tablet with disable_auto_compaction = true from the start
+ // We need to create a fresh tablet with true to avoid cache pollution
issues
+ TTabletSchema initial_schema;
+ initial_schema.__set_disable_auto_compaction(true);
+ TColumn col;
+ col.__set_column_name("test_col_true_to_false_" +
std::to_string(_current_tablet_id));
+ col.__set_column_type(TColumnType());
+ col.column_type.__set_type(TPrimitiveType::INT);
+ col.__set_is_key(true);
+ col.__set_aggregation_type(TAggregationType::NONE);
+ col.__set_col_unique_id(0);
+ initial_schema.__set_columns({col});
+ initial_schema.__set_keys_type(TKeysType::DUP_KEYS);
+
+ TabletMetaSharedPtr tablet_meta_true;
+ tablet_meta_true.reset(new TabletMeta(1, 2, _current_tablet_id + 1000,
15674, 4, 5,
+ initial_schema, 1, {{0, 0}},
UniqueId(9, 10),
+ TTabletType::TABLET_TYPE_DISK,
TCompressionType::LZ4F));
+ _tablet =
+ std::make_shared<CloudTablet>(_engine,
std::make_shared<TabletMeta>(*tablet_meta_true));
+
+ // Verify initial state: disable_auto_compaction = true
+
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ // Create mock tablet meta with disable_auto_compaction = false (with
matching column name)
+ TTabletSchema mock_schema;
+ mock_schema.__set_disable_auto_compaction(false);
+ TColumn mock_col;
+ mock_col.__set_column_name("test_col_true_to_false_" +
std::to_string(_current_tablet_id));
+ mock_col.__set_column_type(TColumnType());
+ mock_col.column_type.__set_type(TPrimitiveType::INT);
+ mock_col.__set_is_key(true);
+ mock_col.__set_aggregation_type(TAggregationType::NONE);
+ mock_col.__set_col_unique_id(0);
+ mock_schema.__set_columns({mock_col});
+ mock_schema.__set_keys_type(TKeysType::DUP_KEYS);
+
+ TabletMetaSharedPtr mock_tablet_meta;
+ mock_tablet_meta.reset(new TabletMeta(1, 2, _current_tablet_id + 1000,
15674, 4, 5, mock_schema,
+ 1, {{0, 0}}, UniqueId(9, 10),
+ TTabletType::TABLET_TYPE_DISK,
TCompressionType::LZ4F));
+
+ // Mock get_tablet_meta to return tablet_meta with disable_auto_compaction
= false
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+
+ // Tell the sync point to return with Status::OK()
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+
+ // Call sync_meta
+ Status st = _tablet->sync_meta();
+ EXPECT_TRUE(st.ok());
+
+ // Verify disable_auto_compaction has been synced to false
+
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
+// Test sync_meta when disable_auto_compaction is unchanged
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaDisableAutoCompactionUnchanged) {
+ // Verify initial state: disable_auto_compaction = false
+
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ // Create mock tablet meta with disable_auto_compaction = false (same as
current)
+ auto mock_tablet_meta = createMockTabletMeta(false);
+
+ // Mock get_tablet_meta to return tablet_meta with same
disable_auto_compaction = false
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+
+ // Tell the sync point to return with Status::OK()
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+
+ // Call sync_meta
+ Status st = _tablet->sync_meta();
+ EXPECT_TRUE(st.ok());
+
+ // Verify disable_auto_compaction remains false
+
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
+// Test sync_meta is skipped when enable_file_cache is false
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaSkippedWhenFileCacheDisabled) {
+ // Disable file cache
+ config::enable_file_cache = false;
+
+ // Set initial state: disable_auto_compaction = false
+
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ bool callback_called = false;
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta",
+ [&callback_called](auto&& args) { callback_called =
true; });
+
+ // Call sync_meta - should return early without calling get_tablet_meta
+ Status st = _tablet->sync_meta();
+ EXPECT_TRUE(st.ok());
+ EXPECT_FALSE(callback_called);
+
+ // Verify disable_auto_compaction is not changed
+
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
+
+// Test sync_meta syncs compaction_policy together with disable_auto_compaction
+TEST_F(CloudTabletSyncMetaTest, TestSyncMetaMultipleProperties) {
+ // Verify initial states
+
EXPECT_FALSE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+ // Default compaction_policy is "size_based"
+ EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "size_based");
+
+ auto sp = SyncPoint::get_instance();
+ sp->clear_all_call_backs();
+ sp->enable_processing();
+
+ // Create mock tablet meta with updated properties
+ auto mock_tablet_meta = createMockTabletMeta(true, "time_series");
+
+ // Mock get_tablet_meta to return tablet_meta with updated properties
+ sp->set_call_back("CloudMetaMgr::get_tablet_meta",
[mock_tablet_meta](auto&& args) {
+ auto* tablet_meta_ptr = try_any_cast<TabletMetaSharedPtr*>(args[1]);
+ *tablet_meta_ptr = mock_tablet_meta;
+
+ // Tell the sync point to return with Status::OK()
+ try_any_cast_ret<Status>(args)->second = true;
+ });
+
+ // Call sync_meta
+ Status st = _tablet->sync_meta();
+ EXPECT_TRUE(st.ok());
+
+ // Verify both properties are synced
+
EXPECT_TRUE(_tablet->tablet_meta()->tablet_schema()->disable_auto_compaction());
+ EXPECT_EQ(_tablet->tablet_meta()->compaction_policy(), "time_series");
+
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+}
} // namespace doris
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index 6bb3810e840..90ebd4076e1 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -1209,15 +1209,40 @@ void
MetaServiceImpl::update_tablet(::google::protobuf::RpcController* controlle
}
schema_pb.set_disable_auto_compaction(tablet_meta_info.disable_auto_compaction());
+ // Directly overwrite schema KV instead of using put_schema_kv,
+ // because put_schema_kv skips writing when the key already
exists.
auto key = meta_schema_key(
{instance_id, tablet_meta.index_id(),
tablet_meta.schema_version()});
- put_schema_kv(code, msg, txn.get(), key, schema_pb);
- if (code != MetaServiceCode::OK) return;
+ LOG_INFO("overwrite schema kv for disable_auto_compaction
update")
+ .tag("key", hex(key))
+ .tag("disable_auto_compaction",
tablet_meta_info.disable_auto_compaction());
+ // Remove old blob chunks first to avoid orphaned KVs when
+ // the value format or size changes across versions.
+ ValueBuf old_val;
+ auto get_err = cloud::blob_get(txn.get(), key, &old_val);
+ if (get_err == TxnErrorCode::TXN_OK) {
+ bool skip_remove = false;
+
TEST_SYNC_POINT_CALLBACK("update_tablet::skip_schema_remove", &skip_remove);
+ if (!skip_remove) {
+ old_val.remove(txn.get());
+ }
+ }
+ uint8_t ver = config::meta_schema_value_version;
+ if (ver > 0) {
+ cloud::blob_put(txn.get(), key, schema_pb, ver);
+ } else {
+ auto schema_value = schema_pb.SerializeAsString();
+ txn->put(key, schema_value);
+ }
if (is_versioned_write) {
auto key = versioned::meta_schema_key(
{instance_id, tablet_meta.index_id(),
tablet_meta.schema_version()});
- put_versioned_schema_kv(code, msg, txn.get(), key,
schema_pb);
- if (code != MetaServiceCode::OK) return;
+ doris::TabletSchemaCloudPB tablet_schema(schema_pb);
+ if (!document_put(txn.get(), key,
std::move(tablet_schema))) {
+ code = MetaServiceCode::PROTOBUF_SERIALIZE_ERR;
+ msg = "failed to serialize versioned tablet schema";
+ return;
+ }
}
}
}
diff --git a/cloud/test/schema_kv_test.cpp b/cloud/test/schema_kv_test.cpp
index 33816770c0a..54733a29a65 100644
--- a/cloud/test/schema_kv_test.cpp
+++ b/cloud/test/schema_kv_test.cpp
@@ -29,6 +29,7 @@
#include "meta-service/meta_service.h"
#include "meta-service/meta_service_schema.h"
#include "meta-store/blob_message.h"
+#include "meta-store/codec.h"
#include "meta-store/document_message.h"
#include "meta-store/keys.h"
#include "meta-store/txn_kv.h"
@@ -768,8 +769,8 @@ TEST(SchemaKVTest, InsertExistedRowsetTest) {
check_get_tablet(meta_service.get(), 10005, 2);
}
-static void check_schema(MetaServiceProxy* meta_service, int64_t tablet_id,
- int32_t schema_version) {
+static void check_schema(MetaServiceProxy* meta_service, int64_t tablet_id,
int32_t schema_version,
+ bool expected_disable_auto_compaction = false) {
brpc::Controller cntl;
GetTabletRequest req;
GetTabletResponse res;
@@ -780,6 +781,9 @@ static void check_schema(MetaServiceProxy* meta_service,
int64_t tablet_id,
EXPECT_TRUE(res.tablet_meta().has_schema()) << tablet_id;
EXPECT_EQ(res.tablet_meta().schema_version(), schema_version) << tablet_id;
EXPECT_EQ(res.tablet_meta().schema().column_size(), 10) << tablet_id;
+ EXPECT_EQ(res.tablet_meta().schema().disable_auto_compaction(),
+ expected_disable_auto_compaction)
+ << tablet_id;
};
static void update_tablet(MetaServiceProxy* meta_service, int64_t tablet_id) {
@@ -813,7 +817,7 @@ TEST(AlterSchemaKVTest, AlterDisableAutoCompactionTest) {
check_get_tablet(meta_service.get(), 10005, 2);
update_tablet(meta_service.get(), 10005);
- check_schema(meta_service.get(), 10005, 2);
+ check_schema(meta_service.get(), 10005, 2, true);
}
//case 2 config::write_schema_kv = false;
@@ -835,7 +839,7 @@ TEST(AlterSchemaKVTest, AlterDisableAutoCompactionTest) {
check_get_tablet(meta_service.get(), 10005, 2);
update_tablet(meta_service.get(), 10005);
- check_schema(meta_service.get(), 10005, 2);
+ check_schema(meta_service.get(), 10005, 2, true);
}
//case 3 config::write_schema_kv = false, create tablet,
config::write_schema_kv = true;
@@ -857,7 +861,7 @@ TEST(AlterSchemaKVTest, AlterDisableAutoCompactionTest) {
check_get_tablet(meta_service.get(), 10005, 2);
config::write_schema_kv = true;
update_tablet(meta_service.get(), 10005);
- check_schema(meta_service.get(), 10005, 2);
+ check_schema(meta_service.get(), 10005, 2, true);
}
//case 4 config::write_schema_kv = false, create tablet,
config::write_schema_kv = true;
@@ -882,7 +886,158 @@ TEST(AlterSchemaKVTest, AlterDisableAutoCompactionTest) {
check_get_tablet(meta_service.get(), 10005, 2);
config::write_schema_kv = true;
update_tablet(meta_service.get(), 10005);
- check_schema(meta_service.get(), 10005, 2);
+ check_schema(meta_service.get(), 10005, 2, true);
+ }
+}
+
+// Helper: count the number of blob chunks for a given base key.
+static int count_blob_chunks(Transaction* txn, const std::string& base_key) {
+ std::string begin_key = base_key;
+ std::string end_key = base_key;
+ encode_int64(INT64_MAX, &end_key);
+ std::unique_ptr<RangeGetIterator> iter;
+ auto err = txn->get(begin_key, end_key, &iter);
+ EXPECT_EQ(err, TxnErrorCode::TXN_OK);
+ if (err != TxnErrorCode::TXN_OK) return -1;
+ int count = 0;
+ while (iter->has_next()) {
+ iter->next();
+ ++count;
+ }
+ return count;
+}
+
+// Verify that ValueBuf::remove() before blob_put is necessary to clean up
+// orphaned blob chunks when the number of chunks decreases.
+TEST(AlterSchemaKVTest, BlobRemoveBeforeOverwriteTest) {
+ auto meta_service = get_meta_service();
+
+ auto sp = SyncPoint::get_instance();
+ DORIS_CLOUD_DEFER {
+ sp->clear_all_call_backs();
+ sp->disable_processing();
+ };
+ sp->set_call_back("get_instance_id", [&](auto&& args) {
+ auto* ret = try_any_cast_ret<std::string>(args);
+ ret->first = instance_id;
+ ret->second = true;
+ });
+ sp->enable_processing();
+
+ constexpr int64_t db_id = 4000;
+ constexpr int64_t table_id = 40001;
+ constexpr int64_t index_id = 40002;
+ constexpr int64_t partition_id = 40003;
+ constexpr int64_t tablet_id = 40004;
+ constexpr int32_t schema_version = 0;
+
+ config::write_schema_kv = true;
+ config::meta_schema_value_version = 1;
+
+ // Create a tablet with a normal 10-column schema.
+ ASSERT_NO_FATAL_FAILURE(create_tablet(meta_service.get(), db_id, table_id,
index_id,
+ partition_id, tablet_id,
next_rowset_id(),
+ schema_version));
+ check_get_tablet(meta_service.get(), tablet_id, schema_version);
+
+ // Compute the schema KV key.
+ auto schema_key = meta_schema_key({instance_id, index_id, schema_version});
+
+ // Overwrite the schema KV with a small split_size to create 3 blob chunks.
+ // The normal schema (~120 bytes) with split_size=50 yields 3 chunks.
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ // First remove existing blob chunks
+ ValueBuf old_val;
+ ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &old_val),
TxnErrorCode::TXN_OK);
+ old_val.remove(txn.get());
+ // Write with small split_size to create 3 chunks
+ doris::TabletSchemaCloudPB schema;
+ ASSERT_TRUE(old_val.to_pb(&schema));
+ std::string serialized = schema.SerializeAsString();
+ size_t split_size = (serialized.size() / 3) + 1; // ensures exactly 3
chunks
+ cloud::blob_put(txn.get(), schema_key, schema,
config::meta_schema_value_version,
+ split_size);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+
+ // Verify we now have 3 blob chunks.
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 3);
+ }
+
+ // Case 1: update_tablet WITH remove (default behavior).
+ // The remove cleans up old 3 chunks, then blob_put writes 1 chunk
+ // (with default split_size=90KB, the ~120 byte schema fits in 1 chunk).
+ update_tablet(meta_service.get(), tablet_id);
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 1)
+ << "With remove: should have 1 chunk (old 3 cleaned up)";
+ // Verify schema is correct
+ ValueBuf val;
+ ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &val),
TxnErrorCode::TXN_OK);
+ doris::TabletSchemaCloudPB schema;
+ ASSERT_TRUE(val.to_pb(&schema));
+ EXPECT_TRUE(schema.disable_auto_compaction());
+ }
+ check_schema(meta_service.get(), tablet_id, schema_version, true);
+
+ // Reset: overwrite schema KV again with 3 blob chunks and
+ // disable_auto_compaction=false so update_tablet will modify it again.
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ ValueBuf old_val;
+ ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &old_val),
TxnErrorCode::TXN_OK);
+ old_val.remove(txn.get());
+ doris::TabletSchemaCloudPB schema;
+ ASSERT_TRUE(old_val.to_pb(&schema));
+ schema.set_disable_auto_compaction(false);
+ std::string serialized = schema.SerializeAsString();
+ size_t split_size = (serialized.size() / 3) + 1;
+ cloud::blob_put(txn.get(), schema_key, schema,
config::meta_schema_value_version,
+ split_size);
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+ }
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ EXPECT_EQ(count_blob_chunks(txn.get(), schema_key), 3);
+ }
+
+ // Case 2: update_tablet WITHOUT remove (inject skip via sync point).
+ // This simulates what would happen if we didn't have the remove logic.
+ sp->set_call_back("update_tablet::skip_schema_remove",
+ [](auto&& args) { *try_any_cast<bool*>(args[0]) = true;
});
+ update_tablet(meta_service.get(), tablet_id);
+ sp->clear_call_back("update_tablet::skip_schema_remove");
+
+ {
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(meta_service->txn_kv()->create_txn(&txn),
TxnErrorCode::TXN_OK);
+ int chunks = count_blob_chunks(txn.get(), schema_key);
+ EXPECT_EQ(chunks, 3) << "Without remove: should have 3 chunks (2
orphaned + 1 overwritten)";
+ // blob_get reads all 3 chunks and concatenates them, producing
+ // corrupted data that cannot be parsed correctly.
+ ValueBuf val;
+ ASSERT_EQ(cloud::blob_get(txn.get(), schema_key, &val),
TxnErrorCode::TXN_OK);
+ doris::TabletSchemaCloudPB schema;
+ // The deserialized schema may parse but contain garbage from
+ // orphaned chunks, or the column count will be wrong.
+ if (val.to_pb(&schema)) {
+ // If it parses, the data is corrupted - the column count or
+ // content won't match the expected schema.
+ LOG(INFO) << "Without remove: parsed schema has " <<
schema.column_size()
+ << " columns (expected 10), disable_auto_compaction="
+ << schema.disable_auto_compaction();
+ } else {
+ LOG(INFO) << "Without remove: schema parse failed as expected
(corrupted data)";
+ }
}
}
diff --git
a/regression-test/suites/cloud_p0/compaction/test_cloud_alter_disable_auto_compaction.groovy
b/regression-test/suites/cloud_p0/compaction/test_cloud_alter_disable_auto_compaction.groovy
new file mode 100644
index 00000000000..799cc05810b
--- /dev/null
+++
b/regression-test/suites/cloud_p0/compaction/test_cloud_alter_disable_auto_compaction.groovy
@@ -0,0 +1,193 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite("test_cloud_alter_disable_auto_compaction", "p0,docker") {
+ if (!isCloudMode()) {
+ logger.info("not cloud mode, skip this test")
+ return
+ }
+
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ ]
+ options.beConfigs += [
+ // Short sync interval for faster property propagation
+ 'tablet_sync_interval_s=5',
+ // Trigger cumulative compaction after 5 rowsets
+ 'cumulative_compaction_min_deltas=5',
+ // Enable file cache (required for sync_meta to work)
+ 'enable_file_cache=true',
+ ]
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = true
+
+ docker(options) {
+ def tableName = "test_cloud_alter_disable_auto_compaction"
+
+ // Get BE address for metrics
+ def backendId_to_backendIP = [:]
+ def backendId_to_backendHttpPort = [:]
+ getBackendIpHttpPort(backendId_to_backendIP,
backendId_to_backendHttpPort)
+ def beIp = backendId_to_backendIP.values()[0]
+ def beHttpPort = backendId_to_backendHttpPort.values()[0]
+ logger.info("BE address: ${beIp}:${beHttpPort}")
+
+ sql """ DROP TABLE IF EXISTS ${tableName}; """
+ sql """
+ CREATE TABLE ${tableName} (
+ `id` int(11) NULL,
+ `name` varchar(255) NULL,
+ `score` int(11) NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ );
+ """
+
+ // Verify initial state: disable_auto_compaction should be false
(default)
+ def result = sql """ SHOW CREATE TABLE ${tableName}; """
+ logger.info("Initial SHOW CREATE TABLE result: ${result}")
+ assertFalse(result[0][1].contains('"disable_auto_compaction" =
"true"'))
+
+ // ============================================================
+ // Test 1: Verify compaction works when enabled (default)
+ // ============================================================
+ logger.info("Test 1: Verify compaction works when enabled")
+
+ // Get initial compaction metric
+ def getCompactionMetric = {
+ return get_be_metric(beIp, beHttpPort, 'compaction_deltas_total',
"cumulative")
+ }
+ def initialCompactionCount = getCompactionMetric()
+ logger.info("Initial compaction count: ${initialCompactionCount}")
+
+ // Insert enough rowsets to trigger compaction (need > 5 for
cumulative)
+ for (int i = 0; i < 8; i++) {
+ sql """ INSERT INTO ${tableName} VALUES (${i}, "test${i}", ${i *
100}); """
+ }
+
+ // Wait for compaction to trigger (may take a few seconds)
+ def compactionTriggered = false
+ for (int retry = 0; retry < 30; retry++) {
+ sleep(2000)
+ def currentCount = getCompactionMetric()
+ logger.info("Retry ${retry}: current compaction count =
${currentCount}")
+ if (currentCount > initialCompactionCount) {
+ compactionTriggered = true
+ break
+ }
+ }
+ assertTrue(compactionTriggered, "Expected compaction to be triggered
when disable_auto_compaction is false")
+
+ // Verify data is correct
+ def count = sql """ SELECT count(*) FROM ${tableName}; """
+ assertEquals(8, count[0][0])
+
+ // ============================================================
+ // Test 2: ALTER TABLE to disable compaction
+ // ============================================================
+ logger.info("Test 2: Verify compaction is disabled after ALTER TABLE")
+
+ sql """ ALTER TABLE ${tableName} SET ("disable_auto_compaction" =
"true"); """
+
+ // Verify the property is updated in FE
+ result = sql """ SHOW CREATE TABLE ${tableName}; """
+ logger.info("After setting true, SHOW CREATE TABLE result: ${result}")
+ assertTrue(result[0][1].contains('"disable_auto_compaction" = "true"'))
+
+ // Wait for sync_meta to propagate the change to BE
+ // tablet_sync_interval_s is set to 5 seconds, so wait a bit longer
+ logger.info("Waiting for sync_meta to propagate
disable_auto_compaction=true to BE...")
+ sleep(10000)
+
+ // Record compaction count after disabling
+ def countAfterDisable = getCompactionMetric()
+ logger.info("Compaction count after disabling: ${countAfterDisable}")
+
+ // Insert more rowsets - these should NOT trigger compaction
+ for (int i = 8; i < 16; i++) {
+ sql """ INSERT INTO ${tableName} VALUES (${i}, "test${i}", ${i *
100}); """
+ }
+
+ // Wait and verify compaction does NOT happen
+ sleep(15000) // Wait longer than typical compaction interval
+ def countAfterInsert = getCompactionMetric()
+ logger.info("Compaction count after inserting with disabled
compaction: ${countAfterInsert}")
+
+ // The compaction count should not have increased significantly
+ // Allow for small variations due to other internal operations
+ assertTrue(countAfterInsert <= countAfterDisable + 1,
+ "Compaction should not be triggered when disable_auto_compaction
is true. " +
+ "Before: ${countAfterDisable}, After: ${countAfterInsert}")
+
+ // Verify data is still correct
+ count = sql """ SELECT count(*) FROM ${tableName}; """
+ assertEquals(16, count[0][0])
+
+ // ============================================================
+ // Test 3: Re-enable compaction and verify it works again
+ // ============================================================
+ logger.info("Test 3: Verify compaction resumes after re-enabling")
+
+ sql """ ALTER TABLE ${tableName} SET ("disable_auto_compaction" =
"false"); """
+
+ // Verify the property is updated
+ result = sql """ SHOW CREATE TABLE ${tableName}; """
+ logger.info("After setting false, SHOW CREATE TABLE result: ${result}")
+ assertFalse(result[0][1].contains('"disable_auto_compaction" =
"true"'))
+
+ // Wait for sync_meta to propagate the change
+ logger.info("Waiting for sync_meta to propagate
disable_auto_compaction=false to BE...")
+ sleep(10000)
+
+ def countBeforeReEnable = getCompactionMetric()
+ logger.info("Compaction count before re-enabling:
${countBeforeReEnable}")
+
+ // Insert more rowsets - these SHOULD trigger compaction
+ for (int i = 16; i < 24; i++) {
+ sql """ INSERT INTO ${tableName} VALUES (${i}, "test${i}", ${i *
100}); """
+ }
+
+ // Wait for compaction to trigger
+ compactionTriggered = false
+ for (int retry = 0; retry < 30; retry++) {
+ sleep(2000)
+ def currentCount = getCompactionMetric()
+ logger.info("Retry ${retry}: current compaction count =
${currentCount}")
+ if (currentCount > countBeforeReEnable) {
+ compactionTriggered = true
+ break
+ }
+ }
+ assertTrue(compactionTriggered, "Expected compaction to resume after
re-enabling")
+
+ // Verify final data
+ count = sql """ SELECT count(*) FROM ${tableName}; """
+ assertEquals(24, count[0][0])
+
+ // Clean up
+ sql """ DROP TABLE IF EXISTS ${tableName}; """
+
+ logger.info("All tests passed!")
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
