[core] Preserve decimal predicates and equivalent Arrow match keys

bindings/python/src/predicate.rs

@@ -302,6 +302,64 @@ fn decimal_datum(value: &Bound<'_, PyAny>, dec: &DecimalType) -> PyResult<Datum>
     })
 }
 
+/// Predicate literals are values to compare, not values to store in the column.
+/// Java Decimal.compareTo permits a different precision and scale. Keep the
+/// strict column conversion above for partition writes and other stored values.
+fn predicate_datum(value: &Bound<'_, PyAny>, data_type: &DataType) -> PyResult<Datum> {
+    let DataType::Decimal(column) = data_type else {
+        return py_to_datum(value, data_type);
+    };
+    if let Ok(value) = decimal_datum(value, column) {
+        return Ok(value);
+    }
+    let decimal_class = value.py().import("decimal")?.getattr("Decimal")?;
+    let value = if value.is_instance_of::<PyInt>() && !value.is_instance_of::<PyBool>() {
+        decimal_class.call1((value,))?
+    } else if value.is_instance(&decimal_class)? {
+        value.clone()
+    } else {
+        return decimal_datum(value, column);
+    };
+    let tuple = value.call_method0("as_tuple")?;
+    let sign: u8 = tuple.getattr("sign")?.extract()?;
+    let mut exponent: i64 = tuple
+        .getattr("exponent")?
+        .extract()
+        .map_err(|_| PyValueError::new_err("expected a finite decimal predicate literal"))?;
+    let digits: Vec<u8> = tuple.getattr("digits")?.extract()?;
+    let start = digits.iter().position(|d| *d != 0).unwrap_or(digits.len());
+    let mut digits = &digits[start..];
+    if digits.is_empty() {
+        return Ok(Datum::Decimal {
+            unscaled: 0,
+            precision: column.precision(),
+            scale: column.scale(),
+        });
+    }
+    while digits.last() == Some(&0) {
+        digits = &digits[..digits.len() - 1];
+        exponent = exponent
+            .checked_add(1)
+            .ok_or_else(|| PyValueError::new_err("decimal exponent overflow"))?;
+    }
+    let scale = exponent.checked_neg().unwrap_or(i64::MAX).max(0);
+    let precision = (digits.len() as i64)
+        .saturating_add(exponent.max(0))
+        .max(scale);
+    if precision > 38 || scale > 38 {
+        return Err(PyValueError::new_err(
+            "decimal predicate literal exceeds Java DECIMAL precision",
+        ));
+    }
+    let magnitude = digits.iter().fold(0_i128, |v, d| v * 10 + i128::from(*d))
+        * 10_i128.pow(exponent.max(0) as u32);
+    Ok(Datum::Decimal {
+        unscaled: if sign == 1 { -magnitude } else { magnitude },
+        precision: precision as u32,
+        scale: scale as u32,
+    })
+}
+
 /// Rescale a finite `decimal.Decimal` to `scale` exactly via `as_tuple()`
 /// (sign, digits, exponent), avoiding Decimal arithmetic whose context
 /// precision could silently round.
@@ -494,7 +552,7 @@ fn leaf_to_predicate(
                     continue;
                 }
                 out.push(Some(
-                    py_to_datum(&item, &data_type)
+                    predicate_datum(&item, &data_type)
                         .map_err(|e| with_field_context(e, &field, &data_type))?,
                 ));
             }
@@ -1540,6 +1598,79 @@ value = datetime.datetime(2024, 1, 1, tzinfo=FloatingTz())",
         DataType::Decimal(paimon::spec::DecimalType::new(precision, scale).unwrap())
     }
 
