airbytehq / airbytehq/airbyte

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

未關閉
#81,351 2 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視
autoteam community connectors/destination/gcs-data-lake connectors/destination/s3-data-lake team/use
主要語言
Python
星號
22.1k
分支
5.4k
平均合併
5 小時
30 天內合併 PR
671

描述

# [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

貢獻指南

開啟貢獻指南

研究方向

Start in airbyte-cdk/bulk/toolkits/load-iceberg-parquet, especially IcebergUtil.createTable and the GcsDataLakeSpecification and S3DataLakeSpecification entry points. Trace IcebergTableSynchronizer.maybeApplySchemaChanges and IcebergTableWriterFactory.create, along with PartitionedAppendWriter and PartitionedDeltaWriter. Done means both destination configurations can declare the requested transforms and tables use the declared partition spec without breaking existing partitioned writers.

由索引模型根據 Issue 內容生成。

評估

技術堆疊
aws, google-cloud, java
領域
cloud, data-engineering, databases
Issue 類型
功能
難度
5/5
預估耗時
一週以上
活躍度
冷清
描述清晰度
基本清楚
新手友好度
35/100

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。