hcrosse opened a new issue, #3617: URL: https://github.com/apache/iceberg-rust/issues/3617
### Apache Iceberg Rust version Main at `896f9d6c50ff52fa9e43aec170d3f14bfb40b512`, tested on Linux amd64 with Rust 1.99.1 and Arrow/Parquet 59.3.2. ### Describe the bug When a Parquet file stores a `timestamp` column as `TIMESTAMP(MILLIS)`, a scan that filters on that column without selecting it silently drops matching rows. Selecting the column instead fails with `DataInvalid => Unsupported Arrow data type: Timestamp(ms)`. I reproduced the row loss with one synthetic row at `1970-01-01T00:00:02`. The file uses `TIMESTAMP(MILLIS)` and the table schema uses `timestamp`. Filtering on that value and selecting only `id` returns no rows with the default reader. Disabling row-group pruning returns one row. Enabling only page-index selection or bloom filtering also drops it. Both [`ArrowReaderBuilder`](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/arrow/reader/mod.rs#L74-L76) and [`TableScanBuilder`](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/scan/mod.rs#L171-L173) enable row-group filtering by default, with page-index selection and bloom filtering off. This does not require opting into pruning. The reader [interprets the raw row-group bounds as microseconds](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/arrow/schema.rs#L1018-L1023). The [page-index conversion preserves the raw integer](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/expr/visitors/page_index_evaluator.rs#L506-L515), and the [bloom check hashes the literal without converting its unit](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/expr/visitors/bloom_filter_evaluator.rs#L271-L272). The stored value is `2000`, but the predicate uses `2000000` microseconds. Selecting `ts` fails even without a predicate. The [projection path calls `arrow_schema_to_schema`](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/arrow/reader/projection.rs#L184-L192), which [rejects the millisecond type](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/arrow/schema.rs#L643-L670). I kept both symptoms in one issue because accepting millisecond columns in projection without fixing pruning would turn this error into silent row loss for scans that select and filter `ts`. Iceberg Java has the same row-group statistics bug, reported in https://github.com/apache/iceberg/issues/18173. Its open fix, https://github.com/apache/iceberg/pull/18321, converts the bounds to microseconds before filtering. ### To Reproduce Create a Rust 2024 binary crate with these dependencies in `Cargo.toml`. Put the program in `src/main.rs` and run `cargo run`. I used Arrow/Parquet 59.3.2, resolved from these version ranges. ```toml [dependencies] arrow-array = "59.2" arrow-schema = "59.2" futures = "0.3" iceberg = { git = "https://github.com/apache/iceberg-rust";, rev = "896f9d6c50ff52fa9e43aec170d3f14bfb40b512" } parquet = "59.2" tokio = { version = "1", features = ["macros", "rt-multi-thread"] } ``` ```rust use std::collections::HashMap; use std::error::Error; use std::fs::File; use std::sync::Arc; use arrow_array::{Int32Array, RecordBatch, TimestampMillisecondArray}; use arrow_schema::{DataType, Field, Schema as ArrowSchema, TimeUnit}; use futures::TryStreamExt; use iceberg::Runtime; use iceberg::arrow::ArrowReaderBuilder; use iceberg::expr::{Bind, Predicate, Reference}; use iceberg::io::FileIO; use iceberg::scan::FileScanTask; use iceberg::spec::{DataFileFormat, Datum, NestedField, PrimitiveType, Schema, SchemaRef, Type}; use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}; use parquet::file::properties::WriterProperties; #[tokio::main] async fn main() -> Result<(), Box<dyn Error>> { let path = std::env::temp_dir().join("ts_millis.parquet"); let path = path.to_str().unwrap(); let field_id = |id: &str| HashMap::from([(PARQUET_FIELD_ID_META_KEY.into(), id.into())]); let millis = DataType::Timestamp(TimeUnit::Millisecond, None); let file_schema = Arc::new(ArrowSchema::new(vec![ Field::new("id", DataType::Int32, false).with_metadata(field_id("1")), Field::new("ts", millis, false).with_metadata(field_id("2")), ])); let batch = RecordBatch::try_new( file_schema.clone(), vec![ Arc::new(Int32Array::from(vec![0])), Arc::new(TimestampMillisecondArray::from(vec![2_000])), ], )?; let props = WriterProperties::builder() .set_bloom_filter_enabled(true) .build(); let mut writer = ArrowWriter::try_new(File::create(path)?, file_schema, Some(props))?; writer.write(&batch)?; writer.close()?; let schema = Arc::new( Schema::builder() .with_fields(vec![ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), NestedField::required(2, "ts", Type::Primitive(PrimitiveType::Timestamp)).into(), ]) .build()?, ); let ts_is_2s = Reference::new("ts").equal_to(Datum::timestamp_micros(2_000_000)); let reader = || ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()); let full = rows(reader(), path, &schema, None).await?; println!("full scan: {full} row(s)"); for (label, reader) in [ ("default reader", reader()), ( "row-group filtering off", reader().with_row_group_filtering_enabled(false), ), ( "page index only", reader() .with_row_group_filtering_enabled(false) .with_row_selection_enabled(true), ), ( "bloom filter only", reader() .with_row_group_filtering_enabled(false) .with_bloom_filter_enabled(true), ), ] { let n = rows(reader, path, &schema, Some(&ts_is_2s)).await?; println!("ts = 1970-01-01T00:00:02, {label}: {n} row(s), expected 1"); } Ok(()) } async fn rows( reader: ArrowReaderBuilder, path: &str, schema: &SchemaRef, filter: Option<&Predicate>, ) -> Result<usize, Box<dyn Error>> { let size = std::fs::metadata(path)?.len(); let task = FileScanTask::builder() .with_file_size_in_bytes(size) .with_start(0) .with_length(size) .with_data_file_path(path.to_string()) .with_data_file_format(DataFileFormat::Parquet) .with_schema(schema.clone()) .with_project_field_ids(vec![1]) .with_predicate(filter.map(|p| p.bind(schema.clone(), true)).transpose()?) .with_case_sensitive(true) .build()?; let batches: Vec<RecordBatch> = reader .build() .read(Box::pin(futures::stream::iter([Ok(task)])))? .stream() .try_collect() .await?; Ok(batches.iter().map(RecordBatch::num_rows).sum()) } ``` Output: ```text $ cargo run # iceberg-rust 896f9d6c, parquet/arrow 59.3.2, rustc 1.99.1, linux/amd64 full scan: 1 row(s) ts = 1970-01-01T00:00:02, default reader: 0 row(s), expected 1 ts = 1970-01-01T00:00:02, row-group filtering off: 1 row(s), expected 1 ts = 1970-01-01T00:00:02, page index only: 0 row(s), expected 1 ts = 1970-01-01T00:00:02, bloom filter only: 0 row(s), expected 1 ``` To see the projection error, pass `vec![2]` or `vec![1, 2]` to `with_project_field_ids`. Every mode then fails, with or without the predicate. The reproduction uses `FileScanTask` and local files, not catalog import or a complete `TableScan`. ### Expected behavior The filtered scan returns the matching row in every pruning mode. Pruning either uses the file's timestamp unit or keeps data it cannot safely rule out. Selecting `ts` returns `1970-01-01T00:00:02` as a microsecond timestamp, as Iceberg Java's [Parquet reader](https://github.com/apache/iceberg/blob/5a2ccf5ba830838ecf182bccb516d5ee48f8b8ed/parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java#L160-L169) does. If millisecond files are out of scope, the reader rejects them in every mode instead of returning incomplete results. ### 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]
