planner: DISTINCT materializes all predicate matches instead of using per-row semijoin/FirstMatch

## Bug Report

Please answer these questions before submitting your issue. Thanks!

### 1. Minimal reproduce step (Required)

[tidb_distinct_correlated_firstmatch_repro.sql](https://github.com/user-attachments/files/30153723/tidb_distinct_correlated_firstmatch_repro.sql)

[tidb_distinct_correlated_firstmatch_repro_result.txt](https://github.com/user-attachments/files/30153724/tidb_distinct_correlated_firstmatch_repro_result.txt)

Run the attached reproduction script:

```bash
mysql -h 127.0.0.1 -P 4000 -u root --comments --table \
< tidb_distinct_correlated_firstmatch_repro.sql \
> tidb_distinct_correlated_firstmatch_repro_result.txt 2>&1
```

The script creates:

- `t0`: 2,000 rows, with an index `i1(c1)`
- `t1`: 20,000 rows
- `t0_no_match`: 100 rows used to verify no-match semantics

`t0.c1` ranges from `10` to `4008`.

Each `t1` row contains a threshold in `t1.c3`. The threshold ranges from `12` to `4010`, with 2,000 distinct threshold values. Therefore, the number of matching `t0` rows varies for different `t1` rows.

The primary query is:

```sql
SELECT DISTINCTROW
t1.c4 AS ref0,
t1.c3 AS ref1,
t1.c2 AS ref2
FROM t1, t0
WHERE t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED);
```

All projected columns and all `DISTINCT` keys come from `t1`.

`t0` only determines whether the current `t1` row has at least one matching row. Additional matching `t0` rows only produce duplicate copies of the same projected `t1` values.

The data shape is:

```text
t0 rows = 2,000
t1 rows = 20,000
distinct thresholds = 2,000
matching join pairs = 20,010,000
final DISTINCT rows = 20,000
```

The script compares the following query forms.

#### Baseline

```sql
SELECT DISTINCTROW
t1.c4,
t1.c3,
t1.c2
FROM t1;
```

#### Original query

```sql
SELECT DISTINCTROW
t1.c4,
t1.c3,
t1.c2
FROM t1, t0
WHERE t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED);
```

#### Equivalent correlated `EXISTS` rewrite

```sql
SELECT DISTINCTROW
t1.c4,
t1.c3,
t1.c2
FROM t1
WHERE EXISTS (
SELECT 1
FROM t0
WHERE t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED)
);
```

#### Correlated `Apply` diagnostic

```sql
SELECT DISTINCTROW
t1.c4,
t1.c3,
t1.c2
FROM t1
WHERE EXISTS (
SELECT /*+ NO_DECORRELATE(), USE_INDEX(t0, i1) */ 1
FROM t0
WHERE t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED)
LIMIT 1
);
```

The reproduction script:

- collects table statistics;
- verifies result counts;
- verifies the behavior when no inner row matches;
- runs `EXPLAIN FORMAT = 'verbose'`;
- runs `EXPLAIN ANALYZE` three times for the baseline, original query, correlated `EXISTS` rewrite, and `NO_DECORRELATE` diagnostic;
- and runs the original query once with HashJoin disabled.

The complete SQL script and execution result are attached.

### 2. What did you expect to see? (Required)

All projected columns and all `DISTINCT` keys come from `t1`.

For each `t1` row, `t0` only determines whether at least one row satisfies the correlated predicate:

```sql
t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED)
```

After the first matching `t0` row has been found, additional matching rows cannot add a new final `DISTINCT` value. They only create duplicate copies of the same projected `t1` row.

The relevant execution semantics are therefore:

```text
for each t1 row:
check whether at least one matching t0 row exists
stop after the first match
```

I expected the optimizer to consider transforming the duplicate-only inner join into a per-outer-row semijoin or FirstMatch-style plan.

A logically equivalent query is:

```sql
SELECT DISTINCTROW
t1.c4,
t1.c3,
t1.c2
FROM t1
WHERE EXISTS (
SELECT 1
FROM t0
WHERE t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED)
);
```

One possible physical implementation would be:

```text
TableScan(t1)
└─per-row semijoin / Apply
└─parameterized lookup on t0
└─stop after the first matching row
```

For example, if the predicate can be converted into a dynamic index range, the inner access could be logically equivalent to:

```text
IndexRangeScan(t0.i1, c1 < current_t1_threshold)
└─Limit 1
```

The optimizer does not need to produce exactly this physical plan. Any equivalent execution strategy that avoids materializing every matching `t1 × t0` pair would be sufficient.

### 3. What did you see instead (Required)

All tested query forms return the same result count:

```text
baseline = 20,000
original = 20,000
exists_rewrite = 20,000
no_decorrelate_apply = 20,000

result_count_equivalence = PASS
```

The no-match test also preserves the expected semantics:

```text
original_rows = 0
exists_rows = 0
apply_rows = 0
```

Statistics were successfully collected:

```text
t0 Row_count = 2,000
t1 Row_count = 20,000
Modify_count = 0
statistics health = 100
```

Therefore, the behavior is not caused by missing or pseudo statistics.

#### Baseline

The baseline only scans and deduplicates `t1`:

```text
HashAgg
└─TableFullScan(t1)
```

Three runs:

```text
6.00 ms
6.66 ms
6.86 ms
```

Median runtime:

```text
6.66 ms
```

#### The original query materializes every predicate match

The original query uses:

```text
HashAgg
└─HashJoin CARTESIAN inner join
├─IndexFullScan(t0.i1)
└─TableFullScan(t1)
```

The correlated inequality is evaluated as a residual join condition:

```text
t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED)
```

The optimizer estimates a very large join result:

```text
HashJoin estRows = 40,000,000
```

The actual join output is:

```text
HashJoin actRows = 20,010,000
```

These 20,010,000 matching rows are then reduced by `HashAgg` to only:

```text
20,000 rows
```

The intermediate join output is approximately 1,000.5 times larger than the final result:

```text
20,010,000 / 20,000 = 1,000.5x
```

Three runs:

```text
3.15 s
3.26 s
3.12 s
```

Median runtime:

```text
3.15 s
```

Compared with the baseline:

```text
3.15 s / 6.66 ms = approximately 473x
```

The optimizer already estimates that the join will produce tens of millions of rows, but no duplicate-aware semijoin/FirstMatch alternative is generated for the original `DISTINCT` query.

#### The correlated `EXISTS` rewrite becomes a semi join, but remains expensive

The equivalent correlated `EXISTS` form is recognized as:

```text
HashAgg
└─HashJoin CARTESIAN semi join
├─IndexFullScan(t0.i1)
└─TableFullScan(t1)
```

The semi join emits only 20,000 rows to the upper `HashAgg`, confirming that only existence is required for each `t1` row.

However, it still evaluates the non-equality predicate using a Cartesian Hash Semi Join rather than an efficient per-row FirstMatch access path.

Three runs:

```text
2.04 s
2.27 s
2.18 s
```

Median runtime:

```text
2.18 s
```

The `EXISTS` rewrite is faster than the original inner join because duplicate matches are no longer emitted to `HashAgg`, but it is still approximately:

```text
2.18 s / 6.66 ms = 327x
```

slower than the baseline.

#### `NO_DECORRELATE()` produces `Apply + LIMIT 1`, but still performs repeated index full scans

With:

```sql
/*+ NO_DECORRELATE(), USE_INDEX(t0, i1) */
```

TiDB produces:

```text
HashAgg
└─Apply CARTESIAN semi join
├─TableFullScan(t1)
└─Limit 1
└─Selection
└─IndexReader
└─IndexFullScan(t0.i1)
```

This plan confirms that per-outer-row early termination is semantically valid.

However, the inner access is still:

```text
IndexFullScan(t0.i1)
→ Selection
→ Limit 1
```

rather than a parameterized range access such as:

```text
IndexRangeScan(t0.i1, c1 < current_threshold)
→ Limit 1
```

A representative run reports:

```text
Apply cacheHitRatio = 90%
Limit actRows = 2,000
IndexReader actRows = 2,048,000
cop tasks = 6,000
runtime = approximately 2.8 seconds
```

The 90% Apply cache hit ratio is expected because the 20,000 outer rows contain 2,000 distinct threshold values.

Despite the cache and inner `LIMIT 1`, TiDB repeatedly reads large portions of the inner index because the correlated inequality is not converted into a parameterized range.

The plan also reports that the string-form `GREATEST` expression cannot be pushed down to TiKV. This may contribute to the inner access limitation, but it does not change the main logical optimization opportunity: the original `DISTINCT` inner join can be reduced to a per-row semijoin/FirstMatch.

#### Disabling HashJoin does not expose a FirstMatch plan

When HashJoin is disabled, TiDB uses a MergeJoin-based plan, but still produces:

```text
MergeJoin actRows = 20,010,000
```

The runtime increases to approximately:

```text
12.1 seconds
```

Therefore, this does not appear to be only a cost-model error between HashJoin and an already available efficient plan.

TiDB does not construct the desired transformation:

```text
DISTINCT over t1 columns
+
predicate-connected duplicate-only t0 input
→ per-t1-row semijoin / FirstMatch
```

#### Impact

The query returns 20,000 rows, but TiDB first produces 20,010,000 matching join rows.

The unnecessary work grows with:

```text
number of outer rows
×
number of matching inner rows per outer row
```

For larger inputs, this behavior can cause:

- severe CPU amplification;
- increased query latency;
- unnecessary intermediate-row processing;
- resource contention;
- and query timeouts.

The current workaround is to manually rewrite the query using a correlated `EXISTS`.

However, even the explicit `EXISTS` form currently uses a Cartesian Hash Semi Join, and the `NO_DECORRELATE` form uses repeated index full scans rather than an efficient parameterized FirstMatch path.

#### Suggested fix

Consider adding a logical transformation for `DISTINCT` and equivalent duplicate-insensitive aggregation plans.

When:

- all projected columns and all `DISTINCT` keys come from one join input;
- the other input contributes no observable output values;
- the other input only determines whether the current outer row has at least one predicate match;
- and multiple matching rows from the other input can only create duplicates removed by `DISTINCT`;

the optimizer could transform the inner join into a per-outer-row semijoin or FirstMatch plan.

For this query, the logical transformation is equivalent to:

```sql
SELECT DISTINCTROW
t1.c4,
t1.c3,
t1.c2
FROM t1
WHERE EXISTS (
SELECT 1
FROM t0
WHERE t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED)
);
```

The optimizer could then consider a parameterized inner access path, or another equivalent execution strategy that stops after the first match.

If applicable, an index access path could use a dynamic range logically equivalent to:

```text
t0.i1 range: c1 < current_t1_threshold
```

with FirstMatch or `LIMIT 1` semantics.

The transformation must preserve no-match semantics: if a particular `t1` row has no matching `t0` row, that `t1` row must not appear in the final result.

Regression tests could cover:

- correlated equality and inequality predicates;
- different numbers of matching inner rows;
- repeated and unique correlated parameter values;
- nullable projected columns;
- `DISTINCT` and equivalent `GROUP BY` forms;
- and behavior with HashJoin enabled or disabled.

#### Relation to #69915 and #69917

This issue is related to:

- [#69915](https://github.com/pingcap/tidb/issues/69915)
- [#69917](https://github.com/pingcap/tidb/issues/69917)

All three issues involve join inputs that can produce duplicate rows without adding new final `DISTINCT` values.

However, the required transformations are different.

In #69915, the additional inputs are independent and predicate-free, and the reproducer contains `ORDER BY` and `LIMIT`. The optimization needs to combine one-time global existence handling with preservation of the ordered `DISTINCT ... ORDER BY ... LIMIT` early-termination path.

In #69917, the additional inputs are also independent and predicate-free. They can be reduced to one-time global existence checks for the entire query, with no per-outer-row dependency.

In this issue, the matching condition depends on the current `t1` row:

```sql
t0.c1 <
CAST(GREATEST(t1.c3, '0.8727017201037127') AS SIGNED)
```

Therefore, `t0` cannot be replaced by a one-time global existence check. Each `t1` row requires its own correlated existence test.

The missing optimization here is specifically:

```text
predicate-connected duplicate-only input
→ per-outer-row semijoin / FirstMatch
```

rather than the global existence reductions involved in #69915 and #69917.

### 4. What is your TiDB version? (Required)

```text
Release Version: v8.5.7
Edition: Community
Git Commit Hash: 202b7f47286a1109b5c957401d34c9358d130ae0
Git Branch: HEAD
UTC Build Time: 2026-07-15 02:06:00
GoVersion: go1.25.10
Race Enabled: false
Check Table Before Drop: false
Store: tikv
```

Start by running the attached SQL reproduction and comparing the EXPLAIN and EXPLAIN ANALYZE plans for the original join, EXISTS rewrite, and NO_DECORRELATE diagnostic. Trace the planner code that handles DISTINCT, joins, semijoins, and Apply, then add regression coverage for the described duplicate-only join and no-match semantics. Done means an equivalent plan avoids materializing every matching pair without changing results.

Written by the indexing model from the issue text.