fix(query): Preserve case for Variant MapAccess in computed columns (#18872)

* fix(query): Preserve case for Variant MapAccess in computed columns

* test nested

* fix test

Author: b41sh

src/query/service/src/interpreters/interpreter_table_recluster.rs

@@ -620,9 +620,7 @@ impl ReclusterTableInterpreter {
         let cluster_keys_len = ast_exprs.len();
         let mut cluster_key_strs = Vec::with_capacity(cluster_keys_len);
         for mut ast in ast_exprs {
-            let mut normalizer = IdentifierNormalizer {
-                ctx: &name_resolution_ctx,
-            };
+            let mut normalizer = IdentifierNormalizer::new(&name_resolution_ctx);
             ast.drive_mut(&mut normalizer);
             cluster_key_strs.push(format!("{:#}", &ast));
         }

src/query/service/tests/it/sql/planner/semantic/name_resolution.rs

@@ -106,7 +106,7 @@ fn test_normalize_identifiers_in_expr() {
     let mut expr = parse_expr(&tokens, Dialect::PostgreSQL).unwrap();
 
     let ctx = NameResolutionContext::default();
-    let mut normalizer = IdentifierNormalizer { ctx: &ctx };
+    let mut normalizer = IdentifierNormalizer::new(&ctx);
 
     expr.drive_mut(&mut normalizer);
 

src/query/sql/src/planner/binder/ddl/table.rs

@@ -2170,9 +2170,7 @@ impl Binder {
             }
 
             let mut cluster_expr = cluster_expr.clone();
-            let mut normalizer = IdentifierNormalizer {
-                ctx: &self.name_resolution_ctx,
-            };
+            let mut normalizer = IdentifierNormalizer::new(&self.name_resolution_ctx);
             cluster_expr.drive_mut(&mut normalizer);
             cluster_keys.push(format!("{:#}", &cluster_expr));
         }

src/query/sql/src/planner/expression/expression_parser.rs

@@ -303,9 +303,7 @@ pub fn parse_computed_expr_to_string(
         )));
     }
     let mut ast = ast.clone();
-    let mut normalizer = IdentifierNormalizer {
-        ctx: &name_resolution_ctx,
-    };
+    let mut normalizer = IdentifierNormalizer::new(&name_resolution_ctx);
     ast.drive_mut(&mut normalizer);
     Ok(format!("{:#}", ast))
 }
@@ -492,9 +490,7 @@ pub fn analyze_cluster_keys(
         exprs.push(expr);
 
         let mut cluster_by = ast.clone();
-        let mut normalizer = IdentifierNormalizer {
-            ctx: &name_resolution_ctx,
-        };
+        let mut normalizer = IdentifierNormalizer::new(&name_resolution_ctx);
         cluster_by.drive_mut(&mut normalizer);
         cluster_keys.push(format!("{:#}", &cluster_by));
     }

src/query/sql/src/planner/semantic/name_resolution.rs

@@ -17,6 +17,7 @@ use std::sync::Arc;
 use databend_common_ast::ast::quote::ident_needs_quote;
 use databend_common_ast::ast::Identifier;
 use databend_common_ast::ast::IdentifierType;
+use databend_common_ast::ast::MapAccessor;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
@@ -110,15 +111,38 @@ pub fn compare_table_name(
 }
 
 #[derive(VisitorMut)]
-#[visitor(Identifier(enter))]
+#[visitor(Identifier(enter), MapAccessor)]
 pub struct IdentifierNormalizer<'a> {
-    pub ctx: &'a NameResolutionContext,
+    ctx: &'a NameResolutionContext,
+    in_map_accessor: bool,
 }
 
-impl IdentifierNormalizer<'_> {
+impl<'a> IdentifierNormalizer<'a> {
+    pub fn new(ctx: &'a NameResolutionContext) -> Self {
+        Self {
+            ctx,
+            in_map_accessor: false,
+        }
+    }
+
     fn enter_identifier(&mut self, ident: &mut Identifier) {
-        let normalized_ident = normalize_identifier(ident, self.ctx);
-        *ident = normalized_ident;
+        // Skip normalization if inside a MapAccessor,
+        // because MapAccessor is used to extract internal fields of nested types,
+        // altering the case may prevent the desired data from being retrieved.
+        if !self.in_map_accessor {
+            let normalized_ident = normalize_identifier(ident, self.ctx);
+            *ident = normalized_ident;
+        }
+    }
+
+    fn enter_map_accessor(&mut self, _accessor: &mut MapAccessor) {
+        // Set flag to true before processing the identifier inside the accessor
+        self.in_map_accessor = true;
+    }
+
+    fn exit_map_accessor(&mut self, _accessor: &mut MapAccessor) {
+        // Reset the flag after processing
+        self.in_map_accessor = false;
     }
 }
 

