| 854 | } |
| 855 | |
| 856 | @VisibleForTesting |
| 857 | CompletableFuture<?> singleExec(Portal portal, ResultReceiver<?> resultReceiver, int maxRows) { |
| 858 | RowConsumerToResultReceiver activeConsumer = portal.activeConsumer(); |
| 859 | if (activeConsumer != null) { |
| 860 | activeConsumer.closeAndFinishIfSuspended(); |
| 861 | return activeConsumer.completionFuture(); |
| 862 | } |
| 863 | portal.preparedStmt().timeoutToken().enable(); |
| 864 | |
| 865 | mostRecentJobID = UUIDs.dirtyUUID(); |
| 866 | final UUID jobId = mostRecentJobID; |
| 867 | var routingProvider = new RoutingProvider(Randomness.get().nextInt(), planner.getAwarenessAttributes()); |
| 868 | var txnCtx = new CoordinatorTxnCtx(sessionSettings); |
| 869 | var params = new RowN(portal.params().toArray()); |
| 870 | TimeoutToken timeoutToken = portal.preparedStmt().timeoutToken(); |
| 871 | timeoutToken.enable(); |
| 872 | var plannerContext = planner.createContext( |
| 873 | routingProvider, |
| 874 | jobId, |
| 875 | txnCtx, |
| 876 | maxRows, |
| 877 | params, |
| 878 | cursors, |
| 879 | currentTransactionState, |
| 880 | timeoutToken |
| 881 | ); |
| 882 | var analyzedStmt = portal.analyzedStatement(); |
| 883 | |
| 884 | String rawStatement = portal.preparedStmt().rawStatement(); |
| 885 | lastStmt = rawStatement; |
| 886 | if (analyzedStmt == null) { |
| 887 | String errorMsg = "Statement must have been analyzed: " + rawStatement; |
| 888 | jobsLogs.logPreExecutionFailure(jobId, id, rawStatement, errorMsg, sessionSettings.sessionUser()); |
| 889 | throw new IllegalStateException(errorMsg); |
| 890 | } |
| 891 | Plan plan; |
| 892 | try { |
| 893 | plan = planner.plan(analyzedStmt, plannerContext); |
| 894 | timeoutToken.check("singleExec:plan"); |
| 895 | } catch (Throwable t) { |
| 896 | jobsLogs.logPreExecutionFailure(jobId, id, rawStatement, SQLExceptions.messageOf(t), sessionSettings.sessionUser()); |
| 897 | throw t; |
| 898 | } |
| 899 | if (!analyzedStmt.isWriteOperation()) { |
| 900 | resultReceiver = new RetryOnFailureResultReceiver<>( |
| 901 | tempErrorRetryCount, |
| 902 | executor.clusterService(), |
| 903 | resultReceiver, |
| 904 | jobId, |
| 905 | (newJobId, resultRec) -> retryQuery( |
| 906 | newJobId, |
| 907 | analyzedStmt, |
| 908 | routingProvider, |
| 909 | new RowConsumerToResultReceiver( |
| 910 | resultRec, |
| 911 | maxRows, |
| 912 | new JobsLogsUpdateListener(newJobId, jobsLogs)), |
| 913 | params, |