chore: Improve Virtual Column Block Meta Generation

Author: b41sh

Cargo.lock

@@ -643,7 +643,9 @@ databend-meta = { git = "https://github.com/databendlabs/databend-meta.git", tag
 databend-meta-client = { git = "https://github.com/databendlabs/databend-meta.git", tag = "v260205.13.2" }
 databend-meta-test-harness = { git = "https://github.com/databendlabs/databend-meta.git", tag = "v260629.2.0" }
 deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "9954bff" }
-jsonb = { git = "https://github.com/databendlabs/jsonb.git", rev = "a16344417d1fec6df4a62d8e77679211e909f86e" }
+#jsonb = { git = "https://github.com/databendlabs/jsonb.git", rev = "a16344417d1fec6df4a62d8e77679211e909f86e" }
+#jsonb = { git = "https://github.com/b41sh/jsonb.git", rev = "816c7ab394b5732ea621e9315723a34a96d9047c" }
+jsonb = { git = "https://github.com/b41sh/jsonb.git", rev = "ec1130a0f6a5e63c6a741e06bc4733e1a27c47d3" }
 lance-arrow = { git = "https://github.com/datafuse-extras/lance", rev = "85f9401d3246d52a287bca21457dbae0466858ee" }
 lance-core = { git = "https://github.com/datafuse-extras/lance", rev = "85f9401d3246d52a287bca21457dbae0466858ee" }
 lance-encoding = { git = "https://github.com/datafuse-extras/lance", rev = "85f9401d3246d52a287bca21457dbae0466858ee" }

Cargo.toml

@@ -21,6 +21,7 @@ mod projection;
 mod pruning_statistics;
 mod pushdown;
 mod stream_column;
+mod virtual_column;
 
 pub use agg_index::*;
 pub use datasource::*;
@@ -31,3 +32,4 @@ pub use projection::Projection;
 pub use pruning_statistics::PruningStatistics;
 pub use pushdown::*;
 pub use stream_column::*;
+pub use virtual_column::*;

src/query/catalog/src/plan/mod.rs

@@ -34,6 +34,7 @@ use rand::thread_rng;
 use sha2::Digest;
 
 use crate::plan::PartStatistics;
+use crate::plan::VirtualColumnLayout;
 use crate::table_context::TableContext;
 
 /// Partition information.
@@ -441,6 +442,7 @@ pub struct ReclusterTask {
     // All input blocks in this task are already ordered by the current cluster key.
     #[serde(default)]
     pub all_ordered: bool,
+    pub virtual_column_layout: Option<VirtualColumnLayout>,
 }
 
 pub type BlockMetaWithHLL = (Arc<BlockMeta>, Option<RawBlockHLL>);

src/query/catalog/src/plan/partition.rs

@@ -21,6 +21,7 @@ use databend_common_ast::ast::SampleConfig;
 use databend_common_exception::Result;
 use databend_common_expression::ColumnId;
 use databend_common_expression::DataSchema;
+use databend_common_expression::Expr;
 use databend_common_expression::FunctionRegistry;
 use databend_common_expression::RemoteExpr;
 use databend_common_expression::SEARCH_MATCHED_COL_NAME;
