perf(read): feed sorted runs into the PK merge LoserTree

crates/paimon/src/arrow/residual.rs

@@ -511,8 +511,7 @@ fn evaluate_between_predicate(
     };
     // Delegate the two bound comparisons to `evaluate_column_predicate` rather
     // than calling `arrow_gt_eq`/`arrow_lt_eq` directly, so Between inherits the
-    // type-faithful comparison paths (e.g. signed-byte order for Binary). Using
-    // Arrow's kernels here directly would reintroduce unsigned binary ordering.
+    // type-faithful comparison paths (e.g. byte ordering for Binary).
     let lo_mask = evaluate_column_predicate(array, &low_scalar, PredicateOperator::GtEq)?;
     let hi_mask = evaluate_column_predicate(array, &high_scalar, PredicateOperator::LtEq)?;
     let between = arrow_arith::boolean::and_kleene(&lo_mask, &hi_mask)?;
@@ -703,22 +702,6 @@ fn evaluate_column_predicate(
 ) -> Result<BooleanArray, ArrowError> {
     let scalar = string_scalar_for_column(scalar, column.data_type())?;
 
-    // Binary ordering must match Paimon's Datum::Bytes semantics (Java signed-byte
-    // order, 0xFF < 0x00), which Arrow's unsigned byte comparison does not. Route
-    // ordering ops on Binary/VarBinary columns through the signed comparator.
-    // Eq/NotEq are order-independent, so Arrow's kernels are correct for them.
-    if matches!(
-        column.data_type(),
-        arrow_schema::DataType::Binary | arrow_schema::DataType::LargeBinary
-    ) && matches!(
-        op,
-        PredicateOperator::Lt
-            | PredicateOperator::LtEq
-            | PredicateOperator::Gt
-            | PredicateOperator::GtEq
-    ) {
-        return evaluate_binary_ordering_predicate(column, &scalar, op);
-    }
     match op {
         PredicateOperator::Eq => arrow_eq(column, &scalar),
         PredicateOperator::NotEq => arrow_neq(column, &scalar),
@@ -745,55 +728,6 @@ fn evaluate_column_predicate(
     }
 }
 
-/// Evaluate an ordering predicate (`Lt`/`LtEq`/`Gt`/`GtEq`) on a Binary column
-/// using Paimon's Java signed-byte order, matching `Datum::Bytes` semantics.
-/// `scalar` is a single-element Binary array (the literal). NULL column rows
-/// produce NULL in the mask (later collapsed to `false` by `sanitize_filter_mask`).
-fn evaluate_binary_ordering_predicate(
-    column: &ArrayRef,
-    scalar: &Scalar<ArrayRef>,
-    op: PredicateOperator,
-) -> Result<BooleanArray, ArrowError> {
-    use arrow_array::cast::AsArray;
-    use std::cmp::Ordering;
-
-    // The scalar wraps a length-1 Binary array holding the literal bytes.
-    let (scalar_array, _) = scalar.get();
-    let literal: &[u8] = if let Some(a) = scalar_array.as_binary_opt::<i32>() {
-        a.value(0)
-    } else if let Some(a) = scalar_array.as_binary_opt::<i64>() {
-        a.value(0)
-    } else {
-        return Err(ArrowError::ComputeError(
-            "binary ordering predicate expects a Binary literal".to_string(),
-        ));
-    };
-
-    // Row-wise comparison via the shared signed-byte comparator (single source of
-    // truth with `Datum` ordering).
-    let compare = |bytes: &[u8]| -> bool {
-        let ord = crate::spec::java_bytes_cmp(bytes, literal);
-        match op {
-            PredicateOperator::Lt => ord == Ordering::Less,
-            PredicateOperator::LtEq => ord != Ordering::Greater,
-            PredicateOperator::Gt => ord == Ordering::Greater,
-            PredicateOperator::GtEq => ord != Ordering::Less,
-            _ => unreachable!("only ordering ops reach here"),
-        }
-    };
-
-    let mask: BooleanArray = if let Some(a) = column.as_binary_opt::<i32>() {
-        a.iter().map(|v| v.map(compare)).collect()
-    } else if let Some(a) = column.as_binary_opt::<i64>() {
-        a.iter().map(|v| v.map(compare)).collect()
-    } else {
-        return Err(ArrowError::ComputeError(
-            "binary ordering predicate expects a Binary column".to_string(),
-        ));
-    };
-    Ok(mask)
-}
-
 /// Arrow comparison and pattern kernels reject mismatched string types. The
 /// shared scalar built from Paimon's logical Char/VarChar type is Utf8, while a
 /// decoded file column may be Utf8, LargeUtf8, or Utf8View. Promote the scalar
@@ -1641,10 +1575,9 @@ mod tests {
     }
 
     #[test]
-    fn test_binary_ordering_uses_java_signed_byte_order() {
-        // Paimon Datum::Bytes orders by signed byte (0xFF < 0x00). Arrow's
-        // unsigned comparison would order 0xFF as the largest. Verify the
-        // residual matches Paimon: filter `col > 0x00` must EXCLUDE 0xFF.
+    fn test_binary_ordering_uses_java_unsigned_byte_order() {
+        // Paimon Datum::Bytes and Arrow both order bytes as unsigned values.
+        // Verify the residual keeps 0xFF for `col > 0x00`.
         use crate::spec::BinaryType;
         let col = DataField::new(
             0,
@@ -1659,7 +1592,7 @@ mod tests {
         let values: Vec<Option<&[u8]>> = vec![Some(&[0x00]), Some(&[0x01]), Some(&[0xFF])];
         let batch =
             RecordBatch::try_new(schema, vec![Arc::new(BinaryArray::from(values))]).unwrap();
-        // col > 0x00 : signed order -> only 0x01 (0xFF is negative, < 0x00).
+        // col > 0x00 : unsigned order -> 0x01 and 0xFF.
         let pred = leaf(
             0,
             DataType::Binary(BinaryType::new(1).unwrap()),
@@ -1677,18 +1610,15 @@ mod tests {
         let got: Vec<&[u8]> = (0..out_col.len()).map(|i| out_col.value(i)).collect();
         assert_eq!(
             got,
-            vec![&[0x01u8][..]],
-            "0xFF must be excluded (signed < 0x00)"
+            vec![&[0x01u8][..], &[0xFFu8][..]],
+            "0xFF must be included (unsigned > 0x00)"
         );
     }
 
     #[test]
-    fn test_binary_between_uses_java_signed_byte_order() {
-        // Between must inherit the signed-byte order too (regression: it called
-        // Arrow's unsigned gt_eq/lt_eq directly). `b BETWEEN 0xFF AND 0x01` is,
-        // under signed order, the range [-1, 1] -> keeps 0xFF(-1), 0x00(0),
-        // 0x01(1) and excludes 0x7F(127). Under Arrow's unsigned order it would be
-        // [255, 1] = empty, so this distinguishes the two.
+    fn test_binary_between_uses_java_unsigned_byte_order() {
+        // `b BETWEEN 0x01 AND 0xFF` uses unsigned byte ordering, keeping all
+        // values at or above 0x01 through 0xFF and excluding 0x00.
         use crate::spec::BinaryType;
         let col = DataField::new(
             0,
@@ -1708,7 +1638,7 @@ mod tests {
             0,
             DataType::Binary(BinaryType::new(1).unwrap()),
             PredicateOperator::Between,
-            vec![Datum::Bytes(vec![0xFF]), Datum::Bytes(vec![0x01])],
+            vec![Datum::Bytes(vec![0x01]), Datum::Bytes(vec![0xFF])],
         );
         let fp = file_predicates(vec![pred], vec![col.clone()]);
         let out =
@@ -1721,8 +1651,8 @@ mod tests {
         let got: Vec<&[u8]> = (0..out_col.len()).map(|i| out_col.value(i)).collect();
         assert_eq!(
             got,
-            vec![&[0xFFu8][..], &[0x00u8][..], &[0x01u8][..]],
-            "signed range [-1, 1] keeps 0xFF/0x00/0x01, excludes 0x7F"
+            vec![&[0xFFu8][..], &[0x01u8][..], &[0x7Fu8][..]],
+            "unsigned range [0x01, 0xFF] excludes only 0x00"
         );
     }
 

crates/paimon/src/predicate_stats.rs

@@ -526,7 +526,7 @@ fn coerce_stats_datum_for_predicate(datum: Datum, predicate_data_type: &DataType
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::spec::{IntType, VarCharType};
+    use crate::spec::{BinaryType, IntType, VarCharType};
 
     struct MockStats {
         row_count: i64,
@@ -677,6 +677,34 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn binary_range_pruning_uses_unsigned_byte_order() {
+        let dt = DataType::Binary(BinaryType::new(1).unwrap());
+        let stats = MockStats {
+            row_count: 10,
+            null_count: Some(0),
+            min: Some(Datum::Bytes(vec![0x80])),
+            max: Some(Datum::Bytes(vec![0xFF])),
+        };
+
+        assert!(data_leaf_may_match(
+            0,
+            &dt,
+            &dt,
+            PredicateOperator::Gt,
+            &[Datum::Bytes(vec![0x00])],
+            &stats,
+        ));
+        assert!(!data_leaf_may_match(
+            0,
+            &dt,
+            &dt,
+            PredicateOperator::Lt,
+            &[Datum::Bytes(vec![0x00])],
+            &stats,
+        ));
+    }
+
     fn int_stats(min: i32, max: i32) -> MockStats {
         MockStats {
             row_count: 10,

crates/paimon/src/spec/mod.rs

@@ -99,7 +99,6 @@ mod predicate;
 pub(crate) use predicate::datum_cmp;
 pub(crate) use predicate::eval_row;
 pub(crate) use predicate::extract_datum;
-pub(crate) use predicate::java_bytes_cmp;
 pub(crate) use predicate::like_match;
 pub use predicate::{
     field_idx_to_partition_idx, Datum, Predicate, PredicateBuilder, PredicateOperator, Transform,

crates/paimon/src/spec/predicate.rs

@@ -210,16 +210,10 @@ fn decimal_cmp(ua: i128, sa: u32, ub: i128, sb: u32) -> Option<Ordering> {
     na.partial_cmp(&nb)
 }
 
-/// Match Java `CompareUtils.compare(byte[], byte[])`, which compares signed
-/// bytes lexicographically.
-pub(crate) fn java_bytes_cmp(a: &[u8], b: &[u8]) -> Ordering {
-    for (&lhs, &rhs) in a.iter().zip(b.iter()) {
-        let cmp = (lhs as i8).cmp(&(rhs as i8));
-        if cmp != Ordering::Equal {
-            return cmp;
-        }
-    }
-    a.len().cmp(&b.len())
+/// Match Java `CompareUtils.compare(byte[], byte[])`, which compares bytes as
+/// unsigned values lexicographically.
+fn java_bytes_cmp(a: &[u8], b: &[u8]) -> Ordering {
+    a.cmp(b)
 }
 
 /// 10^exp as i128.  Returns i128::MAX for exponents that would overflow.
@@ -2286,8 +2280,8 @@ mod tests {
     }
 
     #[test]
-    fn test_datum_partial_ord_bytes_matches_java_signed_byte_order() {
-        assert!(Datum::Bytes(vec![0xFF]) < Datum::Bytes(vec![0x00]));
+    fn test_datum_partial_ord_bytes_matches_java_unsigned_byte_order() {
+        assert!(Datum::Bytes(vec![0x00]) < Datum::Bytes(vec![0xFF]));
     }
 
     #[test]