[fix](compaction) fix the issue of schema loss caused by cloud disable auto compaction

cloud/src/meta-service/meta_service.cpp

@@ -685,10 +685,14 @@ void internal_get_tablet(MetaServiceCode& code, std::string& msg, const std::str
         return;
     }
 
-    if (tablet_meta->has_schema()) { // tablet meta saved before detach schema kv
+    if (tablet_meta->has_schema() &&
+        tablet_meta->schema().column_size() > 0) { // tablet meta saved before detach schema kv
         tablet_meta->set_schema_version(tablet_meta->schema().schema_version());
     }
-    if (!tablet_meta->has_schema() && !skip_schema) {
+
+    if ((!tablet_meta->has_schema() ||
+         (tablet_meta->has_schema() && tablet_meta->schema().column_size() <= 0)) &&
+        !skip_schema) {
         if (!tablet_meta->has_schema_version()) {
             code = MetaServiceCode::INVALID_ARGUMENT;
             msg = "tablet_meta must have either schema or schema_version";
@@ -768,8 +772,32 @@ void MetaServiceImpl::update_tablet(::google::protobuf::RpcController* controlle
             tablet_meta.set_time_series_compaction_level_threshold(
                     tablet_meta_info.time_series_compaction_level_threshold());
         } else if (tablet_meta_info.has_disable_auto_compaction()) {
-            tablet_meta.mutable_schema()->set_disable_auto_compaction(
-                    tablet_meta_info.disable_auto_compaction());
+            if (tablet_meta.has_schema() && tablet_meta.schema().column_size() > 0) {
+                tablet_meta.mutable_schema()->set_disable_auto_compaction(
+                        tablet_meta_info.disable_auto_compaction());
+            } else {
+                auto key = meta_schema_key(
+                        {instance_id, tablet_meta.index_id(), tablet_meta.schema_version()});
+                ValueBuf val_buf;
+                err = cloud::get(txn.get(), key, &val_buf);
+                if (err != TxnErrorCode::TXN_OK) {
+                    code = cast_as<ErrCategory::READ>(err);
+                    msg = fmt::format("failed to get schema, err={}",
+                                      err == TxnErrorCode::TXN_KEY_NOT_FOUND ? "not found"
+                                                                             : "internal error");
+                    return;
+                }
+                doris::TabletSchemaCloudPB schema_pb;
+                if (!parse_schema_value(val_buf, &schema_pb)) {
+                    code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+                    msg = fmt::format("malformed schema value, key={}", key);
+                    return;
+                }
+
+                schema_pb.set_disable_auto_compaction(tablet_meta_info.disable_auto_compaction());
+                put_schema_kv(code, msg, txn.get(), key, schema_pb);
+                if (code != MetaServiceCode::OK) return;
+            }
         }
         int64_t table_id = tablet_meta.table_id();
         int64_t index_id = tablet_meta.index_id();

cloud/test/schema_kv_test.cpp

@@ -40,6 +40,15 @@ static std::string next_rowset_id() {
     return std::to_string(++cnt);
 }
 
+static void fill_schema(doris::TabletSchemaCloudPB* schema, int32_t schema_version) {
+    schema->set_schema_version(schema_version);
+    for (int i = 0; i < 10; ++i) {
+        auto column = schema->add_column();
+        column->set_unique_id(20000 + i);
+        column->set_type("INT");
+    }
+}
+
 static void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t index_id,
                        int64_t partition_id, int64_t tablet_id, const std::string& rowset_id,
                        int32_t schema_version) {
@@ -49,7 +58,7 @@ static void add_tablet(CreateTabletsRequest& req, int64_t table_id, int64_t inde
     tablet->set_partition_id(partition_id);
     tablet->set_tablet_id(tablet_id);
     auto schema = tablet->mutable_schema();
-    schema->set_schema_version(schema_version);
+    fill_schema(schema, schema_version);
     auto first_rowset = tablet->add_rs_metas();
     first_rowset->set_rowset_id(0); // required
     first_rowset->set_rowset_id_v2(rowset_id);
@@ -148,6 +157,9 @@ TEST(DetachSchemaKVTest, TabletTest) {
         saved_tablet.set_partition_id(partition_id);
         saved_tablet.set_tablet_id(tablet_id);
         saved_tablet.mutable_schema()->set_schema_version(1);
+        auto column = saved_tablet.mutable_schema()->add_column();
+        column->set_unique_id(30001);
+        column->set_type("INT");
         std::string tablet_key, tablet_val;
         meta_tablet_key({instance_id, table_id, index_id, partition_id, tablet_id}, &tablet_key);
         ASSERT_TRUE(saved_tablet.SerializeToString(&tablet_val));
@@ -672,4 +684,121 @@ 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) {
+    brpc::Controller cntl;
+    GetTabletRequest req;
+    GetTabletResponse res;
+    req.set_tablet_id(tablet_id);
+    meta_service->get_tablet(&cntl, &req, &res, nullptr);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
+    ASSERT_TRUE(res.has_tablet_meta()) << 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;
+};
+
+static void update_tablet(MetaServiceProxy* meta_service, int64_t tablet_id) {
+    brpc::Controller cntl;
+    UpdateTabletRequest req;
+    UpdateTabletResponse res;
+
+    auto meta_info = req.add_tablet_meta_infos();
+    meta_info->set_disable_auto_compaction(true);
+    meta_info->set_tablet_id(tablet_id);
+
+    meta_service->update_tablet(&cntl, &req, &res, nullptr);
+    ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << tablet_id;
+}
+
+TEST(AlterSchemaKVTest, AlterDisableAutoCompactionTest) {
+    //case 1 config::write_schema_kv = true;
+    {
+        auto meta_service = get_meta_service();
+        config::write_schema_kv = true;
+        //config::meta_schema_value_version = 0;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10004, next_rowset_id(), 0));
+        check_get_tablet(meta_service.get(), 10004, 0);
+        check_schema(meta_service.get(), 10004, 0);
+
+        //config::meta_schema_value_version = 1;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10005, next_rowset_id(), 2));
+        check_get_tablet(meta_service.get(), 10005, 2);
+
+        update_tablet(meta_service.get(), 10005);
+        check_schema(meta_service.get(), 10005, 2);
+    }
+
+    //case 2 config::write_schema_kv = false;
+    {
+        auto meta_service = get_meta_service();
+        config::write_schema_kv = false;
+        auto defer1 =
+                std::make_unique<std::function<void()>>([]() { config::write_schema_kv = true; });
+
+        //config::meta_schema_value_version = 0;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10004, next_rowset_id(), 0));
+        check_get_tablet(meta_service.get(), 10004, 0);
+        check_schema(meta_service.get(), 10004, 0);
+
+        //config::meta_schema_value_version = 1;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10005, next_rowset_id(), 2));
+        check_get_tablet(meta_service.get(), 10005, 2);
+
+        update_tablet(meta_service.get(), 10005);
+        check_schema(meta_service.get(), 10005, 2);
+    }
+
+    //case 3 config::write_schema_kv = false, create tablet, config::write_schema_kv = true;
+    {
+        auto meta_service = get_meta_service();
+        config::write_schema_kv = false;
+        auto defer1 =
+                std::make_unique<std::function<void()>>([]() { config::write_schema_kv = true; });
+
+        //config::meta_schema_value_version = 0;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10004, next_rowset_id(), 0));
+        check_get_tablet(meta_service.get(), 10004, 0);
+        check_schema(meta_service.get(), 10004, 0);
+
+        //config::meta_schema_value_version = 1;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10005, next_rowset_id(), 2));
+        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);
+    }
+
+    //case 4 config::write_schema_kv = false, create tablet, config::write_schema_kv = true;
+    //       meta_schema_value_version = 0, meta_schema_value_version = 1
+    {
+        auto meta_service = get_meta_service();
+        config::write_schema_kv = false;
+        auto defer1 = std::make_unique<std::function<void()>>([]() {
+            config::write_schema_kv = true;
+            config::meta_schema_value_version = 1;
+        });
+
+        config::meta_schema_value_version = 0;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10004, next_rowset_id(), 0));
+        check_get_tablet(meta_service.get(), 10004, 0);
+        check_schema(meta_service.get(), 10004, 0);
+
+        config::meta_schema_value_version = 1;
+        ASSERT_NO_FATAL_FAILURE(
+                create_tablet(meta_service.get(), 10001, 10002, 10003, 10005, next_rowset_id(), 2));
+        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);
+    }
+}
+
 } // namespace doris::cloud

