github-actions[bot] commented on code in PR #66307:
URL: https://github.com/apache/doris/pull/66307#discussion_r3702605393
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalOlapScanToPhysicalOlapScan.java:
##########
@@ -132,12 +140,63 @@ public static DistributionSpec
convertDistribution(LogicalOlapScan olapScan) {
}
}
}
- return new DistributionSpecHash(hashColumns,
ShuffleType.NATURAL, olapScan.getTable().getId(),
- olapScan.getSelectedIndexId(),
Sets.newLinkedHashSet(olapScan.getSelectedPartitionIds()));
+ return createNaturalHashSpec(olapScan, hashDistributionInfo,
hashColumns, output);
}
} else {
// RandomDistributionInfo
return DistributionSpecStorageAny.INSTANCE;
}
}
+
+ private static DistributionSpecHash createNaturalHashSpec(LogicalOlapScan
olapScan,
+ HashDistributionInfo hashDistributionInfo, List<ExprId>
hashColumns, List<Slot> output) {
+ return new DistributionSpecHash(hashColumns, ShuffleType.NATURAL,
olapScan.getTable().getId(),
+ olapScan.getSelectedIndexId(),
Sets.newLinkedHashSet(olapScan.getSelectedPartitionIds()),
+ buildDistributionMappings(olapScan.getTable(),
hashDistributionInfo, output));
+ }
+
+ private static List<DistributionMapping>
buildDistributionMappings(OlapTable table,
+ HashDistributionInfo hashDistributionInfo, List<Slot> output) {
+ ConnectContext context = ConnectContext.get();
+ if (context == null ||
!context.getSessionVariable().isEnableColocateMappingConstraint()) {
+ return ImmutableList.of();
+ }
+
+ Map<String, ExprId> columnExprIds = new HashMap<>();
+ for (Slot slot : output) {
+ SlotReference slotReference = (SlotReference) slot;
+ slotReference.getOriginalColumn().ifPresent(
+ column ->
columnExprIds.put(column.getName().toLowerCase(), slot.getExprId()));
+ }
+ Map<String, Integer> distributionIndices = new HashMap<>();
+ List<Column> distributionColumns =
hashDistributionInfo.getDistributionColumns();
+ for (int i = 0; i < distributionColumns.size(); i++) {
+
distributionIndices.put(distributionColumns.get(i).getName().toLowerCase(), i);
+ }
+
+ TableNameInfo tableNameInfo =
TableNameInfoUtils.fromTableOrNull(table);
+ ImmutableList.Builder<DistributionMapping> mappings =
ImmutableList.builder();
+ for (DistributionMappingConstraint constraint :
Env.getCurrentEnv().getConstraintManager()
+ .getDistributionMappingConstraints(tableNameInfo)) {
Review Comment:
[P1] Migrate mappings when a database is renamed
These constraints are keyed only by `catalog.db.table`, but the leader and
replay database-rename paths never migrate `ConstraintManager`. After `db1 ->
db2`, the real table loses its mapping; if `db1.t` is later recreated, this
lookup returns the stale assertion for an unrelated table because there is no
table-id check. That stale proof can suppress a required Join Exchange and lose
rows. Please migrate the keys atomically on rename/replay (or attach them by
stable IDs) and cover old-name reuse.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalOlapScanToPhysicalOlapScan.java:
##########
@@ -132,12 +140,63 @@ public static DistributionSpec
convertDistribution(LogicalOlapScan olapScan) {
}
}
}
- return new DistributionSpecHash(hashColumns,
ShuffleType.NATURAL, olapScan.getTable().getId(),
- olapScan.getSelectedIndexId(),
Sets.newLinkedHashSet(olapScan.getSelectedPartitionIds()));
+ return createNaturalHashSpec(olapScan, hashDistributionInfo,
hashColumns, output);
}
} else {
// RandomDistributionInfo
return DistributionSpecStorageAny.INSTANCE;
}
}
+
+ private static DistributionSpecHash createNaturalHashSpec(LogicalOlapScan
olapScan,
+ HashDistributionInfo hashDistributionInfo, List<ExprId>
hashColumns, List<Slot> output) {
+ return new DistributionSpecHash(hashColumns, ShuffleType.NATURAL,
olapScan.getTable().getId(),
+ olapScan.getSelectedIndexId(),
Sets.newLinkedHashSet(olapScan.getSelectedPartitionIds()),
+ buildDistributionMappings(olapScan.getTable(),
hashDistributionInfo, output));
+ }
+
+ private static List<DistributionMapping>
buildDistributionMappings(OlapTable table,
+ HashDistributionInfo hashDistributionInfo, List<Slot> output) {
+ ConnectContext context = ConnectContext.get();
+ if (context == null ||
!context.getSessionVariable().isEnableColocateMappingConstraint()) {
+ return ImmutableList.of();
+ }
+
+ Map<String, ExprId> columnExprIds = new HashMap<>();
+ for (Slot slot : output) {
+ SlotReference slotReference = (SlotReference) slot;
+ slotReference.getOriginalColumn().ifPresent(
+ column ->
columnExprIds.put(column.getName().toLowerCase(), slot.getExprId()));
Review Comment:
[P1] Resolve mapping determinants through base-column provenance
For a selected synchronous MV, a column displayed as `d` can actually be `x
AS d`; this map then records the `x` ExprId as the declared determinant `d`.
The existing selected-index hash-key path and the new property derivation both
use `tryGetBaseColumnName()` to avoid that substitution. With a `d -> k`
mapping, a join on `L.x = R.d` can therefore appear to cover bucket key `k` and
suppress a required Exchange, losing matching rows. Please key selected-index
determinants by verified base provenance and add forced-rollup
positive/negative coverage.
##########
fe/fe-core/src/test/java/org/apache/doris/catalog/constraint/ConstraintPersistTest.java:
##########
@@ -207,6 +207,24 @@ void externalTableTest() throws Exception {
Assertions.assertEquals(1, loadedMgr.getConstraints(extTni).size());
}
+ @Test
Review Comment:
[P2] Preserve mappings across recoverable drops
This JSON round-trip does not cover the supported recycle-bin lifecycle.
Non-force table drop removes its `ConstraintManager` entry before recycling,
while table recover/replay restores only the table; database drop/recover has
the same loss after `dropDatabaseConstraints`. Thus recovered data silently
loses its declared mapping (including recovery under a new name). Please
preserve constraints in recycle metadata or defer irreversible removal until
erase/force-drop, and add table/database recover plus replay coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java:
##########
@@ -92,6 +93,13 @@ public void run(ConnectContext ctx, StmtExecutor executor)
throws Exception {
} else if (constraint.isUnique()) {
addConstraintAndInvalidate(
tableNameInfo, new UniqueConstraint(name,
ImmutableSet.copyOf(columns)));
+ } else if (constraint.isDistributionMapping()) {
+ Pair<ImmutableList<String>, TableIf> distributionColumnsAndTable =
+ extractColumnsAndTable(ctx,
constraint.toDistributionProject());
+ Preconditions.checkState(table.getId() ==
distributionColumnsAndTable.second.getId(),
+ "determinant and distribution columns must belong to the
same table");
+ addConstraintAndInvalidate(tableNameInfo, new
DistributionMappingConstraint(
+ name, constraint.getMappingId(), columns,
distributionColumnsAndTable.first));
Review Comment:
[P1] Invalidate SQL-result caches for mapping DDL
Constraint add/drop and their EditLog replay paths invalidate only MTMV
rewrite caches. Nereids SQL-cache validation records table identity and visible
data version, and this metadata-only DDL changes neither; keeping this new
switch enabled also keeps the same variable key. A result cached under a bad
`NOT ENFORCED` mapping can therefore survive `DROP CONSTRAINT` and continue
returning rows from the invalid no-shuffle plan. Please invalidate the affected
table's SQL-cache entries on add/drop and replay, with an identical-query
regression.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java:
##########
@@ -92,6 +93,13 @@ public void run(ConnectContext ctx, StmtExecutor executor)
throws Exception {
} else if (constraint.isUnique()) {
addConstraintAndInvalidate(
tableNameInfo, new UniqueConstraint(name,
ImmutableSet.copyOf(columns)));
+ } else if (constraint.isDistributionMapping()) {
+ Pair<ImmutableList<String>, TableIf> distributionColumnsAndTable =
+ extractColumnsAndTable(ctx,
constraint.toDistributionProject());
+ Preconditions.checkState(table.getId() ==
distributionColumnsAndTable.second.getId(),
+ "determinant and distribution columns must belong to the
same table");
+ addConstraintAndInvalidate(tableNameInfo, new
DistributionMappingConstraint(
Review Comment:
[P1] Serialize mapping creation with schema changes
A heavy DROP COLUMN checks constraints once, sets the table to
`SCHEMA_CHANGE`, and keeps the old schema visible until the job publishes. This
ADD path releases its planner table lock before manager validation;
`ConstraintManager` holds only its own lock and checks neither the table lock
nor `OlapTableState.NORMAL`. It can therefore validate and journal a mapping
after DROP's check but before the column disappears, leaving stale metadata
that can reactivate on name reuse. Please add a table-state/lock fence and a
deterministic DROP-vs-ADD replay test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalOlapScanToPhysicalOlapScan.java:
##########
@@ -132,12 +140,63 @@ public static DistributionSpec
convertDistribution(LogicalOlapScan olapScan) {
}
}
}
- return new DistributionSpecHash(hashColumns,
ShuffleType.NATURAL, olapScan.getTable().getId(),
- olapScan.getSelectedIndexId(),
Sets.newLinkedHashSet(olapScan.getSelectedPartitionIds()));
+ return createNaturalHashSpec(olapScan, hashDistributionInfo,
hashColumns, output);
}
} else {
// RandomDistributionInfo
return DistributionSpecStorageAny.INSTANCE;
}
}
+
+ private static DistributionSpecHash createNaturalHashSpec(LogicalOlapScan
olapScan,
+ HashDistributionInfo hashDistributionInfo, List<ExprId>
hashColumns, List<Slot> output) {
+ return new DistributionSpecHash(hashColumns, ShuffleType.NATURAL,
olapScan.getTable().getId(),
+ olapScan.getSelectedIndexId(),
Sets.newLinkedHashSet(olapScan.getSelectedPartitionIds()),
+ buildDistributionMappings(olapScan.getTable(),
hashDistributionInfo, output));
+ }
+
+ private static List<DistributionMapping>
buildDistributionMappings(OlapTable table,
+ HashDistributionInfo hashDistributionInfo, List<Slot> output) {
+ ConnectContext context = ConnectContext.get();
+ if (context == null ||
!context.getSessionVariable().isEnableColocateMappingConstraint()) {
+ return ImmutableList.of();
+ }
+
+ Map<String, ExprId> columnExprIds = new HashMap<>();
+ for (Slot slot : output) {
+ SlotReference slotReference = (SlotReference) slot;
+ slotReference.getOriginalColumn().ifPresent(
+ column ->
columnExprIds.put(column.getName().toLowerCase(), slot.getExprId()));
+ }
+ Map<String, Integer> distributionIndices = new HashMap<>();
+ List<Column> distributionColumns =
hashDistributionInfo.getDistributionColumns();
+ for (int i = 0; i < distributionColumns.size(); i++) {
+
distributionIndices.put(distributionColumns.get(i).getName().toLowerCase(), i);
+ }
+
+ TableNameInfo tableNameInfo =
TableNameInfoUtils.fromTableOrNull(table);
+ ImmutableList.Builder<DistributionMapping> mappings =
ImmutableList.builder();
+ for (DistributionMappingConstraint constraint :
Env.getCurrentEnv().getConstraintManager()
+ .getDistributionMappingConstraints(tableNameInfo)) {
+ ImmutableList.Builder<ExprId> determinants =
ImmutableList.builder();
+ boolean allDeterminantsAvailable = true;
+ for (String column : constraint.getDeterminantColumnNames()) {
+ ExprId exprId = columnExprIds.get(column.toLowerCase());
+ if (exprId == null) {
+ allDeterminantsAvailable = false;
+ break;
+ }
+ determinants.add(exprId);
+ }
+ if (!allDeterminantsAvailable) {
+ continue;
+ }
+ ImmutableList<Integer> targetIndices =
constraint.getDistributionColumnNames().stream()
Review Comment:
[P1] Fence mapped columns against RENAME COLUMN
`Env.renameColumn` (including replay) rewrites the table schema and
hash-distribution names but does not reject or rewrite these constraints.
Renaming a mapped target leaves its old name absent from `distributionIndices`,
so this immutable-list collection receives `null` and ordinary scans fail when
the feature is enabled. Renaming a determinant also leaves a dormant assertion
that can reactivate on a replacement column. Please atomically rewrite the
mapping or reject mapped-column renames on leader and replay, with both target
and determinant tests.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java:
##########
@@ -539,6 +548,12 @@ public String findConstraintWithColumn(
.contains(columnName)) {
return entry.getKey();
}
+ } else if (c instanceof DistributionMappingConstraint) {
+ DistributionMappingConstraint mapping =
(DistributionMappingConstraint) c;
+ if
(mapping.getDeterminantColumnNames().contains(columnName)
Review Comment:
[P1] Compare mapped columns case-insensitively
Constraint analysis stores the catalog column's canonical spelling, while
DROP COLUMN resolves and removes columns with `equalsIgnoreCase`. These
`List.contains` checks are case-sensitive, so a stored `D1` does not block
`DROP COLUMN d1`. Re-adding that name later makes the lowercase scan lookup
bind the stale assertion to the replacement column, which can suppress a
required Exchange and lose rows. Please use Doris's case-insensitive identifier
semantics here and add mixed-case drop/re-add coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/constraint/ConstraintManager.java:
##########
@@ -216,6 +219,12 @@ public ImmutableList<UniqueConstraint>
getUniqueConstraints(
UniqueConstraint.class);
}
+ /** Returns all distribution mapping constraints for the given table. */
+ public ImmutableList<DistributionMappingConstraint>
getDistributionMappingConstraints(
+ TableNameInfo tableNameInfo) {
+ return getConstraintsByType(toKey(tableNameInfo),
DistributionMappingConstraint.class);
Review Comment:
[P1] Transition mappings during atomic backup restore
`BackupMeta` copies tables/resources but not centralized constraints, while
atomic restore directly unregisters the staging/origin tables and registers the
restored object under the origin name without a `ConstraintManager` transition
(the replay path is the same). Because this getter is name-only, the restored
table's new ID inherits the origin table's old mapping. A single-partition
self-join can then pass the same-table/index/partition shortcut and suppress an
Exchange even when restored data violates that assertion. Please back up and
atomically restore/drop mappings with the table object, including alias and
replay tests.
--
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]