+    #[test]
+    fn decimal_predicate_literals_preserve_comparison_value() {
+        Python::attach(|py| {
+            for (text, unscaled, precision, scale) in [
+                ("1.235", 1235, 4, 3),
+                ("-1.23500000000000000000000000000000000000000", -1235, 4, 3),
+                ("123456789.12", 12345678912, 11, 2),
+                ("1E-38", 1, 38, 38),
+                ("1E+37", 10_i128.pow(37), 38, 0),
+                ("0E+100", 0, 10, 2),
+            ] {
+                let value = py_decimal(py, text);
+                assert_eq!(
+                    predicate_datum(&value, &decimal_type(10, 2)).unwrap(),
+                    Datum::Decimal {
+                        unscaled,
+                        precision,
+                        scale
+                    },
+                    "{text}"
+                );
+                // A comparison literal must not relax stored partition values.
+                assert!(py_to_datum(&value, &decimal_type(10, 2)).is_err(), "{text}");
+            }
+            let value = 123456789_i64.into_pyobject(py).unwrap();
+            assert_eq!(
+                predicate_datum(&value, &decimal_type(10, 2)).unwrap(),
+                Datum::Decimal {
+                    unscaled: 123456789,
+                    precision: 9,
+                    scale: 0
+                }
+            );
+            for text in ["1E-39", "1E+38"] {
+                assert!(predicate_datum(&py_decimal(py, text), &decimal_type(10, 2)).is_err());
+            }
+        });
+    }
+
+    #[test]
+    fn decimal_comparison_and_membership_use_literal_scale() {
+        Python::attach(|py| {
+            let fields = vec![DataField::new(0, "amount".into(), decimal_type(10, 2))];
+            for method in [
+                "equal",
+                "notEqual",
+                "lessThan",
+                "lessOrEqual",
+                "greaterThan",
+                "greaterOrEqual",
+                "in",
+                "notIn",
+            ] {
+                let node = leaf_dict(py, method, "amount", &[]);
+                let literals = PyList::empty(py);
+                literals.append(py_decimal(py, "1.235")).unwrap();
+                node.set_item("literals", literals).unwrap();
+                let predicate = dict_to_predicate(&node, &fields, true).unwrap();
+                let Predicate::Leaf { literals, .. } = predicate else {
+                    panic!("expected leaf")
+                };
+                assert_eq!(
+                    literals,
+                    vec![Datum::Decimal {
+                        unscaled: 1235,
+                        precision: 4,
+                        scale: 3
+                    }]
+                );
+            }
+        });
+    }
+
     #[test]
     fn decimal_field_accepts_exact_scale() {
         Python::attach(|py| {

bindings/python/tests/test_read.py

@@ -1032,17 +1032,38 @@ def test_filter_decimal_literal_prunes_and_reads():
         assert t.column("id").to_pylist() == [1]
 
 
-def test_filter_decimal_rejects_float_and_rounding():
-    from decimal import Decimal
-
+def test_filter_decimal_rejects_float():
     with tempfile.TemporaryDirectory() as warehouse:
         table = _make_temporal_table(warehouse)
         b = table.new_read_builder()
         with pytest.raises(ValueError):
             b.with_filter({"method": "equal", "field": "dec", "literals": [12.34]})
-        with pytest.raises(ValueError):
-            b.with_filter(
-                {"method": "equal", "field": "dec", "literals": [Decimal("0.005")]})
+
+
+@pytest.mark.parametrize("method,literal,expected_ids", [
+    ("equal", "0.005", []),
+    ("equal", "12.345", []),
+    ("equal", "12.340", [1]),
+    ("lessThan", "12.345", [1]),
+    ("greaterThan", "56.775", [2]),
+    ("lessThan", "100000000.00", [1, 2]),
+    ("greaterThan", "-100000000.00", [1, 2]),
+])
+def test_filter_decimal_preserves_literal_precision_and_scale(
+        method, literal, expected_ids):
+    from decimal import Decimal
+
+    with tempfile.TemporaryDirectory() as warehouse:
+        table = _make_temporal_table(warehouse)
+        builder = table.new_read_builder().with_filter(
+            {"method": method, "field": "dec", "literals": [Decimal(literal)]})
+        filtered_splits = builder.new_scan().plan().splits()
+        all_splits = table.new_read_builder().new_scan().plan().splits()
+        # Check both file-stat pruning and row filtering without file pruning.
+        for splits in (filtered_splits, all_splits):
+            ids = [row["id"] for batch in builder.new_read().read(splits)
+                   for row in batch.to_pylist()]
+            assert sorted(ids) == expected_ids
 
 
 def test_filter_unknown_field_raises():

crates/paimon/src/table/key_normalization.rs

@@ -23,10 +23,13 @@ use arrow_schema::DataType;
 use super::update_input::{cast_update_value, decode_dictionary, CastMode};
 use super::write_batch_normalize::normalize_write_array;
 
+mod numeric;
+
 /// Unify Arrow representations without changing key identity. Dictionary
-/// indices are batch-local; integer widths, decimal scales and temporal units
-/// may differ only when conversion is exact. Callers validate the logical type
-/// families and decide whether NULL keys match. Payload write rules are separate.
+/// indices are batch-local; numeric widths, decimal scales and temporal units
+/// may differ only when conversion is exact. Binary keys compare their bytes.
+/// Callers validate logical type families and decide whether NULL keys match.
+/// Payload write rules are separate.
 pub(super) fn normalize_match_key(
     array: &ArrayRef,
     expected: &DataType,
@@ -36,19 +39,54 @@ pub(super) fn normalize_match_key(
         return Ok(new_null_array(expected, decoded.len()));
     }
     let source = decoded.data_type();
-    let exact_family = source.is_integer() && expected.is_integer()
-        || source.is_decimal() && expected.is_decimal()
-        || matches!(
-            (source, expected),
-            (
-                DataType::Date32 | DataType::Date64,
-                DataType::Date32 | DataType::Date64
-            ) | (
-                DataType::Time32(_) | DataType::Time64(_),
-                DataType::Time32(_) | DataType::Time64(_)
-            )
+    if source == expected {
+        return Ok(decoded);
+    }
+    if source.is_numeric() && expected.is_numeric() {
+        return numeric::normalize(&decoded, expected);
+    }
+    let binary = |t: &DataType| {
+        matches!(
+            t,
+            DataType::Binary | DataType::LargeBinary | DataType::FixedSizeBinary(_)
         )
-        || matches!((source, expected), (DataType::Timestamp(_, from), DataType::Timestamp(_, to))
+    };
+    if binary(source) && binary(expected) {
+        if source == &DataType::LargeBinary && expected == &DataType::Binary {
+            return super::write_batch_normalize::normalize_large_binary(
+                decoded.as_any().downcast_ref().unwrap(),
+            );
+        }
+        // Key identity is its bytes, independently of Arrow offset width.
+        // Ordinary append validation still sees the original source array.
+        let values = if matches!(source, DataType::FixedSizeBinary(_)) {
+            normalize_write_array(&decoded, &DataType::Binary)?
+        } else {
+            decoded.clone()
+        };
+        return arrow_cast::cast_with_options(
+            values.as_ref(),
+            expected,
+            &arrow_cast::CastOptions {
+                safe: false,
+                ..Default::default()
+            },
+        )
+        .map_err(|error| crate::Error::DataInvalid {
+            message: error.to_string(),
+            source: None,
+        });
+    }
+    let exact_family = matches!(
+        (source, expected),
+        (
+            DataType::Date32 | DataType::Date64,
+            DataType::Date32 | DataType::Date64
+        ) | (
+            DataType::Time32(_) | DataType::Time64(_),
+            DataType::Time32(_) | DataType::Time64(_)
+        )
+    ) || matches!((source, expected), (DataType::Timestamp(_, from), DataType::Timestamp(_, to))
             if from.is_some() == to.is_some());
     if exact_family {
         cast_update_value(&decoded, expected, CastMode::Assignment)
@@ -62,7 +100,8 @@ mod tests {
     use super::*;
     use arrow_array::types::Int8Type;
     use arrow_array::{
-        Array, Decimal128Array, DictionaryArray, Int32Array, Int64Array, Int8Array, StringArray,
+        Array, BinaryArray, Decimal128Array, DictionaryArray, FixedSizeBinaryArray, Float32Array,
+        Float64Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray, StringArray,
         TimestampMicrosecondArray, TimestampMillisecondArray,
     };
     use arrow_schema::TimeUnit;
@@ -102,6 +141,92 @@ mod tests {
         assert!(normalize_match_key(&text, &DataType::Int32).is_err());
     }
 
+    #[test]
+    fn floating_keys_change_width_only_when_identity_is_preserved() {
+        let input: ArrayRef = Arc::new(Float64Array::from(vec![
+            Some(99.0),
+            Some(1.5),
+            None,
+            Some(-2.25),
+        ]));
+        let input = input.slice(1, 3);
+        let narrow = normalize_match_key(&input, &DataType::Float32).unwrap();
+        assert_eq!(
+            narrow.as_any().downcast_ref::<Float32Array>().unwrap(),
+            &Float32Array::from(vec![Some(1.5), None, Some(-2.25)])
+        );
+        assert_eq!(
+            normalize_match_key(&narrow, &DataType::Float64)
+                .unwrap()
+                .to_data(),
+            input.to_data()
+        );
+        for value in [1.00000001, f64::MAX, f64::MIN_POSITIVE] {
+            let lossy: ArrayRef = Arc::new(Float64Array::from(vec![value]));
+            assert!(normalize_match_key(&lossy, &DataType::Float32).is_err());
+        }
+        let dictionary: ArrayRef = Arc::new(
+            DictionaryArray::<Int8Type>::try_new(
+                Int8Array::from(vec![Some(0), None, Some(1), Some(0)]),
+                Arc::new(Float32Array::from(vec![Some(1.5), None])),
+            )
+            .unwrap(),
+        );
+        let decoded = normalize_match_key(&dictionary, &DataType::Float64).unwrap();
+        assert_eq!(
+            decoded.as_any().downcast_ref::<Float64Array>().unwrap(),
+            &Float64Array::from(vec![Some(1.5), None, None, Some(1.5)])
+        );
+    }
+
+    #[test]
+    fn binary_keys_compare_bytes_across_offsets_and_dictionaries() {
+        let expected =
+            BinaryArray::from(vec![Some(b"\xff\0".as_slice()), None, Some(b"".as_slice())]);
+        let large: ArrayRef = Arc::new(LargeBinaryArray::from(vec![
+            Some(b"padding".as_slice()),
+            Some(b"\xff\0".as_slice()),
+            None,
+            Some(b"".as_slice()),
+        ]));
+        let output = normalize_match_key(&large.slice(1, 3), &DataType::Binary).unwrap();
+        assert_eq!(
+            output.as_any().downcast_ref::<BinaryArray>().unwrap(),
+            &expected
+        );
+        let restored = normalize_match_key(&output, &DataType::LargeBinary).unwrap();
+        assert_eq!(restored.to_data(), large.slice(1, 3).to_data());
+
+        let fixed: ArrayRef = Arc::new(
+            FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+                [Some(b"ab".as_slice()), None, Some(b"cd".as_slice())].into_iter(),
+                2,
+            )
+            .unwrap(),
+        );
+        for target in [DataType::Binary, DataType::LargeBinary] {
+            let normalized = normalize_match_key(&fixed.slice(1, 2), &target).unwrap();
+            assert_eq!(
+                normalized.to_data(),
+                arrow_cast::cast(&fixed.slice(1, 2), &target)
+                    .unwrap()
+                    .to_data()
+            );
+        }
+        let dictionary: ArrayRef = Arc::new(
+            DictionaryArray::<Int8Type>::try_new(
+                Int8Array::from(vec![Some(0), None, Some(1)]),
+                Arc::new(LargeBinaryArray::from(vec![Some(b"ab".as_slice()), None])),
+            )
+            .unwrap(),
+        );
+        let output = normalize_match_key(&dictionary, &DataType::Binary).unwrap();
+        assert_eq!(
+            output.as_any().downcast_ref::<BinaryArray>().unwrap(),
+            &BinaryArray::from(vec![Some(b"ab".as_slice()), None, None])
+        );
+    }
+
     #[test]
     fn decimal_keys_cannot_round_into_existing_keys() {
         let array = |value| {