@@ -57,9 +58,11 @@ pub struct VirtualColumnField {
     pub source_column_id: u32,
     /// The source column name.
     pub source_name: String,
-    /// The virtual column id
-    pub column_id: u32,
-    /// The virtual column name.
+    /// Query-time temporary column id. This is not a persisted segment-local
+    /// virtual column id; each segment may assign a different real column id
+    /// for the same path, so readers must map this id through the segment schema.
+    pub query_column_id: u32,
+    /// Full query field name, including the source column, e.g. `v.user.name` or `v[0].id`.
     pub name: String,
     /// Paths to generate virtual column from source column.
     pub key_paths: OwnedKeyPaths,
@@ -69,6 +72,21 @@ pub struct VirtualColumnField {
     pub data_type: Box<TableDataType>,
 }
 
+/// Query-time identity of a virtual column referenced by a pushed-down filter
+/// or ordering expression. It bridges the query column id to the persisted
+/// source-column/canonical-path identity used to locate segment-local stats.
+#[derive(Clone, Debug)]
+pub struct VirtualPredicateRef {
+    /// Full query/pipeline field name used by expression column references.
+    pub name: String,
+    /// Id of the authoritative source Variant column.
+    pub source_column_id: u32,
+    /// Query-time temporary column id used by runtime filters and readers.
+    pub query_column_id: u32,
+    /// Compact canonical JSON path relative to the source column.
+    pub encoded_path: String,
+}
+
 /// Information about prewhere optimization.
 ///
 /// Prewhere steps:
@@ -207,6 +225,30 @@ pub struct PushDownInfo {
 }
 
 impl PushDownInfo {
+    pub fn virtual_predicate_refs(
+        &self,
+        filter_expr: Option<&Expr<String>>,
+    ) -> Vec<VirtualPredicateRef> {
+        let Some(virtual_column) = &self.virtual_column else {
+            return Vec::new();
+        };
+        let mut referenced_columns = filter_expr.map(Expr::column_refs).unwrap_or_default();
+        for (expr, _, _) in &self.order_by {
+            referenced_columns.extend(expr.column_refs());
+        }
+        virtual_column
+            .virtual_column_fields
+            .iter()
+            .filter(|field| referenced_columns.contains_key(&field.name))
+            .map(|field| VirtualPredicateRef {
+                name: field.name.clone(),
+                source_column_id: field.source_column_id,
+                query_column_id: field.query_column_id,
+                encoded_path: field.key_paths.to_canonical_path(),
+            })
+            .collect()
+    }
+
     pub fn add_internal_column_dependencies<'a>(
         &mut self,
         schema: &TableSchema,

src/query/catalog/src/plan/pushdown.rs

@@ -0,0 +1,45 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed 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.
+
+use databend_common_expression::ColumnId;
+
+#[derive(
+    Clone,
+    Debug,
+    Default,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Hash,
+    serde::Serialize,
+    serde::Deserialize,
+)]
+pub struct VirtualColumnPath {
+    pub source_column_id: ColumnId,
+    pub path: String,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct VirtualColumnLayout {
+    pub direct_paths: Vec<VirtualColumnPath>,
+}
+
+impl VirtualColumnLayout {
+    pub fn contains(&self, source_column_id: ColumnId, path: &str) -> bool {
+        self.direct_paths
+            .iter()
+            .any(|item| item.source_column_id == source_column_id && item.path == path)
+    }
+}

src/query/catalog/src/plan/virtual_column.rs

@@ -247,7 +247,7 @@ fn classify_decimal_conversion(src: DecimalSize, dest: DecimalSize) -> Conversio
     }
 }
 
