[Feature](Variant) support variant sparse feature and schema template with multi indexes

be/src/agent/be_exec_version_manager.cpp

@@ -113,7 +113,7 @@ void BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
 *     b. support const column in serialize/deserialize function: PR #41175
  */
 
-const int BeExecVersionManager::max_be_exec_version = 8;
+const int BeExecVersionManager::max_be_exec_version = 9;
 const int BeExecVersionManager::min_be_exec_version = 0;
 std::map<std::string, std::set<int>> BeExecVersionManager::_function_change_map {};
 std::set<std::string> BeExecVersionManager::_function_restrict_map;

be/src/agent/be_exec_version_manager.h

@@ -32,7 +32,8 @@ constexpr inline int AGGREGATION_2_1_VERSION =
         6; // some aggregation changed the data format after this version
 constexpr inline int USE_CONST_SERDE =
         8; // support const column in serialize/deserialize function: PR #41175
-
+constexpr inline int VARIANT_SPARSE_SERDE =
+        9; // support variant for serializing/deserializing sparse column
 class BeExecVersionManager {
 public:
     BeExecVersionManager() = delete;

be/src/cloud/cloud_rowset_writer.cpp

@@ -60,7 +60,10 @@ Status CloudRowsetWriter::init(const RowsetWriterContext& rowset_writer_context)
         DCHECK_NE(_context.newest_write_timestamp, -1);
         _rowset_meta->set_newest_write_timestamp(_context.newest_write_timestamp);
     }
-    _rowset_meta->set_tablet_schema(_context.tablet_schema);
+    auto schema = _context.tablet_schema->need_record_variant_extended_schema()
+                          ? _context.tablet_schema
+                          : _context.tablet_schema->copy_without_variant_extracted_columns();
+    _rowset_meta->set_tablet_schema(schema);
     _context.segment_collector = std::make_shared<SegmentCollectorT<BaseBetaRowsetWriter>>(this);
     _context.file_writer_creator = std::make_shared<FileWriterCreatorT<BaseBetaRowsetWriter>>(this);
     return Status::OK();
