feat(athena): Part 4 — vendor the dbt-athena 1.11.0 macro package

.changes/unreleased/Features-20260918-12254-athena-part-4.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: '[Athena] Vendor the dbt-athena 1.11.0 macro package into `dbt-loader`; `list_schemas`, `get_columns_in_relation`, `create_schema`, `check_schema_exists` and `get_catalog` are rewritten as plain SQL because their upstream delegation to `adapter.*` recurses under Fusion'
+time: 2026-09-18T12:25:02.970682-07:00
+custom:
+    author: aoelvp94
+    issue: "16252"
+    project: dbt-core

crates/dbt-loader/src/dbt_macro_assets/README.md

@@ -8,6 +8,17 @@ All adapter macros are currently maintained in:
 ### [2026-09-24]
   - dbt-databricks: gate view/metric_view ALTER on matching relation type from PR #1687
 
+### [2026-09-16]
+  - dbt-athena: v1.11.0, vendored from the `dbt-athena` wheel's
+    `dbt/include/athena/` (42 files, 2212 SQL LOC). Note the macros ship in the
+    `dbt-athena` distribution, NOT `dbt-athena-community`, which is an empty
+    shim that merely depends on it.
+    No loader change was needed: `internal_package_names` derives the directory
+    name as `dbt-{adapter_type}`, and Athena inherits no other adapter's macros.
+    `sample_profiles.yml` is not vendored: `is_metadata_file` accepts only
+    `dbt_project.yml`, `packages.yml`, `profile_template.yml` and `__init__.py`
+    in an internal package, and the sample profile is covered by `dbt init`.
+
 ### [2026-08-19]
   - dbt-databricks: view full-refresh precedence from commit 45351e11517d3f37c5ac7a736b5fcba453d3f368
 

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/__init__.py

@@ -0,0 +1,3 @@
+import os
+
+PACKAGE_PATH = os.path.dirname(__file__)

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/dbt_project.yml

@@ -0,0 +1,5 @@
+name: dbt_athena
+version: 1.0
+config-version: 2
+
+macro-paths: ['macros']

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/macros/adapters/columns.sql

@@ -0,0 +1,38 @@
+{#- Fusion runs this macro from AdapterImpl::get_columns_in_relation, so
+    delegating back to adapter.get_columns_in_relation would recurse. Trino's
+    information_schema.columns has no length/precision/scale columns, and covers only
+    its own catalog, so it is qualified with the relation's (S3 Tables:
+    s3tablescatalog/<bucket>). -#}
+{% macro athena__get_columns_in_relation(relation) -%}
+  {% call statement('get_columns_in_relation', fetch_result=True) %}
+    select
+        column_name,
+        data_type,
+        cast(null as integer) as character_maximum_length,
+        cast(null as integer) as numeric_precision,
+        cast(null as integer) as numeric_scale
+    from {% if relation.database %}"{{ relation.database }}".{% endif %}information_schema.columns
+    where lower(table_name) = '{{ relation.identifier | lower }}'
+      {% if relation.schema %} and lower(table_schema) = '{{ relation.schema | lower }}' {% endif %}
+    order by ordinal_position
+  {% endcall %}
+  {% set table = load_result('get_columns_in_relation').table %}
+  {{ return(sql_convert_columns_in_relation(table)) }}
+{% endmacro %}
+
+{% macro athena__get_empty_schema_sql(columns) %}
+    {%- set col_err = [] -%}
+    select
+    {% for i in columns %}
+      {%- set col = columns[i] -%}
+      {%- if col['data_type'] is not defined -%}
+        {{ col_err.append(col['name']) }}
+      {%- else -%}
+        {% set col_name = adapter.quote(col['name']) if col.get('quote') else col['name'] %}
+        cast(null as {{ dml_data_type(col['data_type']) }}) as {{ col_name }}{{ ", " if not loop.last }}
+      {%- endif -%}
+    {%- endfor -%}
+    {%- if (col_err | length) > 0 -%}
+      {{ exceptions.column_type_missing(column_names=col_err) }}
+    {%- endif -%}
+{% endmacro %}

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/macros/adapters/metadata.sql

