fresh-borzoni commented on code in PR #578:
URL: https://github.com/apache/fluss-rust/pull/578#discussion_r3336328403
##########
crates/fluss/src/row/column.rs:
##########
@@ -15,1197 +15,246 @@
// specific language governing permissions and limitations
// under the License.
+//! Arrow scan-path cursor over a typed column-vector batch. Accessors are a
+//! single match arm with no per-call downcasting. Nested `ARRAY` / `MAP` /
+//! `ROW` reads return columnar slice/cursor views borrowed from the buffers.
+
use crate::error::Error::IllegalArgument;
use crate::error::Result;
-use crate::metadata::{DataType, RowType};
+use crate::metadata::{DataField, RowType};
use crate::record::from_arrow_field;
-use crate::row::binary_array::FlussArrayWriter;
-use crate::row::binary_map::FlussMap;
-use crate::row::datum::{Date, Datum, Time, TimestampLtz, TimestampNtz};
-use crate::row::{Decimal, FlussArray, GenericRow, InternalRow};
-use arrow::array::{
- Array, AsArray, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
FixedSizeBinaryArray,
- Float32Array, Float64Array, Int8Array, Int16Array, Int32Array, Int64Array,
ListArray, MapArray,
- RecordBatch, StringArray, StructArray, Time32MillisecondArray,
Time32SecondArray,
- Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
- TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
-};
-use arrow::datatypes::{
- DataType as ArrowDataType, Date32Type, Decimal128Type, Float32Type,
Float64Type, Int8Type,
- Int16Type, Int32Type, Int64Type, Time32MillisecondType, Time32SecondType,
- Time64MicrosecondType, Time64NanosecondType, TimeUnit,
TimestampMicrosecondType,
- TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType,
-};
+use crate::row::column_vector::{TypedBatch, TypedColumn};
+use crate::row::columnar::{ColumnarArray, ColumnarMap};
+use crate::row::datum::{Date, Time, TimestampLtz, TimestampNtz};
+use crate::row::view::{ArrayView, MapView, RowView};
+use crate::row::{DataGetters, Decimal, InternalRow};
+use arrow::array::{Array, RecordBatch};
use std::sync::Arc;
-#[derive(Clone)]
+/// Cursor over one row of a typed Arrow column-vector batch. Cheap to clone;
+/// advance via [`set_row_id`](Self::set_row_id).
+#[derive(Debug, Clone)]
pub struct ColumnarRow {
- record_batch: Arc<RecordBatch>,
- row_type: Arc<RowType>,
+ batch: Arc<TypedBatch>,
row_id: usize,
- fluss_row_type: Option<Arc<RowType>>,
- row_column_indices: Arc<[usize]>,
- row_caches: Box<[std::sync::OnceLock<GenericRow<'static>>]>,
-}
-
-pub(crate) fn fluss_row_column_indices(row_type: &RowType) -> Arc<[usize]> {
- row_type
- .fields()
- .iter()
- .enumerate()
- .filter_map(|(i, f)| matches!(f.data_type,
DataType::Row(_)).then_some(i))
- .collect()
-}
-
-pub(crate) fn arrow_row_column_indices(batch: &RecordBatch) -> Arc<[usize]> {
- batch
- .columns()
- .iter()
- .enumerate()
- .filter_map(|(i, c)| matches!(c.data_type(),
ArrowDataType::Struct(_)).then_some(i))
- .collect()
-}
-
-fn make_row_caches(indices: &[usize]) ->
Box<[std::sync::OnceLock<GenericRow<'static>>]> {
- indices.iter().map(|_| std::sync::OnceLock::new()).collect()
}
impl ColumnarRow {
- pub fn new(
- batch: Arc<RecordBatch>,
- row_type: Arc<RowType>,
- row_id: usize,
- fluss_row_type: Option<Arc<RowType>>,
- ) -> Self {
- let row_column_indices = match &fluss_row_type {
- Some(rt) => fluss_row_column_indices(rt),
- None => arrow_row_column_indices(&batch),
- };
- Self::with_indices(batch, row_type, row_id, fluss_row_type,
row_column_indices)
+ /// Primary constructor — caller already has a typed batch.
+ pub(crate) fn from_typed_batch(batch: Arc<TypedBatch>, row_id: usize) ->
Self {
+ Self { batch, row_id }
}
- pub(crate) fn with_indices(
- batch: Arc<RecordBatch>,
+ /// Builds the typed batch in-line from a `RecordBatch` plus [`RowType`].
+ /// `fluss_row_type`, when provided, overrides `row_type`. If both are
+ /// empty, the Fluss types are inferred from the Arrow schema.
+ pub fn new(
+ record_batch: Arc<RecordBatch>,
row_type: Arc<RowType>,
row_id: usize,
fluss_row_type: Option<Arc<RowType>>,
- row_column_indices: Arc<[usize]>,
- ) -> Self {
- let row_caches = make_row_caches(&row_column_indices);
- ColumnarRow {
- record_batch: batch,
- row_type,
- row_id,
- fluss_row_type,
- row_column_indices,
- row_caches,
- }
- }
-
- pub fn fluss_row_type(&self) -> Option<&Arc<RowType>> {
- self.fluss_row_type.as_ref()
+ ) -> Result<Self> {
+ let schema = pick_schema(&record_batch, &row_type,
fluss_row_type.as_deref())?;
+ let typed = TypedBatch::build(&record_batch, &schema)?;
+ Ok(Self::from_typed_batch(Arc::new(typed), row_id))
}
pub fn set_row_id(&mut self, row_id: usize) {
self.row_id = row_id;
- for lock in self.row_caches.iter_mut() {
- *lock = std::sync::OnceLock::new();
- }
}
pub fn get_row_id(&self) -> usize {
self.row_id
}
+ /// Returns the source `RecordBatch`. Panics on a nested cursor (returned
Review Comment:
Changed it to return Option<&RecordBatch> - nested rows just get None now
instead of panicking, it's better - a good call.
--
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]
