diff --git a/CHANGELOG.md b/CHANGELOG.md index a5cd4776..1071fefd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,11 @@ ## dbt-teradata 1.0.0a ### Features +* Added support for the dbt `function` resource type (dbt 1.11+), enabling `REPLACE FUNCTION` materialization of Teradata SQL scalar UDFs. Aggregate UDFs (`type: aggregate`) raise a clear compile-time error since they are not supported. +* Implement `persist_docs` support ([IDE-26225](https://teradata-pe.atlassian.net/browse/IDE-26225)): model and column `description:` text is now written to the Teradata catalog as native `COMMENT ON TABLE/VIEW/FUNCTION/COLUMN` metadata for table, view, incremental, seed, and snapshot materializations; for `function` materializations, only the relation-level `description:` is applied via `COMMENT ON FUNCTION` (argument/column-level docs are not supported). Comments are escaped, capped at Teradata's 255-character limit, only re-issued when changed, and skipped gracefully for OTF/Iceberg relations. Relation comments now also surface in `dbt docs generate` catalog output for relational resources (tables/views/etc.); dbt-core excludes `function` nodes from the catalog fetch, so function comments do not appear there. ### Fixes +* Fixed snapshots with `hard_deletes='new_record'` reprocessing already-deleted records on every subsequent run, generating a new "deletion" record (with a reused `dbt_scd_id`) each time instead of only once when the deletion is first detected. ### Docs diff --git a/README.md b/README.md index 6ab5f819..b6e414a1 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,7 @@ Refer to [connection parameters](https://github.com/Teradata/python-driver#conne * `table` * `ephemeral` * `incremental` +* `function` (User-Defined Functions — see [UDF materialization support](#user-defined-function-udf-materialization-support)) #### Incremental Materialization The following incremental materialization strategies are supported: @@ -634,6 +635,63 @@ Another e.g. for adding multiple grants: More on Grants can be found at https://docs.getdbt.com/reference/resource-configs/grants +#### Persist docs + +`persist_docs` writes the `description:` text from your model/column YAML into the Teradata data dictionary as native object comments (`COMMENT ON TABLE`, `COMMENT ON VIEW`, `COMMENT ON FUNCTION`, `COMMENT ON COLUMN`). This makes dbt the single source of truth for documentation and surfaces those descriptions in Teradata Studio, BI tools, and data catalogs (e.g. Collibra, Alation) that read metadata from `DBC`. It is supported for `table`, `view`, `incremental`, `seed`, `snapshot`, and (relation-level only — see below) `function` materializations. + +It is disabled by default. Enable it — independently for relations (tables/views) and columns — in `dbt_project.yml`, per folder, or per model: + + dbt_project.yml + ```yaml + models: + : + +persist_docs: + relation: true + columns: true + ``` + + or per model: + ```sql + {{ config(persist_docs={"relation": true, "columns": true}) }} + ``` + +The comment text comes from the `description:` fields: + + models/schema.yml + ```yaml + version: 2 + models: + - name: customers + description: "Cleaned customer dimension, one row per customer." + columns: + - name: customer_id + description: "Primary key of the customer." + ``` + +You can verify the persisted comments directly in Teradata: + ```sql + SELECT CommentString FROM DBC.TablesV + WHERE DatabaseName = '' AND TableName = 'customers'; + + SELECT ColumnName, CommentString FROM DBC.ColumnsV + WHERE DatabaseName = '' AND TableName = 'customers'; + ``` +They also appear in `dbt docs generate` output (`catalog.json`) under each node's `metadata.comment` and per-column `comment`. + +Teradata-specific behavior: + +* **Length limit** – Teradata comments are stored in `DBC ...CommentString` (`VARCHAR(255)`), so comments are capped at **255 characters**. Longer descriptions are truncated with a warning. Override the cap (if your platform permits) with a project var: + ```yaml + vars: + teradata_max_comment_length: 255 + ``` +* **Special characters** – single quotes are escaped by doubling (`'` → `''`). Double quotes, newlines, `--`, and `/* */` are preserved as-is and are safe inside Teradata single-quoted string literals, so they round-trip correctly. +* **Idempotency** – re-running with unchanged descriptions issues no comment DDL; only changed text is re-written (for relations that persist across runs, such as incremental and snapshot — and, per the `REPLACE FUNCTION` behavior noted below, `function`). +* **Open Table Format (OTF/Iceberg)** – models created via `catalog_name` are skipped gracefully (no error), since native `COMMENT ON` does not apply to them. +* **Functions (UDFs)** – only the relation-level comment is supported (`COMMENT ON FUNCTION`, driven by the function's own `description:`). Unlike `REPLACE VIEW`/`CREATE OR REPLACE TABLE`, Teradata's `REPLACE FUNCTION` preserves the existing comment across a rebuild, so unchanged descriptions issue no DDL on re-run. Teradata has no DDL for commenting individual arguments, so argument/column-level docs cannot be applied; if a function node carries column/argument metadata and `persist_docs: {columns: true}` is enabled, the adapter emits a warning and skips those comments rather than raising an error (in practice, dbt-core 1.11's `functions.yml` schema does not currently parse a `columns:` block for functions, so this warning path is rarely reached). Function comments are also not shown in `dbt docs generate` output — dbt-core does not treat `function` nodes as "relational", so they are excluded from the catalog fetch that populates `catalog.json` — but they are visible directly in Teradata (e.g. via `DBC.TablesV.CommentString`) and in tools that read the data dictionary. See [UDF materialization support](#user-defined-function-udf-materialization-support) for details. + +More on persist_docs can be found at https://docs.getdbt.com/reference/resource-configs/persist_docs + ### Cross DB macros Starting with release 1.3, some macros were migrated from [teradata-dbt-utils](https://github.com/Teradata/dbt-teradata-utils) dbt package to the connector. See the table below for the macros supported from the connector. @@ -1071,6 +1129,58 @@ Teradata raises error **7825** ("OTF table not found in external catalog") — o * **OTF cannot be used with the `snapshot` materialization.** Setting `catalog_name` on a snapshot raises a compile-time error (snapshots require update/merge semantics OTF does not provide). * **Column metadata is not available for OTF tables in dbt docs.** OTF tables are not registered in Teradata's `DBC.ColumnsV` view (only native tables are), so column-level metadata (descriptions, data types, constraints) will not appear in dbt docs when you run `dbt docs generate` and serve with `dbt docs serve`. Table-level metadata is still available and functional. +## User-Defined Function (UDF) materialization support + +Starting with dbt 1.11, dbt introduces a `function` resource type. dbt-teradata implements this resource type to materialize Teradata **SQL scalar UDFs** via `REPLACE FUNCTION`. + +Functions are defined the same way as any other dbt `function` resource — a `.sql` file containing the function body, plus a `functions.yml` declaring its arguments, return type, and config: + +```sql +-- functions/add_two_ints.sql +RETURN a + b; +``` + +```yaml +# functions/functions.yml +functions: + - name: add_two_ints + config: + type: scalar + volatility: deterministic + language: sql + arguments: + - name: a + data_type: INTEGER + - name: b + data_type: INTEGER + returns: + data_type: INTEGER +``` + +The function body must be a bare Teradata `RETURN` statement — dbt-teradata wraps it with the rest of the `REPLACE FUNCTION` DDL (argument list, `RETURNS`, `LANGUAGE SQL`, data access clause, volatility, `COLLATION INVOKER`, `INLINE TYPE 1`). Call the function from a model with the `function()` Jinja helper: + +```sql +-- models/use_udf.sql +select {{ function('add_two_ints') }}(10, 32) as total +``` + +### What is supported + +* **Scalar SQL UDFs** (`type: scalar`, `language: sql`) — the only function type/language dbt-teradata materializes. +* **`volatility` config** — `deterministic` maps to `DETERMINISTIC`; `stable` or `non-deterministic` maps to `NOT DETERMINISTIC`. If omitted, neither clause is emitted (Teradata's own default applies). An unrecognized value emits a warning and is ignored. +* **`grants`** — supported via the `execute` privilege, which is rendered as Teradata's `GRANT EXECUTE FUNCTION` DCL (plain `EXECUTE` is for macros/stored procedures, so the adapter adds the required `FUNCTION` keyword automatically for UDFs). Configure it like any other resource, e.g. `grants: {execute: ['reporting_role']}`. Table DML privileges (`select`/`insert`/`update`/`delete`) do not apply to UDFs. +* **`persist_docs` (relation-level only)** — a function's `description:` is written as a native `COMMENT ON FUNCTION` when `persist_docs: {relation: true}` is set (project-wide or per-function `config:`). Unlike `REPLACE VIEW`/`CREATE OR REPLACE TABLE` (which drop-and-recreate the object, wiping any existing comment), Teradata's `REPLACE FUNCTION` preserves the existing comment across a rebuild, so change detection is meaningful here: unchanged descriptions issue no DDL on re-run, and only a changed description re-issues it. Column/argument-level `persist_docs` (`columns: true`) is not supported — Teradata has no DDL for commenting individual arguments — so if a function node carries column/argument metadata, those comments are skipped with a warning instead of raising an error (dbt-core 1.11's `functions.yml` schema does not currently parse a `columns:` block for functions, so this warning path is rarely reached in practice). Function comments do not appear in `dbt docs generate`'s `catalog.json` (dbt-core does not treat `function` nodes as "relational"), but they are visible directly in Teradata, e.g. via `SELECT CommentString FROM DBC.TablesV WHERE TableName = ''`. +* Functions participate in the DAG like any other resource and can be depended on with `ref()`/`function()` from downstream models; `dbt list --resource-type function` and `dbt build --select +` work as expected. +* Re-running is idempotent: the materialization always issues `REPLACE FUNCTION`, so re-running a project with an unchanged function definition does not error. + +### What is not supported + +* **Aggregate UDFs** (`type: aggregate`) — not implemented. Declaring one raises a clear compile-time error rather than emitting invalid DDL; use `type: scalar` for SQL scalar UDFs. +* **Languages other than SQL** — dbt-teradata only emits `LANGUAGE SQL` / `INLINE TYPE 1` DDL. Note that dbt 1.11 itself does not yet parse the `language:` key in `functions.yml` (it always defaults to `sql`), so this is currently a moot point regardless of what you set there. +* **SQL data access clauses other than `CONTAINS SQL`** — Teradata's `LANGUAGE SQL` / `INLINE TYPE 1` UDFs only accept `CONTAINS SQL`; `NO SQL`, `READS SQL DATA`, and `MODIFIES SQL DATA` are for external (C/Java) UDFs and stored procedures and are not applicable here. +* **Reserved-word or special-character argument names.** Argument names in `functions.yml` are emitted unquoted into the `REPLACE FUNCTION` signature (e.g. `RETURN a + b;` requires an unquoted `a INTEGER` parameter, since Teradata folds unquoted identifiers to upper case and the function body references arguments unquoted). This means an argument named after a Teradata reserved word, or containing spaces/special characters, will produce invalid DDL. Use plain, non-reserved identifier names for UDF arguments. +* The dbt user must already have `CREATE FUNCTION` privilege on the target schema; dbt-teradata does not grant it automatically. + ## temporary_metadata_generation_schema (earlier fallback_schema) dbt-teradata internally created temporary tables to fetch the metadata of views for manifest and catalog creation. In case if user does not have permission to create tables on the schema they are working on, they can define a temporary_metadata_generation_schema(to which they have proper create and drop privileges) in dbt_project.yml as variable. diff --git a/dbt/adapters/teradata/column.py b/dbt/adapters/teradata/column.py index fd8561cb..e605c459 100644 --- a/dbt/adapters/teradata/column.py +++ b/dbt/adapters/teradata/column.py @@ -13,6 +13,7 @@ class TeradataColumn(Column): table_name: Optional[str] = None table_type: Optional[str] = None column_index: Optional[int] = None + comment: Optional[str] = None @property def quoted(self) -> str: diff --git a/dbt/include/teradata/macros/adapters.sql b/dbt/include/teradata/macros/adapters.sql index 8a895c3a..1782c3af 100644 --- a/dbt/include/teradata/macros/adapters.sql +++ b/dbt/include/teradata/macros/adapters.sql @@ -284,7 +284,10 @@ END AS table_type, {%- endif -%} - ColumnsV.ColumnID AS column_index + ColumnsV.ColumnID AS column_index, + {#-- "comment" is a Teradata reserved word, so the alias must be quoted. Column + mapping is positional (api.Column(*row)), so the alias name itself is cosmetic. --#} + ColumnsV.CommentString AS "comment" FROM {% if use_qvci == True -%} {{ information_schema_name(relation.schema) }}.ColumnsJQV AS ColumnsV diff --git a/dbt/include/teradata/macros/apply_grants.sql b/dbt/include/teradata/macros/apply_grants.sql index 9d7b8dd2..1befd05a 100644 --- a/dbt/include/teradata/macros/apply_grants.sql +++ b/dbt/include/teradata/macros/apply_grants.sql @@ -12,6 +12,13 @@ {% set TD_db_name= relation.schema %} {% set TD_table_name=relation.identifier %} +{%- if relation.type == 'function' -%} +{#-- UDFs carry the EXECUTE FUNCTION access right ('EF') in DBC.AllRightsV, not the + table DML rights (R/U/I/D). Report it as the 'execute' privilege so it matches + the `grants: {execute: [...]}` config key when dbt diffs current vs. desired grants. --#} +SEL t.Username as grantee, 'execute' as privilege_type FROM DBC.AllRightsV t +WHERE t.DatabaseName='{{TD_db_name}}' and t.Username <> current_user and t.AccessRight = 'EF' and t.tablename='{{TD_table_name}}'; +{%- else -%} with privilege as( SELECT privilege_type, abbreviation FROM (sel 'select' as privilege_type, 'R' as abbreviation) As "DUAL" UNION ALL @@ -24,10 +31,40 @@ SELECT privilege_type, abbreviation FROM (sel 'delete' as privilege_type, 'D' as SEL t.Username as grantee , p.privilege_type FROM DBC.AllRights t cross join privilege p WHERE t.DatabaseName='{{TD_db_name}}' and t.Username <> current_user and t.AccessRight IN ('R','RF','I','U','D') and p.abbreviation=t.AccessRight and t.tablename='{{TD_table_name}}'; +{%- endif -%} {% endmacro %} +{# + -- Teradata requires the `EXECUTE FUNCTION` privilege keyword for UDFs; plain `EXECUTE` + -- is for macros / stored procedures. dbt's grant config key for functions is `execute`, + -- so for function relations we render it as `EXECUTE FUNCTION`. Tables/views are + -- unaffected and keep dbt's default DCL. +#} +{%- macro teradata__function_privilege(privilege) -%} + {%- if privilege | lower == 'execute' -%}execute function{%- else -%}{{ privilege }}{%- endif -%} +{%- endmacro -%} + + +{%- macro teradata__get_grant_sql(relation, privilege, grantees) -%} + {%- if relation.type == 'function' -%} + grant {{ teradata__function_privilege(privilege) }} on {{ relation.render() }} to {{ grantees | join(', ') }} + {%- else -%} + grant {{ privilege }} on {{ relation.render() }} to {{ grantees | join(', ') }} + {%- endif -%} +{%- endmacro -%} + + +{%- macro teradata__get_revoke_sql(relation, privilege, grantees) -%} + {%- if relation.type == 'function' -%} + revoke {{ teradata__function_privilege(privilege) }} on {{ relation.render() }} from {{ grantees | join(', ') }} + {%- else -%} + revoke {{ privilege }} on {{ relation.render() }} from {{ grantees | join(', ') }} + {%- endif -%} +{%- endmacro -%} + + {% macro teradata__call_dcl_statements(dcl_statement_list) %} {# -- We have overridden this macro as teradata doesn't support running multiple dcl statement as a single statement diff --git a/dbt/include/teradata/macros/catalog.sql b/dbt/include/teradata/macros/catalog.sql index 057b08a6..fa96f50f 100644 --- a/dbt/include/teradata/macros/catalog.sql +++ b/dbt/include/teradata/macros/catalog.sql @@ -160,6 +160,7 @@ {%- endmacro %} {% macro teradata__get_catalog_results_sql(view_tmp_tables_mapping) -%} + {% set use_qvci = var("use_qvci", False) | as_bool %} , columns_transformed AS ( SELECT @@ -259,9 +260,44 @@ tables.table_schema = columns_transformed.table_schema AND tables.table_name = columns_transformed.table_name ) - SELECT * + SELECT + joined.table_database, + joined.table_schema, + joined.table_name, + joined.table_type, + {#-- Pull the real relation comment from DBC.TablesV keyed on the remapped + (real) names, so both tables and views (whose columns come from temp + tables) surface their COMMENT ON TABLE/VIEW text in the catalog. --#} + rel_comments.CommentString AS table_comment, + joined.table_owner, + joined.column_name, + joined.column_index, + joined.column_type, + {#-- For views under use_qvci=False, joined.column_comment comes from the temp + table (always NULL); the real COMMENT ON COLUMN text lives in DBC.ColumnsV + under the view's own name, so we backfill it via the col_comments join below. + For tables joined.column_comment is already the real value, so COALESCE keeps + it. Under use_qvci=True the columns come from ColumnsJQV (real view columns), + so joined.column_comment is already correct and the extra join is skipped. --#} + {% if use_qvci -%} + joined.column_comment AS column_comment + {%- else -%} + COALESCE(joined.column_comment, col_comments.CommentString) AS column_comment + {%- endif %} FROM joined - ORDER BY table_schema, table_name, column_index + LEFT OUTER JOIN DBC.TablesV AS rel_comments + ON rel_comments.DatabaseName = joined.table_schema (NOT CASESPECIFIC) + AND rel_comments.TableName = joined.table_name (NOT CASESPECIFIC) + {% if not use_qvci -%} + {#-- Restricted to views (joined.table_type = 'view'); table columns already carry + the real comment in joined.column_comment, so no join is needed for them. --#} + LEFT OUTER JOIN DBC.ColumnsV AS col_comments + ON joined.table_type = 'view' + AND col_comments.DatabaseName = joined.table_schema (NOT CASESPECIFIC) + AND col_comments.TableName = joined.table_name (NOT CASESPECIFIC) + AND col_comments.ColumnName = joined.column_name (NOT CASESPECIFIC) + {% endif -%} + ORDER BY joined.table_schema, joined.table_name, joined.column_index {%- endmacro %} --get_catalog_schemas_where_clause_sql(schemas) copied straight from pre-existing get_catalog(). This uses jinja to loop through the provided schema list and make a big WHERE clause of the form: diff --git a/dbt/include/teradata/macros/materializations/functions/function.sql b/dbt/include/teradata/macros/materializations/functions/function.sql new file mode 100644 index 00000000..fe29ba46 --- /dev/null +++ b/dbt/include/teradata/macros/materializations/functions/function.sql @@ -0,0 +1,126 @@ +{# + Teradata SQL scalar UDF support for the dbt 1.11 `function` resource type. + + Supported: type: scalar, language: sql + Unsupported: type: aggregate — raises compile error via guard macro. + + Note: dbt 1.11 ignores the `language:` key in functions.yml entirely; + FunctionNode.language always defaults to 'sql'. The supported_languages + declaration in the materialization override is therefore a no-op guard + for future-proofing, not an active filter today. + + The model body (.sql file) must be a bare Teradata RETURN statement, e.g.: + RETURN a + b; + + Teradata emits: + REPLACE FUNCTION . () + RETURNS + LANGUAGE SQL + CONTAINS SQL -- only valid data access for LANGUAGE SQL / INLINE TYPE 1 + [DETERMINISTIC|NOT DETERMINISTIC] -- config: volatility (omitted if not set) + COLLATION INVOKER + INLINE TYPE 1 + RETURN ...; + + persist_docs: the function-level `description:` is written via `COMMENT ON + FUNCTION ... AS '...'` (see teradata__persist_docs / teradata__alter_relation_comment + in persist_docs.sql). Column-level persist_docs (`columns: true`) is skipped with a + warning — Teradata has no per-argument comment DDL, only a single comment for the + whole function. +#} + +{# Declare only SQL as supported. dbt 1.11 ignores `language:` in functions.yml + so this is currently future-proofing, but keeps the door closed for any + future dbt version that does parse the language field. #} +{% materialization function, adapter='teradata', supported_languages=['sql'] %} + {{ return(materialization_function_default()) }} +{% endmaterialization %} + + +{# ---- scalar UDF DDL ---- #} +{% macro teradata__scalar_function_sql(target_relation) %} + REPLACE FUNCTION {{ target_relation.render() }} ({{ teradata__formatted_function_args() }}) + RETURNS {{ model.returns.data_type }} + LANGUAGE SQL + {{ teradata__function_sql_data_access() }} + {{ teradata__function_volatility_sql() }} + COLLATION INVOKER + INLINE TYPE 1 + {{ model.compiled_code }} +{% endmacro %} + + +{# ---- argument list: "name TYPE, name TYPE, ..." ---- #} +{% macro teradata__formatted_function_args() %} + {%- set args = [] -%} + {%- for arg in model.arguments -%} + {%- do args.append(arg.name ~ ' ' ~ arg.data_type) -%} + {%- endfor -%} + {{ args | join(', ') }} +{% endmacro %} + + +{# ---- data access clause ---- + Teradata SQL UDFs with LANGUAGE SQL / INLINE TYPE 1 only accept CONTAINS SQL. + NO SQL, READS SQL DATA, and MODIFIES SQL DATA are for external (C/Java) UDFs + and stored procedures — Teradata raises Error 3706 if they are used here. #} +{% macro teradata__function_sql_data_access() %} + CONTAINS SQL +{% endmacro %} + + +{# ---- volatility: maps dbt vocabulary to Teradata DETERMINISTIC / NOT DETERMINISTIC ---- #} +{% macro teradata__function_volatility_sql() %} + {%- set volatility = model.config.get('volatility') -%} + {%- if volatility == 'deterministic' -%} + DETERMINISTIC + {%- elif volatility in ('stable', 'non-deterministic') -%} + NOT DETERMINISTIC + {%- elif volatility is not none -%} + {{ exceptions.warn( + "Unsupported volatility '" ~ volatility ~ "' on function '" ~ model.name ~ "' — ignoring." + ) }} + {%- endif -%} +{% endmacro %} + + +{# ---- execution: run DDL, apply grants, persist docs, commit ---- #} +{% macro teradata__function_execute_build_sql(build_sql, existing_relation, target_relation) %} + {% do set_query_band() %} + + {% set grant_config = config.get('grants') %} + + {% call statement(name="main") %} + {{ build_sql }} + {% endcall %} + + {% set should_revoke = should_revoke(existing_relation, full_refresh_mode=True) %} + {# Grants on a function relation are rendered as EXECUTE FUNCTION DCL — see + teradata__get_grant_sql / teradata__get_revoke_sql in apply_grants.sql. #} + {% do apply_grants(target_relation, grant_config, should_revoke=should_revoke) %} + + {# persist_docs: teradata__persist_docs (persist_docs.sql) handles function relations + by emitting `COMMENT ON FUNCTION ... AS '...'` for the relation-level description. + Column-level persist_docs (`columns: true`) is skipped there with a warning, since + Teradata has no per-argument comment DDL. #} + {% do persist_docs(target_relation, model) %} + + {% do adapter.commit() %} +{% endmacro %} + + +{# ---- aggregate UDF guard: raise a clear error rather than emitting invalid DDL ---- + dbt dispatches on f"{config.type}_function_{config.language}" (see + BaseRelation.get_function_macro_name in dbt-adapters), so for `type: aggregate` / + `language: sql` the dispatched macro name is `aggregate_function_sql` — NOT + `get_aggregate_function_create_replace_signature` (that's an internal helper the + base `default__aggregate_function_sql` composition would call, but no such + top-level `default__aggregate_function_sql` macro exists in dbt-adapters today, + so without this override the failure is an opaque "no macro found" dispatch + error rather than a clear message). #} +{% macro teradata__aggregate_function_sql(target_relation) %} + {{ exceptions.raise_compiler_error( + "Aggregate user-defined functions (type: aggregate) are not supported by dbt-teradata. " + ~ "Use type: scalar for SQL scalar UDFs." + ) }} +{% endmacro %} diff --git a/dbt/include/teradata/macros/materializations/snapshot/helpers.sql b/dbt/include/teradata/macros/materializations/snapshot/helpers.sql index f38ca5b4..2a10e28b 100644 --- a/dbt/include/teradata/macros/materializations/snapshot/helpers.sql +++ b/dbt/include/teradata/macros/materializations/snapshot/helpers.sql @@ -16,6 +16,21 @@ {% macro teradata__snapshot_staging_table(strategy, source_sql, target_relation) -%} {% set columns = config.get('snapshot_table_column_names') or get_snapshot_table_column_names() %} + {%- if strategy.hard_deletes == 'new_record' %} + {# Generate a new unique scd_id for deletion records using the qualified dbt_unique_key + (VARCHAR-castable) + current timestamp to ensure uniqueness per deletion event. #} + {% if strategy.unique_key is string %} + {% set new_scd_id = snapshot_hash_arguments(['snapshotted_data.dbt_unique_key', snapshot_get_time(), "'delete'"]) %} + {% else %} + {% set _new_scd_args = [] %} + {% for key in strategy.unique_key %} + {% do _new_scd_args.append('snapshotted_data.dbt_unique_key_' ~ loop.index) %} + {% endfor %} + {% do _new_scd_args.append(snapshot_get_time()) %} + {% do _new_scd_args.append("'delete'") %} + {% set new_scd_id = snapshot_hash_arguments(_new_scd_args) %} + {% endif %} + {%- endif %} with snapshot_query as ( @@ -130,6 +145,9 @@ left join deletes_source_data as source_data on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }} where {{ unique_key_is_null(strategy.unique_key, "source_data") }} + {%- if strategy.hard_deletes == 'new_record' %} + and coalesce(snapshotted_data.{{ columns.dbt_is_deleted }}, 'False') = 'False' + {%- endif %} ) {%- endif %} @@ -146,20 +164,25 @@ {% endfor -%} {%- if strategy.unique_key | is_list -%} {%- for key in strategy.unique_key -%} - snapshotted_data.{{ key }} as dbt_unique_key_{{ loop.index }}, + snapshotted_data.dbt_unique_key_{{ loop.index }}, {% endfor -%} {%- else -%} snapshotted_data.dbt_unique_key as dbt_unique_key, {% endif -%} {{ snapshot_get_time() }} as {{ columns.dbt_updated_at }}, {{ snapshot_get_time() }} as {{ columns.dbt_valid_from }}, + {%- if config.get('dbt_valid_to_current') %} + {{ config.get('dbt_valid_to_current') }} as {{ columns.dbt_valid_to }}, + {%- else %} snapshotted_data.{{ columns.dbt_valid_to }} as {{ columns.dbt_valid_to }}, - snapshotted_data.{{ columns.dbt_scd_id }}, + {%- endif %} + {{ new_scd_id }} as {{ columns.dbt_scd_id }}, 'True' as {{ columns.dbt_is_deleted }} from snapshotted_data left join deletes_source_data as source_data on {{ unique_key_join_on(strategy.unique_key, "snapshotted_data", "source_data") }} where {{ unique_key_is_null(strategy.unique_key, "source_data") }} + and coalesce(snapshotted_data.{{ columns.dbt_is_deleted }}, 'False') = 'False' ) {%- endif %} diff --git a/dbt/include/teradata/macros/materializations/snapshot/snapshot.sql b/dbt/include/teradata/macros/materializations/snapshot/snapshot.sql index dae95e8d..ba4e0bb4 100644 --- a/dbt/include/teradata/macros/materializations/snapshot/snapshot.sql +++ b/dbt/include/teradata/macros/materializations/snapshot/snapshot.sql @@ -84,7 +84,7 @@ {% endfor %} -- Use separate DELETE + UPDATE + INSERT statements instead of the MERGE statement - {%- if strategy.invalidate_hard_deletes %} + {%- if strategy.invalidate_hard_deletes or strategy.hard_deletes == 'new_record' %} {% set final_sql_delete = teradata__snapshot_merge_sql_delete( target = target_relation, source = staging_table, @@ -107,7 +107,7 @@ ) %} - {%- if strategy.invalidate_hard_deletes %} + {%- if strategy.invalidate_hard_deletes or strategy.hard_deletes == 'new_record' %} {% call statement('main') %} {{ final_sql_delete }} {% endcall %} diff --git a/dbt/include/teradata/macros/persist_docs.sql b/dbt/include/teradata/macros/persist_docs.sql new file mode 100644 index 00000000..f54c0c98 --- /dev/null +++ b/dbt/include/teradata/macros/persist_docs.sql @@ -0,0 +1,205 @@ +{# + persist_docs support for Teradata (IDE-26225). + + Writes model / column `description:` YAML into the Teradata catalog as native + object comments using COMMENT ON TABLE|VIEW|FUNCTION|COLUMN ... AS '...'. + + Teradata specifics handled here: + * Comment string literals are escaped by doubling single quotes. + * Comments are capped at 255 characters (Teradata limit) -> truncated with a warning. + * ANSI mode forbids multi-statement DDL, so each column comment is issued in its + own statement() block rather than a single semicolon-joined batch. + * OTF / Iceberg relations (config.catalog_name set) do not support COMMENT ON and + are skipped gracefully. + * Change detection: unchanged comments issue no DDL on re-run. + * Function (UDF) relations only support the relation-level comment + (COMMENT ON FUNCTION) — Teradata has no per-argument comment DDL, so + persist_docs `columns` config is skipped with a warning for functions. +#} + +{#-- Truncates `comment` to the configured/Teradata comment limit, returning the plain + (unescaped) string. Shared by teradata_escape_comment (DDL emission) and + teradata__persist_docs / teradata__alter_column_comment (change detection), so that + both sides of the "has this comment changed?" comparison are truncated the same way. + Without this, a description longer than the limit would never match what's actually + stored (which is always truncated), so COMMENT ON DDL would be re-issued on every + run for over-length comments on persistent relations (incremental/function), rather + than only when the comment is genuinely new/changed. `warn` suppresses the + truncation warning for the change-detection call so it isn't logged twice (once + during detection, once when the DDL is actually built) when a comment is re-issued. --#} +{%- macro teradata_truncate_comment(comment, warn=True) -%} + {%- if comment is not string -%} + {%- do exceptions.raise_compiler_error('cannot escape a non-string: ' ~ comment) -%} + {%- endif -%} + {#-- Teradata stores comments in DBC ...CommentString (VARCHAR(255)); overridable via var. + `int(255)` falls back to 255 for null/non-numeric values, and a non-positive + override is ignored (also falls back to 255) so we never emit empty/garbled comments. --#} + {%- set max_len = var("teradata_max_comment_length", 255) | int(255) -%} + {%- if max_len <= 0 -%} + {%- set max_len = 255 -%} + {%- endif -%} + {%- if comment | length > max_len -%} + {%- if warn -%} + {{ exceptions.warn("Comment exceeds the configured Teradata limit of " ~ max_len ~ " characters; truncating: " ~ comment[:40] ~ "...") }} + {%- endif -%} + {%- set comment = comment[:max_len] -%} + {%- endif -%} + {{- comment -}} +{%- endmacro -%} + + +{%- macro teradata_escape_comment(comment) -%} + {%- set truncated = teradata_truncate_comment(comment) -%} + {%- set escaped = truncated | replace("'", "''") -%} + {{- "'" ~ escaped ~ "'" -}} +{%- endmacro -%} + + +{% macro teradata__alter_relation_comment(relation, comment) -%} + {%- set escaped = teradata_escape_comment(comment) -%} + {%- if relation.type == 'view' -%} + comment on view {{ relation }} as {{ escaped }} + {%- elif relation.type == 'function' -%} + comment on function {{ relation }} as {{ escaped }} + {%- else -%} + comment on table {{ relation }} as {{ escaped }} + {%- endif -%} +{%- endmacro %} + + +{#-- Fetch the current relation-level comment for change detection. Returns none if absent. --#} +{% macro teradata__get_relation_comment(relation) -%} + {% call statement('get_relation_comment', fetch_result=True) %} + SELECT CommentString FROM DBC.TablesV + WHERE DatabaseName = '{{ relation.schema }}' (NOT CASESPECIFIC) + AND TableName = '{{ relation.identifier }}' (NOT CASESPECIFIC) + {% endcall %} + {%- set result = load_result('get_relation_comment').table -%} + {%- if result and result.rows | length > 0 -%} + {%- set value = result.columns['CommentString'].values()[0] -%} + {{ return(value | trim if value is not none else none) }} + {%- endif -%} + {{ return(none) }} +{%- endmacro %} + + +{#-- Self-contained column validation (does not depend on the global validate_doc_columns, + which only exists in dbt-adapters >= 1.22.10). Warns about documented columns absent + from the database and returns only the columns that exist. Honors the per-column + `quote` flag: quoted identifiers compare case-sensitively, unquoted case-insensitively. --#} +{% macro teradata__validate_doc_columns(relation, column_dict, existing_column_names) -%} + {%- set existing_lower = existing_column_names | map("lower") | list -%} + {%- set missing = [] -%} + {%- set filtered = {} -%} + {%- for col_name in column_dict -%} + {%- if column_dict[col_name]['quote'] -%} + {%- set present = col_name in existing_column_names -%} + {%- else -%} + {%- set present = col_name | lower in existing_lower -%} + {%- endif -%} + {%- if present -%} + {%- do filtered.update({col_name: column_dict[col_name]}) -%} + {%- else -%} + {%- do missing.append(col_name) -%} + {%- endif -%} + {%- endfor -%} + {%- if missing | length > 0 -%} + {{ exceptions.warn("In relation " ~ relation.render() ~ ": The following columns are specified in the schema but are not present in the database: " ~ missing | join(", ")) }} + {%- endif -%} + {{ return(filtered) }} +{%- endmacro %} + + +{#-- Existing column comments keyed by column name, read straight from DBC.ColumnsV for + the *real* relation. This is the correct source for change detection: for views under + use_qvci=False, adapter.get_columns_in_relation reads a comment-less temp table, so + relying on it would report every column as having no comment. Querying DBC.ColumnsV + directly also avoids creating/dropping that temp table for views. --#} +{% macro teradata__get_column_comments(relation) -%} + {% call statement('get_column_comments', fetch_result=True) %} + SELECT ColumnName, CommentString FROM DBC.ColumnsV + WHERE DatabaseName = '{{ relation.schema }}' (NOT CASESPECIFIC) + AND TableName = '{{ relation.identifier }}' (NOT CASESPECIFIC) + {% endcall %} + {%- set result = load_result('get_column_comments').table -%} + {%- set comments = {} -%} + {%- if result -%} + {%- for row in result.rows -%} + {%- do comments.update({row[0]: row[1]}) -%} + {%- endfor -%} + {%- endif -%} + {{ return(comments) }} +{%- endmacro %} + + +{#-- Issues one COMMENT ON COLUMN statement per changed column (ANSI-safe: no batching). + column_dict has already been filtered to existing columns. `existing_comments` is a + {column_name: comment_string} map from teradata__get_column_comments. Note: for view + materializations REPLACE VIEW drops column comments, so on a view re-run the map is + empty and the comments are (correctly) re-applied. --#} +{% macro teradata__alter_column_comment(relation, column_dict, existing_comments) -%} + {#-- Build a lower-cased lookup so change detection mirrors the quote semantics: quoted + identifiers match case-sensitively, unquoted ones case-insensitively (a documented + `ID` still matches a physical `id`, so its comment is not re-issued every run). --#} + {%- set existing_by_lower = {} -%} + {%- for name, cmt in existing_comments.items() -%} + {%- do existing_by_lower.update({name | lower: cmt}) -%} + {%- endfor -%} + {%- for column_name in column_dict -%} + {%- set desc = column_dict[column_name]['description'] -%} + {%- set quoted = column_dict[column_name]['quote'] -%} + {%- set rendered_col = adapter.quote(column_name) if quoted else column_name -%} + {%- set existing_raw = existing_comments.get(column_name) if quoted else existing_by_lower.get(column_name | lower) -%} + {%- set existing_comment = (existing_raw | trim) if existing_raw is not none else none -%} + {#-- Compare trimmed, truncated forms: YAML block scalars carry a trailing newline that + Teradata does not store, and `existing_comment` (read back from the catalog) is + always <= the Teradata comment limit, so comparing against the raw untruncated + `desc` would never match for over-length descriptions, re-issuing DDL every run. --#} + {%- if existing_comment != teradata_truncate_comment(desc | trim, warn=False) -%} + {%- call statement('alter_column_comment_' ~ loop.index, fetch_result=False) -%} + comment on column {{ relation }}.{{ rendered_col }} as {{ teradata_escape_comment(desc) }} + {%- endcall -%} + {%- endif -%} + {%- endfor -%} +{%- endmacro %} + + +{% macro teradata__persist_docs(relation, model, for_relation, for_columns) -%} + {#-- OTF / Iceberg: native COMMENT ON is unsupported -> skip gracefully. --#} + {%- if config.get('catalog_name') is not none -%} + {{ log("persist_docs skipped for OTF relation " ~ relation, info=false) }} + {{ return('') }} + {%- endif -%} + + {% if for_relation and config.persist_relation_docs() and model.description %} + {%- set existing_rel_comment = teradata__get_relation_comment(relation) -%} + {#-- Compare against the truncated form: what's stored can never exceed the Teradata + comment limit, so comparing against the raw untruncated description would never + match for over-length descriptions, re-issuing COMMENT ON DDL on every run. --#} + {%- if existing_rel_comment != teradata_truncate_comment(model.description | trim, warn=False) -%} + {% do run_query(teradata__alter_relation_comment(relation, model.description)) %} + {%- endif -%} + {% endif %} + + {#-- Functions: Teradata has no COMMENT ON DDL, only a single + comment for the whole function, so per-column (per-argument) comments + cannot be honored. Warn and skip rather than emitting invalid DDL. + NOTE: as of dbt-core 1.11, functions.yml's `columns:` block is not parsed + into model.columns (UnparsedFunctionUpdate uses HasColumnProps, not + HasColumnDocs), so model.columns is always {} here in practice and this + branch is currently unreachable via YAML — it is intentional future-proofing + in case a later dbt-core version starts populating model.columns for + functions. Without this guard, that scenario would silently fall through to + the elif below and attempt COMMENT ON COLUMN against a function argument, + which is invalid Teradata DDL and would raise a database error rather than + a graceful warning. Covered directly in tests/unit/test_persist_docs.py + (TestPersistDocsFunctionColumnsSkipped) via a synthetic model.columns dict. --#} + {% if for_columns and config.persist_column_docs() and model.columns and relation.type == 'function' %} + {{ exceptions.warn("persist_docs 'columns' config is not supported for Teradata functions (arguments cannot carry individual comments); skipping column comments for '" ~ relation ~ "'.") }} + {% elif for_columns and config.persist_column_docs() and model.columns %} + {% set existing_comments = teradata__get_column_comments(relation) %} + {% set existing_column_names = existing_comments.keys() | list %} + {% set filtered_columns = teradata__validate_doc_columns(relation, model.columns, existing_column_names) %} + {% do teradata__alter_column_comment(relation, filtered_columns, existing_comments) %} + {% endif %} +{%- endmacro %} diff --git a/requirements_dev.txt b/requirements_dev.txt index 569f1ea8..3f215741 100644 --- a/requirements_dev.txt +++ b/requirements_dev.txt @@ -8,9 +8,9 @@ pytest~=7.0 tox~=3.2 pylava~=0.3.0 teradatasql>=20.00.00.10 -dbt-adapters>=1.17.2 -dbt-common>=1.13,<2.0 -dbt-core>=1.10.0,<2.0 +dbt-adapters>=1.20.0 +dbt-common>=1.37.2,<2.0 +dbt-core>=1.11.0,<2.0 MarkupSafe==2.0.1 pytest-dotenv pytest-cov diff --git a/setup.py b/setup.py index 7050a3c9..87af8873 100644 --- a/setup.py +++ b/setup.py @@ -48,8 +48,8 @@ ], }, install_requires=[ - "dbt-adapters>=1.17.2", - "dbt-common>=1.13,<2.0", + "dbt-adapters>=1.20.0", + "dbt-common>=1.37.2,<2.0", "teradatasql>=20.00.00.10", ], classifiers=[ diff --git a/tests/functional/adapter/simple_snapshot/test_new_record_mode.py b/tests/functional/adapter/simple_snapshot/test_new_record_mode.py index 74951bfe..497ad8eb 100644 --- a/tests/functional/adapter/simple_snapshot/test_new_record_mode.py +++ b/tests/functional/adapter/simple_snapshot/test_new_record_mode.py @@ -1,6 +1,8 @@ import pytest -from dbt.tests.util import check_relations_equal, run_dbt, relation_from_name +from dbt.tests.util import run_dbt, relation_from_name + +# -- SQL fixtures ------------------------------------------------------- _seed_new_record_mode = """ create table {schema}.seed ( @@ -14,7 +16,7 @@ ); """ -create_snapshot_expected_sql=""" +create_snapshot_expected_sql = """ create table {schema}.snapshot_expected ( id INTEGER, first_name VARCHAR(50), @@ -32,10 +34,8 @@ dbt_is_deleted varchar(50) ); """ -seed_insert_sql=""" --- seed inserts --- use the same email for two users to verify that duplicated check_cols values --- are handled appropriately + +seed_insert_sql = """ insert into {schema}.seed (id, first_name, last_name, email, gender, ip_address, updated_at) values (1, 'Judith', 'Kennedy', '(not provided)', 'Female', '54.60.24.128', '2015-12-24 12:19:28'); insert into {schema}.seed (id, first_name, last_name, email, gender, ip_address, updated_at) values @@ -78,36 +78,19 @@ (20, 'Phyllis', 'Fox', null, 'Female', '163.191.232.95', '2016-08-21 10:35:19'); """ -populate_snapshot_expected_sql=""" --- populate snapshot table +populate_snapshot_expected_sql = """ insert into {schema}.snapshot_expected ( - id, - first_name, - last_name, - email, - gender, - ip_address, - updated_at, - dbt_valid_from, - dbt_valid_to, - dbt_updated_at, - dbt_scd_id, - dbt_is_deleted + id, first_name, last_name, email, gender, ip_address, + updated_at, dbt_valid_from, dbt_valid_to, dbt_updated_at, dbt_scd_id, dbt_is_deleted ) - select - id, - first_name, - last_name, - email, - gender, - ip_address, + id, first_name, last_name, email, gender, ip_address, updated_at, - -- fields added by snapshotting updated_at as dbt_valid_from, cast(null as timestamp) as dbt_valid_to, updated_at as dbt_updated_at, - HASHROW(coalesce(cast(id || '-' || first_name as varchar(50)), '') || '|' || coalesce(cast(updated_at as varchar(50)), '')) as test_scd_id, + HASHROW(coalesce(cast(id || '-' || first_name as varchar(50)), '') + || '|' || coalesce(cast(updated_at as varchar(50)), '')) as dbt_scd_id, 'False' as dbt_is_deleted from {schema}.seed; """ @@ -139,53 +122,58 @@ select * from {{ ref('snapshot_actual') }} """ +# `unique_key` here is a list (composite key), unlike `_snapshot_actual_sql` above which uses +# a string expression. This exercises the `strategy.unique_key | is_list` branch of the +# `deletion_records` CTE in teradata__snapshot_staging_table (dbt/include/teradata/macros/ +# materializations/snapshot/helpers.sql), i.e. `snapshotted_data.dbt_unique_key_{{ loop.index }}` +# and the multi-column `new_scd_id` hashing, which is otherwise untested. +_snapshot_actual_composite_key_sql = """ +{% snapshot snapshot_actual %} + + {{ + config( + unique_key=['id', 'first_name'], + ) + }} + + select * from {{target.schema}}.seed + +{% endsnapshot %} +""" + +_snapshots_composite_key_yml = """ +snapshots: + - name: snapshot_actual + config: + strategy: timestamp + updated_at: updated_at + hard_deletes: new_record +""" _invalidate_sql = """ --- update records 11 - 21. Change email and updated_at field update {schema}.seed set updated_at = updated_at + interval '1' hour, - email = case when id = 20 then 'pfoxj@creativecommons.org' else 'new_' || email end + email = case when id = 20 then 'pfoxj@creativecommons.org' else 'new_' || email end where id >= 10 and id <= 20; - --- invalidate records 11 - 21 update {schema}.snapshot_expected set dbt_valid_to = updated_at + interval '1' hour where id >= 10 and id <= 20; - """ _update_sql = """ --- insert v2 of the 11 - 21 records - insert into {schema}.snapshot_expected ( - id, - first_name, - last_name, - email, - gender, - ip_address, - updated_at, - dbt_valid_from, - dbt_valid_to, - dbt_updated_at, - dbt_scd_id, - dbt_is_deleted + id, first_name, last_name, email, gender, ip_address, + updated_at, dbt_valid_from, dbt_valid_to, dbt_updated_at, dbt_scd_id, dbt_is_deleted ) - select - id, - first_name, - last_name, - email, - gender, - ip_address, + id, first_name, last_name, email, gender, ip_address, updated_at, - -- fields added by snapshotting updated_at as dbt_valid_from, cast(null as timestamp) as dbt_valid_to, updated_at as dbt_updated_at, - HASHROW(coalesce(cast(id || '-' || first_name as varchar(50)), '') || '|' || coalesce(cast(updated_at as varchar(50)), '')) as test_scd_id, + HASHROW(coalesce(cast(id || '-' || first_name as varchar(50)), '') + || '|' || coalesce(cast(updated_at as varchar(50)), '')) as dbt_scd_id, 'False' as dbt_is_deleted from {schema}.seed where id >= 10 and id <= 20; @@ -195,6 +183,49 @@ delete from {schema}.seed where id = 1 """ +# -- Helper ------------------------------------------------------- + +def _drop_table_safe(project, table_name): + """Drop a table if it exists.""" + relation = relation_from_name(project.adapter, table_name) + try: + project.run_sql(f"DROP TABLE /*+ IF EXISTS */ {relation}") + except Exception as ex: + # Teradata adapter suppresses "does not exist" errors for /*+ IF EXISTS */ + # (errors 3807, 3854, 3853, 7825, 6321), but in test context via project.run_sql() + # we still need explicit handling for Error 3807 + if "[Error 3807]" not in str(ex): + raise + + +def _reset_tables(project): + """Drop seed and snapshot tables so each test starts clean.""" + _drop_table_safe(project, "seed") + _drop_table_safe(project, "snapshot_actual") + _drop_table_safe(project, "snapshot_expected") + + +def _get_snapshot_rows(project, columns="*", where="1=1"): + """Return rows from the snapshot_actual table.""" + relation = relation_from_name(project.adapter, "snapshot_actual") + return project.run_sql( + f"select {columns} from {relation} where {where}", fetch="all" + ) + + +def _assert_snapshot_success(results): + """Assert that exactly one snapshot ran and it succeeded. + + Teradata snapshots return 'activity: Insert, rows_affected: N' rather than + 'success', so we accept both to support all snapshot materialization outcomes. + """ + assert len(results) == 1 + status = str(results[0].status) + assert status == "success" or status.startswith("activity:"), \ + f"Unexpected snapshot status: {status}" + + +# -- Test class ------------------------------------------------------- class SnapshotNewRecordMode: @pytest.fixture(scope="class") @@ -220,43 +251,364 @@ def update_sql(self): def delete_sql(self): return _delete_sql - def test_snapshot_new_record_mode( - self, project, invalidate_sql, update_sql - ): + def test_snapshot_new_record_mode(self, project, invalidate_sql, update_sql): + """Test initial load + updates with expected-table comparison.""" + _reset_tables(project) project.run_sql(_seed_new_record_mode) project.run_sql(create_snapshot_expected_sql) project.run_sql(seed_insert_sql) project.run_sql(populate_snapshot_expected_sql) + # --- Run 1: initial snapshot load --- results = run_dbt(["snapshot"]) - assert len(results) == 1 + _assert_snapshot_success(results) + # --- Run 2: update records 10-20 and re-snapshot --- project.run_sql(invalidate_sql) project.run_sql(update_sql) results = run_dbt(["snapshot"]) - assert len(results) == 1 + _assert_snapshot_success(results) + # Verify snapshot matches expected table (bidirectional MINUS) relation_actual = relation_from_name(project.adapter, "snapshot_actual") relation_expected = relation_from_name(project.adapter, "snapshot_expected") - result = project.run_sql(f"select id, first_name, last_name, email, gender, ip_address, updated_at, dbt_valid_from, dbt_valid_to, dbt_scd_id, dbt_updated_at, dbt_is_deleted from {relation_actual} \ - minus \ - select id, first_name, last_name, email, gender, ip_address, updated_at, dbt_valid_from, dbt_valid_to, dbt_scd_id, dbt_updated_at, dbt_is_deleted from {relation_expected}", fetch="one") - - # if two expected and actual snapshot tables are equal then the result varible would be None, as there would no difference between the two relations - assert result == None + cols = "id, first_name, last_name, email, gender, ip_address, updated_at, dbt_valid_from, dbt_valid_to, dbt_scd_id, dbt_updated_at, dbt_is_deleted" + + result = project.run_sql( + f"select {cols} from {relation_actual} minus select {cols} from {relation_expected}", + fetch="one", + ) + assert result is None, f"Rows in actual but not expected: {result}" + + result2 = project.run_sql( + f"select {cols} from {relation_expected} minus select {cols} from {relation_actual}", + fetch="one", + ) + assert result2 is None, f"Rows in expected but not actual: {result2}" + + # --- Run 3: delete record id=1 and snapshot --- + project.run_sql(_delete_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + def test_hard_delete_creates_new_record_with_correct_flags(self, project): + """After deleting a source record, the snapshot should contain: + - The original record with dbt_valid_to set (closed) and dbt_is_deleted='False' + - A new deletion record with dbt_is_deleted='True' and dbt_valid_to IS NULL + """ + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + # Initial snapshot + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # All 20 records should be active (dbt_valid_to IS NULL, dbt_is_deleted='False') + rows = _get_snapshot_rows(project, "count(*)", "dbt_valid_to is null and dbt_is_deleted = 'False'") + assert rows[0][0] == 20, f"Expected 20 active records, got {rows[0][0]}" + + # Delete record id=1 + project.run_sql(_delete_sql) + + # Snapshot after delete + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Original record for id=1 should now be closed (dbt_valid_to IS NOT NULL) + closed_rows = _get_snapshot_rows( + project, + "count(*)", + "id = 1 and dbt_valid_to is not null and dbt_is_deleted = 'False'", + ) + assert closed_rows[0][0] == 1, \ + f"Expected 1 closed original record for id=1, got {closed_rows[0][0]}" + + # A new deletion record should exist with dbt_is_deleted='True' + deleted_rows = _get_snapshot_rows( + project, + "count(*)", + "id = 1 and dbt_is_deleted = 'True'", + ) + assert deleted_rows[0][0] == 1, \ + f"Expected 1 deletion record for id=1, got {deleted_rows[0][0]}" + + def test_hard_delete_produces_unique_scd_id(self, project): + """The deletion record must have a different dbt_scd_id than the original record.""" + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Delete record id=1 + project.run_sql(_delete_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Fetch original and deletion rows separately to avoid relying on result ordering + original_rows = _get_snapshot_rows(project, "dbt_scd_id", "id = 1 and dbt_is_deleted = 'False'") + deleted_rows = _get_snapshot_rows(project, "dbt_scd_id", "id = 1 and dbt_is_deleted = 'True'") + assert len(original_rows) == 1, f"Expected 1 original record for id=1, got {len(original_rows)}" + assert len(deleted_rows) == 1, f"Expected 1 deletion record for id=1, got {len(deleted_rows)}" + + scd_id_original = original_rows[0][0] + scd_id_deleted = deleted_rows[0][0] + assert scd_id_original != scd_id_deleted, \ + f"dbt_scd_id must be unique: original={scd_id_original}, deletion={scd_id_deleted}" + + # Verify global uniqueness — no duplicate dbt_scd_id in the entire snapshot + relation = relation_from_name(project.adapter, "snapshot_actual") + dups = project.run_sql( + f"select dbt_scd_id, count(*) as cnt from {relation} group by dbt_scd_id having count(*) > 1", + fetch="all", + ) + assert len(dups) == 0, f"Found duplicate dbt_scd_id values: {dups}" + + def test_snapshot_idempotent_after_delete(self, project): + """Running snapshot multiple times after a delete should NOT create duplicate records.""" + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + # Initial snapshot + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Delete record id=1 + project.run_sql(_delete_sql) + + # First snapshot after delete + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Count total records + count_after_first = _get_snapshot_rows(project, "count(*)") + total_after_first = count_after_first[0][0] + + # Run snapshot 3 more times — count should NOT change + for _ in range(3): + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + count_after_repeats = _get_snapshot_rows(project, "count(*)") + total_after_repeats = count_after_repeats[0][0] + + assert total_after_first == total_after_repeats, ( + f"Record count changed from {total_after_first} to {total_after_repeats} " + f"after 3 idempotent snapshot runs — exponential duplication bug!" + ) - result2 = project.run_sql(f"select id, first_name, last_name, email, gender, ip_address, updated_at, dbt_valid_from, dbt_valid_to, dbt_scd_id, dbt_updated_at, dbt_is_deleted from {relation_expected} \ - minus \ - select id, first_name, last_name, email, gender, ip_address, updated_at, dbt_valid_from, dbt_valid_to, dbt_scd_id, dbt_updated_at, dbt_is_deleted from {relation_actual}", fetch="one") - assert result2 == None - # check_relations_equal(project.adapter, ["snapshot_actual", "snapshot_expected"]) + def test_non_deleted_records_unaffected(self, project): + """Records that were NOT deleted should remain unchanged after delete + snapshot.""" + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + # Delete only id=1 (Judith) project.run_sql(_delete_sql) results = run_dbt(["snapshot"]) - assert len(results) == 1 + _assert_snapshot_success(results) + + # All other 19 records should still be active with dbt_is_deleted='False' + active_rows = _get_snapshot_rows( + project, + "count(*)", + "id <> 1 and dbt_valid_to is null and dbt_is_deleted = 'False'", + ) + assert active_rows[0][0] == 19, \ + f"Expected 19 unaffected active records, got {active_rows[0][0]}" + + # No other record should have dbt_is_deleted='True' + other_deleted = _get_snapshot_rows( + project, + "count(*)", + "id <> 1 and dbt_is_deleted = 'True'", + ) + assert other_deleted[0][0] == 0, \ + f"Expected 0 deletion records for non-deleted sources, got {other_deleted[0][0]}" + class TestSnapshotNewRecordModeTeradata(SnapshotNewRecordMode): - pass \ No newline at end of file + pass + + +class SnapshotNewRecordModeCompositeKey: + """ + Same `hard_deletes: new_record` behavior as `SnapshotNewRecordMode`, but with a + composite/list `unique_key` (['id', 'first_name']) instead of a string expression. + This is the code path a reviewer flagged in PR #241: the `deletion_records` CTE + references `snapshotted_data.dbt_unique_key_{{ loop.index }}` directly instead of + re-deriving it from the raw key column, relying on the `snapshotted_data` CTE already + exposing those columns (via the shared `unique_key_fields` macro). These tests confirm + that assumption holds end-to-end on a live Teradata instance. + """ + + @pytest.fixture(scope="class") + def snapshots(self): + return {"snapshot.sql": _snapshot_actual_composite_key_sql} + + @pytest.fixture(scope="class") + def models(self): + return { + "snapshots.yml": _snapshots_composite_key_yml, + "ref_snapshot.sql": _ref_snapshot_sql, + } + + def test_hard_delete_creates_new_record_with_correct_flags(self, project): + """After deleting a source record, the snapshot should contain: + - The original record with dbt_valid_to set (closed) and dbt_is_deleted='False' + - A new deletion record with dbt_is_deleted='True' and dbt_valid_to IS NULL + """ + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + # Initial snapshot + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # All 20 records should be active (dbt_valid_to IS NULL, dbt_is_deleted='False') + rows = _get_snapshot_rows(project, "count(*)", "dbt_valid_to is null and dbt_is_deleted = 'False'") + assert rows[0][0] == 20, f"Expected 20 active records, got {rows[0][0]}" + + # Delete record id=1 + project.run_sql(_delete_sql) + + # Snapshot after delete + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Original record for id=1 should now be closed (dbt_valid_to IS NOT NULL) + closed_rows = _get_snapshot_rows( + project, + "count(*)", + "id = 1 and dbt_valid_to is not null and dbt_is_deleted = 'False'", + ) + assert closed_rows[0][0] == 1, \ + f"Expected 1 closed original record for id=1, got {closed_rows[0][0]}" + + # A new deletion record should exist with dbt_is_deleted='True' + deleted_rows = _get_snapshot_rows( + project, + "count(*)", + "id = 1 and dbt_is_deleted = 'True'", + ) + assert deleted_rows[0][0] == 1, \ + f"Expected 1 deletion record for id=1, got {deleted_rows[0][0]}" + + def test_hard_delete_produces_unique_scd_id(self, project): + """The deletion record must have a different dbt_scd_id than the original record, + with the scd_id computed from the composite dbt_unique_key_1 / dbt_unique_key_2 columns.""" + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Delete record id=1 + project.run_sql(_delete_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Fetch original and deletion rows separately to avoid relying on result ordering + original_rows = _get_snapshot_rows(project, "dbt_scd_id", "id = 1 and dbt_is_deleted = 'False'") + deleted_rows = _get_snapshot_rows(project, "dbt_scd_id", "id = 1 and dbt_is_deleted = 'True'") + assert len(original_rows) == 1, f"Expected 1 original record for id=1, got {len(original_rows)}" + assert len(deleted_rows) == 1, f"Expected 1 deletion record for id=1, got {len(deleted_rows)}" + + scd_id_original = original_rows[0][0] + scd_id_deleted = deleted_rows[0][0] + assert scd_id_original != scd_id_deleted, \ + f"dbt_scd_id must be unique: original={scd_id_original}, deletion={scd_id_deleted}" + + # Verify global uniqueness — no duplicate dbt_scd_id in the entire snapshot + relation = relation_from_name(project.adapter, "snapshot_actual") + dups = project.run_sql( + f"select dbt_scd_id, count(*) as cnt from {relation} group by dbt_scd_id having count(*) > 1", + fetch="all", + ) + assert len(dups) == 0, f"Found duplicate dbt_scd_id values: {dups}" + + def test_snapshot_idempotent_after_delete(self, project): + """Running snapshot multiple times after a delete should NOT create duplicate records, + with a composite unique_key.""" + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + # Initial snapshot + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Delete record id=1 + project.run_sql(_delete_sql) + + # First snapshot after delete + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Count total records + count_after_first = _get_snapshot_rows(project, "count(*)") + total_after_first = count_after_first[0][0] + + # Run snapshot 3 more times — count should NOT change + for _ in range(3): + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + count_after_repeats = _get_snapshot_rows(project, "count(*)") + total_after_repeats = count_after_repeats[0][0] + + assert total_after_first == total_after_repeats, ( + f"Record count changed from {total_after_first} to {total_after_repeats} " + f"after 3 idempotent snapshot runs — exponential duplication bug!" + ) + + def test_non_deleted_records_unaffected(self, project): + """Records that were NOT deleted should remain unchanged after delete + snapshot.""" + _reset_tables(project) + project.run_sql(_seed_new_record_mode) + project.run_sql(seed_insert_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # Delete only id=1 (Judith) + project.run_sql(_delete_sql) + + results = run_dbt(["snapshot"]) + _assert_snapshot_success(results) + + # All other 19 records should still be active with dbt_is_deleted='False' + active_rows = _get_snapshot_rows( + project, + "count(*)", + "id <> 1 and dbt_valid_to is null and dbt_is_deleted = 'False'", + ) + assert active_rows[0][0] == 19, \ + f"Expected 19 unaffected active records, got {active_rows[0][0]}" + + # No other record should have dbt_is_deleted='True' + other_deleted = _get_snapshot_rows( + project, + "count(*)", + "id <> 1 and dbt_is_deleted = 'True'", + ) + assert other_deleted[0][0] == 0, \ + f"Expected 0 deletion records for non-deleted sources, got {other_deleted[0][0]}" + + +class TestSnapshotNewRecordModeCompositeKeyTeradata(SnapshotNewRecordModeCompositeKey): + pass diff --git a/tests/functional/adapter/teradata_dbt/test_validate_teradata_persist_docs.py b/tests/functional/adapter/teradata_dbt/test_validate_teradata_persist_docs.py new file mode 100644 index 00000000..a9b77031 --- /dev/null +++ b/tests/functional/adapter/teradata_dbt/test_validate_teradata_persist_docs.py @@ -0,0 +1,424 @@ +""" +Functional tests for persist_docs on Teradata (IDE-26225). + +Reuses the shared dbt-tests-adapter BasePersistDocs suite (validates relation + +column comments surface in catalog.json, quote/case-sensitivity handling, and +missing-column warnings) and adds Teradata-specific coverage: + + * per-materialization - view / incremental / seed / snapshot comments + * special characters - quotes, --, /* */ are escaped and round-trip + * idempotency - a re-run with unchanged descriptions issues no COMMENT ON DDL + * changed description - updated text re-issues the comment DDL + * truncation - a >255 char description is truncated, succeeds, and (on a + persistent incremental relation) is idempotent: a second run + with the unchanged over-length description issues no COMMENT + ON DDL, since change detection compares truncated forms. + +OTF/Iceberg skip (models with catalog_name set) is handled in teradata__persist_docs +but is not covered here, as it requires an OTF catalog that is not available in this +functional test environment. +""" +import json +import os + +import pytest + +from dbt.tests.util import run_dbt, run_dbt_and_capture + +from dbt.tests.adapter.persist_docs.test_persist_docs import ( + BasePersistDocsColumnMissing, + BasePersistDocsAllColumnsMissing, + BasePersistDocsQuotedColumnCaseSensitive, + BasePersistDocsQuotedDescriptionNotAppliedOnMismatch, + BasePersistDocsCommentOnQuotedColumn, +) + + +# --------------------------------------------------------------------------- +# Shared dbt-tests-adapter suite +# +# NOTE: The flagship BasePersistDocs.test_has_comments_pglike is intentionally +# NOT subclassed. Its shared fixtures produce a `name` column comment of ~280 +# characters (after resolving a doc() block) and assert the full text round-trips. +# Teradata stores comments in DBC ...CommentString (VARCHAR(255)), so text beyond +# 255 chars cannot round-trip. The Teradata relation/column round-trip is instead +# covered by TestPersistDocsRoundtripTeradata below using <=255 char descriptions. +# The remaining shared classes use short fixtures that fit the limit. +# --------------------------------------------------------------------------- +class TestPersistDocsColumnMissingTeradata(BasePersistDocsColumnMissing): + pass + + +class TestPersistDocsAllColumnsMissingTeradata(BasePersistDocsAllColumnsMissing): + pass + + +class TestPersistDocsQuotedColumnCaseSensitiveTeradata( + BasePersistDocsQuotedColumnCaseSensitive +): + pass + + +class TestPersistDocsQuotedDescriptionNotAppliedOnMismatchTeradata( + BasePersistDocsQuotedDescriptionNotAppliedOnMismatch +): + pass + + +class TestPersistDocsCommentOnQuotedColumnTeradata(BasePersistDocsCommentOnQuotedColumn): + pass + + +# --------------------------------------------------------------------------- +# Teradata-specific coverage +# --------------------------------------------------------------------------- +_MODEL_SQL = "select 1 as id, 'a' as name" + +_LONG_DESC = "x" * 400 # exceeds the 255-char Teradata comment limit + +_SCHEMA_YML = """ +version: 2 +models: + - name: persist_model + description: "Relation level description for persist_docs test" + columns: + - name: id + description: "The id column description" + - name: name + description: "The name column description" +""" + +_SCHEMA_LONG_YML = """ +version: 2 +models: + - name: long_desc_model + description: "{long}" +""".replace("{long}", _LONG_DESC) + + +class TestPersistDocsRoundtripTeradata: + """Teradata-native equivalent of the flagship BasePersistDocs test: relation and + column descriptions (<=255 chars) round-trip into catalog.json via COMMENT ON.""" + + @pytest.fixture(scope="class") + def models(self): + return {"persist_model.sql": _MODEL_SQL, "schema.yml": _SCHEMA_YML} + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "models": { + "test": { + "materialized": "table", + "+persist_docs": {"relation": True, "columns": True}, + } + } + } + + def test_comments_roundtrip_to_catalog(self, project): + run_dbt(["run"]) + run_dbt(["docs", "generate"]) + catalog_path = os.path.join(project.project_root, "target", "catalog.json") + with open(catalog_path) as fp: + catalog = json.load(fp) + node = catalog["nodes"]["model.test.persist_model"] + assert node["metadata"]["comment"].startswith("Relation level description") + assert node["columns"]["id"]["comment"].startswith("The id column description") + assert node["columns"]["name"]["comment"].startswith("The name column description") + + +_INCR_IDEMPOTENT = ( + "{{ config(materialized='incremental') }}\n" + "select 1 as id, cast('a' as varchar(10)) as name" +) + + +class TestPersistDocsIdempotentTeradata: + """A second run with unchanged descriptions must issue no COMMENT ON DDL. + + Uses an *incremental* model: table/view materializations drop-and-recreate the + object every run (so the comment is legitimately re-applied), whereas an + incremental relation persists across runs, which is where change detection + (skip DDL when the comment is unchanged) actually applies. + """ + + @pytest.fixture(scope="class") + def models(self): + return {"persist_model.sql": _INCR_IDEMPOTENT, "schema.yml": _SCHEMA_YML} + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "models": { + "test": { + "materialized": "incremental", + "+persist_docs": {"relation": True, "columns": True}, + } + } + } + + def test_no_ddl_on_unchanged_rerun(self, project): + run_dbt(["run"]) + # Second run: object persists and descriptions are unchanged -> no COMMENT DDL. + # COMMENT DDL is only emitted at DEBUG level, so capture with --debug. + _, logs = run_dbt_and_capture(["--debug", "run"]) + assert "comment on table" not in logs.lower() + assert "comment on column" not in logs.lower() + + +class TestPersistDocsLongCommentTeradata: + """A description longer than 255 chars is truncated, the run succeeds, and a + second run with the same (unchanged, over-length) description is idempotent: + change detection must compare truncated forms on both sides, or it would + re-issue COMMENT ON DDL on every run (see PR #241 review discussion). + + Uses an *incremental* model so the relation persists across runs — table/view + materializations drop-and-recreate the object every run, which would mask an + idempotency regression (the comment would be legitimately re-applied either way). + """ + + @pytest.fixture(scope="class") + def models(self): + return {"long_desc_model.sql": _INCR_IDEMPOTENT, "schema.yml": _SCHEMA_LONG_YML} + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "models": { + "test": { + "materialized": "incremental", + "+persist_docs": {"relation": True, "columns": False}, + } + } + } + + def test_long_comment_truncated_ok(self, project): + results = run_dbt(["run"]) + assert len(results) == 1 + assert results[0].status == "success" + + def test_long_comment_idempotent_on_rerun(self, project): + run_dbt(["run"]) + # Second run: relation persists and the (over-length) description is + # unchanged -> no COMMENT ON DDL. COMMENT DDL is only emitted at DEBUG + # level, so capture with --debug. + _, logs = run_dbt_and_capture(["--debug", "run"]) + assert "comment on table" not in logs.lower() + + +# --------------------------------------------------------------------------- +# Per-materialization coverage (view / incremental / seed) in a single build +# to minimize round-trips against the (latency-heavy) database. +# --------------------------------------------------------------------------- +_VIEW_MODEL = "{{ config(materialized='view') }}\nselect 1 as id, cast('a' as varchar(10)) as nm" +_INCR_MODEL = "{{ config(materialized='incremental') }}\nselect 1 as id, cast('a' as varchar(10)) as nm" +_SEED_CSV = "id,nm\n1,a\n2,b\n" + +_MATS_SCHEMA_YML = """ +version: 2 +models: + - name: view_mat + description: "View materialization relation description" + columns: + - name: id + description: "View id column description" + - name: incr_mat + description: "Incremental materialization relation description" + columns: + - name: id + description: "Incremental id column description" +seeds: + - name: seed_mat + description: "Seed relation description" + columns: + - name: nm + description: "Seed nm column description" +""" + + +class TestPersistDocsMaterializationsTeradata: + """persist_docs works across view, incremental and seed materializations.""" + + @pytest.fixture(scope="class") + def seeds(self): + return {"seed_mat.csv": _SEED_CSV} + + @pytest.fixture(scope="class") + def models(self): + return { + "view_mat.sql": _VIEW_MODEL, + "incr_mat.sql": _INCR_MODEL, + "schema.yml": _MATS_SCHEMA_YML, + } + + @pytest.fixture(scope="class") + def project_config_update(self): + pd = {"relation": True, "columns": True} + return { + "models": {"test": {"+persist_docs": pd}}, + "seeds": {"test": {"+persist_docs": pd}}, + } + + def test_comments_on_all_materializations(self, project): + run_dbt(["seed"]) + run_dbt(["run"]) + run_dbt(["docs", "generate"]) + with open(os.path.join(project.project_root, "target", "catalog.json")) as fp: + catalog = json.load(fp) + nodes = catalog["nodes"] + + view_node = nodes["model.test.view_mat"] + assert view_node["metadata"]["comment"].startswith("View materialization relation") + assert view_node["columns"]["id"]["comment"].startswith("View id column") + + incr_node = nodes["model.test.incr_mat"] + assert incr_node["metadata"]["comment"].startswith("Incremental materialization relation") + assert incr_node["columns"]["id"]["comment"].startswith("Incremental id column") + + seed_node = nodes["seed.test.seed_mat"] + assert seed_node["metadata"]["comment"].startswith("Seed relation description") + assert seed_node["columns"]["nm"]["comment"].startswith("Seed nm column") + + +# --------------------------------------------------------------------------- +# Special characters must be escaped and round-trip intact. +# --------------------------------------------------------------------------- +_SPECIAL_DESC = "It's a \"quoted\" desc; with -- dash and /* block */ tokens" +# Use YAML block scalars so embedded single/double quotes need no YAML escaping. +# `|-` strips the trailing newline, so the value equals _SPECIAL_DESC exactly. +_SPECIAL_SCHEMA_YML = """ +version: 2 +models: + - name: special_model + description: |- + It's a "quoted" desc; with -- dash and /* block */ tokens + columns: + - name: id + description: |- + col with O'Brien apostrophe +""" + + +class TestPersistDocsSpecialCharsTeradata: + """Single quotes, double quotes and SQL-comment tokens are escaped and round-trip.""" + + @pytest.fixture(scope="class") + def models(self): + return {"special_model.sql": _MODEL_SQL, "schema.yml": _SPECIAL_SCHEMA_YML} + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "models": { + "test": { + "materialized": "table", + "+persist_docs": {"relation": True, "columns": True}, + } + } + } + + def test_special_chars_roundtrip(self, project): + run_dbt(["run"]) + run_dbt(["docs", "generate"]) + with open(os.path.join(project.project_root, "target", "catalog.json")) as fp: + catalog = json.load(fp) + node = catalog["nodes"]["model.test.special_model"] + assert node["metadata"]["comment"] == _SPECIAL_DESC + assert node["columns"]["id"]["comment"] == "col with O'Brien apostrophe" + + +# --------------------------------------------------------------------------- +# A changed description re-issues exactly the comment DDL (complement to idempotency). +# --------------------------------------------------------------------------- +_SCHEMA_V1 = """ +version: 2 +models: + - name: change_model + description: "Original description" +""" +_SCHEMA_V2 = """ +version: 2 +models: + - name: change_model + description: "Updated description" +""" + + +class TestPersistDocsChangedDescriptionTeradata: + """Changing the description re-issues the relation COMMENT and the new text persists.""" + + @pytest.fixture(scope="class") + def models(self): + return {"change_model.sql": _MODEL_SQL, "schema.yml": _SCHEMA_V1} + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "models": { + "test": { + "materialized": "table", + "+persist_docs": {"relation": True, "columns": False}, + } + } + } + + def test_changed_description_reissues_ddl(self, project): + run_dbt(["run"]) + # rewrite the schema with a new description, then re-run + schema_path = os.path.join(project.project_root, "models", "schema.yml") + with open(schema_path, "w") as fp: + fp.write(_SCHEMA_V2) + # COMMENT DDL is only emitted at DEBUG level. + _, logs = run_dbt_and_capture(["--debug", "run"]) + assert "comment on table" in logs.lower() + run_dbt(["docs", "generate"]) + with open(os.path.join(project.project_root, "target", "catalog.json")) as fp: + catalog = json.load(fp) + node = catalog["nodes"]["model.test.change_model"] + assert node["metadata"]["comment"].startswith("Updated description") + + +# --------------------------------------------------------------------------- +# Snapshot materialization persists docs. +# --------------------------------------------------------------------------- +_SNAPSHOT_SQL = """ +{% snapshot cmt_snapshot %} +{{ config(target_schema=schema, unique_key='id', strategy='check', check_cols=['nm']) }} +select 1 as id, cast('a' as varchar(10)) as nm +{% endsnapshot %} +""" +_SNAPSHOT_SCHEMA_YML = """ +version: 2 +snapshots: + - name: cmt_snapshot + description: "Snapshot relation description" + columns: + - name: id + description: "Snapshot id column description" +""" + + +class TestPersistDocsSnapshotTeradata: + """persist_docs works for the snapshot materialization.""" + + @pytest.fixture(scope="class") + def snapshots(self): + return {"cmt_snapshot.sql": _SNAPSHOT_SQL} + + @pytest.fixture(scope="class") + def models(self): + return {"schema.yml": _SNAPSHOT_SCHEMA_YML} + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "snapshots": {"test": {"+persist_docs": {"relation": True, "columns": True}}} + } + + def test_snapshot_comment_persists(self, project): + run_dbt(["snapshot"]) + run_dbt(["docs", "generate"]) + with open(os.path.join(project.project_root, "target", "catalog.json")) as fp: + catalog = json.load(fp) + node = catalog["nodes"]["snapshot.test.cmt_snapshot"] + assert node["metadata"]["comment"].startswith("Snapshot relation description") diff --git a/tests/functional/adapter/test_behavior_flags.py b/tests/functional/adapter/test_behavior_flags.py new file mode 100644 index 00000000..9965aa32 --- /dev/null +++ b/tests/functional/adapter/test_behavior_flags.py @@ -0,0 +1,269 @@ +"""Functional tests for IDE-26230: dbt 1.11 legacy-behavior change flags. + +dbt Core 1.11 introduces two opt-in behavior change flags in dbt_project.yml: + + require_unique_project_resource_names (default: false) + false → duplicate names across resource types (e.g. model + seed) emit a + DuplicateNameDistinctNodeTypesDeprecation warning but do not fail. + true → raises DuplicateResourceNameError immediately. + + require_ref_searches_node_package_before_root (default: false) + false → when resolving ref() in a package model, dbt searches the root + project first, then the defining package. + true → dbt searches the defining package first, then the root project. + +These are pure dbt-core behaviors; the adapter is a pass-through. The tests +confirm the Teradata adapter does not obstruct or alter these flag semantics. + +All tests are functional and require a live Teradata Vantage instance. +All tests in this module are skipped on dbt-core < 1.11 (module-level pytestmark). +""" + +import pytest +import dbt.version as _dbt_version +from packaging.version import Version + +from dbt.exceptions import DuplicateResourceNameError +from dbt.tests.util import run_dbt, run_dbt_and_capture + +_dbt_version_tuple = Version(_dbt_version.__version__).release[:2] + +pytestmark = pytest.mark.skipif( + _dbt_version_tuple < (1, 11), + reason=f"Requires dbt-core >= 1.11 (found {_dbt_version.__version__})", +) + + +# ── fixtures ────────────────────────────────────────────────────────── + + +_SEED_CSV = """id,name +1,alice +2,bob +""".strip() + +_MODEL_SIMPLE_SQL = "select 1 as id, 'foo' as name" + +# Model that shares its resource NAME with the seed above but has a different +# alias, so only the name-uniqueness check fires (not the alias-collision check). +# The alias config prevents AmbiguousAliasError so we can observe the +# require_unique_project_resource_names flag behaviour in isolation. +_MODEL_DUPLICATE_SQL = "{{ config(alias='shared_resource_view') }}\nselect 99 as id, 'dup' as name" + +# A harmless unique model used as a sanity-check baseline. +# Note: 'result' is a reserved word in Teradata, so use 'val' instead. +_MODEL_UNIQUE_SQL = "select 42 as val" + + +# ── require_unique_project_resource_names: flag = false (default) ───── + + +class TestRequireUniqueProjectResourceNamesDefault: + """ + With require_unique_project_resource_names=false a seed and a model sharing + the same name must not prevent compilation. dbt emits a deprecation warning + but continues successfully. + + Requires dbt >= 1.11: earlier versions always error on duplicate resource names + regardless of any flag. + """ + + @pytest.fixture(scope="class") + def seeds(self): + return {"shared_resource.csv": _SEED_CSV} + + @pytest.fixture(scope="class") + def models(self): + return { + "shared_resource.sql": _MODEL_DUPLICATE_SQL, + "unique_model.sql": _MODEL_UNIQUE_SQL, + } + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "flags": { + "send_anonymous_usage_stats": False, + "require_unique_project_resource_names": False, + } + } + + def test_compile_succeeds_with_duplicate_names(self, project): + results = run_dbt(["compile"]) + assert results is not None + + def test_deprecation_warning_emitted(self, project): + _, logs = run_dbt_and_capture(["--debug", "compile"]) + assert "DuplicateNameDistinctNodeTypesDeprecation" in logs + + +# ── require_unique_project_resource_names: flag = true ──────────────── + + +class TestRequireUniqueProjectResourceNamesEnabled: + """ + With require_unique_project_resource_names=true a model and seed sharing + the same name must raise an error and stop compilation. + """ + + @pytest.fixture(scope="class") + def seeds(self): + return {"shared_resource.csv": _SEED_CSV} + + @pytest.fixture(scope="class") + def models(self): + return {"shared_resource.sql": _MODEL_DUPLICATE_SQL} + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "flags": { + "send_anonymous_usage_stats": False, + "require_unique_project_resource_names": True, + } + } + + def test_compile_fails_with_duplicate_names(self, project): + with pytest.raises(DuplicateResourceNameError): + run_dbt(["compile"]) + + +# ── require_unique_project_resource_names: no duplicates → always ok ── + + +class TestRequireUniqueProjectResourceNamesNoDuplicates: + """ + When there are no duplicate resource names the flag value (true or false) + must not affect the outcome — compilation always succeeds. + """ + + @pytest.fixture(scope="class") + def models(self): + return { + "model_a.sql": _MODEL_SIMPLE_SQL, + "model_b.sql": _MODEL_UNIQUE_SQL, + } + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "flags": { + "send_anonymous_usage_stats": False, + "require_unique_project_resource_names": True, + } + } + + def test_compile_succeeds_with_no_duplicates(self, project): + results = run_dbt(["compile"]) + assert results is not None + + def test_run_succeeds_with_no_duplicates(self, project): + results = run_dbt(["run"]) + assert len(results) == 2 + + +# ── require_ref_searches_node_package_before_root: flag accepted ─────── + + +_MODEL_REF_SQL = "select * from {{ ref('model_a') }}" + + +class TestRefSearchNodePackageBeforeRootFlagFalse: + """ + With require_ref_searches_node_package_before_root=false (default) dbt + resolves ref() by searching the root project first. This test confirms + the flag is accepted and compilation succeeds. + """ + + @pytest.fixture(scope="class") + def models(self): + return { + "model_a.sql": _MODEL_SIMPLE_SQL, + "model_b.sql": _MODEL_REF_SQL, + } + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "flags": { + "send_anonymous_usage_stats": False, + "require_ref_searches_node_package_before_root": False, + } + } + + def test_compile_succeeds_flag_false(self, project): + results = run_dbt(["compile"]) + assert results is not None + + def test_run_succeeds_flag_false(self, project): + results = run_dbt(["run"]) + assert len(results) == 2 + + +class TestRefSearchNodePackageBeforeRootFlagTrue: + """ + With require_ref_searches_node_package_before_root=true dbt searches the + defining package before the root project when resolving ref(). Within a + single-package project (no packages.yml) the behaviour is identical to + flag=false — all refs resolve normally and compilation succeeds. + """ + + @pytest.fixture(scope="class") + def models(self): + return { + "model_a.sql": _MODEL_SIMPLE_SQL, + "model_b.sql": _MODEL_REF_SQL, + } + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "flags": { + "send_anonymous_usage_stats": False, + "require_ref_searches_node_package_before_root": True, + } + } + + def test_compile_succeeds_flag_true(self, project): + results = run_dbt(["compile"]) + assert results is not None + + def test_run_succeeds_flag_true(self, project): + results = run_dbt(["run"]) + assert len(results) == 2 + + +# ── both flags enabled together ──────────────────────────────────────── + + +class TestBothFlagsEnabledTogether: + """ + Both behavior flags can be set simultaneously without conflict. + A clean project (no duplicate names, no package ambiguity) must + compile and run successfully with both flags set to true. + """ + + @pytest.fixture(scope="class") + def models(self): + return { + "model_a.sql": _MODEL_SIMPLE_SQL, + "model_b.sql": _MODEL_REF_SQL, + } + + @pytest.fixture(scope="class") + def project_config_update(self): + return { + "flags": { + "send_anonymous_usage_stats": False, + "require_unique_project_resource_names": True, + "require_ref_searches_node_package_before_root": True, + } + } + + def test_compile_succeeds_with_both_flags(self, project): + results = run_dbt(["compile"]) + assert results is not None + + def test_run_succeeds_with_both_flags(self, project): + results = run_dbt(["run"]) + assert len(results) == 2 diff --git a/tests/functional/adapter/test_udf.py b/tests/functional/adapter/test_udf.py new file mode 100644 index 00000000..09f7870c --- /dev/null +++ b/tests/functional/adapter/test_udf.py @@ -0,0 +1,621 @@ +""" +Functional tests for the dbt 1.11 `function` resource type on Teradata. + +Tests require a live Teradata / Vantage instance (env vars in conftest.py). +Run with: + pytest tests/functional/adapter/test_udf.py -v + +Implementation notes: + - dbt 1.11 ignores the `language:` key in functions.yml; FunctionNode.language + always defaults to 'sql'. A language-guard test is not feasible. + - `dbt run --select resource_type:function` does NOT pick up function nodes + without downstream model dependents. Use `dbt build --select +` + so the function reaches the DAG as an upstream dependency. + - target/compiled/ contains only the raw body (model.compiled_code). + The full REPLACE FUNCTION DDL is in target/run/ after dbt build. + - The dbt user must have CREATE FUNCTION privilege on the target schema; these + tests grant it explicitly (dbt-teradata does not grant it automatically). +""" +import pathlib +import pytest +from dbt.tests.util import run_dbt, run_dbt_and_capture + + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + +_ADD_TWO_INTS_SQL = "RETURN a + b;" + +_ADD_TWO_INTS_YML = """ +functions: + - name: add_two_ints + config: + type: scalar + volatility: deterministic + language: sql + arguments: + - name: a + data_type: INTEGER + - name: b + data_type: INTEGER + returns: + data_type: INTEGER +""".lstrip() + +_USE_UDF_SQL = "select {{ function('add_two_ints') }}(10, 32) as total" + +_USE_UDF_YML = """ +models: + - name: use_udf + columns: + - name: total + data_tests: + - accepted_values: + values: [42] +""".lstrip() + +_CONCAT_STRINGS_SQL = "RETURN TRIM(prefix) || '_' || TRIM(suffix);" + +_CONCAT_STRINGS_YML = """ +functions: + - name: concat_strings + config: + type: scalar + language: sql + arguments: + - name: prefix + data_type: VARCHAR(50) + - name: suffix + data_type: VARCHAR(50) + returns: + data_type: VARCHAR(101) +""".lstrip() + + +def _read_run_sql(project, project_name, function_name): + """Read the fully-assembled REPLACE FUNCTION DDL from target/run/.""" + return ( + pathlib.Path(project.project_root) + / "target" / "run" / project_name + / "functions" / f"{function_name}.sql" + ).read_text() + + +# --------------------------------------------------------------------------- +# Test: basic scalar UDF create + downstream model invocation +# --------------------------------------------------------------------------- + +class TestScalarUdfBasic: + """REPLACE FUNCTION is issued, function is callable from a model, test passes.""" + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_scalar_udf_basic"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "add_two_ints.sql": _ADD_TWO_INTS_SQL, + "functions.yml": _ADD_TWO_INTS_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return { + "use_udf.sql": _USE_UDF_SQL, + "schema.yml": _USE_UDF_YML, + } + + def test_build_and_test(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_udf"]) + + +# --------------------------------------------------------------------------- +# Test: idempotency — second run must not error (REPLACE FUNCTION is safe) +# --------------------------------------------------------------------------- + +class TestScalarUdfIdempotent: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_scalar_udf_idempotent"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "add_two_ints.sql": _ADD_TWO_INTS_SQL, + "functions.yml": _ADD_TWO_INTS_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_udf.sql": _USE_UDF_SQL} + + def test_second_run_is_clean(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_udf"]) + run_dbt(["build", "--select", "+use_udf"]) + + +# --------------------------------------------------------------------------- +# Test: dbt list shows functions +# --------------------------------------------------------------------------- + +class TestScalarUdfList: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_scalar_udf_list"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "add_two_ints.sql": _ADD_TWO_INTS_SQL, + "functions.yml": _ADD_TWO_INTS_YML, + } + + def test_list_resource_type(self, project): + _, stdout = run_dbt_and_capture(["list", "--resource-type", "function"]) + assert "add_two_ints" in stdout + + +# --------------------------------------------------------------------------- +# Test: multiple argument types compile and execute correctly +# --------------------------------------------------------------------------- + +_CONCAT_USE_SQL = "select {{ function('concat_strings') }}('hello', 'world') as concat_result" + +_CONCAT_USE_YML = """ +models: + - name: use_concat + columns: + - name: concat_result + data_tests: + - accepted_values: + values: ['hello_world'] +""".lstrip() + + +class TestScalarUdfMultipleArgs: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_scalar_udf_multi_args"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "concat_strings.sql": _CONCAT_STRINGS_SQL, + "functions.yml": _CONCAT_STRINGS_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return { + "use_concat.sql": _CONCAT_USE_SQL, + "schema.yml": _CONCAT_USE_YML, + } + + def test_varchar_args(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_concat"]) + + +# --------------------------------------------------------------------------- +# Test: CONTAINS SQL is always emitted (the only valid data access for SQL UDFs) +# --------------------------------------------------------------------------- +# Teradata SQL UDFs with LANGUAGE SQL / INLINE TYPE 1 only support CONTAINS SQL. +# NO SQL, READS SQL DATA, MODIFIES SQL DATA are for external (C/Java) UDFs. + +_CONTAINS_SQL_YML = """ +functions: + - name: contains_sql_udf + config: + type: scalar + language: sql + arguments: + - name: x + data_type: INTEGER + returns: + data_type: INTEGER +""".lstrip() + +_CONTAINS_SQL_BODY = "RETURN x + 1;" +_CONTAINS_SQL_USE = "select {{ function('contains_sql_udf') }}(1) as udf_val" + + +class TestSqlDataAccessConfig: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_data_access"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "contains_sql_udf.sql": _CONTAINS_SQL_BODY, + "functions.yml": _CONTAINS_SQL_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_contains_sql.sql": _CONTAINS_SQL_USE} + + def test_contains_sql_in_run_sql(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_contains_sql"]) + ddl = _read_run_sql(project, "test_udf_data_access", "contains_sql_udf") + assert "CONTAINS SQL" in ddl + + +# --------------------------------------------------------------------------- +# Test: volatility: deterministic renders DETERMINISTIC in executed DDL +# --------------------------------------------------------------------------- + +_VOLATILE_YML = """ +functions: + - name: det_udf + config: + type: scalar + volatility: deterministic + language: sql + arguments: + - name: x + data_type: INTEGER + returns: + data_type: INTEGER +""".lstrip() + +_VOLATILE_SQL = "RETURN x * 2;" +_VOLATILE_USE = "select {{ function('det_udf') }}(5) as val" + + +class TestVolatilityConfig: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_volatility"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "det_udf.sql": _VOLATILE_SQL, + "functions.yml": _VOLATILE_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_det_udf.sql": _VOLATILE_USE} + + def test_deterministic_in_run_sql(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_det_udf"]) + ddl = _read_run_sql(project, "test_udf_volatility", "det_udf") + assert "DETERMINISTIC" in ddl + + +# --------------------------------------------------------------------------- +# Test: default sql_data_access is CONTAINS SQL (not omitted, not NO SQL) +# --------------------------------------------------------------------------- + +_DEFAULT_ACCESS_YML = """ +functions: + - name: default_access_udf + config: + type: scalar + language: sql + arguments: + - name: x + data_type: INTEGER + returns: + data_type: INTEGER +""".lstrip() + +_DEFAULT_ACCESS_SQL = "RETURN x + 0;" +_DEFAULT_ACCESS_USE = "select {{ function('default_access_udf') }}(1) as val" + + +class TestDefaultSqlDataAccess: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_default_access"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "default_access_udf.sql": _DEFAULT_ACCESS_SQL, + "functions.yml": _DEFAULT_ACCESS_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_default_access.sql": _DEFAULT_ACCESS_USE} + + def test_default_access_is_contains_sql(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_default_access"]) + ddl = _read_run_sql(project, "test_udf_default_access", "default_access_udf") + assert "CONTAINS SQL" in ddl + + +# --------------------------------------------------------------------------- +# Test: `grants` on a UDF is applied as GRANT EXECUTE FUNCTION +# --------------------------------------------------------------------------- +# The grant DCL is executed at runtime (not written to target/run/), so we +# verify it by confirming the EXECUTE FUNCTION right ('EF') actually lands in +# DBC.AllRightsV. The old behaviour emitted plain `GRANT EXECUTE`, which is the +# macro/stored-procedure privilege and would not create an 'EF' right on a UDF. + +_GRANTED_UDF_YML = """ +functions: + - name: granted_udf + config: + type: scalar + grants: + execute: ['PUBLIC'] + language: sql + arguments: + - name: x + data_type: INTEGER + returns: + data_type: INTEGER +""".lstrip() + +_GRANTED_UDF_SQL = "RETURN x + 1;" +_GRANTED_UDF_USE = "select {{ function('granted_udf') }}(1) as val" + + +class TestScalarUdfGrants: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_grants"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "granted_udf.sql": _GRANTED_UDF_SQL, + "functions.yml": _GRANTED_UDF_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_granted_udf.sql": _GRANTED_UDF_USE} + + def test_execute_function_grant_applied(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_granted_udf"]) + rights = project.run_sql( + "SELECT AccessRight FROM DBC.AllRightsV " + f"WHERE DatabaseName='{project.test_schema}' " + "AND TRIM(LOWER(TableName))='granted_udf' " + "AND TRIM(LOWER(Username))='public' " + "AND AccessRight='EF'", + fetch="all", + ) + assert len(rights) >= 1 + + +# --------------------------------------------------------------------------- +# Test: aggregate UDFs are rejected with a clear compile-time error +# --------------------------------------------------------------------------- +# `--select +use_aggregate_udf` pulls in two nodes: the function (which errors) +# and the downstream model (which gets marked 'skipped' as a result), so +# `results` has 2 entries here — assert on content across all of them rather +# than an exact list length. + +_AGGREGATE_YML = """ +functions: + - name: aggregate_udf + config: + type: aggregate + language: sql + arguments: + - name: x + data_type: INTEGER + returns: + data_type: INTEGER +""".lstrip() + +_AGGREGATE_SQL = "RETURN x;" +_AGGREGATE_USE = "select {{ function('aggregate_udf') }}(1) as val" + + +class TestAggregateUdfNotSupported: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_aggregate_not_supported"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "aggregate_udf.sql": _AGGREGATE_SQL, + "functions.yml": _AGGREGATE_YML, + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_aggregate_udf.sql": _AGGREGATE_USE} + + def test_aggregate_udf_raises_clear_error(self, project): + # expect_pass=False (rather than pytest.raises(CompilationError)) is safe here: + # BaseRunner.safe_run in dbt-core catches any Exception — including the + # CompilationError raised by raise_compiler_error — around compile_and_execute + # and converts it into a per-node error result instead of letting it propagate. + results = run_dbt(["build", "--select", "+use_aggregate_udf"], expect_pass=False) + assert any("Aggregate user-defined functions" in str(r.message or "") for r in results) + + +# --------------------------------------------------------------------------- +# Test: persist_docs relation-level description writes COMMENT ON FUNCTION +# --------------------------------------------------------------------------- +# Teradata stores UDF comments in DBC.TablesV.CommentString, the same column +# used for tables/views/macros (TableKind='F' for a scalar UDF), so the comment +# is verified directly against DBC.TablesV. Function nodes are not "relational" +# in dbt-core 1.11 (is_relational excludes NodeType.Function), so they are never +# fetched by `dbt docs generate` and therefore do not surface in catalog.json. + +_DOC_UDF_SQL = "RETURN x + 1;" +_DOC_UDF_USE = "select {{ function('doc_udf') }}(1) as val" + +_DOC_UDF_DESCRIPTION = "Adds one to the input integer." +_DOC_UDF_DESCRIPTION_V2 = "Adds one to the given integer value." + + +def _doc_udf_yml(description): + return ( + "functions:\n" + " - name: doc_udf\n" + " config:\n" + " type: scalar\n" + " persist_docs:\n" + " relation: true\n" + f" description: \"{description}\"\n" + " language: sql\n" + " arguments:\n" + " - name: x\n" + " data_type: INTEGER\n" + " returns:\n" + " data_type: INTEGER\n" + ) + + +def _function_comment(project, function_name): + """Read the current COMMENT ON FUNCTION text straight from DBC.TablesV.""" + row = project.run_sql( + "SELECT CommentString FROM DBC.TablesV " + f"WHERE DatabaseName='{project.test_schema}' " + "AND TableKind='F' " + f"AND TRIM(LOWER(TableName))='{function_name}'", + fetch="one", + ) + return row[0].strip() if row and row[0] is not None else None + + +class TestScalarUdfPersistDocsRelation: + """persist_docs: {relation: true} + a `description:` writes COMMENT ON FUNCTION.""" + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_persist_docs"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "doc_udf.sql": _DOC_UDF_SQL, + "functions.yml": _doc_udf_yml(_DOC_UDF_DESCRIPTION), + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_doc_udf.sql": _DOC_UDF_USE} + + def test_comment_on_function_written(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + _, logs = run_dbt_and_capture(["--debug", "build", "--select", "+use_doc_udf"]) + assert "comment on function" in logs.lower() + assert _function_comment(project, "doc_udf") == _DOC_UDF_DESCRIPTION + + +# --------------------------------------------------------------------------- +# Test: unchanged description issues no COMMENT ON FUNCTION DDL on rerun +# --------------------------------------------------------------------------- +# Unlike REPLACE VIEW / CREATE OR REPLACE TABLE (which drop-and-recreate the +# object, wiping DBC.TablesV.CommentString), Teradata's REPLACE FUNCTION +# preserves the existing comment across a rebuild with an unchanged signature +# and body -- verified empirically: COMMENT ON FUNCTION, then REPLACE FUNCTION +# again with identical DDL, leaves CommentString unchanged. So change detection +# in teradata__persist_docs (existing_rel_comment != model.description) is +# meaningful here, unlike for table/view (see TestPersistDocsIdempotentTeradata +# in teradata_dbt/test_validate_teradata_persist_docs.py, which needs an +# *incremental* model for the same reason table/view do not preserve comments). + +class TestScalarUdfPersistDocsIdempotent: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_persist_docs_idempotent"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "doc_udf.sql": _DOC_UDF_SQL, + "functions.yml": _doc_udf_yml(_DOC_UDF_DESCRIPTION), + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_doc_udf.sql": _DOC_UDF_USE} + + def test_no_ddl_on_unchanged_rerun(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_doc_udf"]) + assert _function_comment(project, "doc_udf") == _DOC_UDF_DESCRIPTION + + # Second run: same functions.yml, same description -> no COMMENT DDL. + # COMMENT DDL is only emitted at DEBUG level, so capture with --debug. + _, logs = run_dbt_and_capture(["--debug", "build", "--select", "+use_doc_udf"]) + assert "comment on function" not in logs.lower() + assert _function_comment(project, "doc_udf") == _DOC_UDF_DESCRIPTION + + +# --------------------------------------------------------------------------- +# Test: changing the description re-issues COMMENT ON FUNCTION with new text +# --------------------------------------------------------------------------- + +class TestScalarUdfPersistDocsChanged: + + @pytest.fixture(scope="class") + def project_config_update(self): + return {"name": "test_udf_persist_docs_changed"} + + @pytest.fixture(scope="class") + def functions(self): + return { + "doc_udf.sql": _DOC_UDF_SQL, + "functions.yml": _doc_udf_yml(_DOC_UDF_DESCRIPTION), + } + + @pytest.fixture(scope="class") + def models(self): + return {"use_doc_udf.sql": _DOC_UDF_USE} + + def test_changed_description_reissues_ddl(self, project): + username = project.adapter.config.credentials.username + project.run_sql(f"GRANT CREATE FUNCTION ON {project.test_schema} TO {username}") + run_dbt(["build", "--select", "+use_doc_udf"]) + assert _function_comment(project, "doc_udf") == _DOC_UDF_DESCRIPTION + + functions_yml_path = ( + pathlib.Path(project.project_root) / "functions" / "functions.yml" + ) + functions_yml_path.write_text(_doc_udf_yml(_DOC_UDF_DESCRIPTION_V2)) + + _, logs = run_dbt_and_capture(["--debug", "build", "--select", "+use_doc_udf"]) + assert "comment on function" in logs.lower() + assert _function_comment(project, "doc_udf") == _DOC_UDF_DESCRIPTION_V2 + + +# --------------------------------------------------------------------------- +# Note: persist_docs `columns` config on a function (warn-and-skip) is NOT +# covered functionally here. Verified empirically that dbt-core 1.11's +# UnparsedFunctionUpdate schema (HasColumnProps, not HasColumnDocs) does not +# parse a `columns:` block into model.columns for function nodes -- a +# functions.yml `columns:` list is silently dropped, so the warn-and-skip +# branch in teradata__persist_docs can never be reached via YAML in practice. +# That branch is instead covered directly in tests/unit/test_persist_docs.py +# (TestPersistDocsFunctionColumnsSkipped), which invokes the macro with a +# synthetic model.columns dict. diff --git a/tests/unit/test_engine_env_vars.py b/tests/unit/test_engine_env_vars.py new file mode 100644 index 00000000..a8f4a96a --- /dev/null +++ b/tests/unit/test_engine_env_vars.py @@ -0,0 +1,174 @@ +"""Unit tests for IDE-26224: DBT_ENGINE_* env vars compatibility. + +dbt Core 1.11 renamed two specific environment variables used for state +artifact lookup: + - DBT_STATE → DBT_ENGINE_STATE + - DBT_DEFER_STATE → DBT_ENGINE_DEFER_STATE + +These tests verify that the Teradata adapter credentials and configuration +are completely unaffected by both the old and new variable names being +present in the environment. + +Coverage also includes the broader DBT_ENGINE_* namespace (DBT_ENGINE_FULL_REFRESH, +DBT_ENGINE_TARGET, DBT_ENGINE_PROFILES_DIR) because dbt-core may introduce +additional variables in this prefix over time. Testing each now ensures +the adapter remains isolated from the entire namespace, not just the two +renamed vars specific to IDE-26224. + +The adapter itself never reads these vars; they are handled exclusively by +dbt-core. The tests here guard against accidental coupling. +""" + +import os +from unittest.mock import patch + +from dbt.adapters.teradata.connections import TeradataCredentials + + +def _make_credentials(**overrides): + defaults = { + "server": "localhost", + "schema": "test_schema", + "username": "test_user", + "password": "test_pass", + } + defaults.update(overrides) + return TeradataCredentials(**defaults) + + +class TestEngineEnvVarsIsolation: + """DBT_ENGINE_* env vars (dbt 1.11) must not affect adapter credential setup.""" + + def test_credentials_unaffected_by_dbt_engine_state(self): + with patch.dict(os.environ, clear=True, values={"DBT_ENGINE_STATE": "/some/path/to/state"}): + creds = _make_credentials() + assert creds.server == "localhost" + assert creds.username == "test_user" + + def test_credentials_unaffected_by_dbt_engine_defer_state(self): + with patch.dict(os.environ, clear=True, values={"DBT_ENGINE_DEFER_STATE": "/some/path/to/defer"}): + creds = _make_credentials() + assert creds.server == "localhost" + assert creds.schema == "test_schema" + + def test_credentials_unaffected_by_dbt_engine_full_refresh(self): + with patch.dict(os.environ, clear=True, values={"DBT_ENGINE_FULL_REFRESH": "true"}): + creds = _make_credentials() + assert creds.server == "localhost" + + def test_credentials_unaffected_by_dbt_engine_target(self): + with patch.dict(os.environ, clear=True, values={"DBT_ENGINE_TARGET": "prod"}): + creds = _make_credentials() + assert creds.server == "localhost" + + def test_credentials_unaffected_by_dbt_engine_profiles_dir(self): + with patch.dict(os.environ, clear=True, values={"DBT_ENGINE_PROFILES_DIR": "/some/profiles"}): + creds = _make_credentials() + assert creds.server == "localhost" + + def test_credentials_unaffected_by_multiple_engine_vars(self): + engine_vars = { + "DBT_ENGINE_STATE": "/path/state", + "DBT_ENGINE_DEFER_STATE": "/path/defer", + "DBT_ENGINE_FULL_REFRESH": "true", + "DBT_ENGINE_TARGET": "prod", + "DBT_ENGINE_PROFILES_DIR": "/path/profiles", + } + with patch.dict(os.environ, engine_vars, clear=True): + creds = _make_credentials() + assert creds.server == "localhost" + assert creds.schema == "test_schema" + assert creds.username == "test_user" + + +class TestLegacyStateEnvVarsIsolation: + """Old DBT_STATE and DBT_DEFER_STATE env vars must not break adapter credential setup.""" + + def test_credentials_unaffected_by_dbt_state(self): + with patch.dict(os.environ, clear=True, values={"DBT_STATE": "/some/path/to/state"}): + creds = _make_credentials() + assert creds.server == "localhost" + + def test_credentials_unaffected_by_dbt_defer_state(self): + with patch.dict(os.environ, clear=True, values={"DBT_DEFER_STATE": "/some/path/to/defer"}): + creds = _make_credentials() + assert creds.server == "localhost" + + +class TestOldAndNewEnvVarsCoexist: + """DBT_STATE / DBT_DEFER_STATE and their DBT_ENGINE_* replacements can coexist without breaking the adapter.""" + + def test_both_state_vars_coexist(self): + env_vars = { + "DBT_STATE": "/old/path", + "DBT_ENGINE_STATE": "/new/path", + } + with patch.dict(os.environ, env_vars, clear=True): + creds = _make_credentials() + assert creds.server == "localhost" + assert creds.username == "test_user" + + def test_old_and_new_defer_coexist(self): + env_vars = { + "DBT_DEFER_STATE": "/old/defer", + "DBT_ENGINE_DEFER_STATE": "/new/defer", + } + with patch.dict(os.environ, env_vars, clear=True): + creds = _make_credentials() + assert creds.server == "localhost" + + +class TestTeradataEnvVarsDoNotConflictWithEngineVars: + """Teradata-specific env vars and DBT_ENGINE_* vars can coexist.""" + + def test_explicit_credentials_not_overridden_by_engine_vars(self): + """Explicitly passed credentials must not be silently overridden by + DBT_ENGINE_* or DBT_TERADATA_* env vars present in the environment.""" + env_vars = { + "DBT_ENGINE_STATE": "/some/path", + "DBT_ENGINE_TARGET": "prod", + "DBT_TERADATA_SERVER_NAME": "env-host.example.com", + "DBT_TERADATA_USERNAME": "env_user", + "DBT_TERADATA_PASSWORD": "env_pass", + } + with patch.dict(os.environ, env_vars, clear=True): + creds = _make_credentials( + server="explicit-host.example.com", + username="explicit_user", + password="explicit_pass", + ) + assert creds.server == "explicit-host.example.com" + assert creds.username == "explicit_user" + assert creds.password == "explicit_pass" + assert creds.server != os.environ["DBT_TERADATA_SERVER_NAME"] + assert creds.username != os.environ["DBT_TERADATA_USERNAME"] + + def test_teradata_env_vars_not_auto_read_by_credentials(self): + """DBT_TERADATA_* env vars must not be auto-consumed by TeradataCredentials. + + The adapter reads credentials from the dbt profile, not from env vars + directly. Setting DBT_TERADATA_* vars in the environment must not silently + change the credential values when defaults are used. + """ + env_vars = { + "DBT_TERADATA_SERVER_NAME": "env-host.example.com", + "DBT_TERADATA_USERNAME": "env_user", + "DBT_TERADATA_PASSWORD": "env_pass", + } + with patch.dict(os.environ, env_vars, clear=True): + creds = _make_credentials() + # Defaults must be used, not the env var values + assert creds.server == "localhost" + assert creds.username == "test_user" + assert creds.password == "test_pass" + + def test_credentials_with_all_env_var_types_present(self): + env_vars = { + "DBT_STATE": "/old/state", + "DBT_ENGINE_STATE": "/new/state", + "DBT_TERADATA_SERVER_NAME": "env-host.example.com", + } + with patch.dict(os.environ, env_vars, clear=True): + creds = _make_credentials(server="explicit-host.example.com") + assert creds.server == "explicit-host.example.com" + assert creds.server != os.environ["DBT_TERADATA_SERVER_NAME"] diff --git a/tests/unit/test_persist_docs.py b/tests/unit/test_persist_docs.py new file mode 100644 index 00000000..5f3264c3 --- /dev/null +++ b/tests/unit/test_persist_docs.py @@ -0,0 +1,360 @@ +"""Unit tests for persist_docs macros (IDE-26225). + +Pure unit tests (no database required). They render the Jinja macros in +dbt/include/teradata/macros/persist_docs.sql with lightweight stubs for the +dbt-provided globals (`exceptions`, `var`, `config`) and assert behavior: + + 1. teradata_escape_comment: single-quote escaping, 255-char truncation, + configurable limit via var, and non-string rejection. + 2. teradata_truncate_comment: the plain (unescaped) truncation shared between + DDL emission (teradata_escape_comment) and change detection + (teradata__persist_docs / teradata__alter_column_comment), so over-length + descriptions truncate identically on both sides and are idempotent. + 3. teradata__alter_relation_comment: emits COMMENT ON TABLE vs VIEW vs FUNCTION. + 4. teradata__validate_doc_columns: quote-aware, case-sensitive filtering + plus the "columns not present" warning. + 5. teradata__persist_docs: column-level persist_docs is skipped with a + warning for function (UDF) relations, since Teradata has no per-argument + comment DDL. +""" + +import os + +import pytest +from jinja2 import Environment + + +_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +_PERSIST_DOCS = os.path.join( + _REPO_ROOT, "dbt", "include", "teradata", "macros", "persist_docs.sql" +) + + +class _MacroReturn(Exception): + """Emulates dbt's `return()` which raises to hand a value back from a macro.""" + + def __init__(self, value): + self.value = value + + +def _call_returning(macro, *args): + """Invoke a macro that ends in `{{ return(x) }}` and recover x.""" + try: + macro(*args) + except _MacroReturn as r: + return r.value + raise AssertionError("macro did not call return()") + + +class _Exceptions: + """Stub for dbt's `exceptions` global. Records warnings and raises on error.""" + + def __init__(self): + self.warnings = [] + + def warn(self, msg): + self.warnings.append(str(msg)) + return "" + + def raise_compiler_error(self, msg): + raise ValueError(str(msg)) + + +class _Relation: + """Minimal stand-in for a dbt relation: renders to a quoted name and has a type.""" + + def __init__(self, name='"db"."obj"', rtype="table"): + self._name = name + self.type = rtype + + def render(self): + return self._name + + def __str__(self): + return self._name + + +class _Config: + """Stub for dbt's per-model `config` global used by teradata__persist_docs.""" + + def __init__(self, catalog_name=None, persist_relation=False, persist_columns=False): + self._catalog_name = catalog_name + self._persist_relation = persist_relation + self._persist_columns = persist_columns + + def get(self, key, default=None): + if key == "catalog_name": + return self._catalog_name + return default + + def persist_relation_docs(self): + return self._persist_relation + + def persist_column_docs(self): + return self._persist_columns + + +class _Model: + """Minimal stand-in for the `model` global: description + columns dict.""" + + def __init__(self, description="", columns=None): + self.description = description + self.columns = columns or {} + + +def _load_macros(max_comment_length=255, config=None): + """Load persist_docs.sql and return its rendered Jinja module plus the exceptions stub.""" + exc = _Exceptions() + env = Environment(extensions=["jinja2.ext.do"]) + env.globals["exceptions"] = exc + env.globals["var"] = lambda key, default=None: ( + max_comment_length if key == "teradata_max_comment_length" else default + ) + env.globals["config"] = config if config is not None else _Config() + + def _return(value): + raise _MacroReturn(value) + + env.globals["return"] = _return + with open(_PERSIST_DOCS, encoding="utf-8") as fh: + module = env.from_string(fh.read()).module + return module, exc + + +# --------------------------------------------------------------------------- +# teradata_escape_comment +# --------------------------------------------------------------------------- +class TestEscapeComment: + def test_wraps_in_single_quotes(self): + mod, _ = _load_macros() + assert str(mod.teradata_escape_comment("hello")) == "'hello'" + + def test_doubles_single_quotes(self): + mod, _ = _load_macros() + # o'brien -> 'o''brien' + assert str(mod.teradata_escape_comment("o'brien")) == "'o''brien'" + + def test_triple_quoted_token_roundtrips(self): + mod, _ = _load_macros() + text = "'''abc'''" + assert str(mod.teradata_escape_comment(text)) == "'" + text.replace("'", "''") + "'" + + def test_preserves_newlines_and_sql_comment_tokens(self): + mod, _ = _load_macros() + text = "line1\n-- dashcomment\n/* block */" + out = str(mod.teradata_escape_comment(text)) + assert out == "'" + text + "'" + assert "\n" in out and "--" in out and "/* block */" in out + + def test_truncates_at_255_and_warns(self): + mod, exc = _load_macros() + out = str(mod.teradata_escape_comment("x" * 400)) + # 255 x's wrapped in quotes + assert out == "'" + "x" * 255 + "'" + assert len(out) == 257 + assert any("truncating" in w for w in exc.warnings) + + def test_no_truncation_at_exactly_255(self): + mod, exc = _load_macros() + out = str(mod.teradata_escape_comment("y" * 255)) + assert out == "'" + "y" * 255 + "'" + assert exc.warnings == [] + + def test_respects_configurable_limit(self): + mod, exc = _load_macros(max_comment_length=10) + out = str(mod.teradata_escape_comment("z" * 50)) + assert out == "'" + "z" * 10 + "'" + assert any("10 characters" in w for w in exc.warnings) + + def test_non_string_raises(self): + mod, _ = _load_macros() + with pytest.raises(ValueError): + mod.teradata_escape_comment(123) + + def test_null_limit_falls_back_to_255(self): + # var set to null/None must not coerce to 0 (which would empty the comment) + mod, exc = _load_macros(max_comment_length=None) + out = str(mod.teradata_escape_comment("q" * 300)) + assert out == "'" + "q" * 255 + "'" + + def test_non_positive_limit_falls_back_to_255(self): + mod, exc = _load_macros(max_comment_length=0) + out = str(mod.teradata_escape_comment("q" * 300)) + assert out == "'" + "q" * 255 + "'" + + def test_non_numeric_limit_falls_back_to_255(self): + mod, exc = _load_macros(max_comment_length="not-a-number") + out = str(mod.teradata_escape_comment("q" * 300)) + assert out == "'" + "q" * 255 + "'" + + +# --------------------------------------------------------------------------- +# teradata_truncate_comment +# +# This is the change-detection half of the fix for the review comment on PR #241: +# `teradata__persist_docs` / `teradata__alter_column_comment` compare the *stored* +# comment (always <= the Teradata limit) against `teradata_truncate_comment(desc)` +# rather than the raw `desc`. These tests pin down that `teradata_truncate_comment` +# truncates identically to (and is reused by) `teradata_escape_comment`, so an +# over-length description is idempotent: it is truncated the same way on every run, +# so change detection can find a stable match instead of re-issuing DDL forever. +# --------------------------------------------------------------------------- +class TestTruncateComment: + def test_no_truncation_under_limit(self): + mod, _ = _load_macros() + assert str(mod.teradata_truncate_comment("hello")) == "hello" + + def test_truncates_at_255_by_default(self): + mod, _ = _load_macros() + out = str(mod.teradata_truncate_comment("x" * 400)) + assert out == "x" * 255 + + def test_matches_escape_comment_truncation(self): + # The exact guarantee the review comment asked for: the plain truncated form + # used for change detection must equal what teradata_escape_comment (DDL + # emission) actually truncates and stores, for the same input. + mod, _ = _load_macros() + long_desc = "y" * 400 + truncated = str(mod.teradata_truncate_comment(long_desc, warn=False)) + escaped = str(mod.teradata_escape_comment(long_desc)) + assert escaped == "'" + truncated + "'" + + def test_warn_true_emits_warning(self): + mod, exc = _load_macros() + mod.teradata_truncate_comment("x" * 300) + assert any("truncating" in w for w in exc.warnings) + + def test_warn_false_suppresses_warning(self): + mod, exc = _load_macros() + mod.teradata_truncate_comment("x" * 300, warn=False) + assert exc.warnings == [] + + def test_respects_configurable_limit(self): + mod, _ = _load_macros(max_comment_length=10) + out = str(mod.teradata_truncate_comment("z" * 50, warn=False)) + assert out == "z" * 10 + + def test_non_string_raises(self): + mod, _ = _load_macros() + with pytest.raises(ValueError): + mod.teradata_truncate_comment(123) + + +# --------------------------------------------------------------------------- +# teradata__alter_relation_comment +# --------------------------------------------------------------------------- +class TestAlterRelationComment: + def test_table_emits_comment_on_table(self): + mod, _ = _load_macros() + sql = str(mod.teradata__alter_relation_comment(_Relation(rtype="table"), "desc")).strip() + assert sql.lower().startswith("comment on table") + assert sql.endswith("as 'desc'") + + def test_view_emits_comment_on_view(self): + mod, _ = _load_macros() + sql = str(mod.teradata__alter_relation_comment(_Relation(rtype="view"), "desc")).strip() + assert sql.lower().startswith("comment on view") + + def test_snapshot_treated_as_table(self): + # snapshot is a real dbt relation type; it maps to COMMENT ON TABLE. + mod, _ = _load_macros() + sql = str(mod.teradata__alter_relation_comment(_Relation(rtype="snapshot"), "d")).strip() + assert sql.lower().startswith("comment on table") + + def test_function_emits_comment_on_function(self): + mod, _ = _load_macros() + sql = str( + mod.teradata__alter_relation_comment(_Relation(rtype="function"), "desc") + ).strip() + assert sql.lower().startswith("comment on function") + assert sql.endswith("as 'desc'") + + def test_escapes_embedded_quote(self): + mod, _ = _load_macros() + sql = str(mod.teradata__alter_relation_comment(_Relation(), "a'b")).strip() + assert sql.endswith("as 'a''b'") + + +# --------------------------------------------------------------------------- +# teradata__validate_doc_columns +# --------------------------------------------------------------------------- +def _col(quote=False): + return {"quote": quote, "description": "d"} + + +class TestValidateDocColumns: + def test_unquoted_matches_case_insensitively(self): + mod, exc = _load_macros() + filtered = _call_returning( + mod.teradata__validate_doc_columns, _Relation(), {"ID": _col()}, ["id"] + ) + assert list(filtered.keys()) == ["ID"] + assert exc.warnings == [] + + def test_quoted_matches_case_sensitively(self): + mod, exc = _load_macros() + # quoted "MyCol" is not present as physical "mycol" -> filtered out + warning + filtered = _call_returning( + mod.teradata__validate_doc_columns, + _Relation(), + {"MyCol": _col(quote=True)}, + ["mycol"], + ) + assert list(filtered.keys()) == [] + assert any("MyCol" in w for w in exc.warnings) + + def test_missing_column_warns_with_expected_message(self): + mod, exc = _load_macros() + _call_returning( + mod.teradata__validate_doc_columns, _Relation(), {"ghost": _col()}, ["id", "nm"] + ) + assert any( + "The following columns are specified in the schema but are not present " + "in the database: ghost" in w + for w in exc.warnings + ) + + def test_mixed_present_and_missing(self): + mod, exc = _load_macros() + filtered = _call_returning( + mod.teradata__validate_doc_columns, + _Relation(), + {"id": _col(), "ghost": _col()}, + ["id"], + ) + assert list(filtered.keys()) == ["id"] + assert any("ghost" in w for w in exc.warnings) + + +# --------------------------------------------------------------------------- +# teradata__persist_docs: column docs are skipped (with a warning) for +# function (UDF) relations, since Teradata has no per-argument comment DDL. +# --------------------------------------------------------------------------- +class TestPersistDocsFunctionColumnsSkipped: + def test_function_with_columns_warns_and_skips(self): + # for_relation=False isolates the column-docs branch: no run_query/statement + # stubs are needed since that code path is never reached for a function. + config = _Config(persist_columns=True) + mod, exc = _load_macros(config=config) + model = _Model(columns={"x": _col()}) + mod.teradata__persist_docs(_Relation(rtype="function"), model, False, True) + assert any( + "persist_docs 'columns' config is not supported for Teradata functions" in w + for w in exc.warnings + ) + + def test_function_without_columns_is_silent(self): + # model.columns is empty (the common case: functions.yml has no `columns:` + # block) -> the condition is falsy and no warning is emitted at all. + config = _Config(persist_columns=True) + mod, exc = _load_macros(config=config) + model = _Model(columns={}) + mod.teradata__persist_docs(_Relation(rtype="function"), model, False, True) + assert exc.warnings == [] + + def test_columns_disabled_is_silent_even_with_columns(self): + config = _Config(persist_columns=False) + mod, exc = _load_macros(config=config) + model = _Model(columns={"x": _col()}) + mod.teradata__persist_docs(_Relation(rtype="function"), model, False, True) + assert exc.warnings == []