From cf24b45c53fb4aaacbb4b01c8b1266e36bf20695 Mon Sep 17 00:00:00 2001 From: Beinan Date: Mon, 19 Oct 2026 11:54:05 +0000 Subject: [PATCH] feat: share canonical Arrow ingestion schema fingerprints --- .github/workflows/rust-test.yml | 4 +- .github/workflows/style.yml | 2 +- Cargo.lock | 1 + crates/lance-context-wal/Cargo.toml | 1 + crates/lance-context-wal/README.md | 13 ++ .../lance-context-wal/src/ingest_contract.rs | 5 + .../src/ingest_contract/arrow.rs | 38 ++++++ .../src/ingest_contract/arrow/tests.rs | 123 ++++++++++++++++++ 8 files changed, 185 insertions(+), 2 deletions(-) create mode 100644 crates/lance-context-wal/src/ingest_contract/arrow.rs create mode 100644 crates/lance-context-wal/src/ingest_contract/arrow/tests.rs diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index bc871519..f28c7cf5 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -64,7 +64,9 @@ jobs: cargo run -p lance-context-wal --features lab --bin context-wal-lab -- codec --payload-bytes 1024 --rows 8 --iterations 2 cargo run -p lance-context-wal --features lab --bin context-wal-lab -- disk --directory "$RUNNER_TEMP/wal-disk-smoke" --payload-bytes 1024 --batch 4 --writers 2 --groups-per-writer 2 --rounds 8 - name: WAL placement, admission, record, contract and IPC decoding tests - run: cargo test -p lance-context-wal --features placement,admission,ipc,record,contract --lib + run: | + cargo test -p lance-context-wal --features placement,admission,ipc,record,arrow-contract --lib + cargo test -p lance-context-wal --features arrow-contract,serde_json/preserve_order --lib ingest_contract::arrow - name: Start etcd for HA scheduler test run: | ETCD_VERSION=3.7.0 diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml index 70598437..de408eb0 100644 --- a/.github/workflows/style.yml +++ b/.github/workflows/style.yml @@ -56,7 +56,7 @@ jobs: - name: Clippy (lance-context, default features) run: cargo clippy --manifest-path crates/lance-context/Cargo.toml --all-targets -- -D warnings - name: Clippy (WAL experiments) - run: cargo clippy -p lance-context-wal --features lab,placement,admission,ipc,record,contract --all-targets -- -D warnings + run: cargo clippy -p lance-context-wal --features lab,placement,admission,ipc,record,arrow-contract --all-targets -- -D warnings typos: name: Spell Check diff --git a/Cargo.lock b/Cargo.lock index 307d850b..6013d877 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -333,6 +333,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f633dbfdf39c039ada1bf9e34c694816eb71fbb7dc78f613993b7245e078a1ed" dependencies = [ "bitflags 2.11.1", + "serde", "serde_core", "serde_json", ] diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 33845e33..824b66e2 100644 --- a/crates/lance-context-wal/Cargo.toml +++ b/crates/lance-context-wal/Cargo.toml @@ -8,6 +8,7 @@ description = "Independent Rust WAL experiments owned by lance-context" [features] contract = ["dep:serde", "dep:sha2"] +arrow-contract = ["contract", "dep:arrow-schema", "arrow-schema/serde", "dep:serde_json"] record = ["dep:bytes", "bytes/serde", "dep:serde", "dep:sha2", "dep:uuid"] placement = ["dep:anyhow", "dep:serde", "dep:serde_json", "dep:sha2", "dep:uuid"] admission = ["dep:anyhow", "dep:serde"] diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index 78dd940b..6f2ecfdb 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -20,6 +20,19 @@ parse Arrow. The module requires neither Arrow nor a replication/log engine. cargo test -p lance-context-wal --features contract --lib ``` +The optional `arrow-contract` feature adds +`ingest_contract::arrow_schema_fingerprint(&Schema)` for ingress and materializers. +It recursively sorts JSON object keys, including nested metadata, and preserves +field order, types, nullability and serialized dictionary attributes. Calculate +it when binding a table generation, then retain the fingerprint on replicas. +Only Arrow schema metadata is required; this feature does not decode IPC or load +a replication engine. Arrow serialization changes require compatibility review +before upgrading writers of existing contracts. + +```bash +cargo test -p lance-context-wal --features arrow-contract --lib +``` + The optional `record` feature provides opaque `Batch`/`BatchMeta` envelopes with shared `Bytes` payload ownership. `BatchMeta::check_body_length` validates frame lengths and a caller-supplied data/control limit before payload allocation; diff --git a/crates/lance-context-wal/src/ingest_contract.rs b/crates/lance-context-wal/src/ingest_contract.rs index 6ff7e32e..78bce4e5 100644 --- a/crates/lance-context-wal/src/ingest_contract.rs +++ b/crates/lance-context-wal/src/ingest_contract.rs @@ -7,6 +7,11 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +#[cfg(feature = "arrow-contract")] +mod arrow; +#[cfg(feature = "arrow-contract")] +pub use arrow::arrow_schema_fingerprint; + /// Protocol ceiling. Deployments may set a smaller maximum in each contract; /// ingress and materializers must both have capacity for that maximum. pub const MAX_RECORD_BYTES: u32 = 255 * 1024 * 1024; diff --git a/crates/lance-context-wal/src/ingest_contract/arrow.rs b/crates/lance-context-wal/src/ingest_contract/arrow.rs new file mode 100644 index 00000000..e64caac8 --- /dev/null +++ b/crates/lance-context-wal/src/ingest_contract/arrow.rs @@ -0,0 +1,38 @@ +use arrow_schema::Schema; +use serde_json::Value; + +/// Fingerprint an Arrow 58 schema for a durable ingestion contract. +/// +/// Sort every JSON object's keys, including nested field metadata, while +/// preserving field/array order and all serialized values. Uses the same domain +/// as [`super::schema_fingerprint`]. This identifies the serialized schema, +/// including dictionary attributes, rather than Arrow's looser schema equality. +/// An Arrow serialization change requires compatibility review before upgrading +/// writers of existing contracts. This does not validate an uploaded IPC body. +/// +/// Compute once when binding a table generation; this allocates schema metadata +/// and is not intended for the per-record replication path. +pub fn arrow_schema_fingerprint(schema: &Schema) -> Result<[u8; 32], serde_json::Error> { + Ok(super::schema_fingerprint(&canonical_document(schema)?)) +} + +fn canonical_document(schema: &Schema) -> Result, serde_json::Error> { + fn canonical(value: Value) -> Value { + match value { + Value::Object(fields) => Value::Object( + fields + .into_iter() + .map(|(key, value)| (key, canonical(value))) + .collect::>() + .into_iter() + .collect(), + ), + Value::Array(values) => Value::Array(values.into_iter().map(canonical).collect()), + value => value, + } + } + serde_json::to_vec(&canonical(serde_json::to_value(schema)?)) +} + +#[cfg(test)] +mod tests; diff --git a/crates/lance-context-wal/src/ingest_contract/arrow/tests.rs b/crates/lance-context-wal/src/ingest_contract/arrow/tests.rs new file mode 100644 index 00000000..37d53d70 --- /dev/null +++ b/crates/lance-context-wal/src/ingest_contract/arrow/tests.rs @@ -0,0 +1,123 @@ +use super::*; +use arrow_schema::{DataType, Field}; +use std::{collections::HashMap, sync::Arc}; + +fn metadata(reverse: bool) -> HashMap { + let pairs = if reverse { + [("zeta", "last"), ("alpha", "first")] + } else { + [("alpha", "first"), ("zeta", "last")] + }; + pairs + .into_iter() + .map(|(k, v)| (k.into(), v.into())) + .collect() +} + +#[test] +fn canonical_bytes_match_fixed_arrow_58_document() { + let schema = Schema::new_with_metadata( + vec![Field::new("event", DataType::Utf8, false).with_metadata(metadata(true))], + [("owner".into(), "public".into())].into(), + ); + let expected = concat!( + "{\"fields\":[{\"data_type\":\"Utf8\",\"dict_id\":0,\"dict_is_ordered\":false,", + "\"metadata\":{\"alpha\":\"first\",\"zeta\":\"last\"},\"name\":\"event\",\"nullable\":false}],", + "\"metadata\":{\"owner\":\"public\"}}" + ); + assert_eq!(canonical_document(&schema).unwrap(), expected.as_bytes()); + // SHA-256 over the domain, little-endian u64 byte length and the fixed + // document above, independently calculated with OpenSSL. + assert_eq!( + arrow_schema_fingerprint(&schema).unwrap(), + [ + 0x92, 0xd1, 0xbf, 0x40, 0x10, 0xe1, 0x9a, 0xf8, 0x01, 0x7a, 0xe8, 0x0f, 0x7e, 0xa9, + 0x4b, 0x46, 0x67, 0x3b, 0xa3, 0xc1, 0xee, 0x41, 0x7e, 0x4e, 0xf9, 0x62, 0x49, 0xae, + 0x2e, 0x49, 0x2a, 0xab, + ] + ); +} + +fn nested(reverse: bool) -> Schema { + let child = + Arc::new(Field::new("child", DataType::Utf8, false).with_metadata(metadata(reverse))); + Schema::new_with_metadata( + vec![ + Field::new("nested", DataType::List(child), false).with_metadata(metadata(reverse)), + Field::new("other", DataType::UInt64, false), + ], + metadata(reverse), + ) +} + +#[test] +fn metadata_order_and_json_round_trip_preserve_identity() { + let expected = arrow_schema_fingerprint(&nested(false)).unwrap(); + // Each fresh HashMap has an independent iteration seed. Nested metadata and + // schema metadata must stay stable across service restarts and JSON reloads. + for attempt in 0..32 { + let schema = nested(attempt % 2 == 0); + assert_eq!(arrow_schema_fingerprint(&schema).unwrap(), expected); + let restored: Schema = + serde_json::from_slice(&serde_json::to_vec(&schema).unwrap()).unwrap(); + assert_eq!(arrow_schema_fingerprint(&restored).unwrap(), expected); + } +} + +#[test] +fn identity_preserves_field_order_type_nullability_and_metadata_values() { + let original = nested(false); + let expected = arrow_schema_fingerprint(&original).unwrap(); + let mut variants = vec![ + Schema::new_with_metadata( + original.fields().iter().rev().cloned().collect::>(), + original.metadata().clone(), + ), + original.clone().with_metadata( + metadata(true) + .into_iter() + .map(|(k, v)| (k, v + "!")) + .collect(), + ), + ]; + for changed in [ + Field::new("other", DataType::Int64, false), + Field::new("other", DataType::UInt64, true), + Field::new("renamed", DataType::UInt64, false), + ] { + variants.push(Schema::new_with_metadata( + vec![original.field(0).clone(), changed], + original.metadata().clone(), + )); + } + let changed_child = Arc::new( + Field::new("child", DataType::Utf8, false) + .with_metadata([("alpha".into(), "changed".into())].into()), + ); + variants.push(Schema::new_with_metadata( + vec![ + Field::new("nested", DataType::List(changed_child), false) + .with_metadata(metadata(false)), + original.field(1).clone(), + ], + original.metadata().clone(), + )); + for schema in variants { + assert_ne!(arrow_schema_fingerprint(&schema).unwrap(), expected); + } +} + +#[test] +fn dictionary_attributes_remain_part_of_the_serialized_contract() { + let field = Field::new( + "label", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)), + false, + ); + let plain = Schema::new(vec![field.clone()]); + let ordered = Schema::new(vec![field.with_dict_is_ordered(true)]); + assert_ne!( + arrow_schema_fingerprint(&plain).unwrap(), + arrow_schema_fingerprint(&ordered).unwrap() + ); +}