Google Git
Sign in
apache/fluss-rust/refs/heads/dependabot/github_actions/actions/cache-6/./bindings/cpp/src/lib.rs
blob: 2d62136a3986b985b42df61bc8f6929df35699f6 [file]
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
mod types;
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::{Arc, LazyLock, Mutex};
use std::time::Duration;
use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
use fluss as fcore;
use fluss::PartitionId;
use fluss::client::PrefixKeyLookuper;
use fluss::error::Error;
use fluss::metadata::{Column, DataType, TableInfo};
use fluss::row::{Datum, GenericRow};
use fluss::rpc::FlussError as CoreFlussError;
use fluss::rpc::message::OffsetSpec;
static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
});
#[cxx::bridge(namespace = "fluss::ffi")]
mod ffi {
struct HashMapValue {
key: String,
value: String,
}
struct FfiConfig {
bootstrap_servers: String,
writer_request_max_size: i32,
writer_acks: String,
writer_retries: i32,
writer_batch_size: i32,
writer_dynamic_batch_size_enabled: bool,
writer_dynamic_batch_size_min: i32,
writer_bucket_no_key_assigner: String,
scanner_remote_log_prefetch_num: usize,
remote_file_download_thread_num: usize,
scanner_remote_log_read_concurrency: usize,
scanner_log_max_poll_records: usize,
scanner_log_fetch_max_bytes: i32,
scanner_log_fetch_min_bytes: i32,
scanner_log_fetch_wait_max_time_ms: i32,
scanner_log_fetch_max_bytes_for_bucket: i32,
writer_batch_timeout_ms: i64,
writer_enable_idempotence: bool,
writer_max_inflight_requests_per_bucket: usize,
writer_buffer_memory_size: usize,
writer_buffer_wait_timeout_ms: u64,
connect_timeout_ms: u64,
security_protocol: String,
security_sasl_mechanism: String,
security_sasl_username: String,
security_sasl_password: String,
lookup_queue_size: usize,
lookup_max_batch_size: usize,
lookup_batch_timeout_ms: u64,
lookup_max_inflight_requests: usize,
lookup_max_retries: i32,
}
struct FfiResult {
error_code: i32,
error_message: String,
}
struct FfiTablePath {
database_name: String,
table_name: String,
}
struct FfiColumn {
name: String,
data_type: i32,
nullable: bool,
comment: String,
precision: i32,
scale: i32,
array_nesting: i32,
array_nullability: Vec<u8>,
element_data_type: i32,
element_precision: i32,
element_scale: i32,
}
struct FfiSchema {
columns: Vec<FfiColumn>,
primary_keys: Vec<String>,
}
struct FfiTableDescriptor {
schema: FfiSchema,
partition_keys: Vec<String>,
bucket_count: i32,
bucket_keys: Vec<String>,
properties: Vec<HashMapValue>,
custom_properties: Vec<HashMapValue>,
comment: String,
}
struct FfiTableInfo {
table_id: i64,
schema_id: i32,
table_path: FfiTablePath,
created_time: i64,
modified_time: i64,
primary_keys: Vec<String>,
bucket_keys: Vec<String>,
partition_keys: Vec<String>,
num_buckets: i32,
has_primary_key: bool,
is_partitioned: bool,
properties: Vec<HashMapValue>,
custom_properties: Vec<HashMapValue>,
comment: String,
schema: FfiSchema,
}
struct FfiTableInfoResult {
result: FfiResult,
table_info: FfiTableInfo,
}
// NOTE: FfiDatum, FfiGenericRow, FfiScanRecord, FfiScanRecords, FfiScanRecordsResult
// have been replaced by opaque types below (ScanResultInner, GenericRowInner, LookupResultInner).
struct FfiArrowRecordBatch {
array_ptr: usize,
schema_ptr: usize,
table_id: i64,
partition_id: i64,
bucket_id: i32,
base_offset: i64,
}
struct FfiArrowRecordBatches {
batches: Vec<FfiArrowRecordBatch>,
}
struct FfiArrowRecordBatchesResult {
result: FfiResult,
arrow_batches: FfiArrowRecordBatches,
}
struct FfiLakeSnapshot {
snapshot_id: i64,
bucket_offsets: Vec<FfiBucketOffset>,
}
struct FfiBucketOffset {
table_id: i64,
partition_id: i64,
bucket_id: i32,
offset: i64,
}
struct FfiOffsetQuery {
offset_type: i32,
timestamp: i64,
}
struct FfiBucketInfo {
table_id: i64,
bucket_id: i32,
has_partition_id: bool,
partition_id: i64,
record_count: usize,
}
struct FfiBucketSubscription {
bucket_id: i32,
offset: i64,
}
struct FfiPartitionBucketSubscription {
partition_id: i64,
bucket_id: i32,
offset: i64,
}
struct FfiBucketOffsetPair {
bucket_id: i32,
offset: i64,
}
struct FfiListOffsetsResult {
result: FfiResult,
bucket_offsets: Vec<FfiBucketOffsetPair>,
}
// NOTE: FfiLookupResult replaced by opaque LookupResultInner below.
struct FfiLakeSnapshotResult {
result: FfiResult,
lake_snapshot: FfiLakeSnapshot,
}
struct FfiPartitionKeyValue {
key: String,
value: String,
}
struct FfiPartitionInfo {
partition_id: i64,
partition_name: String,
}
struct FfiListPartitionInfosResult {
result: FfiResult,
partition_infos: Vec<FfiPartitionInfo>,
}
struct FfiDatabaseDescriptor {
comment: String,
properties: Vec<HashMapValue>,
}
struct FfiDatabaseInfo {
database_name: String,
comment: String,
properties: Vec<HashMapValue>,
created_time: i64,
modified_time: i64,
}
struct FfiDatabaseInfoResult {
result: FfiResult,
database_info: FfiDatabaseInfo,
}
struct FfiListDatabasesResult {
result: FfiResult,
database_names: Vec<String>,
}
struct FfiListTablesResult {
result: FfiResult,
table_names: Vec<String>,
}
struct FfiBoolResult {
result: FfiResult,
value: bool,
}
struct FfiServerNode {
node_id: i32,
host: String,
port: u32,
server_type: String,
uid: String,
}
struct FfiServerNodesResult {
result: FfiResult,
server_nodes: Vec<FfiServerNode>,
}
struct FfiPtrResult {
result: FfiResult,
ptr: usize,
}
extern "Rust" {
type Connection;
type Admin;
type Table;
type AppendWriter;
type WriteResult;
type LogScanner;
type BatchScanner;
type UpsertWriter;
type Lookuper;
type PrefixLookuper;
// Opaque types for optimized FFI
type ScanResultInner;
type GenericRowInner;
type LookupResultInner;
type PrefixLookupResultInner;
type ArrayWriterInner;
type MapWriterInner;
type ValueInner;
// Connection
fn new_connection(config: &FfiConfig) -> FfiPtrResult;
unsafe fn delete_connection(conn: *mut Connection);
fn get_admin(self: &Connection) -> FfiPtrResult;
fn get_table(self: &Connection, table_path: &FfiTablePath) -> FfiPtrResult;
// Admin
unsafe fn delete_admin(admin: *mut Admin);
fn create_table(
self: &Admin,
table_path: &FfiTablePath,
descriptor: &FfiTableDescriptor,
ignore_if_exists: bool,
) -> FfiResult;
// Create a table whose columns come from an Arrow schema (C Data
// Interface pointer), supporting nested MAP/ROW columns. `descriptor`
// supplies primary keys + table-level metadata; its `schema.columns`
// are ignored in favour of `arrow_schema_ptr`.
fn create_table_arrow(
self: &Admin,
table_path: &FfiTablePath,
descriptor: &FfiTableDescriptor,
arrow_schema_ptr: usize,
ignore_if_exists: bool,
) -> FfiResult;
fn drop_table(
self: &Admin,
table_path: &FfiTablePath,
ignore_if_not_exists: bool,
) -> FfiResult;
fn get_table_info(self: &Admin, table_path: &FfiTablePath) -> FfiTableInfoResult;
fn get_latest_lake_snapshot(
self: &Admin,
table_path: &FfiTablePath,
) -> FfiLakeSnapshotResult;
fn list_offsets(
self: &Admin,
table_path: &FfiTablePath,
bucket_ids: Vec<i32>,
offset_query: &FfiOffsetQuery,
) -> FfiListOffsetsResult;
fn list_partition_offsets(
self: &Admin,
table_path: &FfiTablePath,
partition_name: String,
bucket_ids: Vec<i32>,
offset_query: &FfiOffsetQuery,
) -> FfiListOffsetsResult;
fn list_partition_infos(
self: &Admin,
table_path: &FfiTablePath,
) -> FfiListPartitionInfosResult;
fn list_partition_infos_with_spec(
self: &Admin,
table_path: &FfiTablePath,
partition_spec: Vec<FfiPartitionKeyValue>,
) -> FfiListPartitionInfosResult;
fn create_partition(
self: &Admin,
table_path: &FfiTablePath,
partition_spec: Vec<FfiPartitionKeyValue>,
ignore_if_exists: bool,
) -> FfiResult;
fn drop_partition(
self: &Admin,
table_path: &FfiTablePath,
partition_spec: Vec<FfiPartitionKeyValue>,
ignore_if_not_exists: bool,
) -> FfiResult;
fn create_database(
self: &Admin,
database_name: &str,
descriptor: &FfiDatabaseDescriptor,
ignore_if_exists: bool,
) -> FfiResult;
fn drop_database(
self: &Admin,
database_name: &str,
ignore_if_not_exists: bool,
cascade: bool,
) -> FfiResult;
fn list_databases(self: &Admin) -> FfiListDatabasesResult;
fn database_exists(self: &Admin, database_name: &str) -> FfiBoolResult;
fn get_database_info(self: &Admin, database_name: &str) -> FfiDatabaseInfoResult;
fn list_tables(self: &Admin, database_name: &str) -> FfiListTablesResult;
fn table_exists(self: &Admin, table_path: &FfiTablePath) -> FfiBoolResult;
fn get_server_nodes(self: &Admin) -> FfiServerNodesResult;
// Table
unsafe fn delete_table(table: *mut Table);
fn new_append_writer(self: &Table) -> FfiPtrResult;
fn create_scanner(self: &Table, column_indices: Vec<usize>, batch: bool) -> FfiPtrResult;
fn create_bucket_batch_scanner(
self: &Table,
column_indices: Vec<usize>,
limit: i32,
table_id: i64,
partition_id: i64,
bucket_id: i32,
) -> FfiPtrResult;
fn get_table_info_from_table(self: &Table) -> FfiTableInfo;
fn get_table_path(self: &Table) -> FfiTablePath;
fn has_primary_key(self: &Table) -> bool;
fn create_upsert_writer(self: &Table, column_indices: Vec<usize>) -> FfiPtrResult;
fn new_lookuper(self: &Table) -> FfiPtrResult;
fn new_prefix_lookuper(self: &Table, lookup_column_names: Vec<String>) -> FfiPtrResult;
// GenericRowInner — opaque row for writes
fn new_generic_row(field_count: usize) -> Box<GenericRowInner>;
fn gr_reset(self: &mut GenericRowInner);
fn gr_set_null(self: &mut GenericRowInner, idx: usize);
fn gr_set_bool(self: &mut GenericRowInner, idx: usize, val: bool);
fn gr_set_i32(self: &mut GenericRowInner, idx: usize, val: i32);
fn gr_set_i64(self: &mut GenericRowInner, idx: usize, val: i64);
fn gr_set_f32(self: &mut GenericRowInner, idx: usize, val: f32);
fn gr_set_f64(self: &mut GenericRowInner, idx: usize, val: f64);
fn gr_set_str(self: &mut GenericRowInner, idx: usize, val: &str);
fn gr_set_bytes(self: &mut GenericRowInner, idx: usize, val: &[u8]);
fn gr_set_date(self: &mut GenericRowInner, idx: usize, days: i32);
fn gr_set_time(self: &mut GenericRowInner, idx: usize, millis: i32);
fn gr_set_ts_ntz(self: &mut GenericRowInner, idx: usize, millis: i64, nanos: i32);
fn gr_set_ts_ltz(self: &mut GenericRowInner, idx: usize, millis: i64, nanos: i32);
fn gr_set_decimal_str(self: &mut GenericRowInner, idx: usize, val: &str);
fn gr_set_array(
self: &mut GenericRowInner,
idx: usize,
writer: &mut ArrayWriterInner,
) -> Result<()>;
fn gr_set_map(
self: &mut GenericRowInner,
idx: usize,
writer: &mut MapWriterInner,
) -> Result<()>;
fn gr_set_row(
self: &mut GenericRowInner,
idx: usize,
row: &mut GenericRowInner,
) -> Result<()>;
// ArrayWriterInner — opaque array builder for writes
fn new_array_writer(
size: usize,
element_leaf_type_id: i32,
precision: u32,
scale: u32,
array_nesting: u32,
) -> Result<Box<ArrayWriterInner>>;
// ROW/MAP element types: a one-field Arrow schema (the flat encoding can't carry them).
fn new_array_writer_arrow(
size: usize,
element_schema_ptr: usize,
) -> Result<Box<ArrayWriterInner>>;
fn aw_size(self: &ArrayWriterInner) -> usize;
fn aw_set_null(self: &mut ArrayWriterInner, idx: usize) -> Result<()>;
fn aw_set_bool(self: &mut ArrayWriterInner, idx: usize, val: bool) -> Result<()>;
fn aw_set_i32(self: &mut ArrayWriterInner, idx: usize, val: i32) -> Result<()>;
fn aw_set_i64(self: &mut ArrayWriterInner, idx: usize, val: i64) -> Result<()>;
fn aw_set_f32(self: &mut ArrayWriterInner, idx: usize, val: f32) -> Result<()>;
fn aw_set_f64(self: &mut ArrayWriterInner, idx: usize, val: f64) -> Result<()>;
fn aw_set_str(self: &mut ArrayWriterInner, idx: usize, val: &str) -> Result<()>;
fn aw_set_bytes(self: &mut ArrayWriterInner, idx: usize, val: &[u8]) -> Result<()>;
fn aw_set_date(self: &mut ArrayWriterInner, idx: usize, days: i32) -> Result<()>;
fn aw_set_time(self: &mut ArrayWriterInner, idx: usize, millis: i32) -> Result<()>;
fn aw_set_ts_ntz(
self: &mut ArrayWriterInner,
idx: usize,
millis: i64,
nanos: i32,
) -> Result<()>;
fn aw_set_ts_ltz(
self: &mut ArrayWriterInner,
idx: usize,
millis: i64,
nanos: i32,
) -> Result<()>;
fn aw_set_decimal_str(self: &mut ArrayWriterInner, idx: usize, val: &str) -> Result<()>;
fn aw_set_array(
self: &mut ArrayWriterInner,
idx: usize,
nested: &mut ArrayWriterInner,
) -> Result<()>;
fn aw_set_row(
self: &mut ArrayWriterInner,
idx: usize,
row: &mut GenericRowInner,
) -> Result<()>;
fn aw_set_map(
self: &mut ArrayWriterInner,
idx: usize,
map: &mut MapWriterInner,
) -> Result<()>;
// MapWriterInner — opaque map builder for writes.
#[allow(clippy::too_many_arguments)]
fn new_map_writer(
capacity: usize,
key_leaf_type_id: i32,
key_precision: u32,
key_scale: u32,
value_leaf_type_id: i32,
value_precision: u32,
value_scale: u32,
value_array_nesting: u32,
) -> Result<Box<MapWriterInner>>;
// ROW/MAP key/value types: a [key, value] Arrow schema (the flat encoding can't).
fn new_map_writer_arrow(
capacity: usize,
kv_schema_ptr: usize,
) -> Result<Box<MapWriterInner>>;
fn mw_key_bool(self: &mut MapWriterInner, val: bool) -> Result<()>;
fn mw_key_i32(self: &mut MapWriterInner, val: i32) -> Result<()>;
fn mw_key_i64(self: &mut MapWriterInner, val: i64) -> Result<()>;
fn mw_key_f32(self: &mut MapWriterInner, val: f32) -> Result<()>;
fn mw_key_f64(self: &mut MapWriterInner, val: f64) -> Result<()>;
fn mw_key_str(self: &mut MapWriterInner, val: &str) -> Result<()>;
fn mw_key_bytes(self: &mut MapWriterInner, val: &[u8]) -> Result<()>;
fn mw_key_date(self: &mut MapWriterInner, days: i32) -> Result<()>;
fn mw_key_time(self: &mut MapWriterInner, millis: i32) -> Result<()>;
fn mw_key_timestamp(self: &mut MapWriterInner, millis: i64, nanos: i32) -> Result<()>;
fn mw_key_decimal_str(self: &mut MapWriterInner, val: &str) -> Result<()>;
fn mw_value_null(self: &mut MapWriterInner) -> Result<()>;
fn mw_value_bool(self: &mut MapWriterInner, val: bool) -> Result<()>;
fn mw_value_i32(self: &mut MapWriterInner, val: i32) -> Result<()>;
fn mw_value_i64(self: &mut MapWriterInner, val: i64) -> Result<()>;
fn mw_value_f32(self: &mut MapWriterInner, val: f32) -> Result<()>;
fn mw_value_f64(self: &mut MapWriterInner, val: f64) -> Result<()>;
fn mw_value_str(self: &mut MapWriterInner, val: &str) -> Result<()>;
fn mw_value_bytes(self: &mut MapWriterInner, val: &[u8]) -> Result<()>;
fn mw_value_date(self: &mut MapWriterInner, days: i32) -> Result<()>;
fn mw_value_time(self: &mut MapWriterInner, millis: i32) -> Result<()>;
fn mw_value_timestamp(self: &mut MapWriterInner, millis: i64, nanos: i32) -> Result<()>;
fn mw_value_decimal_str(self: &mut MapWriterInner, val: &str) -> Result<()>;
fn mw_value_row(self: &mut MapWriterInner, row: &mut GenericRowInner) -> Result<()>;
fn mw_value_map(self: &mut MapWriterInner, map: &mut MapWriterInner) -> Result<()>;
fn mw_value_array(self: &mut MapWriterInner, array: &mut ArrayWriterInner) -> Result<()>;
fn mw_commit(self: &mut MapWriterInner) -> Result<()>;
// AppendWriter
unsafe fn delete_append_writer(writer: *mut AppendWriter);
fn append(self: &mut AppendWriter, row: &GenericRowInner) -> FfiPtrResult;
fn append_arrow_batch(
self: &mut AppendWriter,
array_ptr: usize,
schema_ptr: usize,
) -> FfiPtrResult;
fn flush(self: &mut AppendWriter) -> FfiResult;
// WriteResult
unsafe fn delete_write_result(wr: *mut WriteResult);
fn wait(self: &mut WriteResult) -> FfiResult;
// UpsertWriter
unsafe fn delete_upsert_writer(writer: *mut UpsertWriter);
fn upsert(self: &mut UpsertWriter, row: &GenericRowInner) -> FfiPtrResult;
fn delete_row(self: &mut UpsertWriter, row: &GenericRowInner) -> FfiPtrResult;
fn upsert_flush(self: &mut UpsertWriter) -> FfiResult;
// Lookuper
unsafe fn delete_lookuper(lookuper: *mut Lookuper);
fn lookup(self: &mut Lookuper, pk_row: &GenericRowInner) -> Box<LookupResultInner>;
// LookupResultInner accessors
fn lv_has_error(self: &LookupResultInner) -> bool;
fn lv_error_code(self: &LookupResultInner) -> i32;
fn lv_error_message(self: &LookupResultInner) -> &str;
fn lv_found(self: &LookupResultInner) -> bool;
fn lv_field_count(self: &LookupResultInner) -> usize;
fn lv_column_name(self: &LookupResultInner, field: usize) -> Result<&str>;
fn lv_column_type(self: &LookupResultInner, field: usize) -> Result<i32>;
fn lv_is_null(self: &LookupResultInner, field: usize) -> Result<bool>;
fn lv_get_bool(self: &LookupResultInner, field: usize) -> Result<bool>;
fn lv_get_i32(self: &LookupResultInner, field: usize) -> Result<i32>;
fn lv_get_i64(self: &LookupResultInner, field: usize) -> Result<i64>;
fn lv_get_f32(self: &LookupResultInner, field: usize) -> Result<f32>;
fn lv_get_f64(self: &LookupResultInner, field: usize) -> Result<f64>;
fn lv_get_str(self: &LookupResultInner, field: usize) -> Result<&str>;
fn lv_get_bytes(self: &LookupResultInner, field: usize) -> Result<&[u8]>;
fn lv_get_date_days(self: &LookupResultInner, field: usize) -> Result<i32>;
fn lv_get_time_millis(self: &LookupResultInner, field: usize) -> Result<i32>;
fn lv_get_ts_millis(self: &LookupResultInner, field: usize) -> Result<i64>;
fn lv_get_ts_nanos(self: &LookupResultInner, field: usize) -> Result<i32>;
fn lv_is_ts_ltz(self: &LookupResultInner, field: usize) -> Result<bool>;
fn lv_get_decimal_str(self: &LookupResultInner, field: usize) -> Result<String>;
// ValueInner — recursive value handle for complex column reads.
fn lv_get_value(self: &LookupResultInner, field: usize) -> Result<Box<ValueInner>>;
fn sv_get_value(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<Box<ValueInner>>;
fn v_type(self: &ValueInner) -> i32;
fn v_is_null(self: &ValueInner) -> bool;
fn v_get_bool(self: &ValueInner) -> Result<bool>;
fn v_get_i32(self: &ValueInner) -> Result<i32>;
fn v_get_i64(self: &ValueInner) -> Result<i64>;
fn v_get_f32(self: &ValueInner) -> Result<f32>;
fn v_get_f64(self: &ValueInner) -> Result<f64>;
fn v_get_str(self: &ValueInner) -> Result<String>;
fn v_get_bytes(self: &ValueInner) -> Result<Vec<u8>>;
fn v_get_date_days(self: &ValueInner) -> Result<i32>;
fn v_get_time_millis(self: &ValueInner) -> Result<i32>;
fn v_get_ts_millis(self: &ValueInner) -> Result<i64>;
fn v_get_ts_nanos(self: &ValueInner) -> Result<i32>;
fn v_get_decimal_str(self: &ValueInner) -> Result<String>;
fn v_size(self: &ValueInner) -> Result<usize>;
fn v_field_count(self: &ValueInner) -> Result<usize>;
fn v_at(self: &ValueInner, i: usize) -> Result<Box<ValueInner>>;
fn v_key_at(self: &ValueInner, i: usize) -> Result<Box<ValueInner>>;
fn v_value_at(self: &ValueInner, i: usize) -> Result<Box<ValueInner>>;
fn v_field(self: &ValueInner, i: usize) -> Result<Box<ValueInner>>;
fn v_field_by_name(self: &ValueInner, name: &str) -> Result<Box<ValueInner>>;
// PrefixLookuper
unsafe fn delete_prefix_lookuper(lookuper: *mut PrefixLookuper);
fn prefix_lookup(
self: &mut PrefixLookuper,
prefix_row: &GenericRowInner,
) -> Box<PrefixLookupResultInner>;
// PrefixLookupResultInner accessors — like LookupResultInner but indexed
// by record, since a prefix lookup returns zero-or-more rows.
fn plv_has_error(self: &PrefixLookupResultInner) -> bool;
fn plv_error_code(self: &PrefixLookupResultInner) -> i32;
fn plv_error_message(self: &PrefixLookupResultInner) -> &str;
fn plv_row_count(self: &PrefixLookupResultInner) -> usize;
fn plv_field_count(self: &PrefixLookupResultInner) -> usize;
fn plv_column_name(self: &PrefixLookupResultInner, field: usize) -> Result<&str>;
fn plv_column_type(self: &PrefixLookupResultInner, field: usize) -> Result<i32>;
fn plv_is_null(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<bool>;
fn plv_get_bool(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<bool>;
fn plv_get_i32(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<i32>;
fn plv_get_i64(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<i64>;
fn plv_get_f32(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<f32>;
fn plv_get_f64(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<f64>;
fn plv_get_str(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<&str>;
fn plv_get_bytes(self: &PrefixLookupResultInner, rec: usize, field: usize)
-> Result<&[u8]>;
fn plv_get_date_days(
self: &PrefixLookupResultInner,
rec: usize,
field: usize,
) -> Result<i32>;
fn plv_get_time_millis(
self: &PrefixLookupResultInner,
rec: usize,
field: usize,
) -> Result<i32>;
fn plv_get_ts_millis(
self: &PrefixLookupResultInner,
rec: usize,
field: usize,
) -> Result<i64>;
fn plv_get_ts_nanos(
self: &PrefixLookupResultInner,
rec: usize,
field: usize,
) -> Result<i32>;
fn plv_is_ts_ltz(self: &PrefixLookupResultInner, rec: usize, field: usize) -> Result<bool>;
fn plv_get_decimal_str(
self: &PrefixLookupResultInner,
rec: usize,
field: usize,
) -> Result<String>;
fn plv_get_value(
self: &PrefixLookupResultInner,
rec: usize,
field: usize,
) -> Result<Box<ValueInner>>;
// LogScanner
unsafe fn delete_log_scanner(scanner: *mut LogScanner);
fn subscribe(self: &LogScanner, bucket_id: i32, start_offset: i64) -> FfiResult;
fn subscribe_buckets(
self: &LogScanner,
subscriptions: Vec<FfiBucketSubscription>,
) -> FfiResult;
fn subscribe_partition(
self: &LogScanner,
partition_id: i64,
bucket_id: i32,
start_offset: i64,
) -> FfiResult;
fn subscribe_partition_buckets(
self: &LogScanner,
subscriptions: Vec<FfiPartitionBucketSubscription>,
) -> FfiResult;
fn unsubscribe(self: &LogScanner, bucket_id: i32) -> FfiResult;
fn unsubscribe_partition(self: &LogScanner, partition_id: i64, bucket_id: i32)
-> FfiResult;
fn poll(self: &LogScanner, timeout_ms: i64) -> Box<ScanResultInner>;
fn poll_record_batch(self: &LogScanner, timeout_ms: i64) -> FfiArrowRecordBatchesResult;
fn free_arrow_ffi_structures(array_ptr: usize, schema_ptr: usize);
// BatchScanner
unsafe fn delete_batch_scanner(scanner: *mut BatchScanner);
fn next_batch(self: &BatchScanner) -> FfiArrowRecordBatchesResult;
fn collect_all_batches(self: &BatchScanner) -> FfiArrowRecordBatchesResult;
// ScanResultInner accessors
fn sv_has_error(self: &ScanResultInner) -> bool;
fn sv_error_code(self: &ScanResultInner) -> i32;
fn sv_error_message(self: &ScanResultInner) -> &str;
fn sv_record_count(self: &ScanResultInner) -> usize;
fn sv_column_count(self: &ScanResultInner) -> usize;
fn sv_column_name(self: &ScanResultInner, field: usize) -> Result<&str>;
fn sv_column_type(self: &ScanResultInner, field: usize) -> Result<i32>;
fn sv_offset(self: &ScanResultInner, bucket: usize, rec: usize) -> i64;
fn sv_timestamp(self: &ScanResultInner, bucket: usize, rec: usize) -> i64;
fn sv_change_type(self: &ScanResultInner, bucket: usize, rec: usize) -> i32;
fn sv_field_count(self: &ScanResultInner) -> usize;
fn sv_is_null(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<bool>;
fn sv_get_bool(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<bool>;
fn sv_get_i32(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<i32>;
fn sv_get_i64(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<i64>;
fn sv_get_f32(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<f32>;
fn sv_get_f64(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<f64>;
fn sv_get_str(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<&str>;
fn sv_get_bytes(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<&[u8]>;
fn sv_get_date_days(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<i32>;
fn sv_get_time_millis(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<i32>;
fn sv_get_ts_millis(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<i64>;
fn sv_get_ts_nanos(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<i32>;
fn sv_is_ts_ltz(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<bool>;
fn sv_get_decimal_str(
self: &ScanResultInner,
bucket: usize,
rec: usize,
field: usize,
) -> Result<String>;
fn sv_bucket_infos(self: &ScanResultInner) -> &Vec<FfiBucketInfo>;
}
}
pub struct Connection {
inner: Arc<fcore::client::FlussConnection>,
}
pub struct Admin {
inner: Arc<fcore::client::FlussAdmin>,
}
pub struct Table {
connection: Arc<fcore::client::FlussConnection>,
metadata: Arc<fcore::client::Metadata>,
table_info: fcore::metadata::TableInfo,
table_path: fcore::metadata::TablePath,
has_pk: bool,
}
pub struct AppendWriter {
inner: fcore::client::AppendWriter,
table_info: fcore::metadata::TableInfo,
}
pub struct WriteResult {
inner: Option<fcore::client::WriteResultFuture>,
}
enum ScannerKind {
Record(fcore::client::LogScanner),
Batch(fcore::client::RecordBatchLogScanner),
}
pub struct LogScanner {
scanner: ScannerKind,
/// Fluss columns matching the projected Arrow fields (1:1 by index).
/// For non-projected scanners this is the full table schema columns.
projected_columns: Vec<fcore::metadata::Column>,
}
pub struct BatchScanner {
inner: Mutex<fcore::client::LimitBatchScanner>,
}
pub struct UpsertWriter {
inner: fcore::client::UpsertWriter,
table_info: fcore::metadata::TableInfo,
}
pub struct Lookuper {
inner: fcore::client::Lookuper,
table_info: fcore::metadata::TableInfo,
}
pub struct PrefixLookuper {
inner: PrefixKeyLookuper,
table_info: TableInfo,
/// Full-schema indices of the lookup columns, used to compact the input row.
lookup_column_indices: Vec<usize>,
}
/// Error code for client-side errors that did not originate from the server API protocol.
/// Must be non-zero so that CPP `Result::Ok()` (which checks `error_code == 0`) correctly
/// detects client-side errors as failures. The value -2 is outside the server API error
/// code range (-1 .. 57+), so it will never collide with current or future API codes.
const CLIENT_ERROR_CODE: i32 = -2;
fn ok_result() -> ffi::FfiResult {
ffi::FfiResult {
error_code: 0,
error_message: String::new(),
}
}
fn err_result(code: i32, msg: String) -> ffi::FfiResult {
ffi::FfiResult {
error_code: code,
error_message: msg,
}
}
/// Create a client-side error result (not from server API).
fn client_err(msg: String) -> ffi::FfiResult {
err_result(CLIENT_ERROR_CODE, msg)
}
fn err_from_core_error(e: &Error) -> ffi::FfiResult {
// Transport failures map to `NetworkException` (Java parity,
// retriable).
match e {
Error::FlussAPIError { api_error } => err_result(api_error.code, api_error.message.clone()),
Error::RpcError { .. } => {
err_result(CoreFlussError::NetworkException.code(), e.to_string())
}
_ => client_err(e.to_string()),
}
}
fn ok_ptr(ptr: usize) -> ffi::FfiPtrResult {
ffi::FfiPtrResult {
result: ok_result(),
ptr,
}
}
fn client_err_ptr(msg: String) -> ffi::FfiPtrResult {
ffi::FfiPtrResult {
result: client_err(msg),
ptr: 0usize,
}
}
fn err_ptr_from_core(e: &fcore::error::Error) -> ffi::FfiPtrResult {
ffi::FfiPtrResult {
result: err_from_core_error(e),
ptr: 0usize,
}
}
fn empty_arrow_batches_result(result: ffi::FfiResult) -> ffi::FfiArrowRecordBatchesResult {
ffi::FfiArrowRecordBatchesResult {
result,
arrow_batches: ffi::FfiArrowRecordBatches { batches: vec![] },
}
}
/// Wrap the result of `core_scan_batches_to_ffi` in an `FfiArrowRecordBatchesResult`.
fn arrow_batches_result(
converted: Result<ffi::FfiArrowRecordBatches, String>,
) -> ffi::FfiArrowRecordBatchesResult {
match converted {
Ok(arrow_batches) => ffi::FfiArrowRecordBatchesResult {
result: ok_result(),
arrow_batches,
},
Err(e) => empty_arrow_batches_result(client_err(e)),
}
}
// Connection implementation
fn new_connection(config: &ffi::FfiConfig) -> ffi::FfiPtrResult {
let assigner_type = match config
.writer_bucket_no_key_assigner
.parse::<fluss::config::NoKeyAssigner>()
{
Ok(v) => v,
Err(e) => return client_err_ptr(format!("Invalid bucket assigner type: {e}")),
};
let config_core = fluss::config::Config {
bootstrap_servers: config.bootstrap_servers.to_string(),
writer_request_max_size: config.writer_request_max_size,
writer_acks: config.writer_acks.to_string(),
writer_retries: config.writer_retries,
writer_batch_size: config.writer_batch_size,
writer_dynamic_batch_size_enabled: config.writer_dynamic_batch_size_enabled,
writer_dynamic_batch_size_min: config.writer_dynamic_batch_size_min,
writer_batch_timeout_ms: config.writer_batch_timeout_ms,
writer_bucket_no_key_assigner: assigner_type,
scanner_remote_log_prefetch_num: config.scanner_remote_log_prefetch_num,
remote_file_download_thread_num: config.remote_file_download_thread_num,
scanner_remote_log_read_concurrency: config.scanner_remote_log_read_concurrency,
scanner_log_max_poll_records: config.scanner_log_max_poll_records,
scanner_log_fetch_max_bytes: config.scanner_log_fetch_max_bytes,
scanner_log_fetch_min_bytes: config.scanner_log_fetch_min_bytes,
scanner_log_fetch_wait_max_time_ms: config.scanner_log_fetch_wait_max_time_ms,
scanner_log_fetch_max_bytes_for_bucket: config.scanner_log_fetch_max_bytes_for_bucket,
writer_enable_idempotence: config.writer_enable_idempotence,
writer_max_inflight_requests_per_bucket: config.writer_max_inflight_requests_per_bucket,
writer_buffer_memory_size: config.writer_buffer_memory_size,
writer_buffer_wait_timeout_ms: config.writer_buffer_wait_timeout_ms,
connect_timeout_ms: config.connect_timeout_ms,
security_protocol: config.security_protocol.to_string(),
security_sasl_mechanism: config.security_sasl_mechanism.to_string(),
security_sasl_username: config.security_sasl_username.to_string(),
security_sasl_password: config.security_sasl_password.to_string(),
lookup_queue_size: config.lookup_queue_size,
lookup_max_batch_size: config.lookup_max_batch_size,
lookup_batch_timeout_ms: config.lookup_batch_timeout_ms,
lookup_max_inflight_requests: config.lookup_max_inflight_requests,
lookup_max_retries: config.lookup_max_retries,
};
let conn = RUNTIME.block_on(async { fcore::client::FlussConnection::new(config_core).await });
match conn {
Ok(c) => {
let ptr = Box::into_raw(Box::new(Connection { inner: Arc::new(c) }));
ok_ptr(ptr as usize)
}
Err(e) => err_ptr_from_core(&e),
}
}
unsafe fn delete_connection(conn: *mut Connection) {
if !conn.is_null() {
unsafe {
drop(Box::from_raw(conn));
}
}
}
impl Connection {
fn get_admin(&self) -> ffi::FfiPtrResult {
let admin_result = self.inner.get_admin();
match admin_result {
Ok(admin) => {
let ptr = Box::into_raw(Box::new(Admin { inner: admin }));
ok_ptr(ptr as usize)
}
Err(e) => err_ptr_from_core(&e),
}
}
fn get_table(&self, table_path: &ffi::FfiTablePath) -> ffi::FfiPtrResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let table_result = RUNTIME.block_on(async { self.inner.get_table(&path).await });
match table_result {
Ok(t) => {
let ptr = Box::into_raw(Box::new(Table {
connection: self.inner.clone(),
metadata: t.metadata().clone(),
table_info: t.get_table_info().clone(),
table_path: t.table_path().clone(),
has_pk: t.has_primary_key(),
}));
ok_ptr(ptr as usize)
}
Err(e) => err_ptr_from_core(&e),
}
}
}
// Admin implementation
unsafe fn delete_admin(admin: *mut Admin) {
if !admin.is_null() {
unsafe {
drop(Box::from_raw(admin));
}
}
}
impl Admin {
fn create_table(
&self,
table_path: &ffi::FfiTablePath,
descriptor: &ffi::FfiTableDescriptor,
ignore_if_exists: bool,
) -> ffi::FfiResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let core_descriptor = match types::ffi_descriptor_to_core(descriptor) {
Ok(d) => d,
Err(e) => return client_err(e.to_string()),
};
let result = RUNTIME.block_on(async {
self.inner
.create_table(&path, &core_descriptor, ignore_if_exists)
.await
});
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
fn create_table_arrow(
&self,
table_path: &ffi::FfiTablePath,
descriptor: &ffi::FfiTableDescriptor,
arrow_schema_ptr: usize,
ignore_if_exists: bool,
) -> ffi::FfiResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
// Safety: C++ exports the schema via `arrow::ExportSchema` into a heap
// `FFI_ArrowSchema` whose pointer is passed here; ownership transfers.
let core_descriptor =
match unsafe { types::arrow_ffi_to_core_descriptor(arrow_schema_ptr, descriptor) } {
Ok(d) => d,
Err(e) => return client_err(e.to_string()),
};
let result = RUNTIME.block_on(async {
self.inner
.create_table(&path, &core_descriptor, ignore_if_exists)
.await
});
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
fn drop_table(
&self,
table_path: &ffi::FfiTablePath,
ignore_if_not_exists: bool,
) -> ffi::FfiResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let result =
RUNTIME.block_on(async { self.inner.drop_table(&path, ignore_if_not_exists).await });
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
fn get_table_info(&self, table_path: &ffi::FfiTablePath) -> ffi::FfiTableInfoResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let result = RUNTIME.block_on(async { self.inner.get_table_info(&path).await });
match result {
Ok(info) => ffi::FfiTableInfoResult {
result: ok_result(),
table_info: types::core_table_info_to_ffi(&info),
},
Err(e) => ffi::FfiTableInfoResult {
result: err_from_core_error(&e),
table_info: types::empty_table_info(),
},
}
}
fn get_latest_lake_snapshot(
&self,
table_path: &ffi::FfiTablePath,
) -> ffi::FfiLakeSnapshotResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let result = RUNTIME.block_on(async { self.inner.get_latest_lake_snapshot(&path).await });
match result {
Ok(snapshot) => ffi::FfiLakeSnapshotResult {
result: ok_result(),
lake_snapshot: types::core_lake_snapshot_to_ffi(&snapshot),
},
Err(e) => ffi::FfiLakeSnapshotResult {
result: err_from_core_error(&e),
lake_snapshot: ffi::FfiLakeSnapshot {
snapshot_id: -1,
bucket_offsets: vec![],
},
},
}
}
// Helper function for common list offsets functionality
fn do_list_offsets(
&self,
table_path: &ffi::FfiTablePath,
partition_name: Option<&str>,
bucket_ids: Vec<i32>,
offset_query: &ffi::FfiOffsetQuery,
) -> ffi::FfiListOffsetsResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let offset_spec = match offset_query.offset_type {
0 => OffsetSpec::Earliest,
1 => OffsetSpec::Latest,
2 => OffsetSpec::Timestamp(offset_query.timestamp),
_ => {
return ffi::FfiListOffsetsResult {
result: client_err(format!(
"Invalid offset_type: {}",
offset_query.offset_type
)),
bucket_offsets: vec![],
};
}
};
let result = RUNTIME.block_on(async {
if let Some(part_name) = partition_name {
self.inner
.list_partition_offsets(&path, part_name, &bucket_ids, offset_spec)
.await
} else {
self.inner
.list_offsets(&path, &bucket_ids, offset_spec)
.await
}
});
match result {
Ok(offsets) => {
let bucket_offsets: Vec<ffi::FfiBucketOffsetPair> = offsets
.into_iter()
.map(|(bucket_id, offset)| ffi::FfiBucketOffsetPair { bucket_id, offset })
.collect();
ffi::FfiListOffsetsResult {
result: ok_result(),
bucket_offsets,
}
}
Err(e) => ffi::FfiListOffsetsResult {
result: err_from_core_error(&e),
bucket_offsets: vec![],
},
}
}
fn list_offsets(
&self,
table_path: &ffi::FfiTablePath,
bucket_ids: Vec<i32>,
offset_query: &ffi::FfiOffsetQuery,
) -> ffi::FfiListOffsetsResult {
self.do_list_offsets(table_path, None, bucket_ids, offset_query)
}
fn list_partition_offsets(
&self,
table_path: &ffi::FfiTablePath,
partition_name: String,
bucket_ids: Vec<i32>,
offset_query: &ffi::FfiOffsetQuery,
) -> ffi::FfiListOffsetsResult {
self.do_list_offsets(table_path, Some(&partition_name), bucket_ids, offset_query)
}
fn list_partition_infos(
&self,
table_path: &ffi::FfiTablePath,
) -> ffi::FfiListPartitionInfosResult {
self.do_list_partition_infos(table_path, None)
}
fn list_partition_infos_with_spec(
&self,
table_path: &ffi::FfiTablePath,
partition_spec: Vec<ffi::FfiPartitionKeyValue>,
) -> ffi::FfiListPartitionInfosResult {
let spec_map: std::collections::HashMap<String, String> = partition_spec
.into_iter()
.map(|kv| (kv.key, kv.value))
.collect();
let spec = fcore::metadata::PartitionSpec::new(spec_map);
self.do_list_partition_infos(table_path, Some(&spec))
}
fn create_partition(
&self,
table_path: &ffi::FfiTablePath,
partition_spec: Vec<ffi::FfiPartitionKeyValue>,
ignore_if_exists: bool,
) -> ffi::FfiResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let spec_map: std::collections::HashMap<String, String> = partition_spec
.into_iter()
.map(|kv| (kv.key, kv.value))
.collect();
let partition_spec = fcore::metadata::PartitionSpec::new(spec_map);
let result = RUNTIME.block_on(async {
self.inner
.create_partition(&path, &partition_spec, ignore_if_exists)
.await
});
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
fn drop_partition(
&self,
table_path: &ffi::FfiTablePath,
partition_spec: Vec<ffi::FfiPartitionKeyValue>,
ignore_if_not_exists: bool,
) -> ffi::FfiResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let spec_map: std::collections::HashMap<String, String> = partition_spec
.into_iter()
.map(|kv| (kv.key, kv.value))
.collect();
let partition_spec = fcore::metadata::PartitionSpec::new(spec_map);
let result = RUNTIME.block_on(async {
self.inner
.drop_partition(&path, &partition_spec, ignore_if_not_exists)
.await
});
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
fn create_database(
&self,
database_name: &str,
descriptor: &ffi::FfiDatabaseDescriptor,
ignore_if_exists: bool,
) -> ffi::FfiResult {
let descriptor_opt = types::ffi_database_descriptor_to_core(descriptor);
let result = RUNTIME.block_on(async {
self.inner
.create_database(database_name, descriptor_opt.as_ref(), ignore_if_exists)
.await
});
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
fn drop_database(
&self,
database_name: &str,
ignore_if_not_exists: bool,
cascade: bool,
) -> ffi::FfiResult {
let result = RUNTIME.block_on(async {
self.inner
.drop_database(database_name, ignore_if_not_exists, cascade)
.await
});
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
fn list_databases(&self) -> ffi::FfiListDatabasesResult {
let result = RUNTIME.block_on(async { self.inner.list_databases().await });
match result {
Ok(names) => ffi::FfiListDatabasesResult {
result: ok_result(),
database_names: names,
},
Err(e) => ffi::FfiListDatabasesResult {
result: err_from_core_error(&e),
database_names: vec![],
},
}
}
fn database_exists(&self, database_name: &str) -> ffi::FfiBoolResult {
let result = RUNTIME.block_on(async { self.inner.database_exists(database_name).await });
match result {
Ok(exists) => ffi::FfiBoolResult {
result: ok_result(),
value: exists,
},
Err(e) => ffi::FfiBoolResult {
result: err_from_core_error(&e),
value: false,
},
}
}
fn get_database_info(&self, database_name: &str) -> ffi::FfiDatabaseInfoResult {
let result = RUNTIME.block_on(async { self.inner.get_database_info(database_name).await });
match result {
Ok(info) => ffi::FfiDatabaseInfoResult {
result: ok_result(),
database_info: types::core_database_info_to_ffi(&info),
},
Err(e) => ffi::FfiDatabaseInfoResult {
result: err_from_core_error(&e),
database_info: ffi::FfiDatabaseInfo {
database_name: String::new(),
comment: String::new(),
properties: vec![],
created_time: 0,
modified_time: 0,
},
},
}
}
fn list_tables(&self, database_name: &str) -> ffi::FfiListTablesResult {
let result = RUNTIME.block_on(async { self.inner.list_tables(database_name).await });
match result {
Ok(names) => ffi::FfiListTablesResult {
result: ok_result(),
table_names: names,
},
Err(e) => ffi::FfiListTablesResult {
result: err_from_core_error(&e),
table_names: vec![],
},
}
}
fn table_exists(&self, table_path: &ffi::FfiTablePath) -> ffi::FfiBoolResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let result = RUNTIME.block_on(async { self.inner.table_exists(&path).await });
match result {
Ok(exists) => ffi::FfiBoolResult {
result: ok_result(),
value: exists,
},
Err(e) => ffi::FfiBoolResult {
result: err_from_core_error(&e),
value: false,
},
}
}
fn do_list_partition_infos(
&self,
table_path: &ffi::FfiTablePath,
partial_partition_spec: Option<&fcore::metadata::PartitionSpec>,
) -> ffi::FfiListPartitionInfosResult {
let path = fcore::metadata::TablePath::new(
table_path.database_name.clone(),
table_path.table_name.clone(),
);
let result = RUNTIME.block_on(async {
self.inner
.list_partition_infos_with_spec(&path, partial_partition_spec)
.await
});
match result {
Ok(infos) => {
let partition_infos: Vec<ffi::FfiPartitionInfo> = infos
.into_iter()
.map(|info| ffi::FfiPartitionInfo {
partition_id: info.get_partition_id(),
partition_name: info.get_partition_name(),
})
.collect();
ffi::FfiListPartitionInfosResult {
result: ok_result(),
partition_infos,
}
}
Err(e) => ffi::FfiListPartitionInfosResult {
result: err_from_core_error(&e),
partition_infos: vec![],
},
}
}
fn get_server_nodes(&self) -> ffi::FfiServerNodesResult {
let result = RUNTIME.block_on(async { self.inner.get_server_nodes().await });
match result {
Ok(nodes) => {
let server_nodes: Vec<ffi::FfiServerNode> = nodes
.into_iter()
.map(|node| ffi::FfiServerNode {
node_id: node.id(),
host: node.host().to_string(),
port: node.port(),
server_type: node.server_type().to_string(),
uid: node.uid().to_string(),
})
.collect();
ffi::FfiServerNodesResult {
result: ok_result(),
server_nodes,
}
}
Err(e) => ffi::FfiServerNodesResult {
result: err_from_core_error(&e),
server_nodes: vec![],
},
}
}
}
// Table implementation
unsafe fn delete_table(table: *mut Table) {
if !table.is_null() {
unsafe {
drop(Box::from_raw(table));
}
}
}
impl Table {
fn fluss_table(&self) -> fcore::client::FlussTable<'_> {
fcore::client::FlussTable::new(
&self.connection,
self.metadata.clone(),
self.table_info.clone(),
)
}
fn resolve_projected_columns(
&self,
indices: &[usize],
) -> Result<Vec<fcore::metadata::Column>, String> {
let all_columns = self.table_info.get_schema().columns();
indices
.iter()
.map(|&i| {
all_columns.get(i).cloned().ok_or_else(|| {
format!(
"Invalid column index {i}: schema has {} columns",
all_columns.len()
)
})
})
.collect()
}
fn new_append_writer(&self) -> ffi::FfiPtrResult {
let _enter = RUNTIME.enter();
let table_append = match self.fluss_table().new_append() {
Ok(a) => a,
Err(e) => return err_ptr_from_core(&e),
};
let writer = match table_append.create_writer() {
Ok(w) => w,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(AppendWriter {
inner: writer,
table_info: self.table_info.clone(),
}));
ok_ptr(ptr as usize)
}
fn create_scanner(&self, column_indices: Vec<usize>, batch: bool) -> ffi::FfiPtrResult {
RUNTIME.block_on(async {
let fluss_table = self.fluss_table();
let scan = fluss_table.new_scan();
let (projected_columns, scan) = if column_indices.is_empty() {
(self.table_info.get_schema().columns().to_vec(), scan)
} else {
let cols = match self.resolve_projected_columns(&column_indices) {
Ok(c) => c,
Err(e) => return client_err_ptr(e),
};
let scan = match scan.project(&column_indices) {
Ok(s) => s,
Err(e) => return err_ptr_from_core(&e),
};
(cols, scan)
};
let scanner = if batch {
match scan.create_record_batch_log_scanner() {
Ok(s) => ScannerKind::Batch(s),
Err(e) => return err_ptr_from_core(&e),
}
} else {
match scan.create_log_scanner() {
Ok(s) => ScannerKind::Record(s),
Err(e) => return err_ptr_from_core(&e),
}
};
let ptr = Box::into_raw(Box::new(LogScanner {
scanner,
projected_columns,
}));
ok_ptr(ptr as usize)
})
}
fn create_bucket_batch_scanner(
&self,
column_indices: Vec<usize>,
limit: i32,
table_id: i64,
partition_id: i64,
bucket_id: i32,
) -> ffi::FfiPtrResult {
RUNTIME.block_on(async {
let fluss_table = self.fluss_table();
let scan = fluss_table.new_scan();
let scan = if column_indices.is_empty() {
scan
} else {
match scan.project(&column_indices) {
Ok(s) => s,
Err(e) => return err_ptr_from_core(&e),
}
};
let scan = match scan.limit(limit) {
Ok(s) => s,
Err(e) => return err_ptr_from_core(&e),
};
// A negative partition id encodes "no partition" across the FFI boundary.
let partition = (partition_id >= 0).then_some(partition_id);
let bucket =
fcore::metadata::TableBucket::new_with_partition(table_id, partition, bucket_id);
let scanner = match scan.create_bucket_batch_scanner(bucket) {
Ok(s) => s,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(BatchScanner {
inner: Mutex::new(scanner),
}));
ok_ptr(ptr as usize)
})
}
fn get_table_info_from_table(&self) -> ffi::FfiTableInfo {
types::core_table_info_to_ffi(&self.table_info)
}
fn get_table_path(&self) -> ffi::FfiTablePath {
ffi::FfiTablePath {
database_name: self.table_path.database().to_string(),
table_name: self.table_path.table().to_string(),
}
}
fn has_primary_key(&self) -> bool {
self.has_pk
}
fn create_upsert_writer(&self, column_indices: Vec<usize>) -> ffi::FfiPtrResult {
let _enter = RUNTIME.enter();
let table_upsert = match self.fluss_table().new_upsert() {
Ok(u) => u,
Err(e) => return err_ptr_from_core(&e),
};
let table_upsert = if column_indices.is_empty() {
table_upsert
} else {
match table_upsert.partial_update(Some(column_indices)) {
Ok(u) => u,
Err(e) => return err_ptr_from_core(&e),
}
};
let writer = match table_upsert.create_writer() {
Ok(w) => w,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(UpsertWriter {
inner: writer,
table_info: self.table_info.clone(),
}));
ok_ptr(ptr as usize)
}
fn new_lookuper(&self) -> ffi::FfiPtrResult {
let _enter = RUNTIME.enter();
let table_lookup = match self.fluss_table().new_lookup() {
Ok(l) => l,
Err(e) => return err_ptr_from_core(&e),
};
let lookuper = match table_lookup.create_lookuper() {
Ok(l) => l,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(Lookuper {
inner: lookuper,
table_info: self.table_info.clone(),
}));
ok_ptr(ptr as usize)
}
fn new_prefix_lookuper(&self, lookup_column_names: Vec<String>) -> ffi::FfiPtrResult {
let _enter = RUNTIME.enter();
let table_lookup = match self.fluss_table().new_lookup() {
Ok(l) => l,
Err(e) => return err_ptr_from_core(&e),
};
// `create_lookuper` validates the column names (rejecting unknown ones),
// so by the time it succeeds every name resolves to an index.
let row_type = self.table_info.row_type();
let lookup_column_indices: Vec<usize> = lookup_column_names
.iter()
.filter_map(|name| row_type.get_field_index(name))
.collect();
let lookuper = match table_lookup
.lookup_by(lookup_column_names)
.create_lookuper()
{
Ok(l) => l,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(PrefixLookuper {
inner: lookuper,
table_info: self.table_info.clone(),
lookup_column_indices,
}));
ok_ptr(ptr as usize)
}
}
// AppendWriter implementation
unsafe fn delete_append_writer(writer: *mut AppendWriter) {
if !writer.is_null() {
unsafe {
drop(Box::from_raw(writer));
}
}
}
impl AppendWriter {
fn append(&mut self, row: &GenericRowInner) -> ffi::FfiPtrResult {
let schema = self.table_info.get_schema();
let generic_row = match types::resolve_row_types(&row.row, Some(schema)) {
Ok(r) => r,
Err(e) => return client_err_ptr(e.to_string()),
};
let result_future = match self.inner.append(&generic_row) {
Ok(f) => f,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(WriteResult {
inner: Some(result_future),
}));
ok_ptr(ptr as usize)
}
fn append_arrow_batch(&mut self, array_ptr: usize, schema_ptr: usize) -> ffi::FfiPtrResult {
// Safety: C++ allocates these via `new ArrowArray/ArrowSchema` after a
// successful `ExportRecordBatch`, so both pointers are valid heap
// allocations that we take ownership of here.
let ffi_array = unsafe { *Box::from_raw(array_ptr as *mut FFI_ArrowArray) };
let ffi_schema = unsafe { Box::from_raw(schema_ptr as *mut FFI_ArrowSchema) };
// Safety: `from_ffi` requires that the array and schema conform to the
// Arrow C Data Interface, which is guaranteed by C++'s ExportRecordBatch.
let array_data = match unsafe { arrow::ffi::from_ffi(ffi_array, &ffi_schema) } {
Ok(d) => d,
Err(e) => return client_err_ptr(format!("Failed to import Arrow batch: {e}")),
};
// ffi_array is consumed by from_ffi; ffi_schema is dropped here (Box goes out of scope)
// Reconstruct RecordBatch from the imported StructArray data
let struct_array = arrow::array::StructArray::from(array_data);
let batch = arrow::record_batch::RecordBatch::from(struct_array);
let result_future = match self.inner.append_arrow_batch(batch) {
Ok(f) => f,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(WriteResult {
inner: Some(result_future),
}));
ok_ptr(ptr as usize)
}
fn flush(&mut self) -> ffi::FfiResult {
let result = RUNTIME.block_on(async { self.inner.flush().await });
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
}
unsafe fn delete_write_result(wr: *mut WriteResult) {
if !wr.is_null() {
unsafe {
drop(Box::from_raw(wr));
}
}
}
impl WriteResult {
fn wait(&mut self) -> ffi::FfiResult {
if let Some(future) = self.inner.take() {
let result = RUNTIME.block_on(future);
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
} else {
client_err("WriteResult already consumed".to_string())
}
}
}
// UpsertWriter implementation
unsafe fn delete_upsert_writer(writer: *mut UpsertWriter) {
if !writer.is_null() {
unsafe {
drop(Box::from_raw(writer));
}
}
}
impl UpsertWriter {
/// Pad row with Null to full schema width.
/// This allows callers to only set the fields they care about.
fn pad_row<'a>(&self, mut row: fcore::row::GenericRow<'a>) -> fcore::row::GenericRow<'a> {
let num_columns = self.table_info.get_schema().columns().len();
if row.values.len() < num_columns {
row.values.resize(num_columns, fcore::row::Datum::Null);
}
row
}
fn upsert(&mut self, row: &GenericRowInner) -> ffi::FfiPtrResult {
let schema = self.table_info.get_schema();
let generic_row = match types::resolve_row_types(&row.row, Some(schema)) {
Ok(r) => r,
Err(e) => return client_err_ptr(e.to_string()),
};
let generic_row = self.pad_row(generic_row);
let result_future = match self.inner.upsert(&generic_row) {
Ok(f) => f,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(WriteResult {
inner: Some(result_future),
}));
ok_ptr(ptr as usize)
}
fn delete_row(&mut self, row: &GenericRowInner) -> ffi::FfiPtrResult {
let schema = self.table_info.get_schema();
let generic_row = match types::resolve_row_types(&row.row, Some(schema)) {
Ok(r) => r,
Err(e) => return client_err_ptr(e.to_string()),
};
let generic_row = self.pad_row(generic_row);
let result_future = match self.inner.delete(&generic_row) {
Ok(f) => f,
Err(e) => return err_ptr_from_core(&e),
};
let ptr = Box::into_raw(Box::new(WriteResult {
inner: Some(result_future),
}));
ok_ptr(ptr as usize)
}
fn upsert_flush(&mut self) -> ffi::FfiResult {
let result = RUNTIME.block_on(async { self.inner.flush().await });
match result {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
}
}
// Lookuper implementation
unsafe fn delete_lookuper(lookuper: *mut Lookuper) {
if !lookuper.is_null() {
unsafe {
drop(Box::from_raw(lookuper));
}
}
}
impl Lookuper {
/// Build a dense PK-only row from a (possibly sparse) input row.
/// The user may set PK values at their full schema positions (e.g. [0, 2])
/// via name-based Set(). We compact them into [0, 1, …] to match
/// the lookup_row_type the core KeyEncoder expects.
fn dense_pk_row<'a>(&self, mut row: fcore::row::GenericRow<'a>) -> fcore::row::GenericRow<'a> {
let pk_indices = self.table_info.get_schema().primary_key_indexes();
let mut dense = fcore::row::GenericRow::new(pk_indices.len());
for (dense_idx, &schema_idx) in pk_indices.iter().enumerate() {
if schema_idx < row.values.len() {
dense.values[dense_idx] =
std::mem::replace(&mut row.values[schema_idx], fcore::row::Datum::Null);
}
}
dense
}
fn lookup(&mut self, pk_row: &GenericRowInner) -> Box<LookupResultInner> {
let schema = self.table_info.get_schema();
let generic_row = match types::resolve_row_types(&pk_row.row, Some(schema)) {
Ok(r) => self.dense_pk_row(r),
Err(e) => {
return Box::new(LookupResultInner::from_error(
CLIENT_ERROR_CODE,
e.to_string(),
));
}
};
let lookup_result = match RUNTIME.block_on(self.inner.lookup(&generic_row)) {
Ok(r) => r,
Err(e) => {
let ffi_err = err_from_core_error(&e);
return Box::new(LookupResultInner::from_error(
ffi_err.error_code,
ffi_err.error_message,
));
}
};
let columns = self.table_info.get_schema().columns().to_vec();
match lookup_result.get_single_row() {
Ok(Some(row)) => match types::compacted_row_to_owned(&row, &self.table_info) {
Ok(owned_row) => Box::new(LookupResultInner {
error: None,
found: true,
row: Some(owned_row),
columns,
}),
Err(e) => Box::new(LookupResultInner::from_error(
CLIENT_ERROR_CODE,
e.to_string(),
)),
},
Ok(None) => Box::new(LookupResultInner {
error: None,
found: false,
row: None,
columns,
}),
Err(e) => {
let ffi_err = err_from_core_error(&e);
Box::new(LookupResultInner::from_error(
ffi_err.error_code,
ffi_err.error_message,
))
}
}
}
}
// PrefixLookuper implementation
unsafe fn delete_prefix_lookuper(lookuper: *mut PrefixLookuper) {
if !lookuper.is_null() {
unsafe {
drop(Box::from_raw(lookuper));
}
}
}
impl PrefixLookuper {
/// Compact a sparse input row (prefix columns set at their schema positions)
/// into the dense, lookup-column-ordered row the core prefix encoder expects.
fn dense_prefix_row<'a>(&self, mut row: GenericRow<'a>) -> GenericRow<'a> {
let mut dense = GenericRow::new(self.lookup_column_indices.len());
for (dense_idx, &schema_idx) in self.lookup_column_indices.iter().enumerate() {
if schema_idx < row.values.len() {
dense.values[dense_idx] =
std::mem::replace(&mut row.values[schema_idx], Datum::Null);
}
}
dense
}
fn prefix_lookup(&mut self, prefix_row: &GenericRowInner) -> Box<PrefixLookupResultInner> {
let schema = self.table_info.get_schema();
let generic_row = match types::resolve_row_types(&prefix_row.row, Some(schema)) {
Ok(r) => self.dense_prefix_row(r),
Err(e) => {
return Box::new(PrefixLookupResultInner::from_error(
CLIENT_ERROR_CODE,
e.to_string(),
));
}
};
let lookup_result = match RUNTIME.block_on(self.inner.lookup(&generic_row)) {
Ok(r) => r,
Err(e) => {
let ffi_err = err_from_core_error(&e);
return Box::new(PrefixLookupResultInner::from_error(
ffi_err.error_code,
ffi_err.error_message,
));
}
};
let lookup_rows = match lookup_result.get_rows() {
Ok(rows) => rows,
Err(e) => {
let ffi_err = err_from_core_error(&e);
return Box::new(PrefixLookupResultInner::from_error(
ffi_err.error_code,
ffi_err.error_message,
));
}
};
let mut rows = Vec::with_capacity(lookup_rows.len());
for row in &lookup_rows {
match types::compacted_row_to_owned(row, &self.table_info) {
Ok(owned_row) => rows.push(owned_row),
Err(e) => {
return Box::new(PrefixLookupResultInner::from_error(
CLIENT_ERROR_CODE,
e.to_string(),
));
}
}
}
let columns = self.table_info.get_schema().columns().to_vec();
Box::new(PrefixLookupResultInner {
error: None,
rows,
columns,
})
}
}
// LogScanner implementation
unsafe fn delete_log_scanner(scanner: *mut LogScanner) {
if !scanner.is_null() {
unsafe {
drop(Box::from_raw(scanner));
}
}
}
// Helper function to free the Arrow FFI structures separately (for use after ImportRecordBatch)
pub extern "C" fn free_arrow_ffi_structures(array_ptr: usize, schema_ptr: usize) {
if array_ptr != 0 {
let _array = unsafe { Box::from_raw(array_ptr as *mut FFI_ArrowArray) };
}
if schema_ptr != 0 {
let _schema = unsafe { Box::from_raw(schema_ptr as *mut FFI_ArrowSchema) };
}
}
/// Dispatch a method call to whichever scanner variant is active.
/// Both LogScanner and RecordBatchLogScanner share the same subscribe/unsubscribe interface.
macro_rules! dispatch_scanner {
($self:expr, $method:ident($($arg:expr),*)) => {
match RUNTIME.block_on(async {
match &$self.scanner {
ScannerKind::Record(s) => s.$method($($arg),*).await,
ScannerKind::Batch(s) => s.$method($($arg),*).await,
}
}) {
Ok(_) => ok_result(),
Err(e) => err_from_core_error(&e),
}
};
}
impl LogScanner {
fn subscribe(&self, bucket_id: i32, start_offset: i64) -> ffi::FfiResult {
dispatch_scanner!(self, subscribe(bucket_id, start_offset))
}
fn subscribe_buckets(&self, subscriptions: Vec<ffi::FfiBucketSubscription>) -> ffi::FfiResult {
let bucket_offsets: HashMap<i32, i64> = subscriptions
.into_iter()
.map(|s| (s.bucket_id, s.offset))
.collect();
dispatch_scanner!(self, subscribe_buckets(&bucket_offsets))
}
fn subscribe_partition(
&self,
partition_id: PartitionId,
bucket_id: i32,
start_offset: i64,
) -> ffi::FfiResult {
dispatch_scanner!(
self,
subscribe_partition(partition_id, bucket_id, start_offset)
)
}
fn subscribe_partition_buckets(
&self,
subscriptions: Vec<ffi::FfiPartitionBucketSubscription>,
) -> ffi::FfiResult {
let offsets: HashMap<(PartitionId, i32), i64> = subscriptions
.into_iter()
.map(|s| ((s.partition_id, s.bucket_id), s.offset))
.collect();
dispatch_scanner!(self, subscribe_partition_buckets(&offsets))
}
fn unsubscribe(&self, bucket_id: i32) -> ffi::FfiResult {
dispatch_scanner!(self, unsubscribe(bucket_id))
}
fn unsubscribe_partition(&self, partition_id: PartitionId, bucket_id: i32) -> ffi::FfiResult {
dispatch_scanner!(self, unsubscribe_partition(partition_id, bucket_id))
}
fn poll(&self, timeout_ms: i64) -> Box<ScanResultInner> {
let ScannerKind::Record(ref inner) = self.scanner else {
return Box::new(ScanResultInner::from_error(
CLIENT_ERROR_CODE,
"Record-based scanner not available".to_string(),
));
};
let timeout = Duration::from_millis(timeout_ms.max(0) as u64);
let result = RUNTIME.block_on(async { inner.poll(timeout).await });
match result {
Ok(records) => {
let columns = self.projected_columns.clone();
let mut total_count = 0usize;
let mut buckets = Vec::new();
let mut bucket_infos = Vec::new();
for (table_bucket, bucket_records) in records.into_records_by_buckets() {
let count = bucket_records.len();
total_count += count;
bucket_infos.push(ffi::FfiBucketInfo {
table_id: table_bucket.table_id(),
bucket_id: table_bucket.bucket_id(),
has_partition_id: table_bucket.partition_id().is_some(),
partition_id: table_bucket.partition_id().unwrap_or(0),
record_count: count,
});
buckets.push((table_bucket, bucket_records));
}
Box::new(ScanResultInner {
error: None,
buckets,
columns,
bucket_infos,
total_count,
})
}
Err(e) => {
let ffi_err = err_from_core_error(&e);
Box::new(ScanResultInner::from_error(
ffi_err.error_code,
ffi_err.error_message,
))
}
}
}
fn poll_record_batch(&self, timeout_ms: i64) -> ffi::FfiArrowRecordBatchesResult {
let ScannerKind::Batch(ref inner_batch) = self.scanner else {
return ffi::FfiArrowRecordBatchesResult {
result: client_err("Batch-based scanner not available".to_string()),
arrow_batches: ffi::FfiArrowRecordBatches { batches: vec![] },
};
};
let timeout = Duration::from_millis(timeout_ms.max(0) as u64);
let result = RUNTIME.block_on(async { inner_batch.poll(timeout).await });
match result {
Ok(batches) => arrow_batches_result(types::core_scan_batches_to_ffi(&batches)),
Err(e) => empty_arrow_batches_result(err_from_core_error(&e)),
}
}
}
// BatchScanner implementation
unsafe fn delete_batch_scanner(scanner: *mut BatchScanner) {
if !scanner.is_null() {
unsafe {
drop(Box::from_raw(scanner));
}
}
}
impl BatchScanner {
fn next_batch(&self) -> ffi::FfiArrowRecordBatchesResult {
let mut scanner = self.inner.lock().unwrap();
match RUNTIME.block_on(scanner.next_batch()) {
Ok(Some(batch)) => arrow_batches_result(types::core_scan_batches_to_ffi(
std::slice::from_ref(&batch),
)),
Ok(None) => empty_arrow_batches_result(ok_result()),
Err(e) => empty_arrow_batches_result(err_from_core_error(&e)),
}
}
fn collect_all_batches(&self) -> ffi::FfiArrowRecordBatchesResult {
let mut scanner = self.inner.lock().unwrap();
match RUNTIME.block_on(scanner.collect_all_batches()) {
Ok(batches) => arrow_batches_result(types::core_scan_batches_to_ffi(&batches)),
Err(e) => empty_arrow_batches_result(err_from_core_error(&e)),
}
}
}
// ============================================================================
// 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 gr_set_array(&mut self, idx: usize, writer: &mut ArrayWriterInner) -> Result<(), String> {
self.ensure_size(idx);
writer.complete_if_needed()?;
let arr = writer.completed.take().ok_or_else(|| {
"ArrayWriter invariant violation: completed array missing after finalize".to_string()
})?;
self.row.set_field(idx, fcore::row::Datum::Array(arr));
Ok(())
}
fn gr_set_map(&mut self, idx: usize, writer: &mut MapWriterInner) -> Result<(), String> {
self.ensure_size(idx);
writer.complete_if_needed()?;
let map = writer.completed.take().ok_or_else(|| {
"MapWriter invariant violation: completed map missing after finalize".to_string()
})?;
self.row.set_field(idx, fcore::row::Datum::Map(map));
Ok(())
}
fn gr_set_row(&mut self, idx: usize, row: &mut GenericRowInner) -> Result<(), String> {
self.ensure_size(idx);
// Move the nested row out; the C++ side discards its writer afterwards.
let nested = std::mem::replace(&mut row.row, fcore::row::GenericRow::new(0));
self.row
.set_field(idx, fcore::row::Datum::Row(Box::new(nested)));
Ok(())
}
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;
/// Get column at `field`, or error if out of bounds.
fn get_column(
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<&fcore::metadata::Column, String> {
columns.get(field).ok_or_else(|| {
format!(
"field index {field} out of range ({} columns)",
columns.len()
)
})
}
/// Validate bounds, null, and type compatibility in a single pass.
/// Returns the data type on success for callers that need to dispatch on it.
fn validate<'a>(
row: &dyn InternalRow,
columns: &'a [fcore::metadata::Column],
field: usize,
getter: &str,
allowed: impl FnOnce(&fcore::metadata::DataType) -> bool,
) -> Result<&'a fcore::metadata::DataType, String> {
let col = get_column(columns, field)?;
if row.is_null_at(field).map_err(|e| e.to_string())? {
return Err(format!("field {field} is null"));
}
let dt = col.data_type();
if !allowed(dt) {
return Err(format!(
"{getter}: column {field} has incompatible type {dt}"
));
}
Ok(dt)
}
pub fn column_type(columns: &[fcore::metadata::Column], field: usize) -> Result<i32, String> {
Ok(types::core_data_type_to_ffi(
get_column(columns, field)?.data_type(),
))
}
pub fn column_name(columns: &[fcore::metadata::Column], field: usize) -> Result<&str, String> {
Ok(get_column(columns, field)?.name())
}
pub fn is_null(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<bool, String> {
get_column(columns, field)?;
row.is_null_at(field).map_err(|e| e.to_string())
}
pub fn get_bool(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<bool, String> {
validate(row, columns, field, "get_bool", |dt| {
matches!(dt, fcore::metadata::DataType::Boolean(_))
})?;
row.get_boolean(field).map_err(|e| e.to_string())
}
pub fn get_i32(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<i32, String> {
let dt = validate(row, columns, field, "get_i32", |dt| {
matches!(
dt,
fcore::metadata::DataType::TinyInt(_)
| fcore::metadata::DataType::SmallInt(_)
| fcore::metadata::DataType::Int(_)
)
})?;
match dt {
fcore::metadata::DataType::TinyInt(_) => row
.get_byte(field)
.map(|v| v as i32)
.map_err(|e| e.to_string()),
fcore::metadata::DataType::SmallInt(_) => row
.get_short(field)
.map(|v| v as i32)
.map_err(|e| e.to_string()),
_ => row.get_int(field).map_err(|e| e.to_string()),
}
}
pub fn get_i64(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<i64, String> {
validate(row, columns, field, "get_i64", |dt| {
matches!(dt, fcore::metadata::DataType::BigInt(_))
})?;
row.get_long(field).map_err(|e| e.to_string())
}
pub fn get_f32(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<f32, String> {
validate(row, columns, field, "get_f32", |dt| {
matches!(dt, fcore::metadata::DataType::Float(_))
})?;
row.get_float(field).map_err(|e| e.to_string())
}
pub fn get_f64(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<f64, String> {
validate(row, columns, field, "get_f64", |dt| {
matches!(dt, fcore::metadata::DataType::Double(_))
})?;
row.get_double(field).map_err(|e| e.to_string())
}
pub fn get_str<'a>(
row: &'a dyn InternalRow,
columns: &'a [fcore::metadata::Column],
field: usize,
) -> Result<&'a str, String> {
let dt = validate(row, columns, field, "get_str", |dt| {
matches!(
dt,
fcore::metadata::DataType::Char(_) | fcore::metadata::DataType::String(_)
)
})?;
match dt {
fcore::metadata::DataType::Char(ct) => row
.get_char(field, ct.length() as usize)
.map_err(|e| e.to_string()),
_ => row.get_string(field).map_err(|e| e.to_string()),
}
}
pub fn get_bytes<'a>(
row: &'a dyn InternalRow,
columns: &'a [fcore::metadata::Column],
field: usize,
) -> Result<&'a [u8], String> {
let dt = validate(row, columns, field, "get_bytes", |dt| {
matches!(
dt,
fcore::metadata::DataType::Binary(_) | fcore::metadata::DataType::Bytes(_)
)
})?;
match dt {
fcore::metadata::DataType::Binary(bt) => row
.get_binary(field, bt.length())
.map_err(|e| e.to_string()),
_ => row.get_bytes(field).map_err(|e| e.to_string()),
}
}
pub fn get_date_days(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<i32, String> {
validate(row, columns, field, "get_date_days", |dt| {
matches!(dt, fcore::metadata::DataType::Date(_))
})?;
row.get_date(field)
.map(|d| d.get_inner())
.map_err(|e| e.to_string())
}
pub fn get_time_millis(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<i32, String> {
validate(row, columns, field, "get_time_millis", |dt| {
matches!(dt, fcore::metadata::DataType::Time(_))
})?;
row.get_time(field)
.map(|t| t.get_inner())
.map_err(|e| e.to_string())
}
pub fn get_ts_millis(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<i64, String> {
let dt = validate(row, columns, field, "get_ts_millis", |dt| {
matches!(
dt,
fcore::metadata::DataType::Timestamp(_)
| fcore::metadata::DataType::TimestampLTz(_)
)
})?;
match dt {
fcore::metadata::DataType::TimestampLTz(ts) => row
.get_timestamp_ltz(field, ts.precision())
.map(|v| v.get_epoch_millisecond())
.map_err(|e| e.to_string()),
fcore::metadata::DataType::Timestamp(ts) => row
.get_timestamp_ntz(field, ts.precision())
.map(|v| v.get_millisecond())
.map_err(|e| e.to_string()),
dt => Err(format!("get_ts_millis: unexpected type {dt}")),
}
}
pub fn get_ts_nanos(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<i32, String> {
let dt = validate(row, columns, field, "get_ts_nanos", |dt| {
matches!(
dt,
fcore::metadata::DataType::Timestamp(_)
| fcore::metadata::DataType::TimestampLTz(_)
)
})?;
match dt {
fcore::metadata::DataType::TimestampLTz(ts) => row
.get_timestamp_ltz(field, ts.precision())
.map(|v| v.get_nano_of_millisecond())
.map_err(|e| e.to_string()),
fcore::metadata::DataType::Timestamp(ts) => row
.get_timestamp_ntz(field, ts.precision())
.map(|v| v.get_nano_of_millisecond())
.map_err(|e| e.to_string()),
dt => Err(format!("get_ts_nanos: unexpected type {dt}")),
}
}
pub fn is_ts_ltz(columns: &[fcore::metadata::Column], field: usize) -> Result<bool, String> {
Ok(matches!(
get_column(columns, field)?.data_type(),
fcore::metadata::DataType::TimestampLTz(_)
))
}
pub fn get_decimal_str(
row: &dyn InternalRow,
columns: &[fcore::metadata::Column],
field: usize,
) -> Result<String, String> {
let dt = validate(row, columns, field, "get_decimal_str", |dt| {
matches!(dt, fcore::metadata::DataType::Decimal(_))
})?;
match dt {
fcore::metadata::DataType::Decimal(dd) => {
let decimal = row
.get_decimal(field, dd.precision() as usize, dd.scale() as usize)
.map_err(|e| e.to_string())?;
Ok(decimal.to_big_decimal().to_string())
}
dt => Err(format!("get_decimal_str: unexpected type {dt}")),
}
}
}
// ============================================================================
// array_reader — low-level accessors over an already-resolved FlussArray.
//
// `get_datum` is the single dispatch that turns any array/map element into an
// owned `Datum` for the recursive `ValueInner` handle (bounds-checked, null
// and type validated, recursing into nested ROW/MAP/ARRAY).
// ============================================================================
mod array_reader {
use super::fcore;
use fcore::metadata::DataType as DT;
use fcore::row::Datum;
fn validate_index(
arr: &fcore::row::binary_array::FlussArray,
element: usize,
op: &str,
) -> Result<(), String> {
if element < arr.size() {
Ok(())
} else {
Err(format!(
"{op}: element index out of bounds: element={element}, size={}",
arr.size()
))
}
}
pub fn is_null(
arr: &fcore::row::binary_array::FlussArray,
element: usize,
) -> Result<bool, String> {
validate_index(arr, element, "array_is_null")?;
Ok(arr.is_null_at(element))
}
pub fn get_datum(
arr: &fcore::row::binary_array::FlussArray,
elem_type: &fcore::metadata::DataType,
element: usize,
) -> Result<fcore::row::Datum<'static>, String> {
if is_null(arr, element)? {
return Ok(Datum::Null);
}
let map_err = |e: fcore::error::Error| e.to_string();
Ok(match elem_type {
DT::Boolean(_) => Datum::Bool(arr.get_boolean(element).map_err(map_err)?),
DT::TinyInt(_) => Datum::Int8(arr.get_byte(element).map_err(map_err)?),
DT::SmallInt(_) => Datum::Int16(arr.get_short(element).map_err(map_err)?),
DT::Int(_) => Datum::Int32(arr.get_int(element).map_err(map_err)?),
DT::BigInt(_) => Datum::Int64(arr.get_long(element).map_err(map_err)?),
DT::Float(_) => Datum::Float32(arr.get_float(element).map_err(map_err)?.into()),
DT::Double(_) => Datum::Float64(arr.get_double(element).map_err(map_err)?.into()),
DT::String(_) | DT::Char(_) => Datum::String(std::borrow::Cow::Owned(
arr.get_string(element).map_err(map_err)?.to_string(),
)),
DT::Bytes(_) | DT::Binary(_) => Datum::Blob(std::borrow::Cow::Owned(
arr.get_binary(element).map_err(map_err)?.to_vec(),
)),
DT::Decimal(d) => Datum::Decimal(
arr.get_decimal(element, d.precision(), d.scale())
.map_err(map_err)?,
),
DT::Date(_) => Datum::Date(arr.get_date(element).map_err(map_err)?),
DT::Time(_) => Datum::Time(arr.get_time(element).map_err(map_err)?),
DT::Timestamp(t) => Datum::TimestampNtz(
arr.get_timestamp_ntz(element, t.precision())
.map_err(map_err)?,
),
DT::TimestampLTz(t) => Datum::TimestampLtz(
arr.get_timestamp_ltz(element, t.precision())
.map_err(map_err)?,
),
DT::Array(_) => Datum::Array(arr.get_array(element).map_err(map_err)?),
DT::Map(mt) => Datum::Map(
arr.get_map(element, mt.key_type(), mt.value_type())
.map_err(map_err)?,
),
DT::Row(rt) => {
let cr = arr.get_row(element, rt).map_err(map_err)?;
let cols: Vec<fcore::metadata::Column> = rt
.fields()
.iter()
.map(|f| fcore::metadata::Column::new(f.name(), f.data_type().clone()))
.collect();
Datum::Row(Box::new(
crate::types::internal_row_to_owned_generic(&cr, &cols)
.map_err(|e| e.to_string())?,
))
}
})
}
}
// ============================================================================
// Opaque types: ScanResultInner (scan read path)
// ============================================================================
pub struct ScanResultInner {
error: Option<(i32, String)>,
buckets: Vec<(fcore::metadata::TableBucket, Vec<fcore::record::ScanRecord>)>,
columns: Vec<fcore::metadata::Column>,
bucket_infos: Vec<ffi::FfiBucketInfo>,
total_count: usize,
}
impl ScanResultInner {
fn from_error(code: i32, msg: String) -> Self {
Self {
error: Some((code, msg)),
buckets: Vec::new(),
columns: Vec::new(),
bucket_infos: Vec::new(),
total_count: 0,
}
}
fn resolve(&self, bucket: usize, rec: usize) -> &fcore::record::ScanRecord {
&self.buckets[bucket].1[rec]
}
fn sv_has_error(&self) -> bool {
self.error.is_some()
}
fn sv_error_code(&self) -> i32 {
self.error.as_ref().map_or(0, |e| e.0)
}
fn sv_error_message(&self) -> &str {
self.error.as_ref().map_or("", |e| e.1.as_str())
}
fn sv_record_count(&self) -> usize {
self.total_count
}
fn sv_column_count(&self) -> usize {
self.columns.len()
}
fn sv_column_name(&self, field: usize) -> Result<&str, String> {
row_reader::column_name(&self.columns, field)
}
fn sv_column_type(&self, field: usize) -> Result<i32, String> {
row_reader::column_type(&self.columns, field)
}
fn sv_offset(&self, bucket: usize, rec: usize) -> i64 {
self.resolve(bucket, rec).offset()
}
fn sv_timestamp(&self, bucket: usize, rec: usize) -> i64 {
self.resolve(bucket, rec).timestamp()
}
fn sv_change_type(&self, bucket: usize, rec: usize) -> i32 {
self.resolve(bucket, rec).change_type().to_byte_value() as i32
}
fn sv_field_count(&self) -> usize {
self.columns.len()
}
// Field accessors — C++ validates bounds in BucketRecords/RecordAt, validate() checks field.
fn sv_is_null(&self, bucket: usize, rec: usize, field: usize) -> Result<bool, String> {
row_reader::is_null(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_bool(&self, bucket: usize, rec: usize, field: usize) -> Result<bool, String> {
row_reader::get_bool(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_i32(&self, bucket: usize, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_i32(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_i64(&self, bucket: usize, rec: usize, field: usize) -> Result<i64, String> {
row_reader::get_i64(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_f32(&self, bucket: usize, rec: usize, field: usize) -> Result<f32, String> {
row_reader::get_f32(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_f64(&self, bucket: usize, rec: usize, field: usize) -> Result<f64, String> {
row_reader::get_f64(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_str(&self, bucket: usize, rec: usize, field: usize) -> Result<&str, String> {
row_reader::get_str(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_bytes(&self, bucket: usize, rec: usize, field: usize) -> Result<&[u8], String> {
row_reader::get_bytes(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_date_days(&self, bucket: usize, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_date_days(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_time_millis(&self, bucket: usize, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_time_millis(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_ts_millis(&self, bucket: usize, rec: usize, field: usize) -> Result<i64, String> {
row_reader::get_ts_millis(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_ts_nanos(&self, bucket: usize, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_ts_nanos(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_is_ts_ltz(&self, _bucket: usize, _rec: usize, field: usize) -> Result<bool, String> {
row_reader::is_ts_ltz(&self.columns, field)
}
fn sv_get_decimal_str(
&self,
bucket: usize,
rec: usize,
field: usize,
) -> Result<String, String> {
row_reader::get_decimal_str(self.resolve(bucket, rec).row(), &self.columns, field)
}
fn sv_get_value(
&self,
bucket: usize,
rec: usize,
field: usize,
) -> Result<Box<ValueInner>, String> {
// Scan rows are borrowed Arrow cursors; materialize only the requested
// field's owned datum (not the whole row). Top-level scalar scan reads
// keep their zero-copy index fast-path.
let datum = crate::types::field_to_owned_datum(
self.resolve(bucket, rec).row(),
&self.columns,
field,
)
.map_err(|e| e.to_string())?;
let data_type = self
.columns
.get(field)
.ok_or_else(|| format!("field index {field} out of range"))?
.data_type()
.clone();
Ok(Box::new(ValueInner { datum, data_type }))
}
fn sv_bucket_infos(&self) -> &Vec<ffi::FfiBucketInfo> {
&self.bucket_infos
}
}
// ============================================================================
// Opaque types: LookupResultInner (lookup read path)
// ============================================================================
pub struct LookupResultInner {
error: Option<(i32, String)>,
found: bool,
row: Option<fcore::row::GenericRow<'static>>,
columns: Vec<fcore::metadata::Column>,
}
impl LookupResultInner {
fn from_error(code: i32, msg: String) -> Self {
Self {
error: Some((code, msg)),
found: false,
row: None,
columns: Vec::new(),
}
}
fn lv_has_error(&self) -> bool {
self.error.is_some()
}
fn lv_error_code(&self) -> i32 {
self.error.as_ref().map_or(0, |e| e.0)
}
fn lv_error_message(&self) -> &str {
self.error.as_ref().map_or("", |e| e.1.as_str())
}
fn lv_found(&self) -> bool {
self.found
}
fn lv_field_count(&self) -> usize {
self.columns.len()
}
fn lv_column_type(&self, field: usize) -> Result<i32, String> {
row_reader::column_type(&self.columns, field)
}
fn lv_column_name(&self, field: usize) -> Result<&str, String> {
row_reader::column_name(&self.columns, field)
}
fn lv_row(&self) -> Result<&fcore::row::GenericRow<'static>, String> {
self.row
.as_ref()
.ok_or_else(|| "no row available (not found or error)".to_string())
}
// Field accessors — delegate to shared row_reader helpers.
fn lv_is_null(&self, field: usize) -> Result<bool, String> {
let r = self.lv_row()?;
row_reader::is_null(r, &self.columns, field)
}
fn lv_get_bool(&self, field: usize) -> Result<bool, String> {
let r = self.lv_row()?;
row_reader::get_bool(r, &self.columns, field)
}
fn lv_get_i32(&self, field: usize) -> Result<i32, String> {
let r = self.lv_row()?;
row_reader::get_i32(r, &self.columns, field)
}
fn lv_get_i64(&self, field: usize) -> Result<i64, String> {
let r = self.lv_row()?;
row_reader::get_i64(r, &self.columns, field)
}
fn lv_get_f32(&self, field: usize) -> Result<f32, String> {
let r = self.lv_row()?;
row_reader::get_f32(r, &self.columns, field)
}
fn lv_get_f64(&self, field: usize) -> Result<f64, String> {
let r = self.lv_row()?;
row_reader::get_f64(r, &self.columns, field)
}
fn lv_get_str(&self, field: usize) -> Result<&str, String> {
let r = self.lv_row()?;
row_reader::get_str(r, &self.columns, field)
}
fn lv_get_bytes(&self, field: usize) -> Result<&[u8], String> {
let r = self.lv_row()?;
row_reader::get_bytes(r, &self.columns, field)
}
fn lv_get_date_days(&self, field: usize) -> Result<i32, String> {
let r = self.lv_row()?;
row_reader::get_date_days(r, &self.columns, field)
}
fn lv_get_time_millis(&self, field: usize) -> Result<i32, String> {
let r = self.lv_row()?;
row_reader::get_time_millis(r, &self.columns, field)
}
fn lv_get_ts_millis(&self, field: usize) -> Result<i64, String> {
let r = self.lv_row()?;
row_reader::get_ts_millis(r, &self.columns, field)
}
fn lv_get_ts_nanos(&self, field: usize) -> Result<i32, String> {
let r = self.lv_row()?;
row_reader::get_ts_nanos(r, &self.columns, field)
}
fn lv_is_ts_ltz(&self, field: usize) -> Result<bool, String> {
row_reader::is_ts_ltz(&self.columns, field)
}
fn lv_get_decimal_str(&self, field: usize) -> Result<String, String> {
let r = self.lv_row()?;
row_reader::get_decimal_str(r, &self.columns, field)
}
fn lv_get_value(&self, field: usize) -> Result<Box<ValueInner>, String> {
value_from_owned_row(self.lv_row()?, &self.columns, field)
}
}
// ============================================================================
// Opaque types: PrefixLookupResultInner (prefix lookup read path)
//
// Like LookupResultInner but holds 0..n rows; accessors take a record index.
// ============================================================================
pub struct PrefixLookupResultInner {
error: Option<(i32, String)>,
rows: Vec<GenericRow<'static>>,
columns: Vec<Column>,
}
impl PrefixLookupResultInner {
fn from_error(code: i32, msg: String) -> Self {
Self {
error: Some((code, msg)),
rows: Vec::new(),
columns: Vec::new(),
}
}
fn plv_has_error(&self) -> bool {
self.error.is_some()
}
fn plv_error_code(&self) -> i32 {
self.error.as_ref().map_or(0, |e| e.0)
}
fn plv_error_message(&self) -> &str {
self.error.as_ref().map_or("", |e| e.1.as_str())
}
fn plv_row_count(&self) -> usize {
self.rows.len()
}
fn plv_field_count(&self) -> usize {
self.columns.len()
}
fn plv_column_type(&self, field: usize) -> Result<i32, String> {
row_reader::column_type(&self.columns, field)
}
fn plv_column_name(&self, field: usize) -> Result<&str, String> {
row_reader::column_name(&self.columns, field)
}
fn plv_row(&self, rec: usize) -> Result<&GenericRow<'static>, String> {
self.rows
.get(rec)
.ok_or_else(|| format!("record index {rec} out of range ({} rows)", self.rows.len()))
}
// Field accessors — delegate to shared row_reader helpers.
fn plv_is_null(&self, rec: usize, field: usize) -> Result<bool, String> {
row_reader::is_null(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_bool(&self, rec: usize, field: usize) -> Result<bool, String> {
row_reader::get_bool(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_i32(&self, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_i32(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_i64(&self, rec: usize, field: usize) -> Result<i64, String> {
row_reader::get_i64(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_f32(&self, rec: usize, field: usize) -> Result<f32, String> {
row_reader::get_f32(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_f64(&self, rec: usize, field: usize) -> Result<f64, String> {
row_reader::get_f64(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_str(&self, rec: usize, field: usize) -> Result<&str, String> {
row_reader::get_str(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_bytes(&self, rec: usize, field: usize) -> Result<&[u8], String> {
row_reader::get_bytes(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_date_days(&self, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_date_days(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_time_millis(&self, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_time_millis(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_ts_millis(&self, rec: usize, field: usize) -> Result<i64, String> {
row_reader::get_ts_millis(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_ts_nanos(&self, rec: usize, field: usize) -> Result<i32, String> {
row_reader::get_ts_nanos(self.plv_row(rec)?, &self.columns, field)
}
fn plv_is_ts_ltz(&self, _rec: usize, field: usize) -> Result<bool, String> {
row_reader::is_ts_ltz(&self.columns, field)
}
fn plv_get_decimal_str(&self, rec: usize, field: usize) -> Result<String, String> {
row_reader::get_decimal_str(self.plv_row(rec)?, &self.columns, field)
}
fn plv_get_value(&self, rec: usize, field: usize) -> Result<Box<ValueInner>, String> {
value_from_owned_row(self.plv_row(rec)?, &self.columns, field)
}
}
// ============================================================================
// ValueInner — recursive value handle for complex reads. Holds an owned Datum
// + its DataType: leaf getters match on the datum, navigators (v_at/v_key_at/
// v_value_at/v_field) descend into a child node.
// ============================================================================
pub struct ValueInner {
datum: fcore::row::Datum<'static>,
data_type: fcore::metadata::DataType,
}
/// Builds a `ValueInner` from an owned row; shared by the lookup and prefix-lookup paths.
fn value_from_owned_row(
row: &GenericRow<'static>,
columns: &[Column],
field: usize,
) -> Result<Box<ValueInner>, String> {
let datum = row
.values
.get(field)
.ok_or_else(|| format!("field index {field} out of range"))?
.clone();
let data_type = columns
.get(field)
.ok_or_else(|| format!("field index {field} out of range"))?
.data_type()
.clone();
Ok(Box::new(ValueInner { datum, data_type }))
}
impl ValueInner {
fn type_err(&self, expected: &str) -> String {
format!("value of type {} is not {expected}", self.data_type)
}
fn v_type(&self) -> i32 {
crate::types::core_data_type_to_ffi(&self.data_type)
}
fn v_is_null(&self) -> bool {
matches!(self.datum, fcore::row::Datum::Null)
}
fn v_get_bool(&self) -> Result<bool, String> {
match &self.datum {
fcore::row::Datum::Bool(v) => Ok(*v),
_ => Err(self.type_err("BOOLEAN")),
}
}
fn v_get_i32(&self) -> Result<i32, String> {
match &self.datum {
fcore::row::Datum::Int8(v) => Ok(i32::from(*v)),
fcore::row::Datum::Int16(v) => Ok(i32::from(*v)),
fcore::row::Datum::Int32(v) => Ok(*v),
_ => Err(self.type_err("INT")),
}
}
fn v_get_i64(&self) -> Result<i64, String> {
match &self.datum {
fcore::row::Datum::Int64(v) => Ok(*v),
_ => Err(self.type_err("BIGINT")),
}
}
fn v_get_f32(&self) -> Result<f32, String> {
match &self.datum {
fcore::row::Datum::Float32(v) => Ok(v.0),
_ => Err(self.type_err("FLOAT")),
}
}
fn v_get_f64(&self) -> Result<f64, String> {
match &self.datum {
fcore::row::Datum::Float64(v) => Ok(v.0),
_ => Err(self.type_err("DOUBLE")),
}
}
fn v_get_str(&self) -> Result<String, String> {
match &self.datum {
fcore::row::Datum::String(v) => Ok(v.to_string()),
_ => Err(self.type_err("STRING")),
}
}
fn v_get_bytes(&self) -> Result<Vec<u8>, String> {
match &self.datum {
fcore::row::Datum::Blob(v) => Ok(v.to_vec()),
_ => Err(self.type_err("BYTES")),
}
}
fn v_get_date_days(&self) -> Result<i32, String> {
match &self.datum {
fcore::row::Datum::Date(d) => Ok(d.get_inner()),
_ => Err(self.type_err("DATE")),
}
}
fn v_get_time_millis(&self) -> Result<i32, String> {
match &self.datum {
fcore::row::Datum::Time(t) => Ok(t.get_inner()),
_ => Err(self.type_err("TIME")),
}
}
fn v_get_ts_millis(&self) -> Result<i64, String> {
match &self.datum {
fcore::row::Datum::TimestampNtz(t) => Ok(t.get_millisecond()),
fcore::row::Datum::TimestampLtz(t) => Ok(t.get_epoch_millisecond()),
_ => Err(self.type_err("TIMESTAMP")),
}
}
fn v_get_ts_nanos(&self) -> Result<i32, String> {
match &self.datum {
fcore::row::Datum::TimestampNtz(t) => Ok(t.get_nano_of_millisecond()),
fcore::row::Datum::TimestampLtz(t) => Ok(t.get_nano_of_millisecond()),
_ => Err(self.type_err("TIMESTAMP")),
}
}
fn v_get_decimal_str(&self) -> Result<String, String> {
match &self.datum {
fcore::row::Datum::Decimal(d) => Ok(d.to_big_decimal().to_string()),
_ => Err(self.type_err("DECIMAL")),
}
}
fn v_size(&self) -> Result<usize, String> {
match &self.datum {
fcore::row::Datum::Array(a) => Ok(a.size()),
fcore::row::Datum::Map(m) => Ok(m.size()),
_ => Err(self.type_err("ARRAY/MAP")),
}
}
fn v_field_count(&self) -> Result<usize, String> {
match &self.datum {
fcore::row::Datum::Row(r) => Ok(r.values.len()),
_ => Err(self.type_err("ROW")),
}
}
fn v_at(&self, i: usize) -> Result<Box<ValueInner>, String> {
match &self.datum {
fcore::row::Datum::Array(a) => {
let elem_type = match &self.data_type {
fcore::metadata::DataType::Array(at) => at.get_element_type().clone(),
_ => return Err("internal: ARRAY datum without ARRAY type".to_string()),
};
let datum = array_reader::get_datum(a, &elem_type, i)?;
Ok(Box::new(ValueInner {
datum,
data_type: elem_type,
}))
}
_ => Err(self.type_err("ARRAY")),
}
}
fn v_key_at(&self, i: usize) -> Result<Box<ValueInner>, String> {
match &self.datum {
fcore::row::Datum::Map(m) => {
let kt = m.key_type().clone();
let datum = array_reader::get_datum(m.key_array(), &kt, i)?;
Ok(Box::new(ValueInner {
datum,
data_type: kt,
}))
}
_ => Err(self.type_err("MAP")),
}
}
fn v_value_at(&self, i: usize) -> Result<Box<ValueInner>, String> {
match &self.datum {
fcore::row::Datum::Map(m) => {
let vt = m.value_type().clone();
let datum = array_reader::get_datum(m.value_array(), &vt, i)?;
Ok(Box::new(ValueInner {
datum,
data_type: vt,
}))
}
_ => Err(self.type_err("MAP")),
}
}
fn v_field(&self, i: usize) -> Result<Box<ValueInner>, String> {
match &self.datum {
fcore::row::Datum::Row(r) => {
let ft = match &self.data_type {
fcore::metadata::DataType::Row(rt) => rt
.fields()
.get(i)
.ok_or_else(|| format!("field index {i} out of range"))?
.data_type()
.clone(),
_ => return Err("internal: ROW datum without ROW type".to_string()),
};
let datum = r
.values
.get(i)
.ok_or_else(|| format!("field index {i} out of range"))?
.clone();
Ok(Box::new(ValueInner {
datum,
data_type: ft,
}))
}
_ => Err(self.type_err("ROW")),
}
}
fn v_field_by_name(&self, name: &str) -> Result<Box<ValueInner>, String> {
match &self.data_type {
fcore::metadata::DataType::Row(rt) => {
let idx = rt
.fields()
.iter()
.position(|f| f.name() == name)
.ok_or_else(|| format!("no field named '{name}' in {}", self.data_type))?;
self.v_field(idx)
}
_ => Err(self.type_err("ROW")),
}
}
}
// ============================================================================
// Opaque types: ArrayWriterInner (array builder for writes)
// ============================================================================
pub struct ArrayWriterInner {
writer: Option<fcore::row::binary_array::FlussArrayWriter>,
completed: Option<fcore::row::binary_array::FlussArray>,
element_type: fcore::metadata::DataType,
num_elements: usize,
}
// ============================================================================
// MapWriterInner — opaque map builder (mirrors ArrayWriterInner)
// ============================================================================
pub struct MapWriterInner {
writer: Option<fcore::row::binary_map::FlussMapWriter>,
completed: Option<fcore::row::binary_map::FlussMap>,
key_type: fcore::metadata::DataType,
value_type: fcore::metadata::DataType,
pending_key: Option<fcore::row::Datum<'static>>,
pending_value: Option<fcore::row::Datum<'static>>,
}
fn ts_ntz_from(millis: i64, nanos: i32) -> fcore::row::TimestampNtz {
fcore::row::TimestampNtz::from_millis_nanos(millis, nanos)
.unwrap_or_else(|_| fcore::row::TimestampNtz::new(millis))
}
fn ts_ltz_from(millis: i64, nanos: i32) -> fcore::row::TimestampLtz {
fcore::row::TimestampLtz::from_millis_nanos(millis, nanos)
.unwrap_or_else(|_| fcore::row::TimestampLtz::new(millis))
}
/// Parse a decimal string into a `Datum::Decimal` matching `dt`'s precision and
/// scale. Used by the map key/value decimal setters, which carry no schema.
fn parse_decimal_datum(
dt: &fcore::metadata::DataType,
val: &str,
) -> Result<fcore::row::Datum<'static>, String> {
match dt {
fcore::metadata::DataType::Decimal(d) => {
let bd = bigdecimal::BigDecimal::from_str(val).map_err(|e| e.to_string())?;
let dec = fcore::row::Decimal::from_big_decimal(bd, d.precision(), d.scale())
.map_err(|e| e.to_string())?;
Ok(fcore::row::Datum::Decimal(dec))
}
other => Err(format!(
"decimal setter used on non-DECIMAL map type {other}"
)),
}
}
fn timestamp_datum(
dt: &fcore::metadata::DataType,
millis: i64,
nanos: i32,
) -> Result<fcore::row::Datum<'static>, String> {
match dt {
fcore::metadata::DataType::Timestamp(_) => {
Ok(fcore::row::Datum::TimestampNtz(ts_ntz_from(millis, nanos)))
}
fcore::metadata::DataType::TimestampLTz(_) => {
Ok(fcore::row::Datum::TimestampLtz(ts_ltz_from(millis, nanos)))
}
other => Err(format!(
"timestamp setter used on non-TIMESTAMP map type {other}"
)),
}
}
/// Narrow a C++ `int32` into the `Datum` a map key/value expects, erroring on
/// overflow. Core needs `Int8`/`Int16` for TINYINT/SMALLINT; a widened `Int32`
/// is rejected. Mirrors `ArrayWriterInner::aw_set_i32`.
fn int_datum(dt: &DataType, val: i32, role: &str) -> Result<Datum<'static>, String> {
match dt {
DataType::TinyInt(_) => i8::try_from(val)
.map(Datum::Int8)
.map_err(|_| format!("Value {val} does not fit TINYINT map {role}")),
DataType::SmallInt(_) => i16::try_from(val)
.map(Datum::Int16)
.map_err(|_| format!("Value {val} does not fit SMALLINT map {role}")),
_ => Ok(Datum::Int32(val)),
}
}
#[allow(clippy::too_many_arguments)]
fn new_map_writer(
capacity: usize,
key_leaf_type_id: i32,
key_precision: u32,
key_scale: u32,
value_leaf_type_id: i32,
value_precision: u32,
value_scale: u32,
value_array_nesting: u32,
) -> Result<Box<MapWriterInner>, String> {
let key_type = types::element_type_from_ffi(key_leaf_type_id, key_precision, key_scale, 0)
.map_err(|e| e.to_string())?;
let value_type = types::element_type_from_ffi(
value_leaf_type_id,
value_precision,
value_scale,
value_array_nesting,
)
.map_err(|e| e.to_string())?;
let writer = fcore::row::binary_map::FlussMapWriter::new(capacity, &key_type, &value_type);
Ok(Box::new(MapWriterInner {
writer: Some(writer),
completed: None,
key_type,
value_type,
pending_key: None,
pending_value: None,
}))
}
fn new_map_writer_arrow(
capacity: usize,
kv_schema_ptr: usize,
) -> Result<Box<MapWriterInner>, String> {
let mut types =
unsafe { types::arrow_ffi_to_data_types(kv_schema_ptr) }.map_err(|e| e.to_string())?;
if types.len() != 2 {
return Err(format!(
"map writer Arrow schema must have exactly 2 fields (key, value), got {}",
types.len()
));
}
let value_type = types.swap_remove(1);
let key_type = types.swap_remove(0);
let writer = fcore::row::binary_map::FlussMapWriter::new(capacity, &key_type, &value_type);
Ok(Box::new(MapWriterInner {
writer: Some(writer),
completed: None,
key_type,
value_type,
pending_key: None,
pending_value: None,
}))
}
impl MapWriterInner {
fn writer_mut(&mut self) -> Result<&mut fcore::row::binary_map::FlussMapWriter, String> {
self.writer
.as_mut()
.ok_or_else(|| "MapWriter is already finalized".to_string())
}
fn mw_key_bool(&mut self, val: bool) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::Bool(val));
Ok(())
}
fn mw_key_i32(&mut self, val: i32) -> Result<(), String> {
self.pending_key = Some(int_datum(&self.key_type, val, "key")?);
Ok(())
}
fn mw_key_i64(&mut self, val: i64) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::Int64(val));
Ok(())
}
fn mw_key_f32(&mut self, val: f32) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::Float32(val.into()));
Ok(())
}
fn mw_key_f64(&mut self, val: f64) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::Float64(val.into()));
Ok(())
}
fn mw_key_str(&mut self, val: &str) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::String(std::borrow::Cow::Owned(
val.to_string(),
)));
Ok(())
}
fn mw_key_bytes(&mut self, val: &[u8]) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::Blob(std::borrow::Cow::Owned(
val.to_vec(),
)));
Ok(())
}
fn mw_key_date(&mut self, days: i32) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::Date(fcore::row::Date::new(days)));
Ok(())
}
fn mw_key_time(&mut self, millis: i32) -> Result<(), String> {
self.pending_key = Some(fcore::row::Datum::Time(fcore::row::Time::new(millis)));
Ok(())
}
fn mw_key_timestamp(&mut self, millis: i64, nanos: i32) -> Result<(), String> {
self.pending_key = Some(timestamp_datum(&self.key_type, millis, nanos)?);
Ok(())
}
fn mw_key_decimal_str(&mut self, val: &str) -> Result<(), String> {
self.pending_key = Some(parse_decimal_datum(&self.key_type, val)?);
Ok(())
}
fn mw_value_null(&mut self) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Null);
Ok(())
}
fn mw_value_bool(&mut self, val: bool) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Bool(val));
Ok(())
}
fn mw_value_i32(&mut self, val: i32) -> Result<(), String> {
self.pending_value = Some(int_datum(&self.value_type, val, "value")?);
Ok(())
}
fn mw_value_i64(&mut self, val: i64) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Int64(val));
Ok(())
}
fn mw_value_f32(&mut self, val: f32) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Float32(val.into()));
Ok(())
}
fn mw_value_f64(&mut self, val: f64) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Float64(val.into()));
Ok(())
}
fn mw_value_str(&mut self, val: &str) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::String(std::borrow::Cow::Owned(
val.to_string(),
)));
Ok(())
}
fn mw_value_bytes(&mut self, val: &[u8]) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Blob(std::borrow::Cow::Owned(
val.to_vec(),
)));
Ok(())
}
fn mw_value_date(&mut self, days: i32) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Date(fcore::row::Date::new(days)));
Ok(())
}
fn mw_value_time(&mut self, millis: i32) -> Result<(), String> {
self.pending_value = Some(fcore::row::Datum::Time(fcore::row::Time::new(millis)));
Ok(())
}
fn mw_value_timestamp(&mut self, millis: i64, nanos: i32) -> Result<(), String> {
self.pending_value = Some(timestamp_datum(&self.value_type, millis, nanos)?);
Ok(())
}
fn mw_value_decimal_str(&mut self, val: &str) -> Result<(), String> {
self.pending_value = Some(parse_decimal_datum(&self.value_type, val)?);
Ok(())
}
fn mw_value_row(&mut self, row: &mut GenericRowInner) -> Result<(), String> {
let nested = std::mem::replace(&mut row.row, fcore::row::GenericRow::new(0));
self.pending_value = Some(fcore::row::Datum::Row(Box::new(nested)));
Ok(())
}
fn mw_value_map(&mut self, map: &mut MapWriterInner) -> Result<(), String> {
map.complete_if_needed()?;
let completed = map
.completed
.take()
.ok_or_else(|| "MapWriter invariant violation: nested map missing".to_string())?;
self.pending_value = Some(fcore::row::Datum::Map(completed));
Ok(())
}
fn mw_value_array(&mut self, array: &mut ArrayWriterInner) -> Result<(), String> {
array.complete_if_needed()?;
let completed = array
.completed
.take()
.ok_or_else(|| "ArrayWriter invariant violation: nested array missing".to_string())?;
self.pending_value = Some(fcore::row::Datum::Array(completed));
Ok(())
}
/// Commit the pending key/value as one map entry. Keys cannot be null;
/// an unset value defaults to null.
fn mw_commit(&mut self) -> Result<(), String> {
let key = self
.pending_key
.take()
.ok_or_else(|| "MapWriter: entry key not set before commit".to_string())?;
let value = self.pending_value.take().unwrap_or(fcore::row::Datum::Null);
self.writer_mut()?
.write_entry(key, value)
.map_err(|e| e.to_string())
}
fn complete_if_needed(&mut self) -> Result<(), String> {
if self.completed.is_none() {
let w = self
.writer
.take()
.ok_or_else(|| "MapWriter has already been finalized".to_string())?;
self.completed = Some(w.complete().map_err(|e| e.to_string())?);
}
Ok(())
}
}
fn new_array_writer(
size: usize,
element_leaf_type_id: i32,
precision: u32,
scale: u32,
array_nesting: u32,
) -> Result<Box<ArrayWriterInner>, String> {
let element_type =
types::element_type_from_ffi(element_leaf_type_id, precision, scale, array_nesting)
.map_err(|e| e.to_string())?;
let writer = fcore::row::binary_array::FlussArrayWriter::new(size, &element_type);
Ok(Box::new(ArrayWriterInner {
writer: Some(writer),
completed: None,
element_type,
num_elements: size,
}))
}
fn new_array_writer_arrow(
size: usize,
element_schema_ptr: usize,
) -> Result<Box<ArrayWriterInner>, String> {
let mut types =
unsafe { types::arrow_ffi_to_data_types(element_schema_ptr) }.map_err(|e| e.to_string())?;
let element_type = if types.is_empty() {
return Err("array element Arrow schema must have one field".to_string());
} else {
types.swap_remove(0)
};
let writer = fcore::row::binary_array::FlussArrayWriter::new(size, &element_type);
Ok(Box::new(ArrayWriterInner {
writer: Some(writer),
completed: None,
element_type,
num_elements: size,
}))
}
impl ArrayWriterInner {
fn writer_mut(&mut self) -> Result<&mut fcore::row::binary_array::FlussArrayWriter, String> {
self.writer
.as_mut()
.ok_or_else(|| "ArrayWriter is already finalized".to_string())
}
fn validate_index(&self, idx: usize) -> Result<(), String> {
if idx < self.num_elements {
Ok(())
} else {
Err(format!(
"ArrayWriter index out of bounds: idx={idx}, size={}",
self.num_elements
))
}
}
fn complete_if_needed(&mut self) -> Result<(), String> {
if self.completed.is_none() {
let w = self
.writer
.take()
.ok_or_else(|| "ArrayWriter has already been finalized".to_string())?;
self.completed = Some(w.complete().map_err(|e| e.to_string())?);
}
Ok(())
}
/// Checks writer liveness first, then the element index. Returning the
/// clearest finalization error before a bounds error keeps diagnostics
/// aligned with the caller's intent when a writer is misused after
/// completion.
fn ensure_writable(&self, idx: usize) -> Result<(), String> {
if self.writer.is_none() {
return Err("ArrayWriter is already finalized".to_string());
}
self.validate_index(idx)
}
fn aw_size(&self) -> usize {
self.num_elements
}
fn aw_set_null(&mut self, idx: usize) -> Result<(), String> {
self.ensure_writable(idx)?;
self.writer_mut()?.set_null_at(idx);
Ok(())
}
fn aw_set_bool(&mut self, idx: usize, val: bool) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::Boolean(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected BOOLEAN element, got {}",
self.element_type
));
}
self.writer_mut()?.write_boolean(idx, val);
Ok(())
}
fn aw_set_i32(&mut self, idx: usize, val: i32) -> Result<(), String> {
self.ensure_writable(idx)?;
match &self.element_type {
fcore::metadata::DataType::TinyInt(_) => {
let v = i8::try_from(val)
.map_err(|_| format!("Value {val} does not fit TINYINT element"))?;
self.writer_mut()?.write_byte(idx, v);
}
fcore::metadata::DataType::SmallInt(_) => {
let v = i16::try_from(val)
.map_err(|_| format!("Value {val} does not fit SMALLINT element"))?;
self.writer_mut()?.write_short(idx, v);
}
fcore::metadata::DataType::Int(_) => {
self.writer_mut()?.write_int(idx, val);
}
_ => {
return Err(format!(
"ArrayWriter type mismatch: expected TINYINT/SMALLINT/INT element, got {}",
self.element_type
));
}
}
Ok(())
}
fn aw_set_i64(&mut self, idx: usize, val: i64) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::BigInt(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected BIGINT element, got {}",
self.element_type
));
}
self.writer_mut()?.write_long(idx, val);
Ok(())
}
fn aw_set_f32(&mut self, idx: usize, val: f32) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::Float(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected FLOAT element, got {}",
self.element_type
));
}
self.writer_mut()?.write_float(idx, val);
Ok(())
}
fn aw_set_f64(&mut self, idx: usize, val: f64) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::Double(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected DOUBLE element, got {}",
self.element_type
));
}
self.writer_mut()?.write_double(idx, val);
Ok(())
}
fn aw_set_str(&mut self, idx: usize, val: &str) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(
self.element_type,
fcore::metadata::DataType::String(_) | fcore::metadata::DataType::Char(_)
) {
return Err(format!(
"ArrayWriter type mismatch: expected STRING/CHAR element, got {}",
self.element_type
));
}
self.writer_mut()?.write_string(idx, val);
Ok(())
}
fn aw_set_bytes(&mut self, idx: usize, val: &[u8]) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(
self.element_type,
fcore::metadata::DataType::Bytes(_) | fcore::metadata::DataType::Binary(_)
) {
return Err(format!(
"ArrayWriter type mismatch: expected BYTES/BINARY element, got {}",
self.element_type
));
}
self.writer_mut()?.write_binary_bytes(idx, val);
Ok(())
}
fn aw_set_date(&mut self, idx: usize, days: i32) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::Date(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected DATE element, got {}",
self.element_type
));
}
self.writer_mut()?
.write_date(idx, fcore::row::Date::new(days));
Ok(())
}
fn aw_set_time(&mut self, idx: usize, millis: i32) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::Time(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected TIME element, got {}",
self.element_type
));
}
self.writer_mut()?
.write_time(idx, fcore::row::Time::new(millis));
Ok(())
}
fn aw_set_ts_ntz(&mut self, idx: usize, millis: i64, nanos: i32) -> Result<(), String> {
self.ensure_writable(idx)?;
let precision = match &self.element_type {
fcore::metadata::DataType::Timestamp(ts) => ts.precision(),
_ => {
return Err(format!(
"ArrayWriter type mismatch: expected TIMESTAMP element, got {}",
self.element_type
));
}
};
let ts = fcore::row::TimestampNtz::from_millis_nanos(millis, nanos)
.map_err(|e| e.to_string())?;
self.writer_mut()?.write_timestamp_ntz(idx, &ts, precision);
Ok(())
}
fn aw_set_ts_ltz(&mut self, idx: usize, millis: i64, nanos: i32) -> Result<(), String> {
self.ensure_writable(idx)?;
let precision = match &self.element_type {
fcore::metadata::DataType::TimestampLTz(ts) => ts.precision(),
_ => {
return Err(format!(
"ArrayWriter type mismatch: expected TIMESTAMP_LTZ element, got {}",
self.element_type
));
}
};
let ts = fcore::row::TimestampLtz::from_millis_nanos(millis, nanos)
.map_err(|e| e.to_string())?;
self.writer_mut()?.write_timestamp_ltz(idx, &ts, precision);
Ok(())
}
fn aw_set_decimal_str(&mut self, idx: usize, val: &str) -> Result<(), String> {
self.ensure_writable(idx)?;
let (precision, scale) = match &self.element_type {
fcore::metadata::DataType::Decimal(d) => (d.precision(), d.scale()),
_ => {
return Err(format!(
"ArrayWriter type mismatch: expected DECIMAL element, got {}",
self.element_type
));
}
};
let bd = bigdecimal::BigDecimal::from_str(val).map_err(|e| e.to_string())?;
let decimal = fcore::row::Decimal::from_big_decimal(bd, precision, scale)
.map_err(|e| e.to_string())?;
self.writer_mut()?.write_decimal(idx, &decimal, precision);
Ok(())
}
fn aw_set_array(&mut self, idx: usize, nested: &mut ArrayWriterInner) -> Result<(), String> {
self.ensure_writable(idx)?;
let expected_inner = match &self.element_type {
fcore::metadata::DataType::Array(at) => at.get_element_type(),
_ => {
return Err(format!(
"ArrayWriter type mismatch: expected ARRAY element, got {}",
self.element_type
));
}
};
if !structurally_compatible(expected_inner, &nested.element_type) {
return Err(format!(
"Nested ArrayWriter type mismatch: expected nested element type {}, got {}",
expected_inner, nested.element_type
));
}
nested.complete_if_needed()?;
let arr = nested.completed.as_ref().ok_or_else(|| {
"ArrayWriter invariant violation: nested completed array missing after finalize"
.to_string()
})?;
self.writer_mut()?.write_array(idx, arr);
Ok(())
}
fn aw_set_row(&mut self, idx: usize, row: &mut GenericRowInner) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::Row(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected ROW element, got {}",
self.element_type
));
}
self.writer_mut()?
.write_row(idx, &row.row)
.map_err(|e| e.to_string())
}
fn aw_set_map(&mut self, idx: usize, map: &mut MapWriterInner) -> Result<(), String> {
self.ensure_writable(idx)?;
if !matches!(self.element_type, fcore::metadata::DataType::Map(_)) {
return Err(format!(
"ArrayWriter type mismatch: expected MAP element, got {}",
self.element_type
));
}
map.complete_if_needed()?;
let completed = map.completed.as_ref().ok_or_else(|| {
"ArrayWriter invariant violation: nested completed map missing after finalize"
.to_string()
})?;
self.writer_mut()?.write_map(idx, completed);
Ok(())
}
}
/// Structural type equivalence that ignores nullability flags but preserves
/// variant and precision/scale semantics. Used to compare ArrayWriter element
/// types on the binding boundary. Nullability is ignored in structural comparison
/// because the Rust-side element type is always reconstructed as nullable
/// (encoding doesn't depend on it).
fn structurally_compatible(a: &fcore::metadata::DataType, b: &fcore::metadata::DataType) -> bool {
match (a, b) {
(DataType::Boolean(_), DataType::Boolean(_))
| (DataType::TinyInt(_), DataType::TinyInt(_))
| (DataType::SmallInt(_), DataType::SmallInt(_))
| (DataType::Int(_), DataType::Int(_))
| (DataType::BigInt(_), DataType::BigInt(_))
| (DataType::Float(_), DataType::Float(_))
| (DataType::Double(_), DataType::Double(_))
| (DataType::String(_), DataType::String(_))
| (DataType::Bytes(_), DataType::Bytes(_))
| (DataType::Date(_), DataType::Date(_))
| (DataType::Time(_), DataType::Time(_)) => true,
(DataType::Timestamp(x), DataType::Timestamp(y)) => x.precision() == y.precision(),
(DataType::TimestampLTz(x), DataType::TimestampLTz(y)) => x.precision() == y.precision(),
(DataType::Char(x), DataType::Char(y)) => x.length() == y.length(),
(DataType::Binary(x), DataType::Binary(y)) => x.length() == y.length(),
(DataType::Decimal(x), DataType::Decimal(y)) => {
x.precision() == y.precision() && x.scale() == y.scale()
}
(DataType::Array(x), DataType::Array(y)) => {
structurally_compatible(x.get_element_type(), y.get_element_type())
}
_ => false,
}
}