fix(reader): keep matching rows when filtering a promoted column by comphead · Pull Request #3540 · apache/iceberg-rust

comphead · 2026-10-26T17:49:56Z

Which issue does this PR close?

Part of bug(arrow): filters lose rows after numeric schema promotion #3122. This fixes the row filter, the reader's default path. The open PRs fix(reader): keep pages whose index bounds cannot be compared with the predicate #3507 and fix(reader): widen promoted page-index bounds to the field type #3521 handle the page-index part (with_row_selection_enabled(true)).

What changes are included in this PR?

Predicates are bound to the table schema, but the Arrow row filter evaluated them on batches in the file's types. For a file written before a type promotion, try_cast_literal cast the literal down to the file type. An out-of-range long literal became null, so x < 2147483648 returned no rows. A double literal was rounded to f32, and a timestamp literal was truncated to a date.

get_row_filter converts the filter columns to the table schema with a RecordBatchTransformer, as the reader does for its output. This covers int -> long, float -> double, decimal precision, date -> timestamp/timestamp_ns, and INT(8) or UINT_32 files.
try_cast_literal fails instead of returning null. After the conversion it only bridges representations of one type, such as Binary and LargeBinary.

Costs, on files with a predicate:

Per batch: a filter column whose file type differs from the table type is cast in full, where main cast one literal. This covers promoted columns, large_string/large_binary files (about 0.5 ns per row), and timestamptz columns, where the cast only relabels "UTC" as "+00:00".
Per file: the filter transformer is built from a schema that holds only the filter fields, so its setup grows with the number of filter columns, not the table's width. The output transformer still converts the whole schema, tracked in perf(arrow): build RecordBatchTransformer's plan from the projected fields only #3572.

Are these changes tested?

test_predicate_on_promoted_column covers int, float and date files read as long, double and timestamp, with and without field IDs. It fails on main.
test_predicate_converter_rejects_lossy_literal_cast covers the strict cast.
Skipping the transformer, misaligning its field IDs, or relaxing the cast each fail a test. Dropping the leaf-order sort does not, since it only affects performance.

AI Disclosure

Drafted with Claude Code (Anthropic) and reviewed by the author.

Copilot

🟢 Approval recommended

The focused fix consistently covers all eight affected comparison paths with targeted regression tests and no identified blocking issues.

No open issues 🎉

What changed in this PR

Fixes row filtering on numerically promoted columns, addressing the default-reader portion of #3122. Page-index pruning remains outside this PR’s scope.

Changes:

Casts numeric columns to the table type before comparisons, preventing literal overflow and rounding from changing filter results.
Adds regression coverage for all eight comparison paths across integer and floating-point promotions.

File	Description
`crates/iceberg/src/arrow/reader/row_filter.rs`	Adds promoted-column read tests with exact expected results.
`crates/iceberg/src/arrow/reader/predicate_visitor.rs`	Applies numeric column promotion while preserving nonnumeric literal casting.

_{Review effort: Balanced}

kevinjqliu · 2026-10-27T04:06:27Z

thanks for digging into this! the fix works, but this feels like a workaround to me 😅

I think the root cause is that predicates are bound to the table schema but evaluated against file schema batches. try_cast_literal narrows the literal to the file type, and a failed cast silently becomes null. this PR casts the column back up in each comparison, but only for numeric types, so the lossy literal cast is still there for everything else.

i think we should cast the filter batch to table types once in get_row_filter, using the same rule as RecordBatchTransformer. then try_cast_literal can error instead of returning null.

another option is ArrowReaderOptions::with_schema so parquet decodes directly into table types. not sure it supports int -> long though

Let me know what you think!

comphead · 2026-10-27T15:45:34Z

Thanks @kevinjqliu for the review, it might totally happen that batches’ schema and incoming schema are different, in datafusion, arrow and Comet we have a notion on schema_adapter which actually converges the schema and I think RecordBatchTransformer plays the same role, checking it.

