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 d05cb03a fix: Map precision-zero timestamps to Arrow seconds (#884)
d05cb03a is described below
commit d05cb03ad8818f67ff827ad16a3fc570044668db
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Sep 20 15:36:50 2026 +0800
fix: Map precision-zero timestamps to Arrow seconds (#884)
---
crates/paimon/src/arrow/format/avro.rs | 13 +-
crates/paimon/src/arrow/format/mod.rs | 47 ++++
crates/paimon/src/arrow/format/mosaic.rs | 4 +-
crates/paimon/src/arrow/format/parquet.rs | 226 +++++++++++++++++-
crates/paimon/src/arrow/format/row.rs | 104 ++++++++-
crates/paimon/src/arrow/mod.rs | 17 +-
crates/paimon/src/arrow/partition.rs | 19 +-
crates/paimon/src/arrow/residual.rs | 243 +++++++++++++++++++-
crates/paimon/src/arrow/shredding/variant.rs | 8 +-
crates/paimon/src/spec/binary_row.rs | 44 +++-
crates/paimon/src/table/aggregator/numeric.rs | 55 +++--
crates/paimon/src/table/data_file_reader.rs | 315 +++++++++++++++++++++++++-
12 files changed, 1048 insertions(+), 47 deletions(-)
diff --git a/crates/paimon/src/arrow/format/avro.rs
b/crates/paimon/src/arrow/format/avro.rs
index 4757f0a0..6c17cbca 100644
--- a/crates/paimon/src/arrow/format/avro.rs
+++ b/crates/paimon/src/arrow/format/avro.rs
@@ -27,7 +27,7 @@ use arrow_array::{
BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array,
Float64Array,
Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
ListArray, MapArray,
RecordBatch, StringArray, StructArray, Time32MillisecondArray,
TimestampMicrosecondArray,
- TimestampMillisecondArray, TimestampNanosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::SchemaRef;
@@ -433,7 +433,16 @@ fn build_timestamp_column(
.map(|i| get_field_at(&records[i], idx).and_then(value_as_i64))
.collect();
match precision {
- 0..=3 =>
Arc::new(TimestampMillisecondArray::from(values).with_timezone_opt(tz)),
+ 0 => Arc::new(
+ TimestampSecondArray::from(
+ values
+ .into_iter()
+ .map(|value| value.map(|value| value.div_euclid(1_000)))
+ .collect::<Vec<_>>(),
+ )
+ .with_timezone_opt(tz),
+ ),
+ 1..=3 =>
Arc::new(TimestampMillisecondArray::from(values).with_timezone_opt(tz)),
4..=6 =>
Arc::new(TimestampMicrosecondArray::from(values).with_timezone_opt(tz)),
_ =>
Arc::new(TimestampNanosecondArray::from(values).with_timezone_opt(tz)),
}
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index 4cd42a3c..b4ede7e6 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -331,6 +331,53 @@ pub(crate) async fn create_format_writer(
}
}
+fn timestamp_millis_schema(schema: &SchemaRef) -> SchemaRef {
+ let fields = schema
+ .fields()
+ .iter()
+ .map(timestamp_millis_field)
+ .collect::<Vec<_>>();
+ Arc::new(arrow_schema::Schema::new_with_metadata(
+ fields,
+ schema.metadata().clone(),
+ ))
+}
+
+fn timestamp_millis_field(field: &arrow_schema::FieldRef) ->
arrow_schema::FieldRef {
+ let data_type = timestamp_millis_data_type(field.data_type());
+ if &data_type == field.data_type() {
+ field.clone()
+ } else {
+ Arc::new(field.as_ref().clone().with_data_type(data_type))
+ }
+}
+
+fn timestamp_millis_data_type(data_type: &arrow_schema::DataType) ->
arrow_schema::DataType {
+ use arrow_schema::DataType as ArrowDataType;
+
+ match data_type {
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Second, timezone) => {
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond,
timezone.clone())
+ }
+ ArrowDataType::List(field) =>
ArrowDataType::List(timestamp_millis_field(field)),
+ ArrowDataType::LargeList(field) =>
ArrowDataType::LargeList(timestamp_millis_field(field)),
+ ArrowDataType::FixedSizeList(field, size) => {
+ ArrowDataType::FixedSizeList(timestamp_millis_field(field), *size)
+ }
+ ArrowDataType::Struct(fields) => ArrowDataType::Struct(
+ fields
+ .iter()
+ .map(timestamp_millis_field)
+ .collect::<Vec<_>>()
+ .into(),
+ ),
+ ArrowDataType::Map(field, sorted) => {
+ ArrowDataType::Map(timestamp_millis_field(field), *sorted)
+ }
+ _ => data_type.clone(),
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/crates/paimon/src/arrow/format/mosaic.rs
b/crates/paimon/src/arrow/format/mosaic.rs
index fa0d8f93..2c5a2970 100644
--- a/crates/paimon/src/arrow/format/mosaic.rs
+++ b/crates/paimon/src/arrow/format/mosaic.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use super::{FilePredicates, FormatFileReader};
+use super::{timestamp_millis_schema, FilePredicates, FormatFileReader};
use crate::arrow::build_target_arrow_schema;
use crate::arrow::filtering::{
predicates_may_match_with_schema, remap_predicates_to_file, StatsAccessor,
@@ -320,7 +320,7 @@ fn read_mosaic_batches_blocking(
row_filter_factory: None,
file_fields: existing_scan_fields.clone(),
});
- let read_schema = build_target_arrow_schema(&existing_scan_fields)?;
+ let read_schema =
timestamp_millis_schema(&build_target_arrow_schema(&existing_scan_fields)?);
validate_mosaic_schema(&read_schema)?;
let projected_names = existing_scan_fields
.iter()
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index d4ddd584..a8ddca4b 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -16,7 +16,9 @@
// under the License.
use super::shredding::PhysicalFormatWriterFactory;
-use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
+use super::{
+ timestamp_millis_schema, FilePredicates, FormatFileReader,
FormatFileWriter, FormatWriteResult,
+};
use crate::arrow::filtering::{predicates_may_match_with_schema, StatsAccessor};
use crate::arrow::read_budget::ReadPermit;
use crate::arrow::shredding::map::MapShreddingReadPlan;
@@ -31,6 +33,7 @@ use crate::spec::{
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use arrow_array::{BooleanArray, RecordBatch};
+use arrow_cast::cast;
use async_trait::async_trait;
use bytes::Bytes;
use futures::future::BoxFuture;
@@ -137,6 +140,8 @@ fn row_filter_required_bytes(
/// Streams data directly to storage via `AsyncArrowWriter` + opendal.
pub(crate) struct ParquetFormatWriter {
inner: AsyncArrowWriter<Box<dyn crate::io::AsyncFileWrite>>,
+ /// Logical Arrow schema accepted from the Paimon write path.
+ input_schema: arrow_schema::SchemaRef,
/// Physical Arrow schema the writer was created with; re-encoded with the
/// shredding field metadata at close time.
schema: arrow_schema::SchemaRef,
@@ -201,6 +206,8 @@ impl ParquetFormatWriter {
// Reject a bad codec before allocating the writer.
let codec = parse_compression(compression, zstd_level)?;
let async_write = output.async_writer().await?;
+ let input_schema = schema;
+ let schema = timestamp_millis_schema(&input_schema);
let inner = create_parquet_arrow_writer(async_write, schema.clone(),
codec)?;
let core_options = CoreOptions::new(format_options);
let stats_modes = write_fields
@@ -208,6 +215,7 @@ impl ParquetFormatWriter {
.transpose()?;
Ok(Self {
inner,
+ input_schema,
schema,
write_fields: write_fields.map(|fields| fields.to_vec()),
stats_modes,
@@ -289,6 +297,23 @@ fn parse_compression(codec: &str, zstd_level: i32) ->
crate::Result<Compression>
#[async_trait]
impl FormatFileWriter for ParquetFormatWriter {
async fn write(&mut self, batch: &RecordBatch) -> crate::Result<()> {
+ if batch.schema().as_ref() != self.input_schema.as_ref() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Parquet write batch schema {:?} does not match writer
schema {:?}",
+ batch.schema(),
+ self.input_schema
+ ),
+ source: None,
+ });
+ }
+ let physical_batch;
+ let batch = if self.input_schema.as_ref() == self.schema.as_ref() {
+ batch
+ } else {
+ physical_batch = cast_record_batch_to_schema(batch, &self.schema)?;
+ &physical_batch
+ };
self.inner
.write(batch)
.await
@@ -379,6 +404,48 @@ impl FormatFileWriter for ParquetFormatWriter {
}
}
+fn cast_record_batch_to_schema(
+ batch: &RecordBatch,
+ target_schema: &arrow_schema::SchemaRef,
+) -> crate::Result<RecordBatch> {
+ if batch.num_columns() != target_schema.fields().len() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Parquet write batch has {} columns, expected {}",
+ batch.num_columns(),
+ target_schema.fields().len()
+ ),
+ source: None,
+ });
+ }
+ let columns = batch
+ .columns()
+ .iter()
+ .zip(target_schema.fields())
+ .map(|(column, field)| {
+ if column.data_type() == field.data_type() {
+ Ok(column.clone())
+ } else {
+ cast(column, field.data_type()).map_err(|error|
crate::Error::DataInvalid {
+ message: format!(
+ "Failed to convert Parquet column '{}' from {:?} to
{:?}: {error}",
+ field.name(),
+ column.data_type(),
+ field.data_type()
+ ),
+ source: Some(Box::new(error)),
+ })
+ }
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ RecordBatch::try_new(target_schema.clone(), columns).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("Failed to build Parquet physical batch:
{error}"),
+ source: Some(Box::new(error)),
+ }
+ })
+}
+
#[async_trait]
impl FormatFileReader for ParquetFormatReader {
async fn read_batch_stream(
@@ -2336,14 +2403,15 @@ mod tests {
use crate::arrow::{build_target_arrow_schema, variant_arrow_type,
ReadBudget};
use crate::io::FileIOBuilder;
use crate::spec::{
- ArrayType, BigIntType, DataField, DataType, Datum, IntType, MapType,
PredicateBuilder,
- VarCharType, VariantType,
+ ArrayType, BigIntType, DataField, DataType, Datum, IntType,
LocalZonedTimestampType,
+ MapType, PredicateBuilder, TimestampType, VarCharType, VariantType,
};
use crate::table::RowRange;
use crate::variant::GenericVariant;
use crate::Error;
use arrow_array::{
- Array, BinaryArray, Int32Array, Int64Array, MapArray, RecordBatch,
StringArray, StructArray,
+ Array, BinaryArray, Int32Array, Int64Array, MapArray, RecordBatch,
StringArray,
+ StructArray, TimestampMillisecondArray, TimestampSecondArray,
};
use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
@@ -3136,6 +3204,156 @@ mod tests {
assert_eq!(total_rows, 5);
}
+ fn timestamp_zero_fields() -> Vec<DataField> {
+ vec![
+ DataField::new(
+ 0,
+ "ts".to_string(),
+ DataType::Timestamp(TimestampType::new(0).unwrap()),
+ ),
+ DataField::new(
+ 1,
+ "ts_ltz".to_string(),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(0).unwrap()),
+ ),
+ ]
+ }
+
+ async fn assert_timestamp_zero_filters(bytes: Bytes, fields: &[DataField])
{
+ let cases = [
+ (
+ "ts",
+ Datum::Timestamp {
+ millis: 2_000,
+ nanos: 0,
+ },
+ ),
+ (
+ "ts_ltz",
+ Datum::LocalZonedTimestamp {
+ millis: 2_000,
+ nanos: 0,
+ },
+ ),
+ ];
+ for (column, literal) in cases {
+ let predicate = PredicateBuilder::new(fields)
+ .equal(column, literal)
+ .unwrap();
+ let predicates = FilePredicates {
+ predicates: vec![predicate],
+ row_filter_factory: None,
+ file_fields: fields.to_vec(),
+ };
+ let batches = ParquetFormatReader::default()
+ .read_batch_stream(
+ Box::new(TrackingFileRead::new(bytes.clone())),
+ bytes.len() as u64,
+ fields,
+ Some(&predicates),
+ None,
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(
+ batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ 1,
+ "filter on {column}"
+ );
+ let array = batches[0]
+ .column_by_name(column)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .unwrap();
+ assert_eq!(array.value(0), 2_000, "filter on {column}");
+ }
+ }
+
+ #[tokio::test]
+ async fn
test_timestamp_zero_parquet_write_preserves_millis_encoding_and_filters() {
+ let fields = timestamp_zero_fields();
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(TimestampSecondArray::from(vec![-1, 2, 3])),
+ Arc::new(TimestampSecondArray::from(vec![-1, 2,
3]).with_timezone("UTC")),
+ ],
+ )
+ .unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/timestamp_zero_millis_encoding.parquet";
+ let output = file_io.new_output(path).unwrap();
+ let mut writer: Box<dyn FormatFileWriter> = Box::new(
+ ParquetFormatWriter::new(&output, schema, "zstd", 1,
Some(&fields), &HashMap::new())
+ .await
+ .unwrap(),
+ );
+ writer.write(&batch).await.unwrap();
+ writer.close().await.unwrap();
+
+ let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
+ let metadata = load_metadata_with_page_index(&bytes, false);
+ for (index, adjusted_to_utc) in [(0, false), (1, true)] {
+ let column = metadata.file_metadata().schema_descr().column(index);
+ assert_eq!(column.physical_type(), parquet::basic::Type::INT64);
+ assert_eq!(
+ column.logical_type_ref(),
+ Some(&parquet::basic::LogicalType::Timestamp {
+ is_adjusted_to_u_t_c: adjusted_to_utc,
+ unit: parquet::basic::TimeUnit::MILLIS,
+ })
+ );
+ let ParquetStatistics::Int64(stats) =
+ metadata.row_group(0).column(index).statistics().unwrap()
+ else {
+ panic!("timestamp column must have INT64 statistics");
+ };
+ assert_eq!(stats.min_opt(), Some(&-1_000));
+ assert_eq!(stats.max_opt(), Some(&3_000));
+ }
+ assert_timestamp_zero_filters(bytes, &fields).await;
+ }
+
+ #[tokio::test]
+ async fn test_existing_millis_timestamp_zero_parquet_filters() {
+ let fields = timestamp_zero_fields();
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new(
+ "ts",
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond,
None),
+ true,
+ ),
+ ArrowField::new(
+ "ts_ltz",
+ ArrowDataType::Timestamp(arrow_schema::TimeUnit::Millisecond,
Some("UTC".into())),
+ true,
+ ),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(TimestampMillisecondArray::from(vec![-1_000, 2_000,
3_000])),
+ Arc::new(
+ TimestampMillisecondArray::from(vec![-1_000, 2_000, 3_000])
+ .with_timezone("UTC"),
+ ),
+ ],
+ )
+ .unwrap();
+ let mut bytes = Vec::new();
+ let mut writer = AsyncArrowWriter::try_new(&mut bytes, schema,
None).unwrap();
+ writer.write(&batch).await.unwrap();
+ writer.close().await.unwrap();
+
+ assert_timestamp_zero_filters(Bytes::from(bytes), &fields).await;
+ }
+
#[tokio::test]
async fn test_parquet_inline_fixed_size_list_roundtrip() {
use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
diff --git a/crates/paimon/src/arrow/format/row.rs
b/crates/paimon/src/arrow/format/row.rs
index 0c65645a..9419507e 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -33,13 +33,14 @@ use arrow_array::builder::{
BinaryBuilder, BooleanBuilder, Date32Builder, Decimal128Builder,
Float32Builder,
Float64Builder, Int16Builder, Int32Builder, Int64Builder, Int8Builder,
LargeBinaryBuilder,
StringBuilder, Time32MillisecondBuilder, TimestampMicrosecondBuilder,
- TimestampMillisecondBuilder, TimestampNanosecondBuilder,
+ TimestampMillisecondBuilder, TimestampNanosecondBuilder,
TimestampSecondBuilder,
};
use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Float32Array,
Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
LargeBinaryArray, ListArray,
MapArray, RecordBatch, RecordBatchOptions, StringArray, StructArray,
Time32MillisecondArray,
TimestampMicrosecondArray, TimestampMillisecondArray,
TimestampNanosecondArray,
+ TimestampSecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType as ArrowDataType, Field, Fields, SchemaRef,
TimeUnit};
@@ -507,7 +508,8 @@ fn validate_arrow_map_entries(
fn timestamp_time_unit_for_precision(precision: u32) -> TimeUnit {
match precision {
- 0..=3 => TimeUnit::Millisecond,
+ 0 => TimeUnit::Second,
+ 1..=3 => TimeUnit::Millisecond,
4..=6 => TimeUnit::Microsecond,
_ => TimeUnit::Nanosecond,
}
@@ -901,6 +903,16 @@ fn write_timestamp(
precision: u32,
) -> crate::Result<()> {
let (millis, nanos_of_milli) = match array.data_type() {
+ ArrowDataType::Timestamp(TimeUnit::Second, _) => (
+ downcast::<TimestampSecondArray>(array,
&DataType::Timestamp(Default::default()))?
+ .value(row_idx)
+ .checked_mul(1_000)
+ .ok_or_else(|| Error::DataInvalid {
+ message: ".row timestamp second conversion
overflow".to_string(),
+ source: None,
+ })?,
+ 0,
+ ),
ArrowDataType::Timestamp(TimeUnit::Millisecond, _) => (
downcast::<TimestampMillisecondArray>(array,
&DataType::Timestamp(Default::default()))?
.value(row_idx),
@@ -1173,6 +1185,7 @@ enum ColumnBuilder {
LargeBinary(LargeBinaryBuilder),
Date(Date32Builder),
Time(Time32MillisecondBuilder),
+ TimestampS(TimestampSecondBuilder),
TimestampMs(TimestampMillisecondBuilder),
TimestampUs(TimestampMicrosecondBuilder),
TimestampNs(TimestampNanosecondBuilder),
@@ -1313,6 +1326,7 @@ impl ColumnBuilder {
Self::LargeBinary(b) => b.append_null(),
Self::Date(b) => b.append_null(),
Self::Time(b) => b.append_null(),
+ Self::TimestampS(b) => b.append_null(),
Self::TimestampMs(b) => b.append_null(),
Self::TimestampUs(b) => b.append_null(),
Self::TimestampNs(b) => b.append_null(),
@@ -1372,6 +1386,9 @@ impl ColumnBuilder {
(Self::LargeBinary(b), DataType::Blob(_)) =>
b.append_value(input.read_bytes()?),
(Self::Date(b), DataType::Date(_)) =>
b.append_value(input.read_i32()?),
(Self::Time(b), DataType::Time(_)) =>
b.append_value(input.read_i32()?),
+ (Self::TimestampS(b), DataType::Timestamp(_) |
DataType::LocalZonedTimestamp(_)) => {
+ b.append_value(read_timestamp_value(input, TimeUnit::Second)?);
+ }
(Self::TimestampMs(b), DataType::Timestamp(_) |
DataType::LocalZonedTimestamp(_)) => {
b.append_value(read_timestamp_value(input,
TimeUnit::Millisecond)?);
}
@@ -1499,6 +1516,7 @@ impl ColumnBuilder {
Self::LargeBinary(mut b) => Arc::new(b.finish()),
Self::Date(mut b) => Arc::new(b.finish()),
Self::Time(mut b) => Arc::new(b.finish()),
+ Self::TimestampS(mut b) => Arc::new(b.finish()),
Self::TimestampMs(mut b) => Arc::new(b.finish()),
Self::TimestampUs(mut b) => Arc::new(b.finish()),
Self::TimestampNs(mut b) => Arc::new(b.finish()),
@@ -1750,7 +1768,15 @@ fn map_entries_field(
fn timestamp_builder(precision: u32, timezone: bool, capacity: usize) ->
ColumnBuilder {
match precision {
- 0..=3 => {
+ 0 => {
+ let builder = TimestampSecondBuilder::with_capacity(capacity);
+ if timezone {
+ ColumnBuilder::TimestampS(builder.with_timezone("UTC"))
+ } else {
+ ColumnBuilder::TimestampS(builder)
+ }
+ }
+ 1..=3 => {
let builder = TimestampMillisecondBuilder::with_capacity(capacity);
if timezone {
ColumnBuilder::TimestampMs(builder.with_timezone("UTC"))
@@ -1801,7 +1827,7 @@ fn read_timestamp_value(input: &mut BlockInput<'_>, unit:
TimeUnit) -> crate::Re
source: None,
})?
}
- TimeUnit::Second => millis / 1_000,
+ TimeUnit::Second => millis.div_euclid(1_000),
})
}
@@ -2341,8 +2367,9 @@ mod tests {
use crate::io::FileIOBuilder;
use crate::spec::{
ArrayType, BigIntType, BlobType, BooleanType, DataType, DateType,
Datum, DecimalType,
- DoubleType, FloatType, IntType, MapType, MultisetType, Predicate,
PredicateOperator,
- RowType, TimeType, TimestampType, VarBinaryType, VarCharType,
VariantType,
+ DoubleType, FloatType, IntType, LocalZonedTimestampType, MapType,
MultisetType, Predicate,
+ PredicateOperator, RowType, TimeType, TimestampType, VarBinaryType,
VarCharType,
+ VariantType,
};
use crate::variant::GenericVariant;
use futures::TryStreamExt;
@@ -2641,6 +2668,71 @@ mod tests {
assert_eq!(names.value(1), "ccc");
}
+ #[tokio::test]
+ async fn row_writer_reader_roundtrip_timestamp_zero_as_seconds() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/row-timestamp-zero/data.row";
+ let output = file_io.new_output(path).unwrap();
+ let fields = vec![
+ DataField::new(
+ 0,
+ "ts".to_string(),
+ DataType::Timestamp(TimestampType::new(0).unwrap()),
+ ),
+ DataField::new(
+ 1,
+ "ts_ltz".to_string(),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(0).unwrap()),
+ ),
+ ];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let mut writer = RowFormatWriter::new(&output, schema.clone(),
fields.clone(), 1)
+ .await
+ .unwrap();
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(TimestampSecondArray::from(vec![Some(-1), Some(2),
None])),
+ Arc::new(
+ TimestampSecondArray::from(vec![Some(-1), Some(2),
None]).with_timezone("UTC"),
+ ),
+ ],
+ )
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ Box::new(writer).close().await.unwrap();
+
+ let input = file_io.new_input(path).unwrap();
+ let bytes = input.read().await.unwrap();
+ let actual = RowFormatReader
+ .read_batch_stream(
+ Box::new(BytesFileRead(bytes.clone())),
+ bytes.len() as u64,
+ &fields,
+ None,
+ Some(8),
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(actual.len(), 1);
+ assert_eq!(actual[0].schema().as_ref(), batch.schema().as_ref());
+ for column in 0..2 {
+ let values = actual[0]
+ .column(column)
+ .as_any()
+ .downcast_ref::<TimestampSecondArray>()
+ .unwrap();
+ assert_eq!(values.value(0), -1);
+ assert_eq!(values.value(1), 2);
+ assert!(values.is_null(2));
+ }
+ }
+
#[tokio::test]
async fn row_writer_reader_preserves_binary_and_large_binary_types() {
let file_io = FileIOBuilder::new("memory").build().unwrap();
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index 4792cdfb..637d6ee2 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -149,7 +149,8 @@ pub fn paimon_type_to_arrow(dt: &PaimonDataType) ->
crate::Result<ArrowDataType>
fn timestamp_time_unit(precision: u32) -> crate::Result<TimeUnit> {
match precision {
- 0..=3 => Ok(TimeUnit::Millisecond),
+ 0 => Ok(TimeUnit::Second),
+ 1..=3 => Ok(TimeUnit::Millisecond),
4..=6 => Ok(TimeUnit::Microsecond),
7..=9 => Ok(TimeUnit::Nanosecond),
_ => Err(crate::Error::Unsupported {
@@ -518,6 +519,15 @@ mod tests {
#[test]
fn test_timestamp_roundtrip() {
+ // second precision
+ let ts0 = PaimonDataType::Timestamp(TimestampType::new(0).unwrap());
+ assert_paimon_to_arrow(&ts0,
&ArrowDataType::Timestamp(TimeUnit::Second, None));
+ assert_arrow_to_paimon(
+ &ArrowDataType::Timestamp(TimeUnit::Second, None),
+ true,
+ &ts0,
+ );
+
// millisecond precision
let ts3 = PaimonDataType::Timestamp(TimestampType::new(3).unwrap());
assert_paimon_to_arrow(&ts3,
&ArrowDataType::Timestamp(TimeUnit::Millisecond, None));
@@ -548,6 +558,11 @@ mod tests {
#[test]
fn test_local_zoned_timestamp() {
+ let lzts0 =
PaimonDataType::LocalZonedTimestamp(LocalZonedTimestampType::new(0).unwrap());
+ let arrow0 = ArrowDataType::Timestamp(TimeUnit::Second,
Some("UTC".into()));
+ assert_paimon_to_arrow(&lzts0, &arrow0);
+ assert_arrow_to_paimon(&arrow0, true, &lzts0);
+
let lzts =
PaimonDataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap());
let arrow = ArrowDataType::Timestamp(TimeUnit::Millisecond,
Some("UTC".into()));
assert_paimon_to_arrow(&lzts, &arrow);
diff --git a/crates/paimon/src/arrow/partition.rs
b/crates/paimon/src/arrow/partition.rs
index 5e426cab..88c75118 100644
--- a/crates/paimon/src/arrow/partition.rs
+++ b/crates/paimon/src/arrow/partition.rs
@@ -24,7 +24,7 @@ use arrow_array::{
new_null_array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
Decimal128Array,
Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
StringArray,
Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
- TimestampNanosecondArray,
+ TimestampNanosecondArray, TimestampSecondArray,
};
use std::sync::Arc;
@@ -121,7 +121,15 @@ fn timestamp_array(
num_rows: usize,
) -> crate::Result<ArrayRef> {
let array: ArrayRef = match precision {
- 0..=3 => {
+ 0 => {
+ let value = millis.div_euclid(1_000);
+ let array = TimestampSecondArray::from(vec![Some(value);
num_rows]);
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ 1..=3 => {
let array = TimestampMillisecondArray::from(vec![Some(millis);
num_rows]);
match timezone {
Some(tz) => Arc::new(array.with_timezone(tz)),
@@ -179,6 +187,13 @@ mod tests {
}),
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(9).unwrap()),
),
+ (
+ Some(Datum::LocalZonedTimestamp {
+ millis: -1_000,
+ nanos: 0,
+ }),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(0).unwrap()),
+ ),
];
let datums: Vec<_> = values
.iter()
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index 50d716c4..f1e5fe2a 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -52,7 +52,7 @@ use arrow_array::{
FixedSizeListArray, Float32Array, Float64Array, Int16Array, Int32Array,
Int64Array, Int8Array,
LargeBinaryArray, LargeListArray, ListArray, RecordBatch, Scalar,
StringArray,
Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
- TimestampNanosecondArray,
+ TimestampNanosecondArray, TimestampSecondArray,
};
use arrow_ord::cmp::{
eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt,
lt_eq as arrow_lt_eq,
@@ -336,6 +336,19 @@ pub(crate) fn evaluate_exact_leaf_predicate(
) {
return evaluate_array_membership_predicate(array, data_type, op,
literals);
}
+ // TIMESTAMP(0) is exposed as Arrow seconds, while Parquet keeps Paimon's
+ // millisecond physical representation. Compare both representations in
+ // i128 nanoseconds so the decoded array's unit cannot disagree with the
+ // literal scalar and a fractional literal (for example 1.5 seconds) is not
+ // rounded down to the column precision.
+ if matches!(
+ data_type,
+ DataType::Timestamp(_) | DataType::LocalZonedTimestamp(_)
+ ) && matches!(array.data_type(), arrow_schema::DataType::Timestamp(_, _))
+ && !matches!(op, PredicateOperator::IsNull |
PredicateOperator::IsNotNull)
+ {
+ return evaluate_timestamp_leaf(array, data_type, op, literals);
+ }
// Decimals are compared by mathematical value across scales (Paimon
// `datum_cmp`/`decimal_cmp`). Arrow scalar comparison requires the
literal to
// be representable at the column scale, which fails for a finer-scale
literal
@@ -393,6 +406,152 @@ pub(crate) fn evaluate_exact_leaf_predicate(
}
}
+/// Evaluate timestamps independently of their decoded Arrow unit.
+///
+/// Paimon's [`Datum`] stores epoch milliseconds plus nanoseconds within that
+/// millisecond. Converting both the Arrow value and the literal to `i128`
+/// epoch nanoseconds is lossless for every Arrow timestamp unit and avoids the
+/// range loss that casting a seconds column to Arrow nanoseconds would cause.
+fn evaluate_timestamp_leaf(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ use std::cmp::Ordering;
+
+ let local_zoned = match data_type {
+ DataType::Timestamp(_) => false,
+ DataType::LocalZonedTimestamp(_) => true,
+ _ => unreachable!("guarded by caller"),
+ };
+ let literal_nanos = |literal: &Datum| -> Result<i128, ArrowError> {
+ let (millis, nanos) = match (local_zoned, literal) {
+ (false, Datum::Timestamp { millis, nanos })
+ | (true, Datum::LocalZonedTimestamp { millis, nanos }) =>
(*millis, *nanos),
+ _ => {
+ return Err(ArrowError::ComputeError(
+ "timestamp column compared against a non-matching
timestamp literal"
+ .to_string(),
+ ))
+ }
+ };
+ if !(0..1_000_000).contains(&nanos) {
+ return Err(ArrowError::ComputeError(format!(
+ "timestamp nanos-of-millisecond is out of range: {nanos}"
+ )));
+ }
+ Ok(i128::from(millis) * 1_000_000 + i128::from(nanos))
+ };
+ let value_nanos = |row: usize| -> Result<i128, ArrowError> {
+ let value = match array.data_type() {
+ arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Second,
_) => {
+ i128::from(
+ array
+ .as_any()
+ .downcast_ref::<TimestampSecondArray>()
+ .expect("Arrow timestamp unit matches array type")
+ .value(row),
+ ) * 1_000_000_000
+ }
+
arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Millisecond, _) => {
+ i128::from(
+ array
+ .as_any()
+ .downcast_ref::<TimestampMillisecondArray>()
+ .expect("Arrow timestamp unit matches array type")
+ .value(row),
+ ) * 1_000_000
+ }
+
arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, _) => {
+ i128::from(
+ array
+ .as_any()
+ .downcast_ref::<TimestampMicrosecondArray>()
+ .expect("Arrow timestamp unit matches array type")
+ .value(row),
+ ) * 1_000
+ }
+
arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, _) =>
i128::from(
+ array
+ .as_any()
+ .downcast_ref::<TimestampNanosecondArray>()
+ .expect("Arrow timestamp unit matches array type")
+ .value(row),
+ ),
+ other => {
+ return Err(ArrowError::ComputeError(format!(
+ "timestamp predicate expects an Arrow timestamp column,
got {other:?}"
+ )))
+ }
+ };
+ Ok(value)
+ };
+
+ let literal_values = literals
+ .iter()
+ .map(literal_nanos)
+ .collect::<Result<Vec<_>, _>>()?;
+ let mut result = Vec::with_capacity(array.len());
+ for row in 0..array.len() {
+ if array.is_null(row) {
+ result.push(false);
+ continue;
+ }
+ let value = value_nanos(row)?;
+ let keep = match op {
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq => {
+ let literal = literal_values
+ .first()
+ .ok_or_else(|| unconvertible_literal_error(op,
data_type))?;
+ let ordering = value.cmp(literal);
+ match op {
+ PredicateOperator::Eq => ordering == Ordering::Equal,
+ PredicateOperator::NotEq => ordering != Ordering::Equal,
+ PredicateOperator::Lt => ordering == Ordering::Less,
+ PredicateOperator::LtEq => ordering != Ordering::Greater,
+ PredicateOperator::Gt => ordering == Ordering::Greater,
+ PredicateOperator::GtEq => ordering != Ordering::Less,
+ _ => unreachable!(),
+ }
+ }
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ let contains = literal_values.contains(&value);
+ if matches!(op, PredicateOperator::In) {
+ contains
+ } else {
+ !contains
+ }
+ }
+ PredicateOperator::Between | PredicateOperator::NotBetween => {
+ let (Some(low), Some(high)) = (literal_values.first(),
literal_values.get(1))
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let within =
+ value.cmp(low) != Ordering::Less && value.cmp(high) !=
Ordering::Greater;
+ if matches!(op, PredicateOperator::Between) {
+ within
+ } else {
+ !within
+ }
+ }
+ _ => {
+ return Err(ArrowError::ComputeError(format!(
+ "operator {op:?} is not a timestamp value comparison"
+ )))
+ }
+ };
+ result.push(keep);
+ }
+ Ok(BooleanArray::from(result))
+}
+
fn evaluate_array_membership_predicate(
array: &ArrayRef,
data_type: &DataType,
@@ -545,6 +704,12 @@ fn evaluate_array_element_equality(
PredicateOperator::Eq,
std::slice::from_ref(literal),
)?)),
+ DataType::Timestamp(_) | DataType::LocalZonedTimestamp(_) =>
evaluate_timestamp_leaf(
+ values,
+ element_type,
+ PredicateOperator::Eq,
+ std::slice::from_ref(literal),
+ ),
_ => {
let Some(scalar) = literal_scalar_for_arrow_filter(literal,
element_type)
.map_err(|error| ArrowError::ComputeError(error.to_string()))?
@@ -1114,7 +1279,15 @@ fn timestamp_scalar(
timezone: Option<&'static str>,
) -> crate::Result<Option<ArrayRef>> {
let array: ArrayRef = match precision {
- 0..=3 => {
+ 0 => {
+ let value = millis.div_euclid(1_000);
+ let array = TimestampSecondArray::new_scalar(value).into_inner();
+ match timezone {
+ Some(tz) => Arc::new(array.with_timezone(tz)),
+ None => Arc::new(array),
+ }
+ }
+ 1..=3 => {
let array =
TimestampMillisecondArray::new_scalar(millis).into_inner();
match timezone {
Some(tz) => Arc::new(array.with_timezone(tz)),
@@ -1179,13 +1352,13 @@ fn float64_literal(literal: &Datum) -> Option<f64> {
mod tests {
use super::*;
use crate::spec::{
- row_id_leaf, ArrayType, DecimalType, DoubleType, FloatType, IntType,
VarCharType,
- ROW_ID_FIELD_NAME,
+ row_id_leaf, ArrayType, DecimalType, DoubleType, FloatType, IntType,
+ LocalZonedTimestampType, TimestampType, VarCharType, ROW_ID_FIELD_NAME,
};
use arrow_array::builder::{
Decimal128Builder, Float32Builder, Float64Builder, Int32Builder,
ListBuilder,
};
- use arrow_array::{Int32Array, StringArray};
+ use arrow_array::{Int32Array, StringArray, TimestampMillisecondArray,
TimestampSecondArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use std::sync::Arc;
@@ -1265,6 +1438,66 @@ mod tests {
mask.iter().map(|value| value.unwrap_or(false)).collect()
}
+ #[test]
+ fn
test_timestamp_zero_predicates_preserve_fractional_literals_and_physical_units()
{
+ let cases = [
+ (
+ Arc::new(TimestampSecondArray::from(vec![Some(1), Some(2)]))
as ArrayRef,
+ DataType::Timestamp(TimestampType::new(0).unwrap()),
+ Datum::Timestamp {
+ millis: 1_500,
+ nanos: 0,
+ },
+ ),
+ (
+ Arc::new(TimestampSecondArray::from(vec![Some(1),
Some(2)]).with_timezone("UTC"))
+ as ArrayRef,
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(0).unwrap()),
+ Datum::LocalZonedTimestamp {
+ millis: 1_500,
+ nanos: 0,
+ },
+ ),
+ ];
+
+ for (array, data_type, literal) in cases {
+ let equal = evaluate_exact_leaf_predicate(
+ &array,
+ &data_type,
+ PredicateOperator::Eq,
+ std::slice::from_ref(&literal),
+ )
+ .unwrap();
+ assert_eq!(bool_values(&equal), vec![false, false]);
+
+ let greater_or_equal = evaluate_exact_leaf_predicate(
+ &array,
+ &data_type,
+ PredicateOperator::GtEq,
+ std::slice::from_ref(&literal),
+ )
+ .unwrap();
+ assert_eq!(bool_values(&greater_or_equal), vec![false, true]);
+ }
+
+ let existing_millis: ArrayRef =
Arc::new(TimestampMillisecondArray::from(vec![
+ Some(-1_000),
+ Some(2_000),
+ Some(3_000),
+ ]));
+ let equal = evaluate_exact_leaf_predicate(
+ &existing_millis,
+ &DataType::Timestamp(TimestampType::new(0).unwrap()),
+ PredicateOperator::Eq,
+ &[Datum::Timestamp {
+ millis: 2_000,
+ nanos: 0,
+ }],
+ )
+ .unwrap();
+ assert_eq!(bool_values(&equal), vec![false, true, false]);
+ }
+
#[test]
fn test_array_membership_predicates_match_java_multivalue_semantics() {
let array = int_list_array();
diff --git a/crates/paimon/src/arrow/shredding/variant.rs
b/crates/paimon/src/arrow/shredding/variant.rs
index 85475be5..6246fc80 100644
--- a/crates/paimon/src/arrow/shredding/variant.rs
+++ b/crates/paimon/src/arrow/shredding/variant.rs
@@ -33,7 +33,7 @@ use arrow_array::{
new_null_array, Array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
Decimal128Array,
Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
LargeBinaryArray,
ListArray, RecordBatch, StringArray, StructArray,
TimestampMicrosecondArray,
- TimestampMillisecondArray, TimestampNanosecondArray,
+ TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Fields,
TimeUnit};
@@ -1052,7 +1052,8 @@ fn timestamp_array(
})
.collect::<Vec<_>>();
Ok(match precision {
- 0..=3 =>
Arc::new(TimestampMillisecondArray::from(values).with_timezone_opt(tz)),
+ 0 =>
Arc::new(TimestampSecondArray::from(values).with_timezone_opt(tz)),
+ 1..=3 =>
Arc::new(TimestampMillisecondArray::from(values).with_timezone_opt(tz)),
4..=6 =>
Arc::new(TimestampMicrosecondArray::from(values).with_timezone_opt(tz)),
_ =>
Arc::new(TimestampNanosecondArray::from(values).with_timezone_opt(tz)),
})
@@ -1261,6 +1262,9 @@ fn value_at(array: &dyn Array, row: usize, data_type:
&DataType) -> Result<Optio
fn timestamp_value_at(array: &dyn Array, row: usize) -> Result<i64> {
match array.data_type() {
+ ArrowDataType::Timestamp(TimeUnit::Second, _) => {
+ Ok(downcast_array::<TimestampSecondArray>(array,
"TimestampS")?.value(row))
+ }
ArrowDataType::Timestamp(TimeUnit::Millisecond, _) => {
Ok(downcast_array::<TimestampMillisecondArray>(array,
"TimestampMs")?.value(row))
}
diff --git a/crates/paimon/src/spec/binary_row.rs
b/crates/paimon/src/spec/binary_row.rs
index 17c4820a..15b81289 100644
--- a/crates/paimon/src/spec/binary_row.rs
+++ b/crates/paimon/src/spec/binary_row.rs
@@ -1166,7 +1166,21 @@ fn extract_timestamp_parts_from_arrow(
expected: &str,
) -> crate::Result<(i64, i32)> {
match precision {
- 0..=3 => {
+ 0 => {
+ let arr = col
+ .as_any()
+ .downcast_ref::<arrow_array::TimestampSecondArray>()
+ .ok_or_else(|| type_mismatch_err(&format!("{expected}(s)"),
col_idx))?;
+ let millis =
+ arr.value(row_idx)
+ .checked_mul(1_000)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("{expected}(s) value overflows
milliseconds"),
+ source: None,
+ })?;
+ Ok((millis, 0))
+ }
+ 1..=3 => {
let arr = col
.as_any()
.downcast_ref::<arrow_array::TimestampMillisecondArray>()
@@ -1336,6 +1350,7 @@ enum TypedColumn<'a> {
Decimal128(&'a arrow_array::Decimal128Array, u32, u32), // (array,
precision, scale)
Binary(&'a arrow_array::BinaryArray),
Variant(&'a arrow_array::StructArray),
+ TimestampS(&'a arrow_array::TimestampSecondArray),
TimestampMs(&'a arrow_array::TimestampMillisecondArray),
TimestampUs(&'a arrow_array::TimestampMicrosecondArray),
TimestampNs(&'a arrow_array::TimestampNanosecondArray),
@@ -1442,7 +1457,12 @@ fn downcast_columns<'a>(
TypedColumn::Variant(arr)
}
DataType::Timestamp(ts) => match ts.precision() {
- 0..=3 => TypedColumn::TimestampMs(
+ 0 => TypedColumn::TimestampS(
+ col.as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Timestamp(s)",
col_idx))?,
+ ),
+ 1..=3 => TypedColumn::TimestampMs(
col.as_any()
.downcast_ref()
.ok_or_else(|| type_mismatch_err("Timestamp(ms)",
col_idx))?,
@@ -1465,7 +1485,12 @@ fn downcast_columns<'a>(
},
DataType::LocalZonedTimestamp(ts) => {
match ts.precision() {
- 0..=3 =>
TypedColumn::TimestampMs(col.as_any().downcast_ref().ok_or_else(
+ 0 => {
+
TypedColumn::TimestampS(col.as_any().downcast_ref().ok_or_else(
+ || type_mismatch_err("LocalZonedTimestamp(s)",
col_idx),
+ )?)
+ }
+ 1..=3 =>
TypedColumn::TimestampMs(col.as_any().downcast_ref().ok_or_else(
|| type_mismatch_err("LocalZonedTimestamp(ms)",
col_idx),
)?),
4..=6 =>
TypedColumn::TimestampUs(col.as_any().downcast_ref().ok_or_else(
@@ -1660,6 +1685,19 @@ fn write_typed_value(
}
}
}
+ TypedColumn::TimestampS(arr) => {
+ if arr.is_null(row_idx) {
+ builder.set_null_at(pos);
+ } else {
+ let millis =
arr.value(row_idx).checked_mul(1_000).ok_or_else(|| {
+ crate::Error::DataInvalid {
+ message: "Timestamp(s) value overflows
milliseconds".to_string(),
+ source: None,
+ }
+ })?;
+ builder.write_timestamp_compact(pos, millis);
+ }
+ }
TypedColumn::TimestampMs(arr) => {
if arr.is_null(row_idx) {
builder.set_null_at(pos);
diff --git a/crates/paimon/src/table/aggregator/numeric.rs
b/crates/paimon/src/table/aggregator/numeric.rs
index 5546be3a..280eb2c7 100644
--- a/crates/paimon/src/table/aggregator/numeric.rs
+++ b/crates/paimon/src/table/aggregator/numeric.rs
@@ -45,7 +45,7 @@ use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
Float32Array,
Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, StringArray,
Time32MillisecondArray, TimestampMicrosecondArray,
TimestampMillisecondArray,
- TimestampNanosecondArray,
+ TimestampNanosecondArray, TimestampSecondArray,
};
use arrow_schema::{DataType as ArrowDataType, TimeUnit};
@@ -474,7 +474,8 @@ fn arrow_timestamp_timezone(
fn timestamp_time_unit(precision: u32) -> crate::Result<TimeUnit> {
match precision {
- 0..=3 => Ok(TimeUnit::Millisecond),
+ 0 => Ok(TimeUnit::Second),
+ 1..=3 => Ok(TimeUnit::Millisecond),
4..=6 => Ok(TimeUnit::Microsecond),
7..=9 => Ok(TimeUnit::Nanosecond),
other => Err(crate::Error::Unsupported {
@@ -601,17 +602,10 @@ fn agg_minmax(
MinMaxState::Date32(acc) => update_primitive!(acc, Date32Array),
MinMaxState::Time32Ms(acc) => update_primitive!(acc,
Time32MillisecondArray),
MinMaxState::Timestamp { unit, acc, .. } => match unit {
+ TimeUnit::Second => update_primitive!(acc, TimestampSecondArray),
TimeUnit::Millisecond => update_primitive!(acc,
TimestampMillisecondArray),
TimeUnit::Microsecond => update_primitive!(acc,
TimestampMicrosecondArray),
TimeUnit::Nanosecond => update_primitive!(acc,
TimestampNanosecondArray),
- other => {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "Timestamp with unit {other:?} not expected for field
'{field_name}'"
- ),
- source: None,
- });
- }
},
MinMaxState::Utf8(acc) => {
let v = downcast::<StringArray>(array, field_name)?.value(row_idx);
@@ -690,6 +684,9 @@ fn minmax_result(state: &MinMaxState, agg_name: &str,
field_name: &str) -> crate
} => match unit {
// `with_timezone_opt` keeps the result array's Arrow type equal
to the
// field's, which `RecordBatch::try_new` checks.
+ TimeUnit::Second => {
+
Arc::new(TimestampSecondArray::from(vec![*acc]).with_timezone_opt(timezone.clone()))
+ }
TimeUnit::Millisecond => Arc::new(
TimestampMillisecondArray::from(vec![*acc]).with_timezone_opt(timezone.clone()),
),
@@ -699,14 +696,6 @@ fn minmax_result(state: &MinMaxState, agg_name: &str,
field_name: &str) -> crate
TimeUnit::Nanosecond => Arc::new(
TimestampNanosecondArray::from(vec![*acc]).with_timezone_opt(timezone.clone()),
),
- other => {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "Timestamp with unit {other:?} not expected for field
'{field_name}'"
- ),
- source: None,
- });
- }
},
MinMaxState::Utf8(acc) =>
Arc::new(StringArray::from(vec![acc.clone()])),
MinMaxState::Binary(acc) =>
Arc::new(BinaryArray::from_opt_vec(vec![acc.as_deref()])),
@@ -1463,6 +1452,36 @@ mod tests {
arrow_array::RecordBatch::try_new(Arc::new(schema),
vec![max.result().unwrap()]).unwrap();
}
+ #[test]
+ fn test_min_max_timestamp_zero_uses_seconds() {
+ let dt = DataType::Timestamp(TimestampType::new(0).unwrap());
+ let arr = TimestampSecondArray::from(vec![Some(30), Some(10),
Some(20)]);
+ let mut min = min_agg(dt.clone());
+ let mut max = max_agg(dt);
+ for i in 0..arr.len() {
+ min.agg(&arr, i).unwrap();
+ max.agg(&arr, i).unwrap();
+ }
+ assert_eq!(
+ min.result()
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampSecondArray>()
+ .unwrap()
+ .value(0),
+ 10
+ );
+ assert_eq!(
+ max.result()
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampSecondArray>()
+ .unwrap()
+ .value(0),
+ 30
+ );
+ }
+
#[test]
fn test_min_max_date_and_timestamp_supported() {
// Date32
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index d72703b0..0b3cb03e 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -1965,6 +1965,200 @@ mod tests {
.unwrap()
}
+ #[tokio::test]
+ async fn test_mosaic_timestamp_zero_read_filter_and_row_selection() {
+ use crate::spec::{LocalZonedTimestampType, MapType, TimestampType};
+ use arrow_array::builder::{ListBuilder, TimestampMillisecondBuilder};
+ use arrow_array::{ArrayRef, MapArray, StructArray,
TimestampMillisecondArray};
+ use arrow_buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
+ use arrow_schema::DataType as ArrowDataType;
+
+ let fields_at_precision = |precision| {
+ let timestamp =
DataType::Timestamp(TimestampType::new(precision).unwrap());
+ let timestamp_ltz =
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(precision).unwrap());
+ vec![
+ data_field(0, "ts", timestamp.clone()),
+ data_field(1, "ts_ltz", timestamp_ltz.clone()),
+ data_field(2, "timestamps",
DataType::Array(ArrayType::new(timestamp))),
+ data_field(
+ 3,
+ "timestamp_map",
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ timestamp_ltz,
+ )),
+ ),
+ ]
+ };
+ let read_fields = fields_at_precision(0);
+ let output_schema = build_target_arrow_schema(&read_fields).unwrap();
+ let physical_schema =
build_target_arrow_schema(&fields_at_precision(3)).unwrap();
+ let ArrowDataType::List(element) =
physical_schema.field(2).data_type() else {
+ unreachable!()
+ };
+ let mut timestamps =
+
ListBuilder::new(TimestampMillisecondBuilder::new()).with_field(element.clone());
+ timestamps.values().append_value(-1_000);
+ timestamps.values().append_value(2_000);
+ timestamps.append(true);
+ timestamps.append(true);
+ timestamps.append(false);
+ timestamps.values().append_value(3_000);
+ timestamps.append(true);
+
+ let ArrowDataType::Map(entries_field, sorted) =
physical_schema.field(3).data_type() else {
+ unreachable!()
+ };
+ let ArrowDataType::Struct(entry_fields) = entries_field.data_type()
else {
+ unreachable!()
+ };
+ let entries = StructArray::try_new(
+ entry_fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["before", "after", "last"])),
+ Arc::new(
+ TimestampMillisecondArray::from(vec![-1_000, 2_000, 3_000])
+ .with_timezone("UTC"),
+ ),
+ ],
+ None,
+ )
+ .unwrap();
+ let timestamp_map = MapArray::try_new(
+ entries_field.clone(),
+ OffsetBuffer::new(ScalarBuffer::from(vec![0, 2, 2, 2, 3])),
+ entries,
+ Some(NullBuffer::from(vec![true, true, false, true])),
+ *sorted,
+ )
+ .unwrap();
+ let timestamps_millis = vec![Some(-1_000), Some(2_000), None,
Some(3_000)];
+ let physical_batch = RecordBatch::try_new(
+ physical_schema,
+ vec![
+
Arc::new(TimestampMillisecondArray::from(timestamps_millis.clone())),
+
Arc::new(TimestampMillisecondArray::from(timestamps_millis).with_timezone("UTC")),
+ Arc::new(timestamps.finish()),
+ Arc::new(timestamp_map),
+ ],
+ )
+ .unwrap();
+ let expected_columns = physical_batch
+ .columns()
+ .iter()
+ .zip(output_schema.fields())
+ .map(|(column, field)| arrow_cast::cast(column,
field.data_type()).unwrap())
+ .collect::<Vec<ArrayRef>>();
+ let data = write_mosaic(&physical_batch);
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/mosaic_timestamp_zero";
+ let bucket_path = format!("{table_path}/bucket-0");
+ let file_name = "part-0.mosaic";
+ file_io
+ .new_output(&format!("{bucket_path}/{file_name}"))
+ .unwrap()
+ .write(data.clone())
+ .await
+ .unwrap();
+ let builder = PredicateBuilder::new(&read_fields);
+ let cases = vec![
+ (Vec::new(), None, vec![0, 1, 2, 3]),
+ (
+ vec![builder
+ .greater_or_equal(
+ "ts",
+ Datum::Timestamp {
+ millis: 1_500,
+ nanos: 0,
+ },
+ )
+ .unwrap()],
+ None,
+ vec![1, 3],
+ ),
+ (
+ vec![builder
+ .equal(
+ "ts_ltz",
+ Datum::LocalZonedTimestamp {
+ millis: -1_000,
+ nanos: 0,
+ },
+ )
+ .unwrap()],
+ None,
+ vec![0],
+ ),
+ (
+ vec![builder
+ .array_contains(
+ "timestamps",
+ Datum::Timestamp {
+ millis: 2_000,
+ nanos: 0,
+ },
+ )
+ .unwrap()],
+ None,
+ vec![0],
+ ),
+ (
+ Vec::new(),
+ Some(vec![RowRange::new(0, 0), RowRange::new(3, 3)]),
+ vec![0, 3],
+ ),
+ ];
+ for row_groups in [0, 2] {
+ for (predicates, row_ranges, expected_rows) in &cases {
+ let mut split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path.clone())
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file(file_name, data.len() as
i64, 4, 1)]);
+ if let Some(row_ranges) = row_ranges {
+ split = split.with_row_ranges(row_ranges.clone());
+ }
+ let batches = DataFileReader::new(
+ file_io.clone(),
+ SchemaManager::new(file_io.clone(),
table_path.to_string()),
+ 1,
+ read_fields.clone(),
+ read_fields.clone(),
+ predicates.clone(),
+ )
+ .with_mosaic_prefetch(MosaicPrefetchOptions {
+ row_groups,
+ ..Default::default()
+ })
+ .read(&[split.build().unwrap()])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let actual =
+ arrow_select::concat::concat_batches(&output_schema,
&batches).unwrap();
+ let expected_indices =
arrow_array::UInt32Array::from(expected_rows.clone());
+ let expected = RecordBatch::try_new(
+ output_schema.clone(),
+ expected_columns
+ .iter()
+ .map(|column| {
+ arrow_select::take::take(column,
&expected_indices, None).unwrap()
+ })
+ .collect(),
+ )
+ .unwrap();
+ assert_eq!(
+ actual, expected,
+ "row_groups={row_groups}, predicates={predicates:?},
row_ranges={row_ranges:?}"
+ );
+ }
+ }
+ }
+
#[tokio::test]
async fn test_mosaic_physical_missing_column_is_null_filled() {
let physical_fields = vec![
@@ -2963,10 +3157,13 @@ mod vector_parquet_tests {
use crate::arrow::format::ParquetFormatWriter;
use crate::io::FileIOBuilder;
use crate::spec::stats::BinaryTableStats;
- use crate::spec::{DataFileMeta, DataType, FloatType, VectorType};
+ use crate::spec::{
+ DataFileMeta, DataType, Datum, FloatType, LocalZonedTimestampType,
PredicateBuilder,
+ TimestampType, VectorType,
+ };
use crate::table::source::DataSplitBuilder;
use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
- use arrow_array::{FixedSizeListArray, Float32Array, RecordBatch};
+ use arrow_array::{FixedSizeListArray, Float32Array, RecordBatch,
TimestampSecondArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField};
use futures::TryStreamExt;
@@ -3122,6 +3319,120 @@ mod vector_parquet_tests {
.expect("child should be Float32Array");
assert_eq!(floats2.values(), &[3.0, 4.0]);
}
+
+ #[tokio::test]
+ async fn test_datafilereader_timestamp_zero_parquet_roundtrip_and_filter()
{
+ let read_fields = vec![
+ DataField::new(
+ 0,
+ "ts".to_string(),
+ DataType::Timestamp(TimestampType::new(0).unwrap()),
+ ),
+ DataField::new(
+ 1,
+ "ts_ltz".to_string(),
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(0).unwrap()),
+ ),
+ ];
+ let arrow_schema = build_target_arrow_schema(&read_fields).unwrap();
+ let batch = RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(TimestampSecondArray::from(vec![-1, 2, 3])),
+ Arc::new(TimestampSecondArray::from(vec![-1, 2,
3]).with_timezone("UTC")),
+ ],
+ )
+ .unwrap();
+
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/timestamp_zero_parquet_e2e";
+ let bucket_path = format!("{table_path}/bucket-0");
+ let file_name = "part-0.parquet";
+ let file_path = format!("{bucket_path}/{file_name}");
+ let output = file_io.new_output(&file_path).unwrap();
+ let mut writer: Box<dyn FormatFileWriter> = Box::new(
+ ParquetFormatWriter::new(
+ &output,
+ arrow_schema.clone(),
+ "zstd",
+ 1,
+ Some(&read_fields),
+ &std::collections::HashMap::new(),
+ )
+ .await
+ .unwrap(),
+ );
+ writer.write(&batch).await.unwrap();
+ let file_size = writer.close().await.unwrap().file_size;
+
+ let schema_id = 1;
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(crate::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file(file_name, file_size as i64, 3,
schema_id)])
+ .build()
+ .unwrap();
+ let predicates = vec![
+ PredicateBuilder::new(&read_fields)
+ .equal(
+ "ts",
+ Datum::Timestamp {
+ millis: 2_000,
+ nanos: 0,
+ },
+ )
+ .unwrap(),
+ PredicateBuilder::new(&read_fields)
+ .equal(
+ "ts_ltz",
+ Datum::LocalZonedTimestamp {
+ millis: 2_000,
+ nanos: 0,
+ },
+ )
+ .unwrap(),
+ ];
+ let reader = DataFileReader::new(
+ file_io.clone(),
+ SchemaManager::new(file_io, table_path.to_string()),
+ schema_id,
+ read_fields.clone(),
+ read_fields,
+ predicates,
+ );
+ let batches = reader
+ .read(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].schema().as_ref(), arrow_schema.as_ref());
+ assert_eq!(
+ batches[0]
+ .column_by_name("ts")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampSecondArray>()
+ .unwrap()
+ .values(),
+ &[2]
+ );
+ assert_eq!(
+ batches[0]
+ .column_by_name("ts_ltz")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampSecondArray>()
+ .unwrap()
+ .values(),
+ &[2]
+ );
+ }
}
#[cfg(test)]
