Merge branch 'main' into data-contract-views

Author: adavoudi

.github/workflows/_integration-tests.yml

@@ -342,6 +342,7 @@ jobs:
             AWS_REGION: ${{ vars.AWS_REGION }}
             REDSHIFT_TEST_DBNAME: ${{ vars.REDSHIFT_TEST_DBNAME }}
             REDSHIFT_TEST_CROSS_DBNAME: ${{ vars.REDSHIFT_TEST_CROSS_DBNAME }}
+            REDSHIFT_TEST_DBNAME_W_HYPHEN: ${{ vars.REDSHIFT_TEST_DBNAME_W_HYPHEN }}
             REDSHIFT_TEST_PASS: ${{ secrets.REDSHIFT_TEST_PASS }}
             REDSHIFT_TEST_USER: ${{ vars.REDSHIFT_TEST_USER }}
             REDSHIFT_TEST_PORT: ${{ vars.REDSHIFT_TEST_PORT }}

dbt-adapters/.changes/unreleased/Features-20260415-134809.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Added --empty support for dbt seed
+time: 2026-04-15T13:48:09.244397+05:30
+custom:
+    Author: sriramr98
+    Issue: "1865"

dbt-adapters/.changes/unreleased/Fixes-20260413-120000.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Skip empty catalog tables before merge to prevent agate type conflicts during `dbt docs generate`
+time: 2026-04-13T12:00:00.000000+00:00
+custom:
+  Author: tauhidanjum
+  Issue: "1833"

dbt-adapters/src/dbt/adapters/base/impl.py

