morningman commented on code in PR #68921:
URL: https://github.com/apache/doris/pull/68921#discussion_r4259535526
##########
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:
Confirmed for writes with more than one writer, and fixed in a18e560610e:
`getWriteDistribution()` keeps the random write when any partition field
transforms to BINARY or FIXED (identity or truncate on a binary or fixed
source). `bucket(binary_key)` yields an INT and is still routed.
One correction: the VALUES cases in `test_iceberg_write_binary_partitions`
do not reach this, and they passed on this PR's External run. A VALUES source
has one instance, so the sink fragment gets one too (`UnassignedShuffleJob`
keeps the biggest child's instance count), and `ExchangeSinkOperatorX` sends to
a single destination as RANDOM without calling the partition function. A source
with several instances does hash. On a local cluster, master's Iceberg MERGE,
which uses the same `IcebergInsertPartitionFunction`, already fails this way on
an identity(binary) partition: `Method update_crc32c_batch is not supported for
ColumnVarbinary`. Giving `ColumnVarbinary` a shuffle hash would let these
partitions route and would fix that MERGE failure too; that is a separate
change.
Tests: `getWriteDistributionKeepsRandomWritesForBinaryPartitionValues`
covers identity(binary), truncate(binary), identity(fixed), a spec mixing an
INT and a binary identity, and the routed bucket(binary).
`test_iceberg_write_binary_partitions` now also writes binary identity and
truncate partitions with INSERT ... SELECT from numbers() at eight writers.
--
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]
