[VidyavathiGK]

Purpose

Linked issue: close apache/fluss-rust#572

Per maintainer guidance (see apache/fluss-rust#666), moving the bounded reader stream adapter and bare-RecordBatch bridges into fluss-rust in the consolidated apache/fluss repository.

Provides a bare Arrow RecordBatch bridge for DataFusion and PyArrow (RecordBatchStream and to_record_batch_reader), gives LimitBatchScanner parity with RecordBatchLogReader (schema(&self) and SyncRecordBatchLimitScanner), and provides into_record_batch_stream() across both bounded readers for seamless interop with DataFusion (RecordBatchStreamAdapter) and PyArrow.

Brief change log

• New Stream Module (fluss-rust/crates/fluss/src/client/table/stream.rs):
  • RecordBatchStream: Asynchronous Stream<Item = Result<RecordBatch, ArrowError>> yielding bare Arrow batches (stripping Fluss metadata), with .schema() -> SchemaRef and .to_record_batch_reader(handle). Directly plugs into DataFusion's RecordBatchStreamAdapter or PyArrow async workflows.
  • ScanBatchStream: Named asynchronous Stream<Item = Result<ScanBatch>> wrapper with .schema() and .into_record_batch_stream().
  • SyncRecordBatchStreamReader: Synchronous arrow::record_batch::RecordBatchReader adapter driven by a Tokio runtime handle.
• RecordBatchLogReader (fluss-rust/crates/fluss/src/client/table/reader.rs):
  • Added into_record_batch_stream(self) -> RecordBatchStream.
  • Preserved existing into_stream(), next_batch(), collect_all_batches(), schema(), and to_record_batch_reader(handle) -> SyncRecordBatchLogReader.
• LimitBatchScanner (fluss-rust/crates/fluss/src/client/table/batch_scanner.rs):
  • Added schema: SchemaRef field and schema(&self) -> SchemaRef getter (computed with projection pushdown awareness).
  • Added into_record_batch_stream(self) -> RecordBatchStream.
  • Added to_record_batch_reader(self, handle) -> SyncRecordBatchLimitScanner.
  • Added SyncRecordBatchLimitScanner implementing arrow::record_batch::RecordBatchReader (matching SyncRecordBatchLogReader parity).
  • Preserved existing into_stream(), next_batch(), collect_all_batches(), and bucket().
• Module Exports (fluss-rust/crates/fluss/src/client/table/mod.rs):
  • Re-exported ScanBatchStream, RecordBatchStream, SyncRecordBatchStreamReader, and SyncRecordBatchLimitScanner.

Tests

• Unit tests in stream.rs:
  • test_scan_batch_stream_consumption: verifying item polling, schema accessor, and stream termination.
  • test_into_record_batch_stream: verifying conversion to bare RecordBatches and schema preservation.
  • test_record_batch_stream_error_propagation: verifying error mapping to ArrowError::ExternalError.
  • test_sync_record_batch_stream_reader: verifying synchronous RecordBatchReader iteration contract.
• Integration tests in tests/integration/batch_scanner.rs:
  • batch_scanner_bare_record_batch_bridges: verifying into_record_batch_stream and to_record_batch_reader against real table data in a live cluster.
• Integration tests in tests/integration/record_batch_log_reader.rs:
  • record_batch_log_reader_bare_record_batch_bridges: verifying into_record_batch_stream() and to_record_batch_reader() against a live cluster.

API and Format

• Additive API extensions:
  • LimitBatchScanner: schema(), into_record_batch_stream(), to_record_batch_reader().
  • RecordBatchLogReader: into_record_batch_stream().
  • New public types: ScanBatchStream, RecordBatchStream, SyncRecordBatchStreamReader, SyncRecordBatchLimitScanner.
• Zero breaking changes to existing APIs or binary formats.

Documentation

• Does this change introduce a new feature: yes, bare-RecordBatch bridges and stream adapters for bounded readers to enable seamless DataFusion and PyArrow interoperability.