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/26 09:21:43 UTC
[ignite] branch master updated: IGNITE-6609: SQL: Consistent
expiration handling for PK hash index. This closes #6158.
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 af04149 IGNITE-6609: SQL: Consistent expiration handling for PK hash index. This closes #6158.
af04149 is described below
commit af041491423cc2c91668de3790a81e3631bcfa5c
Author: palmuhal <pa...@gmail.ru>
AuthorDate: Tue Feb 26 12:21:35 2019 +0300
IGNITE-6609: SQL: Consistent expiration handling for PK hash index. This closes #6158.
---
.../query/h2/database/H2PkHashIndex.java | 21 +-
.../processors/query/h2/database/H2TreeIndex.java | 23 ++-
.../cache/index/H2RowExpireTimeIndexSelfTest.java | 230 +++++++++++++++++++++
.../IgniteBinaryCacheQueryTestSuite.java | 2 +
4 files changed, 272 insertions(+), 4 deletions(-)
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2PkHashIndex.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2PkHashIndex.java
index 84393dd..b073da7 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2PkHashIndex.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2PkHashIndex.java
@@ -32,11 +32,12 @@ import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.tree.CacheDataRowStore;
import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2IndexBase;
-import org.apache.ignite.internal.processors.query.h2.opt.H2CacheRow;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2RowDescriptor;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
+import org.apache.ignite.internal.processors.query.h2.opt.H2CacheRow;
import org.apache.ignite.internal.processors.query.h2.opt.QueryContext;
import org.apache.ignite.internal.util.lang.GridCursor;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.spi.indexing.IndexingQueryCacheFilter;
import org.apache.ignite.spi.indexing.IndexingQueryFilter;
import org.h2.engine.Session;
@@ -215,6 +216,9 @@ public class H2PkHashIndex extends GridH2IndexBase {
/** */
private GridCursor<? extends CacheDataRow> curr;
+ /** Creation time of this cursor to check if some row is expired. */
+ private final long time;
+
/**
* @param iter Cursors iterator.
*/
@@ -224,6 +228,8 @@ public class H2PkHashIndex extends GridH2IndexBase {
this.iter = iter;
desc = rowDescriptor();
+
+ time = U.currentTimeMillis();
}
/** {@inheritDoc} */
@@ -251,9 +257,11 @@ public class H2PkHashIndex extends GridH2IndexBase {
for (;;) {
if (curr != null) {
while (curr.next()) {
+ CacheDataRow row = curr.get();
// Need to filter rows by value type because in a single cache
// we can have multiple indexed types.
- if (type.matchType(curr.get().value()))
+ // Also need to skip expired rows.
+ if (type.matchType(row.value()) && !wasExpired(row))
return true;
}
}
@@ -272,6 +280,15 @@ public class H2PkHashIndex extends GridH2IndexBase {
}
}
+ /**
+ * @param row to check.
+ * @return {@code true} if row was expired at the moment this cursor was created; {@code false} if not or if
+ * expire time is not set for this cursor.
+ */
+ private boolean wasExpired(CacheDataRow row) {
+ return row.expireTime() > 0 && row.expireTime() <= time;
+ }
+
/** {@inheritDoc} */
@Override public boolean previous() {
throw DbException.getUnsupportedException("previous");
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
index 263ae67..2a327e2 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
@@ -84,6 +84,7 @@ import org.h2.result.SearchRow;
import org.h2.table.IndexColumn;
import org.h2.table.TableFilter;
import org.h2.value.Value;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import static java.util.Collections.singletonList;
@@ -351,7 +352,10 @@ public class H2TreeIndex extends H2TreeIndexBase {
tree.compareRows((H2Row)lower, (H2Row)upper) == 0) {
H2Row row = tree.findOne((H2Row)lower, filter(qryCtxRegistry.getThreadLocal()), null);
- return (row == null) ? GridH2Cursor.EMPTY : new SingleRowCursor(row);
+ if (row == null || isExpired(row))
+ return GridH2Cursor.EMPTY;
+
+ return new SingleRowCursor(row);
}
else {
return new H2Cursor(tree.find((H2Row)lower,
@@ -450,13 +454,28 @@ public class H2TreeIndex extends H2TreeIndexBase {
H2Tree tree = treeForRead(threadLocalSegment());
QueryContext qctx = qryCtxRegistry.getThreadLocal();
- return new SingleRowCursor(b ? tree.findFirst(filter(qctx)): tree.findLast(filter(qctx)));
+ H2Row found = b ? tree.findFirst(filter(qctx)) : tree.findLast(filter(qctx));
+
+ if (found == null || isExpired(found))
+ return GridH2Cursor.EMPTY;
+
+ return new SingleRowCursor(found);
}
catch (IgniteCheckedException e) {
throw DbException.convert(e);
}
}
+ /**
+ * Determines if provided row can be treated as expired at the current moment.
+ *
+ * @param row row to check.
+ * @throws NullPointerException if provided row is {@code null}.
+ */
+ private static boolean isExpired(@NotNull H2Row row) {
+ return row.expireTime() > 0 && row.expireTime() <= U.currentTimeMillis();
+ }
+
/** {@inheritDoc} */
@Override public void destroy(boolean rmvIdx) {
try {
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/H2RowExpireTimeIndexSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/H2RowExpireTimeIndexSelfTest.java
new file mode 100644
index 0000000..ba9899a
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/H2RowExpireTimeIndexSelfTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.cache.index;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import javax.cache.expiry.CreatedExpiryPolicy;
+import javax.cache.expiry.Duration;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.processors.query.h2.H2Cursor;
+import org.apache.ignite.internal.processors.query.h2.database.H2TreeIndex;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.h2.engine.Session;
+import org.h2.result.SearchRow;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import static java.util.Arrays.asList;
+
+/**
+ * Since sql (unlike cache api) doesn't remove expired rows, we need to check that expired rows are filtered by the
+ * cursor. Background expired rows cleanup is turned off.
+ */
+public class H2RowExpireTimeIndexSelfTest extends GridCommonAbstractTest {
+ /** In so milliseconds since creation row can be treated as expired. */
+ private static final long EXPIRE_IN_MS_FROM_CREATE = 100L;
+
+ /** How many milliseconds we are going to wait til, cache data row become expired. */
+ private static final long WAIT_MS_TIL_EXPIRED = EXPIRE_IN_MS_FROM_CREATE * 2L;
+
+ /**
+ * Start only one grid.
+ */
+ @Override protected void beforeTestsStarted() throws Exception {
+ startGrids(1);
+ }
+
+ /**
+ * Cleans up before test.
+ */
+ @Before
+ public void dropTestCache() {
+ grid(0).destroyCache("notEager");
+ }
+
+ /**
+ * Create test cache. We use {@link CacheConfiguration#isEagerTtl()} == false to have expired rows in the cache for
+ * sure.
+ */
+ private IgniteCache<Integer, Integer> createTestCache() {
+ CacheConfiguration ccfg = defaultCacheConfiguration()
+ .setEagerTtl(false)
+ .setName("notEager")
+ .setQueryEntities(Collections.singleton(
+ new QueryEntity("java.lang.Integer", "java.lang.Integer")
+ .setKeyFieldName("id")
+ .setValueFieldName("val")
+ .addQueryField("id", Integer.class.getName(), null)
+ .addQueryField("val", Integer.class.getName(), null)
+ .setIndexes(Collections.singleton(new QueryIndex("val")))
+ .setTableName("Integer")));
+
+ IgniteCache<Integer, Integer> cache = grid(0).createCache(ccfg);
+
+ return cache;
+ }
+
+ /**
+ * Put values into the table with expire policy. Inserted row become expired in {@link #EXPIRE_IN_MS_FROM_CREATE}
+ * milliseconds.
+ *
+ * @param cache cache to put values in.
+ * @param key key of the row.
+ * @param val value of the row.
+ */
+ private void putExpiredSoon(IgniteCache cache, Integer key, Integer val) {
+ CreatedExpiryPolicy expireSinceCreated = new CreatedExpiryPolicy(new Duration(TimeUnit.MILLISECONDS,
+ EXPIRE_IN_MS_FROM_CREATE));
+
+ IgniteCache<Integer, Integer> expCache = cache.withExpiryPolicy(expireSinceCreated);
+
+ expCache.put(key, val);
+ }
+
+ /**
+ * Put values into the table with expire policy.
+ *
+ * @param cache cache to put values in.
+ * @param key key of the row.
+ * @param val value of the row.
+ */
+ private void putExpireInYear(IgniteCache cache, Integer key, Integer val) {
+ CreatedExpiryPolicy expireSinceCreated = new CreatedExpiryPolicy(new Duration(TimeUnit.MILLISECONDS,
+ TimeUnit.DAYS.toMillis(365)));
+
+ IgniteCache<Integer, Integer> expCache = cache.withExpiryPolicy(expireSinceCreated);
+
+ expCache.put(key, val);
+ }
+
+ /**
+ * Expired row check of the tree index in case {@link H2TreeIndex#find(Session, SearchRow, SearchRow)} optimizes
+ * returned cursor as SingleRowCursor.
+ */
+ @Test
+ public void testTreeIndexSingleRow() throws Exception {
+ IgniteCache<Integer, Integer> cache = createTestCache();
+
+ cache.put(1, 2);
+ cache.put(3, 4);
+
+ putExpireInYear(cache, 5, 6);
+
+ putExpiredSoon(cache, 42, 43);
+
+ U.sleep(WAIT_MS_TIL_EXPIRED);
+
+ {
+ List<List<?>> expired = cache.query(new SqlFieldsQuery("SELECT * FROM \"notEager\".Integer where _key = 42")).getAll();
+
+ Assert.assertTrue("Expired row should not be returned by sql. Result = " + expired, expired.isEmpty());
+ }
+
+ {
+ List<List<?>> expired = cache.query(new SqlFieldsQuery("SELECT * FROM \"notEager\".Integer where id >= 42 and id <= 42")).getAll();
+
+ Assert.assertTrue("Expired row should not be returned by sql. Result = " + expired, expired.isEmpty());
+ }
+
+ {
+ List<List<?>> expired = cache.query(new SqlFieldsQuery("SELECT * FROM \"notEager\".Integer where id >= 5 and id <= 5")).getAll();
+
+ assertEqualsCollections(Collections.singletonList(asList(5, 6)), expired);
+ }
+ }
+
+ /**
+ * Expired row check of the tree index in case {@link H2TreeIndex#find(Session, SearchRow, SearchRow)} doesn't
+ * perform one-row optimization and returns {@link H2Cursor}.
+ */
+ @Test
+ public void testTreeIndexManyRows() throws Exception {
+ IgniteCache<Integer, Integer> cache = createTestCache();
+
+ cache.put(1, 2);
+ cache.put(3, 4);
+
+ putExpireInYear(cache, 5, 6);
+
+ putExpiredSoon(cache, 42, 43);
+ putExpiredSoon(cache, 77, 88);
+
+ U.sleep(WAIT_MS_TIL_EXPIRED);
+
+ {
+ List<List<?>> mixed = cache.query(new SqlFieldsQuery("SELECT * FROM \"notEager\".Integer WHERE id >= 5")).getAll();
+
+ assertEqualsCollections(Collections.singletonList(asList(5, 6)), mixed);
+ }
+
+ {
+ List<List<?>> mixed = cache.query(new SqlFieldsQuery("SELECT * FROM \"notEager\".Integer WHERE id >= 3")).getAll();
+
+ assertEqualsCollections(asList(asList(3, 4), asList(5, 6)), mixed);
+ }
+
+
+ {
+ List<List<?>> expired = cache.query(new SqlFieldsQuery("SELECT * FROM \"notEager\".Integer WHERE id >= 42")).getAll();
+
+ Assert.assertTrue("Expired row should not be returned by sql. Result = " + expired, expired.isEmpty());
+ }
+ }
+
+ /**
+ * Expired row check if hash index is used.
+ */
+ @Test
+ public void testHashIndex() throws Exception {
+ IgniteCache<Integer, Integer> cache = createTestCache();
+
+ cache.put(1, 2);
+ cache.put(3, 4);
+
+ putExpireInYear(cache, 5, 6);
+
+ putExpiredSoon(cache, 42, 43);
+ putExpiredSoon(cache, 77, 88);
+
+ U.sleep(WAIT_MS_TIL_EXPIRED);
+
+ List<List<?>> mixed = cache.query(new SqlFieldsQuery(
+ "SELECT * FROM \"notEager\".Integer USE INDEX (\"_key_PK_hash\")")).getAll();
+
+ List<List<Integer>> exp = asList(
+ asList(1, 2),
+ asList(3, 4),
+ asList(5, 6));
+
+ assertEqualsCollections(exp, mixed);
+
+ List<List<?>> expired = cache.query(new SqlFieldsQuery(
+ "SELECT * FROM \"notEager\".Integer USE INDEX (\"_key_PK_hash\") WHERE id >= 42 and id <= 42")).getAll();
+
+ Assert.assertTrue("Expired row should not be returned by sql. Result = " + expired, expired.isEmpty());
+ }
+
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
index 1753356..85aa86c 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite.java
@@ -154,6 +154,7 @@ import org.apache.ignite.internal.processors.cache.index.H2DynamicIndexingComple
import org.apache.ignite.internal.processors.cache.index.H2DynamicTableSelfTest;
import org.apache.ignite.internal.processors.cache.index.H2RowCachePageEvictionTest;
import org.apache.ignite.internal.processors.cache.index.H2RowCacheSelfTest;
+import org.apache.ignite.internal.processors.cache.index.H2RowExpireTimeIndexSelfTest;
import org.apache.ignite.internal.processors.cache.index.IgniteDecimalSelfTest;
import org.apache.ignite.internal.processors.cache.index.LongIndexNameTest;
import org.apache.ignite.internal.processors.cache.index.OptimizedMarshallerIndexNameTest;
@@ -508,6 +509,7 @@ import org.junit.runners.Suite;
// H2 Rows on-heap cache
H2RowCacheSelfTest.class,
H2RowCachePageEvictionTest.class,
+ H2RowExpireTimeIndexSelfTest.class,
// User operation SQL
SqlParserUserSelfTest.class,