@@ -2039,7 +2039,11 @@ def catch_as_completed(
         # we want to re-raise on ctrl+c and BaseException
         if exc is None:
             catalog = future.result()
-            tables.append(catalog)
+            # Skip empty catalog results to avoid agate type conflicts.
+            # Empty results cause agate to infer text columns (e.g. column_name)
+            # as Number, which conflicts with Text when merged with non-empty tables.
+            if len(catalog) > 0:
+                tables.append(catalog)
         elif isinstance(exc, KeyboardInterrupt) or not isinstance(exc, Exception):
             raise exc
         else:

dbt-adapters/src/dbt/include/global_project/macros/materializations/seeds/seed.sql

@@ -31,7 +31,10 @@
 
   {% set code = 'CREATE' if full_refresh_mode else 'INSERT' %}
   {% set rows_affected = (agate_table.rows | length) %}
-  {% set sql = load_csv_rows(model, agate_table) %}
+  {% set sql = "" %}
+  {% if rows_affected > 0 %}
+    {% set sql = load_csv_rows(model, agate_table) %}
+  {% endif %}
 
   {% call noop_statement('main', code ~ ' ' ~ rows_affected, code, rows_affected) %}
     {{ get_csv_sql(create_table_sql, sql) }};

dbt-adapters/tests/unit/test_catch_as_completed.py

@@ -0,0 +1,105 @@
+from concurrent.futures import Future
+
+import agate
+from dbt_common.clients.agate_helper import DEFAULT_TYPE_TESTER
+
+from dbt.adapters.base.impl import catch_as_completed
+
+
+def _make_catalog_table(rows):
+    """Build an agate table mimicking catalog output."""
+    return agate.Table.from_object(rows, column_types=DEFAULT_TYPE_TESTER)
+
+
+def _resolved_future(value):
+    """Create a Future that is already resolved with the given value."""
+    f = Future()
+    f.set_result(value)
+    return f
+
+
+def _failed_future(exc):
+    """Create a Future that is already resolved with an exception."""
+    f = Future()
+    f.set_exception(exc)
+    return f
+
+
+class TestCatchAsCompleted:
+    def test_empty_table_excluded_from_merge(self):
+        """An empty catalog result should not cause a type conflict when merged
+        with a non-empty result. This is the scenario from issue #1833: one schema
+        returns rows (column_name inferred as Text) and another returns zero rows
+        (column_name inferred as Number by agate default)."""
+        non_empty = _make_catalog_table([{"column_name": "id", "column_type": "integer"}])
+        empty = _make_catalog_table([])
+
+        futures = [_resolved_future(non_empty), _resolved_future(empty)]
+        result, exceptions = catch_as_completed(futures)
+
+        assert len(exceptions) == 0
+        assert len(result) == 1
+        assert result[0]["column_name"] == "id"
+
+    def test_empty_table_with_conflicting_column_types_excluded(self):
+        """An empty table with explicit column schema whose types conflict with
+        a non-empty table should be excluded. This covers the case where an adapter
+        builds an empty table with column definitions (e.g. column_name as Number)
+        that would conflict with Text in non-empty tables during merge."""
+        non_empty = _make_catalog_table([{"column_name": "id", "column_type": "integer"}])
+        # Build an empty table with explicit columns where column_name is Number —
+        # this is the type conflict that causes the RuntimeError in agate.Table.merge()
+        empty_with_schema = agate.Table(
+            rows=[],
+            column_names=["column_name", "column_type"],
+            column_types=[agate.Number(), agate.Number()],
+        )
+
+        futures = [_resolved_future(non_empty), _resolved_future(empty_with_schema)]
+        result, exceptions = catch_as_completed(futures)
+
+        assert len(exceptions) == 0
+        assert len(result) == 1
+        assert result[0]["column_name"] == "id"
+
+    def test_all_empty_tables(self):
+        """When every schema returns empty results, merge should still succeed."""
+        empty1 = _make_catalog_table([])
+        empty2 = _make_catalog_table([])
+
+        futures = [_resolved_future(empty1), _resolved_future(empty2)]
+        result, exceptions = catch_as_completed(futures)
+
+        assert len(exceptions) == 0
+        assert len(result) == 0
+
+    def test_no_futures(self):
+        """No futures at all should return an empty table."""
+        result, exceptions = catch_as_completed([])
+
+        assert len(exceptions) == 0
+        assert len(result) == 0
+
+    def test_multiple_non_empty_tables_merged(self):
+        """Non-empty tables should still merge normally."""
+        table1 = _make_catalog_table([{"column_name": "id", "column_type": "integer"}])
+        table2 = _make_catalog_table([{"column_name": "name", "column_type": "text"}])
+
+        futures = [_resolved_future(table1), _resolved_future(table2)]
+        result, exceptions = catch_as_completed(futures)
+
+        assert len(exceptions) == 0
+        assert len(result) == 2
+
+    def test_exception_collected(self):
+        """Futures that raise exceptions should be collected, not crash."""
+        non_empty = _make_catalog_table([{"column_name": "id", "column_type": "integer"}])
+        futures = [
+            _resolved_future(non_empty),
+            _failed_future(RuntimeError("connection failed")),
+        ]
+        result, exceptions = catch_as_completed(futures)
+
+        assert len(exceptions) == 1
+        assert "connection failed" in str(exceptions[0])
+        assert len(result) == 1

dbt-athena/.changes/unreleased/Features-20260415-135228.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Added support for --empty in dbt seed
+time: 2026-04-15T13:52:28.479796+05:30
+custom:
+    Author: sriramr98
+    Issue: "1865"

dbt-athena/.changes/unreleased/Fixes-20260320-015650.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix get_partition_batches exceeding partition limit when combining non-bucket and bucket partitions
+time: 2026-03-20T01:56:50.322765+09:00
+custom:
+    Author: dtaniwaki
+    Issue: "1783"

dbt-athena/.changes/unreleased/Under the Hood-20260223-225659.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Optimize Athena partition deletions for insert_overwrite strategy
+time: 2026-02-23T22:56:59.91236+02:00
+custom:
+    Author: juhoautio-rovio
+    Issue: "1125"

dbt-athena/src/dbt/adapters/athena/impl.py

@@ -9,7 +9,20 @@
 from functools import lru_cache
 from textwrap import dedent
 from threading import Lock
-from typing import TYPE_CHECKING, Any, Dict, FrozenSet, Iterable, List, Optional, Set, Tuple, Type
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    FrozenSet,
+    Generator,
+    Iterable,
+    List,
+    Optional,
+    Set,
+    Tuple,
+    Type,
+    Union,
+)
 from urllib.parse import urlparse
 from uuid import uuid4
 
