CHA-545 | key the cold segment cache by content hash, not row uuid

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?;

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<u8>,
+    pub content_hash: Uuid,
 }
 
 impl<'a> DurableSegmentWriter<SnapshotSegmentScope<'a>> {
@@ -456,6 +469,7 @@ impl<'a> DurableSegmentWriter<SnapshotSegmentScope<'a>> {
                 row.length,
                 self.scope.storage_format.extension(),
                 &row.statistics,
+                &row.content_hash,
             )
             .await?;
             self.current_group()

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<Option<SnapshotFileStep>, 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 {

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<SnapshotFileStep, ApiError> {
     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<W: FormatWriter>(
         // 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);

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<Option<IndexSidecar>> {
     row.get::<Option<String>, _>("sidecar_object_uri")
         .map(|object_uri| -> Result<IndexSidecar> {
@@ -1471,6 +1475,10 @@ fn decode_child_sidecar(row: &PgRow) -> Result<Option<IndexSidecar>> {
                 format: sidecar_format,
                 segment_index_uuid: row.get::<Uuid, _>("sidecar_segment_index_uuid").to_string(),
                 size_bytes: row.get::<Option<i64>, _>("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::<Uuid, _>("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]),
         }
     }