QuakeWang commented on code in PR #54:
URL: https://github.com/apache/paimon-mosaic/pull/54#discussion_r3371189205
##########
ffi/src/lib.rs:
##########
Review Comment:
This still exports the internal expanded schema. For files containing a
`STRUCT`, `schema.columns` includes physical columns such as `info.__null__`,
`info.name`, and `info.age`, but the reader’s public schema should match the
logical Arrow schema that the user wrote.
A small `id`, `info: struct<name, age>` case currently gives:
```rust
reader.schema.names = ['id', 'info.__null__', 'info.name', 'info.age']
read_row_group(0).schema.names = ['id', 'info']
```
So the Python/C++ visible `reader.schema` does not match the schema of the
batches returned by reads. I think this should use the original logical columns
when `original_columns` is present, instead of reconstructing the public schema
from expanded `schema.columns`.
##########
core/src/schema.rs:
##########
@@ -62,10 +96,328 @@ impl MosaicSchema {
.map(|f| (f.name().clone(), f.data_type().clone(),
f.is_nullable()))
.collect();
Self::validate(&columns)?;
- Ok(Self::new(columns, num_buckets))
+
+ // Expand STRUCT columns into independent flat columns with DFS column
IDs
+ let mut expanded: Vec<(String, u32, DataType, bool)> = Vec::new();
+ let mut struct_mappings: Vec<StructMapping> = Vec::new();
+ let mut null_col_ids: Vec<u32> = Vec::new();
+ let mut id_counter: u32 = 0;
+
+ for (orig_idx, (name, dt, nullable)) in columns.iter().enumerate() {
+ if let DataType::Struct(fields) = dt {
+ if !types::is_timestamp_nanos_struct(fields) {
+ let struct_id = id_counter;
+ id_counter += 1;
+ let start = expanded.len();
+ let has_null_col = *nullable;
+ if has_null_col {
+ let null_name = format!("{}.__null__", name);
+ expanded.push((null_name, struct_id,
DataType::Boolean, false));
+ null_col_ids.push(struct_id);
+ }
+ let field_start = expanded.len();
+ Self::expand_struct_fields_recursive(
+ name,
+ fields,
+ &mut id_counter,
+ &mut expanded,
+ &mut null_col_ids,
+ );
+ let end = expanded.len();
+
+ let expanded_indices: Vec<usize> = (start..end).collect();
+ let field_ids: Vec<u32> = expanded[field_start..end]
+ .iter()
+ .filter(|(_, id, _, _)| !null_col_ids.contains(id))
+ .map(|(_, id, _, _)| *id)
+ .collect();
+
+ struct_mappings.push(StructMapping {
+ struct_id,
+ original_col_index: orig_idx,
+ original_field: schema.field(orig_idx).clone(),
+ expanded_col_indices: expanded_indices,
+ null_col_sorted_idx: None, // filled after sorting
+ field_ids,
+ });
+ continue;
+ }
+ }
+ let col_id = id_counter;
+ id_counter += 1;
+ expanded.push((name.clone(), col_id, dt.clone(), *nullable));
+ }
+
+ let mut schema = Self::new_with_ids(expanded, num_buckets);
+ schema.struct_mappings = struct_mappings;
+ schema.null_col_ids = null_col_ids;
+
+ // Fix struct_mappings expanded_col_indices and null_col_sorted_idx to
sorted positions
+ for mapping in &mut schema.struct_mappings {
+ mapping.expanded_col_indices = mapping
+ .expanded_col_indices
+ .iter()
+ .map(|&input_idx| schema.original_order[input_idx])
+ .collect();
+ if mapping.original_field.is_nullable() {
+ mapping.null_col_sorted_idx =
schema.columns.iter().position(|c| {
+ c.column_id == mapping.struct_id && c.data_type ==
DataType::Boolean
+ });
+ }
+ }
+
+ if !schema.struct_mappings.is_empty() {
+ schema.original_columns = Some(columns);
+ }
+
+ Ok(schema)
+ }
+
+ fn expand_struct_fields_recursive(
+ prefix: &str,
+ fields: &arrow_schema::Fields,
+ id_counter: &mut u32,
+ out: &mut Vec<(String, u32, DataType, bool)>,
+ null_col_ids: &mut Vec<u32>,
+ ) {
+ for field in fields.iter() {
+ let field_id = *id_counter;
+ *id_counter += 1;
+ let full_name = format!("{}.{}", prefix, field.name());
+ let dt = field.data_type();
+ if let DataType::Struct(inner_fields) = dt {
+ if !types::is_timestamp_nanos_struct(inner_fields) {
+ if field.is_nullable() {
+ let null_name = format!("{}.__null__", full_name);
+ out.push((null_name, field_id, DataType::Boolean,
false));
+ null_col_ids.push(field_id);
+ }
+ Self::expand_struct_fields_recursive(
+ &full_name,
+ inner_fields,
+ id_counter,
+ out,
+ null_col_ids,
+ );
+ continue;
+ }
+ }
+ out.push((full_name, field_id, dt.clone(), true));
+ }
+ }
+
+ /// Resolve a projected Arrow Schema to expanded column indices.
+ /// Returns `(output_order, read_indices)`:
+ /// - `output_order`: user-requested indices in request order (for output
column ordering)
+ /// - `read_indices`: output_order + auto-added `__null__` ancestors (for
physical reads)
+ ///
+ /// Each field in `projected` is matched by name against the original
schema.
+ /// STRUCT fields in `projected` can be partial (containing only a subset
of the
+ /// original STRUCT's fields) for field-level projection.
+ pub fn resolve_schema_projection(
+ &self,
+ projected: &Schema,
+ ) -> io::Result<(Vec<usize>, Vec<usize>)> {
+ let mut output = Vec::new();
+
+ for proj_field in projected.fields() {
+ if let Some(mapping) = self
+ .struct_mappings
+ .iter()
+ .find(|m| m.original_field.name() == proj_field.name())
+ {
+ if let DataType::Struct(proj_fields) = proj_field.data_type() {
+ // Partial STRUCT projection: collect IDs for requested
sub-fields
+ let mut local_ids = Vec::new();
+ Self::collect_projected_ids(
+ mapping.original_field.data_type(),
+ &DataType::Struct(proj_fields.clone()),
+ &mut local_ids,
+ );
+ let base = mapping.struct_id + 1;
+ for lid in local_ids {
+ let abs_id = base + lid;
+ if let Some(idx) = self.columns.iter().position(|c|
c.column_id == abs_id) {
+ output.push(idx);
+ }
+ }
+ } else {
+ // Non-STRUCT field matching a STRUCT column — expand all
+ output.extend_from_slice(&mapping.expanded_col_indices);
+ }
+ } else {
+ // Non-STRUCT column: find by name in original schema
+ let orig = self.original_columns.as_deref();
+ let found =
Review Comment:
This checks that the projected field exists in `original_columns`, but then
still resolves it by first matching `ColumnMeta.name` in the expanded columns.
That name is not unique in this format: a top-level column `"a.b"` and the leaf
`b` inside `STRUCT a` can both produce the physical name `"a.b"`.
A failing case is:
```rust
Schema::new(vec![
Field::new("a", DataType::Struct(... field "b" ...), true),
Field::new("a.b", DataType::Int32, false),
])
```
With projection `Schema::new(vec![Field::new("a.b", DataType::Int32,
false)])`, the current lookup can pick the `STRUCT a` leaf and the output field
becomes `"a"` instead of the top-level `"a.b"`. This should resolve top-level
non-STRUCT fields through their original schema position / column id, not by
first matching the physical column name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
