You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/01/22 00:43:34 UTC
[PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
fresh-borzoni opened a new pull request, #196:
URL: https://github.com/apache/fluss-rust/pull/196
closes #191
Enables Arrow serialization for decimal and temporal types, providing feature parity with the Java client.
| Fluss Type | Arrow Type |
|------------|-----------|
| `DECIMAL(p, s)` | `Decimal128(p, s)` |
| `DATE` | `Date32` |
| `TIME` | `Time32/Time64` |
| `TIMESTAMP` | `Timestamp` (all precisions) |
| `TIMESTAMP_LTZ` | `Timestamp` (all precisions) |
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#discussion_r2716603781
##########
crates/fluss/src/row/datum.rs:
##########
@@ -298,16 +450,297 @@ impl Datum<'_> {
Datum::Float64(v) => append_value_to_arrow!(Float64Builder, v.into_inner()),
Datum::String(v) => append_value_to_arrow!(StringBuilder, v.as_ref()),
Datum::Blob(v) => append_value_to_arrow!(BinaryBuilder, v.as_ref()),
- Datum::Decimal(_)
- | Datum::Date(_)
- | Datum::Time(_)
- | Datum::TimestampNtz(_)
- | Datum::TimestampLtz(_) => {
+ Datum::Decimal(decimal) => {
Review Comment:
Should we have test cases for these on both rust and Java side to ensure parity?
Also, could arrow's rust library already be handling these (on Java side, we do use arrow's library).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#discussion_r2719385787
##########
crates/fluss/src/row/datum.rs:
##########
@@ -298,16 +450,297 @@ impl Datum<'_> {
Datum::Float64(v) => append_value_to_arrow!(Float64Builder, v.into_inner()),
Datum::String(v) => append_value_to_arrow!(StringBuilder, v.as_ref()),
Datum::Blob(v) => append_value_to_arrow!(BinaryBuilder, v.as_ref()),
- Datum::Decimal(_)
- | Datum::Date(_)
- | Datum::Time(_)
- | Datum::TimestampNtz(_)
- | Datum::TimestampLtz(_) => {
+ Datum::Decimal(decimal) => {
Review Comment:
Not really, we have internal type and we need to convert it.
Arrow builders doesn't know how to handle our Fluss Decimal type, Java does almost the same, maybe with less checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#discussion_r2718975466
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -172,61 +175,129 @@ pub struct RowAppendRecordBatchBuilder {
}
impl RowAppendRecordBatchBuilder {
- pub fn new(row_type: &RowType) -> Self {
- let schema_ref = to_arrow_schema(row_type);
- let builders = Mutex::new(
- schema_ref
- .fields()
- .iter()
- .map(|field| Self::create_builder(field.data_type()))
- .collect(),
- );
- Self {
+ pub fn new(row_type: &RowType) -> Result<Self> {
+ let schema_ref = to_arrow_schema(row_type)?;
+ let builders: Result<Vec<_>> = schema_ref
+ .fields()
+ .iter()
+ .map(|field| Self::create_builder(field.data_type()))
+ .collect();
+ Ok(Self {
table_schema: schema_ref.clone(),
- arrow_column_builders: builders,
+ arrow_column_builders: Mutex::new(builders?),
Review Comment:
I think we can remove the `Mutex` now since we already make `ArrowLogWriteBatch#build` mut
##########
crates/fluss/src/row/datum.rs:
##########
@@ -298,16 +450,297 @@ impl Datum<'_> {
Datum::Float64(v) => append_value_to_arrow!(Float64Builder, v.into_inner()),
Datum::String(v) => append_value_to_arrow!(StringBuilder, v.as_ref()),
Datum::Blob(v) => append_value_to_arrow!(BinaryBuilder, v.as_ref()),
- Datum::Decimal(_)
- | Datum::Date(_)
- | Datum::Time(_)
- | Datum::TimestampNtz(_)
- | Datum::TimestampLtz(_) => {
+ Datum::Decimal(decimal) => {
+ // Extract target precision and scale from Arrow schema
+ let (p, s) = match data_type {
+ arrow_schema::DataType::Decimal128(p, s) => (*p, *s),
+ _ => {
+ return Err(RowConvertError {
+ message: format!(
+ "Expected Decimal128 Arrow type, got: {:?}",
+ data_type
+ ),
+ });
+ }
+ };
+
+ // Validate scale is non-negative (Fluss doesn't support negative scales)
+ if s < 0 {
+ return Err(RowConvertError {
+ message: format!("Negative decimal scale {} is not supported", s),
+ });
+ }
+
+ let target_precision = p as u32;
+ let target_scale = s as i64; // Safe now: 0..127 â 0i64..127i64
+
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Decimal128Builder>() {
+ use bigdecimal::RoundingMode;
+
+ // Rescale the decimal to match Arrow's target scale
+ let bd = decimal.to_big_decimal();
+ let rescaled = bd.with_scale_round(target_scale, RoundingMode::HalfUp);
+ let (unscaled, _) = rescaled.as_bigint_and_exponent();
+
+ // Validate precision
+ let actual_precision = Decimal::compute_precision(&unscaled);
+ if actual_precision > target_precision as usize {
+ return Err(RowConvertError {
+ message: format!(
+ "Decimal precision overflow: value has {} digits but Arrow expects {} (value: {})",
+ actual_precision, target_precision, rescaled
+ ),
+ });
+ }
+
+ // Convert to i128 for Arrow
+ let i128_val: i128 = match unscaled.try_into() {
+ Ok(v) => v,
+ Err(_) => {
+ return Err(RowConvertError {
+ message: format!("Decimal value exceeds i128 range: {}", rescaled),
+ });
+ }
+ };
+
+ b.append_value(i128_val);
+ return Ok(());
+ }
+
+ return Err(RowConvertError {
+ message: "Builder type mismatch for Decimal128".to_string(),
+ });
+ }
+ Datum::Date(date) => {
+ append_value_to_arrow!(Date32Builder, date.get_inner());
+ }
+ Datum::Time(time) => {
+ // Time is stored as milliseconds since midnight in Fluss
+ // Convert to Arrow's time unit based on schema
+ let millis = time.get_inner();
+
+ match data_type {
+ arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Second) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Time32SecondBuilder>()
+ {
+ // Validate no sub-second precision is lost
+ if millis % MILLIS_PER_SECOND as i32 != 0 {
+ return Err(RowConvertError {
+ message: format!(
+ "Time value {} ms has sub-second precision but schema expects seconds only",
+ millis
+ ),
+ });
+ }
+ b.append_value(millis / MILLIS_PER_SECOND as i32);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Millisecond) => {
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<Time32MillisecondBuilder>()
+ {
+ b.append_value(millis);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Microsecond) => {
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<Time64MicrosecondBuilder>()
+ {
+ let micros = (millis as i64)
+ .checked_mul(MICROS_PER_MILLI)
+ .ok_or_else(|| RowConvertError {
+ message: format!(
+ "Time value {} ms overflows when converting to microseconds",
+ millis
+ ),
+ })?;
+ b.append_value(micros);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Nanosecond) => {
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<Time64NanosecondBuilder>()
+ {
+ let nanos = (millis as i64).checked_mul(NANOS_PER_MILLI).ok_or_else(
+ || RowConvertError {
+ message: format!(
+ "Time value {} ms overflows when converting to nanoseconds",
+ millis
+ ),
+ },
+ )?;
+ b.append_value(nanos);
+ return Ok(());
+ }
+ }
+ _ => {
+ return Err(RowConvertError {
+ message: format!(
+ "Expected Time32/Time64 Arrow type, got: {:?}",
+ data_type
+ ),
+ });
+ }
+ }
+
return Err(RowConvertError {
- message: format!(
- "Type {:?} is not yet supported for Arrow conversion",
- std::mem::discriminant(self)
- ),
+ message: "Builder type mismatch for Time".to_string(),
+ });
+ }
+ Datum::TimestampNtz(ts) => {
+ let millis = ts.get_millisecond();
+ let nanos = ts.get_nano_of_millisecond();
+
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampSecondBuilder>()
+ {
+ // Convert milliseconds to seconds
+ b.append_value(millis / MILLIS_PER_SECOND);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMillisecondBuilder>()
+ {
+ b.append_value(millis);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMicrosecondBuilder>()
+ {
+ // Convert milliseconds + nanos to microseconds
Review Comment:
nit: can we extract convert to `microseconds`, convert to `nanoseconds` as common method so that timstampntz, timestampltz can share it?
##########
crates/fluss/src/row/datum.rs:
##########
@@ -246,19 +285,122 @@ impl TryFrom<&Datum<'_>> for i8 {
}
}
+impl TryFrom<&Datum<'_>> for Decimal {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::Decimal(d) => Ok(d.clone()),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for Date {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::Date(d) => Ok(*d),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for Time {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::Time(t) => Ok(*t),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for TimestampNtz {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::TimestampNtz(ts) => Ok(*ts),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for TimestampLtz {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::TimestampLtz(ts) => Ok(*ts),
+ _ => Err(()),
+ }
+ }
+}
Review Comment:
Considering `Date` is just wraping a int, I think copy is fine for clean code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#issuecomment-3781893230
@luoyuxia @leekeiabstraction PTAL ð
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#discussion_r2714999454
##########
crates/fluss/src/row/datum.rs:
##########
@@ -298,16 +445,186 @@ impl Datum<'_> {
Datum::Float64(v) => append_value_to_arrow!(Float64Builder, v.into_inner()),
Datum::String(v) => append_value_to_arrow!(StringBuilder, v.as_ref()),
Datum::Blob(v) => append_value_to_arrow!(BinaryBuilder, v.as_ref()),
- Datum::Decimal(_)
- | Datum::Date(_)
- | Datum::Time(_)
- | Datum::TimestampNtz(_)
- | Datum::TimestampLtz(_) => {
+ Datum::Decimal(decimal) => {
+ // Extract target precision and scale from Arrow schema
+ let (target_precision, target_scale) = match data_type {
+ arrow_schema::DataType::Decimal128(p, s) => (*p as u32, *s as u32),
+ _ => {
+ return Err(RowConvertError {
+ message: format!(
+ "Expected Decimal128 Arrow type, got: {:?}",
+ data_type
+ ),
+ });
+ }
+ };
+
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Decimal128Builder>() {
+ use bigdecimal::RoundingMode;
+
+ // Rescale the decimal to match Arrow's target scale
+ let bd = decimal.to_big_decimal();
+ let rescaled = bd.with_scale_round(target_scale as i64, RoundingMode::HalfUp);
+ let (unscaled, _) = rescaled.as_bigint_and_exponent();
+
+ // Validate precision
+ let actual_precision = Decimal::compute_precision(&unscaled);
+ if actual_precision > target_precision as usize {
+ return Err(RowConvertError {
+ message: format!(
+ "Decimal precision overflow: value has {} digits but Arrow expects {} (value: {})",
+ actual_precision, target_precision, rescaled
+ ),
+ });
+ }
+
+ // Convert to i128 for Arrow
+ let i128_val: i128 = match unscaled.try_into() {
+ Ok(v) => v,
+ Err(_) => {
+ return Err(RowConvertError {
+ message: format!("Decimal value exceeds i128 range: {}", rescaled),
+ });
+ }
+ };
+
+ b.append_value(i128_val);
+ return Ok(());
+ }
+
return Err(RowConvertError {
- message: format!(
- "Type {:?} is not yet supported for Arrow conversion",
- std::mem::discriminant(self)
- ),
+ message: "Builder type mismatch for Decimal128".to_string(),
+ });
+ }
+ Datum::Date(date) => {
+ append_value_to_arrow!(Date32Builder, date.get_inner());
+ }
+ Datum::Time(time) => {
+ // Time is stored as milliseconds since midnight in Fluss
+ // Convert to Arrow's time unit based on schema
+ let millis = time.get_inner();
+
+ match data_type {
+ arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Second) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int32Builder>() {
+ b.append_value(millis / MILLIS_PER_SECOND as i32);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Millisecond) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int32Builder>() {
+ b.append_value(millis);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Microsecond) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int64Builder>() {
+ b.append_value(millis as i64 * MICROS_PER_MILLI);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Nanosecond) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int64Builder>() {
+ b.append_value(millis as i64 * NANOS_PER_MILLI);
+ return Ok(());
+ }
+ }
+ _ => {
+ return Err(RowConvertError {
+ message: format!(
+ "Expected Time32/Time64 Arrow type, got: {:?}",
+ data_type
+ ),
+ });
+ }
+ }
+
+ return Err(RowConvertError {
+ message: "Builder type mismatch for Time".to_string(),
+ });
+ }
+ Datum::TimestampNtz(ts) => {
+ let millis = ts.get_millisecond();
+ let nanos = ts.get_nano_of_millisecond();
+
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampSecondBuilder>()
+ {
+ // Convert milliseconds to seconds
+ b.append_value(millis / MILLIS_PER_SECOND);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMillisecondBuilder>()
+ {
+ b.append_value(millis);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMicrosecondBuilder>()
+ {
+ // Convert milliseconds + nanos to microseconds
+ let micros = millis * MICROS_PER_MILLI + (nanos as i64) / MICROS_PER_MILLI;
+ b.append_value(micros);
Review Comment:
Timestamp unit conversion multiplies `millis` by `MICROS_PER_MILLI`/`NANOS_PER_MILLI` without overflow checks. For sufficiently large (or corrupted) timestamps this can overflow `i64` (panic in debug, wrap in release) and corrupt the serialized value. Use `checked_mul`/`checked_add` and return `RowConvertError` on overflow.
##########
crates/fluss/src/row/datum.rs:
##########
@@ -298,16 +445,186 @@ impl Datum<'_> {
Datum::Float64(v) => append_value_to_arrow!(Float64Builder, v.into_inner()),
Datum::String(v) => append_value_to_arrow!(StringBuilder, v.as_ref()),
Datum::Blob(v) => append_value_to_arrow!(BinaryBuilder, v.as_ref()),
- Datum::Decimal(_)
- | Datum::Date(_)
- | Datum::Time(_)
- | Datum::TimestampNtz(_)
- | Datum::TimestampLtz(_) => {
+ Datum::Decimal(decimal) => {
+ // Extract target precision and scale from Arrow schema
+ let (target_precision, target_scale) = match data_type {
+ arrow_schema::DataType::Decimal128(p, s) => (*p as u32, *s as u32),
+ _ => {
Review Comment:
`DataType::Decimal128` stores `scale` as an `i8` (and Arrow allows negative scales). Casting `*s` to `u32` will turn negative scales into a huge positive number, causing incorrect rescaling and likely overflow/allocations. Keep `target_scale` as an `i8`/`i64` (no unsigned cast) and pass it directly to `with_scale_round`; also consider validating that the scale is in the range supported by your `Decimal` representation.
##########
crates/fluss/src/row/datum.rs:
##########
@@ -298,16 +445,186 @@ impl Datum<'_> {
Datum::Float64(v) => append_value_to_arrow!(Float64Builder, v.into_inner()),
Datum::String(v) => append_value_to_arrow!(StringBuilder, v.as_ref()),
Datum::Blob(v) => append_value_to_arrow!(BinaryBuilder, v.as_ref()),
- Datum::Decimal(_)
- | Datum::Date(_)
- | Datum::Time(_)
- | Datum::TimestampNtz(_)
- | Datum::TimestampLtz(_) => {
+ Datum::Decimal(decimal) => {
+ // Extract target precision and scale from Arrow schema
+ let (target_precision, target_scale) = match data_type {
+ arrow_schema::DataType::Decimal128(p, s) => (*p as u32, *s as u32),
+ _ => {
+ return Err(RowConvertError {
+ message: format!(
+ "Expected Decimal128 Arrow type, got: {:?}",
+ data_type
+ ),
+ });
+ }
+ };
+
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Decimal128Builder>() {
+ use bigdecimal::RoundingMode;
+
+ // Rescale the decimal to match Arrow's target scale
+ let bd = decimal.to_big_decimal();
+ let rescaled = bd.with_scale_round(target_scale as i64, RoundingMode::HalfUp);
+ let (unscaled, _) = rescaled.as_bigint_and_exponent();
+
+ // Validate precision
+ let actual_precision = Decimal::compute_precision(&unscaled);
+ if actual_precision > target_precision as usize {
+ return Err(RowConvertError {
+ message: format!(
+ "Decimal precision overflow: value has {} digits but Arrow expects {} (value: {})",
+ actual_precision, target_precision, rescaled
+ ),
+ });
+ }
+
+ // Convert to i128 for Arrow
+ let i128_val: i128 = match unscaled.try_into() {
+ Ok(v) => v,
+ Err(_) => {
+ return Err(RowConvertError {
+ message: format!("Decimal value exceeds i128 range: {}", rescaled),
+ });
+ }
+ };
+
+ b.append_value(i128_val);
+ return Ok(());
+ }
+
return Err(RowConvertError {
- message: format!(
- "Type {:?} is not yet supported for Arrow conversion",
- std::mem::discriminant(self)
- ),
+ message: "Builder type mismatch for Decimal128".to_string(),
+ });
+ }
+ Datum::Date(date) => {
+ append_value_to_arrow!(Date32Builder, date.get_inner());
+ }
+ Datum::Time(time) => {
+ // Time is stored as milliseconds since midnight in Fluss
+ // Convert to Arrow's time unit based on schema
+ let millis = time.get_inner();
+
+ match data_type {
+ arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Second) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int32Builder>() {
+ b.append_value(millis / MILLIS_PER_SECOND as i32);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Millisecond) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int32Builder>() {
+ b.append_value(millis);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Microsecond) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int64Builder>() {
+ b.append_value(millis as i64 * MICROS_PER_MILLI);
+ return Ok(());
+ }
+ }
+ arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Nanosecond) => {
+ if let Some(b) = builder.as_any_mut().downcast_mut::<Int64Builder>() {
+ b.append_value(millis as i64 * NANOS_PER_MILLI);
+ return Ok(());
+ }
+ }
+ _ => {
+ return Err(RowConvertError {
+ message: format!(
+ "Expected Time32/Time64 Arrow type, got: {:?}",
+ data_type
+ ),
+ });
+ }
+ }
+
+ return Err(RowConvertError {
+ message: "Builder type mismatch for Time".to_string(),
+ });
+ }
+ Datum::TimestampNtz(ts) => {
+ let millis = ts.get_millisecond();
+ let nanos = ts.get_nano_of_millisecond();
+
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampSecondBuilder>()
+ {
+ // Convert milliseconds to seconds
+ b.append_value(millis / MILLIS_PER_SECOND);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMillisecondBuilder>()
+ {
+ b.append_value(millis);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMicrosecondBuilder>()
+ {
+ // Convert milliseconds + nanos to microseconds
+ let micros = millis * MICROS_PER_MILLI + (nanos as i64) / MICROS_PER_MILLI;
+ b.append_value(micros);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampNanosecondBuilder>()
+ {
+ // Convert milliseconds + nanos to nanoseconds
+ let nanos_total = millis * NANOS_PER_MILLI + nanos as i64;
+ b.append_value(nanos_total);
+ return Ok(());
+ }
+
+ return Err(RowConvertError {
+ message: "Builder type mismatch for TimestampNtz".to_string(),
+ });
+ }
+ Datum::TimestampLtz(ts) => {
+ let millis = ts.get_epoch_millisecond();
+ let nanos = ts.get_nano_of_millisecond();
+
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampSecondBuilder>()
+ {
+ // Convert milliseconds to seconds
+ b.append_value(millis / MILLIS_PER_SECOND);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMillisecondBuilder>()
+ {
+ b.append_value(millis);
+ return Ok(());
+ }
+ if let Some(b) = builder
+ .as_any_mut()
+ .downcast_mut::<TimestampMicrosecondBuilder>()
+ {
+ // Convert milliseconds + nanos to microseconds
+ let micros = millis * MICROS_PER_MILLI + (nanos as i64) / MICROS_PER_MILLI;
+ b.append_value(micros);
Review Comment:
Timestamp unit conversion multiplies `millis` by `MICROS_PER_MILLI`/`NANOS_PER_MILLI` without overflow checks. For sufficiently large (or corrupted) timestamps this can overflow `i64` (panic in debug, wrap in release) and corrupt the serialized value. Use `checked_mul`/`checked_add` and return `RowConvertError` on overflow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#issuecomment-3787905371
@leekeiabstraction @luoyuxia
Thanks for the review. Addressed the comments. PTAL ð
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #196:
URL: https://github.com/apache/fluss-rust/pull/196
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#discussion_r2716411604
##########
crates/fluss/src/row/datum.rs:
##########
@@ -246,19 +285,122 @@ impl TryFrom<&Datum<'_>> for i8 {
}
}
+impl TryFrom<&Datum<'_>> for Decimal {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::Decimal(d) => Ok(d.clone()),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for Date {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::Date(d) => Ok(*d),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for Time {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::Time(t) => Ok(*t),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for TimestampNtz {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::TimestampNtz(ts) => Ok(*ts),
+ _ => Err(()),
+ }
+ }
+}
+
+impl TryFrom<&Datum<'_>> for TimestampLtz {
+ type Error = ();
+
+ #[inline]
+ fn try_from(from: &Datum) -> std::result::Result<Self, Self::Error> {
+ match from {
+ Datum::TimestampLtz(ts) => Ok(*ts),
+ _ => Err(()),
+ }
+ }
+}
Review Comment:
Should we prefer non-copy variant like that for `&'b str`?
```
impl <'b, 'a: 'b> TryFrom<&'b Datum<'a>> for &'b Date {
type Error = ();
#[inline]
fn try_from(from: &'b Datum<'a>) -> std::result::Result<Self, Self::Error> {
match from {
Datum::Date(d) => Ok(d),
_ => Err(()),
}
}
}
```
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -172,61 +175,129 @@ pub struct RowAppendRecordBatchBuilder {
}
impl RowAppendRecordBatchBuilder {
- pub fn new(row_type: &RowType) -> Self {
- let schema_ref = to_arrow_schema(row_type);
- let builders = Mutex::new(
- schema_ref
- .fields()
- .iter()
- .map(|field| Self::create_builder(field.data_type()))
- .collect(),
- );
- Self {
+ pub fn new(row_type: &RowType) -> Result<Self> {
+ let schema_ref = to_arrow_schema(row_type)?;
+ let builders: Result<Vec<_>> = schema_ref
+ .fields()
+ .iter()
+ .map(|field| Self::create_builder(field.data_type()))
+ .collect();
+ Ok(Self {
table_schema: schema_ref.clone(),
- arrow_column_builders: builders,
+ arrow_column_builders: Mutex::new(builders?),
Review Comment:
Can you clarify why Mutex is necessary?
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -1249,4 +1388,201 @@ mod tests {
}
out
}
+
+ #[test]
+ fn test_temporal_and_decimal_builder_validation() {
+ // Test valid builder creation
+ let builder =
+ RowAppendRecordBatchBuilder::create_builder(&ArrowDataType::Decimal128(10, 2)).unwrap();
+ assert!(
+ builder
+ .as_any()
+ .downcast_ref::<Decimal128Builder>()
+ .is_some()
Review Comment:
Should we assert precision and scale as well?
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -1249,4 +1388,201 @@ mod tests {
}
out
}
+
+ #[test]
+ fn test_temporal_and_decimal_builder_validation() {
+ // Test valid builder creation
+ let builder =
+ RowAppendRecordBatchBuilder::create_builder(&ArrowDataType::Decimal128(10, 2)).unwrap();
+ assert!(
+ builder
+ .as_any()
+ .downcast_ref::<Decimal128Builder>()
+ .is_some()
+ );
+
+ // Test error case: invalid precision/scale
+ let result =
+ RowAppendRecordBatchBuilder::create_builder(&ArrowDataType::Decimal128(100, 50));
+ assert!(result.is_err());
+ }
+
+ #[test]
+ fn test_decimal_rescaling_and_validation() -> Result<()> {
+ use crate::row::{Datum, Decimal, GenericRow};
+ use arrow::array::Decimal128Array;
+ use bigdecimal::BigDecimal;
+ use std::str::FromStr;
+
+ // Test 1: Rescaling from scale 3 to scale 2
+ let row_type = RowType::new(vec![DataField::new(
+ "amount".to_string(),
+ DataTypes::decimal(10, 2),
+ None,
+ )]);
+ let mut builder = RowAppendRecordBatchBuilder::new(&row_type)?;
+ let decimal = Decimal::from_big_decimal(BigDecimal::from_str("123.456").unwrap(), 10, 3)?;
+ builder.append(&GenericRow {
+ values: vec![Datum::Decimal(decimal)],
+ })?;
+ let batch = builder.build_arrow_record_batch()?;
+ let array = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap();
+ assert_eq!(array.value(0), 12346); // 123.46 rounded
+ assert_eq!(array.scale(), 2);
+
+ // Test 2: Precision overflow (should error)
+ let row_type = RowType::new(vec![DataField::new(
+ "amount".to_string(),
+ DataTypes::decimal(5, 2),
+ None,
+ )]);
+ let mut builder = RowAppendRecordBatchBuilder::new(&row_type)?;
+ let decimal = Decimal::from_big_decimal(BigDecimal::from_str("123456.78").unwrap(), 10, 2)?;
+ let result = builder.append(&GenericRow {
+ values: vec![Datum::Decimal(decimal)],
+ });
+ assert!(result.is_err());
+ assert!(
+ result
+ .unwrap_err()
+ .to_string()
+ .contains("precision overflow")
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_temporal_types_end_to_end() -> Result<()> {
+ use crate::row::{Date, Datum, Decimal, GenericRow, Time, TimestampLtz, TimestampNtz};
+ use arrow::array::{
+ Date32Array, Decimal128Array, Int32Array, Time32MillisecondArray,
+ Time64NanosecondArray, TimestampMicrosecondArray, TimestampNanosecondArray,
+ };
+ use bigdecimal::BigDecimal;
+ use std::str::FromStr;
+
+ // Schema with decimal, date, time (ms + ns), timestamps (Ξs + ns)
+ let row_type = RowType::new(vec![
+ DataField::new("id".to_string(), DataTypes::int(), None),
+ DataField::new("amount".to_string(), DataTypes::decimal(10, 2), None),
+ DataField::new("date".to_string(), DataTypes::date(), None),
+ DataField::new(
+ "time_ms".to_string(),
+ DataTypes::time_with_precision(3),
+ None,
+ ),
+ DataField::new(
+ "time_ns".to_string(),
+ DataTypes::time_with_precision(9),
+ None,
+ ),
+ DataField::new(
+ "ts_us".to_string(),
+ DataTypes::timestamp_with_precision(6),
+ None,
+ ),
+ DataField::new(
+ "ts_ltz_ns".to_string(),
+ DataTypes::timestamp_ltz_with_precision(9),
+ None,
+ ),
+ ]);
+
+ let mut builder = RowAppendRecordBatchBuilder::new(&row_type)?;
+
+ // Append rows with temporal values
+ builder.append(&GenericRow {
+ values: vec![
+ Datum::Int32(1),
+ Datum::Decimal(Decimal::from_big_decimal(
+ BigDecimal::from_str("123.456").unwrap(),
+ 10,
+ 3,
+ )?),
+ Datum::Date(Date::new(18000)),
+ Datum::Time(Time::new(43200000)),
+ Datum::Time(Time::new(12345)),
Review Comment:
What do these value mean in human readable date/time?
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -1249,4 +1388,201 @@ mod tests {
}
out
}
+
+ #[test]
+ fn test_temporal_and_decimal_builder_validation() {
+ // Test valid builder creation
+ let builder =
+ RowAppendRecordBatchBuilder::create_builder(&ArrowDataType::Decimal128(10, 2)).unwrap();
+ assert!(
+ builder
+ .as_any()
+ .downcast_ref::<Decimal128Builder>()
+ .is_some()
+ );
+
+ // Test error case: invalid precision/scale
+ let result =
+ RowAppendRecordBatchBuilder::create_builder(&ArrowDataType::Decimal128(100, 50));
+ assert!(result.is_err());
+ }
+
+ #[test]
+ fn test_decimal_rescaling_and_validation() -> Result<()> {
+ use crate::row::{Datum, Decimal, GenericRow};
+ use arrow::array::Decimal128Array;
+ use bigdecimal::BigDecimal;
+ use std::str::FromStr;
+
+ // Test 1: Rescaling from scale 3 to scale 2
+ let row_type = RowType::new(vec![DataField::new(
+ "amount".to_string(),
+ DataTypes::decimal(10, 2),
+ None,
+ )]);
+ let mut builder = RowAppendRecordBatchBuilder::new(&row_type)?;
+ let decimal = Decimal::from_big_decimal(BigDecimal::from_str("123.456").unwrap(), 10, 3)?;
+ builder.append(&GenericRow {
+ values: vec![Datum::Decimal(decimal)],
+ })?;
+ let batch = builder.build_arrow_record_batch()?;
+ let array = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap();
+ assert_eq!(array.value(0), 12346); // 123.46 rounded
Review Comment:
This should say 123.456 rounded to 2 decimal points instead?
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -1249,4 +1388,201 @@ mod tests {
}
out
}
+
+ #[test]
+ fn test_temporal_and_decimal_builder_validation() {
+ // Test valid builder creation
+ let builder =
+ RowAppendRecordBatchBuilder::create_builder(&ArrowDataType::Decimal128(10, 2)).unwrap();
+ assert!(
+ builder
+ .as_any()
+ .downcast_ref::<Decimal128Builder>()
+ .is_some()
+ );
+
+ // Test error case: invalid precision/scale
+ let result =
+ RowAppendRecordBatchBuilder::create_builder(&ArrowDataType::Decimal128(100, 50));
+ assert!(result.is_err());
+ }
+
+ #[test]
+ fn test_decimal_rescaling_and_validation() -> Result<()> {
+ use crate::row::{Datum, Decimal, GenericRow};
+ use arrow::array::Decimal128Array;
+ use bigdecimal::BigDecimal;
+ use std::str::FromStr;
+
+ // Test 1: Rescaling from scale 3 to scale 2
+ let row_type = RowType::new(vec![DataField::new(
+ "amount".to_string(),
+ DataTypes::decimal(10, 2),
+ None,
+ )]);
+ let mut builder = RowAppendRecordBatchBuilder::new(&row_type)?;
+ let decimal = Decimal::from_big_decimal(BigDecimal::from_str("123.456").unwrap(), 10, 3)?;
+ builder.append(&GenericRow {
+ values: vec![Datum::Decimal(decimal)],
+ })?;
+ let batch = builder.build_arrow_record_batch()?;
+ let array = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap();
+ assert_eq!(array.value(0), 12346); // 123.46 rounded
+ assert_eq!(array.scale(), 2);
+
+ // Test 2: Precision overflow (should error)
+ let row_type = RowType::new(vec![DataField::new(
+ "amount".to_string(),
+ DataTypes::decimal(5, 2),
+ None,
+ )]);
+ let mut builder = RowAppendRecordBatchBuilder::new(&row_type)?;
+ let decimal = Decimal::from_big_decimal(BigDecimal::from_str("123456.78").unwrap(), 10, 2)?;
+ let result = builder.append(&GenericRow {
+ values: vec![Datum::Decimal(decimal)],
+ });
+ assert!(result.is_err());
+ assert!(
+ result
+ .unwrap_err()
+ .to_string()
+ .contains("precision overflow")
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_temporal_types_end_to_end() -> Result<()> {
Review Comment:
We seem to be testing int and decimal in this test case as well despite name indicating temporal only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [TASK-191] Arrow serialization for decimal and temporal types [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #196:
URL: https://github.com/apache/fluss-rust/pull/196#discussion_r2719404150
##########
crates/fluss/src/row/datum.rs:
##########
@@ -298,16 +450,297 @@ impl Datum<'_> {
Datum::Float64(v) => append_value_to_arrow!(Float64Builder, v.into_inner()),
Datum::String(v) => append_value_to_arrow!(StringBuilder, v.as_ref()),
Datum::Blob(v) => append_value_to_arrow!(BinaryBuilder, v.as_ref()),
- Datum::Decimal(_)
- | Datum::Date(_)
- | Datum::Time(_)
- | Datum::TimestampNtz(_)
- | Datum::TimestampLtz(_) => {
+ Datum::Decimal(decimal) => {
Review Comment:
For parity - I'll create a separate issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org