You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by al...@apache.org on 2026/08/27 18:16:48 UTC
(asterixdb) 01/05: [ASTERIXDB-3667][OTH][BP] Log counts of currently running jobs by kind
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit b40321ae586715979c343248cd49fbe99b3684b5
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Tue Aug 18 06:15:50 2026 -0400
[ASTERIXDB-3667][OTH][BP] Log counts of currently running jobs by kind
- user model changes: no
- storage format changes: no
- interface changes: no
Details: Backport of the JobKind plumbing only; this branch has neither
the global transaction manager nor COPY FROM, so those tagging sites are
absent. AsterixJobProperty is carried over intact (GLOBAL_TX is unused
here) to keep the merge up to morpheus/totoro clean. QueryTranslator
carries no runTrackJob on this branch, so JOB_KIND is set alongside
setRequestId in createAndRunJob and the kind is threaded through
deliverResult/asyncCreateAndRunJob.
(cherry picked from commit f139a63ddea92edb8efbf7034bb8c80d060ff24b)
Ext-ref: MB-73366
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: I80e1270a514fb9683f3f429e54aeaa2c5ee77a9a
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20523
Reviewed-by: Ali Alsuliman <al...@gmail.com>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <mh...@apache.org>
Tested-by: Ali Alsuliman <al...@gmail.com>
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21573
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mb...@apache.org>
---
.../asterix/optimizer/base/AnalysisUtil.java | 4 +-
.../app/active/ActiveNotificationHandler.java | 6 +--
.../asterix/app/active/FeedEventsListener.java | 7 +++-
.../app/external/ExternalLibraryJobUtils.java | 8 +++-
.../asterix/app/translator/QueryTranslator.java | 24 +++++++-----
.../org/apache/asterix/utils/DataverseUtil.java | 3 ++
.../org/apache/asterix/utils/RebalanceUtil.java | 3 ++
.../asterix/test/active/ActiveStatsTest.java | 3 +-
.../test/active/TestClusterControllerActor.java | 4 +-
.../asterix/common/utils/AsterixJobProperty.java | 26 +++++++++++++
.../apache/asterix/metadata/utils/DatasetUtil.java | 7 ++--
.../apache/asterix/metadata/utils/IndexUtil.java | 18 +++++++--
.../apache/hyracks/api/job/HyracksJobProperty.java | 23 +++++++++++
.../org/apache/hyracks/api/job/IJobProperty.java | 24 ++++++++++++
.../java/org/apache/hyracks/api/job/JobKind.java | 29 ++++++++++++++
.../apache/hyracks/api/job/JobSpecification.java | 10 ++---
.../apache/hyracks/control/cc/job/JobManager.java | 45 +++++++++++++++++++---
17 files changed, 209 insertions(+), 35 deletions(-)
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AnalysisUtil.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AnalysisUtil.java
index b6f79d9924..c5416ba31c 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AnalysisUtil.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AnalysisUtil.java
@@ -69,7 +69,9 @@ import org.apache.hyracks.algebricks.core.rewriter.base.IRuleSetKind;
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.job.HyracksJobProperty;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.api.result.IResultSetReader;
import org.apache.hyracks.api.result.ResultSetId;
@@ -285,7 +287,7 @@ public class AnalysisUtil {
compiler.optimize();
JobSpecification jobSpec = compiler.createJob(appCtx, new JobEventListenerFactory(newTxnId, false));
-
+ jobSpec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.SYS_QUERY);
JobId jobId = JobUtils.runJob(appCtx.getHcc(), jobSpec, true);
IResultSetReader resultSetReader = appCtx.getResultSet().createReader(jobId, resultSetId);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
index 01e47b5ecd..347a49e475 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/ActiveNotificationHandler.java
@@ -33,6 +33,7 @@ import org.apache.asterix.active.IActiveNotificationHandler;
import org.apache.asterix.active.message.ActivePartitionMessage;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
+import org.apache.asterix.common.utils.AsterixJobProperty;
import org.apache.asterix.metadata.declared.MetadataProvider;
import org.apache.asterix.metadata.entities.Dataset;
import org.apache.commons.lang3.tuple.Pair;
@@ -54,7 +55,6 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
private static final Logger LOGGER = LogManager.getLogger();
private static final Level level = Level.DEBUG;
- public static final String ACTIVE_ENTITY_PROPERTY_NAME = "ActiveJob";
private final Map<EntityId, IActiveEntityEventsListener> entityEventListeners;
private final Map<JobId, EntityId> jobId2EntityId;
private boolean suspended = false;
@@ -97,10 +97,10 @@ public class ActiveNotificationHandler extends SingleThreadEventProcessor<Active
@Override
public void notifyJobCreation(JobId jobId, JobSpecification jobSpecification,
IJobCapacityController.JobSubmissionStatus status) throws HyracksDataException {
- Object property = jobSpecification.getProperty(ACTIVE_ENTITY_PROPERTY_NAME);
+ Object property = jobSpecification.getProperty(AsterixJobProperty.ACTIVE_ENTITY);
if (!(property instanceof EntityId)) {
if (property != null) {
- LOGGER.debug("{} is not an ingestion job. job property={}", jobId, property);
+ LOGGER.debug("{} is not an ingestion job. found entity={}", jobId, property);
}
return;
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java
index a18cd45d16..f2c4a9cd2a 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java
@@ -18,6 +18,9 @@
*/
package org.apache.asterix.app.active;
+import static org.apache.asterix.common.utils.AsterixJobProperty.ACTIVE_ENTITY;
+import static org.apache.hyracks.api.job.HyracksJobProperty.JOB_KIND;
+
import java.util.EnumSet;
import java.util.List;
@@ -45,6 +48,7 @@ import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartit
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
public class FeedEventsListener extends ActiveEntityEventsListener {
@@ -102,7 +106,8 @@ public class FeedEventsListener extends ActiveEntityEventsListener {
Pair<JobSpecification, AlgebricksAbsolutePartitionConstraint> jobInfo =
FeedOperations.buildStartFeedJob(mdProvider, feed, feedConnections, statementExecutor, hcc);
JobSpecification feedJob = jobInfo.getLeft();
- feedJob.setProperty(ActiveNotificationHandler.ACTIVE_ENTITY_PROPERTY_NAME, entityId);
+ feedJob.setProperty(ACTIVE_ENTITY, entityId);
+ feedJob.setProperty(JOB_KIND, JobKind.INGESTION);
// TODO(Yingyi): currently we do not check IFrameWriter protocol violations for Feed jobs.
// We will need to design general exception handling mechanism for feeds.
setLocations(jobInfo.getRight());
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalLibraryJobUtils.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalLibraryJobUtils.java
index 6c6691b674..4e7a558f83 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalLibraryJobUtils.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalLibraryJobUtils.java
@@ -42,6 +42,8 @@ import org.apache.hyracks.algebricks.common.utils.Pair;
import org.apache.hyracks.algebricks.common.utils.Triple;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.io.FileSplit;
+import org.apache.hyracks.api.job.HyracksJobProperty;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.dataflow.std.file.IFileSplitProvider;
@@ -67,6 +69,10 @@ public class ExternalLibraryJobUtils {
JobSpecification abortJobSpec =
createLibraryAbortJobSpec(dataverseName, libraryName, appCtx, splitsAndConstraint);
+ prepareJobSpec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
+ commitJobSpec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
+ abortJobSpec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
+
return new Triple<>(prepareJobSpec, commitJobSpec, abortJobSpec);
}
@@ -111,7 +117,7 @@ public class ExternalLibraryJobUtils {
AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(jobSpec, opDesc,
splitsAndConstraint.second);
jobSpec.addRoot(opDesc);
-
+ jobSpec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
return jobSpec;
}
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 dede4bea9c..8c5c9acc13 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
@@ -22,6 +22,7 @@ import static org.apache.asterix.common.api.IIdentifierMapper.Modifier.PLURAL;
import static org.apache.asterix.common.utils.IdentifierUtil.dataset;
import static org.apache.asterix.common.utils.IdentifierUtil.dataverse;
import static org.apache.asterix.lang.common.statement.CreateFullTextFilterStatement.FIELD_TYPE_STOPWORDS;
+import static org.apache.hyracks.api.job.HyracksJobProperty.JOB_KIND;
import java.io.File;
import java.io.FileInputStream;
@@ -255,6 +256,7 @@ import org.apache.hyracks.api.io.FileSplit;
import org.apache.hyracks.api.io.UnmanagedFileSplit;
import org.apache.hyracks.api.job.JobFlag;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.api.job.profiling.IOperatorStats;
import org.apache.hyracks.api.result.IResultSet;
@@ -3677,6 +3679,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
bActiveTxn = false;
if (spec != null && !isCompileOnly()) {
+ spec.setProperty(JOB_KIND, JobKind.DML);
runJob(hcc, spec);
}
} catch (Exception e) {
@@ -3733,7 +3736,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
ClientRequest clientRequest = (ClientRequest) requestTracker.get(reqParams.getRequestReference().getUuid());
if (stmtInsertUpsert.getReturnExpression() != null) {
deliverResult(hcc, resultSet, compiler, metadataProvider, locker, resultDelivery, outMetadata, stats,
- reqParams, false, clientRequest);
+ reqParams, false, clientRequest, JobKind.DML);
} else {
locker.lock();
try {
@@ -3741,6 +3744,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
if (jobSpec == null) {
return jobSpec;
}
+ jobSpec.setProperty(JOB_KIND, JobKind.DML);
runJob(hcc, jobSpec);
} finally {
locker.unlock();
@@ -3774,6 +3778,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
bActiveTxn = false;
if (jobSpec != null && !isCompileOnly()) {
+ jobSpec.setProperty(JOB_KIND, JobKind.DML);
runJob(hcc, jobSpec);
}
return jobSpec;
@@ -4736,19 +4741,19 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
}
};
deliverResult(hcc, resultSet, compiler, metadataProvider, locker, resultDelivery, outMetadata, stats,
- requestParameters, true, clientRequest);
+ requestParameters, true, clientRequest, JobKind.USER_QUERY);
}
private void deliverResult(IHyracksClientConnection hcc, IResultSet resultSet, IStatementCompiler compiler,
MetadataProvider metadataProvider, IMetadataLocker locker, ResultDelivery resultDelivery,
ResultMetadata outMetadata, Stats stats, IRequestParameters requestParameters, boolean cancellable,
- ClientRequest clientRequest) throws Exception {
+ ClientRequest clientRequest, JobKind jobKind) throws Exception {
final ResultSetId resultSetId = metadataProvider.getResultSetId();
switch (resultDelivery) {
case ASYNC:
MutableBoolean printed = new MutableBoolean(false);
executorService.submit(() -> asyncCreateAndRunJob(hcc, compiler, locker, resultDelivery,
- requestParameters, cancellable, resultSetId, printed, metadataProvider));
+ requestParameters, cancellable, resultSetId, printed, metadataProvider, jobKind));
synchronized (printed) {
while (!printed.booleanValue()) {
printed.wait();
@@ -4762,7 +4767,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
responsePrinter.addResultPrinter(new ResultsPrinter(appCtx, resultReader,
metadataProvider.findOutputRecordType(), stats, sessionOutput));
responsePrinter.printResults();
- }, requestParameters, cancellable, appCtx, metadataProvider);
+ }, requestParameters, cancellable, appCtx, metadataProvider, jobKind);
break;
case DEFERRED:
createAndRunJob(hcc, jobFlags, null, compiler, locker, resultDelivery, id -> {
@@ -4774,7 +4779,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
outMetadata.getResultSets().add(org.apache.commons.lang3.tuple.Triple.of(id, resultSetId,
metadataProvider.findOutputRecordType()));
}
- }, requestParameters, cancellable, appCtx, metadataProvider);
+ }, requestParameters, cancellable, appCtx, metadataProvider, jobKind);
break;
default:
break;
@@ -4802,7 +4807,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
private void asyncCreateAndRunJob(IHyracksClientConnection hcc, IStatementCompiler compiler, IMetadataLocker locker,
ResultDelivery resultDelivery, IRequestParameters requestParameters, boolean cancellable,
- ResultSetId resultSetId, MutableBoolean printed, MetadataProvider metadataProvider) {
+ ResultSetId resultSetId, MutableBoolean printed, MetadataProvider metadataProvider, JobKind jobKind) {
Mutable<JobId> jobId = new MutableObject<>(JobId.INVALID);
try {
createAndRunJob(hcc, jobFlags, jobId, compiler, locker, resultDelivery, id -> {
@@ -4814,7 +4819,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
printed.setTrue();
printed.notify();
}
- }, requestParameters, cancellable, appCtx, metadataProvider);
+ }, requestParameters, cancellable, appCtx, metadataProvider, jobKind);
} catch (Exception e) {
if (Objects.equals(JobId.INVALID, jobId.getValue())) {
// compilation failed
@@ -4857,7 +4862,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
private static void createAndRunJob(IHyracksClientConnection hcc, EnumSet<JobFlag> jobFlags, Mutable<JobId> jId,
IStatementCompiler compiler, IMetadataLocker locker, ResultDelivery resultDelivery, IResultPrinter printer,
IRequestParameters requestParameters, boolean cancellable, ICcApplicationContext appCtx,
- MetadataProvider metadataProvider) throws Exception {
+ MetadataProvider metadataProvider, JobKind jobKind) throws Exception {
final IRequestTracker requestTracker = appCtx.getRequestTracker();
final ClientRequest clientRequest =
(ClientRequest) requestTracker.get(requestParameters.getRequestReference().getUuid());
@@ -4876,6 +4881,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
// ensure request not cancelled before running job
ensureNotCancelled(clientRequest);
jobSpec.setRequestId(clientRequest.getId());
+ jobSpec.setProperty(JOB_KIND, jobKind);
final JobId jobId = JobUtils.runJob(hcc, jobSpec, jobFlags, false);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Created job {} for query uuid:{}, clientContextID:{}", jobId,
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/DataverseUtil.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/DataverseUtil.java
index f470949a08..141a735955 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/DataverseUtil.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/DataverseUtil.java
@@ -24,6 +24,8 @@ import org.apache.asterix.runtime.utils.RuntimeUtils;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraintHelper;
import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.apache.hyracks.api.job.HyracksJobProperty;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.dataflow.std.file.FileRemoveOperatorDescriptor;
import org.apache.hyracks.dataflow.std.file.IFileSplitProvider;
@@ -40,6 +42,7 @@ public class DataverseUtil {
FileRemoveOperatorDescriptor frod = new FileRemoveOperatorDescriptor(jobSpec, splitsAndConstraint.first, false);
AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(jobSpec, frod, splitsAndConstraint.second);
jobSpec.addRoot(frod);
+ jobSpec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
return jobSpec;
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
index 601cd0260e..78f9be580e 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
@@ -61,6 +61,8 @@ import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.api.dataflow.IConnectorDescriptor;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.job.HyracksJobProperty;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.dataflow.common.data.partition.FieldHashPartitionComputerFactory;
import org.apache.hyracks.dataflow.std.connectors.MToNPartitioningConnectorDescriptor;
@@ -344,6 +346,7 @@ public class RebalanceUtil {
spec.connect(new OneToOneConnectorDescriptor(spec), upsertOp, 0, commitOp, 0);
// Executes the job.
+ spec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DML);
JobUtils.runJob(hcc, spec, true);
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java
index 4e5d5319bf..e267ac2cf0 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java
@@ -42,6 +42,7 @@ import org.apache.asterix.app.nc.NCAppRuntimeContext;
import org.apache.asterix.app.result.ResponsePrinter;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.metadata.DataverseName;
+import org.apache.asterix.common.utils.AsterixJobProperty;
import org.apache.asterix.external.feed.watch.WaitForStateSubscriber;
import org.apache.asterix.external.operators.FeedIntakeOperatorNodePushable;
import org.apache.asterix.metadata.declared.MetadataProvider;
@@ -96,7 +97,7 @@ public class ActiveStatsTest {
// Mock JobSpecification
JobSpecification jobSpec = Mockito.mock(JobSpecification.class);
- Mockito.when(jobSpec.getProperty(ActiveNotificationHandler.ACTIVE_ENTITY_PROPERTY_NAME)).thenReturn(entityId);
+ Mockito.when(jobSpec.getProperty(AsterixJobProperty.ACTIVE_ENTITY)).thenReturn(entityId);
// Mock MetadataProvider
CCExtensionManager extensionManager = (CCExtensionManager) appCtx.getExtensionManager();
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestClusterControllerActor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestClusterControllerActor.java
index 883a0cb069..9ea8573dd5 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestClusterControllerActor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/TestClusterControllerActor.java
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.asterix.active.ActiveEvent;
import org.apache.asterix.active.EntityId;
import org.apache.asterix.app.active.ActiveNotificationHandler;
+import org.apache.asterix.common.utils.AsterixJobProperty;
import org.apache.asterix.metadata.declared.MetadataProvider;
import org.apache.asterix.metadata.entities.Dataset;
import org.apache.hyracks.api.job.JobId;
@@ -48,8 +49,7 @@ public class TestClusterControllerActor extends Actor {
protected void doExecute(MetadataProvider actorMdProvider) throws Exception {
// succeed
JobSpecification jobSpecification = Mockito.mock(JobSpecification.class);
- Mockito.when(jobSpecification.getProperty(ActiveNotificationHandler.ACTIVE_ENTITY_PROPERTY_NAME))
- .thenReturn(entityId);
+ Mockito.when(jobSpecification.getProperty(AsterixJobProperty.ACTIVE_ENTITY)).thenReturn(entityId);
handler.notifyJobCreation(jobId, jobSpecification, IJobCapacityController.JobSubmissionStatus.EXECUTE);
handler.notifyJobStart(jobId, null);
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/AsterixJobProperty.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/AsterixJobProperty.java
new file mode 100644
index 0000000000..760c5f7184
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/AsterixJobProperty.java
@@ -0,0 +1,26 @@
+/*
+ * 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 org.apache.hyracks.api.job.IJobProperty;
+
+public enum AsterixJobProperty implements IJobProperty {
+ ACTIVE_ENTITY,
+ GLOBAL_TX
+}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
index b2e68173cc..fa28415fe3 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
@@ -75,6 +75,8 @@ import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileSplit;
+import org.apache.hyracks.api.job.HyracksJobProperty;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
@@ -298,10 +300,8 @@ public class DatasetUtil {
public static JobSpecification dropDatasetJobSpec(Dataset dataset, MetadataProvider metadataProvider,
Set<IndexDropOperatorDescriptor.DropOption> options) throws AlgebricksException, ACIDException {
LOGGER.info("DROP DATASET: " + dataset);
- if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
- return RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
- }
JobSpecification specPrimary = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
+ specPrimary.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint =
metadataProvider.getSplitProviderAndConstraints(dataset);
IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
@@ -360,6 +360,7 @@ public class DatasetUtil {
AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, indexCreateOp,
splitsAndConstraint.second);
spec.addRoot(indexCreateOp);
+ spec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
return spec;
}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java
index 7e9cfc33bb..834d171ee0 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/IndexUtil.java
@@ -43,7 +43,9 @@ import org.apache.hyracks.algebricks.common.utils.Triple;
import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.job.HyracksJobProperty;
import org.apache.hyracks.api.job.IJobletEventListenerFactory;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.util.OptionalBoolean;
@@ -126,21 +128,27 @@ public class IndexUtil {
Dataset dataset, SourceLocation sourceLoc) throws AlgebricksException {
ISecondaryIndexOperationsHelper secondaryIndexHelper =
SecondaryIndexOperationsHelper.createIndexOperationsHelper(dataset, index, metadataProvider, sourceLoc);
- return secondaryIndexHelper.buildDropJobSpec(EnumSet.noneOf(DropOption.class));
+ JobSpecification spec = secondaryIndexHelper.buildDropJobSpec(EnumSet.noneOf(DropOption.class));
+ spec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
+ return spec;
}
public static JobSpecification buildDropIndexJobSpec(Index index, MetadataProvider metadataProvider,
Dataset dataset, Set<DropOption> options, SourceLocation sourceLoc) throws AlgebricksException {
ISecondaryIndexOperationsHelper secondaryIndexHelper =
SecondaryIndexOperationsHelper.createIndexOperationsHelper(dataset, index, metadataProvider, sourceLoc);
- return secondaryIndexHelper.buildDropJobSpec(options);
+ JobSpecification spec = secondaryIndexHelper.buildDropJobSpec(options);
+ spec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
+ return spec;
}
public static JobSpecification buildSecondaryIndexCreationJobSpec(Dataset dataset, Index index,
MetadataProvider metadataProvider, SourceLocation sourceLoc) throws AlgebricksException {
ISecondaryIndexOperationsHelper secondaryIndexHelper =
SecondaryIndexOperationsHelper.createIndexOperationsHelper(dataset, index, metadataProvider, sourceLoc);
- return secondaryIndexHelper.buildCreationJobSpec();
+ JobSpecification spec = secondaryIndexHelper.buildCreationJobSpec();
+ spec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DDL);
+ return spec;
}
public static JobSpecification buildSecondaryIndexLoadingJobSpec(Dataset dataset, Index index,
@@ -162,7 +170,9 @@ public class IndexUtil {
if (files != null) {
((SecondaryIndexOperationsHelper) secondaryIndexHelper).setExternalFiles(files);
}
- return secondaryIndexHelper.buildLoadingJobSpec();
+ JobSpecification spec = secondaryIndexHelper.buildLoadingJobSpec();
+ spec.setProperty(HyracksJobProperty.JOB_KIND, JobKind.DML);
+ return spec;
}
private static boolean supportsCorrelated(DatasetConfig.IndexType indexType) {
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/HyracksJobProperty.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/HyracksJobProperty.java
new file mode 100644
index 0000000000..d8bf5cba2d
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/HyracksJobProperty.java
@@ -0,0 +1,23 @@
+/*
+ * 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.job;
+
+public enum HyracksJobProperty implements IJobProperty {
+ JOB_KIND
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobProperty.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobProperty.java
new file mode 100644
index 0000000000..6d575796af
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobProperty.java
@@ -0,0 +1,24 @@
+/*
+ * 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.job;
+
+import java.io.Serializable;
+
+public interface IJobProperty extends Serializable {
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobKind.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobKind.java
new file mode 100644
index 0000000000..4c5e8692a9
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobKind.java
@@ -0,0 +1,29 @@
+/*
+ * 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.job;
+
+import java.io.Serializable;
+
+public enum JobKind implements Serializable {
+ USER_QUERY,
+ SYS_QUERY,
+ DDL,
+ DML,
+ INGESTION
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobSpecification.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobSpecification.java
index 2a11d79479..f23520652e 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobSpecification.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobSpecification.java
@@ -70,7 +70,7 @@ public class JobSpecification implements Serializable, IOperatorDescriptorRegist
private transient Map<Object, String> logical2PhysicalMap;
- private final Map<String, Serializable> properties;
+ private final Map<IJobProperty, Serializable> properties;
private final Set<Constraint> userConstraints;
@@ -155,12 +155,12 @@ public class JobSpecification implements Serializable, IOperatorDescriptorRegist
Pair.of(Pair.of(producerOp, producerPort), Pair.of(consumerOp, consumerPort)));
}
- public void setProperty(String name, Serializable value) {
- properties.put(name, value);
+ public void setProperty(IJobProperty property, Serializable value) {
+ properties.put(property, value);
}
- public Serializable getProperty(String name) {
- return properties.get(name);
+ public Serializable getProperty(IJobProperty property) {
+ return properties.get(property);
}
private <T> void extend(List<T> list, int index) {
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
index 64085bc180..031decdf1e 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
@@ -19,6 +19,7 @@
package org.apache.hyracks.control.cc.job;
+import java.io.Serializable;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.time.ZoneId;
@@ -38,7 +39,9 @@ import org.apache.hyracks.api.exceptions.HyracksException;
import org.apache.hyracks.api.exceptions.IError;
import org.apache.hyracks.api.exceptions.IFormattedException;
import org.apache.hyracks.api.job.ActivityClusterGraph;
+import org.apache.hyracks.api.job.HyracksJobProperty;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.JobKind;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.api.job.JobStatus;
import org.apache.hyracks.api.job.resource.IClusterCapacity;
@@ -75,6 +78,11 @@ public class JobManager implements IJobManager {
private final AtomicLong totalFailedJobs;
private final AtomicLong totalCancelledJobs;
private final AtomicLong totalRejectedJobs;
+ private long queryJobs;
+ private long sysQueryJobs;
+ private long ddlJobs;
+ private long dmlJobs;
+ private long ingestionJobs;
private IJobQueue jobQueue;
public JobManager(CCConfig ccConfig, ClusterControllerService ccs, IJobCapacityController jobCapacityController) {
@@ -244,9 +252,9 @@ public class JobManager implements IJobManager {
JobId jobId = run.getJobId();
Throwable caughtException = null;
CCServiceContext serviceCtx = ccs.getContext();
+ JobSpecification spec = run.getJobSpecification();
try {
- serviceCtx.notifyJobFinish(jobId, run.getJobSpecification(), run.getPendingStatus(),
- run.getPendingExceptions());
+ serviceCtx.notifyJobFinish(jobId, spec, run.getPendingStatus(), run.getPendingExceptions());
} catch (Exception e) {
LOGGER.error("Exception notifying job finish {}", jobId, e);
caughtException = e;
@@ -254,6 +262,7 @@ public class JobManager implements IJobManager {
run.setStatus(run.getPendingStatus(), run.getPendingExceptions());
run.setEndTime(System.currentTimeMillis());
if (activeRunMap.remove(jobId) != null) {
+ updateActiveJobCounts(spec, -1);
incrementJobCounters(run, successful);
// non-active jobs have zero capacity
@@ -283,6 +292,31 @@ public class JobManager implements IJobManager {
}
}
+ private void updateActiveJobCounts(JobSpecification spec, int delta) {
+ Serializable property = spec.getProperty(HyracksJobProperty.JOB_KIND);
+ if (property instanceof JobKind) {
+ switch ((JobKind) property) {
+ case USER_QUERY:
+ queryJobs += delta;
+ break;
+ case SYS_QUERY:
+ sysQueryJobs += delta;
+ break;
+ case DDL:
+ ddlJobs += delta;
+ break;
+ case DML:
+ dmlJobs += delta;
+ break;
+ case INGESTION:
+ ingestionJobs += delta;
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
/**
* Increments the job counters depending on the status
*
@@ -394,6 +428,7 @@ public class JobManager implements IJobManager {
run.setStartTimeZoneId(ZoneId.systemDefault().getId());
JobId jobId = run.getJobId();
logJobCapacity(run, "running", Level.INFO);
+ updateActiveJobCounts(run.getJobSpecification(), 1);
activeRunMap.put(jobId, run);
run.setStatus(JobStatus.RUNNING, null);
executeJobInternal(run);
@@ -452,10 +487,10 @@ public class JobManager implements IJobManager {
}
IReadOnlyClusterCapacity clusterCapacity = jobCapacityController.getClusterCapacity();
LOGGER.log(lvl,
- "{} {}, job memory={}, cpu={}, (new) cluster memory={}, cpu={}, currently running={}, queued={}",
+ "{} {}, job memory={} & cpu={}, (new) cluster memory={}, cpu={}, queued={}, currently running={}, user queries={}, system queries={}, ddls={}, dmls={}, ingestions={}",
jobStateDesc, jobRun.getJobId(), requiredMemory, requiredCPUs,
- clusterCapacity.getAggregatedMemoryByteSize(), clusterCapacity.getAggregatedCores(),
- getRunningJobsCount(), jobQueue.size());
+ clusterCapacity.getAggregatedMemoryByteSize(), clusterCapacity.getAggregatedCores(), jobQueue.size(),
+ getRunningJobsCount(), queryJobs, sysQueryJobs, ddlJobs, dmlJobs, ingestionJobs);
}
private void handleException(HyracksException ex) {