src/query/storages/common/index/src/inverted_index.rs

@@ -96,8 +96,10 @@ use tantivy::query::Query;
 use tantivy::query::QueryClone;
 use tantivy::query::TermQuery;
 use tantivy::schema::Field;
+use tantivy::version;
 use tantivy::Directory;
 use tantivy::Term;
+use tantivy::Version;
 use tantivy_common::BinarySerializable;
 use tantivy_common::HasLen;
 use tantivy_common::VInt;
@@ -108,28 +110,12 @@ use tantivy_fst::Streamer;
 
 // tantivy version is used to generate the footer data
 
-// Index major version.
-const INDEX_MAJOR_VERSION: u32 = 0;
-// Index minor version.
-const INDEX_MINOR_VERSION: u32 = 25;
-// Index patch version.
-const INDEX_PATCH_VERSION: u32 = 0;
-// Index format version.
-const INDEX_FORMAT_VERSION: u32 = 7;
 // The magic byte of the footer to identify corruption
 // or an old version of the footer.
 const FOOTER_MAGIC_NUMBER: u32 = 1337;
 
 type CrcHashU32 = u32;
 
-/// Structure version for the index.
-#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-pub struct Version {
-    major: u32,
-    minor: u32,
-    patch: u32,
-    index_format_version: u32,
-}
 /// A Footer is appended every part of data, like tantivy file.
 #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 struct Footer {
