Repository: dbt-athena/dbt-athena (python)

Issue:
BatchDeletePartitions only supports up to 25 partitions per call
failing example model:

```

{{
    config(
            materialized='table',
            ha=true,
            partitioned_by=["partition_date"],
        )
}}

with
    partitions as (
        select sequence(current_date - interval '99' day, current_date, interval '1' day) as date_sequence
), sequence_exploded as (
select cast(single_date as date) as single_date
from partitions cross join unnest (date_sequence) as t(single_date)
), partitioned_table as (
        select
            1 as foo,
            single_date as partition_date
        from sequence_exploded
    )
select *
from partitioned_table
```

Error on second `dbt run` (with existing table):
```
An error occurred (ValidationException) when calling the BatchDeletePartition operation: 1 validation error detected: Value '[PartitionValueList(values=[2023-05-25]), PartitionValueList(values=[2023-03-10]), PartitionValueList(values=[2023-05-03]), PartitionValueList(values=[2023-05-19]), PartitionValueList(values=[2023-05-13]), PartitionValueList(values=[2023-04-11]), PartitionValueList(values=[2023-03-05]), PartitionValueList(values=[2023-03-28]), PartitionValueList(values=[2023-04-17]), PartitionValueList(values=[2023-05-18]), PartitionValueList(values=[2023-05-14]), PartitionValueList(values=[2023-03-29]), PartitionValueList(values=[2023-04-28]), PartitionValueList(values=[2023-04-05]), PartitionValueList(values=[2023-04-26]), PartitionValueList(values=[2023-...
...
...
[2023-03-22]), PartitionValueList(values=[2023-03-13]), PartitionValueList(values=[2023-05-08]), PartitionValueList(values=[2023-04-29]), PartitionValueList(values=[2023-05-23])]' at 'partitionsToDelete' failed to satisfy constraint: Member must have length less than or equal to 25
```

using latest dbt-athena version 163b3d5a71ec131d9e0a7c3888b6d23d05f51b02


Pull request description:
fix: BatchDeletePartitions only accepts up to 25 partitions
via partition list chunking (chunks a 25 partitions)

### Description

- resolve #327
- implement chunking also for creating partitions, here the limit is 100 partitions per batch.
- update `dev-requirements.txt` to include missing packages

## Models used to test - Optional
```

{{
    config(
            materialized='table',
            ha=true,
            partitioned_by=["partition_date"],
        )
}}

with
    partitions as (
        select sequence(current_date - interval '99' day, current_date, interval '1' day) as date_sequence
), sequence_exploded as (
select cast(single_date as date) as single_date
from partitions cross join unnest (date_sequence) as t(single_date)
), partitioned_table as (
        select
            1 as foo,
            single_date as partition_date
        from sequence_exploded
    )
select *
from partitioned_table
```


## Checklist
- [x] You followed [contributing section](https://github.com/dbt-athena/dbt-athena#contributing)
- [x] You kept your Pull Request small and focused on a single feature or bug fix.
- [x] You added unit testing when necessary
- [ ] You added functional testing when necessary


Patch:
diff --git a/dbt/adapters/athena/impl.py b/dbt/adapters/athena/impl.py
index 88d01d5..af0f2cd 100755
--- a/dbt/adapters/athena/impl.py
+++ b/dbt/adapters/athena/impl.py
@@ -34,7 +34,7 @@ from dbt.adapters.athena.relation import (
     get_table_type,
 )
 from dbt.adapters.athena.s3 import S3DataNaming
-from dbt.adapters.athena.utils import clean_sql_comment, get_catalog_id
+from dbt.adapters.athena.utils import clean_sql_comment, get_catalog_id, get_chunks
 from dbt.adapters.base import ConstraintSupport, available
 from dbt.adapters.base.relation import BaseRelation, InformationSchema
 from dbt.adapters.sql import SQLAdapter
@@ -46,6 +46,9 @@ boto3_client_lock = Lock()
 
 
 class AthenaAdapter(SQLAdapter):
+    BATCH_CREATE_PARTITION_API_LIMIT = 100
+    BATCH_DELETE_PARTITION_API_LIMIT = 25
+
     ConnectionManager = AthenaConnectionManager
     Relation = AthenaRelation
 
@@ -522,21 +525,27 @@ class AthenaAdapter(SQLAdapter):
         # if source table has partitions we need to delete and add partitions
         # it source table hasn't any partitions we need to delete target table partitions
         if target_table_partitions:
-            glue_client.batch_delete_partition(
-                DatabaseName=target_relation.schema,
-                TableName=target_relation.identifier,
-                PartitionsToDelete=[{"Values": i["Values"]} for i in target_table_partitions],
-            )
+            for partition_batch in get_chunks(target_table_partitions, AthenaAdapter.BATCH_DELETE_PARTITION_API_LIMIT):
+                glue_client.batch_delete_partition(
+                    DatabaseName=target_relation.schema,
+                    TableName=target_relation.identifier,
+                    PartitionsToDelete=[{"Values": partition["Values"]} for partition in partition_batch],
+                )
 
         if src_table_partitions:
-            glue_client.batch_create_partition(
-                DatabaseName=target_relation.schema,
-                TableName=target_relation.identifier,
-                PartitionInputList=[
-                    {"Values": p["Values"], "StorageDescriptor": p["StorageDescriptor"], "Parameters": p["Parameters"]}
-                    for p in src_table_partitions
-                ],
-            )
+            for partition_batch in get_chunks(src_table_partitions, AthenaAdapter.BATCH_CREATE_PARTITION_API_LIMIT):
+                glue_client.batch_create_partition(
+                    DatabaseName=target_relation.schema,
+                    TableName=target_relation.identifier,
+                    PartitionInputList=[
+                        {
+                            "Values": partition["Values"],
+                            "StorageDescriptor": partition["StorageDescriptor"],
+                            "Parameters": partition["Parameters"],
+                        }
+                        for partition in partition_batch
+                    ],
+                )
 
     def _get_glue_table_versions_to_expire(self, relation: AthenaRelation, to_keep: int):
         """
diff --git a/dbt/adapters/athena/utils.py b/dbt/adapters/athena/utils.py
index b0e9bbd..ee3b20b 100644
--- a/dbt/adapters/athena/utils.py
+++ b/dbt/adapters/athena/utils.py
@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Collection, List, Optional, TypeVar
 
 from mypy_boto3_athena.type_defs import DataCatalogTypeDef
 
@@ -11,3 +11,12 @@ def clean_sql_comment(comment: str) -> str:
 def get_catalog_id(catalog: Optional[DataCatalogTypeDef]) -> Optional[str]:
     if catalog:
         return catalog["Parameters"]["catalog-id"]
+
+
+T = TypeVar("T")
+
+
+def get_chunks(lst: Collection[T], n: int) -> Collection[List[T]]:
+    """Yield successive n-sized chunks from lst."""
+    for i in range(0, len(lst), n):
+        yield lst[i : i + n]


Test patch:
diff --git a/tests/unit/test_adapter.py b/tests/unit/test_adapter.py
index 55397f1..ad66d53 100644
--- a/tests/unit/test_adapter.py
+++ b/tests/unit/test_adapter.py
@@ -728,7 +728,7 @@ class TestAthenaAdapter:
         mock_aws_service.create_table(source_table)
         mock_aws_service.add_partitions_to_table(DATABASE_NAME, source_table)
         mock_aws_service.create_table(target_table)
-        mock_aws_service.add_partitions_to_table(DATABASE_NAME, source_table)
+        mock_aws_service.add_partitions_to_table(DATABASE_NAME, target_table)
         source_relation = self.adapter.Relation.create(
             database=DATA_CATALOG_NAME,
             schema=DATABASE_NAME,
@@ -836,7 +836,7 @@ class TestAthenaAdapter:
         ).get("Partitions")
 
         assert self.adapter.get_glue_table_location(target_relation) == f"s3://{BUCKET}/tables/{source_table}"
-        assert len(target_table_partitions_after) == 3
+        assert len(target_table_partitions_after) == 26
 
     @mock_athena
     @mock_glue
diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py
index b72d931..2385136 100644
--- a/tests/unit/test_utils.py
+++ b/tests/unit/test_utils.py
@@ -1,4 +1,4 @@
-from dbt.adapters.athena.utils import clean_sql_comment
+from dbt.adapters.athena.utils import clean_sql_comment, get_chunks
 
 
 def test_clean_comment():
@@ -12,3 +12,20 @@ def test_clean_comment():
         )
         == "my long comment on several lines with weird spaces and indents."
     )
+
+
+def test_get_chunks_empty():
+    assert len(list(get_chunks([], 5))) == 0
+
+
+def test_get_chunks_uneven():
+    chunks = list(get_chunks([1, 2, 3], 2))
+    assert chunks[0] == [1, 2]
+    assert chunks[1] == [3]
+    assert len(chunks) == 2
+
+
+def test_get_chunks_more_elements_than_chunk():
+    chunks = list(get_chunks([1, 2, 3], 4))
+    assert chunks[0] == [1, 2, 3]
+    assert len(chunks) == 1
diff --git a/tests/unit/utils.py b/tests/unit/utils.py
index eeacd23..60ec02e 100644
--- a/tests/unit/utils.py
+++ b/tests/unit/utils.py
@@ -450,7 +450,7 @@ class MockAWSService:
                 },
                 "Parameters": {"compressionType": "snappy", "classification": "parquet"},
             }
-            for dt in ["2022-01-01", "2022-01-02", "2022-01-03"]
+            for dt in [f"2022-01-{day:02d}" for day in range(1, 27)]
         ]
         glue = boto3.client("glue", region_name=AWS_REGION)
         glue.batch_create_partition(