laskoviymishka commented on code in PR #3269:
URL: https://github.com/apache/iceberg-rust/pull/3269#discussion_r4234772678
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -613,26 +613,24 @@ impl FileScanTaskReader {
record_batch_stream_builder =
record_batch_stream_builder.with_row_filter(row_filter);
if self.row_group_filtering_enabled {
- let predicate_filtered_row_groups =
ArrowReader::get_selected_row_group_indices(
+ let all_row_groups;
Review Comment:
This deferred-init `let all_row_groups; match { … &all_row_groups }` is the
same shape the bloom block just below uses for `all_rgs`, so we've now got two
copies of a fairly subtle borrow pattern. Not blocking, but a small `fn
candidate_row_groups(selected, metadata) -> Cow<'_, [usize]>` would collapse
both and read better — and would let the bloom path reuse the narrowed result
instead of recomputing it.
##########
crates/iceberg/src/arrow/reader/projection.rs:
##########
@@ -524,6 +531,30 @@ mod tests {
};
use crate::{ErrorKind, Runtime};
+ #[test]
+ fn test_name_mapping_duplicate_alias_preserves_first_match() {
Review Comment:
Good that this pins first-match-wins. The subtler case is where the first
`MappedField` for a name has `field_id: None` and a later duplicate has
`Some(id)` — `or_insert` correctly keeps the `None` so no ID gets assigned, but
nothing guards it, so a future "fix" to prefer `Some` would silently change
behavior and this test wouldn't catch it. Worth a second case:
`MappedField::new(None, ["name"], [])` then `MappedField::new(Some(4),
["name"], [])`, asserting no ID lands on the field.
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -634,6 +636,52 @@ mod tests {
assert_eq!(first_val, 100, "Task 2 should start with id=100, not
id=0");
}
+
+ // Combining byte-range and predicate row-group filtering should
narrow the
+ // split's candidates instead of evaluating and intersecting
independent lists.
+ // This task owns row groups 1 and 2, while the predicate can match
row groups
+ // 0 and 2. Only row group 2 should be read.
+ let predicate = Reference::new("id")
+ .less_than(Datum::int(100))
+ .or(Reference::new("id").greater_than_or_equal_to(Datum::int(200)))
+ .bind(schema.clone(), true)
+ .unwrap();
+
+ let task = FileScanTask::builder()
+
.with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
+ .with_start(rg1_start)
+ .with_length(file_end - rg1_start)
+ .with_data_file_path(file_path.clone())
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema.clone())
+ .with_project_field_ids(vec![1])
+ .with_predicate(Some(predicate))
+ .with_record_count(Some(200))
+ .with_case_sensitive(false)
+ .build()
+ .unwrap();
+
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(),
Runtime::current())
+ .with_row_group_filtering_enabled(true)
+ .build();
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as
FileScanTaskStream;
+ let result = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect::<Vec<RecordBatch>>()
+ .await
+ .unwrap();
+
+ let mut ids = Vec::new();
+ for batch in &result {
+ let id_col = batch
+ .column(0)
+ .as_primitive::<arrow_array::types::Int32Type>();
+ ids.extend((0..id_col.len()).map(|index| id_col.value(index)));
+ }
+
+ assert_eq!(ids, (200..300).collect::<Vec<_>>());
Review Comment:
This asserts the output is `200..300`, but the task's byte range already
excludes row group 0 and the row filter drops rows below 200 exactly — so you
get `200..300` whether or not the candidate narrowing does anything, and the
old evaluate-then-intersect code passes it unchanged. So it's a solid
equivalence test, but it can't catch a regression in the new pruning, which is
the thing the comment says it checks. I'd add a direct unit test on
`get_selected_row_group_indices`: pass `&[1, 2]` and assert `vec![2]`, plus a
`&[0, 1]` case asserting `vec![0]`. That also gives the new
`candidate_row_groups` parameter its only direct coverage.
##########
crates/iceberg/src/arrow/reader/row_filter.rs:
##########
@@ -71,11 +71,13 @@ impl ArrowReader {
parquet_metadata: &Arc<ParquetMetaData>,
field_id_map: &HashMap<i32, usize>,
snapshot_schema: &Schema,
+ candidate_row_groups: &[usize],
) -> Result<Vec<usize>> {
let row_groups_metadata = parquet_metadata.row_groups();
- let mut results = Vec::with_capacity(row_groups_metadata.len());
+ let mut results = Vec::with_capacity(candidate_row_groups.len());
- for (idx, row_group_metadata) in
row_groups_metadata.iter().enumerate() {
+ for &idx in candidate_row_groups {
+ let row_group_metadata = &row_groups_metadata[idx];
Review Comment:
Now that this takes `candidate_row_groups: &[usize]` from the caller,
`&row_groups_metadata[idx]` panics on any index that doesn't line up with the
metadata. It's safe today — the only caller builds indices from the same
metadata or `0..num_row_groups()` — but the signature now accepts arbitrary
indices, and the old `.iter().enumerate()` loop couldn't go out of range at
all. Given the never-panic norm I'd return an error instead:
```suggestion
let row_group_metadata =
row_groups_metadata.get(idx).ok_or_else(|| {
Error::new(
ErrorKind::DataInvalid,
format!(
"row group index {idx} out of range ({} row groups)",
row_groups_metadata.len()
),
)
})?;
```
--
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]
