feat(query): support read nested struct in iceberg tables

sundy-li · sundy-li · commit 1b41889aae7e · 2025-12-29T20:12:17.000+08:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -312,13 +312,13 @@ hyper-util = { version = "0.1.9", features = ["client", "client-legacy", "tokio"
 lru = "0.12"
 
 ## in branch dev
-iceberg = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403", features = [
+iceberg = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "a5e780ae", features = [
     "storage-all",
 ] }
-iceberg-catalog-glue = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
-iceberg-catalog-hms = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
-iceberg-catalog-rest = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
-iceberg-catalog-s3tables = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
+iceberg-catalog-glue = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "a5e780ae" }
+iceberg-catalog-hms = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "a5e780ae" }
+iceberg-catalog-rest = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "a5e780ae" }
+iceberg-catalog-s3tables = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "a5e780ae" }
 
 # Explicitly specify compatible AWS SDK versions
 aws-config = "1.5.18"
diff --git a/pyproject.toml b/pyproject.toml
diff --git a/src/query/service/src/interpreters/interpreter_copy_into_location.rs b/src/query/service/src/interpreters/interpreter_copy_into_location.rs
@@ -16,7 +16,6 @@ use std::sync::Arc;
 
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_exception::Result;
-use databend_common_expression::infer_table_schema;
 use databend_common_meta_app::schema::UpdateStreamMetaReq;
 use databend_common_pipeline::core::ExecutionInfo;
 use databend_storages_common_stage::CopyIntoLocationInfo;
