github-actions[bot] commented on code in PR #66685:
URL: https://github.com/apache/doris/pull/66685#discussion_r3783910891
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergTableSink.java:
##########
@@ -135,31 +138,105 @@ public PhysicalProperties getRequirePhysicalProperties()
{
return PhysicalProperties.GATHER;
}
- Set<String> partitionNames = new
java.util.TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ if (targetIcebergTable.spec().isPartitioned()) {
+ if (Config.be_exec_version
+ <
DistributionSpecExternalTableSinkHashPartitioned.MIN_BE_EXEC_VERSION) {
+ return PhysicalProperties.GATHER;
+ }
+ DistributionSpecIcebergTableSinkHashPartitioned distributionSpec
+ = buildPartitionDistributionSpec();
+ // An unsupported transform must not silently hash its raw source
column as though it
+ // were the final Iceberg partition value. GATHER is the safe
fallback until the
+ // transform is supported.
+ return distributionSpec == null
+ ? PhysicalProperties.GATHER
+ : new PhysicalProperties(distributionSpec);
+ }
+ return PhysicalProperties.EXTERNAL_TABLE_SINK_UNPARTITIONED;
+ }
+
+ private DistributionSpecIcebergTableSinkHashPartitioned
buildPartitionDistributionSpec() {
+ List<org.apache.doris.nereids.trees.expressions.Slot> outputSlots =
child().getOutput();
+ if (cols.size() != outputSlots.size()) {
+ return null;
+ }
+
+ Map<String, ExprId> columnExprIds = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ Map<String, Column> columnsByName = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ for (int i = 0; i < cols.size(); i++) {
+ columnExprIds.put(cols.get(i).getName(),
outputSlots.get(i).getExprId());
Review Comment:
[P1] Map Iceberg partition fields to the coerced output order
BindSink keeps `cols` in the explicit INSERT-list order, but
`getOutputProjectByCoercion(insertSchema, ...)` rebuilds the child in Iceberg
schema order. Zipping `cols[i]` with `outputSlots[i]` therefore associates the
wrong ExprId after, for example, `INSERT INTO t (s, d) ...`; a `year(d)`
transform can be serialized against the string slot and fail in the BE instead
of executing a valid insert. Build this map from the target-schema order
actually used by the child (or carry the name-to-ExprId map through binding),
and cover a reordered full column list.
##########
regression-test/suites/paimon_write/test_paimon_write_pk.groovy:
##########
@@ -174,8 +174,8 @@ suite("test_paimon_write_pk", "p0,external,paimon") {
FROM t_pk_string_bucket ORDER BY event_id"""
assertTableEquals("t_pk_string_bucket", "ORDER BY event_id")
- // FT-045: A fixed-bucket PK table uses one writer until Doris has a
- // bucket-aware exchange, so all versions owned by one bucket stay
together.
+ // FT-045: The bucket-aware Exchange may use multiple writers, but
every
Review Comment:
[P1] Update this plan golden for the new fixed-bucket exchange
With `be_exec_version` now defaulting to 12,
`getRequirePhysicalProperties()` returns
`DistributionSpecPaimonTableSinkHashPartitioned` for this HASH_FIXED table even
though strict-consistency DML is disabled. The checked-in
`test_paimon_write_pk.out` still expects `DistributionSpecGather`, so this
enabled suite will fail at `qt_pk_writer_scaling_plan` instead of testing the
new plan. Please update the golden (and preferably assert the bucket-aware
distribution explicitly).
--
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]
