You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2022/08/03 08:40:15 UTC
[ignite-3] branch main updated: IGNITE-17081 Disabled WAL in RocksDB based partition storage. Consistency closure / "snapshots" implemented. (#960)
This is an automated email from the ASF dual-hosted git repository.
ibessonov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 5f68cd0a4 IGNITE-17081 Disabled WAL in RocksDB based partition storage. Consistency closure / "snapshots" implemented. (#960)
5f68cd0a4 is described below
commit 5f68cd0a43986e4cb77d8178f4c5f29b02b10119
Author: ibessonov <be...@gmail.com>
AuthorDate: Wed Aug 3 11:40:11 2022 +0300
IGNITE-17081 Disabled WAL in RocksDB based partition storage. Consistency closure / "snapshots" implemented. (#960)
---
.../apache/ignite/raft/jraft/core/BallotBox.java | 2 +
.../internal/storage/MvPartitionStorage.java | 36 ++
.../internal/storage/engine/StorageEngine.java | 3 +
.../storage/AbstractMvPartitionStorageTest.java | 377 ++++++++++++---------
.../storage/rocksdb/RocksDbFlushListener.java | 106 ++++++
.../storage/rocksdb/RocksDbMetaStorage.java | 15 +-
.../storage/rocksdb/RocksDbMvPartitionStorage.java | 252 +++++++++++---
.../storage/rocksdb/RocksDbStorageEngine.java | 28 +-
.../storage/rocksdb/RocksDbTableStorage.java | 170 +++++++---
.../RocksDbStorageEngineConfigurationSchema.java | 4 +
.../rocksdb/RocksDbMvPartitionStorageTest.java | 1 -
.../rocksdb/RocksDbPartitionStorageTest.java | 105 ------
.../storage/rocksdb/RocksDbStorageEngineTest.java | 10 +-
.../storage/rocksdb/RocksDbTableStorageTest.java | 202 ++++-------
.../table/distributed/raft/PartitionListener.java | 253 +++++++++-----
.../distributed/storage/VersionedRowStore.java | 5 +-
parent/pom.xml | 2 +-
17 files changed, 973 insertions(+), 598 deletions(-)
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/BallotBox.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/BallotBox.java
index 78c1b0eae..bb19f6d40 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/BallotBox.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/BallotBox.java
@@ -175,6 +175,8 @@ public class BallotBox implements Lifecycle<BallotBoxOptions>, Describer {
return false;
}
this.pendingIndex = newPendingIndex;
+ //TODO Fix it properly: https://issues.apache.org/jira/browse/IGNITE-17445
+ this.lastCommittedIndex = newPendingIndex - 1;
this.closureQueue.resetFirstIndex(newPendingIndex);
return true;
}
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
index 7eb39e8f0..ed59c62ea 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.storage;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
import java.util.function.Predicate;
import org.apache.ignite.internal.schema.BinaryRow;
@@ -31,6 +32,41 @@ import org.jetbrains.annotations.Nullable;
* <p>Each MvPartitionStorage instance represents exactly one partition.
*/
public interface MvPartitionStorage extends AutoCloseable {
+ /**
+ * Closure for executing write operations on the storage.
+ *
+ * @param <V> Type of the result returned from the closure.
+ */
+ @SuppressWarnings("PublicInnerClass")
+ @FunctionalInterface
+ interface WriteClosure<V> {
+ V execute() throws StorageException;
+ }
+
+ /**
+ * Executes {@link WriteClosure} atomically, meaning that partial result of an incomplete closure will never be written to the
+ * physical device, thus guaranteeing data consistency after restart. Simply runs the closure in case of a volatile storage.
+ *
+ * @param closure Data access closure to be executed.
+ * @param <V> Type of the result returned from the closure.
+ * @return Closure result.
+ * @throws StorageException If failed to write data to the storage.
+ */
+ default <V> V runConsistently(WriteClosure<V> closure) throws StorageException {
+ return closure.execute();
+ }
+
+ /**
+ * Flushes current state of the data or <i>the state from the nearest future</i> to the storage. It means that the future can be
+ * completed when {@link #persistedIndex()} is higher than {@link #lastAppliedIndex()} at the moment of the method's call. This feature
+ * allows implementing a batch flush for several partitions at once.
+ *
+ * @return Future that's completed when flushing of the data is completed.
+ */
+ default CompletableFuture<Void> flush() {
+ return CompletableFuture.completedFuture(null);
+ }
+
/**
* Index of the highest write command applied to the storage. {@code 0} if index is unknown.
*/
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/StorageEngine.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/StorageEngine.java
index 8e82d7177..1e6ba299b 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/StorageEngine.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/StorageEngine.java
@@ -43,7 +43,10 @@ public interface StorageEngine {
*
* @param tableCfg Table configuration.
* @throws StorageException If an error has occurs while creating the table.
+ *
+ * @deprecated Replaced with {@link MvTableStorage}.
*/
+ @Deprecated
TableStorage createTable(TableConfiguration tableCfg) throws StorageException;
/**
diff --git a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/AbstractMvPartitionStorageTest.java b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/AbstractMvPartitionStorageTest.java
index 83725856f..38445e461 100644
--- a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/AbstractMvPartitionStorageTest.java
+++ b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/AbstractMvPartitionStorageTest.java
@@ -40,7 +40,6 @@ import java.util.stream.StreamSupport;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.tx.Timestamp;
import org.apache.ignite.internal.util.Cursor;
-import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.Test;
/**
@@ -57,6 +56,66 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
private final TestValue value2 = new TestValue(21, "bar2");
protected final BinaryRow binaryRow2 = binaryRow(key, value2);
+ /**
+ * Reads a row inside of consistency closure.
+ */
+ protected BinaryRow read(RowId rowId, UUID txId) {
+ return storage.runConsistently(() -> storage.read(rowId, txId));
+ }
+
+ /**
+ * Reads a row inside of consistency closure.
+ */
+ protected BinaryRow read(RowId rowId, Timestamp timestamp) {
+ return storage.runConsistently(() -> storage.read(rowId, timestamp));
+ }
+
+ /**
+ * Scans partition inside of consistency closure.
+ */
+ protected Cursor<BinaryRow> scan(Predicate<BinaryRow> filter, Timestamp timestamp) {
+ return storage.runConsistently(() -> storage.scan(filter, timestamp));
+ }
+
+ /**
+ * Scans partition inside of consistency closure.
+ */
+ protected Cursor<BinaryRow> scan(Predicate<BinaryRow> filter, UUID txId) {
+ return storage.runConsistently(() -> storage.scan(filter, txId));
+ }
+
+ /**
+ * Inserts a row inside of consistency closure.
+ */
+ private RowId insert(BinaryRow binaryRow, UUID txId) {
+ return storage.runConsistently(() -> storage.insert(binaryRow, txId));
+ }
+
+ /**
+ * Adds/updates a write-intent inside of consistency closure.
+ */
+ private BinaryRow addWrite(RowId rowId, BinaryRow binaryRow, UUID txId) {
+ return storage.runConsistently(() -> storage.addWrite(rowId, binaryRow, txId));
+ }
+
+ /**
+ * Commits write-intent inside of consistency closure.
+ */
+ protected void commitWrite(RowId rowId, Timestamp tsExact) {
+ storage.runConsistently(() -> {
+ storage.commitWrite(rowId, tsExact);
+
+ return null;
+ });
+ }
+
+ /**
+ * Aborts write-intent inside of consistency closure.
+ */
+ protected BinaryRow abortWrite(RowId rowId) {
+ return storage.runConsistently(() -> storage.abortWrite(rowId));
+ }
+
/**
* Tests that reads from empty storage return empty results.
*/
@@ -66,24 +125,26 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
assertEquals(partitionId(), rowId.partitionId());
- assertNull(storage.read(rowId, newTransactionId()));
- assertNull(storage.read(rowId, Timestamp.nextVersion()));
+ assertNull(read(rowId, newTransactionId()));
+ assertNull(read(rowId, Timestamp.nextVersion()));
}
private RowId insertAndAbortWrite() {
- RowId rowId = storage.insert(binaryRow, txId);
+ return storage.runConsistently(() -> {
+ RowId rowId = storage.insert(binaryRow, txId);
- storage.abortWrite(rowId);
+ storage.abortWrite(rowId);
- return rowId;
+ return rowId;
+ });
}
@Test
public void testScanOverEmpty() throws Exception {
insertAndAbortWrite();
- assertEquals(List.of(), convert(storage.scan(row -> true, newTransactionId())));
- assertEquals(List.of(), convert(storage.scan(row -> true, Timestamp.nextVersion())));
+ assertEquals(List.of(), convert(scan(row -> true, newTransactionId())));
+ assertEquals(List.of(), convert(scan(row -> true, Timestamp.nextVersion())));
}
protected int partitionId() {
@@ -99,22 +160,22 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
*/
@Test
public void testAddWrite() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
// Attempt to write from another transaction.
- assertThrows(TxIdMismatchException.class, () -> storage.addWrite(rowId, binaryRow, newTransactionId()));
+ assertThrows(TxIdMismatchException.class, () -> addWrite(rowId, binaryRow, newTransactionId()));
// Write from the same transaction.
- storage.addWrite(rowId, binaryRow, txId);
+ addWrite(rowId, binaryRow, txId);
// Read without timestamp returns uncommited row.
- assertRowMatches(storage.read(rowId, txId), binaryRow);
+ assertRowMatches(read(rowId, txId), binaryRow);
// Read with wrong transaction id should throw exception.
- assertThrows(TxIdMismatchException.class, () -> storage.read(rowId, newTransactionId()));
+ assertThrows(TxIdMismatchException.class, () -> read(rowId, newTransactionId()));
// Read with timestamp returns null.
- assertNull(storage.read(rowId, Timestamp.nextVersion()));
+ assertNull(read(rowId, Timestamp.nextVersion()));
}
/**
@@ -122,12 +183,12 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
*/
@Test
public void testAbortWrite() {
- RowId rowId = storage.insert(binaryRow(key, value), txId);
+ RowId rowId = insert(binaryRow(key, value), txId);
- storage.abortWrite(rowId);
+ abortWrite(rowId);
// Aborted row can't be read.
- assertNull(storage.read(rowId, txId));
+ assertNull(read(rowId, txId));
}
/**
@@ -135,74 +196,74 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
*/
@Test
public void testCommitWrite() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
Timestamp tsBefore = Timestamp.nextVersion();
Timestamp tsExact = Timestamp.nextVersion();
- storage.commitWrite(rowId, tsExact);
+ commitWrite(rowId, tsExact);
Timestamp tsAfter = Timestamp.nextVersion();
// Row is invisible at the time before writing.
- assertNull(storage.read(rowId, tsBefore));
+ assertNull(read(rowId, tsBefore));
// Row is valid at the time during and after writing.
- assertRowMatches(storage.read(rowId, tsExact), binaryRow);
- assertRowMatches(storage.read(rowId, tsAfter), binaryRow);
+ assertRowMatches(read(rowId, tsExact), binaryRow);
+ assertRowMatches(read(rowId, tsAfter), binaryRow);
TestValue newValue = new TestValue(30, "duh");
UUID newTxId = newTransactionId();
BinaryRow newRow = binaryRow(key, newValue);
- storage.addWrite(rowId, newRow, newTxId);
+ addWrite(rowId, newRow, newTxId);
// Same checks, but now there are two different versions.
- assertNull(storage.read(rowId, tsBefore));
+ assertNull(read(rowId, tsBefore));
- assertRowMatches(storage.read(rowId, newTxId), newRow);
+ assertRowMatches(read(rowId, newTxId), newRow);
- assertRowMatches(storage.read(rowId, tsExact), binaryRow);
- assertRowMatches(storage.read(rowId, tsAfter), binaryRow);
- assertRowMatches(storage.read(rowId, Timestamp.nextVersion()), binaryRow);
+ assertRowMatches(read(rowId, tsExact), binaryRow);
+ assertRowMatches(read(rowId, tsAfter), binaryRow);
+ assertRowMatches(read(rowId, Timestamp.nextVersion()), binaryRow);
// Only latest time behavior changes after commit.
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ commitWrite(rowId, Timestamp.nextVersion());
- assertRowMatches(storage.read(rowId, newTxId), newRow);
+ assertRowMatches(read(rowId, newTxId), newRow);
- assertRowMatches(storage.read(rowId, tsExact), binaryRow);
- assertRowMatches(storage.read(rowId, tsAfter), binaryRow);
+ assertRowMatches(read(rowId, tsExact), binaryRow);
+ assertRowMatches(read(rowId, tsAfter), binaryRow);
- assertRowMatches(storage.read(rowId, Timestamp.nextVersion()), newRow);
+ assertRowMatches(read(rowId, Timestamp.nextVersion()), newRow);
// Remove.
UUID removeTxId = newTransactionId();
- storage.addWrite(rowId, null, removeTxId);
+ addWrite(rowId, null, removeTxId);
- assertNull(storage.read(rowId, tsBefore));
+ assertNull(read(rowId, tsBefore));
- assertNull(storage.read(rowId, removeTxId));
+ assertNull(read(rowId, removeTxId));
- assertRowMatches(storage.read(rowId, tsExact), binaryRow);
- assertRowMatches(storage.read(rowId, tsAfter), binaryRow);
+ assertRowMatches(read(rowId, tsExact), binaryRow);
+ assertRowMatches(read(rowId, tsAfter), binaryRow);
- assertRowMatches(storage.read(rowId, Timestamp.nextVersion()), newRow);
+ assertRowMatches(read(rowId, Timestamp.nextVersion()), newRow);
// Commit remove.
Timestamp removeTs = Timestamp.nextVersion();
- storage.commitWrite(rowId, removeTs);
+ commitWrite(rowId, removeTs);
- assertNull(storage.read(rowId, tsBefore));
+ assertNull(read(rowId, tsBefore));
- assertNull(storage.read(rowId, removeTxId));
- assertNull(storage.read(rowId, removeTs));
- assertNull(storage.read(rowId, Timestamp.nextVersion()));
+ assertNull(read(rowId, removeTxId));
+ assertNull(read(rowId, removeTs));
+ assertNull(read(rowId, Timestamp.nextVersion()));
- assertRowMatches(storage.read(rowId, tsExact), binaryRow);
- assertRowMatches(storage.read(rowId, tsAfter), binaryRow);
+ assertRowMatches(read(rowId, tsExact), binaryRow);
+ assertRowMatches(read(rowId, tsAfter), binaryRow);
}
/**
@@ -217,14 +278,14 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
TestValue value2 = new TestValue(20, "yyy");
UUID txId1 = newTransactionId();
- RowId rowId1 = storage.insert(binaryRow(key1, value1), txId1);
+ RowId rowId1 = insert(binaryRow(key1, value1), txId1);
UUID txId2 = newTransactionId();
- RowId rowId2 = storage.insert(binaryRow(key2, value2), txId2);
+ RowId rowId2 = insert(binaryRow(key2, value2), txId2);
// Scan with and without filters.
- assertThrows(TxIdMismatchException.class, () -> convert(storage.scan(row -> true, txId1)));
- assertThrows(TxIdMismatchException.class, () -> convert(storage.scan(row -> true, txId2)));
+ assertThrows(TxIdMismatchException.class, () -> convert(scan(row -> true, txId1)));
+ assertThrows(TxIdMismatchException.class, () -> convert(scan(row -> true, txId2)));
assertEquals(List.of(value1), convert(storage.scan(row -> key(row).intKey == 1, txId1)));
assertEquals(List.of(value2), convert(storage.scan(row -> key(row).intKey == 2, txId2)));
@@ -232,23 +293,23 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
Timestamp ts1 = Timestamp.nextVersion();
Timestamp ts2 = Timestamp.nextVersion();
- storage.commitWrite(rowId1, ts2);
+ commitWrite(rowId1, ts2);
Timestamp ts3 = Timestamp.nextVersion();
Timestamp ts4 = Timestamp.nextVersion();
- storage.commitWrite(rowId2, ts4);
+ commitWrite(rowId2, ts4);
Timestamp ts5 = Timestamp.nextVersion();
// Full scan with various timestamp values.
- assertEquals(List.of(), convert(storage.scan(row -> true, ts1)));
+ assertEquals(List.of(), convert(scan(row -> true, ts1)));
- assertEquals(List.of(value1), convert(storage.scan(row -> true, ts2)));
- assertEquals(List.of(value1), convert(storage.scan(row -> true, ts3)));
+ assertEquals(List.of(value1), convert(scan(row -> true, ts2)));
+ assertEquals(List.of(value1), convert(scan(row -> true, ts3)));
- assertEquals(List.of(value1, value2), convert(storage.scan(row -> true, ts4)));
- assertEquals(List.of(value1, value2), convert(storage.scan(row -> true, ts5)));
+ assertEquals(List.of(value1, value2), convert(scan(row -> true, ts4)));
+ assertEquals(List.of(value1, value2), convert(scan(row -> true, ts5)));
}
@Test
@@ -257,13 +318,13 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
TestValue value2 = new TestValue(20, "yyy");
- RowId rowId1 = storage.insert(binaryRow(new TestKey(1, "1"), value1), txId);
- storage.commitWrite(rowId1, Timestamp.nextVersion());
+ RowId rowId1 = insert(binaryRow(new TestKey(1, "1"), value1), txId);
+ commitWrite(rowId1, Timestamp.nextVersion());
- RowId rowId2 = storage.insert(binaryRow(new TestKey(2, "2"), value2), txId);
- storage.commitWrite(rowId2, Timestamp.nextVersion());
+ RowId rowId2 = insert(binaryRow(new TestKey(2, "2"), value2), txId);
+ commitWrite(rowId2, Timestamp.nextVersion());
- Cursor<BinaryRow> cursor = storage.scan(row -> true, txId);
+ Cursor<BinaryRow> cursor = scan(row -> true, txId);
assertTrue(cursor.hasNext());
//noinspection ConstantConditions
@@ -299,79 +360,79 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
@Test
void readOfUncommittedRowWithCorrespondingTransactionIdReturnsTheRow() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- BinaryRow foundRow = storage.read(rowId, txId);
+ BinaryRow foundRow = read(rowId, txId);
assertRowMatches(foundRow, binaryRow);
}
- protected final void assertRowMatches(@Nullable BinaryRow rowUnderQuestion, BinaryRow expectedRow) {
+ protected final void assertRowMatches(BinaryRow rowUnderQuestion, BinaryRow expectedRow) {
assertThat(rowUnderQuestion, is(notNullValue()));
assertThat(rowUnderQuestion.bytes(), is(equalTo(expectedRow.bytes())));
}
@Test
void readOfUncommittedRowWithDifferentTransactionIdThrows() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- assertThrows(TxIdMismatchException.class, () -> storage.read(rowId, newTransactionId()));
+ assertThrows(TxIdMismatchException.class, () -> read(rowId, newTransactionId()));
}
@Test
void readOfCommittedRowWithAnyTransactionIdReturnsTheRow() {
- RowId rowId = storage.insert(binaryRow, txId);
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, txId);
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow foundRow = storage.read(rowId, newTransactionId());
+ BinaryRow foundRow = read(rowId, newTransactionId());
assertRowMatches(foundRow, binaryRow);
}
@Test
void readsUncommittedVersionEvenWhenThereIsCommittedVersionBeforeIt() {
- RowId rowId1 = storage.insert(binaryRow, txId);
- storage.commitWrite(rowId1, Timestamp.nextVersion());
+ RowId rowId1 = insert(binaryRow, txId);
+ commitWrite(rowId1, Timestamp.nextVersion());
- RowId rowId2 = storage.insert(binaryRow2, txId);
+ RowId rowId2 = insert(binaryRow2, txId);
- BinaryRow foundRow = storage.read(rowId2, txId);
+ BinaryRow foundRow = read(rowId2, txId);
assertRowMatches(foundRow, binaryRow2);
}
@Test
void readsCommittedVersionEvenWhenThereIsCommittedVersionBeforeIt() {
- RowId rowId1 = storage.insert(binaryRow, txId);
- storage.commitWrite(rowId1, Timestamp.nextVersion());
+ RowId rowId1 = insert(binaryRow, txId);
+ commitWrite(rowId1, Timestamp.nextVersion());
- RowId rowId2 = storage.insert(binaryRow2, txId);
- storage.commitWrite(rowId2, Timestamp.nextVersion());
+ RowId rowId2 = insert(binaryRow2, txId);
+ commitWrite(rowId2, Timestamp.nextVersion());
- BinaryRow foundRow = storage.read(rowId2, txId);
+ BinaryRow foundRow = read(rowId2, txId);
assertRowMatches(foundRow, binaryRow2);
}
@Test
void readByExactlyCommitTimestampFindsRow() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
Timestamp commitTimestamp = Timestamp.nextVersion();
- storage.commitWrite(rowId, commitTimestamp);
+ commitWrite(rowId, commitTimestamp);
- BinaryRow foundRow = storage.read(rowId, commitTimestamp);
+ BinaryRow foundRow = read(rowId, commitTimestamp);
assertRowMatches(foundRow, binaryRow);
}
@Test
void readByTimestampAfterCommitTimestampFindsRow() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
Timestamp commitTimestamp = Timestamp.nextVersion();
- storage.commitWrite(rowId, commitTimestamp);
+ commitWrite(rowId, commitTimestamp);
Timestamp afterCommit = Timestamp.nextVersion();
- BinaryRow foundRow = storage.read(rowId, afterCommit);
+ BinaryRow foundRow = read(rowId, afterCommit);
assertRowMatches(foundRow, binaryRow);
}
@@ -380,214 +441,214 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
void readByTimestampBeforeFirstVersionCommitTimestampFindsNothing() {
Timestamp beforeCommit = Timestamp.nextVersion();
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
Timestamp commitTimestamp = Timestamp.nextVersion();
- storage.commitWrite(rowId, commitTimestamp);
+ commitWrite(rowId, commitTimestamp);
- BinaryRow foundRow = storage.read(rowId, beforeCommit);
+ BinaryRow foundRow = read(rowId, beforeCommit);
assertThat(foundRow, is(nullValue()));
}
@Test
void readByTimestampOfLastVersionFindsLastVersion() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
Timestamp firstVersionTs = Timestamp.nextVersion();
- storage.commitWrite(rowId, firstVersionTs);
+ commitWrite(rowId, firstVersionTs);
- storage.addWrite(rowId, binaryRow2, newTransactionId());
+ addWrite(rowId, binaryRow2, newTransactionId());
Timestamp secondVersionTs = Timestamp.nextVersion();
- storage.commitWrite(rowId, secondVersionTs);
+ commitWrite(rowId, secondVersionTs);
- BinaryRow foundRow = storage.read(rowId, secondVersionTs);
+ BinaryRow foundRow = read(rowId, secondVersionTs);
assertRowMatches(foundRow, binaryRow2);
}
@Test
void readByTimestampOfPreviousVersionFindsPreviousVersion() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
Timestamp firstVersionTs = Timestamp.nextVersion();
- storage.commitWrite(rowId, firstVersionTs);
+ commitWrite(rowId, firstVersionTs);
- storage.addWrite(rowId, binaryRow2, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ addWrite(rowId, binaryRow2, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow foundRow = storage.read(rowId, firstVersionTs);
+ BinaryRow foundRow = read(rowId, firstVersionTs);
assertRowMatches(foundRow, binaryRow);
}
@Test
void readByTimestampBetweenVersionsFindsPreviousVersion() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
Timestamp firstVersionTs = Timestamp.nextVersion();
- storage.commitWrite(rowId, firstVersionTs);
+ commitWrite(rowId, firstVersionTs);
Timestamp tsInBetween = Timestamp.nextVersion();
- storage.addWrite(rowId, binaryRow2, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ addWrite(rowId, binaryRow2, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow foundRow = storage.read(rowId, tsInBetween);
+ BinaryRow foundRow = read(rowId, tsInBetween);
assertRowMatches(foundRow, binaryRow);
}
@Test
void readByTimestampIgnoresUncommittedVersion() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- storage.addWrite(rowId, binaryRow2, newTransactionId());
+ addWrite(rowId, binaryRow2, newTransactionId());
Timestamp latestTs = Timestamp.nextVersion();
- BinaryRow foundRow = storage.read(rowId, latestTs);
+ BinaryRow foundRow = read(rowId, latestTs);
assertRowMatches(foundRow, binaryRow);
}
@Test
void addWriteWithDifferentTxIdThrows() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- assertThrows(TxIdMismatchException.class, () -> storage.addWrite(rowId, binaryRow2, newTransactionId()));
+ assertThrows(TxIdMismatchException.class, () -> addWrite(rowId, binaryRow2, newTransactionId()));
}
@Test
void secondUncommittedWriteWithSameTxIdReplacesExistingUncommittedWrite() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- storage.addWrite(rowId, binaryRow2, txId);
+ addWrite(rowId, binaryRow2, txId);
- BinaryRow foundRow = storage.read(rowId, txId);
+ BinaryRow foundRow = read(rowId, txId);
assertRowMatches(foundRow, binaryRow2);
}
@Test
void addWriteReturnsUncommittedVersionIfItExists() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- BinaryRow returnedRow = storage.addWrite(rowId, binaryRow2, txId);
+ BinaryRow returnedRow = addWrite(rowId, binaryRow2, txId);
assertRowMatches(returnedRow, binaryRow);
}
@Test
void addWriteReturnsNullIfNoUncommittedVersionExists() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow returnedRow = storage.addWrite(rowId, binaryRow2, txId);
+ BinaryRow returnedRow = addWrite(rowId, binaryRow2, txId);
assertThat(returnedRow, is(nullValue()));
}
@Test
void afterRemovalReadWithTxIdFindsNothing() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- storage.addWrite(rowId, null, txId);
+ addWrite(rowId, null, txId);
- BinaryRow foundRow = storage.read(rowId, txId);
+ BinaryRow foundRow = read(rowId, txId);
assertThat(foundRow, is(nullValue()));
}
@Test
void afterRemovalReadByLatestTimestampFindsNothing() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- storage.addWrite(rowId, null, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ addWrite(rowId, null, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow foundRow = storage.read(rowId, Timestamp.nextVersion());
+ BinaryRow foundRow = read(rowId, Timestamp.nextVersion());
assertThat(foundRow, is(nullValue()));
}
@Test
void afterRemovalPreviousVersionRemainsAccessibleByTimestamp() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
+ RowId rowId = insert(binaryRow, newTransactionId());
Timestamp firstTimestamp = Timestamp.nextVersion();
- storage.commitWrite(rowId, firstTimestamp);
+ commitWrite(rowId, firstTimestamp);
- storage.addWrite(rowId, null, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ addWrite(rowId, null, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow foundRow = storage.read(rowId, firstTimestamp);
+ BinaryRow foundRow = read(rowId, firstTimestamp);
assertRowMatches(foundRow, binaryRow);
}
@Test
void removalReturnsUncommittedRowVersionIfItExists() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- BinaryRow rowFromRemoval = storage.addWrite(rowId, null, txId);
+ BinaryRow rowFromRemoval = addWrite(rowId, null, txId);
assertRowMatches(rowFromRemoval, binaryRow);
}
@Test
void removalReturnsNullIfNoUncommittedVersionExists() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow rowFromRemoval = storage.addWrite(rowId, null, newTransactionId());
+ BinaryRow rowFromRemoval = addWrite(rowId, null, newTransactionId());
assertThat(rowFromRemoval, is(nullValue()));
}
@Test
void commitWriteMakesVersionAvailableToReadByTimestamp() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ commitWrite(rowId, Timestamp.nextVersion());
- BinaryRow foundRow = storage.read(rowId, Timestamp.nextVersion());
+ BinaryRow foundRow = read(rowId, Timestamp.nextVersion());
assertRowMatches(foundRow, binaryRow);
}
@Test
void commitAndAbortWriteNoOpIfNoUncommittedVersionExists() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- storage.abortWrite(rowId);
+ abortWrite(rowId);
- assertRowMatches(storage.read(rowId, newTransactionId()), binaryRow);
+ assertRowMatches(read(rowId, newTransactionId()), binaryRow);
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ commitWrite(rowId, Timestamp.nextVersion());
- assertRowMatches(storage.read(rowId, newTransactionId()), binaryRow);
+ assertRowMatches(read(rowId, newTransactionId()), binaryRow);
}
@Test
void abortWriteRemovesUncommittedVersion() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
- storage.commitWrite(rowId, Timestamp.nextVersion());
+ RowId rowId = insert(binaryRow, newTransactionId());
+ commitWrite(rowId, Timestamp.nextVersion());
- storage.addWrite(rowId, binaryRow2, txId);
+ addWrite(rowId, binaryRow2, txId);
- storage.abortWrite(rowId);
+ abortWrite(rowId);
- BinaryRow foundRow = storage.read(rowId, txId);
+ BinaryRow foundRow = read(rowId, txId);
assertRowMatches(foundRow, binaryRow);
}
@Test
void abortOfInsertMakesRowNonExistentForReadByTimestamp() {
- RowId rowId = storage.insert(binaryRow, newTransactionId());
+ RowId rowId = insert(binaryRow, newTransactionId());
- storage.abortWrite(rowId);
+ abortWrite(rowId);
- BinaryRow foundRow = storage.read(rowId, Timestamp.nextVersion());
+ BinaryRow foundRow = read(rowId, Timestamp.nextVersion());
assertThat(foundRow, is(nullValue()));
}
@@ -596,25 +657,25 @@ public abstract class AbstractMvPartitionStorageTest<S extends MvPartitionStorag
void abortOfInsertMakesRowNonExistentForReadWithTxId() {
RowId rowId = insertAndAbortWrite();
- BinaryRow foundRow = storage.read(rowId, txId);
+ BinaryRow foundRow = read(rowId, txId);
assertThat(foundRow, is(nullValue()));
}
@Test
void abortWriteReturnsTheRemovedVersion() {
- RowId rowId = storage.insert(binaryRow, txId);
+ RowId rowId = insert(binaryRow, txId);
- BinaryRow returnedRow = storage.abortWrite(rowId);
+ BinaryRow returnedRow = abortWrite(rowId);
assertRowMatches(returnedRow, binaryRow);
}
@Test
void scanWithTxIdThrowsWhenOtherTransactionHasUncommittedChanges() {
- storage.insert(binaryRow, txId);
+ insert(binaryRow, txId);
- Cursor<BinaryRow> cursor = storage.scan(k -> true, newTransactionId());
+ Cursor<BinaryRow> cursor = scan(row -> true, newTransactionId());
assertThrows(TxIdMismatchException.class, cursor::next);
}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbFlushListener.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbFlushListener.java
new file mode 100644
index 000000000..92b1b41bd
--- /dev/null
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbFlushListener.java
@@ -0,0 +1,106 @@
+/*
+ * 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 static org.rocksdb.AbstractEventListener.EnabledEventCallback.ON_FLUSH_BEGIN;
+import static org.rocksdb.AbstractEventListener.EnabledEventCallback.ON_FLUSH_COMPLETED;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.configuration.schemas.table.TableView;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.storage.StorageException;
+import org.rocksdb.AbstractEventListener;
+import org.rocksdb.FlushJobInfo;
+import org.rocksdb.RocksDB;
+
+/**
+ * Represents a listener of RocksDB flush events. Responsible for updating persisted index of partitions.
+ *
+ * @see RocksDbMvPartitionStorage#persistedIndex()
+ * @see RocksDbMvPartitionStorage#refreshPersistedIndex()
+ */
+class RocksDbFlushListener extends AbstractEventListener {
+ /** Logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(RocksDbFlushListener.class);
+
+ /** Table storage. */
+ private final RocksDbTableStorage tableStorage;
+
+ /**
+ * Type of last processed event. Real amount of events doesn't matter in atomic flush mode. All "completed" events go after all "begin"
+ * events, and vice versa.
+ */
+ private final AtomicReference<EnabledEventCallback> lastEventType = new AtomicReference<>(ON_FLUSH_COMPLETED);
+
+ /**
+ * Future that guarantees that last flush was fully processed and the new flush can safely begin.
+ */
+ private volatile CompletableFuture<?> lastFlushProcessed = CompletableFuture.completedFuture(null);
+
+ public RocksDbFlushListener(RocksDbTableStorage tableStorage) {
+ super(EnabledEventCallback.ON_FLUSH_BEGIN, EnabledEventCallback.ON_FLUSH_COMPLETED);
+ this.tableStorage = tableStorage;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void onFlushBegin(RocksDB db, FlushJobInfo flushJobInfo) {
+ if (lastEventType.compareAndSet(ON_FLUSH_COMPLETED, ON_FLUSH_BEGIN)) {
+ lastFlushProcessed.join();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void onFlushCompleted(RocksDB db, FlushJobInfo flushJobInfo) {
+ if (lastEventType.compareAndSet(ON_FLUSH_BEGIN, ON_FLUSH_COMPLETED)) {
+ lastFlushProcessed = CompletableFuture.runAsync(this::refreshPersistedIndexes, tableStorage.engine().threadPool());
+ }
+ }
+
+ private void refreshPersistedIndexes() {
+ if (!tableStorage.busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ TableView tableCfgView = tableStorage.configuration().value();
+
+ for (int partitionId = 0; partitionId < tableCfgView.partitions(); partitionId++) {
+ RocksDbMvPartitionStorage partition = tableStorage.getMvPartition(partitionId);
+
+ if (partition != null) {
+ try {
+ partition.refreshPersistedIndex();
+ } catch (StorageException storageException) {
+ LOG.error(
+ "Filed to refresh persisted applied index value for table {} partition {}",
+ storageException,
+ tableStorage.configuration().name().value(),
+ partitionId
+ );
+ }
+ }
+ }
+ } finally {
+ tableStorage.busyLock.leaveBusy();
+ }
+ }
+}
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 25c6b148a..949cceacf 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
@@ -115,20 +115,7 @@ class RocksDbMetaStorage {
}
}
- /**
- * Removes the given partition ID from the meta Column Family.
- *
- * @param partitionId partition ID
- */
- void removePartitionId(int partitionId) {
- try {
- metaColumnFamily.delete(partitionIdKey(partitionId));
- } catch (RocksDBException e) {
- throw new StorageException("Unable to delete partition " + partitionId + " from the meta Column Family", e);
- }
- }
-
- private static byte[] partitionIdKey(int partitionId) {
+ static byte[] partitionIdKey(int partitionId) {
assert partitionId >= 0 && partitionId <= 0xFFFF : partitionId;
return ByteBuffer.allocate(PARTITION_ID_PREFIX.length + Short.BYTES)
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 9a3235b4e..8418d84c4 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,14 +21,23 @@ import static java.lang.ThreadLocal.withInitial;
import static java.nio.ByteBuffer.allocateDirect;
import static java.nio.ByteOrder.BIG_ENDIAN;
import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static java.util.Arrays.copyOf;
+import static java.util.Arrays.copyOfRange;
+import static org.rocksdb.ReadTier.PERSISTED_TIER;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
+import java.util.Iterator;
+import java.util.Map.Entry;
import java.util.NoSuchElementException;
+import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.function.BiConsumer;
import java.util.function.Predicate;
+import org.apache.ignite.configuration.schemas.table.TableConfiguration;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.ByteBufferRow;
import org.apache.ignite.internal.storage.MvPartitionStorage;
@@ -47,6 +56,8 @@ import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.rocksdb.Slice;
+import org.rocksdb.WriteBatch;
+import org.rocksdb.WriteBatchWithIndex;
import org.rocksdb.WriteOptions;
/**
@@ -56,7 +67,7 @@ import org.rocksdb.WriteOptions;
* or
* <pre><code>
* | rowId (16 bytes, BE) | timestamp (16 bytes, DESC) |</code></pre>
- * depending on transaction status. Pending transactions data doesn't have a timestamp assigned.
+ * depending on transaction status. Pending transactions' data doesn't have a timestamp assigned.
*
* <p/>BE means Big Endian, meaning that lexicographical bytes order matches a natural order of partitions.
*
@@ -79,11 +90,19 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
/** Maximum size of the key. */
private static final int MAX_KEY_SIZE = ROW_PREFIX_SIZE + TIMESTAMP_SIZE;
- /** Threadlocal direct buffer instance to read keys from RocksDB. */
+ /** Thread-local direct buffer instance to read keys from RocksDB. */
private static final ThreadLocal<ByteBuffer> MV_KEY_BUFFER = withInitial(() -> allocateDirect(MAX_KEY_SIZE).order(BIG_ENDIAN));
- /** Threadlocal on-heap byte buffer instance to use for key manipulations. */
- private final ThreadLocal<ByteBuffer> heapKeyBuffer;
+ /** Thread-local write batch for {@link #runConsistently(WriteClosure)}. */
+ private static final ThreadLocal<WriteBatchWithIndex> WRITE_BATCH = new ThreadLocal<>();
+
+ /** Thread-local on-heap byte buffer instance to use for key manipulations. */
+ private static final ThreadLocal<ByteBuffer> HEAP_KEY_BUFFER = withInitial(
+ () -> ByteBuffer.allocate(MAX_KEY_SIZE).order(BIG_ENDIAN)
+ );
+
+ /** Table storage instance. */
+ private final RocksDbTableStorage tableStorage;
/**
* Partition ID (should be treated as an unsigned short).
@@ -102,61 +121,104 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
private final ColumnFamilyHandle meta;
/** Write options. */
- private final WriteOptions writeOpts = new WriteOptions();
+ private final WriteOptions writeOpts = new WriteOptions().setDisableWAL(true);
+
+ /** Read options for regular reads. */
+ private final ReadOptions readOpts = new ReadOptions();
+
+ /** Read options for reading persisted data. */
+ private final ReadOptions persistedTierReadOpts = new ReadOptions().setReadTier(PERSISTED_TIER);
/** Upper bound for scans and reads. */
private final Slice upperBound;
/** Key to store applied index value in meta. */
- private byte[] lastAppliedIndexKey;
+ private final byte[] lastAppliedIndexKey;
+
+ /** On-heap-cached last applied index value. */
+ private volatile long lastAppliedIndex;
+
+ /** The value of {@link #lastAppliedIndex} persisted to the device at this moment. */
+ private volatile long persistedIndex;
+
+ /** Map with flush futures by applied index at the time of the {@link #flush()} call. */
+ private final ConcurrentMap<Long, CompletableFuture<Void>> flushFuturesByAppliedIndex = new ConcurrentHashMap<>();
/**
* Constructor.
*
+ * @param tableStorage Table storage.
* @param partitionId Partition id.
- * @param db RocksDB instance.
- * @param cf Column family handle to store partition data.
- * @param meta Column family handle to store partition metadata.
*/
- public RocksDbMvPartitionStorage(int partitionId, RocksDB db, ColumnFamilyHandle cf, ColumnFamilyHandle meta) {
+ public RocksDbMvPartitionStorage(RocksDbTableStorage tableStorage, int partitionId) {
+ this.tableStorage = tableStorage;
this.partitionId = partitionId;
- this.db = db;
- this.cf = cf;
- this.meta = meta;
-
- heapKeyBuffer = withInitial(() ->
- ByteBuffer.allocate(MAX_KEY_SIZE)
- .order(BIG_ENDIAN)
- );
+ db = tableStorage.db();
+ cf = tableStorage.partitionCfHandle();
+ meta = tableStorage.metaCfHandle();
upperBound = new Slice(partitionEndPrefix());
lastAppliedIndexKey = ("index" + partitionId).getBytes(StandardCharsets.UTF_8);
+
+ lastAppliedIndex = readLastAppliedIndex(readOpts);
+
+ persistedIndex = lastAppliedIndex;
}
/** {@inheritDoc} */
@Override
- public long lastAppliedIndex() {
- byte[] appliedIndexBytes;
+ public <V> V runConsistently(WriteClosure<V> closure) throws StorageException {
+ if (WRITE_BATCH.get() != null) {
+ return closure.execute();
+ } else {
+ try (var writeBatch = new WriteBatchWithIndex()) {
+ WRITE_BATCH.set(writeBatch);
+
+ V res = closure.execute();
+
+ try {
+ db.write(writeOpts, writeBatch);
+ } catch (RocksDBException e) {
+ throw new StorageException("Unable to apply a write batch to RocksDB instance.", e);
+ }
- try {
- appliedIndexBytes = db.get(meta, lastAppliedIndexKey);
- } catch (RocksDBException e) {
- throw new StorageException(e);
+ return res;
+ } finally {
+ WRITE_BATCH.set(null);
+ }
}
+ }
- if (appliedIndexBytes == null) {
- return 0;
- }
+ /** {@inheritDoc} */
+ @Override
+ public CompletableFuture<Void> flush() {
+ CompletableFuture<Void> flushFuture = new CompletableFuture<>();
+
+ CompletableFuture<Void> oldFuture = flushFuturesByAppliedIndex.put(lastAppliedIndex, flushFuture);
+
+ assert oldFuture == null;
+
+ tableStorage.scheduleFlush();
- return ByteUtils.bytesToLong(appliedIndexBytes);
+ return flushFuture;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public long lastAppliedIndex() {
+ return lastAppliedIndex;
}
/** {@inheritDoc} */
@Override
public void lastAppliedIndex(long lastAppliedIndex) throws StorageException {
+ WriteBatchWithIndex writeBatch = requireWriteBatch();
+
try {
- db.put(meta, lastAppliedIndexKey, ByteUtils.longToBytes(lastAppliedIndex));
+ writeBatch.put(meta, lastAppliedIndexKey, ByteUtils.longToBytes(lastAppliedIndex));
+
+ this.lastAppliedIndex = lastAppliedIndex;
} catch (RocksDBException e) {
throw new StorageException(e);
}
@@ -165,7 +227,52 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
/** {@inheritDoc} */
@Override
public long persistedIndex() {
- return lastAppliedIndex();
+ return persistedIndex;
+ }
+
+ /**
+ * Reads a value of {@link #lastAppliedIndex()} from the storage, avoiding memtable, and sets it as a new value of
+ * {@link #persistedIndex()}.
+ *
+ * <p/>All futures returned by {@link #flush()} are completed here if they correspond to the value of {@link #persistedIndex()}
+ * (if flush was called before data started being flushed to the storage).
+ *
+ * @throws StorageException If failed to read index from the storage.
+ */
+ public void refreshPersistedIndex() throws StorageException {
+ long persistedIndex = readLastAppliedIndex(persistedTierReadOpts);
+
+ this.persistedIndex = persistedIndex;
+
+ Set<Entry<Long, CompletableFuture<Void>>> entries = flushFuturesByAppliedIndex.entrySet();
+
+ for (Iterator<Entry<Long, CompletableFuture<Void>>> iterator = entries.iterator(); iterator.hasNext(); ) {
+ Entry<Long, CompletableFuture<Void>> entry = iterator.next();
+
+ if (persistedIndex >= entry.getKey()) {
+ entry.getValue().complete(null);
+
+ iterator.remove();
+ }
+ }
+ }
+
+ /**
+ * Reads the value of {@link #lastAppliedIndex} from the storage.
+ *
+ * @param readOptions Read options to be used for reading.
+ * @return The value of last applied index.
+ */
+ private long readLastAppliedIndex(ReadOptions readOptions) {
+ byte[] appliedIndexBytes;
+
+ try {
+ appliedIndexBytes = db.get(meta, readOptions, lastAppliedIndexKey);
+ } catch (RocksDBException e) {
+ throw new StorageException(e);
+ }
+
+ return appliedIndexBytes == null ? 0 : ByteUtils.bytesToLong(appliedIndexBytes);
}
/** {@inheritDoc} */
@@ -190,6 +297,8 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
throws TxIdMismatchException, StorageException {
assert rowId.partitionId() == partitionId : rowId;
+ WriteBatchWithIndex writeBatch = requireWriteBatch();
+
ByteBuffer keyBuf = prepareHeapKeyBuf(rowId);
BinaryRow res = null;
@@ -198,7 +307,7 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
// Check concurrent transaction data.
byte[] keyBufArray = keyBuf.array();
- byte[] previousValue = db.get(cf, keyBufArray, 0, ROW_PREFIX_SIZE);
+ byte[] previousValue = writeBatch.getFromBatchAndDB(db, cf, readOpts, copyOf(keyBufArray, ROW_PREFIX_SIZE));
// Previous value must belong to the same transaction.
if (previousValue != null) {
@@ -211,12 +320,12 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
// Write empty value as a tombstone.
if (previousValue != null) {
// Reuse old array with transaction id already written to it.
- db.put(cf, writeOpts, keyBufArray, 0, ROW_PREFIX_SIZE, previousValue, 0, TX_ID_SIZE);
+ writeBatch.put(cf, copyOf(keyBufArray, ROW_PREFIX_SIZE), copyOf(previousValue, TX_ID_SIZE));
} else {
// Use tail of the key buffer to save on array allocations.
putTransactionId(keyBufArray, ROW_PREFIX_SIZE, txId);
- db.put(cf, writeOpts, keyBufArray, 0, ROW_PREFIX_SIZE, keyBufArray, ROW_PREFIX_SIZE, TX_ID_SIZE);
+ writeBatch.put(cf, copyOf(keyBufArray, ROW_PREFIX_SIZE), copyOfRange(keyBufArray, ROW_PREFIX_SIZE, MAX_KEY_SIZE));
}
} else {
writeUnversioned(keyBufArray, row, txId);
@@ -237,6 +346,8 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
* @throws RocksDBException If write failed.
*/
private void writeUnversioned(byte[] keyArray, BinaryRow row, UUID txId) throws RocksDBException {
+ WriteBatchWithIndex writeBatch = requireWriteBatch();
+
//TODO IGNITE-16913 Add proper way to write row bytes into array without allocations.
byte[] rowBytes = row.bytes();
@@ -247,7 +358,7 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
value.position(TX_ID_SIZE).put(rowBytes);
// Write binary row data as a value.
- db.put(cf, writeOpts, keyArray, 0, ROW_PREFIX_SIZE, value.array(), 0, value.capacity());
+ writeBatch.put(cf, copyOf(keyArray, ROW_PREFIX_SIZE), copyOf(value.array(), value.capacity()));
}
/** {@inheritDoc} */
@@ -255,10 +366,12 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
public @Nullable BinaryRow abortWrite(RowId rowId) throws StorageException {
assert rowId.partitionId() == partitionId : rowId;
+ WriteBatchWithIndex writeBatch = requireWriteBatch();
+
ByteBuffer keyBuf = prepareHeapKeyBuf(rowId);
try {
- byte[] previousValue = db.get(cf, keyBuf.array(), 0, ROW_PREFIX_SIZE);
+ byte[] previousValue = writeBatch.getFromBatchAndDB(db, cf, readOpts, copyOf(keyBuf.array(), ROW_PREFIX_SIZE));
if (previousValue == null) {
//the chain doesn't contain an uncommitted write intent
@@ -266,7 +379,7 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
}
// Perform unconditional remove for the key without associated timestamp.
- db.delete(cf, writeOpts, keyBuf.array(), 0, ROW_PREFIX_SIZE);
+ writeBatch.delete(cf, copyOf(keyBuf.array(), ROW_PREFIX_SIZE));
return wrapValueIntoBinaryRow(previousValue, true);
} catch (RocksDBException e) {
@@ -279,11 +392,13 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
public void commitWrite(RowId rowId, Timestamp timestamp) throws StorageException {
assert rowId.partitionId() == partitionId : rowId;
+ WriteBatchWithIndex writeBatch = requireWriteBatch();
+
ByteBuffer keyBuf = prepareHeapKeyBuf(rowId);
try {
// Read a value associated with pending write.
- byte[] valueBytes = db.get(cf, keyBuf.array(), 0, ROW_PREFIX_SIZE);
+ byte[] valueBytes = writeBatch.getFromBatchAndDB(db, cf, readOpts, copyOf(keyBuf.array(), ROW_PREFIX_SIZE));
if (valueBytes == null) {
//the chain doesn't contain an uncommitted write intent
@@ -291,12 +406,12 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
}
// Delete pending write.
- db.delete(cf, writeOpts, keyBuf.array(), 0, ROW_PREFIX_SIZE);
+ writeBatch.delete(cf, copyOf(keyBuf.array(), ROW_PREFIX_SIZE));
// Add timestamp to the key, and put the value back into the storage.
putTimestamp(keyBuf, timestamp);
- db.put(cf, writeOpts, keyBuf.array(), 0, MAX_KEY_SIZE, valueBytes, TX_ID_SIZE, valueBytes.length - TX_ID_SIZE);
+ writeBatch.put(cf, copyOf(keyBuf.array(), MAX_KEY_SIZE), copyOfRange(valueBytes, TX_ID_SIZE, valueBytes.length));
} catch (RocksDBException e) {
throw new StorageException("Failed to commit row into storage", e);
}
@@ -316,32 +431,43 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
private @Nullable BinaryRow read(RowId rowId, @Nullable Timestamp timestamp, @Nullable UUID txId)
throws TxIdMismatchException, StorageException {
- assert rowId.partitionId() == partitionId : rowId;
assert timestamp == null ^ txId == null;
+ if (rowId.partitionId() != partitionId) {
+ return null;
+ }
+
+ // We can read data outside of consistency closure. Batch is not required.
+ WriteBatchWithIndex writeBatch = WRITE_BATCH.get();
+
ByteBuffer keyBuf = prepareHeapKeyBuf(rowId);
try (
// Set next partition as an upper bound.
var readOpts = new ReadOptions().setIterateUpperBound(upperBound);
- RocksIterator it = db.newIterator(cf, readOpts)
+ RocksIterator baseIterator = db.newIterator(cf, readOpts);
+ // "count()" check is mandatory. Write batch iterator without any updates just crashes everything.
+ // It's not documented, but this is exactly how it should be used.
+ RocksIterator seekIterator = writeBatch != null && writeBatch.count() > 0
+ ? writeBatch.newIteratorWithBase(cf, baseIterator)
+ : baseIterator;
) {
if (timestamp == null) {
// Seek to the first appearance of row id if timestamp isn't set.
// Since timestamps are sorted from newest to oldest, first occurance will always be the latest version.
// Unfortunately, copy here is unavoidable with current API.
- it.seek(Arrays.copyOf(keyBuf.array(), keyBuf.position()));
+ seekIterator.seek(copyOf(keyBuf.array(), keyBuf.position()));
} else {
// Put timestamp restriction according to N2O timestamps order.
putTimestamp(keyBuf, timestamp);
// This seek will either find a key with timestamp that's less or equal than required value, or a different key whatsoever.
// It is guaranteed by descending order of timestamps.
- it.seek(keyBuf.array());
+ seekIterator.seek(keyBuf.array());
}
// Return null if nothing was found.
- if (invalid(it)) {
+ if (invalid(seekIterator)) {
return null;
}
@@ -350,7 +476,7 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
// Here we prepare direct buffer to read key without timestamp. Shared direct buffer is used to avoid extra memory allocations.
ByteBuffer directBuffer = MV_KEY_BUFFER.get().position(0).limit(MAX_KEY_SIZE);
- int keyLength = it.key(directBuffer);
+ int keyLength = seekIterator.key(directBuffer);
boolean valueHasTxId = keyLength == ROW_PREFIX_SIZE;
@@ -360,7 +486,7 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
}
// Get binary row from the iterator. It has the exact payload that we need.
- byte[] valueBytes = it.value();
+ byte[] valueBytes = seekIterator.value();
assert valueBytes != null;
@@ -629,17 +755,39 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
* Deletes partition data from the storage.
*/
public void destroy() {
- try {
- db.delete(meta, lastAppliedIndexKey);
+ try (WriteBatch writeBatch = new WriteBatch()) {
+ writeBatch.delete(meta, lastAppliedIndexKey);
+
+ writeBatch.delete(meta, RocksDbMetaStorage.partitionIdKey(partitionId));
+
+ writeBatch.deleteRange(cf, partitionStartPrefix(), partitionEndPrefix());
+
+ db.write(writeOpts, writeBatch);
} catch (RocksDBException e) {
- throw new StorageException("Failed to cleanup partition meta on destruction", e);
+ TableConfiguration tableCfg = tableStorage.configuration();
+
+ throw new StorageException("Failed to destroy partition " + partitionId + " of table " + tableCfg.name(), e);
}
}
/** {@inheritDoc} */
@Override
public void close() throws Exception {
- IgniteUtils.closeAll(writeOpts, upperBound);
+ for (CompletableFuture<Void> future : flushFuturesByAppliedIndex.values()) {
+ future.cancel(false);
+ }
+
+ IgniteUtils.closeAll(persistedTierReadOpts, readOpts, writeOpts, upperBound);
+ }
+
+ private WriteBatchWithIndex requireWriteBatch() {
+ WriteBatchWithIndex writeBatch = WRITE_BATCH.get();
+
+ if (writeBatch == null) {
+ throw new StorageException("Attempting to write data outside of data access closure.");
+ }
+
+ return writeBatch;
}
/**
@@ -648,7 +796,7 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
private ByteBuffer prepareHeapKeyBuf(RowId rowId) {
assert rowId instanceof UuidRowId : rowId;
- ByteBuffer keyBuf = heapKeyBuffer.get().position(0);
+ ByteBuffer keyBuf = HEAP_KEY_BUFFER.get().position(0);
((UuidRowId) rowId).writeTo(keyBuf);
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngine.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngine.java
index 788d1ecfa..05192a7f2 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngine.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngine.java
@@ -25,6 +25,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.configuration.notifications.ConfigurationNamedListListener;
import org.apache.ignite.configuration.notifications.ConfigurationNotificationEvent;
@@ -69,6 +70,10 @@ public class RocksDbStorageEngine implements StorageEngine {
new NamedThreadFactory("rocksdb-storage-engine-pool", LOG)
);
+ private final ScheduledExecutorService scheduledPool = Executors.newSingleThreadScheduledExecutor(
+ new NamedThreadFactory("rocksdb-storage-engine-scheduled-pool", LOG)
+ );
+
private final Map<String, RocksDbDataRegion> regions = new ConcurrentHashMap<>();
/**
@@ -82,6 +87,27 @@ public class RocksDbStorageEngine implements StorageEngine {
this.storagePath = storagePath;
}
+ /**
+ * Returns a RocksDB storage engine configuration.
+ */
+ public RocksDbStorageEngineConfiguration configuration() {
+ return engineConfig;
+ }
+
+ /**
+ * Returns a common thread pool for async operations.
+ */
+ public ExecutorService threadPool() {
+ return threadPool;
+ }
+
+ /**
+ * Returns a scheduled thread pool for async operations.
+ */
+ public ScheduledExecutorService scheduledPool() {
+ return scheduledPool;
+ }
+
/** {@inheritDoc} */
@Override
public void start() throws StorageException {
@@ -139,7 +165,7 @@ public class RocksDbStorageEngine implements StorageEngine {
throw new StorageException("Failed to create table store directory for " + tableView.name() + ": " + e.getMessage(), e);
}
- return new RocksDbTableStorage(tablePath, tableCfg, threadPool, dataRegion);
+ return new RocksDbTableStorage(this, tablePath, tableCfg, dataRegion);
}
@Override
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 02b35530a..de3aadc75 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
@@ -33,11 +33,15 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.stream.Collectors;
import org.apache.ignite.configuration.schemas.table.TableConfiguration;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.rocksdb.ColumnFamily;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.PartitionStorage;
import org.apache.ignite.internal.storage.StorageException;
import org.apache.ignite.internal.storage.engine.MvTableStorage;
@@ -47,35 +51,44 @@ import org.apache.ignite.internal.storage.index.SortedIndexStorage;
import org.apache.ignite.internal.storage.rocksdb.index.BinaryRowComparator;
import org.apache.ignite.internal.storage.rocksdb.index.RocksDbSortedIndexStorage;
import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.IgniteUtils;
import org.jetbrains.annotations.Nullable;
import org.rocksdb.ColumnFamilyDescriptor;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
+import org.rocksdb.FlushOptions;
import org.rocksdb.Options;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteOptions;
/**
* Table storage implementation based on {@link RocksDB} instance.
*/
class RocksDbTableStorage implements TableStorage, MvTableStorage {
+ /** Logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(RocksDbTableStorage.class);
+
+ /** RocksDB storage engine instance. */
+ private final RocksDbStorageEngine engine;
+
/** Path for the directory that stores table data. */
private final Path tablePath;
/** Table configuration. */
private final TableConfiguration tableCfg;
- /** Thread pool for async operations. */
- private final Executor threadPool;
-
/** Data region for the table. */
private final RocksDbDataRegion dataRegion;
/** Rocks DB instance. */
private volatile RocksDB db;
+ /** Write options for write operations. */
+ private final WriteOptions writeOptions = new WriteOptions().setDisableWAL(true);
+
/** Meta information. */
private volatile RocksDbMetaStorage meta;
@@ -83,34 +96,77 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
private volatile ColumnFamily partitionCf;
/** Partition storages. */
- private volatile AtomicReferenceArray<PartitionStorage> partitions;
+ private volatile AtomicReferenceArray<RocksDbMvPartitionStorage> partitions;
/** Column families for indexes by their names. */
private final Map<String, RocksDbSortedIndexStorage> sortedIndices = new ConcurrentHashMap<>();
+ /**
+ * Instance of the latest scheduled flush closure.
+ *
+ * @see #scheduleFlush()
+ */
+ private volatile Runnable latestFlushClosure;
+
/** Flag indicating if the storage has been stopped. */
+ @Deprecated
private volatile boolean stopped = false;
+ //TODO Use it instead of the "stopped" flag.
+ /** Busy lock to stop synchronously. */
+ final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Prevents double stopping the component. */
+ private final AtomicBoolean stopGuard = new AtomicBoolean();
+
/**
* Constructor.
*
+ * @param engine RocksDB storage engine instance.
* @param tablePath Path for the directory that stores table data.
* @param tableCfg Table configuration.
- * @param threadPool Thread pool for async operations.
* @param dataRegion Data region for the table.
*/
RocksDbTableStorage(
+ RocksDbStorageEngine engine,
Path tablePath,
TableConfiguration tableCfg,
- Executor threadPool,
RocksDbDataRegion dataRegion
) {
+ this.engine = engine;
this.tablePath = tablePath;
this.tableCfg = tableCfg;
- this.threadPool = threadPool;
this.dataRegion = dataRegion;
}
+ /**
+ * Returns a storage engine instance.
+ */
+ public RocksDbStorageEngine engine() {
+ return engine;
+ }
+
+ /**
+ * Returns a {@link RocksDB} instance.
+ */
+ public RocksDB db() {
+ return db;
+ }
+
+ /**
+ * Returns a column family handle for partitions column family.
+ */
+ public ColumnFamilyHandle partitionCfHandle() {
+ return partitionCf.handle();
+ }
+
+ /**
+ * Returns a column family handle for meta column family.
+ */
+ public ColumnFamilyHandle metaCfHandle() {
+ return meta.columnFamily().handle();
+ }
+
/** {@inheritDoc} */
@Override
public TableConfiguration configuration() {
@@ -133,6 +189,9 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
DBOptions dbOptions = new DBOptions()
.setCreateIfMissing(true)
.setCreateMissingColumnFamilies(true)
+ // Atomic flush must be enabled to guarantee consistency between different column families when WAL is disabled.
+ .setAtomicFlush(true)
+ .setListeners(List.of(new RocksDbFlushListener(this)))
.setWriteBufferManager(dataRegion.writeBufferManager());
try {
@@ -173,23 +232,58 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
partitions = new AtomicReferenceArray<>(tableCfg.value().partitions());
for (int partId : meta.getPartitionIds()) {
- partitions.set(partId, new RocksDbPartitionStorage(db, partitionCf, partId, threadPool));
+ partitions.set(partId, new RocksDbMvPartitionStorage(this, partId));
}
}
+ /**
+ * Schedules a flush of the table. If run several times within a small amount of time, only the last scheduled flush will be executed.
+ */
+ public void scheduleFlush() {
+ Runnable newClosure = new Runnable() {
+ @Override
+ public void run() {
+ if (latestFlushClosure != this) {
+ return;
+ }
+
+ try (FlushOptions flushOptions = new FlushOptions().setWaitForFlush(false)) {
+ db.flush(flushOptions);
+ } catch (RocksDBException e) {
+ LOG.error("Error occurred during the explicit flush for table '{}'", e, tableCfg.name());
+ }
+ }
+ };
+
+ latestFlushClosure = newClosure;
+
+ int delay = engine.configuration().flushDelayMillis().value();
+
+ engine.scheduledPool().schedule(newClosure, delay, TimeUnit.MILLISECONDS);
+ }
+
+
/** {@inheritDoc} */
@Override
public void stop() throws StorageException {
stopped = true;
+ if (!stopGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ busyLock.block();
+
List<AutoCloseable> resources = new ArrayList<>();
resources.add(db);
+ resources.add(writeOptions);
+
resources.addAll(sortedIndices.values());
for (int i = 0; i < partitions.length(); i++) {
- PartitionStorage partition = partitions.get(i);
+ MvPartitionStorage partition = partitions.get(i);
if (partition != null) {
resources.add(partition);
@@ -216,64 +310,46 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
/** {@inheritDoc} */
@Override
public PartitionStorage getOrCreatePartition(int partId) throws StorageException {
- PartitionStorage storage = getPartition(partId);
-
- if (storage != null) {
- return storage;
- }
-
- // Possible races when creating the partitions with the same ID are safe, since both the storage creation and the meta update
- // are cheap and idempotent.
- storage = new RocksDbPartitionStorage(db, partitionCf, partId, threadPool);
-
- partitions.set(partId, storage);
-
- meta.putPartitionId(partId);
-
- return storage;
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
@Nullable
@Override
public PartitionStorage getPartition(int partId) {
- assert !stopped : "Storage has been stopped";
-
- checkPartitionId(partId);
-
- return partitions.get(partId);
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
@Override
public void dropPartition(int partId) throws StorageException {
- PartitionStorage partition = getPartition(partId);
-
- if (partition != null) {
- partitions.set(partId, null);
-
- partition.destroy();
-
- meta.removePartitionId(partId);
- }
+ throw new UnsupportedOperationException();
}
/** {@inheritDoc} */
@Override
public RocksDbMvPartitionStorage getOrCreateMvPartition(int partitionId) throws StorageException {
- getOrCreatePartition(partitionId);
+ RocksDbMvPartitionStorage partition = getMvPartition(partitionId);
+
+ if (partition != null) {
+ return partition;
+ }
+
+ partition = new RocksDbMvPartitionStorage(this, partitionId);
+
+ partitions.set(partitionId, partition);
+
+ meta.putPartitionId(partitionId);
- return getMvPartition(partitionId);
+ return partition;
}
/** {@inheritDoc} */
@Override
public @Nullable RocksDbMvPartitionStorage getMvPartition(int partitionId) {
- if (getPartition(partitionId) == null) {
- return null;
- }
+ checkPartitionId(partitionId);
- return new RocksDbMvPartitionStorage(partitionId, db, partitionCf.handle(), meta.columnFamily().handle());
+ return partitions.get(partitionId);
}
/** {@inheritDoc} */
@@ -281,9 +357,9 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
public CompletableFuture<?> destroyPartition(int partitionId) throws StorageException {
RocksDbMvPartitionStorage mvPartition = getMvPartition(partitionId);
- dropPartition(partitionId);
-
if (mvPartition != null) {
+ partitions.set(partitionId, null);
+
mvPartition.destroy();
}
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/configuration/schema/RocksDbStorageEngineConfigurationSchema.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/configuration/schema/RocksDbStorageEngineConfigurationSchema.java
index b0a3c06da..5e1c75b48 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/configuration/schema/RocksDbStorageEngineConfigurationSchema.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/configuration/schema/RocksDbStorageEngineConfigurationSchema.java
@@ -23,6 +23,7 @@ import org.apache.ignite.configuration.annotation.ConfigValue;
import org.apache.ignite.configuration.annotation.ConfigurationRoot;
import org.apache.ignite.configuration.annotation.Name;
import org.apache.ignite.configuration.annotation.NamedConfigValue;
+import org.apache.ignite.configuration.annotation.Value;
import org.apache.ignite.configuration.validation.ExceptKeys;
import org.apache.ignite.internal.storage.rocksdb.RocksDbStorageEngine;
@@ -34,6 +35,9 @@ public class RocksDbStorageEngineConfigurationSchema {
/** Name of the default data region. */
public static final String DEFAULT_DATA_REGION_NAME = "default";
+ @Value(hasDefault = true)
+ public int flushDelayMillis = 100;
+
/** Default data region. */
@Name(DEFAULT_DATA_REGION_NAME)
@ConfigValue
diff --git a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorageTest.java b/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorageTest.java
index 62cb71b29..cd5bc54a7 100644
--- a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorageTest.java
+++ b/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorageTest.java
@@ -94,5 +94,4 @@ public class RocksDbMvPartitionStorageTest extends AbstractMvPartitionStorageTes
engine == null ? null : engine::stop
);
}
-
}
diff --git a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbPartitionStorageTest.java b/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbPartitionStorageTest.java
deleted file mode 100644
index b1299b93f..000000000
--- a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbPartitionStorageTest.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * 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 static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbStorageEngineConfigurationSchema.DEFAULT_DATA_REGION_NAME;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
-
-import java.nio.file.Path;
-import org.apache.ignite.configuration.schemas.store.UnknownDataStorageConfigurationSchema;
-import org.apache.ignite.configuration.schemas.table.ConstantValueDefaultConfigurationSchema;
-import org.apache.ignite.configuration.schemas.table.FunctionCallDefaultConfigurationSchema;
-import org.apache.ignite.configuration.schemas.table.HashIndexConfigurationSchema;
-import org.apache.ignite.configuration.schemas.table.NullValueDefaultConfigurationSchema;
-import org.apache.ignite.configuration.schemas.table.TableConfiguration;
-import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
-import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.storage.AbstractPartitionStorageTest;
-import org.apache.ignite.internal.storage.engine.StorageEngine;
-import org.apache.ignite.internal.storage.engine.TableStorage;
-import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageChange;
-import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageConfigurationSchema;
-import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageView;
-import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbStorageEngineConfiguration;
-import org.apache.ignite.internal.testframework.WorkDirectory;
-import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
-import org.apache.ignite.internal.util.IgniteUtils;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.extension.ExtendWith;
-
-/**
- * Storage test implementation for {@link RocksDbPartitionStorage}.
- */
-@ExtendWith(WorkDirectoryExtension.class)
-@ExtendWith(ConfigurationExtension.class)
-public class RocksDbPartitionStorageTest extends AbstractPartitionStorageTest {
- private StorageEngine engine;
-
- private TableStorage table;
-
- @BeforeEach
- public void setUp(
- @WorkDirectory Path workDir,
- @InjectConfiguration RocksDbStorageEngineConfiguration engineConfig,
- @InjectConfiguration(
- name = "table",
- polymorphicExtensions = {
- HashIndexConfigurationSchema.class,
- UnknownDataStorageConfigurationSchema.class,
- RocksDbDataStorageConfigurationSchema.class,
- ConstantValueDefaultConfigurationSchema.class,
- FunctionCallDefaultConfigurationSchema.class,
- NullValueDefaultConfigurationSchema.class,
- }
- ) TableConfiguration tableCfg
- ) throws Exception {
- tableCfg.dataStorage().change(c -> c.convert(RocksDbDataStorageChange.class)).get(1, SECONDS);
-
- assertThat(((RocksDbDataStorageView) tableCfg.dataStorage().value()).dataRegion(), equalTo(DEFAULT_DATA_REGION_NAME));
-
- engineConfig.defaultRegion().change(c -> c.changeSize(16 * 1024).changeWriteBufferSize(16 * 1024)).get(1, SECONDS);
-
- engine = new RocksDbStorageEngine(engineConfig, workDir);
-
- engine.start();
-
- table = engine.createTable(tableCfg);
-
- assertThat(table, is(instanceOf(RocksDbTableStorage.class)));
-
- table.start();
-
- storage = table.getOrCreatePartition(0);
-
- assertThat(storage, is(instanceOf(RocksDbPartitionStorage.class)));
- }
-
- @AfterEach
- public void tearDown() throws Exception {
- IgniteUtils.closeAll(
- storage,
- table == null ? null : table::stop,
- engine == null ? null : engine::stop
- );
- }
-}
diff --git a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngineTest.java b/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngineTest.java
index 0939ff1e7..46e56cb3d 100644
--- a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngineTest.java
+++ b/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbStorageEngineTest.java
@@ -31,7 +31,7 @@ import org.apache.ignite.configuration.schemas.table.NullValueDefaultConfigurati
import org.apache.ignite.configuration.schemas.table.TableConfiguration;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageConfiguration;
import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageConfigurationSchema;
import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbStorageEngineConfiguration;
@@ -79,7 +79,7 @@ public class RocksDbStorageEngineTest {
}
) TableConfiguration tableCfg
) {
- TableStorage table = engine.createTable(tableCfg);
+ MvTableStorage table = engine.createMvTable(tableCfg);
table.start();
@@ -88,7 +88,7 @@ public class RocksDbStorageEngineTest {
assertThat(dataStorageConfig.dataRegion().value(), is(DEFAULT_DATA_REGION_NAME));
- table.getOrCreatePartition(1);
+ table.getOrCreateMvPartition(1);
} finally {
table.stop();
}
@@ -115,7 +115,7 @@ public class RocksDbStorageEngineTest {
assertThat(engineConfigChangeFuture, willCompleteSuccessfully());
- TableStorage table = engine.createTable(tableCfg);
+ MvTableStorage table = engine.createMvTable(tableCfg);
table.start();
@@ -124,7 +124,7 @@ public class RocksDbStorageEngineTest {
assertThat(dataStorageConfig.dataRegion().value(), is(customRegionName));
- table.getOrCreatePartition(1);
+ table.getOrCreateMvPartition(1);
} finally {
table.stop();
}
diff --git a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorageTest.java b/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorageTest.java
index 3980900ca..c02e423d3 100644
--- a/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorageTest.java
+++ b/modules/storage-rocksdb/src/test/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorageTest.java
@@ -26,12 +26,11 @@ import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.sameInstance;
-import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -43,10 +42,12 @@ import org.apache.ignite.configuration.schemas.table.NullValueDefaultConfigurati
import org.apache.ignite.configuration.schemas.table.TableConfiguration;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.storage.PartitionStorage;
-import org.apache.ignite.internal.storage.basic.SimpleDataRow;
+import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.storage.BaseMvStoragesTest;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.storage.engine.StorageEngine;
-import org.apache.ignite.internal.storage.engine.TableStorage;
import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageChange;
import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageConfigurationSchema;
import org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbDataStorageView;
@@ -55,6 +56,7 @@ import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.lang.IgniteBiTuple;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -65,19 +67,18 @@ import org.junit.jupiter.api.extension.ExtendWith;
*/
@ExtendWith(WorkDirectoryExtension.class)
@ExtendWith(ConfigurationExtension.class)
-public class RocksDbTableStorageTest {
- @WorkDirectory
- private Path workDir;
-
+public class RocksDbTableStorageTest extends BaseMvStoragesTest {
private StorageEngine engine;
- private TableStorage storage;
+ private MvTableStorage storage;
@BeforeEach
public void setUp(
+ @WorkDirectory Path workDir,
@InjectConfiguration RocksDbStorageEngineConfiguration rocksDbEngineConfig,
@InjectConfiguration(
name = "table",
+ value = "mock.partitions = 1024",
polymorphicExtensions = {
HashIndexConfigurationSchema.class,
UnknownDataStorageConfigurationSchema.class,
@@ -107,7 +108,7 @@ public class RocksDbTableStorageTest {
engine.start();
- storage = engine.createTable(tableCfg);
+ storage = engine.createMvTable(tableCfg);
assertThat(storage, is(instanceOf(RocksDbTableStorage.class)));
@@ -127,23 +128,27 @@ public class RocksDbTableStorageTest {
*/
@Test
void testCreatePartition() {
- PartitionStorage partitionStorage = storage.getPartition(0);
+ MvPartitionStorage absentStorage = storage.getMvPartition(0);
- assertThat(partitionStorage, is(nullValue()));
+ assertThat(absentStorage, is(nullValue()));
- partitionStorage = storage.getOrCreatePartition(0);
+ MvPartitionStorage partitionStorage = storage.getOrCreateMvPartition(0);
assertThat(partitionStorage, is(notNullValue()));
- var testData = new SimpleDataRow("foo".getBytes(StandardCharsets.UTF_8), "bar".getBytes(StandardCharsets.UTF_8));
+ var testData = binaryRow(new TestKey(1, "1"), new TestValue(10, "10"));
- partitionStorage.write(testData);
+ partitionStorage.runConsistently(() -> {
+ UUID txId = UUID.randomUUID();
- partitionStorage = storage.getPartition(0);
+ RowId rowId = partitionStorage.insert(testData, txId);
- assertThat(partitionStorage, is(notNullValue()));
+ assertThat(partitionStorage, is(sameInstance(storage.getMvPartition(0))));
+
+ assertThat(unwrap(partitionStorage.read(rowId, txId)), is(equalTo(unwrap(testData))));
- assertThat(partitionStorage.read(testData), is(equalTo(testData)));
+ return null;
+ });
}
/**
@@ -151,31 +156,33 @@ public class RocksDbTableStorageTest {
*/
@Test
void testPartitionIndependence() throws Exception {
- PartitionStorage partitionStorage0 = storage.getOrCreatePartition(42);
- // using a shifted ID value to test a multi-byte scenario
- PartitionStorage partitionStorage1 = storage.getOrCreatePartition(1 << 8);
+ MvPartitionStorage partitionStorage0 = storage.getOrCreateMvPartition(42);
+ // Using a shifted ID value to test a multibyte scenario.
+ MvPartitionStorage partitionStorage1 = storage.getOrCreateMvPartition(1 << 8);
- var testData = new SimpleDataRow("foo".getBytes(StandardCharsets.UTF_8), "bar".getBytes(StandardCharsets.UTF_8));
+ var testData0 = binaryRow(new TestKey(1, "1"), new TestValue(10, "10"));
- partitionStorage0.write(testData);
+ UUID txId = UUID.randomUUID();
- assertThat(partitionStorage0.read(testData), is(equalTo(testData)));
- assertThat(partitionStorage1.read(testData), is(nullValue()));
+ RowId rowId0 = partitionStorage0.runConsistently(() -> partitionStorage0.insert(testData0, txId));
- var testData2 = new SimpleDataRow("baz".getBytes(StandardCharsets.UTF_8), "quux".getBytes(StandardCharsets.UTF_8));
+ assertThat(unwrap(partitionStorage0.read(rowId0, txId)), is(equalTo(unwrap(testData0))));
+ assertThat(partitionStorage1.read(rowId0, txId), is(nullValue()));
- partitionStorage1.write(testData2);
+ var testData1 = binaryRow(new TestKey(2, "2"), new TestValue(20, "20"));
- assertThat(partitionStorage0.read(testData2), is(nullValue()));
- assertThat(partitionStorage1.read(testData2), is(equalTo(testData2)));
+ RowId rowId1 = partitionStorage1.runConsistently(() -> partitionStorage1.insert(testData1, txId));
- assertThat(toList(partitionStorage0.scan(row -> true)), contains(testData));
- assertThat(toList(partitionStorage1.scan(row -> true)), contains(testData2));
+ assertThat(partitionStorage0.read(rowId1, txId), is(nullValue()));
+ assertThat(unwrap(partitionStorage1.read(rowId1, txId)), is(equalTo(unwrap(testData1))));
+
+ assertThat(toList(partitionStorage0.scan(row -> true, txId)), contains(unwrap(testData0)));
+ assertThat(toList(partitionStorage1.scan(row -> true, txId)), contains(unwrap(testData1)));
}
- private static <T> List<T> toList(Cursor<T> cursor) throws Exception {
+ private List<IgniteBiTuple<TestKey, TestValue>> toList(Cursor<BinaryRow> cursor) throws Exception {
try (cursor) {
- return cursor.stream().collect(Collectors.toList());
+ return cursor.stream().map(this::unwrap).collect(Collectors.toList());
}
}
@@ -184,16 +191,23 @@ public class RocksDbTableStorageTest {
*/
@Test
void testDropPartition() {
- var testData = new SimpleDataRow("foo".getBytes(StandardCharsets.UTF_8), "bar".getBytes(StandardCharsets.UTF_8));
+ var testData = binaryRow(new TestKey(1, "1"), new TestValue(10, "10"));
+
+ UUID txId = UUID.randomUUID();
+
+ MvPartitionStorage partitionStorage0 = storage.getOrCreateMvPartition(42);
- storage.getOrCreatePartition(42).write(testData);
- storage.getOrCreatePartition(1 << 8).write(testData);
+ RowId rowId0 = partitionStorage0.runConsistently(() -> partitionStorage0.insert(testData, txId));
- storage.dropPartition(42);
+ MvPartitionStorage partitionStorage1 = storage.getOrCreateMvPartition(1 << 8);
- assertThat(storage.getPartition(42), is(nullValue()));
- assertThat(storage.getOrCreatePartition(42).read(testData), is(nullValue()));
- assertThat(storage.getPartition(1 << 8).read(testData), is(equalTo(testData)));
+ RowId rowId1 = partitionStorage1.runConsistently(() -> partitionStorage1.insert(testData, txId));
+
+ storage.destroyPartition(42);
+
+ assertThat(storage.getMvPartition(42), is(nullValue()));
+ assertThat(storage.getOrCreateMvPartition(42).read(rowId0, txId), is(nullValue()));
+ assertThat(unwrap(storage.getMvPartition(1 << 8).read(rowId1, txId)), is(equalTo(unwrap(testData))));
}
/**
@@ -206,107 +220,37 @@ public class RocksDbTableStorageTest {
polymorphicExtensions = {HashIndexConfigurationSchema.class, RocksDbDataStorageConfigurationSchema.class}
) TableConfiguration tableCfg
) throws Exception {
- var testData = new SimpleDataRow("foo".getBytes(StandardCharsets.UTF_8), "bar".getBytes(StandardCharsets.UTF_8));
+ var testData = binaryRow(new TestKey(1, "1"), new TestValue(10, "10"));
+
+ UUID txId = UUID.randomUUID();
- storage.getOrCreatePartition(0).write(testData);
+ MvPartitionStorage partitionStorage0 = storage.getOrCreateMvPartition(0);
+
+ RowId rowId0 = partitionStorage0.runConsistently(() -> partitionStorage0.insert(testData, txId));
storage.stop();
tableCfg.dataStorage().change(c -> c.convert(RocksDbDataStorageChange.class)).get(1, TimeUnit.SECONDS);
- storage = engine.createTable(tableCfg);
+ storage = engine.createMvTable(tableCfg);
storage.start();
- assertThat(storage.getPartition(0), is(notNullValue()));
- assertThat(storage.getPartition(1), is(nullValue()));
- assertThat(storage.getPartition(0).read(testData), is(equalTo(testData)));
- }
-
- /**
- * Tests that restoring a snapshot clears all previous data.
- */
- @Test
- void testRestoreSnapshot() {
- PartitionStorage partitionStorage = storage.getOrCreatePartition(0);
-
- var testData1 = new SimpleDataRow("foo".getBytes(StandardCharsets.UTF_8), "bar".getBytes(StandardCharsets.UTF_8));
- var testData2 = new SimpleDataRow("baz".getBytes(StandardCharsets.UTF_8), "quux".getBytes(StandardCharsets.UTF_8));
-
- Path snapshotDir = workDir.resolve("snapshot");
-
- partitionStorage.write(testData1);
-
- assertThat(partitionStorage.snapshot(snapshotDir), willBe(nullValue(Void.class)));
-
- partitionStorage.write(testData2);
-
- partitionStorage.restoreSnapshot(snapshotDir);
-
- assertThat(partitionStorage.read(testData1), is(testData1));
- assertThat(partitionStorage.read(testData2), is(nullValue()));
+ assertThat(storage.getMvPartition(0), is(notNullValue()));
+ assertThat(storage.getMvPartition(1), is(nullValue()));
+ assertThat(unwrap(storage.getMvPartition(0).read(rowId0, txId)), is(equalTo(unwrap(testData))));
}
- /**
- * Tests that loading snapshots for one partition does not influence data in another.
- */
@Test
- void testSnapshotIndependence() {
- PartitionStorage partitionStorage1 = storage.getOrCreatePartition(0);
- PartitionStorage partitionStorage2 = storage.getOrCreatePartition(1);
-
- var testData1 = new SimpleDataRow("foo".getBytes(StandardCharsets.UTF_8), "bar".getBytes(StandardCharsets.UTF_8));
- var testData2 = new SimpleDataRow("baz".getBytes(StandardCharsets.UTF_8), "quux".getBytes(StandardCharsets.UTF_8));
-
- partitionStorage1.writeAll(List.of(testData1, testData2));
- partitionStorage2.writeAll(List.of(testData1, testData2));
-
- // take a snapshot of the first partition
- assertThat(partitionStorage1.snapshot(workDir.resolve("snapshot")), willBe(nullValue(Void.class)));
-
- // remove all data from partitions
- partitionStorage1.removeAll(List.of(testData1, testData2));
- partitionStorage2.removeAll(List.of(testData1, testData2));
-
- assertThat(partitionStorage1.readAll(List.of(testData1, testData2)), is(empty()));
- assertThat(partitionStorage2.readAll(List.of(testData1, testData2)), is(empty()));
-
- // restore a snapshot and check that only the first partition has data
- partitionStorage1.restoreSnapshot(workDir.resolve("snapshot"));
-
- assertThat(partitionStorage1.readAll(List.of(testData1, testData2)), containsInAnyOrder(testData1, testData2));
- assertThat(partitionStorage2.readAll(List.of(testData1, testData2)), is(empty()));
+ void storageAdvertisesItIsPersistent() {
+ assertThat(storage.isVolatile(), is(false));
}
- /**
- * Tests that loading snapshots for one partition does not influence data in another when overwriting existing keys.
- */
- @Test
- void testSnapshotIndependenceOverwritesKeys() {
- PartitionStorage partitionStorage1 = storage.getOrCreatePartition(0);
- PartitionStorage partitionStorage2 = storage.getOrCreatePartition(1);
-
- var testData = new SimpleDataRow("foo".getBytes(StandardCharsets.UTF_8), "bar".getBytes(StandardCharsets.UTF_8));
-
- partitionStorage1.write(testData);
- partitionStorage2.write(testData);
-
- assertThat(partitionStorage2.snapshot(workDir.resolve("snapshot")), willBe(nullValue(Void.class)));
-
- // key is intentionally the same as testData
- var overwriteData = new SimpleDataRow(testData.keyBytes(), "new value".getBytes(StandardCharsets.UTF_8));
-
- // test that snapshot restoration overrides existing keys
- partitionStorage2.write(overwriteData);
-
- partitionStorage2.restoreSnapshot(workDir.resolve("snapshot"));
-
- assertThat(partitionStorage1.read(overwriteData), is(testData));
- assertThat(partitionStorage2.read(overwriteData), is(testData));
- }
+ private IgniteBiTuple<TestKey, TestValue> unwrap(BinaryRow binaryRow) {
+ if (binaryRow == null) {
+ return null;
+ }
- @Test
- void storageAdvertisesItIsPersistent() {
- assertThat(storage.isVolatile(), is(false));
+ return new IgniteBiTuple<>(key(binaryRow), value(binaryRow));
}
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
index ee8699231..2eac8b1f5 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
@@ -143,40 +143,36 @@ public class PartitionListener implements RaftGroupListener {
: "Pending write command has a higher index than already processed commands [commandIndex=" + commandIndex
+ ", storageAppliedIndex=" + storageAppliedIndex + ']';
- // TODO IGNITE-17081 IGNITE-17077
- // Applied index is set non-atomically. This is a wrong and non-recoverable behavior. Will be fixed later.
- storage.lastAppliedIndex(commandIndex);
-
if (command instanceof InsertCommand) {
- clo.result(handleInsertCommand((InsertCommand) command));
+ clo.result(handleInsertCommand((InsertCommand) command, commandIndex));
} else if (command instanceof DeleteCommand) {
- clo.result(handleDeleteCommand((DeleteCommand) command));
+ clo.result(handleDeleteCommand((DeleteCommand) command, commandIndex));
} else if (command instanceof ReplaceCommand) {
- clo.result(handleReplaceCommand((ReplaceCommand) command));
+ clo.result(handleReplaceCommand((ReplaceCommand) command, commandIndex));
} else if (command instanceof UpsertCommand) {
- handleUpsertCommand((UpsertCommand) command);
+ handleUpsertCommand((UpsertCommand) command, commandIndex);
clo.result(null);
} else if (command instanceof InsertAllCommand) {
- clo.result(handleInsertAllCommand((InsertAllCommand) command));
+ clo.result(handleInsertAllCommand((InsertAllCommand) command, commandIndex));
} else if (command instanceof UpsertAllCommand) {
- handleUpsertAllCommand((UpsertAllCommand) command);
+ handleUpsertAllCommand((UpsertAllCommand) command, commandIndex);
clo.result(null);
} else if (command instanceof DeleteAllCommand) {
- clo.result(handleDeleteAllCommand((DeleteAllCommand) command));
+ clo.result(handleDeleteAllCommand((DeleteAllCommand) command, commandIndex));
} else if (command instanceof DeleteExactCommand) {
- clo.result(handleDeleteExactCommand((DeleteExactCommand) command));
+ clo.result(handleDeleteExactCommand((DeleteExactCommand) command, commandIndex));
} else if (command instanceof DeleteExactAllCommand) {
- clo.result(handleDeleteExactAllCommand((DeleteExactAllCommand) command));
+ clo.result(handleDeleteExactAllCommand((DeleteExactAllCommand) command, commandIndex));
} else if (command instanceof ReplaceIfExistCommand) {
- clo.result(handleReplaceIfExistsCommand((ReplaceIfExistCommand) command));
+ clo.result(handleReplaceIfExistsCommand((ReplaceIfExistCommand) command, commandIndex));
} else if (command instanceof GetAndDeleteCommand) {
- clo.result(handleGetAndDeleteCommand((GetAndDeleteCommand) command));
+ clo.result(handleGetAndDeleteCommand((GetAndDeleteCommand) command, commandIndex));
} else if (command instanceof GetAndReplaceCommand) {
- clo.result(handleGetAndReplaceCommand((GetAndReplaceCommand) command));
+ clo.result(handleGetAndReplaceCommand((GetAndReplaceCommand) command, commandIndex));
} else if (command instanceof GetAndUpsertCommand) {
- clo.result(handleGetAndUpsertCommand((GetAndUpsertCommand) command));
+ clo.result(handleGetAndUpsertCommand((GetAndUpsertCommand) command, commandIndex));
} else if (command instanceof ScanInitCommand) {
handleScanInitCommand((CommandClosure<ScanInitCommand>) clo, (ScanInitCommand) command);
} else if (command instanceof ScanRetrieveBatchCommand) {
@@ -184,7 +180,7 @@ public class PartitionListener implements RaftGroupListener {
} else if (command instanceof ScanCloseCommand) {
handleScanCloseCommand((CommandClosure<ScanCloseCommand>) clo, (ScanCloseCommand) command);
} else if (command instanceof FinishTxCommand) {
- clo.result(handleFinishTxCommand((FinishTxCommand) command));
+ clo.result(handleFinishTxCommand((FinishTxCommand) command, commandIndex));
} else {
assert false : "Command was not found [cmd=" + command + ']';
}
@@ -243,192 +239,287 @@ public class PartitionListener implements RaftGroupListener {
* Handler for the {@link InsertCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private boolean handleInsertCommand(InsertCommand cmd) {
- return storage.insert(cmd.getRow(), cmd.getTxId());
+ private boolean handleInsertCommand(InsertCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ boolean res = storage.insert(cmd.getRow(), cmd.getTxId());
+
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return res;
+ });
}
/**
* Handler for the {@link DeleteCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private boolean handleDeleteCommand(DeleteCommand cmd) {
- return storage.delete(cmd.getRow(), cmd.getTxId());
+ private boolean handleDeleteCommand(DeleteCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ boolean res = storage.delete(cmd.getRow(), cmd.getTxId());
+
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return res;
+ });
}
/**
* Handler for the {@link ReplaceCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private boolean handleReplaceCommand(ReplaceCommand cmd) {
- return storage.replace(cmd.getOldRow(), cmd.getRow(), cmd.getTxId());
+ private boolean handleReplaceCommand(ReplaceCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ boolean res = storage.replace(cmd.getOldRow(), cmd.getRow(), cmd.getTxId());
+
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return res;
+ });
}
/**
* Handler for the {@link UpsertCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
*/
- private void handleUpsertCommand(UpsertCommand cmd) {
- storage.upsert(cmd.getRow(), cmd.getTxId());
+ private void handleUpsertCommand(UpsertCommand cmd, long commandIndex) {
+ storage.delegate().runConsistently(() -> {
+ storage.upsert(cmd.getRow(), cmd.getTxId());
+
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return null;
+ });
}
/**
* Handler for the {@link InsertAllCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private MultiRowsResponse handleInsertAllCommand(InsertAllCommand cmd) {
- Collection<BinaryRow> rows = cmd.getRows();
+ private MultiRowsResponse handleInsertAllCommand(InsertAllCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ Collection<BinaryRow> rows = cmd.getRows();
+
+ assert rows != null && !rows.isEmpty();
+
+ List<BinaryRow> res = storage.insertAll(rows, cmd.getTxId());
- assert rows != null && !rows.isEmpty();
+ storage.delegate().lastAppliedIndex(commandIndex);
- return new MultiRowsResponse(storage.insertAll(rows, cmd.getTxId()));
+ return new MultiRowsResponse(res);
+ });
}
/**
* Handler for the {@link UpsertAllCommand}.
*
* @param cmd Command.
- * @return Result.
+ * @param commandIndex Index of the RAFT command.
*/
- private void handleUpsertAllCommand(UpsertAllCommand cmd) {
- Collection<BinaryRow> rows = cmd.getRows();
+ private void handleUpsertAllCommand(UpsertAllCommand cmd, long commandIndex) {
+ storage.delegate().runConsistently(() -> {
+ Collection<BinaryRow> rows = cmd.getRows();
+
+ assert rows != null && !rows.isEmpty();
- assert rows != null && !rows.isEmpty();
+ storage.upsertAll(rows, cmd.getTxId());
- storage.upsertAll(rows, cmd.getTxId());
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return null;
+ });
}
/**
* Handler for the {@link DeleteAllCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private MultiRowsResponse handleDeleteAllCommand(DeleteAllCommand cmd) {
- Collection<BinaryRow> rows = cmd.getRows();
+ private MultiRowsResponse handleDeleteAllCommand(DeleteAllCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ Collection<BinaryRow> rows = cmd.getRows();
+
+ assert rows != null && !rows.isEmpty();
- assert rows != null && !rows.isEmpty();
+ List<BinaryRow> res = storage.deleteAll(rows, cmd.getTxId());
- return new MultiRowsResponse(storage.deleteAll(rows, cmd.getTxId()));
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return new MultiRowsResponse(res);
+ });
}
/**
* Handler for the {@link DeleteExactCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private boolean handleDeleteExactCommand(DeleteExactCommand cmd) {
- BinaryRow row = cmd.getRow();
+ private boolean handleDeleteExactCommand(DeleteExactCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ BinaryRow row = cmd.getRow();
- assert row != null;
- assert row.hasValue();
+ assert row != null;
+ assert row.hasValue();
- return storage.deleteExact(row, cmd.getTxId());
+ boolean res = storage.deleteExact(row, cmd.getTxId());
+
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return res;
+ });
}
/**
* Handler for the {@link DeleteExactAllCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private MultiRowsResponse handleDeleteExactAllCommand(DeleteExactAllCommand cmd) {
- Collection<BinaryRow> rows = cmd.getRows();
+ private MultiRowsResponse handleDeleteExactAllCommand(DeleteExactAllCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ Collection<BinaryRow> rows = cmd.getRows();
- assert rows != null && !rows.isEmpty();
+ assert rows != null && !rows.isEmpty();
- return new MultiRowsResponse(storage.deleteAllExact(rows, cmd.getTxId()));
+ List<BinaryRow> res = storage.deleteAllExact(rows, cmd.getTxId());
+
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return new MultiRowsResponse(res);
+ });
}
/**
* Handler for the {@link ReplaceIfExistCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private boolean handleReplaceIfExistsCommand(ReplaceIfExistCommand cmd) {
- BinaryRow row = cmd.getRow();
+ private boolean handleReplaceIfExistsCommand(ReplaceIfExistCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ BinaryRow row = cmd.getRow();
- assert row != null;
+ assert row != null;
- return storage.replace(row, cmd.getTxId());
+ boolean res = storage.replace(row, cmd.getTxId());
+
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return res;
+ });
}
/**
* Handler for the {@link GetAndDeleteCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private SingleRowResponse handleGetAndDeleteCommand(GetAndDeleteCommand cmd) {
- BinaryRow row = cmd.getRow();
+ private SingleRowResponse handleGetAndDeleteCommand(GetAndDeleteCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ BinaryRow row = cmd.getRow();
+
+ assert row != null;
+
+ BinaryRow res = storage.getAndDelete(row, cmd.getTxId());
- assert row != null;
+ storage.delegate().lastAppliedIndex(commandIndex);
- return new SingleRowResponse(storage.getAndDelete(row, cmd.getTxId()));
+ return new SingleRowResponse(res);
+ });
}
/**
* Handler for the {@link GetAndReplaceCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private SingleRowResponse handleGetAndReplaceCommand(GetAndReplaceCommand cmd) {
- BinaryRow row = cmd.getRow();
+ private SingleRowResponse handleGetAndReplaceCommand(GetAndReplaceCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ BinaryRow row = cmd.getRow();
+
+ assert row != null && row.hasValue();
+
+ BinaryRow res = storage.getAndReplace(row, cmd.getTxId());
- assert row != null && row.hasValue();
+ storage.delegate().lastAppliedIndex(commandIndex);
- return new SingleRowResponse(storage.getAndReplace(row, cmd.getTxId()));
+ return new SingleRowResponse(res);
+ });
}
/**
* Handler for the {@link GetAndUpsertCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private SingleRowResponse handleGetAndUpsertCommand(GetAndUpsertCommand cmd) {
- BinaryRow row = cmd.getRow();
+ private SingleRowResponse handleGetAndUpsertCommand(GetAndUpsertCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ BinaryRow row = cmd.getRow();
+
+ assert row != null && row.hasValue();
+
+ BinaryRow res = storage.getAndUpsert(row, cmd.getTxId());
- assert row != null && row.hasValue();
+ storage.delegate().lastAppliedIndex(commandIndex);
- return new SingleRowResponse(storage.getAndUpsert(row, cmd.getTxId()));
+ return new SingleRowResponse(res);
+ });
}
/**
* Handler for the {@link FinishTxCommand}.
*
* @param cmd Command.
+ * @param commandIndex Index of the RAFT command.
* @return Result.
*/
- private boolean handleFinishTxCommand(FinishTxCommand cmd) {
- UUID txId = cmd.txId();
-
- boolean stateChanged = txManager.changeState(txId, TxState.PENDING, cmd.finish() ? TxState.COMMITED : TxState.ABORTED);
-
- // This code is technically incorrect and assumes that "stateChanged" is always true. This was done because transaction state is not
- // persisted and thus FinishTxCommand couldn't be completed on recovery after node restart ("changeState" uses "replace").
- if (/*txManager.state(txId) == TxState.COMMITED*/cmd.finish()) {
- cmd.lockedKeys().getOrDefault(lockId, new ArrayList<>()).forEach(key -> {
- storage.commitWrite(ByteBuffer.wrap(key), txId);
- });
- } else /*if (txManager.state(txId) == TxState.ABORTED)*/ {
- cmd.lockedKeys().getOrDefault(lockId, new ArrayList<>()).forEach(key -> {
- storage.abortWrite(ByteBuffer.wrap(key));
- });
- }
+ private boolean handleFinishTxCommand(FinishTxCommand cmd, long commandIndex) {
+ return storage.delegate().runConsistently(() -> {
+ UUID txId = cmd.txId();
+
+ boolean stateChanged = txManager.changeState(txId, TxState.PENDING, cmd.finish() ? TxState.COMMITED : TxState.ABORTED);
+
+ // This code is technically incorrect and assumes that "stateChanged" is always true. This was done because transaction state is
+ // not persisted and thus FinishTxCommand couldn't be completed on recovery after node restart ("changeState" uses "replace").
+ if (/*txManager.state(txId) == TxState.COMMITED*/cmd.finish()) {
+ cmd.lockedKeys().getOrDefault(lockId, new ArrayList<>()).forEach(key -> {
+ storage.commitWrite(ByteBuffer.wrap(key), txId);
+ });
+ } else /*if (txManager.state(txId) == TxState.ABORTED)*/ {
+ cmd.lockedKeys().getOrDefault(lockId, new ArrayList<>()).forEach(key -> {
+ storage.abortWrite(ByteBuffer.wrap(key));
+ });
+ }
- return stateChanged;
+ storage.delegate().lastAppliedIndex(commandIndex);
+
+ return stateChanged;
+ });
}
/**
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java
index 5acefb5b8..603ae98f7 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java
@@ -26,7 +26,6 @@ import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -525,9 +524,8 @@ public class VersionedRowStore {
* @param path The path.
* @return Snapshot future.
*/
- // TODO: IGNITE-16644 Support snapshots.
public CompletionStage<Void> snapshot(Path path) {
- return CompletableFuture.completedFuture(null);
+ return storage.flush();
}
/**
@@ -535,7 +533,6 @@ public class VersionedRowStore {
*
* @param path The path.
*/
- // TODO: IGNITE-16644 Support snapshots.
public void restoreSnapshot(Path path) {
}
diff --git a/parent/pom.xml b/parent/pom.xml
index 8096496c0..ce0a518bd 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -92,7 +92,7 @@
<javassist.version>3.28.0-GA</javassist.version>
<esri.geometry.version>2.2.0</esri.geometry.version>
<checker.version>3.10.0</checker.version>
- <rocksdb.version>6.20.3</rocksdb.version>
+ <rocksdb.version>7.3.1</rocksdb.version>
<disruptor.version>3.3.7</disruptor.version>
<metrics.version>4.0.2</metrics.version>
<jctools.version>3.3.0</jctools.version>