fix

Author: csun5285

be/src/olap/rowset/segment_creator.cpp

@@ -103,10 +103,38 @@ Status SegmentFlusher::_internal_parse_variant_columns(vectorized::Block& block)
         return Status::OK();
     }
 
-    vectorized::ParseConfig config;
-    config.enable_flatten_nested = _context.tablet_schema->variant_flatten_nested();
+    std::vector<vectorized::ParseConfig> configs(variant_column_pos.size());
+    for (size_t i = 0; i < variant_column_pos.size(); ++i) {
+        configs[i].enable_flatten_nested = _context.tablet_schema->variant_flatten_nested();
+        const auto& column = _context.tablet_schema->column(variant_column_pos[i]);
+        if (column.is_variant_type()) {
+            // enable doc snapshot mode
+            if (column.variant_enable_doc_snapshot_mode()) {
+                // if has schema template, no need to parse to doc snapshot, when writing data, we will parse to doc snapshot
+                if (column.get_sub_columns().empty()) {
+                    configs[i].parse_to_doc_snapshot = true;
+                } else {
+                    configs[i].parse_to_subcolumns = false;
+                }
+
+                // if min rows is greater than 0, no need to parse to subcolumns
+                // when compaction row size is greater than min rows, parse to subcolumns
+                if (column.variant_doc_snapshot_min_rows() > 0) {
+                    configs[i].parse_to_subcolumns = false;
+                } else {
+                    configs[i].parse_to_subcolumns = true;
+                }
+            } else {
+                // default: only parse to subcolumns
+                configs[i].parse_to_subcolumns = true;
+                configs[i].parse_to_doc_snapshot = false;
+            }
+        } else {
+            return Status::InternalError("column is not variant type, column name: {}", column.name());
+        }
+    }
     RETURN_IF_ERROR(
-            vectorized::schema_util::parse_variant_columns(block, variant_column_pos, config));
+            vectorized::schema_util::parse_variant_columns(block, variant_column_pos, configs));
     return Status::OK();
 }
 

be/src/olap/rowset/segment_v2/segment_writer.h

@@ -76,6 +76,8 @@ struct SegmentWriterOptions {
     RowsetWriterContext* rowset_ctx = nullptr;
     DataWriteType write_type = DataWriteType::TYPE_DEFAULT;
     std::shared_ptr<MowContext> mow_ctx;
+
+    bool 
 };
 
 using TabletSharedPtr = std::shared_ptr<Tablet>;

be/src/olap/rowset/segment_v2/variant/variant_column_writer_impl.cpp

