diff --git a/Project.toml b/Project.toml index 8534986..a0d7a06 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "RustyIceberg" uuid = "390bdf5b-b624-43dc-a846-0ef7a3405804" -version = "0.9.3" +version = "0.9.4" [deps] Arrow = "69666777-d1a9-59fb-9406-91d4454c9d45" diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index cb90a52..4e25d79 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -4,7 +4,7 @@ networks: services: minio: container_name: minio - image: minio/minio:latest + image: quay.io/minio/minio:latest ports: - "9000:9000" # MinIO API port - "9001:9001" # MinIO Console port @@ -24,7 +24,7 @@ services: retries: 5 setup_bucket: - image: minio/mc:latest + image: quay.io/minio/mc:latest depends_on: minio: condition: service_healthy diff --git a/iceberg_rust_ffi/Cargo.lock b/iceberg_rust_ffi/Cargo.lock index 2019a86..5d04ce0 100644 --- a/iceberg_rust_ffi/Cargo.lock +++ b/iceberg_rust_ffi/Cargo.lock @@ -1878,7 +1878,7 @@ dependencies = [ [[package]] name = "iceberg_rust_ffi" -version = "0.9.3" +version = "0.9.4" dependencies = [ "anyhow", "arrow-array", diff --git a/iceberg_rust_ffi/Cargo.toml b/iceberg_rust_ffi/Cargo.toml index 5c26cbb..3d8b6e0 100644 --- a/iceberg_rust_ffi/Cargo.toml +++ b/iceberg_rust_ffi/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "iceberg_rust_ffi" -version = "0.9.3" +version = "0.9.4" edition = "2021" [lib] diff --git a/iceberg_rust_ffi/src/table.rs b/iceberg_rust_ffi/src/table.rs index 56d337a..1f0a507 100644 --- a/iceberg_rust_ffi/src/table.rs +++ b/iceberg_rust_ffi/src/table.rs @@ -453,6 +453,27 @@ pub extern "C" fn iceberg_table_schema(table: *mut IcebergTable) -> *mut c_char } } +/// Current snapshot's summary (`operation` + caller-supplied and computed properties, +/// all flattened into one object) as a JSON string. Null if the table has no current +/// snapshot. +#[no_mangle] +pub extern "C" fn iceberg_table_current_snapshot_summary(table: *mut IcebergTable) -> *mut c_char { + if table.is_null() { + return ptr::null_mut(); + } + let table_ref = unsafe { &*table }; + let Some(snapshot) = table_ref.table.metadata().current_snapshot() else { + return ptr::null_mut(); + }; + match serde_json::to_string(snapshot.summary()) { + Ok(json) => match std::ffi::CString::new(json) { + Ok(c_str) => c_str.into_raw(), + Err(_) => ptr::null_mut(), + }, + Err(_) => ptr::null_mut(), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/iceberg_rust_ffi/src/transaction.rs b/iceberg_rust_ffi/src/transaction.rs index 4efa95e..884bfa0 100644 --- a/iceberg_rust_ffi/src/transaction.rs +++ b/iceberg_rust_ffi/src/transaction.rs @@ -7,9 +7,11 @@ use crate::catalog::IcebergCatalog; use crate::error_codes::{classified_error, classify_iceberg, IcebergErrorCode}; use crate::response::IcebergBoxedResponse; use crate::table::IcebergTable; -use crate::IcebergDataFiles; +use crate::util::parse_properties; +use crate::{IcebergDataFiles, PropertyEntry}; use iceberg::spec::DataFile; use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use std::collections::HashMap; // FFI exports use object_store_ffi::{ @@ -80,6 +82,9 @@ impl IcebergFastAppendAction { pub struct IcebergOverwriteAction { added_files: Vec, deleted_files: Vec, + /// Custom snapshot summary properties, merged with iceberg-rust's computed ones + /// (which win on key collision). + snapshot_properties: HashMap, } unsafe impl Send for IcebergOverwriteAction {} @@ -363,6 +368,45 @@ pub extern "C" fn iceberg_overwrite_action_add_data_files( 0 } +/// Set custom snapshot summary properties for the commit this action produces. Merged +/// with iceberg-rust's computed properties, which win on key collision. +/// +/// Copies `properties`. Returns 0 on success, non-zero on error. +#[no_mangle] +pub extern "C" fn iceberg_overwrite_action_set_snapshot_properties( + action: *mut IcebergOverwriteAction, + properties: *const PropertyEntry, + properties_len: usize, + error_message_out: *mut *mut std::ffi::c_char, +) -> i32 { + let set_error = |msg: &str, out: *mut *mut std::ffi::c_char| { + if !out.is_null() { + if let Ok(c_str) = std::ffi::CString::new(msg) { + unsafe { + *out = c_str.into_raw(); + } + } + } + }; + if action.is_null() { + set_error("Null action pointer provided", error_message_out); + return 1; + } + let props: HashMap = match parse_properties(properties, properties_len) { + Ok(props) => props, + Err(e) => { + set_error( + &format!("Failed to parse snapshot properties: {}", e), + error_message_out, + ); + return 1; + } + }; + let action_ref = unsafe { &mut *action }; + action_ref.snapshot_properties.extend(props); + 0 +} + /// Mark data files for deletion in the overwrite snapshot. /// /// The data_files handle is consumed — its files are moved into the action. @@ -431,6 +475,7 @@ pub extern "C" fn iceberg_overwrite_action_apply( let added = std::mem::take(&mut action_ref.added_files); let deleted = std::mem::take(&mut action_ref.deleted_files); + let snapshot_properties = std::mem::take(&mut action_ref.snapshot_properties); let tx = match tx_ref.take() { Some(t) => t, @@ -450,7 +495,8 @@ pub extern "C" fn iceberg_overwrite_action_apply( let overwrite_action = tx .overwrite() .add_data_files(added) - .delete_data_files(deleted); + .delete_data_files(deleted) + .set_snapshot_properties(snapshot_properties); match overwrite_action.apply(tx) { Ok(new_tx) => { diff --git a/src/RustyIceberg.jl b/src/RustyIceberg.jl index d75c336..8748bac 100644 --- a/src/RustyIceberg.jl +++ b/src/RustyIceberg.jl @@ -23,7 +23,7 @@ export INTERNAL export new_incremental_scan, free_incremental_scan! export scan_incremental_nested!, nested_incremental_arrow_stream export table_open, free_table, new_scan, free_scan! -export table_location, table_uuid, table_format_version, table_last_sequence_number, table_last_updated_ms, table_current_snapshot_id, table_schema +export table_location, table_uuid, table_format_version, table_last_sequence_number, table_last_updated_ms, table_current_snapshot_id, table_schema, table_current_snapshot_summary export IcebergPerfConfig export select_columns!, with_snapshot_id! export with_file_column!, with_pos_column! @@ -49,6 +49,7 @@ export IcebergString, IcebergUuid, IcebergBinary, IcebergDecimal export Transaction, DataFiles, free_transaction!, free_data_files!, commit, transaction, data_file_info export FastAppendAction, free_fast_append_action!, add_data_files, apply, with_fast_append export OverwriteAction, free_overwrite_action!, delete_data_files, with_overwrite, list_data_files +export set_snapshot_properties export DataFileWriter, free_writer!, close_writer, set_encode_workers! export WriterConfig, CompressionCodec, UNCOMPRESSED, SNAPPY, GZIP, LZ4, ZSTD, LZ4_RAW export RowChunk, flush! @@ -640,7 +641,7 @@ Get the current snapshot ID of an Iceberg table. Returns the snapshot ID if the table has at least one committed snapshot, or `nothing` if the table has no snapshots yet (e.g. immediately after creation, before any commit). """ -function table_current_snapshot_id(table::Table)::Union{Int64,Nothing} +function table_current_snapshot_id(table::Table) id = @ccall rust_lib.iceberg_table_current_snapshot_id(table::Table)::Int64 return id == -1 ? nothing : id end @@ -667,4 +668,27 @@ function table_schema(table::Table) return result end +""" + table_current_snapshot_summary(table::Table)::Union{Dict{String,String},Nothing} + +Current snapshot's summary (`operation` + caller-supplied and computed properties). +`nothing` if the table has no current snapshot; throws if `table` has been freed. +""" +function table_current_snapshot_summary(table::Table) + if table == C_NULL + throw(IcebergException( + STATE_RESOURCE_FREED, + "Resource has been freed", + "Table has been freed", + )) + end + ptr = @ccall rust_lib.iceberg_table_current_snapshot_summary(table::Table)::Ptr{Cchar} + if ptr == C_NULL + return nothing + end + json = unsafe_string(ptr) + @ccall rust_lib.iceberg_destroy_cstring(ptr::Ptr{Cchar})::Cint + return Dict{String,String}(JSON.parse(json)) +end + end # module RustyIceberg diff --git a/src/transaction.jl b/src/transaction.jl index 1996be8..364466a 100644 --- a/src/transaction.jl +++ b/src/transaction.jl @@ -417,6 +417,33 @@ function add_data_files(action::OverwriteAction, data_files::DataFiles) return nothing end +""" + set_snapshot_properties(action::OverwriteAction, properties::Dict{String,String}) + +Set custom snapshot summary properties for the commit this action produces. Merged with +iceberg-rust's computed properties, which win on key collision. +""" +function set_snapshot_properties(action::OverwriteAction, properties::Dict{String,String}) + if action.ptr == C_NULL + throw(IcebergException(STATE_RESOURCE_FREED, "Resource has been freed", "OverwriteAction has been freed")) + end + property_entries = [PropertyEntry(pointer(k), pointer(v)) for (k, v) in properties] + properties_len = length(property_entries) + error_message_ptr = Ref{Ptr{Cchar}}(C_NULL) + result = GC.@preserve properties property_entries begin + @ccall rust_lib.iceberg_overwrite_action_set_snapshot_properties( + action.ptr::Ptr{Cvoid}, + (properties_len > 0 ? pointer(property_entries) : C_NULL)::Ptr{PropertyEntry}, + properties_len::Csize_t, + error_message_ptr::Ref{Ptr{Cchar}} + )::Cint + end + if result != 0 + parse_and_throw(error_message_ptr[], "overwrite set_snapshot_properties") + end + return nothing +end + """ delete_data_files(action::OverwriteAction, data_files::DataFiles) diff --git a/test/overwrite_tests.jl b/test/overwrite_tests.jl index 98a82f4..2d91506 100644 --- a/test/overwrite_tests.jl +++ b/test/overwrite_tests.jl @@ -406,3 +406,68 @@ end end println("✅ OverwriteAction error handling tests passed") end + +# --------------------------------------------------------------------------- +# set_snapshot_properties +# --------------------------------------------------------------------------- + +@testset "set_snapshot_properties round-trips through a real commit" begin + mktempdir() do warehouse + cat = catalog_create_memory(warehouse) + table = C_NULL + updated = C_NULL + try + create_namespace(cat, ["ns"]) + table = create_table(cat, ["ns"], "t", _ow_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1, 2], value=[1.0, 2.0])) + end + + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + # "added-data-files" collides with a computed key; the real value + # (asserted below) must win, not this bogus one. + set_snapshot_properties(action, Dict( + "attempt_id" => "test-attempt-123", + "added-data-files" => "999", + )) + end + end + + @test !isnothing(table_current_snapshot_id(updated)) + data = read_table_data(updated) + @test sort(data.id) == [1, 2] + + summary = table_current_snapshot_summary(updated) + @test !isnothing(summary) + @test summary["operation"] == "overwrite" + @test summary["attempt_id"] == "test-attempt-123" + @test summary["added-data-files"] == "1" + @test summary["total-records"] == "2" + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ set_snapshot_properties round-trips through a real commit") +end + +@testset "table_current_snapshot_summary on a null table throws" begin + # `Table` is a raw `Ptr{Cvoid}`, not a wrapper struct -- `free_table` can't null it + # out on the caller's side, so a freed-then-reused Table is unchecked UB here, same + # as every other Table-consuming function in this file. What the null-check in + # table_current_snapshot_summary actually guards against is a genuinely-null Table + # value, matching table_schema's existing behavior -- that's what this tests. + @test_throws RustyIceberg.IcebergException table_current_snapshot_summary(C_NULL) + println("✅ table_current_snapshot_summary on a null table throws") +end + +@testset "set_snapshot_properties on a freed action throws" begin + action = RustyIceberg.OverwriteAction() + free_overwrite_action!(action) + @test_throws RustyIceberg.IcebergException set_snapshot_properties(action, Dict("k" => "v")) + println("✅ set_snapshot_properties on a freed action throws") +end