feat(storage): add variant shredding for parquet

fix(storage): stabilize variant shredding read path

Author: dantengsky

Cargo.lock

@@ -307,7 +307,7 @@ aws-smithy-runtime = "1.7.8"
 aws-smithy-runtime-api = "1.7.3"
 aws-smithy-types = "1.2.13"
 
-indexmap = "2.0.0"
+indexmap = "2.10.0"
 indicatif = "0.17.5"
 itertools = "0.13.0"
 jaq-core = "2.2.1"
@@ -377,7 +377,8 @@ orc-rust = "0.6.0"
 ordered-float = { version = "5.1.0", default-features = false }
 p256 = "0.13"
 parking_lot = "0.12.1"
-parquet = { version = "56", features = ["async"] }
+parquet = { version = "56", features = ["async", "variant_experimental"] }
+parquet-variant = "0.1.0"
 passwords = { version = "3.1.16" }
 paste = "1.0.15"
 percent-encoding = "2.3.1"

Cargo.toml

@@ -117,6 +117,18 @@ static MERGE_INTO_MATCHED_OPERATION_MILLISECONDS: LazyLock<Histogram> = LazyLock
     register_histogram_in_milliseconds("merge_into_matched_operation_milliseconds")
 });
 
+// Variant shredding metrics.
+static VARIANT_SHREDDING_INLINE_COLUMNS: LazyLock<Counter> =
+    LazyLock::new(|| register_counter("variant_shredding_inline_columns"));
+static VARIANT_SHREDDING_INLINE_VALUE_ALL_NULL_COLUMNS: LazyLock<Counter> =
+    LazyLock::new(|| register_counter("variant_shredding_inline_value_all_null_columns"));
+static VARIANT_SHREDDING_READ_TYPED_VALUE_HITS: LazyLock<Counter> =
+    LazyLock::new(|| register_counter("variant_shredding_read_typed_value_hits"));
+static VARIANT_SHREDDING_READ_TYPED_VALUE_MISSES: LazyLock<Counter> =
+    LazyLock::new(|| register_counter("variant_shredding_read_typed_value_misses"));
+static VARIANT_SHREDDING_UNSHRED_MILLISECONDS: LazyLock<Histogram> =
+    LazyLock::new(|| register_histogram_in_milliseconds("variant_shredding_unshred_milliseconds"));
+
 // Fuse engine metrics.
 static COMMIT_MUTATION_UNRESOLVABLE_CONFLICT: LazyLock<Counter> =
     LazyLock::new(|| register_counter("fuse_commit_mutation_unresolvable_conflict"));
