class RightJoin<Row> extends MergeJoinNode<Row> {
        /** Right row factory. */
        private final RowHandler.RowFactory<Row> leftRowFactory;

        /** */
        private Row left;

        /** */
        private Row right;

        /** Used to store similar rows of rights stream in many-to-many join mode. */
        private List<Row> rightMaterialization;

        /** */
        private int rightIdx;

        /** */
        private boolean drainMaterialization;

        /** Whether current right row was matched (hence pushed to downstream) or not. */
        private boolean matched;

        /**
         * @param ctx Execution context.
         * @param rowType Row type.
         * @param comp Join expression comparator.
         * @param distributed If one of the inputs has exchange underneath.
         * @param leftRowFactory Left row factory.
         */
        public RightJoin(
            ExecutionContext<Row> ctx,
            RelDataType rowType,
            Comparator<Row> comp,
            boolean distributed,
            RowHandler.RowFactory<Row> leftRowFactory
        ) {
            super(ctx, rowType, comp, distributed);

            this.leftRowFactory = leftRowFactory;
        }

        /** {@inheritDoc} */
        @Override protected void rewindInternal() {
            left = null;
            right = null;
            rightIdx = 0;
            rightMaterialization = null;
            drainMaterialization = false;

            super.rewindInternal();
        }

        /** {@inheritDoc} */
        @Override protected void join() throws Exception {<FILL_FUNCTION_BODY>}
    }

            inLoop = true;
            try {
                while (requested > 0 && !(left == null && leftInBuf.isEmpty() && waitingLeft != NOT_WAITING)
                    && (right != null || !rightInBuf.isEmpty() || rightMaterialization != null)) {
                    checkState();

                    if (left == null && !leftInBuf.isEmpty())
                        left = leftInBuf.remove();

                    if (right == null) {
                        if (rightInBuf.isEmpty() && waitingRight != NOT_WAITING)
                            break;

                        if (!rightInBuf.isEmpty()) {
                            right = rightInBuf.remove();

                            matched = false;
                        }
                    }

                    if (right == null && rightMaterialization != null && !drainMaterialization) {
                        drainMaterialization = true;
                        left = null;

                        continue;
                    }

                    Row row;
                    if (!drainMaterialization) {
                        if (left == null) {
                            if (!matched) {
                                row = handler.concat(leftRowFactory.create(), right);

                                requested--;
                                downstream().push(row);
                            }

                            right = null;

                            continue;
                        }

                        int cmp = comp.compare(left, right);

                        if (cmp < 0) {
                            left = null;
                            rightIdx = 0;

                            if (rightMaterialization != null)
                                drainMaterialization = true;

                            continue;
                        }
                        else if (cmp > 0) {
                            if (!matched) {
                                row = handler.concat(leftRowFactory.create(), right);

                                requested--;
                                downstream().push(row);
                            }

                            right = null;
                            rightIdx = 0;
                            rightMaterialization = null;

                            continue;
                        }

                        if (rightMaterialization == null && (!rightInBuf.isEmpty() || waitingRight != NOT_WAITING)) {
                            if (rightInBuf.isEmpty())
                                break;

                            if (comp.compare(left, rightInBuf.peek()) == 0)
                                rightMaterialization = new ArrayList<>();
                        }

                        matched = true;

                        row = handler.concat(left, right);

                        if (rightMaterialization != null) {
                            rightMaterialization.add(right);

                            right = null;
                        }
                        else
                            left = null;
                    }
                    else {
                        if (left == null) {
                            if (waitingLeft == NOT_WAITING)
                                rightMaterialization = null;

                            break;
                        }

                        if (rightIdx >= rightMaterialization.size()) {
                            rightIdx = 0;
                            left = null;

                            continue;
                        }

                        Row right = rightMaterialization.get(rightIdx++);

                        int cmp = comp.compare(left, right);

                        if (cmp > 0) {
                            rightIdx = 0;
                            rightMaterialization = null;
                            drainMaterialization = false;

                            continue;
                        }

                        row = handler.concat(left, right);
                    }

                    requested--;
                    downstream().push(row);
                }
            }
            finally {
                inLoop = false;
            }

            if (waitingRight == 0)
                rightSource().request(waitingRight = IN_BUFFER_SIZE);

            if (waitingLeft == 0)
                leftSource().request(waitingLeft = IN_BUFFER_SIZE);

            if (requested > 0 && waitingRight == NOT_WAITING && right == null && rightInBuf.isEmpty() && rightMaterialization == null)
                checkJoinFinished();
        
<methods>public void close() ,public ExecutionContext<Row> context() ,public Downstream<Row> downstream() ,public void onError(java.lang.Throwable) ,public void onRegister(Downstream<Row>) ,public void register(List<Node<Row>>) ,public void rewind() ,public RelDataType rowType() ,public List<Node<Row>> sources() <variables>protected static final int IN_BUFFER_SIZE,protected static final int IO_BATCH_CNT,protected static final int IO_BATCH_SIZE,protected static final int MODIFY_BATCH_SIZE,private boolean closed,private ExecutionContext<Row> ctx,private Downstream<Row> downstream,private RelDataType rowType,private List<Node<Row>> sources,private volatile java.lang.Thread thread