@@ -88,8 +87,7 @@ impl CopyIntoLocationInterpreter {
         let (query_interpreter, update_stream_meta_req) = self.build_query(query).await?;
         let query_physical_plan = query_interpreter.build_physical_plan().await?;
         let query_result_schema = query_interpreter.get_result_schema();
-        let table_schema = infer_table_schema(&query_result_schema)?;
-
+        let table_schema = query_interpreter.get_result_table_schema()?;
         let mut physical_plan = PhysicalPlan::new(CopyIntoLocation {
             input: query_physical_plan,
             project_columns: query_interpreter.get_result_columns(),
diff --git a/src/query/service/src/interpreters/interpreter_select.rs b/src/query/service/src/interpreters/interpreter_select.rs
@@ -24,7 +24,10 @@ use databend_common_exception::Result;
 use databend_common_expression::DataField;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::DataSchemaRefExt;
+use databend_common_expression::TableField;
 use databend_common_expression::TableSchemaRef;
+use databend_common_expression::TableSchemaRefExt;
+use databend_common_expression::infer_schema_type;
 use databend_common_expression::infer_table_schema;
 use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
 use databend_common_meta_store::MetaStore;
@@ -36,6 +39,7 @@ use databend_common_pipeline::core::PipeItem;
 use databend_common_pipeline::core::Pipeline;
 use databend_common_pipeline_transforms::processors::TransformDummy;
 use databend_common_sql::ColumnBinding;
+use databend_common_sql::ColumnEntry;
 use databend_common_sql::MetadataRef;
 use databend_common_sql::executor::physical_plans::FragmentKind;
 use databend_common_sql::parse_result_scan_args;
@@ -113,6 +117,30 @@ impl SelectInterpreter {
         DataSchemaRefExt::create(fields)
     }
 
+    pub fn get_result_table_schema(&self) -> Result<TableSchemaRef> {
+        let metadata = self.metadata.read();
+        let mut fields = Vec::with_capacity(self.bind_context.columns.len());
+        for column_binding in &self.bind_context.columns {
+            let table_data_type = if column_binding.index < metadata.columns().len() {
+                match metadata.column(column_binding.index) {
+                    ColumnEntry::BaseTableColumn(base) => base.data_type.clone(),
+                    ColumnEntry::VirtualColumn(virtual_column) => virtual_column.data_type.clone(),
+                    ColumnEntry::DerivedColumn(derived) => infer_schema_type(&derived.data_type)?,
+                    ColumnEntry::InternalColumn(internal) => {
+                        infer_schema_type(&internal.internal_column.data_type())?
+                    }
+                }
+            } else {
+                infer_schema_type(column_binding.data_type.as_ref())?
+            };
+            fields.push(TableField::new(
+                &column_binding.column_name,
+                table_data_type,
+            ));
+        }
+        Ok(TableSchemaRefExt::create(fields))
+    }
+
     #[fastrace::trace(name = "SelectInterpreter::build_physical_plan")]
     #[async_backtrace::framed]
     pub async fn build_physical_plan(&self) -> Result<PhysicalPlan> {
diff --git a/src/query/storages/iceberg/src/table.rs b/src/query/storages/iceberg/src/table.rs
@@ -27,6 +27,7 @@ use databend_common_catalog::plan::ParquetReadOptions;
 use databend_common_catalog::plan::PartStatistics;
 use databend_common_catalog::plan::Partitions;
 use databend_common_catalog::plan::PartitionsShuffleKind;
+use databend_common_catalog::plan::Projection;
 use databend_common_catalog::plan::PushDownInfo;
 use databend_common_catalog::table::ColumnStatisticsProvider;
 use databend_common_catalog::table::DistributionLevel;
@@ -40,7 +41,9 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::ColumnId;
 use databend_common_expression::DataSchema;
+use databend_common_expression::FieldIndex;
 use databend_common_expression::TableField;
+use databend_common_expression::TableDataType;
 use databend_common_expression::TableSchema;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::TableIdent;
@@ -407,13 +410,9 @@ impl IcebergTable {
 
         if let Some(push_downs) = &push_downs {
             if let Some(projection) = &push_downs.projection {
-                scan = scan.select(
-                    projection
-                        .project_schema(&self.schema())
-                        .fields
-                        .iter()
-                        .map(|v| v.name.clone()),
-                );
+                let select_fields =
+                    Self::projection_to_iceberg_select_fields(projection, &self.schema())?;
+                scan = scan.select(select_fields);
             }
             if let Some(filter) = &push_downs.filters {
                 let (_, predicate) = PredicateBuilder::build(&filter.filter);
@@ -450,6 +449,84 @@ impl IcebergTable {
         ))
     }
 
+    fn projection_to_iceberg_select_fields(
+        projection: &Projection,
+        schema: &TableSchema,
+    ) -> Result<Vec<String>> {
+        match projection {
+            Projection::Columns(_) => Ok(projection
+                .project_schema(schema)
+                .fields
+                .iter()
+                .map(|v| v.name.clone())
+                .collect()),
+            Projection::InnerColumns(path_indices) => {
+                let fields = schema.fields();
+                let mut names = Vec::with_capacity(path_indices.len());
+                for path in path_indices.values() {
+                    names.push(Self::inner_column_path_to_name(fields, path)?);
+                }
+                Ok(names)
+            }
+        }
+    }
+
+    fn inner_column_path_to_name(
+        fields: &[TableField],
+        path: &[FieldIndex],
+    ) -> Result<String> {
+        if path.is_empty() {
+            return Err(ErrorCode::BadArguments(
+                "Inner column path should not be empty".to_string(),
+            ));
+        }
+
+        let field = fields.get(path[0]).ok_or_else(|| {
+            ErrorCode::BadArguments(format!(
+                "Inner column path {:?} is out of range",
+                path
+            ))
+        })?;
+        let mut name_parts = Vec::with_capacity(path.len());
+        name_parts.push(field.name().clone());
+
+        let mut current_type = field.data_type().remove_nullable();
+        for index in path.iter().skip(1) {
+            match &current_type {
+                TableDataType::Tuple {
+                    fields_name,
+                    fields_type,
+                } => {
+                    let inner_name = fields_name.get(*index).ok_or_else(|| {
+                        ErrorCode::BadArguments(format!(
+                            "Inner column path {:?} is out of range for {}",
+                            path,
+                            name_parts.join(".")
+                        ))
+                    })?;
+                    name_parts.push(inner_name.clone());
+                    let inner_type = fields_type.get(*index).ok_or_else(|| {
+                        ErrorCode::BadArguments(format!(
+                            "Inner column path {:?} is out of range for {}",
+                            path,
+                            name_parts.join(".")
+                        ))
+                    })?;
+                    current_type = inner_type.remove_nullable();
+                }
+                _ => {
+                    return Err(ErrorCode::BadArguments(format!(
+                        "Inner column path {:?} is invalid for non-tuple field {}",
+                        path,
+                        name_parts.join(".")
+                    )));
+                }
+            }
+        }
+
+        Ok(name_parts.join("."))
+    }
+
     fn convert_orc_schema(schema: &Schema) -> Schema {
         fn visit_field(field: &arrow_schema::FieldRef) -> arrow_schema::FieldRef {
             Arc::new(
diff --git a/src/query/storages/parquet/src/parquet_reader/reader/builder.rs b/src/query/storages/parquet/src/parquet_reader/reader/builder.rs
@@ -240,9 +240,8 @@ impl<'a> ParquetReaderBuilder<'a> {
             .map(|(proj, _, _, paths)| (proj.clone(), paths.clone()))
             .unwrap();
 
-        let (_, _, output_schema, _) = self.built_output.as_ref().unwrap();
-        let transformer = source_type
-            .need_transformer()
+        let (_, _, output_schema, output_field_paths) = self.built_output.as_ref().unwrap();
+        let transformer = (source_type.need_transformer() && output_field_paths.is_none())
             .then(|| RecordBatchTransformer::build(output_schema.clone()));
         Ok(ParquetWholeFileReader {
             op_registry: self.op_registry.clone(),
@@ -277,8 +276,10 @@ impl<'a> ParquetReaderBuilder<'a> {
         let transformer = source_type
             .need_transformer()
             .then(|| {
-                self.built_output.as_ref().map(|(_, _, output_schema, _)| {
-                    RecordBatchTransformer::build(output_schema.clone())
+                self.built_output.as_ref().and_then(|(_, _, output_schema, output_field_paths)| {
+                    output_field_paths
+                        .is_none()
+                        .then(|| RecordBatchTransformer::build(output_schema.clone()))
                 })
             })
             .flatten();
diff --git a/src/query/storages/parquet/src/source.rs b/src/query/storages/parquet/src/source.rs
@@ -66,7 +66,7 @@ enum State {
     ReadFiles(Vec<(Bytes, String)>),
 }
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
 pub enum ParquetSourceType {
     StageTable,
     ResultCache,
diff --git a/tests/sqllogictests/scripts/prepare_iceberg_test_data.py b/tests/sqllogictests/scripts/prepare_iceberg_test_data.py
@@ -1,4 +1,5 @@
 from pyspark.sql import SparkSession
+from pyspark.sql.types import StructType, StructField, StringType, IntegerType
 
 spark = (
     SparkSession.builder.appName("CSV to Iceberg REST Catalog")
@@ -57,4 +58,54 @@
     f"""INSERT INTO iceberg.test.t1_orc VALUES (0, 0, 'a'), (1, 1, 'b'), (2, 2, 'c'), (3, 3, 'd'), (4, null, null);"""
 )
 
+# create nested table
+spark.sql("DROP TABLE IF EXISTS iceberg.test.t_nested")
+data = [
+    (1, ("Alice", 30), (("A1", 1), 10)),
+    (2, ("Bob", 25), (("B1", 2), 20)),
+    (3, ("Charlie", 35), (("C1", 3), 30)),
+    (4, None, None),
+]
+
+# Create DataFrame and write
+schema = StructType(
+    [
+        StructField("id", IntegerType(), True),
+        StructField(
+            "item",
+            StructType(
+                [
+                    StructField("name", StringType(), True),
+                    StructField("age", IntegerType(), True),
+                ]
+            ),
+            True,
+        ),
+        StructField(
+            "item_2",
+            StructType(
+                [
+                    StructField(
+                        "item",
+                        StructType(
+                            [
+                                StructField("name", StringType(), True),
+                                StructField("level", IntegerType(), True),
+                            ]
+                        ),
+                        True,
+                    ),
+                    StructField("level", IntegerType(), True),
+                ]
+            ),
+            True,
+        ),
+    ]
+)
+
+df = spark.createDataFrame(data, schema)
+df.writeTo("iceberg.test.t_nested").using("iceberg").createOrReplace()
+
+print("Table iceberg.test.t_nested created with sample data")
+
 spark.stop()
diff --git a/tests/sqllogictests/suites/base/03_common/03_0028_copy_into_stage.test b/tests/sqllogictests/suites/base/03_common/03_0028_copy_into_stage.test
@@ -56,6 +56,21 @@ select sum(number) from @hello;
 ----
 45
 
+
+statement ok
+remove @hello
+
+statement ok
+create or replace table abc(number int, item Tuple(name String, age int)) as select number, ('aa', number) from numbers(10);
+
+statement ok
+COPY INTO @hello from (select * from abc) FILE_FORMAT = (type = parquet)
+
+query TI
+select max(item['name']), sum(item['age']) from @hello;
+----
+aa 45
+
 statement ok
 CREATE TABLE world(c1 int , c2 int);
 
@@ -66,7 +81,10 @@ statement ok
 DROP STAGE IF EXISTS hello
 
 statement ok
-drop table world
+drop table if EXISTS abc
+
+statement ok
+drop table if EXISTS world
 
 statement ok
 DROP DATABASE db1
diff --git a/tests/sqllogictests/suites/tpch_iceberg/base.test b/tests/sqllogictests/suites/tpch_iceberg/base.test
diff --git a/uv.lock b/uv.lock

Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,7 @@ enum State {`
`66`	`66`	`ReadFiles(Vec<(Bytes, String)>),`
`67`	`67`	`}`
`68`	`68`
`69`		`-#[derive(Debug, Clone, Copy)]`
	`69`	`+#[derive(Debug, Clone, Copy, Eq, PartialEq)]`
`70`	`70`	`pub enum ParquetSourceType {`
`71`	`71`	`StageTable,`
`72`	`72`	`ResultCache,`