This is an automated email from the ASF dual-hosted git repository.
ibessonov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 22e2df000b IGNITE-19394 "peek" contract improved to avoid missing data
on RW scans. (#2021)
22e2df000b is described below
commit 22e2df000b1e34c9594dd957fffb74879f3386ef
Author: Ivan Bessonov <[email protected]>
AuthorDate: Fri May 5 14:59:33 2023 +0300
IGNITE-19394 "peek" contract improved to avoid missing data on RW scans.
(#2021)
---
.../ignite/internal/storage/index/PeekCursor.java | 18 +-
.../index/AbstractHashIndexStorageTest.java | 229 ++----------
.../storage/index/AbstractIndexStorageTest.java | 383 +++++++++++++++++++++
.../index/AbstractSortedIndexStorageTest.java | 370 +++++---------------
.../index/impl/BinaryTupleRowSerializer.java | 16 +-
.../storage/index/impl/TestHashIndexStorage.java | 89 +++--
.../storage/index/impl/TestSortedIndexStorage.java | 181 ++++------
.../index/sorted/PageMemorySortedIndexStorage.java | 76 ++--
.../AbstractPageMemoryHashIndexStorageTest.java | 39 ++-
.../rocksdb/index/RocksDbSortedIndexStorage.java | 55 ++-
10 files changed, 718 insertions(+), 738 deletions(-)
diff --git
a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/PeekCursor.java
b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/PeekCursor.java
index 2ef036f690..088d69d59d 100644
---
a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/PeekCursor.java
+++
b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/PeekCursor.java
@@ -27,7 +27,21 @@ public interface PeekCursor<T> extends Cursor<T> {
/**
* Returns the next element without advancing the cursor, {@code null} if
there is no next element.
*
- * <p>Usage notes:
+ * <p>The behavior of this method is tightly coupled with the behaviour of
{@link #next()} and {@link #hasNext()}:
+ * <ul>
+ * <li>
+ * {@code peek()} returns exactly the same value as the {@link
#next()} would, if called instead or right after.
+ * </li>
+ * <li>
+ * This property can be applied to the end of the cursor. {@code
peek()} returns {@code null} if {@link #next()} would
+ * throw an {@link java.util.NoSuchElementException}, if it was
called instead or right after.
+ * </li>
+ * <li>
+ * {@link #next()}, called immediately after {@code peek()}, must
thus return the same value, or throw an exception if the
+ * value was {@code null}.
+ * </li>
+ * </ul>
+ * These properties, in conjunction with basic iterator invariants, give
strict constraints on {@link #hasNext()} usage:
* <ul>
* <li>After the cursor is created, {@code #peek()} will return the
actual (up-to-date) next element;</li>
* <li>After calling {@link #hasNext()}, if it returned {@code true},
then {@code peek()} will return the element (cached) that
@@ -36,6 +50,8 @@ public interface PeekCursor<T> extends Cursor<T> {
* <li>After {@link #next()} is called, but before {@link #hasNext()}
is called, {@code peek()} will always return the actual
* (up-to-date) next element without advancing the cursor.</li>
* </ul>
+ * In other words, {@link #hasNext()} call forces the cursor to return
cached value from {@code peek()}, instead of looking for
+ * up-to-date value in the storage.
*/
@Nullable T peek();
}
diff --git
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractHashIndexStorageTest.java
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractHashIndexStorageTest.java
index efde8f475a..8fc9e4c0c8 100644
---
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractHashIndexStorageTest.java
+++
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractHashIndexStorageTest.java
@@ -17,227 +17,77 @@
package org.apache.ignite.internal.storage.index;
-import static java.util.stream.Collectors.toList;
import static
org.apache.ignite.internal.schema.testutils.SchemaConfigurationConverter.addIndex;
-import static
org.apache.ignite.internal.schema.testutils.SchemaConfigurationConverter.convert;
-import static
org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders.column;
-import static
org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders.tableBuilder;
-import static
org.apache.ignite.internal.storage.BaseMvStoragesTest.getOrCreateMvPartition;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.empty;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
import org.apache.ignite.internal.configuration.util.ConfigurationUtil;
-import org.apache.ignite.internal.schema.configuration.TableConfiguration;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.schema.configuration.index.TableIndexView;
import org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders;
-import org.apache.ignite.internal.schema.testutils.definition.ColumnDefinition;
import org.apache.ignite.internal.schema.testutils.definition.ColumnType;
-import org.apache.ignite.internal.schema.testutils.definition.TableDefinition;
import
org.apache.ignite.internal.schema.testutils.definition.index.HashIndexDefinition;
-import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.RowId;
-import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.storage.index.impl.BinaryTupleRowSerializer;
-import org.apache.ignite.internal.util.Cursor;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
/**
* Base class for Hash Index storage tests.
*/
-public abstract class AbstractHashIndexStorageTest {
- protected static final int TEST_PARTITION = 0;
-
- private static final String INT_COLUMN_NAME = "intVal";
-
- private static final String STR_COLUMN_NAME = "strVal";
-
- private MvTableStorage tableStorage;
-
- private MvPartitionStorage partitionStorage;
-
- private HashIndexStorage indexStorage;
-
- private BinaryTupleRowSerializer serializer;
-
- /**
- * Initializes the internal structures needed for tests.
- *
- * <p>This method *MUST* always be called in either subclass' constructor
or setUp method.
- */
- protected final void initialize(MvTableStorage tableStorage,
TablesConfiguration tablesCfg) {
- this.tableStorage = tableStorage;
-
- createTestTable(tableStorage.configuration());
-
- this.partitionStorage = getOrCreateMvPartition(tableStorage,
TEST_PARTITION);
- this.indexStorage = createIndex(tableStorage, tablesCfg);
- this.serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
- }
-
- /**
- * Configures a test table with some indexed columns.
- */
- private static void createTestTable(TableConfiguration tableCfg) {
- ColumnDefinition pkColumn = column("pk",
ColumnType.INT32).asNullable(false).build();
-
- ColumnDefinition[] allColumns = {
- pkColumn,
- column(INT_COLUMN_NAME,
ColumnType.INT32).asNullable(true).build(),
- column(STR_COLUMN_NAME,
ColumnType.string()).asNullable(true).build()
- };
-
- TableDefinition tableDefinition = tableBuilder("schema", "table")
- .columns(allColumns)
- .withPrimaryKey(pkColumn.name())
+public abstract class AbstractHashIndexStorageTest extends
AbstractIndexStorageTest<HashIndexStorage, HashIndexDescriptor> {
+ @Override
+ protected HashIndexStorage createIndexStorage(String name, ColumnType...
columnTypes) {
+ HashIndexDefinition indexDefinition = SchemaBuilders.hashIndex(name)
+
.withColumns(Stream.of(columnTypes).map(AbstractIndexStorageTest::columnName).toArray(String[]::new))
.build();
- CompletableFuture<Void> createTableFuture = tableCfg.change(cfg ->
convert(tableDefinition, cfg));
-
- assertThat(createTableFuture, willCompleteSuccessfully());
- }
-
- /**
- * Configures and creates a storage instance for testing.
- */
- private static HashIndexStorage createIndex(MvTableStorage tableStorage,
TablesConfiguration tablesConf) {
- HashIndexDefinition indexDefinition =
SchemaBuilders.hashIndex("hashIndex")
- .withColumns(INT_COLUMN_NAME, STR_COLUMN_NAME)
- .build();
-
- CompletableFuture<Void> createIndexFuture = tablesConf.indexes()
+ CompletableFuture<Void> createIndexFuture = tablesCfg.indexes()
.change(chg -> chg.create(indexDefinition.name(), idx -> {
- UUID tableId =
ConfigurationUtil.internalId(tablesConf.tables().value(), "foo");
+ UUID tableId =
ConfigurationUtil.internalId(tablesCfg.tables().value(), TABLE_NAME);
addIndex(indexDefinition, tableId, idx);
}));
assertThat(createIndexFuture, willCompleteSuccessfully());
- TableIndexView indexConfig =
tablesConf.indexes().get(indexDefinition.name()).value();
+ TableIndexView indexConfig =
tablesCfg.indexes().get(indexDefinition.name()).value();
return tableStorage.getOrCreateHashIndex(TEST_PARTITION,
indexConfig.id());
}
- /**
- * Tests the {@link HashIndexStorage#get} method.
- */
- @Test
- public void testGet() {
- // First two rows have the same index key, but different row IDs
- IndexRow row1 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
- IndexRow row2 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
- IndexRow row3 = serializer.serializeRow(new Object[]{ 2, "bar" }, new
RowId(TEST_PARTITION));
-
- assertThat(getAll(row1), is(empty()));
- assertThat(getAll(row2), is(empty()));
- assertThat(getAll(row3), is(empty()));
-
- put(row1);
- put(row2);
- put(row3);
-
- assertThat(getAll(row1), containsInAnyOrder(row1.rowId(),
row2.rowId()));
- assertThat(getAll(row2), containsInAnyOrder(row1.rowId(),
row2.rowId()));
- assertThat(getAll(row3), contains(row3.rowId()));
- }
-
- /**
- * Tests that {@link HashIndexStorage#put} does not create row ID
duplicates.
- */
- @Test
- public void testPutIdempotence() {
- IndexRow row = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
-
- put(row);
- put(row);
-
- assertThat(getAll(row), contains(row.rowId()));
- }
-
- /**
- * Tests the {@link HashIndexStorage#remove} method.
- */
- @Test
- public void testRemove() {
- IndexRow row1 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
- IndexRow row2 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
- IndexRow row3 = serializer.serializeRow(new Object[]{ 2, "bar" }, new
RowId(TEST_PARTITION));
-
- put(row1);
- put(row2);
- put(row3);
-
- assertThat(getAll(row1), containsInAnyOrder(row1.rowId(),
row2.rowId()));
- assertThat(getAll(row2), containsInAnyOrder(row1.rowId(),
row2.rowId()));
- assertThat(getAll(row3), contains(row3.rowId()));
-
- remove(row1);
-
- assertThat(getAll(row1), contains(row2.rowId()));
- assertThat(getAll(row2), contains(row2.rowId()));
- assertThat(getAll(row3), contains(row3.rowId()));
-
- remove(row2);
-
- assertThat(getAll(row1), is(empty()));
- assertThat(getAll(row2), is(empty()));
- assertThat(getAll(row3), contains(row3.rowId()));
-
- remove(row3);
-
- assertThat(getAll(row1), is(empty()));
- assertThat(getAll(row2), is(empty()));
- assertThat(getAll(row3), is(empty()));
- }
-
- /**
- * Tests that {@link HashIndexStorage#remove} works normally when removing
a non-existent row.
- */
- @Test
- public void testRemoveIdempotence() {
- IndexRow row = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
-
- assertDoesNotThrow(() -> remove(row));
-
- put(row);
-
- remove(row);
-
- assertThat(getAll(row), is(empty()));
-
- assertDoesNotThrow(() -> remove(row));
+ @Override
+ protected HashIndexDescriptor indexDescriptor(HashIndexStorage index) {
+ return index.indexDescriptor();
}
@Disabled("https://issues.apache.org/jira/browse/IGNITE-17626";)
@Test
public void testDestroy() {
+ HashIndexStorage index = createIndexStorage(INDEX_NAME,
ColumnType.INT32, ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
IndexRow row1 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
IndexRow row2 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
IndexRow row3 = serializer.serializeRow(new Object[]{ 2, "bar" }, new
RowId(TEST_PARTITION));
- put(row1);
- put(row2);
- put(row3);
+ put(index, row1);
+ put(index, row2);
+ put(index, row3);
- CompletableFuture<Void> destroyFuture =
tableStorage.destroyIndex(indexStorage.indexDescriptor().id());
+ CompletableFuture<Void> destroyFuture =
tableStorage.destroyIndex(index.indexDescriptor().id());
waitForDurableCompletion(destroyFuture);
//TODO IGNITE-17626 Index must be invalid, we should assert that
getIndex returns null and that in won't surface upon restart.
// "destroy" is not "clear", you know. Maybe "getAndCreateIndex" will
do it for the test, idk
- assertThat(getAll(row1), is(empty()));
- assertThat(getAll(row2), is(empty()));
- assertThat(getAll(row3), is(empty()));
+ assertThat(getAll(index, row1), is(empty()));
+ assertThat(getAll(index, row2), is(empty()));
+ assertThat(getAll(index, row3), is(empty()));
}
private void waitForDurableCompletion(CompletableFuture<?> future) {
@@ -250,38 +100,7 @@ public abstract class AbstractHashIndexStorageTest {
}
}
- protected Collection<RowId> getAll(IndexRow row) {
- try (Cursor<RowId> cursor = indexStorage.get(row.indexColumns())) {
- return cursor.stream().collect(toList());
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- protected void put(IndexRow row) {
- partitionStorage.runConsistently(locker -> {
- indexStorage.put(row);
-
- return null;
- });
- }
-
- private void remove(IndexRow row) {
- partitionStorage.runConsistently(locker -> {
- indexStorage.remove(row);
-
- return null;
- });
- }
-
- /**
- * Creates an index row.
- *
- * @param intVal Integer column.
- * @param strVal String column.
- * @param rowId Row ID.
- */
- protected IndexRow createIndexRow(int intVal, String strVal, RowId rowId) {
- return serializer.serializeRow(new Object[]{ intVal, strVal }, rowId);
+ protected IndexRow createIndexRow(BinaryTupleRowSerializer serializer,
RowId rowId, Object... values) {
+ return serializer.serializeRow(values, rowId);
}
}
diff --git
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractIndexStorageTest.java
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractIndexStorageTest.java
new file mode 100644
index 0000000000..6f6090f8e4
--- /dev/null
+++
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractIndexStorageTest.java
@@ -0,0 +1,383 @@
+/*
+ * 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.storage.index;
+
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toUnmodifiableList;
+import static
org.apache.ignite.internal.schema.testutils.SchemaConfigurationConverter.convert;
+import static
org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders.column;
+import static
org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders.tableBuilder;
+import static
org.apache.ignite.internal.storage.BaseMvStoragesTest.getOrCreateMvPartition;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.anyOf;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.schema.BinaryTuple;
+import org.apache.ignite.internal.schema.configuration.TableConfiguration;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.testutils.definition.ColumnDefinition;
+import org.apache.ignite.internal.schema.testutils.definition.ColumnType;
+import org.apache.ignite.internal.schema.testutils.definition.TableDefinition;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
+import org.apache.ignite.internal.storage.index.impl.BinaryTupleRowSerializer;
+import org.apache.ignite.internal.util.Cursor;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Base class for index storage tests. Covers common methods, such as {@link
IndexStorage#get(BinaryTuple)} or
+ * {@link IndexStorage#put(IndexRow)}.
+ *
+ * @param <S> Type of specific index implementation.
+ * @param <D> Type of index descriptor for that specific implementation.
+ */
+public abstract class AbstractIndexStorageTest<S extends IndexStorage, D
extends IndexDescriptor> {
+ private static final IgniteLogger log =
Loggers.forClass(AbstractIndexStorageTest.class);
+
+ /** Definitions of all supported column types. */
+ @SuppressWarnings("WeakerAccess") // May be used in "@VariableSource",
that's why it's public.
+ public static final List<ColumnDefinition> ALL_TYPES_COLUMN_DEFINITIONS =
allTypesColumnDefinitions();
+
+ protected static final int TEST_PARTITION = 0;
+
+ protected static final String INDEX_NAME = "TEST_IDX";
+
+ // Short name is convenient for initialization in @InjectConfiguration.
+ protected static final String TABLE_NAME = "foo";
+
+ private static List<ColumnDefinition> allTypesColumnDefinitions() {
+ Stream<ColumnType> allColumnTypes = Stream.of(
+ ColumnType.INT8,
+ ColumnType.INT16,
+ ColumnType.INT32,
+ ColumnType.INT64,
+ ColumnType.FLOAT,
+ ColumnType.DOUBLE,
+ ColumnType.UUID,
+ ColumnType.DATE,
+ ColumnType.bitmaskOf(32),
+ ColumnType.string(),
+ ColumnType.blob(),
+ ColumnType.number(),
+ ColumnType.decimal(),
+ ColumnType.time(),
+ ColumnType.datetime(),
+ ColumnType.timestamp()
+ );
+
+ return allColumnTypes
+ .map(type -> column(columnName(type),
type).asNullable(true).build())
+ .collect(toUnmodifiableList());
+ }
+
+ private final long seed = System.currentTimeMillis();
+
+ protected final Random random = new Random(seed);
+
+ protected MvTableStorage tableStorage;
+
+ protected MvPartitionStorage partitionStorage;
+
+ protected TablesConfiguration tablesCfg;
+
+ @BeforeEach
+ void setUp() {
+ log.info("Using random seed: " + seed);
+ }
+
+ /**
+ * Returns a name of the column with given type.
+ */
+ protected static String columnName(ColumnType type) {
+ return type.typeSpec().name();
+ }
+
+ /**
+ * Initializes the internal structures needed for tests.
+ *
+ * <p>This method *MUST* always be called in either subclass' constructor
or setUp method.
+ */
+ protected final void initialize(MvTableStorage tableStorage,
TablesConfiguration tablesCfg) {
+ this.tablesCfg = tablesCfg;
+ this.tableStorage = tableStorage;
+ this.partitionStorage = getOrCreateMvPartition(tableStorage,
TEST_PARTITION);
+
+ createTestTable(tableStorage.configuration());
+ }
+
+ /**
+ * Configures a test table with columns of all supported types.
+ */
+ private static void createTestTable(TableConfiguration tableCfg) {
+ ColumnDefinition pkColumn = column("pk",
ColumnType.INT32).asNullable(false).build();
+
+ ColumnDefinition[] allColumns = Stream.concat(Stream.of(pkColumn),
ALL_TYPES_COLUMN_DEFINITIONS.stream())
+ .toArray(ColumnDefinition[]::new);
+
+ TableDefinition tableDefinition = tableBuilder("test", TABLE_NAME)
+ .columns(allColumns)
+ .withPrimaryKey(pkColumn.name())
+ .build();
+
+ CompletableFuture<Void> createTableFuture = tableCfg.change(cfg ->
convert(tableDefinition, cfg));
+
+ assertThat(createTableFuture, willCompleteSuccessfully());
+ }
+
+ /**
+ * Creates an IndexStorage instance using the given columns.
+ *
+ * @see #columnName(ColumnType)
+ */
+ protected abstract S createIndexStorage(String name, ColumnType...
columnTypes);
+
+ /**
+ * Provides safe access to the index descriptor of the storage.
+ */
+ protected abstract D indexDescriptor(S index);
+
+ /**
+ * Tests the {@link IndexStorage#get} method.
+ */
+ @Test
+ public void testGet() {
+ S index = createIndexStorage(INDEX_NAME, ColumnType.INT32,
ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
+ // First two rows have the same index key, but different row IDs.
+ IndexRow row1 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
+ IndexRow row2 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
+ IndexRow row3 = serializer.serializeRow(new Object[]{ 2, "bar" }, new
RowId(TEST_PARTITION));
+ IndexRow row4 = serializer.serializeRow(new Object[]{ 3, "baz" }, new
RowId(TEST_PARTITION));
+
+ assertThat(getAll(index, row1), is(empty()));
+ assertThat(getAll(index, row2), is(empty()));
+ assertThat(getAll(index, row3), is(empty()));
+
+ put(index, row1);
+ put(index, row2);
+ put(index, row3);
+
+ assertThat(getAll(index, row1), containsInAnyOrder(row1.rowId(),
row2.rowId()));
+ assertThat(getAll(index, row2), containsInAnyOrder(row1.rowId(),
row2.rowId()));
+ assertThat(getAll(index, row3), contains(row3.rowId()));
+ assertThat(getAll(index, row4), is(empty()));
+ }
+
+ @Test
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-19422";)
+ public void testGetConcurrentPut() {
+ S index = createIndexStorage(INDEX_NAME, ColumnType.INT32,
ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
+ Object[] columnValues = { 1, "foo" };
+ IndexRow row1 = serializer.serializeRow(columnValues, new
RowId(TEST_PARTITION, 1, 1));
+ IndexRow row2 = serializer.serializeRow(columnValues, new
RowId(TEST_PARTITION, 2, 2));
+
+ try (Cursor<RowId> cursor = index.get(row1.indexColumns())) {
+ put(index, row1);
+
+ assertTrue(cursor.hasNext());
+ assertEquals(row1.rowId(), cursor.next());
+
+ put(index, row2);
+
+ assertTrue(cursor.hasNext());
+ assertEquals(row2.rowId(), cursor.next());
+
+ assertFalse(cursor.hasNext());
+ assertThrows(NoSuchElementException.class, cursor::next);
+ }
+ }
+
+ @Test
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-19422";)
+ public void testGetConcurrentReplace() {
+ S index = createIndexStorage(INDEX_NAME, ColumnType.INT32,
ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
+ Object[] columnValues = { 1, "foo" };
+ IndexRow row1 = serializer.serializeRow(columnValues, new
RowId(TEST_PARTITION, 1, 1));
+ IndexRow row2 = serializer.serializeRow(columnValues, new
RowId(TEST_PARTITION, 2, 2));
+
+ put(index, row1);
+
+ try (Cursor<RowId> cursor = index.get(row1.indexColumns())) {
+ assertTrue(cursor.hasNext());
+ assertEquals(row1.rowId(), cursor.next());
+
+ remove(index, row1);
+ put(index, row2);
+
+ assertTrue(cursor.hasNext());
+ assertEquals(row2.rowId(), cursor.next());
+
+ assertFalse(cursor.hasNext());
+ assertThrows(NoSuchElementException.class, cursor::next);
+ }
+ }
+
+ /**
+ * Tests that {@link IndexStorage#put} does not create row ID duplicates.
+ */
+ @Test
+ public void testPutIdempotence() {
+ S index = createIndexStorage(INDEX_NAME, ColumnType.INT32,
ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
+ IndexRow row = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
+
+ put(index, row);
+ put(index, row);
+
+ IndexRow actualRow = getSingle(index, row.indexColumns());
+
+ assertNotNull(actualRow);
+
+ assertThat(actualRow.rowId(), is(equalTo(row.rowId())));
+
+ assertThat(getAll(index, row), contains(row.rowId()));
+ }
+
+ /**
+ * Tests the {@link IndexStorage#remove} method.
+ */
+ @Test
+ public void testRemove() {
+ S index = createIndexStorage(INDEX_NAME, ColumnType.INT32,
ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
+ IndexRow row1 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
+ IndexRow row2 = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
+ IndexRow row3 = serializer.serializeRow(new Object[]{ 2, "bar" }, new
RowId(TEST_PARTITION));
+
+ put(index, row1);
+ put(index, row2);
+ put(index, row3);
+
+ assertThat(getAll(index, row1), containsInAnyOrder(row1.rowId(),
row2.rowId()));
+ assertThat(getAll(index, row2), containsInAnyOrder(row1.rowId(),
row2.rowId()));
+ assertThat(getAll(index, row3), contains(row3.rowId()));
+
+ remove(index, row1);
+
+ assertThat(getAll(index, row1), contains(row2.rowId()));
+ assertThat(getAll(index, row2), contains(row2.rowId()));
+ assertThat(getAll(index, row3), contains(row3.rowId()));
+
+ remove(index, row2);
+
+ assertThat(getAll(index, row1), is(empty()));
+ assertThat(getAll(index, row2), is(empty()));
+ assertThat(getAll(index, row3), contains(row3.rowId()));
+
+ remove(index, row3);
+
+ assertThat(getAll(index, row1), is(empty()));
+ assertThat(getAll(index, row2), is(empty()));
+ assertThat(getAll(index, row3), is(empty()));
+ }
+
+ /**
+ * Tests that {@link IndexStorage#remove} works normally when removing a
non-existent row.
+ */
+ @Test
+ public void testRemoveIdempotence() {
+ S index = createIndexStorage(INDEX_NAME, ColumnType.INT32,
ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
+ IndexRow row = serializer.serializeRow(new Object[]{ 1, "foo" }, new
RowId(TEST_PARTITION));
+
+ assertDoesNotThrow(() -> remove(index, row));
+
+ put(index, row);
+
+ remove(index, row);
+
+ assertThat(getAll(index, row), is(empty()));
+
+ assertDoesNotThrow(() -> remove(index, row));
+ }
+
+ protected static Collection<RowId> getAll(IndexStorage index, IndexRow
row) {
+ try (Cursor<RowId> cursor = index.get(row.indexColumns())) {
+ return cursor.stream().collect(toList());
+ }
+ }
+
+ /**
+ * Extracts a single value by a given key or {@code null} if it does not
exist.
+ */
+ protected static @Nullable IndexRow getSingle(IndexStorage indexStorage,
BinaryTuple fullPrefix) {
+ List<RowId> rowIds = get(indexStorage, fullPrefix);
+
+ assertThat(rowIds, anyOf(empty(), hasSize(1)));
+
+ return rowIds.isEmpty() ? null : new IndexRowImpl(fullPrefix,
rowIds.get(0));
+ }
+
+ protected static List<RowId> get(IndexStorage index, BinaryTuple key) {
+ try (Cursor<RowId> cursor = index.get(key)) {
+ return cursor.stream().collect(toUnmodifiableList());
+ }
+ }
+
+ protected final void put(S indexStorage, IndexRow row) {
+ partitionStorage.runConsistently(locker -> {
+ locker.lock(row.rowId());
+
+ indexStorage.put(row);
+
+ return null;
+ });
+ }
+
+ protected final void remove(S indexStorage, IndexRow row) {
+ partitionStorage.runConsistently(locker -> {
+ locker.lock(row.rowId());
+
+ indexStorage.remove(row);
+
+ return null;
+ });
+ }
+}
diff --git
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractSortedIndexStorageTest.java
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractSortedIndexStorageTest.java
index 5a61474087..6ef872af03 100644
---
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractSortedIndexStorageTest.java
+++
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/AbstractSortedIndexStorageTest.java
@@ -21,10 +21,6 @@ import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toUnmodifiableList;
import static
org.apache.ignite.internal.schema.testutils.SchemaConfigurationConverter.addIndex;
-import static
org.apache.ignite.internal.schema.testutils.SchemaConfigurationConverter.convert;
-import static
org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders.column;
-import static
org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders.tableBuilder;
-import static
org.apache.ignite.internal.storage.BaseMvStoragesTest.getOrCreateMvPartition;
import static
org.apache.ignite.internal.storage.index.SortedIndexStorage.GREATER;
import static
org.apache.ignite.internal.storage.index.SortedIndexStorage.GREATER_OR_EQUAL;
import static org.apache.ignite.internal.storage.index.SortedIndexStorage.LESS;
@@ -34,9 +30,7 @@ import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -51,21 +45,16 @@ import java.util.Collections;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Objects;
-import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.IntStream;
-import java.util.stream.Stream;
import org.apache.ignite.internal.configuration.util.ConfigurationUtil;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.schema.BinaryTuple;
import org.apache.ignite.internal.schema.BinaryTuplePrefix;
import org.apache.ignite.internal.schema.SchemaTestUtils;
-import org.apache.ignite.internal.schema.configuration.TableConfiguration;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.schema.configuration.index.TableIndexView;
import org.apache.ignite.internal.schema.testutils.builder.SchemaBuilders;
import
org.apache.ignite.internal.schema.testutils.builder.SortedIndexDefinitionBuilder;
@@ -73,12 +62,9 @@ import
org.apache.ignite.internal.schema.testutils.builder.SortedIndexDefinition
import org.apache.ignite.internal.schema.testutils.definition.ColumnDefinition;
import org.apache.ignite.internal.schema.testutils.definition.ColumnType;
import
org.apache.ignite.internal.schema.testutils.definition.ColumnType.ColumnTypeSpec;
-import org.apache.ignite.internal.schema.testutils.definition.TableDefinition;
import
org.apache.ignite.internal.schema.testutils.definition.index.ColumnarIndexDefinition;
import
org.apache.ignite.internal.schema.testutils.definition.index.SortedIndexDefinition;
-import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.RowId;
-import org.apache.ignite.internal.storage.engine.MvTableStorage;
import
org.apache.ignite.internal.storage.index.SortedIndexDescriptor.SortedIndexColumnDescriptor;
import org.apache.ignite.internal.storage.index.impl.BinaryTupleRowSerializer;
import org.apache.ignite.internal.storage.index.impl.TestIndexRow;
@@ -93,92 +79,25 @@ import org.junit.jupiter.params.ParameterizedTest;
/**
* Base class for Sorted Index storage tests.
*/
-public abstract class AbstractSortedIndexStorageTest {
+public abstract class AbstractSortedIndexStorageTest extends
AbstractIndexStorageTest<SortedIndexStorage, SortedIndexDescriptor> {
private static final IgniteLogger log =
Loggers.forClass(AbstractSortedIndexStorageTest.class);
- /** Definitions of all supported column types. */
- public static final List<ColumnDefinition> ALL_TYPES_COLUMN_DEFINITIONS =
allTypesColumnDefinitions();
-
- protected static final int TEST_PARTITION = 0;
-
- private static List<ColumnDefinition> allTypesColumnDefinitions() {
- Stream<ColumnType> allColumnTypes = Stream.of(
- ColumnType.INT8,
- ColumnType.INT16,
- ColumnType.INT32,
- ColumnType.INT64,
- ColumnType.FLOAT,
- ColumnType.DOUBLE,
- ColumnType.UUID,
- ColumnType.DATE,
- ColumnType.bitmaskOf(32),
- ColumnType.string(),
- ColumnType.blob(),
- ColumnType.number(),
- ColumnType.decimal(),
- ColumnType.time(),
- ColumnType.datetime(),
- ColumnType.timestamp()
- );
-
- return allColumnTypes
- .map(type -> column(type.typeSpec().name(),
type).asNullable(true).build())
- .collect(toUnmodifiableList());
- }
-
- private final Random random;
-
- private MvTableStorage tableStorage;
-
- private MvPartitionStorage partitionStorage;
-
- private TablesConfiguration tablesCfg;
-
- protected AbstractSortedIndexStorageTest() {
- long seed = System.currentTimeMillis();
-
- log.info("Using random seed: " + seed);
-
- random = new Random(seed);
- }
-
- /**
- * Initializes the internal structures needed for tests.
- *
- * <p>This method *MUST* always be called in either subclass' constructor
or setUp method.
- */
- protected final void initialize(MvTableStorage tableStorage,
TablesConfiguration tablesCfg) {
- this.tablesCfg = tablesCfg;
- this.tableStorage = tableStorage;
- this.partitionStorage = getOrCreateMvPartition(tableStorage,
TEST_PARTITION);
-
- createTestTable(tableStorage.configuration());
- }
-
- /**
- * Configures a test table with columns of all supported types.
- */
- private static void createTestTable(TableConfiguration tableCfg) {
- ColumnDefinition pkColumn = column("pk",
ColumnType.INT32).asNullable(false).build();
-
- ColumnDefinition[] allColumns = Stream.concat(Stream.of(pkColumn),
ALL_TYPES_COLUMN_DEFINITIONS.stream())
- .toArray(ColumnDefinition[]::new);
+ @Override
+ protected SortedIndexStorage createIndexStorage(String name, ColumnType...
columnTypes) {
+ SortedIndexDefinitionBuilder builder =
SchemaBuilders.sortedIndex(name);
- TableDefinition tableDefinition = tableBuilder("test", "foo")
- .columns(allColumns)
- .withPrimaryKey(pkColumn.name())
- .build();
-
- CompletableFuture<Void> createTableFuture = tableCfg.change(cfg ->
convert(tableDefinition, cfg));
+ for (ColumnType columnType : columnTypes) {
+ builder =
builder.addIndexColumn(columnType.typeSpec().name()).asc().done();
+ }
- assertThat(createTableFuture, willCompleteSuccessfully());
+ return createIndexStorage(builder.build());
}
/**
* Creates a Sorted Index using the given columns.
*/
- private SortedIndexStorage createIndexStorage(List<ColumnDefinition>
indexSchema) {
- SortedIndexDefinitionBuilder indexDefinitionBuilder =
SchemaBuilders.sortedIndex("TEST_IDX");
+ protected SortedIndexStorage createIndexStorage(String name,
List<ColumnDefinition> indexSchema) {
+ SortedIndexDefinitionBuilder indexDefinitionBuilder =
SchemaBuilders.sortedIndex(name);
indexSchema.forEach(column -> {
SortedIndexColumnBuilder columnBuilder =
indexDefinitionBuilder.addIndexColumn(column.name());
@@ -203,7 +122,7 @@ public abstract class AbstractSortedIndexStorageTest {
protected SortedIndexStorage createIndexStorage(ColumnarIndexDefinition
indexDefinition) {
CompletableFuture<Void> createIndexFuture =
tablesCfg.indexes().change(chg ->
chg.create(indexDefinition.name(), idx -> {
- UUID tableId =
ConfigurationUtil.internalId(tablesCfg.tables().value(), "foo");
+ UUID tableId =
ConfigurationUtil.internalId(tablesCfg.tables().value(), TABLE_NAME);
addIndex(indexDefinition, tableId, idx);
}));
@@ -215,12 +134,17 @@ public abstract class AbstractSortedIndexStorageTest {
return tableStorage.getOrCreateSortedIndex(TEST_PARTITION,
indexConfig.id());
}
+ @Override
+ protected SortedIndexDescriptor indexDescriptor(SortedIndexStorage index) {
+ return index.indexDescriptor();
+ }
+
/**
* Tests that columns of all types are correctly serialized and
deserialized.
*/
@Test
void testRowSerialization() {
- SortedIndexStorage indexStorage =
createIndexStorage(ALL_TYPES_COLUMN_DEFINITIONS);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ALL_TYPES_COLUMN_DEFINITIONS);
Object[] columns = indexStorage.indexDescriptor().columns().stream()
.map(SortedIndexColumnDescriptor::type)
@@ -238,40 +162,11 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testEmpty() {
- SortedIndexStorage index =
createIndexStorage(shuffledRandomDefinitions());
+ SortedIndexStorage index = createIndexStorage(INDEX_NAME,
shuffledRandomDefinitions());
assertThat(scan(index, null, null, 0), is(empty()));
}
- @Test
- void testGet() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_INDEX")
- .addIndexColumn(ColumnTypeSpec.STRING.name()).asc().done()
- .addIndexColumn(ColumnTypeSpec.INT32.name()).asc().done()
- .build();
-
- SortedIndexStorage index = createIndexStorage(indexDefinition);
-
- var serializer = new BinaryTupleRowSerializer(index.indexDescriptor());
-
- IndexRow val1090 = serializer.serializeRow(new Object[]{ "10", 90 },
new RowId(TEST_PARTITION));
- IndexRow otherVal1090 = serializer.serializeRow(new Object[]{ "10", 90
}, new RowId(TEST_PARTITION));
- IndexRow val1080 = serializer.serializeRow(new Object[]{ "10", 80 },
new RowId(TEST_PARTITION));
- IndexRow val2090 = serializer.serializeRow(new Object[]{ "20", 90 },
new RowId(TEST_PARTITION));
-
- put(index, val1090);
- put(index, otherVal1090);
- put(index, val1080);
-
- assertThat(get(index, val1090.indexColumns()),
containsInAnyOrder(val1090.rowId(), otherVal1090.rowId()));
-
- assertThat(get(index, otherVal1090.indexColumns()),
containsInAnyOrder(val1090.rowId(), otherVal1090.rowId()));
-
- assertThat(get(index, val1080.indexColumns()),
containsInAnyOrder(val1080.rowId()));
-
- assertThat(get(index, val2090.indexColumns()), is(empty()));
- }
-
/**
* Tests the Put-Get-Remove case when an index is created using a single
column.
*/
@@ -281,44 +176,12 @@ public abstract class AbstractSortedIndexStorageTest {
testPutGetRemove(List.of(columnDefinition));
}
- /**
- * Tests that appending an already existing row does no harm.
- */
- @Test
- void testPutIdempotence() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_INDEX")
- .addIndexColumn(ColumnTypeSpec.STRING.name()).asc().done()
- .addIndexColumn(ColumnTypeSpec.INT32.name()).asc().done()
- .build();
-
- SortedIndexStorage index = createIndexStorage(indexDefinition);
-
- var columnValues = new Object[] { "foo", 1 };
- var rowId = new RowId(TEST_PARTITION);
-
- var serializer = new BinaryTupleRowSerializer(index.indexDescriptor());
-
- IndexRow row = serializer.serializeRow(columnValues, rowId);
-
- put(index, row);
- put(index, row);
-
- IndexRow actualRow = getSingle(index, row.indexColumns());
-
- assertThat(actualRow.rowId(), is(equalTo(row.rowId())));
- }
-
/**
* Tests that it is possible to add rows with the same columns but
different Row IDs.
*/
@Test
void testMultiplePuts() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_INDEX")
- .addIndexColumn(ColumnTypeSpec.STRING.name()).asc().done()
- .addIndexColumn(ColumnTypeSpec.INT32.name()).asc().done()
- .build();
-
- SortedIndexStorage index = createIndexStorage(indexDefinition);
+ SortedIndexStorage index = createIndexStorage(INDEX_NAME,
ColumnType.string(), ColumnType.INT32);
var columnValues1 = new Object[] { "foo", 1 };
var columnValues2 = new Object[] { "bar", 3 };
@@ -345,13 +208,8 @@ public abstract class AbstractSortedIndexStorageTest {
* Tests the {@link SortedIndexStorage#remove} method.
*/
@Test
- void testRemove() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_INDEX")
- .addIndexColumn(ColumnTypeSpec.STRING.name()).asc().done()
- .addIndexColumn(ColumnTypeSpec.INT32.name()).asc().done()
- .build();
-
- SortedIndexStorage index = createIndexStorage(indexDefinition);
+ void testRemoveAndScan() {
+ SortedIndexStorage index = createIndexStorage(INDEX_NAME,
ColumnType.string(), ColumnType.INT32);
var columnValues1 = new Object[] { "foo", 1 };
var columnValues2 = new Object[] { "bar", 3 };
@@ -408,7 +266,7 @@ public abstract class AbstractSortedIndexStorageTest {
*/
@RepeatedTest(5)
void testScan() {
- SortedIndexStorage indexStorage =
createIndexStorage(shuffledDefinitions());
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
shuffledDefinitions());
List<TestIndexRow> entries = IntStream.range(0, 10)
.mapToObj(i -> {
@@ -538,7 +396,7 @@ public abstract class AbstractSortedIndexStorageTest {
void testEmptyRange() {
List<ColumnDefinition> indexSchema = shuffledRandomDefinitions();
- SortedIndexStorage indexStorage = createIndexStorage(indexSchema);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
indexSchema);
TestIndexRow entry1 = TestIndexRow.randomRow(indexStorage);
TestIndexRow entry2 = TestIndexRow.randomRow(indexStorage);
@@ -560,7 +418,7 @@ public abstract class AbstractSortedIndexStorageTest {
@ParameterizedTest
@VariableSource("ALL_TYPES_COLUMN_DEFINITIONS")
void testNullValues(ColumnDefinition columnDefinition) {
- SortedIndexStorage storage =
createIndexStorage(List.of(columnDefinition));
+ SortedIndexStorage storage = createIndexStorage(INDEX_NAME,
List.of(columnDefinition));
TestIndexRow entry1 = TestIndexRow.randomRow(storage);
@@ -594,11 +452,7 @@ public abstract class AbstractSortedIndexStorageTest {
*/
@Test
void testScanSimple() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -750,11 +604,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractAddRowBeforeInvokeHasNext() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -777,11 +627,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractAddRowAfterInvokeHasNext() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -830,11 +676,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractInvokeOnlyNext() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -866,11 +708,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractAddRowsOnly() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -929,11 +767,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractForFinishCursor() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -979,11 +813,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractNextMethodOnly() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1030,11 +860,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractRemoveRowsOnly() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1088,11 +914,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractReplaceRow() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1133,11 +955,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractRemoveCachedRow() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1201,11 +1019,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanContractRemoveNextAndAddFirstRow() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1236,14 +1050,9 @@ public abstract class AbstractSortedIndexStorageTest {
assertThrows(NoSuchElementException.class, scan::next);
}
-
@Test
void testScanPeekForFinishedCursor() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1306,11 +1115,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanPeekAddRowsOnly() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1387,11 +1192,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanPeekRemoveRows() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1450,11 +1251,7 @@ public abstract class AbstractSortedIndexStorageTest {
@Test
void testScanPeekReplaceRow() {
- SortedIndexDefinition indexDefinition =
SchemaBuilders.sortedIndex("TEST_IDX")
-
.addIndexColumn(ColumnType.INT32.typeSpec().name()).asc().done()
- .build();
-
- SortedIndexStorage indexStorage = createIndexStorage(indexDefinition);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
@@ -1506,6 +1303,45 @@ public abstract class AbstractSortedIndexStorageTest {
assertNull(scan.peek());
}
+ @Test
+ void testScanPeekRemoveNext() {
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
ColumnType.INT32);
+
+ BinaryTupleRowSerializer serializer = new
BinaryTupleRowSerializer(indexStorage.indexDescriptor());
+
+ PeekCursor<IndexRow> scan = indexStorage.scan(null, null, 0);
+
+ RowId rowId = new RowId(TEST_PARTITION);
+
+ // index = [0]
+ // cursor = ^ with no cached row
+ put(indexStorage, serializer.serializeRow(new Object[]{0}, rowId));
+
+ // index = [0]
+ // cursor = ^ with no cached row
+ assertEquals(SimpleRow.of(0, rowId), SimpleRow.of(scan.peek(),
firstColumn(serializer)));
+
+ // index =
+ // cursor = ^ with no cached row (but it remembers the last peek call)
+ remove(indexStorage, serializer.serializeRow(new Object[]{0}, rowId));
+
+ // "hasNext" and "next" must return the result of last "peek"
operation. This is crucial for RW scans.
+ assertTrue(scan.hasNext());
+ assertEquals(SimpleRow.of(0, rowId), SimpleRow.of(scan.next(),
firstColumn(serializer)));
+
+ // index =
+ // cursor = ^ with no cached row
+ assertNull(scan.peek());
+
+ // index = [1]
+ // cursor = ^ with no cached row (points before [1] because last
returned value was [0])
+ put(indexStorage, serializer.serializeRow(new Object[]{1}, rowId));
+
+ // But, "hasNext" must return "false" to be consistent with the result
of last "peek" operation. This is crucial for RW scans.
+ assertFalse(scan.hasNext());
+ assertThrows(NoSuchElementException.class, scan::next);
+ }
+
private List<ColumnDefinition> shuffledRandomDefinitions() {
return shuffledDefinitions(d -> random.nextBoolean());
}
@@ -1539,7 +1375,7 @@ public abstract class AbstractSortedIndexStorageTest {
* be removed.
*/
private void testPutGetRemove(List<ColumnDefinition> indexSchema) {
- SortedIndexStorage indexStorage = createIndexStorage(indexSchema);
+ SortedIndexStorage indexStorage = createIndexStorage(INDEX_NAME,
indexSchema);
TestIndexRow entry1 = TestIndexRow.randomRow(indexStorage);
TestIndexRow entry2;
@@ -1567,18 +1403,6 @@ public abstract class AbstractSortedIndexStorageTest {
assertThat(getSingle(indexStorage, entry1.indexColumns()),
is(nullValue()));
}
- /**
- * Extracts a single value by a given key or {@code null} if it does not
exist.
- */
- @Nullable
- private static IndexRow getSingle(SortedIndexStorage indexStorage,
BinaryTuple fullPrefix) {
- List<RowId> rowIds = get(indexStorage, fullPrefix);
-
- assertThat(rowIds, anyOf(empty(), hasSize(1)));
-
- return rowIds.isEmpty() ? null : new IndexRowImpl(fullPrefix,
rowIds.get(0));
- }
-
private static BinaryTuplePrefix prefix(SortedIndexStorage index,
Object... vals) {
var serializer = new BinaryTupleRowSerializer(index.indexDescriptor());
@@ -1611,42 +1435,6 @@ public abstract class AbstractSortedIndexStorageTest {
}
}
- protected static List<RowId> get(SortedIndexStorage index, BinaryTuple
key) {
- try (Cursor<RowId> cursor = index.get(key)) {
- return cursor.stream().collect(toUnmodifiableList());
- }
- }
-
- protected void put(SortedIndexStorage indexStorage, IndexRow row) {
- partitionStorage.runConsistently(locker -> {
- locker.lock(row.rowId());
-
- indexStorage.put(row);
-
- return null;
- });
- }
-
- private void remove(SortedIndexStorage indexStorage, IndexRow row) {
- partitionStorage.runConsistently(locker -> {
- locker.lock(row.rowId());
-
- indexStorage.remove(row);
-
- return null;
- });
- }
-
- private static <T> List<T> getRemaining(Cursor<IndexRow> scanCursor,
Function<IndexRow, T> mapper) {
- List<T> result = new ArrayList<>();
-
- while (scanCursor.hasNext()) {
- result.add(mapper.apply(scanCursor.next()));
- }
-
- return result;
- }
-
private static <T> Function<IndexRow, T>
firstColumn(BinaryTupleRowSerializer serializer) {
return indexRow -> (T) serializer.deserializeColumns(indexRow)[0];
}
diff --git
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/BinaryTupleRowSerializer.java
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/BinaryTupleRowSerializer.java
index 5435522aa5..612b80ab5f 100644
---
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/BinaryTupleRowSerializer.java
+++
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/BinaryTupleRowSerializer.java
@@ -30,10 +30,9 @@ import
org.apache.ignite.internal.schema.BinaryTupleSchema.Element;
import org.apache.ignite.internal.schema.NativeType;
import org.apache.ignite.internal.schema.NativeTypeSpec;
import org.apache.ignite.internal.storage.RowId;
-import org.apache.ignite.internal.storage.index.HashIndexDescriptor;
+import org.apache.ignite.internal.storage.index.IndexDescriptor;
import org.apache.ignite.internal.storage.index.IndexRow;
import org.apache.ignite.internal.storage.index.IndexRowImpl;
-import org.apache.ignite.internal.storage.index.SortedIndexDescriptor;
/**
* Class for converting an array of objects into a {@link BinaryTuple} and
vice-versa using a given index schema.
@@ -55,18 +54,9 @@ public class BinaryTupleRowSerializer {
private final BinaryTupleSchema tupleSchema;
/**
- * Creates a new instance for a Sorted Index.
+ * Creates a new instance for an index.
*/
- public BinaryTupleRowSerializer(SortedIndexDescriptor descriptor) {
- this(descriptor.columns().stream()
- .map(colDesc -> new ColumnDescriptor(colDesc.type(),
colDesc.nullable()))
- .collect(toUnmodifiableList()));
- }
-
- /**
- * Creates a new instance for a Hash Index.
- */
- public BinaryTupleRowSerializer(HashIndexDescriptor descriptor) {
+ public BinaryTupleRowSerializer(IndexDescriptor descriptor) {
this(descriptor.columns().stream()
.map(colDesc -> new ColumnDescriptor(colDesc.type(),
colDesc.nullable()))
.collect(toUnmodifiableList()));
diff --git
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestHashIndexStorage.java
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestHashIndexStorage.java
index c31cb3b898..591d79c160 100644
---
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestHashIndexStorage.java
+++
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestHashIndexStorage.java
@@ -17,26 +17,28 @@
package org.apache.ignite.internal.storage.index.impl;
-import static org.apache.ignite.internal.util.IgniteUtils.capacity;
-
import java.nio.ByteBuffer;
-import java.util.HashSet;
import java.util.Iterator;
+import java.util.NavigableSet;
+import java.util.NoSuchElementException;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.stream.Collectors;
import org.apache.ignite.internal.schema.BinaryTuple;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.index.HashIndexDescriptor;
import org.apache.ignite.internal.storage.index.HashIndexStorage;
import org.apache.ignite.internal.storage.index.IndexRow;
+import org.jetbrains.annotations.Nullable;
/**
* Test-only implementation of a {@link HashIndexStorage}.
*/
public class TestHashIndexStorage extends AbstractTestIndexStorage implements
HashIndexStorage {
- private final ConcurrentMap<ByteBuffer, Set<RowId>> index = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<ByteBuffer, NavigableSet<RowId>> index = new
ConcurrentHashMap<>();
private final HashIndexDescriptor descriptor;
@@ -56,7 +58,7 @@ public class TestHashIndexStorage extends
AbstractTestIndexStorage implements Ha
@Override
Iterator<RowId> getRowIdIteratorForGetByBinaryTuple(BinaryTuple key) {
- return index.getOrDefault(key.byteBuffer(), Set.of()).iterator();
+ return new RowIdIterator(key.byteBuffer());
}
@Override
@@ -65,17 +67,12 @@ public class TestHashIndexStorage extends
AbstractTestIndexStorage implements Ha
index.compute(row.indexColumns().byteBuffer(), (k, v) -> {
if (v == null) {
- return Set.of(row.rowId());
- } else if (v.contains(row.rowId())) {
- return v;
- } else {
- var result = new HashSet<RowId>(capacity(v.size() + 1));
+ v = new ConcurrentSkipListSet<>();
+ }
- result.addAll(v);
- result.add(row.rowId());
+ v.add(row.rowId());
- return result;
- }
+ return v;
});
}
@@ -84,19 +81,11 @@ public class TestHashIndexStorage extends
AbstractTestIndexStorage implements Ha
checkStorageClosedOrInProcessOfRebalance();
index.computeIfPresent(row.indexColumns().byteBuffer(), (k, v) -> {
- if (v.contains(row.rowId())) {
- if (v.size() == 1) {
- return null;
- } else {
- var result = new HashSet<>(v);
-
- result.remove(row.rowId());
-
- return result;
- }
- } else {
- return v;
+ if (v.remove(row.rowId()) && v.isEmpty()) {
+ return null;
}
+
+ return v;
});
}
@@ -111,4 +100,52 @@ public class TestHashIndexStorage extends
AbstractTestIndexStorage implements Ha
public Set<RowId> allRowsIds() {
return
index.values().stream().flatMap(Set::stream).collect(Collectors.toSet());
}
+
+ /**
+ * Row IDs iterator that always returns up-to-date values.
+ */
+ private class RowIdIterator implements Iterator<RowId> {
+ private final ByteBuffer key;
+
+ @Nullable Boolean hasNext;
+
+ @Nullable RowId rowId;
+
+ RowIdIterator(ByteBuffer key) {
+ this.key = key;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (hasNext != null) {
+ return hasNext;
+ }
+
+ // Yes, we must read it every time, because concurrency.
+ NavigableSet<RowId> rowIds = index.get(key);
+
+ if (rowIds == null) {
+ rowId = null;
+ } else if (rowId == null) {
+ rowId = rowIds.stream().findFirst().orElse(null);
+ } else {
+ rowId = rowIds.higher(rowId);
+ }
+
+ hasNext = rowId != null;
+
+ return hasNext;
+ }
+
+ @Override
+ public RowId next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ hasNext = null;
+
+ return Objects.requireNonNull(rowId);
+ }
+ }
}
diff --git
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestSortedIndexStorage.java
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestSortedIndexStorage.java
index 6961c07765..44f96fa4e5 100644
---
a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestSortedIndexStorage.java
+++
b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestSortedIndexStorage.java
@@ -17,17 +17,17 @@
package org.apache.ignite.internal.storage.index.impl;
-import static java.util.Collections.emptyNavigableMap;
+import static java.util.Collections.emptyNavigableSet;
+import static java.util.Comparator.comparing;
+import static org.apache.ignite.internal.storage.RowId.highestRowId;
+import static org.apache.ignite.internal.storage.RowId.lowestRowId;
import java.nio.ByteBuffer;
import java.util.Iterator;
-import java.util.Map.Entry;
-import java.util.NavigableMap;
import java.util.NavigableSet;
import java.util.NoSuchElementException;
import java.util.Set;
-import java.util.concurrent.ConcurrentNavigableMap;
-import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ConcurrentSkipListSet;
import java.util.stream.Collectors;
import org.apache.ignite.internal.binarytuple.BinaryTupleCommon;
import org.apache.ignite.internal.schema.BinaryTuple;
@@ -39,19 +39,16 @@ import
org.apache.ignite.internal.storage.index.IndexRowImpl;
import org.apache.ignite.internal.storage.index.PeekCursor;
import org.apache.ignite.internal.storage.index.SortedIndexDescriptor;
import org.apache.ignite.internal.storage.index.SortedIndexStorage;
+import org.apache.ignite.internal.util.TransformingIterator;
import org.jetbrains.annotations.Nullable;
/**
* Test implementation of MV sorted index storage.
*/
public class TestSortedIndexStorage extends AbstractTestIndexStorage
implements SortedIndexStorage {
- private static final Object NULL = new Object();
+ private final int partitionId;
- /**
- * {@code NavigableMap<RowId, Object>} is used as a {@link NavigableSet},
but map was chosen because methods like
- * {@link NavigableSet#first()} throw an {@link NoSuchElementException} if
the set is empty.
- */
- private final ConcurrentNavigableMap<ByteBuffer, NavigableMap<RowId,
Object>> index;
+ private final NavigableSet<IndexRow> index;
private final SortedIndexDescriptor descriptor;
@@ -61,8 +58,14 @@ public class TestSortedIndexStorage extends
AbstractTestIndexStorage implements
public TestSortedIndexStorage(int partitionId, SortedIndexDescriptor
descriptor) {
super(partitionId);
+ BinaryTupleComparator binaryTupleComparator = new
BinaryTupleComparator(descriptor);
+
+ this.partitionId = partitionId;
this.descriptor = descriptor;
- this.index = new ConcurrentSkipListMap<>(new
BinaryTupleComparator(descriptor));
+ this.index = new ConcurrentSkipListSet<>(
+ comparing((IndexRow indexRow) ->
indexRow.indexColumns().byteBuffer(), binaryTupleComparator)
+ .thenComparing(IndexRow::rowId)
+ );
}
@Override
@@ -72,31 +75,27 @@ public class TestSortedIndexStorage extends
AbstractTestIndexStorage implements
@Override
Iterator<RowId> getRowIdIteratorForGetByBinaryTuple(BinaryTuple key) {
- return index.getOrDefault(key.byteBuffer(),
emptyNavigableMap()).keySet().iterator();
+ // These must be two different instances, because "scan" call messes
up headers.
+ BinaryTuplePrefix lowerBound = BinaryTuplePrefix.fromBinaryTuple(key);
+ BinaryTuplePrefix higherBound = BinaryTuplePrefix.fromBinaryTuple(key);
+
+ PeekCursor<IndexRow> peekCursor = scan(lowerBound, higherBound,
GREATER_OR_EQUAL | LESS_OR_EQUAL);
+
+ return new TransformingIterator<>(peekCursor, IndexRow::rowId);
}
@Override
public void put(IndexRow row) {
checkStorageClosed();
- index.compute(row.indexColumns().byteBuffer(), (k, v) -> {
- NavigableMap<RowId, Object> rowIds = v == null ? new
ConcurrentSkipListMap<>() : v;
-
- rowIds.put(row.rowId(), NULL);
-
- return rowIds;
- });
+ index.add(row);
}
@Override
public void remove(IndexRow row) {
checkStorageClosedOrInProcessOfRebalance();
- index.computeIfPresent(row.indexColumns().byteBuffer(), (k, v) -> {
- v.remove(row.rowId());
-
- return v.isEmpty() ? null : v;
- });
+ index.remove(row);
}
@Override
@@ -118,23 +117,35 @@ public class TestSortedIndexStorage extends
AbstractTestIndexStorage implements
setEqualityFlag(upperBound);
}
- NavigableMap<ByteBuffer, NavigableMap<RowId, Object>> navigableMap;
+ NavigableSet<IndexRow> navigableSet;
if (lowerBound == null && upperBound == null) {
- navigableMap = index;
+ navigableSet = index;
} else if (lowerBound == null) {
- navigableMap = index.headMap(upperBound.byteBuffer());
+ navigableSet = index.headSet(prefixToIndexRow(upperBound,
highestRowId(partitionId)), true);
} else if (upperBound == null) {
- navigableMap = index.tailMap(lowerBound.byteBuffer());
+ navigableSet = index.tailSet(prefixToIndexRow(lowerBound,
lowestRowId(partitionId)), true);
} else {
try {
- navigableMap = index.subMap(lowerBound.byteBuffer(),
upperBound.byteBuffer());
+ navigableSet = index.subSet(
+ prefixToIndexRow(lowerBound, lowestRowId(partitionId)),
+ true,
+ prefixToIndexRow(upperBound,
highestRowId(partitionId)),
+ true
+ );
} catch (IllegalArgumentException e) {
- navigableMap = emptyNavigableMap();
+ // Upper bound is below the lower bound.
+ navigableSet = emptyNavigableSet();
}
}
- return new ScanCursor(navigableMap);
+ return new ScanCursor(navigableSet);
+ }
+
+ private IndexRowImpl prefixToIndexRow(BinaryTuplePrefix prefix, RowId
rowId) {
+ var binaryTuple = new BinaryTuple(descriptor.binaryTupleSchema(),
prefix.byteBuffer());
+
+ return new IndexRowImpl(binaryTuple, rowId);
}
private static void setEqualityFlag(BinaryTuplePrefix prefix) {
@@ -150,20 +161,22 @@ public class TestSortedIndexStorage extends
AbstractTestIndexStorage implements
index.clear();
}
+ private static final IndexRow NO_PEEKED_ROW = new IndexRowImpl(null, null);
+
private class ScanCursor implements PeekCursor<IndexRow> {
- private final NavigableMap<ByteBuffer, NavigableMap<RowId, Object>>
indexMap;
+ private final NavigableSet<IndexRow> indexSet;
@Nullable
private Boolean hasNext;
@Nullable
- private Entry<ByteBuffer, NavigableMap<RowId, Object>> currentEntry;
+ private IndexRow currentRow;
@Nullable
- private RowId rowId;
+ private IndexRow peekedRow = NO_PEEKED_ROW;
- private ScanCursor(NavigableMap<ByteBuffer, NavigableMap<RowId,
Object>> indexMap) {
- this.indexMap = indexMap;
+ private ScanCursor(NavigableSet<IndexRow> indexSet) {
+ this.indexSet = indexSet;
}
@Override
@@ -175,26 +188,26 @@ public class TestSortedIndexStorage extends
AbstractTestIndexStorage implements
public boolean hasNext() {
checkStorageClosedOrInProcessOfRebalance();
- advanceIfNeeded();
+ if (hasNext != null) {
+ return hasNext;
+ }
+
+ currentRow = peekedRow == NO_PEEKED_ROW ? peek() : peekedRow;
+ peekedRow = NO_PEEKED_ROW;
+ hasNext = currentRow != null;
return hasNext;
}
@Override
public IndexRow next() {
- checkStorageClosedOrInProcessOfRebalance();
-
- advanceIfNeeded();
-
- boolean hasNext = this.hasNext;
-
- if (!hasNext) {
+ if (!hasNext()) {
throw new NoSuchElementException();
}
this.hasNext = null;
- return new IndexRowImpl(new
BinaryTuple(descriptor.binaryTupleSchema(), currentEntry.getKey()), rowId);
+ return currentRow;
}
@Override
@@ -202,78 +215,20 @@ public class TestSortedIndexStorage extends
AbstractTestIndexStorage implements
checkStorageClosedOrInProcessOfRebalance();
if (hasNext != null) {
- if (hasNext) {
- return new IndexRowImpl(new
BinaryTuple(descriptor.binaryTupleSchema(), currentEntry.getKey()), rowId);
- }
-
- return null;
+ return currentRow;
}
- Entry<ByteBuffer, NavigableMap<RowId, Object>> indexMapEntry0 =
currentEntry == null ? indexMap.firstEntry() : currentEntry;
-
- RowId nextRowId = null;
-
- if (rowId == null) {
- if (indexMapEntry0 != null) {
- nextRowId =
getRowId(indexMapEntry0.getValue().firstEntry());
+ if (currentRow == null) {
+ try {
+ peekedRow = indexSet.first();
+ } catch (NoSuchElementException e) {
+ peekedRow = null;
}
} else {
- Entry<RowId, Object> nextRowIdEntry =
indexMapEntry0.getValue().higherEntry(rowId);
-
- if (nextRowIdEntry != null) {
- nextRowId = nextRowIdEntry.getKey();
- } else {
- indexMapEntry0 =
indexMap.higherEntry(indexMapEntry0.getKey());
-
- if (indexMapEntry0 != null) {
- nextRowId =
getRowId(indexMapEntry0.getValue().firstEntry());
- }
- }
- }
-
- return nextRowId == null
- ? null : new IndexRowImpl(new
BinaryTuple(descriptor.binaryTupleSchema(), indexMapEntry0.getKey()),
nextRowId);
- }
-
- private void advanceIfNeeded() {
- if (hasNext != null) {
- return;
- }
-
- if (currentEntry == null) {
- currentEntry = indexMap.firstEntry();
- }
-
- if (rowId == null) {
- if (currentEntry != null) {
- rowId = getRowId(currentEntry.getValue().firstEntry());
- }
- } else {
- Entry<RowId, Object> nextRowIdEntry =
currentEntry.getValue().higherEntry(rowId);
-
- if (nextRowIdEntry != null) {
- rowId = nextRowIdEntry.getKey();
- } else {
- Entry<ByteBuffer, NavigableMap<RowId, Object>>
nextIndexMapEntry = indexMap.higherEntry(currentEntry.getKey());
-
- if (nextIndexMapEntry == null) {
- hasNext = false;
-
- return;
- } else {
- currentEntry = nextIndexMapEntry;
-
- rowId = getRowId(currentEntry.getValue().firstEntry());
- }
- }
+ peekedRow = indexSet.higher(this.currentRow);
}
- hasNext = rowId != null;
- }
-
- @Nullable
- private RowId getRowId(@Nullable Entry<RowId, ?> rowIdEntry) {
- return rowIdEntry == null ? null : rowIdEntry.getKey();
+ return peekedRow;
}
}
@@ -281,6 +236,6 @@ public class TestSortedIndexStorage extends
AbstractTestIndexStorage implements
* Returns all indexed row ids.
*/
public Set<RowId> allRowsIds() {
- return index.values().stream().flatMap(m ->
m.keySet().stream()).collect(Collectors.toSet());
+ return index.stream().map(IndexRow::rowId).collect(Collectors.toSet());
}
}
diff --git
a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java
b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java
index 801533b1e2..6183195682 100644
---
a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java
+++
b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java
@@ -178,8 +178,8 @@ public class PageMemorySortedIndexStorage extends
AbstractPageMemoryIndexStorage
return new SortedIndexRow(new IndexColumns(partitionId,
tuple.byteBuffer()), rowId);
}
- private IndexRowImpl toIndexRowImpl(SortedIndexRow sortedIndexRow) {
- return new IndexRowImpl(
+ private @Nullable IndexRowImpl toIndexRowImpl(@Nullable SortedIndexRow
sortedIndexRow) {
+ return sortedIndexRow == null ? null : new IndexRowImpl(
new BinaryTuple(descriptor.binaryTupleSchema(),
sortedIndexRow.indexColumns().valueBuffer()),
sortedIndexRow.rowId()
);
@@ -224,6 +224,9 @@ public class PageMemorySortedIndexStorage extends
AbstractPageMemoryIndexStorage
};
}
+ /** Constant that represents the absence of value in {@link ScanCursor}. */
+ private static final SortedIndexRow NO_INDEX_ROW = new
SortedIndexRow(null, null);
+
private class ScanCursor implements PeekCursor<IndexRow> {
@Nullable
private Boolean hasNext;
@@ -237,6 +240,9 @@ public class PageMemorySortedIndexStorage extends
AbstractPageMemoryIndexStorage
@Nullable
private SortedIndexRow treeRow;
+ @Nullable
+ private SortedIndexRow peekedRow = NO_INDEX_ROW;
+
private ScanCursor(@Nullable SortedIndexRowKey lower, @Nullable
SortedIndexRowKey upper) {
this.lower = lower;
this.upper = upper;
@@ -251,9 +257,7 @@ public class PageMemorySortedIndexStorage extends
AbstractPageMemoryIndexStorage
public boolean hasNext() {
return busy(() -> {
try {
- advanceIfNeeded();
-
- return hasNext;
+ return advanceIfNeeded();
} catch (IgniteInternalCheckedException e) {
throw new StorageException("Error while advancing the
cursor", e);
}
@@ -264,11 +268,7 @@ public class PageMemorySortedIndexStorage extends
AbstractPageMemoryIndexStorage
public IndexRow next() {
return busy(() -> {
try {
- advanceIfNeeded();
-
- boolean hasNext = this.hasNext;
-
- if (!hasNext) {
+ if (!advanceIfNeeded()) {
throw new NoSuchElementException();
}
@@ -286,56 +286,44 @@ public class PageMemorySortedIndexStorage extends
AbstractPageMemoryIndexStorage
return busy(() -> {
throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemorySortedIndexStorage.this::createStorageInfo);
- if (hasNext != null) {
- if (hasNext) {
- return toIndexRowImpl(treeRow);
- }
-
- return null;
- }
-
try {
- SortedIndexRow nextTreeRow;
-
- if (treeRow == null) {
- nextTreeRow = lower == null ?
sortedIndexTree.findFirst() : sortedIndexTree.findNext(lower, true);
- } else {
- nextTreeRow = sortedIndexTree.findNext(treeRow, false);
- }
-
- if (nextTreeRow == null || (upper != null &&
compareRows(nextTreeRow, upper) >= 0)) {
- return null;
- }
-
- return toIndexRowImpl(nextTreeRow);
+ return toIndexRowImpl(peekBusy());
} catch (IgniteInternalCheckedException e) {
throw new StorageException("Error when peeking next
element", e);
}
});
}
- private void advanceIfNeeded() throws IgniteInternalCheckedException {
- throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemorySortedIndexStorage.this::createStorageInfo);
-
+ private @Nullable SortedIndexRow peekBusy() throws
IgniteInternalCheckedException {
if (hasNext != null) {
- return;
+ return treeRow;
}
if (treeRow == null) {
- treeRow = lower == null ? sortedIndexTree.findFirst() :
sortedIndexTree.findNext(lower, true);
+ peekedRow = lower == null ? sortedIndexTree.findFirst() :
sortedIndexTree.findNext(lower, true);
} else {
- SortedIndexRow next = sortedIndexTree.findNext(treeRow, false);
+ peekedRow = sortedIndexTree.findNext(treeRow, false);
+ }
- if (next == null) {
- hasNext = false;
+ if (peekedRow == null || (upper != null && compareRows(peekedRow,
upper) >= 0)) {
+ peekedRow = null;
+ }
- return;
- } else {
- treeRow = next;
- }
+ return peekedRow;
+ }
+
+ private boolean advanceIfNeeded() throws
IgniteInternalCheckedException {
+ throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemorySortedIndexStorage.this::createStorageInfo);
+
+ if (hasNext != null) {
+ return hasNext;
}
- hasNext = treeRow != null && (upper == null ||
compareRows(treeRow, upper) < 0);
+ treeRow = (peekedRow == NO_INDEX_ROW) ? peekBusy() : peekedRow;
+ peekedRow = NO_INDEX_ROW;
+
+ hasNext = treeRow != null;
+ return hasNext;
}
private int compareRows(SortedIndexRowKey key1, SortedIndexRowKey
key2) {
diff --git
a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/index/AbstractPageMemoryHashIndexStorageTest.java
b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/index/AbstractPageMemoryHashIndexStorageTest.java
index dcd71bb088..badac4a78b 100644
---
a/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/index/AbstractPageMemoryHashIndexStorageTest.java
+++
b/modules/storage-page-memory/src/test/java/org/apache/ignite/internal/storage/pagememory/index/AbstractPageMemoryHashIndexStorageTest.java
@@ -26,11 +26,13 @@ import static org.hamcrest.Matchers.empty;
import java.util.Random;
import org.apache.ignite.internal.pagememory.PageMemory;
import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.schema.testutils.definition.ColumnType;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.storage.index.AbstractHashIndexStorageTest;
import org.apache.ignite.internal.storage.index.HashIndexStorage;
import org.apache.ignite.internal.storage.index.IndexRow;
+import org.apache.ignite.internal.storage.index.impl.BinaryTupleRowSerializer;
import
org.apache.ignite.internal.storage.pagememory.configuration.schema.BasePageMemoryStorageEngineConfiguration;
import org.junit.jupiter.api.Test;
@@ -59,29 +61,38 @@ abstract class AbstractPageMemoryHashIndexStorageTest
extends AbstractHashIndexS
@Test
void testWithStringsLargerThanMaximumInlineSize() {
- IndexRow indexRow0 = createIndexRow(1, randomString(random,
MAX_BINARY_TUPLE_INLINE_SIZE), new RowId(TEST_PARTITION));
- IndexRow indexRow1 = createIndexRow(1, randomString(random,
MAX_BINARY_TUPLE_INLINE_SIZE), new RowId(TEST_PARTITION));
+ HashIndexStorage index = createIndexStorage(INDEX_NAME,
ColumnType.INT32, ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
- put(indexRow0);
- put(indexRow1);
+ IndexRow indexRow0 = createIndexRow(serializer, new
RowId(TEST_PARTITION), 1, randomString(random, MAX_BINARY_TUPLE_INLINE_SIZE));
+ IndexRow indexRow1 = createIndexRow(serializer, new
RowId(TEST_PARTITION), 1, randomString(random, MAX_BINARY_TUPLE_INLINE_SIZE));
- assertThat(getAll(indexRow0), contains(indexRow0.rowId()));
- assertThat(getAll(indexRow1), contains(indexRow1.rowId()));
+ put(index, indexRow0);
+ put(index, indexRow1);
- assertThat(getAll(createIndexRow(1, "foo", new
RowId(TEST_PARTITION))), empty());
+ assertThat(getAll(index, indexRow0), contains(indexRow0.rowId()));
+ assertThat(getAll(index, indexRow1), contains(indexRow1.rowId()));
+
+ assertThat(getAll(index, createIndexRow(serializer, new
RowId(TEST_PARTITION), 1, "foo")), empty());
}
@Test
void testFragmentedIndexColumns() {
- IndexRow indexRow0 = createIndexRow(1, randomString(random,
baseEngineConfig.pageSize().value() * 2), new RowId(TEST_PARTITION));
- IndexRow indexRow1 = createIndexRow(1, randomString(random,
baseEngineConfig.pageSize().value() * 2), new RowId(TEST_PARTITION));
+ HashIndexStorage index = createIndexStorage(INDEX_NAME,
ColumnType.INT32, ColumnType.string());
+ var serializer = new BinaryTupleRowSerializer(indexDescriptor(index));
+
+ String longString0 = randomString(random,
baseEngineConfig.pageSize().value() * 2);
+ String longString1 = randomString(random,
baseEngineConfig.pageSize().value() * 2);
+
+ IndexRow indexRow0 = createIndexRow(serializer, new
RowId(TEST_PARTITION), 1, longString0);
+ IndexRow indexRow1 = createIndexRow(serializer, new
RowId(TEST_PARTITION), 1, longString1);
- put(indexRow0);
- put(indexRow1);
+ put(index, indexRow0);
+ put(index, indexRow1);
- assertThat(getAll(indexRow0), contains(indexRow0.rowId()));
- assertThat(getAll(indexRow1), contains(indexRow1.rowId()));
+ assertThat(getAll(index, indexRow0), contains(indexRow0.rowId()));
+ assertThat(getAll(index, indexRow1), contains(indexRow1.rowId()));
- assertThat(getAll(createIndexRow(1, "foo", new
RowId(TEST_PARTITION))), empty());
+ assertThat(getAll(index, createIndexRow(serializer, new
RowId(TEST_PARTITION), 1, "foo")), empty());
}
}
diff --git
a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbSortedIndexStorage.java
b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbSortedIndexStorage.java
index afe0c9c295..3cc782171d 100644
---
a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbSortedIndexStorage.java
+++
b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbSortedIndexStorage.java
@@ -202,6 +202,8 @@ public class RocksDbSortedIndexStorage extends
AbstractRocksDbIndexStorage imple
private byte @Nullable [] key;
+ private byte @Nullable [] peekedKey = BYTE_EMPTY_ARRAY;
+
@Override
public void close() {
try {
@@ -213,21 +215,13 @@ public class RocksDbSortedIndexStorage extends
AbstractRocksDbIndexStorage imple
@Override
public boolean hasNext() {
- return busy(() -> {
- advanceIfNeeded();
-
- return hasNext;
- });
+ return busy(this::advanceIfNeeded);
}
@Override
public T next() {
return busy(() -> {
- advanceIfNeeded();
-
- boolean hasNext = this.hasNext;
-
- if (!hasNext) {
+ if (!advanceIfNeeded()) {
throw new NoSuchElementException();
}
@@ -242,31 +236,19 @@ public class RocksDbSortedIndexStorage extends
AbstractRocksDbIndexStorage imple
return busy(() -> {
throwExceptionIfStorageInProgressOfRebalance(state.get(),
RocksDbSortedIndexStorage.this::createStorageInfo);
- if (hasNext != null) {
- if (hasNext) {
- return
mapper.apply(ByteBuffer.wrap(key).order(KEY_BYTE_ORDER));
- }
-
- return null;
- }
-
- refreshAndPrepareRocksIterator();
-
- if (!it.isValid()) {
- RocksUtils.checkIterator(it);
+ byte[] res = peek0();
+ if (res == null) {
return null;
} else {
- return
mapper.apply(ByteBuffer.wrap(it.key()).order(KEY_BYTE_ORDER));
+ return
mapper.apply(ByteBuffer.wrap(res).order(KEY_BYTE_ORDER));
}
});
}
- private void advanceIfNeeded() throws StorageException {
- throwExceptionIfStorageInProgressOfRebalance(state.get(),
RocksDbSortedIndexStorage.this::createStorageInfo);
-
+ private byte @Nullable [] peek0() {
if (hasNext != null) {
- return;
+ return key;
}
refreshAndPrepareRocksIterator();
@@ -274,12 +256,23 @@ public class RocksDbSortedIndexStorage extends
AbstractRocksDbIndexStorage imple
if (!it.isValid()) {
RocksUtils.checkIterator(it);
- hasNext = false;
+ peekedKey = null;
} else {
- key = it.key();
-
- hasNext = true;
+ peekedKey = it.key();
}
+
+ return peekedKey;
+ }
+
+ private boolean advanceIfNeeded() throws StorageException {
+ throwExceptionIfStorageInProgressOfRebalance(state.get(),
RocksDbSortedIndexStorage.this::createStorageInfo);
+
+ //noinspection ArrayEquality
+ key = (peekedKey == BYTE_EMPTY_ARRAY) ? peek0() : peekedKey;
+ peekedKey = BYTE_EMPTY_ARRAY;
+
+ hasNext = key != null;
+ return hasNext;
}
private void refreshAndPrepareRocksIterator() {
