[FLINK-9344] [table] Support INTERSECT and INTERSECT ALL for streaming #5998
Conversation
Xpray
force-pushed
the
FLINK-9344
branch
7 times, most recently
from
b93360a
to
d007ddb
Compare
There was a problem hiding this comment.
Thanks @Xpray for the contribution. It looks pretty good! I just left a few comment and questions.
I guess I am confused by the JIRA ticket description as it doesn't specify whether you are supporting unbounded intersect or windowed intersect, or both.
I guess a brief description would be very helpful here, for other reviewers as well.
--
Rong
| } | ||
|
|
||
| override def toString: String = { | ||
| s"Intersect$intersectType(intersect$intersectType: ($intersectSelectionToString))" |
There was a problem hiding this comment.
s"Intersect$intersectType(intersect: ($intersectSelectionToString))"
I dont think you need to duplicate the type twice
| with DataStreamRel { | ||
|
|
||
| private lazy val intersectType = if (all) { | ||
| "All" |
There was a problem hiding this comment.
" All" might be better formatting since you only attached this to the explainTerm and toString method
| import org.apache.flink.types.Row | ||
| import org.apache.flink.util.Collector | ||
|
|
||
| class StreamIntersectCoProcessFunction( |
| resultType: TypeInformation[Row], | ||
| queryConfig: StreamQueryConfig, | ||
| all: Boolean) | ||
| extends CoProcessFunction[CRow, CRow, CRow] |
There was a problem hiding this comment.
I guess I am confused here:
There's CoGroupedStream with customized CoGroupFunction which is already supported in DataStream API. seems like if we operate on a windowed stream, we can apply the intersect as a CoGroupFunction. Is this function solely targeting the non-windowed intersect case. If so, can we rename the function (also adds to my point: please add Java Doc).
There was a problem hiding this comment.
I think it makes sense to have two implementations of this operator.
- For tables with a time attribute. This implementation works without retraction and can automatically cleanup the state.
- For tables without time attributes. This implementation needs to cleanup state based on retention time and produces retractions.
This PR seems to address both cases, which is fine for now. We can improve for 1. later on. Both cases should be implemented as CoProcessFunction. We should try to be independent of the DataStream window operators, IMO.
There was a problem hiding this comment.
Thanks for the review @walterddr and @fhueske , This PR intends to support NonWindow intersect just like NonWindow innerJoin.
| } | ||
| } | ||
|
|
||
| private def expireOutTimeRow( |
There was a problem hiding this comment.
I don't believe any of your test triggers this code path.
There was a problem hiding this comment.
Consider overriding the queryConfig for triggering this perhaps
|
|
||
| validateEqualsHashCode("intersect", resultType) | ||
|
|
||
| // state to hold left stream element |
There was a problem hiding this comment.
The description is misleading, you are not actually holding the "row" of stream element if I understand correctly.
Xpray
force-pushed
the
FLINK-9344
branch
4 times, most recently
from
bb72574
to
492f7b6
Compare
|
Thanks for the review @walterddr @fhueske , I've updated the PR. |
|
@fhueske , I would like to support minus/minus All after this issue, would you give some suggestion about this issue? |
twalthr
changed the title
[FLINK-9344] [TableAPI & SQL] Support INTERSECT and INTERSECT ALL for streaming
What is the purpose of the change
Support Intersect and Intersect All for Streaming SQL and TableAPI
Brief change log
Verifying this change
This change added tests and can be verified as follows:
cases of intersect operations in both
org.apache.flink.table.runtime.stream.sql.SetOperatorsITCaseandorg.apache.flink.table.runtime.stream.table.SetOperatorsITCaseDoes this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation