[client] Add LookupResult::to_record_batch()

crates/fluss/src/client/table/lookup.rs

@@ -20,13 +20,16 @@ use crate::client::metadata::Metadata;
 use crate::client::table::partition_getter::PartitionGetter;
 use crate::error::{Error, Result};
 use crate::metadata::{PhysicalTablePath, RowType, TableBucket, TableInfo, TablePath};
+use crate::record::ArrowRecordBatchInnerBuilder;
+use crate::record::RowAppendRecordBatchBuilder;
 use crate::record::kv::SCHEMA_ID_LENGTH;
 use crate::row::InternalRow;
 use crate::row::compacted::CompactedRow;
 use crate::row::encode::{KeyEncoder, KeyEncoderFactory};
 use crate::rpc::ApiError;
 use crate::rpc::RpcClient;
 use crate::rpc::message::LookupRequest;
+use arrow::array::RecordBatch;
 use std::sync::Arc;
 
 /// The result of a lookup operation.
@@ -85,6 +88,35 @@ impl LookupResult {
             .map(|bytes| CompactedRow::from_bytes(&self.row_type, &bytes[SCHEMA_ID_LENGTH..]))
             .collect()
     }
+    /// Converts all rows in this result into an Arrow [`RecordBatch`].
+    ///
+    /// This is useful for integration with DataFusion or other Arrow-based tools.
+    ///
+    /// # Returns
+    /// - `Ok(RecordBatch)` - All rows in columnar Arrow format. Returns an empty
+    ///   batch (with the correct schema) if the result set is empty.
+    /// - `Err(Error)` - If the conversion fails.
+    pub fn to_record_batch(&self) -> Result<RecordBatch> {
+        let mut builder = RowAppendRecordBatchBuilder::new(&self.row_type)?;
+
+        for bytes in &self.rows {
+            let payload = bytes.get(SCHEMA_ID_LENGTH..).ok_or_else(|| {
+                Error::RowConvertError {
+                    message: format!(
+                        "LookupResult row payload too short: {} bytes, need at least {} bytes for schema id",
+                        bytes.len(),
+                        SCHEMA_ID_LENGTH
+                    ),
+                }
+            })?;
+
+            let row = CompactedRow::from_bytes(&self.row_type, payload);
+            builder.append(&row)?;
+        }
+
+        let arc_batch = builder.build_arrow_record_batch()?;
+        Ok(Arc::unwrap_or_clone(arc_batch))
+    }
 }
 
 /// Configuration and factory struct for creating lookup operations.