mod error; mod utils; pub use crate::error::deltalake_error_message; use std::ffi::CString; use std::future::IntoFuture; use std::os::raw::{c_char, c_int, c_long, c_void}; use std::ptr; use std::str::FromStr; use std::sync::Arc; #[cfg(all(feature = "jemalloc", not(target_env = "msvc")))] #[global_allocator] static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; use deltalake::kernel::{DataType, PrimitiveType, StructField}; use deltalake::operations::optimize::OptimizeType; use deltalake::operations::write::WriteBuilder; use deltalake::protocol::SaveMode; use deltalake::table::config::TableProperty; use deltalake::DeltaTable; use serde_json; use tokio::runtime::Runtime; // Build Arrow Schema use deltalake::arrow::array::*; use deltalake::arrow::datatypes::DataType as ArrowDataType; use deltalake::arrow::datatypes::{Field, Schema}; use deltalake::arrow::datatypes::{Int8Type, Int16Type, Int32Type, Int64Type}; use deltalake::arrow::datatypes::{UInt8Type, UInt16Type, UInt32Type, UInt64Type}; use deltalake::arrow::record_batch::RecordBatch; use crate::error::{CResult, DeltaGoError}; use crate::utils::{c_str_to_string, path_to_url, string_to_c_str}; // Global runtime for async operations lazy_static::lazy_static! { static ref RUNTIME: Runtime = Runtime::new().unwrap(); } /// Initialize the Delta Lake library #[no_mangle] pub extern "C" fn deltalake_init() { env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("warn")).init(); deltalake::aws::register_handlers(None); deltalake::azure::register_handlers(None); deltalake::gcp::register_handlers(None); deltalake::unity_catalog::register_handlers(None); } /// Get the version of the delta-rs core library #[no_mangle] pub extern "C" fn deltalake_version() -> *const c_char { string_to_c_str(deltalake::crate_version().to_string()) } /// Create a new Delta Lake table #[no_mangle] pub extern "C" fn deltalake_create_table( table_path: *const c_char, save_mode: *const c_char, columns_json: *const c_char, partition_columns_json: *const c_char, table_name: *const c_char, storage_options_json: *const c_char, table_configuration_json: *const c_char, ) -> CResult<*mut c_void> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; // Convert C strings to Rust strings let path = match c_str_to_string(table_path) { Ok(p) => p, Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } }; let mode_str = match c_str_to_string(save_mode) { Ok(m) => m, Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } }; let columns_json_str = match c_str_to_string(columns_json) { Ok(c) => c, Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } }; // Parse partition columns JSON let partition_columns = if partition_columns_json.is_null() { Vec::new() } else { match c_str_to_string(partition_columns_json) { Ok(json_str) => { if json_str.is_empty() { Vec::new() } else { match serde_json::from_str::>(&json_str) { Ok(columns) => columns, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } } } } Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; let name = if table_name.is_null() { None } else { match c_str_to_string(table_name) { Ok(n) => Some(n), Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; // Parse storage options let storage_options = if storage_options_json.is_null() { std::collections::HashMap::new() } else { match c_str_to_string(storage_options_json) { Ok(json_str) => { if json_str.is_empty() { std::collections::HashMap::new() } else { match serde_json::from_str::>( &json_str, ) { Ok(options) => options, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } } } } Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; // Parse table configuration properties let table_configuration = if table_configuration_json.is_null() { std::collections::HashMap::new() } else { match c_str_to_string(table_configuration_json) { Ok(json_str) => { if json_str.is_empty() { std::collections::HashMap::new() } else { match serde_json::from_str::>( &json_str, ) { Ok(config) => config, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } } } } Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; // Parse save mode let save_mode = match mode_str.as_str() { "append" => SaveMode::Append, "overwrite" => SaveMode::Overwrite, "error" => SaveMode::ErrorIfExists, "ignore" => SaveMode::Ignore, _ => { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_configuration(format!( "Invalid save mode: {}", mode_str )))); return result; } }; // Parse columns JSON let columns: Vec = match serde_json::from_str(&columns_json_str) { Ok(c) => c, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } }; // Convert JSON to StructField let mut struct_fields = Vec::new(); for col in columns { let field_name = match col.get("name").and_then(|v| v.as_str()) { Some(name) => name.to_string(), None => { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_schema( "Column must have a 'name' field".to_string(), ))); return result; } }; let field_type_str = match col.get("type").and_then(|v| v.as_str()) { Some(t) => t, None => { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_schema( "Column must have a 'type' field".to_string(), ))); return result; } }; let nullable = col .get("nullable") .and_then(|v| v.as_bool()) .unwrap_or(true); let data_type = match field_type_str { "integer" => DataType::Primitive(PrimitiveType::Integer), "long" => DataType::Primitive(PrimitiveType::Long), "string" => DataType::Primitive(PrimitiveType::String), "boolean" => DataType::Primitive(PrimitiveType::Boolean), "float" => DataType::Primitive(PrimitiveType::Float), "double" => DataType::Primitive(PrimitiveType::Double), "date" => DataType::Primitive(PrimitiveType::Date), "timestamp" => DataType::Primitive(PrimitiveType::Timestamp), _ => { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_schema(format!( "Unsupported data type: {}", field_type_str )))); return result; } }; struct_fields.push(StructField::new(field_name, data_type, nullable)); } // Create Delta table let table_result = RUNTIME.block_on(async { let url = path_to_url(&path).map_err(|e| deltalake::errors::DeltaTableError::Generic(e))?; let delta_table = if storage_options.is_empty() { DeltaTable::try_from_url(url).await? } else { DeltaTable::try_from_url_with_storage_options(url, storage_options).await? }; let mut create_builder = delta_table.create().with_save_mode(save_mode); if let Some(name) = name { create_builder = create_builder.with_table_name(&name); } // Add partition columns if provided if !partition_columns.is_empty() { create_builder = create_builder.with_partition_columns(partition_columns); } // Apply table configuration properties for (key, value) in table_configuration { match TableProperty::from_str(&key) { Ok(prop) => { create_builder = create_builder.with_configuration_property(prop, Some(value.as_str())); } Err(_) => { // If unknown property, skip it with a warning // In production, you might want to log this eprintln!("Warning: Unknown table property '{}', skipping", key); } } } create_builder.with_columns(struct_fields).await }); match table_result { Ok(table) => { result.success = true; result.value = Box::into_raw(Box::new(table)) as *mut c_void; } Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); } } result } /// Open an existing Delta Lake table #[no_mangle] pub extern "C" fn deltalake_open_table( table_path: *const c_char, storage_options_json: *const c_char, ) -> CResult<*mut c_void> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; // Convert table path from C string let path = match c_str_to_string(table_path) { Ok(p) => p, Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } }; // Validate path is not empty if path.trim().is_empty() { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_uri( "Table path cannot be empty".to_string(), ))); return result; } // Parse storage options let storage_options = if storage_options_json.is_null() { std::collections::HashMap::new() } else { match c_str_to_string(storage_options_json) { Ok(json_str) => { match serde_json::from_str::>(&json_str) { Ok(options) => options, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } } } Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; // Open the Delta table with proper error handling using catch_unwind let table_result = std::panic::catch_unwind(|| { RUNTIME.block_on(async { // First, try to create the DeltaOps instance let url = match path_to_url(&path) { Ok(u) => u, Err(e) => return Err(deltalake::errors::DeltaTableError::Generic(e)), }; let delta_table = if storage_options.is_empty() { match DeltaTable::try_from_url(url).await { Ok(table) => table, Err(e) => { // Handle specific errors more gracefully match &e { deltalake::errors::DeltaTableError::NotATable(_) => { return Err(deltalake::errors::DeltaTableError::NotATable(format!( "No Delta Lake table found at path: {}. The path either doesn't exist or is not a valid Delta Lake table.", path ))); } deltalake::errors::DeltaTableError::ObjectStore { .. } => { return Err(deltalake::errors::DeltaTableError::NotATable(format!( "Cannot access path: {}. The directory or file does not exist or is not accessible.", path ))); } _ => return Err(e), } } } } else { let url = match path_to_url(&path) { Ok(u) => u, Err(e) => return Err(deltalake::errors::DeltaTableError::Generic(e)), }; match DeltaTable::try_from_url_with_storage_options(url, storage_options).await { Ok(ops) => ops, Err(e) => { // Handle specific errors more gracefully match &e { deltalake::errors::DeltaTableError::NotATable(_) => { return Err(deltalake::errors::DeltaTableError::NotATable(format!( "No Delta Lake table found at path: {}. The path either doesn't exist or is not a valid Delta Lake table.", path ))); } deltalake::errors::DeltaTableError::ObjectStore { .. } => { return Err(deltalake::errors::DeltaTableError::NotATable(format!( "Cannot access path: {}. The directory or file does not exist or is not accessible.", path ))); } _ => return Err(e), } } } }; // IMPORTANT: DeltaTable::try_from_url returns Ok(table) even when the path is not a Delta table! // This is by design - it allows creating new tables at uninitialized locations. // When load() encounters NotATable error internally, try_from_url still returns Ok // but the table will have version() == None. We must check this to detect invalid tables. if delta_table.version().is_none() { return Err(deltalake::errors::DeltaTableError::NotATable(format!( "Path exists but is not a valid Delta Lake table: {}. Missing _delta_log directory or corrupted metadata.", path ))); } Ok(delta_table) }) }); let final_result = match table_result { Ok(async_result) => async_result, Err(_panic) => { // If a panic occurred, return a proper error Err(deltalake::errors::DeltaTableError::NotATable(format!( "Failed to open Delta Lake table at path: {}. The path may not exist or is not a valid Delta Lake table.", path ))) } }; match final_result { Ok(table) => { result.success = true; result.value = Box::into_raw(Box::new(table)) as *mut c_void; } Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); } } result } /// Open an existing Delta table with configuration options #[no_mangle] pub extern "C" fn deltalake_open_table_with_config( table_path: *const c_char, storage_options_json: *const c_char, require_files: c_int, log_buffer_size: c_int, ) -> CResult<*mut c_void> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; // Convert table path from C string let path = match c_str_to_string(table_path) { Ok(p) => p, Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } }; // Validate path is not empty if path.trim().is_empty() { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_uri( "Table path cannot be empty".to_string(), ))); return result; } // Parse storage options let storage_options = if storage_options_json.is_null() { std::collections::HashMap::new() } else { match c_str_to_string(storage_options_json) { Ok(json_str) => { match serde_json::from_str::>(&json_str) { Ok(options) => options, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } } } Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; // Build table with configuration let table_result = RUNTIME.block_on(async { use deltalake::table::builder::DeltaTableBuilder; // Create builder - from_url now returns Result let url = match path_to_url(&path) { Ok(u) => u, Err(e) => return Err(deltalake::errors::DeltaTableError::Generic(e)), }; let mut builder = DeltaTableBuilder::from_url(url)?; // Apply storage options if provided if !storage_options.is_empty() { builder = builder.with_storage_options(storage_options); } // Apply require_files configuration if require_files == 0 { builder = builder.without_files(); } // Apply log_buffer_size configuration if log_buffer_size > 0 { // with_log_buffer_size returns Result builder = builder.with_log_buffer_size(log_buffer_size as usize)?; } // Load the table builder.load().await }); match table_result { Ok(table) => { result.success = true; result.value = Box::into_raw(Box::new(table)) as *mut c_void; } Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); } } result } /// Check if a Delta table exists at the given path #[no_mangle] pub extern "C" fn deltalake_table_exists( table_path: *const c_char, storage_options_json: *const c_char, ) -> CResult { let mut result = CResult { success: false, value: false, error: std::ptr::null_mut(), }; // Convert table path from C string let path = match c_str_to_string(table_path) { Ok(p) => p, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } }; // Validate path is not empty if path.trim().is_empty() { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_uri( "Table path cannot be empty".to_string(), ))); return result; } // Parse storage options let storage_options = if storage_options_json.is_null() { std::collections::HashMap::new() } else { match c_str_to_string(storage_options_json) { Ok(json_str) => { if json_str.trim().is_empty() { std::collections::HashMap::new() } else { match serde_json::from_str::>( &json_str, ) { Ok(options) => options, Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } } } } Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); return result; } } }; // Check table existence using DeltaTableBuilder and verify_deltatable_existence let existence_result = std::panic::catch_unwind(|| { RUNTIME.block_on(async { use deltalake::DeltaTableBuilder; let url = match path_to_url(&path) { Ok(u) => u, Err(_e) => return Ok(false), // Invalid URL means table doesn't exist }; let builder_result = DeltaTableBuilder::from_url(url); let builder = match builder_result { Ok(b) => { if storage_options.is_empty() { b } else { b.with_storage_options(storage_options) } } Err(_) => return Ok(false), }; match builder.build() { Ok(table) => { // Use verify_deltatable_existence to check if table exists match table.verify_deltatable_existence().await { Ok(exists) => Ok(exists), Err(e) => { // If verification fails, assume table doesn't exist // Some errors like "table not found" should return false, not error let error_msg = e.to_string().to_lowercase(); if error_msg.contains("not found") || error_msg.contains("does not exist") || error_msg.contains("no such file") || error_msg.contains("notfound") { Ok(false) } else { Err(e) } } } } Err(e) => { // If we can't build the table, check if it's a "not found" error let error_msg = e.to_string().to_lowercase(); if error_msg.contains("not found") || error_msg.contains("does not exist") || error_msg.contains("no such file") || error_msg.contains("notfound") { Ok(false) } else { Err(e) } } } }) }); let final_result = match existence_result { Ok(async_result) => async_result, Err(_panic) => { result.error = Box::into_raw(Box::new(DeltaGoError::delta_table_error( "Panic occurred while checking table existence".to_string(), ))); return result; } }; match final_result { Ok(exists) => { result.success = true; result.value = exists; } Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); } } result } /// Get the version of a Delta table #[no_mangle] pub extern "C" fn deltalake_table_version(table: *const c_void) -> CResult { let mut result = CResult { success: false, value: -1, error: std::ptr::null_mut(), }; if table.is_null() { result.error = Box::into_raw(Box::new(DeltaGoError::null_pointer( "Table pointer is null".to_string(), ))); return result; } unsafe { let delta_table = &*(table as *const DeltaTable); result.success = true; result.value = delta_table.version().unwrap_or(0) as c_long; } result } /// Get the files of a Delta table as a JSON string #[no_mangle] pub extern "C" fn deltalake_table_files(table: *const c_void) -> CResult<*mut c_char> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; if table.is_null() { result.error = Box::into_raw(Box::new(DeltaGoError::null_pointer( "Table pointer is null".to_string(), ))); return result; } unsafe { let delta_table = &*(table as *const DeltaTable); // In delta-rs 0.29, get file URIs from the table let files_result = RUNTIME.block_on(async { match delta_table.get_file_uris() { Ok(uris) => { let files: Vec = uris.collect(); Ok(files) } Err(e) => Err(DeltaGoError::from(e)), } }); match files_result { Ok(file_list) => match serde_json::to_string(&file_list) { Ok(json_str) => { result.success = true; result.value = string_to_c_str(json_str) as *mut c_char; } Err(e) => { result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); } }, Err(e) => { result.error = Box::into_raw(Box::new(e)); } } } result } /// Free memory allocated for strings #[no_mangle] pub extern "C" fn deltalake_free_string(ptr: *mut c_char) { if !ptr.is_null() { unsafe { let _ = CString::from_raw(ptr); } } } /// Free error #[no_mangle] pub extern "C" fn deltalake_free_error(error: *mut DeltaGoError) { if !error.is_null() { unsafe { let _ = Box::from_raw(error); } } } /// Write data to a Delta table /// /// CRITICAL MEMORY MANAGEMENT: /// - Takes ownership of the input table pointer immediately /// - On early errors (before table consumed): Returns original pointer in result.value /// - On late errors (after table consumed): Returns null in result.value /// - On success: Returns new table pointer in result.value /// - Caller MUST use the returned pointer and ignore the input pointer after this call #[no_mangle] pub extern "C" fn deltalake_write_table( table: *const c_void, records_json: *const c_char, save_mode: *const c_char, ) -> CResult<*mut c_void> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; if table.is_null() { result.error = Box::into_raw(Box::new(DeltaGoError::null_pointer( "Table pointer is null".to_string(), ))); return result; } // STEP 1: Take ownership immediately to prevent memory leak // From this point, we own the pointer and MUST return it (or null if consumed) let old_table = unsafe { Box::from_raw(table as *mut DeltaTable) }; // STEP 2: Validate inputs - if error, return the original pointer back // Convert save mode from C string let mode_str = match c_str_to_string(save_mode) { Ok(m) => m, Err(e) => { // Error before consuming: return old pointer back to caller result.value = Box::into_raw(old_table) as *mut c_void; result.error = Box::into_raw(Box::new(e)); return result; } }; // Convert records data from C string let records_json_str = match c_str_to_string(records_json) { Ok(j) => j, Err(e) => { // Error before consuming: return old pointer back to caller result.value = Box::into_raw(old_table) as *mut c_void; result.error = Box::into_raw(Box::new(e)); return result; } }; // Parse save mode let save_mode = match mode_str.as_str() { "append" => SaveMode::Append, "overwrite" => SaveMode::Overwrite, "error" => SaveMode::ErrorIfExists, "ignore" => SaveMode::Ignore, _ => { // Error before consuming: return old pointer back to caller result.value = Box::into_raw(old_table) as *mut c_void; result.error = Box::into_raw(Box::new(DeltaGoError::invalid_configuration(format!( "Invalid save mode: {}", mode_str )))); return result; } }; // STEP 3: Parse and validate data BEFORE consuming old_table // If parsing fails, we can still return the original pointer let batch_result = RUNTIME.block_on(async { // Parse JSON records data let records_data: serde_json::Value = serde_json::from_str(&records_json_str).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to parse records JSON: {}", e)) })?; // Extract schema and data from JSON let schema_json = records_data.get("schema").ok_or_else(|| { DeltaGoError::invalid_schema("Missing 'schema' field in records data".to_string()) })?; let data_json = records_data.get("data").ok_or_else(|| { DeltaGoError::invalid_schema("Missing 'data' field in records data".to_string()) })?; // Parse schema let schema_array = schema_json .as_array() .ok_or_else(|| DeltaGoError::invalid_schema("Schema must be an array".to_string()))?; // Build Arrow Schema let mut fields = Vec::new(); for field_json in schema_array { let name = field_json .get("name") .and_then(|v| v.as_str()) .ok_or_else(|| { DeltaGoError::invalid_schema("Field must have a name".to_string()) })?; let type_str = field_json .get("type") .and_then(|v| v.as_str()) .ok_or_else(|| { DeltaGoError::invalid_schema("Field must have a type".to_string()) })?; let nullable = field_json .get("nullable") .and_then(|v| v.as_bool()) .unwrap_or(true); let arrow_type = match type_str { "integer" => ArrowDataType::Int32, "long" => ArrowDataType::Int64, "string" => ArrowDataType::Utf8, "boolean" => ArrowDataType::Boolean, "float" => ArrowDataType::Float32, "double" => ArrowDataType::Float64, _ => { return Err(DeltaGoError::invalid_schema(format!( "Unsupported data type: {}", type_str ))); } }; fields.push(Field::new(name, arrow_type, nullable)); } let schema = Arc::new(Schema::new(fields)); // Parse data rows let data_array = data_json .as_array() .ok_or_else(|| DeltaGoError::invalid_schema("Data must be an array".to_string()))?; // Build columns based on schema let mut columns: Vec> = Vec::new(); for (field_idx, field) in schema.fields().iter().enumerate() { match field.data_type() { ArrowDataType::Int32 => { let mut values = Vec::new(); for row in data_array { let row_array = row.as_array().ok_or_else(|| { DeltaGoError::invalid_schema( "Each data row must be an array".to_string(), ) })?; if field_idx < row_array.len() { if let Some(val) = row_array[field_idx].as_i64() { values.push(val as i32); } else { return Err(DeltaGoError::invalid_schema( "Invalid integer value".to_string(), )); } } else { return Err(DeltaGoError::invalid_schema( "Row data missing for field".to_string(), )); } } columns.push(Arc::new(Int32Array::from(values))); } ArrowDataType::Int64 => { let mut values = Vec::new(); for row in data_array { let row_array = row.as_array().ok_or_else(|| { DeltaGoError::invalid_schema( "Each data row must be an array".to_string(), ) })?; if field_idx < row_array.len() { if let Some(val) = row_array[field_idx].as_i64() { values.push(val); } else { return Err(DeltaGoError::invalid_schema( "Invalid long value".to_string(), )); } } else { return Err(DeltaGoError::invalid_schema( "Row data missing for field".to_string(), )); } } columns.push(Arc::new(Int64Array::from(values))); } ArrowDataType::Utf8 => { let mut values = Vec::new(); for row in data_array { let row_array = row.as_array().ok_or_else(|| { DeltaGoError::invalid_schema( "Each data row must be an array".to_string(), ) })?; if field_idx < row_array.len() { if let Some(val) = row_array[field_idx].as_str() { values.push(val); } else { return Err(DeltaGoError::invalid_schema( "Invalid string value".to_string(), )); } } else { return Err(DeltaGoError::invalid_schema( "Row data missing for field".to_string(), )); } } columns.push(Arc::new(StringArray::from(values))); } ArrowDataType::Boolean => { let mut values = Vec::new(); for row in data_array { let row_array = row.as_array().ok_or_else(|| { DeltaGoError::invalid_schema( "Each data row must be an array".to_string(), ) })?; if field_idx < row_array.len() { if let Some(val) = row_array[field_idx].as_bool() { values.push(val); } else { return Err(DeltaGoError::invalid_schema( "Invalid boolean value".to_string(), )); } } else { return Err(DeltaGoError::invalid_schema( "Row data missing for field".to_string(), )); } } columns.push(Arc::new(BooleanArray::from(values))); } ArrowDataType::Float32 => { let mut values = Vec::new(); for row in data_array { let row_array = row.as_array().ok_or_else(|| { DeltaGoError::invalid_schema( "Each data row must be an array".to_string(), ) })?; if field_idx < row_array.len() { if let Some(val) = row_array[field_idx].as_f64() { values.push(val as f32); } else { return Err(DeltaGoError::invalid_schema( "Invalid float value".to_string(), )); } } else { return Err(DeltaGoError::invalid_schema( "Row data missing for field".to_string(), )); } } columns.push(Arc::new(Float32Array::from(values))); } ArrowDataType::Float64 => { let mut values = Vec::new(); for row in data_array { let row_array = row.as_array().ok_or_else(|| { DeltaGoError::invalid_schema( "Each data row must be an array".to_string(), ) })?; if field_idx < row_array.len() { if let Some(val) = row_array[field_idx].as_f64() { values.push(val); } else { return Err(DeltaGoError::invalid_schema( "Invalid double value".to_string(), )); } } else { return Err(DeltaGoError::invalid_schema( "Row data missing for field".to_string(), )); } } columns.push(Arc::new(Float64Array::from(values))); } _ => { return Err(DeltaGoError::invalid_schema(format!( "Unsupported data type in write: {:?}", field.data_type() ))); } } } // Create RecordBatch let batch = RecordBatch::try_new(schema, columns).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to create record batch: {}", e)) })?; Ok::(batch) }); // Check if batch creation succeeded let batch = match batch_result { Ok(b) => b, Err(e) => { // Error during parsing/validation: old_table not consumed yet, return it result.value = Box::into_raw(old_table) as *mut c_void; result.error = Box::into_raw(Box::new(e)); return result; } }; // STEP 4: Build write operation using WriteBuilder to mirror official bindings // Capture the current log store and snapshot before consuming the table handle. let log_store = old_table.log_store(); let snapshot = old_table .state .as_ref() .map(|state| state.snapshot().clone()); let builder = WriteBuilder::new(log_store, snapshot) .with_save_mode(save_mode) .with_input_batches(vec![batch]); // old_table is now fully consumed; drop it to release memory before the async write executes. drop(old_table); let write_result = RUNTIME .block_on(builder.into_future()) .map_err(DeltaGoError::from); // STEP 5: Handle write result match write_result { Ok(new_table) => { // Success: old_table was consumed, return new table result.success = true; result.value = Box::into_raw(Box::new(new_table)) as *mut c_void; } Err(e) => { // Error after consumption: old_table is gone, return null // This signals to Go that the pointer was consumed and should not be used result.success = false; result.value = std::ptr::null_mut(); result.error = Box::into_raw(Box::new(e)); } } result } /// Read data from a Delta table and return as JSON #[no_mangle] pub extern "C" fn deltalake_read_table_data( table: *const c_void, limit: c_long, ) -> CResult<*mut c_char> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; if table.is_null() { result.error = Box::into_raw(Box::new(DeltaGoError::null_pointer( "Table pointer is null".to_string(), ))); return result; } unsafe { let delta_table = &*(table as *const DeltaTable); let read_result = RUNTIME.block_on(async { use deltalake::arrow::json; use deltalake::datafusion::prelude::*; // Create DataFusion context let ctx = SessionContext::new(); // Register the Delta table ctx.register_table("delta_table", Arc::new(delta_table.clone())) .map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to register table: {}", e)) })?; // Build query with optional limit let query = if limit > 0 { format!("SELECT * FROM delta_table LIMIT {}", limit) } else { "SELECT * FROM delta_table".to_string() }; // Execute query let df = ctx.sql(&query).await.map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to execute query: {}", e)) })?; // Collect results let batches = df.collect().await.map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to collect results: {}", e)) })?; // Convert to JSON if batches.is_empty() { return Ok("[]".to_string()); } let mut json_rows = Vec::new(); for batch in batches { // Convert RecordBatch to JSON let mut buf = Vec::new(); let mut writer = json::ArrayWriter::new(&mut buf); writer.write(&batch).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to write JSON: {}", e)) })?; writer.finish().map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to finish JSON write: {}", e)) })?; let json_str = String::from_utf8(buf).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to parse UTF-8: {}", e)) })?; // Parse and extract rows let json_array: serde_json::Value = serde_json::from_str(&json_str).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to parse JSON: {}", e)) })?; if let serde_json::Value::Array(rows) = json_array { json_rows.extend(rows); } } // Return as JSON array let result_json = serde_json::to_string(&json_rows).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to serialize result: {}", e)) })?; Ok(result_json) }); match read_result { Ok(json_data) => { result.success = true; result.value = string_to_c_str(json_data) as *mut c_char; } Err(e) => { result.error = Box::into_raw(Box::new(e)); } } } result } /// Execute an arbitrary SQL query against a Delta table and return the results as JSON. #[no_mangle] pub extern "C" fn deltalake_query_table( table: *const c_void, sql_query: *const c_char, table_alias: *const c_char, ) -> CResult<*mut c_char> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; if table.is_null() { result.error = Box::into_raw(Box::new(DeltaGoError::null_pointer( "Table pointer is null".to_string(), ))); return result; } if sql_query.is_null() { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_configuration( "SQL query cannot be null".to_string(), ))); return result; } let sql = match c_str_to_string(sql_query) { Ok(query) => query, Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } }; if sql.trim().is_empty() { result.error = Box::into_raw(Box::new(DeltaGoError::invalid_configuration( "SQL query cannot be empty".to_string(), ))); return result; } let alias = if table_alias.is_null() { "delta_table".to_string() } else { match c_str_to_string(table_alias) { Ok(alias) => { let trimmed = alias.trim(); if trimmed.is_empty() { "delta_table".to_string() } else { trimmed.to_string() } } Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; unsafe { let delta_table = &*(table as *const DeltaTable); let query_result = RUNTIME.block_on(async { use deltalake::datafusion::prelude::*; let ctx = SessionContext::new(); let table_arc = Arc::new(delta_table.clone()); ctx.register_table(&alias, table_arc.clone()).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to register table: {}", e)) })?; if alias != "delta_table" { ctx.register_table("delta_table", Arc::new(delta_table.clone())) .map_err(|e| { DeltaGoError::invalid_schema(format!( "Failed to register default table alias: {}", e )) })?; } let df = ctx.sql(&sql).await.map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to execute query: {}", e)) })?; let batches = df.collect().await.map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to collect results: {}", e)) })?; if batches.is_empty() { return Ok("[]".to_string()); } let mut json_rows = Vec::new(); for batch in batches { // Convert all values to strings to preserve precision let schema = batch.schema(); let num_rows = batch.num_rows(); for row_idx in 0..num_rows { let mut row_map = serde_json::Map::new(); for (col_idx, field) in schema.fields().iter().enumerate() { let column = batch.column(col_idx); let field_name = field.name().clone(); // Convert value to string based on type let value_str = if column.is_null(row_idx) { serde_json::Value::Null } else { use deltalake::arrow::array::*; use deltalake::arrow::datatypes::DataType as ArrowDataType; let str_value = match field.data_type() { ArrowDataType::Int8 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Int16 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Int32 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Int64 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::UInt8 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::UInt16 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::UInt32 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::UInt64 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Float32 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Float64 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Utf8 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::LargeUtf8 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Boolean => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Date32 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Date64 => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Timestamp(_, _) => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) .or_else(|| column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string())) .or_else(|| column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string())) .or_else(|| column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string())) }, ArrowDataType::Decimal128(_, _) => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Decimal256(_, _) => { column.as_any().downcast_ref::() .map(|arr| arr.value(row_idx).to_string()) }, ArrowDataType::Dictionary(key_type, value_type) => { // Handle DictionaryArray by extracting the actual value // Dictionary arrays store indices (keys) that reference actual values use deltalake::arrow::array::cast::AsArray; // Try to handle common dictionary combinations match (key_type.as_ref(), value_type.as_ref()) { (ArrowDataType::UInt8, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, (ArrowDataType::UInt16, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, (ArrowDataType::UInt32, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, (ArrowDataType::UInt64, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, (ArrowDataType::Int8, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, (ArrowDataType::Int16, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, (ArrowDataType::Int32, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, (ArrowDataType::Int64, ArrowDataType::Utf8) => { column.as_any().downcast_ref::>() .and_then(|dict_arr| { if dict_arr.is_valid(row_idx) { dict_arr.values().as_string::() .value(dict_arr.keys().value(row_idx) as usize) .to_string().into() } else { None } }) }, _ => { // For other dictionary types, fall back to debug format Some(format!("dict({:?}->{:?}):idx={}", key_type, value_type, row_idx)) } } }, _ => { // For unsupported types, use Arrow's display format Some(format!("{:?}", column.slice(row_idx, 1))) } }; match str_value { Some(s) => serde_json::Value::String(s), None => serde_json::Value::String(format!("unsupported_type_{}", field.data_type())), } }; row_map.insert(field_name, value_str); } json_rows.push(serde_json::Value::Object(row_map)); } } let result_json = serde_json::to_string(&json_rows).map_err(|e| { DeltaGoError::invalid_schema(format!("Failed to serialize result: {}", e)) })?; Ok(result_json) }); match query_result { Ok(json_data) => { result.success = true; result.value = string_to_c_str(json_data) as *mut c_char; } Err(e) => { result.error = Box::into_raw(Box::new(e)); } } } result } #[repr(C)] pub struct DeleteResult { table_handle: *mut c_void, metrics_json: *mut c_char, } /// Delete rows from a Delta table using an optional predicate #[no_mangle] pub extern "C" fn deltalake_delete_where( table_handle: *const c_void, predicate: *const c_char, ) -> CResult<*mut DeleteResult> { let mut result = CResult { success: false, value: std::ptr::null_mut(), error: std::ptr::null_mut(), }; if table_handle.is_null() { result.error = Box::into_raw(Box::new(DeltaGoError::null_pointer( "Table pointer is null".to_string(), ))); return result; } let predicate_string = if predicate.is_null() { None } else { match c_str_to_string(predicate) { Ok(p) => Some(p), Err(e) => { result.error = Box::into_raw(Box::new(e)); return result; } } }; unsafe { let table = &*(table_handle as *const DeltaTable); let delete_result: Result<(*mut DeltaTable, String), DeltaGoError> = RUNTIME.block_on(async { let mut delete_builder = table.clone().delete(); if let Some(ref predicate) = predicate_string { delete_builder = delete_builder.with_predicate(predicate.as_str()); } match delete_builder.await { Ok((updated_table, metrics)) => { let metrics_json = serde_json::to_string(&metrics).map_err(DeltaGoError::from)?; let table_ptr = Box::into_raw(Box::new(updated_table)); Ok((table_ptr, metrics_json)) } Err(e) => Err(DeltaGoError::from(e)), } }); match delete_result { Ok((table_ptr, metrics_json)) => match CString::new(metrics_json) { Ok(metrics_c_string) => { let delete_result = DeleteResult { table_handle: table_ptr as *mut c_void, metrics_json: metrics_c_string.into_raw(), }; result.success = true; result.value = Box::into_raw(Box::new(delete_result)); } Err(e) => { let _ = Box::from_raw(table_ptr); result.error = Box::into_raw(Box::new(DeltaGoError::from(e))); } }, Err(e) => { result.error = Box::into_raw(Box::new(e)); } } } result } /// Free delete result handle /// Note: This does NOT free the table_handle - ownership is transferred to the caller #[no_mangle] pub extern "C" fn deltalake_free_delete_result(result: *mut c_void) { if !result.is_null() { unsafe { let boxed_result = Box::from_raw(result as *mut DeleteResult); // Free the metrics JSON string if !boxed_result.metrics_json.is_null() { let _ = CString::from_raw(boxed_result.metrics_json); } // Do NOT free table_handle - ownership transferred to caller // The caller (Go side) is responsible for calling deltalake_free_table } } } /// Free table handle #[no_mangle] pub extern "C" fn deltalake_free_table(table: *mut c_void) { if !table.is_null() { unsafe { let _ = Box::from_raw(table as *mut DeltaTable); } } } /// Vacuum configuration #[derive(Debug)] pub struct VacuumData { table: DeltaTable, retention_hours: Option, dry_run: bool, } /// Vacuum result from C FFI #[repr(C)] pub struct VacuumResult { table_handle: *mut c_void, metrics_json: *mut c_char, } /// Optimize configuration #[derive(Debug)] pub struct OptimizeData { table: DeltaTable, optimize_type: String, // "compact" or "zorder" zorder_columns: Vec, // Only used for zorder optimization target_size: Option, max_concurrent_tasks: Option, } /// Optimize result from C FFI #[repr(C)] pub struct OptimizeResult { table_handle: *mut c_void, metrics_json: *mut c_char, } /// Create vacuum builder #[no_mangle] pub extern "C" fn deltalake_vacuum_table(table_handle: *const c_void) -> *mut c_void { if table_handle.is_null() { return ptr::null_mut(); } unsafe { let table = &*(table_handle as *const DeltaTable); let vacuum_data = VacuumData { table: table.clone(), retention_hours: None, dry_run: false, }; Box::into_raw(Box::new(vacuum_data)) as *mut c_void } } /// Set retention hours for vacuum #[no_mangle] pub extern "C" fn deltalake_vacuum_set_retention_hours(vacuum_handle: *mut c_void, hours: c_int) { if vacuum_handle.is_null() { return; } unsafe { let vacuum_data = &mut *(vacuum_handle as *mut VacuumData); vacuum_data.retention_hours = Some(hours as u64); } } /// Set dry run mode for vacuum #[no_mangle] pub extern "C" fn deltalake_vacuum_set_dry_run(vacuum_handle: *mut c_void, dry_run: c_int) { if vacuum_handle.is_null() { return; } unsafe { let vacuum_data = &mut *(vacuum_handle as *mut VacuumData); vacuum_data.dry_run = dry_run != 0; } } /// Execute vacuum operation #[no_mangle] pub extern "C" fn deltalake_vacuum_execute(vacuum_handle: *mut c_void) -> *mut VacuumResult { if vacuum_handle.is_null() { return ptr::null_mut(); } unsafe { let vacuum_data = &*(vacuum_handle as *const VacuumData); let vacuum_result: Result<(*mut DeltaTable, String), String> = RUNTIME.block_on(async { // Create vacuum operation using DeltaTable methods directly let mut vacuum_op = vacuum_data.table.clone().vacuum(); // Set retention hours if specified if let Some(hours) = vacuum_data.retention_hours { use chrono::Duration; vacuum_op = vacuum_op .with_retention_period(Duration::hours(hours as i64)) .with_enforce_retention_duration(false); } // Set dry run mode vacuum_op = vacuum_op.with_dry_run(vacuum_data.dry_run); // Set full vacuum mode to remove all files older than retention period vacuum_op = vacuum_op.with_mode(deltalake::operations::vacuum::VacuumMode::Full); // Execute vacuum match vacuum_op.await { Ok((updated_table, metrics)) => { // Create JSON response let response = serde_json::json!({ "files_deleted": metrics.files_deleted, "total_files_deleted": metrics.files_deleted.len() as i64, "dry_run": vacuum_data.dry_run }); let json_string = serde_json::to_string(&response) .map_err(|e| format!("Failed to serialize vacuum metrics: {}", e))?; // Store updated table let table_ptr = Box::into_raw(Box::new(updated_table)); Ok((table_ptr, json_string)) } Err(e) => { // If vacuum operation fails, return original table with empty metrics let response = serde_json::json!({ "files_deleted": Vec::::new(), "total_files_deleted": 0i64, "dry_run": vacuum_data.dry_run, "error": format!("Vacuum failed: {}", e) }); let json_string = serde_json::to_string(&response) .map_err(|e| format!("Failed to serialize error: {}", e))?; let table_ptr = Box::into_raw(Box::new(vacuum_data.table.clone())); Ok((table_ptr, json_string)) } } }); match vacuum_result { Ok((table_ptr, json_string)) => { let json_cstring = match CString::new(json_string) { Ok(cs) => cs.into_raw(), Err(_) => return ptr::null_mut(), }; let result = VacuumResult { table_handle: table_ptr as *mut c_void, metrics_json: json_cstring, }; Box::into_raw(Box::new(result)) } Err(_) => ptr::null_mut(), } } } /// Free vacuum builder #[no_mangle] pub extern "C" fn deltalake_free_vacuum_builder(vacuum_handle: *mut c_void) { if !vacuum_handle.is_null() { unsafe { let _ = Box::from_raw(vacuum_handle as *mut VacuumData); } } } /// Free vacuum result #[no_mangle] pub extern "C" fn deltalake_free_vacuum_result(result: *mut VacuumResult) { if !result.is_null() { unsafe { let boxed_result = Box::from_raw(result); // Free the JSON string if !boxed_result.metrics_json.is_null() { let _ = CString::from_raw(boxed_result.metrics_json); } } } } /// Create optimize builder #[no_mangle] pub extern "C" fn deltalake_optimize_table(table_handle: *const c_void) -> *mut c_void { if table_handle.is_null() { return ptr::null_mut(); } unsafe { let table = &*(table_handle as *const DeltaTable); let optimize_data = OptimizeData { table: table.clone(), optimize_type: "compact".to_string(), // Default to compact zorder_columns: Vec::new(), target_size: None, max_concurrent_tasks: None, }; Box::into_raw(Box::new(optimize_data)) as *mut c_void } } /// Set optimize type (compact or zorder) #[no_mangle] pub extern "C" fn deltalake_optimize_set_type( optimize_handle: *mut c_void, optimize_type: *const c_char, zorder_columns_json: *const c_char, ) -> bool { if optimize_handle.is_null() || optimize_type.is_null() { return false; } let opt_type = match c_str_to_string(optimize_type) { Ok(t) => t, Err(_) => return false, }; unsafe { let optimize_data = &mut *(optimize_handle as *mut OptimizeData); optimize_data.optimize_type = opt_type.clone(); // Handle Z-order columns if provided if opt_type == "zorder" && !zorder_columns_json.is_null() { let zorder_json = match c_str_to_string(zorder_columns_json) { Ok(json) => json, Err(_) => return false, }; match serde_json::from_str::>(&zorder_json) { Ok(columns) => optimize_data.zorder_columns = columns, Err(_) => return false, } } true } } /// Set target file size for optimization #[no_mangle] pub extern "C" fn deltalake_optimize_set_target_size( optimize_handle: *mut c_void, target_size: c_long, ) -> bool { if optimize_handle.is_null() { return false; } unsafe { let optimize_data = &mut *(optimize_handle as *mut OptimizeData); optimize_data.target_size = if target_size > 0 { Some(target_size as i64) } else { None }; true } } /// Set max concurrent tasks for optimization #[no_mangle] pub extern "C" fn deltalake_optimize_set_max_concurrent_tasks( optimize_handle: *mut c_void, max_tasks: c_int, ) -> bool { if optimize_handle.is_null() { return false; } unsafe { let optimize_data = &mut *(optimize_handle as *mut OptimizeData); optimize_data.max_concurrent_tasks = if max_tasks > 0 { Some(max_tasks as usize) } else { None }; true } } /// Execute optimize operation #[no_mangle] pub extern "C" fn deltalake_optimize_execute( optimize_handle: *const c_void, ) -> *mut OptimizeResult { if optimize_handle.is_null() { return ptr::null_mut(); } unsafe { let optimize_data = &*(optimize_handle as *const OptimizeData); let optimize_result = RUNTIME.block_on(async { let mut optimize_builder = optimize_data.table.clone().optimize(); // Set optimize type match optimize_data.optimize_type.as_str() { "compact" => { optimize_builder = optimize_builder.with_type(OptimizeType::Compact); } "zorder" => { if !optimize_data.zorder_columns.is_empty() { optimize_builder = optimize_builder .with_type(OptimizeType::ZOrder(optimize_data.zorder_columns.clone())); } else { return Err(deltalake::DeltaTableError::Generic( "Z-order optimization requires columns".to_string(), )); } } _ => { return Err(deltalake::DeltaTableError::Generic( "Invalid optimize type".to_string(), )); } } // Set target size if provided if let Some(target_size) = optimize_data.target_size { optimize_builder = optimize_builder.with_target_size(target_size as u64); } // Set max concurrent tasks if provided if let Some(max_tasks) = optimize_data.max_concurrent_tasks { optimize_builder = optimize_builder.with_max_concurrent_tasks(max_tasks); } // Execute the optimization match optimize_builder.await { Ok((table, metrics)) => { // Serialize metrics to JSON let json_string = match serde_json::to_string(&metrics) { Ok(json) => json, Err(_) => { return Err(deltalake::DeltaTableError::Generic( "Failed to serialize metrics".to_string(), )) } }; let table_ptr = Box::into_raw(Box::new(table)); Ok((table_ptr, json_string)) } Err(e) => Err(e), } }); match optimize_result { Ok((table_ptr, json_string)) => { let json_cstring = match CString::new(json_string) { Ok(cs) => cs.into_raw(), Err(_) => return ptr::null_mut(), }; let result = OptimizeResult { table_handle: table_ptr as *mut c_void, metrics_json: json_cstring, }; Box::into_raw(Box::new(result)) } Err(_) => ptr::null_mut(), } } } /// Free optimize builder #[no_mangle] pub extern "C" fn deltalake_free_optimize_builder(optimize_handle: *mut c_void) { if !optimize_handle.is_null() { unsafe { let _ = Box::from_raw(optimize_handle as *mut OptimizeData); } } } /// Free optimize result #[no_mangle] pub extern "C" fn deltalake_free_optimize_result(result: *mut OptimizeResult) { if !result.is_null() { unsafe { let boxed_result = Box::from_raw(result); // Free the JSON string if !boxed_result.metrics_json.is_null() { let _ = CString::from_raw(boxed_result.metrics_json); } } } } #[cfg(test)] mod tests { use super::*; use std::ffi::{CStr, CString}; use tempfile::TempDir; #[test] fn test_create_delta_table() { let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); // Create the directory for delta-rs 0.29 std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let table_name_cstr = CString::new("test_table").unwrap(); let columns_json = r#"[ {"name": "id", "type": "integer", "nullable": false}, {"name": "name", "type": "string", "nullable": true}, {"name": "age", "type": "integer", "nullable": true} ]"#; let columns_cstr = CString::new(columns_json).unwrap(); let result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), std::ptr::null(), // No partition columns table_name_cstr.as_ptr(), std::ptr::null(), // No storage options std::ptr::null(), // No table properties ); if !result.success { if !result.error.is_null() { let error = unsafe { &*result.error }; match error { DeltaGoError::DeltaTable(wrapper) => { if !wrapper.message.is_null() { let msg = unsafe { CStr::from_ptr(wrapper.message) }; eprintln!("DeltaTable error: {}", msg.to_string_lossy()); } } _ => eprintln!("Error creating table: {:?}", error), } } } assert!(result.success, "Table creation should succeed"); assert!(!result.value.is_null(), "Table handle should not be null"); assert!(result.error.is_null(), "Error should be null on success"); // Clean up deltalake_free_table(result.value); } #[test] fn test_create_delta_table_invalid_json() { let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let invalid_json_cstr = CString::new("invalid json").unwrap(); let result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), invalid_json_cstr.as_ptr(), std::ptr::null(), // No partition columns std::ptr::null(), std::ptr::null(), // No storage options std::ptr::null(), // No table properties ); assert!( !result.success, "Table creation should fail with invalid JSON" ); assert!( result.value.is_null(), "Table handle should be null on error" ); assert!( !result.error.is_null(), "Error should not be null on failure" ); // Clean up error deltalake_free_error(result.error); } #[test] fn test_create_table_with_storage_options() { let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); // Create the directory std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let table_name_cstr = CString::new("test_table").unwrap(); let columns_json = r#"[ {"name": "id", "type": "integer", "nullable": false}, {"name": "name", "type": "string", "nullable": true} ]"#; let columns_cstr = CString::new(columns_json).unwrap(); let storage_options_json = r#"{ "AWS_REGION": "us-west-2", "AWS_S3_ALLOW_UNSAFE_RENAME": "true" }"#; let storage_options_cstr = CString::new(storage_options_json).unwrap(); let result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), std::ptr::null(), // No partition columns table_name_cstr.as_ptr(), storage_options_cstr.as_ptr(), std::ptr::null(), // No table properties ); assert!( result.success, "Table creation with storage options should succeed" ); assert!(!result.value.is_null(), "Table handle should not be null"); assert!(result.error.is_null(), "Error should be null on success"); // Clean up deltalake_free_table(result.value); } #[test] fn test_open_delta_table() { // First create a table to test opening let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); // Create the directory std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let table_name_cstr = CString::new("test_table").unwrap(); let columns_json = r#"[ {"name": "id", "type": "integer", "nullable": false}, {"name": "name", "type": "string", "nullable": true} ]"#; let columns_cstr = CString::new(columns_json).unwrap(); // Create table let create_result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), std::ptr::null(), // No partition columns table_name_cstr.as_ptr(), std::ptr::null(), // No storage options std::ptr::null(), // No table properties ); assert!(create_result.success, "Table creation should succeed"); deltalake_free_table(create_result.value); // Now test opening the table let storage_options_json = r#"{}"#; let storage_options_cstr = CString::new(storage_options_json).unwrap(); let open_result = deltalake_open_table(path_cstr.as_ptr(), storage_options_cstr.as_ptr()); assert!(open_result.success, "Table opening should succeed"); assert!( !open_result.value.is_null(), "Table handle should not be null" ); assert!( open_result.error.is_null(), "Error should be null on success" ); // Clean up deltalake_free_table(open_result.value); } #[test] fn test_table_version() { // Create a table first let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); // Create the directory std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let columns_json = r#"[{"name": "id", "type": "integer", "nullable": false}]"#; let columns_cstr = CString::new(columns_json).unwrap(); let create_result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), std::ptr::null(), // No partition columns std::ptr::null(), std::ptr::null(), // No storage options std::ptr::null(), // No table properties ); assert!(create_result.success); let table_handle = create_result.value; // Test getting version let version_result = deltalake_table_version(table_handle); assert!( version_result.success, "Getting table version should succeed" ); assert_eq!(version_result.value, 0, "Initial table version should be 0"); assert!( version_result.error.is_null(), "Error should be null on success" ); // Clean up deltalake_free_table(table_handle); } #[test] fn test_table_version_invalid_handle() { let result = deltalake_table_version(std::ptr::null_mut()); assert!( !result.success, "Getting version with null handle should fail" ); assert_eq!(result.value, -1, "Version should be -1 on error"); assert!( !result.error.is_null(), "Error should not be null on failure" ); // Clean up error deltalake_free_error(result.error); } #[test] fn test_table_files() { // Create a table first let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); // Create the directory std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let columns_json = r#"[{"name": "id", "type": "integer", "nullable": false}]"#; let columns_cstr = CString::new(columns_json).unwrap(); let create_result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), std::ptr::null(), // No partition columns std::ptr::null(), std::ptr::null(), // No storage options std::ptr::null(), // No table properties ); assert!(create_result.success); let table_handle = create_result.value; // Test getting files let files_result = deltalake_table_files(table_handle); assert!(files_result.success, "Getting table files should succeed"); assert!( !files_result.value.is_null(), "Files JSON should not be null" ); assert!( files_result.error.is_null(), "Error should be null on success" ); // Convert the result to a Rust string to validate it's valid JSON let files_cstr = unsafe { CStr::from_ptr(files_result.value) }; let files_str = files_cstr.to_str().unwrap(); let _: serde_json::Value = serde_json::from_str(files_str).unwrap(); // Clean up deltalake_free_string(files_result.value as *mut c_char); deltalake_free_table(table_handle); } #[test] fn test_table_files_invalid_handle() { let result = deltalake_table_files(std::ptr::null_mut()); assert!( !result.success, "Getting files with null handle should fail" ); assert!(result.value.is_null(), "Files should be null on error"); assert!( !result.error.is_null(), "Error should not be null on failure" ); // Clean up error deltalake_free_error(result.error); } #[test] fn test_create_table_with_partition_columns() { let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_partitioned_table"); // Create the directory std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let table_name_cstr = CString::new("test_partitioned_table").unwrap(); let columns_json = r#"[ {"name": "id", "type": "integer", "nullable": false}, {"name": "name", "type": "string", "nullable": true}, {"name": "year", "type": "integer", "nullable": false}, {"name": "month", "type": "integer", "nullable": false} ]"#; let columns_cstr = CString::new(columns_json).unwrap(); // Create partition columns JSON let partition_columns_json = r#"["year", "month"]"#; let partition_columns_cstr = CString::new(partition_columns_json).unwrap(); let result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), partition_columns_cstr.as_ptr(), table_name_cstr.as_ptr(), std::ptr::null(), // No storage options std::ptr::null(), // No table properties ); assert!(result.success, "Partitioned table creation should succeed"); assert!(!result.value.is_null(), "Table handle should not be null"); assert!(result.error.is_null(), "Error should be null on success"); // Clean up deltalake_free_table(result.value); } #[test] fn test_write_table() { // Create a table first let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); // Create the directory std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let columns_json = r#"[ {"name": "id", "type": "integer", "nullable": false}, {"name": "name", "type": "string", "nullable": true} ]"#; let columns_cstr = CString::new(columns_json).unwrap(); let create_result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), std::ptr::null(), // No partition columns std::ptr::null(), // No table name std::ptr::null(), // No storage options std::ptr::null(), // No table properties ); assert!(create_result.success); let table_handle = create_result.value; // Test writing data let write_data_json = r#"{ "schema": [ {"name": "id", "type": "integer", "nullable": false}, {"name": "name", "type": "string", "nullable": true} ], "data": [ [1, "Alice"], [2, "Bob"], [3, "Charlie"] ] }"#; let write_data_cstr = CString::new(write_data_json).unwrap(); let append_mode_cstr = CString::new("append").unwrap(); let write_result = deltalake_write_table( table_handle, write_data_cstr.as_ptr(), append_mode_cstr.as_ptr(), ); assert!(write_result.success, "Writing data should succeed"); assert!( !write_result.value.is_null(), "Updated table handle should not be null" ); assert!( write_result.error.is_null(), "Error should be null on success" ); // Clean up - only free the returned table handle // The original table_handle was consumed by deltalake_write_table deltalake_free_table(write_result.value); } #[test] fn test_table_exists() { // First create a table to test existence let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("test_table"); // Create the directory std::fs::create_dir_all(&table_path).unwrap(); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let mode_cstr = CString::new("overwrite").unwrap(); let table_name_cstr = CString::new("test_table").unwrap(); let columns_json = r#"[ {"name": "id", "type": "integer", "nullable": true}, {"name": "name", "type": "string", "nullable": true} ]"#; let columns_cstr = CString::new(columns_json).unwrap(); // Create table let create_result = deltalake_create_table( path_cstr.as_ptr(), mode_cstr.as_ptr(), columns_cstr.as_ptr(), ptr::null(), table_name_cstr.as_ptr(), ptr::null(), std::ptr::null(), // No table properties ); assert!(create_result.success, "Table creation should succeed"); deltalake_free_table(create_result.value); // Test that the table now exists let storage_options_json = r#"{}"#; let storage_options_cstr = CString::new(storage_options_json).unwrap(); let exists_result = deltalake_table_exists(path_cstr.as_ptr(), storage_options_cstr.as_ptr()); assert!( exists_result.success, "Table existence check should succeed" ); assert!(exists_result.value, "Table should exist"); assert!( exists_result.error.is_null(), "Error should be null on success" ); } #[test] fn test_table_not_exists() { let temp_dir = TempDir::new().unwrap(); let table_path = temp_dir.path().join("nonexistent_table"); let path_str = table_path.to_str().unwrap(); let path_cstr = CString::new(path_str).unwrap(); let storage_options_json = r#"{}"#; let storage_options_cstr = CString::new(storage_options_json).unwrap(); let exists_result = deltalake_table_exists(path_cstr.as_ptr(), storage_options_cstr.as_ptr()); assert!( exists_result.success, "Table existence check should succeed" ); assert!(!exists_result.value, "Table should not exist"); assert!( exists_result.error.is_null(), "Error should be null on success" ); } #[test] fn test_table_exists_invalid_path() { let empty_path_cstr = CString::new("").unwrap(); let storage_options_json = r#"{}"#; let storage_options_cstr = CString::new(storage_options_json).unwrap(); let exists_result = deltalake_table_exists(empty_path_cstr.as_ptr(), storage_options_cstr.as_ptr()); assert!( !exists_result.success, "Table existence check should fail with empty path" ); assert!( !exists_result.error.is_null(), "Error should not be null on failure" ); // Clean up error deltalake_free_error(exists_result.error); } }