morningman opened a new pull request, #68921:
URL: https://github.com/apache/doris/pull/68921
### What problem does this PR solve?
Issue Number: None
Related PR: #66685
Problem Summary:
**In short.** This ports the Iceberg part of branch-4.x #66685 to master: an
INSERT into a partitioned Iceberg table is now hash-distributed by the
transformed partition value, so each partition is written by one writer. On
master today the rows are spread randomly, and a single INSERT that touches
more than 128 partitions fails.
**Background.** Every writer of an Iceberg sink keeps one open file per
partition it has seen, capped by the BE config
`table_sink_partition_write_max_partition_nums_per_writer` (default 128).
branch-4.x #66685 routes partitioned Iceberg writes through an
Iceberg-transform exchange. The FE computes the active partition spec's
transforms, the BE applies them before hashing, and all rows of one partition
reach one writer. Master's Iceberg connector only declares
`requiresParallelWrite`, so the engine falls back to random distribution. The
external sink exchange that master already has (connector-defined partition
functions, used by Paimon's fixed buckets) has no Iceberg function.
**The problem, and what it cost.** With random distribution every writer
receives rows of every partition:
| | master today | with this PR |
|---|---|---|
| `INSERT ... SELECT number, number % 200 FROM numbers("number" = "10000")`
into a table partitioned by `p` | fails: `Too many open partitions 128` |
succeeds: 10,000 rows in 200 partitions |
| Data files per partition after one INSERT | up to one per writer | one per
partition (200 identity partitions: 200 files; 300 `day(ts)` partitions: 300
files) |
| Sink exchange in `EXPLAIN SHAPE PLAN` |
`DistributionSpecHiveTableSinkUnPartitioned` (random) |
`DistributionSpecExternalTableSinkHashPartitioned` |
**How this PR fixes it.**
- **Iceberg connector.** `IcebergWritePlanProvider.getWriteDistribution`
returns an external hash distribution. The route columns are the partition
fields' source columns, the partition function is `iceberg_transform` with the
transforms in spec order (for example `day,identity,bucket[4]`), and the writer
assignment is `SKEWED`, so a large partition can still scale out. It reads the
statement's pinned write schema when one is active. It returns `null` (today's
random parallel write) for an unpartitioned table, a partition field on a
nested source (route columns must be top-level sink outputs), or a backend
older than execution version 17.
- **BE.** The external partition function factory registers
`iceberg_transform`. It reuses `IcebergInsertPartitionFunction`, the MERGE
insert-side router, which applies the same `PartitionColumnTransforms` as the
Iceberg writer. A transform the writer cannot apply fails the partitioner the
same way it fails the writer; it never falls back to hashing the raw value.
- **Mixed versions.** Older backends reject the new function name, so BE
execution version 17 marks the capability and FE `max_be_exec_version` moves to
17. The connector reads the version at optimize time through a new
`ConnectorSession#getBeExecVersion()` (default `Integer.MAX_VALUE`; the engine
session returns `Config.be_exec_version`), the same value
`ConnectorWriteHandle#getBeExecVersion()` carries into `planWrite`. With
`be_exec_version` below 17 the plan stays as today: the sink keeps the random
distribution, and the INSERT above still fails with `Too many open partitions
128`. branch-4.x falls back to a single gathered writer instead.
- The SPI addition is a default method; the plugin-surface baseline is
refreshed. The SPI is unreleased, so `connector.plugin.api.version` is
unchanged. No Thrift change: the function name and its options travel in the
existing `TExternalTableSinkHashPartitionInfo` fields.
Original author of the branch-4.x change: @suxiaogang223
**Results.** On a local single-BE cluster with the Iceberg REST docker
environment, the 200-partition INSERT that fails on master succeeds, a
300-partition `day(ts)` INSERT succeeds too, and every partition holds exactly
one data file. Setting `be_exec_version = 16` brings back the random
distribution and the original error, which is the mixed-version fallback.
### Release note
INSERT into a partitioned Iceberg table now routes rows by partition, so one
INSERT can write more than 128 partitions and each writer no longer opens files
for every partition.
### Check List (For Author)
- Test: Regression test / Unit Test
- FE UT: `IcebergWritePlanProviderTest` (4 new cases: transform routing,
multi-field spec order, random-write fallbacks for unpartitioned / nested
source / BE exec version 16, pinned write schema), `ConnectorSessionImplTest`,
all `fe-connector-spi` tests including the plugin-surface baseline,
`PhysicalConnectorTableSinkTest`. Removing the version gate, reading the live
table instead of the pinned schema, and routing a nested source by its
top-level column each fail one of the new cases.
- BE UT: `ExternalTableSinkHashPartitionerTest` and
`ExternalPartitionFunctionFactoryTest` (14 tests, 3 new: transformed-value
routing with `truncate[10]`, fail-closed on an unknown transform, option
validation). Hashing the raw source value instead of the transformed one fails
both partitioner cases.
- Regression:
`external_table_p0/iceberg/write/test_iceberg_write_partition_distribution`
(200 identity partitions and 300 `day(ts)` partitions in one INSERT each; plan
shape, row counts, one data file per partition). Passed locally against the
Iceberg REST docker environment, together with
`test_iceberg_write_order_distribution_properties`,
`test_iceberg_write_transform_partitions` and `test_iceberg_write_partitions`.
- Behavior changed: Yes. Partitioned Iceberg INSERT / INSERT OVERWRITE uses
a partition-hash exchange instead of random distribution once `be_exec_version`
is 17 or newer.
- Does this need documentation: No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
