morningman commented on code in PR #68921:
URL: https://github.com/apache/doris/pull/68921#discussion_r4244946756
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorSession.java:
##########
@@ -114,6 +114,17 @@ default boolean isExternalScanTaskReuseEnabled() {
getSessionProperties().get("enable_external_scan_task_reuse"));
}
+ /**
+ * Query-wide backend execution version selected by the engine, for
planning decisions that depend on a
+ * backend capability (for example a write distribution whose partition
function older backends reject).
+ * Mirrors {@link
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getBeExecVersion()},
which
+ * carries the same value at {@code planWrite}. Offline and third-party
sessions default to the
+ * newest-capability side of any fence.
+ */
+ default int getBeExecVersion() {
Review Comment:
Not bumping here, on purpose, as on #68874. The connector SPI has not
shipped in a release: the FE and every connector plugin are built from the same
tree and released together, so there is no Iceberg plugin built against an
older 13.0 surface for the gate to keep out. Until the SPI ships, methods are
added without moving `connector.plugin.api.version`. The frozen surface
baseline (`connector-plugin-surface.txt`) is refreshed in this PR, so the
surface test still pins the new method.
##########
fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorSessionImpl.java:
##########
@@ -165,6 +166,12 @@ public Map<String, String> getSessionProperties() {
return sessionProperties;
}
+ @Override
Review Comment:
Not changed here. Every capability gate on this path reads the live
`Config.be_exec_version` at plan time:
`PhysicalConnectorTableSink.requireExternalWriterRoutingSupport` (the version
13 gate for the Hive and Paimon routes), the Variant gate in
`PluginDrivenTableSink.bindDataSink`, and
`ConnectorWriteHandle#getBeExecVersion()`, which the new session method
mirrors. The coordinator reads the config again when it builds `TQueryOptions`.
So the window is not specific to this route: a statement planned at one value
and dispatched after an operator changes it can carry any of these
capabilities. It also fails closed here: a backend without `iceberg_transform`
returns `Unsupported external sink partition function 'iceberg_transform'`, and
the INSERT fails without committing anything. Closing the window means taking
one version per statement for all of these gates and for dispatch, which is a
separate change.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java:
##########
@@ -614,6 +618,58 @@ transform, parseTransformParam(transform),
sourceColumnName,
return new ConnectorWritePartitionSpec(spec.specId(), fields);
}
+ /**
+ * Routes a partitioned INSERT / INSERT OVERWRITE by the Iceberg partition
value: the BE
+ * {@code iceberg_transform} partition function hashes each partition
field's transformed source column, so the
+ * rows of one partition reach one writer (SKEWED lets a large partition
scale out). Without it every writer
+ * receives every partition, opens a file per partition it sees, and fails
past
+ * {@code table_sink_partition_write_max_partition_nums_per_writer} open
partitions.
+ *
+ * <p>{@code null} keeps the engine's random parallel write: an
unpartitioned table, a partition field on a nested
+ * source (the route columns are top-level sink outputs), or a backend
older than the routing function.</p>
+ */
+ @Override
+ public ConnectorWriteDistribution getWriteDistribution(ConnectorSession
session,
+ ConnectorTableHandle tableHandle) {
+ if (session.getBeExecVersion() <
SUPPORT_TRANSFORM_ROUTING_EXEC_VERSION) {
+ return null;
+ }
+ IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
+ Optional<IcebergWriteSchemaContext> active =
+ IcebergStatementScope.activeWriteSchema(
+ session, handle.getDbName(), handle.getTableName());
+ Table table = active.isPresent() ? null : resolveTable(session,
handle);
+ PartitionSpec spec = active.isPresent()
+ ? active.get().getPartitionSpec() : table.spec();
+ if (spec == null || !spec.isPartitioned()) {
+ return null;
+ }
+ Schema schema = active.isPresent()
+ ? active.get().getSchema() : table.schema();
+ List<String> routeColumns = new ArrayList<>();
+ List<String> transforms = new ArrayList<>();
+ for (PartitionField field : spec.fields()) {
+ String sourceColumnName = findTopLevelColumnName(schema,
field.sourceId());
+ if (sourceColumnName == null) {
+ return null;
+ }
+ routeColumns.add(sourceColumnName);
+ transforms.add(field.transform().toString());
+ }
+ return ConnectorWriteDistribution.externalHash(routeColumns,
TRANSFORM_PARTITION_FUNCTION,
Review Comment:
Right about the mechanism, and it reaches this PR's own regression suite, so
part of this is fixed in 62f9b3c8cea.
Routing by hash gives each writer the partitions whose hash lands on it, and
`table_sink_partition_write_max_partition_nums_per_writer` caps that share; a
writer whose share goes over 128 still fails. branch-4.x (#66685) routes the
same way, through the same `IcebergInsertPartitionFunction` and `SKEWED`
assignment. Before this PR every writer received rows of every partition, so
your 250-partition write failed on every writer whatever the writer count; now
it fails only when one writer's share is over the cap. On a local
single-backend cluster, by `parallel_pipeline_task_num`:
| Writers | 200 identity partitions | 300 `day(ts)` partitions |
|---|---|---|
| 1 | `Too many open partitions 128` | `Too many open partitions 128` |
| 2 | succeeds | `Too many open partitions 128` |
| 3, 4, 8 | succeeds | succeeds |
Keeping every writer under the cap would need capacity-aware assignment
instead of hashing, which is a change to the `SKEWED` assigner that the Hive
route shares, so it stays out of this port. The release note and the PR
description now say that the cap applies to each writer's share.
The test part: the sink's writers follow the source fragment's parallelism,
and the External pipeline runs fuzzy sessions, which draw
`parallel_pipeline_task_num` from 0 to 7, on a single backend. With one or two
writers `test_iceberg_write_partition_distribution` would fail, so 62f9b3c8cea
pins `parallel_pipeline_task_num = 8` in the suite.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java:
##########
@@ -614,6 +618,58 @@ transform, parseTransformParam(transform),
sourceColumnName,
return new ConnectorWritePartitionSpec(spec.specId(), fields);
}
+ /**
+ * Routes a partitioned INSERT / INSERT OVERWRITE by the Iceberg partition
value: the BE
+ * {@code iceberg_transform} partition function hashes each partition
field's transformed source column, so the
+ * rows of one partition reach one writer (SKEWED lets a large partition
scale out). Without it every writer
+ * receives every partition, opens a file per partition it sees, and fails
past
+ * {@code table_sink_partition_write_max_partition_nums_per_writer} open
partitions.
+ *
+ * <p>{@code null} keeps the engine's random parallel write: an
unpartitioned table, a partition field on a nested
+ * source (the route columns are top-level sink outputs), or a backend
older than the routing function.</p>
+ */
+ @Override
+ public ConnectorWriteDistribution getWriteDistribution(ConnectorSession
session,
+ ConnectorTableHandle tableHandle) {
+ if (session.getBeExecVersion() <
SUPPORT_TRANSFORM_ROUTING_EXEC_VERSION) {
+ return null;
+ }
+ IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
+ Optional<IcebergWriteSchemaContext> active =
+ IcebergStatementScope.activeWriteSchema(
+ session, handle.getDbName(), handle.getTableName());
+ Table table = active.isPresent() ? null : resolveTable(session,
handle);
+ PartitionSpec spec = active.isPresent()
+ ? active.get().getPartitionSpec() : table.spec();
+ if (spec == null || !spec.isPartitioned()) {
+ return null;
+ }
+ Schema schema = active.isPresent()
+ ? active.get().getSchema() : table.schema();
+ List<String> routeColumns = new ArrayList<>();
+ List<String> transforms = new ArrayList<>();
+ for (PartitionField field : spec.fields()) {
+ String sourceColumnName = findTopLevelColumnName(schema,
field.sourceId());
+ if (sourceColumnName == null) {
+ return null;
+ }
+ routeColumns.add(sourceColumnName);
+ transforms.add(field.transform().toString());
Review Comment:
Not changed here. The route hashes through `IcebergInsertPartitionFunction`,
which master already uses for the insert side of Iceberg MERGE and which
branch-4.x's route uses too; this PR only registers it as an external partition
function. It feeds the transformed column to the shuffle hash
(`update_shuffle_hashes`), which hashes a FLOAT or DOUBLE by its bits. The
effect is limited to file count: in an identity FLOAT or DOUBLE partition, NaNs
with different bit patterns can reach different writers, each of which writes
its own file into the same NaN partition, and every row still lands in the
right partition. Canonicalizing NaN belongs in that shared router, where it
would cover MERGE as well, rather than in this routing change.
--
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]
