branch-3.0: [Fix](Variant) fix variant may lost schema info when meet TXN_CONFLICT in cloud mode #47284

cloud/src/meta-service/meta_service.cpp

@@ -939,6 +939,30 @@ static void set_schema_in_existed_rowset(MetaServiceCode& code, std::string& msg
     }
 }
 
+/**
+ * Fills schema information into the rowset meta from the dictionary.
+ * Handles schemas with variant types by retrieving the complete schema from the dictionary.
+ *
+ * @param code Result code indicating the operation status.
+ * @param msg Error description for failed operations.
+ * @param instance_id Identifier for the specific instance.
+ * @param txn Pointer to the transaction object for transactional operations.
+ * @param existed_rowset_meta Rowset meta object to be updated with schema information.
+ */
+static void fill_schema_from_dict(MetaServiceCode& code, std::string& msg,
+                                  const std::string& instance_id, Transaction* txn,
+                                  doris::RowsetMetaCloudPB* existed_rowset_meta) {
+    google::protobuf::RepeatedPtrField<doris::RowsetMetaCloudPB> metas;
+    metas.Add()->CopyFrom(*existed_rowset_meta);
+    // Retrieve schema from the dictionary and update metas.
+    read_schema_dict(code, msg, instance_id, existed_rowset_meta->index_id(), txn, &metas, nullptr);
+    if (code != MetaServiceCode::OK) {
+        return;
+    }
+    // Update the original rowset meta with the complete schema from metas.
+    existed_rowset_meta->CopyFrom(metas.Get(0));
+}
+
 /**
  * 1. Check and confirm tmp rowset kv does not exist
  * 2. Construct recycle rowset kv which contains object path
@@ -1010,6 +1034,10 @@ void MetaServiceImpl::prepare_rowset(::google::protobuf::RpcController* controll
             existed_rowset_meta->set_schema_version(
                     existed_rowset_meta->tablet_schema().schema_version());
         }
+        if (existed_rowset_meta->has_variant_type_in_schema()) {
+            fill_schema_from_dict(code, msg, instance_id, txn.get(), existed_rowset_meta);
+            if (code != MetaServiceCode::OK) return;
+        }
         code = MetaServiceCode::ALREADY_EXISTED;
         msg = "rowset already exists";
         return;
@@ -1134,6 +1162,10 @@ void MetaServiceImpl::commit_rowset(::google::protobuf::RpcController* controlle
             existed_rowset_meta->set_schema_version(
                     existed_rowset_meta->tablet_schema().schema_version());
         }
+        if (existed_rowset_meta->has_variant_type_in_schema()) {
+            fill_schema_from_dict(code, msg, instance_id, txn.get(), existed_rowset_meta);
+            if (code != MetaServiceCode::OK) return;
+        }
         code = MetaServiceCode::ALREADY_EXISTED;
         msg = "rowset already exists";
         return;
@@ -1626,7 +1658,8 @@ void MetaServiceImpl::get_rowset(::google::protobuf::RpcController* controller,
         }
 
         if (need_read_schema_dict) {
-            read_schema_dict(code, msg, instance_id, idx.index_id(), txn.get(), response,
+            read_schema_dict(code, msg, instance_id, idx.index_id(), txn.get(),
+                             response->mutable_rowset_meta(), response->mutable_schema_dict(),
                              request->schema_op());
             if (code != MetaServiceCode::OK) return;
         }

cloud/src/meta-service/meta_service_schema.cpp

@@ -312,13 +312,15 @@ void write_schema_dict(MetaServiceCode& code, std::string& msg, const std::strin
         LOG(INFO) << "Dictionary saved, key=" << hex(dict_key)
                   << " txn_id=" << rowset_meta->txn_id() << " Dict size=" << dict.column_dict_size()
                   << ", Current column ID=" << dict.current_column_dict_id()
-                  << ", Current index ID=" << dict.current_index_dict_id();
+                  << ", Current index ID=" << dict.current_index_dict_id()
+                  << ", Dict bytes=" << dict_val.size();
     }
 }
 
 void read_schema_dict(MetaServiceCode& code, std::string& msg, const std::string& instance_id,
-                      int64_t index_id, Transaction* txn, GetRowsetResponse* response,
-                      GetRowsetRequest::SchemaOp schema_op) {
+                      int64_t index_id, Transaction* txn,
+                      google::protobuf::RepeatedPtrField<doris::RowsetMetaCloudPB>* rsp_metas,
+                      SchemaCloudDictionary* rsp_dict, GetRowsetRequest::SchemaOp schema_op) {
     std::stringstream ss;
 
     // read dict if any rowset has dict key list
@@ -341,8 +343,8 @@ void read_schema_dict(MetaServiceCode& code, std::string& msg, const std::string
               << ", index size=" << dict.index_dict_size();
 
     // Return dict, let backend to fill schema with dict info
-    if (schema_op == GetRowsetRequest::RETURN_DICT) {
-        response->mutable_schema_dict()->Swap(&dict);
+    if (schema_op == GetRowsetRequest::RETURN_DICT && rsp_dict != nullptr) {
+        rsp_dict->Swap(&dict);
         return;
     }
 
@@ -381,7 +383,7 @@ void read_schema_dict(MetaServiceCode& code, std::string& msg, const std::string
     };
 
     // fill rowsets's schema with dict info
-    for (auto& rowset_meta : *response->mutable_rowset_meta()) {
+    for (auto& rowset_meta : *rsp_metas) {
         if (rowset_meta.has_schema_dict_key_list()) {
             fill_schema_with_dict(&rowset_meta);
         }

cloud/src/meta-service/meta_service_schema.h

@@ -36,7 +36,9 @@ void write_schema_dict(MetaServiceCode& code, std::string& msg, const std::strin
 
 // Read schema from dictionary metadata, modified to rowset_metas
 void read_schema_dict(MetaServiceCode& code, std::string& msg, const std::string& instance_id,
-                      int64_t index_id, Transaction* txn, GetRowsetResponse* response,
-                      GetRowsetRequest::SchemaOp schema_op);
+                      int64_t index_id, Transaction* txn,
+                      google::protobuf::RepeatedPtrField<doris::RowsetMetaCloudPB>* rsp_metas,
+                      SchemaCloudDictionary* rsp_dict,
+                      GetRowsetRequest::SchemaOp schema_op = GetRowsetRequest::FILL_WITH_DICT);
 
 } // namespace doris::cloud

regression-test/data/variant_p0/schema_change/schema_change_with_txn_conflict.out

@@ -0,0 +1,5 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+0	{"k1":1,"k2":"hello world","k3":[1234],"k4":1.1,"k5":[[123]]}
+2	{"a":12345}
+

regression-test/suites/variant_p0/schema_change/schema_change_with_txn_conflict.groovy

@@ -0,0 +1,49 @@
+// 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_schema_change_txn_conflict", "nonConcurrent") {
+    def customFeConfig = [
+            schema_change_max_retry_time: 10
+    ]
+    def tableName = "variant_txn_conflict"
+    setFeConfigTemporary(customFeConfig) {
+        try {
+            GetDebugPoint().enableDebugPointForAllBEs("CloudSchemaChangeJob::_convert_historical_rowsets.test_conflict")
+            sql "DROP TABLE IF EXISTS ${tableName}"
+            sql """
+                CREATE TABLE IF NOT EXISTS ${tableName} (
+                    k bigint,
+                    v variant
+                )
+                DUPLICATE KEY(`k`)
+                DISTRIBUTED BY HASH(k) BUCKETS 4
+                properties("replication_num" = "1");
+            """
+            sql """INSERT INTO ${tableName} SELECT *, '{"k1":1, "k2": "hello world", "k3" : [1234], "k4" : 1.10000, "k5" : [[123]]}' FROM numbers("number" = "1")"""
+            sql """ALTER TABLE ${tableName} SET("bloom_filter_columns" = "v")"""
+
+            waitForSchemaChangeDone {
+                sql """SHOW ALTER TABLE COLUMN WHERE IndexName='${tableName}' ORDER BY createtime DESC LIMIT 1"""
+                time 600
+            }
+            sql """insert into ${tableName} values (2, '{"a" : 12345}')"""
+        } catch (Exception e) {
+            GetDebugPoint().disableDebugPointForAllBEs("CloudSchemaChangeJob::_convert_historical_rowsets.test_conflict")
+        }
+    }
+    qt_sql "select * from ${tableName} order by k limit 5"
+}