@@ -1,7 +1,7 @@
 [package]
 name = "sqd-worker"
 license = "AGPL-3.0-or-later"
-version = "2.15.0"
+version = "2.16.0-rc.1"
 edition = "2021"
 
 [features]
@@ -73,7 +73,7 @@ sqd-network-transport = { git = "https://github.com/subsquid/sqd-network.git", r
 
 sqd-query = { git = "https://github.com/subsquid/data.git", rev = "aa62e79", features = ["parquet"] }
 sqd-polars = { git = "https://github.com/subsquid/data.git", rev = "aa62e79" }
-sqd-query-engine = { git = "https://github.com/subsquid/sqd-query-engine.git", rev = "54b9020" }
+sqd-query-engine = { git = "https://github.com/subsquid/sqd-query-engine.git", rev = "6df565a" }
 
 sql_query_plan = {git = "https://github.com/subsquid/qplan.git", rev = "658f88f" }
 
 
@@ -224,16 +224,22 @@ mod tests {
 name: evm
 tables:
   blocks:
+    output:
+      name: block
+      fields: [number]
     block_number_column: number
-    field_name: block
     sort_key: [number]
     columns:
       number:
         type: uint64
   logs:
+    request:
+      name: logs
+      filters: []
+    output:
+      name: log
+      fields: [transaction_index, log_index, data]
     block_number_column: block_number
-    query_name: logs
-    field_name: log
     item_order_keys: [transaction_index, log_index]
     sort_key: [block_number, transaction_index, log_index]
     columns:
@@ -245,7 +251,7 @@ tables:
         type: uint32
       data:
         type: string
-        json_encoding: hex
+        encoding: hex_bytes
         weight: data_size
       data_size:
         type: uint64
@@ -473,14 +479,17 @@ tables:
 name: evm
 tables:
   blocks:
+    output:
+      name: block
+      fields: [number, hash]
     block_number_column: number
     sort_key: [number]
     columns:
       number:
         type: uint64
-        stats: true
       hash:
         type: string
+        encoding: hex_bytes
 "#;
         let app = axum::Router::new()
             .route(
 
@@ -719,6 +719,7 @@ async fn load_dir(
     joined(
         tokio::task::spawn_blocking(move || {
             let mut schemas = HashMap::new();
+            let mut skipped = Vec::new();
             // Reading back what was just staged is disk; what the yaml says is the bundle.
             for entry in std::fs::read_dir(&dir).map_err(BundleFault::transient)? {
                 let path = entry.map_err(BundleFault::transient)?.path();
@@ -732,18 +733,31 @@ async fn load_dir(
                 let yaml = std::fs::read_to_string(&path)
                     .with_context(|| format!("couldn't read schema {}", path.display()))
                     .map_err(BundleFault::transient)?;
-                let description = sqd_query_engine::metadata::parse_dataset_description(&yaml)
-                    .map_err(|e| {
-                        BundleFault::permanent(anyhow::anyhow!(
-                            "couldn't parse schema {}: {e:?}",
-                            path.display()
-                        ))
-                    })?;
-                schemas.insert(id, Arc::new(description));
+                // A schema is load-bearing only for the chunks that name it, and the assignment
+                // check downstream already refuses those. Failing the bundle instead would let
+                // one unreadable schema — a kind this worker holds no chunk of — stop it serving
+                // the datasets it does hold.
+                match sqd_query_engine::metadata::parse_dataset_description(&yaml) {
+                    Ok(description) => {
+                        schemas.insert(id, Arc::new(description));
+                    }
+                    Err(e) => {
+                        metrics::SCHEMA_BUNDLE_SCHEMAS_SKIPPED.inc();
+                        tracing::warn!(
+                            %id,
+                            error = ?e,
+                            "Dropping a schema the bundle carries but this engine can't parse",
+                        );
+                        skipped.push(format!("{id}: {e:?}"));
+                    }
+                }
             }
             if schemas.is_empty() {
+                // `unpack` has already refused a bundle with no `<id>.yaml` entries, so there
+                // were some and every one of them failed: nothing here can serve a query.
                 return Err(BundleFault::permanent(anyhow::anyhow!(
-                    "no schemas found in {dir}"
+                    "no schema in {dir} could be parsed: {}",
+                    skipped.join("; ")
                 )));
             }
             Ok(schemas)
@@ -791,6 +805,9 @@ pub(crate) mod test_support {
 name: evm
 tables:
   blocks:
+    output:
+      name: block
+      fields: [number]
     block_number_column: number
     sort_key: [number]
     columns:
@@ -934,6 +951,65 @@ mod tests {
         assert!(format!("{fault:#}").contains("zstd"), "{fault:#}");
     }
 
+    /// A stale catalog for a kind this worker holds no chunk of used to stop it serving the
+    /// kinds it does hold: one parse fault was a verdict on the whole bundle.
+    #[tokio::test]
+    async fn one_unparseable_schema_leaves_the_rest_of_the_bundle_usable() {
+        let dir = tempfile::tempdir().unwrap();
+        let registry = store(&dir);
+        let solana = SCHEMA.replace("name: evm", "name: solana");
+        let archive = targz(&[
+            ("7.yaml", SCHEMA.as_bytes()),
+            (
+                "9.yaml",
+                b"name: fuel\ntables:\n  blocks:\n    field_name: block\n",
+            ),
+            ("12.yaml", solana.as_bytes()),
+        ]);
+        let bundle = served_bundle(archive).await;
+
+        registry
+            .ensure(&bundle, &reqwest::Client::new())
+            .await
+            .unwrap();
+
+        assert_eq!(registry.get_by_id(SchemaId::new(7)).unwrap().name, "evm");
+        assert_eq!(
+            registry.get_by_id(SchemaId::new(12)).unwrap().name,
+            "solana"
+        );
+        assert!(registry.get_by_id(SchemaId::new(9)).is_err());
+
+        // The id the bundle carries but cannot serve is not one it offers, so an assignment
+        // whose chunks need it is refused rather than answered from a schema that never loaded.
+        assert_eq!(
+            *registry.bundle_ids(),
+            HashSet::from([SchemaId::new(7), SchemaId::new(12)]),
+        );
+        assert_eq!(stored(&dir), vec!["12.yaml", "7.yaml"]);
+        assert_eq!(registry.installed_hash(), Some(bundle.hash));
+    }
+
+    #[tokio::test]
+    async fn a_bundle_no_schema_of_which_parses_is_still_a_permanent_fault() {
+        let dir = tempfile::tempdir().unwrap();
+        let registry = store(&dir);
+        let archive = targz(&[(
+            "9.yaml",
+            b"name: fuel\ntables:\n  blocks:\n    field_name: block\n",
+        )]);
+        let bundle = served_bundle(archive).await;
+
+        let fault = registry
+            .prepare_bundle(&bundle, &reqwest::Client::new())
+            .await
+            .unwrap_err();
+
+        assert!(fault.is_permanent(), "{fault:#?}");
+        assert!(format!("{:#}", fault.into_error()).contains("field_name"));
+        assert!(stored(&dir).is_empty());
+    }
+
     #[tokio::test]
     async fn preparing_a_bundle_does_not_make_it_active() {
         let dir = tempfile::tempdir().unwrap();
 
@@ -69,6 +69,7 @@ lazy_static::lazy_static! {
     pub static ref SCHEMA_BUNDLE_LOADED: Gauge = Default::default();
     pub static ref SCHEMA_BUNDLE_FAILURES: Counter = Default::default();
     pub static ref SCHEMA_BUNDLE_MISMATCHES: Counter = Default::default();
+    pub static ref SCHEMA_BUNDLE_SCHEMAS_SKIPPED: Counter = Default::default();
 
     static ref QUERY_EXECUTED: Family<QueryExecutedLabels, Counter> = Default::default();
     static ref QUERY_RESULT_SIZE: Histogram = Histogram::new(std::iter::empty());
@@ -193,6 +194,11 @@ pub fn register_metrics(registry: &mut Registry, version: String) {
         "Number of times a schema bundle failed to install",
         SCHEMA_BUNDLE_FAILURES.clone(),
     );
+    registry.register(
+        "schema_bundle_schemas_skipped",
+        "Number of schemas dropped from a bundle because they could not be parsed",
+        SCHEMA_BUNDLE_SCHEMAS_SKIPPED.clone(),
+    );
     registry.register(
         "schema_bundle_mismatches",
         "Number of pairs the scheduler published that do not hold together",
 
@@ -17,16 +17,22 @@ pub const SCHEMA_YAML: &str = r#"
 name: evm
 tables:
   blocks:
+    output:
+      name: block
+      fields: [number]
     block_number_column: number
-    field_name: block
     sort_key: [number]
     columns:
       number:
         type: uint64
   logs:
+    request:
+      name: logs
+      filters: []
+    output:
+      name: log
+      fields: [transaction_index, log_index, data]
     block_number_column: block_number
-    query_name: logs
-    field_name: log
     item_order_keys: [transaction_index, log_index]
     sort_key: [block_number, transaction_index, log_index]
     columns:
@@ -38,7 +44,7 @@ tables:
         type: uint32
       data:
         type: string
-        json_encoding: hex
+        encoding: hex_bytes
         weight: data_size
       data_size:
         type: uint64