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 ae858eae [core] Align BIGINT and timestamp schema evolution with Java
(#1107)
ae858eae is described below
commit ae858eae590cc0d0aeeeda9904ec6e974641f7f7
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 12 10:59:02 2026 +0800
[core] Align BIGINT and timestamp schema evolution with Java (#1107)
---
crates/paimon/src/arrow/evolution_cast.rs | 63 ++++
crates/paimon/src/arrow/nested_evolution.rs | 354 ++++++++++++++++++----
crates/paimon/tests/schema_evolution_cast_test.rs | 173 ++++++++++-
3 files changed, 528 insertions(+), 62 deletions(-)
diff --git a/crates/paimon/src/arrow/evolution_cast.rs
b/crates/paimon/src/arrow/evolution_cast.rs
index 5a980386..4648dd6f 100644
--- a/crates/paimon/src/arrow/evolution_cast.rs
+++ b/crates/paimon/src/arrow/evolution_cast.rs
@@ -83,6 +83,69 @@ pub(super) fn numeric_to_integer(source: &ArrayRef, target:
&ArrowType) -> crate
Ok(result)
}
+/// TimestampToNumericPrimitiveCastRule returns floor-divided epoch seconds,
+/// rather than Arrow's physical timestamp ticks. LTZ uses Java Timestamp's
+/// local-date-time round trip, whose epoch day is an int.
+pub(super) fn timestamp_to_bigint(source: &ArrayRef, local_zoned: bool) ->
crate::Result<ArrayRef> {
+ let ArrowType::Timestamp(unit, _) = source.data_type() else {
+ unreachable!("timestamp source validated by evolution dispatch")
+ };
+ let raw = arrow_cast::cast(source, &ArrowType::Int64).map_err(cast_error)?;
+ let values =
Int64Array::from_iter(raw.as_primitive::<Int64Type>().iter().map(|value| {
+ value.and_then(|value| {
+ if !local_zoned {
+ return Some(value.div_euclid(ticks_per_second(unit)));
+ }
+ let millis = match unit {
+ TimeUnit::Second => value.checked_mul(1_000)?,
+ TimeUnit::Millisecond => value,
+ TimeUnit::Microsecond => value.div_euclid(1_000),
+ TimeUnit::Nanosecond => value.div_euclid(1_000_000),
+ };
+ // Timestamp.toLocalDateTime casts the epoch day to int, then
+ // Timestamp.fromLocalDateTime reconstructs its milliseconds.
+ let day = millis.div_euclid(86_400_000) as i32;
+ let millis = i64::from(day) * 86_400_000 +
millis.rem_euclid(86_400_000);
+ Some(millis.div_euclid(1_000))
+ })
+ }));
+ Ok(Arc::new(values))
+}
+
+/// Java NumericPrimitiveToTimestamp multiplies long seconds by 1000 with
+/// signed-long wrapping. Only after that conversion do we rebuild Arrow's
+/// declared timestamp unit, returning NULL if its storage cannot hold it.
+pub(super) fn numeric_epoch_seconds_to_timestamp(
+ source: &ArrayRef,
+ target_arrow: &ArrowType,
+) -> crate::Result<ArrayRef> {
+ let ArrowType::Timestamp(unit, None) = target_arrow else {
+ unreachable!("ordinary timestamp target validated by evolution
dispatch")
+ };
+ let raw = arrow_cast::cast(source, &ArrowType::Int64).map_err(cast_error)?;
+ let values =
Int64Array::from_iter(raw.as_primitive::<Int64Type>().iter().map(|value| {
+ value.and_then(|value| {
+ let millis = value.wrapping_mul(1_000);
+ match unit {
+ TimeUnit::Second => Some(millis.div_euclid(1_000)),
+ TimeUnit::Millisecond => Some(millis),
+ TimeUnit::Microsecond => millis.checked_mul(1_000),
+ TimeUnit::Nanosecond => millis.checked_mul(1_000_000),
+ }
+ })
+ }));
+ arrow_cast::cast(&(Arc::new(values) as ArrayRef),
target_arrow).map_err(cast_error)
+}
+
+fn ticks_per_second(unit: &TimeUnit) -> i64 {
+ match unit {
+ TimeUnit::Second => 1,
+ TimeUnit::Millisecond => 1_000,
+ TimeUnit::Microsecond => 1_000_000,
+ TimeUnit::Nanosecond => 1_000_000_000,
+ }
+}
+
/// Match DateTimeUtils.truncate for the same timestamp kind. Work in i128 so
/// flooring a pre-epoch value cannot overflow the source's i64 unit before
/// converting to a coarser target unit. Cross-kind timezone casts are handled
diff --git a/crates/paimon/src/arrow/nested_evolution.rs
b/crates/paimon/src/arrow/nested_evolution.rs
index e3fad1d3..883b89c3 100644
--- a/crates/paimon/src/arrow/nested_evolution.rs
+++ b/crates/paimon/src/arrow/nested_evolution.rs
@@ -34,8 +34,7 @@
use std::sync::Arc;
use arrow_array::{
- new_null_array, Array, ArrayRef, Int64Array, ListArray, MapArray,
StringArray, StructArray,
- UInt32Array,
+ new_null_array, Array, ArrayRef, ListArray, MapArray, StringArray,
StructArray, UInt32Array,
};
use arrow_cast::cast;
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Fields};
@@ -205,20 +204,24 @@ pub(crate) fn evolve_column(
_ => {}
}
- // Java `NumericPrimitiveToTimestamp` reads an INTEGER/BIGINT column as
epoch
- // *seconds* (`value.longValue() * 1000` millis, fed to `toLocalDateTime`);
- // Arrow's Int64->Timestamp cast instead reinterprets the raw value
directly
- // in the column's unit, so a BIGINT epoch column evolved to TIMESTAMP(6)
- // would read 10^6x too small (~1970). Scale seconds into the target unit
- // first. Scoped to INTEGER/BIGINT into TIMESTAMP *without* local time
zone:
- // Java's `create()` returns null for every other numeric source, and its
- // TIMESTAMP-with-local-zone target resolves the offset with
- // `ZoneId.systemDefault()` (environment dependent) -- both are left to the
- // Arrow fallthrough below rather than guessed at here.
- if matches!(source_type, DataType::Int(_) | DataType::BigInt(_)) {
- if let DataType::Timestamp(ts) = target_type {
- return numeric_epoch_seconds_to_timestamp(source, &target_arrow,
ts.precision());
- }
+ // Apply the epoch-seconds rule at every leaf, including nested containers.
+ // Numeric -> LTZ depends on Java's system time zone and retains its
existing
+ // path; ordinary TIMESTAMP has no such runtime dependency.
+ if matches!(source_type, DataType::Int(_) | DataType::BigInt(_))
+ && matches!(target_type, DataType::Timestamp(_))
+ {
+ return
super::evolution_cast::numeric_epoch_seconds_to_timestamp(source,
&target_arrow);
+ }
+ if matches!(target_type, DataType::BigInt(_))
+ && matches!(
+ source_type,
+ DataType::Timestamp(_) | DataType::LocalZonedTimestamp(_)
+ )
+ {
+ return super::evolution_cast::timestamp_to_bigint(
+ source,
+ matches!(source_type, DataType::LocalZonedTimestamp(_)),
+ );
}
if matches!(
@@ -252,49 +255,6 @@ pub(crate) fn evolve_column(
})
}
-/// Evolve an INTEGER/BIGINT column into a TIMESTAMP (without local time zone)
-/// the way Java `NumericPrimitiveToTimestamp` does: interpret each value as
-/// epoch seconds and scale it into the target time unit. Nulls are preserved,
-/// and a value that overflows `i64` in the target unit becomes NULL, matching
-/// the safe-cast overflow behavior of the surrounding evolution path.
-fn numeric_epoch_seconds_to_timestamp(
- source: &ArrayRef,
- target_arrow: &ArrowDataType,
- precision: u32,
-) -> crate::Result<ArrayRef> {
- // Ticks per second for the target unit (precision buckets match
- // `paimon_type_to_arrow`'s `timestamp_time_unit`).
- let ticks_per_second: i64 = match precision {
- 0 => 1,
- 1..=3 => 1_000,
- 4..=6 => 1_000_000,
- _ => 1_000_000_000,
- };
- let as_i64 =
- cast(source, &ArrowDataType::Int64).map_err(|e|
crate::Error::UnexpectedError {
- message: format!(
- "failed to widen {:?} to Int64 before scaling epoch seconds to
timestamp",
- source.data_type()
- ),
- source: Some(Box::new(e)),
- })?;
- let seconds = as_i64
- .as_any()
- .downcast_ref::<Int64Array>()
- .expect("cast to Int64 yields an Int64Array");
- // Scale seconds into the unit; a null stays null and an overflow becomes
null.
- let scaled: Int64Array = seconds
- .iter()
- .map(|value| value.and_then(|v| v.checked_mul(ticks_per_second)))
- .collect();
- // Reinterpret the already-scaled value straight into the timestamp unit
- // (Arrow's Int64->Timestamp copies it verbatim, no further scaling).
- cast(&(Arc::new(scaled) as ArrayRef), target_arrow).map_err(|e|
crate::Error::UnexpectedError {
- message: format!("failed to build {target_arrow:?} from scaled epoch
seconds"),
- source: Some(Box::new(e)),
- })
-}
-
/// Match Paimon's constructed-value string form, including nested values and
/// NULL containers. Arrow's generic cast cannot convert ROW/ARRAY/MAP to Utf8.
fn render_string_values(
@@ -940,6 +900,282 @@ mod tests {
assert_eq!(millis.value(0), secs * 1_000);
}
+ #[test]
+ fn evolves_timestamp_to_bigint_as_floored_epoch_seconds() {
+ for (precision, ticks) in [(0, 1_i64), (3, 1_000), (6, 1_000_000), (9,
1_000_000_000)] {
+ for local in [false, true] {
+ let source_type = if local {
+ DataType::LocalZonedTimestamp(
+
crate::spec::LocalZonedTimestampType::new(precision).unwrap(),
+ )
+ } else {
+ DataType::Timestamp(TimestampType::new(precision).unwrap())
+ };
+ let fraction = if precision == 0 { 0 } else { 1 };
+ let raw: ArrayRef = Arc::new(Int64Array::from(vec![
+ None,
+ Some(ticks + fraction),
+ Some(-1),
+ Some(-ticks - fraction),
+ Some(0),
+ Some((i64::from(i32::MAX) + 3) * ticks),
+ None,
+ ]));
+ let source = cast(&raw,
&paimon_type_to_arrow(&source_type).unwrap())
+ .unwrap()
+ .slice(1, 6);
+ let out =
+ evolve_column(&source, &source_type,
&DataType::BigInt(BigIntType::new()))
+ .unwrap();
+ let values: Vec<_> = out
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .iter()
+ .collect();
+ assert_eq!(
+ values,
+ vec![
+ Some(1),
+ Some(-1),
+ Some(if precision == 0 { -1 } else { -2 }),
+ Some(0),
+ Some(i64::from(i32::MAX) + 3),
+ None
+ ],
+ "precision={precision} local={local}"
+ );
+ }
+ }
+ }
+
+ #[test]
+ fn
bigint_epoch_seconds_wrap_in_millis_before_timestamp_conversion_like_java() {
+ let source: ArrayRef = Arc::new(Int64Array::from(vec![
+ Some(i64::MAX),
+ Some(i64::MIN),
+ Some(i64::MAX - 1),
+ Some(42),
+ Some(-42),
+ None,
+ ]));
+ for (precision, ticks) in [(0, 1_i64), (3, 1_000), (6, 1_000_000), (9,
1_000_000_000)] {
+ let target =
DataType::Timestamp(TimestampType::new(precision).unwrap());
+ let out =
+ evolve_column(&source, &DataType::BigInt(BigIntType::new()),
&target).unwrap();
+ let raw = cast(&out, &ArrowDataType::Int64).unwrap();
+ let values: Vec<_> = raw
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .iter()
+ .collect();
+ assert_eq!(
+ values,
+ vec![
+ Some(-ticks),
+ Some(0),
+ Some(-2 * ticks),
+ Some(42 * ticks),
+ Some(-42 * ticks),
+ None
+ ],
+ "precision={precision}"
+ );
+ }
+ }
+
+ #[test]
+ fn timestamp_to_bigint_matches_java_extreme_millis_round_trip() {
+ for (local, expected) in [
+ (
+ false,
+ vec![
+ Some(9_223_372_036_854_775),
+ Some(-9_223_372_036_854_776),
+ None,
+ ],
+ ),
+ (
+ true,
+ vec![Some(-53_757_322_505_225), Some(53_757_322_505_224),
None],
+ ),
+ ] {
+ let source_type = if local {
+
DataType::LocalZonedTimestamp(crate::spec::LocalZonedTimestampType::new(3).unwrap())
+ } else {
+ DataType::Timestamp(TimestampType::new(3).unwrap())
+ };
+ let raw: ArrayRef =
+ Arc::new(Int64Array::from(vec![Some(i64::MAX), Some(i64::MIN),
None]));
+ let source = cast(&raw,
&paimon_type_to_arrow(&source_type).unwrap()).unwrap();
+ let output =
+ evolve_column(&source, &source_type,
&DataType::BigInt(BigIntType::new())).unwrap();
+ assert_eq!(
+ output
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>(),
+ expected,
+ "local={local}"
+ );
+ }
+ }
+
+ #[test]
+ fn bigint_to_timestamp_keeps_lenient_arrow_storage_overflow() {
+ let source: ArrayRef = Arc::new(Int64Array::from(vec![
+ Some(9_223_372_036_854_775),
+ Some(-9_223_372_036_854_775),
+ None,
+ ]));
+ for (precision, expected) in [
+ (
+ 0,
+ vec![
+ Some(9_223_372_036_854_775),
+ Some(-9_223_372_036_854_775),
+ None,
+ ],
+ ),
+ (
+ 3,
+ vec![
+ Some(9_223_372_036_854_775_000),
+ Some(-9_223_372_036_854_775_000),
+ None,
+ ],
+ ),
+ (6, vec![None, None, None]),
+ (9, vec![None, None, None]),
+ ] {
+ let output = evolve_column(
+ &source,
+ &DataType::BigInt(BigIntType::new()),
+ &DataType::Timestamp(TimestampType::new(precision).unwrap()),
+ )
+ .unwrap();
+ let raw = cast(&output, &ArrowDataType::Int64).unwrap();
+ assert_eq!(
+ raw.as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>(),
+ expected,
+ "precision={precision}"
+ );
+ }
+ }
+
+ #[test]
+ fn temporal_evolution_recurses_through_nullable_rows_lists_and_maps() {
+ for reverse in [false, true] {
+ let timestamp =
DataType::Timestamp(TimestampType::new(6).unwrap());
+ let bigint = DataType::BigInt(BigIntType::new());
+ let (source_type, target_type, raw, expected) = if reverse {
+ (
+ bigint,
+ timestamp,
+ vec![Some(42), Some(i64::MAX), None],
+ vec![Some(42_000_000), Some(-1_000_000), None],
+ )
+ } else {
+ (
+ timestamp,
+ bigint,
+ vec![Some(42_000_001), Some(-1), None],
+ vec![Some(42), Some(-1), None],
+ )
+ };
+ let raw: ArrayRef = Arc::new(Int64Array::from(raw));
+ let leaf = cast(&raw,
&paimon_type_to_arrow(&source_type).unwrap()).unwrap();
+ let validity = Some(NullBuffer::from(vec![true, false, true]));
+ let values = |array: &ArrayRef| {
+ cast(array, &ArrowDataType::Int64)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>()
+ };
+
+ let source_row = row(vec![field(11, "v", source_type.clone())]);
+ let target_row = row(vec![field(11, "v", target_type.clone())]);
+ let ArrowDataType::Struct(fields) =
paimon_type_to_arrow(&source_row).unwrap() else {
+ panic!()
+ };
+ let source: ArrayRef = Arc::new(
+ StructArray::try_new(fields, vec![leaf.clone()],
validity.clone()).unwrap(),
+ );
+ let out = evolve_column(&source, &source_row,
&target_row).unwrap();
+ let out = as_struct(&out);
+ assert_eq!(out.nulls(), source.nulls());
+ assert_eq!(values(out.column(0)), expected, "ROW
reverse={reverse}");
+
+ let source: ArrayRef = Arc::new(
+ ListArray::try_new(
+ Arc::new(ArrowField::new("element",
leaf.data_type().clone(), true)),
+ arrow_buffer::OffsetBuffer::new(vec![0, 2, 2, 3].into()),
+ leaf.clone(),
+ validity.clone(),
+ )
+ .unwrap(),
+ );
+ let out = evolve_column(
+ &source,
+ &DataType::Array(ArrayType::new(source_type.clone())),
+ &DataType::Array(ArrayType::new(target_type.clone())),
+ )
+ .unwrap();
+ let out = out.as_any().downcast_ref::<ListArray>().unwrap();
+ assert_eq!(out.nulls(), source.nulls());
+ assert_eq!(out.value_offsets(), &[0, 2, 2, 3]);
+ assert_eq!(values(out.values()), expected, "ARRAY
reverse={reverse}");
+
+ let fields = Fields::from(vec![
+ ArrowField::new("key", ArrowDataType::Utf8, false),
+ ArrowField::new("value", leaf.data_type().clone(), true),
+ ]);
+ let entries = StructArray::try_new(
+ fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["positive", "negative",
"null"])) as ArrayRef,
+ leaf,
+ ],
+ None,
+ )
+ .unwrap();
+ let source: ArrayRef = Arc::new(
+ MapArray::try_new(
+ Arc::new(ArrowField::new(
+ "entries",
+ ArrowDataType::Struct(fields),
+ false,
+ )),
+ arrow_buffer::OffsetBuffer::new(vec![0, 2, 2, 3].into()),
+ entries,
+ validity,
+ false,
+ )
+ .unwrap(),
+ );
+ let out = evolve_column(
+ &source,
+ &DataType::Map(MapType::new(string_type(), source_type)),
+ &DataType::Map(MapType::new(string_type(), target_type)),
+ )
+ .unwrap();
+ let out = out.as_any().downcast_ref::<MapArray>().unwrap();
+ assert_eq!(out.nulls(), source.nulls());
+ assert_eq!(out.value_offsets(), &[0, 2, 2, 3]);
+ assert_eq!(values(out.values()), expected, "MAP
reverse={reverse}");
+ }
+ }
+
#[test]
fn evolves_int_epoch_seconds_to_timestamp_like_bigint() {
// Java switches INTEGER and BIGINT together in the same cast rule.
diff --git a/crates/paimon/tests/schema_evolution_cast_test.rs
b/crates/paimon/tests/schema_evolution_cast_test.rs
index 6b3b63f5..de4c1ccd 100644
--- a/crates/paimon/tests/schema_evolution_cast_test.rs
+++ b/crates/paimon/tests/schema_evolution_cast_test.rs
@@ -20,12 +20,13 @@ mod common;
use std::sync::Arc;
use arrow_array::{
- ArrayRef, Int32Array, Int64Array, Int8Array, RecordBatch,
TimestampMillisecondArray,
+ Array, ArrayRef, Int32Array, Int64Array, Int8Array, RecordBatch,
TimestampMicrosecondArray,
+ TimestampMillisecondArray,
};
use futures::TryStreamExt;
use paimon::spec::{
- BigIntType, DataType, Datum, IntType, Predicate, PredicateBuilder, Schema,
SchemaChange,
- TableSchema, TimestampType, TinyIntType,
+ BigIntType, DataType, Datum, IntType, LocalZonedTimestampType, Predicate,
PredicateBuilder,
+ Schema, SchemaChange, TableSchema, TimestampType, TinyIntType,
};
use paimon::table::Table;
@@ -178,3 +179,169 @@ async fn
converted_predicates_filter_after_schema_evolution_for_every_table_mode
);
}
}
+
+#[tokio::test]
+async fn
epoch_seconds_evolution_reads_mixed_files_and_filters_for_every_table_mode() {
+ for mode in ["append", "pk", "de"] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("t", DataType::Timestamp(TimestampType::new(6).unwrap()))
+ .column(
+ "l",
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(6).unwrap()),
+ )
+ .column("s", DataType::BigInt(BigIntType::new()))
+ .option("read.batch-size", "1");
+ if mode == "pk" {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ }
+ if mode == "de" {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ let (io, old) = memory_table(
+ &format!("memory:/epoch_seconds_{mode}"),
+ TableSchema::new(0, &schema.build().unwrap()),
+ );
+ setup_dirs(&io, old.location()).await;
+ persist_table_schema(&io, old.location(), old.schema()).await;
+ let times = TimestampMicrosecondArray::from(vec![
+ Some(42_000_001),
+ Some(-1),
+ None,
+ Some(42_000_001),
+ ]);
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as ArrayRef,
+ ),
+ ("t", Arc::new(times.clone()) as ArrayRef),
+ ("l", Arc::new(times.with_timezone("UTC")) as ArrayRef),
+ (
+ "s",
+ Arc::new(Int64Array::from(vec![
+ Some(42),
+ Some(-42),
+ None,
+ Some(i64::MAX),
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ write_batch(&old, &batch.slice(0, 3)).await;
+ // A single-value old file must survive stats pruning in the new units.
+ write_batch(&old, &batch.slice(3, 1)).await;
+ let schema = old
+ .schema()
+ .apply_changes(vec![
+ SchemaChange::update_column_type("t".into(),
DataType::BigInt(BigIntType::new())),
+ SchemaChange::update_column_type("l".into(),
DataType::BigInt(BigIntType::new())),
+ SchemaChange::update_column_type(
+ "s".into(),
+ DataType::Timestamp(TimestampType::new(6).unwrap()),
+ ),
+ ])
+ .unwrap();
+ persist_table_schema(&io, old.location(), &schema).await;
+ let current = old.copy_with_resolved_schema(schema, "main").unwrap();
+ write_batch(
+ ¤t,
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![5])) as ArrayRef),
+ ("t", Arc::new(Int64Array::from(vec![42])) as ArrayRef),
+ ("l", Arc::new(Int64Array::from(vec![42])) as ArrayRef),
+ (
+ "s",
+
Arc::new(TimestampMicrosecondArray::from(vec![42_000_000])) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await;
+
+ let builder = current.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in batches {
+ let raw = batch
+ .columns()
+ .iter()
+ .skip(1)
+ .map(|array| arrow_cast::cast(array,
&arrow_schema::DataType::Int64).unwrap())
+ .collect::<Vec<_>>();
+ let values = raw
+ .iter()
+ .map(|array|
array.as_any().downcast_ref::<Int64Array>().unwrap())
+ .collect::<Vec<_>>();
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for index in 0..batch.num_rows() {
+ let value = |column: usize| {
+ values[column]
+ .is_valid(index)
+ .then(|| values[column].value(index))
+ };
+ rows.push((ids.value(index), value(0), value(1), value(2)));
+ }
+ }
+ rows.sort_unstable_by_key(|row| row.0);
+ assert_eq!(
+ rows,
+ vec![
+ (1, Some(42), Some(42), Some(42_000_000)),
+ (2, Some(-1), Some(-1), Some(-42_000_000)),
+ (3, None, None, None),
+ (4, Some(42), Some(42), Some(-1_000_000)),
+ (5, Some(42), Some(42), Some(42_000_000)),
+ ],
+ "{mode}"
+ );
+ let pb = PredicateBuilder::new(current.schema().fields());
+ for name in ["t", "l"] {
+ assert_eq!(
+ ids(¤t, pb.equal(name, Datum::Long(42)).unwrap()).await,
+ vec![1, 4, 5],
+ "{mode} {name} predicate-only field"
+ );
+ assert_eq!(
+ ids(¤t, pb.less_than(name,
Datum::Long(0)).unwrap()).await,
+ vec![2],
+ "{mode} {name} pre-epoch"
+ );
+ }
+ assert_eq!(
+ ids(
+ ¤t,
+ pb.equal(
+ "s",
+ Datum::Timestamp {
+ millis: -1000,
+ nanos: 0
+ }
+ )
+ .unwrap()
+ )
+ .await,
+ vec![4],
+ "{mode} wrapped single-value file"
+ );
+ assert_eq!(
+ ids(¤t, pb.is_null("s").unwrap()).await,
+ vec![3],
+ "{mode} NULL"
+ );
+ }
+}
