Address comments

Author: charlesdong1991

bindings/cpp/src/types.rs

@@ -351,6 +351,8 @@ pub fn resolve_row_types(
             Datum::Time(t) => Datum::Time(*t),
             Datum::TimestampNtz(ts) => Datum::TimestampNtz(*ts),
             Datum::TimestampLtz(ts) => Datum::TimestampLtz(*ts),
+            // TODO: C++ bindings need proper CXX wrapper types for FlussArray
+            // before C++ users can construct or inspect array values through FFI.
             Datum::Array(a) => Datum::Array(a.clone()),
         };
         out.set_field(idx, resolved);
@@ -409,6 +411,8 @@ pub fn compacted_row_to_owned(
             fcore::metadata::DataType::Binary(dt) => {
                 Datum::Blob(Cow::Owned(row.get_binary(i, dt.length())?.to_vec()))
             }
+            // TODO: C++ bindings need proper CXX wrapper types for FlussArray
+            // before C++ users can construct or inspect array values through FFI.
             fcore::metadata::DataType::Array(_) => Datum::Array(row.get_array(i)?),
             other => return Err(anyhow!("Unsupported data type for column {i}: {other:?}")),
         };

crates/fluss/src/record/arrow.rs

@@ -332,7 +332,7 @@ impl RowAppendRecordBatchBuilder {
             }
             arrow_schema::DataType::List(field) => {
                 let inner_builder = Self::create_builder(field.data_type(), capacity)?;
-                Ok(Box::new(ListBuilder::new(inner_builder)))
+                Ok(Box::new(ListBuilder::with_capacity(inner_builder, capacity)))
             }
             arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Millisecond, _) => Ok(
                 Box::new(TimestampMillisecondBuilder::with_capacity(capacity)),

crates/fluss/src/row/binary_array.rs

@@ -310,23 +310,21 @@ impl FlussArray {
         if TimestampNtz::is_compact(precision) {
             Ok(TimestampNtz::new(self.get_long(pos)))
         } else {
-            let (offset, _size) = self.get_offset_and_size(pos);
+            let (offset, nanos_of_millis) = self.get_offset_and_size(pos);
             let millis_bytes = self.checked_slice(offset, 8, "timestamp ntz millis")?;
             let millis = i64::from_ne_bytes(millis_bytes.try_into().unwrap());
-            let nanos = _size as i32;
-            TimestampNtz::from_millis_nanos(millis, nanos)
+            TimestampNtz::from_millis_nanos(millis, nanos_of_millis as i32)
         }
     }
 
     pub fn get_timestamp_ltz(&self, pos: usize, precision: u32) -> Result<TimestampLtz> {
         if TimestampLtz::is_compact(precision) {
             Ok(TimestampLtz::new(self.get_long(pos)))
         } else {
-            let (offset, _size) = self.get_offset_and_size(pos);
+            let (offset, nanos_of_millis) = self.get_offset_and_size(pos);
             let millis_bytes = self.checked_slice(offset, 8, "timestamp ltz millis")?;
             let millis = i64::from_ne_bytes(millis_bytes.try_into().unwrap());
-            let nanos = _size as i32;
-            TimestampLtz::from_millis_nanos(millis, nanos)
+            TimestampLtz::from_millis_nanos(millis, nanos_of_millis as i32)
         }
     }
 

crates/fluss/src/row/datum.rs

@@ -504,6 +504,9 @@ fn append_fluss_array_to_list_builder(
 
     for i in 0..arr.size() {
         if arr.is_null_at(i) {
+            // TODO: Datum::Null triggers a chain of downcast attempts in append_to.
+            // For sparse arrays with many nulls, call append_null directly on the
+            // typed inner builder to avoid the overhead.
             let null_datum = Datum::Null;
             null_datum.append_to(values_builder, &element_arrow_type)?;
         } else {