[opt](variant) Reduce sparse variant parse memory (#63970)

## Proposed changes

This PR addresses Variant import memory for sparse dynamic keys.

- Parse plain dynamic non-doc Variant JSON into doc-value KV during
storage parse instead of eagerly expanding every path into parse-time
subcolumns.
- Keep the eager subcolumn parse path for cases that still depend on
parse-time path/type metadata: nested group, deprecated flatten nested,
predefined typed paths, and parent inverted index columns.
- For legacy multi-subcolumn `ColumnVariant` blocks that already reach
the segment writer, append them into a doc-value intermediate when the
writer buffer is still root-only. This avoids copying thousands of
sparse subcolumns into the writer append buffer.
- Stream large doc-value materialization sets one path at a time when
selected materialized doc paths exceed 64, instead of holding all
materialized sparse subcolumns in memory.
- Gate serialized Variant doc-value block payloads by BE exec version,
so WAL blocks written with `be_exec_version=9` replay correctly.
- Add focused Release-gated BE UT/perf coverage. The perf tests stay
skipped by default unless `DORIS_RUN_VARIANT_WRITE_PERF_TEST=1` is set.

## Problem

The downloaded WAL first block is `be_exec_version=9`. It deserializes
as a root-only Variant column:

```text
source_rows=9234 source_subcolumns=1 source_sparse_entries=0 source_doc_value_entries=0
```

So this real WAL does not reproduce the 1000-subcolumn writer-buffer
shape directly. It did expose the old-version doc-value serialization
compatibility issue, which is fixed by the version gate.

Full first-block Release memory result:

```text
cir20431_wal_variant_memory rows=9234 source_rows=9234 source_subcolumns=1 source_sparse_entries=0 source_doc_value_entries=0 source_allocated=67452928 legacy_append_allocated=67649536 doc_value_append_allocated=67518464 doc_vs_legacy=0.998062
```

## Release Writer Perf

Release BE UT build (`BUILD_TYPE_UT=Release`, `-O3 -DNDEBUG`). Best of 3
measured runs after 1 warmup. JSON generation and JSON-to-Variant parse
are excluded; the measured window covers conversion, append, finish,
data/index writes, and file close.

```text
variant_write_perf case=sparse_keys rows=8192 paths_per_row=32 max_subcolumns=2 legacy_us=156634 kv_us=28341 kv_vs_legacy=0.180938 legacy_input_allocated=148013056 kv_input_allocated=10747904 legacy_append_buffer_allocated=148013056 optimized_append_buffer_bytes=6346316 kv_append_buffer_bytes=6346316 legacy_footer_columns=4 kv_footer_columns=4 legacy_materialized=2 kv_materialized=2 legacy_sparse=1 kv_sparse=1 legacy_doc_value=0 kv_doc_value=0 legacy_file_size=245659 kv_file_size=254370
variant_write_perf case=dense_keys rows=8192 paths_per_row=32 max_subcolumns=32 legacy_us=25043 kv_us=17450 kv_vs_legacy=0.696802 legacy_input_allocated=4980736 kv_input_allocated=10747904 legacy_append_buffer_allocated=4980736 optimized_append_buffer_bytes=5816320 kv_append_buffer_bytes=5816320 legacy_footer_columns=34 kv_footer_columns=34 legacy_materialized=32 kv_materialized=32 legacy_sparse=1 kv_sparse=1 legacy_doc_value=0 kv_doc_value=0 legacy_file_size=12963 kv_file_size=12963
```

Interpretation:

- Sparse-key shape: writer append buffer drops from `148013056` bytes to
`6346316` bytes, and write time is `0.181x` legacy.
- Dense-key shape: all 32 paths are materialized, so memory is roughly
comparable (`4980736` bytes vs `5816320` bytes) and write time is
`0.697x` legacy in this Release run.
- Both cases report `legacy_doc_value=0 kv_doc_value=0`, with identical
footer column counts in each case. The non-doc path does not persist
both doc-value and materialized subcolumns; doc-value is used as an
intermediate before writer-side materialization/sparse writing.

## Testing

Current head: `e27bdc09407033191d7d9770dda1ed60d2bb55ef`

- `clang-format` on modified C++ files before commit.
- `git diff --check`
- `env BUILD_TYPE_UT=Release DORIS_RUN_VARIANT_WRITE_PERF_TEST=1
DORIS_CIR20431_WAL_FILE=/tmp/cir20431_wal/walbak/1_1778071896877_16715020621810688_group_commit_b94fdc3cd7568b18_30cfcd56d2cfeb8f
DORIS_CLANG_HOME=/mnt/disk1/claude-max/ldb_toolchain20
PATH=/mnt/disk1/claude-max/ldb_toolchain20/bin:$PATH ./run-be-ut.sh
--run
--filter='VariantColumnWriterReaderTest.test_legacy_subcolumn_append_as_doc_value_buffer:VariantColumnWriterReaderTest.test_storage_parse_kv_write_materialized_and_sparse:VariantColumnWriterReaderTest.test_cir20431_wal_doc_value_buffer_memory:VariantColumnWriterReaderTest.test_storage_parse_kv_write_perf'`
- `env BUILD_TYPE_UT=Release
DORIS_CIR20431_WAL_FILE=/tmp/cir20431_wal/walbak/1_1778071896877_16715020621810688_group_commit_b94fdc3cd7568b18_30cfcd56d2cfeb8f
DORIS_CIR20431_WAL_ROWS=9234
DORIS_CLANG_HOME=/mnt/disk1/claude-max/ldb_toolchain20
PATH=/mnt/disk1/claude-max/ldb_toolchain20/bin:$PATH ./run-be-ut.sh
--run
--filter='VariantColumnWriterReaderTest.test_cir20431_wal_doc_value_buffer_memory'`

The Release test script used its default parallelism (`PARALLEL -- 39`);
no manual `-j` was passed.

Author: eldenmoon

be/src/common/config.cpp

@@ -1166,12 +1166,18 @@ DEFINE_Int32(blocking_pipeline_executor_size, "0");
 DEFINE_mInt32(variant_max_json_key_length, "255");
 DEFINE_mBool(variant_throw_exeception_on_invalid_json, "false");
 DEFINE_mBool(variant_enable_duplicate_json_path_check, "false");
+// Controls storage-layer parse target for plain non-doc VARIANT columns:
+// 0 = auto, 1 = force parse-time subcolumns, 2 = force doc-value KV staging.
+// NestedGroup, deprecated flatten-nested, and persistent doc mode keep their required paths.
+DEFINE_mInt32(variant_storage_parse_mode, "0");
 DEFINE_mBool(enable_vertical_compact_variant_subcolumns, "true");
 DEFINE_mBool(enable_variant_doc_sparse_write_subcolumns, "true");
 DEFINE_mBool(variant_nested_group_discard_scalar_on_conflict, "false");
 
 DEFINE_Validator(variant_max_json_key_length,
                  [](const int config) -> bool { return config > 0 && config <= 65535; });
+DEFINE_Validator(variant_storage_parse_mode,
+                 [](const int config) -> bool { return config >= 0 && config <= 2; });
 
 // block file cache
 DEFINE_Bool(enable_file_cache, "false");

be/src/common/config.h

@@ -1428,6 +1428,9 @@ DECLARE_mInt32(variant_max_json_key_length);
 DECLARE_mBool(variant_throw_exeception_on_invalid_json);
 // Enable duplicate path check when parsing json into variant subcolumns/jsonb.
 DECLARE_mBool(variant_enable_duplicate_json_path_check);
+// Controls storage-layer parse target for plain non-doc VARIANT columns:
+// 0 = auto, 1 = force parse-time subcolumns, 2 = force doc-value KV staging.
+DECLARE_mInt32(variant_storage_parse_mode);
 // Enable vertical compact subcolumns of variant column
 DECLARE_mBool(enable_vertical_compact_variant_subcolumns);
 DECLARE_mBool(enable_variant_doc_sparse_write_subcolumns);

be/src/exec/common/variant_util.cpp

@@ -1988,7 +1988,6 @@ void parse_json_to_variant_impl(IColumn& column, const char* src, size_t length,
         }
         break;
     case ParseConfig::ParseTo::OnlyDocValueColumn: {
-        CHECK(column_variant.enable_doc_mode()) << "OnlyDocValueColumn requires doc mode enabled";
         std::vector<size_t> doc_item_indexes;
         doc_item_indexes.reserve(paths.size());
         phmap::flat_hash_set<StringRef, StringRefHash> seen_paths;
@@ -1998,6 +1997,14 @@ void parse_json_to_variant_impl(IColumn& column, const char* src, size_t length,
             FieldInfo field_info;
             get_field_info(values[i], &field_info);
             if (paths[i].empty()) {
+                // Plain non-doc VARIANT can use doc-value KV as writer-side staging. An
+                // invalid root entry from JSON object/array is neither a scalar root value nor
+                // a doc KV path, so leave this row's doc offset empty. Doc-mode and valid scalar
+                // roots still populate the root subcolumn below.
+                if (!column_variant.enable_doc_mode() &&
+                    field_info.scalar_type_id == PrimitiveType::INVALID_TYPE) {
+                    continue;
+                }
                 auto* subcolumn = column_variant.get_subcolumn(paths[i]);
                 DCHECK(subcolumn != nullptr);
                 flush_defaults(subcolumn);
@@ -2217,6 +2224,49 @@ Status parse_and_materialize_variant_columns(Block& block, const std::vector<uin
             { return _parse_and_materialize_variant_columns(block, variant_pos, configs); });
 }
 
+namespace {
+
+ParseConfig::ParseTo select_storage_variant_parse_target(const TabletColumn& column,
+                                                         const ParseConfig& config) {
+    // NestedGroup consumes the parse-time subcolumn tree to build nested storage structures, so it
+    // must not go through doc-value staging.
+    if (column.variant_enable_nested_group()) {
+        return ParseConfig::ParseTo::OnlySubcolumns;
+    }
+
+    // Persistent doc mode owns doc-value bucket columns in VariantDocWriter. Keep it separate from
+    // the plain non-doc staging optimization, even when typed paths or parent indexes exist.
+    if (column.variant_enable_doc_mode()) {
+        return ParseConfig::ParseTo::OnlyDocValueColumn;
+    }
+
+    // Deprecated flatten-nested still consumes parse-time subcolumns. Predefined typed paths and
+    // parent inverted indexes are handled later by regular doc-value staging: typed paths are
+    // forced into the materialized set unless typed-to-sparse is enabled, and materialized dynamic
+    // subcolumns inherit parent indexes while sparse payloads stay unindexed.
+    if (config.deprecated_enable_flatten_nested) {
+        return ParseConfig::ParseTo::OnlySubcolumns;
+    }
+
+    // Plain dynamic non-doc VARIANT can avoid eagerly creating thousands of parse-time subcolumns.
+    // The segment writer will pick the materialized/sparse split from this doc-value KV staging.
+    // Keep a BE switch so tests and rollouts can compare the old parse-time path with staging under
+    // the same writer and schema.
+    switch (config::variant_storage_parse_mode) {
+    case 0:
+    case 2:
+        return ParseConfig::ParseTo::OnlyDocValueColumn;
+    case 1:
+        return ParseConfig::ParseTo::OnlySubcolumns;
+    default:
+        CHECK(false) << "invalid variant_storage_parse_mode: "
+                     << config::variant_storage_parse_mode;
+        return ParseConfig::ParseTo::OnlyDocValueColumn;
+    }
+}
+
+} // namespace
+
 Status parse_and_materialize_variant_columns(Block& block, const TabletSchema& tablet_schema,
                                              const std::vector<uint32_t>& column_pos) {
     std::vector<uint32_t> variant_column_pos;
@@ -2247,13 +2297,7 @@ Status parse_and_materialize_variant_columns(Block& block, const TabletSchema& t
             return Status::InternalError("column is not variant type, column name: {}",
                                          column.name());
         }
-        // if doc mode is not enabled, no need to parse to doc value column
-        if (!column.variant_enable_doc_mode()) {
-            configs[i].parse_to = ParseConfig::ParseTo::OnlySubcolumns;
-            continue;
-        }
-
-        configs[i].parse_to = ParseConfig::ParseTo::OnlyDocValueColumn;
+        configs[i].parse_to = select_storage_variant_parse_target(column, configs[i]);
     }
 
     RETURN_IF_ERROR(parse_and_materialize_variant_columns(block, variant_column_pos, configs));

be/src/storage/segment/segment_writer.cpp

@@ -1244,11 +1244,11 @@ Status SegmentWriter::_generate_short_key_index(std::vector<IOlapColumnDataAcces
     return Status::OK();
 }
 
-inline bool SegmentWriter::_is_mow() {
+bool SegmentWriter::_is_mow() {
     return _tablet_schema->keys_type() == UNIQUE_KEYS && _opts.enable_unique_key_merge_on_write;
 }
 
-inline bool SegmentWriter::_is_mow_with_cluster_key() {
+bool SegmentWriter::_is_mow_with_cluster_key() {
     return _is_mow() && !_tablet_schema->cluster_key_uids().empty();
 }