You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "zstan (via GitHub)" <gi...@apache.org> on 2024/11/11 13:28:26 UTC
[PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
zstan opened a new pull request, #4699:
URL: https://github.com/apache/ignite-3/pull/4699
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan merged PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837938995
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
thanks !
1. not applicable of course ))
2. i think about such approach too, but supplier over future seems over complicated as for me
, but i can implement in this way of course...
do you see any risks in my current approach ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1840089399
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
Review Comment:
What exactly was done?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1839611209
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
Review Comment:
```suggestion
List<List<Object>> res = sql(0, query, EXECUTING.name());
assertThat(res, Matchers.hasSize(1));
```
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
Review Comment:
why not just check ROW data against STATE of execution?
I mean
```
verifyQueryInfo(res.get(0), await(execution.stateAsync()));
```
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
Review Comment:
the query already contains a qualified table name, so specifying the default schema is redundant
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ // Second Job call on different node.
+ job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+
+ tsBefore = clockService.now().getPhysical();
+
+ targetNode = CLUSTER.node(1);
+
+ execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ tsAfter = clockService.now().getPhysical();
+
+ query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE COORDINATOR_NODE_ID = ? AND STATUS = ?";
+
+ res = sql(0, query, targetNode.name(), EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
Review Comment:
Since we are not only displaying currently running jobs, maybe we should also check the view data after a job is cancelled?
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlQueriesSystemViewTest.java:
##########
@@ -220,7 +220,7 @@ public void multiStatementWithTransaction() throws InterruptedException {
long timeAfter = clockService.now().getPhysical();
// "DDL" and "EXPLAIN" queries close cursor automatically.
- waitForCondition(() -> queryProcessor().runningQueries() == 4, 5_000);
+ assertTrue(waitForCondition(() -> queryProcessor().runningQueries() == 4, 5_000));
Review Comment:
According to the author's idea :sunglasses: , an error should be displayed, where the expected and actual value will be indicated (see the next line of code). But after this change, there will simply be an error without such information, so I would advise you to roll back this change.
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ // Second Job call on different node.
+ job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+
+ tsBefore = clockService.now().getPhysical();
+
+ targetNode = CLUSTER.node(1);
+
+ execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ tsAfter = clockService.now().getPhysical();
+
+ query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE COORDINATOR_NODE_ID = ? AND STATUS = ?";
+
+ res = sql(0, query, targetNode.name(), EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+ } finally {
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningBroadcasts(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(CLUSTER.node(0)).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ Map<ClusterNode, JobExecution<Void>> execution = entryNode.compute().submitBroadcast(
+ Set.of(clusterNode(CLUSTER.node(0)), clusterNode(CLUSTER.node(1))), job, null);
+
+ execution.forEach((k, exec) -> await().until(exec::stateAsync, willBe(jobStateWithStatus(EXECUTING))));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ assertThat(res.size(), is(2));
+ List<String> execNodes = List.of(CLUSTER.node(0).name(), CLUSTER.node(1).name());
+
+ verifyQueryInfo(res.get(0), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+ verifyQueryInfo(res.get(1), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+
+ execution.forEach((k, exec) -> IgniteTestUtils.await(exec.cancelAsync()));
+ } finally {
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningMapReduceTask(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ TaskExecution<Void> execution = entryNode.compute()
+ .submitMapReduce(TaskDescriptor.builder(MapReduceTaskCustom.class).build(), null);
+
+ await().until(execution::stateAsync, willBe(taskStateWithStatus(TaskStatus.EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ assertThat(res.size(), is(2));
+ List<String> execNodes = List.of(CLUSTER.node(0).name(), CLUSTER.node(1).name());
+
+ verifyQueryInfo(res.get(0), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+ verifyQueryInfo(res.get(1), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+ } finally {
+ closeQuiet(entryNode);
+ }
+ }
+
+ private static ClusterNode clusterNode(Ignite node) {
+ return unwrapIgniteImpl(node).node();
+ }
+
+ private static class InfiniteMapReduceJob implements ComputeJob<Void, Void> {
+ @Override
+ public CompletableFuture<Void> executeAsync(JobExecutionContext context, Void input) {
+ return new CompletableFuture<>();
+ }
+ }
+
+ private static class MapReduceTaskCustom implements MapReduceTask<Void, Void, Void, Void> {
+ @Override
+ public CompletableFuture<List<MapReduceJob<Void, Void>>> splitAsync(TaskExecutionContext taskContext, @Nullable Void input) {
+ return completedFuture(List.of(
+ MapReduceJob.<Void, Void>builder()
+ .jobDescriptor(JobDescriptor.builder(InfiniteMapReduceJob.class).build())
+ .nodes(taskContext.ignite().clusterNodes())
+ .build()
+ ));
+ }
+
+ @Override
+ public CompletableFuture<Void> reduceAsync(TaskExecutionContext taskContext, Map<UUID, Void> results) {
+ return completedFuture(null);
+ }
+ }
+
+ private static class InfiniteJob implements ComputeJob<Void, Void> {
+ @Override
+ public @Nullable CompletableFuture<Void> executeAsync(JobExecutionContext context, @Nullable Void arg) {
+ while (true) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ // No op, just return from loop
+ break;
+ }
+ }
+
+ return null;
+ }
+ }
+
+ private static void verifyQueryInfo(
Review Comment:
may be it's better to rename this method to something like
`verifyComputeJobState` or `verifyExecutionState` or `verifyViewRowData`.
The method with this name in `ItSqlQueriesSystemViewTest` checks "query info" (containing in `QueryInfo` class).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1839779266
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlQueriesSystemViewTest.java:
##########
@@ -220,7 +220,7 @@ public void multiStatementWithTransaction() throws InterruptedException {
long timeAfter = clockService.now().getPhysical();
// "DDL" and "EXPLAIN" queries close cursor automatically.
- waitForCondition(() -> queryProcessor().runningQueries() == 4, 5_000);
+ assertTrue(waitForCondition(() -> queryProcessor().runningQueries() == 4, 5_000));
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1839893806
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ // Second Job call on different node.
+ job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+
+ tsBefore = clockService.now().getPhysical();
+
+ targetNode = CLUSTER.node(1);
+
+ execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ tsAfter = clockService.now().getPhysical();
+
+ query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE COORDINATOR_NODE_ID = ? AND STATUS = ?";
+
+ res = sql(0, query, targetNode.name(), EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
Review Comment:
in such a case we need to check inner compute api invariants, i don\`t like it ..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837904566
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I found two possible ways to re-use `SubscriptionUtils`:
1. Works out of the box, but uses "join()" :thinking:
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(List::iterator).join()
);
```
2. Need to replace `CompletableFuture<Iterable<T>>` with `Supplier<CompletableFuture<Iterable<T>>>` in `IterableToPublisherAdapter`.
After that iIn method subscribe - we instead of
````
CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
```
call
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.get().thenApply(Iterable::iterator);
```
After that we can pass Supplier to completable future.
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(states -> states::iterator)
);
Personally I vote for the second approach with supplier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837491598
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I can`t use utility method here, if i miss smth plz show me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1838077634
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
@xtern i fix it like @lowka suggested
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837997147
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I just think that you are not the last person who needs such a publisher and it would be better to add such a feature.
But I agree that Supplier<CompletableFuture<... looks over-complicated, Mb need to think a bit more.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837904566
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I found two possible ways to re-use `SubscriptionUtils`:
1. Works out of the box, but uses "join()" :pensive:
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(List::iterator).join()
);
```
2. Need to replace `CompletableFuture<Iterable<T>>` with `Supplier<CompletableFuture<Iterable<T>>>` in `IterableToPublisherAdapter`.
After that in "subscribe()" instead of
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
```
call
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.get().thenApply(Iterable::iterator);
```
After that we can pass Supplier with CompletableFuture.
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(states -> states::iterator)
);
```
Personally I prefer the second approach with supplier, but I'm not sure about that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1839780989
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ // Second Job call on different node.
+ job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+
+ tsBefore = clockService.now().getPhysical();
+
+ targetNode = CLUSTER.node(1);
+
+ execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ tsAfter = clockService.now().getPhysical();
+
+ query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE COORDINATOR_NODE_ID = ? AND STATUS = ?";
+
+ res = sql(0, query, targetNode.name(), EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+ } finally {
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningBroadcasts(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(CLUSTER.node(0)).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ Map<ClusterNode, JobExecution<Void>> execution = entryNode.compute().submitBroadcast(
+ Set.of(clusterNode(CLUSTER.node(0)), clusterNode(CLUSTER.node(1))), job, null);
+
+ execution.forEach((k, exec) -> await().until(exec::stateAsync, willBe(jobStateWithStatus(EXECUTING))));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ assertThat(res.size(), is(2));
+ List<String> execNodes = List.of(CLUSTER.node(0).name(), CLUSTER.node(1).name());
+
+ verifyQueryInfo(res.get(0), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+ verifyQueryInfo(res.get(1), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+
+ execution.forEach((k, exec) -> IgniteTestUtils.await(exec.cancelAsync()));
+ } finally {
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningMapReduceTask(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ TaskExecution<Void> execution = entryNode.compute()
+ .submitMapReduce(TaskDescriptor.builder(MapReduceTaskCustom.class).build(), null);
+
+ await().until(execution::stateAsync, willBe(taskStateWithStatus(TaskStatus.EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ assertThat(res.size(), is(2));
+ List<String> execNodes = List.of(CLUSTER.node(0).name(), CLUSTER.node(1).name());
+
+ verifyQueryInfo(res.get(0), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+ verifyQueryInfo(res.get(1), execNodes, EXECUTING.name(), tsBefore, tsAfter);
+
+ IgniteTestUtils.await(execution.cancelAsync());
+ } finally {
+ closeQuiet(entryNode);
+ }
+ }
+
+ private static ClusterNode clusterNode(Ignite node) {
+ return unwrapIgniteImpl(node).node();
+ }
+
+ private static class InfiniteMapReduceJob implements ComputeJob<Void, Void> {
+ @Override
+ public CompletableFuture<Void> executeAsync(JobExecutionContext context, Void input) {
+ return new CompletableFuture<>();
+ }
+ }
+
+ private static class MapReduceTaskCustom implements MapReduceTask<Void, Void, Void, Void> {
+ @Override
+ public CompletableFuture<List<MapReduceJob<Void, Void>>> splitAsync(TaskExecutionContext taskContext, @Nullable Void input) {
+ return completedFuture(List.of(
+ MapReduceJob.<Void, Void>builder()
+ .jobDescriptor(JobDescriptor.builder(InfiniteMapReduceJob.class).build())
+ .nodes(taskContext.ignite().clusterNodes())
+ .build()
+ ));
+ }
+
+ @Override
+ public CompletableFuture<Void> reduceAsync(TaskExecutionContext taskContext, Map<UUID, Void> results) {
+ return completedFuture(null);
+ }
+ }
+
+ private static class InfiniteJob implements ComputeJob<Void, Void> {
+ @Override
+ public @Nullable CompletableFuture<Void> executeAsync(JobExecutionContext context, @Nullable Void arg) {
+ while (true) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ // No op, just return from loop
+ break;
+ }
+ }
+
+ return null;
+ }
+ }
+
+ private static void verifyQueryInfo(
Review Comment:
i have nothing against current naming, it`s correct as for me, but ok i change it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837904566
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I found two possible ways to re-use `SubscriptionUtils`:
1. Works out of the box, but uses "join()" :pensive:
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(List::iterator).join()
);
```
2. Need to replace `CompletableFuture<Iterable<T>>` with `Supplier<CompletableFuture<Iterable<T>>>` in `IterableToPublisherAdapter`.
After that iIn method subscribe - we instead of
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
```
call
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.get().thenApply(Iterable::iterator);
```
After that we can pass Supplier to completable future.
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(states -> states::iterator)
);
Personally I vote for the second approach with supplier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837904566
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I found two possible ways to re-use `SubscriptionUtils`:
1. Works out of the box, but uses "join()" :pensive:
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(List::iterator).join()
);
```
2. Need to replace `CompletableFuture<Iterable<T>>` with `Supplier<CompletableFuture<Iterable<T>>>` in `IterableToPublisherAdapter`.
After that iIn method subscribe - we instead of
````
CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
```
call
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.get().thenApply(Iterable::iterator);
```
After that we can pass Supplier to completable future.
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(states -> states::iterator)
);
Personally I vote for the second approach with supplier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837997147
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I just think that you are not the last person who needs such a publisher and it would be nice to add such a feature.
But I agree that Supplier<CompletableFuture<... looks over-complicated, Mb need to think a bit more.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837904566
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I found two possible ways to re-use `SubscriptionUtils`:
1. Works out of the box, but uses "join()" :pensive:
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(List::iterator).join()
);
```
2. Need to replace `CompletableFuture<Iterable<T>>` with `Supplier<CompletableFuture<Iterable<T>>>` in `IterableToPublisherAdapter`.
After that in "subscribe()" instead of
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
```
call
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.get().thenApply(Iterable::iterator);
```
After that we can pass Supplier to completable future.
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(states -> states::iterator)
);
```
Personally I prefer the second approach with supplier, but I'm not sure about that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837904566
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I found two possible ways to re-use `SubscriptionUtils`:
1. Works out of the box, but uses "join()" :pensive:
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(List::iterator).join()
);
```
2. Need to replace `CompletableFuture<Iterable<T>>` with `Supplier<CompletableFuture<Iterable<T>>>` in `IterableToPublisherAdapter`.
After that in "subscribe()" instead of
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
```
call
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.get().thenApply(Iterable::iterator);
```
After that we can pass Supplier to completable future.
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(states -> states::iterator)
);
```
Personally I prefer the second approach with supplier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1840090900
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
Review Comment:
Got it - query has been changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1839889024
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
Review Comment:
done
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "korlov42 (via GitHub)" <gi...@apache.org>.
korlov42 commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1840101218
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,311 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
Review Comment:
I think we don't need to execute job just to check meta returned from cursor. I propose to simplify this test
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.List;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscriber;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.apache.ignite.internal.util.SubscriptionUtils;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final DelegatingPublisher publisher = new DelegatingPublisher();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ publisher.executionManager = executionManager;
+ }
+
+ public void stop() {
+ publisher.executionManager = null;
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
Review Comment:
out guidelines (see `modules/system-view-api/README.md`) suggests to explicitly specify generic type of the column:
```
Please note that we explicitly specify the generic type of the column. It is a good practice to detect the column type mismatch at compile time.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "xtern (via GitHub)" <gi...@apache.org>.
xtern commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1837904566
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeViewProvider.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.compute;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscription;
+import org.apache.ignite.compute.JobState;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposes information about currently running compute tasks hosted by a node. */
+public class ComputeViewProvider {
+ private final CompletableFuture<ExecutionManager> startFuture = new CompletableFuture<>();
+
+ /** Initializes with compute aware manager. */
+ public void init(ExecutionManager executionManager) {
+ startFuture.complete(executionManager);
+ }
+
+ /** Returns system view exposes information about currently running tasks. */
+ public SystemView<?> get() {
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<JobState>nodeViewBuilder()
+ .name("COMPUTE_TASKS")
+ .nodeNameColumnAlias("COORDINATOR_NODE_ID")
+ .addColumn("ID", idType, info -> info.id().toString())
+ .addColumn("STATUS", stringOf(10), info -> info.status().name())
+ .addColumn("CREATE_TIME", timestampType, JobState::createTime)
+ .addColumn("START_TIME", timestampType, JobState::startTime)
+ .addColumn("FINISH_TIME", timestampType, JobState::finishTime)
+ .dataProvider(buildDataProvider())
+ .build();
+ }
+
+ private Publisher<JobState> buildDataProvider() {
Review Comment:
I found two possible ways to re-use `SubscriptionUtils`:
1. Works out of the box, but uses "join()" :pensive:
```
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(List::iterator).join()
);
```
2. Need to replace `CompletableFuture<Iterable<T>>` with `Supplier<CompletableFuture<Iterable<T>>>` in `IterableToPublisherAdapter`.
After that in "subscribe()" instead of
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
```
call
```
CompletableFuture<Iterator<T>> itFuture = iterableFuture.get().thenApply(Iterable::iterator);
```
After that we can pass Supplier to completable future.
Publisher<JobState> publisher = SubscriptionUtils.fromIterable(
() -> startFuture.thenCompose(ExecutionManager::localStatesAsync).thenApply(states -> states::iterator)
);
Personally I vote for the second approach with supplier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-23423 Provide system view for running compute jobs [ignite-3]
Posted by "zstan (via GitHub)" <gi...@apache.org>.
zstan commented on code in PR #4699:
URL: https://github.com/apache/ignite-3/pull/4699#discussion_r1839891868
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
Review Comment:
this check is common for different compute api approaches, i don`t like idea do adopt it for such methods as :
`Map<ClusterNode, JobExecution<Void>> execution = entryNode.compute().submitBroadcast`
##########
modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.ignite.internal.sql.engine;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.compute.JobStatus.CANCELED;
+import static org.apache.ignite.compute.JobStatus.EXECUTING;
+import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
+import static org.apache.ignite.internal.sql.engine.util.Commons.closeQuiet;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStateMatcher.jobStateWithStatus;
+import static org.apache.ignite.internal.testframework.matchers.TaskStateMatcher.taskStateWithStatus;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasLength;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobDescriptor;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobTarget;
+import org.apache.ignite.compute.TaskDescriptor;
+import org.apache.ignite.compute.TaskStatus;
+import org.apache.ignite.compute.task.MapReduceJob;
+import org.apache.ignite.compute.task.MapReduceTask;
+import org.apache.ignite.compute.task.TaskExecution;
+import org.apache.ignite.compute.task.TaskExecutionContext;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.sql.ColumnType;
+import org.hamcrest.Matchers;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * End-to-end tests to verify {@code COMPUTE_TASKS} system view.
+ */
+public class ItComputeSystemViewTest extends BaseSqlIntegrationTest {
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void checkMeta(boolean isClient) {
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS";
+
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ try {
+ // Verify metadata.
+ assertQuery(query)
+ .withDefaultSchema("SYSTEM")
+ .columnMetadata(
+ new MetadataMatcher().name("COORDINATOR_NODE_ID").type(ColumnType.STRING).nullable(false),
+ new MetadataMatcher().name("ID").type(ColumnType.STRING).precision(36).nullable(true),
+ new MetadataMatcher().name("STATUS").type(ColumnType.STRING).nullable(true),
+ new MetadataMatcher().name("CREATE_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("START_TIME").type(ColumnType.TIMESTAMP).nullable(true),
+ new MetadataMatcher().name("FINISH_TIME").type(ColumnType.TIMESTAMP).nullable(true)
+ )
+ .check();
+ } finally {
+ execution.cancelAsync();
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(CANCELED)));
+
+ closeQuiet(entryNode);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void viewRunningJobs(boolean isClient) {
+ Ignite entryNode = isClient ? IgniteClient.builder().addresses("localhost").build() : CLUSTER.node(0);
+ Ignite targetNode = CLUSTER.node(0);
+
+ try {
+ ClockService clockService = unwrapIgniteImpl(targetNode).clockService();
+
+ long tsBefore = clockService.now().getPhysical();
+
+ JobDescriptor<Void, Void> job = JobDescriptor.builder(InfiniteJob.class).units(List.of()).build();
+ JobExecution<Void> execution = entryNode.compute().submit(JobTarget.node(clusterNode(targetNode)), job, null);
+
+ await().until(execution::stateAsync, willBe(jobStateWithStatus(EXECUTING)));
+
+ long tsAfter = clockService.now().getPhysical();
+
+ String query = "SELECT * FROM SYSTEM.COMPUTE_TASKS WHERE STATUS = ?";
+
+ List<List<Object>> res = sql(0, query, EXECUTING.name());
+
+ verifyQueryInfo(res.get(0), List.of(targetNode.name()), EXECUTING.name(), tsBefore, tsAfter);
Review Comment:
this check is common for different compute api approaches, i don\`t like idea do adopt it for such methods as :
`Map<ClusterNode, JobExecution<Void>> execution = entryNode.compute().submitBroadcast`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org