Au-Miner opened a new pull request, #28246:
URL: https://github.com/apache/flink/pull/28246
## What is the purpose of the change
When a `Correlate` has a `Calc` on its right side that adds non-identity
projections (e.g. `LATERAL (SELECT s * 2 FROM UNNEST(...) AS T(s))`), the
physical conversion previously dropped those projections, leaving wrong output
columns. This PR preserves them by wrapping the resulting `Correlate` in a
`Calc` that re-applies the right-side projections.
## Brief change log
- `StreamPhysicalCorrelateRule` / `BatchPhysicalCorrelateRule`: split the
right-side calc into `(scan, projects, condition)`; when projects are
non-identity, build an inner `*PhysicalCorrelate` with the natural join row
type and wrap it in a `*PhysicalCalc` whose program is built by the new
`buildOuterProgram` helper that shifts right-side projections past the left
field count.
- `FlinkCalcMergeRule`: add `BATCH_PHYSICAL_INSTANCE` so the new wrapping
calc can merge with adjacent batch calcs (mirrors the existing
`STREAM_PHYSICAL_INSTANCE`).
- `FlinkBatchRuleSets#PHYSICAL_OPT_RULES`: register
`FlinkCalcMergeRule.BATCH_PHYSICAL_INSTANCE`.
## Verifying this change
Added `testLateralProjectionFromUnnest` in `UnnestTestBase` (exercised by
both stream and batch `UnnestTest`) which asserts that the optimized plan
contains a `Calc(select=[a, *(b0, 2) AS doubled])` over the `Correlate`. Added
`testCorrelateWithRightSideProjection` in `CorrelateITCase` which runs `LATERAL
(SELECT CONCAT(v, '_x') FROM TABLE(STRING_SPLIT(c, '|')) AS T(v))` and asserts
the projected values appear in the sink. Updated `UnnestTest.xml` plans (stream
+ batch) accordingly.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? no
--
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]