@@ -54,6 +67,7 @@
 from dbt.adapters.athena.s3 import S3DataNaming
 from dbt.adapters.athena.utils import (
     AthenaCatalogType,
+    chunk_iterable,
     clean_sql_comment,
     ellipsis_comment,
     get_catalog_id,
@@ -134,6 +148,9 @@ class AthenaConfig(AdapterConfig):
 class AthenaAdapter(SQLAdapter):
     BATCH_CREATE_PARTITION_API_LIMIT = 100
     BATCH_DELETE_PARTITION_API_LIMIT = 25
+    BATCH_DELETE_S3_OBJECTS_API_LIMIT = 1000
+    PARTITION_PROCESSING_CHUNK_SIZE = 1000
+    GET_PARTITIONS_API_EXPRESSION_MAX_LENGTH = 2048
     INTEGER_MAX_VALUE_32_BIT_SIGNED = 0x7FFFFFFF
 
     ConnectionManager = AthenaConnectionManager
@@ -403,7 +420,9 @@ def get_glue_table_location(self, relation: AthenaRelation) -> Optional[str]:
         return None
 
     @available
-    def clean_up_partitions(self, relation: AthenaRelation, where_condition: str) -> None:
+    def clean_up_partitions(
+        self, relation: AthenaRelation, where_condition: Union[str, List[str]]
+    ) -> None:
         conn = self.connections.get_thread_connection()
         creds = conn.credentials
         client = conn.handle
@@ -417,24 +436,79 @@ def clean_up_partitions(self, relation: AthenaRelation, where_condition: str) ->
                 region_name=client.region_name,
                 config=get_boto3_config(num_retries=creds.effective_num_retries),
             )
-        paginator = glue_client.get_paginator("get_partitions")
-        partition_params = {
-            "CatalogId": catalog_id,
-            "DatabaseName": relation.schema,
-            "TableName": relation.identifier,
-            "Expression": where_condition,
-            "ExcludeColumnSchema": True,
-        }
-        partition_pg = paginator.paginate(**partition_params)
-        partitions = partition_pg.build_full_result().get("Partitions")
-        for partition in partitions:
-            self.delete_from_s3(partition["StorageDescriptor"]["Location"])
-            glue_client.delete_partition(
-                CatalogId=catalog_id,
-                DatabaseName=relation.schema,
-                TableName=relation.identifier,
-                PartitionValues=partition["Values"],
-            )
+
+        where_conditions = (
+            [where_condition] if isinstance(where_condition, str) else where_condition
+        )
+
+        def join_or_conditions(conditions: List[str]) -> str:
+            return " or ".join(conditions)
+
+        def get_partition_expressions() -> Generator[List[str], None, None]:
+            current_chunk: List[str] = []
+            for condition in where_conditions:
+                condition_with_brackets = f"({condition})"
+                if (
+                    len(condition_with_brackets)
+                    > AthenaAdapter.GET_PARTITIONS_API_EXPRESSION_MAX_LENGTH
+                ):
+                    raise DbtRuntimeError(
+                        f"Partition condition exceeds the Glue API expression limit of "
+                        f"{AthenaAdapter.GET_PARTITIONS_API_EXPRESSION_MAX_LENGTH} characters: "
+                        f"'{condition_with_brackets[:100]}...'"
+                    )
+                if current_chunk and (
+                    len(join_or_conditions(current_chunk + [condition_with_brackets]))
+                    > AthenaAdapter.GET_PARTITIONS_API_EXPRESSION_MAX_LENGTH
+                ):
+                    yield current_chunk
+                    current_chunk = []
+                current_chunk.append(condition_with_brackets)
+            if current_chunk:
+                yield current_chunk
+
+        def iter_partitions() -> Generator[Dict[str, Any], None, None]:
+            paginator = glue_client.get_paginator("get_partitions")
+            for expression_chunk in get_partition_expressions():
+                expression = join_or_conditions(expression_chunk)
+                partition_params = {
+                    "CatalogId": catalog_id,
+                    "DatabaseName": relation.schema,
+                    "TableName": relation.identifier,
+                    "Expression": expression,
+                    "ExcludeColumnSchema": True,
+                }
+                for page in paginator.paginate(**partition_params):
+                    yield from page.get("Partitions", [])
+
+        def delete_partition_chunk(partitions):
+            self.bulk_delete_from_s3([p["StorageDescriptor"]["Location"] for p in partitions])
+
+            for glue_batch in get_chunks(
+                partitions, AthenaAdapter.BATCH_DELETE_PARTITION_API_LIMIT
+            ):
+                response = glue_client.batch_delete_partition(
+                    CatalogId=catalog_id,
+                    DatabaseName=relation.schema,
+                    TableName=relation.identifier,
+                    PartitionsToDelete=[{"Values": p["Values"]} for p in glue_batch],
+                )
+                if errors := response.get("Errors"):
+                    for err in errors:
+                        LOGGER.error(
+                            f"Failed to delete Glue partition: Values='{err['PartitionValues']}', "
+                            f"Code='{err['ErrorDetail']['ErrorCode']}', "
+                            f"Message='{err['ErrorDetail']['ErrorMessage']}'"
+                        )
+                    raise DbtRuntimeError(
+                        f"Failed to delete {len(errors)} partition(s) from Glue table "
+                        f"'{relation.schema}.{relation.identifier}'"
+                    )
+
+        for partition_params_chunk in chunk_iterable(
+            iter_partitions(), AthenaAdapter.PARTITION_PROCESSING_CHUNK_SIZE
+        ):
+            delete_partition_chunk(partition_params_chunk)
 
     @available
     def clean_up_table(self, relation: AthenaRelation) -> None:
