nooneuse opened a new pull request, #66307:
URL: https://github.com/apache/doris/pull/66307
### What problem does this PR solve?
Problem Summary:
Doris can currently use Colocate Join when the join equality conditions
directly cover all hash distribution keys of two tables in the same stable
colocate group.
However, some data models contain determinant columns that consistently map
to distribution keys across tables. For example, if both tables are distributed
by `tenant_id`, and each `user_id` always belongs to exactly one `tenant_id`,
joining the tables by `user_id` is also colocated when the mapping from
`user_id` to `tenant_id` is consistent across both tables.
Previously, Doris could not declare or use this cross-table mapping
relationship. Such queries therefore required Shuffle Join even though the
matching rows were already located in corresponding buckets.
This PR introduces a `COLOCATE MAPPING` constraint and allows Nereids to use
the declared mapping when proving that a Join can run as a Colocate Join.
The implementation supports:
- A mapping determinant covering one or more distribution key positions.
- Multiple mappings jointly covering a composite distribution key.
- A combination of direct distribution-key equalities and mapping-derived
coverage.
- Additional equality predicates that do not affect an already complete
colocate proof.
- Automatic fallback to Shuffle Join when the join conditions do not cover
all distribution key positions.
- Catalog persistence and ADD, DROP, and SHOW lifecycle management for the
new constraint.
The optimization is controlled by the session variable:
```sql
SET enable_colocate_mapping_constraint = true;
```
It is disabled by default. When disabled, scan distribution properties, join
property requests, and Colocate Join decisions retain their original behavior.
The constraint is declared as `NOT ENFORCED`. Doris trusts the mapping
supplied by the user and does not validate it during data writes. Declaring an
incorrect mapping may produce incorrect query results when the optimization is
enabled.
This change is implemented entirely in FE. It reuses the existing Colocate
Hash Join execution path and does not change BE Hash Join semantics.
### Release note
Added an experimental `COLOCATE MAPPING` constraint that allows Nereids to
derive Colocate Join eligibility from user-declared mappings between join
columns and hash distribution keys.
The optimization is disabled by default and can be enabled per session with:
```sql
SET enable_colocate_mapping_constraint = true;
```
### Feature Overview
A Colocate Join normally requires equality conditions that directly cover
all hash distribution keys. This feature additionally allows those keys to be
covered indirectly through named, cross-table distribution mappings.
For tables distributed by:
```sql
DISTRIBUTED BY HASH(k1, k2)
```
the following join can use Colocate Join:
```sql
ON left_table.d1 = right_table.d1
AND left_table.k2 = right_table.k2
```
when both tables declare the same mapping from `d1` to `k1`.
Two separate mappings may also jointly cover the composite distribution key:
```sql
ON left_table.d1 = right_table.d1
AND left_table.d2 = right_table.d2
```
where `d1` maps to `k1` and `d2` maps to `k2`.
Colocate Join is selected only when:
- Both scans retain their underlying `NATURAL` hash distribution.
- Both tables belong to the same stable colocate group.
- The mapping identifiers match across the two tables.
- Corresponding determinant columns are connected by Join equality
predicates.
- Direct equalities and mapping-derived positions jointly cover every
distribution key position.
If any distribution key position remains uncovered, Doris falls back to a
normal Shuffle Join.
### Applicable Scenarios
This feature is useful when:
- Tables are in the same colocate group.
- Tables use compatible composite or single-column hash distribution keys.
- Business keys consistently determine distribution keys across the involved
tables.
- Queries commonly join by those business keys rather than directly by the
distribution keys.
A typical example is:
```text
user_id -> tenant_id
```
where both tables are distributed by `tenant_id`, but queries frequently
join by `user_id`.
This feature should only be used when the declared mapping has identical
semantics across all participating tables.
### Usage
Create two tables in the same colocate group:
```sql
CREATE TABLE orders (
tenant_id BIGINT,
user_id BIGINT,
order_id BIGINT
)
DUPLICATE KEY(tenant_id, user_id)
DISTRIBUTED BY HASH(tenant_id) BUCKETS 16
PROPERTIES (
"replication_num" = "1",
"colocate_with" = "tenant_group"
);
CREATE TABLE users (
tenant_id BIGINT,
user_id BIGINT,
user_name STRING
)
DUPLICATE KEY(tenant_id, user_id)
DISTRIBUTED BY HASH(tenant_id) BUCKETS 16
PROPERTIES (
"replication_num" = "1",
"colocate_with" = "tenant_group"
);
```
Declare the same logical mapping on both tables:
```sql
ALTER TABLE orders
ADD CONSTRAINT orders_user_mapping
COLOCATE MAPPING tenant_by_user (user_id)
DETERMINES DISTRIBUTION KEY (tenant_id)
NOT ENFORCED;
ALTER TABLE users
ADD CONSTRAINT users_user_mapping
COLOCATE MAPPING tenant_by_user (user_id)
DETERMINES DISTRIBUTION KEY (tenant_id)
NOT ENFORCED;
```
The constraint name is table-local, while the mapping identifier must match
across tables:
```text
Mapping identifier: tenant_by_user
```
Enable the optimization:
```sql
SET enable_colocate_mapping_constraint = true;
```
A Join using the determinant columns can then use Colocate Join:
```sql
SELECT *
FROM orders o
JOIN users u
ON o.user_id = u.user_id;
```
Use `EXPLAIN` to verify the selected distribution strategy:
```sql
EXPLAIN
SELECT *
FROM orders o
JOIN users u
ON o.user_id = u.user_id;
```
The plan should contain:
```text
INNER JOIN(COLOCATE)
```
The constraint can be removed with:
```sql
ALTER TABLE orders
DROP CONSTRAINT orders_user_mapping;
ALTER TABLE users
DROP CONSTRAINT users_user_mapping;
```
### Limitations
- The constraint is `NOT ENFORCED`; Doris does not verify mapping
consistency during writes.
- Mapping-based optimization only applies to underlying natural hash
distribution.
- Mapping propagation is conservative across projections. If a projection
removes any original distribution key, the mapping proof is discarded and the
plan falls back to Shuffle Join.
- Aggregate, Union, multi-hop mapping closure, and expression-based
determinants are not supported for now.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should
merge into -->
--
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]
