This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 7486f7ba [core] Compare FLOAT/DOUBLE partition values like Java (#1190)
7486f7ba is described below
commit 7486f7ba3aefc428c85679052cbed0c7e3ce6e60
Author: JeonDaehong <[email protected]>
AuthorDate: Thu Oct 22 15:42:03 2026 +0900
[core] Compare FLOAT/DOUBLE partition values like Java (#1190)
---
crates/paimon/src/spec/mod.rs | 1 +
crates/paimon/src/spec/predicate.rs | 41 ++++++--
crates/paimon/src/table/partition_filter.rs | 4 +-
crates/paimon/src/table/table_commit.rs | 61 ++++++++++-
crates/paimon/tests/float_partition_filter_test.rs | 116 +++++++++++++++++++++
5 files changed, 209 insertions(+), 14 deletions(-)
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index d800acca..b067858b 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -129,6 +129,7 @@ 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_cmp;
pub(crate) use predicate::like_match;
#[cfg(test)]
pub(crate) use predicate::row_id_leaf;
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index dab8bb18..668e817a 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -2471,6 +2471,27 @@ pub(crate) fn eval_row(predicate: &Predicate, row:
&BinaryRow) -> Result<bool> {
///
/// This function is infallible: all type decoding happens in `extract_datum`
/// before this point, and the operator match is exhaustive.
+/// Comparison as Java's `InternalRowUtils.compare`: FLOAT/DOUBLE follow
+/// `Float.compare` / `Double.compare`, where every NaN is one value above
+/// `+Inf` and `-0.0` is below `0.0`.
+pub(crate) fn java_cmp(lhs: &Datum, rhs: &Datum) -> Option<Ordering> {
+ match (lhs, rhs) {
+ (Datum::Float(a), Datum::Float(b)) => {
+ let canonical = |v: f32| if v.is_nan() { f32::NAN } else { v };
+ Some(canonical(*a).total_cmp(&canonical(*b)))
+ }
+ (Datum::Double(a), Datum::Double(b)) => {
+ let canonical = |v: f64| if v.is_nan() { f64::NAN } else { v };
+ Some(canonical(*a).total_cmp(&canonical(*b)))
+ }
+ _ => datum_cmp(lhs, rhs),
+ }
+}
+
+fn java_eq(lhs: &Datum, rhs: &Datum) -> bool {
+ java_cmp(lhs, rhs) == Some(Ordering::Equal)
+}
+
fn eval_leaf(op: PredicateOperator, datum: Option<&Datum>, literals: &[Datum])
-> bool {
match op {
PredicateOperator::IsNull => datum.is_none(),
@@ -2482,24 +2503,24 @@ fn eval_leaf(op: PredicateOperator, datum:
Option<&Datum>, literals: &[Datum]) -
None => return false,
};
match op {
- PredicateOperator::Eq => literals.first().is_some_and(|lit|
datum_eq(val, lit)),
- PredicateOperator::NotEq => literals.first().is_some_and(|lit|
!datum_eq(val, lit)),
+ PredicateOperator::Eq => literals.first().is_some_and(|lit|
java_eq(val, lit)),
+ PredicateOperator::NotEq => literals.first().is_some_and(|lit|
!java_eq(val, lit)),
PredicateOperator::Lt => {
- literals.first().and_then(|lit| datum_cmp(val, lit)) ==
Some(Ordering::Less)
+ literals.first().and_then(|lit| java_cmp(val, lit)) ==
Some(Ordering::Less)
}
PredicateOperator::LtEq => matches!(
- literals.first().and_then(|lit| datum_cmp(val, lit)),
+ literals.first().and_then(|lit| java_cmp(val, lit)),
Some(Ordering::Less | Ordering::Equal)
),
PredicateOperator::Gt => {
- literals.first().and_then(|lit| datum_cmp(val, lit)) ==
Some(Ordering::Greater)
+ literals.first().and_then(|lit| java_cmp(val, lit)) ==
Some(Ordering::Greater)
}
PredicateOperator::GtEq => matches!(
- literals.first().and_then(|lit| datum_cmp(val, lit)),
+ literals.first().and_then(|lit| java_cmp(val, lit)),
Some(Ordering::Greater | Ordering::Equal)
),
- PredicateOperator::In => literals.iter().any(|lit|
datum_eq(val, lit)),
- PredicateOperator::NotIn => !literals.iter().any(|lit|
datum_eq(val, lit)),
+ PredicateOperator::In => literals.iter().any(|lit|
java_eq(val, lit)),
+ PredicateOperator::NotIn => !literals.iter().any(|lit|
java_eq(val, lit)),
PredicateOperator::StartsWith => match (val, literals.first())
{
(Datum::String(haystack), Some(Datum::String(needle))) => {
haystack.starts_with(needle.as_str())
@@ -2683,11 +2704,11 @@ fn eval_between(value: &Datum, literals: &[Datum]) ->
bool {
unreachable!("BETWEEN must have 2 literals (validated by builder)");
};
let above_low = matches!(
- datum_cmp(value, low),
+ java_cmp(value, low),
Some(Ordering::Greater | Ordering::Equal)
);
let below_high = matches!(
- datum_cmp(value, high),
+ java_cmp(value, high),
Some(Ordering::Less | Ordering::Equal)
);
above_low && below_high
diff --git a/crates/paimon/src/table/partition_filter.rs
b/crates/paimon/src/table/partition_filter.rs
index 07ee5712..ad54f931 100644
--- a/crates/paimon/src/table/partition_filter.rs
+++ b/crates/paimon/src/table/partition_filter.rs
@@ -704,8 +704,10 @@ mod tests {
(
DataType::Double(crate::spec::DoubleType::new()),
vec![Datum::Double(f64::NAN), Datum::Double(1.0)],
+ // Java's Double.compare treats every NaN as equal.
vec![
- (Datum::Double(f64::NAN), false),
+ (Datum::Double(f64::NAN), true),
+ (Datum::Double(f64::from_bits(0xfff8_0000_0000_0000)),
true),
(Datum::Double(1.0), true),
(Datum::Double(2.0), false),
],
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index ce9d8804..87879367 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -24,7 +24,7 @@ use crate::io::FileIO;
use crate::spec::stats::BinaryTableStats;
use crate::spec::FileKind;
use crate::spec::{
- bucket_path, bucket_path_under, extract_datum, merge_active_entries,
BinaryRow,
+ bucket_path, bucket_path_under, extract_datum, java_cmp,
merge_active_entries, BinaryRow,
BinaryRowBuilder, CommitKind, CoreOptions, DataFileMeta, DataType, Datum,
GlobalIndexColumnUpdateAction, IndexManifest, IndexManifestEntry,
Manifest, ManifestEntry,
ManifestFileMeta, ManifestList, PartitionComputer, PartitionStatistics,
Predicate, Snapshot,
@@ -3252,14 +3252,16 @@ impl TableCommit {
let row = BinaryRow::from_serialized_bytes(partition_bytes)?;
for i in 0..num_fields {
match extract_datum(&row, i, &data_types[i])? {
+ // Java order keeps NaN and -0.0 from making these depend
on
+ // the entry order.
Some(datum) => {
mins[i] = Some(match mins[i].take() {
- Some(cur) if cur <= datum => cur,
+ Some(cur) if java_cmp(&cur,
&datum).is_some_and(|o| o.is_le()) => cur,
Some(_) => datum.clone(),
None => datum.clone(),
});
maxs[i] = Some(match maxs[i].take() {
- Some(cur) if cur >= datum => cur,
+ Some(cur) if java_cmp(&cur,
&datum).is_some_and(|o| o.is_ge()) => cur,
Some(_) => datum,
None => datum,
});
@@ -7033,6 +7035,59 @@ mod tests {
assert_eq!(stats.null_counts(), &vec![Some(1)]);
}
+ /// Java's `Double.compare` orders `-0.0` below `0.0` and NaN above all, so
+ /// the bounds must not depend on the order of the entries.
+ #[test]
+ fn compute_partition_stats_orders_floats_like_java() {
+ use crate::spec::{DataType, DoubleType, IntType, Schema};
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("p", DataType::Double(DoubleType::new()))
+ .partition_keys(["p"])
+ .build()
+ .unwrap();
+ let table = Table::new(
+ test_file_io(),
+ Identifier::new("default", "test_table"),
+ "memory:/test_float_partition_stats".to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let commit = TableCommit::new(table, "test-user".to_string());
+ let partition_type = DataType::Double(DoubleType::new());
+ let entry = |value: f64| {
+ let mut builder = BinaryRowBuilder::new(1);
+ builder.write_datum(0, &Datum::Double(value), &partition_type);
+ ManifestEntry::new(
+ FileKind::Add,
+ builder.build_serialized(),
+ 0,
+ 1,
+ test_data_file("data.parquet", 1),
+ 2,
+ )
+ };
+ let bounds = |values: &[f64]| {
+ let entries: Vec<_> = values.iter().map(|value|
entry(*value)).collect();
+ let stats = commit.compute_partition_stats(&entries).unwrap();
+ let min =
BinaryRow::from_serialized_bytes(stats.min_values()).unwrap();
+ let max =
BinaryRow::from_serialized_bytes(stats.max_values()).unwrap();
+ (
+ min.get_double(0).unwrap().to_bits(),
+ max.get_double(0).unwrap().to_bits(),
+ )
+ };
+ let expected = ((-0.0f64).to_bits(), f64::NAN.to_bits());
+ for values in [
+ [0.0, -0.0, f64::NAN, 1.0],
+ [1.0, f64::NAN, 0.0, -0.0],
+ [f64::NAN, 1.0, -0.0, 0.0],
+ [-0.0, 0.0, 1.0, f64::NAN],
+ ] {
+ assert_eq!(bounds(&values), expected, "{values:?}");
+ }
+ }
+
#[tokio::test]
async fn test_manifest_files_roll_by_target_size_and_preserve_entries() {
let file_io = test_file_io();
diff --git a/crates/paimon/tests/float_partition_filter_test.rs
b/crates/paimon/tests/float_partition_filter_test.rs
new file mode 100644
index 00000000..a0eb9aa3
--- /dev/null
+++ b/crates/paimon/tests/float_partition_filter_test.rs
@@ -0,0 +1,116 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Filters on a FLOAT/DOUBLE partition key compare like Java's
+//! `Float.compare` / `Double.compare`: NaN is above every number and equal to
+//! itself, and `-0.0` is not `0.0`.
+
+mod common;
+
+use std::sync::Arc;
+
+use arrow_array::{ArrayRef, Float64Array, Int32Array, RecordBatch};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+use futures::TryStreamExt;
+use paimon::spec::{
+ DataType, Datum, DoubleType, IntType, Predicate, PredicateBuilder, Schema,
TableSchema,
+};
+use paimon::table::Table;
+
+async fn table() -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("p", DataType::Double(DoubleType::new()))
+ .partition_keys(["p"])
+ .build()
+ .unwrap();
+ let path = "memory:/float_partition_filter";
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ write_batch(
+ &table,
+ &RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+ ),
+ (
+ "p",
+ Arc::new(Float64Array::from(vec![0.0, -0.0, f64::NAN, 1.0]))
as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await;
+ table
+}
+
+async fn read_ids(table: &Table, filter: Predicate) -> Vec<i32> {
+ let mut read = table.new_read_builder();
+ read.with_projection(&["id"]).unwrap();
+ read.with_filter(filter);
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids: Vec<i32> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ ids.sort_unstable();
+ ids
+}
+
+#[tokio::test]
+async fn float_partition_filters_follow_java_order() {
+ // Partitions: 0 -> 0.0, 1 -> -0.0, 2 -> NaN, 3 -> 1.0.
+ let table = table().await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let cases = [
+ (
+ pb.greater_than("p", Datum::Double(0.5)).unwrap(),
+ vec![2, 3],
+ ),
+ (pb.equal("p", Datum::Double(f64::NAN)).unwrap(), vec![2]),
+ (
+ pb.not_equal("p", Datum::Double(f64::NAN)).unwrap(),
+ vec![0, 1, 3],
+ ),
+ (pb.equal("p", Datum::Double(0.0)).unwrap(), vec![0]),
+ ];
+ for (predicate, expected) in cases {
+ assert_eq!(
+ read_ids(&table, predicate.clone()).await,
+ expected,
+ "{predicate}"
+ );
+ }
+}