@@ -139,12 +125,7 @@ struct Footer {
 
 impl Footer {
     fn new(crc: CrcHashU32) -> Self {
-        let version = Version {
-            major: INDEX_MAJOR_VERSION,
-            minor: INDEX_MINOR_VERSION,
-            patch: INDEX_PATCH_VERSION,
-            index_format_version: INDEX_FORMAT_VERSION,
-        };
+        let version = version().clone();
         Footer { version, crc }
     }
 

src/query/storages/fuse/src/io/read/inverted_index/inverted_index_loader.rs

@@ -38,6 +38,7 @@ use databend_storages_common_io::MergeIOReader;
 use databend_storages_common_io::ReadSettings;
 use databend_storages_common_table_meta::meta::SingleColumnMeta;
 use databend_storages_common_table_meta::table::TableCompression;
+use log::info;
 use opendal::Operator;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReader;
 use parquet::arrow::parquet_to_arrow_field_levels;
@@ -115,6 +116,7 @@ pub(crate) async fn legacy_load_inverted_index_files<'a>(
     operator: &'a Operator,
 ) -> Result<Vec<Arc<InvertedIndexFile>>> {
     let start = Instant::now();
+    info!("load inverted index directory version 2");
 
     let mut files = vec![];
     let mut ranges = vec![];
@@ -300,6 +302,7 @@ pub(crate) async fn load_inverted_index_directory<'a>(
     // 7. meta.json file
     // 8. .managed.json file
     if version == 1 {
+        info!("load inverted index directory version 1");
         let mut columns = Vec::with_capacity(inverted_index_meta_map.len());
         for (col_name, col_meta) in inverted_index_meta_map {
             let col_range = col_meta.offset..(col_meta.offset + col_meta.len);

src/query/storages/fuse/src/io/write/inverted_index_writer.rs

@@ -42,6 +42,8 @@ use databend_storages_common_table_meta::meta::Location;
 use databend_storages_common_table_meta::table::TableCompression;
 use jsonb::from_raw_jsonb;
 use jsonb::RawJsonb;
+use log::debug;
+use log::info;
 use tantivy::index::SegmentComponent;
 use tantivy::indexer::UserOperation;
 use tantivy::schema::Field;
@@ -106,6 +108,10 @@ pub fn create_inverted_index_builders(table_meta: &TableMeta) -> Vec<InvertedInd
         }
         // ignore invalid index
         if index_fields.len() != index.column_ids.len() {
+            debug!(
+                "Ignoring invalid inverted index: {}, missing columns",
+                index.name
+            );
             continue;
         }
         let index_schema = DataSchema::new(index_fields);
@@ -145,6 +151,19 @@ impl InvertedIndexState {
         inverted_index_builder: &InvertedIndexBuilder,
     ) -> Result<Self> {
         let start = Instant::now();
+
+        let inverted_index_location =
+            TableMetaLocationGenerator::gen_inverted_index_location_from_block_location(
+                &block_location.0,
+                &inverted_index_builder.name,
+                &inverted_index_builder.version,
+            );
+
+        info!(
+            "Start build inverted index for location: {}",
+            inverted_index_location
+        );
+
         let mut writer = InvertedIndexWriter::try_create(
             Arc::new(inverted_index_builder.schema.clone()),
             &inverted_index_builder.options,
@@ -153,18 +172,16 @@ impl InvertedIndexState {
         let data = writer.finalize()?;
 
         // Perf.
+        let size = data.len();
+        let elapsed_ms = start.elapsed().as_millis() as u64;
         {
-            metrics_inc_block_inverted_index_generate_milliseconds(
-                start.elapsed().as_millis() as u64
-            );
+            metrics_inc_block_inverted_index_generate_milliseconds(elapsed_ms);
         }
+        info!(
+            "Finish build inverted index: location={}, size={} bytes in {} ms",
+            inverted_index_location, size, elapsed_ms
+        );
 
-        let inverted_index_location =
-            TableMetaLocationGenerator::gen_inverted_index_location_from_block_location(
-                &block_location.0,
-                &inverted_index_builder.name,
-                &inverted_index_builder.version,
-            );
         Self::try_create(data, inverted_index_location)
     }
 }
@@ -183,7 +200,6 @@ impl InvertedIndexWriter {
         let (index_schema, _) = create_index_schema(schema.clone(), index_options)?;
 
         let index_settings = IndexSettings {
-            // sort_by_field: None,
             ..Default::default()
         };
 
@@ -320,7 +336,6 @@ impl InvertedIndexWriter {
             &mut data,
             // Zstd has the best compression ratio
             TableCompression::Zstd,
-            // Some(metadata),
             None,
         )?;
 

tests/sqllogictests/suites/base/15_procedure/15_0001_execute_immediate.test

@@ -7,7 +7,7 @@ use test_procedure;
 query I
 EXECUTE IMMEDIATE $$
 BEGIN
-    RETURN;
+    RETURN NULL;
 END;
 $$;
 ----

tests/sqllogictests/suites/ee/01_ee_system/01_0001_computed_column.test

@@ -254,22 +254,22 @@ a1 a1-c1 c1
 a2 a2-c2 c2
 
 statement ok
-drop table if exists t_virtual;
+drop table if exists t_virtual_merge;
 
 statement ok
-create table t_virtual(a string null default 'a', b string null as (concat(a, '-', c)) virtual, c string null default 'c');
+create table t_virtual_merge(a string null default 'a', b string null as (concat(a, '-', c)) virtual, c string null default 'c');
 
 statement ok
-insert into t_virtual values ('a1', 'c1'), ('a2', 'c2');
+insert into t_virtual_merge values ('a1', 'c1'), ('a2', 'c2');
 
 statement ok
-insert into t_virtual (a) values ('a3'), ('a4');
+insert into t_virtual_merge (a) values ('a3'), ('a4');
 
 statement ok
-insert into t_virtual (c) values ('c5'), ('c6');
+insert into t_virtual_merge (c) values ('c5'), ('c6');
 
 query T
-select * from t_virtual order by a, c;
+select * from t_virtual_merge order by a, c;
 ----
 a a-c5 c5
 a a-c6 c6
@@ -279,10 +279,10 @@ a3 a3-c c
 a4 a4-c c
 
 statement ok
-merge into t_virtual target using t_virtual source on target.a = source.a when matched and source.a = 'a4' then update set target.a = 'aa', target.c = 'cc'; 
+merge into t_virtual_merge target using t_virtual_merge source on target.a = source.a when matched and source.a = 'a4' then update set target.a = 'aa', target.c = 'cc'; 
 
 query T
-select * from t_virtual order by a, c;
+select * from t_virtual_merge order by a, c;
 ----
 a a-c5 c5
 a a-c6 c6
@@ -291,5 +291,21 @@ a2 a2-c2 c2
 a3 a3-c c
 aa aa-cc cc
 
+statement ok
+drop table if exists t_variant;
+
+statement ok
+create table t_variant(id int, data variant, name string as (data:videoInfo:Name::string) stored, duration int as(DATA:videoInfo:Duration::int) stored);
+
+statement ok
+insert into t_variant values(1, '{"videoInfo":{"Name":"H264","Duration":10}}'),(2, '{"videoInfo":{"Name":"H265","Duration":20}}');
+
+query ITTI
+select * from t_variant;
+----
+1 {"videoInfo":{"Duration":10,"Name":"H264"}} H264 10
+2 {"videoInfo":{"Duration":20,"Name":"H265"}} H265 20
+
 statement ok
 DROP DATABASE test_computed_column
+