Method execute

server/src/main/java/io/crate/session/Session.java:517–602

(String portalName, int maxRows, ResultReceiver<?> resultReceiver)

    }

    @Nullable
    public CompletableFuture<?> execute(String portalName, int maxRows, ResultReceiver<?> resultReceiver) {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("method=execute portalName={} maxRows={}", portalName, maxRows);
        }
        Portal portal = getSafePortal(portalName);
        var activeConsumer = portal.activeConsumer();
        if (activeConsumer != null && activeConsumer.suspended()) {
            activeConsumer.replaceResultReceiver(resultReceiver, maxRows);
            activeConsumer.resume();
            return resultReceiver.completionFuture();
        }

        var analyzedStmt = portal.analyzedStatement();
        if (isReadOnly && analyzedStmt.isWriteOperation()) {
            throw new ReadOnlyException(portal.preparedStmt().rawStatement());
        }
        if (analyzedStmt instanceof AnalyzedBegin) {
            currentTransactionState = TransactionState.IN_TRANSACTION;
            resultReceiver.allFinished();
        } else if (analyzedStmt instanceof AnalyzedCommit) {
            currentTransactionState = TransactionState.IDLE;
            cursors.close(cursor -> cursor.hold() == Hold.WITHOUT);
            resultReceiver.allFinished();
            return resultReceiver.completionFuture();
        } else if (analyzedStmt instanceof AnalyzedDeallocate ad) {
            String stmtToDeallocate = ad.preparedStmtName();
            if (stmtToDeallocate != null) {
                close((byte) 'S', stmtToDeallocate);
            } else {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("deallocating all prepared statements");
                }
                preparedStatements.clear();
            }
            resultReceiver.allFinished();
        } else if (analyzedStmt instanceof AnalyzedDiscard discard) {
            // We don't cache plans, don't have sequences or temporary tables
            // See https://www.postgresql.org/docs/current/sql-discard.html
            if (discard.target() == Target.ALL) {
                close();
            }
            resultReceiver.allFinished();
        } else if (analyzedStmt.isWriteOperation()) {
            /* We defer the execution for any other statements to `sync` messages so that we can efficiently process
             * bulk operations. E.g. If we receive `INSERT INTO (x) VALUES (?)` bindings/execute multiple times
             * We want to create bulk requests internally:                                                          /
             * - To reduce network overhead
             * - To have 1 disk flush per shard instead of 1 disk flush per item
             *
             * Many clients support this by doing something like this:
             *
             *      var preparedStatement = conn.prepareStatement("...")
             *      for (var args in manyArgs):
             *          preparedStatement.execute(args)
             *      conn.commit()
             */
            deferredExecutionsByStmt.computeIfAbsent(