-fn number_common_type(left: NumberDataType, right: NumberDataType) -> DataType {
+pub fn number_common_type(left: NumberDataType, right: NumberDataType) -> DataType {
     if left == right {
         return DataType::Number(left);
     }

src/query/expression/src/conversion.rs

@@ -1103,6 +1103,29 @@ impl<Index: ColumnIndex> Expr<Index> {
 }
 
 impl<Index: ColumnIndex> RemoteExpr<Index> {
+    pub fn column_refs(&self) -> HashMap<Index, DataType> {
+        #[recursive::recursive]
+        fn walk<Index: ColumnIndex>(expr: &RemoteExpr<Index>, refs: &mut HashMap<Index, DataType>) {
+            match expr {
+                RemoteExpr::ColumnRef { id, data_type, .. } => {
+                    refs.insert(id.clone(), data_type.clone());
+                }
+                RemoteExpr::Cast { expr, .. } => walk(expr, refs),
+                RemoteExpr::FunctionCall { args, .. }
+                | RemoteExpr::LambdaFunctionCall { args, .. } => {
+                    for arg in args {
+                        walk(arg, refs);
+                    }
+                }
+                RemoteExpr::Constant { .. } => {}
+            }
+        }
+
+        let mut refs = HashMap::new();
+        walk(self, &mut refs);
+        refs
+    }
+
     pub fn as_expr(&self, fn_registry: &FunctionRegistry) -> Expr<Index> {
         match self {
             RemoteExpr::Constant {

src/query/expression/src/expression.rs

@@ -43,7 +43,11 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_RECLUSTER_DEPTH;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_BLOCK;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_PAGE;
+use databend_common_storages_fuse::FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_DIRECT_COLUMNS;
+use databend_common_storages_fuse::FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_PATH_STATISTICS;
 use databend_common_storages_fuse::MAX_RECLUSTER_DEPTH;
+use databend_common_storages_fuse::MAX_VIRTUAL_COLUMN_DIRECT_COLUMNS;
+use databend_common_storages_fuse::MAX_VIRTUAL_COLUMN_PATH_STATISTICS;
 use databend_common_storages_fuse::MIN_RECLUSTER_DEPTH;
 use databend_storages_common_index::BloomIndex;
 use databend_storages_common_index::RangeIndex;
@@ -94,6 +98,8 @@ pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(
     r.insert(FUSE_OPT_KEY_ENABLE_AUTO_VACUUM);
     r.insert(FUSE_OPT_KEY_ENABLE_AUTO_ANALYZE);
     r.insert(FUSE_OPT_KEY_ENABLE_VIRTUAL_COLUMN);
+    r.insert(FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_DIRECT_COLUMNS);
+    r.insert(FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_PATH_STATISTICS);
     r.insert(FUSE_OPT_KEY_AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD);
 
     r.insert(OPT_KEY_BLOOM_INDEX_COLUMNS);
@@ -219,6 +225,8 @@ pub static UNSET_TABLE_OPTIONS_WHITE_LIST: LazyLock<HashSet<&'static str>> = Laz
     r.insert(FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP);
     r.insert(FUSE_OPT_KEY_AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD);
     r.insert(FUSE_OPT_KEY_ENABLE_VIRTUAL_COLUMN);
+    r.insert(FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_DIRECT_COLUMNS);
+    r.insert(FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_PATH_STATISTICS);
     r.insert(OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH);
     r.insert(FUSE_OPT_KEY_DATA_PAGE_ROWS);
     r.insert(FUSE_OPT_KEY_DATA_PAGE_BYTES);
@@ -230,6 +238,28 @@ pub static UNSET_TABLE_OPTIONS_WHITE_LIST: LazyLock<HashSet<&'static str>> = Laz
     r
 });
 
+pub fn is_valid_virtual_column_layout_options(
+    options: &BTreeMap<String, String>,
+) -> databend_common_exception::Result<()> {
+    if let Some(value) = options.get(FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_DIRECT_COLUMNS) {
+        let value = value.parse::<usize>()?;
+        if value > MAX_VIRTUAL_COLUMN_DIRECT_COLUMNS {
+            return Err(ErrorCode::TableOptionInvalid(format!(
+                "{FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_DIRECT_COLUMNS} must be between 0 and {MAX_VIRTUAL_COLUMN_DIRECT_COLUMNS}",
+            )));
+        }
+    }
+    if let Some(value) = options.get(FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_PATH_STATISTICS) {
+        let value = value.parse::<usize>()?;
+        if value > MAX_VIRTUAL_COLUMN_PATH_STATISTICS {
+            return Err(ErrorCode::TableOptionInvalid(format!(
+                "{FUSE_OPT_KEY_VIRTUAL_COLUMN_MAX_PATH_STATISTICS} must be between 0 and {MAX_VIRTUAL_COLUMN_PATH_STATISTICS}",
+            )));
+        }
+    }
+    Ok(())
+}
+
 pub fn is_valid_create_opt<S: AsRef<str>>(opt_key: S, engine: &Engine) -> bool {
     let opt_key = opt_key.as_ref().to_lowercase();
     let opt_key = opt_key.as_str();

src/query/service/src/interpreters/common/table_option_validation.rs

@@ -92,6 +92,7 @@ use crate::interpreters::common::table_option_validation::is_valid_option_of_typ
 use crate::interpreters::common::table_option_validation::is_valid_random_seed;
 use crate::interpreters::common::table_option_validation::is_valid_recluster_depth;
 use crate::interpreters::common::table_option_validation::is_valid_row_per_block;
+use crate::interpreters::common::table_option_validation::is_valid_virtual_column_layout_options;
 use crate::interpreters::hook::vacuum_hook::hook_clear_m_cte_temp_table;
 use crate::interpreters::hook::vacuum_hook::hook_disk_temp_dir;
 use crate::interpreters::hook::vacuum_hook::hook_vacuum_temp_files;
@@ -564,6 +565,7 @@ impl CreateTableInterpreter {
         is_valid_fuse_parquet_dictionary_opt(&table_meta.options)?;
         // check enable_virtual_column
         is_valid_fuse_virtual_column_opt(&table_meta.options)?;
+        is_valid_virtual_column_layout_options(&table_meta.options)?;
         is_valid_data_page_rows(&table_meta.options)?;
         is_valid_data_page_bytes(&table_meta.options)?;
         is_valid_analyze_histogram_algorithm(&table_meta.options)?;