@@ -103,7 +106,10 @@ Status CloudRowsetWriter::build(RowsetSharedPtr& rowset) {
     // update rowset meta tablet schema if tablet schema updated
     auto rowset_schema = _context.merged_tablet_schema != nullptr ? _context.merged_tablet_schema
                                                                   : _context.tablet_schema;
-    _rowset_meta->set_tablet_schema(rowset_schema);
+    auto schema = rowset_schema->need_record_variant_extended_schema()
+                          ? rowset_schema
+                          : rowset_schema->copy_without_variant_extracted_columns();
+    _rowset_meta->set_tablet_schema(schema);
 
     if (_rowset_meta->newest_write_timestamp() == -1) {
         _rowset_meta->set_newest_write_timestamp(UnixSeconds());

be/src/common/config.cpp

@@ -1344,6 +1344,8 @@ DEFINE_Bool(enable_snapshot_action, "false");
 
 DEFINE_mInt32(variant_max_merged_tablet_schema_size, "2048");
 
+DEFINE_mInt32(variant_max_sparse_column_statistics_size, "10000");
+
 DEFINE_mBool(enable_column_type_check, "true");
 // 128 MB
 DEFINE_mInt64(local_exchange_buffer_mem_limit, "134217728");

be/src/common/config.h

@@ -1299,8 +1299,6 @@ DECLARE_mInt64(LZ4_HC_compression_level);
 DECLARE_mDouble(variant_ratio_of_defaults_as_sparse_column);
 DECLARE_mBool(variant_use_cloud_schema_dict_cache);
 // Threshold to estimate a column is sparsed
-// Notice: TEST ONLY
-DECLARE_mInt64(variant_threshold_rows_to_estimate_sparse_column);
 // Treat invalid json format str as string, instead of throwing exception if false
 DECLARE_mBool(variant_throw_exeception_on_invalid_json);
 
@@ -1409,6 +1407,9 @@ DECLARE_Bool(enable_snapshot_action);
 // The max columns size for a tablet schema
 DECLARE_mInt32(variant_max_merged_tablet_schema_size);
 
+// The max sparse column statistics size for a variant column
+DECLARE_mInt32(variant_max_sparse_column_statistics_size);
+
 DECLARE_mInt64(local_exchange_buffer_mem_limit);
 
 DECLARE_mInt64(enable_debug_log_timeout_secs);

be/src/index-tools/index_tool.cpp

@@ -81,6 +81,7 @@ DEFINE_string(dest_idx_dirs_file, "", "destination segment index files");
 DEFINE_string(dest_seg_num_rows_file, "", "destination segment number of rows");
 DEFINE_string(tablet_path, "", "tablet path");
 DEFINE_string(trans_vec_file, "", "rowid conversion map file");
+DEFINE_string(idx_path, "", "inverted index path");
 
 std::string get_usage(const std::string& progname) {
     std::stringstream ss;
@@ -674,6 +675,9 @@ int main(int argc, char** argv) {
             std::vector<std::string> files;
             int64_t index_id = FLAGS_idx_id;
             std::string index_suffix = "";
+            if (FLAGS_idx_path != "") {
+                index_suffix = FLAGS_idx_path;
+            }
             doris::TabletIndexPB index_pb;
             index_pb.set_index_id(index_id);
             index_pb.set_index_suffix_name(index_suffix);

be/src/olap/base_tablet.cpp

@@ -37,6 +37,7 @@
 #include "olap/delete_bitmap_calculator.h"
 #include "olap/iterators.h"
 #include "olap/memtable.h"
+#include "olap/olap_common.h"
 #include "olap/partial_update_info.h"
 #include "olap/primary_key_index.h"
 #include "olap/rowid_conversion.h"
@@ -171,6 +172,31 @@ TabletSchemaSPtr BaseTablet::tablet_schema_with_merged_max_schema_version(
     return target_schema;
 }
 
+Status BaseTablet::get_extended_compaction_schema(
+        const std::vector<RowsetMetaSharedPtr>& rowset_metas, TabletSchemaSPtr& target_schema) {
+    RowsetMetaSharedPtr max_schema_version_rs = *std::max_element(
+            rowset_metas.begin(), rowset_metas.end(),
+            [](const RowsetMetaSharedPtr& a, const RowsetMetaSharedPtr& b) {
+                return !a->tablet_schema()
+                               ? true
+                               : (!b->tablet_schema()
+                                          ? false
+                                          : a->tablet_schema()->schema_version() <
+                                                    b->tablet_schema()->schema_version());
+            });
+    target_schema = max_schema_version_rs->tablet_schema();
+    if (target_schema->num_variant_columns() > 0) {
+        RowsetIdUnorderedSet rowset_ids;
+        for (const RowsetMetaSharedPtr& rs_meta : rowset_metas) {
+            rowset_ids.emplace(rs_meta->rowset_id());
+        }
+        // extended schema for variant columns
+        RETURN_IF_ERROR(vectorized::schema_util::get_extended_compaction_schema(
+                get_rowset_by_ids(&rowset_ids), target_schema));
+    }
+    return Status::OK();
+}
+
 Status BaseTablet::set_tablet_state(TabletState state) {
     if (_tablet_meta->tablet_state() == TABLET_SHUTDOWN && state != TABLET_SHUTDOWN) {
         return Status::Error<META_INVALID_ARGUMENT>(
@@ -2162,4 +2188,15 @@ int32_t BaseTablet::max_version_config() {
     return max_version;
 }
 
+TabletSchemaSPtr BaseTablet::calculate_variant_extended_schema() const {
+    std::vector<RowsetSharedPtr> rowsets;
+    {
+        std::shared_lock rdlock(_meta_lock);
+        for (const auto& it : _rs_version_map) {
+            rowsets.emplace_back(it.second);
+        }
+    }
+    return vectorized::schema_util::calculate_variant_extended_schema(rowsets, _max_version_schema);
+}
+
 } // namespace doris

be/src/olap/base_tablet.h

@@ -148,6 +148,10 @@ class BaseTablet {
     static TabletSchemaSPtr tablet_schema_with_merged_max_schema_version(
             const std::vector<RowsetMetaSharedPtr>& rowset_metas);
 
+    // Get the extended compaction schema from the rowset metas
+    Status get_extended_compaction_schema(const std::vector<RowsetMetaSharedPtr>& rowset_metas,
+                                          TabletSchemaSPtr& target_schema);
+
     ////////////////////////////////////////////////////////////////////////////
     // begin MoW functions
     ////////////////////////////////////////////////////////////////////////////
@@ -301,6 +305,8 @@ class BaseTablet {
         return _max_version_schema;
     }
 
+    TabletSchemaSPtr calculate_variant_extended_schema() const;
+
     void traverse_rowsets(std::function<void(const RowsetSharedPtr&)> visitor,
                           bool include_stale = false) {
         std::shared_lock rlock(_meta_lock);