bug(arrow): filters lose rows after numeric schema promotion

[mattfaltyn]

Apache Iceberg Rust version

Current main at 5c37021834c5391f499bb04e825e5b3d3282bdee (2026-08-31), Arrow/Parquet 58.4.0, pinned nightly-2026-04-16, macOS arm64.

Describe the bug

Filtered scans can silently discard matching rows from older Parquet files after a valid int → long schema promotion. Unfiltered reads return the correctly promoted values, but a predicate bound to the new table schema is evaluated against the old physical column before RecordBatchTransformer promotes it.

PredicateConverter::try_cast_literal casts the Int64 predicate scalar down to the physical Int32 column. An out-of-range scalar becomes null, so comparisons produce null instead of true and the row filter discards the rows.

For a file containing x: int = [1, 2, 3], read using x: long with the same field ID:

Predicate	Expected	Actual
No predicate	[1, 2, 3]	[1, 2, 3]
x < 4	[1, 2, 3]	[1, 2, 3]
x < 2147483648	[1, 2, 3]	[]
x > -2147483649	[1, 2, 3]	[]
x != 2147483648	[1, 2, 3]	[]

This affects query correctness without producing an error. The reproduction uses the default reader options; it does not require page-index row selection, delete files, a catalog, or external services.

To Reproduce

Create a small binary crate outside the workspace with these dependencies, replacing the Iceberg path with a checkout of the revision above:

[package]
name = "promotion-repro"
version = "0.1.0"
edition = "2024"

[dependencies]
iceberg = { path = "/path/to/iceberg-rust/crates/iceberg" }
arrow-array = "=58.4.0"
arrow-schema = "=58.4.0"
parquet = "=58.4.0"
futures = "0.3"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tempfile = "3"

Put the following in src/main.rs and run cargo +nightly-2026-04-16 run. It writes the old physical schema and reads it through ArrowReader with the evolved schema. The assertion fails for the three out-of-range predicates. cargo +nightly-2026-04-16 run -- --control passes.

Self-contained reproduction

use std::{collections::HashMap, fs::File, sync::Arc};

use arrow_array::{Int32Array, Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use futures::{TryStreamExt, stream};
use iceberg::{
    Runtime,
    arrow::ArrowReaderBuilder,
    expr::{Bind, Predicate, Reference},
    io::FileIO,
    scan::FileScanTask,
    spec::{DataFileFormat, Datum, NestedField, PrimitiveType, Schema, Type},
};
use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("old-int.parquet");
    let field = Field::new("x", DataType::Int32, false).with_metadata(HashMap::from([(
        PARQUET_FIELD_ID_META_KEY.to_string(),
        "1".to_string(),
    )]));
    let batch = RecordBatch::try_new(
        Arc::new(ArrowSchema::new(vec![field])),
        vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
    )?;
    let mut writer = ArrowWriter::try_new(File::create(&path)?, batch.schema(), None)?;
    writer.write(&batch)?;
    writer.close()?;
    let schema = Arc::new(
        Schema::builder()
            .with_schema_id(1)
            .with_fields(vec![Arc::new(NestedField::required(
                1,
                "x",
                Type::Primitive(PrimitiveType::Long),
            ))])
            .build()?,
    );
    let cases: Vec<(&str, Option<Predicate>, Vec<i64>)> = vec![
        ("unfiltered", None, vec![1, 2, 3]),
        (
            "x < 4 (representable control)",
            Some(Reference::new("x").less_than(Datum::long(4))),
            vec![1, 2, 3],
        ),
        (
            "x < 2147483648",
            Some(Reference::new("x").less_than(Datum::long(2147483648_i64))),
            vec![1, 2, 3],
        ),
        (
            "x > -2147483649",
            Some(Reference::new("x").greater_than(Datum::long(-2147483649_i64))),
            vec![1, 2, 3],
        ),
        (
            "x != 2147483648",
            Some(Reference::new("x").not_equal_to(Datum::long(2147483648_i64))),
            vec![1, 2, 3],
        ),
    ];
    let mut failures = 0;
    let control_only = std::env::args().any(|arg| arg == "--control");
    for (index, (name, predicate, expected)) in cases.into_iter().enumerate() {
        if control_only && index >= 2 {
            break;
        }
        let task = FileScanTask::builder()
            .with_file_size_in_bytes(path.metadata()?.len())
            .with_start(0)
            .with_length(path.metadata()?.len())
            .with_data_file_path(path.to_str().unwrap().to_string())
            .with_data_file_format(DataFileFormat::Parquet)
            .with_schema(schema.clone())
            .with_project_field_ids(vec![1])
            .with_case_sensitive(true)
            .with_predicate(
                predicate
                    .map(|p| p.bind(schema.clone(), true))
                    .transpose()?,
            )
            .build();
        let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
        let batches: Vec<RecordBatch> = reader
            .read(Box::pin(stream::iter(vec![Ok(task)])))?
            .stream()
            .try_collect()
            .await?;
        let actual: Vec<i64> = batches
            .iter()
            .flat_map(|b| {
                b.column(0)
                    .as_any()
                    .downcast_ref::<Int64Array>()
                    .unwrap()
                    .values()
                    .iter()
                    .copied()
            })
            .collect();
        println!("{name}: expected={expected:?}, actual={actual:?}");
        failures += usize::from(actual != expected);
    }
    assert_eq!(
        failures, 0,
        "filters must preserve results across int-to-long promotion"
    );
    Ok(())
}

Reproduced twice on the revision above; both controls pass.

Expected behavior

Predicates must preserve their meaning across supported numeric schema promotions. Comparing the promoted column to the original bound scalar must not narrow or null the scalar. Equivalent Arrow string/binary representation casts should continue to work.

Related work checked

I searched open and closed issues/PRs, including comments, for the helper name, numeric promotion, widening, overflow, and the boundary values above. #1307/#1308 concern equivalent Arrow string representations, not numeric narrowing. Open #2961 handles promotion for equality-delete keys but leaves ordinary predicate conversion unchanged. Open #3069 and #3073 do not fix this path. I did not find an existing report or fix for this failure.

Scope of the proposed fix

The fix in #3123 covers ordinary Arrow row filters and numeric promotion of integer/float bounds in optional page-index pruning. The latter constructs promoted numeric Datum values with the old physical literal variant, so enabling with_row_selection_enabled(true) can discard matching rows before the row filter runs.

Adding .with_row_selection_enabled(true) to the ArrowReaderBuilder in the reproduction above exposes this additional failure. The expanded regression also fails against the initial row-filter-only fix (8fe5538): x < 2147483648 returns no rows when page pruning is enabled. Its fixture has multiple pages and row groups, and the reader uses single-row batches. The follow-up in #3123 promotes integer/float page bounds while preserving the existing handling of other index types. All 472 regression configurations pass on the updated PR head (c598504a3).

Decimal row filtering is covered at scales 0 and 2. General decimal page-index decoding remains outside this fix: FIXED_LEN_BYTE_ARRAY indexes are unsupported even without schema evolution. Closed, unmerged #1950 proposed that separate feature. A fresh search of issues and PRs found no existing integer/float page-bound promotion fix.

Willingness to contribute

I can contribute a fix for this bug independently.

AI disclosure

Codex assisted with investigation and this report. The behavior was verified with the executable reproduction above.