github-actions[bot] commented on code in PR #68921:
URL: https://github.com/apache/doris/pull/68921#discussion_r4243841753
##########
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:
[P2] Pin the BE execution version for this write. Config.be_exec_version is
mutable, so a 17-to-16 ADMIN change after getWriteDistribution selects
iceberg_transform but before CoordinatorContext builds TQueryOptions leaves a
version-17 partition function in a version-16 plan. An old BE in that rolling
upgrade rejects the function and the INSERT fails. Capture one statement
version for routing and dispatch, or revalidate before dispatch.
##########
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:
[P1] Account for the per-writer open-partition limit when assigning these
hashes. With two writers and 250 small identity partitions p=2*number, default
CRC32C routing sends 130 distinct partitions to one writer and 120 to the
other. The rebalancer does not run below its byte threshold, so
VIcebergTableWriter fails on that writer's 129th partition even though 250 is
below the combined 256-slot capacity. A two-writer boundary case would expose
this; the current 64-writer four-row test cannot.
##########
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:
[P2] Make the route hash treat NaNs as the same Iceberg partition value. For
an identity DOUBLE partition, the exchange hashes raw IEEE bits: positive and
negative quiet NaNs route to different writers with two destinations. The
writer renders both as the same NaN partition key, so even a tiny write creates
multiple files for one partition. Canonicalize NaNs in the transformed hash
input and cover two distinct NaN payloads in a two-writer test.
##########
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:
[P1] Bump the connector plugin API major for this new session method. The
frozen SPI surface records it, but the pom and ConnectorPluginSurfaceTest still
pin 13.0. Since the loader compares only majors and loads SPI parent-first, a
new Iceberg plugin can be admitted by an older 13.0 FE and hit
NoSuchMethodError when getWriteDistribution calls this method. Bump the API to
14.0 and update the test pin with this surface 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]
