You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/08/12 16:55:12 UTC
[XL] Change in asterixdb[master]: multistatement archived patch
From Ali Alsuliman <al...@gmail.com>:
Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21541?usp=email )
Change subject: multistatement archived patch
......................................................................
multistatement archived patch
Change-Id: I5efd9a2fec92d6176b709beb621c8bef7bb86bdc
---
M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ClientRequest.java
M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ExecutionPlans.java
M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IRequestParameters.java
M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ResultProperties.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/AbstractInternalRequestMessage.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ClientInfoRequestMessage.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementResponseMessage.java
A asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/fields/NcStatementsPrinter.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/RequestParameters.java
A asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/server/StatementPropertiesTransferTest.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
A asterixdb/asterix-app/src/test/java/org/apache/asterix/test/result/NcStatementsPrinterTest.java
A asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.1.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.2.query.sqlpp
M asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/AsyncDeferredQueries.xml
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/async-not-supported-dataverse/async-not-supported-dataverse.1.async.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.1.adm
A asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.2.adm
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_parseonly.xml
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IClientRequest.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
M asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/RequestTracker.java
30 files changed, 1,552 insertions(+), 152 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/41/21541/1
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ClientRequest.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ClientRequest.java
index 6a533b3..141468b 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ClientRequest.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ClientRequest.java
@@ -19,8 +19,17 @@
package org.apache.asterix.translator;
import static org.apache.hyracks.api.job.resource.IJobCapacityController.JobSubmissionStatus.QUEUE;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.asterix.common.api.ICommonRequestParameters;
@@ -35,7 +44,9 @@
import org.apache.hyracks.api.util.ExceptionUtils;
import org.apache.hyracks.util.LogRedactionUtil;
import org.apache.hyracks.util.StorageUtil;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
public class ClientRequest extends BaseClientRequest {
@@ -47,8 +58,18 @@
protected Thread executor;
protected final String statement;
protected final String clientContextId;
- protected final JobState jobState;
+ /** The state of the most recent job; the one a request running a single job reports. */
+ protected volatile JobState jobState;
protected volatile JobId jobId;
+ /**
+ * The state of every job this request has run, keyed by job and in the order they ran; a request runs one job per
+ * statement. A request running a single job reports exactly what it did before, from {@link #jobState}.
+ */
+ private final Map<JobId, JobState> jobStates = Collections.synchronizedMap(new LinkedHashMap<>());
+ /** The jobs of this request whose results have been swept; guarded by {@link #jobStates}. */
+ private final Set<JobId> sweptJobs = new HashSet<>();
+ /** The jobs whose results a client can still fetch; guarded by {@link #jobStates}. */
+ private final Set<JobId> fetchableJobs = new HashSet<>();
private volatile String plan; // can be null
private volatile long compileTimeNanos;
@@ -79,6 +100,7 @@
public synchronized void setJobId(JobId jobId) {
this.jobId = jobId;
+ stateOf(jobId);
setRunning();
}
@@ -86,6 +108,59 @@
return jobId;
}
+ /**
+ * @return whether {@code candidate} is one of this request's jobs. A request runs a job per statement, so
+ * identifying it by its most recent job alone would disown the jobs of all of its earlier statements --
+ * and with them the result handles those statements returned.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public boolean ownsJob(JobId candidate) {
+ return candidate != null && jobStates.containsKey(candidate);
+ }
+
+ /** @return this request's jobs, in the order they ran. */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public Set<JobId> getJobIds() {
+ synchronized (jobStates) {
+ return new LinkedHashSet<>(jobStates.keySet());
+ }
+ }
+
+ /**
+ * Records the jobs whose results a client was handed a handle to, and which therefore have to be swept before this
+ * request has nothing left to fetch. Only these count: a job that registers no result -- a DDL statement's, say --
+ * is never swept, so requiring every job of the request would pin it forever.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void setFetchableJobs(Set<JobId> jobs) {
+ synchronized (jobStates) {
+ fetchableJobs.clear();
+ fetchableJobs.addAll(jobs);
+ }
+ }
+
+ @Override
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public boolean resultSwept(JobId jobId) {
+ synchronized (jobStates) {
+ sweptJobs.add(jobId);
+ // before the fetchable set is known, fall back to every job, which for a single job is the same thing
+ Set<JobId> awaited = fetchableJobs.isEmpty() ? jobStates.keySet() : fetchableJobs;
+ return sweptJobs.containsAll(awaited);
+ }
+ }
+
+ /**
+ * @return the state of {@code forJob}, creating it if this is the first notification about that job, and making it
+ * the state the flat job fields describe
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ private JobState stateOf(JobId forJob) {
+ JobState state = jobStates.computeIfAbsent(forJob, key -> new JobState());
+ jobState = state;
+ return state;
+ }
+
public Thread getExecutor() {
return executor;
}
@@ -94,12 +169,25 @@
protected void doCancel(ICcApplicationContext appCtx) throws HyracksDataException {
// if the request has a job, we abort the job and do not interrupt the thread as it will be notified
// that the job has been cancelled. Otherwise, we interrupt the thread
- if (jobId != null) {
+ Set<JobId> jobs = getJobIds();
+ if (!jobs.isEmpty()) {
IHyracksClientConnection hcc = appCtx.getHcc();
- try {
- hcc.cancelJob(jobId);
- } catch (Exception e) {
- throw HyracksDataException.create(e);
+ // every job of the request, not just the most recent: a request that ran several statements has a job
+ // per statement, and an earlier one may still be holding its result
+ HyracksDataException failure = null;
+ for (JobId job : jobs) {
+ try {
+ hcc.cancelJob(job);
+ } catch (Exception e) {
+ if (failure == null) {
+ failure = HyracksDataException.create(e);
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
}
} else if (executor != null) {
executor.interrupt();
@@ -144,27 +232,32 @@
return json;
}
+ // these are told which job they concern, so they keep that job's own state rather than a single shared one
+
@Override
public void jobCreated(JobId jobId, IReadOnlyClusterCapacity requiredClusterCapacity,
IJobCapacityController.JobSubmissionStatus status) {
- jobState.createTime = System.currentTimeMillis();
- jobState.status = status == QUEUE ? JobStatus.PENDING : JobStatus.RUNNING;
- jobState.requiredCPUs = requiredClusterCapacity.getAggregatedCores();
- jobState.requiredMemoryInBytes = requiredClusterCapacity.getAggregatedMemoryByteSize();
+ JobState state = stateOf(jobId);
+ state.createTime = System.currentTimeMillis();
+ state.status = status == QUEUE ? JobStatus.PENDING : JobStatus.RUNNING;
+ state.requiredCPUs = requiredClusterCapacity.getAggregatedCores();
+ state.requiredMemoryInBytes = requiredClusterCapacity.getAggregatedMemoryByteSize();
}
@Override
public void jobStarted(JobId jobId) {
- jobState.startTime = System.currentTimeMillis();
- jobState.status = JobStatus.RUNNING;
+ JobState state = stateOf(jobId);
+ state.startTime = System.currentTimeMillis();
+ state.status = JobStatus.RUNNING;
}
@Override
public void jobFinished(JobId jobId, JobStatus jobStatus, List<Exception> exceptions) {
- jobState.endTime = System.currentTimeMillis();
- jobState.status = jobStatus;
+ JobState state = stateOf(jobId);
+ state.endTime = System.currentTimeMillis();
+ state.status = jobStatus;
if (exceptions != null && !exceptions.isEmpty()) {
- jobState.errorMsg = processException(exceptions.get(0));
+ state.errorMsg = processException(exceptions.get(0));
}
}
@@ -176,6 +269,20 @@
try {
json.put("jobId", jobId != null ? jobId.toString() : null);
putJobState(json, jobState, redact);
+ // the flat fields above describe the most recent job, which is the only one for a single-statement
+ // request. A request that ran several also lists them, so none of them is invisible here.
+ Map<JobId, JobState> states = new LinkedHashMap<>();
+ synchronized (jobStates) {
+ states.putAll(jobStates);
+ }
+ if (states.size() > 1) {
+ ArrayNode jobs = json.putArray("jobs");
+ states.forEach((job, state) -> {
+ ObjectNode jobNode = jobs.addObject();
+ jobNode.put("jobId", job.toString());
+ putJobState(jobNode, state, redact);
+ });
+ }
} catch (Throwable th) {
// ignore
}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ExecutionPlans.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ExecutionPlans.java
index f33bb01..436c4f1 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ExecutionPlans.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ExecutionPlans.java
@@ -18,8 +18,14 @@
*/
package org.apache.asterix.translator;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
import java.io.Serializable;
+import org.apache.hyracks.util.annotations.AiProvenance;
+
public class ExecutionPlans implements Serializable {
private static final long serialVersionUID = 6853904213354224458L;
@@ -33,6 +39,43 @@
private String statementParameters;
private boolean explainOnly;
+ public ExecutionPlans() {
+ }
+
+ /**
+ * Copies {@code other}, so that the plans of an individual statement can be retained while the shared instance
+ * this was copied from goes on to be overwritten by the next statement of the same request.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public ExecutionPlans(ExecutionPlans other) {
+ expressionTree = other.expressionTree;
+ rewrittenExpressionTree = other.rewrittenExpressionTree;
+ logicalPlan = other.logicalPlan;
+ optimizedLogicalPlan = other.optimizedLogicalPlan;
+ job = other.job;
+ signature = other.signature;
+ statementCategory = other.statementCategory;
+ statementParameters = other.statementParameters;
+ explainOnly = other.explainOnly;
+ }
+
+ /**
+ * Clears every field. Used between the statements of a multi-statement request; without it a statement that
+ * generates no plan of its own reports the plans of whichever earlier statement did.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void reset() {
+ expressionTree = null;
+ rewrittenExpressionTree = null;
+ logicalPlan = null;
+ optimizedLogicalPlan = null;
+ job = null;
+ signature = null;
+ statementCategory = null;
+ statementParameters = null;
+ explainOnly = false;
+ }
+
public String getExpressionTree() {
return expressionTree;
}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IRequestParameters.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IRequestParameters.java
index 4553f35..4628803 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IRequestParameters.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IRequestParameters.java
@@ -22,7 +22,7 @@
import org.apache.asterix.common.api.ICommonRequestParameters;
import org.apache.asterix.om.base.IAObject;
-import org.apache.asterix.translator.IStatementExecutor.StatementProperties;
+import org.apache.asterix.translator.IStatementExecutor.RequestStatementProperties;
import org.apache.asterix.translator.IStatementExecutor.Stats;
import org.apache.hyracks.api.result.IResultSet;
@@ -46,10 +46,10 @@
Stats getStats();
/**
- * @return a reference on which to write properties of executed queries (e.g. what kind of statement was parsed
- * by the parser)
+ * @return a reference on which to record the properties of each executed statement (e.g. what kind of statement
+ * was parsed by the parser), in execution order
*/
- StatementProperties getStatementProperties();
+ RequestStatementProperties getStatementProperties();
/**
* @return a reference to write the metadata of executed queries
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
index c7443d0..6c4a00b 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/IStatementExecutor.java
@@ -18,6 +18,10 @@
*/
package org.apache.asterix.translator;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
@@ -32,6 +36,7 @@
import java.util.function.Function;
import java.util.stream.Collectors;
+import org.apache.asterix.algebra.extension.ExtensionStatement;
import org.apache.asterix.common.api.IResponsePrinter;
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.exceptions.AsterixException;
@@ -45,10 +50,12 @@
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.client.IClusterInfoCollector;
import org.apache.hyracks.api.client.IHyracksClientConnection;
+import org.apache.hyracks.api.exceptions.SourceLocation;
import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.api.result.ResultSetId;
+import org.apache.hyracks.util.annotations.AiProvenance;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -94,14 +101,89 @@
}
class ResultMetadata implements Serializable {
- private static final long serialVersionUID = 2L;
+ private static final long serialVersionUID = 3L;
private final List<ResultSetInfo> resultSets = new ArrayList<>();
+ private final List<StatementInfo> statements = new ArrayList<>();
public List<ResultSetInfo> getResultSets() {
return resultSets;
}
+ /**
+ * @return what each statement of the request produced, in execution order. Unlike {@link #getResultSets()}
+ * this has an entry for every statement that ran, including those producing no result, which is what
+ * makes it usable as the "how many statements were there" signal.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public List<StatementInfo> getStatements() {
+ return statements;
+ }
+ }
+
+ /**
+ * What one statement of a request produced. Collected on the CC and carried back to the NC so the NC can render a
+ * statement-by-statement response without having parsed the request itself.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ class StatementInfo implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final StatementProperties properties;
+ /** null for a statement that produces no result. */
+ private ResultSetInfo resultSet;
+ private ExecutionPlans plans;
+ private String signature;
+ private Stats stats;
+ private Throwable error;
+
+ public StatementInfo(StatementProperties properties) {
+ this.properties = properties;
+ }
+
+ public StatementProperties getProperties() {
+ return properties;
+ }
+
+ public ResultSetInfo getResultSet() {
+ return resultSet;
+ }
+
+ public void setResultSet(ResultSetInfo resultSet) {
+ this.resultSet = resultSet;
+ }
+
+ public ExecutionPlans getPlans() {
+ return plans;
+ }
+
+ public void setPlans(ExecutionPlans plans) {
+ this.plans = plans;
+ }
+
+ public String getSignature() {
+ return signature;
+ }
+
+ public void setSignature(String signature) {
+ this.signature = signature;
+ }
+
+ public Stats getStats() {
+ return stats;
+ }
+
+ public void setStats(Stats stats) {
+ this.stats = stats;
+ }
+
+ public Throwable getError() {
+ return error;
+ }
+
+ public void setError(Throwable error) {
+ this.error = error;
+ }
}
class ResultSetInfo implements Serializable {
@@ -283,6 +365,99 @@
public void setCachedPlan(boolean cachedPlan) {
this.cachedPlan = cachedPlan;
}
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void addProcessedObjects(long delta) {
+ processedObjects += delta;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void addQueueWaitTimeNanos(long delta) {
+ queueWaitTime += delta;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void addCompileTimeNanos(long delta) {
+ compileTime += delta;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void addBufferCachePageReadCount(long delta) {
+ bufferCachePageReadCount += delta;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void addCloudReadRequestsCount(long delta) {
+ cloudReadRequestsCount += delta;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void addCloudPagesReadCount(long delta) {
+ cloudPagesReadCount += delta;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void addCloudPagesPersistedCount(long delta) {
+ cloudPagesPersistedCount += delta;
+ }
+
+ /**
+ * @return a copy of these stats, for retaining the figures of one statement while this instance goes on to be
+ * updated by the next statement of the same request
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public Stats snapshot() {
+ Stats copy = new Stats();
+ copy.count = count;
+ copy.size = size;
+ copy.processedObjects = processedObjects;
+ copy.queueWaitTime = queueWaitTime;
+ copy.profile = profile;
+ copy.profileType = profileType;
+ copy.totalWarningsCount = totalWarningsCount;
+ copy.compileTime = compileTime;
+ copy.bufferCacheHitRatio = bufferCacheHitRatio;
+ copy.bufferCachePageReadCount = bufferCachePageReadCount;
+ copy.cloudReadRequestsCount = cloudReadRequestsCount;
+ copy.cloudPagesReadCount = cloudPagesReadCount;
+ copy.cloudPagesPersistedCount = cloudPagesPersistedCount;
+ copy.cachedPlan = cachedPlan;
+ return copy;
+ }
+
+ /**
+ * Subtracts {@code earlier} from these stats, turning a running request-wide total into the contribution of
+ * the statement that ran since {@code earlier} was taken. Every field that accumulates over a request is
+ * differenced; {@code bufferCacheHitRatio}, {@code cachedPlan} and the profile describe a single job rather
+ * than a total, so they are carried over as-is — which is this statement's own job, since a statement runs one.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public Stats since(Stats earlier) {
+ Stats delta = snapshot();
+ delta.count -= earlier.count;
+ delta.size -= earlier.size;
+ delta.totalWarningsCount -= earlier.totalWarningsCount;
+ delta.processedObjects -= earlier.processedObjects;
+ delta.queueWaitTime -= earlier.queueWaitTime;
+ delta.compileTime -= earlier.compileTime;
+ delta.bufferCachePageReadCount -= earlier.bufferCachePageReadCount;
+ delta.cloudReadRequestsCount -= earlier.cloudReadRequestsCount;
+ delta.cloudPagesReadCount -= earlier.cloudPagesReadCount;
+ delta.cloudPagesPersistedCount -= earlier.cloudPagesPersistedCount;
+ return delta;
+ }
+
+ /** Adds the per-job figures of {@code other} to these, for aggregating a request from its statements. */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public void accumulate(Stats other) {
+ processedObjects += other.processedObjects;
+ queueWaitTime += other.queueWaitTime;
+ compileTime += other.compileTime;
+ bufferCachePageReadCount += other.bufferCachePageReadCount;
+ cloudReadRequestsCount += other.cloudReadRequestsCount;
+ cloudPagesReadCount += other.cloudPagesReadCount;
+ cloudPagesPersistedCount += other.cloudPagesPersistedCount;
+ }
}
class Profile implements Serializable {
@@ -317,6 +492,16 @@
private Statement.Kind kind;
private String name;
+ /** 0-based position among the request's user-supplied statements; -1 while unassigned or for synthetic ones. */
+ private int index = -1;
+ /** {@code true} for a statement the engine injected rather than the client sending it. */
+ private boolean synthetic;
+ private int sourceLine;
+ private int sourceColumn;
+ /** {@code true} once the statement reached its handler. */
+ private boolean executed;
+ /** {@code true} if the statement's handler threw. */
+ private boolean failed;
public Statement.Kind getKind() {
return kind;
@@ -334,6 +519,51 @@
this.name = name;
}
+ public int getIndex() {
+ return index;
+ }
+
+ public void setIndex(int index) {
+ this.index = index;
+ }
+
+ public boolean isSynthetic() {
+ return synthetic;
+ }
+
+ public void setSynthetic(boolean synthetic) {
+ this.synthetic = synthetic;
+ }
+
+ public int getSourceLine() {
+ return sourceLine;
+ }
+
+ public int getSourceColumn() {
+ return sourceColumn;
+ }
+
+ public void setSourceLocation(int sourceLine, int sourceColumn) {
+ this.sourceLine = sourceLine;
+ this.sourceColumn = sourceColumn;
+ }
+
+ public boolean isExecuted() {
+ return executed;
+ }
+
+ public void setExecuted(boolean executed) {
+ this.executed = executed;
+ }
+
+ public boolean isFailed() {
+ return failed;
+ }
+
+ public void setFailed(boolean failed) {
+ this.failed = failed;
+ }
+
public boolean isValid() {
return kind != null && (kind != Statement.Kind.EXTENSION || name != null);
}
@@ -345,6 +575,88 @@
}
/**
+ * The {@link StatementProperties} of every statement of one request, in execution order.
+ * <p>
+ * A request used to carry a single mutable {@code StatementProperties} that each statement overwrote in turn,
+ * which left every consumer reading it outside the statement's own execution — auditing, and any read from the
+ * asynchronous job-submission thread — looking at whichever statement happened to write last.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ class RequestStatementProperties implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final List<StatementProperties> statements = new ArrayList<>();
+ private transient StatementProperties current;
+
+ /**
+ * Appends and returns the properties of the statement that is about to be dispatched, and makes it
+ * {@link #current()}. {@code name} is set only for an extension statement, so that an extension statement's
+ * name cannot leak onto a later one.
+ *
+ * @param index 0-based position among the user-supplied statements, or -1 for a synthetic statement
+ * @param statement the statement about to be dispatched
+ */
+ public StatementProperties begin(int index, Statement statement) {
+ StatementProperties properties = new StatementProperties();
+ properties.setKind(statement.getKind());
+ properties.setIndex(index);
+ properties.setSynthetic(index < 0);
+ if (statement instanceof ExtensionStatement) {
+ properties.setName(((ExtensionStatement) statement).getName());
+ }
+ SourceLocation sourceLocation = statement.getSourceLocation();
+ if (sourceLocation != null) {
+ properties.setSourceLocation(sourceLocation.getLine(), sourceLocation.getColumn());
+ }
+ statements.add(properties);
+ current = properties;
+ return properties;
+ }
+
+ /**
+ * @return the statement currently being executed, or {@code null} outside the execution of any statement.
+ * This is what a consumer acting on behalf of one statement must read — authorization, for instance,
+ * derives the required privilege from the statement kind and would grant the wrong privilege if it
+ * read a request-wide value instead.
+ */
+ public StatementProperties current() {
+ return current;
+ }
+
+ /** Appends already-populated properties, e.g. those a CC reported back to an NC. */
+ public void add(StatementProperties properties) {
+ statements.add(properties);
+ }
+
+ public List<StatementProperties> asList() {
+ return Collections.unmodifiableList(statements);
+ }
+
+ /**
+ * @return the statement most recently passed to {@link #begin}, or an empty instance before the first one.
+ * Never {@code null}, so that a caller wanting the legacy "the request's statement" view — the last
+ * writer under the old single-instance behaviour — needs no null handling.
+ */
+ public StatementProperties last() {
+ return statements.isEmpty() ? new StatementProperties() : statements.get(statements.size() - 1);
+ }
+
+ /** @return the user-supplied statements, i.e. excluding any the engine injected. */
+ public List<StatementProperties> userStatements() {
+ return statements.stream().filter(s -> !s.isSynthetic()).collect(Collectors.toList());
+ }
+
+ public int size() {
+ return statements.size();
+ }
+
+ @Override
+ public String toString() {
+ return String.valueOf(last());
+ }
+ }
+
+ /**
* Compiles and executes a list of statements
*
* @param hcc
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ResultProperties.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ResultProperties.java
index 21322d4..7264a61 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ResultProperties.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/ResultProperties.java
@@ -18,14 +18,21 @@
*/
package org.apache.asterix.translator;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
import java.io.Serializable;
+import org.apache.hyracks.util.annotations.AiProvenance;
+
public class ResultProperties implements Serializable {
private static final long serialVersionUID = -4741260459407538017L;
public static final long DEFAULT_MAX_READS = 1;
public static final long DEFAULT_RESULT_TTL = -1L; // -1 means use system default
private final IStatementExecutor.ResultDelivery delivery;
+ private final IStatementExecutor.ResultDelivery requestedDelivery;
private final long maxReads;
private final long resultTtlInMillis;
@@ -38,7 +45,13 @@
}
public ResultProperties(IStatementExecutor.ResultDelivery delivery, long maxReads, long resultTtlInMillis) {
+ this(delivery, delivery, maxReads, resultTtlInMillis);
+ }
+
+ private ResultProperties(IStatementExecutor.ResultDelivery delivery,
+ IStatementExecutor.ResultDelivery requestedDelivery, long maxReads, long resultTtlInMillis) {
this.delivery = delivery;
+ this.requestedDelivery = requestedDelivery;
this.maxReads = maxReads;
this.resultTtlInMillis = resultTtlInMillis;
}
@@ -47,6 +60,16 @@
return delivery;
}
+ /**
+ * @return the delivery the client asked for, which {@link #getDelivery()} no longer reports once
+ * {@link #getNcToCcResultProperties()} has substituted DEFERRED for it. Anything reasoning about what the
+ * client will do with the result — rather than about how this hop delivers it — wants this one.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ public IStatementExecutor.ResultDelivery getRequestedDelivery() {
+ return requestedDelivery;
+ }
+
public long getMaxReads() {
return maxReads;
}
@@ -59,7 +82,7 @@
if (delivery != IStatementExecutor.ResultDelivery.IMMEDIATE) {
return this;
}
- // switch IMMEDIATE to DEFERRED since the result will be severed by the NC
- return new ResultProperties(IStatementExecutor.ResultDelivery.DEFERRED, maxReads, resultTtlInMillis);
+ // switch IMMEDIATE to DEFERRED since the result will be severed by the NC, but remember what was asked for
+ return new ResultProperties(IStatementExecutor.ResultDelivery.DEFERRED, delivery, maxReads, resultTtlInMillis);
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
index 53abe1b..b0d58ce 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ApiServlet.java
@@ -154,7 +154,7 @@
long startTime = System.currentTimeMillis();
final IRequestParameters requestParameters = new RequestParameters(requestReference, query, resultSet,
new ResultProperties(IStatementExecutor.ResultDelivery.IMMEDIATE), new IStatementExecutor.Stats(),
- new IStatementExecutor.StatementProperties(), null, null, null, null, null, true);
+ new IStatementExecutor.RequestStatementProperties(), null, null, null, null, null, true);
translator.compileAndExecute(hcc, requestParameters);
long endTime = System.currentTimeMillis();
duration = (endTime - startTime) / 1000.00;
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
index 9a042f6..7e21dfd 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NCQueryServiceServlet.java
@@ -19,6 +19,13 @@
package org.apache.asterix.api.http.server;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.REFACTORED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
+import java.nio.charset.Charset;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
@@ -30,10 +37,14 @@
import org.apache.asterix.app.message.CancelQueryRequest;
import org.apache.asterix.app.message.ExecuteStatementRequestMessage;
import org.apache.asterix.app.message.ExecuteStatementResponseMessage;
+import org.apache.asterix.app.result.ExecutionWarning;
import org.apache.asterix.app.result.ResponsePrinter;
import org.apache.asterix.app.result.fields.NcResultPrinter;
+import org.apache.asterix.app.result.fields.NcStatementsPrinter;
import org.apache.asterix.app.result.fields.SignaturePrinter;
+import org.apache.asterix.app.result.fields.WarningsPrinter;
import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.api.ICodedMessage;
import org.apache.asterix.common.api.IRequestReference;
import org.apache.asterix.common.config.GlobalConfig;
import org.apache.asterix.common.exceptions.ErrorCode;
@@ -53,7 +64,9 @@
import org.apache.hyracks.http.api.IServletResponse;
import org.apache.hyracks.http.server.HttpServer;
import org.apache.hyracks.http.server.InterruptOnCloseHandler;
+import org.apache.hyracks.http.server.utils.HttpUtil;
import org.apache.hyracks.ipc.exceptions.IPCException;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.Level;
import io.netty.handler.codec.http.HttpResponseStatus;
@@ -73,7 +86,7 @@
@Override
protected void executeStatement(IServletRequest request, IRequestReference requestReference, String statementsText,
SessionOutput sessionOutput, ResultProperties resultProperties,
- IStatementExecutor.StatementProperties statementProperties, IStatementExecutor.Stats stats,
+ IStatementExecutor.RequestStatementProperties statementProperties, IStatementExecutor.Stats stats,
QueryServiceRequestParameters param, RequestExecutionState executionState,
Map<String, String> optionalParameters, Map<String, byte[]> statementParameters,
ResponsePrinter responsePrinter, List<Warning> warnings) throws Exception {
@@ -136,15 +149,74 @@
executionState.setStatus(ResultStatus.SUCCESS, HttpResponseStatus.OK);
}
updateStatsFromCC(stats, responseMsg);
- if (param.isSignature() && delivery != IStatementExecutor.ResultDelivery.ASYNC && !param.isParseOnly()) {
- responsePrinter.addResultPrinter(SignaturePrinter.newInstance(responseMsg.getExecutionPlans()));
+ if (useMultiStatementResponse(responseMsg)) {
+ responsePrinter.addResultPrinter(newStatementsPrinter(responseMsg, delivery, sessionOutput,
+ HttpUtil.getPreferredCharset(request), param.isSignature() && !param.isParseOnly(),
+ param.isIncludeHost() ? null : requestReference.getUuid()));
+ warnings.addAll(responseMsg.getWarnings());
+ buildMultiStatementResponseResults(responsePrinter, warnings, executionState);
+ } else {
+ if (param.isSignature() && delivery != IStatementExecutor.ResultDelivery.ASYNC && !param.isParseOnly()) {
+ responsePrinter.addResultPrinter(SignaturePrinter.newInstance(responseMsg.getExecutionPlans()));
+ }
+ if (hasResult(responseMsg)) {
+ responsePrinter.addResultPrinter(
+ new NcResultPrinter(appCtx, responseMsg, getResultSet(), delivery, sessionOutput, stats));
+ }
+ warnings.addAll(responseMsg.getWarnings());
+ buildResponseResults(responsePrinter, sessionOutput, responseMsg.getExecutionPlans(), warnings,
+ executionState);
}
- if (hasResult(responseMsg)) {
- responsePrinter.addResultPrinter(
- new NcResultPrinter(appCtx, responseMsg, getResultSet(), delivery, sessionOutput, stats));
+ }
+
+ /**
+ * Whether to report this request statement by statement, rather than merging every statement's output into one set
+ * of request-level fields.
+ * <p>
+ * Off here, and deliberately: this service has always accepted multiple statements in a request, so its response
+ * to one is long-established and clients depend on its shape. A deployment that did not accept them — and so has
+ * no client that can already be relying on a multi-statement response — can turn this on to get the
+ * statement-wise shape without breaking anyone. {@link #hasMultipleStatements} is the predicate to use.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ protected boolean useMultiStatementResponse(ExecuteStatementResponseMessage responseMsg) {
+ return false;
+ }
+
+ /**
+ * Whether the request carried more than one statement of the client's own. Only the CC can tell, since the NC
+ * forwards the request text without parsing it, so this reads the list the CC sent back; an absent list means a CC
+ * that does not report one.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ protected static boolean hasMultipleStatements(ExecuteStatementResponseMessage responseMsg) {
+ IStatementExecutor.ResultMetadata metadata = responseMsg.getMetadata();
+ if (metadata == null || metadata.getStatements() == null) {
+ return false;
}
- warnings.addAll(responseMsg.getWarnings());
- buildResponseResults(responsePrinter, sessionOutput, responseMsg.getExecutionPlans(), warnings, executionState);
+ return metadata.getStatements().stream().filter(s -> !s.getProperties().isSynthetic()).count() > 1;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ protected NcStatementsPrinter newStatementsPrinter(ExecuteStatementResponseMessage responseMsg,
+ IStatementExecutor.ResultDelivery delivery, SessionOutput sessionOutput, Charset resultCharset,
+ boolean printSignature, String requestId) throws Exception {
+ return new NcStatementsPrinter(appCtx, responseMsg.getMetadata().getStatements(), getResultSet(), delivery,
+ sessionOutput, resultCharset, printSignature, requestId);
+ }
+
+ /**
+ * The request-level result fields of a multi-statement response. The plans and the signature belong to an
+ * individual statement, so unlike {@link #buildResponseResults} this contributes neither.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ protected void buildMultiStatementResponseResults(ResponsePrinter responsePrinter, List<Warning> warnings,
+ RequestExecutionState executionState) {
+ if (!warnings.isEmpty()) {
+ List<ICodedMessage> codedWarnings = new ArrayList<>();
+ warnings.forEach(warn -> codedWarnings.add(ExecutionWarning.of(warn)));
+ responsePrinter.addResultPrinter(new WarningsPrinter(codedWarnings));
+ }
}
protected void ensureOptionalParameters(Map<String, String> optionalParameters) throws HyracksDataException {
@@ -226,12 +298,20 @@
stats.setCachedPlan(responseStats.isCachedPlan());
}
- private static void updatePropertiesFromCC(IStatementExecutor.StatementProperties statementProperties,
+ /**
+ * Copies onto the NC the properties of every statement the CC ran, in order. The CC is the only side that parsed
+ * the request, so this is the NC's only account of which statements there were -- and everything the NC derives
+ * per statement, auditing above all, is only per statement if all of them arrive.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = REFACTORED)
+ static void updatePropertiesFromCC(IStatementExecutor.RequestStatementProperties statementProperties,
ExecuteStatementResponseMessage responseMsg) {
- IStatementExecutor.StatementProperties responseStmtProps = responseMsg.getStatementProperties();
- if (responseStmtProps != null) {
- statementProperties.setKind(responseStmtProps.getKind());
- statementProperties.setName(responseStmtProps.getName());
+ IStatementExecutor.ResultMetadata metadata = responseMsg.getMetadata();
+ if (metadata == null) {
+ return;
+ }
+ for (IStatementExecutor.StatementInfo statement : metadata.getStatements()) {
+ statementProperties.add(statement.getProperties());
}
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
index 959c479..3e625ac 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
@@ -19,6 +19,9 @@
package org.apache.asterix.api.http.server;
import static org.apache.asterix.common.exceptions.ErrorCode.NO_STATEMENT_PROVIDED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.REFACTORED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
import java.io.IOException;
import java.io.PrintWriter;
@@ -85,6 +88,7 @@
import org.apache.hyracks.http.api.IServletRequest;
import org.apache.hyracks.http.api.IServletResponse;
import org.apache.hyracks.http.server.utils.HttpUtil;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -254,8 +258,8 @@
.serializeParameterValues(param.getStatementParams(), sessionOutput.config().fmt());
setAccessControlHeaders(request, response);
stats.setProfileType(param.getProfileType());
- IStatementExecutor.StatementProperties statementProperties =
- new IStatementExecutor.StatementProperties();
+ IStatementExecutor.RequestStatementProperties statementProperties =
+ new IStatementExecutor.RequestStatementProperties();
response.setStatus(HttpResponseStatus.OK);
executeStatement(request, requestRef, statement, sessionOutput, resultProperties, statementProperties,
stats, param, executionState, param.getOptionalParams(), statementParams, responsePrinter,
@@ -354,6 +358,7 @@
}
}
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = REFACTORED)
protected ResultUtil.ParseOnlyResult parseStatement(String statementsText) throws CompilationException {
IParserFactory factory = compilationProvider.getParserFactory();
IParser parser = factory.createParser(statementsText);
@@ -362,17 +367,21 @@
Set<VariableExpr> allExtVars = new HashSet<>();
for (Statement stmt : stmts) {
- Query query = (Query) stmt;
- Set<VariableExpr> stmtExtVars = compilationProvider.getRewriterFactory().createQueryRewriter()
- .getExternalVariables(query.getBody());
- allExtVars.addAll(stmtExtVars);
+ // only a query has a body to collect external variables from; anything else (a DDL, a DML, SET, USE)
+ // parses fine and simply contributes no external variables. An unconditional cast here used to fail
+ // with a ClassCastException, surfacing as a 500 for any non-query parse-only request.
+ if (stmt instanceof Query query) {
+ Set<VariableExpr> stmtExtVars = compilationProvider.getRewriterFactory().createQueryRewriter()
+ .getExternalVariables(query.getBody());
+ allExtVars.addAll(stmtExtVars);
+ }
}
return new ResultUtil.ParseOnlyResult(allExtVars);
}
protected void executeStatement(IServletRequest request, IRequestReference requestReference, String statementsText,
SessionOutput sessionOutput, ResultProperties resultProperties,
- IStatementExecutor.StatementProperties statementProperties, Stats stats,
+ IStatementExecutor.RequestStatementProperties statementProperties, Stats stats,
QueryServiceRequestParameters param, RequestExecutionState executionState,
Map<String, String> optionalParameters, Map<String, byte[]> statementParameters,
ResponsePrinter responsePrinter, List<Warning> warnings) throws Exception {
@@ -442,8 +451,9 @@
protected IRequestParameters newRequestParameters(QueryServiceRequestParameters param,
IRequestReference requestReference, String statementsText, IResultSet resultSet,
- ResultProperties resultProperties, Stats stats, IStatementExecutor.StatementProperties statementProperties,
- Map<String, String> optionalParameters, Map<String, IAObject> stmtParams, int stmtCategoryRestriction) {
+ ResultProperties resultProperties, Stats stats,
+ IStatementExecutor.RequestStatementProperties statementProperties, Map<String, String> optionalParameters,
+ Map<String, IAObject> stmtParams, int stmtCategoryRestriction) {
RequestParameters requestParameters = new RequestParameters(requestReference, statementsText, resultSet,
resultProperties, stats, statementProperties, null, param.getClientContextID(), param.getDataverse(),
optionalParameters, stmtParams, param.isMultiStatement(), stmtCategoryRestriction);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
index 79e70b9..8520160 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/java/AsterixJavaClient.java
@@ -154,7 +154,7 @@
RequestReference.of(UUID.randomUUID().toString(), "CC", System.currentTimeMillis());
final IRequestParameters requestParameters = new RequestParameters(requestReference, statement, null,
new ResultProperties(IStatementExecutor.ResultDelivery.IMMEDIATE), new IStatementExecutor.Stats(),
- new IStatementExecutor.StatementProperties(), null, null, null, null, statementParams, true);
+ new IStatementExecutor.RequestStatementProperties(), null, null, null, null, statementParams, true);
translator.compileAndExecute(hcc, requestParameters);
executionPlans = translator.getExecutionPlans();
writer.flush();
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/AbstractInternalRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/AbstractInternalRequestMessage.java
index 41c4906..35ac1a5 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/AbstractInternalRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/AbstractInternalRequestMessage.java
@@ -93,7 +93,7 @@
ResponsePrinter printer = new ResponsePrinter(sessionOutput);
ResultProperties resultProperties = new ResultProperties(IStatementExecutor.ResultDelivery.IMMEDIATE, 1);
IRequestParameters requestParams = new RequestParameters(requestReference, "", null, resultProperties,
- new IStatementExecutor.Stats(), new IStatementExecutor.StatementProperties(), null, null, null,
+ new IStatementExecutor.Stats(), new IStatementExecutor.RequestStatementProperties(), null, null, null,
additionalParams, Collections.emptyMap(), false);
MetadataManager.INSTANCE.init();
IStatementExecutor translator =
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ClientInfoRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ClientInfoRequestMessage.java
index 2e53762..4230eda 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ClientInfoRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ClientInfoRequestMessage.java
@@ -56,7 +56,9 @@
CCMessageBroker messageBroker = (CCMessageBroker) appCtx.getServiceContext().getMessageBroker();
Optional<IClientRequest> clientRequestOpt = appCtx.getRequestTracker().getAsyncOrDeferredRequest(requestId);
ClientInfoResponseMessage response;
- if (clientRequestOpt.isEmpty() || !jobId.equals(((ClientRequest) clientRequestOpt.get()).getJobId())) {
+ // ownsJob, not the most recent job: a request that ran several statements holds a job per statement, and the
+ // handle being resolved may belong to any of them
+ if (clientRequestOpt.isEmpty() || !((ClientRequest) clientRequestOpt.get()).ownsJob(jobId)) {
response = new ClientInfoResponseMessage(ncReqId, false);
} else {
if (metricsRequired) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
index c3b79a9..51ab601 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
@@ -19,6 +19,10 @@
package org.apache.asterix.app.message;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.REFACTORED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
@@ -61,6 +65,7 @@
import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.control.cc.ClusterControllerService;
import org.apache.hyracks.util.LogRedactionUtil;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -141,6 +146,7 @@
}
@Override
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = REFACTORED)
public void handle(ICcApplicationContext ccAppCtx) throws HyracksDataException, InterruptedException {
ICCServiceContext ccSrvContext = ccAppCtx.getServiceContext();
ClusterControllerService ccSrv = (ClusterControllerService) ccSrvContext.getControllerService();
@@ -157,41 +163,38 @@
IStatementExecutorFactory statementExecutorFactory = ccApp.getStatementExecutorFactory();
ExecuteStatementResponseMessage responseMsg =
new ExecuteStatementResponseMessage(requestMessageId, clientContextID, requestReference.getUuid());
- final IStatementExecutor.StatementProperties statementProperties = new IStatementExecutor.StatementProperties();
- responseMsg.setStatementProperties(statementProperties);
+ final IStatementExecutor.RequestStatementProperties statementProperties =
+ new IStatementExecutor.RequestStatementProperties();
+ // these are declared outside the try, and transferred onto the response in the finally, so that a failure
+ // does not discard what already completed. Otherwise the output of the statements that did succeed is
+ // dropped and their result sets are orphaned on the CC until their TTL expires.
+ final List<Warning> warnings = new ArrayList<>();
+ final StringWriter outWriter = new StringWriter(256);
+ final PrintWriter outPrinter = new PrintWriter(outWriter);
+ final IStatementExecutor.ResultMetadata outMetadata = new IStatementExecutor.ResultMetadata();
+ final IStatementExecutor.Stats stats = new IStatementExecutor.Stats();
+ stats.setProfileType(profileType);
+ final long maxWarnings = sessionConfig.getMaxWarnings();
+ long parserTotalWarningsCount = 0;
+ IStatementExecutor translator = null;
try {
- List<Warning> warnings = new ArrayList<>();
IParser parser = compilationProvider.getParserFactory().createParser(statementsText);
List<Statement> statements = parser.parse();
- long maxWarnings = sessionConfig.getMaxWarnings();
parser.getWarnings(warnings, maxWarnings);
- long parserTotalWarningsCount = parser.getTotalWarningsCount();
- StringWriter outWriter = new StringWriter(256);
- PrintWriter outPrinter = new PrintWriter(outWriter);
+ parserTotalWarningsCount = parser.getTotalWarningsCount();
SessionOutput.ResultDecorator resultPrefix = ResultUtil.createPreResultDecorator();
SessionOutput.ResultDecorator resultPostfix = ResultUtil.createPostResultDecorator();
SessionOutput.ResultAppender appendHandle = ResultUtil.createResultHandleAppender(handleUrl);
SessionOutput.ResultAppender appendStatus = ResultUtil.createResultStatusAppender();
SessionOutput sessionOutput = new SessionOutput(sessionConfig, outPrinter, resultPrefix, resultPostfix,
appendHandle, appendStatus);
- IStatementExecutor.ResultMetadata outMetadata = new IStatementExecutor.ResultMetadata();
MetadataManager.INSTANCE.init();
- IStatementExecutor translator = statementExecutorFactory.create(ccAppCtx, statements, sessionOutput,
- compilationProvider, storageComponentProvider, new ResponsePrinter(sessionOutput));
- final IStatementExecutor.Stats stats = new IStatementExecutor.Stats();
- stats.setProfileType(profileType);
+ translator = statementExecutorFactory.create(ccAppCtx, statements, sessionOutput, compilationProvider,
+ storageComponentProvider, new ResponsePrinter(sessionOutput));
Map<String, IAObject> stmtParams = RequestParameters.deserializeParameterValues(statementParameters);
final IRequestParameters requestParameters =
createRequestParameters(statementProperties, stmtParams, outMetadata, stats);
translator.compileAndExecute(ccApp.getHcc(), requestParameters);
- translator.getWarnings(warnings, maxWarnings - warnings.size());
- stats.updateTotalWarningsCount(parserTotalWarningsCount);
- outPrinter.close();
- responseMsg.setResult(outWriter.toString());
- responseMsg.setMetadata(outMetadata);
- responseMsg.setStats(stats);
- responseMsg.setExecutionPlans(translator.getExecutionPlans());
- responseMsg.setWarnings(warnings);
} catch (AlgebricksException | HyracksException | org.apache.asterix.lang.sqlpp.parser.TokenMgrError pe) {
// we trust that "our" exceptions are serializable and have a comprehensible error message
GlobalConfig.ASTERIX_LOGGER.log(Level.WARN, pe.getMessage(), pe);
@@ -199,13 +202,26 @@
} catch (Exception e) {
GlobalConfig.ASTERIX_LOGGER.log(Level.ERROR, "Unexpected exception", e);
responseMsg.setError(e);
+ } finally {
+ // the properties of every statement travel with the result metadata, one entry per statement; there is no
+ // separate field holding a single statement's, which would only ever describe the last one to run
+ if (translator != null) {
+ translator.getWarnings(warnings, maxWarnings - warnings.size());
+ responseMsg.setExecutionPlans(translator.getExecutionPlans());
+ }
+ stats.updateTotalWarningsCount(parserTotalWarningsCount);
+ outPrinter.close();
+ responseMsg.setResult(outWriter.toString());
+ responseMsg.setMetadata(outMetadata);
+ responseMsg.setStats(stats);
+ responseMsg.setWarnings(warnings);
}
sendResponseToNc(messageBroker, responseMsg);
}
- protected IRequestParameters createRequestParameters(IStatementExecutor.StatementProperties statementProperties,
- Map<String, IAObject> stmtParams, IStatementExecutor.ResultMetadata outMetadata,
- IStatementExecutor.Stats stats) {
+ protected IRequestParameters createRequestParameters(
+ IStatementExecutor.RequestStatementProperties statementProperties, Map<String, IAObject> stmtParams,
+ IStatementExecutor.ResultMetadata outMetadata, IStatementExecutor.Stats stats) {
RequestParameters requestParameters = new RequestParameters(requestReference, statementsText, null,
resultProperties, stats, statementProperties, outMetadata, clientContextID, defaultDataverseName,
optionalParameters, stmtParams, multiStatement, statementCategoryRestrictionMask, forceDropDataset,
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementResponseMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementResponseMessage.java
index eaadebe..987d1bd 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementResponseMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementResponseMessage.java
@@ -32,7 +32,7 @@
public final class ExecuteStatementResponseMessage implements INcAddressedMessage {
- private static final long serialVersionUID = 2L;
+ private static final long serialVersionUID = 3L;
private final long requestMessageId;
private final String clientContextID;
@@ -44,8 +44,6 @@
private IStatementExecutor.Stats stats;
- private IStatementExecutor.StatementProperties statementProperties;
-
private Throwable error;
private ExecutionPlans executionPlans;
@@ -99,14 +97,6 @@
this.stats = stats;
}
- public IStatementExecutor.StatementProperties getStatementProperties() {
- return statementProperties;
- }
-
- public void setStatementProperties(IStatementExecutor.StatementProperties statementProperties) {
- this.statementProperties = statementProperties;
- }
-
public ExecutionPlans getExecutionPlans() {
return executionPlans;
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/fields/NcStatementsPrinter.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/fields/NcStatementsPrinter.java
new file mode 100644
index 0000000..12702b8
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/fields/NcStatementsPrinter.java
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.app.result.fields;
+
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
+import java.io.PrintWriter;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.asterix.api.http.server.AbstractQueryApiServlet.ResultStatus;
+import org.apache.asterix.api.http.server.ResultUtil;
+import org.apache.asterix.app.result.ExecutionError;
+import org.apache.asterix.app.result.ResponseMetrics;
+import org.apache.asterix.app.result.ResultHandle;
+import org.apache.asterix.app.result.ResultReader;
+import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.api.ICodedMessage;
+import org.apache.asterix.common.api.IResponseFieldPrinter;
+import org.apache.asterix.translator.ExecutionPlans;
+import org.apache.asterix.translator.IStatementExecutor.ResultDelivery;
+import org.apache.asterix.translator.IStatementExecutor.ResultSetInfo;
+import org.apache.asterix.translator.IStatementExecutor.StatementInfo;
+import org.apache.asterix.translator.IStatementExecutor.StatementProperties;
+import org.apache.asterix.translator.IStatementExecutor.Stats;
+import org.apache.asterix.translator.SessionOutput;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.result.IResultSet;
+import org.apache.hyracks.util.annotations.AiProvenance;
+
+/**
+ * Prints what each statement of a multi-statement request produced, as a {@value #FIELD_NAME} array holding one object
+ * per statement. The fields that describe an individual statement rather than the request as a whole live here: its
+ * signature, plans, rows, whether its plan came from the cache, its own metrics, and its error if it failed.
+ * <p>
+ * A request carrying a single statement is printed by the pre-existing path instead, so that its response is byte for
+ * byte what it was before multi-statement requests were accepted; see {@code NCQueryServiceServlet#executeStatement}.
+ * <p>
+ * Within a statement, {@code metrics} is written after {@code results} because the row count and size are only final
+ * once that statement's rows have been streamed. Any order placing them earlier would mean buffering all of the rows.
+ */
+@AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+public class NcStatementsPrinter implements IResponseFieldPrinter {
+
+ public static final String FIELD_NAME = "statements";
+ public static final String STATEMENT_FIELD_NAME = "statement";
+ public static final String KIND_FIELD_NAME = "kind";
+ /** Indent of a statement object's fields: the array is a top-level field, so its objects are one level deeper. */
+ private static final String FIELD_INDENT = "\t\t\t";
+
+ /** A single field of a statement object. Collected so that the separators between them cannot be got wrong. */
+ private interface FieldEmitter {
+ void emit(PrintWriter pw) throws HyracksDataException;
+ }
+
+ private final IApplicationContext appCtx;
+ private final List<StatementInfo> statements;
+ private final IResultSet resultSet;
+ private final ResultDelivery delivery;
+ private final SessionOutput sessionOutput;
+ private final Charset resultCharset;
+ private final boolean printSignature;
+ /** Included in each statement's handle; null when the deployment identifies a handle by job alone. */
+ private final String requestId;
+
+ public NcStatementsPrinter(IApplicationContext appCtx, List<StatementInfo> statements, IResultSet resultSet,
+ ResultDelivery delivery, SessionOutput sessionOutput, Charset resultCharset, boolean printSignature,
+ String requestId) {
+ this.appCtx = appCtx;
+ this.statements = statements;
+ this.resultSet = resultSet;
+ this.delivery = delivery;
+ this.sessionOutput = sessionOutput;
+ this.resultCharset = resultCharset;
+ this.printSignature = printSignature;
+ this.requestId = requestId;
+ }
+
+ @Override
+ public void print(PrintWriter pw) throws HyracksDataException {
+ pw.print("\t\"");
+ pw.print(FIELD_NAME);
+ pw.print("\": [\n");
+ boolean first = true;
+ for (StatementInfo statement : statements) {
+ // the request's synthetic dataverse decl is not a statement the client sent, so it gets no entry
+ if (statement.getProperties().isSynthetic()) {
+ continue;
+ }
+ if (!first) {
+ pw.print(",\n");
+ }
+ first = false;
+ printStatement(pw, statement);
+ }
+ pw.print("\n\t]");
+ }
+
+ private void printStatement(PrintWriter pw, StatementInfo statement) throws HyracksDataException {
+ StatementProperties properties = statement.getProperties();
+ Stats stats = statement.getStats() == null ? new Stats() : statement.getStats();
+ boolean failed = statement.getError() != null;
+ ResultSetInfo resultSetInfo = statement.getResultSet();
+ List<FieldEmitter> fields = new ArrayList<>();
+
+ fields.add(w -> printField(w, STATEMENT_FIELD_NAME, String.valueOf(properties.getIndex())));
+ String kind = properties.getKind() == null ? null : properties.getKind().getDisplayName();
+ fields.add(w -> printField(w, KIND_FIELD_NAME, kind == null ? "null" : quoted(kind)));
+ if (printSignature && statement.getSignature() != null) {
+ fields.add(new SignaturePrinter(statement.getSignature())::print);
+ }
+ if (resultSetInfo != null) {
+ if (delivery == ResultDelivery.IMMEDIATE) {
+ fields.add(w -> printRows(w, resultSetInfo, stats));
+ } else if (delivery == ResultDelivery.DEFERRED) {
+ // one handle per statement, each naming that statement's own job, so every result can be fetched
+ fields.add(new ResultHandlePrinter(sessionOutput,
+ new ResultHandle(resultSetInfo.getJobId(), resultSetInfo.getResultSetId(), requestId))::print);
+ }
+ }
+ ExecutionPlans plans = statement.getPlans();
+ if (plans != null) {
+ fields.add(new PlansPrinter(plans, sessionOutput.config().getPlanFormat())::print);
+ }
+ fields.add(w -> printField(w, CachedPlanPrinter.FIELD_NAME, String.valueOf(stats.isCachedPlan())));
+ ResultStatus status = failed ? ResultStatus.FATAL : ResultStatus.SUCCESS;
+ fields.add(w -> printField(w, StatusPrinter.FIELD_NAME, quoted(status.str())));
+ if (failed) {
+ fields.add(errorsPrinter(statement.getError())::print);
+ }
+ fields.add(new MetricsPrinter(statementMetrics(stats, failed), resultCharset)::print);
+ if (stats.getJobProfile() != null) {
+ fields.add(new ProfilePrinter(stats.getJobProfile())::print);
+ }
+
+ pw.print("\t\t{\n");
+ for (int i = 0; i < fields.size(); i++) {
+ fields.get(i).emit(pw);
+ if (i + 1 != fields.size()) {
+ pw.print(",\n");
+ }
+ }
+ pw.print("\n\t\t}");
+ }
+
+ /**
+ * Streams this statement's rows. The rows go straight to the writer: the CC has already run every statement of the
+ * request by the time this is reached, so the only thing left to transfer is the row data itself.
+ */
+ private void printRows(PrintWriter pw, ResultSetInfo resultSetInfo, Stats stats) throws HyracksDataException {
+ pw.print(FIELD_INDENT);
+ pw.print(quoted(ResultsPrinter.FIELD_NAME));
+ pw.print(": ");
+ ResultReader reader = new ResultReader(resultSet, resultSetInfo.getJobId(), resultSetInfo.getResultSetId());
+ // a session output with no result decorators: this printer supplies the field name itself, so the decorators
+ // that name the field (and number it once a request has several result sets) must not also run
+ SessionOutput undecorated = new SessionOutput(sessionOutput.config(), pw);
+ ResultUtil.printResults(appCtx, reader, undecorated, stats, resultSetInfo.getRecordType());
+ }
+
+ /**
+ * Builds the errors printer for a failed statement. Overridable so that a deployment which represents errors in
+ * its own form can do so here, as it already does for a request-level failure.
+ */
+ protected IResponseFieldPrinter errorsPrinter(Throwable error) {
+ List<ICodedMessage> errors = Collections.singletonList(ExecutionError.of(error));
+ return new ErrorsPrinter(errors);
+ }
+
+ /**
+ * The metrics of one statement. Elapsed and execution time are request-level measurements taken by the servlet
+ * around the whole request, so they are not reported per statement; compile and queue wait time are.
+ */
+ private static ResponseMetrics statementMetrics(Stats stats, boolean failed) {
+ return ResponseMetrics.of(0, 0, stats.getCount(), stats.getSize(), stats.getProcessedObjects(), failed ? 1 : 0,
+ stats.getTotalWarningsCount(), stats.getCompileTimeNanos(), stats.getQueueWaitTimeNanos(),
+ stats.getBufferCacheHitRatio(), stats.getBufferCachePageReadCount(), stats.getCloudReadRequestsCount(),
+ stats.getCloudPagesReadCount(), stats.getCloudPagesPersistedCount());
+ }
+
+ private static void printField(PrintWriter pw, String name, String rawValue) {
+ pw.print(FIELD_INDENT);
+ pw.print(quoted(name));
+ pw.print(": ");
+ pw.print(rawValue);
+ }
+
+ private static String quoted(String value) {
+ return '"' + value + '"';
+ }
+
+ @Override
+ public String getName() {
+ return FIELD_NAME;
+ }
+}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index c9ddc29..5089b09 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -25,6 +25,10 @@
import static org.apache.hyracks.api.job.HyracksJobProperty.JOB_KIND;
import static org.apache.hyracks.api.job.HyracksJobProperty.RESULT_TTL;
import static org.apache.hyracks.control.nc.result.ResultState.UNLIMITED_READS;
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.REFACTORED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
import java.io.FileInputStream;
import java.io.InputStream;
@@ -316,6 +320,7 @@
import org.apache.hyracks.storage.am.lsm.invertedindex.fulltext.TokenizerCategory;
import org.apache.hyracks.util.LogRedactionUtil;
import org.apache.hyracks.util.OptionalBoolean;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -351,6 +356,18 @@
protected final IQueryPlanCache queryPlanCache;
protected final IGlobalTxManager globalTxManager;
protected final INamespaceResolver namespaceResolver;
+ /**
+ * Running total of compile-time warnings already reported to {@link Stats}. {@link #warningCollector} accumulates
+ * over the whole request, so each statement must contribute only the difference since the previous one.
+ */
+ private long reportedCompileWarnings;
+ /**
+ * The statement this translator injected into {@link #statements} on the request's behalf (the request's dataverse
+ * declaration), held by identity so it can be excluded from the client-visible statement numbering.
+ */
+ private Statement syntheticStatement;
+ /** What the statement currently executing has produced; null outside the execution of any statement. */
+ private StatementInfo currentStatementInfo;
public QueryTranslator(ICcApplicationContext appCtx, List<Statement> statements, SessionOutput output,
ILangCompilationProvider compilationProvider, ExecutorService executorService,
@@ -401,17 +418,38 @@
final ResultDelivery resultDelivery = requestParameters.getResultProperties().getDelivery();
final long maxResultReads = requestParameters.getResultProperties().getMaxReads();
final Stats stats = requestParameters.getStats();
- final StatementProperties statementProperties = requestParameters.getStatementProperties();
+ final RequestStatementProperties statementProperties = requestParameters.getStatementProperties();
final ResultMetadata outMetadata = requestParameters.getOutMetadata();
final Map<String, IAObject> stmtParams = requestParameters.getStatementParameters();
warningCollector.setMaxWarnings(sessionConfig.getMaxWarnings());
- boolean hasResultSet = false;
+ final boolean multiStatementRequest = isMultiStatementRequest();
Exception exception = null;
+ int statementIndex = 0;
try {
for (Statement stmt : statements) {
if (sessionConfig.is(SessionConfig.FORMAT_HTML)) {
sessionOutput.out().println(ApiServlet.HTML_STATEMENT_SEPARATOR);
}
+ // a synthetic statement (the request's dataverse decl) is not one the client sent, so it does not
+ // consume an index
+ boolean synthetic = isSyntheticStatement(stmt, requestParameters);
+ StatementProperties stmtProperties = statementProperties.begin(synthetic ? -1 : statementIndex, stmt);
+ if (!synthetic) {
+ statementIndex++;
+ }
+ // what this statement produces, for a response that reports each statement separately. Kept even for
+ // a synthetic statement so that the recorded list lines up with the properties list.
+ currentStatementInfo = new StatementInfo(stmtProperties);
+ if (outMetadata != null) {
+ outMetadata.getStatements().add(currentStatementInfo);
+ }
+ final Stats statsBeforeStatement = stats.snapshot();
+ if (multiStatementRequest) {
+ // without this a statement that generates no plan of its own reports the plans of whichever
+ // earlier statement did. Only done for a multi-statement request: a trailing exempt statement
+ // (`select 1; set x='y';`) is legal today and clearing on it would blank the query's plans.
+ apiFramework.getExecutionPlans().reset();
+ }
validateOperation(appCtx, activeNamespace, stmt);
MetadataProvider metadataProvider = MetadataProvider.create(appCtx, activeNamespace);
configureMetadataProvider(metadataProvider, config, resultSetIdCounter, outputFile, requestParameters,
@@ -419,7 +457,6 @@
IStatementRewriter stmtRewriter = rewriterFactory.createStatementRewriter();
rewriteStatement(stmt, stmtRewriter, metadataProvider); // Rewrite the statement's AST.
Statement.Kind kind = stmt.getKind();
- statementProperties.setKind(kind);
switch (kind) {
case SET:
handleSetStatement(stmt, config);
@@ -603,8 +640,9 @@
handleDeclareFunctionStatement(metadataProvider, stmt);
break;
case EXTENSION:
+ // the name is recorded by RequestStatementProperties.begin, which sets it only for an
+ // extension statement so that it cannot leak onto a later statement of the same request
final ExtensionStatement extStmt = (ExtensionStatement) stmt;
- statementProperties.setName(extStmt.getName());
if (!isCompileOnly()) {
extStmt.handle(hcc, this, requestParameters, metadataProvider,
resultSetIdCounter.getAndInc());
@@ -622,7 +660,8 @@
throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE, stmt.getSourceLocation(),
"Unexpected statement: " + kind);
}
- hasResultSet |= metadataProvider.getResultSetId() != null;
+ stmtProperties.setExecuted(true);
+ recordStatementOutcome(currentStatementInfo, statsBeforeStatement, stats);
if (shouldInvalidateQueryPlanCache(stmt)) {
queryPlanCache.clear();
}
@@ -630,15 +669,22 @@
} catch (Exception ex) {
// need to handle job and request clean-ups in case of failures
this.appCtx.getRequestTracker().incrementFailedRequests();
+ // attribute the failure to the statement that was executing, so the response and the audit record can
+ // name which of the request's statements failed
+ statementProperties.last().setFailed(true);
+ if (currentStatementInfo != null) {
+ currentStatementInfo.setError(ex);
+ }
exception = ex;
throw ex;
} finally {
+ currentStatementInfo = null;
// async queries are completed after their job completes
if (ResultDelivery.ASYNC != resultDelivery) {
- // this assumes 1-1 mapping between a request and a statement, needs to be adapted for multi-statement
+ // one request completes once, after all of its statements have run
appCtx.getRequestTracker().complete(reqId);
if (trackInAsyncDeferredRequests) {
- completeDeferred((ClientRequest) clientRequest, hasResultSet, exception);
+ completeDeferred((ClientRequest) clientRequest, outMetadata, exception);
}
} else {
if (statements.isEmpty()) {
@@ -655,22 +701,43 @@
}
}
- private void completeDeferred(ClientRequest clientRequest, boolean hasResultSet, Exception exception) {
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = REFACTORED, notes = "Sweep per job rather than on one flag OR'd over the whole request")
+ private void completeDeferred(ClientRequest clientRequest, ResultMetadata outMetadata, Exception exception) {
try {
- JobId jobId = clientRequest.getJobId();
- if (jobId == null) {
- // jobId = null either means:
+ Set<JobId> jobs = clientRequest.getJobIds();
+ if (jobs.isEmpty()) {
+ // no job was created at all, which means either:
// 1. compile error, compile-only, ... resulting in no job created whether query, DML or DDL
// 2. statement currently not setting jobId in the client request, e.g. DDLs
// for 2. it needs to be handled so that the job record is removed also if not producing a result
appCtx.getRequestTracker().removeAsyncOrDeferredRequest(clientRequest.getId());
- } else if (!hasResultSet) {
+ return;
+ }
+ // the jobs whose results the client was handed a handle to; every other job of the request has nothing
+ // left to fetch. Deciding this per job matters once a request runs several statements: a single flag
+ // OR'd over all of them retains the job records of the statements that produced nothing.
+ Set<JobId> retained = new HashSet<>();
+ if (outMetadata != null) {
+ for (ResultSetInfo resultSetInfo : outMetadata.getResultSets()) {
+ retained.add(resultSetInfo.getJobId());
+ }
+ }
+ // only these have to be swept before the request stops being tracked; a job registering no result is
+ // never swept, so awaiting every job of the request would keep it tracked indefinitely
+ clientRequest.setFetchableJobs(retained);
+ ClusterControllerService ccSvs =
+ (ClusterControllerService) appCtx.getServiceContext().getControllerService();
+ for (JobId job : jobs) {
// don't sweep ones that completed successfully and produced a result
// sweeps statements not producing a result, e.g. DMLs without a return clause
// sweeps statements that threw an exception whether query, DML or DDL
- ClusterControllerService ccSvs =
- (ClusterControllerService) appCtx.getServiceContext().getControllerService();
- ccSvs.getResultDirectoryService().removeIfDone(jobId);
+ if (!retained.contains(job)) {
+ ccSvs.getResultDirectoryService().removeIfDone(job);
+ }
+ }
+ if (retained.isEmpty()) {
+ // nothing the client can come back for
+ appCtx.getRequestTracker().removeAsyncOrDeferredRequest(clientRequest.getId());
}
} catch (Throwable th) {
if (exception != null) {
@@ -4616,11 +4683,10 @@
final JobSpecification jobSpec = apiFramework.compileQuery(hcc, metadataProvider, copyTo.getQuery(),
rewrittenResult.second, null, sessionOutput, compiledCopyToStatement, externalVars,
responsePrinter, warningCollector, requestParameters, jobFlags, null);
- // update stats with count of compile-time warnings. needs to be adapted for multi-statement.
- stats.updateTotalWarningsCount(warningCollector.getTotalWarningsCount());
+ updateCompileWarningsCount(stats);
afterCompile();
MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
- stats.setCompileTimeNanos(System.nanoTime() - compileStart);
+ stats.addCompileTimeNanos(System.nanoTime() - compileStart);
clientRequest.setCompileTimeNanos(stats.getCompileTimeNanos());
bActiveTxn = false;
return isCompileOnly() ? null : jobSpec;
@@ -5865,13 +5931,14 @@
}
}
- // update stats with count of compile-time warnings. needs to be adapted for multi-statement.
- stats.updateTotalWarningsCount(warningCollector.getTotalWarningsCount());
+ updateCompileWarningsCount(stats);
afterCompile();
MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
- stats.setCompileTimeNanos(System.nanoTime() - compileStart);
+ // accumulate into the request total, but report this statement's own time to the result metadata
+ long statementCompileTime = System.nanoTime() - compileStart;
+ stats.addCompileTimeNanos(statementCompileTime);
clientRequest.setCompileTimeNanos(stats.getCompileTimeNanos());
- resultMetadata.setCompileTimeNanos(stats.getCompileTimeNanos());
+ resultMetadata.setCompileTimeNanos(statementCompileTime);
bActiveTxn = false;
return query.isExplain() || query.isAdvise() || isCompileOnly() ? null : jobSpec;
} catch (Exception e) {
@@ -6046,8 +6113,14 @@
}
responsePrinter.printResults();
if (outMetadata != null) {
- outMetadata.getResultSets()
- .add(new ResultSetInfo(id, resultSetId, metadataProvider.findOutputRecordType()));
+ ResultSetInfo resultSetInfo =
+ new ResultSetInfo(id, resultSetId, metadataProvider.findOutputRecordType());
+ outMetadata.getResultSets().add(resultSetInfo);
+ // also attach it to the statement that produced it: the flat list above cannot say which
+ // statement a result set came from, since statements producing none leave no entry in it
+ if (currentStatementInfo != null) {
+ currentStatementInfo.setResultSet(resultSetInfo);
+ }
}
}, requestParameters, cancellable, appCtx, metadataProvider, atomicStmt, jobKind);
break;
@@ -6063,13 +6136,17 @@
org.apache.asterix.translator.ResultMetadata resultMetadata =
(org.apache.asterix.translator.ResultMetadata) controllerService.getResultDirectoryService()
.getResultMetadata(jobId, rsId);
- stats.setProcessedObjects(resultMetadata.getProcessedObjects());
- stats.setQueueWaitTimeNanos(resultMetadata.getQueueWaitTimeNanos());
+ // accumulate rather than overwrite: a request runs one job per statement, and these are request totals.
+ // With a single statement, adding to zero is the same as setting, so this is unchanged for such a request.
+ stats.addProcessedObjects(resultMetadata.getProcessedObjects());
+ stats.addQueueWaitTimeNanos(resultMetadata.getQueueWaitTimeNanos());
+ stats.addBufferCachePageReadCount(resultMetadata.getBufferCachePageReadCount());
+ stats.addCloudReadRequestsCount(resultMetadata.getCloudReadRequestsCount());
+ stats.addCloudPagesReadCount(resultMetadata.getCloudPagesReadCount());
+ stats.addCloudPagesPersistedCount(resultMetadata.getCloudPagesPersistedCount());
+ // a ratio, not a counter: still the most recent job's. Exact per statement (each statement runs one job);
+ // for a multi-statement request the request-level figure describes the last statement only.
stats.setBufferCacheHitRatio(resultMetadata.getBufferCacheHitRatio());
- stats.setBufferCachePageReadCount(resultMetadata.getBufferCachePageReadCount());
- stats.setCloudReadRequestsCount(resultMetadata.getCloudReadRequestsCount());
- stats.setCloudPagesReadCount(resultMetadata.getCloudPagesReadCount());
- stats.setCloudPagesPersistedCount(resultMetadata.getCloudPagesPersistedCount());
if (jobFlags.contains(JobFlag.PROFILE_RUNTIME)) {
stats.setJobProfile(resultMetadata.getJobProfile());
apiFramework.generateOptimizedLogicalPlanWithProfile(resultMetadata.getJobProfile());
@@ -6382,6 +6459,34 @@
}
}
+ /**
+ * Records what the statement that just finished produced: its own share of the request's stats, and a snapshot of
+ * its plans and signature taken before the next statement overwrites the shared instance.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ private void recordStatementOutcome(StatementInfo statementInfo, Stats before, Stats after) {
+ if (statementInfo == null) {
+ return;
+ }
+ statementInfo.setStats(after.since(before));
+ ExecutionPlans plans = apiFramework.getExecutionPlans();
+ statementInfo.setPlans(new ExecutionPlans(plans));
+ statementInfo.setSignature(plans.getSignature());
+ }
+
+ /**
+ * Contributes the compile-time warnings of the statement that just compiled to the request stats.
+ * {@link #warningCollector} holds a running total for the whole request while
+ * {@link Stats#updateTotalWarningsCount(long)} adds a delta, so reporting the total verbatim once per statement
+ * would count the warnings of statement <i>k</i> a further <i>N - k</i> times.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ private void updateCompileWarningsCount(Stats stats) {
+ long total = warningCollector.getTotalWarningsCount();
+ stats.updateTotalWarningsCount(total - reportedCompileWarnings);
+ reportedCompileWarnings = total;
+ }
+
protected IClientRequest trackRequest(IRequestParameters requestParameters, boolean trackInAsyncDeferredRequests)
throws HyracksDataException {
final IClientRequest clientRequest = appCtx.getReceptionist().requestReceived(requestParameters);
@@ -6412,13 +6517,55 @@
DataverseDecl requestDataverseDecl = getRequestDataverseDecl(requestParameters);
if (requestDataverseDecl != null) {
statements.add(0, requestDataverseDecl);
+ syntheticStatement = requestDataverseDecl;
}
validateAsyncSupported(requestParameters);
+ validateDeliveryModeSupportsMultiStatement(requestParameters);
}
+ /**
+ * Asynchronous delivery accepts a single statement only. It returns a handle as soon as a job has been submitted
+ * rather than once it has finished, so the statements of a request would overlap rather than run in order, which
+ * leaves a request such as {@code CREATE ...; INSERT ...;} without a defined meaning. Deferred delivery does not
+ * have that problem: it runs each statement to completion, so a request's statements still run in order, and each
+ * one's handle is returned for the client to fetch when it chooses.
+ * <p>
+ * This tests the delivery the client asked for, not this hop's: an NC substitutes DEFERRED for IMMEDIATE when it
+ * forwards a request here because it severs the result itself, and such a request has no handle for a client to
+ * hold, so the restriction does not apply to it.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ private void validateDeliveryModeSupportsMultiStatement(IRequestParameters requestParameters)
+ throws CompilationException {
+ ResultDelivery delivery = requestParameters.getResultProperties().getRequestedDelivery();
+ if (delivery == ResultDelivery.ASYNC && isMultiStatementRequest()) {
+ throw new CompilationException(ErrorCode.UNSUPPORTED_MULTIPLE_STATEMENTS_FOR_DELIVERY_MODE,
+ delivery.getName());
+ }
+ }
+
+ /**
+ * @return whether this request carries more than one statement that the single-statement restriction would have
+ * rejected. Deliberately the exact complement of the {@link ErrorCode#UNSUPPORTED_MULTIPLE_STATEMENTS}
+ * predicate, so that every request which is accepted under that restriction is treated as single here.
+ */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ protected boolean isMultiStatementRequest() {
+ return statements.stream().filter(QueryTranslator::isNotAllowedMultiStatement).count() > 1;
+ }
+
+ /** @return whether {@code statement} was injected by the engine rather than supplied by the client. */
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = GENERATED)
+ protected boolean isSyntheticStatement(Statement statement, IRequestParameters requestParameters) {
+ return statement == syntheticStatement;
+ }
+
+ @AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = REFACTORED)
private void validateAsyncSupported(IRequestParameters requestParameters) throws CompilationException {
if (requestParameters.getResultProperties().getDelivery() == ResultDelivery.ASYNC) {
- Optional<Statement> stmsOpt = statements.stream().findAny().filter(s -> !s.getKind().isSupportAsync());
+ // filter before findAny: findAny() first would collapse the stream to the leading statement, which is
+ // the synthetic request DataverseDecl whenever a default dataverse/query_context is set
+ Optional<Statement> stmsOpt = statements.stream().filter(s -> !s.getKind().isSupportAsync()).findAny();
if (stmsOpt.isPresent()) {
Statement stmt = stmsOpt.get();
throw new CompilationException(ErrorCode.ASYNC_NOT_SUPPORTED_FOR_STATEMENT, stmt.getSourceLocation(),
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/RequestParameters.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/RequestParameters.java
index f278e2e..33e9114 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/RequestParameters.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/RequestParameters.java
@@ -33,7 +33,7 @@
import org.apache.asterix.om.types.BuiltinType;
import org.apache.asterix.translator.IRequestParameters;
import org.apache.asterix.translator.IStatementExecutor;
-import org.apache.asterix.translator.IStatementExecutor.StatementProperties;
+import org.apache.asterix.translator.IStatementExecutor.RequestStatementProperties;
import org.apache.asterix.translator.IStatementExecutor.Stats;
import org.apache.asterix.translator.ResultProperties;
import org.apache.asterix.translator.SessionConfig;
@@ -53,7 +53,7 @@
private final IResultSet resultSet;
private final ResultProperties resultProperties;
private final Stats stats;
- private final StatementProperties statementProperties;
+ private final RequestStatementProperties statementProperties;
private final Map<String, String> optionalParameters;
private final IStatementExecutor.ResultMetadata outMetadata;
private final String clientContextId;
@@ -69,7 +69,7 @@
private boolean skipQueryPlanCache;
public RequestParameters(IRequestReference requestReference, String statement, IResultSet resultSet,
- ResultProperties resultProperties, Stats stats, StatementProperties statementProperties,
+ ResultProperties resultProperties, Stats stats, RequestStatementProperties statementProperties,
IStatementExecutor.ResultMetadata outMetadata, String clientContextId, String defaultDataverseName,
Map<String, String> optionalParameters, Map<String, IAObject> statementParameters, boolean multiStatement) {
this(requestReference, statement, resultSet, resultProperties, stats, statementProperties, outMetadata,
@@ -78,7 +78,7 @@
}
public RequestParameters(IRequestReference requestReference, String statement, IResultSet resultSet,
- ResultProperties resultProperties, Stats stats, StatementProperties statementProperties,
+ ResultProperties resultProperties, Stats stats, RequestStatementProperties statementProperties,
IStatementExecutor.ResultMetadata outMetadata, String clientContextId, String defaultDataverseName,
Map<String, String> optionalParameters, Map<String, IAObject> statementParameters, boolean multiStatement,
int statementCategoryRestrictionMask) {
@@ -88,7 +88,7 @@
}
public RequestParameters(IRequestReference requestReference, String statement, IResultSet resultSet,
- ResultProperties resultProperties, Stats stats, StatementProperties statementProperties,
+ ResultProperties resultProperties, Stats stats, RequestStatementProperties statementProperties,
IStatementExecutor.ResultMetadata outMetadata, String clientContextId, String defaultDataverseName,
Map<String, String> optionalParameters, Map<String, IAObject> statementParameters, boolean multiStatement,
int statementCategoryRestrictionMask, boolean forceDropDataset) {
@@ -98,7 +98,7 @@
}
public RequestParameters(IRequestReference requestReference, String statement, IResultSet resultSet,
- ResultProperties resultProperties, Stats stats, StatementProperties statementProperties,
+ ResultProperties resultProperties, Stats stats, RequestStatementProperties statementProperties,
IStatementExecutor.ResultMetadata outMetadata, String clientContextId, String defaultDataverseName,
Map<String, String> optionalParameters, Map<String, IAObject> statementParameters, boolean multiStatement,
int statementCategoryRestrictionMask, boolean forceDropDataset, boolean skipAdmissionPolicy) {
@@ -135,7 +135,7 @@
}
@Override
- public StatementProperties getStatementProperties() {
+ public RequestStatementProperties getStatementProperties() {
return statementProperties;
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/server/StatementPropertiesTransferTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/server/StatementPropertiesTransferTest.java
new file mode 100644
index 0000000..8259748
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/http/server/StatementPropertiesTransferTest.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.api.http.server;
+
+import static org.apache.hyracks.util.annotations.AiProvenance.Agent.CLAUDE_OPUS_5;
+import static org.apache.hyracks.util.annotations.AiProvenance.ContributionKind.TEST_GENERATED;
+import static org.apache.hyracks.util.annotations.AiProvenance.Tool.CLAUDE_CODE_UI;
+
+import java.util.List;
+
+import org.apache.asterix.app.message.ExecuteStatementResponseMessage;
+import org.apache.asterix.lang.common.base.Statement;
+import org.apache.asterix.translator.IStatementExecutor.RequestStatementProperties;
+import org.apache.asterix.translator.IStatementExecutor.ResultMetadata;
+import org.apache.asterix.translator.IStatementExecutor.StatementInfo;
+import org.apache.asterix.translator.IStatementExecutor.StatementProperties;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * The CC is the only side of a request that parses it, so the properties of the request's statements only reach the NC
+ * if the response carries all of them. Everything the NC derives per statement -- auditing above all -- is only per
+ * statement if that transfer is complete, and a response carrying a single statement's properties silently reduces a
+ * request of any length to one statement.
+ */
+@AiProvenance(agent = CLAUDE_OPUS_5, tool = CLAUDE_CODE_UI, contributionKind = TEST_GENERATED)
+public class StatementPropertiesTransferTest {
+
+ private static ExecuteStatementResponseMessage responseWith(Statement.Kind... kinds) {
+ ExecuteStatementResponseMessage msg = new ExecuteStatementResponseMessage(1L, "ctx", "uuid");
+ ResultMetadata metadata = new ResultMetadata();
+ int index = 0;
+ for (Statement.Kind kind : kinds) {
+ StatementProperties properties = new StatementProperties();
+ properties.setKind(kind);
+ properties.setIndex(index++);
+ metadata.getStatements().add(new StatementInfo(properties));
+ }
+ msg.setMetadata(metadata);
+ return msg;
+ }
+
+ @Test
+ public void everyStatementOfTheRequestReachesTheNc() {
+ ExecuteStatementResponseMessage msg =
+ responseWith(Statement.Kind.CREATE_DATASET, Statement.Kind.INSERT, Statement.Kind.QUERY);
+ RequestStatementProperties received = new RequestStatementProperties();
+
+ NCQueryServiceServlet.updatePropertiesFromCC(received, msg);
+
+ List<StatementProperties> statements = received.userStatements();
+ Assert.assertEquals("every statement the CC ran should reach the NC", 3, statements.size());
+ Assert.assertEquals(Statement.Kind.CREATE_DATASET, statements.get(0).getKind());
+ Assert.assertEquals(Statement.Kind.INSERT, statements.get(1).getKind());
+ Assert.assertEquals(Statement.Kind.QUERY, statements.get(2).getKind());
+ }
+
+ /**
+ * The failure this guards against: carrying one statement's properties leaves the NC unable to tell a request of
+ * several statements from a request of one, so it attributes the whole request to whichever statement came last.
+ */
+ @Test
+ public void severalStatementsAreNotReducedToTheLastOne() {
+ ExecuteStatementResponseMessage msg = responseWith(Statement.Kind.CREATE_DATASET, Statement.Kind.QUERY);
+ RequestStatementProperties received = new RequestStatementProperties();
+
+ NCQueryServiceServlet.updatePropertiesFromCC(received, msg);
+
+ Assert.assertTrue("a request of two statements must not look like a request of one",
+ received.userStatements().size() > 1);
+ Assert.assertEquals("the first statement must not be lost", Statement.Kind.CREATE_DATASET,
+ received.userStatements().get(0).getKind());
+ }
+
+ @Test
+ public void statementOrderIsPreserved() {
+ ExecuteStatementResponseMessage msg =
+ responseWith(Statement.Kind.QUERY, Statement.Kind.INSERT, Statement.Kind.DELETE);
+ RequestStatementProperties received = new RequestStatementProperties();
+
+ NCQueryServiceServlet.updatePropertiesFromCC(received, msg);
+
+ List<StatementProperties> statements = received.userStatements();
+ for (int i = 0; i < statements.size(); i++) {
+ Assert.assertEquals("statement " + i + " is out of order", i, statements.get(i).getIndex());
+ }
+ }
+
+ @Test
+ public void aSingleStatementStillArrives() {
+ ExecuteStatementResponseMessage msg = responseWith(Statement.Kind.QUERY);
+ RequestStatementProperties received = new RequestStatementProperties();
+
+ NCQueryServiceServlet.updatePropertiesFromCC(received, msg);
+
+ Assert.assertEquals(1, received.userStatements().size());
+ Assert.assertEquals(Statement.Kind.QUERY, received.last().getKind());
+ }
+
+ /** A request failing before any statement was dispatched reports none, rather than failing to be reported. */
+ @Test
+ public void noStatementsIsTolerated() {
+ ExecuteStatementResponseMessage msg = responseWith();
+ RequestStatementProperties received = new RequestStatementProperties();
+
+ NCQueryServiceServlet.updatePropertiesFromCC(received, msg);
+
+ Assert.assertTrue(received.userStatements().isEmpty());
+ }
+
+ /** The metadata is absent when the CC could not get far enough to produce any. */
+ @Test
+ public void absentMetadataIsTolerated() {
+ ExecuteStatementResponseMessage msg = new ExecuteStatementResponseMessage(1L, "ctx", "uuid");
+ RequestStatementProperties received = new RequestStatementProperties();
+
+ NCQueryServiceServlet.updatePropertiesFromCC(received, msg);
+
+ Assert.assertTrue(received.userStatements().isEmpty());
+ }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
index 017f103..b57019f 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ResultExtractor.java
@@ -110,6 +110,8 @@
private enum ResultField {
RESULTS("results"),
+ /** A request reported statement by statement; each element carries that statement's own {@code results}. */
+ STATEMENTS("statements"),
REQUEST_ID("requestID"),
METRICS("metrics"),
PROFILE("profile"),
@@ -154,12 +156,13 @@
public static ExtractedResult extract(InputStream resultStream, Charset resultCharset, OutputFormat outputFormat)
throws Exception {
- return extract(resultStream, EnumSet.of(ResultField.RESULTS, ResultField.WARNINGS), resultCharset, outputFormat,
- null);
+ return extract(resultStream, EnumSet.of(ResultField.RESULTS, ResultField.STATEMENTS, ResultField.WARNINGS),
+ resultCharset, outputFormat, null);
}
public static ExtractedResult extract(InputStream resultStream, Charset resultCharset) throws Exception {
- return extract(resultStream, EnumSet.of(ResultField.RESULTS, ResultField.WARNINGS), resultCharset);
+ return extract(resultStream, EnumSet.of(ResultField.RESULTS, ResultField.STATEMENTS, ResultField.WARNINGS),
+ resultCharset);
}
public static InputStream extractMetrics(InputStream resultStream, Charset resultCharset) throws Exception {
@@ -207,6 +210,40 @@
return null;
}
+ /** Appends the rows of one {@code results} value, in the form a test's expected output is written in. */
+ private static void appendResults(StringBuilder resultBuilder, JsonNode fieldValue, boolean isJsonFormat)
+ throws JsonProcessingException {
+ if (fieldValue.size() <= 1) {
+ if (fieldValue.size() == 0) {
+ resultBuilder.append("");
+ } else if (fieldValue.isArray()) {
+ JsonNode oneElement = fieldValue.get(0);
+ if (oneElement.isTextual()) {
+ resultBuilder.append(isJsonFormat ? prettyPrint(oneElement) : oneElement.asText());
+ } else {
+ resultBuilder.append(prettyPrint(oneElement));
+ }
+ } else {
+ resultBuilder.append(prettyPrint(fieldValue));
+ }
+ } else {
+ JsonNode[] fields = Iterators.toArray(fieldValue.elements(), JsonNode.class);
+ if (isJsonFormat) {
+ for (JsonNode f : fields) {
+ resultBuilder.append(prettyPrint(f)).append('\n');
+ }
+ } else {
+ for (JsonNode f : fields) {
+ if (f.isValueNode()) {
+ resultBuilder.append(f.asText()).append('\n');
+ } else {
+ resultBuilder.append(prettyPrint(f)).append('\n');
+ }
+ }
+ }
+ }
+ }
+
private static ExtractedResult extract(InputStream resultStream, EnumSet<ResultField> resultFields,
Charset resultCharset) throws Exception {
return extract(resultStream, resultFields, resultCharset, OutputFormat.ADM, null);
@@ -245,40 +282,27 @@
}
final JsonNode fieldValue = result.get(fieldName);
switch (fieldKind) {
+ case STATEMENTS:
+ // a statement-wise response: concatenate what each statement produced, in order, so that a test
+ // comparing rows sees the same thing it would from a request-level "results"
+ for (JsonNode statement : fieldValue) {
+ JsonNode statementResults = statement.get(ResultField.RESULTS.getFieldName());
+ if (statementResults == null) {
+ continue;
+ }
+ if (!firstResult) {
+ resultBuilder.append('\n');
+ }
+ firstResult = false;
+ appendResults(resultBuilder, statementResults, isJsonFormat);
+ }
+ break;
case RESULTS:
if (!firstResult) {
resultBuilder.append('\n');
}
firstResult = false;
- if (fieldValue.size() <= 1) {
- if (fieldValue.size() == 0) {
- resultBuilder.append("");
- } else if (fieldValue.isArray()) {
- JsonNode oneElement = fieldValue.get(0);
- if (oneElement.isTextual()) {
- resultBuilder.append(isJsonFormat ? prettyPrint(oneElement) : oneElement.asText());
- } else {
- resultBuilder.append(prettyPrint(oneElement));
- }
- } else {
- resultBuilder.append(prettyPrint(fieldValue));
- }
- } else {
- JsonNode[] fields = Iterators.toArray(fieldValue.elements(), JsonNode.class);
- if (isJsonFormat) {
- for (JsonNode f : fields) {
- resultBuilder.append(prettyPrint(f)).append('\n');
- }
- } else {
- for (JsonNode f : fields) {
- if (f.isValueNode()) {
- resultBuilder.append(f.asText()).append('\n');
- } else {
- resultBuilder.append(prettyPrint(f)).append('\n');
- }
- }
- }
- }
+ appendResults(resultBuilder, fieldValue, isJsonFormat);
break;
case REQUEST_ID:
case METRICS:
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/result/NcStatementsPrinterTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/result/NcStatementsPrinterTest.java
new file mode 100644
index 0000000..a157c11
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/result/NcStatementsPrinterTest.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.test.result;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.asterix.api.http.server.ResultUtil;
+import org.apache.asterix.app.result.fields.NcStatementsPrinter;
+import org.apache.asterix.lang.common.base.Statement;
+import org.apache.asterix.translator.ExecutionPlans;
+import org.apache.asterix.translator.IStatementExecutor.ResultDelivery;
+import org.apache.asterix.translator.IStatementExecutor.ResultSetInfo;
+import org.apache.asterix.translator.IStatementExecutor.StatementInfo;
+import org.apache.asterix.translator.IStatementExecutor.StatementProperties;
+import org.apache.asterix.translator.IStatementExecutor.Stats;
+import org.apache.asterix.translator.SessionConfig;
+import org.apache.asterix.translator.SessionOutput;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.result.ResultSetId;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.Test;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * Covers the shape of a statement-wise response. The cases here need no cluster: they are about whether the printer
+ * produces well-formed JSON with each statement's fields in the right place, which is the part where a stray or missing
+ * separator would corrupt every response.
+ */
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED)
+public class NcStatementsPrinterTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static StatementInfo statement(int index, Statement.Kind kind, boolean synthetic, Throwable error) {
+ StatementProperties properties = new StatementProperties();
+ properties.setIndex(index);
+ properties.setKind(kind);
+ properties.setSynthetic(synthetic);
+ StatementInfo info = new StatementInfo(properties);
+ info.setStats(new Stats());
+ info.setPlans(new ExecutionPlans());
+ info.setError(error);
+ return info;
+ }
+
+ /** Prints the statements and parses the result as the single field of an object, as it appears in a response. */
+ private static JsonNode print(List<StatementInfo> statements) throws Exception {
+ return print(statements, ResultDelivery.IMMEDIATE);
+ }
+
+ private static JsonNode print(List<StatementInfo> statements, ResultDelivery delivery) throws Exception {
+ StringWriter out = new StringWriter();
+ PrintWriter pw = new PrintWriter(out);
+ SessionOutput sessionOutput = new SessionOutput(new SessionConfig(SessionConfig.OutputFormat.CLEAN_JSON), pw);
+ sessionOutput.setHandleAppender(ResultUtil.createResultHandleAppender("/analytics/service/result/"));
+ NcStatementsPrinter printer = new NcStatementsPrinter(null, statements, null, delivery, sessionOutput,
+ StandardCharsets.UTF_8, false, "req-1");
+ printer.print(pw);
+ pw.flush();
+ return MAPPER.readTree("{\n" + out + "\n}");
+ }
+
+ @Test
+ public void printsAnObjectPerStatementInOrder() throws Exception {
+ JsonNode response = print(Arrays.asList(statement(0, Statement.Kind.CREATE_DATASET, false, null),
+ statement(1, Statement.Kind.QUERY, false, null), statement(2, Statement.Kind.QUERY, false, null)));
+
+ JsonNode statements = response.get("statements");
+ assertTrue(statements.isArray());
+ assertEquals(3, statements.size());
+ for (int i = 0; i < 3; i++) {
+ assertEquals(i, statements.get(i).get("statement").asInt());
+ }
+ assertEquals(Statement.Kind.CREATE_DATASET.getDisplayName(), statements.get(0).get("kind").asText());
+ assertEquals(Statement.Kind.QUERY.getDisplayName(), statements.get(1).get("kind").asText());
+ }
+
+ @Test
+ public void reportsEachStatementsOwnStatusAndMetrics() throws Exception {
+ JsonNode statements = print(Arrays.asList(statement(0, Statement.Kind.QUERY, false, null),
+ statement(1, Statement.Kind.QUERY, false, new IllegalStateException("boom")))).get("statements");
+
+ assertEquals("success", statements.get(0).get("status").asText());
+ assertFalse(statements.get(0).has("errors"));
+ assertTrue(statements.get(0).has("metrics"));
+
+ assertEquals("fatal", statements.get(1).get("status").asText());
+ assertTrue(statements.get(1).has("errors"));
+ assertEquals(1, statements.get(1).get("metrics").get("errorCount").asInt());
+ }
+
+ @Test
+ public void reportsCachedPlanPerStatement() throws Exception {
+ StatementInfo cached = statement(0, Statement.Kind.QUERY, false, null);
+ cached.getStats().setCachedPlan(true);
+ StatementInfo notCached = statement(1, Statement.Kind.QUERY, false, null);
+
+ JsonNode statements = print(Arrays.asList(cached, notCached)).get("statements");
+ assertTrue(statements.get(0).get("cachedPlan").asBoolean());
+ assertFalse(statements.get(1).get("cachedPlan").asBoolean());
+ }
+
+ @Test
+ public void omitsTheStatementsTheEngineInjected() throws Exception {
+ JsonNode statements = print(Arrays.asList(statement(-1, Statement.Kind.DATAVERSE_DECL, true, null),
+ statement(0, Statement.Kind.QUERY, false, null), statement(1, Statement.Kind.QUERY, false, null)))
+ .get("statements");
+
+ assertEquals(2, statements.size());
+ assertEquals(0, statements.get(0).get("statement").asInt());
+ assertEquals(1, statements.get(1).get("statement").asInt());
+ }
+
+ @Test
+ public void printsWellFormedJsonForASingleEntry() throws Exception {
+ JsonNode statements = print(Arrays.asList(statement(0, Statement.Kind.QUERY, false, null))).get("statements");
+ assertEquals(1, statements.size());
+ }
+
+ /**
+ * Deferred delivery returns a handle per statement, each naming that statement's own job, so that a request that
+ * ran several statements does not leave all but one of its results unreachable.
+ */
+ @Test
+ public void givesEachStatementItsOwnDeferredHandle() throws Exception {
+ StatementInfo first = statement(0, Statement.Kind.QUERY, false, null);
+ first.setResultSet(new ResultSetInfo(new JobId(11), new ResultSetId(0), null));
+ StatementInfo second = statement(1, Statement.Kind.QUERY, false, null);
+ second.setResultSet(new ResultSetInfo(new JobId(12), new ResultSetId(0), null));
+
+ JsonNode statements = print(Arrays.asList(first, second), ResultDelivery.DEFERRED).get("statements");
+
+ assertEquals("/analytics/service/result/req-1/11-0", statements.get(0).get("handle").asText());
+ assertEquals("/analytics/service/result/req-1/12-0", statements.get(1).get("handle").asText());
+ }
+
+ @Test
+ public void printsWellFormedJsonWhenNothingRan() throws Exception {
+ JsonNode statements = print(java.util.Collections.emptyList()).get("statements");
+ assertTrue(statements.isArray());
+ assertEquals(0, statements.size());
+ }
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.1.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.1.query.sqlpp
new file mode 100644
index 0000000..3530c71
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.1.query.sqlpp
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*
+ * Description : parse-only over a request whose statements are not all queries. Only a query carries a body to
+ * : collect statement parameters from; a DDL contributes none. Casting every parsed statement to
+ * : Query used to raise a ClassCastException here, surfacing as a 500.
+ * Expected Res : Success - the parameters of the query statement only
+ */
+
+-- param parse-only:string=true
+
+create collection c1 primary key (id: int);
+select count(*) from c1 where name=$qname;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.2.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.2.query.sqlpp
new file mode 100644
index 0000000..9d36ad2
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/parseonly/003/parseonly_03.2.query.sqlpp
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*
+ * Description : parse-only over a request containing no query at all. Reports an empty parameter list rather
+ * : than failing.
+ * Expected Res : Success - no statement parameters
+ */
+
+-- param parse-only:string=true
+
+create collection c2 primary key (id: int);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/AsyncDeferredQueries.xml b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/AsyncDeferredQueries.xml
index d2b4080..c371e1a 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/AsyncDeferredQueries.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/AsyncDeferredQueries.xml
@@ -76,6 +76,14 @@
<source-location>false</source-location>
</compilation-unit>
</test-case>
+ <test-case FilePath="async-deferred">
+ <compilation-unit name="async-not-supported-dataverse">
+ <parameter name="dataverse" value="test" type="string"/>
+ <output-dir compare="Clean-JSON">async-not-supported-dataverse</output-dir>
+ <expected-error>Async mode not supported for the statement CREATE COLLECTION</expected-error>
+ <source-location>false</source-location>
+ </compilation-unit>
+ </test-case>
<test-case FilePath="async-deferred-improved">
<compilation-unit name="async-failed">
<output-dir compare="Clean-JSON">async-failed</output-dir>
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/async-not-supported-dataverse/async-not-supported-dataverse.1.async.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/async-not-supported-dataverse/async-not-supported-dataverse.1.async.sqlpp
new file mode 100644
index 0000000..64c2f72
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/async-deferred/async-not-supported-dataverse/async-not-supported-dataverse.1.async.sqlpp
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*
+ * Description : async delivery must be rejected for a statement kind that does not support it, even when the
+ * : request supplies a default dataverse. Setting the dataverse prepends a synthetic USE statement,
+ * : and USE does support async, so an async check that only ever inspected the leading statement
+ * : passed this request through instead of rejecting it.
+ * Expected Res : Error - async mode not supported for CREATE COLLECTION
+ */
+
+CREATE COLLECTION col PRIMARY KEY (id: int);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.1.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.1.adm
new file mode 100644
index 0000000..f21a0e3
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.1.adm
@@ -0,0 +1 @@
+{ "statement-parameters": [ "qname" ] }
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.2.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.2.adm
new file mode 100644
index 0000000..75e3de8
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/parseonly/003/parseonly_03.2.adm
@@ -0,0 +1 @@
+{ "statement-parameters": [ ] }
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_parseonly.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_parseonly.xml
index f7cea01..575003a 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_parseonly.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_parseonly.xml
@@ -31,5 +31,10 @@
<expected-error>ASX1001</expected-error>
</compilation-unit>
</test-case>
+ <test-case FilePath="parseonly">
+ <compilation-unit name="003">
+ <output-dir compare="Text">003</output-dir>
+ </compilation-unit>
+ </test-case>
</test-group>
</test-suite>
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IClientRequest.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IClientRequest.java
index 496b2a1..a81d559 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IClientRequest.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IClientRequest.java
@@ -147,4 +147,15 @@
*/
void jobFinished(JobId jobId, JobStatus jobStatus, List<Exception> exceptions);
+ /**
+ * Called when the result of one of this request's jobs has been swept.
+ *
+ * @param jobId the job whose result was swept
+ * @return whether the results of every job of this request have now been swept, i.e. whether the request has
+ * nothing left for a client to fetch and can stop being tracked
+ */
+ default boolean resultSwept(JobId jobId) {
+ return true;
+ }
+
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
index 69ef63b..226018e 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/exceptions/ErrorCode.java
@@ -364,6 +364,7 @@
INVALID_SAMPLE_METHOD(1256),
COMPILATION_ANN_DISTANCE_METRIC_MISMATCH(1257),
COMPILATION_VECTOR_INDEX_CREATION_FAILED(1258),
+ UNSUPPORTED_MULTIPLE_STATEMENTS_FOR_DELIVERY_MODE(1259),
// Feed errors
DATAFLOW_ILLEGAL_STATE(3001),
diff --git a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
index d60d036..95104cb 100644
--- a/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
+++ b/asterixdb/asterix-common/src/main/resources/asx_errormsg/en.properties
@@ -366,6 +366,7 @@
1256 = Sample method option has to be one of "full-scan" or "random"
1257 = ANN distance metric '%1$s' does not match vector index '%2$s' (similarity in metadata: '%3$s')
1258 = Failed to create vector index. %1$s
+1259 = %1$s result delivery is not supported for a request with multiple statements
# Feed Errors
3001 = Illegal state.
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/RequestTracker.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/RequestTracker.java
index 3b1f1bd..c6d3e1f 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/RequestTracker.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/RequestTracker.java
@@ -102,6 +102,15 @@
@Override
public void notifyResultSweep(JobId jobId, String requestId) {
+ if (requestId == null) {
+ return;
+ }
+ // a request that ran several statements has a result per statement, and the client holds a handle to each.
+ // Untracking the request on the first of them to be swept would strand the handles of the others.
+ IClientRequest request = asyncRequests.get(requestId);
+ if (request != null && !request.resultSwept(jobId)) {
+ return;
+ }
removeAsyncOrDeferredRequest(requestId);
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21541?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I5efd9a2fec92d6176b709beb621c8bef7bb86bdc
Gerrit-Change-Number: 21541
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>