github-actions[bot] commented on code in PR #66307:
URL: https://github.com/apache/doris/pull/66307#discussion_r3886142791
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java:
##########
@@ -945,6 +1204,78 @@ private void validateColumnsExist(TableIf table,
}
}
+ private void validateDistributionMappingConstraint(TableNameInfo
tableNameInfo,
+ OlapTable table, DistributionMappingConstraint constraint) {
+ if (table.getCatalogId() != InternalCatalog.INTERNAL_CATALOG_ID) {
+ throw new AnalysisException("Distribution mapping constraint only
supports internal OLAP tables");
+ }
+ if (table.isTemporary()) {
+ throw new AnalysisException("Distribution mapping constraint does
not support temporary tables");
+ }
+ validateColumnsExist(table, constraint.getDeterminantColumnNames(),
toKey(tableNameInfo));
+ validateColumnsExist(table, constraint.getDistributionColumnNames(),
toKey(tableNameInfo));
+
+ TreeSet<String> determinantColumns = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ determinantColumns.addAll(constraint.getDeterminantColumnNames());
+ if (determinantColumns.size() !=
constraint.getDeterminantColumnNames().size()) {
+ throw new AnalysisException("Determinant columns in distribution
mapping constraint must be unique");
+ }
+ TreeSet<String> distributionColumns = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ distributionColumns.addAll(constraint.getDistributionColumnNames());
+ if (distributionColumns.size() !=
constraint.getDistributionColumnNames().size()) {
+ throw new AnalysisException("Distribution columns in distribution
mapping constraint must be unique");
+ }
+
+ if (!(table.getDefaultDistributionInfo() instanceof
HashDistributionInfo)) {
+ throw new AnalysisException("Distribution mapping constraint
requires hash distribution");
+ }
+ if (!constraint.hasCompatibleDistributionColumns(table)) {
+ throw new AnalysisException("Distribution columns in distribution
mapping constraint"
+ + " must be an ordered subset of table distribution
columns");
+ }
+ }
+
+ /** Reject ADD and restore until every registered FE reports this exact
build. */
+ public void validateDistributionMappingFeatureCompatibility() {
+ String currentVersion = getCurrentFrontendVersion();
Review Comment:
[P1] Preserve mappings across later old-FE checkpoints
This exact-build scan only covers FEs registered at ADD time. The new
envelope is readable but not round-trip-preserving for an old binary: after the
mapping commits, ADD FRONTEND can still admit an old follower; its old
TableAttributes ignores the mapping field and its old
ModifyTablePropertyOperationLog reads the mutation as an empty property replay.
If that follower is elected, it can checkpoint an image at/after the ADD
without the mapping, after which journal compaction removes the only record
from which new FEs could recover it. This is a distinct follow-on to the
earlier bootstrap-deserialization thread: the backward-readable fix changes the
failure into silent durable loss. Please keep incompatible FEs out of
admission/leadership while any persisted mapping exists, or use an encoding old
leaders preserve semantically, and cover ADD -> old-FE failover/checkpoint ->
new-FE reload.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -204,12 +251,46 @@ public PhysicalProperties visitPhysicalHashAggregate(
&&
isShuffleCompatible(childOutputProperty.getDistributionSpec())) {
return PhysicalProperties.ANY;
}
+ if
(childOutputProperty.getNaturalDistributionMappingSpec().isPresent()) {
+ return computeAggregateOutputProperties(agg,
childOutputProperty)
Review Comment:
[P1] Do not rebuild mapping locality with distinct group keys
This preserves the hidden proof through every aggregate phase, including the
function-less GLOBAL dedup produced for COUNT(DISTINCT ...). With a hidden-k1
aggregate on one side of a mapping join and serial-source local shuffle
enabled, that dedup downgrades the parent's specific BUCKET requirement to
generic HASH, so the scan is locally rehashed on [k1,k2]. The aggregate above
must then restore BUCKET; because the dedup output is STORAGE_ANY, it falls
back to its grouping expressions [k2,d1] for the BUCKET exchange. Hashing those
determinant/grouping values does not recreate storage buckets hashed by
[k1,k2], yet the aggregate reports BUCKET and the outer colocate join skips
re-alignment with the scan side, so matching rows can land on different tasks.
This Aggregate/dedup path is distinct from the existing Generate/Window/NLJ
barrier thread. Please preserve the specific bucket requirement through the
dedup or drop the hidden proof, and add an asymmetric aggregate-vs-scan result t
est with multiple buckets/tasks and values whose determinant hash differs from
the storage-key hash.
--
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]
