Phase E.2: Compaction integrates tombstones — physical deletion closes GDPR loop

Phase E gave us soft-delete at query time (tombstones hide rows via a
DataFusion filter view). This completes the invariant: after compact,
tombstoned rows are PHYSICALLY absent from the parquet on disk.

delta::compact changes:
- Signature adds tombstones: &[Tombstone]
- After merging base + deltas, apply_tombstone_filter builds a
  BooleanArray keep-mask per batch (True where row_key_value is NOT
  in the tombstone set) and applies arrow::compute::filter_record_batch
- Supports Utf8, Int32, Int64 key columns (matches refresh.rs coverage
  for pg- and csv-derived schemas)
- CompactResult gains tombstones_applied + rows_dropped_by_tombstones
- Caller clears tombstone store on success

Critical correctness fix surfaced during E2E testing:
The original Phase 8 compact concatenated N independent Parquet byte
streams from record_batch_to_parquet() — each with its own footer.
Parquet readers only see the FIRST footer's data; the rest is invisible.
Latent since Phase 8 shipped; triggered by tombstone-filtering produc-
ing multiple batches. Corrupted candidates.parquet on first test run
(restored from UI fixture copy — good argument for test data in repo).

Fix:
- Single ArrowWriter per compaction, writes every batch into one
  properly-footered Parquet
- Snappy compression to match ingest defaults (otherwise rewrite
  inflated file 3× — 10.5MB → 34MB — because no compression was set)
- Verify-before-swap: parse written buf back to confirm row count
  matches expected; refuses to overwrite base_key if verification fails
- Write to {base_key}.compact-{ts}.tmp first, then to base_key; delete
  temp; only then delete delta files. Any error along the way leaves
  the original base intact.

TombstoneStore::clear(dataset) drops all tombstone batch files and
evicts the per-dataset AppendLog from cache. Called after successful
compact.

QueryEngine::catalog() accessor exposes the Registry so queryd
handlers can reach the tombstone store without routing through gateway
state.

E2E on candidates (100K rows, 15 cols):
- Baseline: 10.59 MB, 100000 rows
- Tombstone CAND-000001/2/3 (soft-delete): 99997 visible, 100000 raw
- Compact: tombstones_applied=3, rows_dropped=3, final_rows=99997
- Post: 10.72 MB (Snappy), valid parquet (1 row_group), 99997 rows
- Restart: persists, tombstones list empty, __raw__candidates also
  99997 (the 3 IDs are physically gone from disk)

PRD invariant close: deletion is now actually deletion, not just
masking. GDPR erasure request → tombstone + schedule compact → data
gone.

Deferred:
- Compact-all-datasets cron (currently manual per-dataset via
  POST /query/compact)
- Compaction of tombstone batch files themselves (they grow at
  flush_threshold=1 per tombstone; TombstoneStore::compact exists
  but not auto-called)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

crates/catalogd/src/tombstones.rs

@@ -121,4 +121,31 @@ impl TombstoneStore {
         let log = self.log_for(dataset).await;
         log.compact().await
     }
+
+    /// Remove every tombstone for a dataset. Called after a successful
+    /// parquet compaction has physically deleted those rows — the
+    /// tombstones have done their job and the journal can be cleared.
+    ///
+    /// Implementation: drop the per-dataset AppendLog from the cache and
+    /// delete all its batch files. Next write starts fresh.
+    pub async fn clear(&self, dataset: &str) -> Result<usize, String> {
+        let prefix = format!("{}/", Self::prefix_for(dataset));
+        let keys = storaged::ops::list(&self.store, Some(&prefix)).await?;
+        let matching: Vec<String> = keys
+            .into_iter()
+            .filter(|k| {
+                let basename = k.rsplit('/').next().unwrap_or(k);
+                basename.starts_with("batch_") && basename.ends_with(".jsonl")
+            })
+            .collect();
+        let count = matching.len();
+        for key in &matching {
+            let _ = storaged::ops::delete(&self.store, key).await;
+        }
+        self.logs.write().await.remove(dataset);
+        if count > 0 {
+            tracing::info!("cleared {count} tombstone batch files for '{}'", dataset);
+        }
+        Ok(count)
+    }
 }