walterddr commented on code in PR #9907:
URL: https://github.com/apache/pinot/pull/9907#discussion_r1048717286
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/HashJoinOperator.java:
##########
@@ -152,39 +202,82 @@ private TransferableBlock
buildJoinedDataBlock(TransferableBlock leftBlock)
if (leftBlock.isErrorBlock()) {
_upstreamErrorBlock = leftBlock;
return _upstreamErrorBlock;
- } else if (TransferableBlockUtils.isNoOpBlock(leftBlock) ||
TransferableBlockUtils.isEndOfStream(leftBlock)) {
+ }
+ if (leftBlock.isNoOpBlock() || (leftBlock.isSuccessfulEndOfStreamBlock()
&& !needUnmatchedRows())) {
return leftBlock;
}
+ if (leftBlock.isSuccessfulEndOfStreamBlock()) {
+ // Return remaining non-matched rows for non-inner join.
+ List<Object[]> returnRows = new ArrayList<>();
+ if (needUnmatchedRightRows()) {
+ for (Map.Entry<Key, List<Object[]>> entry :
_broadcastRightTable.entrySet()) {
+ Set<Integer> matchedIdx =
_matchedRightRows.getOrDefault(entry.getKey(), new HashSet<>());
+ int rowSize = entry.getValue().size();
+ Set<Integer> allIdx = IntStream.range(0,
rowSize).boxed().collect(Collectors.toSet());
+ Set<Integer> unmatchedIdx = Sets.difference(allIdx, matchedIdx);
+ for (int unmatchedRow : unmatchedIdx) {
+ Object[] resultRow = joinRow(null,
entry.getValue().get(unmatchedRow));
+ returnRows.add(resultRow);
+ }
Review Comment:
can we avoid using stream? this is on the critical code path (for every
hashkey) so it is better to do a single loop
```suggestion
Set<Integer> matchedIdx =
_matchedRightRows.getOrDefault(entry.getKey(), new HashSet<>());
List<Object[]> rightRows = entry.getValues();
for (int i = 0; i < rightRows.size(); i++) {
if (!matchedIdx.contains(i)) {
Object[] resultRow = joinRow(null, rightRows.get(i));
returnRows.add(resultRow);
}
}
```
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/HashJoinOperator.java:
##########
@@ -43,50 +49,94 @@
/**
* This basic {@code BroadcastJoinOperator} implement a basic broadcast join
algorithm.
+ * This algorithm assumes that the broadcast table has to fit in memory since
we are not supporting any spilling.
*
+ * For left join, inner join, right join and full join,
* <p>It takes the right table as the broadcast side and materialize a hash
table. Then for each of the left table row,
* it looks up for the corresponding row(s) from the hash table and create a
joint row.
*
* <p>For each of the data block received from the left table, it will
generate a joint data block.
- *
- * We currently support left join, inner join and semi join.
+ * We currently support left join, inner join, right join and full join.
* The output is in the format of [left_row, right_row]
*/
// TODO: Move inequi out of hashjoin.
(https://github.com/apache/pinot/issues/9728)
public class HashJoinOperator extends BaseOperator<TransferableBlock> {
private static final String EXPLAIN_NAME = "HASH_JOIN";
- private static final Set<JoinRelType> SUPPORTED_JOIN_TYPES =
ImmutableSet.of(JoinRelType.INNER, JoinRelType.LEFT);
- private final HashMap<Key, List<Object[]>> _broadcastHashTable;
+ private static final Set<JoinRelType> SUPPORTED_JOIN_TYPES =
+ ImmutableSet.of(JoinRelType.INNER, JoinRelType.LEFT, JoinRelType.RIGHT,
JoinRelType.FULL);
+
+ private final HashMap<Key, List<Object[]>> _broadcastRightTable;
+
+ // Used to track matched right rows.
+ // Only used for right join and full join to output non-matched right rows.
+ private final HashMap<Key, HashSet<Integer>> _matchedRightRows;
+
+ // Used to track unmatched left rows.
+ // Only used for left join and full join to output non-matched left rows.
+ private final HashMap<Key, List<Object[]>> _unmatchedLeftRows;
+
private final Operator<TransferableBlock> _leftTableOperator;
private final Operator<TransferableBlock> _rightTableOperator;
private final JoinRelType _joinType;
private final DataSchema _resultSchema;
+ private final int _leftRowSize;
private final int _resultRowSize;
private final List<TransformOperand> _joinClauseEvaluators;
private boolean _isHashTableBuilt;
+
+ // Used by non-inner join.
+ // Needed to indicate we have finished processing all results after
returning last block.
+ // TODO: Remove this special handling by fixing data block EOS abstraction
or operator's invariant.
+ private boolean _isTerminated;
private TransferableBlock _upstreamErrorBlock;
private KeySelector<Object[], Object[]> _leftKeySelector;
private KeySelector<Object[], Object[]> _rightKeySelector;
+ private boolean needUnmatchedRightRows() {
+ return _joinType == JoinRelType.RIGHT || _joinType == JoinRelType.FULL;
+ }
+
+ private boolean needUnmatchedLeftRows() {
+ return _joinType == JoinRelType.LEFT || _joinType == JoinRelType.FULL;
+ }
+
+ private boolean needUnmatchedRows() {
+ return _joinType != JoinRelType.INNER;
+ }
Review Comment:
nit: private util methods are usually put after the public ones in pinot. so
please move these to the end of file
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/HashJoinOperator.java:
##########
@@ -152,39 +202,82 @@ private TransferableBlock
buildJoinedDataBlock(TransferableBlock leftBlock)
if (leftBlock.isErrorBlock()) {
_upstreamErrorBlock = leftBlock;
return _upstreamErrorBlock;
- } else if (TransferableBlockUtils.isNoOpBlock(leftBlock) ||
TransferableBlockUtils.isEndOfStream(leftBlock)) {
+ }
+ if (leftBlock.isNoOpBlock() || (leftBlock.isSuccessfulEndOfStreamBlock()
&& !needUnmatchedRows())) {
return leftBlock;
}
+ if (leftBlock.isSuccessfulEndOfStreamBlock()) {
Review Comment:
nit: suggest putting the logic of the end of stream handling into a separate
function for better readability.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/HashJoinOperator.java:
##########
@@ -152,39 +202,82 @@ private TransferableBlock
buildJoinedDataBlock(TransferableBlock leftBlock)
if (leftBlock.isErrorBlock()) {
_upstreamErrorBlock = leftBlock;
return _upstreamErrorBlock;
- } else if (TransferableBlockUtils.isNoOpBlock(leftBlock) ||
TransferableBlockUtils.isEndOfStream(leftBlock)) {
+ }
+ if (leftBlock.isNoOpBlock() || (leftBlock.isSuccessfulEndOfStreamBlock()
&& !needUnmatchedRows())) {
return leftBlock;
}
+ if (leftBlock.isSuccessfulEndOfStreamBlock()) {
+ // Return remaining non-matched rows for non-inner join.
+ List<Object[]> returnRows = new ArrayList<>();
+ if (needUnmatchedRightRows()) {
+ for (Map.Entry<Key, List<Object[]>> entry :
_broadcastRightTable.entrySet()) {
+ Set<Integer> matchedIdx =
_matchedRightRows.getOrDefault(entry.getKey(), new HashSet<>());
+ int rowSize = entry.getValue().size();
+ Set<Integer> allIdx = IntStream.range(0,
rowSize).boxed().collect(Collectors.toSet());
+ Set<Integer> unmatchedIdx = Sets.difference(allIdx, matchedIdx);
+ for (int unmatchedRow : unmatchedIdx) {
+ Object[] resultRow = joinRow(null,
entry.getValue().get(unmatchedRow));
+ returnRows.add(resultRow);
+ }
+ }
+ }
+ if (needUnmatchedLeftRows()) {
+ for (Map.Entry<Key, List<Object[]>> entry :
_unmatchedLeftRows.entrySet()) {
+ for (Object[] unmatchedLeftRow : entry.getValue()) {
+ Object[] resultRow = joinRow(unmatchedLeftRow, null);
+ returnRows.add(resultRow);
+ }
+ }
+ }
+ _isTerminated = true;
+ return new TransferableBlock(returnRows, _resultSchema,
DataBlock.Type.ROW);
+ }
List<Object[]> rows = new ArrayList<>();
List<Object[]> container = leftBlock.isEndOfStreamBlock() ? new
ArrayList<>() : leftBlock.getContainer();
for (Object[] leftRow : container) {
+ Key key = new Key(_leftKeySelector.getKey(leftRow));
// NOTE: Empty key selector will always give same hash code.
- List<Object[]> hashCollection =
- _broadcastHashTable.getOrDefault(new
Key(_leftKeySelector.getKey(leftRow)), Collections.emptyList());
- // If it is a left join and right table is empty, we return left rows.
- if (hashCollection.isEmpty() && _joinType == JoinRelType.LEFT) {
- rows.add(joinRow(leftRow, null));
- } else {
- // If it is other type of join.
- for (Object[] rightRow : hashCollection) {
- // TODO: Optimize this to avoid unnecessary object copy.
- Object[] resultRow = joinRow(leftRow, rightRow);
- if (_joinClauseEvaluators.isEmpty() ||
_joinClauseEvaluators.stream().allMatch(evaluator ->
- (Boolean)
FunctionInvokeUtils.convert(evaluator.apply(resultRow),
DataSchema.ColumnDataType.BOOLEAN))) {
- rows.add(resultRow);
+ List<Object[]> matchedRightRows = _broadcastRightTable.getOrDefault(key,
null);
+ if (matchedRightRows == null) {
+ if (needUnmatchedLeftRows()) {
+ rows.add(joinRow(leftRow, null));
+ }
Review Comment:
why do we do this here. and then again in line 224-228?
--
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]
