This is an automated email from the ASF dual-hosted git repository.
rpuch 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 9238f27658 IGNITE-23683 Support format versioning when storing lease
data in RocksDB partitions (#4732)
9238f27658 is described below
commit 9238f276582e7f371bf43207b0115e30e73e0458
Author: Roman Puchkovskiy <[email protected]>
AuthorDate: Mon Nov 18 16:34:42 2024 +0400
IGNITE-23683 Support format versioning when storing lease data in RocksDB
partitions (#4732)
---
.../ignite/internal/storage/lease/LeaseInfo.java | 46 ++++++++++++++++
.../storage/lease/LeaseInfoSerializer.java | 48 +++++++++++++++++
.../storage/lease/LeaseInfoSerializerTest.java | 61 ++++++++++++++++++++++
.../storage/rocksdb/RocksDbMvPartitionStorage.java | 53 ++++++-------------
4 files changed, 170 insertions(+), 38 deletions(-)
diff --git
a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/lease/LeaseInfo.java
b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/lease/LeaseInfo.java
new file mode 100644
index 0000000000..a30f3f7f12
--- /dev/null
+++
b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/lease/LeaseInfo.java
@@ -0,0 +1,46 @@
+/*
+ * 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.lease;
+
+import java.util.UUID;
+
+/** Represents information about partition lease. */
+public class LeaseInfo {
+ private final long leaseStartTime;
+ private final UUID primaryReplicaNodeId;
+ private final String primaryReplicaNodeName;
+
+ /** Constructor. */
+ public LeaseInfo(long leaseStartTime, UUID primaryReplicaNodeId, String
primaryReplicaNodeName) {
+ this.leaseStartTime = leaseStartTime;
+ this.primaryReplicaNodeId = primaryReplicaNodeId;
+ this.primaryReplicaNodeName = primaryReplicaNodeName;
+ }
+
+ public long leaseStartTime() {
+ return leaseStartTime;
+ }
+
+ public UUID primaryReplicaNodeId() {
+ return primaryReplicaNodeId;
+ }
+
+ public String primaryReplicaNodeName() {
+ return primaryReplicaNodeName;
+ }
+}
diff --git
a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/lease/LeaseInfoSerializer.java
b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/lease/LeaseInfoSerializer.java
new file mode 100644
index 0000000000..7c2b94d0ad
--- /dev/null
+++
b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/lease/LeaseInfoSerializer.java
@@ -0,0 +1,48 @@
+/*
+ * 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.lease;
+
+import java.io.IOException;
+import java.util.UUID;
+import org.apache.ignite.internal.util.io.IgniteDataInput;
+import org.apache.ignite.internal.util.io.IgniteDataOutput;
+import org.apache.ignite.internal.versioned.VersionedSerializer;
+
+/**
+ * {@link VersionedSerializer} for {@link LeaseInfo} instances.
+ */
+public class LeaseInfoSerializer extends VersionedSerializer<LeaseInfo> {
+ /** Serializer instance. */
+ public static final LeaseInfoSerializer INSTANCE = new
LeaseInfoSerializer();
+
+ @Override
+ protected void writeExternalData(LeaseInfo leaseInfo, IgniteDataOutput
out) throws IOException {
+ out.writeLong(leaseInfo.leaseStartTime());
+ out.writeUuid(leaseInfo.primaryReplicaNodeId());
+ out.writeUTF(leaseInfo.primaryReplicaNodeName());
+ }
+
+ @Override
+ protected LeaseInfo readExternalData(byte protoVer, IgniteDataInput in)
throws IOException {
+ long leaseStartTime = in.readLong();
+ UUID primaryReplicaNodeId = in.readUuid();
+ String primaryReplicaNodeName = in.readUTF();
+
+ return new LeaseInfo(leaseStartTime, primaryReplicaNodeId,
primaryReplicaNodeName);
+ }
+}
diff --git
a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/lease/LeaseInfoSerializerTest.java
b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/lease/LeaseInfoSerializerTest.java
new file mode 100644
index 0000000000..4a1200b7da
--- /dev/null
+++
b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/lease/LeaseInfoSerializerTest.java
@@ -0,0 +1,61 @@
+/*
+ * 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.lease;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import java.util.Base64;
+import java.util.UUID;
+import org.apache.ignite.internal.versioned.VersionedSerialization;
+import org.junit.jupiter.api.Test;
+
+class LeaseInfoSerializerTest {
+ private static final String V1_SERIALIZED_BASE64 =
"Ae++Q9IEAAAAAAAAAQAAAAAAAAACAAAAAAAAAAhwcmltYXJ5";
+
+ private final LeaseInfoSerializer serializer = new LeaseInfoSerializer();
+
+ @Test
+ void serializeAndDeserialize() {
+ LeaseInfo originalInfo = new LeaseInfo(1234, new UUID(1, 2),
"primary");
+
+ byte[] bytes = VersionedSerialization.toBytes(originalInfo,
serializer);
+ LeaseInfo restoredInfo = VersionedSerialization.fromBytes(bytes,
serializer);
+
+ assertThat(restoredInfo.leaseStartTime(), is(1234L));
+ assertThat(restoredInfo.primaryReplicaNodeId(), is(new UUID(1, 2)));
+ assertThat(restoredInfo.primaryReplicaNodeName(), is("primary"));
+ }
+
+ @Test
+ void v1CanBeDeserialized() {
+ byte[] bytes = Base64.getDecoder().decode(V1_SERIALIZED_BASE64);
+ LeaseInfo restoredInfo = VersionedSerialization.fromBytes(bytes,
serializer);
+
+ assertThat(restoredInfo.leaseStartTime(), is(1234L));
+ assertThat(restoredInfo.primaryReplicaNodeId(), is(new UUID(1, 2)));
+ assertThat(restoredInfo.primaryReplicaNodeName(), is("primary"));
+ }
+
+ @SuppressWarnings("unused")
+ private String v1SerializedBase64() {
+ LeaseInfo originalInfo = new LeaseInfo(1234, new UUID(1, 2),
"primary");
+ byte[] v1Bytes = VersionedSerialization.toBytes(originalInfo,
serializer);
+ return Base64.getEncoder().encodeToString(v1Bytes);
+ }
+}
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 60229f23ff..cabfb70810 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
@@ -48,19 +48,11 @@ import static
org.apache.ignite.internal.storage.util.StorageUtils.throwExceptio
import static
org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionDependingOnStorageStateOnRebalance;
import static
org.apache.ignite.internal.storage.util.StorageUtils.throwExceptionIfStorageInProgressOfRebalance;
import static
org.apache.ignite.internal.storage.util.StorageUtils.transitionToTerminalState;
-import static org.apache.ignite.internal.util.ByteUtils.bytesToInt;
import static org.apache.ignite.internal.util.ByteUtils.bytesToLong;
-import static org.apache.ignite.internal.util.ByteUtils.bytesToUuid;
-import static org.apache.ignite.internal.util.ByteUtils.intToBytes;
import static org.apache.ignite.internal.util.ByteUtils.longToBytes;
-import static org.apache.ignite.internal.util.ByteUtils.stringToBytes;
-import static org.apache.ignite.internal.util.ByteUtils.uuidToBytes;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
-import java.nio.charset.StandardCharsets;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.UUID;
@@ -81,12 +73,15 @@ import
org.apache.ignite.internal.storage.StorageRebalanceException;
import org.apache.ignite.internal.storage.TxIdMismatchException;
import org.apache.ignite.internal.storage.engine.MvPartitionMeta;
import org.apache.ignite.internal.storage.gc.GcEntry;
+import org.apache.ignite.internal.storage.lease.LeaseInfo;
+import org.apache.ignite.internal.storage.lease.LeaseInfoSerializer;
import org.apache.ignite.internal.storage.rocksdb.GarbageCollector.AddResult;
import org.apache.ignite.internal.storage.util.LocalLocker;
import org.apache.ignite.internal.storage.util.StorageState;
import org.apache.ignite.internal.tx.TransactionIds;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.internal.versioned.VersionedSerialization;
import org.jetbrains.annotations.Nullable;
import org.rocksdb.AbstractWriteBatch;
import org.rocksdb.ColumnFamilyHandle;
@@ -160,8 +155,6 @@ public class RocksDbMvPartitionStorage implements
MvPartitionStorage {
private static final ThreadLocal<ByteBuffer> TX_STATE_BUFFER =
withInitial(() ->
allocate(DATA_ID_WITH_TX_STATE_SIZE).order(KEY_BYTE_ORDER));
- private static final int UUID_LENGTH_IN_BYTES = 2 * Long.BYTES;
-
/** Table storage instance. */
private final RocksDbTableStorage tableStorage;
@@ -263,17 +256,11 @@ public class RocksDbMvPartitionStorage implements
MvPartitionStorage {
if (leaseBytes == null) {
leaseStartTime = HybridTimestamp.MIN_VALUE.longValue();
} else {
- leaseStartTime = bytesToLong(leaseBytes);
-
- primaryReplicaNodeId = bytesToUuid(leaseBytes, Long.BYTES);
+ LeaseInfo leaseInfo =
VersionedSerialization.fromBytes(leaseBytes, LeaseInfoSerializer.INSTANCE);
- int primaryReplicaNodeNameLength = bytesToInt(leaseBytes,
Long.BYTES + UUID_LENGTH_IN_BYTES);
- primaryReplicaNodeName = new String(
- leaseBytes,
- Long.BYTES + UUID_LENGTH_IN_BYTES + Integer.BYTES,
- primaryReplicaNodeNameLength,
- StandardCharsets.UTF_8
- );
+ leaseStartTime = leaseInfo.leaseStartTime();
+ primaryReplicaNodeId = leaseInfo.primaryReplicaNodeId();
+ primaryReplicaNodeName = leaseInfo.primaryReplicaNodeName();
}
byte[] estimatedSizeBytes = db.get(meta, readOpts,
estimatedSizeKey);
@@ -1125,28 +1112,18 @@ public class RocksDbMvPartitionStorage implements
MvPartitionStorage {
}
private void saveLease(AbstractWriteBatch writeBatch, long leaseStartTime,
UUID primaryReplicaNodeId, String primaryReplicaNodeName) {
- // TODO: IGNITE-23683 - switch to VersionedSerialization.
- try {
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
- outputStream.write(longToBytes(leaseStartTime));
-
- byte[] primaryReplicaNodeIdBytes =
uuidToBytes(primaryReplicaNodeId);
- outputStream.write(primaryReplicaNodeIdBytes);
-
- byte[] primaryReplicaNodeNameBytes =
stringToBytes(primaryReplicaNodeName);
- outputStream.write(intToBytes(primaryReplicaNodeNameBytes.length));
- outputStream.write(primaryReplicaNodeNameBytes);
+ LeaseInfo leaseInfo = new LeaseInfo(leaseStartTime,
primaryReplicaNodeId, primaryReplicaNodeName);
+ byte[] bytes = VersionedSerialization.toBytes(leaseInfo,
LeaseInfoSerializer.INSTANCE);
- writeBatch.put(meta, leaseKey, outputStream.toByteArray());
-
- this.leaseStartTime = leaseStartTime;
- this.primaryReplicaNodeId = primaryReplicaNodeId;
- this.primaryReplicaNodeName = primaryReplicaNodeName;
- } catch (IOException e) {
- throw new StorageException(e);
+ try {
+ writeBatch.put(meta, leaseKey, bytes);
} catch (RocksDBException e) {
throw new IgniteRocksDbException(e);
}
+
+ this.leaseStartTime = leaseStartTime;
+ this.primaryReplicaNodeId = primaryReplicaNodeId;
+ this.primaryReplicaNodeName = primaryReplicaNodeName;
}
@Override
