fix(validation): report empty names/refs and survive nested input

validation/test_validate.py

@@ -420,6 +420,38 @@ def test_root_dialects_and_vendors_are_rejected(self):
         self.assertEqual(result.returncode, 1)
         self.assertIn("'dialects', 'vendors' were unexpected", result.stdout)
 
+    def test_deeply_nested_yaml_fails_cleanly(self):
+        # Deeply nested flow collections exhaust the recursion limit during
+        # composition, surfacing as RecursionError rather than YAMLError. The
+        # validator must exit with a diagnostic, never a raw traceback.
+        result = self.run_validator("[" * 3000 + "]" * 3000 + "\n")
+
+        self.assertEqual(result.returncode, 1)
+        self.assertNotIn("Traceback", result.stderr)
+        self.assertIn("too deeply nested", result.stdout)
+
+    def test_empty_relationship_endpoint_is_reported(self):
+        # "" is a schema-valid string but names no dataset; the reference check
+        # must report it end-to-end instead of silently passing the model.
+        result = self.run_validator(
+            "version: 0.2.0.dev0\n"
+            "name: sales\n"
+            "datasets:\n"
+            "  - name: orders\n"
+            "    source: analytics.orders\n"
+            "relationships:\n"
+            "  - name: orders_to_missing\n"
+            "    from: orders\n"
+            "    to: ''\n"
+            "    from_columns: [customer_id]\n"
+            "    to_columns: [id]\n"
+        )
+
+        self.assertEqual(result.returncode, 1)
+        self.assertNotIn("Traceback", result.stderr)
+        self.assertIn("Validation FAILED", result.stdout)
+        self.assertIn("references unknown dataset ''", result.stdout)
+
 
 if __name__ == "__main__":
     unittest.main()

validation/tests/test_validate.py

@@ -166,6 +166,22 @@ def test_unique_names_are_checked_in_the_root_model() -> None:
     assert errors == ["[Unique] Duplicate dataset name 'orders' in model 'm'"]
 
 
+def test_duplicate_empty_dataset_names_are_reported() -> None:
+    # The schema accepts an empty string as a name, so two empty names must not
+    # slip past duplicate detection via a truthiness filter.
+    empty = {"name": "", "source": "db.s.empty"}
+    errors = _VALIDATE.validate_unique_names(_document([empty, empty], []))
+
+    assert errors == ["[Unique] Duplicate dataset name '' in model 'm'"]
+
+
+def test_missing_names_are_not_treated_as_duplicates() -> None:
+    # A missing name (None) is a schema violation reported elsewhere; the
+    # uniqueness check skips it rather than flagging spurious duplicate None.
+    nameless = {"source": "db.s.nameless"}
+    assert _VALIDATE.validate_unique_names(_document([nameless, nameless], [])) == []
+
+
 def test_sql_checks_traverse_root_fields_and_metrics(monkeypatch: pytest.MonkeyPatch) -> None:
     seen = []
 
@@ -261,6 +277,40 @@ def test_still_reports_unknown_datasets() -> None:
     ]
 
 
