Copilot commented on code in PR #330:
URL: https://github.com/apache/fluss-rust/pull/330#discussion_r2808773737
##########
bindings/cpp/src/table.cpp:
##########
@@ -458,8 +834,7 @@ Result UpsertWriter::Upsert(const GenericRow& row,
WriteResult& out) {
}
try {
- auto ffi_row = utils::to_ffi_generic_row(row);
- auto rust_box = writer_->upsert(ffi_row);
+ auto rust_box = writer_->upsert(*row.inner_);
out = WriteResult(rust_box.into_raw());
Review Comment:
`UpsertWriter::Upsert()` dereferences `row.inner_` without checking that the
`GenericRow` is still available. A moved-from `GenericRow` has `inner_ ==
nullptr` and this will segfault. Please validate `row.Available()` (or
`row.inner_ != nullptr`) and return a client error instead.
##########
bindings/cpp/src/table.cpp:
##########
@@ -529,34 +906,26 @@ Lookuper& Lookuper::operator=(Lookuper&& other) noexcept {
bool Lookuper::Available() const { return lookuper_ != nullptr; }
-Result Lookuper::Lookup(const GenericRow& pk_row, bool& found, GenericRow&
out) {
+Result Lookuper::Lookup(const GenericRow& pk_row, LookupResult& out) {
if (!Available()) {
return utils::make_client_error("Lookuper not available");
}
Review Comment:
`Lookuper::Lookup()` dereferences `pk_row.inner_` without checking that the
`GenericRow` is still available. If the caller passes a moved-from `GenericRow`
(where `inner_ == nullptr`), this will segfault. Please validate
`pk_row.Available()` (or `pk_row.inner_ != nullptr`) and return a client error
instead.
```suggestion
if (!pk_row.Available()) {
return utils::make_client_error("Primary key row not available");
}
```
##########
bindings/cpp/src/table.cpp:
##########
@@ -79,6 +79,321 @@ int Date::Day() const {
return tm.tm_mday;
}
+// ============================================================================
+// GenericRow — write-only row backed by opaque Rust GenericRowInner
+// ============================================================================
+
+GenericRow::GenericRow() {
+ auto box = ffi::new_generic_row(0);
+ inner_ = box.into_raw();
+}
+
+GenericRow::GenericRow(size_t field_count) {
+ auto box = ffi::new_generic_row(field_count);
+ inner_ = box.into_raw();
+}
+
+GenericRow::~GenericRow() noexcept { Destroy(); }
+
+void GenericRow::Destroy() noexcept {
+ if (inner_) {
+ rust::Box<ffi::GenericRowInner>::from_raw(inner_);
+ inner_ = nullptr;
+ }
+}
+
+GenericRow::GenericRow(GenericRow&& other) noexcept
+ : inner_(other.inner_), column_map_(std::move(other.column_map_)) {
+ other.inner_ = nullptr;
+}
+
+GenericRow& GenericRow::operator=(GenericRow&& other) noexcept {
+ if (this != &other) {
+ Destroy();
+ inner_ = other.inner_;
+ column_map_ = std::move(other.column_map_);
+ other.inner_ = nullptr;
+ }
+ return *this;
+}
+
+bool GenericRow::Available() const { return inner_ != nullptr; }
+
+void GenericRow::Reset() {
+ if (inner_) inner_->gr_reset();
+}
+
+void GenericRow::SetNull(size_t idx) {
+ if (inner_) inner_->gr_set_null(idx);
+}
+void GenericRow::SetBool(size_t idx, bool v) {
+ if (inner_) inner_->gr_set_bool(idx, v);
+}
+void GenericRow::SetInt32(size_t idx, int32_t v) {
+ if (inner_) inner_->gr_set_i32(idx, v);
+}
+void GenericRow::SetInt64(size_t idx, int64_t v) {
+ if (inner_) inner_->gr_set_i64(idx, v);
+}
+void GenericRow::SetFloat32(size_t idx, float v) {
+ if (inner_) inner_->gr_set_f32(idx, v);
+}
+void GenericRow::SetFloat64(size_t idx, double v) {
+ if (inner_) inner_->gr_set_f64(idx, v);
+}
+
+void GenericRow::SetString(size_t idx, std::string v) {
+ if (inner_) inner_->gr_set_str(idx, v);
+}
+
+void GenericRow::SetBytes(size_t idx, std::vector<uint8_t> v) {
+ if (inner_) inner_->gr_set_bytes(idx, rust::Slice<const uint8_t>(v.data(),
v.size()));
+}
+
+void GenericRow::SetDate(size_t idx, fluss::Date d) {
+ if (inner_) inner_->gr_set_date(idx, d.days_since_epoch);
+}
+
+void GenericRow::SetTime(size_t idx, fluss::Time t) {
+ if (inner_) inner_->gr_set_time(idx, t.millis_since_midnight);
+}
+
+void GenericRow::SetTimestampNtz(size_t idx, fluss::Timestamp ts) {
+ if (inner_) inner_->gr_set_ts_ntz(idx, ts.epoch_millis,
ts.nano_of_millisecond);
+}
+
+void GenericRow::SetTimestampLtz(size_t idx, fluss::Timestamp ts) {
+ if (inner_) inner_->gr_set_ts_ltz(idx, ts.epoch_millis,
ts.nano_of_millisecond);
+}
+
+void GenericRow::SetDecimal(size_t idx, const std::string& value) {
+ if (inner_) inner_->gr_set_decimal_str(idx, value);
+}
+
+// ============================================================================
+// RowView — zero-copy read-only row view for scan results
+// ============================================================================
+
+size_t RowView::FieldCount() const { return inner_ ?
inner_->sv_field_count(record_idx_) : 0; }
+
+TypeId RowView::GetType(size_t idx) const {
+ return inner_ ? static_cast<TypeId>(inner_->sv_column_type(idx)) :
TypeId::Int;
+}
+
+bool RowView::IsNull(size_t idx) const {
+ return inner_ ? inner_->sv_is_null(record_idx_, idx) : true;
+}
+bool RowView::GetBool(size_t idx) const {
+ return inner_ ? inner_->sv_get_bool(record_idx_, idx) : false;
+}
+int32_t RowView::GetInt32(size_t idx) const {
+ return inner_ ? inner_->sv_get_i32(record_idx_, idx) : 0;
+}
+int64_t RowView::GetInt64(size_t idx) const {
+ return inner_ ? inner_->sv_get_i64(record_idx_, idx) : 0;
+}
+float RowView::GetFloat32(size_t idx) const {
+ return inner_ ? inner_->sv_get_f32(record_idx_, idx) : 0.0f;
+}
+double RowView::GetFloat64(size_t idx) const {
+ return inner_ ? inner_->sv_get_f64(record_idx_, idx) : 0.0;
Review Comment:
`RowView::GetType()` casts `sv_column_type(idx)` directly to `TypeId`. If
Rust returns `0` for out-of-range/unknown columns, this produces an invalid
`TypeId` value (there is no 0). Also, none of the RowView getters validate
`idx` against `FieldCount()`, so out-of-range access can currently panic in
Rust and abort across FFI. Please add an explicit bounds check (throw or return
an error/default) before calling into Rust.
```suggestion
if (!inner_) {
return TypeId::Int;
}
const size_t field_count = FieldCount();
if (idx >= field_count) {
// Out-of-range index: return default without calling into Rust.
return TypeId::Int;
}
return static_cast<TypeId>(inner_->sv_column_type(idx));
}
bool RowView::IsNull(size_t idx) const {
if (!inner_) {
return true;
}
const size_t field_count = FieldCount();
if (idx >= field_count) {
// Out-of-range index: treat as null without calling into Rust.
return true;
}
return inner_->sv_is_null(record_idx_, idx);
}
bool RowView::GetBool(size_t idx) const {
if (!inner_) {
return false;
}
const size_t field_count = FieldCount();
if (idx >= field_count) {
// Out-of-range index: return default value without calling into
Rust.
return false;
}
return inner_->sv_get_bool(record_idx_, idx);
}
int32_t RowView::GetInt32(size_t idx) const {
if (!inner_) {
return 0;
}
const size_t field_count = FieldCount();
if (idx >= field_count) {
// Out-of-range index: return default value without calling into
Rust.
return 0;
}
return inner_->sv_get_i32(record_idx_, idx);
}
int64_t RowView::GetInt64(size_t idx) const {
if (!inner_) {
return 0;
}
const size_t field_count = FieldCount();
if (idx >= field_count) {
// Out-of-range index: return default value without calling into
Rust.
return 0;
}
return inner_->sv_get_i64(record_idx_, idx);
}
float RowView::GetFloat32(size_t idx) const {
if (!inner_) {
return 0.0f;
}
const size_t field_count = FieldCount();
if (idx >= field_count) {
// Out-of-range index: return default value without calling into
Rust.
return 0.0f;
}
return inner_->sv_get_f32(record_idx_, idx);
}
double RowView::GetFloat64(size_t idx) const {
if (!inner_) {
return 0.0;
}
const size_t field_count = FieldCount();
if (idx >= field_count) {
// Out-of-range index: return default value without calling into
Rust.
return 0.0;
}
return inner_->sv_get_f64(record_idx_, idx);
```
##########
bindings/cpp/src/table.cpp:
##########
@@ -398,8 +731,7 @@ Result AppendWriter::Append(const GenericRow& row,
WriteResult& out) {
}
Review Comment:
`AppendWriter::Append()` dereferences `row.inner_` without checking that the
`GenericRow` is still available. A moved-from `GenericRow` (or otherwise
invalid instance) will have `inner_ == nullptr` and this will segfault. Please
validate `row.Available()` (or `row.inner_ != nullptr`) and return a client
error instead of dereferencing a null pointer.
```suggestion
// Ensure the GenericRow is still valid (not moved-from or otherwise
invalid)
if (!row.Available()) {
return utils::make_client_error("GenericRow not available");
}
```
##########
bindings/cpp/src/types.rs:
##########
@@ -110,6 +87,8 @@ fn core_data_type_to_ffi(dt: &fcore::metadata::DataType) ->
i32 {
fcore::metadata::DataType::Timestamp(_) => DATA_TYPE_TIMESTAMP,
fcore::metadata::DataType::TimestampLTz(_) => DATA_TYPE_TIMESTAMP_LTZ,
fcore::metadata::DataType::Decimal(_) => DATA_TYPE_DECIMAL,
+ fcore::metadata::DataType::Char(_) => DATA_TYPE_CHAR,
+ fcore::metadata::DataType::Binary(_) => DATA_TYPE_BINARY,
_ => 0,
Review Comment:
`core_data_type_to_ffi()` falls back to returning `0` for unsupported core
types. On the C++ side this value is cast to `TypeId`, but `TypeId` has no `0`
variant, so callers can end up with an invalid enum value (e.g., when
new/unsupported types are introduced). Consider returning a dedicated "unknown"
type id (and adding `TypeId::Unknown = 0`), or changing this helper to return a
`Result<i32>`/`Option<i32>` and surfacing an error instead of silently
returning `0`.
```suggestion
_ => panic!("Unsupported core data type in core_data_type_to_ffi:
{:?}", dt),
```
##########
bindings/cpp/src/table.cpp:
##########
@@ -480,8 +855,7 @@ Result UpsertWriter::Delete(const GenericRow& row,
WriteResult& out) {
}
try {
- auto ffi_row = utils::to_ffi_generic_row(row);
- auto rust_box = writer_->delete_row(ffi_row);
+ auto rust_box = writer_->delete_row(*row.inner_);
out = WriteResult(rust_box.into_raw());
return utils::make_ok();
Review Comment:
`UpsertWriter::Delete()` dereferences `row.inner_` without checking that the
`GenericRow` is still available. A moved-from `GenericRow` has `inner_ ==
nullptr` and this will segfault. Please validate `row.Available()` (or
`row.inner_ != nullptr`) and return a client error instead.
##########
bindings/cpp/src/table.cpp:
##########
@@ -79,6 +79,321 @@ int Date::Day() const {
return tm.tm_mday;
}
+// ============================================================================
+// GenericRow — write-only row backed by opaque Rust GenericRowInner
+// ============================================================================
+
+GenericRow::GenericRow() {
+ auto box = ffi::new_generic_row(0);
+ inner_ = box.into_raw();
+}
+
+GenericRow::GenericRow(size_t field_count) {
+ auto box = ffi::new_generic_row(field_count);
+ inner_ = box.into_raw();
+}
+
+GenericRow::~GenericRow() noexcept { Destroy(); }
+
+void GenericRow::Destroy() noexcept {
+ if (inner_) {
+ rust::Box<ffi::GenericRowInner>::from_raw(inner_);
+ inner_ = nullptr;
+ }
+}
+
+GenericRow::GenericRow(GenericRow&& other) noexcept
+ : inner_(other.inner_), column_map_(std::move(other.column_map_)) {
+ other.inner_ = nullptr;
+}
+
+GenericRow& GenericRow::operator=(GenericRow&& other) noexcept {
+ if (this != &other) {
+ Destroy();
+ inner_ = other.inner_;
+ column_map_ = std::move(other.column_map_);
+ other.inner_ = nullptr;
+ }
+ return *this;
+}
+
+bool GenericRow::Available() const { return inner_ != nullptr; }
+
+void GenericRow::Reset() {
+ if (inner_) inner_->gr_reset();
+}
+
+void GenericRow::SetNull(size_t idx) {
+ if (inner_) inner_->gr_set_null(idx);
+}
+void GenericRow::SetBool(size_t idx, bool v) {
+ if (inner_) inner_->gr_set_bool(idx, v);
+}
+void GenericRow::SetInt32(size_t idx, int32_t v) {
+ if (inner_) inner_->gr_set_i32(idx, v);
+}
+void GenericRow::SetInt64(size_t idx, int64_t v) {
+ if (inner_) inner_->gr_set_i64(idx, v);
+}
+void GenericRow::SetFloat32(size_t idx, float v) {
+ if (inner_) inner_->gr_set_f32(idx, v);
+}
+void GenericRow::SetFloat64(size_t idx, double v) {
+ if (inner_) inner_->gr_set_f64(idx, v);
+}
+
+void GenericRow::SetString(size_t idx, std::string v) {
+ if (inner_) inner_->gr_set_str(idx, v);
+}
+
+void GenericRow::SetBytes(size_t idx, std::vector<uint8_t> v) {
+ if (inner_) inner_->gr_set_bytes(idx, rust::Slice<const uint8_t>(v.data(),
v.size()));
+}
+
+void GenericRow::SetDate(size_t idx, fluss::Date d) {
+ if (inner_) inner_->gr_set_date(idx, d.days_since_epoch);
+}
+
+void GenericRow::SetTime(size_t idx, fluss::Time t) {
+ if (inner_) inner_->gr_set_time(idx, t.millis_since_midnight);
+}
+
+void GenericRow::SetTimestampNtz(size_t idx, fluss::Timestamp ts) {
+ if (inner_) inner_->gr_set_ts_ntz(idx, ts.epoch_millis,
ts.nano_of_millisecond);
+}
+
+void GenericRow::SetTimestampLtz(size_t idx, fluss::Timestamp ts) {
+ if (inner_) inner_->gr_set_ts_ltz(idx, ts.epoch_millis,
ts.nano_of_millisecond);
+}
+
+void GenericRow::SetDecimal(size_t idx, const std::string& value) {
+ if (inner_) inner_->gr_set_decimal_str(idx, value);
+}
+
+// ============================================================================
+// RowView — zero-copy read-only row view for scan results
+// ============================================================================
+
+size_t RowView::FieldCount() const { return inner_ ?
inner_->sv_field_count(record_idx_) : 0; }
+
+TypeId RowView::GetType(size_t idx) const {
+ return inner_ ? static_cast<TypeId>(inner_->sv_column_type(idx)) :
TypeId::Int;
+}
+
+bool RowView::IsNull(size_t idx) const {
+ return inner_ ? inner_->sv_is_null(record_idx_, idx) : true;
+}
+bool RowView::GetBool(size_t idx) const {
+ return inner_ ? inner_->sv_get_bool(record_idx_, idx) : false;
+}
+int32_t RowView::GetInt32(size_t idx) const {
+ return inner_ ? inner_->sv_get_i32(record_idx_, idx) : 0;
+}
+int64_t RowView::GetInt64(size_t idx) const {
+ return inner_ ? inner_->sv_get_i64(record_idx_, idx) : 0;
+}
+float RowView::GetFloat32(size_t idx) const {
+ return inner_ ? inner_->sv_get_f32(record_idx_, idx) : 0.0f;
+}
+double RowView::GetFloat64(size_t idx) const {
+ return inner_ ? inner_->sv_get_f64(record_idx_, idx) : 0.0;
+}
+
+std::string_view RowView::GetString(size_t idx) const {
+ if (!inner_) return {};
+ auto s = inner_->sv_get_str(record_idx_, idx);
+ return std::string_view(s.data(), s.size());
+}
+
+std::pair<const uint8_t*, size_t> RowView::GetBytes(size_t idx) const {
+ if (!inner_) return {nullptr, 0};
+ auto bytes = inner_->sv_get_bytes(record_idx_, idx);
+ return {bytes.data(), bytes.size()};
+}
+
+Date RowView::GetDate(size_t idx) const {
+ return inner_ ? Date{inner_->sv_get_date_days(record_idx_, idx)} : Date{};
+}
+
+Time RowView::GetTime(size_t idx) const {
+ return inner_ ? Time{inner_->sv_get_time_millis(record_idx_, idx)} :
Time{};
+}
+
+Timestamp RowView::GetTimestamp(size_t idx) const {
+ if (!inner_) return {};
+ return Timestamp{inner_->sv_get_ts_millis(record_idx_, idx),
+ inner_->sv_get_ts_nanos(record_idx_, idx)};
+}
+
+bool RowView::IsDecimal(size_t idx) const { return GetType(idx) ==
TypeId::Decimal; }
+
+std::string RowView::GetDecimalString(size_t idx) const {
+ if (!inner_) return {};
+ return std::string(inner_->sv_get_decimal_str(record_idx_, idx));
+}
+
+// ============================================================================
+// ScanRecords — backed by opaque Rust ScanResultInner
+// ============================================================================
+
+ScanRecords::ScanRecords() noexcept = default;
+
+ScanRecords::~ScanRecords() noexcept { Destroy(); }
+
+void ScanRecords::Destroy() noexcept {
+ if (inner_) {
+ rust::Box<ffi::ScanResultInner>::from_raw(inner_);
+ inner_ = nullptr;
+ }
+}
+
+ScanRecords::ScanRecords(ScanRecords&& other) noexcept
+ : inner_(other.inner_), column_map_(std::move(other.column_map_)) {
+ other.inner_ = nullptr;
+}
+
+ScanRecords& ScanRecords::operator=(ScanRecords&& other) noexcept {
+ if (this != &other) {
+ Destroy();
+ inner_ = other.inner_;
+ column_map_ = std::move(other.column_map_);
+ other.inner_ = nullptr;
+ }
+ return *this;
+}
+
+size_t ScanRecords::Size() const { return inner_ ? inner_->sv_record_count() :
0; }
+
+bool ScanRecords::Empty() const { return Size() == 0; }
+
+void ScanRecords::BuildColumnMap() const {
+ if (!inner_) return;
+ auto map = std::make_shared<detail::ColumnMap>();
+ auto count = inner_->sv_column_count();
+ for (size_t i = 0; i < count; ++i) {
+ auto name = inner_->sv_column_name(i);
+ (*map)[std::string(name.data(), name.size())] = {
+ i, static_cast<TypeId>(inner_->sv_column_type(i))};
+ }
+ column_map_ = std::move(map);
+}
+
+const std::shared_ptr<detail::ColumnMap>& ScanRecords::GetColumnMap() const {
+ if (!column_map_) {
+ BuildColumnMap();
+ }
+ return column_map_;
+}
+
+ScanRecord ScanRecords::operator[](size_t idx) const {
+ if (!inner_ || idx >= inner_->sv_record_count())
+ return ScanRecord{0, std::nullopt, 0,
+ 0, ChangeType::AppendOnly, RowView(nullptr, 0,
nullptr)};
+ return ScanRecord{inner_->sv_bucket_id(idx),
+ inner_->sv_has_partition_id(idx)
+ ?
std::optional<int64_t>(inner_->sv_partition_id(idx))
+ : std::nullopt,
+ inner_->sv_offset(idx),
+ inner_->sv_timestamp(idx),
+ static_cast<ChangeType>(inner_->sv_change_type(idx)),
+ RowView(inner_, idx, GetColumnMap().get())};
+}
+
+ScanRecord ScanRecords::Iterator::operator*() const { return
owner_->operator[](idx_); }
+
+// ============================================================================
+// LookupResult — backed by opaque Rust LookupResultInner
+// ============================================================================
+
+LookupResult::LookupResult() noexcept = default;
+
+LookupResult::~LookupResult() noexcept { Destroy(); }
+
+void LookupResult::Destroy() noexcept {
+ if (inner_) {
+ rust::Box<ffi::LookupResultInner>::from_raw(inner_);
+ inner_ = nullptr;
+ }
+}
+
+LookupResult::LookupResult(LookupResult&& other) noexcept
+ : inner_(other.inner_), column_map_(std::move(other.column_map_)) {
+ other.inner_ = nullptr;
+}
+
+LookupResult& LookupResult::operator=(LookupResult&& other) noexcept {
+ if (this != &other) {
+ Destroy();
+ inner_ = other.inner_;
+ column_map_ = std::move(other.column_map_);
+ other.inner_ = nullptr;
+ }
+ return *this;
+}
+
+void LookupResult::BuildColumnMap() const {
+ if (!inner_) return;
+ auto map = std::make_shared<detail::ColumnMap>();
+ auto count = inner_->lv_field_count();
+ for (size_t i = 0; i < count; ++i) {
+ auto name = inner_->lv_column_name(i);
+ (*map)[std::string(name.data(), name.size())] = {
+ i, static_cast<TypeId>(inner_->lv_column_type(i))};
+ }
+ column_map_ = std::move(map);
+}
+
+bool LookupResult::Found() const { return inner_ && inner_->lv_found(); }
+
+size_t LookupResult::FieldCount() const { return inner_ ?
inner_->lv_field_count() : 0; }
+
+TypeId LookupResult::GetType(size_t idx) const {
+ return inner_ ? static_cast<TypeId>(inner_->lv_column_type(idx)) :
TypeId::Int;
+}
Review Comment:
`LookupResult::GetType()` (and `BuildColumnMap()`) casts the integer type id
from Rust directly to `TypeId`. Rust currently returns `0` for
out-of-range/unknown fields, but `TypeId` has no `0` variant, producing an
invalid enum value. Also consider adding `idx < FieldCount()` checks in the
getters to avoid panics in Rust on out-of-range access.
##########
bindings/cpp/src/lib.rs:
##########
@@ -1417,3 +1510,494 @@ impl LogScanner {
}
}
}
+
+// ============================================================================
+// Opaque types: GenericRowInner (write path)
+// ============================================================================
+
+pub struct GenericRowInner {
+ row: fcore::row::GenericRow<'static>,
+}
+
+fn new_generic_row(field_count: usize) -> Box<GenericRowInner> {
+ Box::new(GenericRowInner {
+ row: fcore::row::GenericRow::new(field_count),
+ })
+}
+
+impl GenericRowInner {
+ fn gr_reset(&mut self) {
+ let len = self.row.values.len();
+ self.row = fcore::row::GenericRow::new(len);
+ }
+
+ fn gr_set_null(&mut self, idx: usize) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Null);
+ }
+
+ fn gr_set_bool(&mut self, idx: usize, val: bool) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Bool(val));
+ }
+
+ fn gr_set_i32(&mut self, idx: usize, val: i32) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Int32(val));
+ }
+
+ fn gr_set_i64(&mut self, idx: usize, val: i64) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Int64(val));
+ }
+
+ fn gr_set_f32(&mut self, idx: usize, val: f32) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx, fcore::row::Datum::Float32(val.into()));
+ }
+
+ fn gr_set_f64(&mut self, idx: usize, val: f64) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx, fcore::row::Datum::Float64(val.into()));
+ }
+
+ fn gr_set_str(&mut self, idx: usize, val: &str) {
+ self.ensure_size(idx);
+ self.row.set_field(
+ idx,
+
fcore::row::Datum::String(std::borrow::Cow::Owned(val.to_string())),
+ );
+ }
+
+ fn gr_set_bytes(&mut self, idx: usize, val: &[u8]) {
+ self.ensure_size(idx);
+ self.row.set_field(
+ idx,
+ fcore::row::Datum::Blob(std::borrow::Cow::Owned(val.to_vec())),
+ );
+ }
+
+ fn gr_set_date(&mut self, idx: usize, days: i32) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx,
fcore::row::Datum::Date(fcore::row::Date::new(days)));
+ }
+
+ fn gr_set_time(&mut self, idx: usize, millis: i32) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx,
fcore::row::Datum::Time(fcore::row::Time::new(millis)));
+ }
+
+ fn gr_set_ts_ntz(&mut self, idx: usize, millis: i64, nanos: i32) {
+ self.ensure_size(idx);
+ // Use from_millis_nanos, falling back to millis-only on error
+ let ts = fcore::row::TimestampNtz::from_millis_nanos(millis, nanos)
+ .unwrap_or_else(|_| fcore::row::TimestampNtz::new(millis));
+ self.row.set_field(idx, fcore::row::Datum::TimestampNtz(ts));
+ }
+
+ fn gr_set_ts_ltz(&mut self, idx: usize, millis: i64, nanos: i32) {
+ self.ensure_size(idx);
+ let ts = fcore::row::TimestampLtz::from_millis_nanos(millis, nanos)
+ .unwrap_or_else(|_| fcore::row::TimestampLtz::new(millis));
+ self.row.set_field(idx, fcore::row::Datum::TimestampLtz(ts));
+ }
+
+ fn gr_set_decimal_str(&mut self, idx: usize, val: &str) {
+ self.ensure_size(idx);
+ // Store as string; resolve_row_types() will parse and validate
against schema
+ self.row.set_field(
+ idx,
+
fcore::row::Datum::String(std::borrow::Cow::Owned(val.to_string())),
+ );
+ }
+
+ fn ensure_size(&mut self, idx: usize) {
+ if self.row.values.len() <= idx {
+ self.row.values.resize(idx + 1, fcore::row::Datum::Null);
+ }
+ }
+}
+
+// ============================================================================
+// Shared row-reading helpers (used by both ScanResultInner and
LookupResultInner)
+// ============================================================================
+
+mod row_reader {
+ use fcore::row::InternalRow;
+ use fluss as fcore;
+
+ use crate::types;
+
+ pub fn column_type(columns: &[fcore::metadata::Column], field: usize) ->
i32 {
+ columns
+ .get(field)
+ .map_or(0, |c| types::core_data_type_to_ffi(c.data_type()))
+ }
+
+ pub fn column_name(columns: &[fcore::metadata::Column], field: usize) ->
&str {
+ columns.get(field).map_or("", |c| c.name())
+ }
+
+ pub fn is_null(row: &dyn InternalRow, _field: usize) -> bool {
+ row.is_null_at(_field)
+ }
+
+ pub fn get_bool(row: &dyn InternalRow, field: usize) -> bool {
+ row.get_boolean(field)
+ }
Review Comment:
`row_reader` delegates to `InternalRow` getters like
`get_boolean/get_int/...` without bounds checks. `InternalRow` implementations
(e.g., `GenericRow`) `unwrap()/expect()` on index access, so an out-of-range
`field` (or mismatch between `columns.len()` and `row.get_field_count()`) will
panic and can abort across the CXX FFI boundary. Please guard `field <
row.get_field_count()` in these helpers (and return a default / error signal)
to avoid panics from C++-provided indices.
##########
bindings/cpp/src/lib.rs:
##########
@@ -1417,3 +1510,494 @@ impl LogScanner {
}
}
}
+
+// ============================================================================
+// Opaque types: GenericRowInner (write path)
+// ============================================================================
+
+pub struct GenericRowInner {
+ row: fcore::row::GenericRow<'static>,
+}
+
+fn new_generic_row(field_count: usize) -> Box<GenericRowInner> {
+ Box::new(GenericRowInner {
+ row: fcore::row::GenericRow::new(field_count),
+ })
+}
+
+impl GenericRowInner {
+ fn gr_reset(&mut self) {
+ let len = self.row.values.len();
+ self.row = fcore::row::GenericRow::new(len);
+ }
+
+ fn gr_set_null(&mut self, idx: usize) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Null);
+ }
+
+ fn gr_set_bool(&mut self, idx: usize, val: bool) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Bool(val));
+ }
+
+ fn gr_set_i32(&mut self, idx: usize, val: i32) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Int32(val));
+ }
+
+ fn gr_set_i64(&mut self, idx: usize, val: i64) {
+ self.ensure_size(idx);
+ self.row.set_field(idx, fcore::row::Datum::Int64(val));
+ }
+
+ fn gr_set_f32(&mut self, idx: usize, val: f32) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx, fcore::row::Datum::Float32(val.into()));
+ }
+
+ fn gr_set_f64(&mut self, idx: usize, val: f64) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx, fcore::row::Datum::Float64(val.into()));
+ }
+
+ fn gr_set_str(&mut self, idx: usize, val: &str) {
+ self.ensure_size(idx);
+ self.row.set_field(
+ idx,
+
fcore::row::Datum::String(std::borrow::Cow::Owned(val.to_string())),
+ );
+ }
+
+ fn gr_set_bytes(&mut self, idx: usize, val: &[u8]) {
+ self.ensure_size(idx);
+ self.row.set_field(
+ idx,
+ fcore::row::Datum::Blob(std::borrow::Cow::Owned(val.to_vec())),
+ );
+ }
+
+ fn gr_set_date(&mut self, idx: usize, days: i32) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx,
fcore::row::Datum::Date(fcore::row::Date::new(days)));
+ }
+
+ fn gr_set_time(&mut self, idx: usize, millis: i32) {
+ self.ensure_size(idx);
+ self.row
+ .set_field(idx,
fcore::row::Datum::Time(fcore::row::Time::new(millis)));
+ }
+
+ fn gr_set_ts_ntz(&mut self, idx: usize, millis: i64, nanos: i32) {
+ self.ensure_size(idx);
+ // Use from_millis_nanos, falling back to millis-only on error
+ let ts = fcore::row::TimestampNtz::from_millis_nanos(millis, nanos)
+ .unwrap_or_else(|_| fcore::row::TimestampNtz::new(millis));
+ self.row.set_field(idx, fcore::row::Datum::TimestampNtz(ts));
+ }
+
+ fn gr_set_ts_ltz(&mut self, idx: usize, millis: i64, nanos: i32) {
+ self.ensure_size(idx);
+ let ts = fcore::row::TimestampLtz::from_millis_nanos(millis, nanos)
+ .unwrap_or_else(|_| fcore::row::TimestampLtz::new(millis));
+ self.row.set_field(idx, fcore::row::Datum::TimestampLtz(ts));
+ }
+
+ fn gr_set_decimal_str(&mut self, idx: usize, val: &str) {
+ self.ensure_size(idx);
+ // Store as string; resolve_row_types() will parse and validate
against schema
+ self.row.set_field(
+ idx,
+
fcore::row::Datum::String(std::borrow::Cow::Owned(val.to_string())),
+ );
+ }
+
+ fn ensure_size(&mut self, idx: usize) {
+ if self.row.values.len() <= idx {
+ self.row.values.resize(idx + 1, fcore::row::Datum::Null);
+ }
+ }
+}
+
+// ============================================================================
+// Shared row-reading helpers (used by both ScanResultInner and
LookupResultInner)
+// ============================================================================
+
+mod row_reader {
+ use fcore::row::InternalRow;
+ use fluss as fcore;
+
+ use crate::types;
+
+ pub fn column_type(columns: &[fcore::metadata::Column], field: usize) ->
i32 {
+ columns
+ .get(field)
+ .map_or(0, |c| types::core_data_type_to_ffi(c.data_type()))
Review Comment:
`column_type()` returns `0` when `field` is out of range. C++ casts this
integer directly to `TypeId`, but `TypeId` has no `0` value, so this can
produce an invalid enum. Consider returning a dedicated "unknown" id (and
exposing it in C++), or make the accessor return an error/Option so
out-of-range is handled explicitly.
```suggestion
let column = columns.get(field).unwrap_or_else(|| {
panic!(
"row_reader::column_type: field index {} out of range (len =
{})",
field,
columns.len()
)
});
types::core_data_type_to_ffi(column.data_type())
```
--
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]
