rohityadav1993 commented on code in PR #19122: URL: https://github.com/apache/pinot/pull/19122#discussion_r3860754702
########## pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMergeJoinOperator.java: ########## @@ -0,0 +1,609 @@ +/** + * 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.pinot.query.runtime.operator; + +import com.google.common.base.Preconditions; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nullable; +import org.apache.calcite.rel.core.JoinRelType; +import org.apache.pinot.calcite.rel.hint.PinotHintOptions; +import org.apache.pinot.calcite.rel.hint.PinotHintOptions.JoinHintOptions; +import org.apache.pinot.common.datatable.StatMap; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.common.utils.config.QueryOptionsUtils; +import org.apache.pinot.query.planner.logical.RexExpression; +import org.apache.pinot.query.planner.plannode.JoinNode; +import org.apache.pinot.query.planner.plannode.PlanNode; +import org.apache.pinot.query.runtime.blocks.MseBlock; +import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock; +import org.apache.pinot.query.runtime.blocks.SuccessMseBlock; +import org.apache.pinot.query.runtime.operator.join.JoinedRowView; +import org.apache.pinot.query.runtime.operator.operands.TransformOperand; +import org.apache.pinot.query.runtime.operator.operands.TransformOperandFactory; +import org.apache.pinot.query.runtime.plan.OpChainExecutionContext; +import org.apache.pinot.spi.exception.QueryErrorCode; +import org.apache.pinot.spi.utils.BooleanUtils; +import org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey; +import org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner.JoinOverFlowMode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// The `SortedMergeJoinOperator` implements a streaming sorted merge join. +/// +/// Unlike [HashJoinOperator], it does not materialize the right side into an in-memory hash table. Instead it +/// assumes both the left and right inputs are already sorted in ascending order on their respective join keys and +/// advances two cursors in lock-step (a two-pointer merge). Only one block per side is held in memory at a time, plus a +/// small buffer for the run of right rows that share the current join key (needed to support one-to-many and +/// many-to-many matches). +/// +/// This makes memory usage proportional to the largest single-key run on the right side rather than the entire right +/// input, which is the key advantage for pre-sorted, pre-partitioned data layouts. +/// +/// Preconditions (enforced by the planner, assumed here): +/// - Both inputs are sorted ascending on the join keys (left keys for the left input, right keys for the right +/// input). +/// - The join is an equi-join (non-empty join keys). Non-equi conditions are still applied as residual filters. +/// +/// Rows whose join key contains a `null` value never match (per SQL semantics) and are skipped; for LEFT joins +/// such left rows are emitted with `null` padding on the right. +/// +/// Only INNER and LEFT joins are currently supported. +public class SortedMergeJoinOperator extends MultiStageOperator { + private static final Logger LOGGER = LoggerFactory.getLogger(SortedMergeJoinOperator.class); + private static final String EXPLAIN_NAME = "SORTED_MERGE_JOIN"; + private static final String MERGE_LOOP_SCOPE = "SortedMergeJoinOperator#mergeLoop"; + private static final String EMIT_MATCHED_KEY_SCOPE = "SortedMergeJoinOperator#emitMatchedKey"; + private static final String BUFFER_RIGHT_RUN_SCOPE = "SortedMergeJoinOperator#bufferRightRun"; + private static final Set<JoinRelType> SUPPORTED_JOIN_TYPES = Set.of(JoinRelType.INNER, JoinRelType.LEFT); + // Target number of output rows per emitted block. A single equi-key run may overshoot this; that is acceptable. + private static final int TARGET_BLOCK_SIZE_ROWS = 1024; + protected static final int DEFAULT_MAX_ROWS_IN_JOIN = 1024 * 1024; // 2^20, around 1MM rows + protected static final JoinOverFlowMode DEFAULT_JOIN_OVERFLOW_MODE = JoinOverFlowMode.THROW; + + private final Cursor _leftCursor; + private final Cursor _rightCursor; + private final JoinRelType _joinType; + private final boolean _needUnmatchedLeftRows; + private final int[] _leftKeyIds; + private final int[] _rightKeyIds; + // Stored type of each join key column, resolved once so key comparison dispatches on a concrete type instead of a + // raw Comparable (avoids cross-type ClassCastExceptions and boxing-heavy generic compares on the hot path). + private final ColumnDataType[] _keyStoredTypes; + private final DataSchema _resultSchema; + private final int _leftColumnSize; + private final int _resultColumnSize; + private final List<TransformOperand> _nonEquiEvaluators; + private final boolean _hasNonEquiConditions; + private final int _maxRowsInJoin; Review Comment: Self review: Can be omitted as it is not a limiting factor now in joins due to now right table materialisation -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
