diff --git a/crates/penca-api/src/lifecycle/compact.rs b/crates/penca-api/src/lifecycle/compact.rs index ef5c1264..12263a48 100644 --- a/crates/penca-api/src/lifecycle/compact.rs +++ b/crates/penca-api/src/lifecycle/compact.rs @@ -159,6 +159,7 @@ impl<'a> PersistScope<'a> { // honoring the ceiling AND the slice arithmetic has to be reworked // for short reads. max_commit_seq_num: Some(row.get("max_commit_seq_num")), + content_hash: row.get("content_hash"), }) } } @@ -320,6 +321,15 @@ where } else { 0 }; + // Recomputed per slice rather than carried over from the input row. + // The rows are preserved 1:1, but `concat_batches` normalizes every + // input to one `segment_schema`, so a slice of the merged batch can + // decode to a wider type than its input file did. Carrying the old hash + // would leave two different decodes under one cache key. + let slice_hash = penca_core::digest::segment_content_hash( + &merged.slice(cumulative as usize, meta.row_count as usize), + ) + .map_err(ApiError::Arrow)?; LifecycleManager::repoint_table_persist_segment( &tx, &catalog_str, @@ -331,6 +341,7 @@ where proportional_size, storage_format_text, &merged_stats, + &slice_hash, false, ) .await?; diff --git a/crates/penca-api/src/lifecycle/durable_writer.rs b/crates/penca-api/src/lifecycle/durable_writer.rs index be8db2c6..ee8bc5a2 100644 --- a/crates/penca-api/src/lifecycle/durable_writer.rs +++ b/crates/penca-api/src/lifecycle/durable_writer.rs @@ -24,6 +24,7 @@ use penca_db::driver::pg::PgDriver; use penca_format::writer::FormatWriter; use penca_storage_cold::ColdStorageClient; use penca_storage_meta::LifecycleManager; +use uuid::Uuid; use crate::error::ApiError; @@ -219,6 +220,12 @@ impl<'a> SegmentScope for PersistSegmentScope<'a> { async fn insert_segment(&self, pool: &PgDriver, step: &Self::Step) -> Result<(), ApiError> { let statistics = penca_dl::stats::compute_segment_statistics(&step.batch); + // Hashed here, on the in-memory batch, rather than after `write_file`: + // the digest identifies the logical content handed to the format writer, + // which is what the segment cache stores post-decode. It is deliberately + // not a checksum of the object on S3 — see `segment_content_hash`. + let content_hash = + penca_core::digest::segment_content_hash(&step.batch).map_err(ApiError::Arrow)?; LifecycleManager::insert_table_persist_segment( pool, self.catalog_str, @@ -235,6 +242,7 @@ impl<'a> SegmentScope for PersistSegmentScope<'a> { step.num_rows, self.storage_format.extension(), &statistics, + &content_hash, ) .await?; Ok(()) @@ -324,6 +332,8 @@ impl<'a> SegmentScope for SnapshotSegmentScope<'a> { async fn insert_segment(&self, pool: &PgDriver, step: &Self::Step) -> Result<(), ApiError> { let statistics = penca_dl::stats::compute_segment_statistics(&step.batch); + let content_hash = + penca_core::digest::segment_content_hash(&step.batch).map_err(ApiError::Arrow)?; LifecycleManager::insert_snapshot_segment( pool, self.catalog_str, @@ -341,6 +351,7 @@ impl<'a> SegmentScope for SnapshotSegmentScope<'a> { step.num_rows, self.storage_format.extension(), &statistics, + &content_hash, ) .await?; Ok(()) @@ -409,9 +420,10 @@ pub(super) struct SnapshotFileStep { /// One per-partition segment metadata row inside a packed file. /// `offset`/`length` are the partition's row range within the file — /// `length` doubles as the catalog `row_count` (a packed row IS its -/// row range; both columns are NOT NULL). `size_bytes` -/// and `statistics` are computed over the slice only, so pruning -/// stats stay partition-tight. +/// row range; both columns are NOT NULL). `size_bytes`, +/// `statistics` and `content_hash` are computed over the slice only, so +/// pruning stats stay partition-tight and the hash identifies the rows +/// this one row covers rather than the whole packed file. pub(super) struct SnapshotSegmentRowSpec { pub seg_uuid_str: String, pub chunk_idx: u32, @@ -425,6 +437,7 @@ pub(super) struct SnapshotSegmentRowSpec { pub length: i64, pub size_bytes: i64, pub statistics: Vec, + pub content_hash: Uuid, } impl<'a> DurableSegmentWriter> { @@ -456,6 +469,7 @@ impl<'a> DurableSegmentWriter> { row.length, self.scope.storage_format.extension(), &row.statistics, + &row.content_hash, ) .await?; self.current_group() diff --git a/crates/penca-api/src/lifecycle/packer.rs b/crates/penca-api/src/lifecycle/packer.rs index 25d9679b..90babb5a 100644 --- a/crates/penca-api/src/lifecycle/packer.rs +++ b/crates/penca-api/src/lifecycle/packer.rs @@ -268,9 +268,10 @@ impl SegmentPacker { } /// Concatenate the buffered partitions into one file step with one - /// segment row per partition. `size_bytes` and `statistics` are - /// computed per partition slice, not per file, so pruning stats - /// stay partition-tight. + /// segment row per partition. `size_bytes`, `statistics` and + /// `content_hash` are computed per partition slice, not per file, so + /// pruning stats stay partition-tight and each row's hash identifies + /// its own rows rather than the packed file they share. fn flush(&mut self) -> Result, ApiError> { if self.buffered.is_empty() { return Ok(None); @@ -297,6 +298,8 @@ impl SegmentPacker { length: num_rows, size_bytes: batch_in_memory_bytes(batch)?, statistics: penca_dl::stats::compute_segment_statistics(batch), + content_hash: penca_core::digest::segment_content_hash(batch) + .map_err(ApiError::Arrow)?, }); self.chunk_idx += 1; offset += num_rows; @@ -327,6 +330,8 @@ impl SegmentPacker { length: num_rows, size_bytes: chunk_bytes, statistics: penca_dl::stats::compute_segment_statistics(&chunk), + content_hash: penca_core::digest::segment_content_hash(&chunk) + .map_err(ApiError::Arrow)?, }; self.chunk_idx += 1; Ok(SnapshotFileStep { diff --git a/crates/penca-api/src/lifecycle/snapshot_op.rs b/crates/penca-api/src/lifecycle/snapshot_op.rs index 19452475..4b7a4808 100644 --- a/crates/penca-api/src/lifecycle/snapshot_op.rs +++ b/crates/penca-api/src/lifecycle/snapshot_op.rs @@ -82,7 +82,7 @@ fn empty_merge_placeholder_step( base_uri: &str, storage_format_text: &str, user_schema: &SchemaRef, -) -> SnapshotFileStep { +) -> Result { let placeholder_batch = RecordBatch::new_empty(penca_merge::snapshot_read_schema(user_schema)); let seg_uuid = table_snapshot_segment_uuid(snap_uuid, 0); let uri = snapshot_segment_uri( @@ -94,7 +94,11 @@ fn empty_merge_placeholder_step( storage_format_text, ); let statistics = penca_dl::stats::compute_segment_statistics(&placeholder_batch); - SnapshotFileStep { + // Every empty placeholder over one schema is the same zero-row content, so + // they legitimately collapse onto one cache entry. + let content_hash = + penca_core::digest::segment_content_hash(&placeholder_batch).map_err(ApiError::Arrow)?; + Ok(SnapshotFileStep { snap_uuid_str: snap_str.to_string(), uri, file_batch: placeholder_batch, @@ -106,8 +110,9 @@ fn empty_merge_placeholder_step( length: 0, size_bytes: 0, statistics, + content_hash, }], - } + }) } impl LifecycleManager { @@ -1539,7 +1544,7 @@ impl LifecycleManager { &self.base_uri, ctx.storage_format_text, ctx.user_schema, - ); + )?; seg_writer .write_segment_group(pool, writer, &placeholder) .await?; @@ -1912,6 +1917,10 @@ async fn build_one_segment_sidecar( // sorted the keys, so the bounds are the first/last entry). Deferred so // the seek owns the exact bound encoding rather than guessing it here. &[], + // The sidecar's own content, not the base segment's: two segments with + // different rows can build byte-identical indexes, and it is the index + // batch that this row's cache entry holds. + &penca_core::digest::segment_content_hash(&sidecar).map_err(ApiError::Arrow)?, ) .await?; built_uris.push(uri); diff --git a/crates/penca-api/src/query/meta_plan.rs b/crates/penca-api/src/query/meta_plan.rs index c72e864e..b362c242 100644 --- a/crates/penca-api/src/query/meta_plan.rs +++ b/crates/penca-api/src/query/meta_plan.rs @@ -784,7 +784,7 @@ impl QueryManager { }; let snapshot_sql = format!( "SELECT seg.table_snapshot_segment_uuid, seg.object_uri, \ - seg.\"offset\", seg.length, seg.format, \ + seg.\"offset\", seg.length, seg.format, seg.content_hash, \ snap.snapshotted_at_micros, snap.commit_seq_num, \ seg.table_snapshot_uuid, seg.row_count, \ seg.size_bytes, seg.metadata, seg.statistics, \ @@ -794,7 +794,8 @@ impl QueryManager { c.object_uri AS sidecar_object_uri, \ c.\"offset\" AS sidecar_offset, c.length AS sidecar_length, \ c.format AS sidecar_format, c.size_bytes AS sidecar_size_bytes, \ - c.segment_index_uuid AS sidecar_segment_index_uuid \ + c.segment_index_uuid AS sidecar_segment_index_uuid, \ + c.content_hash AS sidecar_content_hash \ FROM {seg_table} seg \ INNER JOIN {snap_table} snap \ ON seg.table_snapshot_uuid = snap.table_snapshot_uuid \ @@ -909,6 +910,7 @@ impl QueryManager { size_bytes: row.get("size_bytes"), metadata_json: metadata.to_string(), statistics: statistics.unwrap_or_default(), + content_hash: row.get("content_hash"), row_uuid_index_sidecar: None, index_sidecars: Vec::new(), }); @@ -1264,7 +1266,7 @@ impl QueryManager { "SELECT tfm.log_kind, \ seg.table_persist_segment_uuid AS segment_uuid, \ seg.object_uri, seg.\"offset\", seg.length, \ - seg.row_count, seg.format, \ + seg.row_count, seg.format, seg.content_hash, \ seg.min_tx_commit_micros, \ seg.max_tx_commit_micros, \ seg.max_commit_seq_num, \ @@ -1340,6 +1342,7 @@ impl QueryManager { // segment this equals the file's true maximum, so the ceiling // costs a no-op filter and needs no per-row special case. max_commit_seq_num: Some(row.get("max_commit_seq_num")), + content_hash: row.get("content_hash"), }; match log_kind { @@ -1450,9 +1453,10 @@ pub(crate) struct HotTableNames { /// Decode one row's LEFT-JOINed child-sidecar columns into an /// [`IndexSidecar`], or `None` when no child matched (NULL /// `sidecar_object_uri`). The object_uri-presence gate stands in for "the -/// whole child row matched": offset/length/format/segment_index_uuid are all -/// NOT NULL in the schema, so the non-Option `row.get`s cannot hit a NULL -/// while object_uri is present. Revisit if a nullable child column is added. +/// whole child row matched": offset/length/format/segment_index_uuid/ +/// content_hash are all NOT NULL in the schema, so the non-Option `row.get`s +/// cannot hit a NULL while object_uri is present. Revisit if a nullable child +/// column is added. fn decode_child_sidecar(row: &PgRow) -> Result> { row.get::, _>("sidecar_object_uri") .map(|object_uri| -> Result { @@ -1471,6 +1475,10 @@ fn decode_child_sidecar(row: &PgRow) -> Result> { format: sidecar_format, segment_index_uuid: row.get::("sidecar_segment_index_uuid").to_string(), size_bytes: row.get::, _>("sidecar_size_bytes").unwrap_or(0), + // Non-Option deliberately, unlike `size_bytes` directly above: + // an `unwrap_or(Uuid::nil())` here would hand every sidecar the + // same cache key and serve segment X's index for segment Y. + content_hash: row.get::("sidecar_content_hash"), }) }) .transpose() @@ -1785,6 +1793,7 @@ mod assemble_tests { offset: None, length: None, max_commit_seq_num: None, + content_hash: penca_core::naming::deterministic_uuid_from(&[uuid]), } } @@ -1956,6 +1965,7 @@ mod assemble_tests { format: Format::Parquet, segment_index_uuid: uri.to_string(), size_bytes: 1, + content_hash: penca_core::naming::deterministic_uuid_from(&[uri]), } } diff --git a/crates/penca-core/src/digest.rs b/crates/penca-core/src/digest.rs new file mode 100644 index 00000000..e9db1bd9 --- /dev/null +++ b/crates/penca-core/src/digest.rs @@ -0,0 +1,609 @@ +//! Content digest of a decoded cold segment (CHA-545). + +use std::sync::Arc; + +use arrow::array::{ + Array, ArrayRef, BinaryViewArray, FixedSizeListArray, GenericListArray, LargeListArray, + ListArray, OffsetSizeTrait, RecordBatch, StringViewArray, +}; +use arrow::buffer::OffsetBuffer; +use arrow::datatypes::{DataType, FieldRef}; +use arrow::error::ArrowError; +use arrow::ipc::MetadataVersion; +use arrow::ipc::writer::{DictionaryHandling, IpcWriteOptions, StreamWriter}; +use uuid::Uuid; +use xxhash_rust::xxh3::xxh3_128; + +/// Alignment for the canonical IPC encoding. Pinned rather than defaulted so +/// the digest does not silently change under an arrow-rs upgrade. +const IPC_ALIGNMENT: usize = 8; + +/// `xxh3_128` of a segment's **typed in-memory Arrow batch**, as a [`Uuid`]. +/// +/// The input is the decoded batch, never the encoded file bytes, so everything +/// that distinguishes two decoded batches is in the digest: values, types, column +/// names, nullability, column order, row count. That is the soundness property a +/// cache keyed by this value needs — equal hash must mean equal decode, or two +/// unrelated segments would share one entry. +/// +/// It is **not** a defence against a reader whose schema differs from the +/// writer's. A reference copy inherits this value verbatim (`fork_copy`, snapshot +/// carry-forward) and it is never recomputed, so a fork that `ALTER`s a column +/// still reads the parent's bytes under the parent's hash — no digest could +/// separate the two, because neither row's was ever taken under its own read +/// schema. Serving both from one entry safely is the cached *value's* job: it +/// holds the file-native decode and callers shape after the lookup. See +/// `penca_dl::cache::SegmentCache`. +/// +/// **Write-time only.** A digest taken before the format writer encodes a batch +/// is *not* guaranteed to equal the digest of a later decode of that file: +/// Parquet may widen a type, re-dictionary-encode, or normalize on round-trip. +/// Dedup never compares the two — the value is computed once at write and +/// inherited by reference copies, never recomputed. CHA-545 names checksum +/// reuse as a possible future use of this digest; that use requires +/// establishing round-trip stability first, and this function does not. +/// +/// **Slice-invariant.** A batch from `slice()` stays a view onto its parent's +/// buffers, which physically hold bytes outside the slice, but the IPC writer +/// encodes only the logical range. Measured on arrow-rs 57.3 across the +/// supported set (see [`crate::types`]): primitives, decimals, +/// dates/times/timestamps, `Boolean` at a non-byte-aligned offset, +/// `Utf8`/`LargeUtf8`/`Binary`, and `List`/`LargeList`/`FixedSizeList` all hold +/// unaided. `Utf8View`/`BinaryView` do not, and [`compact_view_buffers`] below +/// is what makes them. `Dictionary` fails the same way under +/// `DictionaryHandling::Resend` but is rejected at the type boundary and cannot +/// reach here. +/// +/// Changing the IPC options or the hash changes every future digest. That is +/// safe — stored digests are opaque identity, never recomputed and compared +/// against a stored value — but it must stay deliberate, which is why nothing +/// here relies on a library default. +pub fn segment_content_hash(batch: &RecordBatch) -> Result { + let options = IpcWriteOptions::try_new(IPC_ALIGNMENT, false, MetadataVersion::V5)? + // Pinned for the same reason as the alignment: the default is `Resend` + // today, and a future flip to `Delta` would make a batch's digest + // depend on what the writer had already emitted. + .with_dictionary_handling(DictionaryHandling::Resend); + let compacted = compact_view_buffers(batch)?; + let mut bytes = Vec::new(); + let mut writer = StreamWriter::try_new_with_options(&mut bytes, &compacted.schema(), options)?; + writer.write(&compacted)?; + writer.finish()?; + drop(writer); + + Ok(Uuid::from_u128(xxh3_128(&bytes))) +} + +/// Rebuild every byte-view column so its data buffers hold only the rows the +/// batch covers. +/// +/// The IPC writer truncates each buffer it can, with one exception: for +/// `Utf8View`/`BinaryView` it emits every variadic data buffer **whole**, since +/// proving no surviving view still points into a pruned buffer is expensive +/// (arrow-ipc 57.3 `writer::write_array_data` says so and points at `gc`). A +/// digest of a slice would otherwise encode the whole parent's string payload, +/// and both `compact` and the snapshot packer take one digest per slice of a +/// single `concat_batches` result — so a wave of `N` inputs would re-serialize +/// every input's bytes `N` times, on exactly the many-small-segments workload +/// compaction exists for. +/// +/// Compacting first bounds each digest to its own rows. It also makes the +/// digest *canonical* for these types rather than merely cheaper: `gc` lays the +/// referenced bytes out in row order in buffers sized by the values alone, so +/// two arrays holding the same values agree even when their parents laid those +/// values out differently — which is the dedup this digest exists for. +/// +/// [`CanonicalType`](crate::types::CanonicalType) admits a view at the top level +/// or as the child of a single-level list, and rejects nesting below that, so +/// one level of descent is total over the supported set. +fn compact_view_buffers(batch: &RecordBatch) -> Result { + if !batch + .schema() + .fields() + .iter() + .any(|f| holds_byte_view(f.data_type())) + { + return Ok(batch.clone()); + } + let columns = batch + .columns() + .iter() + .map(compact_column) + .collect::, _>>()?; + // `try_new` re-validates against the schema, which is what catches a + // compaction that changed a length or a type rather than just a layout. + RecordBatch::try_new(batch.schema(), columns) +} + +/// Whether a column of this type carries variadic data buffers the IPC writer +/// would emit whole. +fn holds_byte_view(dt: &DataType) -> bool { + match dt { + DataType::Utf8View | DataType::BinaryView => true, + DataType::List(child) | DataType::LargeList(child) | DataType::FixedSizeList(child, _) => { + matches!(child.data_type(), DataType::Utf8View | DataType::BinaryView) + } + _ => false, + } +} + +fn compact_column(array: &ArrayRef) -> Result { + match array.data_type() { + DataType::Utf8View => Ok(Arc::new(downcast::(array)?.gc())), + DataType::BinaryView => Ok(Arc::new(downcast::(array)?.gc())), + DataType::List(child) if holds_byte_view(array.data_type()) => { + compact_list(downcast::(array)?, child) + } + DataType::LargeList(child) if holds_byte_view(array.data_type()) => { + compact_list(downcast::(array)?, child) + } + DataType::FixedSizeList(child, size) if holds_byte_view(array.data_type()) => { + // Unlike the offset flavours, `FixedSizeListArray::slice` slices its + // child too, so `values()` is already this array's own range. + let list = downcast::(array)?; + // Carry the length rather than let `try_new` re-derive it: + // `CanonicalType` accepts size 0, and at size 0 `try_new` reads the + // length off the null buffer, so a null-free column would rebuild + // with zero rows. + Ok(Arc::new(FixedSizeListArray::try_new_with_length( + child.clone(), + *size, + compact_column(list.values())?, + list.nulls().cloned(), + list.len(), + )?)) + } + _ => Ok(Arc::clone(array)), + } +} + +/// `GenericListArray::slice` keeps the whole child and slices only the offsets, +/// so compacting the child as-is would compact rows this list does not cover. +/// Narrow it to the range the offsets span, then rebase them onto it. +fn compact_list( + list: &GenericListArray, + child: &FieldRef, +) -> Result { + let offsets = list.offsets(); + let start = offsets[0].as_usize(); + let end = offsets[list.len()].as_usize(); + let values = compact_column(&list.values().slice(start, end - start))?; + let rebased: Vec = offsets.iter().map(|o| *o - offsets[0]).collect(); + Ok(Arc::new(GenericListArray::::try_new( + child.clone(), + OffsetBuffer::new(rebased.into()), + values, + list.nulls().cloned(), + )?)) +} + +fn downcast(array: &ArrayRef) -> Result<&A, ArrowError> { + array.as_any().downcast_ref::().ok_or_else(|| { + ArrowError::InvalidArgumentError(format!( + "{} column is not the array kind its type declares", + array.data_type() + )) + }) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use arrow::array::{Int64Array, ListArray, StringArray, StringViewArray}; + use arrow::datatypes::{DataType, Field, Int64Type, Schema}; + + use super::*; + + fn batch(fields: Vec, columns: Vec) -> RecordBatch { + RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).expect("valid fixture") + } + + fn i64_col(name: &str, values: Vec>) -> (Field, arrow::array::ArrayRef) { + ( + Field::new(name, DataType::Int64, true), + Arc::new(Int64Array::from(values)), + ) + } + + fn one_i64(name: &str, values: Vec>) -> RecordBatch { + let (f, c) = i64_col(name, values); + batch(vec![f], vec![c]) + } + + fn hash(b: &RecordBatch) -> Uuid { + segment_content_hash(b).expect("digest") + } + + #[test] + fn identical_batches_hash_equal() { + let a = one_i64("a", vec![Some(1), Some(2), Some(3)]); + let b = one_i64("a", vec![Some(1), Some(2), Some(3)]); + assert_eq!(hash(&a), hash(&a), "same batch hashed twice is stable"); + assert_eq!( + hash(&a), + hash(&b), + "independently built identical batches agree" + ); + } + + #[test] + fn differing_values_hash_differently() { + let a = one_i64("a", vec![Some(1), Some(2), Some(3)]); + let b = one_i64("a", vec![Some(1), Some(9), Some(3)]); + assert_ne!( + hash(&a), + hash(&b), + "one differing cell must change the hash" + ); + } + + #[test] + fn null_differs_from_value_in_the_same_cell() { + let a = one_i64("a", vec![Some(1), Some(2)]); + let b = one_i64("a", vec![Some(1), None]); + assert_ne!( + hash(&a), + hash(&b), + "null is not the same content as a value" + ); + } + + /// The schema-divergence case the whole ticket exists for: same column name, + /// same values as rendered, different type. A byte-level hash of the file + /// would collide these; hashing the typed batch must not. + #[test] + fn same_values_under_different_column_type_hash_differently() { + let as_int = one_i64("a", vec![Some(1), Some(2)]); + let as_utf8 = batch( + vec![Field::new("a", DataType::Utf8, true)], + vec![Arc::new(StringArray::from(vec![Some("1"), Some("2")]))], + ); + assert_ne!( + hash(&as_int), + hash(&as_utf8), + "Int64 and Utf8 columns are different decoded objects" + ); + } + + #[test] + fn column_name_is_part_of_the_hash() { + let a = one_i64("a", vec![Some(1), Some(2)]); + let b = one_i64("b", vec![Some(1), Some(2)]); + assert_ne!(hash(&a), hash(&b), "column name must change the hash"); + } + + #[test] + fn nullability_flag_is_part_of_the_hash() { + let nullable = batch( + vec![Field::new("a", DataType::Int64, true)], + vec![Arc::new(Int64Array::from(vec![1_i64, 2]))], + ); + let non_nullable = batch( + vec![Field::new("a", DataType::Int64, false)], + vec![Arc::new(Int64Array::from(vec![1_i64, 2]))], + ); + assert_ne!( + hash(&nullable), + hash(&non_nullable), + "nullability is schema, and schema is part of the identity" + ); + } + + #[test] + fn column_order_is_part_of_the_hash() { + let (fa, ca) = i64_col("a", vec![Some(1), Some(2)]); + let (fb, cb) = i64_col("b", vec![Some(3), Some(4)]); + let ab = batch(vec![fa.clone(), fb.clone()], vec![ca.clone(), cb.clone()]); + let ba = batch(vec![fb, fa], vec![cb, ca]); + assert_ne!(hash(&ab), hash(&ba), "column order must change the hash"); + } + + #[test] + fn row_count_is_part_of_the_hash() { + let empty = one_i64("a", vec![]); + let one = one_i64("a", vec![Some(1)]); + let many = one_i64("a", vec![Some(1), Some(2), Some(3)]); + assert_ne!(hash(&empty), hash(&one)); + assert_ne!(hash(&one), hash(&many)); + assert_eq!( + hash(&empty), + hash(&one_i64("a", vec![])), + "zero rows still hashes deterministically" + ); + } + + /// A sliced batch and an independently built batch of the same logical rows + /// must hash equal — the digest is over content, not representation. + /// + /// Uses Utf8 because that is where a slice demonstrably stays a *view* — the + /// guard below asserts it still shares its parent's values buffer, so that + /// buffer physically holds bytes outside the slice. (`RecordBatch::slice` + /// reports `offset() == 0` even here, so asserting on the offset would + /// silently pass on a fixture that proves nothing.) + #[test] + fn slice_hashes_equal_to_an_independently_built_batch() { + let parent = batch( + vec![Field::new("s", DataType::Utf8, true)], + vec![Arc::new(StringArray::from(vec!["aa", "bb", "cc", "dd"]))], + ); + let sliced = parent.slice(1, 2); + let standalone = batch( + vec![Field::new("s", DataType::Utf8, true)], + vec![Arc::new(StringArray::from(vec!["bb", "cc"]))], + ); + + assert_eq!( + parent.column(0).to_data().buffers()[1].as_ptr(), + sliced.column(0).to_data().buffers()[1].as_ptr(), + "fixture must still be a view onto the parent's buffer, or this proves nothing" + ); + assert_eq!( + hash(&sliced), + hash(&standalone), + "the digest is over logical content, not buffer position" + ); + } + + /// The primitive twin. `slice()` rebases the data pointer here rather than + /// keeping a logical offset, so this is the cheaper case — asserted anyway + /// because both paths reach `segment_content_hash` in production. + #[test] + fn primitive_slice_hashes_equal_to_an_independently_built_batch() { + let parent = one_i64("a", vec![Some(1), None, Some(3), Some(4)]); + assert_eq!( + hash(&parent.slice(1, 2)), + hash(&one_i64("a", vec![None, Some(3)])), + "sliced primitive rows hash as their own content" + ); + } + + /// The nested twin, and the reason a `List` of a non-view child needs no + /// compaction step. Slicing a `List` rebases the offsets buffer but leaves + /// the *child* array whole — the guard below pins that it still holds all + /// four rows' values, shared with the parent — and the IPC writer still + /// encodes only the logical range. + /// + /// This is the case an earlier revision assumed a `concat_batches` call was + /// protecting. It was not: `concat` short-circuits single-array input to + /// `slice(0, len)`, which rebuilds nothing. + #[test] + fn list_slice_hashes_equal_to_an_independently_built_batch() { + fn lists(rows: Vec>) -> RecordBatch { + let arr = ListArray::from_iter_primitive::( + rows.into_iter() + .map(|r| Some(r.into_iter().map(Some).collect::>())), + ); + let field = Field::new( + "l", + DataType::List(Arc::new(Field::new("item", DataType::Int64, true))), + true, + ); + batch(vec![field], vec![Arc::new(arr)]) + } + + let parent = lists(vec![vec![1], vec![2, 2], vec![3, 3, 3], vec![4]]); + let sliced = parent.slice(1, 2); + + let sliced_data = sliced.column(0).to_data(); + let parent_data = parent.column(0).to_data(); + assert_eq!( + sliced_data.child_data()[0].len(), + parent_data.child_data()[0].len(), + "fixture must keep the parent's whole child array, or this proves nothing" + ); + assert_eq!( + hash(&sliced), + hash(&lists(vec![vec![2, 2], vec![3, 3, 3]])), + "a sliced list column hashes as its own logical rows" + ); + } + + /// Above the 12-byte inline threshold, so the values live in a data buffer + /// rather than in the view word itself — which is the only case where a + /// view array has variadic buffers to carry. + const LONG: [&str; 3] = [ + "aaaaaaaaaaaaaaaaaaaa", + "bbbbbbbbbbbbbbbbbbbb", + "cccccccccccccccccccc", + ]; + + fn views(v: Vec<&str>) -> RecordBatch { + batch( + vec![Field::new("s", DataType::Utf8View, true)], + vec![Arc::new(StringViewArray::from(v))], + ) + } + + /// The type the IPC writer will *not* truncate: it emits each variadic data + /// buffer whole rather than prove no surviving view still points into a + /// pruned one, so a slice's encoding would otherwise carry bytes belonging + /// to rows it does not cover. [`compact_view_buffers`] is what closes that, + /// so the fixture guard asserting the slice really does still share the + /// parent's data buffer is what makes the equality below non-trivial. + #[test] + fn sliced_view_column_hashes_equal_to_an_independently_built_batch() { + let parent = views(LONG.to_vec()); + let sliced = parent.slice(1, 2); + + assert_eq!( + parent.column(0).to_data().buffers()[1].as_ptr(), + sliced.column(0).to_data().buffers()[1].as_ptr(), + "fixture must still share the parent's data buffer, or this proves nothing" + ); + assert_eq!( + hash(&sliced), + hash(&views(vec![LONG[1], LONG[2]])), + "a sliced view column hashes as its own logical rows" + ); + // One long value anywhere in the parent gives the array a data buffer, + // so a slice of only *short* rows carries it too — compaction has to be + // driven by the parent's layout, not by the sliced rows' lengths. + assert_eq!( + hash(&views(vec!["aa", "bb", LONG[0]]).slice(0, 2)), + hash(&views(vec!["aa", "bb"])), + "an all-inline slice of a buffer-carrying parent is invariant too" + ); + } + + /// Independently written segments holding the same rows must agree even when + /// their parents laid the bytes out differently — that agreement *is* the + /// dedup. Views make this a real risk rather than a tautology: the same + /// values reached through different builders can sit at different buffer + /// offsets, which `gc` normalizes away. + #[test] + fn view_column_hashes_equal_across_differently_laid_out_parents() { + assert_eq!( + hash(&views(vec![LONG[0], LONG[1]])), + hash(&views(vec![LONG[2], LONG[0], LONG[1]]).slice(1, 2)), + "same values, different parent layout, one hash" + ); + } + + /// The `BinaryView` twin of the arm above. Same buffer mechanics, separate + /// downcast — a swapped array type here would be caught by nothing else. + #[test] + fn sliced_binary_view_column_hashes_equal_to_an_independently_built_batch() { + let binaries = |v: Vec<&str>| { + batch( + vec![Field::new("b", DataType::BinaryView, true)], + vec![Arc::new(BinaryViewArray::from_iter_values( + v.into_iter().map(str::as_bytes), + ))], + ) + }; + assert_eq!( + hash(&binaries(LONG.to_vec()).slice(1, 2)), + hash(&binaries(vec![LONG[1], LONG[2]])), + "a sliced binary-view column hashes as its own logical rows" + ); + } + + /// A single-level list of views, in both offset widths. + /// [`crate::types::CanonicalType`] admits a view there, and slicing leaves + /// *both* the child array and its data buffers whole — so the compaction has + /// to narrow the child to the offsets' range before it can help. + fn list_of_views(rows: Vec>) -> RecordBatch { + let child = Arc::new(Field::new("item", DataType::Utf8View, true)); + let mut values: Vec<&str> = Vec::new(); + let mut offsets: Vec = vec![O::zero()]; + for row in &rows { + values.extend(row.iter().copied()); + offsets.push(O::from_usize(values.len()).expect("fixture fits the offset width")); + } + let arr = GenericListArray::::try_new( + Arc::clone(&child), + OffsetBuffer::new(offsets.into()), + Arc::new(StringViewArray::from(values)), + None, + ) + .expect("valid fixture"); + let data_type = if O::IS_LARGE { + DataType::LargeList(child) + } else { + DataType::List(child) + }; + batch(vec![Field::new("l", data_type, true)], vec![Arc::new(arr)]) + } + + fn assert_sliced_list_of_views_is_invariant() { + let parent = list_of_views::(vec![ + vec![LONG[0]], + vec![LONG[1], LONG[2]], + vec![LONG[0], LONG[2]], + ]); + let sliced = parent.slice(1, 1); + + assert_eq!( + sliced.column(0).to_data().child_data()[0].len(), + parent.column(0).to_data().child_data()[0].len(), + "fixture must keep the parent's whole child array, or this proves nothing" + ); + assert_eq!( + hash(&sliced), + hash(&list_of_views::(vec![vec![LONG[1], LONG[2]]])), + "a sliced list-of-view column hashes as its own logical rows" + ); + } + + #[test] + fn sliced_list_of_view_column_hashes_equal_to_an_independently_built_batch() { + assert_sliced_list_of_views_is_invariant::(); + } + + #[test] + fn sliced_large_list_of_view_column_hashes_equal_to_an_independently_built_batch() { + assert_sliced_list_of_views_is_invariant::(); + } + + fn fixed_size_list_of_views(rows: Vec<[&str; 2]>) -> RecordBatch { + let child = Arc::new(Field::new("item", DataType::Utf8View, true)); + let values: Vec<&str> = rows.iter().flatten().copied().collect(); + let arr = FixedSizeListArray::try_new( + Arc::clone(&child), + 2, + Arc::new(StringViewArray::from(values)), + None, + ) + .expect("valid fixture"); + batch( + vec![Field::new("l", DataType::FixedSizeList(child, 2), true)], + vec![Arc::new(arr)], + ) + } + + /// The third nesting flavour, which takes a structurally different path: it + /// narrows nothing, resting entirely on `FixedSizeListArray::slice` slicing + /// its child. The guard below pins that — if it ever stopped holding, the + /// compaction would `gc` the parent's whole child and silently re-absorb the + /// neighbouring rows' bytes. + #[test] + fn sliced_fixed_size_list_of_view_column_hashes_equal_to_an_independently_built_batch() { + let parent = fixed_size_list_of_views(vec![ + [LONG[0], LONG[1]], + [LONG[1], LONG[2]], + [LONG[2], LONG[0]], + ]); + let sliced = parent.slice(1, 1); + + assert_eq!( + sliced.column(0).to_data().child_data()[0].len(), + 2, + "fixture assumes `slice` narrows the child to the sliced rows" + ); + assert_eq!( + hash(&sliced), + hash(&fixed_size_list_of_views(vec![[LONG[1], LONG[2]]])), + "a sliced fixed-size-list-of-view column hashes as its own logical rows" + ); + } + + /// At `size == 0` — which `CanonicalType` accepts — `FixedSizeListArray` + /// carries no values to derive a length from and `try_new` reads it off the + /// null buffer, so rebuilding a null-free column through it would yield zero + /// rows. Two segments of different row counts would then digest identically: + /// a false dedup match on the one value that is supposed to be identity. + #[test] + fn zero_size_fixed_size_list_of_view_keeps_its_row_count() { + let empty_rows = |rows: usize| { + let child = Arc::new(Field::new("item", DataType::Utf8View, true)); + let arr = FixedSizeListArray::try_new_with_length( + Arc::clone(&child), + 0, + Arc::new(StringViewArray::from(Vec::<&str>::new())), + None, + rows, + ) + .expect("valid fixture"); + batch( + vec![Field::new("l", DataType::FixedSizeList(child, 0), true)], + vec![Arc::new(arr)], + ) + }; + assert_ne!( + hash(&empty_rows(3)), + hash(&empty_rows(2)), + "row count must survive compaction even with no values to count" + ); + } +} diff --git a/crates/penca-core/src/lib.rs b/crates/penca-core/src/lib.rs index 2d24e651..47e12f54 100644 --- a/crates/penca-core/src/lib.rs +++ b/crates/penca-core/src/lib.rs @@ -1,4 +1,5 @@ pub mod config; +pub mod digest; pub mod error; pub mod format; pub mod log_kind; diff --git a/crates/penca-core/src/plan.rs b/crates/penca-core/src/plan.rs index ae0c9213..2ae3c025 100644 --- a/crates/penca-core/src/plan.rs +++ b/crates/penca-core/src/plan.rs @@ -13,6 +13,7 @@ //! [`CommittedAtBounds`] rather than the proto `TimestampFilter`. use crate::Format; +use uuid::Uuid; /// Microsecond window on `commit_micros`: inclusive lower /// (`min_micros`), exclusive upper (`max_micros`); both optional. Native @@ -96,6 +97,12 @@ pub struct PersistSegment { /// `..PersistSegment::default()`: a cold `tx_log` segment reuses this type /// but holds commit metadata, not data rows, and has no ceiling. pub max_commit_seq_num: Option, + /// `xxh3_128` of the segment's typed in-memory Arrow batch, recorded at + /// write time and inherited verbatim by reference copies. Keys the segment + /// cache, so a fork and its parent share one decoded entry for a byte range + /// they both reference — which the row uuid cannot express, because a + /// reference copy mints a new uuid over unchanged bytes (CHA-545). + pub content_hash: Uuid, } /// The internal `row_uuid` index sidecar attached to a snapshot segment — the @@ -113,11 +120,18 @@ pub struct IndexSidecar { pub length: i64, /// Columnar file format of the sidecar. pub format: Format, - /// Globally-unique id of the sidecar — its segment-cache key, a distinct - /// deterministic-UUID namespace from `table_snapshot_segment_uuid`. + /// Globally-unique id of the sidecar row, in a distinct deterministic-UUID + /// namespace from `table_snapshot_segment_uuid`. Row identity only — the + /// segment-cache key is `content_hash`, because a reference copy mints a + /// fresh uuid over bytes it did not rewrite (CHA-545). pub segment_index_uuid: String, /// In-memory Arrow footprint, for the shared segment cache's byte budget. pub size_bytes: i64, + /// `xxh3_128` of the sidecar's typed in-memory Arrow batch. Same role as on + /// [`SnapshotSegment`]: sidecars are read through the same cache and + /// reference-copied by the same paths, so they duplicate for the same + /// reason and dedup by the same key (CHA-545). + pub content_hash: Uuid, } /// A snapshot segment in cold storage (read-optimized baseline). @@ -154,6 +168,10 @@ pub struct SnapshotSegment { /// tables, never a planner candidate). The internal identity sidecar stays /// in its dedicated `row_uuid_index_sidecar` slot. pub index_sidecars: Vec<(String, IndexSidecar)>, + /// `xxh3_128` of the segment's typed in-memory Arrow batch, recorded at + /// write time and inherited verbatim by reference copies (carry-forward, + /// fork copy). Keys the segment cache — see [`PersistSegment::content_hash`]. + pub content_hash: Uuid, } /// A user secondary index declared for the plan's snapshot (CHA-485) — the @@ -276,6 +294,15 @@ pub struct Plan { // can `..Default::default()`. `Format` itself stays default-free by // design (no `Unspecified` — see `format.rs`), so the segments pick // `Format::Parquet` as the placeholder; any test that cares sets it. +// +// `content_hash` defaults to `Uuid::nil()`, which would be a shared cache key +// if it ever reached the cache. It cannot: the only production caller of these +// `Default`s is the cold `tx_log` carrier path, whose segments hold commit +// metadata and are read by `read_tx_log_batches`, never through `SegmentCache` +// (CHA-545). Every cache-read segment is built field-by-field from a metadata +// row whose `content_hash` is `NOT NULL`. `IndexSidecar` deliberately has no +// `Default` for the same reason — it has no such non-cached carrier path, so a +// nil hash there would be reachable. impl Default for PersistSegment { fn default() -> Self { Self { @@ -289,6 +316,7 @@ impl Default for PersistSegment { offset: None, length: None, max_commit_seq_num: None, + content_hash: Uuid::nil(), } } } @@ -309,6 +337,7 @@ impl Default for SnapshotSegment { statistics: Vec::new(), row_uuid_index_sidecar: None, index_sidecars: Vec::new(), + content_hash: Uuid::nil(), } } } diff --git a/crates/penca-db/src/dialect/pg.rs b/crates/penca-db/src/dialect/pg.rs index 634f521a..176c9b86 100644 --- a/crates/penca-db/src/dialect/pg.rs +++ b/crates/penca-db/src/dialect/pg.rs @@ -487,6 +487,10 @@ impl PgDialect { // active merged file, `true` for rows of a previously-sealed merged // file. Sealed rows never participate in another compact wave, and the // false → true transition is one-way. + // `content_hash` — on this and the two other cold-artifact tables below + // — is the segment cache key (CHA-545). NOT NULL with no default: + // every writer computes it, and a catalog predating the column is + // recreated rather than migrated, so there is no legacy row to default. driver .execute_no_result(&format!( r#"CREATE TABLE IF NOT EXISTS {qi} ( @@ -504,6 +508,7 @@ impl PgDialect { length BIGINT, row_count BIGINT NOT NULL, format TEXT NOT NULL, + content_hash UUID NOT NULL, size_bytes BIGINT DEFAULT 0, metadata JSONB DEFAULT '{{}}'::jsonb, statistics BYTEA, @@ -626,6 +631,7 @@ impl PgDialect { length BIGINT NOT NULL, size_bytes BIGINT DEFAULT 0, format TEXT NOT NULL, + content_hash UUID NOT NULL, metadata JSONB DEFAULT '{{}}'::jsonb, statistics BYTEA, row_count BIGINT NOT NULL, @@ -752,6 +758,7 @@ impl PgDialect { "offset" BIGINT NOT NULL, length BIGINT NOT NULL, format TEXT NOT NULL, + content_hash UUID NOT NULL, size_bytes BIGINT DEFAULT 0, statistics BYTEA, written_at_micros BIGINT DEFAULT {epoch}, diff --git a/crates/penca-dl/src/cache.rs b/crates/penca-dl/src/cache.rs index a6eeb5bf..bfd2ff14 100644 --- a/crates/penca-dl/src/cache.rs +++ b/crates/penca-dl/src/cache.rs @@ -2,15 +2,13 @@ //! //! A repeat read of the same cold segment within the process lifetime is served //! as an `Arc::clone` of the already-decoded Arrow batches, skipping the S3 GET + -//! Parquet/Lance decode. It holds both snapshot segments (keyed by -//! `table_snapshot_segment_uuid`) and persist data segments (keyed by -//! `segment_uuid`) under one byte budget. Each per-uuid segment file is -//! immutable, so the key→value mapping is stable and needs no invalidation: a -//! snapshot uuid resolves to identical bytes for the life of the process, and -//! although a *resolved persist tier* is mutable under retention compaction, an -//! individual persist *file* (keyed by uuid) is not. There is no TTL — -//! immutability makes W-TinyLFU eviction the whole reclaim mechanism for both -//! tiers. +//! Parquet/Lance decode. It holds snapshot segments, persist data segments and +//! index sidecars under one byte budget, all keyed by `content_hash`. +//! The mapping is stable and needs no invalidation: the hash names one decode +//! by construction, and cold artifacts are immutable — although a *resolved +//! persist tier* is mutable under retention compaction, an individual persist +//! *file* is not. There is no TTL — immutability makes W-TinyLFU eviction the +//! whole reclaim mechanism for every tier. //! //! Eviction is W-TinyLFU (frequency-based, scan-resistant, aged) via `moka`, //! bounded by a byte budget: each entry is weighed by its segment's @@ -30,25 +28,59 @@ use std::sync::Arc; use arrow::record_batch::RecordBatch; use moka::sync::Cache; +use uuid::Uuid; -/// In-process W-TinyLFU cache of decoded cold segments, keyed by segment uuid -/// (`table_snapshot_segment_uuid` for snapshot, `segment_uuid` for persist) and -/// bounded by a byte budget. +/// In-process W-TinyLFU cache of decoded cold segments, keyed by `content_hash` +/// and bounded by a byte budget. /// -/// The key is the segment uuid alone: it is globally unique, immutable, and -/// resolves to identical bytes, so it fully identifies the decoded value. -/// `format` is intentionally absent from the key — it is functionally -/// determined by the uuid, and every value is a format-agnostic decoded -/// [`RecordBatch`]; `format` is consulted only on the miss path (by the -/// caller) to pick the reader. +/// `content_hash` is the digest of the typed in-memory Arrow batch, recorded +/// once at write time and inherited verbatim by every reference copy — snapshot +/// carry-forward (CHA-531) and a fork's cold materialization (CHA-539) both mint +/// a new row uuid over bytes nobody rewrote. Keying by uuid stored one decode +/// per *row*; keying by hash stores one per distinct *content*, which is what +/// lets a fork and its parent share a single entry for a shared slice (CHA-545). +/// +/// **The cached value must be the file-native decode**, not one caller's +/// shaping. A hash-keyed entry is shared across callers whose schemas can +/// differ — a fork that retypes a column still reads the parent's bytes — so a +/// caller-shaped value would hand the second caller the first caller's types. +/// Callers shape after the lookup via `penca_format::reader::shape_to_schema`; +/// `test_fork_and_parent_diverge_a_columns_type_over_one_shared_slice` is the +/// regression guard. +/// +/// The read-time schema still governs the *output* — it governs it at that +/// shaping step, per caller. Only the decode has to be segment-scoped, and only +/// because the decode is what gets shared. Fingerprinting the read schema into +/// the key instead is correct, but then the key moves whenever the schema does +/// while the data does not: one `ALTER TABLE ADD COLUMN` re-fingerprints a +/// footprint nobody rewrote, and a fork stops sharing with its parent at their +/// first divergent `ALTER` — the case this key exists for. See +/// `docs/design-decisions.md` — "Cold segments are cached by content hash". +/// +/// One flat key space, no per-artifact-class prefix. With every artifact keyed +/// by a hash of its own typed content, two entries collide only when their +/// decoded batches are identical, in which case sharing one entry is the +/// correct answer rather than a bug — a base segment and a sidecar that decode +/// to the same batch may safely share. +/// +/// The storage format is deliberately *not* in the key. `content_hash` digests +/// the typed batch *before* a `FormatWriter` encodes it, so after an +/// `OBJECT_STORAGE_FORMAT` flip one hash can name both a Parquet file and a +/// Lance file — and they are supposed to decode to the same batch, since a write +/// followed by a read returns what was written. A divergence there is a broken +/// round trip to fix in that format (or a type to drop from `CanonicalType`), not +/// a fact for this key to encode: adding the format would trade one real +/// cross-format share for a permanent second entry, and would hide the defect +/// rather than surface it. CHA-548 adds the round-trip identity coverage that +/// asserts the invariant this key assumes. /// /// Cheaply cloneable: `moka::sync::Cache` is internally an `Arc`, so callers /// typically hold a `SegmentCache` behind one outer `Arc` shared /// across the process. pub struct SegmentCache { - /// Value carries its own weight so the weigher can charge the + /// The value carries its own weight so the weigher can charge the /// caller-supplied `size_bytes` rather than the batch's runtime memory. - inner: Cache, u32)>, + inner: Cache, u32)>, budget_bytes: u64, } @@ -63,7 +95,7 @@ impl SegmentCache { .max_capacity(budget_bytes) // Weight in the same byte unit as `max_capacity` so the budget is a // real RAM bound; the stored `u32` is the entry's `size_bytes`. - .weigher(|_uuid: &String, (_batch, weight): &(Arc, u32)| *weight) + .weigher(|_key: &Uuid, (_batch, weight): &(Arc, u32)| *weight) .build(); Self { inner, @@ -103,21 +135,21 @@ impl SegmentCache { && weight_bytes <= u32::MAX as u64 } - /// Fetch a decoded segment by uuid, bumping its frequency estimate. A hit - /// is an `Arc::clone` — no buffer copy. - pub fn get(&self, uuid: &str) -> Option> { - self.inner.get(uuid).map(|(batch, _weight)| batch) + /// Fetch a decoded segment, bumping its frequency estimate. A hit is an + /// `Arc::clone` — no buffer copy. + pub fn get(&self, content_hash: &Uuid) -> Option> { + self.inner.get(content_hash).map(|(batch, _weight)| batch) } - /// Insert a decoded segment under its uuid, charged `weight_bytes` against - /// the budget. No-op when the segment is not [`admits`](Self::admits)-ible. - /// moka enforces `max_capacity` via W-TinyLFU; there is no manual eviction - /// loop here. - pub fn insert(&self, uuid: String, batch: Arc, weight_bytes: u64) { + /// Insert a decoded segment charged `weight_bytes` against the budget. No-op + /// when the segment is not [`admits`](Self::admits)-ible. moka enforces + /// `max_capacity` via W-TinyLFU; there is no manual eviction loop here. + pub fn insert(&self, content_hash: Uuid, batch: Arc, weight_bytes: u64) { if !self.admits(weight_bytes) { return; } - self.inner.insert(uuid, (batch, weight_bytes as u32)); + self.inner + .insert(content_hash, (batch, weight_bytes as u32)); } /// Force pending eviction/maintenance to run synchronously. moka does @@ -143,6 +175,7 @@ mod tests { use arrow::array::Int32Array; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; + use uuid::Uuid; use super::SegmentCache; @@ -153,6 +186,11 @@ mod tests { Arc::new(RecordBatch::try_new(schema, vec![Arc::new(col)]).unwrap()) } + /// A distinct content hash per `n`. + fn hash(n: u128) -> Uuid { + Uuid::from_u128(n) + } + #[test] fn admits_predicate() { let cache = SegmentCache::new(100); @@ -171,10 +209,10 @@ mod tests { #[test] fn zero_weight_segment_is_not_cached() { let cache = SegmentCache::new(1 << 20); - cache.insert("zero".into(), batch(4), 0); + cache.insert(hash(1), batch(4), 0); cache.run_pending(); assert!( - cache.get("zero").is_none(), + cache.get(&hash(1)).is_none(), "weight-0 segment must not be pinned in the cache" ); } @@ -194,22 +232,25 @@ mod tests { ); assert!(cache.admits(u32::MAX as u64), "exactly u32::MAX is fine"); - cache.insert("huge".into(), batch(8), over_u32); + cache.insert(hash(2), batch(8), over_u32); cache.run_pending(); - assert!(cache.get("huge").is_none(), "over-u32 weight never stored"); + assert!( + cache.get(&hash(2)).is_none(), + "over-u32 weight never stored" + ); } #[test] fn over_budget_insert_is_noop() { let cache = SegmentCache::new(100); - cache.insert("too-big".into(), batch(8), 200); + cache.insert(hash(3), batch(8), 200); cache.run_pending(); - assert!(cache.get("too-big").is_none(), "over-budget never stored"); + assert!(cache.get(&hash(3)).is_none(), "over-budget never stored"); let disabled = SegmentCache::disabled(); - disabled.insert("x".into(), batch(8), 1); + disabled.insert(hash(4), batch(8), 1); disabled.run_pending(); - assert!(disabled.get("x").is_none(), "disabled never stores"); + assert!(disabled.get(&hash(4)).is_none(), "disabled never stores"); } #[test] @@ -220,7 +261,7 @@ mod tests { // moka's W-TinyLFU choice, not Penca's contract). let cache = SegmentCache::new(100); for i in 0..5 { - cache.insert(format!("seg-{i}"), batch(10), 40); + cache.insert(hash(i), batch(10), 40); } cache.run_pending(); assert!( @@ -234,9 +275,9 @@ mod tests { fn hit_returns_arc_clone_same_buffers() { let cache = SegmentCache::new(1_000); let original = batch(16); - cache.insert("seg".into(), original.clone(), 40); + cache.insert(hash(5), original.clone(), 40); cache.run_pending(); - let hit = cache.get("seg").expect("cached"); + let hit = cache.get(&hash(5)).expect("cached"); // Same backing column buffer — a hit is a refcount bump, no copy. assert_eq!( original.column(0).to_data().buffers()[0].as_ptr(), diff --git a/crates/penca-dl/src/driver.rs b/crates/penca-dl/src/driver.rs index e72b5587..70899503 100644 --- a/crates/penca-dl/src/driver.rs +++ b/crates/penca-dl/src/driver.rs @@ -19,7 +19,7 @@ use async_trait::async_trait; use datafusion::execution::SendableRecordBatchStream; use datafusion::execution::context::{SessionContext, SessionState}; use penca_core::{ColdStoragePlan, IndexSidecar, PersistSegment, SnapshotSegment}; -use penca_format::reader::{FormatError, FormatReader}; +use penca_format::reader::{FormatError, FormatReader, shape_to_schema}; use penca_storage_cold::{COMMIT_SEQ_NUM_COLUMN, ColdStorageError}; use tracing::Instrument as _; use uuid::Uuid; @@ -203,43 +203,39 @@ impl DatafusionDlDriver { } } -/// Cacheable miss: decode the WHOLE segment (all columns, no filter -/// pushdown) so the cached entry is reusable across any projection, insert -/// it under `weight`, and return the full superset. The caller has already -/// decided this segment is admissible. -async fn read_and_cache_full( +/// Cacheable miss: decode the WHOLE segment file (all columns, no filter +/// pushdown, no caller shaping) so the cached entry is reusable across any +/// projection AND any caller schema (CHA-545), offer it to the cache under +/// `content_hash`, and return the native batch. The caller shapes the result +/// itself once it holds it. +/// +/// Takes `uri` + `(offset, length)` rather than any segment type, mirroring +/// [`FormatReader::read_segment_native`] — persist segments, snapshot segments +/// and index sidecars all reach it, and none of them is the reader's or the +/// cache's concern. +/// +/// [`SegmentCache::insert`] self-gates on +/// [`admits`](SegmentCache::admits), so an artifact too large for the budget is +/// decoded-but-not-cached rather than evicting everything and then itself. A +/// caller with a narrower read available should check `admits` first and take +/// that path instead. +async fn decode_and_cache_native( reader: &R, cache: &SegmentCache, - segment: &SnapshotSegment, - full_schema: &SchemaRef, + uri: &str, + offset: Option, + length: Option, + content_hash: Uuid, weight: u64, -) -> Result { - let full_cols: Vec<&str> = full_schema - .fields() - .iter() - .map(|f| f.name().as_str()) - .collect(); +) -> Result, DlError> { let batch = reader - .read_segment( - &segment.uri, - Some(segment.offset), - Some(segment.length), - full_schema, - Some(&full_cols), - ) + .read_segment_native(uri, offset, length) .await .map_err(ColdStorageError::from)?; let batch = Arc::new(batch); - cache.insert( - segment.table_snapshot_segment_uuid.clone(), - Arc::clone(&batch), - weight, - ); - tracing::debug!( - rows = batch.num_rows(), - "snapshot segment cached full decode" - ); - Ok((*batch).clone()) + cache.insert(content_hash, Arc::clone(&batch), weight); + tracing::debug!(rows = batch.num_rows(), "segment decoded and cached"); + Ok(batch) } /// Non-cacheable miss: a projected read of just `out_schema`, not cached. @@ -272,36 +268,59 @@ async fn read_projected_uncached( Ok(batch) } -/// Cache-aware read of a single snapshot segment. Returns the full decoded -/// superset on hit / miss-cached, or the projected `out_schema` batch on the -/// non-cacheable (oversized) path. No predicate is pushed (ADR 0023); the -/// caller projects / null-fills downstream. +/// Shape a natively-decoded segment to `schema`, in [`DlError`] terms. +/// +/// Always projection-less: the cache stores the file-native decode (CHA-545) and +/// every consumer shapes to its own *full* schema after the lookup, leaving +/// column pruning to DataFusion (ADR 0023). +fn shape_native(batch: &RecordBatch, schema: &SchemaRef) -> Result { + Ok(shape_to_schema(batch, schema, None).map_err(ColdStorageError::from)?) +} + +/// One cache-aware segment read, before any shaping — the two outcomes differ +/// in what shaping they still owe. +enum CachedSegment { + /// The file-native decode, cached (or freshly cached) as-is. Still owes + /// [`shape_native`] to the caller's `full_schema`; deferring that to the + /// caller is what lets the index-seek path `take` its O(matches) rows + /// first and null-fill only those. + Native(Arc), + /// The non-cacheable (oversized) path, already read projected to + /// `out_schema`. Owes nothing. + Projected(RecordBatch), +} + +/// Cache-aware read of a single snapshot segment, unshaped. No predicate is +/// pushed (ADR 0023); the caller projects / null-fills downstream. +/// +/// The cached value is the file-native decode, so shaping happens after the +/// lookup and two branches sharing one entry each get their own types rather +/// than whichever branch decoded first (CHA-545). #[tracing::instrument( level = "debug", skip_all, fields( segment_uuid = %segment.table_snapshot_segment_uuid, + content_hash = %segment.content_hash, format = %segment.format, cache = tracing::field::Empty, ), )] -pub(crate) async fn read_cached_snapshot_segment( +async fn read_cached_snapshot_segment_unshaped( readers: &HashMap, cache: &SegmentCache, segment: &SnapshotSegment, - full_schema: &SchemaRef, out_schema: &SchemaRef, -) -> Result { +) -> Result { let span = tracing::Span::current(); - let uuid = segment.table_snapshot_segment_uuid.as_str(); + let code = segment.format.as_wire_code(); - if let Some(full) = cache.get(uuid) { + if let Some(full) = cache.get(&segment.content_hash) { span.record("cache", "hit"); tracing::debug!(rows = full.num_rows(), "snapshot segment cache hit"); - return Ok((*full).clone()); + return Ok(CachedSegment::Native(full)); } - let code = segment.format.as_wire_code(); let reader = readers .get(&code) .ok_or(ColdStorageError::UnknownFormat(code))?; @@ -311,45 +330,40 @@ pub(crate) async fn read_cached_snapshot_segment( if cache.admits(weight) { span.record("cache", "miss-cached"); - read_and_cache_full(reader, cache, segment, full_schema, weight).await + Ok(CachedSegment::Native( + decode_and_cache_native( + reader, + cache, + &segment.uri, + Some(segment.offset), + Some(segment.length), + segment.content_hash, + weight, + ) + .await?, + )) } else { span.record("cache", "miss-uncached"); - read_projected_uncached(reader, segment, out_schema).await + Ok(CachedSegment::Projected( + read_projected_uncached(reader, segment, out_schema).await?, + )) } } -/// Cacheable persist miss: decode the WHOLE persist segment (all columns) so the -/// cached entry serves any projection, insert it under `weight`, and return the -/// full superset. -async fn read_and_cache_full_persist( - reader: &R, +/// Cache-aware read of a single snapshot segment. Returns the full decoded +/// superset on hit / miss-cached, or the projected `out_schema` batch on the +/// non-cacheable (oversized) path. +pub(crate) async fn read_cached_snapshot_segment( + readers: &HashMap, cache: &SegmentCache, - segment: &PersistSegment, + segment: &SnapshotSegment, full_schema: &SchemaRef, - weight: u64, + out_schema: &SchemaRef, ) -> Result { - let full_cols: Vec<&str> = full_schema - .fields() - .iter() - .map(|f| f.name().as_str()) - .collect(); - let batch = reader - .read_segment( - &segment.uri, - segment.offset, - segment.length, - full_schema, - Some(&full_cols), - ) - .await - .map_err(ColdStorageError::from)?; - let batch = Arc::new(batch); - cache.insert(segment.segment_uuid.clone(), Arc::clone(&batch), weight); - tracing::debug!( - rows = batch.num_rows(), - "persist segment decoded and cached" - ); - Ok((*batch).clone()) + match read_cached_snapshot_segment_unshaped(readers, cache, segment, out_schema).await? { + CachedSegment::Native(native) => shape_native(&native, full_schema), + CachedSegment::Projected(batch) => Ok(batch), + } } /// Non-cacheable persist miss: a projected read of just `out_schema`, not cached. @@ -408,20 +422,20 @@ async fn read_projected_uncached_persist( } /// Cache-aware read of a single persist segment. A persist segment file is -/// immutable once written and keyed by its globally-unique `segment_uuid`, so it -/// shares the process-lifetime [`SegmentCache`] with snapshot segments under one -/// byte budget, with NO TTL — W-TinyLFU eviction plus the `admits` budget gate -/// is the whole mechanism. (The *resolved* persist tier is mutable under -/// retention compaction, which is why the tier is re-resolved live on every -/// read; the per-uuid *file bytes* this caches are not.) Returns the full -/// decoded superset on hit / miss-cached, or the projected `out_schema` batch on -/// the non-cacheable (oversized) path; the caller projects / null-fills -/// downstream. +/// immutable once written, so it shares the process-lifetime [`SegmentCache`] +/// with snapshot segments under one byte budget and one `content_hash` key +/// space, with NO TTL — W-TinyLFU eviction plus the `admits` budget gate is the +/// whole mechanism. (The *resolved* persist tier is mutable under retention +/// compaction, which is why the tier is re-resolved live on every read; the +/// *file bytes* this caches are not.) Returns the full decoded superset on hit / +/// miss-cached, or the projected `out_schema` batch on the non-cacheable +/// (oversized) path; the caller projects / null-fills downstream. #[tracing::instrument( level = "debug", skip_all, fields( segment_uuid = %segment.segment_uuid, + content_hash = %segment.content_hash, format = %segment.format, cache = tracing::field::Empty, ), @@ -434,15 +448,14 @@ pub(crate) async fn read_cached_persist_segment( out_schema: &SchemaRef, ) -> Result { let span = tracing::Span::current(); - let uuid = segment.segment_uuid.as_str(); + let code = segment.format.as_wire_code(); - if let Some(full) = cache.get(uuid) { + if let Some(full) = cache.get(&segment.content_hash) { span.record("cache", "hit"); tracing::debug!(rows = full.num_rows(), "persist segment cache hit"); - return Ok((*full).clone()); + return shape_native(&full, full_schema); } - let code = segment.format.as_wire_code(); let reader = readers .get(&code) .ok_or(ColdStorageError::UnknownFormat(code))?; @@ -453,7 +466,17 @@ pub(crate) async fn read_cached_persist_segment( if cache.admits(weight) { span.record("cache", "miss-cached"); - read_and_cache_full_persist(reader, cache, segment, full_schema, weight).await + let native = decode_and_cache_native( + reader, + cache, + &segment.uri, + segment.offset, + segment.length, + segment.content_hash, + weight, + ) + .await?; + shape_native(&native, full_schema) } else { span.record("cache", "miss-uncached"); read_projected_uncached_persist(reader, segment, out_schema, full_schema).await @@ -461,13 +484,22 @@ pub(crate) async fn read_cached_persist_segment( } /// Load a sorted `(key, row_offset)` index sidecar through the shared snapshot -/// cache, keyed by its own `segment_index_uuid` — a distinct deterministic-UUID -/// namespace from the base segment uuid, so the two never collide in one cache. +/// cache, under the sidecar's own `content_hash` — a digest of the sidecar +/// batch, not of the base segment it indexes. Sidecars and base segments share +/// one flat key space; see [`SegmentCache`] for why a same-content collision +/// between the two classes would be correct rather than a bug. +/// +/// Cached natively and shaped to `key_types` after the lookup, for the same +/// reason as a base segment: the file was written with the *writing* branch's +/// key column types, `key_types` comes from the *reading* branch's schema, and +/// a fork that diverges an indexed column's type makes the two disagree +/// (CHA-545). #[tracing::instrument( level = "debug", skip_all, fields( segment_index_uuid = %sidecar.segment_index_uuid, + content_hash = %sidecar.content_hash, cache = tracing::field::Empty, ), )] @@ -478,37 +510,42 @@ async fn read_cached_index_sidecar( key_types: &[arrow::datatypes::DataType], ) -> Result { let span = tracing::Span::current(); - if let Some(batch) = cache.get(&sidecar.segment_index_uuid) { - span.record("cache", "hit"); - return Ok((*batch).clone()); - } - span.record("cache", "miss"); let code = sidecar.format.as_wire_code(); - let reader = readers - .get(&code) - .ok_or(ColdStorageError::UnknownFormat(code))?; // The sidecar's key schema is the indexed columns' native types; the // identity/name sidecars are the all-Utf8 special case. let schema = penca_format::index::segment_index_schema(key_types); - let batch = reader - .read_segment( - &sidecar.object_uri, - Some(sidecar.offset), - Some(sidecar.length), - &schema, - None, - ) - .await - .map_err(ColdStorageError::from)?; - let batch = Arc::new(batch); - // `insert` self-gates on `cache.admits(weight)`, so an oversize sidecar is - // decoded-but-not-cached rather than evicting the whole budget. - cache.insert( - sidecar.segment_index_uuid.clone(), - Arc::clone(&batch), - sidecar.size_bytes.max(0) as u64, + if let Some(batch) = cache.get(&sidecar.content_hash) { + span.record("cache", "hit"); + return shape_native(&batch, &schema); + } + let weight = sidecar.size_bytes.max(0) as u64; + // Telemetry only — unlike a base segment there is no narrower read to fall + // back to, so a non-admissible sidecar is still decoded whole (`insert` + // self-gates). Reporting the base paths' three-valued vocabulary is what + // makes a sidecar that can *never* cache — re-decoded from S3 on every read + // — distinguishable from an ordinary first touch. + span.record( + "cache", + if cache.admits(weight) { + "miss-cached" + } else { + "miss-uncached" + }, ); - Ok((*batch).clone()) + let reader = readers + .get(&code) + .ok_or(ColdStorageError::UnknownFormat(code))?; + let batch = decode_and_cache_native( + reader, + cache, + &sidecar.object_uri, + Some(sidecar.offset), + Some(sidecar.length), + sidecar.content_hash, + weight, + ) + .await?; + shape_native(&batch, &schema) } /// Index-driven selective read: binary-search the segment's index sidecar for @@ -568,6 +605,11 @@ async fn seek_entry_offsets( /// decode entirely on zero matches. A candidate segment that passes coarse /// pruning but doesn't contain the probed key must not pay a full base /// decode just to `take` zero rows (the common cross-segment-lookup miss). +/// +/// `take` runs BEFORE the shaping tail so the seek path stays O(matches): a +/// segment written before an `ALTER TABLE ADD COLUMN` null-fills the added +/// column, and null-filling the whole cached batch first would allocate a +/// segment-length array on every probe. async fn take_matched_rows( readers: &HashMap, cache: &SegmentCache, @@ -579,11 +621,14 @@ async fn take_matched_rows( if offsets.is_empty() { return Ok(RecordBatch::new_empty(full_schema.clone())); } - let base = - read_cached_snapshot_segment(readers, cache, segment, full_schema, out_schema).await?; let indices = arrow::array::Int64Array::from(offsets); - let taken = arrow::compute::take_record_batch(&base, &indices)?; - Ok(taken) + match read_cached_snapshot_segment_unshaped(readers, cache, segment, out_schema).await? { + CachedSegment::Native(native) => { + let taken = arrow::compute::take_record_batch(&native, &indices)?; + shape_native(&taken, full_schema) + } + CachedSegment::Projected(batch) => Ok(arrow::compute::take_record_batch(&batch, &indices)?), + } } /// Seek SEVERAL resolved entries against one segment and decode the @@ -954,6 +999,18 @@ mod tests { Ok(self.batch.project(&indices)?) } + + /// Delegates so the read counter stays in one place: a native decode is + /// a storage hit like any other, and the cache tests count both. + async fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> Result { + self.read_segment(uri, offset, length, &self.batch.schema(), None) + .await + } } fn test_schema() -> SchemaRef { @@ -974,11 +1031,15 @@ mod tests { .unwrap() } + /// `content_hash` is derived from the name so distinct fixtures stay + /// distinct under content-hash cache keying; tests asserting dedup + /// overwrite it with a deliberately shared value. fn segment(uuid: &str, size_bytes: i64) -> SnapshotSegment { SnapshotSegment { table_snapshot_segment_uuid: uuid.to_string(), format: Format::Parquet, size_bytes, + content_hash: penca_core::naming::deterministic_uuid_from(&[uuid]), ..Default::default() } } @@ -1036,6 +1097,17 @@ mod tests { .unwrap_or_else(|| panic!("unexpected read of {uri}")) .clone()) } + + async fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> Result { + // `read_segment` ignores its schema argument here — it routes on uri. + self.read_segment(uri, offset, length, &Arc::new(Schema::empty()), None) + .await + } } fn routing_driver( @@ -1076,6 +1148,7 @@ mod tests { by_uri.insert(side_uri.clone(), sidecar_batch); SnapshotSegment { table_snapshot_segment_uuid: format!("seg-{name}"), + content_hash: penca_core::naming::deterministic_uuid_from(&[&format!("seg-{name}")]), uri: base_uri, format: Format::Parquet, length: keys.len() as i64, @@ -1088,6 +1161,9 @@ mod tests { format: Format::Parquet, segment_index_uuid: format!("idx-{name}"), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&[&format!( + "idx-{name}" + )]), }), ..Default::default() } @@ -1154,6 +1230,55 @@ mod tests { assert_eq!(res, expected); } + /// The seek path `take`s the matched offsets before it null-fills, so this + /// pins that the added column's nulls land on the taken rows and the other + /// columns stay aligned with them. + #[tokio::test] + async fn seek_snapshot_point_null_fills_a_column_absent_from_the_file() { + let cache = Arc::new(SegmentCache::new(1 << 20)); + let mut by_uri = HashMap::new(); + let seg = indexed_segment( + &mut by_uri, + &test_schema(), // what the segment file actually holds + "added", + &["r0", "r1", "r2"], + &[0, 1, 2], + ); + let dl = routing_driver(cache, by_uri); + + // `added` arrived via a later ALTER TABLE ADD COLUMN: it is in the + // table schema but not in this segment's file. + let full_schema: SchemaRef = Arc::new(Schema::new(vec![ + Field::new("row_uuid", DataType::Utf8, false), + Field::new("v", DataType::Int32, false), + Field::new("added", DataType::Int32, true), + ])); + + let res = dl + .seek_snapshot_point( + std::slice::from_ref(&seg), + &[vec!["r1".to_string()]], + None, + &[], + &full_schema, + &full_schema, + ) + .await + .unwrap() + .expect("indexed segment => Some"); + + let expected = RecordBatch::try_new( + full_schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["r1"])), + Arc::new(Int32Array::from(vec![1])), + Arc::new(Int32Array::from(vec![None::])), + ], + ) + .unwrap(); + assert_eq!(res, expected); + } + #[tokio::test] async fn seek_snapshot_point_empty_segments_yields_empty_batch() { let cache = Arc::new(SegmentCache::new(1 << 20)); @@ -1239,6 +1364,7 @@ mod tests { by_uri.insert(side_uri.clone(), sidecar_batch); SnapshotSegment { table_snapshot_segment_uuid: format!("seg-{name}"), + content_hash: penca_core::naming::deterministic_uuid_from(&[&format!("seg-{name}")]), uri: base_uri, format: Format::Parquet, length: key0s.len() as i64, @@ -1251,6 +1377,9 @@ mod tests { format: Format::Parquet, segment_index_uuid: format!("idx-{name}"), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&[&format!( + "idx-{name}" + )]), }), ..Default::default() } @@ -1521,6 +1650,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "idx-identity-never-read".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["idx-identity-never-read"]), }); let dl = routing_driver(cache, by_uri); @@ -1550,6 +1680,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "idx-identity".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["idx-identity"]), }); let name_index = Uuid::new_v4(); let res = dl @@ -1591,6 +1722,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "idx-other".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["idx-other"]), }, )]; let requested = Uuid::new_v4(); // != the keyed index present @@ -1635,6 +1767,9 @@ mod tests { format: Format::Parquet, segment_index_uuid: "idx-keyed-never-read".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&[ + "idx-keyed-never-read", + ]), }, )]; let dl = routing_driver(cache, by_uri); @@ -1756,6 +1891,135 @@ mod tests { ); } + /// Two metadata rows over one byte range — exactly what snapshot + /// carry-forward (CHA-531) and a fork's reference copy (CHA-539) produce: a + /// NEW row uuid over an unchanged `(uri, offset, length)`. Keyed by uuid the + /// cache stores both decodes of identical bytes; keyed by `content_hash` it + /// stores one (CHA-545). + /// + /// One test per artifact class because the scope rule is precisely that all + /// three behave identically — a fix that dedups segments but leaves sidecars + /// uuid-keyed would pass a single-class test. + #[tokio::test] + async fn snapshot_segments_sharing_content_hash_decode_once() { + let schema = test_schema(); + let cache = Arc::new(SegmentCache::new(1 << 20)); + let (dl, reads) = driver_with(cache, test_batch(&schema)); + + let shared = Uuid::from_u128(0xc0ffee); + let mut parent = segment("parent-seg", 128); + parent.uri = "s3://t/shared.parquet".into(); + parent.content_hash = shared; + let mut fork = segment("fork-seg", 128); + fork.uri = parent.uri.clone(); + fork.content_hash = shared; + + let first = read_seg(&dl, &parent, &schema, &schema).await.unwrap(); + let second = read_seg(&dl, &fork, &schema, &schema).await.unwrap(); + + assert_eq!( + reads.load(Ordering::SeqCst), + 1, + "distinct segment uuids over one content hash must share a cache entry" + ); + assert_eq!(first, second, "both rows resolve to the same decoded batch"); + } + + #[tokio::test] + async fn persist_segments_sharing_content_hash_decode_once() { + let schema = test_schema(); + let cache = Arc::new(SegmentCache::new(1 << 20)); + let (dl, reads) = driver_with(cache, test_batch(&schema)); + + let shared = Uuid::from_u128(0xc0ffee); + let seg = |uuid: &str| PersistSegment { + segment_uuid: uuid.to_string(), + uri: "s3://t/shared.parquet".into(), + format: Format::Parquet, + size_bytes: 256, + content_hash: shared, + ..Default::default() + }; + + let first = read_cached_persist_segment( + dl.readers.as_ref(), + &dl.cache, + &seg("parent-p"), + &schema, + &schema, + ) + .await + .unwrap(); + let second = read_cached_persist_segment( + dl.readers.as_ref(), + &dl.cache, + &seg("fork-p"), + &schema, + &schema, + ) + .await + .unwrap(); + + assert_eq!( + reads.load(Ordering::SeqCst), + 1, + "distinct persist segment uuids over one content hash must share a cache entry" + ); + assert_eq!(first, second, "both rows resolve to the same decoded batch"); + } + + /// The fixture is sidecar-shaped because the sidecar read shapes its cached + /// native batch to `segment_index_schema(key_types)` after the lookup. + #[tokio::test] + async fn index_sidecars_sharing_content_hash_decode_once() { + let schema = penca_format::index::segment_index_schema(&[DataType::Utf8]); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["k0", "k1", "k2"])), + Arc::new(arrow::array::Int64Array::from(vec![0i64, 1, 2])), + ], + ) + .unwrap(); + let cache = Arc::new(SegmentCache::new(1 << 20)); + let (dl, reads) = driver_with(cache, batch); + + let shared = Uuid::from_u128(0xbeef); + let sidecar = |uuid: &str| IndexSidecar { + object_uri: "s3://t/shared.idx".into(), + offset: 0, + length: 3, + format: Format::Parquet, + segment_index_uuid: uuid.to_string(), + size_bytes: 256, + content_hash: shared, + }; + + let first = read_cached_index_sidecar( + dl.readers.as_ref(), + &dl.cache, + &sidecar("parent-idx"), + &[DataType::Utf8], + ) + .await + .unwrap(); + let second = read_cached_index_sidecar( + dl.readers.as_ref(), + &dl.cache, + &sidecar("fork-idx"), + &[DataType::Utf8], + ) + .await + .unwrap(); + + assert_eq!( + reads.load(Ordering::SeqCst), + 1, + "distinct sidecar uuids over one content hash must share a cache entry" + ); + assert_eq!(first, second, "both rows resolve to the same decoded batch"); + } + #[tokio::test] async fn oversized_segment_not_cached_uses_uncached_read() { let schema = test_schema(); @@ -1771,7 +2035,10 @@ mod tests { 2, "oversized segment is never cached — both accesses re-read storage" ); - assert!(cache.get("big").is_none(), "oversized segment not stored"); + assert!( + cache.get(&seg.content_hash).is_none(), + "oversized segment not stored" + ); } #[tokio::test] @@ -1791,7 +2058,11 @@ mod tests { // moka evicted one of {a,b} to honor the budget (we don't assert which // — that is moka's W-TinyLFU choice). Re-reading the evicted key must // hit storage again. - let evicted = if cache.get("a").is_none() { "a" } else { "b" }; + let evicted = if cache.get(&segment("a", 150).content_hash).is_none() { + "a" + } else { + "b" + }; read_seg(&dl, &segment(evicted, 150), &schema, &schema) .await .unwrap(); @@ -2211,6 +2482,38 @@ mod tests { ); } + /// Shaping moved after the cache lookup (CHA-545), so a caller asking for a + /// non-nullable column the file lacks must fail the same way whether or not + /// an earlier caller already warmed the entry. The first read caches the + /// native decode and *then* fails to shape it, so the second read takes the + /// hit path — the one that would silently return an unshaped batch if the + /// tail were skipped there. + #[tokio::test] + async fn non_nullable_missing_column_errors_on_a_cache_hit_too() { + let demanding: SchemaRef = Arc::new(Schema::new(vec![ + Field::new("row_uuid", DataType::Utf8, false), + Field::new("absent", DataType::Int32, false), + ])); + let cache = Arc::new(SegmentCache::new(1 << 20)); + let (dl, reads) = driver_with(cache, test_batch(&test_schema())); + let seg = segment("seg-missing", 128); + + for pass in ["miss", "hit"] { + let err = read_seg(&dl, &seg, &demanding, &demanding) + .await + .expect_err("non-nullable column absent from the segment must error"); + assert!( + err.to_string().contains("absent"), + "{pass} pass must name the missing column, got: {err}" + ); + } + assert_eq!( + reads.load(Ordering::SeqCst), + 1, + "the second pass errored off the cached entry, not a re-read" + ); + } + #[tokio::test] async fn scan_snapshot_schema_tolerance() { // Segment decoded against the OLDER narrow schema {row_uuid, name}; diff --git a/crates/penca-dl/src/provider.rs b/crates/penca-dl/src/provider.rs index ca480f59..3310c659 100644 --- a/crates/penca-dl/src/provider.rs +++ b/crates/penca-dl/src/provider.rs @@ -815,6 +815,7 @@ mod tests { offset: None, length: None, max_commit_seq_num: None, + content_hash: penca_core::naming::deterministic_uuid_from(&[uuid]), } } @@ -1021,6 +1022,15 @@ mod tests { ) -> Result { Ok(self.batch.clone()) } + + async fn read_segment_native( + &self, + _uri: &str, + _offset: Option, + _length: Option, + ) -> Result { + Ok(self.batch.clone()) + } } fn snapshot_seg(uuid: &str, size_bytes: i64) -> SnapshotSegment { @@ -1028,6 +1038,7 @@ mod tests { table_snapshot_segment_uuid: uuid.to_string(), format: Format::Parquet, size_bytes, + content_hash: penca_core::naming::deterministic_uuid_from(&[uuid]), ..Default::default() } } @@ -1282,6 +1293,17 @@ mod tests { self.reads.lock().unwrap().push(uri.to_string()); Ok(self.batches.get(uri).cloned().expect("uri registered")) } + + /// Delegates so one read is recorded per storage hit, native or not. + async fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> Result { + self.read_segment(uri, offset, length, &Arc::new(Schema::empty()), None) + .await + } } #[tokio::test] @@ -1323,6 +1345,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sidecar1".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["sidecar1"]), }), ..Default::default() }; @@ -1396,6 +1419,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sidecar1".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["sidecar1"]), }), ..Default::default() }; @@ -1483,6 +1507,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sc".to_string(), size_bytes: 64, + content_hash: penca_core::naming::deterministic_uuid_from(&["sc"]), }); let provider = SnapshotTableProvider::new( vec![seg], @@ -1551,6 +1576,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sidecar-identity".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["sidecar-identity"]), }), index_sidecars: vec![( user_index_uuid.to_string(), @@ -1561,6 +1587,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sidecar-user".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["sidecar-user"]), }, )], ..Default::default() @@ -1642,6 +1669,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sidecar-identity".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["sidecar-identity"]), }), // No keyed sidecars: the user entry below cannot resolve. ..Default::default() @@ -1693,6 +1721,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "id-sc".to_string(), size_bytes: 1, + content_hash: penca_core::naming::deterministic_uuid_from(&["id-sc"]), }), index_sidecars: vec![( user.to_string(), @@ -1703,6 +1732,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "user-sc".to_string(), size_bytes: 1, + content_hash: penca_core::naming::deterministic_uuid_from(&["user-sc"]), }, )], ..Default::default() @@ -1777,6 +1807,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sidecar-value".to_string(), size_bytes: 256, + content_hash: penca_core::naming::deterministic_uuid_from(&["sidecar-value"]), }, )], ..Default::default() @@ -1888,6 +1919,7 @@ mod tests { format: Format::Parquet, segment_index_uuid: "sidecar-x".to_string(), size_bytes: 128, + content_hash: penca_core::naming::deterministic_uuid_from(&["sidecar-x"]), }, )], ..Default::default() @@ -1965,6 +1997,7 @@ mod by_plan_order_tests { uri: uri.to_string(), format: Format::Parquet, size_bytes: 64, + content_hash: penca_core::naming::deterministic_uuid_from(&[uuid]), ..Default::default() } } @@ -1989,6 +2022,17 @@ mod by_plan_order_tests { } Ok(self.batches[uri].clone()) } + + /// Delegates so the deliberate per-uri sleep applies to native reads too. + async fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> Result { + self.read_segment(uri, offset, length, &Arc::new(Schema::empty()), None) + .await + } } fn session_for( diff --git a/crates/penca-format/src/reader/lance.rs b/crates/penca-format/src/reader/lance.rs index 1d734887..27c0895f 100644 --- a/crates/penca-format/src/reader/lance.rs +++ b/crates/penca-format/src/reader/lance.rs @@ -16,7 +16,8 @@ use lance_io::ReadBatchParams; use lance_io::scheduler::{ScanScheduler, SchedulerConfig}; use super::{ - FormatError, FormatReader, empty_batch, null_fill_to_schema, present_columns, project_schema, + FormatError, FormatReader, empty_batch, present_columns, project_schema, requested_columns, + shape_to_schema, }; use crate::uri::uri_to_object_path; @@ -76,10 +77,11 @@ impl LanceFormatReader { /// `params` selects the row range (`RangeFull` for whole-file, /// `Range(start..end)` for a slice). When `projection` is `Some`, /// dispatches through `read_stream_projected`; otherwise through - /// `read_stream`. + /// `read_stream`. `empty_schema` is used only when the slice decodes to no + /// batches at all; otherwise the decoded batches' own schema is kept. async fn read_with_params( reader: &FileReader, - output_schema: &SchemaRef, + empty_schema: &SchemaRef, projection: Option, params: ReadBatchParams, ) -> Result { @@ -113,9 +115,51 @@ impl LanceFormatReader { } if batches.is_empty() { - return Ok(empty_batch(output_schema)); + return Ok(empty_batch(empty_schema)); } - Ok(concat_batches(output_schema, &batches)?) + Ok(concat_batches(&batches[0].schema(), &batches)?) + } + + /// The read itself, in the file's own schema. `projection`, when `Some`, + /// is narrowed to the columns physically present and pushed into the Lance + /// reader; `None` reads every column the file has. + /// + /// Shaping to a caller's schema is the caller's job — `read_segment` runs + /// it immediately, `read_segment_native`'s callers run it after their + /// cache lookup. + async fn read_in_file_schema( + &self, + uri: &str, + offset: Option, + length: Option, + projection: Option<&[&str]>, + ) -> Result { + let reader = self.open_file(uri).await?; + let file_schema: SchemaRef = Arc::new(arrow::datatypes::Schema::from( + reader.metadata().file_schema.as_ref(), + )); + // A column added by a later `ALTER TABLE ADD COLUMN` is absent from an + // older segment, and `ReaderProjection::from_column_names` would reject + // the missing name; the shaping tail null-fills it instead. + // `present_names` is never empty: every segment carries `row_uuid` and + // every read requests it, so the read yields a row count. + let (proj, read_schema) = match projection { + Some(names) => { + let present_names = present_columns(&file_schema, names); + ( + Self::build_projection(&reader, Some(&present_names))?, + project_schema(&file_schema, Some(&present_names))?, + ) + } + None => (None, file_schema), + }; + let params = match (offset, length) { + (Some(offset), Some(length)) => { + ReadBatchParams::Range(offset as usize..offset as usize + length as usize) + } + _ => ReadBatchParams::RangeFull, + }; + Self::read_with_params(&reader, &read_schema, proj, params).await } } @@ -149,33 +193,25 @@ impl FormatReader for LanceFormatReader { schema: &SchemaRef, projection: Option<&[&str]>, ) -> Result { - let output_schema = project_schema(schema, projection)?; - let reader = self.open_file(uri).await?; - // Project only the requested columns physically present in this - // segment file, then null-fill the rest to `output_schema`. A column - // added by a later `ALTER TABLE ADD COLUMN` is absent from an older - // segment; `ReaderProjection::from_column_names` would otherwise reject - // the missing name. `present_names` is never empty: every segment carries - // `row_uuid` and every read requests it, so the read yields a row count. - let file_arrow: SchemaRef = Arc::new(arrow::datatypes::Schema::from( - reader.metadata().file_schema.as_ref(), - )); - let column_names: Vec<&str> = match projection { - Some(cols) => cols.to_vec(), - None => schema.fields().iter().map(|f| f.name().as_str()).collect(), - }; - let present_names = present_columns(&file_arrow, &column_names); - let present_schema = project_schema(schema, Some(&present_names))?; - let proj = Self::build_projection(&reader, Some(&present_names))?; - let params = match (offset, length) { - (Some(offset), Some(length)) => { - ReadBatchParams::Range(offset as usize..offset as usize + length as usize) - } - _ => ReadBatchParams::RangeFull, - }; - let present = Self::read_with_params(&reader, &present_schema, proj, params).await?; - let out = null_fill_to_schema(&present, &output_schema)?; + let column_names = requested_columns(schema, projection); + let present = self + .read_in_file_schema(uri, offset, length, Some(&column_names)) + .await?; + let out = shape_to_schema(&present, schema, projection)?; tracing::Span::current().record("rows", out.num_rows()); Ok(out) } + + #[tracing::instrument( + skip_all, + fields(uri = %uri, offset = ?offset, length = ?length, format = "lance"), + )] + async fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> Result { + self.read_in_file_schema(uri, offset, length, None).await + } } diff --git a/crates/penca-format/src/reader/mod.rs b/crates/penca-format/src/reader/mod.rs index 8a50d80d..9c55ed0e 100644 --- a/crates/penca-format/src/reader/mod.rs +++ b/crates/penca-format/src/reader/mod.rs @@ -55,6 +55,31 @@ pub trait FormatReader: Send + Sync { schema: &SchemaRef, projection: Option<&[&str]>, ) -> impl Future> + Send; + + /// Read one segment file as it was written: the file's own columns, in the + /// file's own types, with no projection and no null-fill. The `(offset, + /// length)` slice still applies, so row count and row order match what + /// [`read_segment`](Self::read_segment) would return for the same slice. + /// + /// "The file's own types" means the schema the format engine embedded and + /// returns, not the table's stored `arrow_schema` — nothing here casts, so + /// the types are the encoder's answer rather than the catalog's. Both + /// encoders are expected to give the same answer, which is what lets + /// `SegmentCache` key on the content hash alone; CHA-548 is the coverage + /// that asserts it, and the open question in `docs/design-decisions.md` is + /// what would make it structural. + /// + /// Shape the result to a caller's schema afterwards with + /// [`shape_to_schema`] — the same tail `read_segment` runs internally. + /// Splitting the two is what lets one decode be cached and served to + /// callers whose schemas disagree: a caller-shaped cache entry would hand + /// the second caller the first caller's types (CHA-545). + fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> impl Future> + Send; } /// Errors from format read/write operations. @@ -115,6 +140,18 @@ impl FormatReader for AnyFormatReader { } } } + + async fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> Result { + match self { + Self::Parquet(r) => r.read_segment_native(uri, offset, length).await, + Self::Lance(r) => r.read_segment_native(uri, offset, length).await, + } + } } /// Create an empty `RecordBatch` matching the given schema. @@ -122,6 +159,18 @@ pub fn empty_batch(schema: &SchemaRef) -> RecordBatch { RecordBatch::new_empty(schema.clone()) } +/// Adapt a natively-decoded `batch` to `schema`/`projection`: the tail every +/// [`FormatReader::read_segment`] impl runs after its own read, exposed so a +/// caller that decoded via [`FormatReader::read_segment_native`] can run it +/// later — after a cache lookup — with identical behavior. +pub fn shape_to_schema( + batch: &RecordBatch, + schema: &SchemaRef, + projection: Option<&[&str]>, +) -> Result { + null_fill_to_schema(batch, &project_schema(schema, projection)?) +} + /// Resolve the effective output schema given an optional column projection. /// /// Returns `schema` unchanged when `projection` is `None`; otherwise returns @@ -174,6 +223,20 @@ pub(crate) fn null_fill_to_schema( Ok(RecordBatch::try_new(output_schema.clone(), columns)?) } +/// The columns a segment read should request from the file: the projection when +/// given, otherwise every column in `schema`. The `None` case is what makes a +/// projection-less [`FormatReader::read_segment`] return the caller's whole +/// schema, null-filling any column the file predates. +pub(crate) fn requested_columns<'a>( + schema: &'a SchemaRef, + projection: Option<&[&'a str]>, +) -> Vec<&'a str> { + match projection { + Some(cols) => cols.to_vec(), + None => schema.fields().iter().map(|f| f.name().as_str()).collect(), + } +} + /// The subset of `names` that actually exist as columns in `file_schema`, /// preserving the requested order. Used by readers to project only the columns /// physically present in a segment before null-filling the rest. @@ -222,6 +285,21 @@ mod tests { assert!(matches!(err, FormatError::UnknownProjectionColumn(name) if name == "zzz")); } + #[test] + fn requested_columns_passes_projection_through_and_expands_none() { + let schema = test_schema(); + assert_eq!( + requested_columns(&schema, Some(&["c", "a"])), + vec!["c", "a"], + "a projection is returned verbatim, in the requested order" + ); + assert_eq!( + requested_columns(&schema, None), + vec!["a", "b", "c"], + "no projection means every column in the caller's schema" + ); + } + #[test] fn present_columns_keeps_only_existing_in_file_order() { // file has {a, b}; request {b, missing, a} -> keep {b, a} in request order. diff --git a/crates/penca-format/src/reader/parquet.rs b/crates/penca-format/src/reader/parquet.rs index 265a5f2a..c8e048fe 100644 --- a/crates/penca-format/src/reader/parquet.rs +++ b/crates/penca-format/src/reader/parquet.rs @@ -18,7 +18,8 @@ use parquet::arrow::arrow_reader::{RowSelection, RowSelector}; use parquet::arrow::async_reader::{ParquetObjectReader, ParquetRecordBatchStreamBuilder}; use super::{ - FormatError, FormatReader, empty_batch, null_fill_to_schema, present_columns, project_schema, + FormatError, FormatReader, empty_batch, present_columns, project_schema, requested_columns, + shape_to_schema, }; use crate::uri::uri_to_object_path; @@ -45,6 +46,59 @@ impl ParquetFormatReader { pub fn new(store: Arc, base_uri: String) -> Self { Self { store, base_uri } } + + /// The read itself, in the file's own schema. `projection`, when `Some`, + /// is narrowed to the columns physically present and pushed down as a + /// [`ProjectionMask`] so only those byte ranges are fetched; `None` reads + /// every column the file has. + /// + /// Shaping to a caller's schema is the caller's job — `read_segment` runs + /// it immediately, `read_segment_native`'s callers run it after their + /// cache lookup. + async fn read_in_file_schema( + &self, + uri: &str, + offset: Option, + length: Option, + projection: Option<&[&str]>, + ) -> Result { + let path = uri_to_object_path(&self.base_uri, uri); + let reader = ParquetObjectReader::new(self.store.clone(), path); + let mut builder = ParquetRecordBatchStreamBuilder::new(reader).await?; + let file_schema = builder.schema().clone(); + + // A column added by a later `ALTER TABLE ADD COLUMN` is absent from an + // older segment; it is null-filled by the shaping tail rather than + // erroring the projection here. `present_names` is never empty: every + // Penca segment carries `row_uuid` and every read requests it + // (`snapshot_read_schema` prepends it), so the read below always + // yields a row count to null-fill against. + let read_schema = match projection { + Some(names) => { + let present_names = present_columns(&file_schema, names); + let parquet_schema = builder.parquet_schema().clone(); + let mask = ProjectionMask::columns(&parquet_schema, present_names.iter().copied()); + builder = builder.with_projection(mask); + project_schema(&file_schema, Some(&present_names))? + } + None => file_schema, + }; + + if let (Some(offset), Some(length)) = (offset, length) { + builder = builder.with_row_selection(slice_selection(offset as usize, length as usize)); + } + + let mut stream = builder.build()?; + let mut batches = Vec::new(); + while let Some(batch) = stream.next().await { + batches.push(batch?); + } + + if batches.is_empty() { + return Ok(empty_batch(&read_schema)); + } + Ok(concat_batches(&batches[0].schema(), &batches)?) + } } impl FormatReader for ParquetFormatReader { @@ -75,46 +129,25 @@ impl FormatReader for ParquetFormatReader { schema: &SchemaRef, projection: Option<&[&str]>, ) -> Result { - let output_schema = project_schema(schema, projection)?; - let column_names: Vec<&str> = match projection { - Some(cols) => cols.to_vec(), - None => schema.fields().iter().map(|f| f.name().as_str()).collect(), - }; - - let path = uri_to_object_path(&self.base_uri, uri); - let reader = ParquetObjectReader::new(self.store.clone(), path); - let mut builder = ParquetRecordBatchStreamBuilder::new(reader).await?; - - let parquet_schema = builder.parquet_schema().clone(); - // Project only the requested columns that physically exist in - // this segment file. A column added by a later `ALTER TABLE ADD COLUMN` - // is absent from an older segment; it is null-filled to `output_schema` - // after the read rather than erroring the projection. `present_names` - // is never empty: every Penca segment carries `row_uuid` and every - // read requests it (`snapshot_read_schema` prepends it), so the read - // below always yields a row count to null-fill against. - let present_names = present_columns(builder.schema(), &column_names); - let output_mask = ProjectionMask::columns(&parquet_schema, present_names.iter().copied()); - builder = builder.with_projection(output_mask); - - if let (Some(offset), Some(length)) = (offset, length) { - builder = builder.with_row_selection(slice_selection(offset as usize, length as usize)); - } - - let mut stream = builder.build()?; - let mut batches = Vec::new(); - while let Some(batch) = stream.next().await { - batches.push(batch?); - } - - if batches.is_empty() { - tracing::Span::current().record("rows", 0); - return Ok(empty_batch(&output_schema)); - } - - let present = concat_batches(&batches[0].schema(), &batches)?; - let out = null_fill_to_schema(&present, &output_schema)?; + let column_names = requested_columns(schema, projection); + let present = self + .read_in_file_schema(uri, offset, length, Some(&column_names)) + .await?; + let out = shape_to_schema(&present, schema, projection)?; tracing::Span::current().record("rows", out.num_rows()); Ok(out) } + + #[tracing::instrument( + skip_all, + fields(uri = %uri, offset = ?offset, length = ?length, format = "parquet"), + )] + async fn read_segment_native( + &self, + uri: &str, + offset: Option, + length: Option, + ) -> Result { + self.read_in_file_schema(uri, offset, length, None).await + } } diff --git a/crates/penca-merge/benches/floor_support.rs b/crates/penca-merge/benches/floor_support.rs index a07f5f15..a2c066c4 100644 --- a/crates/penca-merge/benches/floor_support.rs +++ b/crates/penca-merge/benches/floor_support.rs @@ -84,6 +84,15 @@ impl FormatReader for InMemoryFormatReader { ) -> Result { Ok(self.batch.clone()) } + + async fn read_segment_native( + &self, + _uri: &str, + _offset: Option, + _length: Option, + ) -> Result { + Ok(self.batch.clone()) + } } /// Build a real `DatafusionDlDriver` whose only reader (Parquet) serves `batch` @@ -228,6 +237,19 @@ impl FormatReader for SeekFormatReader { Ok(self.base.clone()) } } + + async fn read_segment_native( + &self, + uri: &str, + _offset: Option, + _length: Option, + ) -> Result { + if uri == "mem://sidecar" { + Ok(self.sidecar.clone()) + } else { + Ok(self.base.clone()) + } + } } /// A `DatafusionDlDriver` whose reader serves `base` + its `row_uuid` sidecar. @@ -259,7 +281,9 @@ pub fn base_segment_with_sidecar(size_bytes: i64, rows: i64) -> SnapshotSegment format: Format::Parquet, segment_index_uuid: "floor-sidecar".to_string(), size_bytes: 1 << 16, + content_hash: penca_core::naming::deterministic_uuid_from(&["floor-sidecar"]), }), + content_hash: penca_core::naming::deterministic_uuid_from(&["floor-seg"]), ..Default::default() } } diff --git a/crates/penca-merge/src/lib.rs b/crates/penca-merge/src/lib.rs index 2633ea3b..97308a76 100644 --- a/crates/penca-merge/src/lib.rs +++ b/crates/penca-merge/src/lib.rs @@ -1631,6 +1631,7 @@ mod tests { statistics: Vec::new(), row_uuid_index_sidecar: None, index_sidecars: Vec::new(), + content_hash: penca_core::naming::deterministic_uuid_from(&[uuid]), } } @@ -1661,6 +1662,7 @@ mod tests { offset: None, length: None, max_commit_seq_num: None, + content_hash: penca_core::naming::deterministic_uuid_from(&[uuid]), } } diff --git a/crates/penca-storage-cold/src/lib.rs b/crates/penca-storage-cold/src/lib.rs index 2bc3e658..61a4b78f 100644 --- a/crates/penca-storage-cold/src/lib.rs +++ b/crates/penca-storage-cold/src/lib.rs @@ -414,6 +414,9 @@ mod tests { format: Format::Parquet, segment_index_uuid: "idx".to_string(), size_bytes: 0, + // Never reaches a cache — this asserts format dispatch fails before + // any I/O — so the nil `Uuid::default()` cannot alias anything. + content_hash: Default::default(), }; let schema = penca_format::index::segment_index_schema(&[DataType::Utf8]); let err = ColdStorageClient::read_segment_index(&readers, &sidecar, &schema) diff --git a/crates/penca-storage-meta/src/fork_copy.rs b/crates/penca-storage-meta/src/fork_copy.rs index b73cc8f6..011cea4e 100644 --- a/crates/penca-storage-meta/src/fork_copy.rs +++ b/crates/penca-storage-meta/src/fork_copy.rs @@ -253,10 +253,12 @@ impl LifecycleManager { "INSERT INTO {seg} \ (table_snapshot_segment_uuid, table_snapshot_uuid, branch_uuid, \ table_uuid, chunk_idx, object_uri, \"offset\", length, size_bytes, \ - format, metadata, statistics, row_count, commit_micros) \ + format, metadata, statistics, row_count, content_hash, \ + commit_micros) \ SELECT m.new_uuid, $1, $2, old.table_uuid, m.chunk_idx, old.object_uri, \ old.\"offset\", old.length, old.size_bytes, old.format, \ - old.metadata, old.statistics, old.row_count, $4 \ + old.metadata, old.statistics, old.row_count, \ + old.content_hash, $4 \ FROM {seg_old} old \ JOIN unnest({new_arr}, {old_arr}, {chunk_arr}) \ AS m(new_uuid, old_uuid, chunk_idx) \ @@ -369,10 +371,10 @@ impl LifecycleManager { "INSERT INTO {sidecar} \ (segment_index_uuid, branch_uuid, segment_uuid, \ table_snapshot_index_uuid, object_uri, \"offset\", length, \ - format, size_bytes, statistics, commit_micros) \ + format, size_bytes, statistics, content_hash, commit_micros) \ SELECT m.new_sidecar, $1, m.new_seg, m.new_parent, old.object_uri, \ old.\"offset\", old.length, old.format, old.size_bytes, \ - old.statistics, $3 \ + old.statistics, old.content_hash, $3 \ FROM {sidecar_old} old \ JOIN unnest({a1}, {a2}, {a3}, {a4}, {a5}) \ AS m(new_sidecar, new_seg, new_parent, old_seg, old_parent) \ @@ -611,13 +613,13 @@ impl LifecycleManager { table_uuid, chunk_idx, min_tx_commit_micros, max_tx_commit_micros, \ min_commit_seq_num, max_commit_seq_num, object_uri, \"offset\", \ length, row_count, format, size_bytes, metadata, statistics, \ - is_sealed, commit_micros) \ + content_hash, is_sealed, commit_micros) \ SELECT m.new_uuid, $1, $2, old.table_uuid, old.chunk_idx, \ old.min_tx_commit_micros, old.max_tx_commit_micros, \ old.min_commit_seq_num, LEAST(old.max_commit_seq_num, $4), \ old.object_uri, old.\"offset\", old.length, old.row_count, \ old.format, old.size_bytes, old.metadata, old.statistics, \ - TRUE, $5 \ + old.content_hash, TRUE, $5 \ FROM {seg_old} old \ JOIN unnest({new_arr}, {old_arr}) AS m(new_uuid, old_uuid) \ ON old.table_persist_segment_uuid = m.old_uuid \ diff --git a/crates/penca-storage-meta/src/persist.rs b/crates/penca-storage-meta/src/persist.rs index 4e881fbb..c0d40acf 100644 --- a/crates/penca-storage-meta/src/persist.rs +++ b/crates/penca-storage-meta/src/persist.rs @@ -332,6 +332,7 @@ impl LifecycleManager { row_count: i64, format_text: &str, statistics: &[u8], + content_hash: &Uuid, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); let branch = parse_uuid(branch_uuid); @@ -344,13 +345,14 @@ impl LifecycleManager { (table_persist_segment_uuid, table_persist_uuid, branch_uuid, table_uuid, \ chunk_idx, min_tx_commit_micros, max_tx_commit_micros, \ min_commit_seq_num, max_commit_seq_num, \ - object_uri, row_count, format, statistics) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) \ + object_uri, row_count, format, statistics, content_hash) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) \ ON CONFLICT (branch_uuid, table_persist_segment_uuid) DO UPDATE \ SET object_uri = EXCLUDED.object_uri, \ row_count = EXCLUDED.row_count, \ format = EXCLUDED.format, \ - statistics = EXCLUDED.statistics", + statistics = EXCLUDED.statistics, \ + content_hash = EXCLUDED.content_hash", table = qi(&table), ); driver @@ -370,6 +372,7 @@ impl LifecycleManager { SqlValue::Int64(row_count), SqlValue::Text(format_text.to_string()), SqlValue::Bytes(statistics.to_vec()), + SqlValue::Uuid(*content_hash), ], ) .await?; @@ -413,6 +416,7 @@ impl LifecycleManager { size_bytes: i64, format_text: &str, statistics: &[u8], + content_hash: &Uuid, seal_now: bool, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); @@ -426,9 +430,10 @@ impl LifecycleManager { length = $3, \ size_bytes = $4, \ format = $5, \ - statistics = $6{seal_clause} \ - WHERE branch_uuid = $7 \ - AND table_persist_segment_uuid = $8", + statistics = $6, \ + content_hash = $7{seal_clause} \ + WHERE branch_uuid = $8 \ + AND table_persist_segment_uuid = $9", table = qi(&table), ); driver @@ -441,6 +446,7 @@ impl LifecycleManager { SqlValue::Int64(size_bytes), SqlValue::Text(format_text.to_string()), SqlValue::Bytes(statistics.to_vec()), + SqlValue::Uuid(*content_hash), SqlValue::Uuid(branch), SqlValue::uuid_str(table_persist_segment_uuid)?, ], @@ -700,6 +706,7 @@ impl LifecycleManager { let mut sql = format!( "SELECT seg.table_persist_segment_uuid, seg.table_persist_uuid, seg.object_uri, \ seg.\"offset\", seg.length, seg.format, seg.row_count, seg.size_bytes, \ + seg.content_hash, \ seg.table_uuid, seg.min_tx_commit_micros, seg.max_tx_commit_micros, \ seg.max_commit_seq_num, seg.is_sealed, tfm.log_kind \ FROM {seg} seg \ diff --git a/crates/penca-storage-meta/src/segment_index.rs b/crates/penca-storage-meta/src/segment_index.rs index 0378008c..5671854a 100644 --- a/crates/penca-storage-meta/src/segment_index.rs +++ b/crates/penca-storage-meta/src/segment_index.rs @@ -255,6 +255,7 @@ impl LifecycleManager { format_text: &str, size_bytes: i64, statistics: &[u8], + content_hash: &Uuid, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); let branch = parse_uuid(branch_uuid); @@ -262,8 +263,9 @@ impl LifecycleManager { let sql = format!( "INSERT INTO {table} \ (segment_index_uuid, branch_uuid, segment_uuid, table_snapshot_index_uuid, \ - object_uri, \"offset\", length, format, size_bytes, statistics) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) \ + object_uri, \"offset\", length, format, size_bytes, statistics, \ + content_hash) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ ON CONFLICT (branch_uuid, segment_index_uuid) DO UPDATE \ SET segment_uuid = EXCLUDED.segment_uuid, \ table_snapshot_index_uuid = EXCLUDED.table_snapshot_index_uuid, \ @@ -272,7 +274,8 @@ impl LifecycleManager { length = EXCLUDED.length, \ format = EXCLUDED.format, \ size_bytes = EXCLUDED.size_bytes, \ - statistics = EXCLUDED.statistics", + statistics = EXCLUDED.statistics, \ + content_hash = EXCLUDED.content_hash", table = qi(&table), ); driver @@ -289,6 +292,7 @@ impl LifecycleManager { SqlValue::Text(format_text.to_string()), SqlValue::Int64(size_bytes), SqlValue::Bytes(statistics.to_vec()), + SqlValue::Uuid(*content_hash), ], ) .await?; @@ -568,10 +572,11 @@ impl LifecycleManager { let sql = format!( "INSERT INTO {table} \ (segment_index_uuid, branch_uuid, segment_uuid, table_snapshot_index_uuid, \ - object_uri, \"offset\", length, format, size_bytes, statistics) \ + object_uri, \"offset\", length, format, size_bytes, statistics, \ + content_hash) \ SELECT n.new_sidecar, $1, n.new_seg, $2, \ old.object_uri, old.\"offset\", old.length, old.format, \ - old.size_bytes, old.statistics \ + old.size_bytes, old.statistics, old.content_hash \ FROM UNNEST({new_seg_arr}, {new_sidecar_arr}, {prior_sidecar_arr}) \ AS n(new_seg, new_sidecar, prior_sidecar) \ JOIN {source_table} old \ @@ -586,7 +591,8 @@ impl LifecycleManager { length = EXCLUDED.length, \ format = EXCLUDED.format, \ size_bytes = EXCLUDED.size_bytes, \ - statistics = EXCLUDED.statistics", + statistics = EXCLUDED.statistics, \ + content_hash = EXCLUDED.content_hash", table = qi(&table), source_table = qi(&source_table), ); diff --git a/crates/penca-storage-meta/src/snapshot.rs b/crates/penca-storage-meta/src/snapshot.rs index f1b3e179..edf773f0 100644 --- a/crates/penca-storage-meta/src/snapshot.rs +++ b/crates/penca-storage-meta/src/snapshot.rs @@ -132,6 +132,7 @@ impl LifecycleManager { row_count: i64, format_text: &str, statistics: &[u8], + content_hash: &Uuid, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); let branch = parse_uuid(branch_uuid); @@ -139,15 +140,17 @@ impl LifecycleManager { let sql = format!( "INSERT INTO {table} \ (table_snapshot_segment_uuid, table_snapshot_uuid, branch_uuid, table_uuid, \ - chunk_idx, object_uri, \"offset\", length, row_count, format, statistics) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ + chunk_idx, object_uri, \"offset\", length, row_count, format, statistics, \ + content_hash) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \ ON CONFLICT (branch_uuid, table_snapshot_segment_uuid) DO UPDATE \ SET object_uri = EXCLUDED.object_uri, \ \"offset\" = EXCLUDED.\"offset\", \ length = EXCLUDED.length, \ row_count = EXCLUDED.row_count, \ format = EXCLUDED.format, \ - statistics = EXCLUDED.statistics", + statistics = EXCLUDED.statistics, \ + content_hash = EXCLUDED.content_hash", table = qi(&table), ); driver @@ -165,6 +168,7 @@ impl LifecycleManager { SqlValue::Int64(row_count), SqlValue::Text(format_text.to_string()), SqlValue::Bytes(statistics.to_vec()), + SqlValue::Uuid(*content_hash), ], ) .await?; @@ -566,10 +570,11 @@ impl LifecycleManager { "INSERT INTO {table} \ (table_snapshot_segment_uuid, table_snapshot_uuid, branch_uuid, \ table_uuid, chunk_idx, object_uri, \"offset\", length, \ - row_count, size_bytes, format, metadata, statistics) \ + row_count, size_bytes, format, metadata, statistics, content_hash) \ SELECT new.uuid, $1, $2, old.table_uuid, new.idx, \ old.object_uri, old.\"offset\", old.length, old.row_count, \ - old.size_bytes, old.format, old.metadata, old.statistics \ + old.size_bytes, old.format, old.metadata, old.statistics, \ + old.content_hash \ FROM UNNEST({new_arr}, {idx_arr}, {prior_arr}) \ AS new(uuid, idx, old_uuid) \ JOIN {source_table} old \ @@ -585,7 +590,8 @@ impl LifecycleManager { size_bytes = EXCLUDED.size_bytes, \ format = EXCLUDED.format, \ metadata = EXCLUDED.metadata, \ - statistics = EXCLUDED.statistics \ + statistics = EXCLUDED.statistics, \ + content_hash = EXCLUDED.content_hash \ RETURNING table_snapshot_segment_uuid", table = qi(&table), source_table = qi(&source_table), diff --git a/docs/design-decisions.md b/docs/design-decisions.md index c82e70dc..271c155f 100644 --- a/docs/design-decisions.md +++ b/docs/design-decisions.md @@ -180,3 +180,180 @@ reads footers, `register_listing_table` hits object storage, Tracked upstream: a real `clone_ctx()` in `datafusion-python` would let Python match Rust's mechanics exactly; until then the factory is the correct shape. + +--- + +## Cold segments are cached by content hash, not by row uuid + +Every cold-artifact metadata row — persist segments, snapshot segments, and +segment index sidecars — carries a `content_hash`: an `xxh3_128` digest of the +**typed in-memory Arrow batch**, computed once at write time +(`penca_core::digest::segment_content_hash`). `SegmentCache` is keyed by that +hash alone. + +**The problem.** Reference copies mint a new row uuid over an unchanged +`(object_uri, offset, length)`: `insert_carried_snapshot_segments` and +`insert_carried_segment_indexes` carry a snapshot forward (CHA-531), and +`fork_copy` materializes a fork's cold references at fork time (CHA-539). +Nobody rewrites the bytes. A uuid-keyed cache therefore stored N identical +decodes of one byte range — N× the memory under a fixed byte budget, and N cold +reads to fill it. A fork's cold footprint is *mostly* its parent's files, so N +grows with the branch count, which is the thing Penca expects to be cheap. + +**Why hash the decoded batch, not the encoded file bytes.** The key has to name +one decoded batch — equal hash must mean equal decode, or two unrelated segments +would share an entry. Hashing the typed batch gives that directly, and it is what +is available where the digest is taken: write time, *before* the format writer +encodes. Hashing the stored object would mean reading back what was just written. +It also makes dedup insensitive to encoding choices, so two independently written +segments holding the same rows share an entry even if the writer picked different +row-group boundaries or dictionary encodings. + +**Why the storage format is deliberately not in the key.** Digesting the batch +pre-encode is exactly what lets one hash name files in two formats, once +`OBJECT_STORAGE_FORMAT` has been flipped between two writes of the same content. +Those two files decode to the same Arrow batch under Penca's storage contract — a +write followed by a read returns the batch that was written — so the hash alone +names one decode and the two files share one entry. A Parquet/Lance divergence +would not be a fact for the key to record; it would mean one format's round trip +is *wrong*, which is fixed in that format (or by dropping the offending type from +`CanonicalType`), not absorbed by a wider key. + +That contract is currently unasserted: +`crates/penca-format/tests/format_reader.rs` checks row and column counts through +the shaped path, nothing compares a decoded batch against the batch that was +written, and nothing compares the two formats against each other. CHA-548 adds +that coverage. Pairing the format into the key in the meantime was considered and +rejected — it would permanently give up a real cross-format share to pre-empt a +defect nobody has demonstrated, and if the defect did exist it would mask it, +turning the loud cross-format type mismatch that CHA-548 is meant to produce into +silence. + +**What the key deliberately does *not* do is separate a reference copy from its +source.** Carry-forward and fork copy select `old.content_hash` verbatim +(`fork_copy.rs`, `snapshot.rs`, `segment_index.rs`) and it is never recomputed — +that inheritance *is* the dedup. So when a fork `ALTER`s a column, its rows still +read the parent's bytes under the parent's hash: parent and fork share one cache +entry while disagreeing about that column's type, and no hashing scheme could +separate them, because neither row's digest was ever taken under its own read +schema. + +That is what forces the cached *value* to be the file-native decode, with +caller-shaping (projection + null-fill of columns added by a later +`ALTER TABLE ADD COLUMN`) moved *after* the lookup — +`FormatReader::read_segment_native` plus `reader::shape_to_schema`. A +caller-shaped entry carries the schema of whichever branch decoded first, so the +second branch's read fails on a type mismatch its own metadata never justified. + +Concretely, with a file holding `{row_uuid, name, value}` that both branches +reference, parent having run `ADD COLUMN extra Int64` and fork `ADD COLUMN extra +Utf8`: + + caching the shaped batch — the bug + fork miss → decode + fill → {row_uuid, name, value, extra: Utf8} + parent HIT → gets extra: Utf8, its own metadata says Int64 ✗ + + caching the native decode — what ships + fork miss → decode → {row_uuid, name, value} + → shape to fork → extra: Utf8 ✓ + parent HIT → {row_uuid, name, value} + → shape to parent → extra: Int64 ✓ + +The read-time schema still governs the output — it just governs it at the +shaping step, per caller, rather than at the decode. Only the decode has to be +segment-scoped, because only the decode is shared. + +### Rejected alternative: fold the read schema into the key + +The symmetric design adds a fingerprint of the read-time schema to the key and +caches a value already adapted to it. Both halves are known before the read, so +it is implementable, and it is correct: the fingerprint separates the two +branches above. It comes in two strengths, and the weaker one is not worth +arguing against. + +Caching the fully **shaped** table — projection included — fragments the key +space by query: `SELECT name` and `SELECT name, value` over one segment become +different entries, combinatorial in the column subsets queries touch. That is +decisive but it is also avoidable, so it is not the real comparison. + +The strong form caches the **type-cast** table: present columns cast to the +read-time schema's declared types, with projection and null-fill still applied +per caller after the lookup. Its cost is that the key still moves when the +schema does, and only the *key* moves — the data does not: + +- **`ALTER TABLE ADD COLUMN` evicts a footprint that did not change.** The added + column is in no file and affects no cast, yet it re-fingerprints every segment + of the table, so the whole cold footprint is re-fetched and re-decoded without + a byte having been rewritten. Narrowing the fingerprint to only the columns a + file actually holds would fix this, but which columns those are is not known + until after the read. +- **It collapses the dedup this entry exists for.** A fork and its parent share + entries only while their schemas agree; the first `ALTER` on either side + duplicates the entire shared footprint — the sharing degrades exactly when a + branch is used for what branches are for. That is the motivating case, not an + edge case. + +Its advantage is real but currently unrealizable: a cached cast would let a read +succeed where the file's decoded type differs from the declared one, which the +shipped path cannot do — `null_fill_to_schema` takes present columns verbatim and +`RecordBatch::try_new` then rejects a mismatch. Penca's schema evolution is +`ADD COLUMN` only, so no supported operation produces that divergence. If type +evolution is added later, the answer is to cast in the shaping tail, where it +applies per caller and is visible as a data-semantics decision — not to hide it +behind a cache key. + +### What directs the decode today, and what should + +Neither reader consults the stored schema. Parquet decodes under the file's +embedded Arrow schema (`builder.schema()`) and Lance under +`reader.metadata().file_schema`; the caller's `SchemaRef` only selects column +*names* to request (`requested_columns`) and drives the shaping tail. No cast or +type coercion happens anywhere in `penca-format`'s readers or writers, so the +decoded types are whatever the encoder chose to write and the decoder chose to +return. + +That is what the cache key assumes away: the decoded types are the encoder's +answer, and one key for one hash trusts both encoders to give the same answer. +`shape_to_schema` would not absorb a divergence either — `null_fill_to_schema` +takes present columns by name verbatim, and `RecordBatch::try_new` then validates +their types against the output schema, so a mismatch is a hard read error, not a +cast. CHA-548 tests the assumption; directing the decode from the stored schema +(below) would remove the need for it. + +**Why the scope is uniform across artifact classes.** Base segments and index +sidecars both come out of object storage through the same `SegmentCache` and are +both reference-copied by carry-forward and fork copy, so they duplicate for the +same reason and dedup the same way. Sidecars were briefly scoped out on the +argument that `segment_index_schema(key_types)` makes the cached value +caller-dependent — but that argument is wrong for the same reason it is wrong +for base segments, and the native-decode split answers it. + +`content_hash` is `NOT NULL` with no default on all three tables +(`penca-db/src/dialect/pg.rs`). Every writer computes it and a catalog predating +the column is recreated rather than migrated, so there is no legacy row to +default and no uuid-fallback key space to keep alive. + +**Non-goal.** `tx_log_persist_segment_metadata` has no `content_hash`: it is +never cache-read and never reference-copied. + +**Open question.** CHA-545 names checksum reuse as a possible second use of this +digest. It is not available yet: the digest is taken on the in-memory batch +before the format writer encodes it, so verifying a stored file against it means +re-deriving the digest from a decode — sound only once round-trip identity is +established. CHA-548 is that gate too. + +**Open question.** The decode should be directed by the segment's *stored* +write-time schema rather than by the file's embedded one. `__penca_system__.tables` +already holds it — `arrow_schema` is Arrow IPC and every row is a complete table +definition at a point in time — and a segment row carries `branch_uuid`, +`table_uuid`, and `max_commit_seq_num`, so the definition in force when it was +written is an as-of read away. Doing that would make the decoded schema a function +of metadata instead of of the encoder's round-trip behavior, let a file whose +physical schema contradicts its declared one be rejected rather than silently +trusted, and turn the cross-format assumption above into a guarantee rather than +something CHA-548 has to keep testing — both formats would decode to the declared +types by construction. It must be the segment's write-time schema and not the +reader's: the cache key is a property of the segment alone, so anything that +varies per reader cannot direct a shared decode. The cost is resolving that +schema per segment, which is plumbing through the read path rather than a change +local to `penca-format`. diff --git a/docs/schema-reference.md b/docs/schema-reference.md index dea19fd5..5cbdde38 100644 --- a/docs/schema-reference.md +++ b/docs/schema-reference.md @@ -378,6 +378,7 @@ the `log_kind` classification (CHA-218: only `upsert_log` and | `length` | int64 (set at compact time) | | `row_count` | int64 | | `format` | text (parquet, lance) | +| `content_hash` | UUID NOT NULL (CHA-545, segment-cache key) | | `size_bytes` | int64 | | `metadata` | JSONB | | `statistics` | JSONB | @@ -387,6 +388,17 @@ the `log_kind` classification (CHA-218: only `upsert_log` and Indices: implicit `(branch_uuid, table_persist_segment_uuid)` PK + per-branch partition. +`content_hash` is the `xxh3_128` of the segment's typed in-memory Arrow +batch, computed once at write time and inherited verbatim by every +reference copy (carry-forward, CHA-539 fork copy). It keys the in-process +`SegmentCache`, so a fork and its parent share one decoded entry for a byte +range they both reference — which the row uuid cannot express, since a +reference copy mints a fresh uuid over bytes it did not rewrite. `NOT NULL` +with no default: every writer computes it, and a catalog predating the +column is recreated rather than migrated (there is no in-place migration +path for catalog metadata), so there is no legacy row to default. Not +indexed — reads carry the value through, they never look a row up by it. + Each level's `written_at_micros` / `commit_micros` pair supports the three-level commit decoupling. Rows with `commit_micros IS NULL` (at any level) are invisible to reads @@ -417,6 +429,12 @@ committed rows gates `PurgeTxLog`; reads seek the sorted `commit_seq_num` / | `format` | text (parquet, lance) | | `committed_at_micros` | int64 (NULL until per-segment commit) | +Deliberately carries **no** `content_hash`, unlike the three cold-artifact +tables around it (14, 16, 17): a cold `tx_log` file is read by +`read_tx_log_batches`, never through `SegmentCache`, and is never +reference-copied — so it has neither of the two properties the hash exists +to serve. + **15. Table snapshot metadata** — `{catalog_uuid}_table_snapshot_metadata` Per-catalog, LIST-partitioned by `branch_uuid` (CHA-198). One row per @@ -464,6 +482,7 @@ completes). | `length` | int64 NOT NULL (row count of the range) | | `size_bytes` | int64 | | `format` | text (lance, parquet) | +| `content_hash` | UUID NOT NULL (CHA-545, segment-cache key — see table 14) | | `metadata` | JSON (format-specific, e.g., row group size) | | `statistics` | JSON (column stats: min/max for filterable columns) | | `row_count` | int64 | @@ -520,6 +539,7 @@ forward by reference with its base segment and participates in the ref-counted G | `offset` | int64 | | `length` | int64 | | `format` | text (lance, parquet) | +| `content_hash` | UUID NOT NULL (CHA-545, segment-cache key — see table 14) | | `size_bytes` | int64 | | `statistics` | bytes (indexed-key min/max bounds; binary, decoded in-planner by the CHA-454 seek in the `SnapshotTableProvider`) | | `written_at_micros` | int64 (micros, auto-generated) | @@ -531,6 +551,13 @@ Indices: implicit `(branch_uuid, …_uuid)` PKs + per-branch partitions; child `index_uuid IS NULL` (role: internal vs user secondary) and `index_type` (physical layout, on `__penca_system__.indexes`) are orthogonal. +Sidecars carry `content_hash` for the same reason base segments do: they are +read out of object storage through the same `SegmentCache` and copied by +reference by both carry-forward and the CHA-539 fork copy. `segment_index_uuid` +is stable across a carry-forward *within* a branch, but a fork derives it from +the child's own `segment_uuid` and so mints a new id over unchanged bytes — +which is exactly the duplication the hash collapses. + ## Data tables (per-branch, per-table) Both user tables and the two system tables (`__penca_system__.schemas`, diff --git a/tests/integration/integration_branch_persist_at_fork_test.py b/tests/integration/integration_branch_persist_at_fork_test.py index 3f585242..38e90918 100644 --- a/tests/integration/integration_branch_persist_at_fork_test.py +++ b/tests/integration/integration_branch_persist_at_fork_test.py @@ -1106,3 +1106,137 @@ def test_fork_audit_does_not_double_match_a_straddling_parent_tx_log(): f"{post_fork!r} was committed on main above the fork and must not " f"appear in the child's audit. Full list: {names}" ) + + +def test_fork_and_parent_diverge_a_columns_type_over_one_shared_slice(): + """Two branches whose schemas diverged must not share a cached decode. + + CHA-545 keys the segment cache on `content_hash`, so a fork's copied row and + the parent's row over the same bytes share one entry — that is the point, + since a fork's cold footprint is mostly the parent's files and the copy + inherits the parent's hash rather than minting its own. + + But the cached value is not the file's native decode: `read_segment` + null-fills to the CALLER's schema, so an entry carries the schema of + whichever branch decoded it first. Since CHA-539 the fork names the parent's + `(uri, offset, length)` while ALTERing independently, so that entry can be + handed to a branch expecting a different type for the same column: whichever + branch missed first caches `extra` as an all-null array of its own type and + the other branch's projection fails on the mismatch — a query error driven by + cache state, on a branch whose own metadata is perfectly consistent. What + keeps that from happening is caching the file's NATIVE decode and null-filling + to the caller's schema after the lookup, not before it. + + Reads child-then-parent, and that one order is sufficient AND deterministic: + within a process the child misses, caches `extra` as Utf8, and the parent's + read is the hit that used to fail with "expected Int64 but found Utf8". Only + the PARENT assertion is the regression guard — the child's read is always the + miss, so its assertion holds with or without the fix and is a sanity check. + The reverse direction (parent poisons child) is not covered here; it needs a + second fixture on a fresh catalog, since after this pair the file is already + cached under its content hash. + """ + client = make_client() + schema_uuid, table_uuid, catalog_uuid, main_branch = setup_schema(client) + scope = { + "catalog_uuid": catalog_uuid, + "schema_uuid": schema_uuid, + "table_uuid": table_uuid, + } + + fork_seq = _write_committed_rows( + client, + branch_uuid=main_branch, + rows={"name": ["shared"], "value": [1]}, + **scope, + ) + client.persist(branch_uuid=main_branch, **scope) + client.snapshot(branch_uuid=main_branch, **scope) + + child = client.create_branch( + "kid", "t", "fork", commit_seq_num=fork_seq, catalog_uuid=catalog_uuid + ).branch_uuid + + # Same new column, incompatible types, one on each branch — over the single + # base slice the fork inherited by reference. + parent_schema = pa.schema( + [*list(USER_SCHEMA), pa.field("extra", pa.int64(), nullable=True)] + ) + child_schema = pa.schema( + [*list(USER_SCHEMA), pa.field("extra", pa.string(), nullable=True)] + ) + client.update_table( + parent_schema, + branch_uuid=main_branch, + primary_keys=["name"], + author="test", + comment="parent adds extra BIGINT", + **scope, + ) + client.update_table( + child_schema, + branch_uuid=child, + primary_keys=["name"], + author="test", + comment="fork adds extra TEXT", + **scope, + ) + + # The precondition the scenario rests on: the fork's cold row and the + # parent's must land on the SAME cache entry, or the two branches decode + # independently, every assertion below still passes, and this test covers + # nothing. Metadata-only, so it does not warm the cache it is about to probe. + # + # `content_hash` is the join that matters — it is the cache key (CHA-545). + # Matching `object_uri`/`offset`/`length` too pins WHY they collide: the + # fork's row is a reference copy of the parent's slice that inherited its + # hash, not two writes that happened to produce the same bytes. Joining on + # addressing alone would pass even if the copy minted a fresh hash, which + # is the failure mode that would silently retire this test. + # + # BOTH tiers, because which one carries the shared slice depends on the + # fixture: here the snapshot covers the whole persist, so `copy_inherited_ + # persist` finds nothing above the baseline watermark and the inheritance is + # entirely snapshot segments. Checking only the persist table looked correct + # and reported "no shared slice" on a fixture that has one. + shared = 0 + for base in (TABLE_PERSIST_SEGMENT_METADATA, TABLE_SNAPSHOT_SEGMENT_METADATA): + rows = get_pg_driver().execute( + SQL( + "SELECT count(*) FROM {tbl} p JOIN {tbl} c" + " ON p.content_hash = c.content_hash AND p.object_uri = c.object_uri" + ' AND p."offset" IS NOT DISTINCT FROM c."offset"' + " AND p.length IS NOT DISTINCT FROM c.length" + " WHERE p.branch_uuid = %s AND c.branch_uuid = %s AND p.table_uuid = %s" + ).format(tbl=Identifier(f"{catalog_uuid}_{base}")), + (main_branch, child, table_uuid), + ) + shared += rows[0][0] + + if shared == 0: + raise RuntimeError( + "setup failed: the fork holds no cold row sharing a content_hash " + "with the parent's slice in either tier, so both branches would " + "decode independently and the cross-branch cache sharing this test " + "pins is unreachable" + ) + + # Child first, then parent: the child's decode populates the cache and the + # parent must not be served it. + child_rows = client.read_data(branch_uuid=child, **scope) + parent_rows = client.read_data(branch_uuid=main_branch, **scope) + + assert child_rows.column("name").to_pylist() == ["shared"], ( + "the fork lost its inherited row" + ) + assert parent_rows.column("name").to_pylist() == ["shared"], ( + "the parent lost its own row" + ) + assert pa.types.is_string(child_rows.schema.field("extra").type), ( + "the fork's read must honour ITS schema for the diverged column, not the " + f"branch that decoded the shared slice first: {child_rows.schema}" + ) + assert pa.types.is_int64(parent_rows.schema.field("extra").type), ( + "the parent's read must honour ITS schema for the diverged column, not " + f"the branch that decoded the shared slice first: {parent_rows.schema}" + ) diff --git a/tests/integration/integration_cold_user_index_build_test.py b/tests/integration/integration_cold_user_index_build_test.py index 091640e0..bb6c5d48 100644 --- a/tests/integration/integration_cold_user_index_build_test.py +++ b/tests/integration/integration_cold_user_index_build_test.py @@ -222,6 +222,35 @@ def _user_child(catalog_uuid, branch_uuid, segment_uuid, index_uuid): return None +def _segment_content_hash(catalog_uuid, branch_uuid, segment_uuid): + """A base segment's ``content_hash``, as text.""" + rows = get_pg_driver().execute( + SQL( + "SELECT content_hash::text FROM {tbl}" + " WHERE branch_uuid = %s AND table_snapshot_segment_uuid = %s" + ).format(tbl=Identifier(f"{catalog_uuid}_{TABLE_SNAPSHOT_SEGMENT_METADATA}")), + (branch_uuid, segment_uuid), + ) + return rows[0][0] + + +def _sidecar_content_hash(catalog_uuid, branch_uuid, segment_uuid, link): + """The ``content_hash`` of ``segment_uuid``'s sidecar under one parent + index, as text. ``link`` is the parent ``table_snapshot_index_uuid`` + returned by :func:`_user_child`.""" + rows = get_pg_driver().execute( + SQL( + "SELECT content_hash::text FROM {tbl}" + " WHERE branch_uuid = %s AND segment_uuid = %s" + " AND table_snapshot_index_uuid = %s" + ).format( + tbl=Identifier(f"{catalog_uuid}_{TABLE_SNAPSHOT_SEGMENT_INDEX_METADATA}") + ), + (branch_uuid, segment_uuid, link), + ) + return rows[0][0] + + class TestUserIndexBuild: """A cold snapshot materializes user ``CREATE INDEX`` definitions: a committed parent (``index_uuid`` non-NULL) + one committed sidecar per base @@ -509,12 +538,18 @@ def test_user_index_sidecar_carries_forward_by_reference(self): table_uuid, pa.table({"name": ["alice", "carol"], "value": [1, 3]}, schema=USER_SCHEMA), ) - # Map each snap1 base file -> its user sidecar uri. + # Map each snap1 base file -> its user sidecar uri, and both rows' + # content hashes for the inheritance assertion below. snap1_sidecar_by_file = {} + snap1_hashes_by_file = {} for seg, uri, off in _base_segment_tuples(catalog_uuid, branch, snap1): child = _user_child(catalog_uuid, branch, seg, index_uuid) assert child is not None, "snap1 must build a user sidecar per segment" snap1_sidecar_by_file[(uri, off)] = child[1] # object_uri + snap1_hashes_by_file[(uri, off)] = ( + _segment_content_hash(catalog_uuid, branch, seg), + _sidecar_content_hash(catalog_uuid, branch, seg, child[0]), + ) # snap2 rewrites only alice -> carol's segment carries forward, and its # already-built sidecar must carry by reference (same object_uri). @@ -540,3 +575,19 @@ def test_user_index_sidecar_carries_forward_by_reference(self): f" by reference, no rebuild): {child[1]} !=" f" {snap1_sidecar_by_file[(uri, off)]}" ) + # CHA-545: a carried row is a fresh uuid over bytes nobody rewrote, + # so it must inherit the prior row's content_hash verbatim. That + # inheritance is the whole dedup — recomputing, or defaulting, would + # give the same bytes two cache entries. + prior_seg_hash, prior_sidecar_hash = snap1_hashes_by_file[(uri, off)] + assert _segment_content_hash(catalog_uuid, branch, seg) == prior_seg_hash, ( + "a carried base segment must inherit its prior row's" + f" content_hash, expected {prior_seg_hash}" + ) + assert ( + _sidecar_content_hash(catalog_uuid, branch, seg, child[0]) + == prior_sidecar_hash + ), ( + "a carried sidecar must inherit its prior row's content_hash," + f" expected {prior_sidecar_hash}" + ) diff --git a/tests/integration/integration_lifecycle_test.py b/tests/integration/integration_lifecycle_test.py index e0c470b7..e021a0fd 100644 --- a/tests/integration/integration_lifecycle_test.py +++ b/tests/integration/integration_lifecycle_test.py @@ -665,11 +665,11 @@ def test_uncommitted_segment_invisible_to_reads(self): SQL( "INSERT INTO {tbl}" " (table_persist_segment_uuid, table_persist_uuid, branch_uuid," - " table_uuid," + " table_uuid, content_hash," " min_tx_commit_micros, max_tx_commit_micros," " min_commit_seq_num, max_commit_seq_num," " object_uri, row_count, format)" - " VALUES (%s, %s, %s, %s, 0, 0," + " VALUES (%s, %s, %s, %s, %s, 0, 0," " 0, 0," " 'fake://upsert', 999, 'parquet')" ).format(tbl=Identifier(segment_parent)), @@ -678,6 +678,10 @@ def test_uncommitted_segment_invisible_to_reads(self): fake_table_persist_uuid, branch.branch_uuid, table_uuid, + # NOT NULL with no default (CHA-545). Irrelevant to what this + # test asserts, but random rather than fixed so the fake row + # can never collide with a real segment's cache key. + str(uuid4()), ), )