github-actions[bot] commented on code in PR #66307:
URL: https://github.com/apache/doris/pull/66307#discussion_r3888243043
##########
COLOCATE_MAPPING_DESIGN_EN.md:
##########
@@ -0,0 +1,437 @@
+# COLOCATE MAPPING Constraint Design Document
+
+## Document Scope
+
+This document describes the purpose, architecture, module design, query and
DDL flows, lifecycle behavior, compatibility strategy, operational constraints,
and test scope of `COLOCATE MAPPING Constraint`.
+
+The feature is implemented in FE metadata and the Nereids optimizer. It adds
no BE operator and changes no FE-BE execution protocol.
+
+## Terminology
+
+| Term | Meaning |
+|---|---|
+| `COLOCATE MAPPING Constraint` | A user declaration that a business-key
column set determines one or more physical distribution columns. Nereids can
use it to prove that some joins do not require a Shuffle. |
+| `determinant` | The ordered business column set on the left side of a
mapping. It need not be unique, but equal determinant values must always imply
equal target distribution-key values. |
+| `mapping` | A determinant-to-distribution-position relationship.
`DistributionMappingConstraint` is the persisted object and
`DistributionMapping` is the planning-time object. |
+| `mapping ID` | The identifier used to match the same business mapping across
tables. Matching also requires the same determinant arity and target
distribution positions. |
+| Constraint name | A table-local name used by ADD, DROP, and SHOW. It is not
the cross-table semantic identifier. |
+| `NOT ENFORCED` | Doris trusts the declaration and does not validate existing
or incoming data. A false declaration can cause an unsafe Shuffle elimination
and incorrect query results. |
+| Distribution key | The ordered columns in `DISTRIBUTED BY HASH(...)`. Column
order is part of the distribution semantics. |
+| Target distribution position | The zero-based position of a mapped column in
the ordered Hash distribution key. The final proof must cover every position. |
+| Natural distribution | The storage Bucket layout delivered by an OLAP Scan.
It is no longer natural after an Exchange redistributes rows. |
+| Natural mapping proof | The FE-only `NaturalDistributionMappingSpec` that
records the physical table, selected index, partitions, Bucket positions, and
currently visible determinants. It cannot create an Exchange. |
+| Stable column unique ID | Persistent column identity used to distinguish an
original column from a same-name replacement. Legacy tables without stable IDs
use base schema version as a conservative substitute. |
+| Table-local metadata | A complete JSON mapping snapshot is stored under a
reserved internal key in the `TableProperty.properties` map owned by the
`OlapTable` and follows the physical table object instead of being indexed by
table name in global `ConstraintManager.constraintsMap`. |
+| Rolling-upgrade gate | Before ADD or Restore, every registered FE must
report the exact current `version-shortHash`. Query planning ignores mappings
and falls back to regular planning when a version is different or unknown. |
+
+## Feature Overview
+
+For two Hash-distributed tables in the same stable Colocate Group, Doris
normally requires join equalities to cover all distribution columns directly.
If both tables are distributed by `tenant_id`, the direct proof is:
+
+```sql
+ON orders.tenant_id = users.tenant_id
+```
+
+When the business invariant is `user_id -> tenant_id`, a join on `user_id`
also connects matching rows in corresponding Buckets:
+
+```sql
+ON orders.user_id = users.user_id
+```
+
+The feature lets the user declare that invariant explicitly:
+
+```sql
+ALTER TABLE orders
+ADD CONSTRAINT orders_user_mapping
+COLOCATE MAPPING tenant_by_user (user_id)
+DETERMINES DISTRIBUTION KEY (tenant_id)
+NOT ENFORCED;
+```
+
+When both tables declare compatible mappings and every proof condition
succeeds, Nereids can select the existing `COLOCATE` Hash Join. Otherwise it
retains the existing Shuffle or Broadcast alternatives.
+
+The feature is disabled by default:
+
+```sql
+SET enable_colocate_mapping_constraint = true;
+```
+
+### Code Composition
+
+The following figures use `git diff --numstat` for the current worktree
against the `upstream/master` merge-base and include the latest persistence and
Aggregate-barrier tests. Each file is assigned once by primary responsibility,
so the percentages describe change composition rather than complexity or risk.
+
+| Category | Files | Added/Deleted Lines | Share of All Changes | Share of
Production Changes |
+|---|---:|---:|---:|---:|
+| Core feature logic | 21 | 1,279 / 95 | 25.3% | 72.1% |
+| Metadata and lifecycle compatibility | 8 | 521 / 11 | 9.8% | 27.9% |
+| Unit tests, regression cases, and results | 16 | 3,496 / 25 | 64.9% | N/A |
+| Total | 45 | 5,296 / 131 | 100% | N/A |
+
+Core logic is concentrated in:
+
+- Syntax and commands: `DorisLexer.g4`, `DorisParser.g4`,
`LogicalPlanBuilder.java`, `Constraint.java`, `AddConstraintCommand.java`,
`DropConstraintCommand.java`, and `ShowConstraintsCommand.java`.
+- Constraint model: `DistributionMappingConstraint.java` and the catalog
`Constraint.java`.
+- Scan and properties: `LogicalOlapScanToPhysicalOlapScan.java`,
`DistributionMapping.java`, `DistributionSpecHash.java`,
`NaturalDistributionMappingSpec.java`, and `PhysicalProperties.java`.
+- Propagation and final join proof: `PhysicalHashAggregate.java`,
`ChildOutputPropertyDeriver.java`, `RequestPropertyDeriver.java`,
`ChildrenPropertiesRegulator.java`, `CostAndEnforcerJob.java`, and
`JoinUtils.java`.
+- Session control: `SessionVariable.java`.
+
+Metadata and lifecycle compatibility is concentrated in:
+
+- Table-local storage and access: `TableProperty.java` and
`ConstraintManager.java`.
+- Journal and replay: `ModifyTablePropertyOperationLog.java`, `EditLog.java`,
and `Env.java`.
+- DDL protection and restore: `SchemaChangeHandler.java` and `RestoreJob.java`.
+- FE version visibility: `Frontend.java`.
+
+Unlike the earlier patch-oriented design, the current implementation does not
modify external catalogs, HMS events, `RefreshManager`, MTMV rename,
recycle-bin, or Replace/Swap code paths. Most lifecycle correctness follows
from one rule: the mapping belongs to the physical table object.
+
+## 1. Purpose
+
+### 1.1 Applicable Scenarios
+
+- Internal OLAP tables are in the same stable Colocate Group.
+- Their Hash distribution layouts are compatible.
+- Queries commonly join on business keys that stably determine the physical
distribution keys.
+- The user can validate the declared invariant through data-production
controls or independent audits.
+
+### 1.2 Supported Scope
+
+- Single-column and composite determinants.
+- One mapping covering one or more ordered distribution-key positions.
+- Multiple mappings jointly covering a composite distribution key.
+- Direct distribution-key equalities combined with mapping-derived coverage.
+- Conservative propagation through Project and supported ordinary non-DISTINCT
Aggregate plans that preserve natural Bucket locality.
+- Direct Slot aliases and non-truncating, hash-value-preserving character
widening casts in Project output.
+- Safe fallback to existing join distribution alternatives when ordinary proof
conditions fail.
+
+### 1.3 Explicit Non-Goals
+
+- External catalogs, HMS tables, `RemoteOlapTable`, and temporary tables.
+- Non-Hash-distributed tables.
+- Runtime enforcement of the `NOT ENFORCED` data invariant.
+- Expression determinants, multi-hop inference, or mapping-closure inference.
+- Using a mapping to construct Exchange, Shuffle, or Bucket Shuffle.
+- Propagation through Union, Intersect, Except, Repeat/Grouping Sets, or a
path already redistributed by Exchange.
+- Propagation across DISTINCT aggregate functions, MultiDistinct phases, or
pure deduplication Aggregates. Such queries remain executable, but upper
planning that depends on proof crossing this boundary falls back.
+- Automatic mapping copies for CREATE TABLE LIKE or CTAS.
+
+## 2. Overall Design
+
+### 2.1 Design Principles
+
+1. **Mappings belong to physical table objects**: a complete snapshot lives
under a reserved internal key in the `TableProperty.properties` map owned by
the `OlapTable`; renames and object lifecycle operations do not maintain a
second name-based index.
+2. **Proof, not execution capability**: a mapping proves existing Bucket
locality and reuses the current Colocate Hash Join.
+3. **Two-stage conservative validation**: each child must satisfy a
non-enforceable mapping request, followed by a final cross-table mapping and
colocate validation.
+4. **Do not use what cannot be proven**: an ordinary proof miss,
schema-incompatible mapping, or mixed FE version prevents construction of a
mapping proof and makes the query fall back to regular planning; metadata
mutation and Restore still fail strictly.
+5. **Central lifecycle rules**: table-local ownership, schema identity
binding, and a single version gate replace per-operation patches.
+6. **Backward-readable and preservable persistence**: the complete snapshot
uses the existing table-property map and journal envelope, so old FE replays
and checkpoints the internal key as opaque data. It still cannot understand or
use mappings, so the version gate prevents it from participating in feature use.
+
+### 2.2 Metadata Model
+
+`DistributionMappingConstraint` persists:
+
+- The table-local constraint name and cross-table mapping ID.
+- Ordered determinant and target distribution-column names.
+- The base schema version at binding time.
+- Stable unique IDs for determinant and target columns.
+- Type signatures for determinant and target columns.
+
+ADD binds the declaration to the current schema. Planning and Restore then
verify:
+
+- The table is still Hash distributed.
+- Target columns remain an ordered subset of the current distribution key.
+- Names, types, and stable unique IDs still match.
+- If a referenced column has no stable unique ID, the base schema version is
unchanged.
+
+The last rule intentionally makes any schema-version change invalidate a
mapping on legacy tables without stable IDs. This avoids accidentally binding
to a dropped-and-recreated same-name column; the operator must DROP and
recreate the mapping.
+
+### 2.3 Persistence Model
+
+- Persistent source:
`TableProperty.properties["__distribution_mapping_constraints"]` stores a
complete JSON snapshot in deterministic constraint-name order. The runtime
mapping map is derived from this property and is not persisted separately.
+- ADD/DROP journal: every operation rewrites the complete snapshot through the
standard properties field of `OP_MODIFY_TABLE_PROPERTIES`; dropping the final
mapping explicitly writes `[]`.
+- Old-FE replay/checkpoint: old code merges the internal key as an opaque
table property and serializes it again, preserving the latest ADD or DROP
snapshot without understanding, showing, or using mappings.
+- New-FE replay: the table is located by database and table ID, then the
complete snapshot is applied under the table write lock and decoded into the
derived mapping map.
+- Binlog: FE versions that support the feature do not publish mapping
snapshots as ordinary table-property binlogs.
+
+### 2.4 Planning Proof Model
+
+```text
+Olap Scan
+ -> validate FE versions and mapping/schema binding
+ -> convert persisted constraints into DistributionMapping objects
+ -> create NATURAL hash property + NaturalDistributionMappingSpec
+ -> conservatively remap proof through Project/ordinary non-DISTINCT Aggregate
+ -> drop proof at DISTINCT/MultiDistinct/pure-deduplication Aggregate
+ -> generate a COLOCATE_MAPPING_REQUIRE join candidate
+ -> validate both child properties and the final cross-table proof
+ -> reuse the existing COLOCATE Hash Join
+```
+
+`NaturalDistributionMappingSpec` separately carries hidden physical Bucket
facts. A Project or ordinary non-DISTINCT Aggregate may remove an original
distribution-key Slot without pretending that an executable Hash distribution
still exists. The proof is only valid for mapping-based Colocate eligibility
and is never enforceable by Exchange.
+
+## 3. Module Design
+
+### 3.1 SQL and Command Layer
+
+Syntax:
+
+```sql
+ALTER TABLE <table>
+ADD CONSTRAINT <constraint_name>
+COLOCATE MAPPING <mapping_id> (<determinants>)
+DETERMINES DISTRIBUTION KEY (<distribution_columns>)
+NOT ENFORCED;
+```
+
+The command layer performs privilege checks, column resolution,
object-identity revalidation, and ADD/DROP/SHOW dispatch. Mappings are accepted
only for internal, non-temporary OLAP tables.
+
+### 3.2 ConstraintManager and Locking
+
+`ConstraintManager` is the access API, but mappings are not inserted into its
global `constraintsMap`. It:
+
+- Validates and reads or writes the table-local map.
+- Prevents a mapping and an existing PK/FK/UNIQUE constraint from sharing a
name.
+- Creates schema identity bindings.
+- Applies the centralized version gate and planning-time compatibility
validation.
+- Merges global constraints and table-local mappings for SHOW.
+
+ADD/DROP lock order is database read lock -> table write lock ->
ConstraintManager lock. The in-memory mutation and journal submission occur
while metadata is locked; `EditLogItem.await()` runs after all metadata locks
are released.
+
+### 3.3 Schema Change
+
+For a determinant or target distribution column referenced by a mapping:
+
+- DROP COLUMN is rejected.
+- RENAME COLUMN is rejected.
+- MODIFY COLUMN is rejected.
+- Hash-to-Random distribution conversion is rejected.
+
+A same-name column change limited to a Rollup does not alter the base-table
mapping and is not incorrectly blocked.
+
+If an old FE replays an unsupported schema operation and bypasses these
front-door checks, a new FE still detects the incompatibility through schema
identity validation, ignores the table's mappings, emits a rate-limited
warning, and falls back to regular planning instead of consuming a stale proof.
+
+### 3.4 Scan and Property Propagation
+
+Scan constructs a mapping proof only when the session switch is enabled,
mappings exist, and version/schema validation succeeds. A mapping enters the
physical property only when every determinant Slot is available.
+
+Project supports:
+
+- A directly forwarded Slot.
+- `Alias(Slot)`.
+- `Alias(Cast(Slot))` only for a non-truncating character widening cast that
preserves the value and Hash bytes.
+
+General expressions, narrowing casts, or a missing determinant drop the
affected mapping.
+
+Aggregate propagation is allowed only when:
+
+- It is not a DISTINCT aggregate function, MultiDistinct phase, or pure
deduplication Aggregate.
+- Its child still carries natural Bucket locality and no Exchange has cut it.
+- It does not originate from Repeat or Grouping Sets.
+- Every Group By expression is a direct Slot.
+- Direct distribution keys and complete determinants in Group By cover every
distribution-key position.
+- The output retains determinants required by the parent Join.
+
+DISTINCT, MultiDistinct, and pure deduplication Aggregates both stop
mapping-property requests from reaching their children and clear mapping proofs
from their outputs. This prevents a DISTINCT plan from forwarding a locality
proof that is valid only for a different row grouping. The query remains
supported; only the Mapping optimization is disabled.
+
+### 3.5 Join Proof
+
+The mapping candidate uses non-enforceable `COLOCATE_MAPPING_REQUIRE`. If a
child does not already satisfy it, the optimizer cannot insert an Exchange to
manufacture success and must discard the candidate.
+
+Final validation requires:
+
+- Equal distribution-key counts.
+- The same stable Colocate Group, or the existing
same-table/same-index/single-partition exception.
+- Slot-to-Slot equality for every hash conjunct.
+- Matching mapping ID, determinant arity, and target distribution positions.
+- Join equalities connecting determinant columns in declaration order.
+- Complete coverage of every distribution position by direct equality or
mapping.
+
+Failure of any condition prevents a mapping-based Colocate Join.
+
+### 3.6 Cache and Execution
+
+- SQL cache is disabled only when a Scan actually constructs a mapping proof.
+- Queries that do not consume mappings keep existing cache behavior.
+- No MTMV rename or rewrite-cache lifecycle patch is required because mappings
have no global name-based reference.
+- BE executes the existing Colocate Hash Join; there is no new runtime column,
Thrift field, or BE state.
+
+## 4. Critical Execution Flows
+
+### 4.1 Query Flow
+
+1. With the session switch disabled, mapping metadata does not affect planning.
+2. A Scan with mappings checks every FE's exact `version-shortHash`; a mixed
or unknown version ignores the table's mappings and falls back to regular
planning.
+3. The Scan validates the persisted schema binding; if any mapping is
incompatible, it ignores all mappings on that table, emits a rate-limited
warning, and falls back to regular planning.
+4. Only after both checks pass does it create a natural-distribution proof
containing the available determinants.
+5. Project and eligible ordinary non-DISTINCT Aggregates conservatively
propagate the proof; DISTINCT/MultiDistinct/pure-deduplication Aggregates,
Exchange, and set-operation boundaries drop it.
+6. Join adds a mapping candidate while preserving the original
Shuffle/Broadcast candidates.
+7. The candidate survives only if both children satisfy the request and final
cross-table validation succeeds.
+8. Execution uses the existing Colocate Hash Join.
+
+### 4.2 ADD CONSTRAINT Flow
+
+1. Resolve determinant and target columns and check ALTER privilege.
+2. Revalidate that the analyzed table object is still the current database
entry.
+3. Under database read and table write locks, validate table state, type,
columns, Hash layout, target order, and constraint-name collisions.
+4. Require every registered FE to report the exact current build.
+5. Bind names to schema version, stable IDs, and type signatures.
+6. Update the complete, constraint-name-sorted snapshot in
`TableProperty.properties` and submit the backward-readable journal.
+7. Release metadata locks before waiting for journal durability.
+
+### 4.3 DROP CONSTRAINT Flow
+
+1. Resolve the current table and check ALTER privilege.
+2. If the name identifies a table-local mapping, use the mapping path;
otherwise use the existing PK/FK/UNIQUE path.
+3. Remove the mapping, write the resulting complete snapshot, and submit the
journal under the table and manager locks; the final removal writes `[]`.
+4. Release locks and wait for persistence.
+
+DROP intentionally has no version gate so mappings can be removed before
downgrade or while FE versions are mixed.
+
+### 4.4 Table and Database Rename
+
+A mapping has no database-name or table-name reference and follows the same
physical `OlapTable` object:
+
+```text
+rename name binding
+ -> same OlapTable object
+ -> same TableProperty
+ -> same mapping metadata
+```
+
+No mapping index needs to be updated and no stale old-name entry can remain.
Column rename is different because mappings bind column identity, so it is
rejected.
+
+### 4.5 Truncate, Drop, and Recover
+
+- Truncate's metadata copy preserves `TableProperty`, so mappings survive; new
data must still honor the declared invariant.
+- Drop places the physical table object and its mappings in the recycle bin.
+- Recover restores that same object and its mappings.
+- A same-name CREATE after Drop creates a different object and does not
inherit the old mapping.
+
+### 4.6 Replace and Swap
+
+Ownership follows the physical object rather than the name:
+
+- `REPLACE ... swap=false`: the replacement object takes the target name and
keeps the replacement object's mappings; the old object and its mappings are
discarded.
+- `REPLACE ... swap=true`: the two physical objects exchange names, and each
mapping remains attached to its original object.
+
+### 4.7 CREATE TABLE LIKE and CTAS
+
+Both statements create new physical table objects and do not copy mappings. A
`NOT ENFORCED` mapping is a business assertion about a data-production contract
and cannot be inferred merely from schema similarity.
+
+### 4.8 Backup and Restore
+
+Backup deep-copies the `OlapTable`, including the mapping snapshot in
`TableProperty.properties`.
+
+Before changing target table state or creating replicas, Restore preflights
every backed-up OLAP table that contains mappings:
Review Comment:
[P2] Document the unconditional Atomic Restore rejection
The implementation now rejects every Atomic Restore whose selected backup
table contains a mapping, regardless of FE version or schema compatibility,
while this section (and the rolling-upgrade/behavior summaries later in both EN
and CN documents) says Restore is preflighted and becomes available after
convergence. Could we distinguish the supported non-atomic Restore path here
and explicitly document the unconditional Atomic Restore rejection in both
files? As written, operators following the design docs will expect a mapped
Atomic Restore to pass these two gates, but
`RestoreJob.validateDistributionMappingConstraintsForRestore()` always cancels
it.
##########
fe/fe-core/src/main/java/org/apache/doris/backup/RestoreJob.java:
##########
@@ -1063,6 +1067,35 @@ private void checkAndPrepareMeta() {
setState(RestoreJobState.CREATING);
}
+ private boolean validateDistributionMappingConstraintsForRestore() {
+ boolean featureCompatibilityValidated = false;
+ for (String tableName : jobInfo.backupOlapTableObjects.keySet()) {
+ OlapTable restoredTable = (OlapTable)
backupMeta.getTable(tableName);
+ if
(env.getConstraintManager().getDistributionMappingConstraints(restoredTable).isEmpty())
{
Review Comment:
[P1] Reconcile the live mapping before restoring existing-table data
This skips a mapping-free backup, but a non-atomic Restore into an existing
table keeps the live `OlapTable`/`TableProperty` while restoring its
partitions. The schema signature does not compare mapping assertions, and these
mappings are `NOT ENFORCED`, so a schema-compatible backup can contain equal
determinants in different target buckets while the destination's old proof
remains active; mapping-enabled joins can then skip a required Exchange and
return wrong results. Please reject a mismatch between the live and backup
mapping snapshots (including a missing backup snapshot), or durably clear the
live mappings before publishing restored data, and cover this existing-table
case.
##########
fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java:
##########
@@ -1004,7 +1004,10 @@ public static void loadJournal(Env env, Long logId,
JournalEntity journal) {
case OperationType.OP_MODIFY_REPLICATION_NUM: {
ModifyTablePropertyOperationLog log =
(ModifyTablePropertyOperationLog) journal.getData();
env.replayModifyTableProperty(opCode, log);
- env.getBinlogManager().addModifyTableProperty(log, logId);
+ // Mapping constraints only reuse this legacy envelope for
old-FE readability.
+ if (!log.hasDistributionMappingConstraintMutation()) {
Review Comment:
[P1] Keep the mapping lifecycle symmetric for CCR
A full-sync Restore copies the table-local mapping snapshot, and
`setBeingSyncedProperties()` does not strip it, but a later source-side DROP
writes the complete `[]` snapshot through this branch and is deliberately
omitted from the feature-binlog stream. Subsequent replicated rows may
therefore violate the old assertion while the destination still consumes it as
a locality proof and can return wrong join results. Please either
replicate/apply the versioned mapping snapshot or strip/disable mappings on
`is_being_synced` restores, with a full-sync -> DROP -> violating-load
regression.
--
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]