@@ -1068,8 +1068,9 @@ Status VariantCompactionDocSnapshotWriter::finalize() {
     size_t num_rows = variant_column->size();
     auto converter = std::make_unique<vectorized::OlapBlockDataConvertor>();
     int column_id = 0;
-    int variant_doc_snapshot_min_rows = parent_column.variant_doc_snapshot_min_rows();
-    if (num_rows >= variant_doc_snapshot_min_rows) {
+    int64_t variant_doc_snapshot_min_rows = parent_column.variant_doc_snapshot_min_rows();
+    if (variant_doc_snapshot_min_rows >= 0
+            && num_rows >= static_cast<size_t>(variant_doc_snapshot_min_rows)) {
         std::unordered_map<std::string_view, vectorized::ColumnVariant::Subcolumn> subcolumns;
 
         auto [column_key, column_value] = variant_column->get_doc_snapshot_data_paths_and_values();

be/src/olap/tablet_schema.h

@@ -233,9 +233,9 @@ class TabletColumn : public MetadataAdder<TabletColumn> {
 
     bool variant_enable_doc_snapshot_mode() const { return _variant_enable_doc_snapshot_mode; }
 
-    int32_t variant_doc_snapshot_min_rows() const { return _variant_doc_snapshot_min_rows; }
+    int64_t variant_doc_snapshot_min_rows() const { return _variant_doc_snapshot_min_rows; }
 
-    void set_variant_doc_snapshot_min_rows(int32_t variant_doc_snapshot_min_rows) {
+    void set_variant_doc_snapshot_min_rows(int64_t variant_doc_snapshot_min_rows) {
         _variant_doc_snapshot_min_rows = variant_doc_snapshot_min_rows;
     }
 
@@ -292,7 +292,7 @@ class TabletColumn : public MetadataAdder<TabletColumn> {
 
     bool _variant_enable_doc_snapshot_mode = false;
 
-    int32_t _variant_doc_snapshot_min_rows = 0;
+    int64_t _variant_doc_snapshot_min_rows = 0;
 };
 
 bool operator==(const TabletColumn& a, const TabletColumn& b);

be/src/vec/common/schema_util.cpp

@@ -530,7 +530,7 @@ Status get_least_common_schema(const std::vector<TabletSchemaSPtr>& schemas,
 }
 
 Status _parse_variant_columns(Block& block, const std::vector<int>& variant_pos,
-                              const ParseConfig& config) {
+                              const std::vector<ParseConfig>& configs) {
     for (int i = 0; i < variant_pos.size(); ++i) {
         auto column_ref = block.get_by_position(variant_pos[i]).column;
         bool is_nullable = column_ref->is_nullable();
@@ -544,13 +544,11 @@ Status _parse_variant_columns(Block& block, const std::vector<int>& variant_pos,
 
         MutableColumnPtr variant_column;
         if (var.is_doc_snapshot_mode()) {
-            LOG(INFO) << "parse doc snapshot variant column";
             // doc snapshot mode, we need to parse the doc snapshot column
-            parse_binary_to_variant(var, config);
+            parse_binary_to_variant(var);
             continue;
         }
         if (!var.is_scalar_variant()) {
-            LOG(INFO) << "already parsed variant column";
             // already parsed
             continue;
         }
@@ -580,7 +578,7 @@ Status _parse_variant_columns(Block& block, const std::vector<int>& variant_pos,
         if (scalar_root_column->is_column_string()) {
             variant_column = ColumnVariant::create(0);
             parse_json_to_variant(*variant_column.get(),
-                                  assert_cast<const ColumnString&>(*scalar_root_column), config);
+                                  assert_cast<const ColumnString&>(*scalar_root_column), configs[i]);
         } else {
             // Root maybe other types rather than string like ColumnVariant(Int32).
             // In this case, we should finlize the root and cast to JSON type
@@ -603,10 +601,10 @@ Status _parse_variant_columns(Block& block, const std::vector<int>& variant_pos,
 }
 
 Status parse_variant_columns(Block& block, const std::vector<int>& variant_pos,
-                             const ParseConfig& config) {
+                             const std::vector<ParseConfig>& configs) {
     // Parse each variant column from raw string column
     RETURN_IF_CATCH_EXCEPTION({
-        return vectorized::schema_util::_parse_variant_columns(block, variant_pos, config);
+        return vectorized::schema_util::_parse_variant_columns(block, variant_pos, configs);
     });
 }
 

be/src/vec/common/schema_util.h

@@ -98,7 +98,7 @@ TabletColumn get_column_by_type(const vectorized::DataTypePtr& data_type, const
 // 2. finalize variant column to each subcolumn least commn types, default ignore sparse sub columns
 // 3. encode sparse sub columns
 Status parse_variant_columns(Block& block, const std::vector<int>& variant_pos,
-                             const ParseConfig& config);
+                             const std::vector<ParseConfig>& configs);
 
 // check if the tuple_paths has ambiguous paths
 // situation:

be/src/vec/json/json_parser.h

@@ -124,6 +124,12 @@ enum class ExtractType {
 
 struct ParseConfig {
     bool enable_flatten_nested = false;
+    // parse variant column to subcolumns or doc snapshot
+    // if parse_to_subcolumns is true, the variant column will be parsed to subcolumns
+    // if parse_to_doc_snapshot is true, the variant column will be parsed to doc snapshot
+    // if both are true, the variant column will be parsed to both subcolumns and doc snapshot
+    bool parse_to_subcolumns = true;
+    bool parse_to_doc_snapshot = false;
 };
 /// Result of parsing of a document.
 /// Contains all paths extracted from document

be/src/vec/json/parse2column.cpp

@@ -165,35 +165,56 @@ void parse_json_to_variant(IColumn& column, const char* src, size_t length,
         check_paths.insert(check_paths.end(), paths.begin(), paths.end());
         THROW_IF_ERROR(vectorized::schema_util::check_variant_has_no_ambiguous_paths(check_paths));
     }
-    for (size_t i = 0; i < paths.size(); ++i) {
-        FieldInfo field_info;
-        schema_util::get_field_info(values[i], &field_info);
-        if (field_info.scalar_type_id == PrimitiveType::INVALID_TYPE) {
-            continue;
-        }
-        if (column_variant.get_subcolumn(paths[i], i) == nullptr) {
-            if (paths[i].has_nested_part()) {
-                column_variant.add_nested_subcolumn(paths[i], field_info, old_num_rows);
-            } else {
-                column_variant.add_sub_column(paths[i], old_num_rows);
+    auto [doc_snapshot_data_paths, doc_snapshot_data_values] = column_variant.get_doc_snapshot_data_paths_and_values();
+    auto& doc_snapshot_data_offsets = column_variant.serialized_doc_snapshot_column_offsets();
+    if (config.parse_to_subcolumns) {
+        for (size_t i = 0; i < paths.size(); ++i) {
+            FieldInfo field_info;
+            schema_util::get_field_info(values[i], &field_info);
+            if (field_info.scalar_type_id == PrimitiveType::INVALID_TYPE) {
+                continue;
+            }
+            if (column_variant.get_subcolumn(paths[i], i) == nullptr) {
+                if (paths[i].has_nested_part()) {
+                    column_variant.add_nested_subcolumn(paths[i], field_info, old_num_rows);
+                } else {
+                    column_variant.add_sub_column(paths[i], old_num_rows);
+                }
+            }
+            auto* subcolumn = column_variant.get_subcolumn(paths[i], i);
+            if (!subcolumn) {
+                throw doris::Exception(ErrorCode::INVALID_ARGUMENT, "Failed to find sub column {}",
+                                       paths[i].get_path());
+            }
+            if (subcolumn->cur_num_of_defaults() > 0) {
+                subcolumn->insert_many_defaults(subcolumn->cur_num_of_defaults());
+                subcolumn->reset_current_num_of_defaults();
+            }
+            if (subcolumn->size() != old_num_rows) {
+                throw doris::Exception(ErrorCode::INVALID_ARGUMENT,
+                                       "subcolumn {} size missmatched, may contains duplicated entry",
+                                       paths[i].get_path());
+            }
+            subcolumn->insert(std::move(values[i]), std::move(field_info));
+            if (!paths[i].empty() && config.parse_to_doc_snapshot) {
+                subcolumn->serialize_to_sparse_column(doc_snapshot_data_paths, paths[i].get_path(), doc_snapshot_data_values, old_num_rows);
             }
         }
-        auto* subcolumn = column_variant.get_subcolumn(paths[i], i);
-        if (!subcolumn) {
-            throw doris::Exception(ErrorCode::INVALID_ARGUMENT, "Failed to find sub column {}",
-                                   paths[i].get_path());
-        }
-        if (subcolumn->cur_num_of_defaults() > 0) {
-            subcolumn->insert_many_defaults(subcolumn->cur_num_of_defaults());
-            subcolumn->reset_current_num_of_defaults();
-        }
-        if (subcolumn->size() != old_num_rows) {
-            throw doris::Exception(ErrorCode::INVALID_ARGUMENT,
-                                   "subcolumn {} size missmatched, may contains duplicated entry",
-                                   paths[i].get_path());
+    } else {
+        CHECK(config.parse_to_doc_snapshot);
+        for (size_t i = 0; i < paths.size(); ++i) {
+            FieldInfo field_info;
+            schema_util::get_field_info(values[i], &field_info);
+            if (paths[i].empty()) {
+                column_variant.get_subcolumn(paths[i])->insert(std::move(values[i]), std::move(field_info));
+                continue;
+            }
+            ColumnVariant::Subcolumn tmp_subcolumn(0, true);
+            tmp_subcolumn.insert(std::move(values[i]), std::move(field_info));
+            tmp_subcolumn.serialize_to_sparse_column(doc_snapshot_data_paths, paths[i].get_path(), doc_snapshot_data_values, 0);
         }
-        subcolumn->insert(std::move(values[i]), std::move(field_info));
     }
+    doc_snapshot_data_offsets.push_back(doc_snapshot_data_paths->size());
     // /// Insert default values to missed subcolumns.
     const auto& subcolumns = column_variant.get_subcolumns();
     for (const auto& entry : subcolumns) {
@@ -215,7 +236,6 @@ void parse_json_to_variant(IColumn& column, const char* src, size_t length,
     auto sparse_column = column_variant.get_sparse_column();
     if (sparse_column->size() == old_num_rows) {
         sparse_column->assume_mutable()->insert_default();
-        column_variant.get_doc_snapshot_column()->assume_mutable()->insert_default();
     }
 #ifndef NDEBUG
     column_variant.check_consistency();
@@ -240,7 +260,7 @@ void parse_json_to_variant(IColumn& column, const ColumnString& raw_json_column,
 
 
 // pasre the doc snapshot column to subcolumns
-void parse_binary_to_variant(ColumnVariant& column_variant, const ParseConfig& config) {
+void parse_binary_to_variant(ColumnVariant& column_variant) {
     std::unordered_map<std::string_view, vectorized::ColumnVariant::Subcolumn> subcolumns;
 
     auto [column_key, column_value] = column_variant.get_doc_snapshot_data_paths_and_values();

be/src/vec/json/parse2column.h

@@ -43,5 +43,5 @@ void parse_json_to_variant(IColumn& column, const ColumnString& raw_json_column,
 void parse_json_to_variant(IColumn& column, const StringRef& jsons, JsonParser* parser,
                            const ParseConfig& config);
 
-void parse_binary_to_variant(ColumnVariant& column_variant, const ParseConfig& config);
+void parse_binary_to_variant(ColumnVariant& column_variant);
 } // namespace doris::vectorized

fe/fe-common/src/main/java/org/apache/doris/catalog/ScalarType.java

@@ -1050,12 +1050,12 @@ public boolean getVariantEnableDocSnapshotMode() {
         return false; // The old variant type had a default value of false.
     }
 
-    public int getVariantDocSnapshotMinRows() {
+    public long getVariantDocSnapshotMinRows() {
         // In the past, variant metadata used the ScalarType type.
         // Now, we use VariantType, which inherits from ScalarType, as the new metadata storage.
         if (this instanceof VariantType) {
             return ((VariantType) this).getVariantDocSnapshotMinRows();
         }
-        return 0; // The old variant type had a default value of 0.
+        return 0L; // The old variant type had a default value of 0.
     }
 }