From 423d5aa97e07b3f4c39f270334f21d77c81c41f9 Mon Sep 17 00:00:00 2001 From: David Mautz Date: Wed, 22 Jul 2026 18:21:25 -0500 Subject: [PATCH 1/4] feat(models): port production UMF model extensions from pulseflow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the model-level deltas that accumulated in the pulseflow fork of tablespec back upstream, adapted to this repo's structure. These fields power upsert merge semantics, SQL plan generation, and cross-pipeline references in production healthcare pipelines. New capabilities: - MergeCondition + UMFColumn.merge_strategy/merge_source/merge_condition: column-level upsert merge semantics (LEAST/GREATEST/COALESCE, gated by a source-row condition) - UMFColumn.internal: helper columns that participate in derivation but are excluded from final output/DDL/schemas - UMFColumn.nullable now accepts plain booleans alongside Nullable contexts (is_nullable_for_all_contexts already handled bool) - TableReference (models/pipeline.py) + parse_table_reference() on ForeignKey, OutgoingRelationship, IncomingRelationship, and DerivationCandidate for pipeline-qualified references - ForeignKey.join_filter and OutgoingRelationship.alternative_joins (OR-join paths) for SQL plan generation - DerivationCandidate.union_value: per-UNION-branch literal values - UMFMetadata: base_table_filter, base_join_column, final_filter, union_base_tables, union_type, union_exclude_base, union_coalesce_base, final_dedup — base-view and final-assembly controls for generated tables - IngestionConfig.update_mode ('upsert' | 'update_only') for _U file merge behavior; order_by docs extended to cover data columns and meta_source_offset tie-breaking - UMF.effective_primary_key / DEFAULT_PRIMARY_KEY (meta_checksum) as the default merge key - ValidationRules warns when ingested-stage expectation types are placed in raw validation_rules (shift-left guidance) - Domain type registry is now cached at module level instead of being re-constructed on every column validation Deliberately not ported (upstream direction wins): PySpark-style data_type validation, bool→LOB nullable normalization, lowercase table_name constraint, and advisory no-primary-key warnings (would break filterwarnings=error test policy). Co-Authored-By: Claude Fable 5 --- src/tablespec/__init__.py | 4 + src/tablespec/models/__init__.py | 12 ++ src/tablespec/models/pipeline.py | 62 +++++- src/tablespec/models/umf.py | 252 ++++++++++++++++++++++- tests/unit/test_umf_models.py | 333 +++++++++++++++++++++++++++++++ tests/unit/test_validator.py | 6 + 6 files changed, 661 insertions(+), 8 deletions(-) diff --git a/src/tablespec/__init__.py b/src/tablespec/__init__.py index d45b3972..bc4cb103 100644 --- a/src/tablespec/__init__.py +++ b/src/tablespec/__init__.py @@ -18,6 +18,7 @@ IngestionConfig, IngestionExclusionRule, JoinViaSpec, + MergeCondition, Nullable, OutgoingRelationship, OutputConfig, @@ -28,6 +29,7 @@ RelationshipSummary, Relationships, Survivorship, + TableReference, UMFColumn, UMFColumnDerivation, UMFMetadata, @@ -147,6 +149,7 @@ "IngestionConfig", "IngestionExclusionRule", "JoinViaSpec", + "MergeCondition", "Nullable", "OutgoingRelationship", "OutputConfig", @@ -157,6 +160,7 @@ "RelationshipSummary", "Relationships", "Survivorship", + "TableReference", "UMFColumnDerivation", "ValidationRule", "ValidationRules", diff --git a/src/tablespec/models/__init__.py b/src/tablespec/models/__init__.py index 595c7d63..1d50152e 100644 --- a/src/tablespec/models/__init__.py +++ b/src/tablespec/models/__init__.py @@ -1,6 +1,12 @@ """UMF data models for tablespec.""" +from tablespec.models.pipeline import ( + PipelineDependency, + PipelineMetadata, + TableReference, +) from tablespec.models.umf import ( + DEFAULT_PRIMARY_KEY, INGESTED_QUALITY_CHECK_TYPES, RAW_VALIDATION_TYPES, REDUNDANT_VALIDATION_TYPES, @@ -23,6 +29,7 @@ JoinViaSpec, JsonProjection, JsonSource, + MergeCondition, Nullable, OutgoingRelationship, OutputConfig, @@ -46,6 +53,7 @@ ) __all__ = [ + "DEFAULT_PRIMARY_KEY", "INGESTED_QUALITY_CHECK_TYPES", "RAW_VALIDATION_TYPES", "REDUNDANT_VALIDATION_TYPES", @@ -68,10 +76,13 @@ "JoinViaSpec", "JsonProjection", "JsonSource", + "MergeCondition", "Nullable", "OutgoingRelationship", "OutputConfig", "ParquetSource", + "PipelineDependency", + "PipelineMetadata", "PostUpsertRule", "QualityCheck", "QualityChecks", @@ -80,6 +91,7 @@ "Relationships", "SourceSpec", "Survivorship", + "TableReference", "UMFColumn", "UMFColumnDerivation", "UMFMetadata", diff --git a/src/tablespec/models/pipeline.py b/src/tablespec/models/pipeline.py index 1f76f854..f74e04a5 100644 --- a/src/tablespec/models/pipeline.py +++ b/src/tablespec/models/pipeline.py @@ -36,4 +36,64 @@ class PipelineMetadata(BaseModel): ) -__all__ = ["PipelineDependency", "PipelineMetadata"] +class TableReference(BaseModel): + """Parsed table reference with optional pipeline qualification. + + Supports two formats: + - Bare: "table_name" (references table in same pipeline) + - Qualified: "pipeline_name.table_name" (references table in another pipeline) + """ + + pipeline: str | None = Field( + default=None, + description="Pipeline name (None = same pipeline as referencing table)", + ) + table: str = Field(description="Table name within pipeline") + + @classmethod + def parse(cls, ref: str) -> "TableReference": + """Parse table reference string. + + Args: + ref: Table reference string ("table" or "pipeline.table") + + Returns: + Parsed TableReference + + Examples: + >>> TableReference.parse("outreach_list") + TableReference(pipeline=None, table='outreach_list') + >>> TableReference.parse("reference_data.icd_codes") + TableReference(pipeline='reference_data', table='icd_codes') + """ + if "." in ref: + parts = ref.split(".", 1) + return cls(pipeline=parts[0], table=parts[1]) + return cls(pipeline=None, table=ref) + + def is_external(self) -> bool: + """Check if this references an external pipeline.""" + return self.pipeline is not None + + def to_qualified_name(self, current_pipeline: str | None = None) -> str: + """Convert to qualified name string. + + Args: + current_pipeline: Optional current pipeline name (for resolving bare refs) + + Returns: + Qualified name ("pipeline.table") when a pipeline is known, else bare name + """ + target_pipeline = self.pipeline or current_pipeline + if target_pipeline: + return f"{target_pipeline}.{self.table}" + return self.table + + def __str__(self) -> str: + """String representation (qualified or bare).""" + if self.pipeline: + return f"{self.pipeline}.{self.table}" + return self.table + + +__all__ = ["PipelineDependency", "PipelineMetadata", "TableReference"] diff --git a/src/tablespec/models/umf.py b/src/tablespec/models/umf.py index e77183ee..c7a2e5c6 100644 --- a/src/tablespec/models/umf.py +++ b/src/tablespec/models/umf.py @@ -8,7 +8,7 @@ import warnings from datetime import datetime from pathlib import Path -from typing import Annotated, Any, Literal, Self +from typing import TYPE_CHECKING, Annotated, Any, Literal, Self from pydantic import ( BaseModel, @@ -115,6 +115,35 @@ def classify_validation_type( return "unknown" +# Default merge key for upsert operations when no explicit primary_key is set. +# meta_checksum is the row-level content hash added during ingestion, so it +# provides row-level deduplication semantics. +DEFAULT_PRIMARY_KEY: list[str] = ["meta_checksum"] + +# Cached domain type registry for validation (lazy loaded) +_domain_type_registry_cache: "DomainTypeRegistry | None" = None + + +def _get_domain_type_registry() -> "DomainTypeRegistry": + """Get cached DomainTypeRegistry instance for validation. + + Lazy loads the registry on first use so importing the UMF models does not + pay the registry construction cost, and repeated column validation does + not re-parse domain_types.yaml. + """ + global _domain_type_registry_cache # noqa: PLW0603 + if _domain_type_registry_cache is None: + from tablespec.inference.domain_types import DomainTypeRegistry + + _domain_type_registry_cache = DomainTypeRegistry() + return _domain_type_registry_cache + + +if TYPE_CHECKING: + from tablespec.inference.domain_types import DomainTypeRegistry + from tablespec.models.pipeline import TableReference + + class FilenamePattern(BaseModel): """Filename pattern for extracting metadata from filenames.""" @@ -470,6 +499,13 @@ class DerivationCandidate(BaseModel): "meta_source_name, loinc code, etc. " "E.g., ['result', 'meta_source_name', 'meta_snapshot_dt'].", ) + union_value: str | int | float | bool | None = Field( + default=None, + description="Literal value to emit for this column in the UNION part " + "corresponding to this table. Used for synthetic columns that don't exist " + "in source tables but need different values per UNION branch " + "(e.g., a flag that is false for the base table, true for the union table).", + ) @model_validator(mode="after") def validate_column_or_expression(self) -> "DerivationCandidate": @@ -479,6 +515,16 @@ def validate_column_or_expression(self) -> "DerivationCandidate": raise ValueError(msg) return self + def parse_table_reference(self) -> "TableReference": + """Parse table field into TableReference with optional pipeline qualification. + + Returns: + TableReference with pipeline and table components + """ + from tablespec.models.pipeline import TableReference + + return TableReference.parse(self.table) + class Survivorship(BaseModel): """Survivorship strategy for multi-source column derivation.""" @@ -503,6 +549,21 @@ class Survivorship(BaseModel): ) +class MergeCondition(BaseModel): + """Condition that gates when a column-level merge strategy applies. + + When the condition is met (source column value is in the specified values), + the merge strategy is applied. When not met, the target value is preserved. + """ + + column: str = Field( + description="Column to check in source data (e.g., 'load_mode')" + ) + values: list[str] = Field( + description="Values that activate the merge strategy (e.g., ['I', 'A'])" + ) + + class UMFColumnDerivation(BaseModel): """Column derivation metadata for multi-source survivorship. @@ -586,7 +647,12 @@ class UMFColumn(BaseModel): default=None, description="Excel column position or identifier" ) description: str | None = Field(default=None, description="Column description") - nullable: Nullable | None = Field(default=None, description="Nullability by LOB") + nullable: bool | Nullable | None = Field( + default=None, + description="Nullability specification. Can be a boolean (true/false) for " + "simple cases, or a Nullable mapping with per-context values " + "(e.g., LOB keys for healthcare pipelines).", + ) sample_values: list[str] | None = Field( default=None, description="Sample values for the column" ) @@ -638,6 +704,15 @@ class UMFColumn(BaseModel): "Use for columns whose values change every run (e.g., rundate) but should not " "trigger a record to be treated as 'changed'.", ) + internal: bool = Field( + default=False, + description="When True, this is an internal helper column: it still participates " + "in SQL aggregation/derivation (its aggregation view is built and joined so it " + "can be referenced by other derivations), but it is EXCLUDED from the final " + "output table, DDL, PySpark/JSON schema, GX baseline, and final schema " + "validation. Use for winning-row anchors/bundles that should not appear in " + "the result.", + ) reporting_requirement: str | None = Field( default=None, description="Reporting requirement classification: 'R' (Required), 'O' (Optional), 'S' (Suggested)", @@ -733,6 +808,32 @@ class UMFColumn(BaseModel): "strings are converted to SQL NULL. Use this for columns where 'NULL' " "is a valid data value (e.g., someone's actual last name is 'NULL').", ) + merge_strategy: ( + Literal["replace", "keep_minimum", "keep_maximum", "keep_existing"] | None + ) = Field( + default=None, + description="Column-level merge strategy during upsert. " + "Default (None/'replace') overwrites with source value. " + "'keep_minimum' uses LEAST(target, source) to retain the smaller value. " + "'keep_maximum' uses GREATEST(target, source) to retain the larger value. " + "'keep_existing' uses COALESCE(target, source) to only fill if target is NULL.", + ) + merge_source: str | None = Field( + default=None, + description="Column name to initialize this column from when it is not present " + "in source data. Used with merge_strategy to seed derived tracking columns " + "from existing data columns (e.g., a load-date column seeded from a " + "file-date column on first insert).", + ) + merge_condition: MergeCondition | None = Field( + default=None, + description="Condition that gates when merge_strategy applies during upsert " + "updates. When the source row meets the condition, the merge strategy is " + "applied. When not met, the target (existing) value is preserved. " + "Only affects UPDATE path — INSERT always uses source values. " + "Example: merge_condition with column='load_mode', values=['I','A'] " + "only applies the strategy for those load modes.", + ) @field_validator("length") @classmethod @@ -813,9 +914,7 @@ def validate_domain_type_compatibility(self) -> Self: return self try: - from tablespec.inference.domain_types import DomainTypeRegistry - - registry = DomainTypeRegistry() + registry = _get_domain_type_registry() except (ImportError, FileNotFoundError): # Registry not available - skip validation return self @@ -890,6 +989,28 @@ class ValidationRules(BaseModel): default=None, description="Expectations pending implementation" ) + @model_validator(mode="after") + def warn_misclassified_expectations(self) -> Self: + """Warn if expectations contain ingested-stage types that belong in quality_checks.""" + if self.expectations is None: + return self + + misclassified = [ + exp_type + for exp in self.expectations + if (exp_type := exp.get("type", "")) in INGESTED_QUALITY_CHECK_TYPES + ] + + if misclassified: + message = ( + "ValidationRules contains ingested-stage expectations that should " + f"be in quality_checks: {misclassified}" + ) + warnings.warn(message, UserWarning, stacklevel=2) + logger.debug(message) + + return self + class QualityCheck(BaseModel): """Single data quality check executed on typed/ingested data. @@ -1008,6 +1129,10 @@ class ForeignKey(BaseModel): default=None, description="SQL join type: 'left' (default) or 'inner'", ) + join_filter: str | None = Field( + default=None, + description="SQL WHERE clause appended to JOIN ON clause (e.g., 'clientid = 2')", + ) @field_validator("references_table", mode="before") @classmethod @@ -1039,6 +1164,16 @@ def parse_references_column(cls, v, info) -> str | None: return parts[1] return v + def parse_table_reference(self) -> "TableReference": + """Parse references_table into TableReference with optional pipeline qualification. + + Returns: + TableReference with pipeline and table components + """ + from tablespec.models.pipeline import TableReference + + return TableReference.parse(self.references_table) + class ReferencedBy(BaseModel): """Reverse foreign key relationship.""" @@ -1109,6 +1244,21 @@ class OutgoingRelationship(BaseModel): cardinality: Cardinality | None = Field( default=None, description="Cardinality specification for relationship" ) + alternative_joins: list[dict[str, str]] | None = Field( + default=None, + description="Alternative join conditions using OR logic. Each dict specifies " + "source_column and target_column for additional join paths.", + ) + + def parse_table_reference(self) -> "TableReference": + """Parse target_table into TableReference with optional pipeline qualification. + + Returns: + TableReference with pipeline and table components + """ + from tablespec.models.pipeline import TableReference + + return TableReference.parse(self.target_table) class IncomingRelationship(BaseModel): @@ -1133,6 +1283,16 @@ class IncomingRelationship(BaseModel): default=None, description="Cardinality specification for relationship" ) + def parse_table_reference(self) -> "TableReference": + """Parse source_table into TableReference with optional pipeline qualification. + + Returns: + TableReference with pipeline and table components + """ + from tablespec.models.pipeline import TableReference + + return TableReference.parse(self.source_table) + class RelationshipSummary(BaseModel): """Summary statistics for table relationships.""" @@ -1209,10 +1369,56 @@ class UMFMetadata(BaseModel): description="Explicit base table name for SQL generation. Overrides automatic " "base table inference.", ) + base_table_filter: str | None = Field( + default=None, + description="SQL WHERE clause applied to the base table when building the " + "base view. Used by generated tables that need to pre-filter base-table " + "rows before joins run.", + ) + base_join_column: str | None = Field( + default=None, + description="Column on `base_table` to use as the join key when building " + "the base view. Overrides the auto-inferred key column. Set when the base " + "table has multiple key-like columns and the wrong one would be " + "auto-selected.", + ) + final_filter: str | None = Field( + default=None, + description="SQL WHERE clause applied after the final assembly view is built. " + "Runs against the fully-joined derivation output, so it can reference " + "derived columns that don't exist on the base table. Use base_table_filter " + "for filters that reference only base-table columns (faster, filters " + "earlier).", + ) source_tables: list[str] | None = Field( default=None, description="Source table names for union_sources strategy.", ) + union_base_tables: list[str] | None = Field( + default=None, + description="Additional tables to UNION ALL into the base view. " + "Tables must have columns matching the base_table's selected columns. " + "Used to merge rows from multiple sources into the generated base.", + ) + union_type: Literal["union_all", "union"] | None = Field( + default=None, + description="SQL union operator for union_base_tables. " + "Defaults to 'union_all' (preserves all rows). " + "Use 'union' for deduplication across sources.", + ) + union_exclude_base: bool = Field( + default=False, + description="When true, union_base_tables rows are excluded if their " + "primary key already exists in the base table (anti-join). " + "Prevents duplicate rows when populations may overlap.", + ) + union_coalesce_base: bool = Field( + default=False, + description="When true (with union_exclude_base), overlapping rows " + "are merged using COALESCE(base_col, union_col) instead of " + "excluding the union table rows. Base table values preferred; " + "union table values fill NULLs.", + ) unpivot_columns: list[str] | None = Field( default=None, description="Source columns to UNPIVOT into rows.", @@ -1226,6 +1432,13 @@ class UMFMetadata(BaseModel): description="Deduplication strategy for generated tables. " "'latest': deduplicate on primary_key, keeping the row with the most recent load date.", ) + final_dedup: Literal["distinct"] | None = Field( + default=None, + description="Final-assembly deduplication strategy (runs after all joins). " + "'distinct': emit SELECT DISTINCT so exact-duplicate rows produced by join " + "fan-out are collapsed. Use when a joined table has higher cardinality than " + "the base and you don't need every joined row.", + ) output_config: OutputConfig | None = Field( default=None, description="Output file configuration" ) @@ -1300,11 +1513,23 @@ class IngestionConfig(BaseModel): description="'snapshot': Filter to latest file, overwrite table. " "'incremental': Keep latest per PK, upsert to table.", ) + update_mode: Literal["upsert", "update_only"] = Field( + default="upsert", + description="How rows with load_mode='U' are merged into the ingested table. " + "'upsert' (default): insert if PK unmatched, update if matched — use when _U " + "files may contain the only copy of the data for a delivery. " + "'update_only': update matched PKs and drop unmatched rows with a warning — use " + "for record-of-truth tables where an unmatched _U row indicates a data problem.", + ) order_by: list[str] | None = Field( default=None, description="Columns to determine 'latest' (sorted descending). Can include: " - "filename-extracted columns (e.g., 'file_date_yyyymmdd') or " - "metadata columns (e.g., 'meta_source_name', 'meta_snapshot_dt', 'meta_load_dt'). " + "filename-extracted columns (e.g., 'file_date_yyyymmdd'), " + "metadata columns (e.g., 'meta_source_name', 'meta_snapshot_dt', 'meta_load_dt', " + "'meta_source_offset'), or data columns present in the row (e.g., 'run_date', " + "'updated_at'). Data columns are useful when the per-row recency signal lives " + "inside the file rather than in the filename. Use 'meta_source_offset' as a " + "stable tie-breaker for intra-file row order. " "For snapshot mode: filters to rows with MAX values of these columns. " "For incremental mode: orders rows when deduping by primary key.", ) @@ -1656,6 +1881,19 @@ def validate_version_format(cls, v) -> str: raise ValueError(msg) from e return v + @property + def effective_primary_key(self) -> list[str]: + """Return the primary key, defaulting to meta_checksum if not configured. + + This property provides the merge key for upsert operations. If no explicit + primary_key is configured, it defaults to meta_checksum (row-level hash) + to enable row-level deduplication. + + Returns: + Primary key columns, or DEFAULT_PRIMARY_KEY if not configured. + """ + return self.primary_key if self.primary_key else DEFAULT_PRIMARY_KEY + model_config = ConfigDict( validate_assignment=True, extra="forbid", diff --git a/tests/unit/test_umf_models.py b/tests/unit/test_umf_models.py index f3c592c0..400887da 100644 --- a/tests/unit/test_umf_models.py +++ b/tests/unit/test_umf_models.py @@ -797,3 +797,336 @@ def test_no_expectation_fields_does_not_warn(self): table_name="test", columns=[UMFColumn(name="id", data_type="INTEGER")], ) + + +class TestMergeCondition: + """Test MergeCondition and column-level merge fields.""" + + def test_merge_condition_model(self): + from tablespec.models.umf import MergeCondition + + cond = MergeCondition(column="load_mode", values=["I", "A"]) + assert cond.column == "load_mode" + assert cond.values == ["I", "A"] + + def test_column_merge_strategy_fields(self): + from tablespec.models.umf import MergeCondition + + col = UMFColumn( + name="chase_load_date", + data_type="DATE", + merge_strategy="keep_minimum", + merge_source="file_date_yyyymmdd", + merge_condition=MergeCondition(column="load_mode", values=["I"]), + ) + assert col.merge_strategy == "keep_minimum" + assert col.merge_source == "file_date_yyyymmdd" + assert col.merge_condition is not None + assert col.merge_condition.values == ["I"] + + def test_merge_strategy_defaults_to_none(self): + col = UMFColumn(name="a", data_type="VARCHAR") + assert col.merge_strategy is None + assert col.merge_source is None + assert col.merge_condition is None + + def test_invalid_merge_strategy_rejected(self): + with pytest.raises(ValidationError): + UMFColumn(name="a", data_type="VARCHAR", merge_strategy="not_a_strategy") + + +class TestInternalColumn: + """Test the internal helper-column flag.""" + + def test_internal_defaults_false(self): + assert UMFColumn(name="a", data_type="VARCHAR").internal is False + + def test_internal_column(self): + col = UMFColumn(name="winning_row_anchor", data_type="VARCHAR", internal=True) + assert col.internal is True + + +class TestBooleanNullable: + """Test that nullable accepts plain booleans alongside Nullable contexts.""" + + def test_nullable_true(self): + col = UMFColumn(name="a", data_type="VARCHAR", nullable=True) + assert col.is_nullable_for_all_contexts() is True + assert col.is_required_for_any_context() is False + + def test_nullable_false(self): + col = UMFColumn(name="a", data_type="VARCHAR", nullable=False) + assert col.is_nullable_for_all_contexts() is False + assert col.is_required_for_any_context() is True + + def test_nullable_model_still_works(self): + col = UMFColumn( + name="a", data_type="VARCHAR", nullable=Nullable(MD=False, MP=True, ME=True) + ) + assert col.is_nullable_for_all_contexts() is False + + +class TestParseTableReference: + """Test parse_table_reference on relationship and derivation models.""" + + def test_foreign_key_bare_reference(self): + fk = ForeignKey( + column="member_id", references_table="member_roster", references_column="id" + ) + ref = fk.parse_table_reference() + assert ref.pipeline is None + assert ref.table == "member_roster" + assert ref.is_external() is False + + def test_foreign_key_qualified_reference(self): + fk = ForeignKey( + column="icd_code", + references_table="reference_data.icd_codes", + references_column="code", + ) + ref = fk.parse_table_reference() + assert ref.pipeline == "reference_data" + assert ref.table == "icd_codes" + assert ref.is_external() is True + + def test_outgoing_relationship_reference(self): + from tablespec.models.umf import OutgoingRelationship + + rel = OutgoingRelationship( + target_table="other_pipeline.target", + source_column="a", + target_column="b", + type="foreign_to_primary", + confidence=0.9, + ) + ref = rel.parse_table_reference() + assert ref.pipeline == "other_pipeline" + assert ref.table == "target" + + def test_incoming_relationship_reference(self): + from tablespec.models.umf import IncomingRelationship + + rel = IncomingRelationship( + source_table="src_table", + source_column="a", + target_column="b", + type="foreign_to_foreign", + confidence=0.9, + ) + ref = rel.parse_table_reference() + assert ref.pipeline is None + assert ref.table == "src_table" + + def test_derivation_candidate_reference(self): + cand = DerivationCandidate(table="pipe.tab", column="c", priority=1) + ref = cand.parse_table_reference() + assert (ref.pipeline, ref.table) == ("pipe", "tab") + + +class TestTableReference: + """Test TableReference parsing and formatting.""" + + def test_parse_bare(self): + from tablespec.models.pipeline import TableReference + + ref = TableReference.parse("outreach_list") + assert ref.pipeline is None + assert ref.table == "outreach_list" + assert str(ref) == "outreach_list" + + def test_parse_qualified(self): + from tablespec.models.pipeline import TableReference + + ref = TableReference.parse("reference_data.icd_codes") + assert ref.pipeline == "reference_data" + assert ref.table == "icd_codes" + assert str(ref) == "reference_data.icd_codes" + + def test_to_qualified_name_resolves_bare_with_current_pipeline(self): + from tablespec.models.pipeline import TableReference + + ref = TableReference.parse("outreach_list") + assert ref.to_qualified_name("my_pipeline") == "my_pipeline.outreach_list" + assert ref.to_qualified_name() == "outreach_list" + + def test_qualified_ignores_current_pipeline(self): + from tablespec.models.pipeline import TableReference + + ref = TableReference.parse("other.t") + assert ref.to_qualified_name("my_pipeline") == "other.t" + + def test_splits_on_first_dot_only(self): + from tablespec.models.pipeline import TableReference + + ref = TableReference.parse("a.b.c") + assert ref.pipeline == "a" + assert ref.table == "b.c" + + +class TestRelationshipJoinExtensions: + """Test join_filter and alternative_joins fields.""" + + def test_foreign_key_join_filter(self): + fk = ForeignKey( + column="member_id", + references_table="enterprise", + references_column="member_id", + join_filter="clientid = 2", + ) + assert fk.join_filter == "clientid = 2" + + def test_alternative_joins(self): + from tablespec.models.umf import OutgoingRelationship + + rel = OutgoingRelationship( + target_table="t", + source_column="a", + target_column="b", + type="foreign_to_primary", + confidence=1.0, + alternative_joins=[{"source_column": "c", "target_column": "d"}], + ) + assert rel.alternative_joins == [{"source_column": "c", "target_column": "d"}] + + +class TestDerivationCandidateUnionValue: + """Test union_value literal on derivation candidates.""" + + def test_union_value_literal(self): + cand = DerivationCandidate( + table="crosswalk", column="c", priority=1, union_value=True + ) + assert cand.union_value is True + + def test_union_value_defaults_none(self): + cand = DerivationCandidate(table="t", column="c", priority=1) + assert cand.union_value is None + + +class TestUMFMetadataSqlGenerationFields: + """Test SQL-generation control fields on UMFMetadata.""" + + def test_base_and_final_filters(self): + from tablespec.models.umf import UMFMetadata + + meta = UMFMetadata( + base_table="charge_file", + base_table_filter="assess_type IN ('QUA','RA')", + base_join_column="mrn", + final_filter="disposition_id IS NOT NULL", + ) + assert meta.base_table_filter == "assess_type IN ('QUA','RA')" + assert meta.base_join_column == "mrn" + assert meta.final_filter == "disposition_id IS NOT NULL" + + def test_union_base_tables_config(self): + from tablespec.models.umf import UMFMetadata + + meta = UMFMetadata( + union_base_tables=["crosswalk_outreach_list"], + union_type="union_all", + union_exclude_base=True, + union_coalesce_base=True, + ) + assert meta.union_base_tables == ["crosswalk_outreach_list"] + assert meta.union_type == "union_all" + assert meta.union_exclude_base is True + assert meta.union_coalesce_base is True + + def test_union_defaults(self): + from tablespec.models.umf import UMFMetadata + + meta = UMFMetadata() + assert meta.union_base_tables is None + assert meta.union_type is None + assert meta.union_exclude_base is False + assert meta.union_coalesce_base is False + assert meta.final_dedup is None + + def test_final_dedup_distinct(self): + from tablespec.models.umf import UMFMetadata + + meta = UMFMetadata(final_dedup="distinct") + assert meta.final_dedup == "distinct" + + def test_invalid_union_type_rejected(self): + from tablespec.models.umf import UMFMetadata + + with pytest.raises(ValidationError): + UMFMetadata(union_type="cross_join") + + +class TestIngestionUpdateMode: + """Test IngestionConfig.update_mode.""" + + def test_update_mode_defaults_to_upsert(self): + from tablespec.models.umf import IngestionConfig + + assert IngestionConfig().update_mode == "upsert" + + def test_update_only(self): + from tablespec.models.umf import IngestionConfig + + assert IngestionConfig(update_mode="update_only").update_mode == "update_only" + + def test_invalid_update_mode_rejected(self): + from tablespec.models.umf import IngestionConfig + + with pytest.raises(ValidationError): + IngestionConfig(update_mode="insert_only") + + +class TestEffectivePrimaryKey: + """Test UMF.effective_primary_key default merge key.""" + + def test_explicit_primary_key(self): + umf = UMF( + version="1.0", + table_name="t", + columns=[UMFColumn(name="id", data_type="INTEGER")], + primary_key=["id"], + ) + assert umf.effective_primary_key == ["id"] + + def test_defaults_to_meta_checksum(self): + from tablespec.models.umf import DEFAULT_PRIMARY_KEY + + umf = UMF( + version="1.0", + table_name="t", + columns=[UMFColumn(name="id", data_type="INTEGER")], + ) + assert umf.effective_primary_key == DEFAULT_PRIMARY_KEY + assert umf.effective_primary_key == ["meta_checksum"] + + +class TestValidationRulesMisclassificationWarning: + """Test ValidationRules warns on ingested-stage expectations.""" + + def test_warns_on_ingested_stage_expectation(self): + from tablespec.models.umf import ValidationRules + + with pytest.warns(UserWarning, match="ingested-stage expectations"): + ValidationRules( + expectations=[{"type": "expect_column_values_to_be_between"}] + ) + + def test_no_warning_for_raw_expectations(self): + import warnings as _warnings + + from tablespec.models.umf import ValidationRules + + with _warnings.catch_warnings(): + _warnings.simplefilter("error", UserWarning) + ValidationRules( + expectations=[{"type": "expect_column_values_to_match_regex"}] + ) + + def test_no_warning_when_expectations_absent(self): + import warnings as _warnings + + from tablespec.models.umf import ValidationRules + + with _warnings.catch_warnings(): + _warnings.simplefilter("error", UserWarning) + ValidationRules(table_level=None, column_level=None) diff --git a/tests/unit/test_validator.py b/tests/unit/test_validator.py index b2487556..739a9af2 100644 --- a/tests/unit/test_validator.py +++ b/tests/unit/test_validator.py @@ -224,6 +224,9 @@ def test_missing_expectation_type_field( assert success is False assert any("Missing 'type' field" in e for e in errors) + @pytest.mark.filterwarnings( + "ignore:ValidationRules contains ingested-stage expectations" + ) @patch("tablespec.validator.validate_naming_conventions", return_value=[]) @patch("tablespec.validator.validate_provenance_columns", return_value=[]) @patch("tablespec.validator.validate_domain_types", return_value=[]) @@ -296,6 +299,9 @@ def test_meta_prefix_columns_skipped( success, errors = validate_table(table_file, ctx) assert success is True + @pytest.mark.filterwarnings( + "ignore:ValidationRules contains ingested-stage expectations" + ) @patch("tablespec.validator.validate_naming_conventions", return_value=[]) @patch("tablespec.validator.validate_provenance_columns", return_value=[]) @patch("tablespec.validator.validate_domain_types", return_value=[]) From f5bbf235a6265e9ea1b1bbb81b085589d92dfe88 Mon Sep 17 00:00:00 2001 From: David Mautz Date: Fri, 24 Jul 2026 08:06:52 -0500 Subject: [PATCH 2/4] feat(models): add UMFColumn.report_sheet with Excel round-trip Workbook tab assignment for multi-sheet report authoring: columns can declare which Excel worksheet tab they belong to in a generated multi-sheet workbook report. Authoring-time provenance consumed by report-configuration tooling; bounded to Excel's 31-character tab-name limit. The Columns-sheet exporter writes a trailing "Report Sheet" header and the importer locates it by header name (consistent with the existing header-mapped reader), so older workbooks without the column import unchanged. Co-Authored-By: Claude Fable 5 --- src/tablespec/excel_converter.py | 16 +++++++++++++++- src/tablespec/models/umf.py | 9 +++++++++ tests/unit/test_excel_converter.py | 19 +++++++++++++++++++ tests/unit/test_umf_models.py | 18 ++++++++++++++++++ 4 files changed, 61 insertions(+), 1 deletion(-) diff --git a/src/tablespec/excel_converter.py b/src/tablespec/excel_converter.py index 2406d5f9..ad6e78e6 100644 --- a/src/tablespec/excel_converter.py +++ b/src/tablespec/excel_converter.py @@ -669,6 +669,7 @@ def _create_columns_sheet(self, umf: UMF) -> None: "Format", "Notes", "_Validation", + "Report Sheet", ] headers.extend(post_nullable_headers) @@ -688,6 +689,7 @@ def col_letter(idx: int) -> str: reporting_idx = desc_idx + 5 format_idx = desc_idx + 6 notes_idx = desc_idx + 7 + report_sheet_idx = desc_idx + 9 # desc_idx + 8 is the _Validation column # Data row = 2 @@ -770,12 +772,18 @@ def col_letter(idx: int) -> str: ws[f"{col_letter(notes_idx)}{row}"] = "" self._apply_font_to_cell(ws[f"{col_letter(notes_idx)}{row}"], default_font) + # Report Sheet (workbook tab assignment for multi-sheet reports) + ws[f"{col_letter(report_sheet_idx)}{row}"] = col.report_sheet or "" + self._apply_font_to_cell( + ws[f"{col_letter(report_sheet_idx)}{row}"], default_font + ) + row += 1 # Adjust column widths pre_nullable_widths = [15, 18, 20, 12, 10, 11, 8] nullable_widths = [12] * num_contexts - post_nullable_widths = [20, 20, 12, 18, 15, 12, 15, 30, 15] + post_nullable_widths = [20, 20, 12, 18, 15, 12, 15, 30, 15, 15] all_widths = pre_nullable_widths + nullable_widths + post_nullable_widths for i, width in enumerate(all_widths, 1): ws.column_dimensions[get_column_letter(i)].width = width @@ -1938,6 +1946,7 @@ def _get(row: tuple, idx: int | None) -> Any: derivation_expression_idx = header_map.get("derivation expression") format_idx = header_map.get("format") notes_idx = header_map.get("notes") + report_sheet_idx = header_map.get("report sheet") columns = [] for row in ws.iter_rows(min_row=2, values_only=False): @@ -2066,6 +2075,11 @@ def _get(row: tuple, idx: int | None) -> Any: line.strip() for line in notes_val.split("\n") if line.strip() ] + # Report Sheet (workbook tab assignment for multi-sheet reports) + report_sheet_val = _get(row, report_sheet_idx) + if report_sheet_val: + col_dict["report_sheet"] = str(report_sheet_val).strip() + columns.append(col_dict) return columns diff --git a/src/tablespec/models/umf.py b/src/tablespec/models/umf.py index c7a2e5c6..930cd873 100644 --- a/src/tablespec/models/umf.py +++ b/src/tablespec/models/umf.py @@ -718,6 +718,15 @@ class UMFColumn(BaseModel): description="Reporting requirement classification: 'R' (Required), 'O' (Optional), 'S' (Suggested)", pattern=r"^(R|O|S)$", ) + report_sheet: str | None = Field( + default=None, + description="Excel worksheet tab this column is assigned to in a multi-sheet " + "workbook report. Authoring-time provenance from the source spec workbook; " + "consumed by report-configuration tooling, not by runtime output generation. " + "Length is bounded by Excel's 31-character tab-name limit.", + min_length=1, + max_length=31, + ) derived_from: str | None = Field( default=None, description="DEPRECATED: Use 'derivation' field instead. " diff --git a/tests/unit/test_excel_converter.py b/tests/unit/test_excel_converter.py index 34d38c84..47d4dc33 100644 --- a/tests/unit/test_excel_converter.py +++ b/tests/unit/test_excel_converter.py @@ -872,6 +872,25 @@ def test_round_trip_nullable_preserved(self, tmp_path): assert nullable["MP"] is False assert nullable["ME"] is True + def test_round_trip_report_sheet_preserved(self, tmp_path): + umf = _make_minimal_umf( + columns=[ + UMFColumn(name="col_id", data_type="INTEGER", report_sheet="Summary"), + UMFColumn(name="col_name", data_type="VARCHAR", length=100), + ] + ) + exporter = UMFToExcelConverter() + wb = exporter.convert(umf) + out = tmp_path / "rt_report_sheet.xlsx" + wb.save(out) + + importer = ExcelToUMFConverter() + columns = importer._extract_columns(openpyxl.load_workbook(out)) + col_id_data = next(c for c in columns if c["name"] == "col_id") + assert col_id_data["report_sheet"] == "Summary" + col_name_data = next(c for c in columns if c["name"] == "col_name") + assert "report_sheet" not in col_name_data + # --------------------------------------------------------------------------- # Edge cases diff --git a/tests/unit/test_umf_models.py b/tests/unit/test_umf_models.py index 400887da..6a5a1483 100644 --- a/tests/unit/test_umf_models.py +++ b/tests/unit/test_umf_models.py @@ -1130,3 +1130,21 @@ def test_no_warning_when_expectations_absent(self): with _warnings.catch_warnings(): _warnings.simplefilter("error", UserWarning) ValidationRules(table_level=None, column_level=None) + + +class TestReportSheet: + """Test UMFColumn.report_sheet workbook tab assignment.""" + + def test_report_sheet_defaults_to_none(self): + col = UMFColumn(name="test_col", data_type="VARCHAR") + assert col.report_sheet is None + + def test_report_sheet_accepts_valid_name(self): + col = UMFColumn(name="test_col", data_type="VARCHAR", report_sheet="Summary") + assert col.report_sheet == "Summary" + + def test_report_sheet_rejects_out_of_range_lengths(self): + with pytest.raises(ValidationError): + UMFColumn(name="test_col", data_type="VARCHAR", report_sheet="") + with pytest.raises(ValidationError): + UMFColumn(name="test_col", data_type="VARCHAR", report_sheet="x" * 32) From 99c5c919aa630cd988a014aab6c2912289eea62a Mon Sep 17 00:00:00 2001 From: David Mautz Date: Fri, 24 Jul 2026 08:13:37 -0500 Subject: [PATCH 3/4] feat(excel): emit derivation sheets only for tables that have derivations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Source-layer specs (plain ingested tables) have no survivorship or derivation metadata, so their exported workbooks no longer carry empty Survivorship/Derivations tabs — matching the existing conditional emission of Validation Rules, Relationships, and File Format sheets. The importer already tolerates absent sheets, so round-trip behavior is unchanged for both spec shapes. Co-Authored-By: Claude Fable 5 --- src/tablespec/excel_converter.py | 9 ++++++-- tests/unit/test_excel_converter.py | 34 ++++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/src/tablespec/excel_converter.py b/src/tablespec/excel_converter.py index ad6e78e6..fca4b059 100644 --- a/src/tablespec/excel_converter.py +++ b/src/tablespec/excel_converter.py @@ -490,8 +490,13 @@ def convert(self, umf: UMF) -> openpyxl.Workbook: self._create_readme_sheet(umf) self._create_schema_sheet(umf) self._create_columns_sheet(umf) - self._create_survivorship_sheet(umf) - self._create_derivations_sheet(umf) + + # Derivation-related sheets only apply to generated/derived tables; + # source-layer specs without survivorship metadata get a lean workbook. + if any(col.derivation or col.provenance_policy for col in umf.columns): + self._create_survivorship_sheet(umf) + if any(col.derivation for col in umf.columns): + self._create_derivations_sheet(umf) if self._get_expectation_dicts_for_export(umf): self._create_validation_sheet(umf) diff --git a/tests/unit/test_excel_converter.py b/tests/unit/test_excel_converter.py index 47d4dc33..940e7e4a 100644 --- a/tests/unit/test_excel_converter.py +++ b/tests/unit/test_excel_converter.py @@ -872,6 +872,40 @@ def test_round_trip_nullable_preserved(self, tmp_path): assert nullable["MP"] is False assert nullable["ME"] is True + def test_source_spec_omits_derivation_sheets(self, tmp_path): + """Tables without derivation/survivorship metadata get a lean workbook.""" + umf = _make_minimal_umf() + wb = UMFToExcelConverter().convert(umf) + assert "Survivorship" not in wb.sheetnames + assert "Derivations" not in wb.sheetnames + + # And the lean workbook still imports cleanly + out = tmp_path / "lean.xlsx" + wb.save(out) + columns = ExcelToUMFConverter()._extract_columns(openpyxl.load_workbook(out)) + assert {c["name"] for c in columns} == {"col_id", "col_name"} + + def test_generated_spec_includes_derivation_sheets(self): + """Tables with column derivations keep Survivorship + Derivations tabs.""" + umf = _make_minimal_umf( + columns=[ + UMFColumn( + name="col_id", + data_type="INTEGER", + derivation=UMFColumnDerivation( + candidates=[ + DerivationCandidate( + table="source_table", column="id", priority=1 + ) + ] + ), + ), + ] + ) + wb = UMFToExcelConverter().convert(umf) + assert "Survivorship" in wb.sheetnames + assert "Derivations" in wb.sheetnames + def test_round_trip_report_sheet_preserved(self, tmp_path): umf = _make_minimal_umf( columns=[ From 520d878289f6a7a589e61459437f4c60d72f750d Mon Sep 17 00:00:00 2001 From: David Mautz Date: Fri, 24 Jul 2026 10:03:51 -0500 Subject: [PATCH 4/4] fix(models): wire internal-column and boolean-nullable semantics end to end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ported UMFColumn.internal flag and bool-widened nullable field were model-only; their consumers didn't honor them: - Schema generators (SQL DDL, PySpark, JSON): skip internal helper columns — they participate in derivation but are excluded from the output table. - GX baseline: skip internal columns in column-level expectations, column-count/ordered-list structural checks, and cross-column date pairs; a plain `nullable: false` now emits the global not-null expectation (it was silently skipped as falsy). - Excel export: a boolean nullable writes its value into every context cell instead of defaulting all contexts to False. - Compatibility checker: boolean nullable maps to the synthetic "*" context instead of crashing on model_fields_set, so True -> False is still reported as a breaking tightening. Co-Authored-By: Claude Fable 5 --- src/tablespec/compatibility.py | 10 ++++- src/tablespec/excel_converter.py | 7 +++- src/tablespec/gx_baseline.py | 29 ++++++++++++-- src/tablespec/schemas/generators.py | 6 +++ tests/unit/test_baseline_fixes.py | 57 ++++++++++++++++++++++++++++ tests/unit/test_compatibility.py | 26 +++++++++++++ tests/unit/test_excel_converter.py | 18 +++++++++ tests/unit/test_schema_generators.py | 29 ++++++++++++++ 8 files changed, 174 insertions(+), 8 deletions(-) diff --git a/src/tablespec/compatibility.py b/src/tablespec/compatibility.py index 85ec8090..77f31132 100644 --- a/src/tablespec/compatibility.py +++ b/src/tablespec/compatibility.py @@ -42,10 +42,16 @@ class CompatibilityReport: # --------------------------------------------------------------------------- -def _nullable_contexts(n: Nullable | None) -> dict[str, bool]: - """Return a {context_key: bool} dict for a Nullable, treating None as empty.""" +def _nullable_contexts(n: bool | Nullable | None) -> dict[str, bool]: + """Return a {context_key: bool} dict for a Nullable, treating None as empty. + + A plain boolean nullable applies to every context and is represented by + the synthetic "*" key so tightening (True -> False) is still detected. + """ if n is None: return {} + if isinstance(n, bool): + return {"*": n} # Nullable uses model_config extra="allow", so extra fields are in __pydantic_extra__ result: dict[str, bool] = {} for key in n.model_fields_set | set(n.__pydantic_extra__ or {}): diff --git a/src/tablespec/excel_converter.py b/src/tablespec/excel_converter.py index fca4b059..dd7e455f 100644 --- a/src/tablespec/excel_converter.py +++ b/src/tablespec/excel_converter.py @@ -729,9 +729,12 @@ def col_letter(idx: int) -> str: self._apply_font_to_cell(ws[f"G{row}"], default_font) # Nullable - write dynamic context columns - if col.nullable: + if col.nullable is not None: nullable_data: dict[str, bool] = {} - if isinstance(col.nullable, Nullable): + if isinstance(col.nullable, bool): + # Simple boolean applies to every context + nullable_data = dict.fromkeys(context_keys, col.nullable) + elif isinstance(col.nullable, Nullable): nullable_data = col.nullable.model_dump(exclude_none=True) elif isinstance(col.nullable, dict): nullable_data = col.nullable diff --git a/src/tablespec/gx_baseline.py b/src/tablespec/gx_baseline.py index a4d57168..11268d97 100644 --- a/src/tablespec/gx_baseline.py +++ b/src/tablespec/gx_baseline.py @@ -226,9 +226,11 @@ def generate_baseline_expectations( if include_structural: expectations.extend(self._generate_structural_expectations(umf_data)) - # Column-level baseline expectations + # Column-level baseline expectations (skip internal helper columns - not in output) context_column = umf_data.get("context_column") for column in umf_data.get("columns", []): + if column.get("internal", False): + continue expectations.extend( self.generate_baseline_column_expectations( column, context_column=context_column @@ -262,7 +264,11 @@ def _generate_structural_expectations( """ expectations = [] - columns = umf_data.get("columns", []) + # Internal helper columns are excluded from the output table, so + # structural checks must count only the externally visible columns. + columns = [ + col for col in umf_data.get("columns", []) if not col.get("internal", False) + ] column_names = [col["name"] for col in columns] # Expect specific column count @@ -312,7 +318,9 @@ def _generate_cross_column_expectations( """ expectations = [] - columns = umf_data.get("columns", []) + columns = [ + col for col in umf_data.get("columns", []) if not col.get("internal", False) + ] # Build lookup of date/datetime columns date_columns: dict[str, dict[str, Any]] = {} @@ -391,7 +399,20 @@ def generate_baseline_column_expectations( # Note: expect_column_to_exist and expect_column_values_to_be_of_type # are no longer generated here as they are in REDUNDANT_VALIDATION_TYPES nullable = column.get("nullable", {}) - if nullable: + if nullable is False: + # Simple boolean form: required in all contexts + expectations.append( + { + "type": "expect_column_values_to_not_be_null", + "kwargs": {"column": column_name}, + "meta": { + "description": f"Column {column_name} is required (nullable=false)", + "severity": "critical", + "generated_from": "baseline", + }, + } + ) + elif nullable: if isinstance(nullable, dict): # Dict format: {context: is_nullable} e.g. {"MD": False, "MP": True} required_contexts = [ diff --git a/src/tablespec/schemas/generators.py b/src/tablespec/schemas/generators.py index de328b36..250ec15c 100644 --- a/src/tablespec/schemas/generators.py +++ b/src/tablespec/schemas/generators.py @@ -68,6 +68,8 @@ def generate_sql_ddl(umf_data: dict[str, Any]) -> str: columns = [] for col in umf_data["columns"]: + if col.get("internal", False): + continue # internal helper columns are not part of the output table col_name = col["name"] data_type = col.get("data_type", "VARCHAR") is_nullable = _resolve_nullable(col.get("nullable")) @@ -169,6 +171,8 @@ def generate_pyspark_schema(umf_data: dict[str, Any]) -> str: fields = [] for col in umf_data["columns"]: + if col.get("internal", False): + continue # internal helper columns are not part of the output schema col_name = col["name"] data_type = col.get("data_type", "VARCHAR") nullable = _resolve_nullable(col.get("nullable")) @@ -202,6 +206,8 @@ def generate_json_schema(umf_data: dict[str, Any]) -> dict[str, Any]: } for col in umf_data["columns"]: + if col.get("internal", False): + continue # internal helper columns are not part of the output schema col_name = col["name"] col_desc = col.get("description", "") nullable = _resolve_nullable(col.get("nullable")) diff --git a/tests/unit/test_baseline_fixes.py b/tests/unit/test_baseline_fixes.py index 71206ae0..01ee8196 100644 --- a/tests/unit/test_baseline_fixes.py +++ b/tests/unit/test_baseline_fixes.py @@ -425,3 +425,60 @@ def test_profiling_wired_into_full_generate(self): gen = BaselineExpectationGenerator() exps = gen.generate_baseline_expectations(umf_data) assert any(e["type"] == "expect_column_values_to_be_unique" for e in exps) + + +class TestInternalColumnsExcludedFromBaseline: + """Internal helper columns never appear in baseline expectations.""" + + def test_internal_columns_skipped_everywhere(self): + umf_data = { + "table_name": "t", + "columns": [ + {"name": "id", "data_type": "INTEGER", "nullable": False}, + { + "name": "helper", + "data_type": "VARCHAR", + "internal": True, + "nullable": False, + "max_length": 10, + }, + ], + } + gen = BaselineExpectationGenerator() + expectations = gen.generate_baseline_expectations(umf_data) + + for exp in expectations: + kwargs = exp.get("kwargs", {}) + assert kwargs.get("column") != "helper" + assert "helper" not in kwargs.get("column_list", []) + if exp["type"] == "expect_table_column_count_to_equal": + assert kwargs["value"] == 1 + + +class TestBooleanNullableBaseline: + """Plain boolean nullable generates the right not-null expectations.""" + + def _not_null_columns(self, expectations): + return { + exp["kwargs"]["column"] + for exp in expectations + if exp["type"] == "expect_column_values_to_not_be_null" + } + + def test_nullable_false_bool_generates_not_null(self): + umf_data = { + "table_name": "t", + "columns": [{"name": "id", "data_type": "INTEGER", "nullable": False}], + } + gen = BaselineExpectationGenerator() + expectations = gen.generate_baseline_expectations(umf_data) + assert "id" in self._not_null_columns(expectations) + + def test_nullable_true_bool_generates_no_not_null(self): + umf_data = { + "table_name": "t", + "columns": [{"name": "id", "data_type": "INTEGER", "nullable": True}], + } + gen = BaselineExpectationGenerator() + expectations = gen.generate_baseline_expectations(umf_data) + assert "id" not in self._not_null_columns(expectations) diff --git a/tests/unit/test_compatibility.py b/tests/unit/test_compatibility.py index 2b4b12c6..38099b75 100644 --- a/tests/unit/test_compatibility.py +++ b/tests/unit/test_compatibility.py @@ -498,3 +498,29 @@ def test_reflexivity_with_shared_strategy(self, umf): assert report.is_backward_compatible assert report.is_forward_compatible assert len(report.issues) == 0 + + +class TestBooleanNullable: + """Plain boolean nullable is handled without crashing and detects tightening.""" + + def test_bool_tightening_is_breaking(self): + old = _umf([_col("x", "VARCHAR", nullable=True)]) + new = _umf([_col("x", "VARCHAR", nullable=False)]) + report = check_compatibility(old, new) + assert not report.is_backward_compatible + tightened = _issues_by_change(report, "nullable_tightened") + assert len(tightened) == 1 + + def test_bool_relaxation_is_safe(self): + old = _umf([_col("x", "VARCHAR", nullable=False)]) + new = _umf([_col("x", "VARCHAR", nullable=True)]) + report = check_compatibility(old, new) + assert report.is_backward_compatible + + def test_adding_nullable_bool_column_is_info(self): + old = _umf([_col("id", "INTEGER", length=None)]) + new = _umf( + [_col("id", "INTEGER", length=None), _col("x", "VARCHAR", nullable=True)] + ) + report = check_compatibility(old, new) + assert report.is_backward_compatible diff --git a/tests/unit/test_excel_converter.py b/tests/unit/test_excel_converter.py index 940e7e4a..0c0bcf03 100644 --- a/tests/unit/test_excel_converter.py +++ b/tests/unit/test_excel_converter.py @@ -872,6 +872,24 @@ def test_round_trip_nullable_preserved(self, tmp_path): assert nullable["MP"] is False assert nullable["ME"] is True + def test_round_trip_boolean_nullable(self, tmp_path): + """Plain boolean nullable exports its value into every context cell.""" + umf = _make_minimal_umf( + columns=[ + UMFColumn(name="col_req", data_type="VARCHAR", nullable=False), + UMFColumn(name="col_null", data_type="VARCHAR", nullable=True), + ] + ) + wb = UMFToExcelConverter().convert(umf) + out = tmp_path / "rt_bool_nullable.xlsx" + wb.save(out) + + columns = ExcelToUMFConverter()._extract_columns(openpyxl.load_workbook(out)) + col_req = next(c for c in columns if c["name"] == "col_req") + assert all(v is False for v in col_req["nullable"].values()) + col_null = next(c for c in columns if c["name"] == "col_null") + assert all(v is True for v in col_null["nullable"].values()) + def test_source_spec_omits_derivation_sheets(self, tmp_path): """Tables without derivation/survivorship metadata get a lean workbook.""" umf = _make_minimal_umf() diff --git a/tests/unit/test_schema_generators.py b/tests/unit/test_schema_generators.py index 17b6e914..b170ccd5 100644 --- a/tests/unit/test_schema_generators.py +++ b/tests/unit/test_schema_generators.py @@ -613,3 +613,32 @@ def test_json_schema_never_crashes_and_returns_valid_json(self, data): json_str = json.dumps(result) parsed = json.loads(json_str) assert parsed == result + + +class TestInternalColumnsExcluded: + """Internal helper columns are excluded from all generated schemas.""" + + @pytest.fixture + def umf_with_internal(self): + return { + "table_name": "test_table", + "columns": [ + {"name": "id", "data_type": "INTEGER", "nullable": False}, + {"name": "helper", "data_type": "STRING", "internal": True}, + ], + } + + def test_sql_ddl_excludes_internal(self, umf_with_internal): + ddl = generate_sql_ddl(umf_with_internal) + assert "id" in ddl + assert "helper" not in ddl + + def test_pyspark_schema_excludes_internal(self, umf_with_internal): + code = generate_pyspark_schema(umf_with_internal) + assert '"id"' in code + assert "helper" not in code + + def test_json_schema_excludes_internal(self, umf_with_internal): + schema = generate_json_schema(umf_with_internal) + assert "id" in schema["properties"] + assert "helper" not in schema["properties"]