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/08/29 09:14:23 UTC
[PR] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
fresh-borzoni opened a new pull request, #4147:
URL: https://github.com/apache/fluss/pull/4147
## Summary
- `append_arrow_batch` took the caller's column types on trust and wrote the buffers verbatim. Readers rebuild with the table's schema, so a mismatched type is reinterpreted rather than converted: `timestamp[ns]` in a `TIMESTAMP(3)` column reads back as milliseconds, and `decimal(10,4)` in a `DECIMAL(10,2)` shifts two digits. Writing `123.4567` and reading `12345.67` produces no error anywhere — the wrong values just land in the log and get tiered to the lake.
- This rejects the batch instead, naming the column and both types.
- Nested field names and nullability are normalised before comparing, since neither changes how a value is read. Without that, every map batch from `MapBuilder` would be refused: it emits `keys`/`values` where `to_arrow_schema` uses `key`/`value`.
- Goes back to #34, which added `append_arrow_batch` without comparing the caller's schema against the table's.
**Behaviour change:** a producer writing mismatched types now gets an error where the write used to succeed. The data was being misread all along, but the error is new. Casting to the table's schema was the alternative — `write_pandas` already does that — but it copies the column, which this API exists to avoid, and `ns` to `ms` loses precision silently.
## Test Plan
The nullability harness added in #3950 can't catch this: a mistyped column decodes fine, it just decodes to a different number. So the new test writes each batch out, reads it back, and asserts the values are identical. Reverting the check fails it:
```
timestamp[ns] into TIMESTAMP(3): values changed on the way through
left: PrimitiveArray<Timestamp(ns)> right: PrimitiveArray<Timestamp(ms)>
```
Rejected: timestamp unit (`ns`/`us` into `TIMESTAMP(3)`), decimal scale and precision, time unit, fixed-size-binary width.
Accepted: matching types, and output from `ListBuilder`, `MapBuilder` and a `StructArray` with a differently named field.
Run locally against the pinned test cluster:
- rust unit: 731 pass
- rust integration: 73 pass, 5 fail — identical on unmodified `main`, all `table.statistics.columns` cluster-version errors
- python bindings: 84 pass, 9 fail — same cluster-version cause, none from this check
- C++ bindings: 78 of 80 pass, same two cluster-version failures
- clippy and fmt clean
🤖 AI-assisted changes - reviewed by human developer
--
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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4147:
URL: https://github.com/apache/fluss/pull/4147#discussion_r3886431090
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -2999,6 +3073,349 @@ mod tests {
),
&row_type,
);
+ assert_validation_matches_reader(
+ &format!("ARRAY<{name} NOT NULL>: no nulls"),
+ &single_col_batch("tags", list_array(clean, vec![0, 2], None)),
+ &row_type,
+ );
+ }
+ }
+
+ /// The batch must be accepted and read back as what it was given. A mistyped
+ /// column decodes fine, it just decodes to a different number.
+ fn assert_round_trips(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ validate_append_record_batch(batch, row_type)
+ .unwrap_or_else(|e| panic!("{label} must be accepted: {e}"));
+ let bytes = encode_batch_bypassing_validation(batch);
+ let read_context = ReadContext::new(
+ to_arrow_schema(row_type).expect("arrow schema"),
+ Arc::new(row_type.clone()),
+ false,
+ );
+ let decoded = LogRecordsBatches::new(bytes)
+ .next()
+ .expect("one batch")
+ .expect("batch header")
Review Comment:
the first says a batch exists, the second says it parsed as it's an Option<Result<_>>
--
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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4147:
URL: https://github.com/apache/fluss/pull/4147#discussion_r3886345596
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -3001,4 +3057,202 @@ mod tests {
);
}
}
+
+ /// An accepted batch must read back what it was given. A mistyped column
+ /// decodes fine, it just decodes to a different number.
+ fn assert_round_trips_or_rejected(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ if validate_append_record_batch(batch, row_type).is_err() {
Review Comment:
Good catch, fixed 👍
--
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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #4147:
URL: https://github.com/apache/fluss/pull/4147#discussion_r3886267342
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -3001,4 +3057,202 @@ mod tests {
);
}
}
+
+ /// An accepted batch must read back what it was given. A mistyped column
+ /// decodes fine, it just decodes to a different number.
+ fn assert_round_trips_or_rejected(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ if validate_append_record_batch(batch, row_type).is_err() {
Review Comment:
so early return will mean that "must accept" cases will pass even if validation rejects them?
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -3001,4 +3057,202 @@ mod tests {
);
}
}
+
+ /// An accepted batch must read back what it was given. A mistyped column
+ /// decodes fine, it just decodes to a different number.
+ fn assert_round_trips_or_rejected(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ if validate_append_record_batch(batch, row_type).is_err() {
+ return;
+ }
+ let bytes = encode_batch_bypassing_validation(batch);
+ let read_context = ReadContext::new(
+ to_arrow_schema(row_type).expect("arrow schema"),
+ Arc::new(row_type.clone()),
+ false,
+ );
+ let decoded = LogRecordsBatches::new(bytes)
+ .next()
+ .expect("one batch")
+ .expect("batch header")
+ .record_batch(&read_context)
+ .expect("an accepted batch must decode");
+ for i in 0..batch.num_columns() {
+ assert_eq!(
+ batch.column(i).as_ref(),
+ decoded.column(i).as_ref(),
+ "{label}: column {i} changed on the way through"
+ );
+ }
+ }
+
+ fn assert_rejected(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ let err = validate_append_record_batch(batch, row_type)
+ .expect_err(&format!("{label} must be rejected"));
+ assert!(
+ err.to_string().contains("but the table declares"),
+ "{label}: unexpected error {err}"
+ );
+ assert_round_trips_or_rejected(label, batch, row_type);
+ }
+
+ #[test]
+ fn column_types_must_match_the_table() {
Review Comment:
can we add nested reject as well? e.g. list of timestamp ns into array<timestamp(3)> or something like that
--
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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #4147:
URL: https://github.com/apache/fluss/pull/4147#discussion_r3886405470
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -2999,6 +3073,349 @@ mod tests {
),
&row_type,
);
+ assert_validation_matches_reader(
+ &format!("ARRAY<{name} NOT NULL>: no nulls"),
+ &single_col_batch("tags", list_array(clean, vec![0, 2], None)),
+ &row_type,
+ );
+ }
+ }
+
+ /// The batch must be accepted and read back as what it was given. A mistyped
+ /// column decodes fine, it just decodes to a different number.
+ fn assert_round_trips(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ validate_append_record_batch(batch, row_type)
+ .unwrap_or_else(|e| panic!("{label} must be accepted: {e}"));
+ let bytes = encode_batch_bypassing_validation(batch);
+ let read_context = ReadContext::new(
+ to_arrow_schema(row_type).expect("arrow schema"),
+ Arc::new(row_type.clone()),
+ false,
+ );
+ let decoded = LogRecordsBatches::new(bytes)
+ .next()
+ .expect("one batch")
+ .expect("batch header")
Review Comment:
i wonder why run expect twice?
--
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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #4147:
URL: https://github.com/apache/fluss/pull/4147
--
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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #4147:
URL: https://github.com/apache/fluss/pull/4147#issuecomment-5461539511
@leekeiabstraction @charlesdong1991 PTAL, correctness issue, so I'd better include it into the release
--
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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #4147:
URL: https://github.com/apache/fluss/pull/4147#issuecomment-5461840284
@charlesdong1991 addressed 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] [rust] Reject append_arrow_batch columns whose type differs from the table [fluss]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #4147:
URL: https://github.com/apache/fluss/pull/4147#discussion_r3886756937
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -671,9 +674,185 @@ fn parse_ipc_message(
Ok((batch_metadata, body_buffer, message.version()))
}
-pub(crate) fn validate_append_record_batch(batch: &RecordBatch, row_type: &RowType) -> Result<()> {
- let typed = TypedBatch::build(batch, row_type)?;
- typed.check_not_null(row_type)
+/// Drops what does not change how a value is read: nested field names, which
+/// readers take by position, nullability, which `check_not_null` covers, and the
+/// name of a timestamp's zone.
+fn erase_read_irrelevant(data_type: &ArrowDataType) -> ArrowDataType {
+ // Arrow names the list element and the map entries struct itself; builders
+ // disagree on what to call them, and nothing is identified by them.
+ let anonymous =
+ |f: &Arc<Field>| Arc::new(Field::new("", erase_read_irrelevant(f.data_type()), true));
+ // A field the table named. Its name says which value it is, so it is kept.
+ let named = |f: &Arc<Field>| {
+ Arc::new(Field::new(
+ f.name(),
+ erase_read_irrelevant(f.data_type()),
+ true,
+ ))
+ };
+ match data_type {
+ ArrowDataType::List(f) => ArrowDataType::List(anonymous(f)),
+ // A map's key and value are Arrow's own names and builders disagree on
+ // them (`keys`/`values` vs `key`/`value`), so they are dropped too. That
+ // leaves a swap undetectable when key and value share a type - callers
+ // must build entries as (key, value).
+ ArrowDataType::Map(f, _) => {
+ let entries = match f.data_type() {
+ ArrowDataType::Struct(kv) => {
+ ArrowDataType::Struct(kv.iter().map(anonymous).collect::<Vec<_>>().into())
+ }
+ other => erase_read_irrelevant(other),
+ };
+ ArrowDataType::Map(Arc::new(Field::new("", entries, true)), false)
+ }
+ ArrowDataType::Struct(fields) => {
+ ArrowDataType::Struct(fields.iter().map(named).collect::<Vec<_>>().into())
+ }
+ // Zoned or not is a real difference; which zone it names is not, since
+ // the stored value is the same instant either way.
+ ArrowDataType::Timestamp(unit, zone) => {
+ ArrowDataType::Timestamp(*unit, zone.as_ref().map(|_| "".into()))
+ }
+ other => other.clone(),
+ }
+}
+
+/// A mismatched Arrow type is reread as something else - `timestamp[ns]` in a
+/// `TIMESTAMP(3)` column comes back as milliseconds, `decimal(10,4)` in a
+/// `DECIMAL(10,2)` shifts two digits.
+fn check_column_types(batch: &RecordBatch, row_type: &RowType) -> Result<()> {
+ let expected = to_arrow_schema(row_type)?;
+ if batch.num_columns() != expected.fields().len() {
+ return Err(IllegalArgument {
+ message: format!(
+ "RecordBatch has {} columns but the table has {}",
+ batch.num_columns(),
+ expected.fields().len()
+ ),
+ });
+ }
+ for (i, field) in expected.fields().iter().enumerate() {
+ // Columns are matched by position, so a batch whose columns are in a
+ // different order would silently write into the wrong ones.
+ let actual_name = batch.schema().field(i).name().clone();
+ if actual_name != *field.name() {
+ return Err(IllegalArgument {
+ message: format!(
+ "Column {i} is named '{actual_name}' but the table declares '{}'",
+ field.name()
+ ),
+ });
+ }
+ let actual = batch.column(i).data_type();
+ if erase_read_irrelevant(actual) != erase_read_irrelevant(field.data_type()) {
+ return Err(IllegalArgument {
+ message: format!(
+ "Column '{}' has Arrow type {} but the table declares {}",
+ field.name(),
+ actual,
+ field.data_type()
+ ),
+ });
+ }
+ }
+ Ok(())
+}
+
+/// True when `from` differs from `to` only in how it stores offsets or encodes
+/// values, so converting it cannot change what the values are. `large_` variants
+/// and views differ in offset width; a dictionary is the same values, encoded.
+fn is_lossless_encoding_difference(from: &ArrowDataType, to: &ArrowDataType) -> bool {
+ use ArrowDataType::*;
+ match (from, to) {
+ (LargeUtf8 | Utf8View, Utf8) => true,
+ (LargeBinary | BinaryView, Binary) => true,
+ (LargeList(f) | List(f), List(t)) => is_same_or_lossless(f.data_type(), t.data_type()),
+ (Dictionary(_, v), to) => is_same_or_lossless(v, to),
+ // Casting a struct relabels its fields to the target's, so the names
+ // have to line up already - otherwise a swap would be converted into
+ // the table's shape and lose which value was which.
+ (Struct(from_fields), Struct(to_fields)) => {
+ from_fields.len() == to_fields.len()
+ && from_fields.iter().zip(to_fields.iter()).all(|(f, t)| {
+ f.name() == t.name() && is_same_or_lossless(f.data_type(), t.data_type())
+ })
+ }
+ // A map's entries are positional, so only the key and value types are
+ // compared - their names differ between builders.
+ (Map(from_entries, _), Map(to_entries, _)) => {
+ match (from_entries.data_type(), to_entries.data_type()) {
+ (Struct(from_kv), Struct(to_kv)) => {
+ from_kv.len() == to_kv.len()
+ && from_kv
+ .iter()
+ .zip(to_kv.iter())
+ .all(|(f, t)| is_same_or_lossless(f.data_type(), t.data_type()))
+ }
+ _ => false,
+ }
+ }
+ _ => false,
+ }
+}
+
+fn is_same_or_lossless(from: &ArrowDataType, to: &ArrowDataType) -> bool {
+ erase_read_irrelevant(from) == erase_read_irrelevant(to)
+ || is_lossless_encoding_difference(from, to)
+}
+
+/// Returns `batch` ready to append: checked against `row_type`, and with any
+/// column whose type differs only by encoding converted to the table's. Columns
+/// that already match are passed through untouched, so a batch built against the
+/// table's schema is not copied.
+pub(crate) fn prepare_append_record_batch(
+ batch: &RecordBatch,
+ row_type: &RowType,
+) -> Result<RecordBatch> {
+ let batch = convert_lossless_columns(batch, row_type)?;
+ check_column_types(&batch, row_type)?;
+ TypedBatch::build(&batch, row_type)?.check_not_null(row_type)?;
+ Ok(batch)
+}
+
+fn convert_lossless_columns(batch: &RecordBatch, row_type: &RowType) -> Result<RecordBatch> {
+ let expected = to_arrow_schema(row_type)?;
+ if batch.num_columns() != expected.fields().len() {
+ return Ok(batch.clone());
+ }
+ let needs_convert = |i: usize, want: &ArrowDataType| {
+ let have = batch.column(i).data_type();
+ erase_read_irrelevant(have) != erase_read_irrelevant(want)
+ && is_lossless_encoding_difference(have, want)
+ };
+ if !expected
+ .fields()
+ .iter()
+ .enumerate()
+ .any(|(i, f)| needs_convert(i, f.data_type()))
+ {
+ return Ok(batch.clone());
+ }
+
+ let mut columns = Vec::with_capacity(batch.num_columns());
+ for (i, field) in expected.fields().iter().enumerate() {
+ let column = batch.column(i);
+ if needs_convert(i, field.data_type()) {
+ columns.push(arrow::compute::cast(column, field.data_type())?);
+ } else {
+ columns.push(Arc::clone(column));
+ }
+ }
+ Ok(RecordBatch::try_new(
+ Arc::new(arrow_schema::Schema::new(
+ expected
+ .fields()
+ .iter()
+ .zip(columns.iter())
+ .map(|(f, c)| Field::new(f.name(), c.data_type().clone(), true))
Review Comment:
this rebuilds batch schema using table's column names, and name check runs after conversion... if any column needs converting, i think caller's name will be overwritten
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -671,9 +674,185 @@ fn parse_ipc_message(
Ok((batch_metadata, body_buffer, message.version()))
}
-pub(crate) fn validate_append_record_batch(batch: &RecordBatch, row_type: &RowType) -> Result<()> {
- let typed = TypedBatch::build(batch, row_type)?;
- typed.check_not_null(row_type)
+/// Drops what does not change how a value is read: nested field names, which
+/// readers take by position, nullability, which `check_not_null` covers, and the
+/// name of a timestamp's zone.
+fn erase_read_irrelevant(data_type: &ArrowDataType) -> ArrowDataType {
+ // Arrow names the list element and the map entries struct itself; builders
+ // disagree on what to call them, and nothing is identified by them.
+ let anonymous =
+ |f: &Arc<Field>| Arc::new(Field::new("", erase_read_irrelevant(f.data_type()), true));
+ // A field the table named. Its name says which value it is, so it is kept.
+ let named = |f: &Arc<Field>| {
+ Arc::new(Field::new(
+ f.name(),
+ erase_read_irrelevant(f.data_type()),
+ true,
+ ))
+ };
+ match data_type {
+ ArrowDataType::List(f) => ArrowDataType::List(anonymous(f)),
+ // A map's key and value are Arrow's own names and builders disagree on
+ // them (`keys`/`values` vs `key`/`value`), so they are dropped too. That
+ // leaves a swap undetectable when key and value share a type - callers
+ // must build entries as (key, value).
+ ArrowDataType::Map(f, _) => {
+ let entries = match f.data_type() {
+ ArrowDataType::Struct(kv) => {
+ ArrowDataType::Struct(kv.iter().map(anonymous).collect::<Vec<_>>().into())
+ }
+ other => erase_read_irrelevant(other),
+ };
+ ArrowDataType::Map(Arc::new(Field::new("", entries, true)), false)
+ }
+ ArrowDataType::Struct(fields) => {
+ ArrowDataType::Struct(fields.iter().map(named).collect::<Vec<_>>().into())
+ }
+ // Zoned or not is a real difference; which zone it names is not, since
+ // the stored value is the same instant either way.
+ ArrowDataType::Timestamp(unit, zone) => {
+ ArrowDataType::Timestamp(*unit, zone.as_ref().map(|_| "".into()))
+ }
+ other => other.clone(),
+ }
+}
+
+/// A mismatched Arrow type is reread as something else - `timestamp[ns]` in a
+/// `TIMESTAMP(3)` column comes back as milliseconds, `decimal(10,4)` in a
+/// `DECIMAL(10,2)` shifts two digits.
+fn check_column_types(batch: &RecordBatch, row_type: &RowType) -> Result<()> {
+ let expected = to_arrow_schema(row_type)?;
+ if batch.num_columns() != expected.fields().len() {
+ return Err(IllegalArgument {
+ message: format!(
+ "RecordBatch has {} columns but the table has {}",
+ batch.num_columns(),
+ expected.fields().len()
+ ),
+ });
+ }
+ for (i, field) in expected.fields().iter().enumerate() {
+ // Columns are matched by position, so a batch whose columns are in a
+ // different order would silently write into the wrong ones.
+ let actual_name = batch.schema().field(i).name().clone();
+ if actual_name != *field.name() {
+ return Err(IllegalArgument {
+ message: format!(
+ "Column {i} is named '{actual_name}' but the table declares '{}'",
+ field.name()
+ ),
+ });
+ }
+ let actual = batch.column(i).data_type();
+ if erase_read_irrelevant(actual) != erase_read_irrelevant(field.data_type()) {
+ return Err(IllegalArgument {
+ message: format!(
+ "Column '{}' has Arrow type {} but the table declares {}",
+ field.name(),
+ actual,
+ field.data_type()
+ ),
+ });
+ }
+ }
+ Ok(())
+}
+
+/// True when `from` differs from `to` only in how it stores offsets or encodes
+/// values, so converting it cannot change what the values are. `large_` variants
+/// and views differ in offset width; a dictionary is the same values, encoded.
+fn is_lossless_encoding_difference(from: &ArrowDataType, to: &ArrowDataType) -> bool {
+ use ArrowDataType::*;
+ match (from, to) {
+ (LargeUtf8 | Utf8View, Utf8) => true,
+ (LargeBinary | BinaryView, Binary) => true,
+ (LargeList(f) | List(f), List(t)) => is_same_or_lossless(f.data_type(), t.data_type()),
Review Comment:
this complex case is quite important, but is untested... can you add two accept tests, ROW with a large_string child
and ARRAY given list <large_string> for instance?
##########
fluss-rust/crates/fluss/src/record/arrow.rs:
##########
@@ -2999,6 +3199,482 @@ mod tests {
),
&row_type,
);
+ assert_validation_matches_reader(
+ &format!("ARRAY<{name} NOT NULL>: no nulls"),
+ &single_col_batch("tags", list_array(clean, vec![0, 2], None)),
+ &row_type,
+ );
+ }
+ }
+
+ /// The batch must be accepted and read back as what it was given. A mistyped
+ /// column decodes fine, it just decodes to a different number.
+ fn assert_round_trips(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ prepare_append_record_batch(batch, row_type)
+ .unwrap_or_else(|e| panic!("{label} must be accepted: {e}"));
+ let bytes = encode_batch_bypassing_validation(batch);
+ let read_context = ReadContext::new(
+ to_arrow_schema(row_type).expect("arrow schema"),
+ Arc::new(row_type.clone()),
+ false,
+ );
+ let decoded = LogRecordsBatches::new(bytes)
+ .next()
+ .expect("a batch was written")
+ .expect("the batch parses")
+ .record_batch(&read_context)
+ .expect("an accepted batch must decode");
+ for i in 0..batch.num_columns() {
+ assert_eq!(
+ batch.column(i).as_ref(),
+ decoded.column(i).as_ref(),
+ "{label}: column {i} changed on the way through"
+ );
+ }
+ }
+
+ fn assert_rejected(label: &str, batch: &RecordBatch, row_type: &RowType) {
+ let err = prepare_append_record_batch(batch, row_type)
+ .expect_err(&format!("{label} must be rejected"));
+ assert!(
+ err.to_string().contains("but the table declares"),
+ "{label}: unexpected error {err}"
+ );
+ }
+
+ #[test]
+ fn column_types_must_match_the_table() {
+ let ts3 = RowType::new(vec![DataField::new(
+ "ts",
+ DataTypes::timestamp_with_precision(3),
+ None,
+ )]);
+ assert_round_trips(
+ "timestamp[ms] into TIMESTAMP(3)",
+ &single_col_batch(
+ "ts",
+ Arc::new(TimestampMillisecondArray::from(vec![1_700_i64])),
+ ),
+ &ts3,
+ );
+ assert_rejected(
+ "timestamp[ns] into TIMESTAMP(3)",
+ &single_col_batch(
+ "ts",
+ Arc::new(TimestampNanosecondArray::from(vec![1_700_i64])),
+ ),
+ &ts3,
+ );
+ assert_rejected(
+ "timestamp[us] into TIMESTAMP(3)",
+ &single_col_batch(
+ "ts",
+ Arc::new(TimestampMicrosecondArray::from(vec![1_700_i64])),
+ ),
+ &ts3,
+ );
+
+ let dec = RowType::new(vec![DataField::new("d", DataTypes::decimal(10, 2), None)]);
+ let decimal = |p, s| {
+ single_col_batch(
+ "d",
+ Arc::new(
+ Decimal128Array::from(vec![1_234_567_i128])
+ .with_precision_and_scale(p, s)
+ .unwrap(),
+ ),
+ )
+ };
+ assert_round_trips("decimal(10,2) into DECIMAL(10,2)", &decimal(10, 2), &dec);
+ assert_rejected("decimal(10,4) into DECIMAL(10,2)", &decimal(10, 4), &dec);
+ assert_rejected("decimal(12,2) into DECIMAL(10,2)", &decimal(12, 2), &dec);
+
+ let time3 = RowType::new(vec![DataField::new(
+ "t",
+ DataTypes::time_with_precision(3),
+ None,
+ )]);
+ assert_round_trips(
+ "time32[ms] into TIME(3)",
+ &single_col_batch("t", Arc::new(Time32MillisecondArray::from(vec![1_000]))),
+ &time3,
+ );
+ assert_rejected(
+ "time64[us] into TIME(3)",
+ &single_col_batch("t", Arc::new(Time64MicrosecondArray::from(vec![1_000_i64]))),
+ &time3,
+ );
+
+ let bin8 = RowType::new(vec![DataField::new("b", DataTypes::binary(8), None)]);
+ assert_rejected(
+ "fixed_size_binary(4) into BINARY(8)",
+ &single_col_batch(
+ "b",
+ Arc::new(
+ FixedSizeBinaryArray::try_from_iter(vec![vec![1_u8, 2, 3, 4]].into_iter())
+ .unwrap(),
+ ),
+ ),
+ &bin8,
+ );
+ }
+
+ /// Types that differ only in offset width or encoding are converted to the
+ /// table's, since converting cannot change the values. A column that already
+ /// matches keeps its buffers.
+ #[test]
+ fn encoding_differences_are_converted_not_rejected() {
+ use arrow::array::{
+ Array, DictionaryArray, Int32Array, LargeBinaryArray, LargeListArray, LargeStringArray,
+ StringViewArray,
+ };
+ use arrow::buffer::OffsetBuffer;
+
+ let string_col = RowType::new(vec![DataField::new("s", DataTypes::string(), None)]);
+ for (label, column) in [
+ (
+ "large_string",
+ Arc::new(LargeStringArray::from(vec!["a", "b"])) as ArrayRef,
+ ),
+ (
+ "string_view",
+ Arc::new(StringViewArray::from(vec!["a", "b"])) as ArrayRef,
+ ),
+ (
+ "dictionary",
+ Arc::new(
+ vec!["a", "b"]
+ .into_iter()
+ .collect::<DictionaryArray<arrow::datatypes::Int32Type>>(),
+ ) as ArrayRef,
+ ),
+ ] {
+ let out = prepare_append_record_batch(&single_col_batch("s", column), &string_col)
+ .unwrap_or_else(|e| panic!("{label} must be accepted: {e}"));
+ assert_eq!(
+ out.column(0).data_type(),
+ &ArrowDataType::Utf8,
+ "{label} must arrive as the table's type"
+ );
+ let strings = out
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow::array::StringArray>()
+ .expect("converted to StringArray");
+ assert_eq!(strings.value(0), "a");
+ assert_eq!(strings.value(1), "b");
+ }
+
+ let bytes_col = RowType::new(vec![DataField::new("b", DataTypes::bytes(), None)]);
+ let out = prepare_append_record_batch(
+ &single_col_batch("b", Arc::new(LargeBinaryArray::from(vec![&b"hi"[..]]))),
+ &bytes_col,
+ )
+ .expect("large_binary must be accepted");
+ assert_eq!(out.column(0).data_type(), &ArrowDataType::Binary);
+
+ let list_col = RowType::new(vec![DataField::new(
+ "l",
+ DataTypes::array(DataTypes::int()),
+ None,
+ )]);
+ let large_list: ArrayRef = Arc::new(LargeListArray::new(
+ Arc::new(Field::new("item", ArrowDataType::Int32, true)),
+ OffsetBuffer::new(vec![0_i64, 2].into()),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ None,
+ ));
+ let out = prepare_append_record_batch(&single_col_batch("l", large_list), &list_col)
+ .expect("large_list must be accepted");
+ assert!(matches!(out.column(0).data_type(), ArrowDataType::List(_)));
+
+ // A matching column is passed through, not rebuilt.
+ let matching = single_col_batch("s", Arc::new(arrow::array::StringArray::from(vec!["a"])));
+ let out = prepare_append_record_batch(&matching, &string_col).expect("accepted");
+ assert!(
+ Arc::ptr_eq(matching.column(0), out.column(0)),
+ "a column that already matches must keep its buffers"
+ );
+ }
+
+ /// A ROW's fields are matched by position too, so the same swap is possible
+ /// one level down.
+ #[test]
+ fn row_field_names_must_match_the_table() {
+ use arrow::array::StructArray;
+
+ let row_type = RowType::new(vec![DataField::new(
+ "person",
+ DataTypes::row(vec![
+ DataField::new("name_id", DataTypes::int(), None),
+ DataField::new("surname_id", DataTypes::int(), None),
+ ]),
+ None,
+ )]);
+ let person = |first: &str, second: &str| {
+ single_col_batch(
+ "person",
+ Arc::new(StructArray::from(vec![
+ (
+ Arc::new(Field::new(first, ArrowDataType::Int32, true)),
+ Arc::new(arrow::array::Int32Array::from(vec![1])) as ArrayRef,
+ ),
+ (
+ Arc::new(Field::new(second, ArrowDataType::Int32, true)),
+ Arc::new(arrow::array::Int32Array::from(vec![999])) as ArrayRef,
+ ),
+ ])),
+ )
+ };
+
+ assert_round_trips(
+ "ROW fields in table order",
+ &person("name_id", "surname_id"),
+ &row_type,
+ );
+ assert_rejected(
+ "ROW fields swapped",
+ &person("surname_id", "name_id"),
+ &row_type,
+ );
+ }
+
+ /// Columns are matched by position. Two same-typed columns supplied in the
+ /// wrong order would otherwise be written into each other's place.
+ #[test]
+ fn column_names_must_match_the_table() {
+ let row_type = RowType::new(vec![
+ DataField::new("user_id", DataTypes::int(), None),
+ DataField::new("account_id", DataTypes::int(), None),
+ ]);
+ let batch = |first: &str, second: &str| {
+ RecordBatch::try_new(
+ Arc::new(arrow_schema::Schema::new(vec![
+ Field::new(first, ArrowDataType::Int32, true),
Review Comment:
echoing to the previous comment, can we also test with a case that requires converting?
--
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