You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mh...@apache.org on 2019/06/24 20:28:14 UTC
[asterixdb] branch master updated: [ASTERIXDB-2598][RT] Add Support
For Runtime Warnings
This is an automated email from the ASF dual-hosted git repository.
mhubail pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new 6d6b290 [ASTERIXDB-2598][RT] Add Support For Runtime Warnings
6d6b290 is described below
commit 6d6b2904ff6eb3f0a3f6f642aafb99ef101e7b47
Author: Murtadha Hubail <mh...@apache.org>
AuthorDate: Thu Jun 20 13:52:20 2019 +0300
[ASTERIXDB-2598][RT] Add Support For Runtime Warnings
- user model changes: yes
- storage format changes: no
- interface changes: yes
Details:
- Add the ability to add runtime warnings per task
and return the generated warnings as part of the
task profile on task completion.
- On successful job completion, aggregate warnings
from all task profiles of a job.
- Return the generated warnings in the query service
response as an array of "warnings" each with a code,
which is currently hard-coded to 1, and a message.
- Fix propagating source location to scalar aggregate
functions.
- Add a flag in test cases definition to indicate whether
or not to check for expected warnings and default it to
false.
- Generate warnings when min/max functions encounter
incomparable type or unsupported input.
- Add support in test framework to extract warnings
along with extracting the result field.
- Add support in test framework to validate generated
and expected warnings.
- Add test cases for min/max generated warnings.
Change-Id: I52fa5b807799487d62e67a8861068e1547aa629a
Reviewed-on: https://asterix-gerrit.ics.uci.edu/3451
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <mh...@apache.org>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Dmitry Lychagin <dm...@couchbase.com>
---
.../asterix/translator/IStatementExecutor.java | 8 ++
.../apache/asterix/api/common/ResultMetadata.java | 12 +++
.../api/http/server/NCQueryServiceServlet.java | 7 +-
.../api/http/server/QueryServiceServlet.java | 24 +++--
.../message/ExecuteStatementRequestMessage.java | 1 +
.../message/ExecuteStatementResponseMessage.java | 13 +++
.../asterix/app/result/ExecutionWarning.java | 5 +
.../asterix/app/result/JobResultCallback.java | 12 ++-
.../asterix/app/translator/QueryTranslator.java | 9 ++
.../asterix/app/result/ResultPrinterTest.java | 3 +-
.../test/common/CancellationTestExecutor.java | 6 ++
.../asterix/test/common/ExtractedResult.java | 30 +++---
.../org/apache/asterix/test/common/IPollTask.java | 4 +-
.../asterix/test/common/ResultExtractor.java | 31 ++++--
.../apache/asterix/test/common/TestExecutor.java | 84 +++++++++++-----
.../min-max-incompatible-types.1.query.sqlpp | 18 +---
.../min-max-incompatible-types.2.query.sqlpp | 18 +---
.../min-max-incompatible-types.1.adm | 1 +
.../min-max-incompatible-types.2.adm | 1 +
.../test/resources/runtimets/testsuite_sqlpp.xml | 9 ++
.../apache/asterix/common/utils/WarningUtil.java | 37 +++++++
.../scalar/AbstractScalarAggregateDescriptor.java | 7 ++
.../std/AbstractMinMaxAggregateFunction.java | 30 ++++--
.../src/main/resources/Catalog.xsd | 14 +++
.../hyracks/api/context/IHyracksTaskContext.java | 10 +-
.../api/exceptions/IFormattedException.java | 7 ++
.../hyracks/api/exceptions/SourceLocation.java | 32 ++++++-
.../org/apache/hyracks/api/exceptions/Warning.java | 106 +++++++++++++++++++++
.../common/job/profiling/om/TaskProfile.java | 29 +++++-
.../java/org/apache/hyracks/control/nc/Joblet.java | 2 +-
.../java/org/apache/hyracks/control/nc/Task.java | 14 +++
.../control/nc/work/NotifyTaskCompleteWork.java | 4 +-
.../hyracks/test/support/TestTaskContext.java | 8 ++
.../util/LSMInvertedIndexTestUtils.java | 6 ++
34 files changed, 488 insertions(+), 114 deletions(-)
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 c27a30f..c5e9beb 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
@@ -37,6 +37,7 @@ import org.apache.commons.lang3.tuple.Triple;
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.Warning;
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.api.result.ResultSetId;
@@ -163,4 +164,11 @@ public interface IStatementExecutor {
* @return the responer printer
*/
IResponsePrinter getResponsePrinter();
+
+ /**
+ * Gets the warnings generated during compiling and exeucting a request
+ *
+ * @return the list of warnings
+ */
+ List<Warning> getWarnings();
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/ResultMetadata.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/ResultMetadata.java
index 69f46a4..6fb37d3 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/ResultMetadata.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/ResultMetadata.java
@@ -18,7 +18,10 @@
*/
package org.apache.asterix.api.common;
+import java.util.Set;
+
import org.apache.asterix.translator.SessionConfig;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.result.IResultMetadata;
public class ResultMetadata implements IResultMetadata {
@@ -26,6 +29,7 @@ public class ResultMetadata implements IResultMetadata {
private final SessionConfig.OutputFormat format;
private long jobDuration;
private long processedObjects;
+ private Set<Warning> warnings;
public ResultMetadata(SessionConfig.OutputFormat format) {
this.format = format;
@@ -47,10 +51,18 @@ public class ResultMetadata implements IResultMetadata {
this.jobDuration = jobDuration;
}
+ public void setWarnings(Set<Warning> warnings) {
+ this.warnings = warnings;
+ }
+
public long getJobDuration() {
return jobDuration;
}
+ public Set<Warning> getWarnings() {
+ return warnings;
+ }
+
@Override
public String toString() {
return "ResultMetadata{" + "format=" + format + ", jobDuration=" + jobDuration + ", processedObjects="
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 254f92a..b1361d9 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,7 @@
package org.apache.asterix.api.http.server;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
@@ -45,6 +46,7 @@ import org.apache.asterix.translator.IStatementExecutor;
import org.apache.asterix.translator.ResultProperties;
import org.apache.asterix.translator.SessionOutput;
import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.http.api.IChannelClosedHandler;
import org.apache.hyracks.http.api.IServletRequest;
import org.apache.hyracks.http.server.HttpServer;
@@ -71,7 +73,7 @@ public class NCQueryServiceServlet extends QueryServiceServlet {
SessionOutput sessionOutput, ResultProperties resultProperties, IStatementExecutor.Stats stats,
QueryServiceRequestParameters param, RequestExecutionState execution,
Map<String, String> optionalParameters, Map<String, byte[]> statementParameters,
- ResponsePrinter responsePrinter) throws Exception {
+ ResponsePrinter responsePrinter, List<Warning> warnings) throws Exception {
// Running on NC -> send 'execute' message to CC
INCServiceContext ncCtx = (INCServiceContext) serviceCtx;
INCMessageBroker ncMb = (INCMessageBroker) ncCtx.getMessageBroker();
@@ -121,7 +123,8 @@ public class NCQueryServiceServlet extends QueryServiceServlet {
responsePrinter.addResultPrinter(
new NcResultPrinter(appCtx, responseMsg, getResultSet(), delivery, sessionOutput, stats));
}
- buildResponseResults(responsePrinter, sessionOutput, responseMsg.getExecutionPlans());
+ warnings.addAll(responseMsg.getWarnings());
+ buildResponseResults(responsePrinter, sessionOutput, responseMsg.getExecutionPlans(), warnings);
}
private void cancelQuery(INCMessageBroker messageBroker, String nodeId, String uuid, String clientContextID,
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 4eb3524..0d5de57 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
@@ -31,6 +31,7 @@ import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
@@ -43,6 +44,7 @@ import java.util.function.Function;
import org.apache.asterix.algebra.base.ILangExtension;
import org.apache.asterix.app.result.ExecutionError;
+import org.apache.asterix.app.result.ExecutionWarning;
import org.apache.asterix.app.result.ResponseMertics;
import org.apache.asterix.app.result.ResponsePrinter;
import org.apache.asterix.app.result.fields.ClientContextIdPrinter;
@@ -88,6 +90,7 @@ import org.apache.asterix.translator.SessionOutput;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
import org.apache.hyracks.api.exceptions.HyracksException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.control.common.controllers.CCConfig;
import org.apache.hyracks.http.api.IServletRequest;
import org.apache.hyracks.http.api.IServletResponse;
@@ -468,7 +471,7 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
long errorCount = 1;
Stats stats = new Stats();
RequestExecutionState execution = new RequestExecutionState();
- List<ICodedMessage> warnings = Collections.emptyList();
+ List<Warning> warnings = new ArrayList<>();
Charset resultCharset = HttpUtil.setContentType(response, HttpUtil.ContentType.APPLICATION_JSON, request);
PrintWriter httpWriter = response.writer();
SessionOutput sessionOutput = createSessionOutput(httpWriter);
@@ -503,7 +506,7 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
setAccessControlHeaders(request, response);
response.setStatus(execution.getHttpStatus());
executeStatement(requestRef, statementsText, sessionOutput, resultProperties, stats, param, execution,
- optionalParams, statementParams, responsePrinter);
+ optionalParams, statementParams, responsePrinter, warnings);
}
errorCount = 0;
} catch (Exception | TokenMgrError | org.apache.asterix.aqlplus.parser.TokenMgrError e) {
@@ -539,20 +542,22 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
}
protected void buildResponseResults(ResponsePrinter responsePrinter, SessionOutput sessionOutput,
- ExecutionPlans plans) {
+ ExecutionPlans plans, List<Warning> warnings) {
responsePrinter.addResultPrinter(new PlansPrinter(plans, sessionOutput.config().getPlanFormat()));
+ if (!warnings.isEmpty()) {
+ List<ICodedMessage> codedWarnings = new ArrayList<>();
+ warnings.forEach(warn -> codedWarnings.add(ExecutionWarning.of(warn)));
+ responsePrinter.addResultPrinter(new WarningsPrinter(codedWarnings));
+ }
}
protected void buildResponseFooters(long elapsedStart, long errorCount, Stats stats,
- RequestExecutionState execution, List<ICodedMessage> warnings, Charset resultCharset,
+ RequestExecutionState execution, List<Warning> warnings, Charset resultCharset,
ResponsePrinter responsePrinter, ResultDelivery delivery) {
if (ResultDelivery.ASYNC != delivery) {
// in case of ASYNC delivery, the status is printed by query translator
responsePrinter.addFooterPrinter(new StatusPrinter(execution.getResultStatus()));
}
- if (!warnings.isEmpty()) {
- responsePrinter.addFooterPrinter(new WarningsPrinter(warnings));
- }
final ResponseMertics mertics = ResponseMertics.of(System.nanoTime() - elapsedStart, execution.duration(),
stats.getCount(), stats.getSize(), stats.getProcessedObjects(), errorCount, warnings.size());
responsePrinter.addFooterPrinter(new MetricsPrinter(mertics, resultCharset));
@@ -580,7 +585,7 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
SessionOutput sessionOutput, ResultProperties resultProperties, Stats stats,
QueryServiceRequestParameters param, RequestExecutionState execution,
Map<String, String> optionalParameters, Map<String, byte[]> statementParameters,
- ResponsePrinter responsePrinter) throws Exception {
+ ResponsePrinter responsePrinter, List<Warning> warnings) throws Exception {
IClusterManagementWork.ClusterState clusterState =
((ICcApplicationContext) appCtx).getClusterStateManager().getState();
if (clusterState != IClusterManagementWork.ClusterState.ACTIVE) {
@@ -600,7 +605,8 @@ public class QueryServiceServlet extends AbstractQueryApiServlet {
optionalParameters, stmtParams, param.isMultiStatement());
translator.compileAndExecute(getHyracksClientConnection(), requestParameters);
execution.end();
- buildResponseResults(responsePrinter, sessionOutput, translator.getExecutionPlans());
+ warnings.addAll(translator.getWarnings());
+ buildResponseResults(responsePrinter, sessionOutput, translator.getExecutionPlans(), warnings);
}
protected void handleExecuteStatementException(Throwable t, RequestExecutionState state,
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 b0b94c5..5663b12 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
@@ -143,6 +143,7 @@ public final class ExecuteStatementRequestMessage implements ICcAddressedMessage
responseMsg.setMetadata(outMetadata);
responseMsg.setStats(stats);
responseMsg.setExecutionPlans(translator.getExecutionPlans());
+ responseMsg.setWarnings(translator.getWarnings());
} catch (AlgebricksException | HyracksException | TokenMgrError
| org.apache.asterix.aqlplus.parser.TokenMgrError pe) {
// we trust that "our" exceptions are serializable and have a comprehensible error message
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 94dd541..7a7661d 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
@@ -19,6 +19,8 @@
package org.apache.asterix.app.message;
+import java.util.List;
+
import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.common.messaging.api.INcAddressedMessage;
import org.apache.asterix.common.messaging.api.MessageFuture;
@@ -26,6 +28,7 @@ import org.apache.asterix.messaging.NCMessageBroker;
import org.apache.asterix.translator.ExecutionPlans;
import org.apache.asterix.translator.IStatementExecutor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.Warning;
public final class ExecuteStatementResponseMessage implements INcAddressedMessage {
private static final long serialVersionUID = 1L;
@@ -42,6 +45,8 @@ public final class ExecuteStatementResponseMessage implements INcAddressedMessag
private ExecutionPlans executionPlans;
+ private List<Warning> warnings;
+
public ExecuteStatementResponseMessage(long requestMessageId) {
this.requestMessageId = requestMessageId;
}
@@ -95,6 +100,14 @@ public final class ExecuteStatementResponseMessage implements INcAddressedMessag
this.executionPlans = executionPlans;
}
+ public List<Warning> getWarnings() {
+ return warnings;
+ }
+
+ public void setWarnings(List<Warning> warnings) {
+ this.warnings = warnings;
+ }
+
@Override
public String toString() {
return String.format("%s(id=%s): %d characters", getClass().getSimpleName(), requestMessageId,
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ExecutionWarning.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ExecutionWarning.java
index 29eb098..5552bb8 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ExecutionWarning.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ExecutionWarning.java
@@ -19,6 +19,7 @@
package org.apache.asterix.app.result;
import org.apache.asterix.common.api.ICodedMessage;
+import org.apache.hyracks.api.exceptions.Warning;
public class ExecutionWarning implements ICodedMessage {
@@ -30,6 +31,10 @@ public class ExecutionWarning implements ICodedMessage {
this.message = message;
}
+ public static ICodedMessage of(Warning warning) {
+ return new ExecutionWarning(1, warning.getMessage());
+ }
+
@Override
public int getCode() {
return code;
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/JobResultCallback.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/JobResultCallback.java
index ecf92a0..ab17cdd 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/JobResultCallback.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/JobResultCallback.java
@@ -19,9 +19,12 @@
package org.apache.asterix.app.result;
import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
import org.apache.asterix.api.common.ResultMetadata;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.result.IJobResultCallback;
import org.apache.hyracks.api.result.ResultJobRecord;
@@ -60,11 +63,12 @@ public class JobResultCallback implements IJobResultCallback {
}
final ResultMetadata metadata = (ResultMetadata) resultSetMetaData.getMetadata();
metadata.setJobDuration(resultJobRecord.getJobDuration());
- metadata.setProcessedObjects(getJobProccssedObjects(jobId));
+ aggregateJobStats(jobId, metadata);
}
- private long getJobProccssedObjects(JobId jobId) {
+ private void aggregateJobStats(JobId jobId, ResultMetadata metadata) {
long processedObjects = 0;
+ Set<Warning> warnings = new HashSet<>();
IJobManager jobManager =
((ClusterControllerService) appCtx.getServiceContext().getControllerService()).getJobManager();
final JobRun run = jobManager.get(jobId);
@@ -75,9 +79,11 @@ public class JobResultCallback implements IJobResultCallback {
final Collection<TaskProfile> jobletTasksProfile = jp.getTaskProfiles().values();
for (TaskProfile tp : jobletTasksProfile) {
processedObjects += tp.getStatsCollector().getAggregatedStats().getTupleCounter().get();
+ warnings.addAll(tp.getWarnings());
}
}
}
- return processedObjects;
+ metadata.setProcessedObjects(processedObjects);
+ metadata.setWarnings(warnings);
}
}
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 60e98f9..37c7d82 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
@@ -194,6 +194,7 @@ import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.io.FileSplit;
import org.apache.hyracks.api.io.UnmanagedFileSplit;
import org.apache.hyracks.api.job.JobFlag;
@@ -230,6 +231,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
protected final EnumSet<JobFlag> jobFlags = EnumSet.noneOf(JobFlag.class);
protected final IMetadataLockManager lockManager;
protected final IResponsePrinter responsePrinter;
+ protected final List<Warning> warnings;
public QueryTranslator(ICcApplicationContext appCtx, List<Statement> statements, SessionOutput output,
ILangCompilationProvider compilationProvider, ExecutorService executorService,
@@ -246,6 +248,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
activeDataverse = MetadataBuiltinEntities.DEFAULT_DATAVERSE;
this.executorService = executorService;
this.responsePrinter = responsePrinter;
+ warnings = new ArrayList<>();
if (appCtx.getServiceContext().getAppConfig().getBoolean(CCConfig.Option.ENFORCE_FRAME_WRITER_PROTOCOL)) {
this.jobFlags.add(JobFlag.ENFORCE_CONTRACT);
}
@@ -2544,6 +2547,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
(org.apache.asterix.api.common.ResultMetadata) controllerService.getResultDirectoryService()
.getResultMetadata(jobId, rsId);
stats.setProcessedObjects(resultMetadata.getProcessedObjects());
+ warnings.addAll(resultMetadata.getWarnings());
}
private void asyncCreateAndRunJob(IHyracksClientConnection hcc, IStatementCompiler compiler, IMetadataLocker locker,
@@ -2917,6 +2921,11 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
return getActiveDataverseName(dataverse != null ? dataverse.getValue() : null);
}
+ @Override
+ public List<Warning> getWarnings() {
+ return warnings;
+ }
+
/**
* Abort the ongoing metadata transaction logging the error cause
*
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/result/ResultPrinterTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/result/ResultPrinterTest.java
index fe64dd1..1624db3 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/result/ResultPrinterTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/result/ResultPrinterTest.java
@@ -77,7 +77,8 @@ public class ResultPrinterTest {
boolean exceptionThrown = false;
try {
// ensure result is valid json and error will be returned and not results.
- ResultExtractor.extract(IOUtils.toInputStream(resultStr, StandardCharsets.UTF_8), StandardCharsets.UTF_8);
+ ResultExtractor.extract(IOUtils.toInputStream(resultStr, StandardCharsets.UTF_8), StandardCharsets.UTF_8)
+ .getResult();
} catch (Exception e) {
exceptionThrown = true;
Assert.assertTrue(e.getMessage().contains(expectedException.getMessage()));
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java
index 564672a..a34a29c 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/CancellationTestExecutor.java
@@ -126,4 +126,10 @@ public class CancellationTestExecutor extends TestExecutor {
return true;
}
}
+
+ @Override
+ protected void ensureWarnings(int actualWarnCount, int expectedWarnCount, TestCase.CompilationUnit cUnit)
+ throws Exception {
+ // skip checking warnings as currently cancelled queries with warnings might not run successfully at all
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/SourceLocation.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ExtractedResult.java
similarity index 63%
copy from hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/SourceLocation.java
copy to asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ExtractedResult.java
index fd3992c..96f82a3 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/SourceLocation.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/ExtractedResult.java
@@ -16,29 +16,29 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.asterix.test.common;
-package org.apache.hyracks.api.exceptions;
+import java.io.InputStream;
+import java.util.List;
-import java.io.Serializable;
+public class ExtractedResult {
-public final class SourceLocation implements Serializable {
+ private InputStream result;
+ private List<String> warnings;
- private static final long serialVersionUID = 1L;
-
- private final int line;
-
- private final int column;
+ public void setResult(InputStream result) {
+ this.result = result;
+ }
- public SourceLocation(int line, int column) {
- this.line = line;
- this.column = column;
+ public void setWarnings(List<String> warnings) {
+ this.warnings = warnings;
}
- public int getLine() {
- return line;
+ public InputStream getResult() {
+ return result;
}
- public int getColumn() {
- return column;
+ public List<String> getWarnings() {
+ return warnings;
}
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/IPollTask.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/IPollTask.java
index 6d32518..ab90244 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/IPollTask.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/IPollTask.java
@@ -43,9 +43,11 @@ public interface IPollTask {
* @param expectedResultFileCtxs
* @param testFile
* @param actualPath
+ * @param actualWarnCount
*/
void execute(TestCaseContext testCaseCtx, TestFileContext ctx, Map<String, Object> variableCtx, String statement,
boolean isDmlRecoveryTest, ProcessBuilder pb, CompilationUnit cUnit, MutableInt queryCount,
- List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath) throws Exception;
+ List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath, MutableInt actualWarnCount)
+ throws Exception;
}
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 864a339..eb708ce 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
@@ -20,9 +20,11 @@ package org.apache.asterix.test.common;
import java.io.InputStream;
import java.nio.charset.Charset;
+import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import org.apache.asterix.common.exceptions.AsterixException;
@@ -34,6 +36,7 @@ import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Iterators;
@@ -82,16 +85,16 @@ public class ResultExtractor {
private static final Logger LOGGER = LogManager.getLogger();
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
- public static InputStream extract(InputStream resultStream, Charset resultCharset) throws Exception {
- return extract(resultStream, EnumSet.of(ResultField.RESULTS), resultCharset);
+ public static ExtractedResult extract(InputStream resultStream, Charset resultCharset) throws Exception {
+ return extract(resultStream, EnumSet.of(ResultField.RESULTS, ResultField.WARNINGS), resultCharset);
}
public static InputStream extractMetrics(InputStream resultStream, Charset resultCharset) throws Exception {
- return extract(resultStream, EnumSet.of(ResultField.METRICS), resultCharset);
+ return extract(resultStream, EnumSet.of(ResultField.METRICS), resultCharset).getResult();
}
public static InputStream extractPlans(InputStream resultStream, Charset resultCharset) throws Exception {
- return extract(resultStream, EnumSet.of(ResultField.PLANS), resultCharset);
+ return extract(resultStream, EnumSet.of(ResultField.PLANS), resultCharset).getResult();
}
public static String extractHandle(InputStream resultStream, Charset responseCharset) throws Exception {
@@ -110,8 +113,9 @@ public class ResultExtractor {
return null;
}
- private static InputStream extract(InputStream resultStream, EnumSet<ResultField> resultFields,
+ private static ExtractedResult extract(InputStream resultStream, EnumSet<ResultField> resultFields,
Charset resultCharset) throws Exception {
+ ExtractedResult extractedResult = new ExtractedResult();
final String resultStr = IOUtils.toString(resultStream, resultCharset);
final ObjectNode result = OBJECT_MAPPER.readValue(resultStr, ObjectNode.class);
@@ -168,14 +172,16 @@ public class ResultExtractor {
case STATUS:
case TYPE:
case PLANS:
- case WARNINGS:
resultBuilder.append(OBJECT_MAPPER.writeValueAsString(fieldValue));
+ case WARNINGS:
+ extractWarnings(fieldValue, extractedResult);
break;
default:
throw new IllegalStateException("Unexpected result field: " + fieldKind);
}
}
- return IOUtils.toInputStream(resultBuilder, resultCharset);
+ extractedResult.setResult(IOUtils.toInputStream(resultBuilder, resultCharset));
+ return extractedResult;
}
private static void checkForErrors(ObjectNode result) throws Exception {
@@ -188,4 +194,15 @@ public class ResultExtractor {
throw new Exception(errors.asText());
}
}
+
+ private static void extractWarnings(JsonNode warningsValue, ExtractedResult exeResult) {
+ List<String> warnings = new ArrayList<>();
+ if (warningsValue.isArray()) {
+ final ArrayNode warningsArray = (ArrayNode) warningsValue;
+ for (JsonNode warn : warningsArray) {
+ warnings.add(warn.get("msg").asText());
+ }
+ }
+ exeResult.setWarnings(warnings);
+ }
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
index cbe41e8..5da5199 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
@@ -152,7 +152,10 @@ public class TestExecutor {
Collections.unmodifiableSet(new HashSet<>(Arrays.asList("store", "validate")));
private static final int MAX_NON_UTF_8_STATEMENT_SIZE = 64 * 1024;
- private final IPollTask plainExecutor = this::executeTestFile;
+ private final IPollTask plainExecutor = (testCaseCtx, ctx, variableCtx, statement, isDmlRecoveryTest, pb, cUnit,
+ queryCount, expectedResultFileCtxs, testFile, actualPath, actualWarnCount) -> executeTestFile(testCaseCtx,
+ ctx, variableCtx, statement, isDmlRecoveryTest, pb, cUnit, queryCount, expectedResultFileCtxs,
+ testFile, actualPath, actualWarnCount);
public static final String DELIVERY_ASYNC = "async";
public static final String DELIVERY_DEFERRED = "deferred";
@@ -903,9 +906,8 @@ public class TestExecutor {
public void executeTestFile(TestCaseContext testCaseCtx, TestFileContext ctx, Map<String, Object> variableCtx,
String statement, boolean isDmlRecoveryTest, ProcessBuilder pb, CompilationUnit cUnit,
- MutableInt queryCount, List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath)
- throws Exception {
- URI uri;
+ MutableInt queryCount, List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath,
+ MutableInt actualWarnCount) throws Exception {
InputStream resultStream;
File qbcFile;
boolean failed = false;
@@ -933,11 +935,11 @@ public class TestExecutor {
case "pollquery":
poll(testCaseCtx, ctx, variableCtx, statement, isDmlRecoveryTest, pb, cUnit, queryCount,
expectedResultFileCtxs, testFile, actualPath, ctx.getType().substring("poll".length()),
- plainExecutor);
+ actualWarnCount, plainExecutor);
break;
case "polldynamic":
polldynamic(testCaseCtx, ctx, variableCtx, statement, isDmlRecoveryTest, pb, cUnit, queryCount,
- expectedResultFileCtxs, testFile, actualPath);
+ expectedResultFileCtxs, testFile, actualPath, actualWarnCount);
break;
case "query":
case "async":
@@ -957,9 +959,12 @@ public class TestExecutor {
: expectedResultFileCtxs.get(queryCount.intValue()).getFile();
File actualResultFile = expectedResultFile == null ? null
: testCaseCtx.getActualResultFile(cUnit, expectedResultFile, new File(actualPath));
- executeQuery(OutputFormat.forCompilationUnit(cUnit), statement, variableCtx, ctx.getType(), testFile,
- expectedResultFile, actualResultFile, queryCount, expectedResultFileCtxs.size(),
- cUnit.getParameter(), ComparisonEnum.TEXT);
+ ExtractedResult extractedResult = executeQuery(OutputFormat.forCompilationUnit(cUnit), statement,
+ variableCtx, ctx.getType(), testFile, expectedResultFile, actualResultFile, queryCount,
+ expectedResultFileCtxs.size(), cUnit.getParameter(), ComparisonEnum.TEXT);
+ if (testCaseCtx.getTestCase().isCheckWarnings()) {
+ validateWarnings(extractedResult.getWarnings(), cUnit.getExpectedWarn(), actualWarnCount);
+ }
break;
case "store":
// This is a query that returns the expected output of a subsequent query
@@ -1232,7 +1237,7 @@ public class TestExecutor {
} else if ("uri".equals(extension)) {
resultStream = executeURI(reqType, URI.create(variablesReplaced), fmt, params, statusCodePredicate, body);
if (extracResult) {
- resultStream = ResultExtractor.extract(resultStream, UTF_8);
+ resultStream = ResultExtractor.extract(resultStream, UTF_8).getResult();
}
} else {
throw new IllegalArgumentException("Unexpected format for method " + reqType + ": " + extension);
@@ -1261,9 +1266,9 @@ public class TestExecutor {
queryCount.increment();
}
- public void executeQuery(OutputFormat fmt, String statement, Map<String, Object> variableCtx, String reqType,
- File testFile, File expectedResultFile, File actualResultFile, MutableInt queryCount, int numResultFiles,
- List<Parameter> params, ComparisonEnum compare) throws Exception {
+ public ExtractedResult executeQuery(OutputFormat fmt, String statement, Map<String, Object> variableCtx,
+ String reqType, File testFile, File expectedResultFile, File actualResultFile, MutableInt queryCount,
+ int numResultFiles, List<Parameter> params, ComparisonEnum compare) throws Exception {
String delivery = DELIVERY_IMMEDIATE;
if (reqType.equalsIgnoreCase("async")) {
delivery = DELIVERY_ASYNC;
@@ -1275,6 +1280,7 @@ public class TestExecutor {
boolean isJsonEncoded = isJsonEncoded(extractHttpRequestType(statement));
Charset responseCharset = expectedResultFile == null ? UTF_8 : nextCharset();
InputStream resultStream;
+ ExtractedResult extractedResult = null;
if (DELIVERY_IMMEDIATE.equals(delivery)) {
resultStream = executeQueryService(statement, fmt, uri, params, isJsonEncoded, responseCharset, null,
isCancellable(reqType));
@@ -1286,7 +1292,8 @@ public class TestExecutor {
resultStream = ResultExtractor.extractPlans(resultStream, responseCharset);
break;
default:
- resultStream = ResultExtractor.extract(resultStream, responseCharset);
+ extractedResult = ResultExtractor.extract(resultStream, responseCharset);
+ resultStream = extractedResult.getResult();
break;
}
} else {
@@ -1308,7 +1315,7 @@ public class TestExecutor {
writeOutputToFile(actualResultFile, resultStream);
if (expectedResultFile == null) {
if (reqType.equals("store")) {
- return;
+ return extractedResult;
}
Assert.fail("no result file for " + testFile.toString() + "; queryCount: " + queryCount
+ ", filectxs.size: " + numResultFiles);
@@ -1320,22 +1327,23 @@ public class TestExecutor {
}
// Deletes the matched result file.
actualResultFile.getParentFile().delete();
+ return extractedResult;
}
private void polldynamic(TestCaseContext testCaseCtx, TestFileContext ctx, Map<String, Object> variableCtx,
String statement, boolean isDmlRecoveryTest, ProcessBuilder pb, CompilationUnit cUnit,
- MutableInt queryCount, List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath)
- throws Exception {
+ MutableInt queryCount, List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath,
+ MutableInt actualWarnCount) throws Exception {
IExpectedResultPoller poller = getExpectedResultPoller(statement);
final String key = getKey(statement);
poll(testCaseCtx, ctx, variableCtx, statement, isDmlRecoveryTest, pb, cUnit, queryCount, expectedResultFileCtxs,
- testFile, actualPath, "validate", new IPollTask() {
+ testFile, actualPath, "validate", actualWarnCount, new IPollTask() {
@Override
public void execute(TestCaseContext testCaseCtx, TestFileContext ctx,
Map<String, Object> variableCtx, String statement, boolean isDmlRecoveryTest,
ProcessBuilder pb, CompilationUnit cUnit, MutableInt queryCount,
- List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath)
- throws Exception {
+ List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath,
+ MutableInt actualWarnCount) throws Exception {
File actualResultFile = new File(actualPath, testCaseCtx.getTestCase().getFilePath()
+ File.separatorChar + cUnit.getName() + '.' + ctx.getSeqNum() + ".polled.adm");
if (actualResultFile.exists() && !actualResultFile.delete()) {
@@ -1373,7 +1381,7 @@ public class TestExecutor {
private void poll(TestCaseContext testCaseCtx, TestFileContext ctx, Map<String, Object> variableCtx,
String statement, boolean isDmlRecoveryTest, ProcessBuilder pb, CompilationUnit cUnit,
MutableInt queryCount, List<TestFileContext> expectedResultFileCtxs, File testFile, String actualPath,
- String newType, IPollTask pollTask) throws Exception {
+ String newType, MutableInt actualWarnCount, IPollTask pollTask) throws Exception {
// polltimeoutsecs=nnn, polldelaysecs=nnn
int timeoutSecs = getTimeoutSecs(statement);
int retryDelaySecs = getRetryDelaySecs(statement);
@@ -1396,7 +1404,7 @@ public class TestExecutor {
try {
startSemaphore.release();
pollTask.execute(testCaseCtx, ctx, variableCtx, statement, isDmlRecoveryTest, pb, cUnit,
- queryCount, expectedResultFileCtxs, testFile, actualPath);
+ queryCount, expectedResultFileCtxs, testFile, actualPath, actualWarnCount);
} finally {
endSemaphore.release();
}
@@ -1475,7 +1483,7 @@ public class TestExecutor {
private InputStream executeUpdateOrDdl(String statement, OutputFormat outputFormat, URI serviceUri)
throws Exception {
InputStream resultStream = executeQueryService(statement, serviceUri, outputFormat, UTF_8);
- return ResultExtractor.extract(resultStream, UTF_8);
+ return ResultExtractor.extract(resultStream, UTF_8).getResult();
}
protected static boolean isExpected(Exception e, CompilationUnit cUnit) {
@@ -1684,6 +1692,8 @@ public class TestExecutor {
List<CompilationUnit> cUnits = testCaseCtx.getTestCase().getCompilationUnit();
for (CompilationUnit cUnit : cUnits) {
List<String> expectedErrors = cUnit.getExpectedError();
+ int expectedWarnCount = cUnit.getExpectedWarn().size();
+ MutableInt actualWarnCount = new MutableInt(0);
LOGGER.info(
"Starting [TEST]: " + testCaseCtx.getTestCase().getFilePath() + "/" + cUnit.getName() + " ... ");
Map<String, Object> variableCtx = new HashMap<>();
@@ -1703,7 +1713,7 @@ public class TestExecutor {
try {
if (!testFile.getName().startsWith(DIAGNOSE)) {
executeTestFile(testCaseCtx, ctx, variableCtx, statement, isDmlRecoveryTest, pb, cUnit,
- queryCount, expectedResultFileCtxs, testFile, actualPath);
+ queryCount, expectedResultFileCtxs, testFile, actualPath, actualWarnCount);
}
} catch (TestLoop loop) {
// rewind the iterator until we find our target
@@ -1735,6 +1745,7 @@ public class TestExecutor {
throw new Exception(
"Test \"" + cUnit.getName() + "\" FAILED; expected exception was not thrown...");
}
+ ensureWarnings(actualWarnCount.getValue(), expectedWarnCount, cUnit);
LOGGER.info(
"[TEST]: " + testCaseCtx.getTestCase().getFilePath() + "/" + cUnit.getName() + " PASSED ");
if (passedGroup != null) {
@@ -1757,7 +1768,7 @@ public class TestExecutor {
final File file = ctx.getFile();
final String statement = readTestFile(file);
executeTestFile(testCaseCtx, ctx, variableCtx, statement, false, pb, cUnit, new MutableInt(-1),
- Collections.emptyList(), file, null);
+ Collections.emptyList(), file, null, new MutableInt(-1));
}
}
} catch (Exception diagnosticFailure) {
@@ -1954,6 +1965,13 @@ public class TestExecutor {
LOGGER.info("Cluster state now " + desiredState);
}
+ protected void ensureWarnings(int actualWarnCount, int expectedWarnCount, CompilationUnit cUnit) throws Exception {
+ if (actualWarnCount < expectedWarnCount) {
+ LOGGER.error("Test {} failed to raise (an) expected warning(s)", cUnit.getName());
+ throw new Exception("Test \"" + cUnit.getName() + "\" FAILED; expected warning(s) was not returned...");
+ }
+ }
+
private void executeStorageCommand(String[] command) throws Exception {
String srcNode = command[0];
String api = command[1];
@@ -2078,13 +2096,27 @@ public class TestExecutor {
final URI uri = getQueryServiceUri(testFile);
final InputStream inputStream = executeQueryService(statement, OutputFormat.forCompilationUnit(cUnit), uri,
cUnit.getParameter(), true, responseCharset, null, false);
- return ResultExtractor.extract(inputStream, responseCharset);
+ return ResultExtractor.extract(inputStream, responseCharset).getResult();
}
private URI getQueryServiceUri(String extension) throws URISyntaxException {
return extension.endsWith(AQL) ? getEndpoint(Servlets.QUERY_AQL) : getEndpoint(Servlets.QUERY_SERVICE);
}
+ private void validateWarnings(List<String> actualWarnings, List<String> expectedWarn, MutableInt actualWarnCount)
+ throws Exception {
+ if (actualWarnCount.getValue() > expectedWarn.size()) {
+ throw new IllegalStateException("returned warnings exceeded expected warnings");
+ }
+ for (String actualWarn : actualWarnings) {
+ if (expectedWarn.stream().anyMatch(actualWarn::contains)) {
+ actualWarnCount.increment();
+ } else {
+ throw new Exception("unexpected warning was encountered (" + actualWarn + ")");
+ }
+ }
+ }
+
private static String toQueryServiceHandle(String handle) {
return handle.replace("/aql/", "/service/");
}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/warnings/min-max-incompatible-types/min-max-incompatible-types.1.query.sqlpp
similarity index 68%
copy from hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java
copy to asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/warnings/min-max-incompatible-types/min-max-incompatible-types.1.query.sqlpp
index 994915f..a392016 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/warnings/min-max-incompatible-types/min-max-incompatible-types.1.query.sqlpp
@@ -16,21 +16,5 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.hyracks.api.exceptions;
-public interface IFormattedException {
-
- /**
- * Gets the component of this {@link IFormattedException}
- *
- * @return the exception component
- */
- String getComponent();
-
- /**
- * Gets the error code of this {@link IFormattedException}
- *
- * @return the error code
- */
- int getErrorCode();
-}
\ No newline at end of file
+SELECT VALUE ARRAY_MIN([1, '2']);
\ No newline at end of file
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/warnings/min-max-incompatible-types/min-max-incompatible-types.2.query.sqlpp
similarity index 68%
copy from hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java
copy to asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/warnings/min-max-incompatible-types/min-max-incompatible-types.2.query.sqlpp
index 994915f..b6eb3ed 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/warnings/min-max-incompatible-types/min-max-incompatible-types.2.query.sqlpp
@@ -16,21 +16,5 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.hyracks.api.exceptions;
-public interface IFormattedException {
-
- /**
- * Gets the component of this {@link IFormattedException}
- *
- * @return the exception component
- */
- String getComponent();
-
- /**
- * Gets the error code of this {@link IFormattedException}
- *
- * @return the error code
- */
- int getErrorCode();
-}
\ No newline at end of file
+SELECT VALUE MIN(ds.InternalDetails) FROM Metadata.`Dataset` ds;
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/warnings/min-max-incompatible-types/min-max-incompatible-types.1.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/warnings/min-max-incompatible-types/min-max-incompatible-types.1.adm
new file mode 100644
index 0000000..ec747fa
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/warnings/min-max-incompatible-types/min-max-incompatible-types.1.adm
@@ -0,0 +1 @@
+null
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/warnings/min-max-incompatible-types/min-max-incompatible-types.2.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/warnings/min-max-incompatible-types/min-max-incompatible-types.2.adm
new file mode 100644
index 0000000..ec747fa
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/warnings/min-max-incompatible-types/min-max-incompatible-types.2.adm
@@ -0,0 +1 @@
+null
\ No newline at end of file
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml
index 349da99..282c03a 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp.xml
@@ -12354,4 +12354,13 @@
</compilation-unit>
</test-case>
</test-group>
+ <test-group name="warnings">
+ <test-case FilePath="warnings" check-warnings="true">
+ <compilation-unit name="min-max-incompatible-types">
+ <output-dir compare="Text">min-max-incompatible-types</output-dir>
+ <expected-warn>ASX0003: Type incompatibility: function min/max gets incompatible input values: bigint and string</expected-warn>
+ <expected-warn>ASX0004: Unsupported type: min/max cannot process input type object</expected-warn>
+ </compilation-unit>
+ </test-case>
+ </test-group>
</test-suite>
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/WarningUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/WarningUtil.java
new file mode 100644
index 0000000..e52df21
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/WarningUtil.java
@@ -0,0 +1,37 @@
+/*
+ * 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.common.utils;
+
+import java.io.Serializable;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.exceptions.Warning;
+import org.apache.hyracks.api.util.ErrorMessageUtil;
+
+public class WarningUtil {
+
+ private WarningUtil() {
+ }
+
+ public static Warning forAsterix(SourceLocation srcLocation, int code, Serializable... params) {
+ return Warning.of(ErrorCode.ASTERIX, srcLocation, code, ErrorMessageUtil.formatMessage(ErrorCode.ASTERIX, code,
+ ErrorCode.getErrorMessage(code), srcLocation, params));
+ }
+}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/AbstractScalarAggregateDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/AbstractScalarAggregateDescriptor.java
index db31b1c..0b0f8d8 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/AbstractScalarAggregateDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/scalar/AbstractScalarAggregateDescriptor.java
@@ -38,6 +38,7 @@ import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
import org.apache.hyracks.algebricks.runtime.evaluators.ColumnAccessEvalFactory;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.SourceLocation;
public abstract class AbstractScalarAggregateDescriptor extends AbstractScalarFunctionDynamicDescriptor {
private static final long serialVersionUID = 1L;
@@ -49,6 +50,12 @@ public abstract class AbstractScalarAggregateDescriptor extends AbstractScalarFu
}
@Override
+ public void setSourceLocation(SourceLocation sourceLoc) {
+ super.setSourceLocation(sourceLoc);
+ aggFuncDesc.setSourceLocation(sourceLoc);
+ }
+
+ @Override
public IScalarEvaluatorFactory createEvaluatorFactory(final IScalarEvaluatorFactory[] args)
throws AlgebricksException {
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java
index e76b8c9..463a676 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/aggregates/std/AbstractMinMaxAggregateFunction.java
@@ -20,6 +20,8 @@ package org.apache.asterix.runtime.aggregates.std;
import java.io.IOException;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.utils.WarningUtil;
import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
import org.apache.asterix.dataflow.data.nontagged.comparators.ComparatorUtil;
import org.apache.asterix.om.types.ATypeTag;
@@ -47,12 +49,14 @@ public abstract class AbstractMinMaxAggregateFunction extends AbstractAggregateF
private final boolean isMin;
private final IAType aggFieldType;
protected final Type type;
+ protected final IHyracksTaskContext context;
protected ATypeTag aggType;
private ILogicalBinaryComparator cmp;
AbstractMinMaxAggregateFunction(IScalarEvaluatorFactory[] args, IHyracksTaskContext context, boolean isMin,
SourceLocation sourceLoc, Type type, IAType aggFieldType) throws HyracksDataException {
super(sourceLoc);
+ this.context = context;
this.eval = args[0].createScalarEvaluator(context);
this.isMin = isMin;
this.aggFieldType = aggFieldType;
@@ -83,29 +87,29 @@ public abstract class AbstractMinMaxAggregateFunction extends AbstractAggregateF
} else if (aggType == ATypeTag.SYSTEM_NULL) {
// First value encountered. Set type, comparator, and initial value.
if (ILogicalBinaryComparator.inequalityUndefined(typeTag)) {
- handleInvalidInput();
+ handleUnsupportedInput(typeTag);
return;
}
aggType = typeTag;
cmp = ComparatorUtil.createLogicalComparator(aggFieldType, aggFieldType, false);
outputVal.assign(inputVal);
} else if (!ATypeHierarchy.isCompatible(typeTag, aggType)) {
- handleInvalidInput();
+ handleIncompatibleInput(typeTag);
} else {
// the two values are compatible non-null/non-missing values
if (aggType == typeTag) {
- compareAndUpdate(cmp, inputVal, outputVal);
+ compareAndUpdate(cmp, inputVal, outputVal, typeTag);
return;
}
if (ATypeHierarchy.canPromote(aggType, typeTag)) {
// switch to new comp & aggregation type (i.e. current min/max is int and new input is double)
castValue(ATypeHierarchy.getTypePromoteComputer(aggType, typeTag), outputVal, tempValForCasting);
outputVal.assign(tempValForCasting);
- compareAndUpdate(cmp, inputVal, outputVal);
+ compareAndUpdate(cmp, inputVal, outputVal, typeTag);
aggType = typeTag;
} else {
castValue(ATypeHierarchy.getTypePromoteComputer(typeTag, aggType), inputVal, tempValForCasting);
- compareAndUpdate(cmp, tempValForCasting, outputVal);
+ compareAndUpdate(cmp, tempValForCasting, outputVal, typeTag);
}
}
}
@@ -151,12 +155,18 @@ public abstract class AbstractMinMaxAggregateFunction extends AbstractAggregateF
return aggType == ATypeTag.NULL;
}
- private void handleInvalidInput() {
- aggType = ATypeTag.NULL;
+ private void handleIncompatibleInput(ATypeTag typeTag) {
+ context.warn(WarningUtil.forAsterix(sourceLoc, ErrorCode.TYPE_INCOMPATIBLE, "min/max", aggType, typeTag));
+ this.aggType = ATypeTag.NULL;
}
- private void compareAndUpdate(ILogicalBinaryComparator c, IPointable newVal, ArrayBackedValueStorage currentVal)
- throws HyracksDataException {
+ private void handleUnsupportedInput(ATypeTag typeTag) {
+ context.warn(WarningUtil.forAsterix(sourceLoc, ErrorCode.TYPE_UNSUPPORTED, "min/max", typeTag));
+ this.aggType = ATypeTag.NULL;
+ }
+
+ private void compareAndUpdate(ILogicalBinaryComparator c, IPointable newVal, ArrayBackedValueStorage currentVal,
+ ATypeTag typeTag) throws HyracksDataException {
// newVal is never NULL/MISSING here. it's already checked up. current value is the first encountered non-null.
ILogicalBinaryComparator.Result result = c.compare(newVal, currentVal);
switch (result) {
@@ -179,7 +189,7 @@ public abstract class AbstractMinMaxAggregateFunction extends AbstractAggregateF
aggType = ATypeTag.NULL;
return;
case INCOMPARABLE:
- handleInvalidInput();
+ handleIncompatibleInput(typeTag);
return;
default:
// EQ, do nothing
diff --git a/asterixdb/asterix-test-framework/src/main/resources/Catalog.xsd b/asterixdb/asterix-test-framework/src/main/resources/Catalog.xsd
index b6bc7bf..5212244 100644
--- a/asterixdb/asterix-test-framework/src/main/resources/Catalog.xsd
+++ b/asterixdb/asterix-test-framework/src/main/resources/Catalog.xsd
@@ -109,6 +109,10 @@
<xs:selector xpath=".//test:expected-error"/>
<xs:field xpath="."/>
</xs:unique>
+ <xs:unique name="unique-expected-warn">
+ <xs:selector xpath=".//test:expected-warn"/>
+ <xs:field xpath="."/>
+ </xs:unique>
</xs:element>
<xs:element ref="test:test-group" minOccurs="0" maxOccurs="unbounded"/>
@@ -180,6 +184,15 @@
</xs:annotation>
</xs:element>
+ <!-- Zero or more expected warnings for this test -->
+
+ <xs:element name="expected-warn" type="xs:string" minOccurs="0" maxOccurs="unbounded">
+ <xs:annotation>
+ <xs:documentation>
+ Zero or more expected warnings for this query.
+ </xs:documentation>
+ </xs:annotation>
+ </xs:element>
</xs:sequence>
<!-- This name is always equal to the name of the test case -->
@@ -200,6 +213,7 @@
<xs:attribute name="FilePath" type="test:SimplifiedRelativeFilePath" use="required"/>
<xs:attribute name="category" type="test:category-enum"/>
<xs:attribute name="repeat" type="xs:positiveInteger" default="1" />
+ <xs:attribute name="check-warnings" type="xs:boolean" default="false"/>
</xs:complexType>
<!-- category-enum type -->
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java
index 930c31d..abe901a 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java
@@ -25,6 +25,7 @@ import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.dataflow.TaskAttemptId;
import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.exceptions.HyracksException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.io.IWorkspaceFileFactory;
import org.apache.hyracks.api.job.IOperatorEnvironment;
import org.apache.hyracks.api.job.JobFlag;
@@ -53,9 +54,16 @@ public interface IHyracksTaskContext
Object getSharedObject();
- public byte[] getJobParameter(byte[] name, int start, int length) throws HyracksException;
+ byte[] getJobParameter(byte[] name, int start, int length) throws HyracksException;
Set<JobFlag> getJobFlags();
IStatsCollector getStatsCollector();
+
+ /**
+ * Adds a warning to this {@link IHyracksTaskContext}
+ *
+ * @param warning
+ */
+ void warn(Warning warning);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java
index 994915f..0f873e1 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/IFormattedException.java
@@ -33,4 +33,11 @@ public interface IFormattedException {
* @return the error code
*/
int getErrorCode();
+
+ /**
+ * Gets the message of this exception
+ *
+ * @return the exception message
+ */
+ String getMessage();
}
\ No newline at end of file
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/SourceLocation.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/SourceLocation.java
index fd3992c..3accc38 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/SourceLocation.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/SourceLocation.java
@@ -19,7 +19,11 @@
package org.apache.hyracks.api.exceptions;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.io.Serializable;
+import java.util.Objects;
public final class SourceLocation implements Serializable {
@@ -38,7 +42,33 @@ public final class SourceLocation implements Serializable {
return line;
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SourceLocation that = (SourceLocation) o;
+ return line == that.line && column == that.column;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(line, column);
+ }
+
public int getColumn() {
return column;
}
-}
+
+ public void writeFields(DataOutput output) throws IOException {
+ output.writeInt(line);
+ output.writeInt(column);
+ }
+
+ public static SourceLocation create(DataInput dataInput) throws IOException {
+ return new SourceLocation(dataInput.readInt(), dataInput.readInt());
+ }
+}
\ No newline at end of file
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/Warning.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/Warning.java
new file mode 100644
index 0000000..78fe3d8
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/exceptions/Warning.java
@@ -0,0 +1,106 @@
+/*
+ * 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.hyracks.api.exceptions;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Objects;
+
+import org.apache.hyracks.api.util.ErrorMessageUtil;
+
+public class Warning implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private final String component;
+ private final SourceLocation srcLocation;
+ private final int code;
+ private final String message;
+
+ private Warning(String component, SourceLocation srcLocation, int code, String message) {
+ this.component = component;
+ this.srcLocation = srcLocation;
+ this.code = code;
+ this.message = message;
+ }
+
+ public static Warning of(String component, SourceLocation srcLocation, int code, String message) {
+ Objects.requireNonNull(srcLocation, "warnings must have source location");
+ return new Warning(component, srcLocation, code, message);
+ }
+
+ public static Warning forHyracks(SourceLocation srcLocation, int code, Serializable... params) {
+ return Warning.of(ErrorCode.HYRACKS, srcLocation, code, ErrorMessageUtil.formatMessage(ErrorCode.HYRACKS, code,
+ ErrorCode.getErrorMessage(code), srcLocation, params));
+ }
+
+ public String getComponent() {
+ return component;
+ }
+
+ public int getCode() {
+ return code;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public SourceLocation getSourceLocation() {
+ return srcLocation;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Warning warning = (Warning) o;
+ return Objects.equals(message, warning.message);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(message);
+ }
+
+ public void writeFields(DataOutput output) throws IOException {
+ output.writeUTF(component);
+ output.writeInt(code);
+ output.writeUTF(message);
+ srcLocation.writeFields(output);
+ }
+
+ public static Warning create(DataInput input) throws IOException {
+ String comp = input.readUTF();
+ int code = input.readInt();
+ String msg = input.readUTF();
+ return new Warning(comp, SourceLocation.create(input), code, msg);
+ }
+
+ @Override
+ public String toString() {
+ return "Warning{" + "component='" + component + '\'' + ", srcLocation=" + srcLocation + ", code=" + code
+ + ", message='" + message + '\'' + '}';
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/job/profiling/om/TaskProfile.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/job/profiling/om/TaskProfile.java
index f977654..2a20624 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/job/profiling/om/TaskProfile.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/job/profiling/om/TaskProfile.java
@@ -22,10 +22,13 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
import org.apache.hyracks.api.dataflow.TaskAttemptId;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.job.profiling.IStatsCollector;
import org.apache.hyracks.api.partitions.PartitionId;
import org.apache.hyracks.control.common.job.profiling.StatsCollector;
@@ -44,6 +47,8 @@ public class TaskProfile extends AbstractProfile {
private IStatsCollector statsCollector;
+ private Set<Warning> warnings;
+
public static TaskProfile create(DataInput dis) throws IOException {
TaskProfile taskProfile = new TaskProfile();
taskProfile.readFields(dis);
@@ -55,10 +60,11 @@ public class TaskProfile extends AbstractProfile {
}
public TaskProfile(TaskAttemptId taskAttemptId, Map<PartitionId, PartitionProfile> partitionSendProfile,
- IStatsCollector statsCollector) {
+ IStatsCollector statsCollector, Set<Warning> warnings) {
this.taskAttemptId = taskAttemptId;
this.partitionSendProfile = new HashMap<>(partitionSendProfile);
this.statsCollector = statsCollector;
+ this.warnings = warnings;
}
public TaskAttemptId getTaskId() {
@@ -115,6 +121,10 @@ public class TaskProfile extends AbstractProfile {
return statsCollector;
}
+ public Set<Warning> getWarnings() {
+ return warnings;
+ }
+
@Override
public void readFields(DataInput input) throws IOException {
super.readFields(input);
@@ -127,6 +137,8 @@ public class TaskProfile extends AbstractProfile {
partitionSendProfile.put(key, value);
}
statsCollector = StatsCollector.create(input);
+ warnings = new HashSet<>();
+ deserializeWarnings(input, warnings);
}
@Override
@@ -139,5 +151,20 @@ public class TaskProfile extends AbstractProfile {
entry.getValue().writeFields(output);
}
statsCollector.writeFields(output);
+ serializeWarnings(output);
+ }
+
+ private void serializeWarnings(DataOutput output) throws IOException {
+ output.writeInt(warnings.size());
+ for (Warning warning : warnings) {
+ warning.writeFields(output);
+ }
+ }
+
+ private static void deserializeWarnings(DataInput input, Set<Warning> warnings) throws IOException {
+ int warnCount = input.readInt();
+ for (int i = 0; i < warnCount; i++) {
+ warnings.add(Warning.create(input));
+ }
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java
index 1d2b77a..8f02f2d 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Joblet.java
@@ -206,7 +206,7 @@ public class Joblet implements IHyracksJobletContext, ICounterContext {
counterMap.forEach((key, value) -> counters.put(key, value.get()));
for (Task task : taskMap.values()) {
TaskProfile taskProfile = new TaskProfile(task.getTaskAttemptId(),
- new Hashtable<>(task.getPartitionSendProfile()), new StatsCollector());
+ new Hashtable<>(task.getPartitionSendProfile()), new StatsCollector(), task.getWarnings());
task.dumpProfile(taskProfile);
jProfile.getTaskProfiles().put(task.getTaskAttemptId(), taskProfile);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java
index 252fe97..b8c1496 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java
@@ -29,6 +29,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
@@ -46,6 +47,7 @@ import org.apache.hyracks.api.dataflow.state.IStateObject;
import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.HyracksException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.io.IWorkspaceFileFactory;
@@ -115,6 +117,8 @@ public class Task implements IHyracksTaskContext, ICounterContext, Runnable {
private volatile boolean completed = false;
+ private final Set<Warning> warnings;
+
public Task(Joblet joblet, Set<JobFlag> jobFlags, TaskAttemptId taskId, String displayName,
ExecutorService executor, NodeControllerService ncs,
List<List<PartitionChannel>> inputChannelsFromConnectors) {
@@ -133,6 +137,7 @@ public class Task implements IHyracksTaskContext, ICounterContext, Runnable {
this.ncs = ncs;
this.inputChannelsFromConnectors = inputChannelsFromConnectors;
statsCollector = new StatsCollector();
+ warnings = ConcurrentHashMap.newKeySet();
}
public void setTaskRuntime(IPartitionCollector[] collectors, IOperatorNodePushable operator) {
@@ -468,10 +473,19 @@ public class Task implements IHyracksTaskContext, ICounterContext, Runnable {
return statsCollector;
}
+ @Override
+ public void warn(Warning warning) {
+ warnings.add(warning);
+ }
+
public boolean isCompleted() {
return completed;
}
+ public Set<Warning> getWarnings() {
+ return warnings;
+ }
+
@Override
public String toString() {
return "{ \"class\" : \"" + getClass().getSimpleName() + "\", \"node\" : \"" + ncs.getId() + "\" \"jobId\" : \""
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/NotifyTaskCompleteWork.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/NotifyTaskCompleteWork.java
index 554c660..b8c5030 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/NotifyTaskCompleteWork.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/work/NotifyTaskCompleteWork.java
@@ -38,8 +38,8 @@ public class NotifyTaskCompleteWork extends AbstractWork {
@Override
public void run() {
- TaskProfile taskProfile =
- new TaskProfile(task.getTaskAttemptId(), task.getPartitionSendProfile(), task.getStatsCollector());
+ TaskProfile taskProfile = new TaskProfile(task.getTaskAttemptId(), task.getPartitionSendProfile(),
+ task.getStatsCollector(), task.getWarnings());
try {
ncs.getClusterController(task.getJobletContext().getJobId().getCcId()).notifyTaskComplete(
task.getJobletContext().getJobId(), task.getTaskAttemptId(), ncs.getId(), taskProfile);
diff --git a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java
index 174d5cd..e328e3d 100644
--- a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java
@@ -22,6 +22,7 @@ import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.EnumSet;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
@@ -32,6 +33,7 @@ import org.apache.hyracks.api.dataflow.TaskAttemptId;
import org.apache.hyracks.api.dataflow.state.IStateObject;
import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.job.JobFlag;
@@ -49,6 +51,7 @@ public class TestTaskContext implements IHyracksTaskContext {
private Map<Object, IStateObject> stateObjectMap = new HashMap<>();
private Object sharedObject;
private final IStatsCollector statsCollector = new StatsCollector();
+ private final Set<Warning> warnings = new HashSet<>();
public TestTaskContext(TestJobletContext jobletContext, TaskAttemptId taskId) {
this.jobletContext = jobletContext;
@@ -175,4 +178,9 @@ public class TestTaskContext implements IHyracksTaskContext {
public IStatsCollector getStatsCollector() {
return statsCollector;
}
+
+ @Override
+ public void warn(Warning warning) {
+ warnings.add(warning);
+ }
}
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-invertedindex-test/src/test/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/LSMInvertedIndexTestUtils.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-invertedindex-test/src/test/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/LSMInvertedIndexTestUtils.java
index 87c2d8f..9021fc5 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-invertedindex-test/src/test/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/LSMInvertedIndexTestUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-invertedindex-test/src/test/java/org/apache/hyracks/storage/am/lsm/invertedindex/util/LSMInvertedIndexTestUtils.java
@@ -48,6 +48,7 @@ import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.HyracksException;
+import org.apache.hyracks.api.exceptions.Warning;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IIOManager;
import org.apache.hyracks.api.job.JobFlag;
@@ -764,6 +765,11 @@ public class LSMInvertedIndexTestUtils {
public IStatsCollector getStatsCollector() {
return null;
}
+
+ @Override
+ public void warn(Warning warning) {
+ // no-op
+ }
}
}