class NestedLoopOperation implements CompletionListenable {

    private final CapturingRowConsumer leftConsumer;
    private final CapturingRowConsumer rightConsumer;
    private final RowConsumer resultConsumer;

    public NestedLoopOperation(int numLeftCols,
                               int numRightCols,
                               RowConsumer nlResultConsumer,
                               Predicate<Row> joinPredicate,
                               JoinType joinType,
                               CircuitBreaker circuitBreaker,
                               RamAccounting ramAccounting,
                               List<DataType<?>> leftSideColumnTypes,
                               long estimatedRowsSizeLeft,
                               long estimatedNumberOfRowsLeft,
                               boolean blockNestedLoop) {
        this.resultConsumer = nlResultConsumer;
        this.leftConsumer = new CapturingRowConsumer(nlResultConsumer.requiresScroll(), nlResultConsumer.completionFuture());
        this.rightConsumer = new CapturingRowConsumer(true, nlResultConsumer.completionFuture());
        CompletableFuture.allOf(leftConsumer.capturedBatchIterator(), rightConsumer.capturedBatchIterator())
            .whenComplete((result, failure) -> {
                if (failure == null) {
                    BatchIterator<Row> nlIterator = createNestedLoopIterator(
                        leftConsumer.capturedBatchIterator().join(),
                        numLeftCols,
                        rightConsumer.capturedBatchIterator().join(),
                        numRightCols,
                        joinType,
                        joinPredicate,
                        circuitBreaker,
                        ramAccounting,
                        leftSideColumnTypes,
                        estimatedRowsSizeLeft,
                        estimatedNumberOfRowsLeft,
                        blockNestedLoop
                    );
                    nlResultConsumer.accept(nlIterator, null);
                } else {
                    nlResultConsumer.accept(null, failure);
                }
            });
    }

    @Override
    public CompletableFuture<?> completionFuture() {
        return resultConsumer.completionFuture();
    }

    public RowConsumer leftConsumer() {
        return leftConsumer;
    }

    public RowConsumer rightConsumer() {
        return rightConsumer;
    }

    @VisibleForTesting
    static BatchIterator<Row> createNestedLoopIterator(BatchIterator<Row> left,
                                                       int leftNumCols,
                                                       BatchIterator<Row> right,
                                                       int rightNumCols,
                                                       JoinType joinType,
                                                       Predicate<Row> joinCondition,
                                                       CircuitBreaker circuitBreaker,
                                                       RamAccounting ramAccounting,
                                                       List<DataType<?>> leftSideColumnTypes,
                                                       long estimatedRowsSizeLeft,
                                                       long estimatedNumberOfRowsLeft,
                                                       boolean blockNestedLoop) {
        final CombinedRow combiner = new CombinedRow(leftNumCols, rightNumCols);
        switch (joinType) {
            case CROSS:
                return buildCrossJoinBatchIterator(
                    left,
                    right,
                    combiner,
                    circuitBreaker,
                    ramAccounting,
                    leftSideColumnTypes,
                    estimatedRowsSizeLeft,
                    blockNestedLoop
                );

            case INNER:
                return new FilteringBatchIterator<>(
                    buildCrossJoinBatchIterator(
                        left,
                        right,
                        combiner,
                        circuitBreaker,
                        ramAccounting,
                        leftSideColumnTypes,
                        estimatedRowsSizeLeft,
                        blockNestedLoop),
                        joinCondition);

            case LEFT:
                return new LeftJoinNLBatchIterator<>(left, right, combiner, joinCondition);

            case RIGHT:
                return new RightJoinNLBatchIterator<>(left, right, combiner, joinCondition);

            case FULL:
                return new FullOuterJoinNLBatchIterator<>(left, right, combiner, joinCondition);

            case SEMI:
                return new SemiJoinNLBatchIterator<>(left, right, combiner, joinCondition);

            case ANTI:
                return new AntiJoinNLBatchIterator<>(left, right, combiner, joinCondition);

            default:
                throw new AssertionError("Invalid joinType: " + joinType);
        }
    }

    private static BatchIterator<Row> buildCrossJoinBatchIterator(BatchIterator<Row> left,
                                                                  BatchIterator<Row> right,
                                                                  CombinedRow combiner,
                                                                  CircuitBreaker circuitBreaker,
                                                                  RamAccounting ramAccounting,
                                                                  List<DataType<?>> leftSideColumnTypes,
                                                                  long estimatedRowsSizeLeft,
                                                                  boolean blockNestedLoop) {<FILL_FUNCTION_BODY>}
}

        if (blockNestedLoop) {
            var blockSizeCalculator = new RamBlockSizeCalculator(
                Paging.PAGE_SIZE,
                circuitBreaker,
                estimatedRowsSizeLeft
            );
            var rowAccounting = new TypedCellsAccounting(leftSideColumnTypes, ramAccounting, 0);
            return new CrossJoinBlockNLBatchIterator(left, right, combiner, blockSizeCalculator, rowAccounting);
        } else {
            return new CrossJoinNLBatchIterator<>(left, right, combiner);
        }
    
<no_super_class>