KurtYoung commented on a change in pull request #8107: [FLINK-12094][table-runtime-blink] Introduce sort merge join operator to blink batch URL: https://github.com/apache/flink/pull/8107#discussion_r272038683
########## File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/SortMergeJoinOperator.java ########## @@ -0,0 +1,350 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.runtime.join; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.memory.MemorySegment; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.runtime.io.disk.iomanager.IOManager; +import org.apache.flink.runtime.memory.MemoryAllocationException; +import org.apache.flink.runtime.memory.MemoryManager; +import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.table.dataformat.BaseRow; +import org.apache.flink.table.dataformat.BinaryRow; +import org.apache.flink.table.dataformat.GenericRow; +import org.apache.flink.table.dataformat.JoinedRow; +import org.apache.flink.table.generated.GeneratedJoinCondition; +import org.apache.flink.table.generated.GeneratedNormalizedKeyComputer; +import org.apache.flink.table.generated.GeneratedProjection; +import org.apache.flink.table.generated.GeneratedRecordComparator; +import org.apache.flink.table.generated.JoinCondition; +import org.apache.flink.table.generated.Projection; +import org.apache.flink.table.generated.RecordComparator; +import org.apache.flink.table.runtime.TableStreamOperator; +import org.apache.flink.table.runtime.sort.BinaryExternalSorter; +import org.apache.flink.table.runtime.util.ResettableExternalBuffer; +import org.apache.flink.table.runtime.util.StreamRecordCollector; +import org.apache.flink.table.typeutils.AbstractRowSerializer; +import org.apache.flink.table.typeutils.BinaryRowSerializer; +import org.apache.flink.util.Collector; +import org.apache.flink.util.MutableObjectIterator; + +import java.util.List; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * An implementation that realizes the joining through a sort-merge join strategy. + * 1.In most cases, its performance is weaker than HashJoin. + * 2.It is more stable than HashJoin, and most of the data can be sorted stably. + * 3.SortMergeJoin should be the best choice if sort can be omitted in the case of multi-level join + * cascade with the same key. + * + * <p>NOTE: SEMI and ANTI join output input1 instead of input2. (Contrary to {@link HashJoinOperator}). + */ +public class SortMergeJoinOperator extends TableStreamOperator<BaseRow> + implements TwoInputStreamOperator<BaseRow, BaseRow, BaseRow> { + + private final long reservedSortMemory1; + private final long reservedSortMemory2; + private final long externalBufferMemory; + private final SortMergeJoinType type; + private final boolean leftIsSmaller; + private final boolean[] filterNulls; + + // generated code to cook + private GeneratedJoinCondition condFuncCode; + private GeneratedProjection projectionCode1; + private GeneratedProjection projectionCode2; + private GeneratedNormalizedKeyComputer computer1; + private GeneratedRecordComparator comparator1; + private GeneratedNormalizedKeyComputer computer2; + private GeneratedRecordComparator comparator2; + private GeneratedRecordComparator genKeyComparator; + + private transient MemoryManager memManager; + private transient IOManager ioManager; + private transient BinaryRowSerializer serializer1; + private transient BinaryRowSerializer serializer2; + private transient BinaryExternalSorter sorter1; + private transient BinaryExternalSorter sorter2; + private transient SortMergeJoinIterator joinIterator1; Review comment: you cannot list all variables you *might* used, and choose which one to use in different scenarios... ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
