You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ho...@apache.org on 2026/10/12 09:07:10 UTC
(fluss) branch main updated: [rust] Expose Arrow batch commit timestamps in client bindings (#4575)
This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 51303ad67 [rust] Expose Arrow batch commit timestamps in client bindings (#4575)
51303ad67 is described below
commit 51303ad67cdfa9b641f8b6ca501489d11ed1420c
Author: lz <97...@qq.com>
AuthorDate: Mon Oct 12 17:06:58 2026 +0800
[rust] Expose Arrow batch commit timestamps in client bindings (#4575)
---
fluss-rust/bindings/cpp/include/fluss.hpp | 12 +-
fluss-rust/bindings/cpp/src/lib.rs | 1 +
fluss-rust/bindings/cpp/src/table.cpp | 12 +-
fluss-rust/bindings/cpp/src/types.rs | 1 +
fluss-rust/bindings/cpp/test/test_log_table.cpp | 54 +++++++
fluss-rust/bindings/python/fluss/__init__.pyi | 12 ++
fluss-rust/bindings/python/src/lib.rs | 1 +
fluss-rust/bindings/python/src/table.rs | 8 +-
.../bindings/python/test/test_batch_scanner.py | 3 +
fluss-rust/bindings/python/test/test_log_table.py | 22 +++
.../fluss/src/client/table/log_fetch_buffer.rs | 146 +++++++++++++++---
fluss-rust/crates/fluss/src/client/table/reader.rs | 19 +++
.../crates/fluss/src/client/table/scanner.rs | 15 +-
fluss-rust/crates/fluss/src/record/mod.rs | 28 ++++
.../fluss/tests/integration/batch_scanner.rs | 4 +
.../crates/fluss/tests/integration/log_table.rs | 167 ++++++++++++++++++++-
.../website/docs/user-guide/cpp/api-reference.md | 5 +
.../docs/user-guide/python/api-reference.md | 17 ++-
.../website/docs/user-guide/rust/api-reference.md | 19 ++-
website/docs/apis/cpp/api-reference.md | 5 +
website/docs/apis/python/api-reference.md | 17 ++-
website/docs/apis/rust/api-reference.md | 19 ++-
22 files changed, 528 insertions(+), 59 deletions(-)
diff --git a/fluss-rust/bindings/cpp/include/fluss.hpp b/fluss-rust/bindings/cpp/include/fluss.hpp
index bc655375c..93b1fd001 100644
--- a/fluss-rust/bindings/cpp/include/fluss.hpp
+++ b/fluss-rust/bindings/cpp/include/fluss.hpp
@@ -1392,6 +1392,11 @@ struct ArrowBatchImporter;
class ArrowRecordBatch {
public:
+ // Reported by GetCommitTimestamp() when the batch carries no single commit
+ // timestamp, which is the case for a limit scan because it merges several
+ // log batches into one Arrow batch.
+ static constexpr int64_t NO_COMMIT_TIMESTAMP = -1;
+
std::shared_ptr<arrow::RecordBatch> GetArrowRecordBatch() const { return batch_; }
bool Available() const;
@@ -1405,13 +1410,15 @@ class ArrowRecordBatch {
int32_t GetBucketId() const;
int64_t GetBaseOffset() const;
int64_t GetLastOffset() const;
+ // Epoch milliseconds, or NO_COMMIT_TIMESTAMP when unavailable.
+ int64_t GetCommitTimestamp() const;
private:
friend class LogScanner;
friend struct detail::ArrowBatchImporter;
explicit ArrowRecordBatch(std::shared_ptr<arrow::RecordBatch> batch, int64_t table_id,
int64_t partition_id, int32_t bucket_id,
- int64_t base_offset) noexcept;
+ int64_t base_offset, int64_t commit_timestamp) noexcept;
std::shared_ptr<arrow::RecordBatch> batch_{nullptr};
@@ -1419,6 +1426,9 @@ class ArrowRecordBatch {
int64_t partition_id_;
int32_t bucket_id_;
int64_t base_offset_;
+ // Defaults to the sentinel so any future construction path that forgets to
+ // set it reports "unknown" rather than an arbitrary value.
+ int64_t commit_timestamp_{NO_COMMIT_TIMESTAMP};
};
struct ArrowRecordBatches {
diff --git a/fluss-rust/bindings/cpp/src/lib.rs b/fluss-rust/bindings/cpp/src/lib.rs
index 69b8f15e9..7793bcb08 100644
--- a/fluss-rust/bindings/cpp/src/lib.rs
+++ b/fluss-rust/bindings/cpp/src/lib.rs
@@ -250,6 +250,7 @@ mod ffi {
partition_id: i64,
bucket_id: i32,
base_offset: i64,
+ commit_timestamp: i64,
}
struct FfiArrowRecordBatches {
diff --git a/fluss-rust/bindings/cpp/src/table.cpp b/fluss-rust/bindings/cpp/src/table.cpp
index 36b228273..d620c86b3 100644
--- a/fluss-rust/bindings/cpp/src/table.cpp
+++ b/fluss-rust/bindings/cpp/src/table.cpp
@@ -2189,12 +2189,13 @@ Result LogScanner::Poll(int64_t timeout_ms, ScanRecords& out) {
ArrowRecordBatch::ArrowRecordBatch(std::shared_ptr<arrow::RecordBatch> batch, int64_t table_id,
int64_t partition_id, int32_t bucket_id,
- int64_t base_offset) noexcept
+ int64_t base_offset, int64_t commit_timestamp) noexcept
: batch_(std::move(batch)),
table_id_(table_id),
partition_id_(partition_id),
bucket_id_(bucket_id),
- base_offset_(base_offset) {}
+ base_offset_(base_offset),
+ commit_timestamp_(commit_timestamp) {}
bool ArrowRecordBatch::Available() const { return batch_ != nullptr; }
@@ -2228,6 +2229,11 @@ int64_t ArrowRecordBatch::GetLastOffset() const {
return this->base_offset_ + this->NumRows() - 1;
}
+int64_t ArrowRecordBatch::GetCommitTimestamp() const {
+ if (!Available()) return NO_COMMIT_TIMESTAMP;
+ return this->commit_timestamp_;
+}
+
namespace detail {
// Imports FFI Arrow batches (C Data Interface) into C++ ArrowRecordBatch
// wrappers. A friend of ArrowRecordBatch so it can build the wrapper's private
@@ -2253,7 +2259,7 @@ struct ArrowBatchImporter {
}
out.batches.push_back(std::unique_ptr<ArrowRecordBatch>(new ArrowRecordBatch(
import_result.ValueOrDie(), ffi_batch.table_id, ffi_batch.partition_id,
- ffi_batch.bucket_id, ffi_batch.base_offset)));
+ ffi_batch.bucket_id, ffi_batch.base_offset, ffi_batch.commit_timestamp)));
}
return utils::make_ok();
}
diff --git a/fluss-rust/bindings/cpp/src/types.rs b/fluss-rust/bindings/cpp/src/types.rs
index c0a4f11b4..697e4810d 100644
--- a/fluss-rust/bindings/cpp/src/types.rs
+++ b/fluss-rust/bindings/cpp/src/types.rs
@@ -750,6 +750,7 @@ pub fn core_scan_batches_to_ffi(
partition_id: batch.bucket().partition_id().unwrap_or(-1),
bucket_id: batch.bucket().bucket_id(),
base_offset: batch.base_offset(),
+ commit_timestamp: batch.commit_timestamp(),
});
}
diff --git a/fluss-rust/bindings/cpp/test/test_log_table.cpp b/fluss-rust/bindings/cpp/test/test_log_table.cpp
index c88e34227..2c6c47b83 100644
--- a/fluss-rust/bindings/cpp/test/test_log_table.cpp
+++ b/fluss-rust/bindings/cpp/test/test_log_table.cpp
@@ -26,6 +26,7 @@
#include <limits>
#include <thread>
#include <tuple>
+#include <unordered_map>
#include "test_utils.h"
@@ -555,6 +556,9 @@ TEST_F(LogTableTest, LimitScan) {
for (const auto& b : first) {
EXPECT_EQ(b->GetTableId(), table_id);
EXPECT_EQ(b->GetBucketId(), 0);
+ // A limit scan merges every log batch it read into one Arrow batch, so
+ // there is no single commit timestamp to report.
+ EXPECT_EQ(b->GetCommitTimestamp(), fluss::ArrowRecordBatch::NO_COMMIT_TIMESTAMP);
rows += b->NumRows();
}
// The server may return fewer rows than the limit, but never more.
@@ -1039,6 +1043,56 @@ TEST_F(LogTableTest, TestPollBatches) {
<< "Projected batch should have 1 column (id), not 2";
}
+ // Test 6: the record and batch paths read the same commit timestamp off the
+ // same log batch header, so they must agree for every offset a batch
+ // covers. This is exact and clock-independent, unlike a "looks like a
+ // recent timestamp" bound, and it also catches a batch that was handed some
+ // other log batch's timestamp.
+ {
+ fluss::Table record_table;
+ ASSERT_OK(conn.GetTable(table_path, record_table));
+ auto record_scan = record_table.NewScan();
+ fluss::LogScanner record_scanner;
+ ASSERT_OK(record_scan.CreateLogScanner(record_scanner));
+ ASSERT_OK(record_scanner.Subscribe(0, 0));
+
+ std::vector<std::pair<int64_t, int64_t>> record_stamps;
+ fluss_test::PollRecords(
+ record_scanner, 8,
+ [](const fluss::ScanRecord& rec) { return std::make_pair(rec.offset, rec.timestamp); },
+ record_stamps);
+ ASSERT_FALSE(record_stamps.empty());
+ const std::unordered_map<int64_t, int64_t> timestamp_by_offset(record_stamps.begin(),
+ record_stamps.end());
+
+ fluss::Table batch_table;
+ ASSERT_OK(conn.GetTable(table_path, batch_table));
+ auto batch_scan = batch_table.NewScan();
+ fluss::RecordBatchLogScanner batch_scanner;
+ ASSERT_OK(batch_scan.CreateRecordBatchLogScanner(batch_scanner));
+ ASSERT_OK(batch_scanner.Subscribe(0, 0));
+
+ fluss::ArrowRecordBatches parity_batches;
+ ASSERT_OK(batch_scanner.Poll(10000, parity_batches));
+ ASSERT_FALSE(parity_batches.Empty());
+
+ int checked = 0;
+ for (const auto& batch : parity_batches) {
+ for (int64_t offset = batch->GetBaseOffset(); offset <= batch->GetLastOffset();
+ ++offset) {
+ auto found = timestamp_by_offset.find(offset);
+ ASSERT_TRUE(found != timestamp_by_offset.end())
+ << "the record path never reported offset " << offset;
+ EXPECT_EQ(batch->GetCommitTimestamp(), found->second)
+ << "batch at offset " << batch->GetBaseOffset() << " reports "
+ << batch->GetCommitTimestamp() << ", record at offset " << offset
+ << " reports " << found->second;
+ ++checked;
+ }
+ }
+ EXPECT_GT(checked, 0) << "no batch offsets were checked";
+ }
+
ASSERT_OK(adm.DropTable(table_path, false));
}
diff --git a/fluss-rust/bindings/python/fluss/__init__.pyi b/fluss-rust/bindings/python/fluss/__init__.pyi
index 8a6784669..cc5cacbda 100644
--- a/fluss-rust/bindings/python/fluss/__init__.pyi
+++ b/fluss-rust/bindings/python/fluss/__init__.pyi
@@ -102,6 +102,14 @@ class RecordBatch:
def last_offset(self) -> int:
"""The offset of the last record in this batch."""
...
+ @property
+ def commit_timestamp(self) -> int:
+ """Server-side commit timestamp in epoch milliseconds.
+
+ `NO_COMMIT_TIMESTAMP` when the batch carries no single timestamp, which
+ is the case for a limit scan that merges several log batches.
+ """
+ ...
def __str__(self) -> str: ...
def __repr__(self) -> str: ...
@@ -1495,4 +1503,8 @@ class OffsetSpec:
# Constant for earliest offset (-2)
EARLIEST_OFFSET: int
+# Reported by `RecordBatch.commit_timestamp` when the batch carries no single
+# commit timestamp, for example a limit scan that merges several log batches (-1)
+NO_COMMIT_TIMESTAMP: int
+
__version__: str
diff --git a/fluss-rust/bindings/python/src/lib.rs b/fluss-rust/bindings/python/src/lib.rs
index f884f127a..a22325364 100644
--- a/fluss-rust/bindings/python/src/lib.rs
+++ b/fluss-rust/bindings/python/src/lib.rs
@@ -141,6 +141,7 @@ fn _fluss(m: &Bound<'_, PyModule>) -> PyResult<()> {
// Register constants
m.add("EARLIEST_OFFSET", fcore::client::EARLIEST_OFFSET)?;
+ m.add("NO_COMMIT_TIMESTAMP", fcore::record::NO_COMMIT_TIMESTAMP)?;
// Register exception types and error codes
m.add_class::<FlussError>()?;
diff --git a/fluss-rust/bindings/python/src/table.rs b/fluss-rust/bindings/python/src/table.rs
index 9563ab365..3fe00301e 100644
--- a/fluss-rust/bindings/python/src/table.rs
+++ b/fluss-rust/bindings/python/src/table.rs
@@ -126,6 +126,9 @@ pub struct RecordBatch {
base_offset: i64,
#[pyo3(get)]
last_offset: i64,
+ /// Server-side commit timestamp in epoch milliseconds, or -1 if unavailable.
+ #[pyo3(get)]
+ commit_timestamp: i64,
}
#[pymethods]
@@ -143,10 +146,11 @@ impl RecordBatch {
fn __str__(&self) -> String {
format!(
- "RecordBatch(bucket={}, base_offset={}, last_offset={}, rows={})",
+ "RecordBatch(bucket={}, base_offset={}, last_offset={}, commit_timestamp={}, rows={})",
self.bucket.__str__(),
self.base_offset,
self.last_offset,
+ self.commit_timestamp,
self.batch.num_rows()
)
}
@@ -159,10 +163,12 @@ impl RecordBatch {
impl RecordBatch {
/// Create a RecordBatch from core ScanBatch
pub fn from_scan_batch(scan_batch: fcore::record::ScanBatch) -> Self {
+ let commit_timestamp = scan_batch.commit_timestamp();
RecordBatch {
bucket: TableBucket::from_core(scan_batch.bucket().clone()),
base_offset: scan_batch.base_offset(),
last_offset: scan_batch.last_offset(),
+ commit_timestamp,
batch: Arc::new(scan_batch.into_batch()),
}
}
diff --git a/fluss-rust/bindings/python/test/test_batch_scanner.py b/fluss-rust/bindings/python/test/test_batch_scanner.py
index 6e30fff6b..f7ba30bfb 100644
--- a/fluss-rust/bindings/python/test/test_batch_scanner.py
+++ b/fluss-rust/bindings/python/test/test_batch_scanner.py
@@ -55,6 +55,9 @@ async def test_returns_appended_rows_then_none(connection, admin):
first = await scanner.next_batch()
assert first is not None
assert first.bucket == bucket
+ # A limit scan merges every log batch it read into one Arrow batch, so there
+ # is no single commit timestamp to report.
+ assert first.commit_timestamp == fluss.NO_COMMIT_TIMESTAMP
# The server may return fewer rows than the limit, but never more.
assert 0 < first.batch.num_rows <= 3
assert await scanner.next_batch() is None
diff --git a/fluss-rust/bindings/python/test/test_log_table.py b/fluss-rust/bindings/python/test/test_log_table.py
index cf923ae0e..7c94b30d4 100644
--- a/fluss-rust/bindings/python/test/test_log_table.py
+++ b/fluss-rust/bindings/python/test/test_log_table.py
@@ -402,6 +402,28 @@ async def test_poll_batches(connection, admin, wait_for_table_ready):
assert len(batches) > 0
assert batches[0].batch.num_columns == 1
+ # The record and batch paths read the same commit timestamp off the same log
+ # batch header, so they must agree for every offset a batch covers. This is
+ # exact and clock-independent, unlike a "looks like a recent timestamp" bound.
+ record_scanner = await table.new_scan().create_log_scanner()
+ record_scanner.subscribe(bucket_id=0, start_offset=0)
+ records = await _poll_records(record_scanner, expected_count=8)
+ timestamp_by_offset = {record.offset: record.timestamp for record in records}
+
+ checked = 0
+ for batch in batches:
+ for offset in range(batch.base_offset, batch.last_offset + 1):
+ assert offset in timestamp_by_offset, (
+ f"the record path never reported offset {offset}"
+ )
+ assert batch.commit_timestamp == timestamp_by_offset[offset], (
+ f"batch at offset {batch.base_offset} reports "
+ f"{batch.commit_timestamp}, record at offset {offset} reports "
+ f"{timestamp_by_offset[offset]}"
+ )
+ checked += 1
+ assert checked > 0, "no batch offsets were checked"
+
await admin.drop_table(table_path, ignore_if_not_exists=False)
diff --git a/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs b/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs
index 1e90c8a64..99ebbda81 100644
--- a/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs
+++ b/fluss-rust/crates/fluss/src/client/table/log_fetch_buffer.rs
@@ -59,15 +59,29 @@ pub(crate) struct FetchErrorContext {
pub(crate) log_message: String,
}
+/// One Arrow batch decoded from a single log batch, together with the log
+/// metadata the scanner needs to turn it into a [`ScanBatch`].
+///
+/// [`ScanBatch`]: crate::record::ScanBatch
+#[derive(Debug)]
+pub(crate) struct FetchedBatch {
+ /// The decoded Arrow data, already sliced to the rows the caller asked for.
+ pub(crate) batch: RecordBatch,
+ /// Offset of the first record in [`Self::batch`], which is the log batch's
+ /// base offset unless leading rows were sliced off.
+ pub(crate) base_offset: i64,
+ /// Commit timestamp of the log batch this Arrow batch was decoded from.
+ pub(crate) commit_timestamp: i64,
+}
+
/// Represents a completed fetch that can be consumed
-pub trait CompletedFetch: Send + Sync {
+pub(crate) trait CompletedFetch: Send + Sync {
fn table_bucket(&self) -> &TableBucket;
fn api_error(&self) -> Option<&ApiError>;
fn fetch_error_context(&self) -> Option<&FetchErrorContext>;
fn take_error(&mut self) -> Option<Error>;
fn fetch_records(&mut self, max_records: usize) -> Result<FetchResult<Vec<ScanRecord>>>;
- fn fetch_batches(&mut self, max_batches: usize)
- -> Result<FetchResult<Vec<(RecordBatch, i64)>>>;
+ fn fetch_batches(&mut self, max_batches: usize) -> Result<FetchResult<Vec<FetchedBatch>>>;
fn is_consumed(&self) -> bool;
fn records_read(&self) -> usize;
fn drain(&mut self);
@@ -609,9 +623,8 @@ impl DefaultCompletedFetch {
source: None,
}
}
- /// Get the next batch with its base offset.
- /// Returns (RecordBatch, base_offset) where base_offset is the offset of the first record.
- fn next_fetched_batch(&mut self) -> Result<FetchStep<(RecordBatch, i64)>> {
+ /// Get the next batch with its base offset and commit timestamp.
+ fn next_fetched_batch(&mut self) -> Result<FetchStep<FetchedBatch>> {
loop {
if self.pending_record_batch.is_none() {
let Some(log_batch_result) = self.log_record_batch.next() else {
@@ -643,6 +656,7 @@ impl DefaultCompletedFetch {
// Calculate the effective base offset for this batch
let log_base_offset = log_batch.base_log_offset();
+ let commit_timestamp = log_batch.commit_timestamp();
let effective_base_offset = if self.next_fetch_offset > log_base_offset {
let skip_count = (self.next_fetch_offset - log_base_offset) as usize;
if skip_count >= record_batch.num_rows() {
@@ -657,7 +671,11 @@ impl DefaultCompletedFetch {
self.next_fetch_offset = log_batch.next_log_offset();
self.records_read += record_batch.num_rows();
- return Ok(FetchStep::InProgress((record_batch, effective_base_offset)));
+ return Ok(FetchStep::InProgress(FetchedBatch {
+ batch: record_batch,
+ base_offset: effective_base_offset,
+ commit_timestamp,
+ }));
}
}
@@ -759,10 +777,7 @@ impl CompletedFetch for DefaultCompletedFetch {
Ok(self.finish_records(scan_records))
}
- fn fetch_batches(
- &mut self,
- max_batches: usize,
- ) -> Result<FetchResult<Vec<(RecordBatch, i64)>>> {
+ fn fetch_batches(&mut self, max_batches: usize) -> Result<FetchResult<Vec<FetchedBatch>>> {
if let Some(error) = self.error.take() {
return Err(error);
}
@@ -784,7 +799,7 @@ impl CompletedFetch for DefaultCompletedFetch {
for _ in 0..max_batches {
match self.next_fetched_batch()? {
- FetchStep::InProgress(batch_with_offset) => batches.push(batch_with_offset),
+ FetchStep::InProgress(fetched_batch) => batches.push(fetched_batch),
FetchStep::End => break,
FetchStep::SchemaRequired(schema_id) => {
if batches.is_empty() {
@@ -877,10 +892,7 @@ impl CompletedFetch for RemoteCompletedFetch {
self.inner.fetch_records(max_records)
}
- fn fetch_batches(
- &mut self,
- max_batches: usize,
- ) -> Result<FetchResult<Vec<(RecordBatch, i64)>>> {
+ fn fetch_batches(&mut self, max_batches: usize) -> Result<FetchResult<Vec<FetchedBatch>>> {
self.inner.fetch_batches(max_batches)
}
@@ -1021,14 +1033,15 @@ mod tests {
TableInfo, TablePath,
};
use crate::record::{
- APPEND_ONLY_FLAG_MASK, ATTRIBUTES_OFFSET, LENGTH_LENGTH, LENGTH_OFFSET, LOG_OVERHEAD,
- MemoryLogRecordsArrowBuilder, RECORDS_OFFSET, ReadContext, to_arrow_schema,
+ APPEND_ONLY_FLAG_MASK, ATTRIBUTES_OFFSET, BASE_OFFSET_LENGTH, BASE_OFFSET_OFFSET,
+ COMMIT_TIMESTAMP_LENGTH, COMMIT_TIMESTAMP_OFFSET, LENGTH_LENGTH, LENGTH_OFFSET,
+ LOG_OVERHEAD, MemoryLogRecordsArrowBuilder, RECORDS_OFFSET, ReadContext, to_arrow_schema,
};
use crate::row::GenericRow;
use crate::test_utils::{
build_table_info, build_table_info_with_columns, uncompressed_arrow_batch_config,
};
- use arrow::array::{Array, StringArray};
+ use arrow::array::{Array, Int32Array, StringArray};
use std::sync::Arc;
fn expect_data<T>(result: FetchResult<T>) -> T {
@@ -1137,7 +1150,7 @@ mod tests {
.into_iter()
.next()
.expect("one fixed-schema batch")
- .0)
+ .batch)
}
struct ErrorPendingFetch {
@@ -1261,6 +1274,97 @@ mod tests {
Ok(())
}
+ #[test]
+ fn fetch_batches_preserves_each_log_batch_timestamp_after_slicing() -> Result<()> {
+ let row_type = RowType::new(vec![DataField::new("id", DataTypes::int(), None)]);
+ let table_path = TablePath::new("db", "tbl");
+ let table_info = Arc::new(build_table_info(table_path.clone(), 1, 1));
+ let physical_path = Arc::new(PhysicalTablePath::of(Arc::new(table_path)));
+ let mut data = Vec::new();
+
+ for (offset, values, timestamp) in
+ [(0_i64, &[0_i32, 1][..], 1234_i64), (2, &[2_i32][..], 5678)]
+ {
+ let mut builder = MemoryLogRecordsArrowBuilder::new(
+ uncompressed_arrow_batch_config(1, &row_type, usize::MAX),
+ false,
+ )?;
+ for &value in values {
+ let mut row = GenericRow::new(1);
+ row.set_field(0, value);
+ builder.append(&WriteRecord::for_append(
+ Arc::clone(&table_info),
+ Arc::clone(&physical_path),
+ 1,
+ &row,
+ ))?;
+ }
+ let mut bytes = builder.build()?;
+ // Stamp the header fields a leader assigns, which the client-side
+ // builder leaves zeroed. Both of them sit before `CRC_OFFSET` and
+ // the checksum only covers `SCHEMA_ID_OFFSET` onwards, so the batch
+ // stays checksum-valid; the assertion below pins that.
+ bytes[BASE_OFFSET_OFFSET..BASE_OFFSET_OFFSET + BASE_OFFSET_LENGTH]
+ .copy_from_slice(&offset.to_le_bytes());
+ bytes[COMMIT_TIMESTAMP_OFFSET..COMMIT_TIMESTAMP_OFFSET + COMMIT_TIMESTAMP_LENGTH]
+ .copy_from_slice(×tamp.to_le_bytes());
+ data.extend(bytes);
+ }
+
+ // Guard the assumption above so this test cannot silently decode
+ // corrupt input if the header layout or checksum coverage changes.
+ for log_batch in LogRecordsBatches::new(data.clone()) {
+ assert!(
+ log_batch?.is_valid(),
+ "the patched log batch should still pass its checksum"
+ );
+ }
+
+ let schema = to_arrow_schema(&row_type)?;
+ let row_type = Arc::new(row_type);
+ let resolver = Arc::new(ReadContextResolver::new(
+ 1,
+ Arc::new(ReadContext::new(
+ schema.clone(),
+ Arc::clone(&row_type),
+ false,
+ )),
+ Arc::new(ReadContext::new(schema, row_type, true)),
+ None,
+ ));
+ let mut fetch = DefaultCompletedFetch::new(
+ TableBucket::new(1, 0),
+ LogRecordsBatches::new(data.clone()),
+ data.len(),
+ resolver,
+ false,
+ 1,
+ 3,
+ );
+
+ let batches = expect_data(fetch.fetch_batches(10)?);
+ assert_eq!(batches.len(), 2);
+ assert_eq!(
+ (batches[0].base_offset, batches[0].commit_timestamp),
+ (1, 1234)
+ );
+ assert_eq!(
+ (batches[1].base_offset, batches[1].commit_timestamp),
+ (2, 5678)
+ );
+ for (fetched, expected) in batches.iter().zip([1, 2]) {
+ assert_eq!(fetched.batch.num_rows(), 1);
+ let ids = fetched
+ .batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(ids.value(0), expected);
+ }
+ Ok(())
+ }
+
/// An empty fetch carrying only a filtered range, as the server sends when
/// it prunes every batch it scanned.
fn filtered_empty_fetch(
@@ -1462,7 +1566,7 @@ mod tests {
let batches = expect_data(fetch.fetch_batches(10)?);
assert_eq!(batches.len(), 1);
- let batch = &batches[0].0;
+ let batch = &batches[0].batch;
assert_eq!(batch.schema(), new_arrow_schema);
assert_eq!(batch.num_columns(), 2);
assert_eq!(batch.column(1).null_count(), 1);
diff --git a/fluss-rust/crates/fluss/src/client/table/reader.rs b/fluss-rust/crates/fluss/src/client/table/reader.rs
index 4a435e6c4..c94a57af4 100644
--- a/fluss-rust/crates/fluss/src/client/table/reader.rs
+++ b/fluss-rust/crates/fluss/src/client/table/reader.rs
@@ -980,9 +980,11 @@ fn filter_batches(
let kept_batch = if last_offset >= stop_at {
let num_to_keep = (stop_at - base_offset) as usize;
+ let commit_timestamp = scan_batch.commit_timestamp();
let b = scan_batch.into_batch();
let limit = num_to_keep.min(b.num_rows());
ScanBatch::new(bucket.clone(), b.slice(0, limit), base_offset)
+ .with_commit_timestamp(commit_timestamp)
} else {
scan_batch
};
@@ -1358,6 +1360,23 @@ mod tests {
assert_eq!(*sb.bucket(), bucket(0));
}
+ #[test]
+ fn filter_preserves_commit_timestamps_across_buckets_and_slicing() {
+ let mut offsets = HashMap::from([(bucket(0), 12), (bucket(1), 10)]);
+ let mut buffer = VecDeque::new();
+ let batches = vec![
+ ScanBatch::new(bucket(0), make_batch(&[1, 2, 3]), 10).with_commit_timestamp(1234),
+ ScanBatch::new(bucket(1), make_batch(&[4, 5]), 0).with_commit_timestamp(5678),
+ ];
+
+ filter_batches(batches, &mut offsets, &mut buffer);
+
+ assert_eq!(buffer[0].batch().num_rows(), 2);
+ assert_eq!(buffer[0].commit_timestamp(), 1234);
+ assert_eq!(buffer[1].batch().num_rows(), 2);
+ assert_eq!(buffer[1].commit_timestamp(), 5678);
+ }
+
#[test]
fn position_at_or_past_stop_completes_bucket() {
let mut offsets = HashMap::from([(bucket(0), 10), (bucket(1), 10)]);
diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs
index 7365090e4..a56a536b3 100644
--- a/fluss-rust/crates/fluss/src/client/table/scanner.rs
+++ b/fluss-rust/crates/fluss/src/client/table/scanner.rs
@@ -2332,7 +2332,7 @@ impl LogFetcher {
if fetch_offset == current_offset {
match next_in_line_fetch.fetch_batches(max_batches)? {
- FetchResult::Data(batches_with_offsets) => {
+ FetchResult::Data(fetched_batches) => {
let next_fetch_offset = next_in_line_fetch.next_fetch_offset();
if next_fetch_offset > current_offset {
@@ -2340,12 +2340,17 @@ impl LogFetcher {
.update_offset(&table_bucket, next_fetch_offset);
}
- // Convert to ScanBatch with bucket info
+ // Convert to ScanBatch with bucket and log-batch metadata.
Ok(FetchResult::Data(
- batches_with_offsets
+ fetched_batches
.into_iter()
- .map(|(batch, base_offset)| {
- ScanBatch::new(table_bucket.clone(), batch, base_offset)
+ .map(|fetched| {
+ ScanBatch::new(
+ table_bucket.clone(),
+ fetched.batch,
+ fetched.base_offset,
+ )
+ .with_commit_timestamp(fetched.commit_timestamp)
})
.collect(),
))
diff --git a/fluss-rust/crates/fluss/src/record/mod.rs b/fluss-rust/crates/fluss/src/record/mod.rs
index fa50664bb..93475a97d 100644
--- a/fluss-rust/crates/fluss/src/record/mod.rs
+++ b/fluss-rust/crates/fluss/src/record/mod.rs
@@ -176,6 +176,10 @@ impl ScanRecords {
}
}
+/// Reported by [`ScanBatch::commit_timestamp`] when the batch carries no single
+/// commit timestamp, mirroring the `-1` sentinel of [`ScanRecord::timestamp`].
+pub const NO_COMMIT_TIMESTAMP: i64 = -1;
+
/// A batch of records with metadata about bucket and offsets.
///
/// This is the batch-level equivalent of [`ScanRecord`], providing efficient
@@ -189,6 +193,9 @@ pub struct ScanBatch {
batch: RecordBatch,
/// Offset of the first record in this batch
base_offset: i64,
+ /// Server-side commit timestamp in milliseconds since epoch, or
+ /// [`NO_COMMIT_TIMESTAMP`] when unavailable.
+ commit_timestamp: i64,
}
impl ScanBatch {
@@ -197,9 +204,17 @@ impl ScanBatch {
bucket,
batch,
base_offset,
+ commit_timestamp: NO_COMMIT_TIMESTAMP,
}
}
+ /// Attaches the commit timestamp of the log batch this Arrow batch was
+ /// decoded from. Without it a batch reports [`NO_COMMIT_TIMESTAMP`].
+ pub fn with_commit_timestamp(mut self, commit_timestamp: i64) -> Self {
+ self.commit_timestamp = commit_timestamp;
+ self
+ }
+
pub fn bucket(&self) -> &TableBucket {
&self.bucket
}
@@ -216,6 +231,14 @@ impl ScanBatch {
self.base_offset
}
+ /// Returns the server-side commit timestamp in milliseconds since epoch.
+ /// Returns [`NO_COMMIT_TIMESTAMP`] when the batch carries no single
+ /// timestamp, which is the case for a bounded limit scan because it may
+ /// combine several log batches into one Arrow batch.
+ pub fn commit_timestamp(&self) -> i64 {
+ self.commit_timestamp
+ }
+
pub fn num_records(&self) -> usize {
self.batch.num_rows()
}
@@ -322,6 +345,11 @@ mod tests {
let scan_batch = ScanBatch::new(bucket.clone(), batch, 100);
assert_eq!(scan_batch.num_records(), 3);
assert_eq!(scan_batch.last_offset(), 102);
+ assert_eq!(scan_batch.commit_timestamp(), NO_COMMIT_TIMESTAMP);
+
+ let timestamped = ScanBatch::new(bucket.clone(), scan_batch.batch().clone(), 100)
+ .with_commit_timestamp(1234);
+ assert_eq!(timestamped.commit_timestamp(), 1234);
// Empty batch -> last_offset = base_offset - 1
let empty_batch = RecordBatch::new_empty(schema);
diff --git a/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs b/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs
index 57002cde8..80e88bc0a 100644
--- a/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs
+++ b/fluss-rust/crates/fluss/tests/integration/batch_scanner.rs
@@ -24,6 +24,7 @@ mod batch_scanner_test {
AddColumn, AlterTableChanges, ColumnPositionType, DataTypes, JsonSerde, LogFormat, Schema,
TableBucket, TableDescriptor, TablePath,
};
+ use fluss::record::NO_COMMIT_TIMESTAMP;
use fluss::row::GenericRow;
use futures::TryStreamExt;
use std::collections::HashMap;
@@ -83,6 +84,9 @@ mod batch_scanner_test {
.expect("first batch should be Some");
assert_eq!(first.bucket(), &bucket);
+ // A limit scan concatenates every log batch the server returned into one
+ // Arrow batch, so there is no single commit timestamp to report.
+ assert_eq!(first.commit_timestamp(), NO_COMMIT_TIMESTAMP);
// The server may return fewer rows than the limit on the first call,
// but must never exceed it.
assert!(
diff --git a/fluss-rust/crates/fluss/tests/integration/log_table.rs b/fluss-rust/crates/fluss/tests/integration/log_table.rs
index afbacd1c3..070e64fc1 100644
--- a/fluss-rust/crates/fluss/tests/integration/log_table.rs
+++ b/fluss-rust/crates/fluss/tests/integration/log_table.rs
@@ -31,14 +31,16 @@ mod table_test {
};
use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::datatypes::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
- use fluss::client::{EARLIEST_OFFSET, FlussAdmin, FlussTable, TableScan};
+ use fluss::client::{
+ EARLIEST_OFFSET, FlussAdmin, FlussTable, RecordBatchLogScanner, TableScan,
+ };
use fluss::error::FlussError;
use fluss::metadata::{
AddColumn, AlterTableChanges, ColumnPositionType, DataField, DataTypes, JsonSerde,
PartitionSpec, Schema, TableDescriptor, TablePath,
};
use fluss::predicate::{Literal, Predicate, col};
- use fluss::record::ScanRecord;
+ use fluss::record::{ScanBatch, ScanRecord};
use fluss::row::binary_array::FlussArrayWriter;
use fluss::row::binary_map::FlussMapWriter;
use fluss::row::{
@@ -1670,6 +1672,167 @@ mod table_test {
assert_eq!(proj_batches[0].batch().num_columns(), 1);
}
+ /// Asserts each batch reports exactly the commit timestamp the record path
+ /// reported for every offset that batch covers, and that timestamps never
+ /// move backwards within a bucket.
+ fn assert_batch_timestamps_match_records(
+ batches: &[ScanBatch],
+ timestamps_by_offset: &HashMap<i64, i64>,
+ ) {
+ assert!(!batches.is_empty(), "expected at least one batch to check");
+ let mut previous = i64::MIN;
+ for batch in batches {
+ let timestamp = batch.commit_timestamp();
+ for offset in batch.base_offset()..=batch.last_offset() {
+ let expected = *timestamps_by_offset
+ .get(&offset)
+ .unwrap_or_else(|| panic!("the record path never reported offset {offset}"));
+ assert_eq!(
+ timestamp,
+ expected,
+ "batch starting at offset {} reports commit timestamp {timestamp}, but the \
+ record at offset {offset} reports {expected}",
+ batch.base_offset()
+ );
+ }
+ // Log batches are appended to a bucket in order, so the leader
+ // never stamps a later batch with an earlier timestamp.
+ assert!(
+ timestamp >= previous,
+ "commit timestamp {timestamp} moved backwards from {previous}"
+ );
+ previous = timestamp;
+ }
+ }
+
+ /// The record and Arrow-batch scan paths read the same `commit_timestamp`
+ /// header field from the same log batch, so the batch API must report
+ /// exactly what the record API reports for the offsets it covers. Comparing
+ /// the two paths pins that invariant without depending on the wall clock,
+ /// and unlike a "looks like a recent timestamp" bound it also catches a
+ /// batch that was handed some other log batch's timestamp.
+ #[tokio::test]
+ async fn record_batch_scanner_agrees_with_record_scanner_on_commit_timestamp() {
+ let cluster = get_shared_cluster();
+ let connection = cluster.get_fluss_connection().await;
+ let admin = connection.get_admin().expect("Failed to get admin");
+
+ let table_path = TablePath::new("fluss", "test_batch_commit_timestamp_parity");
+ let schema = Schema::builder()
+ .column("id", DataTypes::int())
+ .column("name", DataTypes::string())
+ .build()
+ .unwrap();
+ create_table(
+ &admin,
+ &table_path,
+ &TableDescriptor::builder()
+ .schema(schema)
+ // A single bucket keeps every append in one log, so batch
+ // offsets and record offsets line up without interleaving.
+ .distributed_by(Some(1), vec!["id".to_string()])
+ .build()
+ .unwrap(),
+ )
+ .await;
+ wait_for_table_ready(&admin, &table_path).await;
+
+ let table = connection.get_table(&table_path).await.unwrap();
+ let writer = table.new_append().unwrap().create_writer().unwrap();
+ // Flush between appends so the leader stores three separate log
+ // batches, each stamped with its own commit timestamp.
+ writer
+ .append_arrow_batch(
+ record_batch!(("id", Int32, [1, 2]), ("name", Utf8, ["a", "b"])).unwrap(),
+ )
+ .unwrap();
+ writer.flush().await.unwrap();
+ writer
+ .append_arrow_batch(
+ record_batch!(("id", Int32, [3, 4]), ("name", Utf8, ["c", "d"])).unwrap(),
+ )
+ .unwrap();
+ writer.flush().await.unwrap();
+ writer
+ .append_arrow_batch(
+ record_batch!(("id", Int32, [5, 6]), ("name", Utf8, ["e", "f"])).unwrap(),
+ )
+ .unwrap();
+ writer.flush().await.unwrap();
+
+ // Reference: the timestamp the record path reports for each offset.
+ let record_scanner = table.new_scan().create_log_scanner().unwrap();
+ record_scanner.subscribe(0, EARLIEST_OFFSET).await.unwrap();
+ let timestamps_by_offset: HashMap<i64, i64> =
+ poll_until_count(6, DEFAULT_POLL_TIMEOUT, Duration::from_secs(5), async |d| {
+ record_scanner
+ .poll(d)
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|record| (record.offset(), record.timestamp()))
+ .collect()
+ })
+ .await
+ .into_iter()
+ .collect();
+ assert_eq!(
+ timestamps_by_offset.len(),
+ 6,
+ "record path should cover offsets 0..=5, got {timestamps_by_offset:?}"
+ );
+
+ // Reading whole batches must agree with the record path.
+ let scanner = table.new_scan().create_record_batch_log_scanner().unwrap();
+ scanner.subscribe(0, 0).await.unwrap();
+ let batches = collect_batches(&scanner, 6).await;
+ assert_batch_timestamps_match_records(&batches, ×tamps_by_offset);
+
+ // Subscribing mid-batch slices leading rows off the first batch; the
+ // slice must keep the timestamp of the log batch it came from.
+ let sliced_scanner = table.new_scan().create_record_batch_log_scanner().unwrap();
+ sliced_scanner.subscribe(0, 1).await.unwrap();
+ let sliced = collect_batches(&sliced_scanner, 5).await;
+ assert_eq!(
+ sliced[0].base_offset(),
+ 1,
+ "first batch should start at the subscribed offset"
+ );
+ assert_batch_timestamps_match_records(&sliced, ×tamps_by_offset);
+
+ // Projection prunes columns but must not disturb the metadata.
+ let projected_scanner = table
+ .new_scan()
+ .project_by_name(&["id"])
+ .unwrap()
+ .create_record_batch_log_scanner()
+ .unwrap();
+ projected_scanner.subscribe(0, 0).await.unwrap();
+ let projected = collect_batches(&projected_scanner, 6).await;
+ assert_eq!(projected[0].batch().num_columns(), 1);
+ assert_batch_timestamps_match_records(&projected, ×tamps_by_offset);
+ }
+
+ /// Polls a record-batch scanner until it has yielded `expected_rows` rows.
+ async fn collect_batches(
+ scanner: &RecordBatchLogScanner,
+ expected_rows: usize,
+ ) -> Vec<ScanBatch> {
+ let mut collected: Vec<ScanBatch> = Vec::new();
+ let deadline = tokio::time::Instant::now() + DEFAULT_POLL_TIMEOUT;
+ while collected.iter().map(ScanBatch::num_records).sum::<usize>() < expected_rows
+ && tokio::time::Instant::now() < deadline
+ {
+ collected.extend(scanner.poll(Duration::from_secs(5)).await.unwrap());
+ }
+ assert_eq!(
+ collected.iter().map(ScanBatch::num_records).sum::<usize>(),
+ expected_rows,
+ "scanner did not yield {expected_rows} rows in time"
+ );
+ collected
+ }
+
async fn create_region_partitioned_log_table(admin: &FlussAdmin, table_path: &TablePath) {
let table_descriptor = TableDescriptor::builder()
.schema(
diff --git a/fluss-rust/website/docs/user-guide/cpp/api-reference.md b/fluss-rust/website/docs/user-guide/cpp/api-reference.md
index 9fb6f6fd0..acbee1bdb 100644
--- a/fluss-rust/website/docs/user-guide/cpp/api-reference.md
+++ b/fluss-rust/website/docs/user-guide/cpp/api-reference.md
@@ -888,6 +888,11 @@ Read-only result of a prefix lookup — zero or more matched rows. Each row is a
| `GetBucketId() -> int32_t` | Bucket ID |
| `GetBaseOffset() -> int64_t` | First record offset |
| `GetLastOffset() -> int64_t` | Last record offset |
+| `GetCommitTimestamp() -> int64_t` | Commit time in epoch milliseconds |
+
+`GetCommitTimestamp()` returns `ArrowRecordBatch::NO_COMMIT_TIMESTAMP` (`-1`)
+when the batch carries no single timestamp, which is the case for a limit scan
+because it merges several log batches into one Arrow batch.
## `ArrowRecordBatches`
diff --git a/fluss-rust/website/docs/user-guide/python/api-reference.md b/fluss-rust/website/docs/user-guide/python/api-reference.md
index 6b5c876fb..08bcae352 100644
--- a/fluss-rust/website/docs/user-guide/python/api-reference.md
+++ b/fluss-rust/website/docs/user-guide/python/api-reference.md
@@ -273,12 +273,17 @@ for record in scan_records:
## `RecordBatch`
-| Property | Description |
-|----------------------------|------------------------------|
-| `.batch -> pa.RecordBatch` | Arrow RecordBatch data |
-| `.bucket -> TableBucket` | Bucket this batch belongs to |
-| `.base_offset -> int` | First record offset |
-| `.last_offset -> int` | Last record offset |
+| Property | Description |
+|----------------------------|-----------------------------------|
+| `.batch -> pa.RecordBatch` | Arrow RecordBatch data |
+| `.bucket -> TableBucket` | Bucket this batch belongs to |
+| `.base_offset -> int` | First record offset |
+| `.last_offset -> int` | Last record offset |
+| `.commit_timestamp -> int` | Commit time in epoch milliseconds |
+
+`.commit_timestamp` is `fluss.NO_COMMIT_TIMESTAMP` (`-1`) when the batch carries
+no single timestamp, which is the case for a limit scan because it merges several
+log batches into one Arrow batch.
## `Schema`
diff --git a/fluss-rust/website/docs/user-guide/rust/api-reference.md b/fluss-rust/website/docs/user-guide/rust/api-reference.md
index cee8ed4dc..43d43a737 100644
--- a/fluss-rust/website/docs/user-guide/rust/api-reference.md
+++ b/fluss-rust/website/docs/user-guide/rust/api-reference.md
@@ -299,13 +299,18 @@ for record in records {
## `ScanBatch`
-| Method | Description |
-|------------------------------------|--------------------------------|
-| `fn bucket(&self) -> &TableBucket` | Bucket this batch belongs to |
-| `fn batch(&self) -> &RecordBatch` | Arrow RecordBatch data |
-| `fn base_offset(&self) -> i64` | First record offset |
-| `fn last_offset(&self) -> i64` | Last record offset |
-| `fn num_records(&self) -> usize` | Number of records in the batch |
+| Method | Description |
+|-------------------------------------|-----------------------------------|
+| `fn bucket(&self) -> &TableBucket` | Bucket this batch belongs to |
+| `fn batch(&self) -> &RecordBatch` | Arrow RecordBatch data |
+| `fn base_offset(&self) -> i64` | First record offset |
+| `fn last_offset(&self) -> i64` | Last record offset |
+| `fn commit_timestamp(&self) -> i64` | Commit time in epoch milliseconds |
+| `fn num_records(&self) -> usize` | Number of records in the batch |
+
+`commit_timestamp` is `fluss::record::NO_COMMIT_TIMESTAMP` (`-1`) when the batch
+carries no single timestamp, which is the case for a limit scan because it
+merges several log batches into one Arrow batch.
## `TableUpsert`
diff --git a/website/docs/apis/cpp/api-reference.md b/website/docs/apis/cpp/api-reference.md
index 0a254ae97..c6ce12b35 100644
--- a/website/docs/apis/cpp/api-reference.md
+++ b/website/docs/apis/cpp/api-reference.md
@@ -449,6 +449,11 @@ Read-only result of a prefix lookup — zero or more matched rows. Each row is a
| `GetBucketId() -> int32_t` | Bucket ID |
| `GetBaseOffset() -> int64_t` | First record offset |
| `GetLastOffset() -> int64_t` | Last record offset |
+| `GetCommitTimestamp() -> int64_t` | Commit time in epoch milliseconds |
+
+`GetCommitTimestamp()` returns `ArrowRecordBatch::NO_COMMIT_TIMESTAMP` (`-1`)
+when the batch carries no single timestamp, which is the case for a limit scan
+because it merges several log batches into one Arrow batch.
## `ArrowRecordBatches`
diff --git a/website/docs/apis/python/api-reference.md b/website/docs/apis/python/api-reference.md
index 36d1e570b..c343464be 100644
--- a/website/docs/apis/python/api-reference.md
+++ b/website/docs/apis/python/api-reference.md
@@ -319,12 +319,17 @@ for record in scan_records:
## `RecordBatch`
-| Property | Description |
-|----------------------------|------------------------------|
-| `.batch -> pa.RecordBatch` | Arrow RecordBatch data |
-| `.bucket -> TableBucket` | Bucket this batch belongs to |
-| `.base_offset -> int` | First record offset |
-| `.last_offset -> int` | Last record offset |
+| Property | Description |
+|----------------------------|-----------------------------------|
+| `.batch -> pa.RecordBatch` | Arrow RecordBatch data |
+| `.bucket -> TableBucket` | Bucket this batch belongs to |
+| `.base_offset -> int` | First record offset |
+| `.last_offset -> int` | Last record offset |
+| `.commit_timestamp -> int` | Commit time in epoch milliseconds |
+
+`.commit_timestamp` is `fluss.NO_COMMIT_TIMESTAMP` (`-1`) when the batch carries
+no single timestamp, which is the case for a limit scan because it merges several
+log batches into one Arrow batch.
## `Schema`
diff --git a/website/docs/apis/rust/api-reference.md b/website/docs/apis/rust/api-reference.md
index fc7628873..dd46f3f39 100644
--- a/website/docs/apis/rust/api-reference.md
+++ b/website/docs/apis/rust/api-reference.md
@@ -281,13 +281,18 @@ for record in records {
## `ScanBatch`
-| Method | Description |
-|------------------------------------|--------------------------------|
-| `fn bucket(&self) -> &TableBucket` | Bucket this batch belongs to |
-| `fn batch(&self) -> &RecordBatch` | Arrow RecordBatch data |
-| `fn base_offset(&self) -> i64` | First record offset |
-| `fn last_offset(&self) -> i64` | Last record offset |
-| `fn num_records(&self) -> usize` | Number of records in the batch |
+| Method | Description |
+|-------------------------------------|-----------------------------------|
+| `fn bucket(&self) -> &TableBucket` | Bucket this batch belongs to |
+| `fn batch(&self) -> &RecordBatch` | Arrow RecordBatch data |
+| `fn base_offset(&self) -> i64` | First record offset |
+| `fn last_offset(&self) -> i64` | Last record offset |
+| `fn commit_timestamp(&self) -> i64` | Commit time in epoch milliseconds |
+| `fn num_records(&self) -> usize` | Number of records in the batch |
+
+`commit_timestamp` is `fluss::record::NO_COMMIT_TIMESTAMP` (`-1`) when the batch
+carries no single timestamp, which is the case for a limit scan because it
+merges several log batches into one Arrow batch.
## `TableUpsert`