lookup index validation: accept a type-scoped partial index on the hierarchy ancestor owning the key's columns

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

fixtures/database.json

@@ -1111,5 +1111,147 @@
         }
       }
     ]
+  },
+  {
+    "description": "Validation - type-scoped lookup key enforced by a partial index on the hierarchy ancestor owning the column",
+    "database": {
+      "indexes": [
+        {
+          "table": "entity",
+          "columns": ["name"],
+          "predicate": "(type = 'registry'::text)"
+        }
+      ],
+      "types": [
+        {
+          "id": "33333333-3333-3333-3333-333333333331",
+          "type": "type",
+          "name": "entity",
+          "module": "test",
+          "source": "test",
+          "hierarchy": ["entity"],
+          "variations": ["entity", "registry"],
+          "fields": ["id", "type", "name"],
+          "schemas": {
+            "entity": {
+              "type": "object",
+              "properties": {
+                "id": { "type": "string", "format": "uuid" },
+                "type": { "type": "string" },
+                "name": { "type": "string" }
+              }
+            }
+          }
+        },
+        {
+          "id": "33333333-3333-3333-3333-333333333332",
+          "type": "type",
+          "name": "registry",
+          "module": "test",
+          "source": "test",
+          "hierarchy": ["entity", "registry"],
+          "variations": ["registry"],
+          "fields": ["id", "type"],
+          "lookup_fields": ["type", "name"],
+          "schemas": {
+            "registry": {
+              "type": "entity",
+              "properties": {
+                "id": { "type": "string", "format": "uuid" }
+              }
+            }
+          }
+        }
+      ]
+    },
+    "tests": [
+      {
+        "description": "Compiles successfully: the key's type component is covered by the index's partial predicate",
+        "action": "compile",
+        "expect": {
+          "success": true,
+          "schemas": {
+            "entity": {},
+            "entity.filter": {},
+            "registry": {},
+            "registry.filter": {}
+          }
+        }
+      }
+    ]
+  },
+  {
+    "description": "Validation - type-scoped lookup key is NOT satisfied by an unscoped ancestor index",
+    "database": {
+      "indexes": [
+        {
+          "table": "entity",
+          "columns": ["name"]
+        }
+      ],
+      "types": [
+        {
+          "id": "44444444-4444-4444-4444-444444444441",
+          "type": "type",
+          "name": "entity",
+          "module": "test",
+          "source": "test",
+          "hierarchy": ["entity"],
+          "variations": ["entity", "registry"],
+          "fields": ["id", "type", "name"],
+          "schemas": {
+            "entity": {
+              "type": "object",
+              "properties": {
+                "id": { "type": "string", "format": "uuid" },
+                "type": { "type": "string" },
+                "name": { "type": "string" }
+              }
+            }
+          }
+        },
+        {
+          "id": "44444444-4444-4444-4444-444444444442",
+          "type": "type",
+          "name": "registry",
+          "module": "test",
+          "source": "test",
+          "hierarchy": ["entity", "registry"],
+          "variations": ["registry"],
+          "fields": ["id", "type"],
+          "lookup_fields": ["type", "name"],
+          "schemas": {
+            "registry": {
+              "type": "entity",
+              "properties": {
+                "id": { "type": "string", "format": "uuid" }
+              }
+            }
+          }
+        }
+      ]
+    },
+    "tests": [
+      {
+        "description": "Fails with MISSING_LOOKUP_INDEX: a full entity(name) index does not carry the key's type scoping",
+        "action": "compile",
+        "expect": {
+          "success": false,
+          "errors": [
+            {
+              "code": "MISSING_LOOKUP_INDEX",
+              "values": {
+                "type": "registry",
+                "lookup_fields": "[\"type\", \"name\"]"
+              },
+              "details": {
+                "path": "/types/registry",
+                "schema": "registry"
+              }
+            }
+          ]
+        }
+      }
+    ]
   }
 ]

src/database/mod.rs

@@ -35,6 +35,10 @@ use crate::drop::{Drop, Error, ErrorDetails};
 pub struct IndexInfo {
   pub table: String,
   pub columns: Vec<String>,
+  /// Partial-index WHERE clause as rendered by pg_get_expr(indpred, indrelid),
+  /// e.g. "(type = 'punc'::text)". Absent for full indexes.
+  #[serde(default)]
+  pub predicate: Option<String>,
 }
 
 #[derive(serde::Serialize)]
