You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2023/03/28 05:41:02 UTC
[ignite-3] branch main updated: IGNITE-19119 Prepare org.apache.ignite.internal.storage.index.IndexStorage to built index (#1839)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill 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 37d1e0f847 IGNITE-19119 Prepare org.apache.ignite.internal.storage.index.IndexStorage to built index (#1839)
37d1e0f847 is described below
commit 37d1e0f84705be380aa2fc5ddcc364c02e4f347b
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Tue Mar 28 08:40:56 2023 +0300
IGNITE-19119 Prepare org.apache.ignite.internal.storage.index.IndexStorage to built index (#1839)
---
.../internal/storage/index/IndexStorage.java | 26 ++-
.../storage/AbstractMvTableStorageTest.java | 64 ++++++
.../internal/storage/BaseMvStoragesTest.java | 1 -
.../internal/storage/impl/TestMvTableStorage.java | 4 +-
...xStorage.java => AbstractTestIndexStorage.java} | 114 +++--------
.../storage/index/impl/TestHashIndexStorage.java | 108 +----------
.../storage/index/impl/TestSortedIndexStorage.java | 113 +----------
.../index/AbstractPageMemoryIndexStorage.java | 215 +++++++++++++++++++++
.../index/hash/PageMemoryHashIndexStorage.java | 136 ++-----------
.../storage/pagememory/index/meta/IndexMeta.java | 31 +--
.../meta/{IndexMeta.java => IndexMetaKey.java} | 36 +---
.../pagememory/index/meta/IndexMetaTree.java | 6 +-
.../UpdateLastRowIdUuidToBuiltInvokeClosure.java | 60 ++++++
.../pagememory/index/meta/io/IndexMetaInnerIo.java | 9 +-
.../pagememory/index/meta/io/IndexMetaIo.java | 51 ++++-
.../pagememory/index/meta/io/IndexMetaLeafIo.java | 9 +-
.../index/sorted/PageMemorySortedIndexStorage.java | 140 ++------------
.../mv/AbstractPageMemoryMvPartitionStorage.java | 46 ++---
.../mv/PersistentPageMemoryMvPartitionStorage.java | 5 +-
.../mv/VolatilePageMemoryMvPartitionStorage.java | 5 +-
.../internal/storage/rocksdb/GarbageCollector.java | 6 +-
.../ignite/internal/storage/rocksdb/HashIndex.java | 7 +-
.../storage/rocksdb/PartitionDataHelper.java | 40 ++--
.../storage/rocksdb/RocksDbMetaStorage.java | 108 +++++++++--
.../storage/rocksdb/RocksDbMvPartitionStorage.java | 4 +-
.../storage/rocksdb/RocksDbStorageUtils.java | 69 +++++++
.../storage/rocksdb/RocksDbTableStorage.java | 26 ++-
.../internal/storage/rocksdb/SortedIndex.java | 7 +-
.../rocksdb/index/AbstractRocksDbIndexStorage.java | 205 ++++++++++++++++++++
.../rocksdb/index/RocksDbHashIndexStorage.java | 161 ++-------------
.../rocksdb/index/RocksDbSortedIndexStorage.java | 170 +++-------------
.../distributed/ItTxDistributedTestSingleNode.java | 2 +-
.../internal/table/distributed/IndexBaseTest.java | 16 +-
.../table/distributed/IndexCleanupTest.java | 22 +--
.../internal/table/distributed/IndexGcTest.java | 10 +-
.../raft/PartitionCommandListenerTest.java | 2 +-
.../PartitionReplicaListenerIndexLockingTest.java | 4 +-
.../replication/PartitionReplicaListenerTest.java | 6 +-
.../table/impl/DummyInternalTableImpl.java | 2 +-
39 files changed, 1034 insertions(+), 1012 deletions(-)
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/IndexStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/IndexStorage.java
index 0addff3490..bced7f0004 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/IndexStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/index/IndexStorage.java
@@ -18,9 +18,11 @@
package org.apache.ignite.internal.storage.index;
import org.apache.ignite.internal.schema.BinaryTuple;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageException;
import org.apache.ignite.internal.util.Cursor;
+import org.jetbrains.annotations.Nullable;
/**
* Common interface for all Index Storage implementations.
@@ -37,7 +39,7 @@ public interface IndexStorage {
* Adds the given index row to the index.
*
* @apiNote This method <b>must</b> always be called inside the corresponding partition's
- * {@link org.apache.ignite.internal.storage.MvPartitionStorage#runConsistently} closure.
+ * {@link MvPartitionStorage#runConsistently} closure.
*
* @throws StorageException If failed to put data.
*/
@@ -49,9 +51,29 @@ public interface IndexStorage {
* <p>Removing a non-existent row is a no-op.
*
* @apiNote This method <b>must</b> always be called inside the corresponding partition's
- * {@link org.apache.ignite.internal.storage.MvPartitionStorage#runConsistently} closure.
+ * {@link MvPartitionStorage#runConsistently} closure.
*
* @throws StorageException If failed to remove data.
*/
void remove(IndexRow row) throws StorageException;
+
+ /**
+ * Returns the row ID for which the index needs to be built, {@code null} means that the index building has completed.
+ *
+ * <p>If index building has not started yet, it will return {@link RowId#lowestRowId(int)}.
+ *
+ * @throws StorageException If failed to get the row ID.
+ */
+ @Nullable RowId getNextRowIdToBuild() throws StorageException;
+
+ /**
+ * Sets the row ID for which the index needs to be built, {@code null} means that the index is built.
+ *
+ * @apiNote This method <b>must</b> always be called inside the corresponding partition's
+ * {@link MvPartitionStorage#runConsistently} closure.
+ *
+ * @param rowId Row ID.
+ * @throws StorageException If failed to set the row ID.
+ */
+ void setNextRowIdToBuild(@Nullable RowId rowId) throws StorageException;
}
diff --git a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java
index 0e3fcebcb2..598055b5ab 100644
--- a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java
+++ b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java
@@ -729,6 +729,70 @@ public abstract class AbstractMvTableStorageTest extends BaseMvStoragesTest {
assertThat(tableStorage.destroyPartition(PARTITION_ID), willCompleteSuccessfully());
}
+ @Test
+ void testNextRowIdToBuilt() {
+ MvPartitionStorage mvPartitionStorage = getOrCreateMvPartition(PARTITION_ID);
+
+ IndexStorage hashIndexStorage = tableStorage.getOrCreateIndex(PARTITION_ID, hashIdx.id());
+ IndexStorage sortedIndexStorage = tableStorage.getOrCreateIndex(PARTITION_ID, sortedIdx.id());
+
+ assertThat(hashIndexStorage.getNextRowIdToBuild(), equalTo(RowId.lowestRowId(PARTITION_ID)));
+ assertThat(sortedIndexStorage.getNextRowIdToBuild(), equalTo(RowId.lowestRowId(PARTITION_ID)));
+
+ RowId rowId0 = new RowId(PARTITION_ID);
+ RowId rowId1 = new RowId(PARTITION_ID);
+
+ mvPartitionStorage.runConsistently(() -> {
+ hashIndexStorage.setNextRowIdToBuild(rowId0);
+ sortedIndexStorage.setNextRowIdToBuild(rowId1);
+
+ return null;
+ });
+
+ assertThat(hashIndexStorage.getNextRowIdToBuild(), equalTo(rowId0));
+ assertThat(sortedIndexStorage.getNextRowIdToBuild(), equalTo(rowId1));
+ }
+
+ @Test
+ void testNextRowIdToBuiltAfterRestart() {
+ MvPartitionStorage mvPartitionStorage = getOrCreateMvPartition(PARTITION_ID);
+
+ IndexStorage hashIndexStorage = tableStorage.getOrCreateIndex(PARTITION_ID, hashIdx.id());
+ IndexStorage sortedIndexStorage = tableStorage.getOrCreateIndex(PARTITION_ID, sortedIdx.id());
+
+ RowId rowId0 = new RowId(PARTITION_ID);
+ RowId rowId1 = new RowId(PARTITION_ID);
+
+ mvPartitionStorage.runConsistently(() -> {
+ hashIndexStorage.setNextRowIdToBuild(rowId0);
+ sortedIndexStorage.setNextRowIdToBuild(rowId1);
+
+ return null;
+ });
+
+ assertThat(mvPartitionStorage.flush(), willCompleteSuccessfully());
+
+ // Restart storages.
+ tableStorage.stop();
+
+ tableStorage = createMvTableStorage(tableStorage.tablesConfiguration());
+
+ tableStorage.start();
+
+ getOrCreateMvPartition(PARTITION_ID);
+
+ IndexStorage hashIndexStorageRestarted = tableStorage.getOrCreateIndex(PARTITION_ID, hashIdx.id());
+ IndexStorage sortedIndexStorageRestarted = tableStorage.getOrCreateIndex(PARTITION_ID, sortedIdx.id());
+
+ if (tableStorage.isVolatile()) {
+ assertThat(hashIndexStorageRestarted.getNextRowIdToBuild(), equalTo(RowId.lowestRowId(PARTITION_ID)));
+ assertThat(sortedIndexStorageRestarted.getNextRowIdToBuild(), equalTo(RowId.lowestRowId(PARTITION_ID)));
+ } else {
+ assertThat(hashIndexStorageRestarted.getNextRowIdToBuild(), equalTo(rowId0));
+ assertThat(sortedIndexStorageRestarted.getNextRowIdToBuild(), equalTo(rowId1));
+ }
+ }
+
private static void createTestIndexes(TablesConfiguration tablesConfig) {
List<IndexDefinition> indexDefinitions = List.of(
SchemaBuilders.sortedIndex(SORTED_INDEX_NAME)
diff --git a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvStoragesTest.java b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvStoragesTest.java
index 6c69eee36e..aefeb91f86 100644
--- a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvStoragesTest.java
+++ b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvStoragesTest.java
@@ -84,7 +84,6 @@ public abstract class BaseMvStoragesTest {
}
}
-
protected static IndexRow indexRow(IndexDescriptor indexDescriptor, BinaryRow binaryRow, RowId rowId) {
int[] columnIndexes = indexDescriptor.columns().stream()
.mapToInt(indexColumnDescriptor -> {
diff --git a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvTableStorage.java b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvTableStorage.java
index 251af1fde2..fc7bbb8811 100644
--- a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvTableStorage.java
+++ b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvTableStorage.java
@@ -70,7 +70,7 @@ public class TestMvTableStorage implements MvTableStorage {
}
TestSortedIndexStorage getOrCreateStorage(Integer partitionId) {
- return storageByPartitionId.computeIfAbsent(partitionId, id -> new TestSortedIndexStorage(descriptor));
+ return storageByPartitionId.computeIfAbsent(partitionId, id -> new TestSortedIndexStorage(id, descriptor));
}
}
@@ -87,7 +87,7 @@ public class TestMvTableStorage implements MvTableStorage {
}
TestHashIndexStorage getOrCreateStorage(Integer partitionId) {
- return storageByPartitionId.computeIfAbsent(partitionId, id -> new TestHashIndexStorage(descriptor));
+ return storageByPartitionId.computeIfAbsent(partitionId, id -> new TestHashIndexStorage(id, descriptor));
}
}
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/AbstractTestIndexStorage.java
similarity index 55%
copy from modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/TestHashIndexStorage.java
copy to modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/index/impl/AbstractTestIndexStorage.java
index 49563660e2..b84ecd5c22 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/AbstractTestIndexStorage.java
@@ -17,53 +17,34 @@
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.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-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.StorageClosedException;
import org.apache.ignite.internal.storage.StorageRebalanceException;
-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.apache.ignite.internal.storage.index.IndexStorage;
import org.apache.ignite.internal.util.Cursor;
+import org.jetbrains.annotations.Nullable;
/**
- * Test-only implementation of a {@link HashIndexStorage}.
+ * Test-only abstract index storage class.
*/
-public class TestHashIndexStorage implements HashIndexStorage {
- private final ConcurrentMap<ByteBuffer, Set<RowId>> index = new ConcurrentHashMap<>();
-
- private final HashIndexDescriptor descriptor;
-
+abstract class AbstractTestIndexStorage implements IndexStorage {
private volatile boolean closed;
private volatile boolean rebalance;
- /**
- * Constructor.
- */
- public TestHashIndexStorage(HashIndexDescriptor descriptor) {
- this.descriptor = descriptor;
- }
+ private volatile @Nullable RowId nextRowIdToBuild;
- @Override
- public HashIndexDescriptor indexDescriptor() {
- return descriptor;
+ AbstractTestIndexStorage(int partitionId) {
+ nextRowIdToBuild = RowId.lowestRowId(partitionId);
}
@Override
public Cursor<RowId> get(BinaryTuple key) {
checkStorageClosedOrInProcessOfRebalance();
- Iterator<RowId> iterator = index.getOrDefault(key.byteBuffer(), Set.of()).iterator();
+ Iterator<RowId> iterator = getRowIdIteratorForGetByBinaryTuple(key);
return new Cursor<>() {
@Override
@@ -88,51 +69,17 @@ public class TestHashIndexStorage implements HashIndexStorage {
}
@Override
- public void put(IndexRow row) {
- checkStorageClosed();
-
- 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));
-
- result.addAll(v);
- result.add(row.rowId());
-
- return result;
- }
- });
- }
-
- @Override
- public void remove(IndexRow row) {
+ public @Nullable RowId getNextRowIdToBuild() {
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;
- }
- });
+ return nextRowIdToBuild;
}
@Override
- public void destroy() {
- closed = true;
+ public void setNextRowIdToBuild(@Nullable RowId rowId) {
+ checkStorageClosedOrInProcessOfRebalance();
- clear0();
+ nextRowIdToBuild = rowId;
}
/**
@@ -144,23 +91,15 @@ public class TestHashIndexStorage implements HashIndexStorage {
clear0();
}
- private void clear0() {
- index.clear();
- }
+ public void destroy() {
+ closed = true;
- private void checkStorageClosed() {
- if (closed) {
- throw new StorageClosedException();
- }
+ clear0();
}
- private void checkStorageClosedOrInProcessOfRebalance() {
- checkStorageClosed();
+ abstract Iterator<RowId> getRowIdIteratorForGetByBinaryTuple(BinaryTuple key);
- if (rebalance) {
- throw new StorageRebalanceException("Storage in the process of rebalancing");
- }
- }
+ abstract void clear0();
/**
* Starts rebalancing of the storage.
@@ -199,10 +138,17 @@ public class TestHashIndexStorage implements HashIndexStorage {
rebalance = false;
}
- /**
- * Returns all indexed row ids.
- */
- public Set<RowId> allRowsIds() {
- return index.values().stream().flatMap(Set::stream).collect(Collectors.toSet());
+ void checkStorageClosed() {
+ if (closed) {
+ throw new StorageClosedException();
+ }
+ }
+
+ void checkStorageClosedOrInProcessOfRebalance() {
+ checkStorageClosed();
+
+ if (rebalance) {
+ throw new StorageRebalanceException("Storage in the process of rebalancing");
+ }
}
}
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 49563660e2..c31cb3b898 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
@@ -28,29 +28,24 @@ import java.util.concurrent.ConcurrentMap;
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.StorageClosedException;
-import org.apache.ignite.internal.storage.StorageRebalanceException;
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.apache.ignite.internal.util.Cursor;
/**
* Test-only implementation of a {@link HashIndexStorage}.
*/
-public class TestHashIndexStorage implements HashIndexStorage {
+public class TestHashIndexStorage extends AbstractTestIndexStorage implements HashIndexStorage {
private final ConcurrentMap<ByteBuffer, Set<RowId>> index = new ConcurrentHashMap<>();
private final HashIndexDescriptor descriptor;
- private volatile boolean closed;
-
- private volatile boolean rebalance;
-
/**
* Constructor.
*/
- public TestHashIndexStorage(HashIndexDescriptor descriptor) {
+ public TestHashIndexStorage(int partitionId, HashIndexDescriptor descriptor) {
+ super(partitionId);
+
this.descriptor = descriptor;
}
@@ -60,31 +55,8 @@ public class TestHashIndexStorage implements HashIndexStorage {
}
@Override
- public Cursor<RowId> get(BinaryTuple key) {
- checkStorageClosedOrInProcessOfRebalance();
-
- Iterator<RowId> iterator = index.getOrDefault(key.byteBuffer(), Set.of()).iterator();
-
- return new Cursor<>() {
- @Override
- public void close() {
- // No-op.
- }
-
- @Override
- public boolean hasNext() {
- checkStorageClosedOrInProcessOfRebalance();
-
- return iterator.hasNext();
- }
-
- @Override
- public RowId next() {
- checkStorageClosedOrInProcessOfRebalance();
-
- return iterator.next();
- }
- };
+ Iterator<RowId> getRowIdIteratorForGetByBinaryTuple(BinaryTuple key) {
+ return index.getOrDefault(key.byteBuffer(), Set.of()).iterator();
}
@Override
@@ -129,76 +101,10 @@ public class TestHashIndexStorage implements HashIndexStorage {
}
@Override
- public void destroy() {
- closed = true;
-
- clear0();
- }
-
- /**
- * Removes all index data.
- */
- public void clear() {
- checkStorageClosedOrInProcessOfRebalance();
-
- clear0();
- }
-
- private void clear0() {
+ void clear0() {
index.clear();
}
- private void checkStorageClosed() {
- if (closed) {
- throw new StorageClosedException();
- }
- }
-
- private void checkStorageClosedOrInProcessOfRebalance() {
- checkStorageClosed();
-
- if (rebalance) {
- throw new StorageRebalanceException("Storage in the process of rebalancing");
- }
- }
-
- /**
- * Starts rebalancing of the storage.
- */
- public void startRebalance() {
- checkStorageClosed();
-
- rebalance = true;
-
- clear0();
- }
-
- /**
- * Aborts rebalance of the storage.
- */
- public void abortRebalance() {
- checkStorageClosed();
-
- if (!rebalance) {
- return;
- }
-
- rebalance = false;
-
- clear0();
- }
-
- /**
- * Completes rebalance of the storage.
- */
- public void finishRebalance() {
- checkStorageClosed();
-
- assert rebalance;
-
- rebalance = false;
- }
-
/**
* Returns all indexed row ids.
*/
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 b23297eb2a..6961c07765 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
@@ -33,22 +33,18 @@ import org.apache.ignite.internal.binarytuple.BinaryTupleCommon;
import org.apache.ignite.internal.schema.BinaryTuple;
import org.apache.ignite.internal.schema.BinaryTuplePrefix;
import org.apache.ignite.internal.storage.RowId;
-import org.apache.ignite.internal.storage.StorageClosedException;
-import org.apache.ignite.internal.storage.StorageException;
-import org.apache.ignite.internal.storage.StorageRebalanceException;
import org.apache.ignite.internal.storage.index.BinaryTupleComparator;
import org.apache.ignite.internal.storage.index.IndexRow;
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.Cursor;
import org.jetbrains.annotations.Nullable;
/**
* Test implementation of MV sorted index storage.
*/
-public class TestSortedIndexStorage implements SortedIndexStorage {
+public class TestSortedIndexStorage extends AbstractTestIndexStorage implements SortedIndexStorage {
private static final Object NULL = new Object();
/**
@@ -59,14 +55,12 @@ public class TestSortedIndexStorage implements SortedIndexStorage {
private final SortedIndexDescriptor descriptor;
- private volatile boolean closed;
-
- private volatile boolean rebalance;
-
/**
* Constructor.
*/
- public TestSortedIndexStorage(SortedIndexDescriptor descriptor) {
+ public TestSortedIndexStorage(int partitionId, SortedIndexDescriptor descriptor) {
+ super(partitionId);
+
this.descriptor = descriptor;
this.index = new ConcurrentSkipListMap<>(new BinaryTupleComparator(descriptor));
}
@@ -77,31 +71,8 @@ public class TestSortedIndexStorage implements SortedIndexStorage {
}
@Override
- public Cursor<RowId> get(BinaryTuple key) throws StorageException {
- checkStorageClosedOrInProcessOfRebalance();
-
- Iterator<RowId> iterator = index.getOrDefault(key.byteBuffer(), emptyNavigableMap()).keySet().iterator();
-
- return new Cursor<>() {
- @Override
- public void close() {
- // No-op.
- }
-
- @Override
- public boolean hasNext() {
- checkStorageClosedOrInProcessOfRebalance();
-
- return iterator.hasNext();
- }
-
- @Override
- public RowId next() {
- checkStorageClosedOrInProcessOfRebalance();
-
- return iterator.next();
- }
- };
+ Iterator<RowId> getRowIdIteratorForGetByBinaryTuple(BinaryTuple key) {
+ return index.getOrDefault(key.byteBuffer(), emptyNavigableMap()).keySet().iterator();
}
@Override
@@ -174,25 +145,8 @@ public class TestSortedIndexStorage implements SortedIndexStorage {
buffer.put(0, (byte) (flags | BinaryTupleCommon.EQUALITY_FLAG));
}
- /**
- * Destroys the storage and the data in it.
- */
- public void destroy() {
- closed = true;
-
- clear0();
- }
-
- /**
- * Removes all index data.
- */
- public void clear() {
- checkStorageClosedOrInProcessOfRebalance();
-
- index.clear();
- }
-
- private void clear0() {
+ @Override
+ void clear0() {
index.clear();
}
@@ -323,57 +277,6 @@ public class TestSortedIndexStorage implements SortedIndexStorage {
}
}
- private void checkStorageClosed() {
- if (closed) {
- throw new StorageClosedException();
- }
- }
-
- private void checkStorageClosedOrInProcessOfRebalance() {
- checkStorageClosed();
-
- if (rebalance) {
- throw new StorageRebalanceException("Storage in the process of rebalancing");
- }
- }
-
- /**
- * Starts rebalancing for the storage.
- */
- public void startRebalance() {
- checkStorageClosed();
-
- rebalance = true;
-
- clear0();
- }
-
- /**
- * Aborts rebalance of the storage.
- */
- public void abortRebalance() {
- checkStorageClosed();
-
- if (!rebalance) {
- return;
- }
-
- rebalance = false;
-
- clear0();
- }
-
- /**
- * Completes rebalance of the storage.
- */
- public void finishRebalance() {
- checkStorageClosed();
-
- assert rebalance;
-
- rebalance = false;
- }
-
/**
* Returns all indexed row ids.
*/
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/AbstractPageMemoryIndexStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/AbstractPageMemoryIndexStorage.java
new file mode 100644
index 0000000000..2d0c315b72
--- /dev/null
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/AbstractPageMemoryIndexStorage.java
@@ -0,0 +1,215 @@
+/*
+ * 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.pagememory.index;
+
+import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageState;
+import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
+import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
+
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.StorageException;
+import org.apache.ignite.internal.storage.StorageRebalanceException;
+import org.apache.ignite.internal.storage.index.IndexStorage;
+import org.apache.ignite.internal.storage.pagememory.index.freelist.IndexColumnsFreeList;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaKey;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree;
+import org.apache.ignite.internal.storage.pagememory.index.meta.UpdateLastRowIdUuidToBuiltInvokeClosure;
+import org.apache.ignite.internal.storage.util.StorageState;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteInternalCheckedException;
+import org.apache.ignite.lang.IgniteStringFormatter;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Abstract index storage based on Page Memory.
+ */
+public abstract class AbstractPageMemoryIndexStorage implements IndexStorage {
+ /** Index ID. */
+ private final UUID indexId;
+
+ /** Partition id. */
+ protected final int partitionId;
+
+ /** Lowest possible RowId according to signed long ordering. */
+ protected final RowId lowestRowId;
+
+ /** Highest possible RowId according to signed long ordering. */
+ protected final RowId highestRowId;
+
+ /** Free list to store index columns. */
+ protected volatile IndexColumnsFreeList freeList;
+
+ /** Busy lock. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Current state of the storage. */
+ protected final AtomicReference<StorageState> state = new AtomicReference<>(StorageState.RUNNABLE);
+
+ /** Index meta tree instance. */
+ private final IndexMetaTree indexMetaTree;
+
+ /** Row ID for which the index needs to be built, {@code null} means that the index building has completed. */
+ private volatile @Nullable RowId nextRowIdToBuilt;
+
+ protected AbstractPageMemoryIndexStorage(
+ IndexMeta indexMeta,
+ int partitionId,
+ IndexColumnsFreeList freeList,
+ IndexMetaTree indexMetaTree
+ ) {
+ this.indexId = indexMeta.indexId();
+ this.partitionId = partitionId;
+ this.freeList = freeList;
+ this.indexMetaTree = indexMetaTree;
+
+ lowestRowId = RowId.lowestRowId(partitionId);
+
+ highestRowId = RowId.highestRowId(partitionId);
+
+ nextRowIdToBuilt = indexMeta.nextRowIdUuidToBuild() == null ? null : new RowId(partitionId, indexMeta.nextRowIdUuidToBuild());
+ }
+
+ @Override
+ public @Nullable RowId getNextRowIdToBuild() {
+ return busy(() -> {
+ throwExceptionIfStorageInProgressOfRebalance(state.get(), this::createStorageInfo);
+
+ return nextRowIdToBuilt;
+ });
+ }
+
+ @Override
+ public void setNextRowIdToBuild(@Nullable RowId rowId) {
+ busy(() -> {
+ throwExceptionIfStorageInProgressOfRebalance(state.get(), this::createStorageInfo);
+
+ UUID rowIdUuid = rowId == null ? null : rowId.uuid();
+
+ try {
+ indexMetaTree.invoke(new IndexMetaKey(indexId), null, new UpdateLastRowIdUuidToBuiltInvokeClosure(rowIdUuid));
+ } catch (IgniteInternalCheckedException e) {
+ throw new StorageException("Error updating last row ID uuid to built: [{}, rowId={}]", e, createStorageInfo(), rowId);
+ }
+
+ nextRowIdToBuilt = rowId;
+
+ return null;
+ });
+ }
+
+ /**
+ * Closes the hash index storage.
+ */
+ public void close() {
+ if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
+ StorageState state = this.state.get();
+
+ assert state == StorageState.CLOSED : state;
+
+ return;
+ }
+
+ busyLock.block();
+
+ closeStructures();
+ }
+
+ /**
+ * Prepares storage for rebalancing.
+ *
+ * @throws StorageRebalanceException If there was an error when starting the rebalance.
+ */
+ public void startRebalance() {
+ if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.REBALANCE)) {
+ throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
+ }
+
+ // Changed storage states and expect all storage operations to stop soon.
+ busyLock.block();
+
+ try {
+ closeStructures();
+ } finally {
+ busyLock.unblock();
+ }
+ }
+
+ /**
+ * Completes the rebalancing of the storage.
+ *
+ * @throws StorageRebalanceException If there is an error while completing the storage rebalance.
+ */
+ public void completeRebalance() {
+ if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
+ throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
+ }
+ }
+
+ /**
+ * Prepares the storage for cleanup.
+ *
+ * <p>After cleanup (successful or not), method {@link #finishCleanup()} must be called.
+ */
+ public void startCleanup() {
+ if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLEANUP)) {
+ throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
+ }
+
+ // Changed storage states and expect all storage operations to stop soon.
+ busyLock.block();
+
+ try {
+ closeStructures();
+ } finally {
+ busyLock.unblock();
+ }
+ }
+
+ /**
+ * Finishes cleanup up the storage.
+ */
+ public void finishCleanup() {
+ state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE);
+ }
+
+ protected <V> V busy(Supplier<V> supplier) {
+ if (!busyLock.enterBusy()) {
+ throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
+ }
+
+ try {
+ return supplier.get();
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ protected String createStorageInfo() {
+ return IgniteStringFormatter.format("indexId={}, partitionId={}", indexId, partitionId);
+ }
+
+ /**
+ * Closes internal structures.
+ */
+ protected abstract void closeStructures();
+}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java
index 7ad2203078..26feeb34cb 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java
@@ -17,13 +17,9 @@
package org.apache.ignite.internal.storage.pagememory.index.hash;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageState;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageNotInCleanupOrRebalancedState;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Supplier;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.pagememory.util.GradualTaskExecutor;
@@ -31,69 +27,49 @@ import org.apache.ignite.internal.pagememory.util.PageIdUtils;
import org.apache.ignite.internal.schema.BinaryTuple;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageException;
-import org.apache.ignite.internal.storage.StorageRebalanceException;
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.apache.ignite.internal.storage.pagememory.index.AbstractPageMemoryIndexStorage;
import org.apache.ignite.internal.storage.pagememory.index.freelist.IndexColumns;
import org.apache.ignite.internal.storage.pagememory.index.freelist.IndexColumnsFreeList;
-import org.apache.ignite.internal.storage.util.StorageState;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree;
import org.apache.ignite.internal.util.Cursor;
-import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.lang.IgniteInternalCheckedException;
-import org.apache.ignite.lang.IgniteStringFormatter;
/**
* Implementation of Hash index storage using Page Memory.
*/
-public class PageMemoryHashIndexStorage implements HashIndexStorage {
+public class PageMemoryHashIndexStorage extends AbstractPageMemoryIndexStorage implements HashIndexStorage {
private static final IgniteLogger LOG = Loggers.forClass(PageMemoryHashIndexStorage.class);
/** Index descriptor. */
private final HashIndexDescriptor descriptor;
- /** Free list to store index columns. */
- private volatile IndexColumnsFreeList freeList;
-
/** Hash index tree instance. */
private volatile HashIndexTree hashIndexTree;
- /** Partition id. */
- private final int partitionId;
-
- /** Lowest possible RowId according to signed long ordering. */
- private final RowId lowestRowId;
-
- /** Highest possible RowId according to signed long ordering. */
- private final RowId highestRowId;
-
- /** Busy lock. */
- private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
-
- /** Current state of the storage. */
- private final AtomicReference<StorageState> state = new AtomicReference<>(StorageState.RUNNABLE);
-
/**
* Constructor.
*
+ * @param indexMeta Index meta.
* @param descriptor Hash index descriptor.
* @param freeList Free list to store index columns.
* @param hashIndexTree Hash index tree instance.
+ * @param indexMetaTree Index meta tree instance.
*/
public PageMemoryHashIndexStorage(
+ IndexMeta indexMeta,
HashIndexDescriptor descriptor,
IndexColumnsFreeList freeList,
- HashIndexTree hashIndexTree
+ HashIndexTree hashIndexTree,
+ IndexMetaTree indexMetaTree
) {
+ super(indexMeta, hashIndexTree.partitionId(), freeList, indexMetaTree);
+
this.descriptor = descriptor;
- this.freeList = freeList;
this.hashIndexTree = hashIndexTree;
-
- partitionId = hashIndexTree.partitionId();
-
- lowestRowId = RowId.lowestRowId(partitionId);
-
- highestRowId = RowId.highestRowId(partitionId);
}
@Override
@@ -225,66 +201,11 @@ public class PageMemoryHashIndexStorage implements HashIndexStorage {
}
}
- /**
- * Closes the hash index storage.
- */
- public void close() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
- StorageState state = this.state.get();
-
- assert state == StorageState.CLOSED : state;
-
- return;
- }
-
- busyLock.block();
-
+ @Override
+ public void closeStructures() {
hashIndexTree.close();
}
- private <V> V busy(Supplier<V> supplier) {
- if (!busyLock.enterBusy()) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- try {
- return supplier.get();
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- /**
- * Prepares storage for rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when starting the rebalance.
- */
- public void startRebalance() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.REBALANCE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- try {
- this.hashIndexTree.close();
- } finally {
- busyLock.unblock();
- }
- }
-
- /**
- * Completes the rebalancing of the storage.
- *
- * @throws StorageRebalanceException If there is an error while completing the storage rebalance.
- */
- public void completeRebalance() {
- if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
- }
-
/**
* Updates the internal data structures of the storage on rebalance or cleanup.
*
@@ -298,35 +219,4 @@ public class PageMemoryHashIndexStorage implements HashIndexStorage {
this.freeList = freeList;
this.hashIndexTree = hashIndexTree;
}
-
- private String createStorageInfo() {
- return IgniteStringFormatter.format("indexId={}, partitionId={}", descriptor.id(), partitionId);
- }
-
- /**
- * Prepares the storage for cleanup.
- *
- * <p>After cleanup (successful or not), method {@link #finishCleanup()} must be called.
- */
- public void startCleanup() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLEANUP)) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- try {
- hashIndexTree.close();
- } finally {
- busyLock.unblock();
- }
- }
-
- /**
- * Finishes cleanup up the storage.
- */
- public void finishCleanup() {
- state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE);
- }
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMeta.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMeta.java
index 36e3015b4e..def8834b7f 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMeta.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMeta.java
@@ -21,43 +21,48 @@ import java.util.UUID;
import org.apache.ignite.internal.tostring.IgniteToStringExclude;
import org.apache.ignite.internal.tostring.S;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.jetbrains.annotations.Nullable;
/**
- * Index tree meta information.
+ * Index meta information.
*/
-public class IndexMeta {
- private final UUID id;
-
+public class IndexMeta extends IndexMetaKey {
@IgniteToStringExclude
private final long metaPageId;
+ private final @Nullable UUID nextRowIdUuidToBuild;
+
/**
* Constructor.
*
* @param id Index ID.
* @param metaPageId Index tree meta page ID.
+ * @param nextRowIdUuidToBuild Row ID uuid for which the index needs to be built, {@code null} means that the index building has
+ * completed.
*/
- public IndexMeta(UUID id, long metaPageId) {
- this.id = id;
+ public IndexMeta(UUID id, long metaPageId, @Nullable UUID nextRowIdUuidToBuild) {
+ super(id);
+
this.metaPageId = metaPageId;
+ this.nextRowIdUuidToBuild = nextRowIdUuidToBuild;
}
/**
- * Returns the index ID.
+ * Returns page ID of the index tree meta page.
*/
- public UUID id() {
- return id;
+ public long metaPageId() {
+ return metaPageId;
}
/**
- * Returns page ID of the index tree meta page.
+ * Returns row ID uuid for which the index needs to be built, {@code null} means that the index building has completed.
*/
- public long metaPageId() {
- return metaPageId;
+ public @Nullable UUID nextRowIdUuidToBuild() {
+ return nextRowIdUuidToBuild;
}
@Override
public String toString() {
- return S.toString(IndexMeta.class, this, "metaPageId", IgniteUtils.hexLong(metaPageId));
+ return S.toString(IndexMeta.class, this, "indexId=", indexId(), "metaPageId", IgniteUtils.hexLong(metaPageId));
}
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMeta.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaKey.java
similarity index 55%
copy from modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMeta.java
copy to modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaKey.java
index 36e3015b4e..90d55abd89 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMeta.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaKey.java
@@ -18,46 +18,26 @@
package org.apache.ignite.internal.storage.pagememory.index.meta;
import java.util.UUID;
-import org.apache.ignite.internal.tostring.IgniteToStringExclude;
-import org.apache.ignite.internal.tostring.S;
-import org.apache.ignite.internal.util.IgniteUtils;
/**
- * Index tree meta information.
+ * Index meta key.
*/
-public class IndexMeta {
- private final UUID id;
-
- @IgniteToStringExclude
- private final long metaPageId;
+public class IndexMetaKey {
+ private final UUID indexId;
/**
* Constructor.
*
- * @param id Index ID.
- * @param metaPageId Index tree meta page ID.
+ * @param indexId Index ID.
*/
- public IndexMeta(UUID id, long metaPageId) {
- this.id = id;
- this.metaPageId = metaPageId;
+ public IndexMetaKey(UUID indexId) {
+ this.indexId = indexId;
}
/**
* Returns the index ID.
*/
- public UUID id() {
- return id;
- }
-
- /**
- * Returns page ID of the index tree meta page.
- */
- public long metaPageId() {
- return metaPageId;
- }
-
- @Override
- public String toString() {
- return S.toString(IndexMeta.class, this, "metaPageId", IgniteUtils.hexLong(metaPageId));
+ public UUID indexId() {
+ return indexId;
}
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java
index 10bf1c6f36..e2897a1028 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/IndexMetaTree.java
@@ -33,7 +33,7 @@ import org.jetbrains.annotations.Nullable;
/**
* {@link BplusTree} implementation for storing {@link IndexMeta}.
*/
-public class IndexMetaTree extends BplusTree<IndexMeta, IndexMeta> {
+public class IndexMetaTree extends BplusTree<IndexMetaKey, IndexMeta> {
/**
* Constructor.
*
@@ -67,14 +67,14 @@ public class IndexMetaTree extends BplusTree<IndexMeta, IndexMeta> {
}
@Override
- protected int compare(BplusIo<IndexMeta> io, long pageAddr, int idx, IndexMeta row) {
+ protected int compare(BplusIo<IndexMetaKey> io, long pageAddr, int idx, IndexMetaKey row) {
IndexMetaIo indexMetaIo = (IndexMetaIo) io;
return indexMetaIo.compare(pageAddr, idx, row);
}
@Override
- public IndexMeta getRow(BplusIo<IndexMeta> io, long pageAddr, int idx, Object x) {
+ public IndexMeta getRow(BplusIo<IndexMetaKey> io, long pageAddr, int idx, @Nullable Object x) {
IndexMetaIo indexMetaIo = (IndexMetaIo) io;
return indexMetaIo.getRow(pageAddr, idx);
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/UpdateLastRowIdUuidToBuiltInvokeClosure.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/UpdateLastRowIdUuidToBuiltInvokeClosure.java
new file mode 100644
index 0000000000..a56bb7f6ab
--- /dev/null
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/UpdateLastRowIdUuidToBuiltInvokeClosure.java
@@ -0,0 +1,60 @@
+/*
+ * 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.pagememory.index.meta;
+
+import java.util.UUID;
+import org.apache.ignite.internal.pagememory.tree.IgniteTree.InvokeClosure;
+import org.apache.ignite.internal.pagememory.tree.IgniteTree.OperationType;
+import org.apache.ignite.lang.IgniteInternalCheckedException;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Closure for update {@link IndexMeta#nextRowIdUuidToBuild()}.
+ */
+public class UpdateLastRowIdUuidToBuiltInvokeClosure implements InvokeClosure<IndexMeta> {
+ private final @Nullable UUID newLastRowIdUuidToBuilt;
+
+ private IndexMeta newRow;
+
+ /**
+ * Constructor.
+ *
+ * @param newLastRowIdUuidToBuilt Row ID uuid for which the index needs to be built, {@code null} means that the index building has
+ * completed.
+ */
+ public UpdateLastRowIdUuidToBuiltInvokeClosure(@Nullable UUID newLastRowIdUuidToBuilt) {
+ this.newLastRowIdUuidToBuilt = newLastRowIdUuidToBuilt;
+ }
+
+ @Override
+ public void call(@Nullable IndexMeta oldRow) throws IgniteInternalCheckedException {
+ assert oldRow != null;
+
+ newRow = new IndexMeta(oldRow.indexId(), oldRow.metaPageId(), newLastRowIdUuidToBuilt);
+ }
+
+ @Override
+ public IndexMeta newRow() {
+ return newRow;
+ }
+
+ @Override
+ public OperationType operationType() {
+ return OperationType.PUT;
+ }
+}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaInnerIo.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaInnerIo.java
index f4c46a8202..5d45d99523 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaInnerIo.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaInnerIo.java
@@ -23,12 +23,13 @@ import org.apache.ignite.internal.pagememory.tree.io.BplusInnerIo;
import org.apache.ignite.internal.pagememory.tree.io.BplusIo;
import org.apache.ignite.internal.storage.pagememory.index.IndexPageTypes;
import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaKey;
import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree;
/**
* IO routines for {@link IndexMetaTree} inner pages.
*/
-public class IndexMetaInnerIo extends BplusInnerIo<IndexMeta> implements IndexMetaIo {
+public class IndexMetaInnerIo extends BplusInnerIo<IndexMetaKey> implements IndexMetaIo {
/** I/O versions. */
public static final IoVersions<IndexMetaInnerIo> VERSIONS = new IoVersions<>(new IndexMetaInnerIo(1));
@@ -42,17 +43,17 @@ public class IndexMetaInnerIo extends BplusInnerIo<IndexMeta> implements IndexMe
}
@Override
- public void store(long dstPageAddr, int dstIdx, BplusIo<IndexMeta> srcIo, long srcPageAddr, int srcIdx) {
+ public void store(long dstPageAddr, int dstIdx, BplusIo<IndexMetaKey> srcIo, long srcPageAddr, int srcIdx) {
IndexMetaIo.super.store(dstPageAddr, dstIdx, srcPageAddr, srcIdx);
}
@Override
- public void storeByOffset(long pageAddr, int off, IndexMeta row) {
+ public void storeByOffset(long pageAddr, int off, IndexMetaKey row) {
IndexMetaIo.super.storeByOffset(pageAddr, off, row);
}
@Override
- public IndexMeta getLookupRow(BplusTree<IndexMeta, ?> tree, long pageAddr, int idx) {
+ public IndexMeta getLookupRow(BplusTree<IndexMetaKey, ?> tree, long pageAddr, int idx) {
return getRow(pageAddr, idx);
}
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaIo.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaIo.java
index 19130dc07d..5dca670ba3 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaIo.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaIo.java
@@ -24,6 +24,7 @@ import java.util.UUID;
import org.apache.ignite.internal.pagememory.tree.io.BplusIo;
import org.apache.ignite.internal.pagememory.util.PageUtils;
import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaKey;
/**
* Interface for {@link IndexMeta} B+Tree-related IO.
@@ -31,7 +32,8 @@ import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
* <p>Defines a following data layout:
* <ul>
* <li>Index ID - {@link UUID} (16 bytes);</li>
- * <li>Index root page ID - long (8 bytes).</li>
+ * <li>Index root page ID - long (8 bytes);</li>
+ * <li>Row ID uuid for which the index needs to be built - {@link UUID} (16 bytes).</li>
* </ul>
*/
public interface IndexMetaIo {
@@ -44,10 +46,22 @@ public interface IndexMetaIo {
/** Index tree meta page id offset - long (8 bytes). */
int INDEX_TREE_META_PAGE_ID_OFFSET = INDEX_ID_LSB_OFFSET + Long.BYTES;
+ /**
+ * Offset of the {@link UUID#getMostSignificantBits() most significant bits} of Row ID uuid for which the index needs to be built (8
+ * bytes).
+ */
+ int NEXT_ROW_ID_TO_BUILT_MSB_OFFSET = INDEX_TREE_META_PAGE_ID_OFFSET + Long.BYTES;
+
+ /**
+ * Offset of the {@link UUID#getLeastSignificantBits() least significant bits} of Row ID uuid for which the index needs to be built (8
+ * bytes).
+ */
+ int NEXT_ROW_ID_TO_BUILT_LSB_OFFSET = NEXT_ROW_ID_TO_BUILT_MSB_OFFSET + Long.BYTES;
+
/** Payload size in bytes. */
int SIZE_IN_BYTES = 2 * Long.BYTES /* Index ID - {@link UUID} (16 bytes) */
+ Long.BYTES /* Index root page ID - long (8 bytes) */
- + Short.BYTES /* Inline size in bytes - short (2 bytes) */;
+ + 2 * Long.BYTES /* Row ID uuid for which the index needs to be built - {@link UUID} (16 bytes) */;
/**
* Returns an offset of the element inside the page.
@@ -64,16 +78,16 @@ public interface IndexMetaIo {
* @param indexMeta Lookup index meta.
* @return Comparison result.
*/
- default int compare(long pageAddr, int idx, IndexMeta indexMeta) {
+ default int compare(long pageAddr, int idx, IndexMetaKey indexMeta) {
int elementOffset = offset(idx);
- int cmp = Long.compare(getLong(pageAddr, elementOffset + INDEX_ID_MSB_OFFSET), indexMeta.id().getMostSignificantBits());
+ int cmp = Long.compare(getLong(pageAddr, elementOffset + INDEX_ID_MSB_OFFSET), indexMeta.indexId().getMostSignificantBits());
if (cmp != 0) {
return cmp;
}
- return Long.compare(getLong(pageAddr, elementOffset + INDEX_ID_LSB_OFFSET), indexMeta.id().getLeastSignificantBits());
+ return Long.compare(getLong(pageAddr, elementOffset + INDEX_ID_LSB_OFFSET), indexMeta.indexId().getLeastSignificantBits());
}
/**
@@ -90,7 +104,14 @@ public interface IndexMetaIo {
long indexTreeMetaPageId = getLong(pageAddr, elementOffset + INDEX_TREE_META_PAGE_ID_OFFSET);
- return new IndexMeta(new UUID(indexIdMsb, indexIdLsb), indexTreeMetaPageId);
+ long nextRowIdUuidToBuiltMsb = getLong(pageAddr, elementOffset + NEXT_ROW_ID_TO_BUILT_MSB_OFFSET);
+ long nextRowIdUuidToBuiltLsb = getLong(pageAddr, elementOffset + NEXT_ROW_ID_TO_BUILT_LSB_OFFSET);
+
+ UUID nextRowIdUuid = nextRowIdUuidToBuiltMsb == 0L && nextRowIdUuidToBuiltLsb == 0L
+ ? null
+ : new UUID(nextRowIdUuidToBuiltMsb, nextRowIdUuidToBuiltLsb);
+
+ return new IndexMeta(new UUID(indexIdMsb, indexIdLsb), indexTreeMetaPageId, nextRowIdUuid);
}
/**
@@ -110,10 +131,22 @@ public interface IndexMetaIo {
*
* @see BplusIo#storeByOffset(long, int, Object)
*/
- default void storeByOffset(long pageAddr, int off, IndexMeta row) {
- putLong(pageAddr, off + INDEX_ID_MSB_OFFSET, row.id().getMostSignificantBits());
- putLong(pageAddr, off + INDEX_ID_LSB_OFFSET, row.id().getLeastSignificantBits());
+ default void storeByOffset(long pageAddr, int off, IndexMetaKey rowKey) {
+ assert rowKey instanceof IndexMeta : rowKey;
+
+ IndexMeta row = (IndexMeta) rowKey;
+
+ putLong(pageAddr, off + INDEX_ID_MSB_OFFSET, row.indexId().getMostSignificantBits());
+ putLong(pageAddr, off + INDEX_ID_LSB_OFFSET, row.indexId().getLeastSignificantBits());
putLong(pageAddr, off + INDEX_TREE_META_PAGE_ID_OFFSET, row.metaPageId());
+
+ UUID nextRowIdUuidToBuild = row.nextRowIdUuidToBuild();
+
+ long nextRowIdUuidToBuildMsb = nextRowIdUuidToBuild == null ? 0L : nextRowIdUuidToBuild.getMostSignificantBits();
+ long nextRowIdUuidToBuildLsb = nextRowIdUuidToBuild == null ? 0L : nextRowIdUuidToBuild.getLeastSignificantBits();
+
+ putLong(pageAddr, off + NEXT_ROW_ID_TO_BUILT_MSB_OFFSET, nextRowIdUuidToBuildMsb);
+ putLong(pageAddr, off + NEXT_ROW_ID_TO_BUILT_LSB_OFFSET, nextRowIdUuidToBuildLsb);
}
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaLeafIo.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaLeafIo.java
index f08902d342..72bbbe5844 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaLeafIo.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/meta/io/IndexMetaLeafIo.java
@@ -23,12 +23,13 @@ import org.apache.ignite.internal.pagememory.tree.io.BplusIo;
import org.apache.ignite.internal.pagememory.tree.io.BplusLeafIo;
import org.apache.ignite.internal.storage.pagememory.index.IndexPageTypes;
import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaKey;
import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree;
/**
* IO routines for {@link IndexMetaTree} leaf pages.
*/
-public class IndexMetaLeafIo extends BplusLeafIo<IndexMeta> implements IndexMetaIo {
+public class IndexMetaLeafIo extends BplusLeafIo<IndexMetaKey> implements IndexMetaIo {
/** I/O versions. */
public static final IoVersions<IndexMetaLeafIo> VERSIONS = new IoVersions<>(new IndexMetaLeafIo(1));
@@ -42,17 +43,17 @@ public class IndexMetaLeafIo extends BplusLeafIo<IndexMeta> implements IndexMeta
}
@Override
- public void store(long dstPageAddr, int dstIdx, BplusIo<IndexMeta> srcIo, long srcPageAddr, int srcIdx) {
+ public void store(long dstPageAddr, int dstIdx, BplusIo<IndexMetaKey> srcIo, long srcPageAddr, int srcIdx) {
IndexMetaIo.super.store(dstPageAddr, dstIdx, srcPageAddr, srcIdx);
}
@Override
- public void storeByOffset(long pageAddr, int off, IndexMeta row) {
+ public void storeByOffset(long pageAddr, int off, IndexMetaKey row) {
IndexMetaIo.super.storeByOffset(pageAddr, off, row);
}
@Override
- public IndexMeta getLookupRow(BplusTree<IndexMeta, ?> tree, long pageAddr, int idx) {
+ public IndexMeta getLookupRow(BplusTree<IndexMetaKey, ?> tree, long pageAddr, int idx) {
return getRow(pageAddr, idx);
}
}
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 111a1f6f78..801533b1e2 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
@@ -17,16 +17,12 @@
package org.apache.ignite.internal.storage.pagememory.index.sorted;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageState;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageNotInCleanupOrRebalancedState;
import java.nio.ByteBuffer;
import java.util.NoSuchElementException;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
-import java.util.function.Supplier;
import org.apache.ignite.internal.binarytuple.BinaryTupleCommon;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -37,68 +33,52 @@ import org.apache.ignite.internal.schema.BinaryTuplePrefix;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageClosedException;
import org.apache.ignite.internal.storage.StorageException;
-import org.apache.ignite.internal.storage.StorageRebalanceException;
import org.apache.ignite.internal.storage.index.IndexRow;
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.storage.pagememory.index.AbstractPageMemoryIndexStorage;
import org.apache.ignite.internal.storage.pagememory.index.freelist.IndexColumns;
import org.apache.ignite.internal.storage.pagememory.index.freelist.IndexColumnsFreeList;
-import org.apache.ignite.internal.storage.util.StorageState;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
+import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree;
import org.apache.ignite.internal.util.Cursor;
-import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.lang.IgniteInternalCheckedException;
-import org.apache.ignite.lang.IgniteStringFormatter;
import org.jetbrains.annotations.Nullable;
/**
* Implementation of Sorted index storage using Page Memory.
*/
-public class PageMemorySortedIndexStorage implements SortedIndexStorage {
+public class PageMemorySortedIndexStorage extends AbstractPageMemoryIndexStorage implements SortedIndexStorage {
private static final IgniteLogger LOG = Loggers.forClass(PageMemorySortedIndexStorage.class);
/** Index descriptor. */
private final SortedIndexDescriptor descriptor;
- /** Free list to store index columns. */
- private volatile IndexColumnsFreeList freeList;
-
/** Sorted index tree instance. */
private volatile SortedIndexTree sortedIndexTree;
- /** Partition id. */
- private final int partitionId;
-
- /** Lowest possible RowId according to signed long ordering. */
- private final RowId lowestRowId;
-
- /** Highest possible RowId according to signed long ordering. */
- private final RowId highestRowId;
-
- /** Busy lock. */
- private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
-
- /** Current state of the storage. */
- private final AtomicReference<StorageState> state = new AtomicReference<>(StorageState.RUNNABLE);
-
/**
* Constructor.
*
+ * @param indexMeta Index meta.
* @param descriptor Sorted index descriptor.
* @param freeList Free list to store index columns.
* @param sortedIndexTree Sorted index tree instance.
+ * @param indexMetaTree Index meta tree instance.
*/
- public PageMemorySortedIndexStorage(SortedIndexDescriptor descriptor, IndexColumnsFreeList freeList, SortedIndexTree sortedIndexTree) {
+ public PageMemorySortedIndexStorage(
+ IndexMeta indexMeta,
+ SortedIndexDescriptor descriptor,
+ IndexColumnsFreeList freeList,
+ SortedIndexTree sortedIndexTree,
+ IndexMetaTree indexMetaTree
+ ) {
+ super(indexMeta, sortedIndexTree.partitionId(), freeList, indexMetaTree);
+
this.descriptor = descriptor;
- this.freeList = freeList;
this.sortedIndexTree = sortedIndexTree;
-
- partitionId = sortedIndexTree.partitionId();
-
- lowestRowId = RowId.lowestRowId(partitionId);
-
- highestRowId = RowId.highestRowId(partitionId);
}
@Override
@@ -205,20 +185,8 @@ public class PageMemorySortedIndexStorage implements SortedIndexStorage {
);
}
- /**
- * Closes the sorted index storage.
- */
- public void close() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
- StorageState state = this.state.get();
-
- assert state == StorageState.CLOSED : state;
-
- return;
- }
-
- busyLock.block();
-
+ @Override
+ public void closeStructures() {
sortedIndexTree.close();
}
@@ -378,49 +346,6 @@ public class PageMemorySortedIndexStorage implements SortedIndexStorage {
}
}
- private <V> V busy(Supplier<V> supplier) {
- if (!busyLock.enterBusy()) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- try {
- return supplier.get();
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- /**
- * Prepares storage for rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when starting the rebalance.
- */
- public void startRebalance() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.REBALANCE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- try {
- this.sortedIndexTree.close();
- } finally {
- busyLock.unblock();
- }
- }
-
- /**
- * Completes the rebalancing of the storage.
- *
- * @throws StorageRebalanceException If there is an error while completing the storage rebalance.
- */
- public void completeRebalance() {
- if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
- }
-
/**
* Updates the internal data structures of the storage on rebalance or cleanup.
*
@@ -435,10 +360,6 @@ public class PageMemorySortedIndexStorage implements SortedIndexStorage {
this.sortedIndexTree = sortedIndexTree;
}
- private String createStorageInfo() {
- return IgniteStringFormatter.format("indexId={}, partitionId={}", descriptor.id(), partitionId);
- }
-
/**
* Starts destruction of the data stored by this index partition.
*
@@ -470,31 +391,4 @@ public class PageMemorySortedIndexStorage implements SortedIndexStorage {
indexRow.indexColumns().link(PageIdUtils.NULL_LINK);
}
}
-
- /**
- * Prepares the storage for cleanup.
- *
- * <p>After cleanup (successful or not), method {@link #finishCleanup()} must be called.
- */
- public void startCleanup() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLEANUP)) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- try {
- sortedIndexTree.close();
- } finally {
- busyLock.unblock();
- }
- }
-
- /**
- * Finishes cleanup up the storage.
- */
- public void finishCleanup() {
- state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE);
- }
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
index ca6b4b7160..3a801f0210 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
@@ -180,7 +180,7 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
NamedListView<TableIndexView> indexesCfgView = tableStorage.tablesConfiguration().indexes().value();
for (IndexMeta indexMeta : cursor) {
- TableIndexView indexCfgView = indexesCfgView.get(indexMeta.id());
+ TableIndexView indexCfgView = indexesCfgView.get(indexMeta.indexId());
if (indexCfgView instanceof HashIndexView) {
hashIndexes.put(indexCfgView.id(), createOrRestoreHashIndex(indexMeta));
@@ -195,10 +195,7 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
return null;
} catch (Exception e) {
- throw new StorageException(
- IgniteStringFormatter.format("Failed to process SQL indexes during partition start: [{}]", createStorageInfo()),
- e
- );
+ throw new StorageException("Failed to process SQL indexes during partition start: [{}]", e, createStorageInfo());
}
});
}
@@ -216,7 +213,7 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
* @param indexId Index UUID.
*/
public PageMemoryHashIndexStorage getOrCreateHashIndex(UUID indexId) {
- return busy(() -> hashIndexes.computeIfAbsent(indexId, uuid -> createOrRestoreHashIndex(new IndexMeta(indexId, 0L))));
+ return busy(() -> hashIndexes.computeIfAbsent(indexId, uuid -> createOrRestoreHashIndex(createIndexMetaForNewIndex(uuid))));
}
/**
@@ -225,17 +222,17 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
* @param indexId Index UUID.
*/
public PageMemorySortedIndexStorage getOrCreateSortedIndex(UUID indexId) {
- return busy(() -> sortedIndexes.computeIfAbsent(indexId, uuid -> createOrRestoreSortedIndex(new IndexMeta(indexId, 0L))));
+ return busy(() -> sortedIndexes.computeIfAbsent(indexId, uuid -> createOrRestoreSortedIndex(createIndexMetaForNewIndex(uuid))));
}
private PageMemoryHashIndexStorage createOrRestoreHashIndex(IndexMeta indexMeta) {
throwExceptionIfStorageNotInRunnableState();
- var indexDescriptor = new HashIndexDescriptor(indexMeta.id(), tableStorage.tablesConfiguration().value());
+ var indexDescriptor = new HashIndexDescriptor(indexMeta.indexId(), tableStorage.tablesConfiguration().value());
HashIndexTree hashIndexTree = createHashIndexTree(indexDescriptor, indexMeta);
- return new PageMemoryHashIndexStorage(indexDescriptor, indexFreeList, hashIndexTree);
+ return new PageMemoryHashIndexStorage(indexMeta, indexDescriptor, indexFreeList, hashIndexTree, indexMetaTree);
}
HashIndexTree createHashIndexTree(HashIndexDescriptor indexDescriptor, IndexMeta indexMeta) {
@@ -262,32 +259,25 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
);
if (initNew) {
- boolean replaced = indexMetaTree.putx(new IndexMeta(indexMeta.id(), metaPageId));
+ boolean replaced = indexMetaTree.putx(new IndexMeta(indexMeta.indexId(), metaPageId, indexMeta.nextRowIdUuidToBuild()));
assert !replaced;
}
return hashIndexTree;
} catch (IgniteInternalCheckedException e) {
- throw new StorageException(
- IgniteStringFormatter.format(
- "Error creating hash index tree: [{}, indexId={}]",
- createStorageInfo(),
- indexMeta.id()
- ),
- e
- );
+ throw new StorageException("Error creating hash index tree: [{}, indexId={}]", e, createStorageInfo(), indexMeta.indexId());
}
}
private PageMemorySortedIndexStorage createOrRestoreSortedIndex(IndexMeta indexMeta) {
throwExceptionIfStorageNotInRunnableState();
- var indexDescriptor = new SortedIndexDescriptor(indexMeta.id(), tableStorage.tablesConfiguration().value());
+ var indexDescriptor = new SortedIndexDescriptor(indexMeta.indexId(), tableStorage.tablesConfiguration().value());
SortedIndexTree sortedIndexTree = createSortedIndexTree(indexDescriptor, indexMeta);
- return new PageMemorySortedIndexStorage(indexDescriptor, indexFreeList, sortedIndexTree);
+ return new PageMemorySortedIndexStorage(indexMeta, indexDescriptor, indexFreeList, sortedIndexTree, indexMetaTree);
}
SortedIndexTree createSortedIndexTree(SortedIndexDescriptor indexDescriptor, IndexMeta indexMeta) {
@@ -314,22 +304,14 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
);
if (initNew) {
- boolean replaced = indexMetaTree.putx(new IndexMeta(indexMeta.id(), metaPageId));
+ boolean replaced = indexMetaTree.putx(new IndexMeta(indexMeta.indexId(), metaPageId, indexMeta.nextRowIdUuidToBuild()));
assert !replaced;
}
return sortedIndexTree;
} catch (IgniteInternalCheckedException e) {
- throw new StorageException(
- IgniteStringFormatter.format(
- "Error creating sorted index tree: [table={}, partitionId={}, indexId={}]",
- tableStorage.getTableName(),
- partitionId,
- indexMeta.id()
- ),
- e
- );
+ throw new StorageException("Error creating sorted index tree: [{}, indexId={}]", e, createStorageInfo(), indexMeta.indexId());
}
}
@@ -1001,4 +983,8 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
return removeWriteOnGc.getResult();
}
+
+ IndexMeta createIndexMetaForNewIndex(UUID indexId) {
+ return new IndexMeta(indexId, 0L, RowId.lowestRowId(partitionId).uuid());
+ }
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java
index 181a0b0a70..756ade63e6 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java
@@ -44,7 +44,6 @@ import org.apache.ignite.internal.storage.pagememory.configuration.schema.Persis
import org.apache.ignite.internal.storage.pagememory.index.freelist.IndexColumns;
import org.apache.ignite.internal.storage.pagememory.index.freelist.IndexColumnsFreeList;
import org.apache.ignite.internal.storage.pagememory.index.hash.PageMemoryHashIndexStorage;
-import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree;
import org.apache.ignite.internal.storage.pagememory.index.sorted.PageMemorySortedIndexStorage;
import org.apache.ignite.internal.storage.pagememory.mv.gc.GcQueue;
@@ -388,14 +387,14 @@ public class PersistentPageMemoryMvPartitionStorage extends AbstractPageMemoryMv
for (PageMemoryHashIndexStorage indexStorage : hashIndexes.values()) {
indexStorage.updateDataStructures(
indexFreeList,
- createHashIndexTree(indexStorage.indexDescriptor(), new IndexMeta(indexStorage.indexDescriptor().id(), 0L))
+ createHashIndexTree(indexStorage.indexDescriptor(), createIndexMetaForNewIndex(indexStorage.indexDescriptor().id()))
);
}
for (PageMemorySortedIndexStorage indexStorage : sortedIndexes.values()) {
indexStorage.updateDataStructures(
indexFreeList,
- createSortedIndexTree(indexStorage.indexDescriptor(), new IndexMeta(indexStorage.indexDescriptor().id(), 0L))
+ createSortedIndexTree(indexStorage.indexDescriptor(), createIndexMetaForNewIndex(indexStorage.indexDescriptor().id()))
);
}
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java
index 318e1afffb..8d6c6b8212 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/VolatilePageMemoryMvPartitionStorage.java
@@ -36,7 +36,6 @@ import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.StorageException;
import org.apache.ignite.internal.storage.pagememory.VolatilePageMemoryTableStorage;
import org.apache.ignite.internal.storage.pagememory.index.hash.PageMemoryHashIndexStorage;
-import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMeta;
import org.apache.ignite.internal.storage.pagememory.index.meta.IndexMetaTree;
import org.apache.ignite.internal.storage.pagememory.index.sorted.PageMemorySortedIndexStorage;
import org.apache.ignite.internal.storage.pagememory.mv.gc.GcQueue;
@@ -326,14 +325,14 @@ public class VolatilePageMemoryMvPartitionStorage extends AbstractPageMemoryMvPa
for (PageMemoryHashIndexStorage indexStorage : hashIndexes.values()) {
indexStorage.updateDataStructures(
indexFreeList,
- createHashIndexTree(indexStorage.indexDescriptor(), new IndexMeta(indexStorage.indexDescriptor().id(), 0L))
+ createHashIndexTree(indexStorage.indexDescriptor(), createIndexMetaForNewIndex(indexStorage.indexDescriptor().id()))
);
}
for (PageMemorySortedIndexStorage indexStorage : sortedIndexes.values()) {
indexStorage.updateDataStructures(
indexFreeList,
- createSortedIndexTree(indexStorage.indexDescriptor(), new IndexMeta(indexStorage.indexDescriptor().id(), 0L))
+ createSortedIndexTree(indexStorage.indexDescriptor(), createIndexMetaForNewIndex(indexStorage.indexDescriptor().id()))
);
}
}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java
index 326375519a..de50ca07ab 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java
@@ -20,15 +20,15 @@ package org.apache.ignite.internal.storage.rocksdb;
import static java.lang.ThreadLocal.withInitial;
import static java.nio.ByteBuffer.allocateDirect;
import static org.apache.ignite.internal.hlc.HybridTimestamp.HYBRID_TIMESTAMP_SIZE;
-import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.KEY_BYTE_ORDER;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.MAX_KEY_SIZE;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.MV_KEY_BUFFER;
-import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.PARTITION_ID_SIZE;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.ROW_ID_OFFSET;
-import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.ROW_ID_SIZE;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.TABLE_ROW_BYTE_ORDER;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.readTimestampNatural;
import static org.apache.ignite.internal.storage.rocksdb.RocksDbMvPartitionStorage.invalid;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.KEY_BYTE_ORDER;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.PARTITION_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.ROW_ID_SIZE;
import java.nio.ByteBuffer;
import org.apache.ignite.internal.hlc.HybridTimestamp;
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/HashIndex.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/HashIndex.java
index 9697178356..4b9ad3d50a 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/HashIndex.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/HashIndex.java
@@ -39,9 +39,12 @@ class HashIndex {
private final ConcurrentMap<Integer, RocksDbHashIndexStorage> storages = new ConcurrentHashMap<>();
- HashIndex(ColumnFamily indexCf, HashIndexDescriptor descriptor) {
+ private final RocksDbMetaStorage indexMetaStorage;
+
+ HashIndex(ColumnFamily indexCf, HashIndexDescriptor descriptor, RocksDbMetaStorage indexMetaStorage) {
this.indexCf = indexCf;
this.descriptor = descriptor;
+ this.indexMetaStorage = indexMetaStorage;
}
/**
@@ -50,7 +53,7 @@ class HashIndex {
HashIndexStorage getOrCreateStorage(RocksDbMvPartitionStorage partitionStorage) {
return storages.computeIfAbsent(
partitionStorage.partitionId(),
- partId -> new RocksDbHashIndexStorage(descriptor, indexCf, partitionStorage)
+ partId -> new RocksDbHashIndexStorage(descriptor, indexCf, partitionStorage, indexMetaStorage)
);
}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/PartitionDataHelper.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/PartitionDataHelper.java
index 9551f5e186..970c662ff5 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/PartitionDataHelper.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/PartitionDataHelper.java
@@ -20,6 +20,11 @@ package org.apache.ignite.internal.storage.rocksdb;
import static java.lang.ThreadLocal.withInitial;
import static java.nio.ByteBuffer.allocateDirect;
import static org.apache.ignite.internal.hlc.HybridTimestamp.HYBRID_TIMESTAMP_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.KEY_BYTE_ORDER;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.PARTITION_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.ROW_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.getRowIdUuid;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.putRowIdUuid;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
@@ -35,12 +40,6 @@ import org.rocksdb.Slice;
/** Helper for the partition data. */
class PartitionDataHelper implements ManuallyCloseable {
- /** Commit partition id size. */
- static final int PARTITION_ID_SIZE = Short.BYTES;
-
- /** UUID size in bytes. */
- static final int ROW_ID_SIZE = 2 * Long.BYTES;
-
/** Position of row id inside the key. */
static final int ROW_ID_OFFSET = Short.BYTES;
@@ -73,8 +72,6 @@ class PartitionDataHelper implements ManuallyCloseable {
static final ByteOrder TABLE_ROW_BYTE_ORDER = ByteBufferRow.ORDER;
- static final ByteOrder KEY_BYTE_ORDER = ByteOrder.BIG_ENDIAN;
-
/** Thread-local direct buffer instance to read keys from RocksDB. */
static final ThreadLocal<ByteBuffer> MV_KEY_BUFFER = withInitial(() -> allocateDirect(MAX_KEY_SIZE).order(KEY_BYTE_ORDER));
@@ -119,44 +116,35 @@ class PartitionDataHelper implements ManuallyCloseable {
}
void putDataKey(ByteBuffer dataKeyBuffer, RowId rowId, HybridTimestamp timestamp) {
+ assert rowId.partitionId() == partitionId : "rowPartitionId=" + rowId.partitionId() + ", storagePartitionId=" + partitionId;
+
dataKeyBuffer.putShort((short) partitionId);
- putRowId(dataKeyBuffer, rowId);
+ putRowIdUuid(dataKeyBuffer, rowId.uuid());
putTimestampDesc(dataKeyBuffer, timestamp);
dataKeyBuffer.flip();
}
void putGcKey(ByteBuffer gcKeyBuffer, RowId rowId, HybridTimestamp timestamp) {
+ assert rowId.partitionId() == partitionId : "rowPartitionId=" + rowId.partitionId() + ", storagePartitionId=" + partitionId;
+
gcKeyBuffer.putShort((short) partitionId);
putTimestampNatural(gcKeyBuffer, timestamp);
- putRowId(gcKeyBuffer, rowId);
+ putRowIdUuid(gcKeyBuffer, rowId.uuid());
gcKeyBuffer.flip();
}
void putRowId(ByteBuffer keyBuffer, RowId rowId) {
- assert rowId.partitionId() == partitionId : rowId;
- assert keyBuffer.order() == KEY_BYTE_ORDER;
+ assert rowId.partitionId() == partitionId : "rowPartitionId=" + rowId.partitionId() + ", storagePartitionId=" + partitionId;
- keyBuffer.putLong(normalize(rowId.mostSignificantBits()));
- keyBuffer.putLong(normalize(rowId.leastSignificantBits()));
+ putRowIdUuid(keyBuffer, rowId.uuid());
}
RowId getRowId(ByteBuffer keyBuffer, int offset) {
assert partitionId == (keyBuffer.getShort(0) & 0xFFFF);
- return new RowId(partitionId, normalize(keyBuffer.getLong(offset)), normalize(keyBuffer.getLong(offset + Long.BYTES)));
- }
-
- /**
- * Converts signed long into a new long value, that when written in Big Endian, will preserve the comparison order if compared
- * lexicographically as an array of unsigned bytes. For example, values {@code -1} and {@code 0}, when written in BE, will become
- * {@code 0xFF..F} and {@code 0x00..0}, and lose their ascending order.
- *
- * <p>Flipping the sign bit will change the situation: {@code -1 -> 0x7F..F} and {@code 0 -> 0x80..0}.
- */
- static long normalize(long value) {
- return value ^ (1L << 63);
+ return new RowId(partitionId, getRowIdUuid(keyBuffer, offset));
}
/**
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java
index e108d52e25..32cb978b64 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java
@@ -17,15 +17,24 @@
package org.apache.ignite.internal.storage.rocksdb;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.KEY_BYTE_ORDER;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.PARTITION_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.ROW_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.getRowIdUuid;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.putIndexId;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.putRowIdUuid;
import static org.apache.ignite.internal.util.ArrayUtils.BYTE_EMPTY_ARRAY;
import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-import java.nio.charset.StandardCharsets;
+import java.util.UUID;
import java.util.stream.Stream;
import org.apache.ignite.internal.rocksdb.ColumnFamily;
import org.apache.ignite.internal.rocksdb.RocksUtils;
+import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageException;
+import org.jetbrains.annotations.Nullable;
+import org.rocksdb.AbstractWriteBatch;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
@@ -35,15 +44,17 @@ import org.rocksdb.Slice;
* Wrapper around the "meta" Column Family inside a RocksDB-based storage, which stores some auxiliary information needed for internal
* storage logic.
*/
-class RocksDbMetaStorage {
- /**
- * Name of the key that corresponds to a list of existing partition IDs of a storage.
- */
- private static final byte[] PARTITION_ID_PREFIX = "part".getBytes(StandardCharsets.UTF_8);
+public class RocksDbMetaStorage {
+ /** Name of the key that corresponds to a list of existing partition IDs of a storage. */
+ private static final byte[] PARTITION_ID_PREFIX = "part".getBytes(UTF_8);
- /**
- * Name of the key that is out of range of the partition ID key prefix, used as an exclusive bound.
- */
+ /** Index meta key prefix. */
+ private static final byte[] INDEX_META_KEY_PREFIX = "index-meta".getBytes(UTF_8);
+
+ /** Index meta key size in bytes. */
+ private static final int INDEX_META_KEY_SIZE = INDEX_META_KEY_PREFIX.length + PARTITION_ID_SIZE + ROW_ID_SIZE;
+
+ /** Name of the key that is out of range of the partition ID key prefix, used as an exclusive bound. */
private static final byte[] PARTITION_ID_PREFIX_END = RocksUtils.incrementPrefix(PARTITION_ID_PREFIX);
private final ColumnFamily metaColumnFamily;
@@ -106,10 +117,83 @@ class RocksDbMetaStorage {
static byte[] partitionIdKey(int partitionId) {
assert partitionId >= 0 && partitionId <= 0xFFFF : partitionId;
- return ByteBuffer.allocate(PARTITION_ID_PREFIX.length + Short.BYTES)
- .order(ByteOrder.BIG_ENDIAN)
+ return ByteBuffer.allocate(PARTITION_ID_PREFIX.length + PARTITION_ID_SIZE)
+ .order(KEY_BYTE_ORDER)
.put(PARTITION_ID_PREFIX)
.putShort((short) partitionId)
.array();
}
+
+ /**
+ * Returns the row ID for which the index needs to be built, {@code null} means that the index building has completed.
+ *
+ * @param indexId Index ID.
+ * @param partitionId Partition ID.
+ * @param ifAbsent Will be returned if next the row ID for which the index needs to be built has never been saved.
+ */
+ public @Nullable RowId getNextRowIdToBuilt(UUID indexId, int partitionId, RowId ifAbsent) {
+ try {
+ byte[] lastBuiltRowIdBytes = metaColumnFamily.get(indexMetaKey(partitionId, indexId));
+
+ if (lastBuiltRowIdBytes == null) {
+ return ifAbsent;
+ }
+
+ if (lastBuiltRowIdBytes.length == 0) {
+ return null;
+ }
+
+ return new RowId(partitionId, getRowIdUuid(ByteBuffer.wrap(lastBuiltRowIdBytes), 0));
+ } catch (RocksDBException e) {
+ throw new StorageException(
+ "Failed to read next row ID to built: [partitionId={}, indexId={}]",
+ e,
+ partitionId, indexId
+ );
+ }
+ }
+
+ /**
+ * Puts row ID for which the index needs to be built, {@code null} means index building is finished.
+ *
+ * @param writeBatch Write batch.
+ * @param partitionId Partition ID.
+ * @param indexId Index ID.
+ * @param rowId Row ID.
+ */
+ public void putNextRowIdToBuilt(AbstractWriteBatch writeBatch, UUID indexId, int partitionId, @Nullable RowId rowId) {
+ try {
+ writeBatch.put(metaColumnFamily.handle(), indexMetaKey(partitionId, indexId), indexLastBuildRowId(rowId));
+ } catch (RocksDBException e) {
+ throw new StorageException(
+ "Failed to save next row ID to built: [partitionId={}, indexId={}, rowId={}]",
+ e,
+ partitionId, indexId, rowId
+ );
+ }
+ }
+
+ private static byte[] indexMetaKey(int partitionId, UUID indexId) {
+ assert partitionId >= 0 && partitionId <= 0xFFFF : partitionId;
+
+ ByteBuffer buffer = ByteBuffer.allocate(INDEX_META_KEY_SIZE).order(KEY_BYTE_ORDER);
+
+ buffer.put(INDEX_META_KEY_PREFIX).putShort((short) partitionId);
+
+ putIndexId(buffer, indexId);
+
+ return buffer.array();
+ }
+
+ private static byte[] indexLastBuildRowId(@Nullable RowId rowId) {
+ if (rowId == null) {
+ return BYTE_EMPTY_ARRAY;
+ }
+
+ ByteBuffer buffer = ByteBuffer.allocate(ROW_ID_SIZE).order(KEY_BYTE_ORDER);
+
+ putRowIdUuid(buffer, rowId.uuid());
+
+ return buffer.array();
+ }
}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java
index 452579db0d..f0193e8823 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java
@@ -21,7 +21,6 @@ import static java.lang.ThreadLocal.withInitial;
import static java.nio.ByteBuffer.allocate;
import static java.util.Arrays.copyOf;
import static java.util.Arrays.copyOfRange;
-import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.KEY_BYTE_ORDER;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.MAX_KEY_SIZE;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.MV_KEY_BUFFER;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.PARTITION_ID_OFFSET;
@@ -32,10 +31,11 @@ import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.TAB
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.TX_ID_OFFSET;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.VALUE_HEADER_SIZE;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.VALUE_OFFSET;
-import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.normalize;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.putTimestampDesc;
import static org.apache.ignite.internal.storage.rocksdb.PartitionDataHelper.readTimestampDesc;
import static org.apache.ignite.internal.storage.rocksdb.RocksDbMetaStorage.partitionIdKey;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.KEY_BYTE_ORDER;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.normalize;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageState;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageUtils.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageUtils.java
new file mode 100644
index 0000000000..309ff84709
--- /dev/null
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageUtils.java
@@ -0,0 +1,69 @@
+/*
+ * 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.rocksdb;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.UUID;
+import org.apache.ignite.internal.storage.RowId;
+
+/**
+ * Helper class for storages based on RocksDB.
+ */
+public class RocksDbStorageUtils {
+ /** Key byte order. */
+ public static final ByteOrder KEY_BYTE_ORDER = ByteOrder.BIG_ENDIAN;
+
+ /** {@link RowId} size in bytes. */
+ public static final int ROW_ID_SIZE = Long.BYTES * 2;
+
+ /** Partition ID size in bytes. */
+ public static final int PARTITION_ID_SIZE = Short.BYTES;
+
+ /** Index ID size in bytes. */
+ public static final int INDEX_ID_SIZE = 2 * Long.BYTES;
+
+ static void putRowIdUuid(ByteBuffer keyBuffer, UUID rowIdUuid) {
+ assert keyBuffer.order() == KEY_BYTE_ORDER;
+
+ keyBuffer.putLong(normalize(rowIdUuid.getMostSignificantBits()));
+ keyBuffer.putLong(normalize(rowIdUuid.getLeastSignificantBits()));
+ }
+
+ static void putIndexId(ByteBuffer keyBuffer, UUID indexId) {
+ assert keyBuffer.order() == KEY_BYTE_ORDER;
+
+ keyBuffer.putLong(normalize(indexId.getMostSignificantBits()));
+ keyBuffer.putLong(normalize(indexId.getLeastSignificantBits()));
+ }
+
+ static UUID getRowIdUuid(ByteBuffer keyBuffer, int offset) {
+ return new UUID(normalize(keyBuffer.getLong(offset)), normalize(keyBuffer.getLong(offset + Long.BYTES)));
+ }
+
+ /**
+ * Converts signed long into a new long value, that when written in Big Endian, will preserve the comparison order if compared
+ * lexicographically as an array of unsigned bytes. For example, values {@code -1} and {@code 0}, when written in BE, will become
+ * {@code 0xFF..F} and {@code 0x00..0}, and lose their ascending order.
+ *
+ * <p>Flipping the sign bit will change the situation: {@code -1 -> 0x7F..F} and {@code 0 -> 0x80..0}.
+ */
+ static long normalize(long value) {
+ return value ^ (1L << 63);
+ }
+}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java
index 745bd6dbfb..23159f6959 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java
@@ -35,7 +35,10 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -234,6 +237,8 @@ public class RocksDbTableStorage implements MvTableStorage {
try {
db = RocksDB.open(dbOptions, tablePath.toAbsolutePath().toString(), cfDescriptors, cfHandles);
+ Map<UUID, ColumnFamily> sortedIndexColumnFamilyByIndexId = new HashMap<>();
+
// read all existing Column Families from the db and parse them according to type: meta, partition data or index.
for (ColumnFamilyHandle cfHandle : cfHandles) {
ColumnFamily cf = ColumnFamily.wrap(db, cfHandle);
@@ -260,17 +265,12 @@ public class RocksDbTableStorage implements MvTableStorage {
break;
case SORTED_INDEX:
- UUID indexId = sortedIndexId(cf.name());
-
- var indexDescriptor = new SortedIndexDescriptor(indexId, tablesCfg.value());
-
- sortedIndices.put(indexId, new SortedIndex(cf, indexDescriptor));
+ sortedIndexColumnFamilyByIndexId.put(sortedIndexId(cf.name()), cf);
break;
default:
- throw new StorageException("Unidentified column family [name=" + cf.name() + ", table="
- + getTableName() + ']');
+ throw new StorageException("Unidentified column family [name={}, table={}]", cf.name(), getTableName());
}
}
@@ -278,6 +278,14 @@ public class RocksDbTableStorage implements MvTableStorage {
assert partitionCf != null;
assert hashIndexCf != null;
+ for (Entry<UUID, ColumnFamily> entry : sortedIndexColumnFamilyByIndexId.entrySet()) {
+ UUID indexId = entry.getKey();
+
+ var indexDescriptor = new SortedIndexDescriptor(indexId, tablesCfg.value());
+
+ sortedIndices.put(indexId, new SortedIndex(entry.getValue(), indexDescriptor, meta));
+ }
+
flusher.init(db, cfHandles);
} catch (RocksDBException e) {
throw new StorageException("Failed to initialize RocksDB instance", e);
@@ -472,7 +480,7 @@ public class RocksDbTableStorage implements MvTableStorage {
flusher.addColumnFamily(columnFamily.handle());
- return new SortedIndex(columnFamily, indexDescriptor);
+ return new SortedIndex(columnFamily, indexDescriptor, meta);
}
@Override
@@ -481,7 +489,7 @@ public class RocksDbTableStorage implements MvTableStorage {
HashIndex storages = hashIndices.computeIfAbsent(indexId, id -> {
var indexDescriptor = new HashIndexDescriptor(indexId, tablesCfg.value());
- return new HashIndex(hashIndexCf, indexDescriptor);
+ return new HashIndex(hashIndexCf, indexDescriptor, meta);
});
RocksDbMvPartitionStorage partitionStorage = mvPartitionStorages.get(partitionId);
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/SortedIndex.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/SortedIndex.java
index df7d14fc9e..54a221d265 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/SortedIndex.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/SortedIndex.java
@@ -41,9 +41,12 @@ class SortedIndex implements ManuallyCloseable {
private final ConcurrentMap<Integer, RocksDbSortedIndexStorage> storages = new ConcurrentHashMap<>();
- SortedIndex(ColumnFamily indexCf, SortedIndexDescriptor descriptor) {
+ private final RocksDbMetaStorage indexMetaStorage;
+
+ SortedIndex(ColumnFamily indexCf, SortedIndexDescriptor descriptor, RocksDbMetaStorage indexMetaStorage) {
this.descriptor = descriptor;
this.indexCf = indexCf;
+ this.indexMetaStorage = indexMetaStorage;
}
/**
@@ -52,7 +55,7 @@ class SortedIndex implements ManuallyCloseable {
SortedIndexStorage getOrCreateStorage(RocksDbMvPartitionStorage partitionStorage) {
return storages.computeIfAbsent(
partitionStorage.partitionId(),
- partId -> new RocksDbSortedIndexStorage(descriptor, indexCf, partitionStorage)
+ partId -> new RocksDbSortedIndexStorage(descriptor, indexCf, partitionStorage, indexMetaStorage)
);
}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/AbstractRocksDbIndexStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/AbstractRocksDbIndexStorage.java
new file mode 100644
index 0000000000..ef447336c3
--- /dev/null
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/AbstractRocksDbIndexStorage.java
@@ -0,0 +1,205 @@
+/*
+ * 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.rocksdb.index;
+
+import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageState;
+import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
+import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
+
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.StorageRebalanceException;
+import org.apache.ignite.internal.storage.index.IndexStorage;
+import org.apache.ignite.internal.storage.rocksdb.RocksDbMetaStorage;
+import org.apache.ignite.internal.storage.rocksdb.RocksDbMvPartitionStorage;
+import org.apache.ignite.internal.storage.util.StorageState;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteStringFormatter;
+import org.jetbrains.annotations.Nullable;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteBatch;
+import org.rocksdb.WriteBatchWithIndex;
+
+/**
+ * Abstract index storage base on RocksDB.
+ */
+abstract class AbstractRocksDbIndexStorage implements IndexStorage {
+ private final UUID indexId;
+
+ final RocksDbMvPartitionStorage partitionStorage;
+
+ private final RocksDbMetaStorage indexMetaStorage;
+
+ /** Busy lock. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Current state of the storage. */
+ protected final AtomicReference<StorageState> state = new AtomicReference<>(StorageState.RUNNABLE);
+
+ /** Row ID for which the index needs to be built, {@code null} means that the index building has completed. */
+ private volatile @Nullable RowId nextRowIdToBuilt;
+
+ AbstractRocksDbIndexStorage(UUID indexId, RocksDbMvPartitionStorage partitionStorage, RocksDbMetaStorage indexMetaStorage) {
+ this.indexId = indexId;
+ this.partitionStorage = partitionStorage;
+ this.indexMetaStorage = indexMetaStorage;
+
+ int partitionId = partitionStorage.partitionId();
+
+ nextRowIdToBuilt = indexMetaStorage.getNextRowIdToBuilt(indexId, partitionId, RowId.lowestRowId(partitionId));
+ }
+
+ @Override
+ public @Nullable RowId getNextRowIdToBuild() {
+ return busy(() -> {
+ throwExceptionIfStorageInProgressOfRebalance(state.get(), this::createStorageInfo);
+
+ return nextRowIdToBuilt;
+ });
+ }
+
+ @Override
+ public void setNextRowIdToBuild(@Nullable RowId rowId) {
+ busy(() -> {
+ throwExceptionIfStorageInProgressOfRebalance(state.get(), this::createStorageInfo);
+
+ WriteBatchWithIndex writeBatch = partitionStorage.currentWriteBatch();
+
+ indexMetaStorage.putNextRowIdToBuilt(writeBatch, indexId, partitionStorage.partitionId(), rowId);
+
+ nextRowIdToBuilt = rowId;
+
+ return null;
+ });
+ }
+
+ /**
+ * Closes the hash index storage.
+ */
+ public void close() {
+ if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
+ StorageState state = this.state.get();
+
+ assert state == StorageState.CLOSED : state;
+
+ return;
+ }
+
+ busyLock.block();
+ }
+
+ /**
+ * Prepares the storage for rebalancing.
+ *
+ * @throws StorageRebalanceException If there was an error when starting the rebalance.
+ */
+ public void startRebalance(WriteBatch writeBatch) {
+ if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.REBALANCE)) {
+ throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
+ }
+
+ // Changed storage states and expect all storage operations to stop soon.
+ busyLock.block();
+
+ try {
+ destroyData(writeBatch);
+ } catch (RocksDBException e) {
+ throw new StorageRebalanceException("Error when trying to start rebalancing storage: " + createStorageInfo(), e);
+ } finally {
+ busyLock.unblock();
+ }
+ }
+
+ /**
+ * Aborts storage rebalancing.
+ *
+ * @throws StorageRebalanceException If there was an error when aborting the rebalance.
+ */
+ public void abortReblance(WriteBatch writeBatch) {
+ if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
+ throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
+ }
+
+ try {
+ destroyData(writeBatch);
+ } catch (RocksDBException e) {
+ throw new StorageRebalanceException("Error when trying to abort rebalancing storage: " + createStorageInfo(), e);
+ }
+ }
+
+ /**
+ * Completes storage rebalancing.
+ *
+ * @throws StorageRebalanceException If there was an error when finishing the rebalance.
+ */
+ public void finishRebalance() {
+ if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
+ throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
+ }
+ }
+
+ /**
+ * Prepares the storage for cleanup.
+ *
+ * <p>After cleanup (successful or not), method {@link #finishCleanup()} must be called.
+ */
+ public void startCleanup(WriteBatch writeBatch) throws RocksDBException {
+ if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLEANUP)) {
+ throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
+ }
+
+ // Changed storage states and expect all storage operations to stop soon.
+ busyLock.block();
+
+ destroyData(writeBatch);
+ }
+
+ /**
+ * Finishes cleanup up the storage.
+ */
+ public void finishCleanup() {
+ if (state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE)) {
+ busyLock.unblock();
+ }
+ }
+
+ <V> V busy(Supplier<V> supplier) {
+ if (!busyLock.enterBusy()) {
+ throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
+ }
+
+ try {
+ return supplier.get();
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ String createStorageInfo() {
+ return IgniteStringFormatter.format("indexId={}, partitionId={}", indexId, partitionStorage.partitionId());
+ }
+
+ /**
+ * Deletes the data associated with the index, using passed write batch for the operation.
+ *
+ * @throws RocksDBException If failed to delete data.
+ */
+ abstract void destroyData(WriteBatch writeBatch) throws RocksDBException;
+}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbHashIndexStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbHashIndexStorage.java
index 255f96ae02..2e94ac5995 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbHashIndexStorage.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/index/RocksDbHashIndexStorage.java
@@ -18,33 +18,29 @@
package org.apache.ignite.internal.storage.rocksdb.index;
import static org.apache.ignite.internal.rocksdb.RocksUtils.incrementPrefix;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageState;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.INDEX_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.KEY_BYTE_ORDER;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.PARTITION_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.ROW_ID_SIZE;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
import static org.apache.ignite.internal.util.ArrayUtils.BYTE_EMPTY_ARRAY;
import static org.apache.ignite.internal.util.ByteUtils.bytesToLong;
import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
import java.util.UUID;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Supplier;
import org.apache.ignite.internal.rocksdb.ColumnFamily;
import org.apache.ignite.internal.rocksdb.RocksIteratorAdapter;
import org.apache.ignite.internal.rocksdb.RocksUtils;
import org.apache.ignite.internal.schema.BinaryTuple;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageException;
-import org.apache.ignite.internal.storage.StorageRebalanceException;
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.apache.ignite.internal.storage.rocksdb.RocksDbMetaStorage;
import org.apache.ignite.internal.storage.rocksdb.RocksDbMvPartitionStorage;
-import org.apache.ignite.internal.storage.util.StorageState;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.util.HashUtils;
-import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.apache.ignite.lang.IgniteStringFormatter;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
@@ -67,49 +63,40 @@ import org.rocksdb.WriteOptions;
*
* <p>We use an empty array as values, because all required information can be extracted from the key.
*/
-public class RocksDbHashIndexStorage implements HashIndexStorage {
- /**
- * Length of the fixed part of the key: Index ID + Partition ID + Hash.
- */
- public static final int FIXED_PREFIX_LENGTH = 2 * Long.BYTES + Short.BYTES + Integer.BYTES;
+public class RocksDbHashIndexStorage extends AbstractRocksDbIndexStorage implements HashIndexStorage {
+ /** Length of the fixed part of the key: Index ID + Partition ID + Hash. */
+ public static final int FIXED_PREFIX_LENGTH = INDEX_ID_SIZE + PARTITION_ID_SIZE + Integer.BYTES;
private final HashIndexDescriptor descriptor;
private final ColumnFamily indexCf;
- private final RocksDbMvPartitionStorage partitionStorage;
-
- /**
- * Constant prefix of every index key.
- */
+ /** Constant prefix of every index key. */
private final byte[] constantPrefix;
- /** Busy lock. */
- private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
-
- /** Current state of the storage. */
- private final AtomicReference<StorageState> state = new AtomicReference<>(StorageState.RUNNABLE);
-
/**
* Creates a new Hash Index storage.
*
* @param descriptor Index descriptor.
* @param indexCf Column family that stores the index data.
* @param partitionStorage Partition storage of the partition that is being indexed (needed for consistency guarantees).
+ * @param indexMetaStorage Index meta storage.
*/
public RocksDbHashIndexStorage(
HashIndexDescriptor descriptor,
ColumnFamily indexCf,
- RocksDbMvPartitionStorage partitionStorage
+ RocksDbMvPartitionStorage partitionStorage,
+ RocksDbMetaStorage indexMetaStorage
) {
+ super(descriptor.id(), partitionStorage, indexMetaStorage);
+
this.descriptor = descriptor;
this.indexCf = indexCf;
- this.partitionStorage = partitionStorage;
UUID indexId = descriptor.id();
- this.constantPrefix = ByteBuffer.allocate(2 * Long.BYTES + Short.BYTES)
- .order(ByteOrder.BIG_ENDIAN)
+ this.constantPrefix = ByteBuffer.allocate(INDEX_ID_SIZE + PARTITION_ID_SIZE)
+ .order(KEY_BYTE_ORDER)
.putLong(indexId.getMostSignificantBits())
.putLong(indexId.getLeastSignificantBits())
.putShort((short) partitionStorage.partitionId())
@@ -235,7 +222,7 @@ public class RocksDbHashIndexStorage implements HashIndexStorage {
ByteBuffer keyBytes = prefix.byteBuffer();
return ByteBuffer.allocate(FIXED_PREFIX_LENGTH + keyBytes.remaining() + extraLength)
- .order(ByteOrder.BIG_ENDIAN)
+ .order(KEY_BYTE_ORDER)
.put(constantPrefix)
.putInt(HashUtils.hash32(keyBytes))
.put(keyBytes);
@@ -245,32 +232,13 @@ public class RocksDbHashIndexStorage implements HashIndexStorage {
RowId rowId = row.rowId();
// We don't store the Partition ID as it is already a part of the key.
- return rocksPrefix(row.indexColumns(), 2 * Long.BYTES)
+ return rocksPrefix(row.indexColumns(), ROW_ID_SIZE)
.putLong(rowId.mostSignificantBits())
.putLong(rowId.leastSignificantBits())
.array();
}
- /**
- * Closes the hash index storage.
- */
- public void close() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
- StorageState state = this.state.get();
-
- assert state == StorageState.CLOSED : state;
-
- return;
- }
-
- busyLock.block();
- }
-
- /**
- * Deletes the data associated with the index, using passed write batch for the operation.
- *
- * @throws RocksDBException If failed to delete data.
- */
+ @Override
public void destroyData(WriteBatch writeBatch) throws RocksDBException {
byte[] rangeEnd = incrementPrefix(constantPrefix);
@@ -278,95 +246,4 @@ public class RocksDbHashIndexStorage implements HashIndexStorage {
writeBatch.deleteRange(indexCf.handle(), constantPrefix, rangeEnd);
}
-
- private <V> V busy(Supplier<V> supplier) {
- if (!busyLock.enterBusy()) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- try {
- return supplier.get();
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- private String createStorageInfo() {
- return IgniteStringFormatter.format("indexId={}, partitionId={}", descriptor.id(), partitionStorage.partitionId());
- }
-
- /**
- * Prepares the storage for rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when starting the rebalance.
- */
- public void startRebalance(WriteBatch writeBatch) {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.REBALANCE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- try {
- destroyData(writeBatch);
- } catch (RocksDBException e) {
- throw new StorageRebalanceException("Error when trying to start rebalancing storage: " + createStorageInfo(), e);
- } finally {
- busyLock.unblock();
- }
- }
-
- /**
- * Aborts storage rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when aborting the rebalance.
- */
- public void abortReblance(WriteBatch writeBatch) {
- if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
-
- try {
- destroyData(writeBatch);
- } catch (RocksDBException e) {
- throw new StorageRebalanceException("Error when trying to abort rebalancing storage: " + createStorageInfo(), e);
- }
- }
-
- /**
- * Completes storage rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when finishing the rebalance.
- */
- public void finishRebalance() {
- if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
- }
-
- /**
- * Prepares the storage for cleanup.
- *
- * <p>After cleanup (successful or not), method {@link #finishCleanup()} must be called.
- */
- public void startCleanup(WriteBatch writeBatch) throws RocksDBException {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLEANUP)) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- destroyData(writeBatch);
- }
-
- /**
- * Finishes cleanup up the storage.
- */
- public void finishCleanup() {
- if (state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE)) {
- busyLock.unblock();
- }
- }
}
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 bb7e923122..17d082dfae 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
@@ -18,8 +18,9 @@
package org.apache.ignite.internal.storage.rocksdb.index;
import static org.apache.ignite.internal.rocksdb.RocksUtils.incrementPrefix;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageState;
-import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.KEY_BYTE_ORDER;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.PARTITION_ID_SIZE;
+import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.ROW_ID_SIZE;
import static org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
import static org.apache.ignite.internal.util.ArrayUtils.BYTE_EMPTY_ARRAY;
import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
@@ -27,9 +28,7 @@ import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.NoSuchElementException;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
-import java.util.function.Supplier;
import org.apache.ignite.internal.binarytuple.BinaryTupleCommon;
import org.apache.ignite.internal.rocksdb.ColumnFamily;
import org.apache.ignite.internal.rocksdb.RocksUtils;
@@ -37,17 +36,14 @@ import org.apache.ignite.internal.schema.BinaryTuple;
import org.apache.ignite.internal.schema.BinaryTuplePrefix;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageException;
-import org.apache.ignite.internal.storage.StorageRebalanceException;
import org.apache.ignite.internal.storage.index.IndexRow;
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.storage.rocksdb.RocksDbMetaStorage;
import org.apache.ignite.internal.storage.rocksdb.RocksDbMvPartitionStorage;
-import org.apache.ignite.internal.storage.util.StorageState;
import org.apache.ignite.internal.util.Cursor;
-import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.apache.ignite.lang.IgniteStringFormatter;
import org.jetbrains.annotations.Nullable;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDBException;
@@ -68,38 +64,29 @@ import org.rocksdb.WriteBatchWithIndex;
*
* <p>We use an empty array as values, because all required information can be extracted from the key.
*/
-public class RocksDbSortedIndexStorage implements SortedIndexStorage {
- private static final int ROW_ID_SIZE = Long.BYTES * 2;
-
- private static final ByteOrder ORDER = ByteOrder.BIG_ENDIAN;
-
+public class RocksDbSortedIndexStorage extends AbstractRocksDbIndexStorage implements SortedIndexStorage {
private final SortedIndexDescriptor descriptor;
private final ColumnFamily indexCf;
- private final RocksDbMvPartitionStorage partitionStorage;
-
- /** Busy lock. */
- private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
-
- /** Current state of the storage. */
- private final AtomicReference<StorageState> state = new AtomicReference<>(StorageState.RUNNABLE);
-
/**
* Creates a storage.
*
* @param descriptor Sorted Index descriptor.
* @param indexCf Column family that stores the index data.
* @param partitionStorage Partition storage of the corresponding index.
+ * @param indexMetaStorage Index meta storage.
*/
public RocksDbSortedIndexStorage(
SortedIndexDescriptor descriptor,
ColumnFamily indexCf,
- RocksDbMvPartitionStorage partitionStorage
+ RocksDbMvPartitionStorage partitionStorage,
+ RocksDbMetaStorage indexMetaStorage
) {
+ super(descriptor.id(), partitionStorage, indexMetaStorage);
+
this.descriptor = descriptor;
this.indexCf = indexCf;
- this.partitionStorage = partitionStorage;
}
@Override
@@ -246,7 +233,7 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
this.hasNext = null;
- return mapper.apply(ByteBuffer.wrap(key).order(ORDER));
+ return mapper.apply(ByteBuffer.wrap(key).order(KEY_BYTE_ORDER));
});
}
@@ -257,7 +244,7 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
if (hasNext != null) {
if (hasNext) {
- return mapper.apply(ByteBuffer.wrap(key).order(ORDER));
+ return mapper.apply(ByteBuffer.wrap(key).order(KEY_BYTE_ORDER));
}
return null;
@@ -270,7 +257,7 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
return null;
} else {
- return mapper.apply(ByteBuffer.wrap(it.key()).order(ORDER));
+ return mapper.apply(ByteBuffer.wrap(it.key()).order(KEY_BYTE_ORDER));
}
});
}
@@ -321,9 +308,9 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
private static void setEqualityFlag(byte[] prefix) {
// Flags start after the partition ID.
- byte flags = prefix[Short.BYTES];
+ byte flags = prefix[PARTITION_ID_SIZE];
- prefix[Short.BYTES] = (byte) (flags | BinaryTupleCommon.EQUALITY_FLAG);
+ prefix[PARTITION_ID_SIZE] = (byte) (flags | BinaryTupleCommon.EQUALITY_FLAG);
}
private IndexRow decodeRow(ByteBuffer bytes) {
@@ -345,8 +332,8 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
private byte[] rocksPrefix(BinaryTuplePrefix prefix) {
ByteBuffer bytes = prefix.byteBuffer();
- return ByteBuffer.allocate(Short.BYTES + bytes.remaining())
- .order(ORDER)
+ return ByteBuffer.allocate(PARTITION_ID_SIZE + bytes.remaining())
+ .order(KEY_BYTE_ORDER)
.putShort((short) partitionStorage.partitionId())
.put(bytes)
.array();
@@ -355,8 +342,8 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
private byte[] rocksKey(IndexRow row) {
ByteBuffer bytes = row.indexColumns().byteBuffer();
- return ByteBuffer.allocate(Short.BYTES + bytes.remaining() + ROW_ID_SIZE)
- .order(ORDER)
+ return ByteBuffer.allocate(PARTITION_ID_SIZE + bytes.remaining() + ROW_ID_SIZE)
+ .order(KEY_BYTE_ORDER)
.putShort((short) partitionStorage.partitionId())
.put(bytes)
.putLong(row.rowId().mostSignificantBits())
@@ -367,35 +354,19 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
private static ByteBuffer binaryTupleSlice(ByteBuffer key) {
return key.duplicate()
// Discard partition ID.
- .position(Short.BYTES)
+ .position(PARTITION_ID_SIZE)
// Discard row ID.
.limit(key.limit() - ROW_ID_SIZE)
.slice()
.order(ByteOrder.LITTLE_ENDIAN);
}
- /**
- * Closes the sorted index storage.
- */
- public void close() {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
- StorageState state = this.state.get();
-
- assert state == StorageState.CLOSED : state;
-
- return;
- }
-
- busyLock.block();
- }
-
- /**
- * Deletes the data associated with the index, using passed write batch for the operation.
- *
- * @throws RocksDBException If failed to delete data.
- */
+ @Override
public void destroyData(WriteBatch writeBatch) throws RocksDBException {
- byte[] constantPrefix = ByteBuffer.allocate(2).order(ByteOrder.BIG_ENDIAN).putShort((short) partitionStorage.partitionId()).array();
+ byte[] constantPrefix = ByteBuffer.allocate(PARTITION_ID_SIZE)
+ .order(KEY_BYTE_ORDER)
+ .putShort((short) partitionStorage.partitionId())
+ .array();
byte[] rangeEnd = incrementPrefix(constantPrefix);
@@ -403,95 +374,4 @@ public class RocksDbSortedIndexStorage implements SortedIndexStorage {
writeBatch.deleteRange(indexCf.handle(), constantPrefix, rangeEnd);
}
-
- private <V> V busy(Supplier<V> supplier) {
- if (!busyLock.enterBusy()) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- try {
- return supplier.get();
- } finally {
- busyLock.leaveBusy();
- }
- }
-
- private String createStorageInfo() {
- return IgniteStringFormatter.format("indexId={}, partitionId={}", descriptor.id(), partitionStorage.partitionId());
- }
-
- /**
- * Prepares the storage for rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when starting the rebalance.
- */
- public void startRebalance(WriteBatch writeBatch) {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.REBALANCE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- try {
- destroyData(writeBatch);
- } catch (RocksDBException e) {
- throw new StorageRebalanceException("Error when trying to start rebalancing storage: " + createStorageInfo(), e);
- } finally {
- busyLock.unblock();
- }
- }
-
- /**
- * Aborts storage rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when aborting the rebalance.
- */
- public void abortReblance(WriteBatch writeBatch) {
- if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
-
- try {
- destroyData(writeBatch);
- } catch (RocksDBException e) {
- throw new StorageRebalanceException("Error when trying to abort rebalancing storage: " + createStorageInfo(), e);
- }
- }
-
- /**
- * Completes storage rebalancing.
- *
- * @throws StorageRebalanceException If there was an error when finishing the rebalance.
- */
- public void finishRebalance() {
- if (!state.compareAndSet(StorageState.REBALANCE, StorageState.RUNNABLE)) {
- throwExceptionDependingOnStorageStateOnRebalance(state.get(), createStorageInfo());
- }
- }
-
- /**
- * Prepares the storage for cleanup.
- *
- * <p>After cleanup (successful or not), method {@link #finishCleanup()} must be called.
- */
- public void startCleanup(WriteBatch writeBatch) throws RocksDBException {
- if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLEANUP)) {
- throwExceptionDependingOnStorageState(state.get(), createStorageInfo());
- }
-
- // Changed storage states and expect all storage operations to stop soon.
- busyLock.block();
-
- destroyData(writeBatch);
- }
-
- /**
- * Finishes cleanup up the storage.
- */
- public void finishCleanup() {
- if (state.compareAndSet(StorageState.CLEANUP, StorageState.RUNNABLE)) {
- busyLock.unblock();
- }
- }
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
index b39b64f61b..93d6e253af 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
@@ -427,7 +427,7 @@ public class ItTxDistributedTestSingleNode extends TxAbstractTest {
Lazy<TableSchemaAwareIndexStorage> pkStorage = new Lazy<>(() -> new TableSchemaAwareIndexStorage(
indexId,
- new TestHashIndexStorage(null),
+ new TestHashIndexStorage(partId, null),
row2Tuple
));
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
index 17c1009b91..30ffda06d3 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
@@ -55,6 +55,8 @@ import org.junit.jupiter.api.extension.ExtendWith;
*/
@ExtendWith(ConfigurationExtension.class)
public abstract class IndexBaseTest extends BaseMvStoragesTest {
+ protected static final int PARTITION_ID = 0;
+
private static final BinaryTupleSchema TUPLE_SCHEMA = BinaryTupleSchema.createRowSchema(schemaDescriptor);
private static final BinaryTupleSchema PK_INDEX_SCHEMA = BinaryTupleSchema.createKeySchema(schemaDescriptor);
@@ -84,7 +86,7 @@ public abstract class IndexBaseTest extends BaseMvStoragesTest {
UUID sortedIndexId = UUID.randomUUID();
UUID hashIndexId = UUID.randomUUID();
- pkInnerStorage = new TestHashIndexStorage(null);
+ pkInnerStorage = new TestHashIndexStorage(PARTITION_ID, null);
TableSchemaAwareIndexStorage pkStorage = new TableSchemaAwareIndexStorage(
pkIndexId,
@@ -92,7 +94,7 @@ public abstract class IndexBaseTest extends BaseMvStoragesTest {
PK_INDEX_BINARY_TUPLE_CONVERTER::toTuple
);
- sortedInnerStorage = new TestSortedIndexStorage(new SortedIndexDescriptor(sortedIndexId, List.of(
+ sortedInnerStorage = new TestSortedIndexStorage(PARTITION_ID, new SortedIndexDescriptor(sortedIndexId, List.of(
new SortedIndexColumnDescriptor("INTVAL", NativeTypes.INT32, false, true),
new SortedIndexColumnDescriptor("STRVAL", NativeTypes.STRING, false, true)
)));
@@ -103,7 +105,7 @@ public abstract class IndexBaseTest extends BaseMvStoragesTest {
USER_INDEX_BINARY_TUPLE_CONVERTER::toTuple
);
- hashInnerStorage = new TestHashIndexStorage(new HashIndexDescriptor(hashIndexId, List.of(
+ hashInnerStorage = new TestHashIndexStorage(PARTITION_ID, new HashIndexDescriptor(hashIndexId, List.of(
new HashIndexColumnDescriptor("INTVAL", NativeTypes.INT32, false),
new HashIndexColumnDescriptor("STRVAL", NativeTypes.STRING, false)
)));
@@ -114,9 +116,9 @@ public abstract class IndexBaseTest extends BaseMvStoragesTest {
USER_INDEX_BINARY_TUPLE_CONVERTER::toTuple
);
- storage = new TestMvPartitionStorage(1);
+ storage = new TestMvPartitionStorage(PARTITION_ID);
- storageUpdateHandler = new StorageUpdateHandler(1, new TestPartitionDataStorage(storage),
+ storageUpdateHandler = new StorageUpdateHandler(PARTITION_ID, new TestPartitionDataStorage(storage),
() -> Map.of(
pkIndexId, pkStorage,
sortedIndexId, sortedIndexStorage,
@@ -133,7 +135,7 @@ public abstract class IndexBaseTest extends BaseMvStoragesTest {
}
static void addWrite(StorageUpdateHandler handler, UUID rowUuid, @Nullable BinaryRow row) {
- TablePartitionId partitionId = new TablePartitionId(UUID.randomUUID(), 1);
+ TablePartitionId partitionId = new TablePartitionId(UUID.randomUUID(), PARTITION_ID);
handler.handleUpdate(
TX_ID,
@@ -224,7 +226,7 @@ public abstract class IndexBaseTest extends BaseMvStoragesTest {
};
void addWrite(StorageUpdateHandler handler, UUID rowUuid, @Nullable BinaryRow row) {
- TablePartitionId tablePartitionId = new TablePartitionId(UUID.randomUUID(), 1);
+ TablePartitionId tablePartitionId = new TablePartitionId(UUID.randomUUID(), PARTITION_ID);
addWrite(handler, tablePartitionId, rowUuid, row);
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexCleanupTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexCleanupTest.java
index 12b7780727..1ad1139177 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexCleanupTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexCleanupTest.java
@@ -36,7 +36,7 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testAbort(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -59,7 +59,7 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testTombstoneCleansUpIndexes(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -88,7 +88,7 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testAbortTombstone(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -108,8 +108,8 @@ public class IndexCleanupTest extends IndexBaseTest {
void testAbortConsecutiveTxWithMatchingIndexes(AddWrite writer) {
UUID rowUuid1 = UUID.randomUUID();
UUID rowUuid2 = UUID.randomUUID();
- RowId rowId1 = new RowId(1, rowUuid1);
- RowId rowId2 = new RowId(1, rowUuid2);
+ RowId rowId1 = new RowId(PARTITION_ID, rowUuid1);
+ RowId rowId2 = new RowId(PARTITION_ID, rowUuid2);
var key1 = new TestKey(1, "foo");
var key2 = new TestKey(2, "baz");
@@ -141,9 +141,9 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testAbortConsecutiveTxWithMatchingIndexesSameRow(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
- var key = new TestKey(1, "foo");
+ var key = new TestKey(PARTITION_ID, "foo");
BinaryRow row1 = binaryRow(key, new TestValue(2, "bar"));
BinaryRow row2 = binaryRow(key, new TestValue(3, "baz"));
@@ -165,7 +165,7 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testIndexNotRemovedOnTombstone(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -186,7 +186,7 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testIndexNotRemovedWhileAbortingIfPreviousVersionExists(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -207,7 +207,7 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testIndexNotRemovedWhileWritingIfPreviousVersionExists(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -227,7 +227,7 @@ public class IndexCleanupTest extends IndexBaseTest {
@EnumSource(AddWrite.class)
void testIndexNotRemovedWhileWritingIfMultiplePreviousVersionsExists(AddWrite writer) {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java
index b8f09c2229..a01a443b7f 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java
@@ -32,7 +32,7 @@ public class IndexGcTest extends IndexBaseTest {
@Test
void testRemoveStaleEntryWithSameIndex() {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -52,7 +52,7 @@ public class IndexGcTest extends IndexBaseTest {
@Test
void testRemoveStaleEntriesWithDifferentIndexes() {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
var key = new TestKey(1, "foo");
@@ -87,7 +87,7 @@ public class IndexGcTest extends IndexBaseTest {
@Test
void testRemoveTombstonesRowNullNull() {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -114,7 +114,7 @@ public class IndexGcTest extends IndexBaseTest {
@Test
void testRemoveTombstonesRowNullRow() {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
@@ -139,7 +139,7 @@ public class IndexGcTest extends IndexBaseTest {
@Test
void testRemoveTombstonesRowRowNull() {
UUID rowUuid = UUID.randomUUID();
- RowId rowId = new RowId(1, rowUuid);
+ RowId rowId = new RowId(PARTITION_ID, rowUuid);
BinaryRow row = defaultRow();
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
index efa846bb69..fb847edfbd 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
@@ -140,7 +140,7 @@ public class PartitionCommandListenerTest {
/** Primary index. */
private final TableSchemaAwareIndexStorage pkStorage = new TableSchemaAwareIndexStorage(
UUID.randomUUID(),
- new TestHashIndexStorage(null),
+ new TestHashIndexStorage(PARTITION_ID, null),
BinaryRowConverter.keyExtractor(SCHEMA)
);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
index 76e018c0ea..8cde444fe6 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
@@ -131,7 +131,7 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
TableSchemaAwareIndexStorage hashIndexStorage = new TableSchemaAwareIndexStorage(
PK_INDEX_ID,
- new TestHashIndexStorage(null),
+ new TestHashIndexStorage(PART_ID, null),
row2HashKeyConverter
);
pkStorage = new Lazy<>(() -> hashIndexStorage);
@@ -146,7 +146,7 @@ public class PartitionReplicaListenerIndexLockingTest extends IgniteAbstractTest
TableSchemaAwareIndexStorage sortedIndexStorage = new TableSchemaAwareIndexStorage(
SORTED_INDEX_ID,
- new TestSortedIndexStorage(
+ new TestSortedIndexStorage(PART_ID,
new SortedIndexDescriptor(
SORTED_INDEX_ID,
List.of(new SortedIndexColumnDescriptor(
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index 5b191a1b28..837307f0b9 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -295,11 +295,11 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
pkStorage = new Lazy<>(() -> new TableSchemaAwareIndexStorage(
pkIndexId,
- new TestHashIndexStorage(null),
+ new TestHashIndexStorage(partId, null),
row2Tuple
));
- SortedIndexStorage indexStorage = new TestSortedIndexStorage(new SortedIndexDescriptor(sortedIndexId, List.of(
+ SortedIndexStorage indexStorage = new TestSortedIndexStorage(partId, new SortedIndexDescriptor(sortedIndexId, List.of(
new SortedIndexColumnDescriptor("intVal", NativeTypes.INT32, false, true)
)));
@@ -307,7 +307,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
hashIndexStorage = new TableSchemaAwareIndexStorage(
hashIndexId,
- new TestHashIndexStorage(new HashIndexDescriptor(hashIndexId, List.of(
+ new TestHashIndexStorage(partId, new HashIndexDescriptor(hashIndexId, List.of(
new HashIndexColumnDescriptor("intVal", NativeTypes.INT32, false)
))),
row -> null
diff --git a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index eb49f6da65..ac641836e7 100644
--- a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++ b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -255,7 +255,7 @@ public class DummyInternalTableImpl extends InternalTableImpl {
Lazy<TableSchemaAwareIndexStorage> pkStorage = new Lazy<>(() -> new TableSchemaAwareIndexStorage(
indexId,
- new TestHashIndexStorage(null),
+ new TestHashIndexStorage(PART_ID, null),
row2Tuple
));