Federation layer 2: header routing + cross-bucket SQL

Three pieces of the multi-bucket federation made real:

1. Catalog migration (POST /catalog/migrate-buckets)
   - One-shot normalizer for ObjectRef.bucket field
   - Empty -> "primary"; legacy "data"/"local" -> "primary"
   - Idempotent; re-running on canonical state is no-op
   - Ran on existing catalog: 12 refs renamed from "data", 2 already
     "primary", all 14 now canonical

2. X-Lakehouse-Bucket header middleware on ingest
   - resolve_bucket() helper extracts header, returns
     (bucket_name, store) or 404 with valid bucket list
   - ingest_file and ingest_db_stream now route writes per-request
   - Defaults to "primary" when header absent
   - pipeline::ingest_file_to_bucket records the actual bucket on the
     ObjectRef so catalog stays the source of truth for "where does this
     data live"
   - Verified: ingest with X-Lakehouse-Bucket: testing lands in
     data/_testing/, ingest without header lands in data/, bad header
     returns 404 with hint

3. queryd registers every bucket with DataFusion
   - QueryEngine now holds Arc<BucketRegistry> instead of single store
   - build_context iterates all buckets, registers each as a separate
     ObjectStore under URL scheme "lakehouse-{bucket}://"
   - ListingTable URLs include the per-object bucket scheme so
     DataFusion routes scans automatically based on ObjectRef.bucket
   - Profile bucket names like "profile:user" sanitized to
     "lakehouse-profile-user" since URL host segments can't contain ":"
   - Tolerant of duplicate manifest entries (pre-existing
     pipeline::ingest_file behavior creates a fresh dataset id per
     ingest); duplicates skipped with debug log
   - Backward compat: legacy "lakehouse://data/" URL still registered
     pointing at primary

Success gate: cross-bucket CROSS JOIN
  SELECT p.name, p.role, a.species
  FROM people_test p          (bucket: testing)
  CROSS JOIN animals a        (bucket: primary)
  LIMIT 5
returns rows correctly. DataFusion routed each scan to its bucket's
ObjectStore based on the URL scheme.

No regressions: SELECT COUNT(*) FROM candidates still returns 100000
from the primary bucket.

Deferred to Phase 17:
- POST /profile/{user}/activate (HNSW hot-load on profile switch)
- vectord storage paths becoming bucket-scoped (trial journals,
  eval sets per-profile)

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

crates/queryd/src/context.rs

@@ -5,6 +5,7 @@ use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::prelude::*;
 use object_store::ObjectStore;
 use std::sync::Arc;
+use storaged::registry::BucketRegistry;
 use url::Url;
 
 use crate::cache::MemCache;
@@ -12,27 +13,56 @@ use crate::delta;
 
 const STORE_SCHEME: &str = "lakehouse";
 
+/// URL scheme used to register a bucket's object store with DataFusion.
+/// Bucket names safely embed in URL host components after sanitizing
+/// `:` (used in `profile:user`) to `-`.
+fn bucket_scheme(bucket: &str) -> String {
+    format!("{STORE_SCHEME}-{}", bucket.replace(':', "-"))
+}
+
+/// Build a `ListingTableUrl`-shaped string for an object in a bucket.
+fn bucket_object_url(bucket: &str, key: &str) -> String {
+    format!("{}://data/{}", bucket_scheme(bucket), key)
+}
+
+/// Build the registration base URL for a bucket.
+fn bucket_base_url(bucket: &str) -> String {
+    format!("{}://data/", bucket_scheme(bucket))
+}
+
 /// Query engine with in-memory cache and delta merge support.
