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 e2f1973 fix(core): align BinaryRow timestamp write path with
euclidean representation (#485)
e2f1973 is described below
commit e2f1973bcc4a6359b3b0eefba3c67915e36f0e24
Author: Wuhen- Li <[email protected]>
AuthorDate: Thu Jul 9 10:06:19 2026 +0800
fix(core): align BinaryRow timestamp write path with euclidean
representation (#485)
The BinaryRow write/extract path converted microsecond timestamps with
truncating division while the DataFusion literal pushdown path uses
euclidean division. For pre-epoch sub-millisecond values the two paths
disagreed: e.g. -1us was stored as (millis=0, nanos=-1000) but the pushed
literal became (millis=-1, nanos=999_000), making pushed equality/range
predicates false-negative.
Switch the three write-path spots in binary_row.rs to div_euclid/rem_euclid
so stored timestamps match the pushdown representation and Paimon Java's
canonical (millisecond, nanoOfMillisecond in [0, 999999]) form.
Add a unit test (fails pre-fix) verifying the write path normalizes a
negative sub-millisecond timestamp, plus an end-to-end DataFusion scan test
covering pushed equality/range predicates on pre-epoch fractional
timestamps.
---
.../integrations/datafusion/tests/read_tables.rs | 81 ++++++++++++++++++++++
crates/paimon/src/spec/binary_row.rs | 67 ++++++++++++++++--
2 files changed, 142 insertions(+), 6 deletions(-)
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index f2b8894..cb9a68f 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -499,6 +499,87 @@ async fn
test_temporal_filter_pushdown_via_datafusion_scan() {
);
}
+/// Regression test for pre-epoch, sub-millisecond timestamps.
+///
+/// The BinaryRow write path and the DataFusion literal pushdown path must
+/// agree on the `(millis, nanos)` representation of a timestamp. Previously
the
+/// write path used truncating division while pushdown used euclidean division,
+/// so `1969-12-31 23:59:59.999999` (`-1us`) was stored as `(0, -1000)` but the
+/// pushed literal became `(-1, 999_000)`. A pushed equality/range predicate on
+/// such a value was therefore false-negative and dropped matching rows.
+#[tokio::test]
+async fn test_negative_temporal_filter_pushdown_via_datafusion_scan() {
+ let (_tmp, sql_context) = common::setup_sql_context().await;
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.negative_temporal_pushdown (
+ id INT,
+ name STRING,
+ ts TIMESTAMP(6)
+ )",
+ )
+ .await
+ .expect("CREATE TABLE should succeed")
+ .collect()
+ .await
+ .expect("CREATE TABLE should collect");
+ // id=1 sits 1us before the epoch (negative micros), id=2 sits at the
epoch,
+ // id=3 is a normal post-epoch value. Each row is written in its own
INSERT so
+ // it lands in a separate data file, forcing per-file min/max stats
pruning to
+ // rely on the pushed predicate rather than a range that happens to
overlap.
+ for values in [
+ "(1, 'neg_us', TIMESTAMP '1969-12-31 23:59:59.999999')",
+ "(2, 'epoch', TIMESTAMP '1970-01-01 00:00:00.000000')",
+ "(3, 'post', TIMESTAMP '1970-01-01 00:00:00.000001')",
+ ] {
+ sql_context
+ .sql(&format!(
+ "INSERT INTO paimon.test_db.negative_temporal_pushdown VALUES
{values}"
+ ))
+ .await
+ .expect("INSERT should succeed")
+ .collect()
+ .await
+ .expect("INSERT should collect");
+ }
+
+ // Equality on the pre-epoch fractional timestamp must return exactly id=1.
+ let eq_sql = "SELECT id, name FROM
paimon.test_db.negative_temporal_pushdown \
+ WHERE ts = TIMESTAMP '1969-12-31 23:59:59.999999'";
+ let plan = sql_context
+ .sql(eq_sql)
+ .await
+ .expect("SQL planning should succeed")
+ .create_physical_plan()
+ .await
+ .expect("Physical plan creation should succeed");
+ let plan_text = format_physical_plan(&plan);
+ let scan_lines = paimon_scan_lines(&plan_text);
+ assert!(
+ scan_lines
+ .iter()
+ .any(|line| line.contains("predicate=ts = TS(")),
+ "Temporal predicate should be pushed into PaimonTableScan,
plan:\n{plan_text}"
+ );
+
+ let rows = common::collect_id_name(&sql_context, eq_sql).await;
+ assert_eq!(
+ rows,
+ vec![(1, "neg_us".to_string())],
+ "Pushed equality on a pre-epoch sub-millisecond timestamp must match
the stored row"
+ );
+
+ // Range predicate across the epoch boundary must keep only the pre-epoch
row.
+ let range_sql = "SELECT id, name FROM
paimon.test_db.negative_temporal_pushdown \
+ WHERE ts < TIMESTAMP '1970-01-01 00:00:00.000000'";
+ let rows = common::collect_id_name(&sql_context, range_sql).await;
+ assert_eq!(
+ rows,
+ vec![(1, "neg_us".to_string())],
+ "Pushed range predicate must order pre-epoch fractional timestamps
correctly"
+ );
+}
+
#[tokio::test]
async fn test_limit_pushdown_on_data_evolution_table_returns_merged_rows() {
let batches = collect_query("SELECT id, name FROM
paimon.default.data_evolution_table LIMIT 3")
diff --git a/crates/paimon/src/spec/binary_row.rs
b/crates/paimon/src/spec/binary_row.rs
index cc6601b..98bb1c2 100644
--- a/crates/paimon/src/spec/binary_row.rs
+++ b/crates/paimon/src/spec/binary_row.rs
@@ -799,8 +799,8 @@ pub fn extract_datum_from_arrow(
.ok_or_else(|| type_mismatch_err("Timestamp(us)",
col_idx))?;
let micros = arr.value(row_idx);
Datum::Timestamp {
- millis: micros / 1000,
- nanos: ((micros % 1000) * 1000) as i32,
+ millis: micros.div_euclid(1_000),
+ nanos: (micros.rem_euclid(1_000) * 1_000) as i32,
}
}
}
@@ -821,8 +821,8 @@ pub fn extract_datum_from_arrow(
.ok_or_else(||
type_mismatch_err("LocalZonedTimestamp(us)", col_idx))?;
let micros = arr.value(row_idx);
Datum::LocalZonedTimestamp {
- millis: micros / 1000,
- nanos: ((micros % 1000) * 1000) as i32,
+ millis: micros.div_euclid(1_000),
+ nanos: (micros.rem_euclid(1_000) * 1_000) as i32,
}
}
}
@@ -1264,8 +1264,8 @@ fn write_typed_value(
builder.set_null_at(pos);
} else {
let micros = arr.value(row_idx);
- let millis = micros / 1000;
- let nanos = ((micros % 1000) * 1000) as i32;
+ let millis = micros.div_euclid(1_000);
+ let nanos = (micros.rem_euclid(1_000) * 1_000) as i32;
builder.write_timestamp_non_compact(pos, millis, nanos);
}
}
@@ -1738,4 +1738,59 @@ mod tests {
);
}
}
+
+ #[test]
+ fn test_negative_sub_millisecond_timestamp_uses_euclidean_parts() {
+ use arrow_array::TimestampMicrosecondArray;
+ use arrow_schema::{DataType as ArrowDT, Field, Schema, TimeUnit};
+ use std::sync::Arc;
+
+ // 1 microsecond before the epoch: 1969-12-31 23:59:59.999999.
+ // Truncating division would store (millis=0, nanos=-1000), which is
+ // inconsistent with the DataFusion literal pushdown path that uses
+ // euclidean division and produces (millis=-1, nanos=999_000). The
+ // mismatch made pushed predicates on pre-epoch fractional timestamps
+ // false-negative, so the write path must normalize the same way.
+ let micros: i64 = -1;
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "ts",
+ ArrowDT::Timestamp(TimeUnit::Microsecond, None),
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![Arc::new(TimestampMicrosecondArray::from(vec![Some(
+ micros,
+ )]))],
+ )
+ .unwrap();
+
+ let ts_type =
DataType::Timestamp(crate::spec::TimestampType::new(6).unwrap());
+ let fields = vec![crate::spec::DataField::new(0, "ts".into(),
ts_type.clone())];
+ let indices = vec![0];
+
+ // Single-row extraction path (`extract_datum_from_arrow`).
+ let datum = extract_datum_from_arrow(&batch, 0, 0, &ts_type)
+ .unwrap()
+ .unwrap();
+ assert_eq!(
+ datum,
+ Datum::Timestamp {
+ millis: -1,
+ nanos: 999_000,
+ },
+ "extract_datum_from_arrow should normalize negative
sub-millisecond timestamps"
+ );
+
+ // Batch write path must persist the same euclidean parts so that a
+ // pushed-down literal compares equal to the stored value.
+ let row = BinaryRow::from_arrow(&batch, 0, &indices, &fields).unwrap();
+ let (millis, nanos) = row.get_timestamp_raw(0, 6).unwrap();
+ assert_eq!(
+ (millis, nanos),
+ (-1, 999_000),
+ "binary-row write path must store euclidean timestamp parts"
+ );
+ }
}
