Copilot commented on code in PR #639:
URL: https://github.com/apache/fluss-rust/pull/639#discussion_r3473093970
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -990,38 +1001,79 @@ impl LogRecordBatch {
LittleEndian::read_i32(&self.data[offset..offset +
RECORDS_COUNT_LENGTH])
}
+ /// Splits the batch body into its per-record change types and the trailing
+ /// Arrow IPC payload (see [`APPEND_ONLY_FLAG_MASK`] for the layout).
+ fn decode_change_types(&self) -> Result<(BatchChangeTypes, &[u8])> {
+ let body = self
+ .data
+ .get(RECORDS_OFFSET..)
+ .ok_or_else(|| Error::UnexpectedError {
+ message: format!(
+ "Corrupt log record batch: data length {} is less than
RECORDS_OFFSET {}",
+ self.data.len(),
+ RECORDS_OFFSET
+ ),
+ source: None,
+ })?;
+
+ if self.is_append_only() {
+ return Ok((BatchChangeTypes::Uniform(ChangeType::AppendOnly),
body));
+ }
+
+ let record_count = self.record_count() as usize;
+ let (change_type_bytes, arrow_data) =
+ body.split_at_checked(record_count)
+ .ok_or_else(|| Error::UnexpectedError {
+ message: format!(
+ "Corrupt changelog batch: body length {} is smaller
than its \
+ {record_count}-record change-type vector",
+ body.len()
+ ),
+ source: None,
+ })?;
Review Comment:
`record_count` is read as an `i32` and immediately cast to `usize`. If the
batch header is corrupted and `record_count` is negative, the cast produces a
huge `usize`, which then feeds `Vec::with_capacity(record_count)` and can
panic/abort due to an attempted enormous allocation. It also makes the
subsequent error message misleading.
Consider validating `record_count >= 0` before casting, returning an error
for negative values.
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -2349,4 +2436,164 @@ mod tests {
Ok(())
}
+
+ /// Builds an append-only `(id INT, name STRING)` Arrow log batch from
`rows`.
+ /// The writer always emits append-only batches, so changelog tests derive
+ /// their bytes from this with [`splice_change_type_vector`].
+ fn build_append_only_batch(rows: &[(i32, &str)]) -> (RowType, Vec<u8>) {
+ let row_type = RowType::new(vec![
+ DataField::new("id".to_string(), DataTypes::int(), None),
+ DataField::new("name".to_string(), DataTypes::string(), None),
+ ]);
+ let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+ let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1));
+ let physical_table_path =
Arc::new(PhysicalTablePath::of(Arc::new(table_path)));
+
+ let mut builder = MemoryLogRecordsArrowBuilder::new(
+ 1,
+ &row_type,
+ false,
+ ArrowCompressionInfo {
+ compression_type: ArrowCompressionType::None,
+ compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
+ },
+ usize::MAX,
+ Arc::new(ArrowCompressionRatioEstimator::default()),
+ )
+ .unwrap();
+
+ for (id, name) in rows {
+ let mut row = GenericRow::new(2);
+ row.set_field(0, *id);
+ row.set_field(1, *name);
+ let record = WriteRecord::for_append(
+ Arc::clone(&table_info),
+ physical_table_path.clone(),
+ 1,
+ &row,
+ );
+ builder.append(&record).unwrap();
+ }
+
+ (row_type, builder.build().unwrap())
+ }
+
+ /// Turns an append-only batch into a wire-valid changelog batch: clears
the
+ /// append-only flag, splices one change-type byte per record between the
+ /// header and the Arrow payload, then fixes up the length field and CRC.
+ fn splice_change_type_vector(append_only: &[u8], change_types:
&[ChangeType]) -> Vec<u8> {
+ let mut data = append_only.to_vec();
+ data[ATTRIBUTES_OFFSET] = 0;
+ let change_bytes = change_types.iter().map(|ct| ct.to_byte_value());
Review Comment:
This helper clears the entire attributes byte (`data[ATTRIBUTES_OFFSET] =
0`). If additional attribute bits are ever introduced, this would accidentally
wipe them. Since the intent is only to clear the append-only bit, it’s safer to
mask off `APPEND_ONLY_FLAG_MASK` instead of zeroing the whole byte.
##########
crates/fluss/src/client/table/log_fetch_buffer.rs:
##########
@@ -944,4 +969,76 @@ mod tests {
Ok(())
}
+
+ /// A `-U`/`+U` pair must not be split across polls: even when
`max_records`
+ /// falls between them, `fetch_records` pulls the matching `+U` so the
batch
+ /// ends on a complete pair (mirrors Java `CompletedFetch.fetchRecords`).
+ #[test]
+ fn fetch_records_keeps_update_before_after_pair_together() -> Result<()> {
+ let row_type = RowType::new(vec![DataField::new("id",
DataTypes::int(), None)]);
+ let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+ let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1));
+ let physical_table_path =
Arc::new(PhysicalTablePath::of(Arc::new(table_path)));
+
+ let mut builder = MemoryLogRecordsArrowBuilder::new(
+ 1,
+ &row_type,
+ false,
+ ArrowCompressionInfo {
+ compression_type: ArrowCompressionType::None,
+ compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
+ },
+ usize::MAX,
+ Arc::new(ArrowCompressionRatioEstimator::default()),
+ )?;
+ for id in [10_i32, 20, 20] {
+ let mut row = GenericRow::new(1);
+ row.set_field(0, id);
+ let record = WriteRecord::for_append(
+ Arc::clone(&table_info),
+ physical_table_path.clone(),
+ 1,
+ &row,
+ );
+ builder.append(&record)?;
+ }
+ let append_only = builder.build()?;
+
+ // Synthesize a changelog batch carrying +I, -U, +U.
+ let change_types = [
+ ChangeType::Insert,
+ ChangeType::UpdateBefore,
+ ChangeType::UpdateAfter,
+ ];
+ let mut data = Vec::with_capacity(append_only.len() +
change_types.len());
+ data.extend_from_slice(&append_only[..RECORDS_OFFSET]);
+ data.extend(change_types.iter().map(ChangeType::to_byte_value));
+ data.extend_from_slice(&append_only[RECORDS_OFFSET..]);
+ data[ATTRIBUTES_OFFSET] = 0;
+ let new_len = ((data.len() - LOG_OVERHEAD) as i32).to_le_bytes();
Review Comment:
This test clears the whole attributes byte (`data[ATTRIBUTES_OFFSET] = 0`)
even though it only needs to clear the append-only flag. Masking off bit 0
better reflects the intent and won’t accidentally wipe other attribute bits if
they’re added later.
##########
crates/fluss/src/record/arrow.rs:
##########
@@ -2349,4 +2436,164 @@ mod tests {
Ok(())
}
+
+ /// Builds an append-only `(id INT, name STRING)` Arrow log batch from
`rows`.
+ /// The writer always emits append-only batches, so changelog tests derive
+ /// their bytes from this with [`splice_change_type_vector`].
+ fn build_append_only_batch(rows: &[(i32, &str)]) -> (RowType, Vec<u8>) {
+ let row_type = RowType::new(vec![
+ DataField::new("id".to_string(), DataTypes::int(), None),
+ DataField::new("name".to_string(), DataTypes::string(), None),
+ ]);
+ let table_path = TablePath::new("db".to_string(), "tbl".to_string());
+ let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1));
+ let physical_table_path =
Arc::new(PhysicalTablePath::of(Arc::new(table_path)));
+
+ let mut builder = MemoryLogRecordsArrowBuilder::new(
+ 1,
+ &row_type,
+ false,
+ ArrowCompressionInfo {
+ compression_type: ArrowCompressionType::None,
+ compression_level: DEFAULT_NON_ZSTD_COMPRESSION_LEVEL,
+ },
+ usize::MAX,
+ Arc::new(ArrowCompressionRatioEstimator::default()),
+ )
+ .unwrap();
+
+ for (id, name) in rows {
+ let mut row = GenericRow::new(2);
+ row.set_field(0, *id);
+ row.set_field(1, *name);
+ let record = WriteRecord::for_append(
+ Arc::clone(&table_info),
+ physical_table_path.clone(),
+ 1,
+ &row,
+ );
+ builder.append(&record).unwrap();
+ }
+
+ (row_type, builder.build().unwrap())
+ }
+
+ /// Turns an append-only batch into a wire-valid changelog batch: clears
the
+ /// append-only flag, splices one change-type byte per record between the
+ /// header and the Arrow payload, then fixes up the length field and CRC.
+ fn splice_change_type_vector(append_only: &[u8], change_types:
&[ChangeType]) -> Vec<u8> {
+ let mut data = append_only.to_vec();
+ data[ATTRIBUTES_OFFSET] = 0;
+ let change_bytes = change_types.iter().map(|ct| ct.to_byte_value());
+ data.splice(RECORDS_OFFSET..RECORDS_OFFSET, change_bytes);
+
+ let new_length = (data.len() - LOG_OVERHEAD) as i32;
+ data[LENGTH_OFFSET..LENGTH_OFFSET + LENGTH_LENGTH]
+ .copy_from_slice(&new_length.to_le_bytes());
+
+ let crc = crc32c(&data[SCHEMA_ID_OFFSET..]);
+ data[CRC_OFFSET..CRC_OFFSET +
CRC_LENGTH].copy_from_slice(&crc.to_le_bytes());
+ data
+ }
+
+ #[test]
+ fn decode_changelog_batch_applies_per_record_change_types() -> Result<()> {
+ let (row_type, append_only) =
+ build_append_only_batch(&[(1, "alice"), (2, "bob"), (3, "carol")]);
+ let read_context = ReadContext::new(to_arrow_schema(&row_type)?,
Arc::new(row_type), false);
+
+ // Append-only batch: every record decodes as AppendOnly (regression
guard).
+ let batch = LogRecordsBatches::new(append_only.clone())
+ .next()
+ .expect("append-only batch")?;
+ assert!(batch.is_append_only());
+ let records: Vec<_> = batch.records(&read_context)?.collect();
+ assert_eq!(records.len(), 3);
+ assert!(
+ records
+ .iter()
+ .all(|r| *r.change_type() == ChangeType::AppendOnly)
+ );
+
+ // Changelog variant: the spliced change-type vector drives per-record
types.
+ let change_types = [
+ ChangeType::Insert,
+ ChangeType::UpdateAfter,
+ ChangeType::Delete,
+ ];
+ let changelog = splice_change_type_vector(&append_only, &change_types);
+ let batch = LogRecordsBatches::new(changelog)
+ .next()
+ .expect("changelog batch")?;
+ assert!(!batch.is_append_only());
+ assert_eq!(batch.record_count(), 3);
+
+ let records: Vec<_> = batch.records(&read_context)?.collect();
+ let got: Vec<ChangeType> = records.iter().map(|r|
*r.change_type()).collect();
+ assert_eq!(got, change_types.to_vec());
+
+ // The row payload and offsets survive the splice unchanged.
+ let mut ids = Vec::new();
+ for record in &records {
+ ids.push(record.row().get_int(0)?);
+ }
+ assert_eq!(ids, vec![1, 2, 3]);
+ let offsets: Vec<i64> = records.iter().map(|r| r.offset()).collect();
+ assert_eq!(offsets, vec![0, 1, 2]);
+
+ // Batch-level access skips the change-type vector and still decodes
rows.
+ let batch =
LogRecordsBatches::new(splice_change_type_vector(&append_only, &change_types))
+ .next()
+ .expect("changelog batch")?;
+ assert_eq!(batch.record_batch(&read_context)?.num_rows(), 3);
+
+ Ok(())
+ }
+
+ #[test]
+ fn decode_changelog_batch_rejects_invalid_change_type_byte() {
+ let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2,
"b")]);
+ let read_context = ReadContext::new(
+ to_arrow_schema(&row_type).unwrap(),
+ Arc::new(row_type),
+ false,
+ );
+
+ let mut changelog =
+ splice_change_type_vector(&append_only, &[ChangeType::Insert,
ChangeType::Insert]);
+ // Corrupt the second change-type byte to an out-of-range value.
+ changelog[RECORDS_OFFSET + 1] = 99;
+
+ let batch = LogRecordBatch::new(Bytes::from(changelog));
+ let err = batch
+ .records(&read_context)
+ .err()
+ .expect("expected decode to reject an invalid change-type byte");
+ assert!(matches!(err, Error::UnexpectedError { .. }));
+ assert!(err.to_string().contains("change type"));
+ }
+
+ #[test]
+ fn decode_changelog_batch_rejects_truncated_change_type_vector() {
+ let (row_type, append_only) = build_append_only_batch(&[(1, "a"), (2,
"b")]);
+ let read_context = ReadContext::new(
+ to_arrow_schema(&row_type).unwrap(),
+ Arc::new(row_type),
+ false,
+ );
+
+ // Clear the append-only flag, then cut the body shorter than the
+ // record_count change-type bytes the decoder now expects.
+ let mut data = append_only;
+ data[ATTRIBUTES_OFFSET] = 0;
+ data.truncate(RECORDS_OFFSET + 1);
Review Comment:
Here the test clears the whole attributes byte (`data[ATTRIBUTES_OFFSET] =
0`) even though it only needs to clear the append-only flag bit. Masking is
safer and matches the intent if other attribute bits are introduced later.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
