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 was to first address the most common use cases we've seen causing large state issues and try to somehow limit the scope -- 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
