feat: share canonical Arrow ingestion schema fingerprints

.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

.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

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"]

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;

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;

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<Vec<u8>, 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::<std::collections::BTreeMap<_, _>>()
+                    .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;

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<String, String> {
+    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::<Vec<_>>(),
+            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()
+    );
+}