+@pytest.mark.parametrize("endpoint", ["from", "to"])
+def test_empty_endpoint_is_reported_as_unknown_dataset(endpoint: str) -> None:
+    # "" is a schema-valid string but never names a real dataset; a truthiness
+    # guard would skip it, so the empty endpoint must be reported, not ignored.
+    rel = _relationship(to_columns=["id"])
+    rel[endpoint] = ""
+    errors = validate_references(_document([_ORDERS, _CUSTOMERS], [rel]))
+
+    assert errors == [
+        "[Reference] Relationship 'orders_to_customers' in model 'm' references unknown dataset ''"
+    ]
+
+
+def test_missing_endpoints_are_skipped() -> None:
+    # A missing from/to (None) is a schema violation reported elsewhere; the
+    # reference check must not invent a reference to the string 'None'.
+    rel = _relationship(to_columns=["id"])
+    del rel["from"]
+    del rel["to"]
+
+    assert validate_references(_document([_ORDERS, _CUSTOMERS], [rel])) == []
+
+
+@pytest.mark.skipif(not _VALIDATE.SQLGLOT_AVAILABLE, reason="sqlglot is not installed")
+def test_deeply_nested_sql_reports_a_diagnostic_instead_of_crashing() -> None:
+    # Pathologically nested SQL exhausts sqlglot's recursion limit; the
+    # validator must turn that into a diagnostic rather than propagating
+    # RecursionError and aborting the run.
+    expression = "(" * 5000 + "1" + ")" * 5000
+    result = _VALIDATE.validate_sql_expression(expression, "ANSI_SQL", "ctx")
+
+    assert result == "[SQL] ctx: expression is too deeply nested to parse"
+
+
 def test_tolerates_null_unique_keys() -> None:
     # `unique_keys:` present but empty parses to None; the check must not crash.
     dataset = {"name": "customers", "source": "db.s.customers",

validation/validate.py

@@ -176,25 +176,28 @@ def validate_unique_names(data: dict) -> list[str]:
 
     model_name = model.get("name", "<unnamed>")
 
+    # Guard on "is not None", not truthiness: "" is a schema-valid name (no
+    # minLength) and must not slip past duplicate detection.
+
     # Check unique dataset names
-    dataset_names = [d.get("name") for d in model.get("datasets", []) if d.get("name")]
+    dataset_names = [d.get("name") for d in model.get("datasets", []) if d.get("name") is not None]
     for dup in find_duplicates(dataset_names):
         errors.append(f"[Unique] Duplicate dataset name '{dup}' in model '{model_name}'")
 
     # Check unique field names within each dataset
     for dataset in model.get("datasets", []):
         dataset_name = dataset.get("name", "<unnamed>")
-        field_names = [f.get("name") for f in dataset.get("fields", []) if f.get("name")]
+        field_names = [f.get("name") for f in dataset.get("fields", []) if f.get("name") is not None]
         for dup in find_duplicates(field_names):
             errors.append(f"[Unique] Duplicate field name '{dup}' in dataset '{dataset_name}'")
 
     # Check unique metric names
-    metric_names = [m.get("name") for m in model.get("metrics", []) if m.get("name")]
+    metric_names = [m.get("name") for m in model.get("metrics", []) if m.get("name") is not None]
     for dup in find_duplicates(metric_names):
         errors.append(f"[Unique] Duplicate metric name '{dup}' in model '{model_name}'")
 
     # Check unique relationship names
-    rel_names = [r.get("name") for r in model.get("relationships", []) if r.get("name")]
+    rel_names = [r.get("name") for r in model.get("relationships", []) if r.get("name") is not None]
     for dup in find_duplicates(rel_names):
         errors.append(f"[Unique] Duplicate relationship name '{dup}' in model '{model_name}'")
 
@@ -211,16 +214,20 @@ def validate_references(data: dict) -> list[str]:
     errors = []
 
     model_name = model.get("name", "<unnamed>")
+    # Exclude falsy dataset names so a relationship pointing at "" is reported
+    # as unknown rather than matched.
     datasets = {d.get("name"): d for d in model.get("datasets", []) if d.get("name")}
 
     for rel in model.get("relationships", []):
         rel_name = rel.get("name", "<unnamed>")
         from_ds = rel.get("from")
         to_ds = rel.get("to")
 
-        if from_ds and from_ds not in datasets:
+        # "is not None", not truthiness: "" is a declared-but-invalid reference
+        # that must be reported, not skipped.
+        if from_ds is not None and from_ds not in datasets:
             errors.append(f"[Reference] Relationship '{rel_name}' in model '{model_name}' references unknown dataset '{from_ds}'")
-        if to_ds and to_ds not in datasets:
+        if to_ds is not None and to_ds not in datasets:
             errors.append(f"[Reference] Relationship '{rel_name}' in model '{model_name}' references unknown dataset '{to_ds}'")
 
         # The spec defines to_columns as "Primary/unique key columns in the
@@ -292,7 +299,9 @@ def validate_sql_expression(expr: str, dialect: str, context: str) -> str | None
         # Try parsing as expression first (for field expressions like "column_name")
         sqlglot.parse_one(expr, dialect=sqlglot_dialect)
         return None
-    except (ParseError, TokenError):
+    except Exception:
+        # Any failure (a parse error, or a RecursionError on deeply nested
+        # input) falls through to the SELECT-wrapped retry below.
         pass
 
     try:
@@ -301,6 +310,11 @@ def validate_sql_expression(expr: str, dialect: str, context: str) -> str | None
         return None
     except (ParseError, TokenError) as e:
         return f"[SQL] {context}: {str(e).split(chr(10))[0]}"
+    except RecursionError:
+        # Deeply nested input exhausts the recursion limit, not a parser error.
+        return f"[SQL] {context}: expression is too deeply nested to parse"
+    except Exception as e:
+        return f"[SQL] {context}: {str(e).split(chr(10))[0] or type(e).__name__}"
 
 
 def validate_sql(data: dict) -> list[str]:
@@ -384,6 +398,10 @@ def main():
         except yaml.YAMLError as e:
             print(f"Error: Invalid YAML: {e}")
             sys.exit(1)
+        except RecursionError:
+            # Deeply nested input surfaces as RecursionError, not YAMLError.
+            print("Error: Invalid YAML: input is too deeply nested to parse")
+            sys.exit(1)
 
     # Run validations
     errors = []