From 4a6881602cf104c2ad1856cf0fb0fd9d52d51af9 Mon Sep 17 00:00:00 2001 From: Gerald Berger Date: Thu, 17 Sep 2026 11:06:15 +0200 Subject: [PATCH 1/3] Add assert_requirements bindings Wraps RelationalAI/iceberg-rust#83's OverwriteAction.assert_requirements for Julia: supply arbitrary TableRequirements (as typed TableRequirement subtypes: AssertTableUuid, AssertRefSnapshotId, etc.), passed to the Rust side as a JSON array via a new iceberg_overwrite_action_assert_requirements FFI export. A supplied requirement of an auto-derived kind replaces the derived one; any other kind is added; repeated calls accumulate. This is needed because Transaction::commit's do_commit always refreshes the table from the catalog and silently rebases onto whatever is current before computing its auto-derived RefSnapshotIdMatch requirement, so a retried (or racing) commit would otherwise keep asserting whatever is current rather than the base it actually started from, and a genuine conflict could never surface. assert_requirements lets a caller pin that assertion to an externally-resolved base snapshot instead. Bumps iceberg_rust_ffi's iceberg-rust git rev to f02c8db2f890d4447a2995 738668573769aa98a3 (that PR's branch head, rebased onto main after #84's upstream sync + pipeline.rs architecture convergence merged) to pick up the new API; move this to a `main` rev once the PR merges. Tests: assert_requirements_tests.jl covers the override/accumulate/reject semantics, including two tests that provoke a genuine CatalogCommitConflicts by pinning a stale/wrong requirement and asserting the catalog itself rejects the commit. Co-Authored-By: Claude Sonnet 5 --- iceberg_rust_ffi/Cargo.lock | 145 ++++++++++++---- iceberg_rust_ffi/Cargo.toml | 13 +- iceberg_rust_ffi/src/transaction.rs | 71 +++++++- src/RustyIceberg.jl | 4 + src/transaction.jl | 165 +++++++++++++++++++ test/assert_requirements_tests.jl | 246 ++++++++++++++++++++++++++++ test/runtests.jl | 3 + 7 files changed, 606 insertions(+), 41 deletions(-) create mode 100644 test/assert_requirements_tests.jl diff --git a/iceberg_rust_ffi/Cargo.lock b/iceberg_rust_ffi/Cargo.lock index 803bbe3..274af22 100644 --- a/iceberg_rust_ffi/Cargo.lock +++ b/iceberg_rust_ffi/Cargo.lock @@ -66,6 +66,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "aliasable" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd" + [[package]] name = "alloc-no-stdlib" version = "2.0.4" @@ -98,26 +104,26 @@ checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" [[package]] name = "apache-avro" -version = "0.21.0" +version = "0.22.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf" +checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c" dependencies = [ "bigdecimal", "bon", "crc32fast", - "digest 0.10.7", + "digest 0.11.3", "log", - "miniz_oxide", + "miniz_oxide 0.9.1", "num-bigint 0.4.8", + "ouroboros", "quad-rand", - "rand 0.9.5", + "rand 0.10.2", "regex-lite", "serde", "serde_bytes", "serde_json", "snap", - "strum 0.27.2", - "strum_macros 0.27.2", + "strum 0.28.0", "thiserror 2.0.20", "uuid", "zstd", @@ -1091,12 +1097,6 @@ dependencies = [ "syn 3.0.4", ] -[[package]] -name = "dissimilar" -version = "1.0.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aeda16ab4059c5fd2a83f2b9c9e9c981327b18aa8e3b313f7e6563799d4f093e" - [[package]] name = "dlv-list" version = "0.5.2" @@ -1136,7 +1136,7 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a1e6a265c649f3f5979b601d26f1d05ada116434c87741c9493cb56218f76cbc" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.119", @@ -1195,16 +1195,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "expect-test" -version = "1.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63af43ff4431e848fb47472a920f14fa71c24de13255a5692e93d4e90302acb0" -dependencies = [ - "dissimilar", - "once_cell", -] - [[package]] name = "fastnum" version = "0.7.5" @@ -1246,7 +1236,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", - "miniz_oxide", + "miniz_oxide 0.8.9", "zlib-rs", ] @@ -1509,6 +1499,12 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "heck" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8" + [[package]] name = "heck" version = "0.5.0" @@ -1787,7 +1783,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=f02c8db2f890d4447a2995738668573769aa98a3#f02c8db2f890d4447a2995738668573769aa98a3" dependencies = [ "aes-gcm", "anyhow", @@ -1810,7 +1806,6 @@ dependencies = [ "chrono", "crc32fast", "derive_builder", - "expect-test", "fastnum", "flate2", "fnv", @@ -1846,7 +1841,7 @@ dependencies = [ [[package]] name = "iceberg-catalog-rest" version = "0.10.1" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=f02c8db2f890d4447a2995738668573769aa98a3#f02c8db2f890d4447a2995738668573769aa98a3" dependencies = [ "async-trait", "chrono", @@ -1866,7 +1861,7 @@ dependencies = [ [[package]] name = "iceberg-property-macro" version = "0.10.1" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=f02c8db2f890d4447a2995738668573769aa98a3#f02c8db2f890d4447a2995738668573769aa98a3" dependencies = [ "proc-macro2", "quote", @@ -1876,7 +1871,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=f02c8db2f890d4447a2995738668573769aa98a3#f02c8db2f890d4447a2995738668573769aa98a3" dependencies = [ "anyhow", "async-trait", @@ -1884,6 +1879,7 @@ dependencies = [ "cfg-if", "futures", "iceberg", + "iceberg-property-macro", "opendal", "reqsign-aws-v4", "reqsign-core", @@ -1895,7 +1891,7 @@ dependencies = [ [[package]] name = "iceberg-storage-refreshable" version = "0.10.1" -source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=ec07ed0f7c08390702d841298e4331976df93b45#ec07ed0f7c08390702d841298e4331976df93b45" +source = "git+https://github.com/RelationalAI/iceberg-rust.git?rev=f02c8db2f890d4447a2995738668573769aa98a3#f02c8db2f890d4447a2995738668573769aa98a3" dependencies = [ "async-trait", "bytes", @@ -1910,7 +1906,7 @@ dependencies = [ [[package]] name = "iceberg_rust_ffi" -version = "0.9.6" +version = "0.9.7" dependencies = [ "anyhow", "arrow-array", @@ -2490,6 +2486,15 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", +] + [[package]] name = "mio" version = "1.2.2" @@ -2987,6 +2992,30 @@ dependencies = [ "hashbrown 0.14.5", ] +[[package]] +name = "ouroboros" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0f050db9c44b97a94723127e6be766ac5c340c48f2c4bb3ffa11713744be59" +dependencies = [ + "aliasable", + "ouroboros_macro", + "static_assertions", +] + +[[package]] +name = "ouroboros_macro" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c7028bdd3d43083f6d8d4d5187680d0d3560d54df4cc9d752005268b41e64d0" +dependencies = [ + "heck 0.4.1", + "proc-macro2", + "proc-macro2-diagnostics", + "quote", + "syn 2.0.119", +] + [[package]] name = "parking" version = "2.2.1" @@ -3225,6 +3254,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proc-macro2-diagnostics" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af066a9c399a26e020ada66a034357a868728e72cd426f3adcd35f80d88d88c8" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "version_check", + "yansi", +] + [[package]] name = "quad-rand" version = "0.2.3" @@ -4220,7 +4262,7 @@ version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.119", @@ -4273,6 +4315,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + [[package]] name = "strsim" version = "0.11.1" @@ -4297,13 +4345,22 @@ dependencies = [ "strum_macros 0.27.2", ] +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros 0.28.0", +] + [[package]] name = "strum_macros" version = "0.26.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "rustversion", @@ -4316,7 +4373,19 @@ version = "0.27.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" dependencies = [ - "heck", + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.119", @@ -5162,6 +5231,12 @@ dependencies = [ "rustix", ] +[[package]] +name = "yansi" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" + [[package]] name = "yoke" version = "0.8.3" diff --git a/iceberg_rust_ffi/Cargo.toml b/iceberg_rust_ffi/Cargo.toml index 5b5efe5..1e5551d 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.6" +version = "0.9.7" edition = "2021" [lib] @@ -12,10 +12,13 @@ default = ["julia"] julia = [] [dependencies] -iceberg = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45" } -iceberg-catalog-rest = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45" } -iceberg-storage-opendal = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45", features = ["opendal-azdls"] } -iceberg-storage-refreshable = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "ec07ed0f7c08390702d841298e4331976df93b45" } +# Pinned to RelationalAI/iceberg-rust#83 (gb/base-snapshot-assertion-and-stage-commit, +# rebased onto main after #84 merged), which adds OverwriteAction::assert_requirements. +# Move this back to a `main` rev once that PR merges. +iceberg = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "f02c8db2f890d4447a2995738668573769aa98a3" } +iceberg-catalog-rest = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "f02c8db2f890d4447a2995738668573769aa98a3" } +iceberg-storage-opendal = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "f02c8db2f890d4447a2995738668573769aa98a3", features = ["opendal-azdls"] } +iceberg-storage-refreshable = { git = "https://github.com/RelationalAI/iceberg-rust.git", rev = "f02c8db2f890d4447a2995738668573769aa98a3" } object_store_ffi = { git = "https://github.com/RelationalAI/object_store_ffi", rev = "bbbbd2f0ab0d609416014cef0d124efa99a515f3", default-features = false } tokio = { version = "1.0", features = ["full"] } futures = "0.3" diff --git a/iceberg_rust_ffi/src/transaction.rs b/iceberg_rust_ffi/src/transaction.rs index 884bfa0..0053367 100644 --- a/iceberg_rust_ffi/src/transaction.rs +++ b/iceberg_rust_ffi/src/transaction.rs @@ -11,6 +11,7 @@ use crate::util::parse_properties; use crate::{IcebergDataFiles, PropertyEntry}; use iceberg::spec::DataFile; use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::TableRequirement; use std::collections::HashMap; // FFI exports @@ -85,6 +86,9 @@ pub struct IcebergOverwriteAction { /// Custom snapshot summary properties, merged with iceberg-rust's computed ones /// (which win on key collision). snapshot_properties: HashMap, + /// Extra requirements to assert on commit, on top of (or in place of) the ones this + /// action derives automatically. See `OverwriteAction::assert_requirements`. + extra_requirements: Vec, } unsafe impl Send for IcebergOverwriteAction {} @@ -407,6 +411,69 @@ pub extern "C" fn iceberg_overwrite_action_set_snapshot_properties( 0 } +/// Assert additional table requirements before this commit is attempted, alongside the ones +/// this action derives automatically from the table it's applied to (currently: a UUID match, +/// and a ref-snapshot-id match for the branch this commit targets). +/// +/// `requirements_json` must be a JSON array of Iceberg REST catalog `TableRequirement` objects, +/// e.g. `[{"type": "assert-ref-snapshot-id", "ref": "main", "snapshot-id": 100}]` -- the exact +/// shape the REST catalog protocol itself uses. A supplied requirement of the same kind as one +/// this action would otherwise derive automatically (a UUID match, or a ref-snapshot-id match +/// for the same ref) replaces the derived one, rather than producing a duplicate -- e.g. to +/// assert a pinned base snapshot id that differs from whatever the table this transaction was +/// created from currently reports, supply your own `assert-ref-snapshot-id` requirement here. +/// Any other requirement kind is added alongside the derived ones. Calling this more than once +/// accumulates requirements rather than replacing the whole set. +/// +/// Copies `requirements_json`. Returns 0 on success, non-zero on error (including malformed JSON). +#[no_mangle] +pub extern "C" fn iceberg_overwrite_action_assert_requirements( + action: *mut IcebergOverwriteAction, + requirements_json: *const std::ffi::c_char, + 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; + } + if requirements_json.is_null() { + set_error("Null requirements_json pointer provided", error_message_out); + return 1; + } + let json_str = match unsafe { std::ffi::CStr::from_ptr(requirements_json) }.to_str() { + Ok(s) => s, + Err(e) => { + set_error( + &format!("requirements_json is not valid UTF-8: {}", e), + error_message_out, + ); + return 1; + } + }; + let requirements: Vec = match serde_json::from_str(json_str) { + Ok(r) => r, + Err(e) => { + set_error( + &format!("Failed to parse requirements_json: {}", e), + error_message_out, + ); + return 1; + } + }; + let action_ref = unsafe { &mut *action }; + action_ref.extra_requirements.extend(requirements); + 0 +} + /// Mark data files for deletion in the overwrite snapshot. /// /// The data_files handle is consumed — its files are moved into the action. @@ -476,6 +543,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 extra_requirements = std::mem::take(&mut action_ref.extra_requirements); let tx = match tx_ref.take() { Some(t) => t, @@ -496,7 +564,8 @@ pub extern "C" fn iceberg_overwrite_action_apply( .overwrite() .add_data_files(added) .delete_data_files(deleted) - .set_snapshot_properties(snapshot_properties); + .set_snapshot_properties(snapshot_properties) + .assert_requirements(extra_requirements); match overwrite_action.apply(tx) { Ok(new_tx) => { diff --git a/src/RustyIceberg.jl b/src/RustyIceberg.jl index 8748bac..a9f8a8a 100644 --- a/src/RustyIceberg.jl +++ b/src/RustyIceberg.jl @@ -50,6 +50,10 @@ export Transaction, DataFiles, free_transaction!, free_data_files!, commit, tran 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 TableRequirement, AssertTableDoesNotExist, AssertTableUuid, AssertRefSnapshotId +export AssertCurrentSchemaId, AssertLastAssignedFieldId, AssertLastAssignedPartitionId +export AssertDefaultSpecId, AssertDefaultSortOrderId +export assert_requirements export DataFileWriter, free_writer!, close_writer, set_encode_workers! export WriterConfig, CompressionCodec, UNCOMPRESSED, SNAPPY, GZIP, LZ4, ZSTD, LZ4_RAW export RowChunk, flush! diff --git a/src/transaction.jl b/src/transaction.jl index 364466a..226755f 100644 --- a/src/transaction.jl +++ b/src/transaction.jl @@ -3,6 +3,126 @@ # This module provides Julia wrappers for the iceberg-rust Transaction API, # enabling Julia to create transactions, add data files, and commit changes. +# --------------------------------------------------------------------------- +# TableRequirement +# --------------------------------------------------------------------------- + +""" + TableRequirement + +Abstract type for a precondition the Iceberg REST catalog protocol can assert before +accepting a commit. Each concrete subtype corresponds to one `type` of the protocol's +`TableRequirement` object; see the Iceberg REST catalog OpenAPI spec. + +Pass a vector of these to [`assert_requirements`](@ref). +""" +abstract type TableRequirement end + +""" + AssertTableDoesNotExist() + +The table must not already exist. Used for create transactions. +""" +struct AssertTableDoesNotExist <: TableRequirement end + +""" + AssertTableUuid(uuid::AbstractString) + +The table's UUID must match `uuid`. +""" +struct AssertTableUuid <: TableRequirement + uuid::String +end + +""" + AssertRefSnapshotId(ref::AbstractString, snapshot_id::Union{Integer,Nothing}) + +The branch or tag named `ref` must currently point at `snapshot_id`. `snapshot_id === +nothing` asserts that `ref` does not exist yet. + +This is the requirement to override via [`assert_requirements`](@ref) to pin a commit to +an externally-resolved base snapshot, e.g. so every retry of the same logical job asserts +the same base regardless of what the table's current state looks like by the time that +retry runs. +""" +struct AssertRefSnapshotId <: TableRequirement + ref::String + snapshot_id::Union{Int64,Nothing} +end +AssertRefSnapshotId(ref::AbstractString, snapshot_id::Union{Integer,Nothing}) = + AssertRefSnapshotId(String(ref), snapshot_id === nothing ? nothing : Int64(snapshot_id)) + +""" + AssertCurrentSchemaId(current_schema_id::Integer) + +The table's current schema id must match `current_schema_id`. +""" +struct AssertCurrentSchemaId <: TableRequirement + current_schema_id::Int32 +end + +""" + AssertLastAssignedFieldId(last_assigned_field_id::Integer) + +The table's last assigned column id must match `last_assigned_field_id`. +""" +struct AssertLastAssignedFieldId <: TableRequirement + last_assigned_field_id::Int32 +end + +""" + AssertLastAssignedPartitionId(last_assigned_partition_id::Integer) + +The table's last assigned partition id must match `last_assigned_partition_id`. +""" +struct AssertLastAssignedPartitionId <: TableRequirement + last_assigned_partition_id::Int32 +end + +""" + AssertDefaultSpecId(default_spec_id::Integer) + +The table's default partition spec id must match `default_spec_id`. +""" +struct AssertDefaultSpecId <: TableRequirement + default_spec_id::Int32 +end + +""" + AssertDefaultSortOrderId(default_sort_order_id::Integer) + +The table's default sort order id must match `default_sort_order_id`. +""" +struct AssertDefaultSortOrderId <: TableRequirement + default_sort_order_id::Int64 +end + +# Julia value -> the exact JSON shape the Iceberg REST catalog protocol uses (kebab-case +# fields, `type` tag), ready for `JSON.json`. +_to_json_dict(::AssertTableDoesNotExist) = Dict{String,Any}("type" => "assert-create") +_to_json_dict(r::AssertTableUuid) = Dict{String,Any}("type" => "assert-table-uuid", "uuid" => r.uuid) +_to_json_dict(r::AssertRefSnapshotId) = + Dict{String,Any}("type" => "assert-ref-snapshot-id", "ref" => r.ref, "snapshot-id" => r.snapshot_id) +_to_json_dict(r::AssertCurrentSchemaId) = + Dict{String,Any}("type" => "assert-current-schema-id", "current-schema-id" => r.current_schema_id) +_to_json_dict(r::AssertLastAssignedFieldId) = + Dict{String,Any}("type" => "assert-last-assigned-field-id", "last-assigned-field-id" => r.last_assigned_field_id) +_to_json_dict(r::AssertLastAssignedPartitionId) = Dict{String,Any}( + "type" => "assert-last-assigned-partition-id", + "last-assigned-partition-id" => r.last_assigned_partition_id, +) +_to_json_dict(r::AssertDefaultSpecId) = + Dict{String,Any}("type" => "assert-default-spec-id", "default-spec-id" => r.default_spec_id) +_to_json_dict(r::AssertDefaultSortOrderId) = Dict{String,Any}( + "type" => "assert-default-sort-order-id", + "default-sort-order-id" => r.default_sort_order_id, +) + +# Structural equality for TableRequirement subtypes -- these are small value types, so two +# instances with the same fields should compare equal even when they aren't the same object. +Base.:(==)(a::T, b::T) where {T<:TableRequirement} = + all(f -> getfield(a, f) == getfield(b, f), fieldnames(T)) + """ Transaction @@ -444,6 +564,51 @@ function set_snapshot_properties(action::OverwriteAction, properties::Dict{Strin return nothing end +""" + assert_requirements(action::OverwriteAction, requirements::AbstractVector{<:TableRequirement}) + +Assert additional table requirements before this commit is attempted, alongside the ones +this action derives automatically from the table it's applied to (currently: +[`AssertTableUuid`](@ref), and an [`AssertRefSnapshotId`](@ref) for the branch this commit +targets). + +A supplied requirement of the same kind as one this action would otherwise derive +automatically replaces the derived one, rather than producing a duplicate -- e.g. to +assert a pinned base snapshot id that differs from whatever the table this transaction was +created from currently reports, supply your own [`AssertRefSnapshotId`](@ref) here. Any +other requirement kind is added alongside the derived ones. Calling this more than once +accumulates requirements rather than replacing the whole set. + +# Example +```julia +# Pin this commit to a base snapshot resolved earlier (e.g. at job dispatch time), so a +# retry of the same job keeps asserting the same base even if this table's current state +# has since moved on. +with_overwrite(tx) do action + add_data_files(action, new_files) + assert_requirements(action, [AssertRefSnapshotId("main", pinned_base_snapshot_id)]) +end +``` +""" +function assert_requirements(action::OverwriteAction, requirements::AbstractVector{<:TableRequirement}) + if action.ptr == C_NULL + throw(IcebergException(STATE_RESOURCE_FREED, "Resource has been freed", "OverwriteAction has been freed")) + end + json = JSON.json([_to_json_dict(r) for r in requirements]) + error_message_ptr = Ref{Ptr{Cchar}}(C_NULL) + result = GC.@preserve json begin + @ccall rust_lib.iceberg_overwrite_action_assert_requirements( + action.ptr::Ptr{Cvoid}, + json::Cstring, + error_message_ptr::Ref{Ptr{Cchar}} + )::Cint + end + if result != 0 + parse_and_throw(error_message_ptr[], "overwrite assert_requirements") + end + return nothing +end + """ delete_data_files(action::OverwriteAction, data_files::DataFiles) diff --git a/test/assert_requirements_tests.jl b/test/assert_requirements_tests.jl new file mode 100644 index 0000000..1e2faee --- /dev/null +++ b/test/assert_requirements_tests.jl @@ -0,0 +1,246 @@ +using RustyIceberg +using Test + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +function _ar_schema() + Schema([ + Field(Int32(1), "id", "long"; required=true), + Field(Int32(2), "value", "double"; required=false), + ]) +end + +# --------------------------------------------------------------------------- +# assert_requirements: overriding an auto-derived requirement +# --------------------------------------------------------------------------- + +@testset "assert_requirements overrides ref-snapshot-id-match end-to-end" begin + mktempdir() do warehouse + cat = catalog_create_memory(warehouse) + table = C_NULL + updated1 = C_NULL + updated2 = C_NULL + try + create_namespace(cat, ["ns"]) + table = create_table(cat, ["ns"], "t", _ar_schema()) + + files1 = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + updated1 = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files1) + end + end + base_id = table_current_snapshot_id(updated1) + + # A second commit, asserting the *same* base explicitly via assert_requirements + # (rather than letting the action auto-derive it). Since the table hasn't + # moved since `updated1` was loaded, this should succeed exactly like a normal + # commit would. + files2 = RustyIceberg.with_data_file_writer(updated1; prefix="d2") do w + write(w, (id=Int64[2], value=[2.0])) + end + updated2 = RustyIceberg.with_transaction(updated1, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files2) + assert_requirements(action, [AssertRefSnapshotId("main", base_id)]) + end + end + @test table_current_snapshot_id(updated2) != base_id + data = read_table_data(updated2) + @test sort(data.id) == [1, 2] + finally + table != C_NULL && free_table(table) + updated1 != C_NULL && free_table(updated1) + updated2 != C_NULL && free_table(updated2) + free_catalog!(cat) + end + end + println("✅ assert_requirements overrides ref-snapshot-id-match end-to-end") +end + +@testset "assert_requirements override is rejected when it no longer matches the catalog" begin + mktempdir() do warehouse + cat = catalog_create_memory(warehouse) + table = C_NULL + updated1 = C_NULL + try + create_namespace(cat, ["ns"]) + table = create_table(cat, ["ns"], "t", _ar_schema()) + + files1 = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + updated1 = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files1) + end + end + base_id = table_current_snapshot_id(updated1) + + # Assert a base snapshot id that is deliberately wrong (doesn't match what the + # catalog actually has). Since assert_requirements-supplied requirements are not + # locally pre-validated (that's the whole point -- see the PR this is testing), + # this must reach the catalog and be rejected there, not fail earlier for the + # wrong reason. + files2 = RustyIceberg.with_data_file_writer(updated1; prefix="d2") do w + write(w, (id=Int64[2], value=[2.0])) + end + bogus_base = base_id + 12345 + @test_throws RustyIceberg.IcebergException begin + RustyIceberg.with_transaction(updated1, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files2) + assert_requirements(action, [AssertRefSnapshotId("main", bogus_base)]) + end + end + end + + # And the catalog must genuinely be untouched by the rejected attempt. + reloaded = load_table(cat, ["ns"], "t") + try + @test table_current_snapshot_id(reloaded) == base_id + finally + free_table(reloaded) + end + finally + table != C_NULL && free_table(table) + updated1 != C_NULL && free_table(updated1) + free_catalog!(cat) + end + end + println("✅ assert_requirements override is rejected when it no longer matches the catalog") +end + +@testset "assert_requirements adds a new requirement kind alongside derived ones" 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", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + # CurrentSchemaIdMatch isn't a kind this action derives on its own, so it + # should ride alongside the auto-derived UuidMatch/RefSnapshotIdMatch, not + # replace anything -- and since it's correct, the commit still succeeds. + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertCurrentSchemaId(0)]) + end + end + @test !isnothing(table_current_snapshot_id(updated)) + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ assert_requirements adds a new requirement kind alongside derived ones") +end + +@testset "assert_requirements: wrong extra requirement kind is rejected by the catalog" begin + mktempdir() do warehouse + cat = catalog_create_memory(warehouse) + table = C_NULL + try + create_namespace(cat, ["ns"]) + table = create_table(cat, ["ns"], "t", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + @test_throws RustyIceberg.IcebergException begin + RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertCurrentSchemaId(999)]) + end + end + end + finally + table != C_NULL && free_table(table) + free_catalog!(cat) + end + end + println("✅ assert_requirements: wrong extra requirement kind is rejected by the catalog") +end + +@testset "assert_requirements accumulates across multiple calls" 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", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertCurrentSchemaId(0)]) + assert_requirements(action, [AssertLastAssignedFieldId(2)]) + end + end + @test !isnothing(table_current_snapshot_id(updated)) + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ assert_requirements accumulates across multiple calls") +end + +# --------------------------------------------------------------------------- +# error handling +# --------------------------------------------------------------------------- + +@testset "assert_requirements on a freed action throws" begin + action = RustyIceberg.OverwriteAction() + free_overwrite_action!(action) + @test_throws RustyIceberg.IcebergException assert_requirements(action, [AssertTableUuid("00000000-0000-0000-0000-000000000000")]) + println("✅ assert_requirements on a freed action throws") +end + +@testset "assert_requirements with an empty vector is a no-op" 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", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, TableRequirement[]) + end + end + @test !isnothing(table_current_snapshot_id(updated)) + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ assert_requirements with an empty vector is a no-op") +end diff --git a/test/runtests.jl b/test/runtests.jl index 1950dc5..59980d3 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -51,6 +51,9 @@ include("writer_tests.jl") # Include overwrite tests include("overwrite_tests.jl") +# Include assert_requirements tests +include("assert_requirements_tests.jl") + end # End of testset println("\n🎉 All tests completed!") From 348bfcd95ca57d5960aba066fa85a8a355b5fd1a Mon Sep 17 00:00:00 2001 From: Gerald Berger Date: Thu, 15 Oct 2026 11:22:23 +0200 Subject: [PATCH 2/3] Cover the remaining 5 TableRequirement variants end-to-end AssertTableUuid, AssertLastAssignedPartitionId, AssertDefaultSpecId, and AssertDefaultSortOrderId are now exercised through a real commit, round-tripping their JSON through the real Rust TableRequirement deserializer and catalog check. AssertTableDoesNotExist can only ever fail against an existing table, so its round-trip is covered by asserting that the catalog correctly rejects it. Previously only 3 of 8 variants reached the real FFI boundary in a test; the other 5 were verified correct only by manual inspection against iceberg-rust's enum definition during self-review, which a future rename or type change there could silently break without any test catching it. Co-Authored-By: Claude Sonnet 5 --- test/assert_requirements_tests.jl | 160 ++++++++++++++++++++++++++++++ 1 file changed, 160 insertions(+) diff --git a/test/assert_requirements_tests.jl b/test/assert_requirements_tests.jl index 1e2faee..5bdcfad 100644 --- a/test/assert_requirements_tests.jl +++ b/test/assert_requirements_tests.jl @@ -205,6 +205,134 @@ end println("✅ assert_requirements accumulates across multiple calls") end +# --------------------------------------------------------------------------- +# assert_requirements: the remaining requirement kinds, exercised end-to-end +# --------------------------------------------------------------------------- +# +# Each of these sends its requirement through the real JSON -> Rust deserializer and +# catalog check (not just Julia-side construction/equality), so a future rename or type +# change in iceberg-rust's `TableRequirement` enum would be caught here. + +@testset "assert_requirements AssertTableUuid matches the real table uuid" 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", _ar_schema()) + uuid = table_uuid(table) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertTableUuid(uuid)]) + end + end + @test !isnothing(table_current_snapshot_id(updated)) + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ assert_requirements AssertTableUuid matches the real table uuid") +end + +@testset "assert_requirements AssertLastAssignedPartitionId matches a fresh unpartitioned table" 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", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + # A fresh, unpartitioned table's last assigned partition id is 999 + # (iceberg-rust's `UNPARTITIONED_LAST_ASSIGNED_ID`); real partition fields + # allocated later would start at 1000. + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertLastAssignedPartitionId(999)]) + end + end + @test !isnothing(table_current_snapshot_id(updated)) + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ assert_requirements AssertLastAssignedPartitionId matches a fresh unpartitioned table") +end + +@testset "assert_requirements AssertDefaultSpecId matches the unpartitioned default spec" 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", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertDefaultSpecId(0)]) + end + end + @test !isnothing(table_current_snapshot_id(updated)) + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ assert_requirements AssertDefaultSpecId matches the unpartitioned default spec") +end + +@testset "assert_requirements AssertDefaultSortOrderId matches the unsorted default order" 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", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertDefaultSortOrderId(0)]) + end + end + @test !isnothing(table_current_snapshot_id(updated)) + finally + table != C_NULL && free_table(table) + updated != C_NULL && free_table(updated) + free_catalog!(cat) + end + end + println("✅ assert_requirements AssertDefaultSortOrderId matches the unsorted default order") +end + # --------------------------------------------------------------------------- # error handling # --------------------------------------------------------------------------- @@ -216,6 +344,38 @@ end println("✅ assert_requirements on a freed action throws") end +@testset "assert_requirements AssertTableDoesNotExist is rejected against an existing table" begin + mktempdir() do warehouse + cat = catalog_create_memory(warehouse) + table = C_NULL + try + create_namespace(cat, ["ns"]) + table = create_table(cat, ["ns"], "t", _ar_schema()) + + files = RustyIceberg.with_data_file_writer(table) do w + write(w, (id=Int64[1], value=[1.0])) + end + + # AssertTableDoesNotExist only ever succeeds against a nonexistent table (it's + # meant for create transactions); here it must reach the real catalog check and + # be rejected, which is the only way to exercise this requirement kind's JSON + # round-trip against an existing table. + @test_throws RustyIceberg.IcebergException begin + RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertTableDoesNotExist()]) + end + end + end + finally + table != C_NULL && free_table(table) + free_catalog!(cat) + end + end + println("✅ assert_requirements AssertTableDoesNotExist is rejected against an existing table") +end + @testset "assert_requirements with an empty vector is a no-op" begin mktempdir() do warehouse cat = catalog_create_memory(warehouse) From b5e2ab7f92861a59aaeee08d878c1ffa3c82a90a Mon Sep 17 00:00:00 2001 From: Gerald Berger Date: Tue, 20 Oct 2026 10:57:17 +0200 Subject: [PATCH 3/3] Tighten rejection tests to assert the actual catalog-conflict error code @test_throws RustyIceberg.IcebergException only checked the exception type, so a wrong JSON shape (e.g. an unrecognized "type" tag) causing a local deserialization failure would have passed these tests too, instead of proving the catalog itself rejected the commit. Capture the exception and assert exc.code == CATALOG_COMMIT_CONFLICT instead, following test/error_tests.jl's established pattern. Also fixed "accumulates across multiple calls": both calls previously asserted correct requirements, so the test passed even if the first call's requirement were silently dropped. Now the first call asserts a wrong schema id and the second a correct field id, so the commit must be rejected for retaining the first call's requirement to be observable at all. Co-Authored-By: Claude Sonnet 5 --- test/assert_requirements_tests.jl | 53 +++++++++++++++++++++++++------ 1 file changed, 44 insertions(+), 9 deletions(-) diff --git a/test/assert_requirements_tests.jl b/test/assert_requirements_tests.jl index 5bdcfad..ae9c129 100644 --- a/test/assert_requirements_tests.jl +++ b/test/assert_requirements_tests.jl @@ -90,14 +90,19 @@ end write(w, (id=Int64[2], value=[2.0])) end bogus_base = base_id + 12345 - @test_throws RustyIceberg.IcebergException begin + exc = try RustyIceberg.with_transaction(updated1, cat) do tx with_overwrite(tx) do action add_data_files(action, files2) assert_requirements(action, [AssertRefSnapshotId("main", bogus_base)]) end end + nothing + catch e + e isa RustyIceberg.IcebergException ? e : rethrow() end + @test !isnothing(exc) + @test exc.code == RustyIceberg.CATALOG_COMMIT_CONFLICT # And the catalog must genuinely be untouched by the rejected attempt. reloaded = load_table(cat, ["ns"], "t") @@ -159,14 +164,19 @@ end write(w, (id=Int64[1], value=[1.0])) end - @test_throws RustyIceberg.IcebergException begin + exc = try RustyIceberg.with_transaction(table, cat) do tx with_overwrite(tx) do action add_data_files(action, files) assert_requirements(action, [AssertCurrentSchemaId(999)]) end end + nothing + catch e + e isa RustyIceberg.IcebergException ? e : rethrow() end + @test !isnothing(exc) + @test exc.code == RustyIceberg.CATALOG_COMMIT_CONFLICT finally table != C_NULL && free_table(table) free_catalog!(cat) @@ -183,19 +193,39 @@ end try create_namespace(cat, ["ns"]) table = create_table(cat, ["ns"], "t", _ar_schema()) + base_id = table_current_snapshot_id(table) files = RustyIceberg.with_data_file_writer(table) do w write(w, (id=Int64[1], value=[1.0])) end - updated = RustyIceberg.with_transaction(table, cat) do tx - with_overwrite(tx) do action - add_data_files(action, files) - assert_requirements(action, [AssertCurrentSchemaId(0)]) - assert_requirements(action, [AssertLastAssignedFieldId(2)]) + # The first call's requirement is deliberately wrong; if a second call to + # assert_requirements discarded it instead of accumulating onto it, this commit + # would incorrectly succeed on the strength of the second (correct) requirement + # alone. `updated` is still assigned (and freed in `finally`) if that happens, so + # a regression here doesn't also leak a table handle. + exc = try + updated = RustyIceberg.with_transaction(table, cat) do tx + with_overwrite(tx) do action + add_data_files(action, files) + assert_requirements(action, [AssertCurrentSchemaId(999)]) + assert_requirements(action, [AssertLastAssignedFieldId(2)]) + end end + nothing + catch e + e isa RustyIceberg.IcebergException ? e : rethrow() + end + @test !isnothing(exc) + @test exc.code == RustyIceberg.CATALOG_COMMIT_CONFLICT + + # And the catalog must genuinely be untouched by the rejected attempt. + reloaded = load_table(cat, ["ns"], "t") + try + @test table_current_snapshot_id(reloaded) == base_id + finally + free_table(reloaded) end - @test !isnothing(table_current_snapshot_id(updated)) finally table != C_NULL && free_table(table) updated != C_NULL && free_table(updated) @@ -360,14 +390,19 @@ end # meant for create transactions); here it must reach the real catalog check and # be rejected, which is the only way to exercise this requirement kind's JSON # round-trip against an existing table. - @test_throws RustyIceberg.IcebergException begin + exc = try RustyIceberg.with_transaction(table, cat) do tx with_overwrite(tx) do action add_data_files(action, files) assert_requirements(action, [AssertTableDoesNotExist()]) end end + nothing + catch e + e isa RustyIceberg.IcebergException ? e : rethrow() end + @test !isnothing(exc) + @test exc.code == RustyIceberg.CATALOG_COMMIT_CONFLICT finally table != C_NULL && free_table(table) free_catalog!(cat)