@@ -540,6 +614,68 @@ def delete_from_s3(self, s3_path: str) -> None:
         else:
             LOGGER.debug("S3 path does not exist")
 
+    def bulk_delete_from_s3(self, s3_paths: List[str]) -> None:
+        if not s3_paths:
+            LOGGER.debug("No S3 paths provided for deletion")
+            return
+
+        conn = self.connections.get_thread_connection()
+        creds = conn.credentials
+        client = conn.handle
+        s3_resource = client.session.resource(
+            "s3",
+            region_name=client.region_name,
+            config=get_boto3_config(num_retries=creds.effective_num_retries),
+        )
+
+        # Group paths by bucket to support partitions spread across multiple buckets
+        paths_by_bucket: Dict[str, List[str]] = {}
+        for s3_path in s3_paths:
+            bucket, _ = self._parse_s3_path(s3_path)
+            paths_by_bucket.setdefault(bucket, []).append(s3_path)
+
+        def filter_objects_by_prefixes(
+            s3_bucket: Any, bucket_paths: List[str]
+        ) -> Generator[Any, None, None]:
+            for s3_path in bucket_paths:
+                LOGGER.debug(f"Listing files for deletion: {s3_path}")
+                _, prefix = self._parse_s3_path(s3_path)
+                yield from s3_bucket.objects.filter(Prefix=prefix)
+
+        def chunk_object_keys(objects_iter) -> Generator[List[Dict[str, str]], None, None]:
+            chunk = []
+            for obj in objects_iter:
+                chunk.append({"Key": obj.key})
+                if len(chunk) >= AthenaAdapter.BATCH_DELETE_S3_OBJECTS_API_LIMIT:
+                    yield chunk
+                    chunk = []
+            if chunk:
+                yield chunk
+
+        for bucket_name, bucket_paths in paths_by_bucket.items():
+            s3_bucket = s3_resource.Bucket(bucket_name)
+            for object_keys in chunk_object_keys(
+                filter_objects_by_prefixes(s3_bucket, bucket_paths)
+            ):
+                if object_keys:
+                    LOGGER.debug(f"Calling delete_objects for {len(object_keys)} objects")
+                    response = s3_bucket.delete_objects(Delete={"Objects": object_keys})
+                    deleted_count = len(response.get("Deleted", []))
+                    error_count = len(response.get("Errors", []))
+                    LOGGER.debug(
+                        f"delete_objects result: {deleted_count} deleted, {error_count} errors"
+                    )
+                    if errors := response.get("Errors"):
+                        for err in errors:
+                            LOGGER.error(
+                                f"Failed to delete S3 object: Key='{err['Key']}', "
+                                f"Code='{err['Code']}', Message='{err['Message']}', "
+                                f"Bucket='{bucket_name}'"
+                            )
+                        raise DbtRuntimeError(
+                            f"Failed to delete {len(errors)} object(s) from S3 bucket '{bucket_name}'"
+                        )
+
     @staticmethod
     def _parse_s3_path(s3_path: str) -> Tuple[str, str]:
         """