regression-test/suites/schema_change_p0/alter_disable_auto_compaction.groovy

@@ -0,0 +1,63 @@
+// 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.
+
+suite("test_alter_property_mow") {
+    def tableName = "test_alter_property_mow"
+
+    sql """ DROP TABLE IF EXISTS ${tableName} """
+
+    sql """
+        CREATE TABLE IF NOT EXISTS ${tableName}
+        (
+            k1 INT,
+            k2 int,
+            v1 bigint,
+            v2 INT,
+            v3 varchar(32) null,
+        )
+        UNIQUE KEY (k1, k2)
+        DISTRIBUTED BY HASH(k2) BUCKETS 1
+        PROPERTIES (
+        'replication_num' = '1'
+        )
+    """
+
+    sql """
+        ALTER TABLE ${tableName} SET ('disable_auto_compaction'='true');
+    """
+
+    sql """
+        ALTER TABLE ${tableName} SET ('disable_auto_compaction'='false');
+    """
+
+    sql """
+        insert into ${tableName} (k1, k2, v1, v2, v3) values(2, 2, 3, 4, 'a')
+    """
+
+    sql """
+        insert into ${tableName} (k1, k2, v1, v2, v3) values(1, 2, 3, 4, 'a')
+    """
+
+    sql """
+        insert into ${tableName} (k1, k2, v1, v2, v3) values(1, 2, 3, 4, 'a')
+    """
+
+    sql """
+        select * from ${tableName}
+    """
+}
+