+/// Federation layer 2: holds a `BucketRegistry` so cross-bucket queries
+/// route reads to the right ObjectStore based on `ObjectRef.bucket`.
 #[derive(Clone)]
 pub struct QueryEngine {
     registry: Registry,
+    buckets: Arc<BucketRegistry>,
+    /// Primary bucket store cached for backward-compat callers (compact,
+    /// workspace ops) that haven't been migrated to bucket-aware paths.
     store: Arc<dyn ObjectStore>,
     cache: MemCache,
 }
 
 impl QueryEngine {
-    pub fn new(registry: Registry, store: Arc<dyn ObjectStore>, cache: MemCache) -> Self {
-        Self { registry, store, cache }
+    pub fn new(registry: Registry, buckets: Arc<BucketRegistry>, cache: MemCache) -> Self {
+        let store = buckets.default_store();
+        Self { registry, buckets, store, cache }
     }
 
     pub fn cache(&self) -> &MemCache {
         &self.cache
     }
 
+    /// Backward-compat: returns the primary bucket's store. Bucket-aware
+    /// callers should use `engine.buckets()` directly.
     pub fn store(&self) -> &Arc<dyn ObjectStore> {
         &self.store
     }
 
+    pub fn buckets(&self) -> &Arc<BucketRegistry> {
+        &self.buckets
+    }
+
     /// Execute a SQL query. Uses cache for hot data, falls back to Parquet.
     pub async fn query(&self, sql: &str) -> Result<Vec<arrow::array::RecordBatch>, String> {
         let ctx = self.build_context().await?;
@@ -43,11 +73,16 @@ impl QueryEngine {
 
     /// Pin a dataset into the memory cache.
     pub async fn pin_dataset(&self, name: &str) -> Result<(), String> {
-        // Read from Parquet
         let ctx = SessionContext::new();
-        let base_url = Url::parse(&format!("{STORE_SCHEME}://data/"))
-            .map_err(|e| format!("invalid store url: {e}"))?;
-        ctx.runtime_env().register_object_store(&base_url, self.store.clone());
+
+        // Register every bucket so a multi-bucket dataset can be pinned.
+        for info in self.buckets.list().await {
+            let url = Url::parse(&bucket_base_url(&info.name))
+                .map_err(|e| format!("invalid store url: {e}"))?;
+            let store = self.buckets.get(&info.name)
+                .map_err(|e| format!("registry inconsistency: {e}"))?;
+            ctx.runtime_env().register_object_store(&url, store);
+        }
 
         let dataset = self.registry.get_by_name(name).await
             .ok_or_else(|| format!("dataset not found: {name}"))?;
@@ -58,7 +93,10 @@ impl QueryEngine {
 
         let opts = ListingOptions::new(Arc::new(ParquetFormat::default()));
         let table_paths: Vec<ListingTableUrl> = dataset.objects.iter()
-            .filter_map(|o| ListingTableUrl::parse(&format!("{STORE_SCHEME}://data/{}", o.key)).ok())
+            .filter_map(|o| {
+                let bucket = if o.bucket.is_empty() { "primary" } else { &o.bucket };
+                ListingTableUrl::parse(&bucket_object_url(bucket, &o.key)).ok()
+            })
             .collect();
 
         let schema = opts.infer_schema(&ctx.state(), &table_paths[0]).await
@@ -79,9 +117,28 @@ impl QueryEngine {
     async fn build_context(&self) -> Result<SessionContext, String> {
         let ctx = SessionContext::new();
 
-        let base_url = Url::parse(&format!("{STORE_SCHEME}://data/"))
-            .map_err(|e| format!("invalid store url: {e}"))?;
-        ctx.runtime_env().register_object_store(&base_url, self.store.clone());
+        // Federation layer 2: register every configured bucket as its own
+        // DataFusion ObjectStore under a distinct URL scheme. Each
+        // dataset's ObjectRef.bucket determines which store DataFusion
+        // routes to at scan time.
+        let bucket_infos = self.buckets.list().await;
+        for info in &bucket_infos {
+            let url_str = bucket_base_url(&info.name);
+            let url = Url::parse(&url_str)
+                .map_err(|e| format!("invalid store url '{url_str}': {e}"))?;
+            // unknown-bucket here would be a config invariant violation,
+            // since we just listed them — propagate as an error to surface it.
+            let store = self.buckets.get(&info.name)
+                .map_err(|e| format!("registry inconsistency: {e}"))?;
+            ctx.runtime_env().register_object_store(&url, store);
+        }
+
+        // Backward-compat: also register the legacy `lakehouse://data/`
+        // URL pointing at primary, in case any code path still constructs
+        // ListingTableUrls without a bucket prefix.
+        let legacy_url = Url::parse(&format!("{STORE_SCHEME}://data/"))
+            .map_err(|e| format!("invalid legacy store url: {e}"))?;
+        ctx.runtime_env().register_object_store(&legacy_url, self.store.clone());
 
         let datasets = self.registry.list().await;
         for dataset in &datasets {
@@ -89,9 +146,9 @@ impl QueryEngine {
                 continue;
             }
 
-            // Check cache first
+            // Check cache first — cached batches are bucket-agnostic since
+            // they're already materialized in memory.
             if let Some(cached) = self.cache.get(&dataset.name).await {
-                // Load any delta files and merge
                 let delta_batches = delta::load_deltas(&self.store, &dataset.name).await.unwrap_or_default();
 
                 let mut all_batches = cached.batches;
@@ -106,10 +163,15 @@ impl QueryEngine {
                 continue;
             }
 
-            // Fall back to Parquet file
+            // Build ListingTable URLs that include the per-object bucket
+            // scheme. DataFusion routes each scan to the right store
+            // automatically based on the URL.
             let opts = ListingOptions::new(Arc::new(ParquetFormat::default()));
             let table_paths: Vec<ListingTableUrl> = dataset.objects.iter()
-                .filter_map(|o| ListingTableUrl::parse(&format!("{STORE_SCHEME}://data/{}", o.key)).ok())
+                .filter_map(|o| {
+                    let bucket = if o.bucket.is_empty() { "primary" } else { &o.bucket };
+                    ListingTableUrl::parse(&bucket_object_url(bucket, &o.key)).ok()
+                })
                 .collect();
 
             if table_paths.is_empty() {
@@ -124,8 +186,19 @@ impl QueryEngine {
             let table = ListingTable::try_new(config)
                 .map_err(|e| format!("table creation failed for {}: {e}", dataset.name))?;
 
-            ctx.register_table(&dataset.name, Arc::new(table))
-                .map_err(|e| format!("table registration failed for {}: {e}", dataset.name))?;
+            // Tolerate duplicate manifest entries for the same name —
+            // pre-existing pipeline::ingest_file behavior creates a fresh
+            // dataset id on every ingest. First registration wins; later
+            // ones are skipped with a warning rather than failing the
+            // whole context build.
+            if let Err(e) = ctx.register_table(&dataset.name, Arc::new(table)) {
+                let msg = e.to_string();
+                if msg.contains("already exists") {
+                    tracing::debug!("skip duplicate manifest registration: {}", dataset.name);
+                } else {
+                    return Err(format!("table registration failed for {}: {}", dataset.name, msg));
+                }
+            }
         }
 
         Ok(ctx)