ArrowReaderOptions::with_schema for int -> long is not currently supported

Predicates are bound to the table schema, but the row filter evaluated them on batches in the file's physical types. For a file written before an int -> long, float -> double, or decimal precision promotion, try_cast_literal cast the literal down to the file type, so an out-of-range literal became null and dropped every row, and a double literal was rounded to f32. Convert the filter columns to the table schema with a RecordBatchTransformer, as the reader does for its output, and make try_cast_literal fail instead of returning null. Page-index pruning is left to apache#3507 and apache#3521. Part of apache#3122.

comphead · 2026-10-27T18:32:45Z

@kevinjqliu please take another look

Copilot

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Copilot review overview

2 open issues

CastOptions { safe: false } is counterintuitive given the intent described in the comment… · New
In assert_promoted_reads, with_length(0) is ambiguous and makes the test harder to understand… · New

_{Review effort: Lite}

comphead · 2026-10-28T16:56:39Z

+    let options = CastOptions {
+        safe: false,
+        ..Default::default()
+    };
+    let literal_array = cast_with_options(literal_array, column_type, &options)?;


Added a one-line comment above the options.

comphead · 2026-10-28T16:56:41Z

+        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()


The helper now reuses read_once, so this code is gone.

mbutrovich

Thanks @comphead, first pass.

mbutrovich · 2026-10-28T16:16:06Z

+        let mut transformer =
+            RecordBatchTransformerBuilder::new(snapshot_schema, &field_ids).build();


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 converts the whole snapshot schema to Arrow and maps every top-level field, even when the result is a pass-through. The output transformer 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?

Thanks, agreed on both. The description now names the per-batch cast, including the timestamptz relabel, and the per-file plan. I'd like to scope generate_batch_transform to the projected fields in a follow-up, since it changes the output transformer too and several open PRs touch that file.

I'd like to scope generate_batch_transform to the projected fields in a follow-up, since it changes the output transformer too and several open PRs touch that file.

Could we fix the per-file cost for the filter transformer here without touching record_batch_transformer.rs? get_row_filter could hand the builder a schema that holds only the filter fields. The output transformer and generate_batch_transform stay as they are, so the follow-up is only needed for the output side:

let filter_schema = Schema::builder() .with_schema_id(snapshot_schema.schema_id()) .with_fields( field_ids .iter() .filter_map(|id| snapshot_schema.field_by_id(*id).cloned()), ) .build()?; let mut transformer = RecordBatchTransformerBuilder::new(Arc::new(filter_schema), &field_ids).build();

I tried this on the head commit, and the arrow:: tests (278) and scan tests (105) pass. I timed building the transformer and processing one 8192-row int -> long batch for a single filter column, in a release build:

Table width Full schema Filter fields only

1 column 3.3 us 2.3 us

101 columns 15.1 us 1.9 us

1001 columns 135.1 us 2.0 us

The cost now grows with the number of filter columns, not the table's width. The new cost is one small Schema build per file, and the 1-column row is where it gains least. It isn't slower there. Since this PR adds the second transformer, keeping its setup flat would mean the PR doesn't add a per-file cost that depends on width. Should the follow-up for the output transformer also get an issue, so we can link it from the description?

Done in 03639f1, thanks for measuring. I used as_struct().field_by_id, the top-level lookup, so a nested field can't become a top-level column that the transformer fills with nulls. The converter rejects nested predicates before this today, but #2906 would lift that. Opened #3572 for the output transformer and linked it from the description.

mbutrovich · 2026-10-28T16:16:17Z

+    /// 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);


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), 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.

Added. assert_promoted_reads now writes each file with and without a field ID. read_once takes no name mapping, so the second file goes through the positional fallback, which feeds the same with_schema hint.

mbutrovich · 2026-10-28T16:16:33Z

+        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}");
+        }


Could this reuse read_once further down in this module? It builds the same task and reader.

Suggested change

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}");

}

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}");

}

Done, thanks.