@@ -937,3 +949,23 @@ pub fn metrics_inc_block_virtual_column_write_bytes(c: u64) {
 pub fn metrics_inc_block_virtual_column_write_milliseconds(c: u64) {
     BLOCK_VIRTUAL_COLUMN_WRITE_MILLISECONDS.observe(c as f64);
 }
+
+pub fn metrics_inc_variant_shredding_inline_columns(c: u64) {
+    VARIANT_SHREDDING_INLINE_COLUMNS.inc_by(c);
+}
+
+pub fn metrics_inc_variant_shredding_inline_value_all_null_columns(c: u64) {
+    VARIANT_SHREDDING_INLINE_VALUE_ALL_NULL_COLUMNS.inc_by(c);
+}
+
+pub fn metrics_inc_variant_shredding_read_typed_value_hits(c: u64) {
+    VARIANT_SHREDDING_READ_TYPED_VALUE_HITS.inc_by(c);
+}
+
+pub fn metrics_inc_variant_shredding_read_typed_value_misses(c: u64) {
+    VARIANT_SHREDDING_READ_TYPED_VALUE_MISSES.inc_by(c);
+}
+
+pub fn metrics_inc_variant_shredding_unshred_milliseconds(c: u64) {
+    VARIANT_SHREDDING_UNSHRED_MILLISECONDS.observe(c as f64);
+}

src/common/metrics/src/metrics/storage.rs

@@ -132,49 +132,44 @@ async fn test_virtual_column_pruner_reader() -> anyhow::Result<()> {
         .await?
         .expect("virtual block meta index");
 
-    let mut plan_kinds = HashSet::new();
-    for plans in virtual_block_meta_index.virtual_column_read_plan.values() {
-        for plan in plans {
-            collect_plan_kinds(plan, &mut plan_kinds);
-        }
-    }
-    assert!(plan_kinds.contains("Direct"));
-    assert!(plan_kinds.contains("FromParent"));
-    assert!(plan_kinds.contains("Shared"));
-    assert!(plan_kinds.contains("Object"));
+    let id_column_id = column_ids[0];
+    let text_column_id = column_ids[1];
+    let id_plans = virtual_block_meta_index
+        .virtual_column_read_plan
+        .get(&id_column_id)
+        .expect("id plans");
+    assert!(
+        id_plans
+            .iter()
+            .any(|plan| matches!(plan, VirtualColumnReadPlan::Direct { .. }))
+    );
+    let text_plans = virtual_block_meta_index
+        .virtual_column_read_plan
+        .get(&text_column_id)
+        .expect("text plans");
+    assert!(
+        text_plans
+            .iter()
+            .any(|plan| matches!(plan, VirtualColumnReadPlan::Direct { .. }))
+    );
 
     let info_column_id = column_ids[2];
     let tags0_column_id = column_ids[3];
     let extra_column_id = column_ids[4];
-
     let info_plans = virtual_block_meta_index
         .virtual_column_read_plan
-        .get(&info_column_id)
-        .expect("user.info plans");
-    assert!(
-        info_plans
-            .iter()
-            .any(|plan| matches!(plan, VirtualColumnReadPlan::Object { .. }))
-    );
+        .get(&info_column_id);
+    assert!(info_plans.is_none());
 
     let tags0_plans = virtual_block_meta_index
         .virtual_column_read_plan
-        .get(&tags0_column_id)
-        .expect("tags[0] plans");
-    assert!(tags0_plans.iter().any(|plan| matches!(
-        plan,
-        VirtualColumnReadPlan::FromParent { suffix_path, .. } if suffix_path == "{0}"
-    )));
+        .get(&tags0_column_id);
+    assert!(tags0_plans.is_none());
 
     let extra_plans = virtual_block_meta_index
         .virtual_column_read_plan
-        .get(&extra_column_id)
-        .expect("user.extra plans");
-    assert!(
-        extra_plans
-            .iter()
-            .any(|plan| matches!(plan, VirtualColumnReadPlan::Shared { .. }))
-    );
+        .get(&extra_column_id);
+    assert!(extra_plans.is_none());
 
     let plan = table
         .read_plan(ctx.clone(), Some(push_down), None, false, true)
@@ -273,27 +268,6 @@ fn format_virtual_column_name(source: &str, key_paths: &OwnedKeyPaths) -> String
     name
 }
 
-fn collect_plan_kinds(plan: &VirtualColumnReadPlan, kinds: &mut HashSet<&'static str>) {
-    match plan {
-        VirtualColumnReadPlan::Direct { .. } => {
-            kinds.insert("Direct");
-        }
-        VirtualColumnReadPlan::FromParent { parent, .. } => {
-            kinds.insert("FromParent");
-            collect_plan_kinds(parent, kinds);
-        }
-        VirtualColumnReadPlan::Shared { .. } => {
-            kinds.insert("Shared");
-        }
-        VirtualColumnReadPlan::Object { entries } => {
-            kinds.insert("Object");
-            for (_, child) in entries {
-                collect_plan_kinds(child, kinds);
-            }
-        }
-    }
-}
-
 fn assert_variant_column(column: &Column, expected: &[Option<&str>]) {
     assert_eq!(column.len(), expected.len());
     for (idx, expected_json) in expected.iter().enumerate() {

src/query/ee/tests/it/storages/fuse/operations/virtual_column_pruner_reader.rs

@@ -86,8 +86,8 @@ async fn test_virtual_column_builder() -> anyhow::Result<()> {
     assert!(!result.data.is_empty());
     assert_eq!(
         result.draft_virtual_block_meta.virtual_column_metas.len(),
-        4
-    ); // Expect a, b.c, b.d, e
+        5
+    ); // Expect a, b.c, b.d, e[0].f, e[1]
 
     // Check v['a']
     let meta_a = find_virtual_col(
@@ -122,16 +122,27 @@ async fn test_virtual_column_builder() -> anyhow::Result<()> {
     assert_eq!(meta_bd.name, "['b']['d']");
     assert_eq!(meta_bd.data_type, VariantDataType::Boolean);
 
-    // Check v['e']
-    let meta_bd = find_virtual_col(
+    // Check v['e'][0]['f']
+    let meta_e0f = find_virtual_col(
         &result.draft_virtual_block_meta.virtual_column_metas,
         1,
-        "['e']",
+        "['e'][0]['f']",
     )
-    .expect("Virtual column ['e'] not found");
-    assert_eq!(meta_bd.source_column_id, 1);
-    assert_eq!(meta_bd.name, "['e']");
-    assert_eq!(meta_bd.data_type, VariantDataType::Jsonb);
+    .expect("Virtual column ['e'][0]['f'] not found");
+    assert_eq!(meta_e0f.source_column_id, 1);
+    assert_eq!(meta_e0f.name, "['e'][0]['f']");
+    assert_eq!(meta_e0f.data_type, VariantDataType::UInt64);
+
+    // Check v['e'][1]
+    let meta_e1 = find_virtual_col(
+        &result.draft_virtual_block_meta.virtual_column_metas,
+        1,
+        "['e'][1]",
+    )
+    .expect("Virtual column ['e'][1] not found");
+    assert_eq!(meta_e1.source_column_id, 1);
+    assert_eq!(meta_e1.name, "['e'][1]");
+    assert_eq!(meta_e1.data_type, VariantDataType::UInt64);
 
     let block = DataBlock::new(
         vec![
@@ -293,8 +304,8 @@ async fn test_virtual_column_builder() -> anyhow::Result<()> {
     builder.add_block(&block)?;
     let result = builder.finalize(&write_settings, &location)?;
 
-    // all columns should be add to shared_column due to > 70% nulls
-    assert!(!result.data.is_empty());
+    // Paths with > 70% nulls are discarded.
+    assert!(result.data.is_empty());
     assert!(
         result
             .draft_virtual_block_meta

src/query/ee/tests/it/storages/fuse/operations/virtual_columns_builder.rs

@@ -19,6 +19,7 @@ async-backtrace = { workspace = true }
 base64 = { workspace = true }
 borsh = { workspace = true }
 bumpalo = { workspace = true }
+chrono = { workspace = true }
 comfy-table = { workspace = true }
 databend-common-ast = { workspace = true }
 databend-common-base = { workspace = true }
@@ -46,10 +47,11 @@ jsonb = { workspace = true }
 lexical-core = { workspace = true }
 log = { workspace = true }
 match-template = { workspace = true }
-memchr = { workspace = true, default-features = false }
+memchr = { default-features = false, workspace = true }
 micromarshal = { workspace = true }
 num-bigint = { workspace = true }
 num-traits = { workspace = true }
+parquet-variant = { workspace = true }
 rand = { workspace = true }
 rand_distr = { workspace = true }
 recursive = { workspace = true }