andygrove opened a new issue, #3557: URL: https://github.com/apache/iceberg-rust/issues/3557
_Drafted with LLM assistance (Claude Code); verified with a Rust reproducer against iceberg-rust `b154603` and `main` at `340b6fa`, and end to end through Comet against Spark 4.1 with Iceberg Java 1.11._ ## What happens A projected map column whose value struct has no leaf in the Parquet file fails the read with `DataInvalid => Map field must have exactly 2 fields`. Spark with Iceberg Java reads the same table and returns a result. The case that reaches it is a table imported with `snapshot`, `migrate` or `add_files`: the files have no field ids, the table has a name mapping, and a nested field in the file has a different physical name than the mapping lists. The mapping gives the map's key an id, but the value struct's child gets none. [`get_arrow_projection_mask_with_field_ids`](https://github.com/apache/iceberg-rust/blob/b154603a710d730d2f43f8af4db257cc137237ef/crates/iceberg/src/arrow/reader/projection.rs#L172-L200) first pre-projects the file's Arrow schema to the requested leaf ids with `Fields::filter_leaves`, then converts the result with `arrow_schema_to_schema`: - The requested leaves for the map are its key and the value struct's children. - The key is in the file and has an id from the mapping, so it is kept. - The value struct's child has no id in the file, so it is dropped. - The map's entries struct is left with one field, and the conversion rejects it at [`arrow/schema.rs`](https://github.com/apache/iceberg-rust/blob/b154603a710d730d2f43f8af4db257cc137237ef/crates/iceberg/src/arrow/schema.rs#L213-L215). A comment in that function says a struct whose requested leaves are all missing "reads as NULL" ([iceberg-rust#3378](https://github.com/apache/iceberg-rust/issues/3378)). That holds for a struct. A map whose value leaves are all missing never gets that far, because the pre-projection fails first. I did not check the same shape in a file that has field ids, where the value struct's children were added after the file was written. By reading the code it takes the same path, so it may fail the same way. The failure is an error rather than a wrong answer. Comet's native Iceberg scan is built on this reader, so a query that Spark answers fails in Comet. Comet falls back to Spark for nested columns of name-mapped tables because of it: https://github.com/apache/datafusion-comet/issues/6790, https://github.com/apache/datafusion-comet/pull/7028. ## Reproducer Added to the tests in `crates/iceberg/src/arrow/reader/projection.rs`. The `x` iteration passes. The `old_x` iteration fails with the error above: ```rust #[tokio::test] async fn map_value_with_all_leaves_missing() { use arrow_array::{Int32Array, MapArray, StringArray, StructArray}; use arrow_buffer::{OffsetBuffer, ScalarBuffer}; use arrow_schema::Fields; // 1: id int, 2: m map<3: string, 4: struct<5: x int>>, mapped by name let schema = Arc::new( Schema::builder() .with_schema_id(1) .with_fields(vec![ NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Int)).into(), NestedField::optional( 2, "m", Type::Map(crate::spec::MapType::new( NestedField::map_key_element(3, Type::Primitive(PrimitiveType::String)) .into(), NestedField::map_value_element( 4, Type::Struct(StructType::new(vec![ NestedField::optional(5, "x", Type::Primitive(PrimitiveType::Int)) .into(), ])), false, ) .into(), )), ) .into(), ]) .build() .unwrap(), ); let name_mapping = Arc::new(NameMapping::new(vec![ MappedField::new(Some(1), vec!["id".to_string()], vec![]), MappedField::new(Some(2), vec!["m".to_string()], vec![ MappedField::new(Some(3), vec!["key".to_string()], vec![]), MappedField::new(Some(4), vec!["value".to_string()], vec![MappedField::new( Some(5), vec!["x".to_string()], vec![], )]), ]), ])); // The value struct's child is named `x`, as in the mapping, and then `old_x`. No field ids. for physical in ["x", "old_x"] { let value = StructArray::from(vec![( Arc::new(Field::new(physical, DataType::Int32, true)), Arc::new(Int32Array::from(vec![10, 20, 30])) as ArrayRef, )]); let entry_fields = Fields::from(vec![ Field::new("key", DataType::Utf8, false), Field::new("value", value.data_type().clone(), true), ]); let entries = StructArray::new( entry_fields.clone(), vec![ Arc::new(StringArray::from(vec!["a", "b", "c"])) as ArrayRef, Arc::new(value) as ArrayRef, ], None, ); let map = MapArray::new( Arc::new(Field::new("key_value", DataType::Struct(entry_fields), false)), OffsetBuffer::new(ScalarBuffer::from(vec![0i32, 1, 2, 3])), entries, None, false, ); let to_write = RecordBatch::try_from_iter_with_nullable(vec![ ("id", Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef, true), ("m", Arc::new(map) as ArrayRef, true), ]) .unwrap(); let tmp_dir = TempDir::new().unwrap(); let file_path = format!("{}/1.parquet", tmp_dir.path().to_str().unwrap()); let mut writer = ArrowWriter::try_new(File::create(&file_path).unwrap(), to_write.schema(), None) .unwrap(); writer.write(&to_write).unwrap(); writer.close().unwrap(); let tasks = Box::pin(futures::stream::iter([Ok(FileScanTask::builder() .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len()) .with_start(0) .with_length(0) .with_data_file_path(file_path.clone()) .with_data_file_format(DataFileFormat::Parquet) .with_schema(schema.clone()) .with_project_field_ids(vec![2]) .with_case_sensitive(false) .with_name_mapping(Some(name_mapping.clone())) .build() .unwrap())])) as FileScanTaskStream; let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build(); let result = reader .read(tasks) .unwrap() .stream() .try_collect::<Vec<RecordBatch>>() .await; assert!(result.is_ok(), "physical={physical}: {:?}", result.err()); } } ``` To see it through Comet (Spark 4.1, Iceberg 1.11, this iceberg-rust revision): write Parquet with no field ids and the schema `id INT, m MAP<STRING, STRUCT<nested: STRUCT<old_x: INT>>>`, register a table with `CREATE TABLE ... (id INT, m MAP<STRING, STRUCT<nested: STRUCT<x: INT>>>) USING parquet LOCATION ...`, import it with `SparkActions.snapshotTable`, and run `SELECT m['k'].nested FROM t` with nested schema pruning on. Spark returns a result. The native scan fails with the error above. Reads of the whole map, of `id` with the map, and of the other projections I tried match Spark. ## Expected behavior The read succeeds. The missing value leaves come back as NULLs the way a struct's missing leaves do ([iceberg-rust#3378](https://github.com/apache/iceberg-rust/issues/3378)). Whether that means NULL values or structs of NULLs should follow what Iceberg Java returns for the same file. I haven't compared the two. A fix probably needs more than the pre-projection. If the map's entries struct is kept with its key only, `ProjectionMask::leaves` selects just the key column, and the batch that reaches the transformer has a one-field entries struct to reconcile with the table's two-field one. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
