Add better support for BigQuery partitions

[waterworthd-cim]

Feature description

dlt has a biqquery_adaptor that accepts a partition column name, this is rather limiting, big query supports different partition types (i.e. timestamp | date | datetime | int64) at different granularities (i.e. hour | day | month | year) or integer ranges (i.e. start + end + interval), along with options to raise an exception if a query doesn't filter by the partition key, and/or to expire partitions after n days etc.

Are you a dlt user?

Yes, I run dlt in production.

Use case

This is important for cost management, I've observed a large step change in BigQuery costs using incremental merge replication on large tables - using proper partitioning rectifies this but for all but in most cases I've had to manually create the destinations tables in order to properly partition.

Proposed solution

One option is to change the type of the partition argument to a dict similar to dbt (https://docs.getdbt.com/reference/resource-configs/bigquery-configs) rather than a column name and generate the PARTITION BY clause

The other option is to leave partition as a str, but accept the BigQuery syntax literally ie.

DATE_TRUNC(transaction_date, MONTH)
  OPTIONS (
    partition_expiration_days = 3,
    require_partition_filter = TRUE)

see https://cloud.google.com/bigquery/docs/creating-partitioned-tables

Related issues

No response

[rudolfix]

@waterworthd-cim is bigquery adapter lacking functionality:

def bigquery_adapter(
    data: Any,
    partition: TColumnNames = None,
    cluster: TColumnNames = None,
    round_half_away_from_zero: TColumnNames = None,
    round_half_even: TColumnNames = None,
    table_description: Optional[str] = None,
    table_expiration_datetime: Optional[str] = None,
    insert_api: Optional[Literal["streaming", "default"]] = None,
    autodetect_schema: Optional[bool] = None,
    partition_expiration_days: Optional[int] = None,
) -> DltResource:
    """

you import adapters from dlt.destination.adapters AFAIK we have a section on it in our docs too

[waterworthd-cim]

@rudolfix yes, unless I'm missing something

dlt requires partition to be a column name, not an expression. This means really the only use case appears to be a partition on a date column (partitioning on a timestamp / datetime / integer without binning isn't generally all that useful). Pretty much all other BigQuery partition schemes require additional arguments, or cluster should be an bq expression - for example if I want to partition a timestamp column into monthly bins the bq syntax is

DATE_TRUNC(transaction_date, MONTH)
  OPTIONS (
    partition_expiration_days = 3,
    require_partition_filter = TRUE)

The require_partition_filter is supported, partition_expiration_days doesn't appear to be, but the main issue is bigquery_adapter only appears to support partition="transaction_date" and not partition="DATE_TRUNC(transaction_date, MONTH)" (or construct this from a dict like dbt). Similarly I don't see a way of partitioning by an integer column since you need to specify a partition expression rather than partition column

PARTITION BY
  RANGE_BUCKET(customer_id, GENERATE_ARRAY(0, 100, 10))
  OPTIONS (
    require_partition_filter = TRUE);

There's also an option to partition by "ingestion time" which I've never used, but again it doesn't appear to be supported.

[rudolfix]

@waterworthd-cim OK this makes sense. if we allow to specify the clause coming after PARTITION BY directly as sql (ie. in partition argument) then it will cover all your use cases, right?

[waterworthd-cim]

Yeah, assuming you mean just the expression between PARTITION BY and OPTIONS - that's all I need, atm it appears to convert DATE_TRUNC(transaction_date, MONTH) to date_trunc_transaction_date_monthx and then look for a column of that name.

Plus whilst I don't need it, I think it would pay to add the require_partition_filter option as well (as an argument to be consistent with the others?

[rudolfix]

this or full expression including options... we need to figure out how to mod the adapter, having this legacy parameters...

[franviera92]

Hello dear, we currently use Airbyte and our downstream flow takes as reference the _airbyte_extracted_at field, which is created by Airbyte to partition. I would like this to be already integrated with DLT or, failing that, to do what the colleagues mention.