mbutrovich commented on code in PR #3540:
URL: https://github.com/apache/iceberg-rust/pull/3540#discussion_r4403117490
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -62,8 +70,18 @@ impl ArrowReader {
// After collecting required leaf column indices used in the predicate,
// creates the projection mask for the Arrow predicates.
let projection_mask = ProjectionMask::leaves(parquet_schema,
column_indices.clone());
- let predicate_func = visit(&mut converter, predicates)?;
- let arrow_predicate = ArrowPredicateFn::new(projection_mask,
predicate_func);
+ let mut predicate_func = visit(&mut converter, predicates)?;
+ // Predicates are bound to the table schema, so evaluate them on the
filter columns
+ // converted to it, as the reader's output is (e.g. a file written
before an `int` ->
+ // `long` promotion).
+ let mut transformer =
+ RecordBatchTransformerBuilder::new(snapshot_schema,
&field_ids).build();
Review Comment:
This moves the type conversion from the literal to the column, which adds
cost on two paths.
Per batch, a filter column whose file type differs from the table type is
now cast in full, where `main` cast one scalar literal. That covers promoted
columns and `large_string` or `large_binary` files (the 0.5 ns per row in the
description), and it scales with the row count. `timestamptz` columns take this
path too, because parquet-rs reads UTC-adjusted timestamps with the timezone
`"UTC"` and the table type uses `"+00:00"`. That cast only relabels the
timezone and leaves the values alone, so it costs the same for any batch size.
Per file, this transformer builds its plan on the first batch.
[`generate_batch_transform`](https://github.com/apache/iceberg-rust/blob/0be54f9076c096e13e3737a6016ab277c495222d/crates/iceberg/src/arrow/record_batch_transformer.rs#L798-L800)
converts the whole snapshot schema to Arrow and maps every top-level field,
even when the result is a pass-through. The [output
transformer](https://github.com/apache/iceberg-rust/blob/0be54f9076c096e13e3737a6016ab277c495222d/crates/iceberg/src/arrow/reader/pipeline.rs#L333-L334)
already does this, so every file read with a predicate now does it twice. The
cost grows with the table's width, not with the number of filter columns.
The degenerate case is a wide table stored as many small files, such as a
table with hundreds of columns that is heavily partitioned or bucketed, or
written by frequent small commits. Every file pays the setup, and there are few
rows to spread it over. A selective filter on a `large_string` column is the
per-batch counterpart. Building the plan from only the fields in `field_ids`
would remove the per-file part, and the output transformer would gain from it
too. Could the description name these costs either way?
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -460,6 +478,89 @@ mod tests {
}
}
+ /// Writes `column` as field 1, reads it back as `iceberg_type` under each
predicate, and
+ /// checks the values that survive (`None` when no row does).
+ async fn assert_promoted_reads(
+ column: ArrayRef,
+ iceberg_type: PrimitiveType,
+ cases: Vec<(Predicate, Option<ArrayRef>)>,
+ ) {
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/column.parquet", tmp.path().to_str().unwrap());
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![field_with_id(
+ "v",
+ column.data_type().clone(),
+ 1,
+ )]));
+ let batch = RecordBatch::try_new(arrow_schema.clone(),
vec![column]).unwrap();
+ write_row_groups(&path, arrow_schema, vec![batch], false);
+
+ let schema = value_schema(Type::Primitive(iceberg_type));
+ let file_size = std::fs::metadata(&path).unwrap().len();
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(),
Runtime::current()).build();
+ for (predicate, expected) in cases {
+ let task = FileScanTask::builder()
+ .with_file_size_in_bytes(file_size)
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(path.clone())
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema.clone())
+ .with_project_field_ids(vec![1])
+ .with_predicate(Some(predicate.bind(schema.clone(),
true).unwrap()))
+ .with_case_sensitive(true)
+ .build()
+ .unwrap();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as
FileScanTaskStream;
+ let batches: Vec<RecordBatch> = reader
+ .clone()
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+ let actual = collapse(&batches).map(|batch|
Arc::clone(batch.column(0)));
+ assert_eq!(actual, expected, "{predicate}");
+ }
+ }
+
+ /// A file written before a type promotion keeps the old physical type, so
the row
+ /// filter must compare its values promoted to the table type.
+ #[tokio::test]
+ async fn test_predicate_on_promoted_column() {
+ let longs = |values: Vec<i64>| Some(Arc::new(Int64Array::from(values))
as ArrayRef);
+ let all = longs(vec![1, 2]);
+ let above = || Datum::long(i64::from(i32::MAX) + 1);
+ let below = || Datum::long(i64::from(i32::MIN) - 1);
+ let v = || Reference::new("v");
+ assert_promoted_reads(
+ Arc::new(Int32Array::from(vec![1, 2])),
+ PrimitiveType::Long,
+ vec![
+ (v().less_than(above()), all.clone()),
+ (v().less_than_or_equal_to(above()), all.clone()),
+ (v().greater_than_or_equal_to(below()), all.clone()),
+ (v().not_equal_to(above()), all),
+ (v().is_in([Datum::long(1), above()]), longs(vec![1])),
+ (v().is_not_in([Datum::long(1), above()]), longs(vec![2])),
+ ],
+ )
+ .await;
+
+ // 0.1f32 promotes to a double that is greater than 0.1, not equal to
it.
+ let promoted = Arc::new(Float64Array::from(vec![f64::from(0.1f32)]))
as ArrayRef;
+ assert_promoted_reads(
+ Arc::new(Float32Array::from(vec![0.1])),
+ PrimitiveType::Double,
+ vec![
+ (v().greater_than(Datum::double(0.1)), Some(promoted)),
+ (v().equal_to(Datum::double(0.1)), None),
+ ],
+ )
+ .await;
+ }
Review Comment:
This change also fixes the v3 [`date` to `timestamp` and `timestamp_ns`
promotions](https://github.com/apache/iceberg/blob/6275107e38cc87a7d416ef231019ceb51556f196/format/spec.md?plain=1#L395-L403).
On `main`, a `date` file holding 1970-01-01 and 2024-01-01, read as
`timestamp` with `v < 2024-01-01T12:00:00`, returns only 1970-01-01. The
literal is cast down to `Date32` and truncated to 2024-01-01. `timestamp_ns`
loses the same row. Should we add a case for it here, and mention it in the
description? For example, after the `double` case:
```rust
// 2024-01-01 promotes to midnight, which is before noon on that day.
let noon = || Datum::timestamp_from_str("2024-01-01T12:00:00").unwrap();
let midnight =
Arc::new(TimestampMicrosecondArray::from(vec![1_704_067_200_000_000]));
assert_promoted_reads(
Arc::new(Date32Array::from(vec![19723])),
PrimitiveType::Timestamp,
vec![(v().less_than(noon()), Some(midnight as ArrayRef))],
)
.await;
```
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -460,6 +478,89 @@ mod tests {
}
}
+ /// Writes `column` as field 1, reads it back as `iceberg_type` under each
predicate, and
+ /// checks the values that survive (`None` when no row does).
+ async fn assert_promoted_reads(
+ column: ArrayRef,
+ iceberg_type: PrimitiveType,
+ cases: Vec<(Predicate, Option<ArrayRef>)>,
+ ) {
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/column.parquet", tmp.path().to_str().unwrap());
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![field_with_id(
+ "v",
+ column.data_type().clone(),
+ 1,
+ )]));
+ let batch = RecordBatch::try_new(arrow_schema.clone(),
vec![column]).unwrap();
+ write_row_groups(&path, arrow_schema, vec![batch], false);
+
+ let schema = value_schema(Type::Primitive(iceberg_type));
+ let file_size = std::fs::metadata(&path).unwrap().len();
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(),
Runtime::current()).build();
+ for (predicate, expected) in cases {
+ let task = FileScanTask::builder()
+ .with_file_size_in_bytes(file_size)
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(path.clone())
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema.clone())
+ .with_project_field_ids(vec![1])
+ .with_predicate(Some(predicate.bind(schema.clone(),
true).unwrap()))
+ .with_case_sensitive(true)
+ .build()
+ .unwrap();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as
FileScanTaskStream;
+ let batches: Vec<RecordBatch> = reader
+ .clone()
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+ let actual = collapse(&batches).map(|batch|
Arc::clone(batch.column(0)));
+ assert_eq!(actual, expected, "{predicate}");
+ }
Review Comment:
Could this reuse `read_once` further down in this module? It builds the same
task and reader.
```suggestion
let schema = value_schema(Type::Primitive(iceberg_type));
for (predicate, expected) in cases {
let (batches, _) =
read_once(&path, schema.clone(), vec![1], predicate.clone(),
false).await;
let actual = collapse(&batches).map(|batch|
Arc::clone(batch.column(0)));
assert_eq!(actual, expected, "{predicate}");
}
```
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -460,6 +478,89 @@ mod tests {
}
}
+ /// Writes `column` as field 1, reads it back as `iceberg_type` under each
predicate, and
+ /// checks the values that survive (`None` when no row does).
+ async fn assert_promoted_reads(
+ column: ArrayRef,
+ iceberg_type: PrimitiveType,
+ cases: Vec<(Predicate, Option<ArrayRef>)>,
+ ) {
+ let tmp = TempDir::new().unwrap();
+ let path = format!("{}/column.parquet", tmp.path().to_str().unwrap());
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![field_with_id(
+ "v",
+ column.data_type().clone(),
+ 1,
+ )]));
+ let batch = RecordBatch::try_new(arrow_schema.clone(),
vec![column]).unwrap();
+ write_row_groups(&path, arrow_schema, vec![batch], false);
Review Comment:
What do you think about also reading a file that has no field IDs, through a
name mapping? The transformer finds each filter column by the field-id metadata
that the `with_schema` hint puts on the filter batch. A column it can't find
becomes null or its `initial-default` rather than an error
([`record_batch_transformer.rs`](https://github.com/apache/iceberg-rust/blob/0be54f9076c096e13e3737a6016ab277c495222d/crates/iceberg/src/arrow/record_batch_transformer.rs#L1116-L1147)),
so a regression there would drop rows silently again. I tried a name-mapped
`int` file read as `long` with `v < 2147483648`. It returns both rows on the
head commit and none on `main`. Positional fallback behaves the same. A
`field_ids: bool` parameter on `assert_promoted_reads` could cover it.
--
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]
