You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2019/02/08 09:26:34 UTC
[ignite] branch master updated: IGNITE-10347: SQL: system view with
running queries. This closes #5845.
This is an automated email from the ASF dual-hosted git repository.
vozerov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 9ec36bf IGNITE-10347: SQL: system view with running queries. This closes #5845.
9ec36bf is described below
commit 9ec36bf4d7875700bdf604e837254ddc353e9b87
Author: Yuriy Gerzhedovich <yg...@gridgain.com>
AuthorDate: Fri Feb 8 12:26:26 2019 +0300
IGNITE-10347: SQL: system view with running queries. This closes #5845.
---
.../processors/query/RunningQueryManager.java | 19 +++-
.../processors/query/h2/IgniteH2Indexing.java | 9 ++
.../processors/query/h2/SchemaManager.java | 2 +
.../h2/sys/view/SqlAbstractLocalSystemView.java | 33 ++++++-
.../h2/sys/view/SqlSystemViewRunningQueries.java | 101 ++++++++++++++++++++
.../processors/query/RunningQueriesTest.java | 2 +-
.../processors/query/SqlSystemViewsSelfTest.java | 105 +++++++++++++++++++++
7 files changed, 265 insertions(+), 6 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
index 0c7a5a2..807488c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
@@ -21,6 +21,7 @@ package org.apache.ignite.internal.processors.query;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -117,6 +118,22 @@ public class RunningQueryManager {
}
/**
+ * Return SQL queries which executing right now.
+ *
+ * @return List of SQL running queries.
+ */
+ public List<GridRunningQueryInfo> runningSqlQueries() {
+ List<GridRunningQueryInfo> res = new ArrayList<>();
+
+ for (GridRunningQueryInfo run : runs.values()) {
+ if (isSqlQuery(run))
+ res.add(run);
+ }
+
+ return res;
+ }
+
+ /**
* Check belongs running query to an SQL type.
*
* @param runningQryInfo Running query info object.
@@ -130,7 +147,7 @@ public class RunningQueryManager {
* Return long running user queries.
*
* @param duration Duration of long query.
- * @return List of queries which running longer than given duration.
+ * @return Collection of queries which running longer than given duration.
*/
public Collection<GridRunningQueryInfo> longRunningQueries(long duration) {
Collection<GridRunningQueryInfo> res = new ArrayList<>();
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index 5170c2a..6d346c0 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -2704,6 +2704,15 @@ public class IgniteH2Indexing implements GridQueryIndexing {
rdcQryExec.onDisconnected(reconnectFut);
}
+ /**
+ * Return SQL running queries.
+ *
+ * @return SQL running queries.
+ */
+ public List<GridRunningQueryInfo> runningSqlQueries() {
+ return runningQueryMgr.runningSqlQueries();
+ }
+
/** {@inheritDoc} */
@Override public Collection<GridRunningQueryInfo> runningQueries(long duration) {
return runningQueryMgr.longRunningQueries(duration);
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/SchemaManager.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/SchemaManager.java
index f681036..bf52d2f 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/SchemaManager.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/SchemaManager.java
@@ -56,6 +56,7 @@ import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewCach
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodeAttributes;
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodeMetrics;
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodes;
+import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewRunningQueries;
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewTables;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.util.typedef.F;
@@ -167,6 +168,7 @@ public class SchemaManager {
views.add(new SqlSystemViewCaches(ctx));
views.add(new SqlSystemViewCacheGroups(ctx));
views.add(new SqlSystemViewCacheGroupsIOStatistics(ctx));
+ views.add(new SqlSystemViewRunningQueries(ctx));
views.add(new SqlSystemViewTables(ctx, this));
return views;
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java
index 7070502..cfc1ba9 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java
@@ -17,22 +17,29 @@
package org.apache.ignite.internal.processors.query.h2.sys.view;
+import java.util.TimeZone;
import java.util.UUID;
import org.apache.ignite.internal.GridKernalContext;
import org.h2.engine.Session;
import org.h2.result.Row;
import org.h2.result.SearchRow;
import org.h2.table.Column;
+import org.h2.util.DateTimeUtils;
import org.h2.value.Value;
import org.h2.value.ValueNull;
import org.h2.value.ValueString;
import org.h2.value.ValueTime;
import org.h2.value.ValueTimestamp;
+import org.h2.value.ValueTimestampTimeZone;
/**
* Local system view base class (which uses only local node data).
*/
+@SuppressWarnings("IfMayBeConditional")
public abstract class SqlAbstractLocalSystemView extends SqlAbstractSystemView {
+
+ public static final int MILLIS_IN_MIN = 60_000;
+
/**
* @param tblName Table name.
* @param desc Description.
@@ -40,7 +47,7 @@ public abstract class SqlAbstractLocalSystemView extends SqlAbstractSystemView {
* @param indexes Indexes.
* @param cols Columns.
*/
- public SqlAbstractLocalSystemView(String tblName, String desc, GridKernalContext ctx, String[] indexes,
+ protected SqlAbstractLocalSystemView(String tblName, String desc, GridKernalContext ctx, String[] indexes,
Column... cols) {
super(tblName, desc, ctx, cols, indexes);
@@ -56,7 +63,8 @@ public abstract class SqlAbstractLocalSystemView extends SqlAbstractSystemView {
* @param indexedCols Indexed columns.
* @param cols Columns.
*/
- public SqlAbstractLocalSystemView(String tblName, String desc, GridKernalContext ctx, String indexedCols, Column... cols) {
+ protected SqlAbstractLocalSystemView(String tblName, String desc, GridKernalContext ctx, String indexedCols,
+ Column... cols) {
this(tblName, desc, ctx, new String[] {indexedCols}, cols);
}
@@ -66,8 +74,9 @@ public abstract class SqlAbstractLocalSystemView extends SqlAbstractSystemView {
* @param ctx Context.
* @param cols Columns.
*/
- public SqlAbstractLocalSystemView(String tblName, String desc, GridKernalContext ctx, Column ... cols) {
- this(tblName, desc, ctx, new String[] {}, cols);
+ @SuppressWarnings("ZeroLengthArrayAllocation")
+ protected SqlAbstractLocalSystemView(String tblName, String desc, GridKernalContext ctx, Column ... cols) {
+ this(tblName, desc, ctx, new String[] {} , cols);
}
/**
@@ -158,4 +167,20 @@ public abstract class SqlAbstractLocalSystemView extends SqlAbstractSystemView {
else
return ValueTimestamp.fromMillis(millis);
}
+
+ /**
+ * Converts millis to H2 ValueTimestamp in default time zone.
+ *
+ * @param millis Millis.
+ */
+ protected static Value valueTimestampZoneFromMillis(long millis) {
+ long dateVal = DateTimeUtils.dateValueFromDate(millis);
+ long nanos = DateTimeUtils.nanosFromDate(millis);
+ int tzOff = TimeZone.getDefault().getRawOffset();
+
+ if(tzOff % MILLIS_IN_MIN == 0)
+ return ValueTimestampTimeZone.fromDateValueAndNanos(dateVal, nanos, (short)(tzOff / MILLIS_IN_MIN));
+ else
+ return DateTimeUtils.timestampTimeZoneFromLocalDateValueAndNanos(dateVal, nanos);
+ }
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlSystemViewRunningQueries.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlSystemViewRunningQueries.java
new file mode 100644
index 0000000..d53056e
--- /dev/null
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlSystemViewRunningQueries.java
@@ -0,0 +1,101 @@
+/*
+ * 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.processors.query.h2.sys.view;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.processors.query.GridRunningQueryInfo;
+import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
+import org.h2.engine.Session;
+import org.h2.result.Row;
+import org.h2.result.SearchRow;
+import org.h2.value.Value;
+
+/**
+ * System view: running queries
+ */
+public class SqlSystemViewRunningQueries extends SqlAbstractLocalSystemView {
+ /**
+ * @param ctx Grid context.
+ */
+ public SqlSystemViewRunningQueries(GridKernalContext ctx) {
+ super("LOCAL_SQL_RUNNING_QUERIES", "Running queries", ctx, new String[] {"QUERY_ID"},
+ newColumn("QUERY_ID"),
+ newColumn("SQL"),
+ newColumn("SCHEMA_NAME"),
+ newColumn("LOCAL", Value.BOOLEAN),
+ newColumn("START_TIME", Value.TIMESTAMP_TZ),
+ newColumn("DURATION", Value.LONG)
+ );
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterator<Row> getRows(Session ses, SearchRow first, SearchRow last) {
+ SqlSystemViewColumnCondition qryIdCond = conditionForColumn("QUERY_ID", first, last);
+
+ List<GridRunningQueryInfo> runningSqlQueries = ((IgniteH2Indexing)ctx.query().getIndexing()).runningSqlQueries();
+
+ if (qryIdCond.isEquality()) {
+ String qryId = qryIdCond.valueForEquality().getString();
+
+ runningSqlQueries = runningSqlQueries.stream()
+ .filter((r) -> r.globalQueryId().equals(qryId))
+ .findFirst()
+ .map(Collections::singletonList)
+ .orElse(Collections.emptyList());
+ }
+
+ if (runningSqlQueries.isEmpty())
+ return Collections.emptyIterator();
+
+ long now = System.currentTimeMillis();
+
+ List<Row> rows = new ArrayList<>(runningSqlQueries.size());
+
+ for (GridRunningQueryInfo info : runningSqlQueries) {
+ long duration = now - info.startTime();
+
+ rows.add(
+ createRow(ses,
+ info.globalQueryId(),
+ info.query(),
+ info.schemaName(),
+ info.local(),
+ valueTimestampZoneFromMillis(info.startTime()),
+ duration
+ )
+ );
+ }
+
+ return rows.iterator();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean canGetRowCount() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getRowCount() {
+ return ((IgniteH2Indexing)ctx.query().getIndexing()).runningSqlQueries().size();
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/RunningQueriesTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/RunningQueriesTest.java
index 24e383d..15c5abe 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/RunningQueriesTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/RunningQueriesTest.java
@@ -109,6 +109,7 @@ public class RunningQueriesTest extends AbstractIndexingCommonTest {
IgniteCache<Integer, Integer> cache = ignite.getOrCreateCache(new CacheConfiguration<Integer, Integer>()
.setName(DEFAULT_CACHE_NAME)
.setQueryEntities(Collections.singletonList(new QueryEntity(Integer.class, Integer.class)))
+ .setIndexedTypes(Integer.class,Integer.class)
);
cache.put(100000, 0);
@@ -242,7 +243,6 @@ public class RunningQueriesTest extends AbstractIndexingCommonTest {
});
assertNoRunningQueries();
-
}
/**
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java
index d4e9278..3c384c0 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.query;
import java.sql.Time;
import java.sql.Timestamp;
+import java.time.LocalDateTime;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@@ -39,7 +40,10 @@ import org.apache.ignite.cache.affinity.AffinityKeyMapper;
import org.apache.ignite.cache.eviction.EvictableEntry;
import org.apache.ignite.cache.eviction.EvictionFilter;
import org.apache.ignite.cache.eviction.EvictionPolicy;
+import org.apache.ignite.cache.query.FieldsQueryCursor;
+import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.cluster.ClusterMetrics;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
@@ -61,9 +65,11 @@ import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.testframework.GridTestUtils;
+import org.h2.api.TimestampWithTimeZone;
import org.junit.Test;
import static java.util.Arrays.asList;
+import static org.junit.Assert.assertNotEquals;
/**
* Tests for ignite SQL system views.
@@ -187,6 +193,105 @@ public class SqlSystemViewsSelfTest extends AbstractIndexingCommonTest {
assertEquals(nodeId, ((List<?>)cache.query(qry).getAll().get(0)).get(0));
}
+
+ /**
+ * Test running queries system view.
+ */
+ @Test
+ public void testRunningQueriesView() throws Exception {
+ IgniteEx ignite = startGrid(0);
+
+ IgniteCache cache = ignite.createCache(
+ new CacheConfiguration<>(DEFAULT_CACHE_NAME).setIndexedTypes(Integer.class, String.class)
+ );
+
+ cache.put(100,"200");
+
+ String sql = "SELECT SQL, QUERY_ID, SCHEMA_NAME, LOCAL, START_TIME, DURATION FROM IGNITE.LOCAL_SQL_RUNNING_QUERIES";
+
+ FieldsQueryCursor notClosedFieldQryCursor = cache.query(new SqlFieldsQuery(sql).setLocal(true));
+
+ List<?> cur = cache.query(new SqlFieldsQuery(sql).setLocal(true)).getAll();
+
+ assertEquals(2, cur.size());
+
+ List<?> res0 = (List<?>)cur.get(0);
+ List<?> res1 = (List<?>)cur.get(1);
+
+ TimestampWithTimeZone tsTz = (TimestampWithTimeZone)res0.get(4);
+
+ LocalDateTime now = LocalDateTime.now();
+
+ int sysTZOff = TimeZone.getDefault().getOffset(System.currentTimeMillis()) / 60 / 1000;
+
+ assertEquals(sysTZOff, tsTz.getTimeZoneOffsetMins());
+
+ long diffInMinutes = Math.abs((tsTz.getNanosSinceMidnight() / 1_000_000_000 / 60) - (now.getHour() * 60 + now.getMinute()));
+
+ assertTrue(diffInMinutes < 3);
+
+ assertEquals(sql, res0.get(0));
+
+ assertEquals(sql, res1.get(0));
+
+ assertTrue((Boolean)res0.get(3));
+
+ String id0 = (String)res0.get(1);
+ String id1 = (String)res1.get(1);
+
+ assertNotEquals(id0, id1);
+
+
+ String qryPrefix = ignite.localNode().id() + "_";
+
+ String qryId1 = qryPrefix + "1";
+ String qryId2 = qryPrefix + "2";
+
+ assertTrue(id0.equals(qryId1) || id1.equals(qryId1));
+
+ assertTrue(id0.equals(qryId2) || id1.equals(qryId2));
+
+ assertEquals(2, cache.query(new SqlFieldsQuery(sql)).getAll().size());
+
+ notClosedFieldQryCursor.close();
+
+ assertEquals(1, cache.query(new SqlFieldsQuery(sql)).getAll().size());
+
+ cache.put(100,"200");
+
+ QueryCursor notClosedQryCursor = cache.query(new SqlQuery<>(String.class, "_key=100"));
+
+ String expSqlQry = "SELECT \"default\".\"STRING\"._KEY, \"default\".\"STRING\"._VAL FROM " +
+ "\"default\".\"STRING\" WHERE _key=100";
+
+ cur = cache.query(new SqlFieldsQuery(sql)).getAll();
+
+ assertEquals(2, cur.size());
+
+ res0 = (List<?>)cur.get(0);
+ res1 = (List<?>)cur.get(1);
+
+ assertTrue(expSqlQry, res0.get(0).equals(expSqlQry) || res1.get(0).equals(expSqlQry));
+
+ assertFalse((Boolean)res0.get(3));
+
+ assertFalse((Boolean)res1.get(3));
+
+ notClosedQryCursor.close();
+
+ sql = "SELECT SQL, QUERY_ID FROM IGNITE.LOCAL_SQL_RUNNING_QUERIES WHERE QUERY_ID='" + qryPrefix + "7'";
+
+ assertEquals(qryPrefix + "7", ((List<?>)cache.query(new SqlFieldsQuery(sql)).getAll().get(0)).get(1));
+
+ sql = "SELECT SQL FROM IGNITE.LOCAL_SQL_RUNNING_QUERIES WHERE DURATION > 100000";
+
+ assertTrue(cache.query(new SqlFieldsQuery(sql)).getAll().isEmpty());
+
+ sql = "SELECT SQL FROM IGNITE.LOCAL_SQL_RUNNING_QUERIES WHERE QUERY_ID='UNKNOWN'";
+
+ assertTrue(cache.query(new SqlFieldsQuery(sql)).getAll().isEmpty());
+ }
+
/**
* Test that we can't use cache tables and system views in the same query.
*/