[FLINK-37916][table] Add StreamPhysicalMultiJoin

[gustavodemorais]

What is the purpose of the change

Adds StreamPhysicalMultiJoin that creates a StreamExecMultiJoin

Brief change log

• Adds StreamPhysicalMultiJoin implementing StreamPhysicalRel
• Add necessary private methods to instantiate params

Verifying this change

This change will be only adds StreamExecMultiJoin. Tests will be added as soon as we can use it/test it.

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? (not applicable)

[flinkbot]

CI report:

• 6028dbb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[gustavodemorais]

Rebased

[snuyanzin]

not sure people who will review FLINK-37936 will be able to detect these TODOs...
should they be explicitly mentioned in JIRA?

[gustavodemorais]

Yep, wanted to do it and forgot. Updated the Jira!

[snuyanzin]

2 quetsions

1. are we sure that ordinalInParent is always inside boundaries?
   2.how likely it could be that value at that index is same as p?

[gustavodemorais]

1. I expect so, I did the same we have for PTFs

   flink/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java

   Line 137 in 86ee4b5

   final List<RelNode> newInputs = new ArrayList<>(inputs);

2. I don't expect it's likely, since we're replacing it

[gustavodemorais]

Took a look at it again and

1. I've added a assertion check since I saw some other places do it
2. As you've mentioned we're not sure what to do if they call with the same p, so I think it's safer to let the caller control this

I think we should be ok either way. But I think adding the first assertion should be safe!

[snuyanzin]

IIRC usually there should be indexes instead of names
otherwise there might be memory issues if there are too many fields with long names
also I'm not sure the output would be clear for cases where there are several nested subqueries and each time fields have a new alias (for instance set manually in sql)

[gustavodemorais]

You raise good points. I've also here followed what we did for PTFs

flink/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalProcessTableFunction.java

Line 190 in 86ee4b5

.item("select", String.join(",", getRowType().getFieldNames()))

Wdyt @twalthr?

[twalthr]

Looking at CommonPhysicalJoin we are also using names. Let's fix this problem when it occurs again and keep the PR as it is for now?

[snuyanzin]

fair enough

[twalthr]

Mostly minor stuff. Should be good from my side in the next iteration.

[twalthr]

Looking at CommonPhysicalJoin we are also using names. Let's fix this problem when it occurs again and keep the PR as it is for now?

[twalthr]

Move this line out of the loop / stream.

[gustavodemorais]

I was actually doing input.getCluster().getMetadataQuery() but this might not be necessary?

[gustavodemorais]

I'm intuitively using the input

final FlinkRelMetadataQuery fmq =
                FlinkRelMetadataQuery.reuseOrCreate(input.getCluster().getMetadataQuery());
        return fmq.getUpsertKeys(input);

And that's why it's inside

[gustavodemorais]

I've moved it to an own method so the code and kept as it is for now. Similar to what we do here