gustavodemorais commented on code in PR #26313: URL: https://github.com/apache/flink/pull/26313#discussion_r2039620658
########## flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java: ########## @@ -0,0 +1,474 @@ +package org.apache.flink.table.runtime.operators.join.stream; + +// TODO Gustavo Confirm we should create a private custom enum for join types instead of using +// Calcite's JoinRelType +import org.apache.flink.api.common.state.ValueState; +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.streaming.api.operators.AbstractInput; +import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2; +import org.apache.flink.streaming.api.operators.Input; +import org.apache.flink.streaming.api.operators.MultipleInputStreamOperator; +import org.apache.flink.streaming.api.operators.StreamOperatorParameters; +import org.apache.flink.streaming.api.operators.TimestampedCollector; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.utils.JoinedRowData; +import org.apache.flink.table.runtime.generated.MultiJoinCondition; +import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStateView; +import org.apache.flink.table.runtime.operators.join.stream.state.JoinRecordStateViews; +import org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.types.RowKind; + +import org.apache.calcite.rel.core.JoinRelType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +/** + * Streaming multi-way join operator which supports inner join and left/right/full outer join. It + * eliminates the intermediate state necessary for a chain of multiple binary joins. In other words, Review Comment: Yeah, I'll refactor so we use an internal JoinType with the supported joins. For this initial implementation of the multi-way join operator, I deliberately focused on INNER and LEFT joins, as outlined in the FLIP's "Limitations and Scope" section. The goal is to have a first multi join operator supporting these two types of joins, which are also the most used ones. Semi and anti joins are out of the scope of the FLIP and would first require additional changes to the optimizer since they're not supported in the FlinkJoinToMultiJoinRule. https://github.com/apache/flink/blob/6878fced84e86f7fa14358d04b3e5b4e0b5e3042/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkJoinToMultiJoinRule.java#L91 -- 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: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
