Fix cluster hint overriding partition hint on bigquery

dlt/destinations/impl/bigquery/bigquery_adapter.py

@@ -141,7 +141,7 @@ def bigquery_adapter(
             column.pop(PARTITION_HINT, None)  # type: ignore[typeddict-item]
 
         if isinstance(partition, str):
-            column_hints[partition] = {"name": partition, PARTITION_HINT: True}  # type: ignore[typeddict-unknown-key]
+            column_hints.setdefault(partition, {"name": partition})[PARTITION_HINT] = True  # type: ignore[typeddict-unknown-key]
 
         if isinstance(partition, PartitionTransformation):
             partition_hint: Dict[str, str] = {}
@@ -156,7 +156,7 @@ def bigquery_adapter(
                 "`cluster` must be a list of column names or a single column name as a string."
             )
         for column_name in cluster:
-            column_hints[column_name] = {"name": column_name, CLUSTER_HINT: True}  # type: ignore[typeddict-unknown-key]
+            column_hints.setdefault(column_name, {"name": column_name})[CLUSTER_HINT] = True  # type: ignore[typeddict-unknown-key]
         additional_table_hints[CLUSTER_COLUMNS_HINT] = cluster
 
     # Implementing rounding logic flags
@@ -169,7 +169,7 @@ def bigquery_adapter(
                 " name."
             )
         for column_name in round_half_away_from_zero:
-            column_hints[column_name] = {"name": column_name, ROUND_HALF_AWAY_FROM_ZERO_HINT: True}  # type: ignore[typeddict-unknown-key]
+            column_hints.setdefault(column_name, {"name": column_name})[ROUND_HALF_AWAY_FROM_ZERO_HINT] = True  # type: ignore[typeddict-unknown-key]
 
     if round_half_even:
         if isinstance(round_half_even, str):
@@ -179,7 +179,7 @@ def bigquery_adapter(
                 "`round_half_even` must be a list of column names or a single column name."
             )
         for column_name in round_half_even:
-            column_hints[column_name] = {"name": column_name, ROUND_HALF_EVEN_HINT: True}  # type: ignore[typeddict-unknown-key]
+            column_hints.setdefault(column_name, {"name": column_name})[ROUND_HALF_EVEN_HINT] = True  # type: ignore[typeddict-unknown-key]
 
     if round_half_away_from_zero and round_half_even:
         if intersection_columns := set(round_half_away_from_zero).intersection(

tests/load/bigquery/test_bigquery_table_builder.py

@@ -292,6 +292,25 @@ def partitioned_table():
     assert expected_clause in sql_partitioned
 
 
+def test_adapter_hints_comprehensive_single_column() -> None:
+    @dlt.resource
+    def partitioned_table():
+        yield {
+            "user_id": 10000,
+            "name": "user 1",
+            "created_at": "2021-01-01T00:00:00Z",
+            "category": "category 1",
+            "score": 100.0,
+        }
+
+    # hints should merge even across multiple adapter calls
+    bigquery_adapter(partitioned_table, partition="user_id")
+    bigquery_adapter(partitioned_table, cluster="user_id")
+    assert partitioned_table.columns == {
+        "user_id": {"name": "user_id", PARTITION_HINT: True, CLUSTER_HINT: True},
+    }
+
+
 @pytest.mark.parametrize(
     "destination_config",
     destinations_configs(default_sql_configs=True, subset=["bigquery"]),
@@ -1001,6 +1020,36 @@ def sources() -> List[DltResource]:
         assert ["col1"] == hints_cluster_fields, "`hints` table IS NOT clustered by `col1`."
 
 
+@pytest.mark.parametrize(
+    "destination_config",
+    destinations_configs(default_sql_configs=True, subset=["bigquery"]),
+    ids=lambda x: x.name,
+)
+def test_adapter_hints_clustering_and_partitioning(
+    destination_config: DestinationTestConfiguration,
+) -> None:
+    @dlt.resource(columns=[{"name": "col1", "data_type": "bigint"}])
+    def data() -> Iterator[Dict[str, str]]:
+        yield from [{"col1": str(i)} for i in range(10)]
+
+    bigquery_adapter(data, partition="col1", cluster="col1")
+
+    pipeline = destination_config.setup_pipeline(f"bigquery_{uniq_id()}", dev_mode=True)
+    pipeline.run(data)
+
+    with pipeline.sql_client() as c:
+        nc: google.cloud.bigquery.client.Client = c.native_connection
+        fqtn = c.make_qualified_table_name("data", quote=False)
+        table = nc.get_table(fqtn)
+
+        assert table.clustering_fields == [
+            "col1"
+        ], f"Expected clustering fields ['col1'], got {table.clustering_fields}"
+        assert (
+            table.range_partitioning is not None and table.range_partitioning.field == "col1"
+        ), f"Expected partition field 'col1', got {table.range_partitioning}"
+
+
 def test_adapter_hints_empty() -> None:
     @dlt.resource(columns=[{"name": "int_col", "data_type": "bigint"}])
     def some_data() -> Iterator[Dict[str, str]]: