hcrosse opened a new issue, #3616: URL: https://github.com/apache/iceberg-rust/issues/3616
### Apache Iceberg Rust version main (`896f9d6c`), v0.10.0, v0.9.1 ### Describe the bug Since v0.9.1, iceberg-rust returns timestamps 1000x too large, with no error, when a Parquet file stores nanoseconds under an Iceberg `timestamp` field. Under `timestamptz`, the same file fails to read. This is a regression in v0.9.1. I reproduced it on main at [`896f9d6c`](https://github.com/apache/iceberg-rust/commit/896f9d6c50ff52fa9e43aec170d3f14bfb40b512), v0.10.0 and v0.9.1, and v0.9.0 reads both types correctly. The coercion came from https://github.com/apache/iceberg-rust/pull/2301, which was backported into v0.9.1 as [`e1fb7edf0`](https://github.com/apache/iceberg-rust/commit/e1fb7edf008285c1fccefe1eb4273f1dc724b22a). I did not run v0.10.1, but its `int96.rs` is identical to v0.10.0 and main, so it is very likely affected. The file uses physical `INT64` with the `TIMESTAMP(NANOS)` logical type, not INT96. One second after the epoch reads as 16 minutes and 40 seconds. The `timestamptz` read fails with `Incompatible supplied Arrow schema`. [`coerce_int96_timestamps`](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/arrow/int96.rs#L77-L104) chooses a microsecond Arrow hint for nanosecond Arrow fields declared as Iceberg `timestamp` or `timestamptz`, without checking the Parquet physical type. For the INT64 column without a timezone, arrow-rs accepts the hint and labels the raw nanosecond integers as microseconds. For the UTC variant, it rejects the hint. Applying the hint only to physical INT96 columns would leave INT64 columns to the reader's existing nanosecond-to-microsecond cast. I have not tested that change. Iceberg Java's generic reader checks for physical INT96 before it uses a timestamp's unit: https://github.com/apache/iceberg/blob/5a2ccf5ba830838ecf182bccb516d5ee48f8b8ed/parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java#L97-L124. #### How such files arise With Iceberg 1.12.0 and Spark 3.5.9 at default settings, I registered the nanosecond file under a `timestamp` field using Spark's `add_files` procedure without an error. The [procedure documentation](https://github.com/apache/iceberg/blob/5a2ccf5ba830838ecf182bccb516d5ee48f8b8ed/docs/docs/spark-procedures.md#L697-L720) says that it does not validate file schemas. iceberg-rust main read those tables with the same wrong values and the same `timestamptz` error. PyIceberg 0.12.0 and main `e3564a54` returned the correct values. Spark returned values 1000x too large, with vectorization on and off. Related: https://github.com/apache/iceberg-rust/issues/3611 concerns a different INT96 bug in the same code. ### To Reproduce Save this as `src/main.rs` in a Rust 2024 crate and run `cargo run -q`. It writes synthetic nanosecond files, registers them with `fast_append` in a `MemoryCatalog`, and scans the tables through public APIs. Use these dependencies in `Cargo.toml`: ```toml [dependencies] iceberg = { git = "https://github.com/apache/iceberg-rust";, rev = "896f9d6c50ff52fa9e43aec170d3f14bfb40b512" } arrow-array = "=59.3.0" arrow-schema = "=59.3.0" parquet = "=59.3.0" futures = "0.3" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } ``` ```rust use std::collections::HashMap; use std::error::Error; use std::sync::Arc; use arrow_array::cast::AsArray; use arrow_array::types::TimestampMicrosecondType; use arrow_array::{RecordBatch, TimestampNanosecondArray}; use arrow_schema::{DataType, Field, Schema as ArrowSchema, TimeUnit}; use futures::TryStreamExt; use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; use iceberg::spec::{ DataContentType, DataFileBuilder, DataFileFormat, NestedField, PrimitiveType, Schema, Type, }; use iceberg::transaction::{ApplyTransactionAction, Transaction}; use iceberg::{Catalog, CatalogBuilder, MemoryCatalog, NamespaceIdent, TableCreation}; use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}; // 1 s, 2 s and 2024-01-01T00:00:00.123456, in nanoseconds. const NANOS: [i64; 3] = [1_000_000_000, 2_000_000_000, 1_704_067_200_123_456_000]; #[tokio::main] async fn main() -> Result<(), Box<dyn Error>> { let props = HashMap::from([( MEMORY_CATALOG_WAREHOUSE.to_string(), "memory:///warehouse".to_string(), )]); let catalog = MemoryCatalogBuilder::default() .load("memory", props) .await?; let namespace = NamespaceIdent::new("ns".to_string()); catalog.create_namespace(&namespace, HashMap::new()).await?; println!("expected (us): {:?}", NANOS.map(|ns| ns / 1000)); for (name, iceberg_type, tz) in [ ("timestamp", PrimitiveType::Timestamp, None), ( "timestamptz", PrimitiveType::Timestamptz, Some("UTC".into()), ), ] { match register_and_scan(&catalog, &namespace, name, iceberg_type, tz).await { Ok(micros) => println!("{name} (us): {micros:?}"), Err(e) => println!("{name} error: {e}"), } } Ok(()) } /// Appends a Parquet file holding an INT64 TIMESTAMP(NANOS) column to a table whose /// field is microsecond `iceberg_type`, then scans the table. async fn register_and_scan( catalog: &MemoryCatalog, namespace: &NamespaceIdent, name: &str, iceberg_type: PrimitiveType, tz: Option<Arc<str>>, ) -> Result<Vec<i64>, Box<dyn Error>> { let field = NestedField::optional(1, "ts", Type::Primitive(iceberg_type)); let schema = Schema::builder().with_fields(vec![field.into()]).build()?; let creation = TableCreation::builder() .name(name.to_string()) .schema(schema) .build(); let table = catalog.create_table(namespace, creation).await?; let field_id = HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]); let field = Field::new( "ts", DataType::Timestamp(TimeUnit::Nanosecond, tz.clone()), true, ); let schema = Arc::new(ArrowSchema::new(vec![field.with_metadata(field_id)])); let column = TimestampNanosecondArray::from(NANOS.to_vec()).with_timezone_opt(tz); let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(column)])?; let mut parquet = Vec::new(); let mut writer = ArrowWriter::try_new(&mut parquet, schema, None)?; writer.write(&batch)?; writer.close()?; let path = format!("{}/data/nanos.parquet", table.metadata().location()); let size = parquet.len() as u64; table .file_io() .new_output(&path)? .write(parquet.into()) .await?; let data_file = DataFileBuilder::default() .content(DataContentType::Data) .file_path(path) .file_format(DataFileFormat::Parquet) .file_size_in_bytes(size) .record_count(NANOS.len() as u64) .build()?; let tx = Transaction::new(&table); let tx = tx.fast_append().add_data_files(vec![data_file]).apply(tx)?; let table = tx.commit(catalog).await?; let batches: Vec<RecordBatch> = table .scan() .build()? .to_arrow() .await? .try_collect() .await?; let column = |b: &RecordBatch| { b.column(0) .as_primitive::<TimestampMicrosecondType>() .values() .to_vec() }; Ok(batches.iter().flat_map(column).collect()) } ``` Output on main `896f9d6c`: ```text expected (us): [1000000, 2000000, 1704067200123456] timestamp (us): [1000000000, 2000000000, 1704067200123456000] timestamptz error: Unexpected => Failed to create ArrowReaderMetadata with the configured reader options (missing_field_ids: false, install_row_number: false, schema: Field { "ts": nullable Timestamp(µs, "UTC"), metadata: {"PARQUET:field_id": "1"} }), source: Arrow: Incompatible supplied Arrow schema: data type mismatch for field ts: requested Timestamp(µs, "UTC") but found Timestamp(ns, "UTC") ``` For releases, I replaced the `iceberg` line with `tag = "v0.10.0"`, `"v0.9.1"` or `"v0.9.0"`. I pinned the three arrow-rs crates to 58.1.0 for v0.10.0 and 57.3.0 for v0.9.x. v0.10.0 and v0.9.1 print the same values. Their `timestamptz` error reads `Failed to create ArrowReaderMetadata with INT96-coerced schema`. v0.9.0 prints: ```text expected (us): [1000000, 2000000, 1704067200123456] timestamp (us): [1000000, 2000000, 1704067200123456] timestamptz (us): [1000000, 2000000, 1704067200123456] ``` I have not tested a fix, filtered scans or pruning. ### Expected behavior The scan should return 1000000, 2000000 and 1704067200123456 microseconds for both `timestamp` and `timestamptz`, as v0.9.0 does. ### Willingness to contribute I would be willing to contribute a fix for this bug with guidance from the Iceberg community -- 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]
