github-actions[bot] commented on code in PR #68921:
URL: https://github.com/apache/doris/pull/68921#discussion_r4258371976
##########
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] Support binary partition values before selecting this exchange. At the
default execution version 18, INSERT/INSERT OVERWRITE into an identity BINARY
partition or truncate(1, binary_key) now selects iceberg_transform. Both
transforms produce ColumnVarbinary, but IcebergInsertPartitionFunction calls
update_crcs_with_value or update_crc32c_batch, and ColumnVarbinary inherits
IColumn implementations that throw NOT_IMPLEMENTED_ERROR. Consequently
previously supported binary writes fail during routing, including the seven-row
identity/truncate cases in test_iceberg_write_binary_partitions.groovy.
Bucket(binary) returns INT and is unaffected. Add shuffle hashing for the
transformed binary bytes and cover both hash methods before enabling this route
for these partitions.
--
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]