@@ -0,0 +1,79 @@
+{#- dbt-athena builds the catalog Python-side from Glue (adapter.get_catalog).
+    Fusion runs this macro from `compile --write-catalog` / docs and expects the
+    Postgres-shaped result (one row per column, table columns repeated), so it is
+    a query over Trino's information_schema instead. Glue exposes no table owner
+    or comment there; both are returned empty. -#}
+{% macro athena__get_catalog(information_schema, schemas) -%}
+  {%- set relations = [] -%}
+  {%- for schema in schemas -%}
+    {%- do relations.append({'schema': schema}) -%}
+  {%- endfor -%}
+  {{ return(athena__get_catalog_relations(information_schema, relations)) }}
+{%- endmacro %}
+
+
+{#- Fusion runs this macro from AdapterImpl::list_schemas, so delegating to
+    adapter.list_schemas would recurse. Query information_schema instead. The
+    database arrives rendered (quoted); strip the quotes for the literal. -#}
+{% macro athena__list_schemas(database) -%}
+  {% call statement('list_schemas', fetch_result=True, auto_begin=False) %}
+    select schema_name from information_schema.schemata
+    {% if database %} where lower(catalog_name) = '{{ database | replace('"', '') | lower }}' {% endif %}
+  {% endcall %}
+  {{ return(load_result('list_schemas').table) }}
+{% endmacro %}
+
+
+{% macro athena__list_relations_without_caching(schema_relation) %}
+  {{ return(adapter.list_relations_without_caching(schema_relation)) }}
+{% endmacro %}
+
+{% macro athena__get_catalog_relations(information_schema, relations) -%}
+  {%- call statement('catalog', fetch_result=True) -%}
+    select
+        '{{ information_schema.database }}' as table_database,
+        t.table_schema as table_schema,
+        t.table_name as table_name,
+        t.table_type as table_type,
+        cast(null as varchar) as table_comment,
+        c.column_name as column_name,
+        c.ordinal_position as column_index,
+        c.data_type as column_type,
+        c.comment as column_comment,
+        cast(null as varchar) as table_owner
+    from information_schema.tables t
+    join information_schema.columns c
+      on c.table_catalog = t.table_catalog
+     and c.table_schema = t.table_schema
+     and c.table_name = t.table_name
+    where (
+      {%- for relation in relations -%}
+        {%- if relation.identifier -%}
+          (lower(t.table_schema) = lower('{{ relation.schema }}') and
+           lower(t.table_name) = lower('{{ relation.identifier }}'))
+        {%- else -%}
+          lower(t.table_schema) = lower('{{ relation.schema }}')
+        {%- endif -%}
+        {%- if not loop.last %} or {% endif -%}
+      {%- endfor -%}
+    )
+    order by t.table_schema, t.table_name, c.ordinal_position
+  {%- endcall -%}
+  {{ return(load_result('catalog').table) }}
+{%- endmacro %}
+
+{#- Fusion's `default__check_schema_exists` calls
+    `information_schema.replace(information_schema_view=...)`, which Fusion
+    relations do not implement; every adapter package overrides it. Athena's
+    Glue-backed `information_schema.schemata` is case-insensitive in practice
+    (everything is stored lowercased), so compare lowercased. -#}
+{% macro athena__check_schema_exists(information_schema, schema) -%}
+  {% call statement('check_schema_exists', fetch_result=True, auto_begin=False) -%}
+    select count(*) from information_schema.schemata
+    where lower(schema_name) = '{{ schema | lower }}'
+    {%- if information_schema.database %}
+      and lower(catalog_name) = '{{ information_schema.database | replace('"', '') | lower }}'
+    {%- endif %}
+  {%- endcall %}
+  {{ return(load_result('check_schema_exists').table) }}
+{% endmacro %}

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/macros/adapters/persist_docs.sql

@@ -0,0 +1,18 @@
+{% macro athena__persist_docs(relation, model, for_relation=true, for_columns=true) -%}
+  {% set persist_relation_docs = for_relation and config.persist_relation_docs()%}
+  {% set persist_column_docs = for_columns and config.persist_column_docs() and model.columns %}
+  {% if persist_column_docs %}
+    {% set existing_columns = adapter.get_columns_in_relation(relation) | map(attribute="name") | list %}
+    {% do validate_doc_columns(relation, model.columns, existing_columns) %}
+  {% endif %}
+  {% if persist_relation_docs or persist_column_docs %}
+    {% do adapter.persist_docs_to_glue(
+            relation=relation,
+            model=model,
+            persist_relation_docs=persist_relation_docs,
+            persist_column_docs=persist_column_docs,
+            skip_archive_table_version=true
+         )
+    %}
+  {% endif %}
+{% endmacro %}

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/macros/adapters/python_submissions.sql

@@ -0,0 +1,95 @@
+{%- macro athena__py_save_table_as(compiled_code, target_relation, optional_args={}) -%}
+    {% set location = optional_args.get("location") %}
+    {% set format = optional_args.get("format", "parquet") %}
+    {% set mode = optional_args.get("mode", "overwrite") %}
+    {% set write_compression = optional_args.get("write_compression", "snappy") %}
+    {% set partitioned_by = optional_args.get("partitioned_by") %}
+    {% set bucketed_by = optional_args.get("bucketed_by") %}
+    {% set sorted_by = optional_args.get("sorted_by") %}
+    {% set merge_schema = optional_args.get("merge_schema", true) %}
+    {% set bucket_count = optional_args.get("bucket_count") %}
+    {% set field_delimiter = optional_args.get("field_delimiter") %}
+    {% set spark_ctas = optional_args.get("spark_ctas", "") %}
+
+import pyspark
+
+
+{{ compiled_code }}
+def materialize(spark_session, df, target_relation):
+    import pandas
+    if isinstance(df, pyspark.sql.dataframe.DataFrame):
+        pass
+    elif isinstance(df, pandas.core.frame.DataFrame):
+        df = spark_session.createDataFrame(df)
+    else:
+        msg = f"{type(df)} is not a supported type for dbt Python materialization"
+        raise Exception(msg)
+
+{% if spark_ctas|length > 0 %}
+    df.createOrReplaceTempView("{{ target_relation.schema}}_{{ target_relation.identifier }}_tmpvw")
+    spark_session.sql("""
+    {{ spark_ctas }}
+    select * from {{ target_relation.schema}}_{{ target_relation.identifier }}_tmpvw
+    """)
+{% else %}
+    writer = df.write \
+    .format("{{ format }}") \
+    .mode("{{ mode }}") \
+    .option("path", "{{ location }}") \
+    .option("compression", "{{ write_compression }}") \
+    .option("mergeSchema", "{{ merge_schema }}") \
+    .option("delimiter", "{{ field_delimiter }}")
+    if {{ partitioned_by }} is not None:
+        writer = writer.partitionBy({{ partitioned_by }})
+    if {{ bucketed_by }} is not None:
+        writer = writer.bucketBy({{ bucket_count }},{{ bucketed_by }})
+    if {{ sorted_by }} is not None:
+        writer = writer.sortBy({{ sorted_by }})
+
+    writer.saveAsTable(
+        name="{{ target_relation.schema}}.{{ target_relation.identifier }}",
+    )
+{% endif %}
+
+    return "Success: {{ target_relation.schema}}.{{ target_relation.identifier }}"
+
+{{ athena__py_get_spark_dbt_object() }}
+
+dbt = SparkdbtObj()
+df = model(dbt, spark)
+materialize(spark, df, dbt.this)
+{%- endmacro -%}
+
+{%- macro athena__py_execute_query(query) -%}
+{{ athena__py_get_spark_dbt_object() }}
+
+def execute_query(spark_session):
+    spark_session.sql("""{{ query }}""")
+    return "OK"
+
+dbt = SparkdbtObj()
+execute_query(spark)
+{%- endmacro -%}
+
+{%- macro athena__py_get_spark_dbt_object() -%}
+def get_spark_df(identifier):
+    """
+    Override the arguments to ref and source dynamically
+
+    spark.table('awsdatacatalog.analytics_dev.model')
+    Raises pyspark.sql.utils.AnalysisException:
+    spark_catalog requires a single-part namespace,
+    but got [awsdatacatalog, analytics_dev]
+
+    So the override removes the catalog component and only
+    provides the schema and identifer to spark.table()
+    """
+    return spark.table(".".join(identifier.split(".")[1:]).replace('"', ''))
+
+class SparkdbtObj(dbtObj):
+    def __init__(self):
+        super().__init__(load_df_function=get_spark_df)
+        self.source = lambda *args: source(*args, dbt_load_df_function=get_spark_df)
+        self.ref = lambda *args: ref(*args, dbt_load_df_function=get_spark_df)
+
+{%- endmacro -%}

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/macros/adapters/relation.sql

@@ -0,0 +1,70 @@
+{% macro athena__drop_relation(relation) -%}
+  {%- set native_drop = config.get('native_drop', default=false) -%}
+  {%- set rel_type_object = adapter.get_glue_table_type(relation) -%}
+  {%- set rel_type = none if rel_type_object == none else rel_type_object.value -%}
+  {%- set natively_droppable = rel_type == 'iceberg_table' or relation.type == 'view' -%}
+
+  {%- if adapter.is_s3_tables_database(relation.database) -%}
+    {%- do drop_relation_s3_tables(relation) -%}
+  {%- elif native_drop and natively_droppable -%}
+    {%- do drop_relation_sql(relation) -%}
+  {%- else -%}
+    {%- do drop_relation_glue(relation) -%}
+  {%- endif -%}
+{% endmacro %}
+
+{% macro drop_relation_glue(relation) -%}
+  {%- do log('Dropping relation via Glue and S3 APIs') -%}
+  {%- do adapter.clean_up_table(relation) -%}
+  {%- do adapter.delete_from_glue_catalog(relation) -%}
+{% endmacro %}
+
+{% macro drop_relation_s3_tables(relation) -%}
+  {# S3 Tables manages its own storage, so we must NOT clean up S3 directly.
+     A SQL `DROP TABLE` can't target a federated sub-catalog either (Athena's
+     Hive DDL is limited to <schema>.<table> in the query's default catalog),
+     so we drop the metadata via Glue, which delegates to S3 Tables and removes
+     the underlying data. #}
+  {%- do log('Dropping S3 Tables relation via Glue Data Catalog') -%}
+  {%- do adapter.delete_from_glue_catalog(relation) -%}
+{% endmacro %}
+
+{% macro drop_relation_sql(relation) -%}
+
+  {%- do log('Dropping relation via SQL only') -%}
+  {% call statement('drop_relation', auto_begin=False) -%}
+    {%- if relation.type == 'view' -%}
+      drop {{ relation.type }} if exists {{ relation.render() }}
+    {%- else -%}
+      drop {{ relation.type }} if exists {{ relation.render_hive() }}
+    {% endif %}
+  {%- endcall %}
+{% endmacro %}
+
+{% macro set_table_classification(relation) -%}
+  {%- set format = config.get('format', default='parquet') -%}
+  {% call statement('set_table_classification', auto_begin=False) -%}
+    alter table {{ relation.render_hive() }} set tblproperties ('classification' = '{{ format }}')
+  {%- endcall %}
+{%- endmacro %}
+
+{% macro make_temp_relation(base_relation, suffix='__dbt_tmp', temp_schema=none) %}
+  {%- set temp_identifier = base_relation.identifier ~ suffix -%}
+  {%- set temp_relation = base_relation.incorporate(path={"identifier": temp_identifier}) -%}
+
+  {%- if temp_schema is not none -%}
+    {%- set temp_relation = temp_relation.incorporate(path={
+      "identifier": temp_identifier,
+      "schema": temp_schema
+      }) -%}
+      {%- do create_schema(temp_relation) -%}
+  {% endif %}
+
+  {{ return(temp_relation) }}
+{% endmacro %}
+
+{% macro athena__rename_relation(from_relation, to_relation) %}
+  {% call statement('rename_relation') -%}
+    alter table {{ from_relation.render_hive() }} rename to `{{ to_relation.schema }}`.`{{ to_relation.identifier }}`
+  {%- endcall %}
+{%- endmacro %}

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/macros/adapters/schema.sql

@@ -0,0 +1,21 @@
+{#- Athena CREATE SCHEMA is Hive DDL: schema name only, backtick-quoted (what
+    dbt-athena's Relation.render_hive() produces). Fusion relations have no
+    render_hive(); a catalog-qualified, double-quoted name is rejected. -#}
+{% macro athena__create_schema(relation) -%}
+  {%- call statement('create_schema') -%}
+    create schema if not exists `{{ relation.schema }}`
+  {% endcall %}
+  {{ adapter.add_lf_tags_to_database(relation) }}
+{% endmacro %}
+
+
+{% macro athena__drop_schema(relation) -%}
+  {%- call statement('drop_schema') -%}
+    drop schema if exists {{ relation.without_identifier().render_hive() }} cascade
+  {% endcall %}
+{% endmacro %}
+
+
+{% macro drop_glue_database(database_name, catalog_name='awsdatacatalog') -%}
+  {{ adapter.drop_glue_database(database_name, catalog_name) }}
+{% endmacro %}

crates/dbt-loader/src/dbt_macro_assets/dbt-athena/macros/materializations/hooks.sql

@@ -0,0 +1,17 @@
+{% macro run_hooks(hooks, inside_transaction=True) %}
+  {% set re = modules.re %}
+  {% for hook in hooks | selectattr('transaction', 'equalto', inside_transaction) %}
+    {% set rendered = render(hook.get('sql')) | trim %}
+    {% if (rendered | length) > 0 %}
+      {%- if re.match("optimize\W+\S+\W+rewrite data using bin_pack", rendered.lower(), re.MULTILINE) -%}
+        {%- do adapter.run_operation_with_potential_multiple_runs(rendered, "optimize") -%}
+      {%- elif re.match("vacuum\W+\S+", rendered.lower(), re.MULTILINE) -%}
+        {%- do adapter.run_operation_with_potential_multiple_runs(rendered, "vacuum") -%}
+      {%- else -%}
+        {% call statement(auto_begin=inside_transaction) %}
+          {{ rendered }}
+        {% endcall %}
+      {%- endif -%}
+    {% endif %}
+  {% endfor %}
+{% endmacro %}