airbytehq / airbytehq/airbyte

[destination-gcs-data-lake][destination-s3-data-lake] Support a configurable Iceberg partition spec (enable BigQuery/BigLake manifest pruning)

Open
#81,351 2 comments 0 reactions 0 assignees View on GitHub
autoteam community connectors/destination/gcs-data-lake connectors/destination/s3-data-lake team/use
Dominant language
Python
Stars
22.1k
Forks
5.3k
PR merge metrics
PR metrics pending

Description

# [destination-gcs-data-lake][destination-s3-data-lake] Support a configurable Iceberg partition spec (enable BigQuery/BigLake manifest pruning)

## Problem

The Iceberg data-lake destinations create **unpartitioned** Iceberg tables. The only physical layout the connector applies is a sort order on the primary key (`IcebergUtil.createTable` → `buildTable(...).withSortOrder(...).create()`), with no `PartitionSpec`.

For read engines that federate over these tables — in our case Google BigQuery reading a BigLake/Lakehouse Iceberg REST catalog read-only — an unpartitioned table means **no partition/manifest pruning is possible**. Every query over a large table scans all manifests/data files. BigQuery explicitly supports pruning only via Iceberg partition transforms:

> "BigQuery supports manifest pruning using all Iceberg partition transformation functions."
> — Create Apache Iceberg external tables, https://cloud.google.com/bigquery/docs/iceberg-external-tables

This is a concrete cost regression versus hive-partitioned formats (e.g. an `extract_date=YYYY-MM-DD` GCS path layout on a pipeline using GCS destination connector), where a date predicate limited the bytes scanned. On large tables the difference in scan cost per query is large. There is currently no way to recover date pruning with the Iceberg data-lake destinations because there is no partitioning at all.

## Current behavior

Verified against `destination-gcs-data-lake` **v1.0.10** (CDK 1.0.13, Apache Iceberg 1.7.0) and `destination-s3-data-lake` **v0.3.52**, which share the `airbyte-cdk/bulk/toolkits/load-iceberg-parquet` toolkit.

- **No partition config option.** Neither `GcsDataLakeSpecification` nor `S3DataLakeSpecification` exposes any partition setting. There is no `withPartitionSpec` / `PartitionSpec` usage anywhere in the create/spec paths of the connectors or the toolkit.
- **Tables are created unpartitioned.** `IcebergUtil.createTable` sets a file format, applies a sort order derived from identifier fields, and calls `.create()` — it never sets a `PartitionSpec`.
- **Schema sync never touches the spec.** `IcebergTableSynchronizer.maybeApplySchemaChanges` evolves columns, sort order, and identifier fields only; it does not modify the partition spec.

Note the toolkit ALREADY contains partitioned writers (`PartitionedAppendWriter` / `PartitionedDeltaWriter`) and `IcebergTableWriterFactory.create` already branches on `table.spec().isUnpartitioned` and uses them when the table is partitioned. The write side is capable; only the table is never *given* a partition spec.

## Proposed feature

Add a per-stream Iceberg partition-spec configuration to the shared `load-iceberg-parquet` toolkit (so it applies to both `destination-gcs-data-lake` and `destination-s3-data-lake`). Concretely:

1. A connector config that declares a partition spec: one or more `(column, transform)` pairs where `transform ∈ { identity, year, month, day, hour, bucket[N], truncate[W] }` (the Iceberg transforms — https://iceberg.apache.org/spec/#partition-transforms).
2. On table creation, pass this spec to `catalog.buildTable(...).withPartitionSpec(...)`.
3. For already-existing tables, optionally apply Iceberg **partition evolution** to add the declared fields (old data keeps its old spec; new data is written under the new spec — native Iceberg behavior: https://iceberg.apache.org/spec/#partition-evolution).

Note that `_airbyte_extracted_at` cannot currently be used with `day()`/`month()` directly, because in the output Iceberg schema it is a **`long` (epoch millis)**, not a `timestamp` (`Meta.EXTRACTED_AT = IntegerType` → `Types.LongType`); Iceberg's `day`/`month`/`year`/`hour` transforms accept only `date`/`timestamp(tz)` source types. Then, consider changing `_airbyte_extracted_at`'s Iceberg type to `timestamptz` and default to `day(_airbyte_extracted_at)` to make a sensible default work.

## Why it matters

BigQuery (and other engines) prune only via Iceberg partition transforms; an unpartitioned table forecloses that entirely. A `day`/`month` partition on the natural date column restores date-range pruning and cuts scan bytes/cost on large tables, matching what hive-partitioned formats provides. Iceberg + BigLake are fully capable of partitioning today; the gap is solely that the connector never sets a spec.

## Alternatives considered

- **Out-of-band partition spec** (pre-create the table, or `ALTER TABLE … ADD PARTITION FIELD` via Spark/Trino/PyIceberg on the Lakehouse catalog). This works with the current code: `IcebergUtil.createTable` only `loadTable`s an existing table without resetting its spec, the synchronizer never modifies the spec, and `IcebergTableWriterFactory` dispatches to the partitioned writers when `table.spec().isPartitioned()`. However this is unsupported (no config, no tests, no docs), fragile to future connector changes, and puts spec management outside Airbyte — hence this request for a first-class config.
- **Rely on sort order only.** Sort order improves within-file layout but does not enable manifest/partition pruning in BigQuery.

## Environment

- destination-gcs-data-lake 1.0.10 / destination-s3-data-lake 0.3.52
- Shared toolkit: airbyte-cdk/bulk/toolkits/load-iceberg-parquet (CDK 1.0.13, Apache Iceberg 1.7.0)
- Read engine: Google BigQuery over a BigLake Iceberg REST catalog (read-only federation)

---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/13024

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.