mbutrovich · 2026-10-28T16:16:49Z

+        // 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;
+    }


This change also fixes the v3 date to timestamp and timestamp_ns promotions. 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:

// 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;

Added, thanks. I also checked timestamp_ns locally, and the description lists both.

comphead · 2026-10-28T16:22:30Z

Thanks @mbutrovich for the review, checking it

…the strict cast Reuse read_once in assert_promoted_reads, read each file with and without a field id, and add a date -> timestamp case. Explain why try_cast_literal sets safe: false.

… only The transformer converts its whole schema to Arrow on each file's first batch. Give it a schema of the top-level filter fields so that setup no longer grows with the table's width.

JeonDaehong · 2026-11-02T13:37:33Z

+            .with_fields(
+                field_ids
+                    .iter()
+                    .filter_map(|id| snapshot_schema.as_struct().field_by_id(*id).cloned()),


as_struct().field_by_id only looks at top-level fields, so a nested leaf never makes it into filter_schema. Not an issue on main yet, but I merged this with #2906 locally and 5 of its nested predicate tests fail with field not found (they pass on #2906 alone). Maybe build the filter schema from the struct parents of those leaves?

comphead force-pushed the fix-3122-promoted-predicates branch from b92d957 to 0f76a20 Compare October 26, 2026 19:50

github-actions Bot added the datafusion label Oct 26, 2026

comphead marked this pull request as ready for review October 26, 2026 22:48

Copilot AI balanced review requested due to automatic review settings October 26, 2026 22:48

Copilot started reviewing on behalf of comphead October 26, 2026 22:49

View session

Copilot AI reviewed Oct 26, 2026

View reviewed changes

comphead force-pushed the fix-3122-promoted-predicates branch from 0f76a20 to 0be54f9 Compare October 27, 2026 16:16

comphead requested a balanced review from Copilot October 27, 2026 18:32

Copilot AI reviewed Oct 27, 2026

View reviewed changes

Copilot stopped reviewing on behalf of comphead due to an error October 27, 2026 18:51
The job was not acquired by a runner even after multiple attempts.

comphead requested a review from kevinjqliu October 28, 2026 15:25

mbutrovich suggested changes Oct 28, 2026

View reviewed changes

test(reader): cover files without field ids and date promotion; name …

5a6cf1d

…the strict cast Reuse read_once in assert_promoted_reads, read each file with and without a field id, and add a date -> timestamp case. Explain why try_cast_literal sets safe: false.

comphead requested a review from mbutrovich October 28, 2026 17:03

perf(reader): build the row filter transformer from the filter fields…

03639f1

… only The transformer converts its whole schema to Arrow on each file's first batch. Give it a schema of the top-level filter fields so that setup no longer grows with the table's width.

This was referenced Oct 28, 2026

perf(arrow): build RecordBatchTransformer's plan from the projected fields only #3572

Open

Native Iceberg scan fails a VERSION AS OF read of a column whose type was promoted after the snapshot apache/datafusion-comet#6993

Open

JeonDaehong reviewed Nov 2, 2026

View reviewed changes

This was referenced Nov 3, 2026

perf(reader): skip per-row predicate evaluation when statistics prove every row matches #3613

Open

Allow dictionary output for selected string columns in ArrowReader #3614

Open

		let mut transformer =
		RecordBatchTransformerBuilder::new(snapshot_schema, &field_ids).build();

Table width	Full schema	Filter fields only
1 column	3.3 us	2.3 us
101 columns	15.1 us	1.9 us
1001 columns	135.1 us	2.0 us

Conversation

comphead commented Oct 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

What changes are included in this PR?

Are these changes tested?

AI Disclosure

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟢 Approval recommended

Uh oh!

kevinjqliu commented Oct 27, 2026

Uh oh!

comphead commented Oct 27, 2026

Uh oh!

comphead commented Oct 27, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead commented Oct 28, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

comphead commented Oct 26, 2026 •

edited

Loading