planner: add newly created col for window projection (#52378)

[ti-chi-bot]

This is an automated cherry-pick of #52378

What problem does this PR solve?

Issue Number: close #42734

Problem Summary:

We will get a panic error during we execute the following SQL:

use test
CREATE TABLE temperature_data (temperature double);
CREATE TABLE humidity_data (humidity double);
CREATE TABLE weather_report (report_id double, report_date varchar(100));
INSERT INTO temperature_data VALUES (1.0);
INSERT INTO humidity_data VALUES (0.5);
INSERT INTO weather_report VALUES (2.0, 'test')
SELECT EXISTS
  (SELECT FIRST_VALUE(temp_data.temperature) OVER weather_window AS first_temperature,
                                                  MIN(report_data.report_id) OVER weather_window AS min_report_id
   FROM temperature_data AS temp_data WINDOW weather_window AS (PARTITION BY EXISTS
                                                                  (SELECT report_data.report_date AS report_date
                                                                   FROM humidity_data AS humidity_data
                                                                   WHERE temp_data.temperature >= humidity_data.humidity ))) AS is_exist
FROM weather_report AS report_data;

The problem happened in the TryToGetChildProp of the LogicalProjection plan.

// TryToGetChildProp will check if this sort property can be pushed or not.
// When a sort column will be replaced by scalar function, we refuse it.
// When a sort column will be replaced by a constant, we just remove it.
func (p *LogicalProjection) TryToGetChildProp(prop *property.PhysicalProperty) (*property.PhysicalProperty, bool) {
	newProp := prop.CloneEssentialFields()
	newCols := make([]property.SortItem, 0, len(prop.SortItems))
	for _, col := range prop.SortItems {
		idx := p.schema.ColumnIndex(col.Col)
+		switch expr := p.Exprs[idx].(type) {
		case *expression.Column:
			newCols = append(newCols, property.SortItem{Col: expr, Desc: col.Desc})
		case *expression.ScalarFunction:
			return nil, false
		}
	}
	newProp.SortItems = newCols
	return newProp, true
}

We cannot find the sort item from the projection's schema.

After I debugged it, I found that we will try to find the Coulmn#14 in the projection's schema. But we don't have it.

What changed and how does it work?

To understand this problem we need to take a look at the query plan after we fixed it :(

+-------------------------------------------+----------+-----------+---------------------+-------------------------------------------------------------------------------------------------------------------+
| id                                        | estRows  | task      | access object       | operator info                                                                                                     |
+-------------------------------------------+----------+-----------+---------------------+-------------------------------------------------------------------------------------------------------------------+
| Projection_15                             | 10000.00 | root      |                     | Column#20                                                                                                         |
| └─Apply_17                                | 10000.00 | root      |                     | CARTESIAN left outer semi join                                                                                    |
|   ├─TableReader_19(Build)                 | 10000.00 | root      |                     | data:TableFullScan_18                                                                                             |
|   │ └─TableFullScan_18                    | 10000.00 | cop[tikv] | table:report_data   | keep order:false, stats:pseudo                                                                                    |
|   └─Shuffle_29(Probe)                     | 10000.00 | root      |                     | execution info: concurrency:2, data sources:[Projection_22]                                                       |
|     └─Window_20                           | 10000.00 | root      |                     | first_value(test.temperature_data.temperature)->Column#16, min(Column#15)->Column#17 over(partition by Column#14) |
|       └─Sort_28                           | 10000.00 | root      |                     | Column#14                                                                                                         |
|         └─ShuffleReceiver_30              | 1.00     | root      |                     |                                                                                                                   |
|           └─Projection_22                 | 10000.00 | root      |                     | test.temperature_data.temperature, Column#14, test.weather_report.report_id->Column#15                            |
|             └─HashJoin_23                 | 10000.00 | root      |                     | CARTESIAN left outer semi join, other cond:ge(test.temperature_data.temperature, test.humidity_data.humidity)     |
|               ├─TableReader_27(Build)     | 10000.00 | root      |                     | data:TableFullScan_26                                                                                             |
|               │ └─TableFullScan_26        | 10000.00 | cop[tikv] | table:humidity_data | keep order:false, stats:pseudo                                                                                    |
|               └─TableReader_25(Probe)     | 10000.00 | root      |                     | data:TableFullScan_24                                                                                             |
|                 └─TableFullScan_24        | 10000.00 | cop[tikv] | table:temp_data     | keep order:false, stats:pseudo                                                                                    |
+-------------------------------------------+----------+-----------+---------------------+-------------------------------------------------------------------------------------------------------------------+
14 rows in set (0.00 sec)

As you can see the window is partitioned by Column#14 and it comes from the Projection_22.

Column#14 evaluates from the exist-subquery:

PARTITION BY EXISTS (
           SELECT
             report_data.report_date AS report_date
           FROM
             humidity_data AS humidity_data
           WHERE temp_data.temperature >= humidity_data.humidity
)

When we built this subquery we found it is a correlated query because we used temp_data.temperature as the predicate.

if b.disableSubQueryPreprocessing || len(ExtractCorrelatedCols4LogicalPlan(np)) > 0 || hasCTEConsumerInSubPlan(np) {
		planCtx.plan, er.err = b.buildSemiApply(planCtx.plan, np, nil, er.asScalar, v.Not, semiJoinRewrite, noDecorrelate)
		if er.err != nil || !er.asScalar {
			return v, true
		}
		er.ctxStackAppend(planCtx.plan.Schema().Columns[planCtx.plan.Schema().Len()-1], planCtx.plan.OutputNames()[planCtx.plan.Schema().Len()-1])

Then the problem came out from the buildByItemsForWindow, because we used the column from the semi-apply plan as our sort item during the expression rewrite then we forget to add this column into the projection's schema:

	for _, item := range items {
		newExpr, _ := item.Expr.Accept(transformer)
		item.Expr = newExpr.(ast.ExprNode)
		it, np, err := b.rewrite(ctx, item.Expr, p, aggMap, true)
		if err != nil {
			return nil, nil, err
		}
		p = np
		if it.GetType().GetType() == mysql.TypeNull {
			continue
		}
		if col, ok := it.(*expression.Column); ok {
+.          // This column comes from the semi-apply
+			retItems = append(retItems, property.SortItem{Col: col, Desc: item.Desc})
+           continue
        }

So the fix is that we need to append this col to the top-level projection schema:

	for _, item := range items {
		newExpr, _ := item.Expr.Accept(transformer)
		item.Expr = newExpr.(ast.ExprNode)
		it, np, err := b.rewrite(ctx, item.Expr, p, aggMap, true)
		if err != nil {
			return nil, nil, err
		}
		p = np
		if it.GetType().GetType() == mysql.TypeNull {
			continue
		}
		if col, ok := it.(*expression.Column); ok {
			retItems = append(retItems, property.SortItem{Col: col, Desc: item.Desc})
+			// If the column is already in the schema, we don't need to add it again.
+			if !proj.schema.Contains(col) {
+				proj.Exprs = append(proj.Exprs, col)
+				proj.schema.Append(col)
+				proj.names = append(proj.names, types.EmptyName)
+			}
			continue
		}

And also we need to avoid adding the same column twice, for example:

   SELECT
     EXISTS (
       SELECT
         FIRST_VALUE(temp_data.temperature) OVER weather_window AS first_temperature,
         MIN(report_data.report_id) OVER weather_window AS min_report_id
       FROM
         temperature_data AS temp_data
       WINDOW weather_window AS (
         PARTITION BY temp_data.temperature 
       )
     ) AS is_exist
   FROM
     weather_report AS report_data;

As you can see we partition the window by itself then we already have it in the projection's
schema. So we don't need to add it again.
The query plan looks like this

+----------------------------------+---------+-----------+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------+
| id                               | estRows | task      | access object     | operator info                                                                                                                             |
+----------------------------------+---------+-----------+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------+
| Projection_11                    | 1.00    | root      |                   | Column#16                                                                                                                                 |
| └─Apply_13                       | 1.00    | root      |                   | CARTESIAN left outer semi join                                                                                                            |
|   ├─TableReader_15(Build)        | 1.00    | root      |                   | data:TableFullScan_14                                                                                                                     |
|   │ └─TableFullScan_14           | 1.00    | cop[tikv] | table:report_data | keep order:false, stats:pseudo                                                                                                            |
|   └─Window_16(Probe)             | 1.00    | root      |                   | first_value(test.temperature_data.temperature)->Column#12, min(Column#11)->Column#13 over(partition by test.temperature_data.temperature) |
|     └─Sort_21                    | 1.00    | root      |                   | test.temperature_data.temperature                                                                                                         |
|       └─Projection_18            | 1.00    | root      |                   | test.temperature_data.temperature, test.weather_report.report_id->Column#11                                                               |
|         └─TableReader_20         | 1.00    | root      |                   | data:TableFullScan_19                                                                                                                     |
|           └─TableFullScan_19     | 1.00    | cop[tikv] | table:temp_data   | keep order:false, stats:pseudo                                                                                                            |
+----------------------------------+---------+-----------+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------+

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)
• No need to test

  • I checked and no code files have been changed.

Side effects

• Performance regression: Consumes more CPU
• Performance regression: Consumes more Memory
• Breaking backward compatibility

Documentation

• Affects user behaviors
• Contains syntax changes
• Contains variable changes
• Contains experimental features
• Changes MySQL compatibility

Release note

Please refer to Release Notes Language Style Guide to write a quality release note.

None

[ti-chi-bot]

PR needs rebase.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: hawkingrei, hi-rustin

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• OWNERS [hawkingrei,hi-rustin]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[ti-chi-bot]

[LGTM Timeline notifier]

Timeline:

• 2024-04-16 09:33:40.27750767 +0000 UTC m=+330639.388554119: ☑️ agreed by hi-rustin.
• 2024-04-16 09:38:32.724424421 +0000 UTC m=+330931.835470870: ☑️ agreed by hawkingrei.

[codecov]

Codecov Report

❗ No coverage uploaded for pull request base (release-7.1@fd9583b). Click here to learn what that means.
The diff coverage is 100.0000%.

Additional details and impacted files

@@               Coverage Diff                @@
##             release-7.1     #52489   +/-   ##
================================================
  Coverage               ?   73.4204%           
================================================
  Files                  ?       1211           
  Lines                  ?     379890           
  Branches               ?          0           
================================================
  Hits                   ?     278917           
  Misses                 ?      83143           
  Partials               ?      17830