From c8031817642f45be37d5fb3f63fdb31e2d129050 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Mon, 7 Sep 2026 05:30:47 +0000 Subject: [PATCH 01/13] Wire model configuration telemetry --- dbt/adapters/databricks/telemetry/builder.py | 347 +++++++++++++++++++ dbt/adapters/databricks/telemetry/hooks.py | 1 + dbt/adapters/databricks/telemetry/models.py | 157 +++++++++ tests/unit/telemetry/test_builder.py | 213 ++++++++++++ tests/unit/telemetry/test_encoder.py | 12 + tests/unit/telemetry/test_hooks.py | 18 + 6 files changed, 748 insertions(+) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index 022b2e496..e2f9e21aa 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -1,3 +1,5 @@ +from collections import Counter +from dataclasses import dataclass, field from importlib.metadata import version as _pkg_version from typing import Any, Callable, Optional @@ -32,6 +34,125 @@ "run_operation": models.DbtCommand.RUN_OPERATION, } +_MATERIALIZATION_MAP = { + "table": models.Materialization.TABLE, + "view": models.Materialization.VIEW, + "incremental": models.Materialization.INCREMENTAL, + "ephemeral": models.Materialization.EPHEMERAL, + "materialized_view": models.Materialization.MATERIALIZED_VIEW, + "streaming_table": models.Materialization.STREAMING_TABLE, + "metric_view": models.Materialization.METRIC_VIEW, +} + +_LANGUAGE_MAP = { + "sql": models.Language.SQL, + "python": models.Language.PYTHON, +} + +_INCREMENTAL_STRATEGY_MAP = { + "merge": models.IncrementalStrategy.MERGE, + "append": models.IncrementalStrategy.APPEND, + "delete+insert": models.IncrementalStrategy.DELETE_INSERT, + "delete_insert": models.IncrementalStrategy.DELETE_INSERT, + "insert_overwrite": models.IncrementalStrategy.INSERT_OVERWRITE, + "replace_where": models.IncrementalStrategy.REPLACE_WHERE, + "microbatch": models.IncrementalStrategy.MICROBATCH, +} + +_PYTHON_SUBMISSION_METHOD_MAP = { + "serverless_cluster": models.PythonSubmissionMethod.SERVERLESS_CLUSTER, + "job_cluster": models.PythonSubmissionMethod.JOB_CLUSTER, + "all_purpose_cluster": models.PythonSubmissionMethod.ALL_PURPOSE_CLUSTER, + "workflow_job": models.PythonSubmissionMethod.WORKFLOW_JOB, +} + +_CATALOG_TYPE_MAP = { + "unity": models.CatalogType.UNITY_CATALOG, + "unity_catalog": models.CatalogType.UNITY_CATALOG, + "hive_metastore": models.CatalogType.HIVE_METASTORE, +} + +_FILE_FORMAT_MAP = { + "delta": models.EffectiveStorageFormat.DELTA, + "parquet": models.EffectiveStorageFormat.PARQUET, + "hudi": models.EffectiveStorageFormat.HUDI, +} + +_CONSTRAINT_CONFIG_MAP = { + "not_null": models.ModelConfig.NOT_NULL_CONSTRAINT, + "check": models.ModelConfig.CHECK_CONSTRAINT, + "primary_key": models.ModelConfig.PRIMARY_KEY_CONSTRAINT, + "foreign_key": models.ModelConfig.FOREIGN_KEY_CONSTRAINT, + "custom": models.ModelConfig.CUSTOM_CONSTRAINT, +} + +_STORAGE_FORMAT_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, +} + + +@dataclass +class _ModelConfigAccumulator: + scope: models.ModelConfigScope + model_count: int = 0 + materializations: Counter = field(default_factory=Counter) + languages: Counter = field(default_factory=Counter) + incremental_model_count: int = 0 + incremental_strategies: Counter = field(default_factory=Counter) + incremental_config_usage: Counter = field(default_factory=Counter) + storage_formats: Counter = field(default_factory=Counter) + catalog_types: Counter = field(default_factory=Counter) + compute_types: Counter = field(default_factory=Counter) + python_model_count: int = 0 + python_submission_methods: Counter = field(default_factory=Counter) + config_usage: Counter = field(default_factory=Counter) + + def to_model(self) -> models.ModelConfigStats: + return models.ModelConfigStats( + scope=self.scope, + model_count=self.model_count, + materialization_counts=_count_rows( + self.materializations, models.Materialization, models.MaterializationCount + ), + language_counts=_count_rows(self.languages, models.Language, models.LanguageCount), + incremental_model_stats=models.IncrementalModelStats( + model_count=self.incremental_model_count, + strategy_counts=_count_rows( + self.incremental_strategies, + models.IncrementalStrategy, + models.IncrementalStrategyCount, + ), + config_usage=_count_rows( + self.incremental_config_usage, + models.ModelConfig, + models.ModelConfigUsage, + ), + ), + effective_storage_format_counts=_count_rows( + self.storage_formats, + models.EffectiveStorageFormat, + models.EffectiveStorageFormatCount, + ), + catalog_type_counts=_count_rows( + self.catalog_types, models.CatalogType, models.CatalogTypeCount + ), + effective_compute_type_counts=_count_rows( + self.compute_types, models.ComputeType, models.ComputeTypeCount + ), + python_model_stats=models.PythonModelStats( + model_count=self.python_model_count, + submission_method_counts=_count_rows( + self.python_submission_methods, + models.PythonSubmissionMethod, + models.PythonSubmissionMethodCount, + ), + ), + config_usage=_count_rows( + self.config_usage, models.ModelConfig, models.ModelConfigUsage + ), + ) + def classify_compute_type(http_path: Optional[str]) -> models.ComputeType: if not http_path: @@ -148,6 +269,225 @@ def aggregate_manifest(manifest: Any) -> models.ManifestStats: return stats +def _count_rows(counts: Counter, enum_type: Any, row_type: type) -> list: + return [row_type(value, counts[value]) for value in enum_type if counts[value]] + + +def _value(obj: Any, name: str, default: Any = None) -> Any: + if obj is None: + return default + if isinstance(obj, dict): + return obj.get(name, default) + getter = getattr(obj, "get", None) + if getter is not None: + try: + return getter(name, default) + except TypeError: + value = getter(name) + return default if value is None else value + return getattr(obj, name, default) + + +def _normalized(value: Any) -> str: + raw = getattr(value, "value", value) + return str(raw or "").strip().lower().replace("-", "_").replace(" ", "_") + + +def _enabled(value: Any) -> bool: + if isinstance(value, str): + return value.strip().lower() == "true" + return bool(value) + + +def _materialization(config: Any) -> models.Materialization: + value = _normalized(_value(config, "materialized")) + return _MATERIALIZATION_MAP.get(value, models.Materialization.OTHER) + + +def _language(node: Any) -> models.Language: + value = _normalized(getattr(node, "language", "sql")) + return _LANGUAGE_MAP.get(value, models.Language.OTHER) + + +def _incremental_strategy(config: Any) -> models.IncrementalStrategy: + value = _normalized(_value(config, "incremental_strategy") or "merge") + return _INCREMENTAL_STRATEGY_MAP.get(value, models.IncrementalStrategy.OTHER) + + +def _python_submission_method(config: Any) -> models.PythonSubmissionMethod: + value = _normalized(_value(config, "submission_method") or "all_purpose_cluster") + return _PYTHON_SUBMISSION_METHOD_MAP.get(value, models.PythonSubmissionMethod.OTHER) + + +def _catalog_type(catalog_relation: Any) -> models.CatalogType: + if catalog_relation is None: + return models.CatalogType.TYPE_UNSPECIFIED + value = _normalized(getattr(catalog_relation, "catalog_type", None)) + return _CATALOG_TYPE_MAP.get(value, models.CatalogType.OTHER) + + +def _storage_format( + catalog_relation: Any, use_managed_iceberg: bool +) -> models.EffectiveStorageFormat: + if catalog_relation is None: + return models.EffectiveStorageFormat.TYPE_UNSPECIFIED + if _normalized(getattr(catalog_relation, "table_format", None)) == "iceberg": + if use_managed_iceberg: + return models.EffectiveStorageFormat.MANAGED_ICEBERG + return models.EffectiveStorageFormat.UNIFORM_ICEBERG + file_format = _normalized(getattr(catalog_relation, "file_format", None)) + if not file_format: + return models.EffectiveStorageFormat.TYPE_UNSPECIFIED + return _FILE_FORMAT_MAP.get(file_format, models.EffectiveStorageFormat.OTHER) + + +def _compute_type(config: Any, creds: DatabricksCredentials) -> models.ComputeType: + compute_name = _value(config, "databricks_compute") + if not compute_name: + return classify_compute_type(getattr(creds, "http_path", None)) + compute = getattr(creds, "compute", None) or {} + compute_config = compute.get(compute_name) + return classify_compute_type(_value(compute_config, "http_path")) + + +def _columns(node: Any) -> list[Any]: + columns = getattr(node, "columns", None) or {} + return list(columns.values()) if isinstance(columns, dict) else list(columns) + + +def _column_extra(column: Any) -> dict: + extra = _value(column, "_extra", {}) + return extra if isinstance(extra, dict) else {} + + +def _constraint_name(constraint: Any) -> str: + return _normalized(_value(constraint, "type")) + + +def _constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: + constraint_names = { + _constraint_name(constraint) for constraint in (getattr(node, "constraints", None) or []) + } + columns = _columns(node) + for column in columns: + constraint_names.update( + _constraint_name(constraint) for constraint in (_value(column, "constraints", []) or []) + ) + + if _enabled(_value(config, "persist_constraints")): + meta = getattr(node, "meta", None) or {} + for constraint in _value(meta, "constraints", []) or []: + constraint_type = _constraint_name(constraint) + constraint_names.add(constraint_type or "check") + for column in columns: + legacy_constraint = _value(_value(column, "meta", {}), "constraint") + if legacy_constraint: + constraint_names.add( + _constraint_name(legacy_constraint) or _normalized(legacy_constraint) + ) + + return { + model_config + for name in constraint_names + if (model_config := _CONSTRAINT_CONFIG_MAP.get(name)) is not None + } + + +def _shared_config_usage(node: Any, config: Any) -> set[models.ModelConfig]: + usage = _constraint_configs(node, config) + auto_liquid_cluster = _enabled(_value(config, "auto_liquid_cluster")) + if _value(config, "liquid_clustered_by") or auto_liquid_cluster: + usage.add(models.ModelConfig.LIQUID_CLUSTERING) + if auto_liquid_cluster: + usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) + if _value(config, "zorder"): + usage.add(models.ModelConfig.ZORDER) + if _value(config, "databricks_tags"): + usage.add(models.ModelConfig.DATABRICKS_RELATION_TAGS) + columns = _columns(node) + if any(_column_extra(column).get("databricks_tags") for column in columns): + usage.add(models.ModelConfig.COLUMN_TAGS) + if any(_column_extra(column).get("column_mask") for column in columns): + usage.add(models.ModelConfig.COLUMN_MASKS) + if _value(config, "row_filter"): + usage.add(models.ModelConfig.ROW_FILTER) + if _value(config, "databricks_compute"): + usage.add(models.ModelConfig.NAMED_COMPUTE_ROUTING) + return usage + + +def _incremental_config_usage(config: Any) -> set[models.ModelConfig]: + usage = set() + if _enabled(_value(config, "merge_with_schema_evolution")): + usage.add(models.ModelConfig.MERGE_SCHEMA_EVOLUTION) + if _value(config, "not_matched_by_source_action"): + usage.add(models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE) + return usage + + +def _catalog_relation(node: Any, builder: Callable[[Any], Any]) -> Any: + try: + return builder(node) + except Exception: + return None + + +def aggregate_model_configs( + manifest: Any, + creds: DatabricksCredentials, + behavior_flag: Callable[[str], bool], + catalog_relation_builder: Callable[[Any], Any], +) -> list[models.ModelConfigStats]: + root_project = getattr(getattr(manifest, "metadata", None), "project_name", None) + accumulators = { + models.ModelConfigScope.ROOT_PROJECT: _ModelConfigAccumulator( + models.ModelConfigScope.ROOT_PROJECT + ), + models.ModelConfigScope.INSTALLED_PACKAGES: _ModelConfigAccumulator( + models.ModelConfigScope.INSTALLED_PACKAGES + ), + } + use_managed_iceberg = bool(behavior_flag("use_managed_iceberg")) + for node in (getattr(manifest, "nodes", None) or {}).values(): + if _resource_type(node) != "model": + continue + scope = ( + models.ModelConfigScope.ROOT_PROJECT + if getattr(node, "package_name", None) == root_project + else models.ModelConfigScope.INSTALLED_PACKAGES + ) + acc = accumulators[scope] + config = getattr(node, "config", None) + materialization = _materialization(config) + language = _language(node) + + acc.model_count += 1 + acc.materializations[materialization] += 1 + acc.languages[language] += 1 + acc.config_usage.update(_shared_config_usage(node, config)) + + if materialization == models.Materialization.INCREMENTAL: + acc.incremental_model_count += 1 + acc.incremental_strategies[_incremental_strategy(config)] += 1 + acc.incremental_config_usage.update(_incremental_config_usage(config)) + + if language == models.Language.PYTHON: + acc.python_model_count += 1 + acc.python_submission_methods[_python_submission_method(config)] += 1 + + if materialization != models.Materialization.EPHEMERAL: + relation = _catalog_relation(node, catalog_relation_builder) + acc.catalog_types[_catalog_type(relation)] += 1 + acc.compute_types[_compute_type(config, creds)] += 1 + if materialization in _STORAGE_FORMAT_MATERIALIZATIONS: + acc.storage_formats[_storage_format(relation, use_managed_iceberg)] += 1 + + return [ + accumulators[models.ModelConfigScope.ROOT_PROJECT].to_model(), + accumulators[models.ModelConfigScope.INSTALLED_PACKAGES].to_model(), + ] + + def ephemeral_resource_ids(manifest: Any) -> set[str]: """Return ephemeral IDs for local counting only.""" result = set() @@ -223,6 +563,7 @@ def build_post_parse_log( config: Any, creds: DatabricksCredentials, behavior_flag: Callable[[str], bool], + catalog_relation_builder: Callable[[Any], Any], ) -> models.TelemetryLog: invocation_id = _invocation_id(manifest) payload = models.PostParsePayload( @@ -230,6 +571,12 @@ def build_post_parse_log( manifest_stats=aggregate_manifest(manifest), connection_config=build_connection_config(creds), project_config=build_project_config(behavior_flag), + model_config_stats=aggregate_model_configs( + manifest, + creds, + behavior_flag, + catalog_relation_builder, + ), ) return models.TelemetryLog( invocation_id=invocation_id, diff --git a/dbt/adapters/databricks/telemetry/hooks.py b/dbt/adapters/databricks/telemetry/hooks.py index e711967fc..7ad1c9ff7 100644 --- a/dbt/adapters/databricks/telemetry/hooks.py +++ b/dbt/adapters/databricks/telemetry/hooks.py @@ -68,6 +68,7 @@ def on_post_parse(adapter: Any, manifest: Any) -> None: config=config, creds=creds, behavior_flag=adapter.get_behavior_flag_no_warn, + catalog_relation_builder=adapter.build_catalog_relation, ) if not log.invocation_id: return diff --git a/dbt/adapters/databricks/telemetry/models.py b/dbt/adapters/databricks/telemetry/models.py index f84549c00..a93ce7842 100644 --- a/dbt/adapters/databricks/telemetry/models.py +++ b/dbt/adapters/databricks/telemetry/models.py @@ -82,6 +82,87 @@ class TerminationReason(Enum): INTERNAL_ERROR = "INTERNAL_ERROR" +class ModelConfigScope(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + ROOT_PROJECT = "ROOT_PROJECT" + INSTALLED_PACKAGES = "INSTALLED_PACKAGES" + + +class Materialization(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + TABLE = "TABLE" + VIEW = "VIEW" + INCREMENTAL = "INCREMENTAL" + EPHEMERAL = "EPHEMERAL" + MATERIALIZED_VIEW = "MATERIALIZED_VIEW" + STREAMING_TABLE = "STREAMING_TABLE" + METRIC_VIEW = "METRIC_VIEW" + OTHER = "OTHER" + + +class Language(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + SQL = "SQL" + PYTHON = "PYTHON" + OTHER = "OTHER" + + +class IncrementalStrategy(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + MERGE = "MERGE" + APPEND = "APPEND" + DELETE_INSERT = "DELETE_INSERT" + INSERT_OVERWRITE = "INSERT_OVERWRITE" + REPLACE_WHERE = "REPLACE_WHERE" + MICROBATCH = "MICROBATCH" + OTHER = "OTHER" + + +class EffectiveStorageFormat(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + DELTA = "DELTA" + MANAGED_ICEBERG = "MANAGED_ICEBERG" + UNIFORM_ICEBERG = "UNIFORM_ICEBERG" + PARQUET = "PARQUET" + HUDI = "HUDI" + OTHER = "OTHER" + + +class CatalogType(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + UNITY_CATALOG = "UNITY_CATALOG" + HIVE_METASTORE = "HIVE_METASTORE" + OTHER = "OTHER" + + +class PythonSubmissionMethod(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + SERVERLESS_CLUSTER = "SERVERLESS_CLUSTER" + JOB_CLUSTER = "JOB_CLUSTER" + ALL_PURPOSE_CLUSTER = "ALL_PURPOSE_CLUSTER" + WORKFLOW_JOB = "WORKFLOW_JOB" + OTHER = "OTHER" + + +class ModelConfig(Enum): + TYPE_UNSPECIFIED = "TYPE_UNSPECIFIED" + LIQUID_CLUSTERING = "LIQUID_CLUSTERING" + AUTO_LIQUID_CLUSTERING = "AUTO_LIQUID_CLUSTERING" + ZORDER = "ZORDER" + DATABRICKS_RELATION_TAGS = "DATABRICKS_RELATION_TAGS" + COLUMN_TAGS = "COLUMN_TAGS" + COLUMN_MASKS = "COLUMN_MASKS" + ROW_FILTER = "ROW_FILTER" + NOT_NULL_CONSTRAINT = "NOT_NULL_CONSTRAINT" + CHECK_CONSTRAINT = "CHECK_CONSTRAINT" + PRIMARY_KEY_CONSTRAINT = "PRIMARY_KEY_CONSTRAINT" + FOREIGN_KEY_CONSTRAINT = "FOREIGN_KEY_CONSTRAINT" + CUSTOM_CONSTRAINT = "CUSTOM_CONSTRAINT" + NAMED_COMPUTE_ROUTING = "NAMED_COMPUTE_ROUTING" + MERGE_SCHEMA_EVOLUTION = "MERGE_SCHEMA_EVOLUTION" + MERGE_NOT_MATCHED_BY_SOURCE = "MERGE_NOT_MATCHED_BY_SOURCE" + + @dataclass class ResourceCounts: model_count: int = 0 @@ -133,12 +214,88 @@ class ProjectConfig: use_describe_as_json_for_relation_metadata: bool = False +@dataclass +class ModelConfigUsage: + config: ModelConfig = ModelConfig.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class MaterializationCount: + materialization: Materialization = Materialization.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class LanguageCount: + language: Language = Language.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class IncrementalStrategyCount: + incremental_strategy: IncrementalStrategy = IncrementalStrategy.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class EffectiveStorageFormatCount: + effective_storage_format: EffectiveStorageFormat = EffectiveStorageFormat.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class CatalogTypeCount: + catalog_type: CatalogType = CatalogType.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class ComputeTypeCount: + compute_type: ComputeType = ComputeType.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class PythonSubmissionMethodCount: + submission_method: PythonSubmissionMethod = PythonSubmissionMethod.TYPE_UNSPECIFIED + count: int = 0 + + +@dataclass +class IncrementalModelStats: + model_count: int = 0 + strategy_counts: list[IncrementalStrategyCount] = field(default_factory=list) + config_usage: list[ModelConfigUsage] = field(default_factory=list) + + +@dataclass +class PythonModelStats: + model_count: int = 0 + submission_method_counts: list[PythonSubmissionMethodCount] = field(default_factory=list) + + +@dataclass +class ModelConfigStats: + scope: ModelConfigScope = ModelConfigScope.TYPE_UNSPECIFIED + model_count: int = 0 + materialization_counts: list[MaterializationCount] = field(default_factory=list) + language_counts: list[LanguageCount] = field(default_factory=list) + incremental_model_stats: IncrementalModelStats = field(default_factory=IncrementalModelStats) + effective_storage_format_counts: list[EffectiveStorageFormatCount] = field(default_factory=list) + catalog_type_counts: list[CatalogTypeCount] = field(default_factory=list) + effective_compute_type_counts: list[ComputeTypeCount] = field(default_factory=list) + python_model_stats: PythonModelStats = field(default_factory=PythonModelStats) + config_usage: list[ModelConfigUsage] = field(default_factory=list) + + @dataclass class PostParsePayload: invocation_config: InvocationConfig manifest_stats: ManifestStats connection_config: ConnectionConfig project_config: ProjectConfig + model_config_stats: list[ModelConfigStats] = field(default_factory=list) @dataclass diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index f3cb959f0..9d40e9866 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -28,6 +28,26 @@ def _node(resource_type, package_name="root", test_metadata=None): ) +def _model( + materialized, + *, + package_name="root", + language="sql", + columns=None, + constraints=None, + **config, +): + return SimpleNamespace( + resource_type="model", + package_name=package_name, + language=language, + config={"materialized": materialized, **config}, + columns=columns or {}, + constraints=constraints or [], + meta={}, + ) + + class TestReportedClassifications: @pytest.mark.parametrize( "http_path, expected", @@ -156,6 +176,199 @@ def test_root_installed_and_test_kinds(self): assert ms.enabled_total.other_count == 1 +class TestAggregateModelConfigs: + def test_aggregates_scopes_defaults_and_config_adoption(self): + columns = { + "id": { + "constraints": [{"type": "not_null"}], + "_extra": { + "databricks_tags": {"sensitivity": "high"}, + "column_mask": {"function": "mask_id"}, + }, + } + } + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "table": _model( + "table", + liquid_clustered_by=["id"], + auto_liquid_cluster=True, + zorder=["id"], + databricks_tags={"team": "data"}, + row_filter={"function": "filter_id", "columns": ["id"]}, + columns=columns, + constraints=[ + {"type": "check"}, + {"type": "primary_key"}, + {"type": "foreign_key"}, + {"type": "custom"}, + ], + ), + "incremental": _model( + "incremental", + table_format="iceberg", + databricks_compute="cluster", + merge_with_schema_evolution=True, + not_matched_by_source_action="delete", + ), + "python": _model( + "view", + language="python", + submission_method="serverless_cluster", + ), + "dependency": _model("ephemeral", package_name="package"), + "test": _node("test"), + }, + ) + creds = _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}) + + def build_relation(node): + config = node.config + return SimpleNamespace( + catalog_type="unity", + table_format=config.get("table_format", "default"), + file_format=config.get("file_format", "delta"), + ) + + stats = builder.aggregate_model_configs( + manifest, + creds, + lambda flag: False, + build_relation, + ) + root, installed = stats + + assert root.scope == models.ModelConfigScope.ROOT_PROJECT + assert root.model_count == 3 + assert {row.materialization: row.count for row in root.materialization_counts} == { + models.Materialization.TABLE: 1, + models.Materialization.VIEW: 1, + models.Materialization.INCREMENTAL: 1, + } + assert {row.language: row.count for row in root.language_counts} == { + models.Language.SQL: 2, + models.Language.PYTHON: 1, + } + assert root.incremental_model_stats.model_count == 1 + assert root.incremental_model_stats.strategy_counts == [ + models.IncrementalStrategyCount(models.IncrementalStrategy.MERGE, 1) + ] + assert {row.config: row.count for row in root.incremental_model_stats.config_usage} == { + models.ModelConfig.MERGE_SCHEMA_EVOLUTION: 1, + models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE: 1, + } + assert { + row.effective_storage_format: row.count for row in root.effective_storage_format_counts + } == { + models.EffectiveStorageFormat.DELTA: 1, + models.EffectiveStorageFormat.UNIFORM_ICEBERG: 1, + } + assert root.catalog_type_counts == [ + models.CatalogTypeCount(models.CatalogType.UNITY_CATALOG, 3) + ] + assert {row.compute_type: row.count for row in root.effective_compute_type_counts} == { + models.ComputeType.SQL_WAREHOUSE: 2, + models.ComputeType.ALL_PURPOSE_CLUSTER: 1, + } + assert root.python_model_stats == models.PythonModelStats( + model_count=1, + submission_method_counts=[ + models.PythonSubmissionMethodCount( + models.PythonSubmissionMethod.SERVERLESS_CLUSTER, 1 + ) + ], + ) + assert {row.config: row.count for row in root.config_usage} == { + config: 1 + for config in ( + models.ModelConfig.LIQUID_CLUSTERING, + models.ModelConfig.AUTO_LIQUID_CLUSTERING, + models.ModelConfig.ZORDER, + models.ModelConfig.DATABRICKS_RELATION_TAGS, + models.ModelConfig.COLUMN_TAGS, + models.ModelConfig.COLUMN_MASKS, + models.ModelConfig.ROW_FILTER, + models.ModelConfig.NOT_NULL_CONSTRAINT, + models.ModelConfig.CHECK_CONSTRAINT, + models.ModelConfig.PRIMARY_KEY_CONSTRAINT, + models.ModelConfig.FOREIGN_KEY_CONSTRAINT, + models.ModelConfig.CUSTOM_CONSTRAINT, + models.ModelConfig.NAMED_COMPUTE_ROUTING, + ) + } + + assert installed.scope == models.ModelConfigScope.INSTALLED_PACKAGES + assert installed.model_count == 1 + assert installed.materialization_counts == [ + models.MaterializationCount(models.Materialization.EPHEMERAL, 1) + ] + assert installed.effective_storage_format_counts == [] + assert installed.catalog_type_counts == [] + assert installed.effective_compute_type_counts == [] + + def test_managed_iceberg_and_unresolved_named_compute(self): + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "iceberg": _model("table", table_format="iceberg", databricks_compute="missing") + }, + ) + relation = SimpleNamespace( + catalog_type="hive_metastore", table_format="iceberg", file_format="delta" + ) + + root = builder.aggregate_model_configs( + manifest, + _creds(compute={}), + lambda flag: flag == "use_managed_iceberg", + lambda node: relation, + )[0] + + assert root.effective_storage_format_counts == [ + models.EffectiveStorageFormatCount(models.EffectiveStorageFormat.MANAGED_ICEBERG, 1) + ] + assert root.catalog_type_counts == [ + models.CatalogTypeCount(models.CatalogType.HIVE_METASTORE, 1) + ] + assert root.effective_compute_type_counts == [ + models.ComputeTypeCount(models.ComputeType.TYPE_UNSPECIFIED, 1) + ] + + def test_legacy_constraints_require_persist_constraints(self): + legacy = _model( + "table", + persist_constraints=True, + columns={"id": {"meta": {"constraint": "not_null"}}}, + ) + legacy.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + ignored = _model( + "table", + persist_constraints=False, + columns={"id": {"meta": {"constraint": "not_null"}}}, + ) + ignored.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"legacy": legacy, "ignored": ignored}, + ) + relation = SimpleNamespace( + catalog_type="unity", table_format="default", file_format="delta" + ) + + root = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag: False, + lambda node: relation, + )[0] + + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + models.ModelConfig.CHECK_CONSTRAINT: 1, + } + + class TestBuildPostRunLog: @pytest.mark.parametrize( "exc_type, results, fail_fast, task_success, status, reason", diff --git a/tests/unit/telemetry/test_encoder.py b/tests/unit/telemetry/test_encoder.py index 397d36b65..abfc74c3b 100644 --- a/tests/unit/telemetry/test_encoder.py +++ b/tests/unit/telemetry/test_encoder.py @@ -20,6 +20,15 @@ def _log(): configured_auth_family=models.AuthFamily.PAT, ), project_config=models.ProjectConfig(use_materialization_v2=True), + model_config_stats=[ + models.ModelConfigStats( + scope=models.ModelConfigScope.ROOT_PROJECT, + model_count=1, + materialization_counts=[ + models.MaterializationCount(models.Materialization.TABLE, 1) + ], + ) + ], ), ) @@ -65,6 +74,9 @@ def test_post_parse_envelope(self): assert "post_run" not in entry assert entry["post_parse"]["invocation_config"]["dbt_command"] == "RUN" assert entry["post_parse"]["connection_config"]["default_compute_type"] == "SQL_WAREHOUSE" + model_stats = entry["post_parse"]["model_config_stats"][0] + assert model_stats["scope"] == "ROOT_PROJECT" + assert model_stats["materialization_counts"] == [{"materialization": "TABLE", "count": 1}] @pytest.mark.parametrize( "workspace_id, expected", diff --git a/tests/unit/telemetry/test_hooks.py b/tests/unit/telemetry/test_hooks.py index 37cdd7888..77b9471fb 100644 --- a/tests/unit/telemetry/test_hooks.py +++ b/tests/unit/telemetry/test_hooks.py @@ -70,6 +70,24 @@ def test_hook_exceptions_do_not_escape(monkeypatch): hooks.on_post_parse(adapter, SimpleNamespace()) # must not raise +def test_post_parse_passes_catalog_relation_builder(monkeypatch): + coord = _enable_hooks(monkeypatch) + coord.needs_post_parse.return_value = True + build = Mock(return_value=_parse_log()) + monkeypatch.setattr(hooks.builder, "build_post_parse_log", build) + adapter = SimpleNamespace( + config=SimpleNamespace(credentials=SimpleNamespace()), + get_behavior_flag_no_warn=lambda _: False, + build_catalog_relation=Mock(), + ) + manifest = SimpleNamespace() + + hooks.on_post_parse(adapter, manifest) + + assert build.call_args.kwargs["catalog_relation_builder"] is adapter.build_catalog_relation + coord.set_post_parse.assert_called_once() + + def test_run_end_exception_finalizes_stored_invocation_not_current_global(monkeypatch): coord = Coordinator() logs = [] From 759833e8c37fad3b9a5c44456b89c4678f1561c2 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Mon, 7 Sep 2026 05:52:30 +0000 Subject: [PATCH 02/13] Add telemetry changelog entry --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0382be37e..cbb6853a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - Add opt-in POST_PARSE invocation telemetry for eligible commands via `connection_parameters.enable_dbt_telemetry` ([#1647](https://github.com/databricks/dbt-databricks/pull/1647)) - Add POST_RUN outcome telemetry to the opt-in invocation telemetry path ([#1648](https://github.com/databricks/dbt-databricks/pull/1648)) +- Add aggregate model configuration statistics to the opt-in POST_PARSE telemetry event ([#1665](https://github.com/databricks/dbt-databricks/pull/1665)) ### Fixes From 92d533673b2abdfb67aff0440037f355154b8123 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Mon, 7 Sep 2026 08:10:15 +0000 Subject: [PATCH 03/13] Align model config telemetry schema --- dbt/adapters/databricks/telemetry/builder.py | 8 +------- dbt/adapters/databricks/telemetry/models.py | 1 - tests/unit/telemetry/test_builder.py | 6 ++---- 3 files changed, 3 insertions(+), 12 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index e2f9e21aa..224e65c12 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -100,7 +100,6 @@ class _ModelConfigAccumulator: languages: Counter = field(default_factory=Counter) incremental_model_count: int = 0 incremental_strategies: Counter = field(default_factory=Counter) - incremental_config_usage: Counter = field(default_factory=Counter) storage_formats: Counter = field(default_factory=Counter) catalog_types: Counter = field(default_factory=Counter) compute_types: Counter = field(default_factory=Counter) @@ -123,11 +122,6 @@ def to_model(self) -> models.ModelConfigStats: models.IncrementalStrategy, models.IncrementalStrategyCount, ), - config_usage=_count_rows( - self.incremental_config_usage, - models.ModelConfig, - models.ModelConfigUsage, - ), ), effective_storage_format_counts=_count_rows( self.storage_formats, @@ -469,7 +463,7 @@ def aggregate_model_configs( if materialization == models.Materialization.INCREMENTAL: acc.incremental_model_count += 1 acc.incremental_strategies[_incremental_strategy(config)] += 1 - acc.incremental_config_usage.update(_incremental_config_usage(config)) + acc.config_usage.update(_incremental_config_usage(config)) if language == models.Language.PYTHON: acc.python_model_count += 1 diff --git a/dbt/adapters/databricks/telemetry/models.py b/dbt/adapters/databricks/telemetry/models.py index a93ce7842..c1ab7f398 100644 --- a/dbt/adapters/databricks/telemetry/models.py +++ b/dbt/adapters/databricks/telemetry/models.py @@ -266,7 +266,6 @@ class PythonSubmissionMethodCount: class IncrementalModelStats: model_count: int = 0 strategy_counts: list[IncrementalStrategyCount] = field(default_factory=list) - config_usage: list[ModelConfigUsage] = field(default_factory=list) @dataclass diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 9d40e9866..3b95571d7 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -254,10 +254,6 @@ def build_relation(node): assert root.incremental_model_stats.strategy_counts == [ models.IncrementalStrategyCount(models.IncrementalStrategy.MERGE, 1) ] - assert {row.config: row.count for row in root.incremental_model_stats.config_usage} == { - models.ModelConfig.MERGE_SCHEMA_EVOLUTION: 1, - models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE: 1, - } assert { row.effective_storage_format: row.count for row in root.effective_storage_format_counts } == { @@ -295,6 +291,8 @@ def build_relation(node): models.ModelConfig.FOREIGN_KEY_CONSTRAINT, models.ModelConfig.CUSTOM_CONSTRAINT, models.ModelConfig.NAMED_COMPUTE_ROUTING, + models.ModelConfig.MERGE_SCHEMA_EVOLUTION, + models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE, ) } From 42ee82213bc24cc6f2e258bfc4da68ed7f8b3ada Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Mon, 7 Sep 2026 21:38:14 +0530 Subject: [PATCH 04/13] fix: classify HMS catalogs and count only effective model config adoption --- dbt/adapters/databricks/telemetry/builder.py | 98 +++++++++++++++++--- tests/unit/telemetry/test_builder.py | 61 +++++++++++- 2 files changed, 144 insertions(+), 15 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index 224e65c12..681cd459b 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -91,6 +91,56 @@ models.Materialization.INCREMENTAL, } +_HMS_CATALOG_NAMES = {"hive_metastore"} + +_TAG_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, + models.Materialization.VIEW, + models.Materialization.MATERIALIZED_VIEW, + models.Materialization.STREAMING_TABLE, + models.Materialization.METRIC_VIEW, +} +_COLUMN_TAG_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, + models.Materialization.VIEW, + models.Materialization.MATERIALIZED_VIEW, + models.Materialization.STREAMING_TABLE, +} +_LIQUID_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, + models.Materialization.MATERIALIZED_VIEW, + models.Materialization.STREAMING_TABLE, +} +_ZORDER_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, +} +_MASK_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, +} +_ROW_FILTER_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, + models.Materialization.MATERIALIZED_VIEW, + models.Materialization.STREAMING_TABLE, +} +_CONSTRAINT_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, +} +_NAMED_COMPUTE_MATERIALIZATIONS = { + models.Materialization.TABLE, + models.Materialization.INCREMENTAL, + models.Materialization.VIEW, + models.Materialization.MATERIALIZED_VIEW, + models.Materialization.STREAMING_TABLE, + models.Materialization.METRIC_VIEW, +} + @dataclass class _ModelConfigAccumulator: @@ -313,7 +363,16 @@ def _python_submission_method(config: Any) -> models.PythonSubmissionMethod: return _PYTHON_SUBMISSION_METHOD_MAP.get(value, models.PythonSubmissionMethod.OTHER) -def _catalog_type(catalog_relation: Any) -> models.CatalogType: +def _catalog_type(catalog_relation: Any, node: Any) -> models.CatalogType: + # Physical hive_metastore is HMS even when the default Unity integration + # supplies catalog_type="unity". + physical = _normalized( + getattr(catalog_relation, "catalog_name", None) if catalog_relation is not None else None + ) + if not physical: + physical = _normalized(getattr(node, "database", None)) + if physical in _HMS_CATALOG_NAMES: + return models.CatalogType.HIVE_METASTORE if catalog_relation is None: return models.CatalogType.TYPE_UNSPECIFIED value = _normalized(getattr(catalog_relation, "catalog_type", None)) @@ -387,25 +446,36 @@ def _constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: } -def _shared_config_usage(node: Any, config: Any) -> set[models.ModelConfig]: - usage = _constraint_configs(node, config) +def _shared_config_usage( + node: Any, config: Any, materialization: models.Materialization +) -> set[models.ModelConfig]: + usage: set[models.ModelConfig] = set() + if materialization in _CONSTRAINT_MATERIALIZATIONS: + usage.update(_constraint_configs(node, config)) auto_liquid_cluster = _enabled(_value(config, "auto_liquid_cluster")) - if _value(config, "liquid_clustered_by") or auto_liquid_cluster: + has_liquid = bool(_value(config, "liquid_clustered_by") or auto_liquid_cluster) + if has_liquid and materialization in _LIQUID_MATERIALIZATIONS: usage.add(models.ModelConfig.LIQUID_CLUSTERING) - if auto_liquid_cluster: - usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) - if _value(config, "zorder"): + if auto_liquid_cluster: + usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) + if _value(config, "zorder") and materialization in _ZORDER_MATERIALIZATIONS and not has_liquid: usage.add(models.ModelConfig.ZORDER) - if _value(config, "databricks_tags"): + if _value(config, "databricks_tags") and materialization in _TAG_MATERIALIZATIONS: usage.add(models.ModelConfig.DATABRICKS_RELATION_TAGS) columns = _columns(node) - if any(_column_extra(column).get("databricks_tags") for column in columns): + if ( + any(_column_extra(column).get("databricks_tags") for column in columns) + and materialization in _COLUMN_TAG_MATERIALIZATIONS + ): usage.add(models.ModelConfig.COLUMN_TAGS) - if any(_column_extra(column).get("column_mask") for column in columns): + if ( + any(_column_extra(column).get("column_mask") for column in columns) + and materialization in _MASK_MATERIALIZATIONS + ): usage.add(models.ModelConfig.COLUMN_MASKS) - if _value(config, "row_filter"): + if _value(config, "row_filter") and materialization in _ROW_FILTER_MATERIALIZATIONS: usage.add(models.ModelConfig.ROW_FILTER) - if _value(config, "databricks_compute"): + if _value(config, "databricks_compute") and materialization in _NAMED_COMPUTE_MATERIALIZATIONS: usage.add(models.ModelConfig.NAMED_COMPUTE_ROUTING) return usage @@ -458,7 +528,7 @@ def aggregate_model_configs( acc.model_count += 1 acc.materializations[materialization] += 1 acc.languages[language] += 1 - acc.config_usage.update(_shared_config_usage(node, config)) + acc.config_usage.update(_shared_config_usage(node, config, materialization)) if materialization == models.Materialization.INCREMENTAL: acc.incremental_model_count += 1 @@ -471,7 +541,7 @@ def aggregate_model_configs( if materialization != models.Materialization.EPHEMERAL: relation = _catalog_relation(node, catalog_relation_builder) - acc.catalog_types[_catalog_type(relation)] += 1 + acc.catalog_types[_catalog_type(relation, node)] += 1 acc.compute_types[_compute_type(config, creds)] += 1 if materialization in _STORAGE_FORMAT_MATERIALIZATIONS: acc.storage_formats[_storage_format(relation, use_managed_iceberg)] += 1 diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 3b95571d7..8563d0851 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -3,6 +3,8 @@ import pytest from dbt_common.exceptions import DbtRuntimeError +from dbt.adapters.databricks import constants +from dbt.adapters.databricks.catalogs._unity import UnityCatalogIntegration from dbt.adapters.databricks.telemetry import builder, models @@ -33,6 +35,7 @@ def _model( *, package_name="root", language="sql", + database="main", columns=None, constraints=None, **config, @@ -41,6 +44,7 @@ def _model( resource_type="model", package_name=package_name, language=language, + database=database, config={"materialized": materialized, **config}, columns=columns or {}, constraints=constraints or [], @@ -280,7 +284,6 @@ def build_relation(node): for config in ( models.ModelConfig.LIQUID_CLUSTERING, models.ModelConfig.AUTO_LIQUID_CLUSTERING, - models.ModelConfig.ZORDER, models.ModelConfig.DATABRICKS_RELATION_TAGS, models.ModelConfig.COLUMN_TAGS, models.ModelConfig.COLUMN_MASKS, @@ -366,6 +369,62 @@ def test_legacy_constraints_require_persist_constraints(self): models.ModelConfig.CHECK_CONSTRAINT: 1, } + def test_physical_hive_metastore_is_classified_as_hms(self): + node = _model("table", database="hive_metastore") + node.schema = "dbt" + node.identifier = "hms" + relation = UnityCatalogIntegration(constants.DEFAULT_UNITY_CATALOG).build_relation(node) + assert relation.catalog_type == "unity" + assert relation.catalog_name == "hive_metastore" + + root = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"hms": node}, + ), + _creds(), + lambda flag: False, + lambda _: relation, + )[0] + + assert root.catalog_type_counts == [ + models.CatalogTypeCount(models.CatalogType.HIVE_METASTORE, 1) + ] + + def test_config_usage_follows_runtime_applicability(self): + columns = {"id": {"_extra": {"column_mask": {"function": "mask_id"}}}} + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "view_zorder": _model( + "view", + zorder=["id"], + liquid_clustered_by=["id"], + columns=columns, + row_filter={"function": "f", "columns": ["id"]}, + ), + "table_both": _model("table", zorder=["id"], liquid_clustered_by=["id"]), + "table_zorder": _model("table", zorder=["id"]), + "ephemeral_compute": _model( + "ephemeral", databricks_compute="cluster", zorder=["id"] + ), + }, + ) + + root = builder.aggregate_model_configs( + manifest, + _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}), + lambda flag: False, + lambda node: SimpleNamespace( + catalog_type="unity", table_format="default", file_format="delta" + ), + )[0] + + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.LIQUID_CLUSTERING: 1, + models.ModelConfig.ZORDER: 1, + } + class TestBuildPostRunLog: @pytest.mark.parametrize( From 25418831c602939ca56ac807029115dc1c17fc71 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Mon, 7 Sep 2026 21:51:39 +0530 Subject: [PATCH 05/13] test: keep telemetry tests pointed at independent contracts Drop kitchen-sink and kwargs-forwarding coverage that restated collector maps, and retain cases that can fail without a matching production change. --- tests/unit/telemetry/test_builder.py | 247 +++++++++------------------ tests/unit/telemetry/test_config.py | 4 - tests/unit/telemetry/test_encoder.py | 2 +- tests/unit/telemetry/test_hooks.py | 18 -- 4 files changed, 84 insertions(+), 187 deletions(-) diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 8563d0851..bbd75c582 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -52,6 +52,10 @@ def _model( ) +def _unity_delta_relation(_node=None): + return SimpleNamespace(catalog_type="unity", table_format="default", file_format="delta") + + class TestReportedClassifications: @pytest.mark.parametrize( "http_path, expected", @@ -81,17 +85,17 @@ def test_compute_type(self, http_path, expected): @pytest.mark.parametrize( "creds, expected", [ - pytest.param(_creds(token="dapi"), models.AuthFamily.PAT, id="pat"), + pytest.param( + _creds(token="dapi", azure_client_id="a", azure_client_secret="b"), + models.AuthFamily.PAT, + id="token_wins_over_azure_sp", + ), pytest.param( _creds(azure_client_id="a", azure_client_secret="b"), models.AuthFamily.AZURE_SERVICE_PRINCIPAL, id="azure_sp", ), - pytest.param( - _creds(auth_type="oauth"), - models.AuthFamily.OAUTH_U2M, - id="u2m", - ), + pytest.param(_creds(), models.AuthFamily.OAUTH_U2M, id="no_secret_is_u2m"), pytest.param( _creds(client_id="c", client_secret="s"), models.AuthFamily.LEGACY_CLIENT_SECRET_AMBIGUOUS, @@ -154,183 +158,96 @@ def test_named_compute_o_parameter_sets_spog_flag(self): class TestAggregateManifest: - def test_root_installed_and_test_kinds(self): + def test_generic_tests_are_split_from_singular(self): manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root", invocation_id="inv-1"), + metadata=SimpleNamespace(project_name="root"), nodes={ - "m1": _node("model", "root"), - "m2": _node("model", "dep_pkg"), - "t_generic": _node("test", "root", test_metadata={"name": "not_null"}), - "t_singular": _node("test", "root"), - "op": _node("operation", "root"), + "t_generic": _node("test", test_metadata={"name": "not_null"}), + "t_singular": _node("test"), }, - sources={}, - exposures={}, - metrics={}, - saved_queries={}, - functions={}, - semantic_models={}, - unit_tests={}, ) ms = builder.aggregate_manifest(manifest) - assert ms.enabled_root_project.model_count == 1 - assert ms.enabled_installed_packages.model_count == 1 assert ms.enabled_total.generic_data_test_count == 1 assert ms.enabled_total.data_test_count == 2 - assert ms.enabled_total.other_count == 1 class TestAggregateModelConfigs: - def test_aggregates_scopes_defaults_and_config_adoption(self): - columns = { - "id": { - "constraints": [{"type": "not_null"}], - "_extra": { - "databricks_tags": {"sensitivity": "high"}, - "column_mask": {"function": "mask_id"}, + def test_package_models_are_not_folded_into_root(self): + root, installed = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "root_model": _model("table"), + "pkg_model": _model("view", package_name="pkg"), }, - } - } - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={ - "table": _model( - "table", - liquid_clustered_by=["id"], - auto_liquid_cluster=True, - zorder=["id"], - databricks_tags={"team": "data"}, - row_filter={"function": "filter_id", "columns": ["id"]}, - columns=columns, - constraints=[ - {"type": "check"}, - {"type": "primary_key"}, - {"type": "foreign_key"}, - {"type": "custom"}, - ], - ), - "incremental": _model( - "incremental", - table_format="iceberg", - databricks_compute="cluster", - merge_with_schema_evolution=True, - not_matched_by_source_action="delete", - ), - "python": _model( - "view", - language="python", - submission_method="serverless_cluster", - ), - "dependency": _model("ephemeral", package_name="package"), - "test": _node("test"), - }, + ), + _creds(), + lambda flag: False, + _unity_delta_relation, ) - creds = _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}) - - def build_relation(node): - config = node.config - return SimpleNamespace( - catalog_type="unity", - table_format=config.get("table_format", "default"), - file_format=config.get("file_format", "delta"), - ) + assert [row.materialization for row in root.materialization_counts] == [ + models.Materialization.TABLE + ] + assert [row.materialization for row in installed.materialization_counts] == [ + models.Materialization.VIEW + ] - stats = builder.aggregate_model_configs( - manifest, - creds, + def test_incremental_defaults_to_merge_and_resolves_named_compute(self): + root = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"inc": _model("incremental", databricks_compute="cluster")}, + ), + _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}), lambda flag: False, - build_relation, - ) - root, installed = stats - - assert root.scope == models.ModelConfigScope.ROOT_PROJECT - assert root.model_count == 3 - assert {row.materialization: row.count for row in root.materialization_counts} == { - models.Materialization.TABLE: 1, - models.Materialization.VIEW: 1, - models.Materialization.INCREMENTAL: 1, - } - assert {row.language: row.count for row in root.language_counts} == { - models.Language.SQL: 2, - models.Language.PYTHON: 1, - } - assert root.incremental_model_stats.model_count == 1 + _unity_delta_relation, + )[0] assert root.incremental_model_stats.strategy_counts == [ models.IncrementalStrategyCount(models.IncrementalStrategy.MERGE, 1) ] - assert { - row.effective_storage_format: row.count for row in root.effective_storage_format_counts - } == { - models.EffectiveStorageFormat.DELTA: 1, - models.EffectiveStorageFormat.UNIFORM_ICEBERG: 1, - } - assert root.catalog_type_counts == [ - models.CatalogTypeCount(models.CatalogType.UNITY_CATALOG, 3) + assert root.effective_compute_type_counts == [ + models.ComputeTypeCount(models.ComputeType.ALL_PURPOSE_CLUSTER, 1) ] - assert {row.compute_type: row.count for row in root.effective_compute_type_counts} == { - models.ComputeType.SQL_WAREHOUSE: 2, - models.ComputeType.ALL_PURPOSE_CLUSTER: 1, - } - assert root.python_model_stats == models.PythonModelStats( - model_count=1, - submission_method_counts=[ - models.PythonSubmissionMethodCount( - models.PythonSubmissionMethod.SERVERLESS_CLUSTER, 1 - ) - ], - ) - assert {row.config: row.count for row in root.config_usage} == { - config: 1 - for config in ( - models.ModelConfig.LIQUID_CLUSTERING, - models.ModelConfig.AUTO_LIQUID_CLUSTERING, - models.ModelConfig.DATABRICKS_RELATION_TAGS, - models.ModelConfig.COLUMN_TAGS, - models.ModelConfig.COLUMN_MASKS, - models.ModelConfig.ROW_FILTER, - models.ModelConfig.NOT_NULL_CONSTRAINT, - models.ModelConfig.CHECK_CONSTRAINT, - models.ModelConfig.PRIMARY_KEY_CONSTRAINT, - models.ModelConfig.FOREIGN_KEY_CONSTRAINT, - models.ModelConfig.CUSTOM_CONSTRAINT, - models.ModelConfig.NAMED_COMPUTE_ROUTING, - models.ModelConfig.MERGE_SCHEMA_EVOLUTION, - models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE, - ) - } - assert installed.scope == models.ModelConfigScope.INSTALLED_PACKAGES - assert installed.model_count == 1 - assert installed.materialization_counts == [ - models.MaterializationCount(models.Materialization.EPHEMERAL, 1) + def test_python_model_defaults_to_all_purpose_submission(self): + root = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"py": _model("table", language="python")}, + ), + _creds(), + lambda flag: False, + _unity_delta_relation, + )[0] + assert root.python_model_stats.submission_method_counts == [ + models.PythonSubmissionMethodCount(models.PythonSubmissionMethod.ALL_PURPOSE_CLUSTER, 1) ] - assert installed.effective_storage_format_counts == [] - assert installed.catalog_type_counts == [] - assert installed.effective_compute_type_counts == [] - def test_managed_iceberg_and_unresolved_named_compute(self): + @pytest.mark.parametrize( + "use_managed, expected_format", + [ + pytest.param(False, models.EffectiveStorageFormat.UNIFORM_ICEBERG, id="uniform"), + pytest.param(True, models.EffectiveStorageFormat.MANAGED_ICEBERG, id="managed"), + ], + ) + def test_iceberg_storage_and_unresolved_named_compute(self, use_managed, expected_format): manifest = SimpleNamespace( metadata=SimpleNamespace(project_name="root"), nodes={ "iceberg": _model("table", table_format="iceberg", databricks_compute="missing") }, ) - relation = SimpleNamespace( - catalog_type="hive_metastore", table_format="iceberg", file_format="delta" - ) + relation = SimpleNamespace(table_format="iceberg", file_format="delta") root = builder.aggregate_model_configs( manifest, _creds(compute={}), - lambda flag: flag == "use_managed_iceberg", + lambda flag: use_managed if flag == "use_managed_iceberg" else False, lambda node: relation, )[0] assert root.effective_storage_format_counts == [ - models.EffectiveStorageFormatCount(models.EffectiveStorageFormat.MANAGED_ICEBERG, 1) - ] - assert root.catalog_type_counts == [ - models.CatalogTypeCount(models.CatalogType.HIVE_METASTORE, 1) + models.EffectiveStorageFormatCount(expected_format, 1) ] assert root.effective_compute_type_counts == [ models.ComputeTypeCount(models.ComputeType.TYPE_UNSPECIFIED, 1) @@ -353,15 +270,12 @@ def test_legacy_constraints_require_persist_constraints(self): metadata=SimpleNamespace(project_name="root"), nodes={"legacy": legacy, "ignored": ignored}, ) - relation = SimpleNamespace( - catalog_type="unity", table_format="default", file_format="delta" - ) root = builder.aggregate_model_configs( manifest, _creds(), lambda flag: False, - lambda node: relation, + _unity_delta_relation, )[0] assert {row.config: row.count for row in root.config_usage} == { @@ -405,6 +319,16 @@ def test_config_usage_follows_runtime_applicability(self): ), "table_both": _model("table", zorder=["id"], liquid_clustered_by=["id"]), "table_zorder": _model("table", zorder=["id"]), + "table_merge": _model( + "table", + merge_with_schema_evolution=True, + not_matched_by_source_action="delete", + ), + "incremental_merge": _model( + "incremental", + merge_with_schema_evolution=True, + not_matched_by_source_action="delete", + ), "ephemeral_compute": _model( "ephemeral", databricks_compute="cluster", zorder=["id"] ), @@ -415,30 +339,25 @@ def test_config_usage_follows_runtime_applicability(self): manifest, _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}), lambda flag: False, - lambda node: SimpleNamespace( - catalog_type="unity", table_format="default", file_format="delta" - ), + _unity_delta_relation, )[0] assert {row.config: row.count for row in root.config_usage} == { models.ModelConfig.LIQUID_CLUSTERING: 1, models.ModelConfig.ZORDER: 1, + models.ModelConfig.MERGE_SCHEMA_EVOLUTION: 1, + models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE: 1, } + assert not any( + row.compute_type == models.ComputeType.ALL_PURPOSE_CLUSTER + for row in root.effective_compute_type_counts + ) class TestBuildPostRunLog: @pytest.mark.parametrize( "exc_type, results, fail_fast, task_success, status, reason", [ - pytest.param( - None, - [], - False, - None, - models.InvocationStatus.SUCCESS, - models.TerminationReason.NORMAL, - id="success", - ), pytest.param( None, [("model.p.m1", "error")], diff --git a/tests/unit/telemetry/test_config.py b/tests/unit/telemetry/test_config.py index 1c35bcd31..6738e7ca0 100644 --- a/tests/unit/telemetry/test_config.py +++ b/tests/unit/telemetry/test_config.py @@ -30,11 +30,7 @@ class TestCommandEligibility: @pytest.mark.parametrize( "command, eligible", [ - ("build", True), ("run", True), - ("test", True), - ("seed", True), - ("snapshot", True), ("compile", False), ("source freshness", False), ("run-operation", False), diff --git a/tests/unit/telemetry/test_encoder.py b/tests/unit/telemetry/test_encoder.py index abfc74c3b..7604238ff 100644 --- a/tests/unit/telemetry/test_encoder.py +++ b/tests/unit/telemetry/test_encoder.py @@ -76,7 +76,7 @@ def test_post_parse_envelope(self): assert entry["post_parse"]["connection_config"]["default_compute_type"] == "SQL_WAREHOUSE" model_stats = entry["post_parse"]["model_config_stats"][0] assert model_stats["scope"] == "ROOT_PROJECT" - assert model_stats["materialization_counts"] == [{"materialization": "TABLE", "count": 1}] + assert model_stats["materialization_counts"][0]["materialization"] == "TABLE" @pytest.mark.parametrize( "workspace_id, expected", diff --git a/tests/unit/telemetry/test_hooks.py b/tests/unit/telemetry/test_hooks.py index 77b9471fb..37cdd7888 100644 --- a/tests/unit/telemetry/test_hooks.py +++ b/tests/unit/telemetry/test_hooks.py @@ -70,24 +70,6 @@ def test_hook_exceptions_do_not_escape(monkeypatch): hooks.on_post_parse(adapter, SimpleNamespace()) # must not raise -def test_post_parse_passes_catalog_relation_builder(monkeypatch): - coord = _enable_hooks(monkeypatch) - coord.needs_post_parse.return_value = True - build = Mock(return_value=_parse_log()) - monkeypatch.setattr(hooks.builder, "build_post_parse_log", build) - adapter = SimpleNamespace( - config=SimpleNamespace(credentials=SimpleNamespace()), - get_behavior_flag_no_warn=lambda _: False, - build_catalog_relation=Mock(), - ) - manifest = SimpleNamespace() - - hooks.on_post_parse(adapter, manifest) - - assert build.call_args.kwargs["catalog_relation_builder"] is adapter.build_catalog_relation - coord.set_post_parse.assert_called_once() - - def test_run_end_exception_finalizes_stored_invocation_not_current_global(monkeypatch): coord = Coordinator() logs = [] From 729f378455908d5744e92619d028931f0744e711 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Mon, 7 Sep 2026 22:35:42 +0530 Subject: [PATCH 06/13] fix: count only model configs that runtime actually applies Merge knobs, Z-order, and constraints were inflating adoption for strategies and formats the macros ignore, and v2 catalog_database HMS routing was labeled Unity. --- dbt/adapters/databricks/telemetry/builder.py | 72 ++++++++++++----- tests/unit/telemetry/test_builder.py | 84 ++++++++++++++++++++ 2 files changed, 138 insertions(+), 18 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index 681cd459b..f2b27dcbb 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -363,15 +363,19 @@ def _python_submission_method(config: Any) -> models.PythonSubmissionMethod: return _PYTHON_SUBMISSION_METHOD_MAP.get(value, models.PythonSubmissionMethod.OTHER) +def _physical_catalog_name(catalog_relation: Any, node: Any) -> str: + if catalog_relation is not None: + for attr in ("catalog_database", "catalog_name"): + value = _normalized(getattr(catalog_relation, attr, None)) + if value: + return value + return _normalized(getattr(node, "database", None)) + + def _catalog_type(catalog_relation: Any, node: Any) -> models.CatalogType: - # Physical hive_metastore is HMS even when the default Unity integration - # supplies catalog_type="unity". - physical = _normalized( - getattr(catalog_relation, "catalog_name", None) if catalog_relation is not None else None - ) - if not physical: - physical = _normalized(getattr(node, "database", None)) - if physical in _HMS_CATALOG_NAMES: + # Physical hive_metastore is HMS even when the Unity integration supplies + # catalog_type="unity", including the v2 catalog_database override. + if _physical_catalog_name(catalog_relation, node) in _HMS_CATALOG_NAMES: return models.CatalogType.HIVE_METASTORE if catalog_relation is None: return models.CatalogType.TYPE_UNSPECIFIED @@ -379,6 +383,14 @@ def _catalog_type(catalog_relation: Any, node: Any) -> models.CatalogType: return _CATALOG_TYPE_MAP.get(value, models.CatalogType.OTHER) +def _resolved_file_format(catalog_relation: Any, use_managed_iceberg: bool) -> str: + if catalog_relation is None: + return "" + if _normalized(getattr(catalog_relation, "table_format", None)) == "iceberg": + return "parquet" if use_managed_iceberg else "delta" + return _normalized(getattr(catalog_relation, "file_format", None)) or "delta" + + def _storage_format( catalog_relation: Any, use_managed_iceberg: bool ) -> models.EffectiveStorageFormat: @@ -417,7 +429,15 @@ def _constraint_name(constraint: Any) -> str: return _normalized(_value(constraint, "type")) -def _constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: +def _constraints_activated(config: Any) -> bool: + if _enabled(_value(config, "persist_constraints")): + return True + return _enabled(_value(_value(config, "contract"), "enforced")) + + +def _constraint_configs(node: Any, config: Any, is_delta: bool) -> set[models.ModelConfig]: + if not is_delta or not _constraints_activated(config): + return set() constraint_names = { _constraint_name(constraint) for constraint in (getattr(node, "constraints", None) or []) } @@ -447,18 +467,23 @@ def _constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: def _shared_config_usage( - node: Any, config: Any, materialization: models.Materialization + node: Any, config: Any, materialization: models.Materialization, is_delta: bool ) -> set[models.ModelConfig]: usage: set[models.ModelConfig] = set() if materialization in _CONSTRAINT_MATERIALIZATIONS: - usage.update(_constraint_configs(node, config)) + usage.update(_constraint_configs(node, config, is_delta)) auto_liquid_cluster = _enabled(_value(config, "auto_liquid_cluster")) has_liquid = bool(_value(config, "liquid_clustered_by") or auto_liquid_cluster) if has_liquid and materialization in _LIQUID_MATERIALIZATIONS: usage.add(models.ModelConfig.LIQUID_CLUSTERING) if auto_liquid_cluster: usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) - if _value(config, "zorder") and materialization in _ZORDER_MATERIALIZATIONS and not has_liquid: + if ( + _value(config, "zorder") + and materialization in _ZORDER_MATERIALIZATIONS + and not has_liquid + and is_delta + ): usage.add(models.ModelConfig.ZORDER) if _value(config, "databricks_tags") and materialization in _TAG_MATERIALIZATIONS: usage.add(models.ModelConfig.DATABRICKS_RELATION_TAGS) @@ -480,11 +505,16 @@ def _shared_config_usage( return usage -def _incremental_config_usage(config: Any) -> set[models.ModelConfig]: +def _incremental_config_usage( + config: Any, strategy: models.IncrementalStrategy +) -> set[models.ModelConfig]: + if strategy != models.IncrementalStrategy.MERGE: + return set() usage = set() if _enabled(_value(config, "merge_with_schema_evolution")): usage.add(models.ModelConfig.MERGE_SCHEMA_EVOLUTION) - if _value(config, "not_matched_by_source_action"): + action = _normalized(_value(config, "not_matched_by_source_action")) + if action == "delete" or action.startswith("update"): usage.add(models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE) return usage @@ -528,19 +558,25 @@ def aggregate_model_configs( acc.model_count += 1 acc.materializations[materialization] += 1 acc.languages[language] += 1 - acc.config_usage.update(_shared_config_usage(node, config, materialization)) + relation = ( + _catalog_relation(node, catalog_relation_builder) + if materialization != models.Materialization.EPHEMERAL + else None + ) + is_delta = _resolved_file_format(relation, use_managed_iceberg) == "delta" + acc.config_usage.update(_shared_config_usage(node, config, materialization, is_delta)) if materialization == models.Materialization.INCREMENTAL: + strategy = _incremental_strategy(config) acc.incremental_model_count += 1 - acc.incremental_strategies[_incremental_strategy(config)] += 1 - acc.config_usage.update(_incremental_config_usage(config)) + acc.incremental_strategies[strategy] += 1 + acc.config_usage.update(_incremental_config_usage(config, strategy)) if language == models.Language.PYTHON: acc.python_model_count += 1 acc.python_submission_methods[_python_submission_method(config)] += 1 if materialization != models.Materialization.EPHEMERAL: - relation = _catalog_relation(node, catalog_relation_builder) acc.catalog_types[_catalog_type(relation, node)] += 1 acc.compute_types[_compute_type(config, creds)] += 1 if materialization in _STORAGE_FORMAT_MATERIALIZATIONS: diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index bbd75c582..0d538d0f6 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -305,6 +305,80 @@ def test_physical_hive_metastore_is_classified_as_hms(self): models.CatalogTypeCount(models.CatalogType.HIVE_METASTORE, 1) ] + def test_v2_catalog_database_hive_metastore_is_classified_as_hms(self): + node = _model("table", database="main") + node.schema = "analytics" + node.identifier = "model_one" + integration = UnityCatalogIntegration( + SimpleNamespace( + name="v2_routed_catalog", + catalog_type="unity", + catalog_name="logical_catalog_label", + catalog_database="hive_metastore", + table_format="default", + external_volume=None, + file_format="delta", + adapter_properties={}, + ) + ) + relation = integration.build_relation(node) + assert relation.catalog_type == "unity" + assert relation.catalog_name == "logical_catalog_label" + assert relation.catalog_database == "hive_metastore" + + root = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"model": node}, + ), + _creds(), + lambda flag: False, + integration.build_relation, + )[0] + + assert root.catalog_type_counts == [ + models.CatalogTypeCount(models.CatalogType.HIVE_METASTORE, 1) + ] + + def test_zorder_and_constraints_follow_delta_and_activation_gates(self): + columns = {"id": {"constraints": [{"type": "not_null"}]}} + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "parquet": _model( + "table", + file_format="parquet", + zorder=["id"], + columns=columns, + ), + "active": _model( + "table", + zorder=["id"], + contract={"enforced": True}, + columns=columns, + ), + }, + ) + + def build_relation(node): + return SimpleNamespace( + catalog_type="unity", + table_format="default", + file_format=node.config.get("file_format", "delta"), + ) + + root = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag: False, + build_relation, + )[0] + + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.ZORDER: 1, + models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + } + def test_config_usage_follows_runtime_applicability(self): columns = {"id": {"_extra": {"column_mask": {"function": "mask_id"}}}} manifest = SimpleNamespace( @@ -329,6 +403,16 @@ def test_config_usage_follows_runtime_applicability(self): merge_with_schema_evolution=True, not_matched_by_source_action="delete", ), + "incremental_append": _model( + "incremental", + incremental_strategy="append", + merge_with_schema_evolution=True, + not_matched_by_source_action="delete", + ), + "incremental_invalid_action": _model( + "incremental", + not_matched_by_source_action="drop", + ), "ephemeral_compute": _model( "ephemeral", databricks_compute="cluster", zorder=["id"] ), From 54bcb1442dbc6d438399564eb824b1b0c45dafe9 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Wed, 9 Sep 2026 12:12:48 +0530 Subject: [PATCH 07/13] fix: drop V2 legacy constraints and shadowed auto clustering from telemetry V2 never applies persist_constraints metadata, and CLUSTER BY AUTO is ignored when explicit liquid clustering columns are set. --- dbt/adapters/databricks/telemetry/builder.py | 32 +++++++++------ tests/unit/telemetry/test_builder.py | 41 +++++++++++++++++++- 2 files changed, 60 insertions(+), 13 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index f2b27dcbb..dd89e05d7 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -429,14 +429,16 @@ def _constraint_name(constraint: Any) -> str: return _normalized(_value(constraint, "type")) -def _constraints_activated(config: Any) -> bool: - if _enabled(_value(config, "persist_constraints")): +def _constraints_activated(config: Any, use_materialization_v2: bool) -> bool: + if _enabled(_value(_value(config, "contract"), "enforced")): return True - return _enabled(_value(_value(config, "contract"), "enforced")) + return not use_materialization_v2 and _enabled(_value(config, "persist_constraints")) -def _constraint_configs(node: Any, config: Any, is_delta: bool) -> set[models.ModelConfig]: - if not is_delta or not _constraints_activated(config): +def _constraint_configs( + node: Any, config: Any, is_delta: bool, use_materialization_v2: bool +) -> set[models.ModelConfig]: + if not is_delta or not _constraints_activated(config, use_materialization_v2): return set() constraint_names = { _constraint_name(constraint) for constraint in (getattr(node, "constraints", None) or []) @@ -447,7 +449,7 @@ def _constraint_configs(node: Any, config: Any, is_delta: bool) -> set[models.Mo _constraint_name(constraint) for constraint in (_value(column, "constraints", []) or []) ) - if _enabled(_value(config, "persist_constraints")): + if not use_materialization_v2 and _enabled(_value(config, "persist_constraints")): meta = getattr(node, "meta", None) or {} for constraint in _value(meta, "constraints", []) or []: constraint_type = _constraint_name(constraint) @@ -467,16 +469,21 @@ def _constraint_configs(node: Any, config: Any, is_delta: bool) -> set[models.Mo def _shared_config_usage( - node: Any, config: Any, materialization: models.Materialization, is_delta: bool + node: Any, + config: Any, + materialization: models.Materialization, + is_delta: bool, + use_materialization_v2: bool, ) -> set[models.ModelConfig]: usage: set[models.ModelConfig] = set() if materialization in _CONSTRAINT_MATERIALIZATIONS: - usage.update(_constraint_configs(node, config, is_delta)) + usage.update(_constraint_configs(node, config, is_delta, use_materialization_v2)) auto_liquid_cluster = _enabled(_value(config, "auto_liquid_cluster")) - has_liquid = bool(_value(config, "liquid_clustered_by") or auto_liquid_cluster) + explicit_liquid = bool(_value(config, "liquid_clustered_by")) + has_liquid = explicit_liquid or auto_liquid_cluster if has_liquid and materialization in _LIQUID_MATERIALIZATIONS: usage.add(models.ModelConfig.LIQUID_CLUSTERING) - if auto_liquid_cluster: + if auto_liquid_cluster and not explicit_liquid: usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) if ( _value(config, "zorder") @@ -542,6 +549,7 @@ def aggregate_model_configs( ), } use_managed_iceberg = bool(behavior_flag("use_managed_iceberg")) + use_materialization_v2 = bool(behavior_flag("use_materialization_v2")) for node in (getattr(manifest, "nodes", None) or {}).values(): if _resource_type(node) != "model": continue @@ -564,7 +572,9 @@ def aggregate_model_configs( else None ) is_delta = _resolved_file_format(relation, use_managed_iceberg) == "delta" - acc.config_usage.update(_shared_config_usage(node, config, materialization, is_delta)) + acc.config_usage.update( + _shared_config_usage(node, config, materialization, is_delta, use_materialization_v2) + ) if materialization == models.Materialization.INCREMENTAL: strategy = _incremental_strategy(config) diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 0d538d0f6..1bda9d4a3 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -283,6 +283,36 @@ def test_legacy_constraints_require_persist_constraints(self): models.ModelConfig.CHECK_CONSTRAINT: 1, } + def test_v2_ignores_legacy_persist_constraints(self): + legacy = _model( + "table", + persist_constraints=True, + columns={"id": {"meta": {"constraint": "not_null"}}}, + ) + legacy.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + contracted = _model( + "table", + persist_constraints=True, + contract={"enforced": True}, + columns={"id": {"constraints": [{"type": "not_null"}]}}, + ) + contracted.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"legacy": legacy, "contracted": contracted}, + ) + + root = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag: flag == "use_materialization_v2", + _unity_delta_relation, + )[0] + + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + } + def test_physical_hive_metastore_is_classified_as_hms(self): node = _model("table", database="hive_metastore") node.schema = "dbt" @@ -391,8 +421,14 @@ def test_config_usage_follows_runtime_applicability(self): columns=columns, row_filter={"function": "f", "columns": ["id"]}, ), - "table_both": _model("table", zorder=["id"], liquid_clustered_by=["id"]), + "table_both": _model( + "table", + zorder=["id"], + liquid_clustered_by=["id"], + auto_liquid_cluster=True, + ), "table_zorder": _model("table", zorder=["id"]), + "table_auto": _model("table", auto_liquid_cluster=True), "table_merge": _model( "table", merge_with_schema_evolution=True, @@ -427,7 +463,8 @@ def test_config_usage_follows_runtime_applicability(self): )[0] assert {row.config: row.count for row in root.config_usage} == { - models.ModelConfig.LIQUID_CLUSTERING: 1, + models.ModelConfig.LIQUID_CLUSTERING: 2, + models.ModelConfig.AUTO_LIQUID_CLUSTERING: 1, models.ModelConfig.ZORDER: 1, models.ModelConfig.MERGE_SCHEMA_EVOLUTION: 1, models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE: 1, From a661d5bec198e2902362aa902632b89355d1f576 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Wed, 9 Sep 2026 12:47:41 +0530 Subject: [PATCH 08/13] fix: count V1 persist_constraints as replacing modern constraints Match get_model_constraints and get_column_constraints so telemetry does not union legacy meta with contract constraints at the same level. --- dbt/adapters/databricks/telemetry/builder.py | 49 +++++++++++--------- tests/unit/telemetry/test_builder.py | 29 ++++++++++++ 2 files changed, 57 insertions(+), 21 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index dd89e05d7..662dbc93b 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -429,6 +429,21 @@ def _constraint_name(constraint: Any) -> str: return _normalized(_value(constraint, "type")) +def _constraint_type_key(constraint: Any) -> str: + if isinstance(constraint, str): + return _normalized(constraint) + name = _constraint_name(constraint) + if name: + return name + if _value(constraint, "name") or _value(constraint, "condition"): + return "check" + return "" + + +def _constraint_type_keys(constraints: Any) -> set[str]: + return {key for constraint in (constraints or []) if (key := _constraint_type_key(constraint))} + + def _constraints_activated(config: Any, use_materialization_v2: bool) -> bool: if _enabled(_value(_value(config, "contract"), "enforced")): return True @@ -440,27 +455,19 @@ def _constraint_configs( ) -> set[models.ModelConfig]: if not is_delta or not _constraints_activated(config, use_materialization_v2): return set() - constraint_names = { - _constraint_name(constraint) for constraint in (getattr(node, "constraints", None) or []) - } - columns = _columns(node) - for column in columns: - constraint_names.update( - _constraint_name(constraint) for constraint in (_value(column, "constraints", []) or []) - ) - - if not use_materialization_v2 and _enabled(_value(config, "persist_constraints")): - meta = getattr(node, "meta", None) or {} - for constraint in _value(meta, "constraints", []) or []: - constraint_type = _constraint_name(constraint) - constraint_names.add(constraint_type or "check") - for column in columns: - legacy_constraint = _value(_value(column, "meta", {}), "constraint") - if legacy_constraint: - constraint_names.add( - _constraint_name(legacy_constraint) or _normalized(legacy_constraint) - ) - + persist = not use_materialization_v2 and _enabled(_value(config, "persist_constraints")) + meta = getattr(node, "meta", None) or {} + legacy_model = _value(meta, "constraints") + if persist and isinstance(legacy_model, (list, tuple)): + constraint_names = _constraint_type_keys(legacy_model) + else: + constraint_names = _constraint_type_keys(getattr(node, "constraints", None)) + for column in _columns(node): + legacy_column = _value(_value(column, "meta", {}), "constraint") + if persist and legacy_column: + constraint_names.update(_constraint_type_keys([legacy_column])) + else: + constraint_names.update(_constraint_type_keys(_value(column, "constraints", []))) return { model_config for name in constraint_names diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 1bda9d4a3..d58eb1968 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -313,6 +313,35 @@ def test_v2_ignores_legacy_persist_constraints(self): models.ModelConfig.NOT_NULL_CONSTRAINT: 1, } + def test_v1_legacy_constraints_replace_modern_at_the_same_level(self): + node = _model( + "table", + persist_constraints=True, + contract={"enforced": True}, + columns={ + "id": { + "constraints": [{"type": "foreign_key"}], + "meta": {"constraint": "not_null"}, + } + }, + constraints=[{"type": "primary_key"}], + ) + node.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + root = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"mixed": node}, + ), + _creds(), + lambda flag: False, + _unity_delta_relation, + )[0] + + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.CHECK_CONSTRAINT: 1, + models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + } + def test_physical_hive_metastore_is_classified_as_hms(self): node = _model("table", database="hive_metastore") node.schema = "dbt" From 4b6860393065d48d63eca8f2c81db1fa14a51422 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Wed, 9 Sep 2026 14:09:38 +0530 Subject: [PATCH 09/13] fix: count model configs on the paths that actually apply them Count MV contract constraints, streaming-table column NOT NULL and masks, V2 table/incremental masks, and drop V1 Python-table auto-liquid and row-filter false positives. --- dbt/adapters/databricks/telemetry/builder.py | 82 ++++++++-- tests/unit/telemetry/test_builder.py | 158 +++++++++++++++++++ 2 files changed, 226 insertions(+), 14 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index 662dbc93b..af8f37995 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -444,12 +444,31 @@ def _constraint_type_keys(constraints: Any) -> set[str]: return {key for constraint in (constraints or []) if (key := _constraint_type_key(constraint))} +def _contract_enforced(config: Any) -> bool: + return _enabled(_value(_value(config, "contract"), "enforced")) + + def _constraints_activated(config: Any, use_materialization_v2: bool) -> bool: - if _enabled(_value(_value(config, "contract"), "enforced")): + if _contract_enforced(config): return True return not use_materialization_v2 and _enabled(_value(config, "persist_constraints")) +def _constraint_configs_from_names(constraint_names: set[str]) -> set[models.ModelConfig]: + return { + model_config + for name in constraint_names + if (model_config := _CONSTRAINT_CONFIG_MAP.get(name)) is not None + } + + +def _modern_constraint_type_keys(node: Any) -> set[str]: + constraint_names = _constraint_type_keys(getattr(node, "constraints", None)) + for column in _columns(node): + constraint_names.update(_constraint_type_keys(_value(column, "constraints", []))) + return constraint_names + + def _constraint_configs( node: Any, config: Any, is_delta: bool, use_materialization_v2: bool ) -> set[models.ModelConfig]: @@ -468,30 +487,60 @@ def _constraint_configs( constraint_names.update(_constraint_type_keys([legacy_column])) else: constraint_names.update(_constraint_type_keys(_value(column, "constraints", []))) - return { - model_config - for name in constraint_names - if (model_config := _CONSTRAINT_CONFIG_MAP.get(name)) is not None - } + return _constraint_configs_from_names(constraint_names) + + +def _materialized_view_constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: + if not _contract_enforced(config): + return set() + return _constraint_configs_from_names(_modern_constraint_type_keys(node)) + + +def _streaming_table_constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: + if not _contract_enforced(config): + return set() + for column in _columns(node): + if "not_null" in _constraint_type_keys(_value(column, "constraints", [])): + return {models.ModelConfig.NOT_NULL_CONSTRAINT} + return set() + + +def _counts_column_masks( + materialization: models.Materialization, use_materialization_v2: bool +) -> bool: + if materialization == models.Materialization.STREAMING_TABLE: + return True + return use_materialization_v2 and materialization in _MASK_MATERIALIZATIONS def _shared_config_usage( node: Any, config: Any, materialization: models.Materialization, + language: models.Language, is_delta: bool, use_materialization_v2: bool, ) -> set[models.ModelConfig]: usage: set[models.ModelConfig] = set() if materialization in _CONSTRAINT_MATERIALIZATIONS: usage.update(_constraint_configs(node, config, is_delta, use_materialization_v2)) + elif materialization == models.Materialization.MATERIALIZED_VIEW: + usage.update(_materialized_view_constraint_configs(node, config)) + elif materialization == models.Materialization.STREAMING_TABLE: + usage.update(_streaming_table_constraint_configs(node, config)) auto_liquid_cluster = _enabled(_value(config, "auto_liquid_cluster")) explicit_liquid = bool(_value(config, "liquid_clustered_by")) has_liquid = explicit_liquid or auto_liquid_cluster + v1_python_table = ( + not use_materialization_v2 + and language == models.Language.PYTHON + and materialization == models.Materialization.TABLE + ) if has_liquid and materialization in _LIQUID_MATERIALIZATIONS: - usage.add(models.ModelConfig.LIQUID_CLUSTERING) - if auto_liquid_cluster and not explicit_liquid: - usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) + if not (v1_python_table and not explicit_liquid): + usage.add(models.ModelConfig.LIQUID_CLUSTERING) + if auto_liquid_cluster and not explicit_liquid: + usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) if ( _value(config, "zorder") and materialization in _ZORDER_MATERIALIZATIONS @@ -507,12 +556,15 @@ def _shared_config_usage( and materialization in _COLUMN_TAG_MATERIALIZATIONS ): usage.add(models.ModelConfig.COLUMN_TAGS) - if ( - any(_column_extra(column).get("column_mask") for column in columns) - and materialization in _MASK_MATERIALIZATIONS + if any(_column_extra(column).get("column_mask") for column in columns) and _counts_column_masks( + materialization, use_materialization_v2 ): usage.add(models.ModelConfig.COLUMN_MASKS) - if _value(config, "row_filter") and materialization in _ROW_FILTER_MATERIALIZATIONS: + if ( + _value(config, "row_filter") + and materialization in _ROW_FILTER_MATERIALIZATIONS + and not v1_python_table + ): usage.add(models.ModelConfig.ROW_FILTER) if _value(config, "databricks_compute") and materialization in _NAMED_COMPUTE_MATERIALIZATIONS: usage.add(models.ModelConfig.NAMED_COMPUTE_ROUTING) @@ -580,7 +632,9 @@ def aggregate_model_configs( ) is_delta = _resolved_file_format(relation, use_managed_iceberg) == "delta" acc.config_usage.update( - _shared_config_usage(node, config, materialization, is_delta, use_materialization_v2) + _shared_config_usage( + node, config, materialization, language, is_delta, use_materialization_v2 + ) ) if materialization == models.Materialization.INCREMENTAL: diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index d58eb1968..859a75caf 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -342,6 +342,164 @@ def test_v1_legacy_constraints_replace_modern_at_the_same_level(self): models.ModelConfig.NOT_NULL_CONSTRAINT: 1, } + def test_materialized_view_counts_enforced_contract_constraints(self): + contracted = _model( + "materialized_view", + persist_constraints=True, + contract={"enforced": True}, + file_format="parquet", + columns={"id": {"constraints": [{"type": "not_null"}]}}, + constraints=[{"type": "primary_key"}], + ) + contracted.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + unenforced = _model( + "materialized_view", + persist_constraints=True, + columns={"id": {"constraints": [{"type": "not_null"}]}}, + constraints=[{"type": "check", "expression": "id > 0"}], + ) + unenforced.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"contracted": contracted, "unenforced": unenforced}, + ) + + def build_relation(node): + return SimpleNamespace( + catalog_type="unity", + table_format="default", + file_format=node.config.get("file_format", "delta"), + ) + + expected = { + models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + models.ModelConfig.PRIMARY_KEY_CONSTRAINT: 1, + } + for use_v2 in (False, True): + root = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag, use_v2=use_v2: use_v2 if flag == "use_materialization_v2" else False, + build_relation, + )[0] + assert {row.config: row.count for row in root.config_usage} == expected + + def test_streaming_table_counts_only_enforced_column_not_null(self): + mixed = _model( + "streaming_table", + persist_constraints=True, + contract={"enforced": True}, + columns={ + "id": { + "constraints": [ + {"type": "not_null"}, + {"type": "check", "expression": "id > 0"}, + {"type": "primary_key"}, + {"type": "foreign_key"}, + {"type": "custom", "expression": "id <> 99"}, + ] + } + }, + constraints=[ + {"type": "not_null", "columns": ["id"]}, + {"type": "check", "name": "model_check", "expression": "id < 100"}, + {"type": "primary_key"}, + ], + ) + mixed.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} + unenforced = _model( + "streaming_table", + columns={"id": {"constraints": [{"type": "not_null"}]}}, + ) + model_only = _model( + "streaming_table", + contract={"enforced": True}, + constraints=[{"type": "not_null", "columns": ["id"]}], + ) + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"mixed": mixed, "unenforced": unenforced, "model_only": model_only}, + ) + for use_v2 in (False, True): + root = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag, use_v2=use_v2: use_v2 if flag == "use_materialization_v2" else False, + _unity_delta_relation, + )[0] + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + } + + def test_column_masks_follow_v2_and_streaming_paths(self): + columns = {"id": {"_extra": {"column_mask": {"function": "mask_id"}}}} + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "table": _model("table", columns=columns), + "incremental": _model("incremental", columns=columns), + "streaming": _model("streaming_table", columns=columns), + "view": _model("view", columns=columns), + "mv": _model("materialized_view", columns=columns), + }, + ) + expected = { + False: {models.ModelConfig.COLUMN_MASKS: 1}, + True: {models.ModelConfig.COLUMN_MASKS: 3}, + } + for use_v2, usage in expected.items(): + root = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag, use_v2=use_v2: use_v2 if flag == "use_materialization_v2" else False, + _unity_delta_relation, + )[0] + assert {row.config: row.count for row in root.config_usage} == usage + + def test_v1_python_table_ignores_auto_liquid_and_row_filter(self): + row_filter = {"function": "f", "columns": ["id"]} + manifest = SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "py_table_auto": _model( + "table", + language="python", + auto_liquid_cluster=True, + row_filter=row_filter, + ), + "py_table_explicit": _model("table", language="python", liquid_clustered_by=["id"]), + "py_incremental": _model( + "incremental", + language="python", + auto_liquid_cluster=True, + row_filter=row_filter, + ), + "sql_table": _model("table", auto_liquid_cluster=True, row_filter=row_filter), + }, + ) + v1 = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag: False, + _unity_delta_relation, + )[0] + assert {row.config: row.count for row in v1.config_usage} == { + models.ModelConfig.LIQUID_CLUSTERING: 3, + models.ModelConfig.AUTO_LIQUID_CLUSTERING: 2, + models.ModelConfig.ROW_FILTER: 2, + } + v2 = builder.aggregate_model_configs( + manifest, + _creds(), + lambda flag: flag == "use_materialization_v2", + _unity_delta_relation, + )[0] + assert {row.config: row.count for row in v2.config_usage} == { + models.ModelConfig.LIQUID_CLUSTERING: 4, + models.ModelConfig.AUTO_LIQUID_CLUSTERING: 3, + models.ModelConfig.ROW_FILTER: 3, + } + def test_physical_hive_metastore_is_classified_as_hms(self): node = _model("table", database="hive_metastore") node.schema = "dbt" From b06091e8a53048e8b6396c83822097b82067e1ac Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Wed, 9 Sep 2026 14:32:47 +0530 Subject: [PATCH 10/13] fix: omit materialized-view CHECK from config telemetry Dedicated CHECK is routed to alter_constraints, which the MV path never applies; keep NOT NULL, PK, FK, and CUSTOM. --- dbt/adapters/databricks/telemetry/builder.py | 2 +- tests/unit/telemetry/test_builder.py | 16 +++++++++++++--- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index af8f37995..c67231c83 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -493,7 +493,7 @@ def _constraint_configs( def _materialized_view_constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: if not _contract_enforced(config): return set() - return _constraint_configs_from_names(_modern_constraint_type_keys(node)) + return _constraint_configs_from_names(_modern_constraint_type_keys(node) - {"check"}) def _streaming_table_constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 859a75caf..bf0796b42 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -349,14 +349,22 @@ def test_materialized_view_counts_enforced_contract_constraints(self): contract={"enforced": True}, file_format="parquet", columns={"id": {"constraints": [{"type": "not_null"}]}}, - constraints=[{"type": "primary_key"}], + constraints=[ + {"type": "primary_key"}, + {"type": "check", "name": "positive", "expression": "id > 0"}, + {"type": "foreign_key"}, + {"type": "custom", "expression": "CONSTRAINT custom_positive CHECK (id > 0)"}, + ], ) - contracted.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + contracted.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} unenforced = _model( "materialized_view", persist_constraints=True, columns={"id": {"constraints": [{"type": "not_null"}]}}, - constraints=[{"type": "check", "expression": "id > 0"}], + constraints=[ + {"type": "check", "expression": "id > 0"}, + {"type": "primary_key"}, + ], ) unenforced.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} manifest = SimpleNamespace( @@ -374,6 +382,8 @@ def build_relation(node): expected = { models.ModelConfig.NOT_NULL_CONSTRAINT: 1, models.ModelConfig.PRIMARY_KEY_CONSTRAINT: 1, + models.ModelConfig.FOREIGN_KEY_CONSTRAINT: 1, + models.ModelConfig.CUSTOM_CONSTRAINT: 1, } for use_v2 in (False, True): root = builder.aggregate_model_configs( From 56d329f772ff6f3f56bef3d314e6a6e53cbdc7c8 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Wed, 9 Sep 2026 15:46:20 +0530 Subject: [PATCH 11/13] test: drop telemetry coverage that restates implementation Keep only contracts that can fail without rewriting the collector maps, and remove lookup-table and default-forwarding cases. --- tests/unit/telemetry/test_builder.py | 83 ---------------------------- tests/unit/telemetry/test_config.py | 27 --------- tests/unit/telemetry/test_encoder.py | 16 ------ 3 files changed, 126 deletions(-) diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index bf0796b42..82bf92a1a 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -57,31 +57,6 @@ def _unity_delta_relation(_node=None): class TestReportedClassifications: - @pytest.mark.parametrize( - "http_path, expected", - [ - pytest.param( - "/sql/1.0/warehouses/a?o=9", - models.ComputeType.SQL_WAREHOUSE, - id="warehouse", - ), - pytest.param( - "/sql/1.0/endpoints/a", - models.ComputeType.SQL_WAREHOUSE, - id="legacy_endpoint", - ), - pytest.param( - "/sql/protocolv1/o/1/2", - models.ComputeType.ALL_PURPOSE_CLUSTER, - id="cluster", - ), - pytest.param("/unknown", models.ComputeType.OTHER, id="other"), - pytest.param(None, models.ComputeType.TYPE_UNSPECIFIED, id="missing"), - ], - ) - def test_compute_type(self, http_path, expected): - assert builder.classify_compute_type(http_path) == expected - @pytest.mark.parametrize( "creds, expected", [ @@ -109,19 +84,12 @@ def test_auth_family(self, creds, expected): @pytest.mark.parametrize( "warn_error, options, expected", [ - pytest.param(None, None, models.WarnErrorPolicy.WARN_ERROR_DISABLED, id="disabled"), pytest.param( True, SimpleNamespace(error=[], warn=[], silence=["X"]), models.WarnErrorPolicy.WARN_ERROR_ALL, id="legacy_takes_precedence", ), - pytest.param( - False, - SimpleNamespace(error="all", warn=[], silence=[]), - models.WarnErrorPolicy.WARN_ERROR_ALL, - id="error_all", - ), pytest.param( False, SimpleNamespace(error="all", warn=["SomeWarning"], silence=[]), @@ -172,57 +140,6 @@ def test_generic_tests_are_split_from_singular(self): class TestAggregateModelConfigs: - def test_package_models_are_not_folded_into_root(self): - root, installed = builder.aggregate_model_configs( - SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={ - "root_model": _model("table"), - "pkg_model": _model("view", package_name="pkg"), - }, - ), - _creds(), - lambda flag: False, - _unity_delta_relation, - ) - assert [row.materialization for row in root.materialization_counts] == [ - models.Materialization.TABLE - ] - assert [row.materialization for row in installed.materialization_counts] == [ - models.Materialization.VIEW - ] - - def test_incremental_defaults_to_merge_and_resolves_named_compute(self): - root = builder.aggregate_model_configs( - SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={"inc": _model("incremental", databricks_compute="cluster")}, - ), - _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}), - lambda flag: False, - _unity_delta_relation, - )[0] - assert root.incremental_model_stats.strategy_counts == [ - models.IncrementalStrategyCount(models.IncrementalStrategy.MERGE, 1) - ] - assert root.effective_compute_type_counts == [ - models.ComputeTypeCount(models.ComputeType.ALL_PURPOSE_CLUSTER, 1) - ] - - def test_python_model_defaults_to_all_purpose_submission(self): - root = builder.aggregate_model_configs( - SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={"py": _model("table", language="python")}, - ), - _creds(), - lambda flag: False, - _unity_delta_relation, - )[0] - assert root.python_model_stats.submission_method_counts == [ - models.PythonSubmissionMethodCount(models.PythonSubmissionMethod.ALL_PURPOSE_CLUSTER, 1) - ] - @pytest.mark.parametrize( "use_managed, expected_format", [ diff --git a/tests/unit/telemetry/test_config.py b/tests/unit/telemetry/test_config.py index 6738e7ca0..6d71db066 100644 --- a/tests/unit/telemetry/test_config.py +++ b/tests/unit/telemetry/test_config.py @@ -17,33 +17,6 @@ def _creds(connection_parameters, **overrides): return SimpleNamespace(**values) -class TestOptIn: - def test_defaults_off(self): - assert config.is_enabled(_creds({})) is False - assert config.is_enabled(_creds(None)) is False - - def test_explicit_opt_in(self): - assert config.is_enabled(_creds({"enable_dbt_telemetry": True})) is True - - -class TestCommandEligibility: - @pytest.mark.parametrize( - "command, eligible", - [ - ("run", True), - ("compile", False), - ("source freshness", False), - ("run-operation", False), - ("parse", False), - ], - ) - def test_command_eligibility(self, monkeypatch, command, eligible): - from dbt import flags - - monkeypatch.setattr(flags, "get_flags", lambda: SimpleNamespace(WHICH=command)) - assert config.is_eligible_command() is eligible - - class TestTransportEligibility: @pytest.mark.parametrize( "overrides, reusable", diff --git a/tests/unit/telemetry/test_encoder.py b/tests/unit/telemetry/test_encoder.py index 7604238ff..6d9844170 100644 --- a/tests/unit/telemetry/test_encoder.py +++ b/tests/unit/telemetry/test_encoder.py @@ -20,15 +20,6 @@ def _log(): configured_auth_family=models.AuthFamily.PAT, ), project_config=models.ProjectConfig(use_materialization_v2=True), - model_config_stats=[ - models.ModelConfigStats( - scope=models.ModelConfigScope.ROOT_PROJECT, - model_count=1, - materialization_counts=[ - models.MaterializationCount(models.Materialization.TABLE, 1) - ], - ) - ], ), ) @@ -72,11 +63,6 @@ def test_post_parse_envelope(self): assert "dbt_databricks_telemetry_log" in fe["entry"] assert entry["event_type"] == "POST_PARSE" assert "post_run" not in entry - assert entry["post_parse"]["invocation_config"]["dbt_command"] == "RUN" - assert entry["post_parse"]["connection_config"]["default_compute_type"] == "SQL_WAREHOUSE" - model_stats = entry["post_parse"]["model_config_stats"][0] - assert model_stats["scope"] == "ROOT_PROJECT" - assert model_stats["materialization_counts"][0]["materialization"] == "TABLE" @pytest.mark.parametrize( "workspace_id, expected", @@ -98,8 +84,6 @@ def test_post_run_wire_contract(self): rc = entry["post_run"]["result_counts"] assert rc["pass"] == 5 assert "pass_" not in rc - assert entry["post_run"]["run_outcome"]["invocation_status"] == "HANDLED_ERROR" - assert entry["post_run"]["results_by_resource_type"][0]["resource_type"] == "MODEL" def test_unavailable_aggregates_are_omitted(self): log = models.TelemetryLog( From 8e41a10a1f18c63361d51a8867a74cd355712682 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Thu, 10 Sep 2026 19:43:50 +0530 Subject: [PATCH 12/13] fix: count config_usage from resolved parse-time declarations Stop predicting runtime applicability so POST_PARSE reports configured intent, including conflicts; keep resolution on effective_* fields. --- dbt/adapters/databricks/telemetry/builder.py | 199 +-------- tests/unit/telemetry/test_builder.py | 417 +++++-------------- 2 files changed, 118 insertions(+), 498 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index c67231c83..c63078156 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -93,54 +93,6 @@ _HMS_CATALOG_NAMES = {"hive_metastore"} -_TAG_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, - models.Materialization.VIEW, - models.Materialization.MATERIALIZED_VIEW, - models.Materialization.STREAMING_TABLE, - models.Materialization.METRIC_VIEW, -} -_COLUMN_TAG_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, - models.Materialization.VIEW, - models.Materialization.MATERIALIZED_VIEW, - models.Materialization.STREAMING_TABLE, -} -_LIQUID_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, - models.Materialization.MATERIALIZED_VIEW, - models.Materialization.STREAMING_TABLE, -} -_ZORDER_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, -} -_MASK_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, -} -_ROW_FILTER_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, - models.Materialization.MATERIALIZED_VIEW, - models.Materialization.STREAMING_TABLE, -} -_CONSTRAINT_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, -} -_NAMED_COMPUTE_MATERIALIZATIONS = { - models.Materialization.TABLE, - models.Materialization.INCREMENTAL, - models.Materialization.VIEW, - models.Materialization.MATERIALIZED_VIEW, - models.Materialization.STREAMING_TABLE, - models.Materialization.METRIC_VIEW, -} - @dataclass class _ModelConfigAccumulator: @@ -444,143 +396,45 @@ def _constraint_type_keys(constraints: Any) -> set[str]: return {key for constraint in (constraints or []) if (key := _constraint_type_key(constraint))} -def _contract_enforced(config: Any) -> bool: - return _enabled(_value(_value(config, "contract"), "enforced")) - - -def _constraints_activated(config: Any, use_materialization_v2: bool) -> bool: - if _contract_enforced(config): - return True - return not use_materialization_v2 and _enabled(_value(config, "persist_constraints")) - - -def _constraint_configs_from_names(constraint_names: set[str]) -> set[models.ModelConfig]: - return { - model_config - for name in constraint_names - if (model_config := _CONSTRAINT_CONFIG_MAP.get(name)) is not None - } - - -def _modern_constraint_type_keys(node: Any) -> set[str]: +def _constraint_configs(node: Any) -> set[models.ModelConfig]: constraint_names = _constraint_type_keys(getattr(node, "constraints", None)) - for column in _columns(node): - constraint_names.update(_constraint_type_keys(_value(column, "constraints", []))) - return constraint_names - - -def _constraint_configs( - node: Any, config: Any, is_delta: bool, use_materialization_v2: bool -) -> set[models.ModelConfig]: - if not is_delta or not _constraints_activated(config, use_materialization_v2): - return set() - persist = not use_materialization_v2 and _enabled(_value(config, "persist_constraints")) meta = getattr(node, "meta", None) or {} - legacy_model = _value(meta, "constraints") - if persist and isinstance(legacy_model, (list, tuple)): - constraint_names = _constraint_type_keys(legacy_model) - else: - constraint_names = _constraint_type_keys(getattr(node, "constraints", None)) + constraint_names.update(_constraint_type_keys(_value(meta, "constraints"))) for column in _columns(node): + constraint_names.update(_constraint_type_keys(_value(column, "constraints", []))) legacy_column = _value(_value(column, "meta", {}), "constraint") - if persist and legacy_column: + if legacy_column: constraint_names.update(_constraint_type_keys([legacy_column])) - else: - constraint_names.update(_constraint_type_keys(_value(column, "constraints", []))) - return _constraint_configs_from_names(constraint_names) - - -def _materialized_view_constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: - if not _contract_enforced(config): - return set() - return _constraint_configs_from_names(_modern_constraint_type_keys(node) - {"check"}) - - -def _streaming_table_constraint_configs(node: Any, config: Any) -> set[models.ModelConfig]: - if not _contract_enforced(config): - return set() - for column in _columns(node): - if "not_null" in _constraint_type_keys(_value(column, "constraints", [])): - return {models.ModelConfig.NOT_NULL_CONSTRAINT} - return set() - - -def _counts_column_masks( - materialization: models.Materialization, use_materialization_v2: bool -) -> bool: - if materialization == models.Materialization.STREAMING_TABLE: - return True - return use_materialization_v2 and materialization in _MASK_MATERIALIZATIONS + return { + model_config + for name in constraint_names + if (model_config := _CONSTRAINT_CONFIG_MAP.get(name)) is not None + } -def _shared_config_usage( - node: Any, - config: Any, - materialization: models.Materialization, - language: models.Language, - is_delta: bool, - use_materialization_v2: bool, -) -> set[models.ModelConfig]: - usage: set[models.ModelConfig] = set() - if materialization in _CONSTRAINT_MATERIALIZATIONS: - usage.update(_constraint_configs(node, config, is_delta, use_materialization_v2)) - elif materialization == models.Materialization.MATERIALIZED_VIEW: - usage.update(_materialized_view_constraint_configs(node, config)) - elif materialization == models.Materialization.STREAMING_TABLE: - usage.update(_streaming_table_constraint_configs(node, config)) +def _config_usage(node: Any, config: Any) -> set[models.ModelConfig]: + usage = _constraint_configs(node) auto_liquid_cluster = _enabled(_value(config, "auto_liquid_cluster")) - explicit_liquid = bool(_value(config, "liquid_clustered_by")) - has_liquid = explicit_liquid or auto_liquid_cluster - v1_python_table = ( - not use_materialization_v2 - and language == models.Language.PYTHON - and materialization == models.Materialization.TABLE - ) - if has_liquid and materialization in _LIQUID_MATERIALIZATIONS: - if not (v1_python_table and not explicit_liquid): - usage.add(models.ModelConfig.LIQUID_CLUSTERING) - if auto_liquid_cluster and not explicit_liquid: - usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) - if ( - _value(config, "zorder") - and materialization in _ZORDER_MATERIALIZATIONS - and not has_liquid - and is_delta - ): + if _value(config, "liquid_clustered_by") or auto_liquid_cluster: + usage.add(models.ModelConfig.LIQUID_CLUSTERING) + if auto_liquid_cluster: + usage.add(models.ModelConfig.AUTO_LIQUID_CLUSTERING) + if _value(config, "zorder"): usage.add(models.ModelConfig.ZORDER) - if _value(config, "databricks_tags") and materialization in _TAG_MATERIALIZATIONS: + if _value(config, "databricks_tags"): usage.add(models.ModelConfig.DATABRICKS_RELATION_TAGS) columns = _columns(node) - if ( - any(_column_extra(column).get("databricks_tags") for column in columns) - and materialization in _COLUMN_TAG_MATERIALIZATIONS - ): + if any(_column_extra(column).get("databricks_tags") for column in columns): usage.add(models.ModelConfig.COLUMN_TAGS) - if any(_column_extra(column).get("column_mask") for column in columns) and _counts_column_masks( - materialization, use_materialization_v2 - ): + if any(_column_extra(column).get("column_mask") for column in columns): usage.add(models.ModelConfig.COLUMN_MASKS) - if ( - _value(config, "row_filter") - and materialization in _ROW_FILTER_MATERIALIZATIONS - and not v1_python_table - ): + if _value(config, "row_filter"): usage.add(models.ModelConfig.ROW_FILTER) - if _value(config, "databricks_compute") and materialization in _NAMED_COMPUTE_MATERIALIZATIONS: + if _value(config, "databricks_compute"): usage.add(models.ModelConfig.NAMED_COMPUTE_ROUTING) - return usage - - -def _incremental_config_usage( - config: Any, strategy: models.IncrementalStrategy -) -> set[models.ModelConfig]: - if strategy != models.IncrementalStrategy.MERGE: - return set() - usage = set() if _enabled(_value(config, "merge_with_schema_evolution")): usage.add(models.ModelConfig.MERGE_SCHEMA_EVOLUTION) - action = _normalized(_value(config, "not_matched_by_source_action")) - if action == "delete" or action.startswith("update"): + if _value(config, "not_matched_by_source_action"): usage.add(models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE) return usage @@ -608,7 +462,6 @@ def aggregate_model_configs( ), } use_managed_iceberg = bool(behavior_flag("use_managed_iceberg")) - use_materialization_v2 = bool(behavior_flag("use_materialization_v2")) for node in (getattr(manifest, "nodes", None) or {}).values(): if _resource_type(node) != "model": continue @@ -630,18 +483,12 @@ def aggregate_model_configs( if materialization != models.Materialization.EPHEMERAL else None ) - is_delta = _resolved_file_format(relation, use_managed_iceberg) == "delta" - acc.config_usage.update( - _shared_config_usage( - node, config, materialization, language, is_delta, use_materialization_v2 - ) - ) + acc.config_usage.update(_config_usage(node, config)) if materialization == models.Materialization.INCREMENTAL: strategy = _incremental_strategy(config) acc.incremental_model_count += 1 acc.incremental_strategies[strategy] += 1 - acc.config_usage.update(_incremental_config_usage(config, strategy)) if language == models.Language.PYTHON: acc.python_model_count += 1 diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 82bf92a1a..08922433a 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -170,261 +170,138 @@ def test_iceberg_storage_and_unresolved_named_compute(self, use_managed, expecte models.ComputeTypeCount(models.ComputeType.TYPE_UNSPECIFIED, 1) ] - def test_legacy_constraints_require_persist_constraints(self): - legacy = _model( - "table", - persist_constraints=True, - columns={"id": {"meta": {"constraint": "not_null"}}}, - ) - legacy.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} - ignored = _model( - "table", - persist_constraints=False, - columns={"id": {"meta": {"constraint": "not_null"}}}, - ) - ignored.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={"legacy": legacy, "ignored": ignored}, + def test_config_usage_unions_all_declared_constraint_sources(self): + node = _model( + "view", + columns={ + "id": { + "constraints": [{"type": "foreign_key"}], + "meta": {"constraint": "not_null"}, + } + }, + constraints=[ + {"type": "primary_key"}, + {"type": "check", "expression": "id > 0"}, + ], ) - + node.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} root = builder.aggregate_model_configs( - manifest, + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"mixed": node}, + ), _creds(), lambda flag: False, _unity_delta_relation, )[0] - assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.PRIMARY_KEY_CONSTRAINT: 1, + models.ModelConfig.FOREIGN_KEY_CONSTRAINT: 1, models.ModelConfig.NOT_NULL_CONSTRAINT: 1, models.ModelConfig.CHECK_CONSTRAINT: 1, } - def test_v2_ignores_legacy_persist_constraints(self): - legacy = _model( - "table", - persist_constraints=True, - columns={"id": {"meta": {"constraint": "not_null"}}}, - ) - legacy.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} - contracted = _model( - "table", - persist_constraints=True, - contract={"enforced": True}, - columns={"id": {"constraints": [{"type": "not_null"}]}}, - ) - contracted.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={"legacy": legacy, "contracted": contracted}, - ) - + def test_config_usage_counts_conflicting_clustering_declarations(self): root = builder.aggregate_model_configs( - manifest, + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "both": _model( + "table", + zorder=["id"], + liquid_clustered_by=["id"], + auto_liquid_cluster=True, + ) + }, + ), _creds(), - lambda flag: flag == "use_materialization_v2", + lambda flag: False, _unity_delta_relation, )[0] - assert {row.config: row.count for row in root.config_usage} == { - models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + models.ModelConfig.ZORDER: 1, + models.ModelConfig.LIQUID_CLUSTERING: 1, + models.ModelConfig.AUTO_LIQUID_CLUSTERING: 1, } - def test_v1_legacy_constraints_replace_modern_at_the_same_level(self): - node = _model( - "table", - persist_constraints=True, - contract={"enforced": True}, - columns={ - "id": { - "constraints": [{"type": "foreign_key"}], - "meta": {"constraint": "not_null"}, - } - }, - constraints=[{"type": "primary_key"}], - ) - node.meta = {"constraints": [{"name": "positive", "condition": "id > 0"}]} + def test_config_usage_counts_merge_options_regardless_of_strategy(self): root = builder.aggregate_model_configs( SimpleNamespace( metadata=SimpleNamespace(project_name="root"), - nodes={"mixed": node}, + nodes={ + "table": _model( + "table", + merge_with_schema_evolution=True, + not_matched_by_source_action="delete", + ), + "append": _model( + "incremental", + incremental_strategy="append", + merge_with_schema_evolution=True, + not_matched_by_source_action="delete", + ), + "invalid": _model( + "incremental", + not_matched_by_source_action="drop", + ), + }, ), _creds(), lambda flag: False, _unity_delta_relation, )[0] - assert {row.config: row.count for row in root.config_usage} == { - models.ModelConfig.CHECK_CONSTRAINT: 1, - models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + models.ModelConfig.MERGE_SCHEMA_EVOLUTION: 2, + models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE: 3, } - def test_materialized_view_counts_enforced_contract_constraints(self): - contracted = _model( - "materialized_view", - persist_constraints=True, - contract={"enforced": True}, - file_format="parquet", - columns={"id": {"constraints": [{"type": "not_null"}]}}, - constraints=[ - {"type": "primary_key"}, - {"type": "check", "name": "positive", "expression": "id > 0"}, - {"type": "foreign_key"}, - {"type": "custom", "expression": "CONSTRAINT custom_positive CHECK (id > 0)"}, - ], - ) - contracted.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} - unenforced = _model( - "materialized_view", - persist_constraints=True, - columns={"id": {"constraints": [{"type": "not_null"}]}}, - constraints=[ - {"type": "check", "expression": "id > 0"}, - {"type": "primary_key"}, - ], - ) - unenforced.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={"contracted": contracted, "unenforced": unenforced}, - ) - - def build_relation(node): - return SimpleNamespace( - catalog_type="unity", - table_format="default", - file_format=node.config.get("file_format", "delta"), - ) - - expected = { - models.ModelConfig.NOT_NULL_CONSTRAINT: 1, - models.ModelConfig.PRIMARY_KEY_CONSTRAINT: 1, - models.ModelConfig.FOREIGN_KEY_CONSTRAINT: 1, - models.ModelConfig.CUSTOM_CONSTRAINT: 1, - } - for use_v2 in (False, True): - root = builder.aggregate_model_configs( - manifest, - _creds(), - lambda flag, use_v2=use_v2: use_v2 if flag == "use_materialization_v2" else False, - build_relation, - )[0] - assert {row.config: row.count for row in root.config_usage} == expected - - def test_streaming_table_counts_only_enforced_column_not_null(self): - mixed = _model( - "streaming_table", - persist_constraints=True, - contract={"enforced": True}, - columns={ - "id": { - "constraints": [ - {"type": "not_null"}, - {"type": "check", "expression": "id > 0"}, - {"type": "primary_key"}, - {"type": "foreign_key"}, - {"type": "custom", "expression": "id <> 99"}, - ] - } - }, - constraints=[ - {"type": "not_null", "columns": ["id"]}, - {"type": "check", "name": "model_check", "expression": "id < 100"}, - {"type": "primary_key"}, - ], - ) - mixed.meta = {"constraints": [{"name": "legacy", "condition": "id > 0"}]} - unenforced = _model( - "streaming_table", - columns={"id": {"constraints": [{"type": "not_null"}]}}, - ) - model_only = _model( - "streaming_table", - contract={"enforced": True}, - constraints=[{"type": "not_null", "columns": ["id"]}], - ) - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={"mixed": mixed, "unenforced": unenforced, "model_only": model_only}, - ) - for use_v2 in (False, True): - root = builder.aggregate_model_configs( - manifest, - _creds(), - lambda flag, use_v2=use_v2: use_v2 if flag == "use_materialization_v2" else False, - _unity_delta_relation, - )[0] - assert {row.config: row.count for row in root.config_usage} == { - models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + def test_config_usage_counts_declarations_on_any_materialization(self): + extra = { + "id": { + "_extra": {"column_mask": {"function": "mask_id"}, "databricks_tags": {"k": "v"}} } - - def test_column_masks_follow_v2_and_streaming_paths(self): - columns = {"id": {"_extra": {"column_mask": {"function": "mask_id"}}}} - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={ - "table": _model("table", columns=columns), - "incremental": _model("incremental", columns=columns), - "streaming": _model("streaming_table", columns=columns), - "view": _model("view", columns=columns), - "mv": _model("materialized_view", columns=columns), - }, - ) - expected = { - False: {models.ModelConfig.COLUMN_MASKS: 1}, - True: {models.ModelConfig.COLUMN_MASKS: 3}, } - for use_v2, usage in expected.items(): - root = builder.aggregate_model_configs( - manifest, - _creds(), - lambda flag, use_v2=use_v2: use_v2 if flag == "use_materialization_v2" else False, - _unity_delta_relation, - )[0] - assert {row.config: row.count for row in root.config_usage} == usage - - def test_v1_python_table_ignores_auto_liquid_and_row_filter(self): row_filter = {"function": "f", "columns": ["id"]} - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={ - "py_table_auto": _model( - "table", - language="python", - auto_liquid_cluster=True, - row_filter=row_filter, - ), - "py_table_explicit": _model("table", language="python", liquid_clustered_by=["id"]), - "py_incremental": _model( - "incremental", - language="python", - auto_liquid_cluster=True, - row_filter=row_filter, - ), - "sql_table": _model("table", auto_liquid_cluster=True, row_filter=row_filter), - }, - ) - v1 = builder.aggregate_model_configs( - manifest, - _creds(), + root = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={ + "view": _model( + "view", + zorder=["id"], + liquid_clustered_by=["id"], + columns=extra, + row_filter=row_filter, + databricks_tags={"a": "b"}, + databricks_compute="cluster", + ), + "ephemeral": _model("ephemeral", databricks_compute="cluster", zorder=["id"]), + "py": _model( + "table", + language="python", + auto_liquid_cluster=True, + row_filter=row_filter, + columns=extra, + ), + }, + ), + _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}), lambda flag: False, _unity_delta_relation, )[0] - assert {row.config: row.count for row in v1.config_usage} == { - models.ModelConfig.LIQUID_CLUSTERING: 3, - models.ModelConfig.AUTO_LIQUID_CLUSTERING: 2, + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.ZORDER: 2, + models.ModelConfig.LIQUID_CLUSTERING: 2, + models.ModelConfig.AUTO_LIQUID_CLUSTERING: 1, + models.ModelConfig.COLUMN_MASKS: 2, models.ModelConfig.ROW_FILTER: 2, + models.ModelConfig.DATABRICKS_RELATION_TAGS: 1, + models.ModelConfig.COLUMN_TAGS: 2, + models.ModelConfig.NAMED_COMPUTE_ROUTING: 2, } - v2 = builder.aggregate_model_configs( - manifest, - _creds(), - lambda flag: flag == "use_materialization_v2", - _unity_delta_relation, - )[0] - assert {row.config: row.count for row in v2.config_usage} == { - models.ModelConfig.LIQUID_CLUSTERING: 4, - models.ModelConfig.AUTO_LIQUID_CLUSTERING: 3, - models.ModelConfig.ROW_FILTER: 3, + assert {row.compute_type: row.count for row in root.effective_compute_type_counts} == { + models.ComputeType.ALL_PURPOSE_CLUSTER: 1, + models.ComputeType.SQL_WAREHOUSE: 1, } def test_physical_hive_metastore_is_classified_as_hms(self): @@ -484,110 +361,6 @@ def test_v2_catalog_database_hive_metastore_is_classified_as_hms(self): models.CatalogTypeCount(models.CatalogType.HIVE_METASTORE, 1) ] - def test_zorder_and_constraints_follow_delta_and_activation_gates(self): - columns = {"id": {"constraints": [{"type": "not_null"}]}} - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={ - "parquet": _model( - "table", - file_format="parquet", - zorder=["id"], - columns=columns, - ), - "active": _model( - "table", - zorder=["id"], - contract={"enforced": True}, - columns=columns, - ), - }, - ) - - def build_relation(node): - return SimpleNamespace( - catalog_type="unity", - table_format="default", - file_format=node.config.get("file_format", "delta"), - ) - - root = builder.aggregate_model_configs( - manifest, - _creds(), - lambda flag: False, - build_relation, - )[0] - - assert {row.config: row.count for row in root.config_usage} == { - models.ModelConfig.ZORDER: 1, - models.ModelConfig.NOT_NULL_CONSTRAINT: 1, - } - - def test_config_usage_follows_runtime_applicability(self): - columns = {"id": {"_extra": {"column_mask": {"function": "mask_id"}}}} - manifest = SimpleNamespace( - metadata=SimpleNamespace(project_name="root"), - nodes={ - "view_zorder": _model( - "view", - zorder=["id"], - liquid_clustered_by=["id"], - columns=columns, - row_filter={"function": "f", "columns": ["id"]}, - ), - "table_both": _model( - "table", - zorder=["id"], - liquid_clustered_by=["id"], - auto_liquid_cluster=True, - ), - "table_zorder": _model("table", zorder=["id"]), - "table_auto": _model("table", auto_liquid_cluster=True), - "table_merge": _model( - "table", - merge_with_schema_evolution=True, - not_matched_by_source_action="delete", - ), - "incremental_merge": _model( - "incremental", - merge_with_schema_evolution=True, - not_matched_by_source_action="delete", - ), - "incremental_append": _model( - "incremental", - incremental_strategy="append", - merge_with_schema_evolution=True, - not_matched_by_source_action="delete", - ), - "incremental_invalid_action": _model( - "incremental", - not_matched_by_source_action="drop", - ), - "ephemeral_compute": _model( - "ephemeral", databricks_compute="cluster", zorder=["id"] - ), - }, - ) - - root = builder.aggregate_model_configs( - manifest, - _creds(compute={"cluster": {"http_path": "/sql/protocolv1/o/1/cluster"}}), - lambda flag: False, - _unity_delta_relation, - )[0] - - assert {row.config: row.count for row in root.config_usage} == { - models.ModelConfig.LIQUID_CLUSTERING: 2, - models.ModelConfig.AUTO_LIQUID_CLUSTERING: 1, - models.ModelConfig.ZORDER: 1, - models.ModelConfig.MERGE_SCHEMA_EVOLUTION: 1, - models.ModelConfig.MERGE_NOT_MATCHED_BY_SOURCE: 1, - } - assert not any( - row.compute_type == models.ComputeType.ALL_PURPOSE_CLUSTER - for row in root.effective_compute_type_counts - ) - class TestBuildPostRunLog: @pytest.mark.parametrize( From ea54e4e0cb8794873001f942e3ada9d116491370 Mon Sep 17 00:00:00 2001 From: Sai Shree Pradhan Date: Fri, 11 Sep 2026 15:59:31 +0530 Subject: [PATCH 13/13] fix: skip non-sequence constraint metadata in POST_PARSE Arbitrary meta.constraints values were iterable and dropped the whole event. --- dbt/adapters/databricks/telemetry/builder.py | 16 +++++------- tests/unit/telemetry/test_builder.py | 17 ++++++++++++ tests/unit/telemetry/test_config.py | 27 ++++++++++++++++++++ 3 files changed, 51 insertions(+), 9 deletions(-) diff --git a/dbt/adapters/databricks/telemetry/builder.py b/dbt/adapters/databricks/telemetry/builder.py index c63078156..af3ca41f3 100644 --- a/dbt/adapters/databricks/telemetry/builder.py +++ b/dbt/adapters/databricks/telemetry/builder.py @@ -335,14 +335,6 @@ def _catalog_type(catalog_relation: Any, node: Any) -> models.CatalogType: return _CATALOG_TYPE_MAP.get(value, models.CatalogType.OTHER) -def _resolved_file_format(catalog_relation: Any, use_managed_iceberg: bool) -> str: - if catalog_relation is None: - return "" - if _normalized(getattr(catalog_relation, "table_format", None)) == "iceberg": - return "parquet" if use_managed_iceberg else "delta" - return _normalized(getattr(catalog_relation, "file_format", None)) or "delta" - - def _storage_format( catalog_relation: Any, use_managed_iceberg: bool ) -> models.EffectiveStorageFormat: @@ -393,7 +385,9 @@ def _constraint_type_key(constraint: Any) -> str: def _constraint_type_keys(constraints: Any) -> set[str]: - return {key for constraint in (constraints or []) if (key := _constraint_type_key(constraint))} + if not isinstance(constraints, (list, tuple)): + return set() + return {key for constraint in constraints if (key := _constraint_type_key(constraint))} def _constraint_configs(node: Any) -> set[models.ModelConfig]: @@ -413,6 +407,10 @@ def _constraint_configs(node: Any) -> set[models.ModelConfig]: def _config_usage(node: Any, config: Any) -> set[models.ModelConfig]: + """POST_PARSE measures resolved configuration intent. + + Declarations are counted independently of runtime applicability. + """ usage = _constraint_configs(node) auto_liquid_cluster = _enabled(_value(config, "auto_liquid_cluster")) if _value(config, "liquid_clustered_by") or auto_liquid_cluster: diff --git a/tests/unit/telemetry/test_builder.py b/tests/unit/telemetry/test_builder.py index 08922433a..a08078a0f 100644 --- a/tests/unit/telemetry/test_builder.py +++ b/tests/unit/telemetry/test_builder.py @@ -201,6 +201,23 @@ def test_config_usage_unions_all_declared_constraint_sources(self): models.ModelConfig.CHECK_CONSTRAINT: 1, } + def test_config_usage_ignores_non_sequence_legacy_constraint_metadata(self): + invalid = _model("table") + invalid.meta = {"constraints": True} + declared = _model("table", constraints=[{"type": "not_null"}]) + root = builder.aggregate_model_configs( + SimpleNamespace( + metadata=SimpleNamespace(project_name="root"), + nodes={"invalid": invalid, "declared": declared}, + ), + _creds(), + lambda flag: False, + _unity_delta_relation, + )[0] + assert {row.config: row.count for row in root.config_usage} == { + models.ModelConfig.NOT_NULL_CONSTRAINT: 1, + } + def test_config_usage_counts_conflicting_clustering_declarations(self): root = builder.aggregate_model_configs( SimpleNamespace( diff --git a/tests/unit/telemetry/test_config.py b/tests/unit/telemetry/test_config.py index 6d71db066..6738e7ca0 100644 --- a/tests/unit/telemetry/test_config.py +++ b/tests/unit/telemetry/test_config.py @@ -17,6 +17,33 @@ def _creds(connection_parameters, **overrides): return SimpleNamespace(**values) +class TestOptIn: + def test_defaults_off(self): + assert config.is_enabled(_creds({})) is False + assert config.is_enabled(_creds(None)) is False + + def test_explicit_opt_in(self): + assert config.is_enabled(_creds({"enable_dbt_telemetry": True})) is True + + +class TestCommandEligibility: + @pytest.mark.parametrize( + "command, eligible", + [ + ("run", True), + ("compile", False), + ("source freshness", False), + ("run-operation", False), + ("parse", False), + ], + ) + def test_command_eligibility(self, monkeypatch, command, eligible): + from dbt import flags + + monkeypatch.setattr(flags, "get_flags", lambda: SimpleNamespace(WHICH=command)) + assert config.is_eligible_command() is eligible + + class TestTransportEligibility: @pytest.mark.parametrize( "overrides, reusable",