@@ -288,13 +292,29 @@ impl Database {
       }
     }
 
-    // Phase 5: Verify unique indexes for defined lookup fields
+    // Phase 5: Verify unique indexes for defined lookup fields. A key is
+    // enforced by exactly one unique index, in one of two physical forms:
+    //  1. On the declaring table over exactly the declared columns.
+    //  2. On the hierarchy table that owns the key's columns (e.g. the entity
+    //     base's `name`), with the key's `type` component expressed as the
+    //     index's partial predicate (`WHERE type = '<declaring type>'`) —
+    //     uniqueness scoped to the declaring type, free for every other type
+    //     sharing the ancestor column.
     for (_, type_def) in &self.types {
       if !type_def.lookup_fields.is_empty() {
+        let has_type_component = type_def.lookup_fields.iter().any(|c| c == "type");
+        let column_fields: Vec<&String> = type_def
+          .lookup_fields
+          .iter()
+          .filter(|c| c.as_str() != "type")
+          .collect();
         let mut index_found = false;
         for index in &self.indexes {
-          if index.table == type_def.name {
+          // Form 1: own-table, exact columns, no partial predicate (a partial
-            if index.columns.len() == type_def.lookup_fields.len()
+          // index enforces less than the key claims).
+          if index.table == type_def.name
+            && index.predicate.is_none()
+            && index.columns.len() == type_def.lookup_fields.len()
             && index
               .columns
               .iter()
@@ -303,6 +323,29 @@ impl Database {
             index_found = true;
             break;
           }
+          // Form 2: hierarchy table owning the non-type columns; a `type`
+          // component must be covered by the partial predicate, and a key
+          // without one must not be narrowed by any predicate.
+          if type_def.hierarchy.contains(&index.table)
+            && index.columns.len() == column_fields.len()
+            && index
+              .columns
+              .iter()
+              .all(|c| column_fields.iter().any(|f| f.as_str() == c))
+          {
+            let predicate_ok = if has_type_component {
+              index
+                .predicate
+                .as_deref()
+                .map(|p| Self::predicate_scopes_type(p, &type_def.name))
+                .unwrap_or(false)
+            } else {
+              index.predicate.is_none()
+            };
+            if predicate_ok {
+              index_found = true;
+              break;
+            }
           }
         }
         if !index_found {
@@ -326,6 +369,18 @@ impl Database {
     }
   }
 
+  /// Whether a partial-index predicate scopes rows to exactly the given type,
+  /// i.e. it is `type = '<type_name>'` modulo pg_get_expr rendering noise
+  /// (parentheses, whitespace, identifier quoting, ::text casts).
+  fn predicate_scopes_type(predicate: &str, type_name: &str) -> bool {
+    let normalized: String = predicate
+      .chars()
+      .filter(|c| !c.is_whitespace() && *c != '(' && *c != ')' && *c != '"')
+      .collect::<String>()
+      .replace("::text", "");
+    normalized == format!("type='{}'", type_name)
+  }
+
   /// Synthesizes Composed Filter References for all table-backed boundaries.
   fn compile_filters(&mut self, errors: &mut Vec<crate::drop::Error>) -> Vec<(String, String)> {
     let mut filter_schemas = Vec::new();

src/tests/fixtures.rs

@@ -3431,6 +3431,18 @@ fn test_database_8_0() {
     crate::tests::runner::run_test_case(&path, 8, 0).unwrap();
 }
 
+#[test]
+fn test_database_9_0() {
+    let path = format!("{}/fixtures/database.json", env!("CARGO_MANIFEST_DIR"));
+    crate::tests::runner::run_test_case(&path, 9, 0).unwrap();
+}
+
+#[test]
+fn test_database_10_0() {
+    let path = format!("{}/fixtures/database.json", env!("CARGO_MANIFEST_DIR"));
+    crate::tests::runner::run_test_case(&path, 10, 0).unwrap();
+}
+
 #[test]
 fn test_cases_0_0() {
     let path = format!("{}/fixtures/cases.json", env!("CARGO_MANIFEST_DIR"));