[VidyavathiGK]

Purpose

Linked issue: close #572

Provides into_stream() on both bounded readers (LimitBatchScanner and RecordBatchLogReader) returning an asynchronous Stream<Item = Result<ScanBatch>> (ScanBatchStream), along with a bare Arrow RecordBatch bridge for DataFusion and PyArrow (RecordBatchStream and to_record_batch_reader). This gives the client a unified, consistent streaming and interop surface while keeping next_batch() and collect_all_batches() additive and intact.

Brief change log

• New Stream Module (crates/fluss/src/client/table/stream.rs):
  • ScanBatchStream: Asynchronous Stream<Item = Result<ScanBatch>> wrapping bounded scanner/reader batches, with .schema() -> SchemaRef, .into_record_batch_stream(), and .to_record_batch_reader(handle).
  • 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.
  • SyncRecordBatchStreamReader: Synchronous arrow::record_batch::RecordBatchReader adapter driven by a Tokio runtime handle.
• RecordBatchLogReader (crates/fluss/src/client/table/reader.rs):
  • Added into_stream(self) -> ScanBatchStream (fused unfold over next_batch()).
  • Added into_record_batch_stream(self) -> RecordBatchStream.
  • Preserved existing next_batch(), collect_all_batches(), schema(), and to_record_batch_reader(handle) -> SyncRecordBatchLogReader.
• LimitBatchScanner (crates/fluss/src/client/table/batch_scanner.rs):
  • Added schema: SchemaRef field and schema(&self) -> SchemaRef getter (computed with projection pushdown awareness).
  • Added into_stream(self) -> ScanBatchStream (fused unfold over next_batch()).
  • 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 next_batch(), collect_all_batches(), and bucket().
• Module Exports (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_into_stream: verifying stream consumption from a live cluster.
  • batch_scanner_bare_record_batch_bridges: verifying into_record_batch_stream and to_record_batch_reader against real table data.
• Integration tests in tests/integration/record_batch_log_reader.rs:
  • record_batch_log_reader_stream_and_bridges: verifying into_stream(), into_record_batch_stream(), and to_record_batch_reader() against a live cluster.

API and Format

• Additive API extensions:
  • LimitBatchScanner: schema(), into_stream(), into_record_batch_stream(), to_record_batch_reader().
  • RecordBatchLogReader: into_stream(), 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, stream adapters for bounded readers to enable seamless DataFusion and PyArrow interoperability.

[pbanakar]

Hi @VidyavathiGK This repository is moved to fluss main repo, lives under fluss-rust module, so this repo is Archived and read-only now so Please open a PR to Fluss main repo https://github.com/apache/fluss