diff --git a/crates/paimon/src/spec/aggregation.rs b/crates/paimon/src/spec/aggregation.rs index 5f2b4c288..b78096c26 100644 --- a/crates/paimon/src/spec/aggregation.rs +++ b/crates/paimon/src/spec/aggregation.rs @@ -430,9 +430,13 @@ pub(crate) fn validate_aggregator_for_type( | DataType::Float(_) | DataType::Double(_) ), + // Java `FieldMinAggFactory` / `FieldMaxAggFactory` accept anything + // `TypeCheckUtils#isComparable` allows, i.e. everything except MAP, + // MULTISET, ROW, ARRAY, VECTOR, VARIANT and BLOB. "min" | "max" => matches!( dt, - DataType::TinyInt(_) + DataType::Boolean(_) + | DataType::TinyInt(_) | DataType::SmallInt(_) | DataType::Int(_) | DataType::BigInt(_) @@ -442,8 +446,11 @@ pub(crate) fn validate_aggregator_for_type( | DataType::Date(_) | DataType::Time(_) | DataType::Timestamp(_) + | DataType::LocalZonedTimestamp(_) | DataType::Char(_) | DataType::VarChar(_) + | DataType::Binary(_) + | DataType::VarBinary(_) ), "bool_and" | "bool_or" => matches!(dt, DataType::Boolean(_)), // Java `FieldListaggAggFactory` only accepts unbounded VARCHAR (STRING); @@ -842,8 +849,9 @@ mod tests { #[test] fn validation_table_matches_constructors() { use crate::spec::{ - BigIntType, BooleanType, DateType, DecimalType, DoubleType, FloatType, SmallIntType, - TimeType, TimestampType, TinyIntType, + BigIntType, BinaryType, BlobType, BooleanType, DateType, DecimalType, DoubleType, + FloatType, LocalZonedTimestampType, SmallIntType, TimeType, TimestampType, TinyIntType, + VarBinaryType, }; let names = [ @@ -875,6 +883,14 @@ mod tests { DataType::Date(DateType::new()), DataType::Time(TimeType::new(3).unwrap()), DataType::Timestamp(TimestampType::new(6).unwrap()), + // Carries a timezone in Arrow, unlike TIMESTAMP — min/max must build a + // result array that still matches the schema. + DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()), + DataType::Binary(BinaryType::new(4).unwrap()), + DataType::VarBinary(VarBinaryType::new(8).unwrap()), + // Maps to the same Arrow Binary as the two above, but Java's + // `isComparable` excludes it, so min/max must keep rejecting it. + DataType::Blob(BlobType::new()), // Bounded VARCHAR (listagg must reject) and unbounded STRING // (listagg must accept) — exercises both sides of the listagg rule. DataType::VarChar(VarCharType::new(255).unwrap()), diff --git a/crates/paimon/src/table/aggregator/numeric.rs b/crates/paimon/src/table/aggregator/numeric.rs index f48511936..f093827bf 100644 --- a/crates/paimon/src/table/aggregator/numeric.rs +++ b/crates/paimon/src/table/aggregator/numeric.rs @@ -42,11 +42,12 @@ use std::sync::Arc; use arrow_array::builder::Decimal128Builder; use arrow_array::{ - Array, ArrayRef, Date32Array, Decimal128Array, Float32Array, Float64Array, Int16Array, - Int32Array, Int64Array, Int8Array, StringArray, Time32MillisecondArray, - TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray, + Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array, + Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, StringArray, + Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray, + TimestampNanosecondArray, }; -use arrow_schema::TimeUnit; +use arrow_schema::{DataType as ArrowDataType, TimeUnit}; use super::{unsupported_type_error, FieldAggregator}; use crate::spec::DataType; @@ -386,6 +387,7 @@ impl FieldAggregator for ProductAgg { /// `None` means "no non-null value seen yet for the current group". #[derive(Debug)] enum MinMaxState { + Bool(Option), I8(Option), I16(Option), I32(Option), @@ -401,11 +403,20 @@ enum MinMaxState { /// Paimon `TIME` is encoded as Arrow `Time32(Millisecond)` regardless of /// declared precision, so a single accumulator variant suffices. Time32Ms(Option), + /// Covers `TIMESTAMP` and `TIMESTAMP WITH LOCAL TIME ZONE`, which Java + /// compares through the same `Timestamp#compareTo` branch. `timezone` is + /// `Some("UTC")` for the latter, mirroring `paimon_type_to_arrow`, and has to + /// be carried here because the result array must still match the field's + /// Arrow type. Timestamp { unit: TimeUnit, + timezone: Option>, acc: Option, }, Utf8(Option), + /// `BINARY` and `VARBINARY`. Not `BLOB`, which Arrow also maps to `Binary` + /// but Java's `TypeCheckUtils#isComparable` excludes. + Binary(Option>), } fn make_minmax_state( @@ -414,6 +425,7 @@ fn make_minmax_state( op: &str, ) -> crate::Result { Ok(match data_type { + DataType::Boolean(_) => MinMaxState::Bool(None), DataType::TinyInt(_) => MinMaxState::I8(None), DataType::SmallInt(_) => MinMaxState::I16(None), DataType::Int(_) => MinMaxState::I32(None), @@ -429,13 +441,37 @@ fn make_minmax_state( DataType::Time(_) => MinMaxState::Time32Ms(None), DataType::Timestamp(t) => MinMaxState::Timestamp { unit: timestamp_time_unit(t.precision())?, + timezone: arrow_timestamp_timezone(field_name, data_type)?, + acc: None, + }, + DataType::LocalZonedTimestamp(t) => MinMaxState::Timestamp { + unit: timestamp_time_unit(t.precision())?, + timezone: arrow_timestamp_timezone(field_name, data_type)?, acc: None, }, DataType::Char(_) | DataType::VarChar(_) => MinMaxState::Utf8(None), + DataType::Binary(_) | DataType::VarBinary(_) => MinMaxState::Binary(None), other => return Err(unsupported_type_error(op, field_name, other)), }) } +/// Read the Arrow timezone back out of the very mapping the read schema is built +/// with, so a min/max result array cannot drift from its field's Arrow type. +fn arrow_timestamp_timezone( + field_name: &str, + data_type: &DataType, +) -> crate::Result>> { + match crate::arrow::paimon_type_to_arrow(data_type)? { + ArrowDataType::Timestamp(_, timezone) => Ok(timezone), + other => Err(crate::Error::DataInvalid { + message: format!( + "Aggregate column '{field_name}' maps to Arrow type {other:?}, expected a timestamp" + ), + source: None, + }), + } +} + fn timestamp_time_unit(precision: u32) -> crate::Result { match precision { 0..=3 => Ok(TimeUnit::Millisecond), @@ -525,6 +561,36 @@ fn agg_minmax( }}; } match state { + MinMaxState::Bool(acc) => { + // Java compares BOOLEAN with `Boolean.compare`, i.e. false < true + // (`InternalRowUtils#compare`), which is what Rust's `bool: Ord` does. + let v = downcast::(array, field_name)?.value(row_idx); + *acc = Some(match acc.take() { + None => v, + Some(prev) => { + // Ordering rather than `<`/`>`: comparing bools with order + // operators is what `clippy::bool_comparison` objects to, and + // this keeps the tie handling identical to the other arms. + let ordering = v.cmp(&prev); + let take_new = if keep_smaller { + if reversed { + ordering.is_lt() + } else { + ordering.is_le() + } + } else if reversed { + ordering.is_ge() + } else { + ordering.is_gt() + }; + if take_new { + v + } else { + prev + } + } + }); + } MinMaxState::I8(acc) => update_primitive!(acc, Int8Array), MinMaxState::I16(acc) => update_primitive!(acc, Int16Array), MinMaxState::I32(acc) => update_primitive!(acc, Int32Array), @@ -534,7 +600,7 @@ fn agg_minmax( MinMaxState::Decimal128 { acc, .. } => update_primitive!(acc, Decimal128Array), MinMaxState::Date32(acc) => update_primitive!(acc, Date32Array), MinMaxState::Time32Ms(acc) => update_primitive!(acc, Time32MillisecondArray), - MinMaxState::Timestamp { unit, acc } => match unit { + MinMaxState::Timestamp { unit, acc, .. } => match unit { TimeUnit::Millisecond => update_primitive!(acc, TimestampMillisecondArray), TimeUnit::Microsecond => update_primitive!(acc, TimestampMicrosecondArray), TimeUnit::Nanosecond => update_primitive!(acc, TimestampNanosecondArray), @@ -571,12 +637,39 @@ fn agg_minmax( } }); } + MinMaxState::Binary(acc) => { + // Java's `byteArrayCompare` is unsigned lexicographic with a length + // tiebreak, which is exactly `[u8]: Ord`. + let v = downcast::(array, field_name)?.value(row_idx); + *acc = Some(match acc.take() { + None => v.to_vec(), + Some(prev) => { + let take_new = if keep_smaller { + if reversed { + v < prev.as_slice() + } else { + v <= prev.as_slice() + } + } else if reversed { + v >= prev.as_slice() + } else { + v > prev.as_slice() + }; + if take_new { + v.to_vec() + } else { + prev + } + } + }); + } } Ok(()) } fn minmax_result(state: &MinMaxState, agg_name: &str, field_name: &str) -> crate::Result { Ok(match state { + MinMaxState::Bool(acc) => Arc::new(BooleanArray::from(vec![*acc])), MinMaxState::I8(acc) => Arc::new(Int8Array::from(vec![*acc])), MinMaxState::I16(acc) => Arc::new(Int16Array::from(vec![*acc])), MinMaxState::I32(acc) => Arc::new(Int32Array::from(vec![*acc])), @@ -590,10 +683,22 @@ fn minmax_result(state: &MinMaxState, agg_name: &str, field_name: &str) -> crate } => decimal_array(*precision, *scale, *acc, agg_name, field_name)?, MinMaxState::Date32(acc) => Arc::new(Date32Array::from(vec![*acc])), MinMaxState::Time32Ms(acc) => Arc::new(Time32MillisecondArray::from(vec![*acc])), - MinMaxState::Timestamp { unit, acc } => match unit { - TimeUnit::Millisecond => Arc::new(TimestampMillisecondArray::from(vec![*acc])), - TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from(vec![*acc])), - TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from(vec![*acc])), + MinMaxState::Timestamp { + unit, + timezone, + acc, + } => match unit { + // `with_timezone_opt` keeps the result array's Arrow type equal to the + // field's, which `RecordBatch::try_new` checks. + TimeUnit::Millisecond => Arc::new( + TimestampMillisecondArray::from(vec![*acc]).with_timezone_opt(timezone.clone()), + ), + TimeUnit::Microsecond => Arc::new( + TimestampMicrosecondArray::from(vec![*acc]).with_timezone_opt(timezone.clone()), + ), + TimeUnit::Nanosecond => Arc::new( + TimestampNanosecondArray::from(vec![*acc]).with_timezone_opt(timezone.clone()), + ), other => { return Err(crate::Error::DataInvalid { message: format!( @@ -604,11 +709,13 @@ fn minmax_result(state: &MinMaxState, agg_name: &str, field_name: &str) -> crate } }, MinMaxState::Utf8(acc) => Arc::new(StringArray::from(vec![acc.clone()])), + MinMaxState::Binary(acc) => Arc::new(BinaryArray::from_opt_vec(vec![acc.as_deref()])), }) } fn reset_minmax(state: &mut MinMaxState) { match state { + MinMaxState::Bool(acc) => *acc = None, MinMaxState::I8(acc) => *acc = None, MinMaxState::I16(acc) => *acc = None, MinMaxState::I32(acc) => *acc = None, @@ -620,6 +727,7 @@ fn reset_minmax(state: &mut MinMaxState) { MinMaxState::Time32Ms(acc) => *acc = None, MinMaxState::Timestamp { acc, .. } => *acc = None, MinMaxState::Utf8(acc) => *acc = None, + MinMaxState::Binary(acc) => *acc = None, } } @@ -808,8 +916,9 @@ fn decimal_array( mod tests { use super::*; use crate::spec::{ - BigIntType, CharType, DateType, DecimalType, DoubleType, FloatType, IntType, SmallIntType, - TimeType, TimestampType, TinyIntType, VarCharType, + BigIntType, BinaryType, CharType, DateType, DecimalType, DoubleType, FloatType, IntType, + LocalZonedTimestampType, SmallIntType, TimeType, TimestampType, TinyIntType, VarBinaryType, + VarCharType, }; use arrow_array::builder::Decimal128Builder; @@ -841,6 +950,24 @@ mod tests { } } + fn collect_bool(arr: ArrayRef) -> Option { + let a = arr.as_any().downcast_ref::().unwrap(); + if a.is_null(0) { + None + } else { + Some(a.value(0)) + } + } + + fn collect_binary(arr: ArrayRef) -> Option> { + let a = arr.as_any().downcast_ref::().unwrap(); + if a.is_null(0) { + None + } else { + Some(a.value(0).to_vec()) + } + } + fn collect_string(arr: ArrayRef) -> Option { let a = arr.as_any().downcast_ref::().unwrap(); if a.is_null(0) { @@ -1260,12 +1387,82 @@ mod tests { #[test] fn test_min_rejects_unsupported_type() { - // Boolean has no <, > defined for min/max in Paimon basic mode. - let err = - MinAgg::new("v", &DataType::Boolean(crate::spec::BooleanType::new())).unwrap_err(); + // Java's `TypeCheckUtils#isComparable` rejects exactly MAP, MULTISET, ROW, + // ARRAY, VECTOR, VARIANT and BLOB. BLOB is the interesting one here: it + // shares Arrow's `Binary` with BINARY and VARBINARY, which min/max accepts. + let err = MinAgg::new("v", &DataType::Blob(crate::spec::BlobType::new())).unwrap_err(); assert!(matches!(err, crate::Error::ConfigInvalid { message } if message.contains("min"))); } + #[test] + fn test_min_max_boolean() { + // `Boolean.compare` in Java's `InternalRowUtils#compare`: false < true. + let arr = BooleanArray::from(vec![Some(true), Some(false), Some(true)]); + let mut min = min_agg(DataType::Boolean(crate::spec::BooleanType::new())); + let mut max = max_agg(DataType::Boolean(crate::spec::BooleanType::new())); + for i in 0..arr.len() { + min.agg(&arr, i).unwrap(); + max.agg(&arr, i).unwrap(); + } + assert_eq!(collect_bool(min.result().unwrap()), Some(false)); + assert_eq!(collect_bool(max.result().unwrap()), Some(true)); + } + + #[test] + fn test_min_max_binary_compares_bytes_as_unsigned() { + // `byteArrayCompare` masks with 0xff, so 0x80 is above 0x7f, and a prefix + // sorts before the longer value it prefixes. + let arr = BinaryArray::from_opt_vec(vec![ + Some(&[0x7fu8, 0x01][..]), + Some(&[0x80u8][..]), + Some(&[0x7fu8][..]), + ]); + for dt in [ + DataType::Binary(BinaryType::new(2).unwrap()), + DataType::VarBinary(VarBinaryType::new(8).unwrap()), + ] { + 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!(collect_binary(min.result().unwrap()), Some(vec![0x7f])); + assert_eq!(collect_binary(max.result().unwrap()), Some(vec![0x80])); + } + } + + #[test] + fn test_min_max_local_zoned_timestamp_keeps_its_arrow_timezone() { + let dt = DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()); + let arr = TimestampMillisecondArray::from(vec![Some(30), Some(10), Some(20)]) + .with_timezone_opt(Some("UTC")); + let mut min = min_agg(dt.clone()); + let mut max = max_agg(dt.clone()); + for i in 0..arr.len() { + min.agg(&arr, i).unwrap(); + max.agg(&arr, i).unwrap(); + } + let min_result = min.result().unwrap(); + assert_eq!( + min_result + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 10 + ); + + // The trap: an accumulator that dropped the timezone still holds the right + // number, but the field's Arrow type no longer matches and the batch the + // merge engine builds is rejected. + let arrow_type = crate::arrow::paimon_type_to_arrow(&dt).unwrap(); + assert_eq!(min_result.data_type(), &arrow_type); + let schema = + arrow_schema::Schema::new(vec![arrow_schema::Field::new("v", arrow_type, true)]); + arrow_array::RecordBatch::try_new(Arc::new(schema), vec![max.result().unwrap()]).unwrap(); + } + #[test] fn test_min_max_date_and_timestamp_supported() { // Date32