class ResultSetReceiver extends BaseResultReceiver {

    private final String query;
    private final DelayableWriteChannel channel;
    private final List<PGType<?>> columnTypes;
    private final TransactionState transactionState;
    private final AccessControl accessControl;
    private final Channel directChannel;
    private final DelayedWrites delayedWrites;

    @Nullable
    private final FormatCodes.FormatCode[] formatCodes;

    private long rowCount = 0;

    ResultSetReceiver(String query,
                      DelayableWriteChannel channel,
                      DelayedWrites delayedWrites,
                      TransactionState transactionState,
                      AccessControl accessControl,
                      List<PGType<?>> columnTypes,
                      @Nullable FormatCodes.FormatCode[] formatCodes) {
        this.query = query;
        this.channel = channel;
        this.delayedWrites = delayedWrites;
        this.directChannel = channel.bypassDelay();
        this.transactionState = transactionState;
        this.accessControl = accessControl;
        this.columnTypes = columnTypes;
        this.formatCodes = formatCodes;
    }

    @Override
    public void setNextRow(Row row) {
        rowCount++;
        Messages.sendDataRow(directChannel, row, columnTypes, formatCodes);
        if (rowCount % 1000 == 0) {
            directChannel.flush();
        }
    }

    @Override
    public void batchFinished() {<FILL_FUNCTION_BODY>}

    @Override
    public void allFinished() {
        ChannelFuture sendCommandComplete = Messages.sendCommandComplete(directChannel, query, rowCount);
        channel.writePendingMessages(delayedWrites);
        channel.flush();
        sendCommandComplete.addListener(f -> super.allFinished());
    }

    @Override
    public void fail(@NotNull Throwable throwable) {
        ChannelFuture sendErrorResponse = Messages.sendErrorResponse(directChannel, accessControl, throwable);
        channel.writePendingMessages(delayedWrites);
        channel.flush();
        sendErrorResponse.addListener(f -> super.fail(throwable));
    }
}

        ChannelFuture sendPortalSuspended = Messages.sendPortalSuspended(directChannel);
        channel.writePendingMessages(delayedWrites);
        channel.flush();

        // Trigger the completion future but by-pass `sendCompleteComplete`
        // This resultReceiver shouldn't be used anymore. The next `execute` message
        // from the client will create a new one.
        sendPortalSuspended.addListener(f -> super.allFinished());
    
<methods>public non-sealed void <init>() ,public void allFinished() ,public void batchFinished() ,public CompletableFuture<java.lang.Void> completionFuture() ,public void fail(java.lang.Throwable) ,public void setNextRow(io.crate.data.Row) <variables>private CompletableFuture<java.lang.Void> completionFuture