You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by ha...@apache.org on 2020/10/01 21:02:23 UTC
[hadoop-ozone] branch master updated: HDDS-3869. Use different
column families for datanode block and metadata (#1298)
This is an automated email from the ASF dual-hosted git repository.
hanishakoneru pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 275653e HDDS-3869. Use different column families for datanode block and metadata (#1298)
275653e is described below
commit 275653e0a997852be44ed55ccd8dabcc255cb399
Author: Ethan Rose <33...@users.noreply.github.com>
AuthorDate: Thu Oct 1 17:02:07 2020 -0400
HDDS-3869. Use different column families for datanode block and metadata (#1298)
---
.../java/org/apache/hadoop/ozone/OzoneConsts.java | 34 +-
.../container/common/helpers/ChunkInfoList.java | 56 ++
.../container/common/impl/ContainerDataYaml.java | 3 +
.../container/common/interfaces/BlockIterator.java | 5 +-
.../container/common/interfaces/Container.java | 7 -
.../commandhandler/DeleteBlocksCommandHandler.java | 69 ++-
.../container/common/utils/ContainerCache.java | 34 +-
.../container/common/utils/ReferenceCountedDB.java | 10 +-
.../container/keyvalue/KeyValueBlockIterator.java | 156 -----
.../container/keyvalue/KeyValueContainer.java | 26 +-
.../container/keyvalue/KeyValueContainerCheck.java | 12 +-
.../container/keyvalue/KeyValueContainerData.java | 48 +-
.../container/keyvalue/helpers/BlockUtils.java | 2 +-
.../keyvalue/helpers/KeyValueContainerUtil.java | 194 ++++---
.../container/keyvalue/impl/BlockManagerImpl.java | 65 ++-
.../background/BlockDeletingService.java | 53 +-
.../metadata/AbstractDatanodeDBDefinition.java | 74 +++
.../container/metadata/AbstractDatanodeStore.java | 297 ++++++++++
.../ozone/container/metadata/BlockDataCodec.java | 47 ++
.../container/metadata/ChunkInfoListCodec.java | 45 ++
.../metadata/DatanodeSchemaOneDBDefinition.java | 91 +++
.../metadata/DatanodeSchemaTwoDBDefinition.java | 81 +++
.../ozone/container/metadata/DatanodeStore.java | 94 ++++
.../metadata/DatanodeStoreSchemaOneImpl.java | 49 ++
.../metadata/DatanodeStoreSchemaTwoImpl.java | 44 ++
.../ozone/container/metadata/DatanodeTable.java | 130 +++++
.../metadata/SchemaOneChunkInfoListCodec.java | 68 +++
.../metadata/SchemaOneDeletedBlocksTable.java | 180 ++++++
.../container/metadata/SchemaOneKeyCodec.java | 106 ++++
.../ozone/container/metadata/package-info.java | 22 +
.../container/common/TestBlockDeletingService.java | 110 +++-
.../ozone/container/common/TestContainerCache.java | 25 +-
.../common/TestKeyValueContainerData.java | 3 +
.../TestSchemaOneBackwardsCompatibility.java | 626 +++++++++++++++++++++
.../common/impl/TestContainerDataYaml.java | 3 +
.../keyvalue/TestKeyValueBlockIterator.java | 340 +++++++----
.../container/keyvalue/TestKeyValueContainer.java | 75 +--
.../keyvalue/TestKeyValueContainerCheck.java | 18 +-
.../container/ozoneimpl/TestContainerReader.java | 43 +-
.../container/ozoneimpl/TestOzoneContainer.java | 14 +-
.../test/resources/123-dn-container.db/000024.sst | Bin 0 -> 1022 bytes
.../test/resources/123-dn-container.db/000026.sst | Bin 0 -> 827 bytes
.../test/resources/123-dn-container.db/000032.sst | Bin 0 -> 896 bytes
.../test/resources/123-dn-container.db/000034.log | 0
.../src/test/resources/123-dn-container.db/CURRENT | 1 +
.../test/resources/123-dn-container.db/IDENTITY | 1 +
.../resources/123-dn-container.db/MANIFEST-000033 | Bin 0 -> 297 bytes
.../resources/123-dn-container.db/OPTIONS-000033 | 165 ++++++
.../resources/123-dn-container.db/OPTIONS-000036 | 165 ++++++
.../src/test/resources/123.container | 10 +
.../hadoop/hdds/utils/MetadataKeyFilters.java | 42 +-
.../apache/hadoop/hdds/utils/db/DBDefinition.java | 14 +-
.../org/apache/hadoop/hdds/utils/db/DBStore.java | 8 +-
.../hadoop/hdds/utils/db/DBStoreBuilder.java | 41 +-
.../org/apache/hadoop/hdds/utils/db/RDBStore.java | 17 +-
.../org/apache/hadoop/hdds/utils/db/RDBTable.java | 92 +++
.../org/apache/hadoop/hdds/utils/db/Table.java | 56 ++
.../apache/hadoop/hdds/utils/db/TypedTable.java | 46 ++
.../src/main/proto/DatanodeClientProtocol.proto | 4 +
hadoop-hdds/pom.xml | 2 +
.../ozone/TestStorageContainerManagerHelper.java | 31 +-
.../client/rpc/TestOzoneRpcClientAbstract.java | 21 +-
.../client/rpc/TestValidateBCSIDOnRestart.java | 8 +-
.../commandhandler/TestBlockDeletion.java | 51 +-
.../ozone/om/ratis/OzoneManagerStateMachine.java | 2 +-
65 files changed, 3396 insertions(+), 740 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
index d980065..a7aca16 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.ozone;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.ratis.thirdparty.io.grpc.Context;
@@ -139,30 +138,17 @@ public final class OzoneConsts {
}
}
+ // Block ID prefixes used in datanode containers.
public static final String DELETING_KEY_PREFIX = "#deleting#";
- public static final String DELETED_KEY_PREFIX = "#deleted#";
- public static final String DELETE_TRANSACTION_KEY_PREFIX = "#delTX#";
- public static final String BLOCK_COMMIT_SEQUENCE_ID_PREFIX = "#BCSID";
+ // Metadata keys for datanode containers.
+ public static final String DELETE_TRANSACTION_KEY = "#delTX";
+ public static final String BLOCK_COMMIT_SEQUENCE_ID = "#BCSID";
public static final String BLOCK_COUNT = "#BLOCKCOUNT";
public static final String CONTAINER_BYTES_USED = "#BYTESUSED";
public static final String PENDING_DELETE_BLOCK_COUNT =
"#PENDINGDELETEBLOCKCOUNT";
-
- public static final byte[] DB_BLOCK_COUNT_KEY =
- StringUtils.string2Bytes(OzoneConsts.BLOCK_COUNT);
- public static final byte[] DB_CONTAINER_BYTES_USED_KEY =
- StringUtils.string2Bytes(OzoneConsts.CONTAINER_BYTES_USED);
- public static final byte[] DB_PENDING_DELETE_BLOCK_COUNT_KEY =
- StringUtils.string2Bytes(PENDING_DELETE_BLOCK_COUNT);
- public static final byte[] DB_CONTAINER_DELETE_TRANSACTION_KEY =
- StringUtils.string2Bytes(DELETE_TRANSACTION_KEY_PREFIX);
- public static final byte[] DB_BLOCK_COMMIT_SEQUENCE_ID_KEY =
- StringUtils.string2Bytes(BLOCK_COMMIT_SEQUENCE_ID_PREFIX);
-
-
-
/**
* OM LevelDB prefixes.
*
@@ -258,6 +244,18 @@ public final class OzoneConsts {
public static final String DATA_SCAN_TIMESTAMP = "dataScanTimestamp";
public static final String ORIGIN_PIPELINE_ID = "originPipelineId";
public static final String ORIGIN_NODE_ID = "originNodeId";
+ public static final String SCHEMA_VERSION = "schemaVersion";
+
+ // Supported .container datanode schema versions.
+ // Since containers in older schema versions are currently not reformatted to
+ // newer schema versions, a datanode may have containers with a mix of schema
+ // versions, requiring this property to be tracked on a per container basis.
+ // V1: All data in default column family.
+ public static final String SCHEMA_V1 = "1";
+ // V2: Metadata, block data, and deleted blocks in their own column families.
+ public static final String SCHEMA_V2 = "2";
+ // Most recent schema version that all new containers should be created with.
+ public static final String SCHEMA_LATEST = SCHEMA_V2;
// Supported store types.
public static final String OZONE = "ozone";
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkInfoList.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkInfoList.java
new file mode 100644
index 0000000..3d6ecd8
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/ChunkInfoList.java
@@ -0,0 +1,56 @@
+/*
+ * 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.hadoop.ozone.container.common.helpers;
+
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Helper class to convert between protobuf lists and Java lists of
+ * {@link ContainerProtos.ChunkInfo} objects.
+ */
+public class ChunkInfoList {
+ private List<ContainerProtos.ChunkInfo> chunks;
+
+ public ChunkInfoList(List<ContainerProtos.ChunkInfo> chunks) {
+ this.chunks = chunks;
+ }
+
+ public List<ContainerProtos.ChunkInfo> asList() {
+ return Collections.unmodifiableList(chunks);
+ }
+
+ /**
+ * @return A new {@link ChunkInfoList} created from protobuf data.
+ */
+ public static ChunkInfoList getFromProtoBuf(
+ ContainerProtos.ChunkInfoList chunksProto) {
+ return new ChunkInfoList(chunksProto.getChunksList());
+ }
+
+ /**
+ * @return A protobuf message of this object.
+ */
+ public ContainerProtos.ChunkInfoList getProtoBufMessage() {
+ return ContainerProtos.ChunkInfoList.newBuilder()
+ .addAllChunks(this.chunks)
+ .build();
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java
index e721693..74cbbc0 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java
@@ -280,6 +280,9 @@ public final class ContainerDataYaml {
String state = (String) nodes.get(OzoneConsts.STATE);
kvData
.setState(ContainerProtos.ContainerDataProto.State.valueOf(state));
+ String schemaVersion = (String) nodes.get(OzoneConsts.SCHEMA_VERSION);
+ kvData.setSchemaVersion(schemaVersion);
+
return kvData;
}
}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/BlockIterator.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/BlockIterator.java
index f6931e3..9cabb89 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/BlockIterator.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/BlockIterator.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.ozone.container.common.interfaces;
+import java.io.Closeable;
import java.io.IOException;
import java.util.NoSuchElementException;
@@ -27,7 +28,7 @@ import java.util.NoSuchElementException;
* interface.
* @param <T>
*/
-public interface BlockIterator<T> {
+public interface BlockIterator<T> extends Closeable {
/**
* This checks if iterator has next element. If it has returns true,
@@ -52,6 +53,4 @@ public interface BlockIterator<T> {
* @throws IOException
*/
T nextBlock() throws IOException, NoSuchElementException;
-
-
}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java
index 9d90ace..67a7a16 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java
@@ -127,13 +127,6 @@ public interface Container<CONTAINERDATA extends ContainerData> extends RwLock {
void updateDeleteTransactionId(long deleteTransactionId);
/**
- * Returns blockIterator for the container.
- * @return BlockIterator
- * @throws IOException
- */
- BlockIterator blockIterator() throws IOException;
-
- /**
* Import the container from an external archive.
*/
void importContainerData(InputStream stream,
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java
index bd5e7b7..ebccf4d 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java
@@ -16,8 +16,6 @@
*/
package org.apache.hadoop.ozone.container.common.statemachine.commandhandler;
-import com.google.common.primitives.Longs;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.protocol.proto
@@ -31,7 +29,10 @@ import org.apache.hadoop.hdds.protocol.proto
.DeleteBlockTransactionResult;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
import org.apache.hadoop.ozone.container.common.helpers
.DeletedContainerBlocksSummary;
import org.apache.hadoop.ozone.container.common.interfaces.Container;
@@ -47,7 +48,7 @@ import org.apache.hadoop.ozone.protocol.commands.DeleteBlockCommandStatus;
import org.apache.hadoop.ozone.protocol.commands.DeleteBlocksCommand;
import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
import org.apache.hadoop.util.Time;
-import org.apache.hadoop.hdds.utils.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,8 +59,6 @@ import java.util.function.Consumer;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
.Result.CONTAINER_NOT_FOUND;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_CONTAINER_DELETE_TRANSACTION_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_PENDING_DELETE_BLOCK_COUNT_KEY;
/**
* Handle block deletion commands.
@@ -209,30 +208,35 @@ public class DeleteBlocksCommandHandler implements CommandHandler {
int newDeletionBlocks = 0;
try(ReferenceCountedDB containerDB =
BlockUtils.getDB(containerData, conf)) {
- for (Long blk : delTX.getLocalIDList()) {
- BatchOperation batch = new BatchOperation();
- byte[] blkBytes = Longs.toByteArray(blk);
- byte[] blkInfo = containerDB.getStore().get(blkBytes);
+ Table<String, BlockData> blockDataTable =
+ containerDB.getStore().getBlockDataTable();
+ Table<String, ChunkInfoList> deletedBlocksTable =
+ containerDB.getStore().getDeletedBlocksTable();
+
+ for (Long blkLong : delTX.getLocalIDList()) {
+ String blk = blkLong.toString();
+ BatchOperation batch = containerDB.getStore()
+ .getBatchHandler().initBatchOperation();
+ BlockData blkInfo = blockDataTable.get(blk);
if (blkInfo != null) {
- byte[] deletingKeyBytes =
- StringUtils.string2Bytes(OzoneConsts.DELETING_KEY_PREFIX + blk);
- byte[] deletedKeyBytes =
- StringUtils.string2Bytes(OzoneConsts.DELETED_KEY_PREFIX + blk);
- if (containerDB.getStore().get(deletingKeyBytes) != null
- || containerDB.getStore().get(deletedKeyBytes) != null) {
+ String deletingKey = OzoneConsts.DELETING_KEY_PREFIX + blk;
+
+ if (blockDataTable.get(deletingKey) != null
+ || deletedBlocksTable.get(blk) != null) {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format(
- "Ignoring delete for block %d in container %d."
+ "Ignoring delete for block %s in container %d."
+ " Entry already added.", blk, containerId));
}
continue;
}
// Found the block in container db,
// use an atomic update to change its state to deleting.
- batch.put(deletingKeyBytes, blkInfo);
- batch.delete(blkBytes);
+ blockDataTable.putWithBatch(batch, deletingKey, blkInfo);
+ blockDataTable.deleteWithBatch(batch, blk);
try {
- containerDB.getStore().writeBatch(batch);
+ containerDB.getStore().getBatchHandler()
+ .commitBatchOperation(batch);
newDeletionBlocks++;
if (LOG.isDebugEnabled()) {
LOG.debug("Transited Block {} to DELETING state in container {}",
@@ -255,22 +259,27 @@ public class DeleteBlocksCommandHandler implements CommandHandler {
if (newDeletionBlocks > 0) {
// Finally commit the DB counters.
- BatchOperation batchOperation = new BatchOperation();
-
+ BatchOperation batchOperation = containerDB.getStore().getBatchHandler()
+ .initBatchOperation();
+ Table<String, Long> metadataTable = containerDB.getStore()
+ .getMetadataTable();
+
// In memory is updated only when existing delete transactionID is
// greater.
if (delTX.getTxID() > containerData.getDeleteTransactionId()) {
// Update in DB pending delete key count and delete transaction ID.
- batchOperation.put(DB_CONTAINER_DELETE_TRANSACTION_KEY,
- Longs.toByteArray(delTX.getTxID()));
+ metadataTable.putWithBatch(batchOperation,
+ OzoneConsts.DELETE_TRANSACTION_KEY, delTX.getTxID());
}
-
- batchOperation.put(DB_PENDING_DELETE_BLOCK_COUNT_KEY, Longs.toByteArray(
- containerData.getNumPendingDeletionBlocks() + newDeletionBlocks));
-
- containerDB.getStore().writeBatch(batchOperation);
-
-
+
+ long pendingDeleteBlocks = containerData.getNumPendingDeletionBlocks() +
+ newDeletionBlocks;
+ metadataTable.putWithBatch(batchOperation,
+ OzoneConsts.PENDING_DELETE_BLOCK_COUNT, pendingDeleteBlocks);
+
+ containerDB.getStore().getBatchHandler()
+ .commitBatchOperation(batchOperation);
+
// update pending deletion blocks count and delete transaction ID in
// in-memory container status
containerData.updateDeleteTransactionId(delTX.getTxID());
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ContainerCache.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ContainerCache.java
index f4d8f43..df5db9f 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ContainerCache.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ContainerCache.java
@@ -18,20 +18,21 @@
package org.apache.hadoop.ozone.container.common.utils;
-import java.io.File;
import java.io.IOException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import com.google.common.util.concurrent.Striped;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
-import org.apache.hadoop.hdds.utils.MetadataStore;
-import org.apache.hadoop.hdds.utils.MetadataStoreBuilder;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import com.google.common.base.Preconditions;
import org.apache.commons.collections.MapIterator;
import org.apache.commons.collections.map.LRUMap;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStore;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStoreSchemaOneImpl;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStoreSchemaTwoImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -115,11 +116,14 @@ public final class ContainerCache extends LRUMap {
* @param containerID - ID of the container.
* @param containerDBType - DB type of the container.
* @param containerDBPath - DB path of the container.
+ * @param schemaVersion - Schema version of the container.
* @param conf - Hadoop Configuration.
* @return ReferenceCountedDB.
*/
public ReferenceCountedDB getDB(long containerID, String containerDBType,
- String containerDBPath, ConfigurationSource conf)
+ String containerDBPath,
+ String schemaVersion,
+ ConfigurationSource conf)
throws IOException {
Preconditions.checkState(containerID >= 0,
"Container ID cannot be negative.");
@@ -139,14 +143,20 @@ public final class ContainerCache extends LRUMap {
}
try {
- MetadataStore metadataStore =
- MetadataStoreBuilder.newBuilder()
- .setDbFile(new File(containerDBPath))
- .setCreateIfMissing(false)
- .setConf(conf)
- .setDBType(containerDBType)
- .build();
- db = new ReferenceCountedDB(metadataStore, containerDBPath);
+ DatanodeStore store;
+
+ if (schemaVersion.equals(OzoneConsts.SCHEMA_V1)) {
+ store = new DatanodeStoreSchemaOneImpl(conf,
+ containerID, containerDBPath);
+ } else if (schemaVersion.equals(OzoneConsts.SCHEMA_V2)) {
+ store = new DatanodeStoreSchemaTwoImpl(conf,
+ containerID, containerDBPath);
+ } else {
+ throw new IllegalArgumentException(
+ "Unrecognized database schema version: " + schemaVersion);
+ }
+
+ db = new ReferenceCountedDB(store, containerDBPath);
} catch (Exception e) {
LOG.error("Error opening DB. Container:{} ContainerPath:{}",
containerID, containerDBPath, e);
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ReferenceCountedDB.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ReferenceCountedDB.java
index 5ab91a8..87e4bff 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ReferenceCountedDB.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/ReferenceCountedDB.java
@@ -21,7 +21,7 @@ package org.apache.hadoop.ozone.container.common.utils;
import com.google.common.base.Preconditions;
import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.hadoop.hdds.utils.MetadataStore;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,10 +39,10 @@ public class ReferenceCountedDB implements Closeable {
private static final Logger LOG =
LoggerFactory.getLogger(ReferenceCountedDB.class);
private final AtomicInteger referenceCount;
- private final MetadataStore store;
+ private final DatanodeStore store;
private final String containerDBPath;
- public ReferenceCountedDB(MetadataStore store, String containerDBPath) {
+ public ReferenceCountedDB(DatanodeStore store, String containerDBPath) {
this.referenceCount = new AtomicInteger(0);
this.store = store;
this.containerDBPath = containerDBPath;
@@ -76,7 +76,7 @@ public class ReferenceCountedDB implements Closeable {
referenceCount.get());
}
try {
- store.close();
+ store.stop();
return true;
} catch (Exception e) {
LOG.error("Error closing DB. Container: " + containerDBPath, e);
@@ -87,7 +87,7 @@ public class ReferenceCountedDB implements Closeable {
}
}
- public MetadataStore getStore() {
+ public DatanodeStore getStore() {
return store;
}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueBlockIterator.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueBlockIterator.java
deleted file mode 100644
index 9a945af..0000000
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueBlockIterator.java
+++ /dev/null
@@ -1,156 +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.hadoop.ozone.container.keyvalue;
-
-import org.apache.hadoop.hdds.annotation.InterfaceAudience;
-import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.ozone.OzoneConsts;
-import org.apache.hadoop.ozone.container.common.helpers.BlockData;
-import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
-import org.apache.hadoop.ozone.container.common.impl.ContainerData;
-import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml;
-import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
-import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
-import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerLocationUtil;
-import org.apache.hadoop.hdds.utils.MetaStoreIterator;
-import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
-import org.apache.hadoop.hdds.utils.MetadataKeyFilters.KeyPrefixFilter;
-import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
-import org.apache.hadoop.hdds.utils.MetadataStore.KeyValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Closeable;
-import java.io.File;
-import java.io.IOException;
-import java.util.NoSuchElementException;
-
-
-/**
- * Block Iterator for KeyValue Container. This block iterator returns blocks
- * which match with the {@link MetadataKeyFilters.KeyPrefixFilter}. If no
- * filter is specified, then default filter used is
- * {@link MetadataKeyFilters#getNormalKeyFilter()}
- */
-@InterfaceAudience.Public
-public class KeyValueBlockIterator implements BlockIterator<BlockData>,
- Closeable {
-
- private static final Logger LOG = LoggerFactory.getLogger(
- KeyValueBlockIterator.class);
-
- private MetaStoreIterator<KeyValue> blockIterator;
- private final ReferenceCountedDB db;
- private static KeyPrefixFilter defaultBlockFilter = MetadataKeyFilters
- .getNormalKeyFilter();
- private KeyPrefixFilter blockFilter;
- private BlockData nextBlock;
- private long containerId;
-
- /**
- * KeyValueBlockIterator to iterate blocks in a container.
- * @param id - container id
- * @param path - container base path
- * @throws IOException
- */
-
- public KeyValueBlockIterator(long id, File path)
- throws IOException {
- this(id, path, defaultBlockFilter);
- }
-
- /**
- * KeyValueBlockIterator to iterate blocks in a container.
- * @param id - container id
- * @param path - container base path
- * @param filter - Block filter, filter to be applied for blocks
- * @throws IOException
- */
- public KeyValueBlockIterator(long id, File path, KeyPrefixFilter filter)
- throws IOException {
- containerId = id;
- File metdataPath = new File(path, OzoneConsts.METADATA);
- File containerFile = ContainerUtils.getContainerFile(metdataPath
- .getParentFile());
- ContainerData containerData = ContainerDataYaml.readContainerFile(
- containerFile);
- KeyValueContainerData keyValueContainerData = (KeyValueContainerData)
- containerData;
- keyValueContainerData.setDbFile(KeyValueContainerLocationUtil
- .getContainerDBFile(metdataPath, containerId));
- db = BlockUtils.getDB(keyValueContainerData, new
- OzoneConfiguration());
- blockIterator = db.getStore().iterator();
- blockFilter = filter;
- }
-
- /**
- * This method returns blocks matching with the filter.
- * @return next block or null if no more blocks
- * @throws IOException
- */
- @Override
- public BlockData nextBlock() throws IOException, NoSuchElementException {
- if (nextBlock != null) {
- BlockData currentBlock = nextBlock;
- nextBlock = null;
- return currentBlock;
- }
- if(hasNext()) {
- return nextBlock();
- }
- throw new NoSuchElementException("Block Iterator reached end for " +
- "ContainerID " + containerId);
- }
-
- @Override
- public boolean hasNext() throws IOException {
- if (nextBlock != null) {
- return true;
- }
- while (blockIterator.hasNext()) {
- KeyValue block = blockIterator.next();
- if (blockFilter.filterKey(null, block.getKey(), null)) {
- nextBlock = BlockUtils.getBlockData(block.getValue());
- if (LOG.isTraceEnabled()) {
- LOG.trace("Block matching with filter found: blockID is : {} for " +
- "containerID {}", nextBlock.getLocalID(), containerId);
- }
- return true;
- }
- }
- return false;
- }
-
- @Override
- public void seekToFirst() {
- nextBlock = null;
- blockIterator.seekToFirst();
- }
-
- @Override
- public void seekToLast() {
- nextBlock = null;
- blockIterator.seekToLast();
- }
-
- public void close() {
- db.close();
- }
-}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
index 533cb34..1fff749 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java
@@ -87,10 +87,10 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
public KeyValueContainer(KeyValueContainerData containerData,
ConfigurationSource
ozoneConfig) {
- Preconditions.checkNotNull(containerData, "KeyValueContainerData cannot " +
- "be null");
- Preconditions.checkNotNull(ozoneConfig, "Ozone configuration cannot " +
- "be null");
+ Preconditions.checkNotNull(containerData,
+ "KeyValueContainerData cannot be null");
+ Preconditions.checkNotNull(ozoneConfig,
+ "Ozone configuration cannot be null");
this.config = ozoneConfig;
this.containerData = containerData;
}
@@ -126,8 +126,13 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
//Create Metadata path chunks path and metadata db
File dbFile = getContainerDBFile();
- KeyValueContainerUtil.createContainerMetaData(containerMetaDataPath,
- chunksPath, dbFile, config);
+
+ // This method is only called when creating new containers.
+ // Therefore, always use the newest schema version.
+ containerData.setSchemaVersion(OzoneConsts.SCHEMA_LATEST);
+ KeyValueContainerUtil.createContainerMetaData(containerID,
+ containerMetaDataPath, chunksPath, dbFile,
+ containerData.getSchemaVersion(), config);
//Set containerData for the KeyValueContainer.
containerData.setChunksPath(chunksPath.getPath());
@@ -379,7 +384,7 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
private void flushAndSyncDB() throws StorageContainerException {
try {
try (ReferenceCountedDB db = BlockUtils.getDB(containerData, config)) {
- db.getStore().flushDB(true);
+ db.getStore().flushLog(true);
LOG.info("Container {} is synced with bcsId {}.",
containerData.getContainerID(),
containerData.getBlockCommitSequenceId());
@@ -451,12 +456,6 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
}
@Override
- public KeyValueBlockIterator blockIterator() throws IOException{
- return new KeyValueBlockIterator(containerData.getContainerID(), new File(
- containerData.getContainerPath()));
- }
-
- @Override
public void importContainerData(InputStream input,
ContainerPacker<KeyValueContainerData> packer) throws IOException {
writeLock();
@@ -488,6 +487,7 @@ public class KeyValueContainer implements Container<KeyValueContainerData> {
containerData.setState(originalContainerData.getState());
containerData
.setContainerDBType(originalContainerData.getContainerDBType());
+ containerData.setSchemaVersion(originalContainerData.getSchemaVersion());
//rewriting the yaml file with new checksum calculation.
update(originalContainerData.getMetadata(), true);
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerCheck.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerCheck.java
index d6c4ff0..20c1455 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerCheck.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerCheck.java
@@ -19,7 +19,6 @@
package org.apache.hadoop.ozone.container.keyvalue;
import com.google.common.base.Preconditions;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdfs.util.Canceler;
@@ -32,6 +31,7 @@ import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
import org.apache.hadoop.ozone.container.common.impl.ChunkLayOutVersion;
import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
import org.apache.hadoop.ozone.container.keyvalue.helpers.ChunkUtils;
import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerLocationUtil;
@@ -232,8 +232,7 @@ public class KeyValueContainerCheck {
try(ReferenceCountedDB db =
BlockUtils.getDB(onDiskContainerData, checkConfig);
- KeyValueBlockIterator kvIter = new KeyValueBlockIterator(containerID,
- new File(onDiskContainerData.getContainerPath()))) {
+ BlockIterator<BlockData> kvIter = db.getStore().getBlockIterator()) {
while(kvIter.hasNext()) {
BlockData block = kvIter.nextBlock();
@@ -243,8 +242,11 @@ public class KeyValueContainerCheck {
if (!chunkFile.exists()) {
// concurrent mutation in Block DB? lookup the block again.
- byte[] bdata = db.getStore().get(
- Longs.toByteArray(block.getBlockID().getLocalID()));
+ String localBlockID =
+ Long.toString(block.getBlockID().getLocalID());
+ BlockData bdata = db.getStore()
+ .getBlockDataTable()
+ .get(localBlockID);
if (bdata != null) {
throw new IOException("Missing chunk file "
+ chunkFile.getAbsolutePath());
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerData.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerData.java
index 68f01fb..5c20e6e 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerData.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerData.java
@@ -24,11 +24,11 @@ import com.google.common.collect.Lists;
import java.io.IOException;
import java.util.Collections;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos
.ContainerDataProto;
-import org.apache.hadoop.hdds.utils.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.container.common.impl.ChunkLayOutVersion;
import org.apache.hadoop.ozone.container.common.impl.ContainerData;
import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
@@ -42,12 +42,13 @@ import java.util.concurrent.atomic.AtomicLong;
import static java.lang.Math.max;
import static org.apache.hadoop.ozone.OzoneConsts.CONTAINER_DB_TYPE_ROCKSDB;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_BLOCK_COUNT_KEY;
import static org.apache.hadoop.ozone.OzoneConsts.CHUNKS_PATH;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_CONTAINER_BYTES_USED_KEY;
import static org.apache.hadoop.ozone.OzoneConsts.CONTAINER_DB_TYPE;
import static org.apache.hadoop.ozone.OzoneConsts.METADATA_PATH;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_PENDING_DELETE_BLOCK_COUNT_KEY;
+import static org.apache.hadoop.ozone.OzoneConsts.SCHEMA_VERSION;
+import static org.apache.hadoop.ozone.OzoneConsts.CONTAINER_BYTES_USED;
+import static org.apache.hadoop.ozone.OzoneConsts.BLOCK_COUNT;
+import static org.apache.hadoop.ozone.OzoneConsts.PENDING_DELETE_BLOCK_COUNT;
/**
* This class represents the KeyValueContainer metadata, which is the
@@ -70,6 +71,8 @@ public class KeyValueContainerData extends ContainerData {
private File dbFile = null;
+ private String schemaVersion;
+
/**
* Number of pending deletion blocks in KeyValueContainer.
*/
@@ -86,6 +89,7 @@ public class KeyValueContainerData extends ContainerData {
KV_YAML_FIELDS.add(METADATA_PATH);
KV_YAML_FIELDS.add(CHUNKS_PATH);
KV_YAML_FIELDS.add(CONTAINER_DB_TYPE);
+ KV_YAML_FIELDS.add(SCHEMA_VERSION);
}
/**
@@ -110,6 +114,21 @@ public class KeyValueContainerData extends ContainerData {
this.deleteTransactionId = 0;
}
+ /**
+ * @param version The schema version indicating the table layout of the
+ * container's database.
+ */
+ public void setSchemaVersion(String version) {
+ schemaVersion = version;
+ }
+
+ /**
+ * @return The schema version describing the container database's table
+ * layout.
+ */
+ public String getSchemaVersion() {
+ return schemaVersion;
+ }
/**
* Sets Container dbFile. This should be called only during creation of
@@ -267,15 +286,16 @@ public class KeyValueContainerData extends ContainerData {
public void updateAndCommitDBCounters(
ReferenceCountedDB db, BatchOperation batchOperation,
int deletedBlockCount) throws IOException {
+ Table<String, Long> metadataTable = db.getStore().getMetadataTable();
+
// Set Bytes used and block count key.
- batchOperation.put(DB_CONTAINER_BYTES_USED_KEY,
- Longs.toByteArray(getBytesUsed()));
- batchOperation.put(DB_BLOCK_COUNT_KEY, Longs.toByteArray(
- getKeyCount() - deletedBlockCount));
- batchOperation.put(DB_PENDING_DELETE_BLOCK_COUNT_KEY, Longs.toByteArray(
- getNumPendingDeletionBlocks() - deletedBlockCount));
- db.getStore().writeBatch(batchOperation);
+ metadataTable.putWithBatch(batchOperation, CONTAINER_BYTES_USED,
+ getBytesUsed());
+ metadataTable.putWithBatch(batchOperation, BLOCK_COUNT,
+ getKeyCount() - deletedBlockCount);
+ metadataTable.putWithBatch(batchOperation, PENDING_DELETE_BLOCK_COUNT,
+ (long)(getNumPendingDeletionBlocks() - deletedBlockCount));
+
+ db.getStore().getBatchHandler().commitBatchOperation(batchOperation);
}
-
-
}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java
index 99ead01..caff36c 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java
@@ -62,7 +62,7 @@ public final class BlockUtils {
try {
return cache.getDB(containerData.getContainerID(), containerData
.getContainerDBType(), containerData.getDbFile().getAbsolutePath(),
- conf);
+ containerData.getSchemaVersion(), conf);
} catch (IOException ex) {
String message = String.format("Error opening DB. Container:%s " +
"ContainerPath:%s", containerData.getContainerID(), containerData
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerUtil.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerUtil.java
index 2958e9e..e807c9b 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerUtil.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/KeyValueContainerUtil.java
@@ -22,32 +22,27 @@ import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils;
-import org.apache.hadoop.ozone.container.keyvalue.KeyValueBlockIterator;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
-import org.apache.hadoop.hdds.utils.MetadataStore;
-import org.apache.hadoop.hdds.utils.MetadataStoreBuilder;
import com.google.common.base.Preconditions;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStore;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStoreSchemaOneImpl;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStoreSchemaTwoImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_BLOCK_COMMIT_SEQUENCE_ID_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_BLOCK_COUNT_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_CONTAINER_BYTES_USED_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_CONTAINER_DELETE_TRANSACTION_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_PENDING_DELETE_BLOCK_COUNT_KEY;
-
/**
* Class which defines utility methods for KeyValueContainer.
*/
@@ -63,14 +58,31 @@ public final class KeyValueContainerUtil {
KeyValueContainerUtil.class);
/**
- * creates metadata path, chunks path and metadata DB for the specified
- * container.
*
* @param containerMetaDataPath
* @throws IOException
*/
- public static void createContainerMetaData(File containerMetaDataPath, File
- chunksPath, File dbFile, ConfigurationSource conf) throws IOException {
+
+ /**
+ * creates metadata path, chunks path and metadata DB for the specified
+ * container.
+ *
+ * @param containerMetaDataPath Path to the container's metadata directory.
+ * @param chunksPath Path were chunks for this container should be stored.
+ * @param dbFile Path to the container's .db file.
+ * @param schemaVersion The schema version of the container. Since this
+ * method is used when creating new containers, the
+ * {@link OzoneConsts#SCHEMA_LATEST} variable can be
+ * used to construct the container. If this method has
+ * not been updated after a schema version addition
+ * and does not recognize the latest SchemaVersion, an
+ * {@link IllegalArgumentException} is thrown.
+ * @param conf The configuration to use for this container.
+ * @throws IOException
+ */
+ public static void createContainerMetaData(long containerID,
+ File containerMetaDataPath, File chunksPath, File dbFile,
+ String schemaVersion, ConfigurationSource conf) throws IOException {
Preconditions.checkNotNull(containerMetaDataPath);
Preconditions.checkNotNull(conf);
@@ -91,8 +103,18 @@ public final class KeyValueContainerUtil {
" Path: " + chunksPath);
}
- MetadataStore store = MetadataStoreBuilder.newBuilder().setConf(conf)
- .setCreateIfMissing(true).setDbFile(dbFile).build();
+ DatanodeStore store;
+ if (schemaVersion.equals(OzoneConsts.SCHEMA_V1)) {
+ store = new DatanodeStoreSchemaOneImpl(conf,
+ containerID, dbFile.getAbsolutePath());
+ } else if (schemaVersion.equals(OzoneConsts.SCHEMA_V2)) {
+ store = new DatanodeStoreSchemaTwoImpl(conf,
+ containerID, dbFile.getAbsolutePath());
+ } else {
+ throw new IllegalArgumentException(
+ "Unrecognized schema version for container: " + schemaVersion);
+ }
+
ReferenceCountedDB db =
new ReferenceCountedDB(store, dbFile.getAbsolutePath());
//add db handler into cache
@@ -159,122 +181,138 @@ public final class KeyValueContainerUtil {
}
kvContainerData.setDbFile(dbFile);
+ if (kvContainerData.getSchemaVersion() == null) {
+ // If this container has not specified a schema version, it is in the old
+ // format with one default column family.
+ kvContainerData.setSchemaVersion(OzoneConsts.SCHEMA_V1);
+ }
+
boolean isBlockMetadataSet = false;
try(ReferenceCountedDB containerDB = BlockUtils.getDB(kvContainerData,
config)) {
+ Table<String, Long> metadataTable =
+ containerDB.getStore().getMetadataTable();
+
// Set pending deleted block count.
- byte[] pendingDeleteBlockCount =
- containerDB.getStore().get(DB_PENDING_DELETE_BLOCK_COUNT_KEY);
+ Long pendingDeleteBlockCount =
+ metadataTable.get(OzoneConsts.PENDING_DELETE_BLOCK_COUNT);
if (pendingDeleteBlockCount != null) {
kvContainerData.incrPendingDeletionBlocks(
- Longs.fromByteArray(pendingDeleteBlockCount));
+ pendingDeleteBlockCount);
} else {
// Set pending deleted block count.
MetadataKeyFilters.KeyPrefixFilter filter =
- new MetadataKeyFilters.KeyPrefixFilter()
- .addFilter(OzoneConsts.DELETING_KEY_PREFIX);
+ MetadataKeyFilters.getDeletingKeyFilter();
int numPendingDeletionBlocks =
- containerDB.getStore().getSequentialRangeKVs(null,
- Integer.MAX_VALUE, filter)
- .size();
+ containerDB.getStore().getBlockDataTable()
+ .getSequentialRangeKVs(null, Integer.MAX_VALUE, filter)
+ .size();
kvContainerData.incrPendingDeletionBlocks(numPendingDeletionBlocks);
}
// Set delete transaction id.
- byte[] delTxnId =
- containerDB.getStore().get(DB_CONTAINER_DELETE_TRANSACTION_KEY);
+ Long delTxnId =
+ metadataTable.get(OzoneConsts.DELETE_TRANSACTION_KEY);
if (delTxnId != null) {
kvContainerData
- .updateDeleteTransactionId(Longs.fromByteArray(delTxnId));
+ .updateDeleteTransactionId(delTxnId);
}
// Set BlockCommitSequenceId.
- byte[] bcsId = containerDB.getStore().get(
- DB_BLOCK_COMMIT_SEQUENCE_ID_KEY);
+ Long bcsId = metadataTable.get(
+ OzoneConsts.BLOCK_COMMIT_SEQUENCE_ID);
if (bcsId != null) {
kvContainerData
- .updateBlockCommitSequenceId(Longs.fromByteArray(bcsId));
+ .updateBlockCommitSequenceId(bcsId);
}
// Set bytes used.
// commitSpace for Open Containers relies on usedBytes
- byte[] bytesUsed =
- containerDB.getStore().get(DB_CONTAINER_BYTES_USED_KEY);
+ Long bytesUsed =
+ metadataTable.get(OzoneConsts.CONTAINER_BYTES_USED);
if (bytesUsed != null) {
isBlockMetadataSet = true;
- kvContainerData.setBytesUsed(Longs.fromByteArray(bytesUsed));
+ kvContainerData.setBytesUsed(bytesUsed);
}
// Set block count.
- byte[] blockCount = containerDB.getStore().get(DB_BLOCK_COUNT_KEY);
+ Long blockCount = metadataTable.get(OzoneConsts.BLOCK_COUNT);
if (blockCount != null) {
isBlockMetadataSet = true;
- kvContainerData.setKeyCount(Longs.fromByteArray(blockCount));
+ kvContainerData.setKeyCount(blockCount);
}
}
if (!isBlockMetadataSet) {
- initializeUsedBytesAndBlockCount(kvContainerData);
+ initializeUsedBytesAndBlockCount(kvContainerData, config);
}
}
/**
* Initialize bytes used and block count.
- * @param kvContainerData
+ * @param kvData
* @throws IOException
*/
private static void initializeUsedBytesAndBlockCount(
- KeyValueContainerData kvContainerData) throws IOException {
+ KeyValueContainerData kvData, ConfigurationSource config)
+ throws IOException {
- MetadataKeyFilters.KeyPrefixFilter filter =
- new MetadataKeyFilters.KeyPrefixFilter();
-
- // Ignore all blocks except those with no prefix, or those with
- // #deleting# prefix.
- filter.addFilter(OzoneConsts.DELETED_KEY_PREFIX, true)
- .addFilter(OzoneConsts.DELETE_TRANSACTION_KEY_PREFIX, true)
- .addFilter(OzoneConsts.BLOCK_COMMIT_SEQUENCE_ID_PREFIX, true)
- .addFilter(OzoneConsts.BLOCK_COUNT, true)
- .addFilter(OzoneConsts.CONTAINER_BYTES_USED, true)
- .addFilter(OzoneConsts.PENDING_DELETE_BLOCK_COUNT, true);
+ final String errorMessage = "Failed to parse block data for" +
+ " Container " + kvData.getContainerID();
long blockCount = 0;
- try (KeyValueBlockIterator blockIter = new KeyValueBlockIterator(
- kvContainerData.getContainerID(),
- new File(kvContainerData.getContainerPath()), filter)) {
- long usedBytes = 0;
-
-
- boolean success = true;
- while (success) {
- try {
- if (blockIter.hasNext()) {
- BlockData block = blockIter.nextBlock();
- long blockLen = 0;
-
- List< ContainerProtos.ChunkInfo > chunkInfoList = block.getChunks();
- for (ContainerProtos.ChunkInfo chunk : chunkInfoList) {
- ChunkInfo info = ChunkInfo.getFromProtoBuf(chunk);
- blockLen += info.getLen();
- }
-
- usedBytes += blockLen;
- blockCount++;
- } else {
- success = false;
+ long usedBytes = 0;
+
+ try(ReferenceCountedDB db = BlockUtils.getDB(kvData, config)) {
+ // Count all regular blocks.
+ try (BlockIterator<BlockData> blockIter =
+ db.getStore().getBlockIterator(
+ MetadataKeyFilters.getUnprefixedKeyFilter())) {
+
+ while (blockIter.hasNext()) {
+ blockCount++;
+ try {
+ usedBytes += getBlockLength(blockIter.nextBlock());
+ } catch (IOException ex) {
+ LOG.error(errorMessage);
+ }
+ }
+ }
+
+ // Count all deleting blocks.
+ try (BlockIterator<BlockData> blockIter =
+ db.getStore().getBlockIterator(
+ MetadataKeyFilters.getDeletingKeyFilter())) {
+
+ while (blockIter.hasNext()) {
+ blockCount++;
+ try {
+ usedBytes += getBlockLength(blockIter.nextBlock());
+ } catch (IOException ex) {
+ LOG.error(errorMessage);
}
- } catch (IOException ex) {
- LOG.error("Failed to parse block data for Container {}",
- kvContainerData.getContainerID());
}
}
- kvContainerData.setBytesUsed(usedBytes);
- kvContainerData.setKeyCount(blockCount);
}
+
+ kvData.setBytesUsed(usedBytes);
+ kvData.setKeyCount(blockCount);
+ }
+
+ private static long getBlockLength(BlockData block) throws IOException {
+ long blockLen = 0;
+ List<ContainerProtos.ChunkInfo> chunkInfoList = block.getChunks();
+
+ for (ContainerProtos.ChunkInfo chunk : chunkInfoList) {
+ ChunkInfo info = ChunkInfo.getFromProtoBuf(chunk);
+ blockLen += info.getLen();
+ }
+
+ return blockLen;
}
/**
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java
index 51fa1c9..f4f3895 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java
@@ -19,19 +19,20 @@
package org.apache.hadoop.ozone.container.keyvalue.impl;
import com.google.common.base.Preconditions;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
import org.apache.hadoop.ozone.container.common.interfaces.Container;
import org.apache.hadoop.ozone.container.keyvalue.interfaces.BlockManager;
import org.apache.hadoop.ozone.container.common.utils.ContainerCache;
-import org.apache.hadoop.hdds.utils.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
import org.slf4j.Logger;
@@ -40,14 +41,10 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.NO_SUCH_BLOCK;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNKNOWN_BCSID;
import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.BCSID_MISMATCH;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_BLOCK_COMMIT_SEQUENCE_ID_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_BLOCK_COUNT_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_CONTAINER_BYTES_USED_KEY;
/**
* This class is for performing block related operations on the KeyValue
@@ -128,10 +125,12 @@ public class BlockManagerImpl implements BlockManager {
return data.getSize();
}
// update the blockData as well as BlockCommitSequenceId here
- BatchOperation batch = new BatchOperation();
- batch.put(Longs.toByteArray(data.getLocalID()),
- data.getProtoBufMessage().toByteArray());
- batch.put(DB_BLOCK_COMMIT_SEQUENCE_ID_KEY, Longs.toByteArray(bcsId));
+ BatchOperation batch = db.getStore().getBatchHandler()
+ .initBatchOperation();
+ db.getStore().getBlockDataTable().putWithBatch(
+ batch, Long.toString(data.getLocalID()), data);
+ db.getStore().getMetadataTable().putWithBatch(
+ batch, OzoneConsts.BLOCK_COMMIT_SEQUENCE_ID, bcsId);
// Set Bytes used, this bytes used will be updated for every write and
// only get committed for every put block. In this way, when datanode
@@ -139,16 +138,18 @@ public class BlockManagerImpl implements BlockManager {
// block length is used, And also on restart the blocks committed to DB
// is only used to compute the bytes used. This is done to keep the
// current behavior and avoid DB write during write chunk operation.
- batch.put(DB_CONTAINER_BYTES_USED_KEY,
- Longs.toByteArray(container.getContainerData().getBytesUsed()));
+ db.getStore().getMetadataTable().putWithBatch(
+ batch, OzoneConsts.CONTAINER_BYTES_USED,
+ container.getContainerData().getBytesUsed());
// Set Block Count for a container.
if (incrKeyCount) {
- batch.put(DB_BLOCK_COUNT_KEY,
- Longs.toByteArray(container.getContainerData().getKeyCount() + 1));
+ db.getStore().getMetadataTable().putWithBatch(
+ batch, OzoneConsts.BLOCK_COUNT,
+ container.getContainerData().getKeyCount() + 1);
}
- db.getStore().writeBatch(batch);
+ db.getStore().getBatchHandler().commitBatchOperation(batch);
container.updateBlockCommitSequenceId(bcsId);
// Increment block count finally here for in-memory.
@@ -257,19 +258,22 @@ public class BlockManagerImpl implements BlockManager {
// are not atomic. Leaving it here since the impact is refusing
// to delete a Block which might have just gotten inserted after
// the get check.
- byte[] blockKey = Longs.toByteArray(blockID.getLocalID());
+ // Throw an exception if block data not found in the block data table.
getBlockByID(db, blockID);
// Update DB to delete block and set block count and bytes used.
- BatchOperation batch = new BatchOperation();
- batch.delete(blockKey);
+ BatchOperation batch = db.getStore().getBatchHandler()
+ .initBatchOperation();
+ String localID = Long.toString(blockID.getLocalID());
+ db.getStore().getBlockDataTable().deleteWithBatch(batch, localID);
// Update DB to delete block and set block count.
// No need to set bytes used here, as bytes used is taken care during
// delete chunk.
- batch.put(DB_BLOCK_COUNT_KEY,
- Longs.toByteArray(container.getContainerData().getKeyCount() - 1));
- db.getStore().writeBatch(batch);
+ long blockCount = container.getContainerData().getKeyCount() - 1;
+ db.getStore().getMetadataTable()
+ .putWithBatch(batch, OzoneConsts.BLOCK_COUNT, blockCount);
+ db.getStore().getBatchHandler().commitBatchOperation(batch);
// Decrement block count here
container.getContainerData().decrKeyCount();
@@ -299,13 +303,12 @@ public class BlockManagerImpl implements BlockManager {
(KeyValueContainerData) container.getContainerData();
try (ReferenceCountedDB db = BlockUtils.getDB(cData, config)) {
result = new ArrayList<>();
- byte[] startKeyInBytes = Longs.toByteArray(startLocalID);
- List<Map.Entry<byte[], byte[]>> range = db.getStore()
- .getSequentialRangeKVs(startKeyInBytes, count,
- MetadataKeyFilters.getNormalKeyFilter());
- for (Map.Entry<byte[], byte[]> entry : range) {
- BlockData value = BlockUtils.getBlockData(entry.getValue());
- BlockData data = new BlockData(value.getBlockID());
+ List<? extends Table.KeyValue<String, BlockData>> range =
+ db.getStore().getBlockDataTable()
+ .getSequentialRangeKVs(Long.toString(startLocalID), count,
+ MetadataKeyFilters.getUnprefixedKeyFilter());
+ for (Table.KeyValue<String, BlockData> entry: range) {
+ BlockData data = new BlockData(entry.getValue().getBlockID());
result.add(data);
}
return result;
@@ -324,14 +327,14 @@ public class BlockManagerImpl implements BlockManager {
private byte[] getBlockByID(ReferenceCountedDB db, BlockID blockID)
throws IOException {
- byte[] blockKey = Longs.toByteArray(blockID.getLocalID());
+ String blockKey = Long.toString(blockID.getLocalID());
- byte[] blockData = db.getStore().get(blockKey);
+ BlockData blockData = db.getStore().getBlockDataTable().get(blockKey);
if (blockData == null) {
throw new StorageContainerException(NO_SUCH_BLOCK_ERR_MSG,
NO_SUCH_BLOCK);
}
- return blockData;
+ return blockData.getProtoBufMessage().toByteArray();
}
}
\ No newline at end of file
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java
index 07b3f6b..84334de 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingService.java
@@ -28,7 +28,6 @@ import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
@@ -38,10 +37,13 @@ import org.apache.hadoop.hdds.utils.BackgroundService;
import org.apache.hadoop.hdds.utils.BackgroundTask;
import org.apache.hadoop.hdds.utils.BackgroundTaskQueue;
import org.apache.hadoop.hdds.utils.BackgroundTaskResult;
-import org.apache.hadoop.hdds.utils.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
import org.apache.hadoop.hdds.utils.MetadataKeyFilters.KeyPrefixFilter;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
import org.apache.hadoop.ozone.container.common.impl.ContainerData;
import org.apache.hadoop.ozone.container.common.impl.TopNOrderedContainerDeletionChoosingPolicy;
import org.apache.hadoop.ozone.container.common.interfaces.Container;
@@ -252,12 +254,14 @@ public class BlockDeletingService extends BackgroundService {
long startTime = Time.monotonicNow();
// Scan container's db and get list of under deletion blocks
try (ReferenceCountedDB meta = BlockUtils.getDB(containerData, conf)) {
+ Table<String, BlockData> blockDataTable =
+ meta.getStore().getBlockDataTable();
+
// # of blocks to delete is throttled
- KeyPrefixFilter filter =
- new KeyPrefixFilter().addFilter(OzoneConsts.DELETING_KEY_PREFIX);
- List<Map.Entry<byte[], byte[]>> toDeleteBlocks =
- meta.getStore().getSequentialRangeKVs(null, blockLimitPerTask,
- filter);
+ KeyPrefixFilter filter = MetadataKeyFilters.getDeletingKeyFilter();
+ List<? extends Table.KeyValue<String, BlockData>> toDeleteBlocks =
+ blockDataTable.getSequentialRangeKVs(null, blockLimitPerTask,
+ filter);
if (toDeleteBlocks.isEmpty()) {
LOG.debug("No under deletion block found in container : {}",
containerData.getContainerID());
@@ -276,33 +280,36 @@ public class BlockDeletingService extends BackgroundService {
Handler handler = Objects.requireNonNull(ozoneContainer.getDispatcher()
.getHandler(container.getContainerType()));
- toDeleteBlocks.forEach(entry -> {
- String blockName = StringUtils.bytes2String(entry.getKey());
+ for (Table.KeyValue<String, BlockData> entry: toDeleteBlocks) {
+ String blockName = entry.getKey();
LOG.debug("Deleting block {}", blockName);
try {
- ContainerProtos.BlockData data =
- ContainerProtos.BlockData.parseFrom(entry.getValue());
- handler.deleteBlock(container, BlockData.getFromProtoBuf(data));
+ handler.deleteBlock(container, entry.getValue());
succeedBlocks.add(blockName);
} catch (InvalidProtocolBufferException e) {
LOG.error("Failed to parse block info for block {}", blockName, e);
} catch (IOException e) {
LOG.error("Failed to delete files for block {}", blockName, e);
}
- });
+ }
// Once files are deleted... replace deleting entries with deleted
// entries
- BatchOperation batch = new BatchOperation();
- succeedBlocks.forEach(entry -> {
- String blockId =
- entry.substring(OzoneConsts.DELETING_KEY_PREFIX.length());
- String deletedEntry = OzoneConsts.DELETED_KEY_PREFIX + blockId;
- batch.put(StringUtils.string2Bytes(deletedEntry),
- StringUtils.string2Bytes(blockId));
- batch.delete(StringUtils.string2Bytes(entry));
- });
-
+ BatchOperation batch = meta.getStore().getBatchHandler()
+ .initBatchOperation();
+ Table<String, ChunkInfoList> deletedBlocksTable =
+ meta.getStore().getDeletedBlocksTable();
+ for (String entry: succeedBlocks) {
+ List<ContainerProtos.ChunkInfo> chunkList =
+ blockDataTable.get(entry).getChunks();
+ String blockId = entry.substring(
+ OzoneConsts.DELETING_KEY_PREFIX.length());
+
+ deletedBlocksTable.putWithBatch(
+ batch, blockId,
+ new ChunkInfoList(chunkList));
+ blockDataTable.deleteWithBatch(batch, entry);
+ }
int deleteBlockCount = succeedBlocks.size();
containerData.updateAndCommitDBCounters(meta, batch, deleteBlockCount);
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/AbstractDatanodeDBDefinition.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/AbstractDatanodeDBDefinition.java
new file mode 100644
index 0000000..8895475
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/AbstractDatanodeDBDefinition.java
@@ -0,0 +1,74 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition;
+import org.apache.hadoop.hdds.utils.db.DBDefinition;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+
+import java.io.File;
+
+/**
+ * This class serves as an intermediate format for all possible database
+ * layouts for datanodes.
+ */
+public abstract class AbstractDatanodeDBDefinition implements DBDefinition {
+
+ private File dbDir;
+
+ /**
+ * @param dbPath The absolute path to the .db file corresponding to this
+ * {@link DBDefinition}.
+ */
+ protected AbstractDatanodeDBDefinition(String dbPath) {
+ this.dbDir = new File(dbPath);
+ }
+
+ @Override
+ public String getName() {
+ return dbDir.getName();
+ }
+
+ @Override
+ public File getDBLocation(ConfigurationSource conf) {
+ return dbDir.getParentFile();
+ }
+
+ @Override
+ public String getLocationConfigKey() {
+ throw new UnsupportedOperationException(
+ "No location config key available for datanode databases.");
+ }
+
+ @Override
+ public DBColumnFamilyDefinition[] getColumnFamilies() {
+ return new DBColumnFamilyDefinition[] {getBlockDataColumnFamily(),
+ getMetadataColumnFamily(), getDeletedBlocksColumnFamily()};
+ }
+
+ public abstract DBColumnFamilyDefinition<String, BlockData>
+ getBlockDataColumnFamily();
+
+ public abstract DBColumnFamilyDefinition<String, Long>
+ getMetadataColumnFamily();
+
+ public abstract DBColumnFamilyDefinition<String, ChunkInfoList>
+ getDeletedBlocksColumnFamily();
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/AbstractDatanodeStore.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/AbstractDatanodeStore.java
new file mode 100644
index 0000000..6c258ed
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/AbstractDatanodeStore.java
@@ -0,0 +1,297 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.StringUtils;
+import org.apache.hadoop.hdds.annotation.InterfaceAudience;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters.KeyPrefixFilter;
+import org.apache.hadoop.hdds.utils.db.*;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
+import org.rocksdb.DBOptions;
+import org.rocksdb.Statistics;
+import org.rocksdb.StatsLevel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.NoSuchElementException;
+
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_METADATA_STORE_ROCKSDB_STATISTICS;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_METADATA_STORE_ROCKSDB_STATISTICS_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_METADATA_STORE_ROCKSDB_STATISTICS_OFF;
+
+/**
+ * Implementation of the {@link DatanodeStore} interface that contains
+ * functionality common to all more derived datanode store implementations.
+ */
+public abstract class AbstractDatanodeStore implements DatanodeStore {
+
+ private Table<String, Long> metadataTable;
+
+ private Table<String, BlockData> blockDataTable;
+
+ private Table<String, BlockData> blockDataTableWithIterator;
+
+ private Table<String, ChunkInfoList> deletedBlocksTable;
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractDatanodeStore.class);
+ private DBStore store;
+ private final AbstractDatanodeDBDefinition dbDef;
+ private final long containerID;
+
+ /**
+ * Constructs the metadata store and starts the DB services.
+ *
+ * @param config - Ozone Configuration.
+ * @throws IOException - on Failure.
+ */
+ protected AbstractDatanodeStore(ConfigurationSource config, long containerID,
+ AbstractDatanodeDBDefinition dbDef)
+ throws IOException {
+ this.dbDef = dbDef;
+ this.containerID = containerID;
+ start(config);
+ }
+
+ @Override
+ public void start(ConfigurationSource config)
+ throws IOException {
+ if (this.store == null) {
+ DBOptions options = new DBOptions();
+ options.setCreateIfMissing(true);
+ options.setCreateMissingColumnFamilies(true);
+
+ String rocksDbStat = config.getTrimmed(
+ OZONE_METADATA_STORE_ROCKSDB_STATISTICS,
+ OZONE_METADATA_STORE_ROCKSDB_STATISTICS_DEFAULT);
+
+ if (!rocksDbStat.equals(OZONE_METADATA_STORE_ROCKSDB_STATISTICS_OFF)) {
+ Statistics statistics = new Statistics();
+ statistics.setStatsLevel(StatsLevel.valueOf(rocksDbStat));
+ options.setStatistics(statistics);
+ }
+
+ this.store = DBStoreBuilder.newBuilder(config, dbDef)
+ .setDBOption(options)
+ .build();
+
+ // Use the DatanodeTable wrapper to disable the table iterator on
+ // existing Table implementations retrieved from the DBDefinition.
+ // See the DatanodeTable's Javadoc for an explanation of why this is
+ // necessary.
+ metadataTable = new DatanodeTable<>(
+ dbDef.getMetadataColumnFamily().getTable(this.store));
+ checkTableStatus(metadataTable, metadataTable.getName());
+
+ // The block iterator this class returns will need to use the table
+ // iterator internally, so construct a block data table instance
+ // that does not have the iterator disabled by DatanodeTable.
+ blockDataTableWithIterator =
+ dbDef.getBlockDataColumnFamily().getTable(this.store);
+
+ blockDataTable = new DatanodeTable<>(blockDataTableWithIterator);
+ checkTableStatus(blockDataTable, blockDataTable.getName());
+
+ deletedBlocksTable = new DatanodeTable<>(
+ dbDef.getDeletedBlocksColumnFamily().getTable(this.store));
+ checkTableStatus(deletedBlocksTable, deletedBlocksTable.getName());
+ }
+ }
+
+ @Override
+ public void stop() throws Exception {
+ if (store != null) {
+ store.close();
+ store = null;
+ }
+ }
+
+ @Override
+ public DBStore getStore() {
+ return this.store;
+ }
+
+ @Override
+ public BatchOperationHandler getBatchHandler() {
+ return this.store;
+ }
+
+ @Override
+ public Table<String, Long> getMetadataTable() {
+ return metadataTable;
+ }
+
+ @Override
+ public Table<String, BlockData> getBlockDataTable() {
+ return blockDataTable;
+ }
+
+ @Override
+ public Table<String, ChunkInfoList> getDeletedBlocksTable() {
+ return deletedBlocksTable;
+ }
+
+ @Override
+ public BlockIterator<BlockData> getBlockIterator() {
+ return new KeyValueBlockIterator(containerID,
+ blockDataTableWithIterator.iterator());
+ }
+
+ @Override
+ public BlockIterator<BlockData> getBlockIterator(KeyPrefixFilter filter) {
+ return new KeyValueBlockIterator(containerID,
+ blockDataTableWithIterator.iterator(), filter);
+ }
+
+ @Override
+ public void flushDB() throws IOException {
+ store.flushDB();
+ }
+
+ @Override
+ public void flushLog(boolean sync) throws IOException {
+ store.flushLog(sync);
+ }
+
+ @Override
+ public void compactDB() throws IOException {
+ store.compactDB();
+ }
+
+ private static void checkTableStatus(Table<?, ?> table, String name)
+ throws IOException {
+ String logMessage = "Unable to get a reference to %s table. Cannot " +
+ "continue.";
+ String errMsg = "Inconsistent DB state, Table - %s. Please check the" +
+ " logs for more info.";
+ if (table == null) {
+ LOG.error(String.format(logMessage, name));
+ throw new IOException(String.format(errMsg, name));
+ }
+ }
+
+ /**
+ * Block Iterator for KeyValue Container. This block iterator returns blocks
+ * which match with the {@link MetadataKeyFilters.KeyPrefixFilter}. If no
+ * filter is specified, then default filter used is
+ * {@link MetadataKeyFilters#getUnprefixedKeyFilter()}
+ */
+ @InterfaceAudience.Public
+ private static class KeyValueBlockIterator implements
+ BlockIterator<BlockData>, Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ KeyValueBlockIterator.class);
+
+ private final TableIterator<String, ? extends Table.KeyValue<String,
+ BlockData>>
+ blockIterator;
+ private static final KeyPrefixFilter DEFAULT_BLOCK_FILTER =
+ MetadataKeyFilters.getUnprefixedKeyFilter();
+ private final KeyPrefixFilter blockFilter;
+ private BlockData nextBlock;
+ private final long containerID;
+
+ /**
+ * KeyValueBlockIterator to iterate unprefixed blocks in a container.
+ * @param iterator - The underlying iterator to apply the block filter to.
+ */
+ KeyValueBlockIterator(long containerID,
+ TableIterator<String, ? extends Table.KeyValue<String, BlockData>>
+ iterator) {
+ this.containerID = containerID;
+ this.blockIterator = iterator;
+ this.blockFilter = DEFAULT_BLOCK_FILTER;
+ }
+
+ /**
+ * KeyValueBlockIterator to iterate blocks in a container.
+ * @param iterator - The underlying iterator to apply the block filter to.
+ * @param filter - Block filter, filter to be applied for blocks
+ */
+ KeyValueBlockIterator(long containerID,
+ TableIterator<String, ? extends Table.KeyValue<String, BlockData>>
+ iterator, KeyPrefixFilter filter) {
+ this.containerID = containerID;
+ this.blockIterator = iterator;
+ this.blockFilter = filter;
+ }
+
+ /**
+ * This method returns blocks matching with the filter.
+ * @return next block or null if no more blocks
+ * @throws IOException
+ */
+ @Override
+ public BlockData nextBlock() throws IOException, NoSuchElementException {
+ if (nextBlock != null) {
+ BlockData currentBlock = nextBlock;
+ nextBlock = null;
+ return currentBlock;
+ }
+ if(hasNext()) {
+ return nextBlock();
+ }
+ throw new NoSuchElementException("Block Iterator reached end for " +
+ "ContainerID " + containerID);
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (nextBlock != null) {
+ return true;
+ }
+ while (blockIterator.hasNext()) {
+ Table.KeyValue<String, BlockData> keyValue = blockIterator.next();
+ byte[] keyBytes = StringUtils.string2Bytes(keyValue.getKey());
+ if (blockFilter.filterKey(null, keyBytes, null)) {
+ nextBlock = keyValue.getValue();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Block matching with filter found: blockID is : {} for " +
+ "containerID {}", nextBlock.getLocalID(), containerID);
+ }
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public void seekToFirst() {
+ nextBlock = null;
+ blockIterator.seekToFirst();
+ }
+
+ @Override
+ public void seekToLast() {
+ nextBlock = null;
+ blockIterator.seekToLast();
+ }
+
+ @Override
+ public void close() throws IOException {
+ blockIterator.close();
+ }
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/BlockDataCodec.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/BlockDataCodec.java
new file mode 100644
index 0000000..fc5af7c
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/BlockDataCodec.java
@@ -0,0 +1,47 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+
+import java.io.IOException;
+
+/**
+ * Supports encoding and decoding {@link BlockData} objects.
+ */
+public class BlockDataCodec implements Codec<BlockData> {
+
+ @Override
+ public byte[] toPersistedFormat(BlockData blockData) {
+ return blockData.getProtoBufMessage().toByteArray();
+ }
+
+ @Override
+ public BlockData fromPersistedFormat(byte[] rawData) throws IOException {
+ // Convert raw bytes -> protobuf version of BlockData -> BlockData object.
+ return BlockData.getFromProtoBuf(
+ ContainerProtos.BlockData.parseFrom(rawData));
+ }
+
+ @Override
+ public BlockData copyObject(BlockData object) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/ChunkInfoListCodec.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/ChunkInfoListCodec.java
new file mode 100644
index 0000000..1b54beb
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/ChunkInfoListCodec.java
@@ -0,0 +1,45 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+
+import java.io.IOException;
+
+/**
+ * Supports encoding and decoding {@link ChunkInfoList} objects.
+ */
+public class ChunkInfoListCodec implements Codec<ChunkInfoList> {
+ @Override
+ public byte[] toPersistedFormat(ChunkInfoList chunkList) {
+ return chunkList.getProtoBufMessage().toByteArray();
+ }
+
+ @Override
+ public ChunkInfoList fromPersistedFormat(byte[] rawData) throws IOException {
+ return ChunkInfoList.getFromProtoBuf(
+ ContainerProtos.ChunkInfoList.parseFrom(rawData));
+ }
+
+ @Override
+ public ChunkInfoList copyObject(ChunkInfoList object) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeSchemaOneDBDefinition.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeSchemaOneDBDefinition.java
new file mode 100644
index 0000000..faf399d
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeSchemaOneDBDefinition.java
@@ -0,0 +1,91 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.StringUtils;
+import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition;
+import org.apache.hadoop.hdds.utils.db.LongCodec;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+import org.rocksdb.RocksDB;
+
+/**
+ * This class allows support of the schema version one RocksDB layout for
+ * datanode, where
+ * all data is kept in the same default table. Clients can use this class as
+ * if the database is in the new format (which has separate column families for
+ * block data, metadata and deleted block IDs), even
+ * though all tables map back to the default table in this implementation.
+ */
+public class DatanodeSchemaOneDBDefinition
+ extends AbstractDatanodeDBDefinition {
+ // In the underlying database, tables are retrieved by name, and then the
+ // codecs/classes are applied on top. By defining different DBDefinitions
+ // with different codecs that all map to the default table, clients are
+ // unaware they are using the same table for both interpretations of the data.
+
+ // Note that the current RDBStore implementation requires all column
+ // families to use the same codec instance for each key/value type.
+ public static final DBColumnFamilyDefinition<String, BlockData>
+ BLOCK_DATA =
+ new DBColumnFamilyDefinition<>(
+ StringUtils.bytes2String(RocksDB.DEFAULT_COLUMN_FAMILY),
+ String.class,
+ new SchemaOneKeyCodec(),
+ BlockData.class,
+ new BlockDataCodec());
+
+ public static final DBColumnFamilyDefinition<String, Long>
+ METADATA =
+ new DBColumnFamilyDefinition<>(
+ StringUtils.bytes2String(RocksDB.DEFAULT_COLUMN_FAMILY),
+ String.class,
+ new SchemaOneKeyCodec(),
+ Long.class,
+ new LongCodec());
+
+ public static final DBColumnFamilyDefinition<String, ChunkInfoList>
+ DELETED_BLOCKS =
+ new DBColumnFamilyDefinition<>(
+ StringUtils.bytes2String(RocksDB.DEFAULT_COLUMN_FAMILY),
+ String.class,
+ new SchemaOneKeyCodec(),
+ ChunkInfoList.class,
+ new SchemaOneChunkInfoListCodec());
+
+ protected DatanodeSchemaOneDBDefinition(String dbPath) {
+ super(dbPath);
+ }
+
+ @Override
+ public DBColumnFamilyDefinition<String, BlockData>
+ getBlockDataColumnFamily() {
+ return BLOCK_DATA;
+ }
+
+ @Override
+ public DBColumnFamilyDefinition<String, Long> getMetadataColumnFamily() {
+ return METADATA;
+ }
+
+ @Override
+ public DBColumnFamilyDefinition<String, ChunkInfoList>
+ getDeletedBlocksColumnFamily() {
+ return DELETED_BLOCKS;
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeSchemaTwoDBDefinition.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeSchemaTwoDBDefinition.java
new file mode 100644
index 0000000..2ac56f2
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeSchemaTwoDBDefinition.java
@@ -0,0 +1,81 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition;
+import org.apache.hadoop.hdds.utils.db.LongCodec;
+import org.apache.hadoop.hdds.utils.db.StringCodec;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+
+/**
+ * This class defines the RocksDB structure for datanodes following schema
+ * version 2, where the block data, metadata, and deleted block ids are put in
+ * their own separate column families.
+ */
+public class DatanodeSchemaTwoDBDefinition extends
+ AbstractDatanodeDBDefinition {
+
+ public static final DBColumnFamilyDefinition<String, BlockData>
+ BLOCK_DATA =
+ new DBColumnFamilyDefinition<>(
+ "block_data",
+ String.class,
+ new StringCodec(),
+ BlockData.class,
+ new BlockDataCodec());
+
+ public static final DBColumnFamilyDefinition<String, Long>
+ METADATA =
+ new DBColumnFamilyDefinition<>(
+ "metadata",
+ String.class,
+ new StringCodec(),
+ Long.class,
+ new LongCodec());
+
+ public static final DBColumnFamilyDefinition<String, ChunkInfoList>
+ DELETED_BLOCKS =
+ new DBColumnFamilyDefinition<>(
+ "deleted_blocks",
+ String.class,
+ new StringCodec(),
+ ChunkInfoList.class,
+ new ChunkInfoListCodec());
+
+ protected DatanodeSchemaTwoDBDefinition(String dbPath) {
+ super(dbPath);
+ }
+
+ @Override
+ public DBColumnFamilyDefinition<String, BlockData>
+ getBlockDataColumnFamily() {
+ return BLOCK_DATA;
+ }
+
+ @Override
+ public DBColumnFamilyDefinition<String, Long> getMetadataColumnFamily() {
+ return METADATA;
+ }
+
+ @Override
+ public DBColumnFamilyDefinition<String, ChunkInfoList>
+ getDeletedBlocksColumnFamily() {
+ return DELETED_BLOCKS;
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStore.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStore.java
new file mode 100644
index 0000000..5a0ce7a
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStore.java
@@ -0,0 +1,94 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.db.BatchOperationHandler;
+import org.apache.hadoop.hdds.utils.db.DBStore;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
+
+import java.io.IOException;
+
+/**
+ * Interface for interacting with datanode databases.
+ */
+public interface DatanodeStore {
+
+ /**
+ * Start datanode manager.
+ *
+ * @param configuration - Configuration
+ * @throws IOException - Unable to start datanode store.
+ */
+ void start(ConfigurationSource configuration) throws IOException;
+
+ /**
+ * Stop datanode manager.
+ */
+ void stop() throws Exception;
+
+ /**
+ * Get datanode store.
+ *
+ * @return datanode store.
+ */
+ @VisibleForTesting
+ DBStore getStore();
+
+ /**
+ * A Table that keeps the block data.
+ *
+ * @return Table
+ */
+ Table<String, BlockData> getBlockDataTable();
+
+ /**
+ * A Table that keeps the metadata.
+ *
+ * @return Table
+ */
+ Table<String, Long> getMetadataTable();
+
+ /**
+ * A Table that keeps IDs of blocks deleted from the block data table.
+ *
+ * @return Table
+ */
+ Table<String, ChunkInfoList> getDeletedBlocksTable();
+
+ /**
+ * Helper to create and write batch transactions.
+ */
+ BatchOperationHandler getBatchHandler();
+
+ void flushLog(boolean sync) throws IOException;
+
+ void flushDB() throws IOException;
+
+ void compactDB() throws IOException;
+
+ BlockIterator<BlockData> getBlockIterator();
+
+ BlockIterator<BlockData>
+ getBlockIterator(MetadataKeyFilters.KeyPrefixFilter filter);
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaOneImpl.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaOneImpl.java
new file mode 100644
index 0000000..97b9b25
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaOneImpl.java
@@ -0,0 +1,49 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+
+import java.io.IOException;
+
+/**
+ * Constructs a datanode store in accordance with schema version 1, which
+ * places all data in the default column family.
+ */
+public class DatanodeStoreSchemaOneImpl extends AbstractDatanodeStore {
+ /**
+ * Constructs the metadata store and starts the DB Services.
+ *
+ * @param config - Ozone Configuration.
+ * @throws IOException - on Failure.
+ */
+ public DatanodeStoreSchemaOneImpl(ConfigurationSource config,
+ long containerID, String dbPath)
+ throws IOException {
+ super(config, containerID, new DatanodeSchemaOneDBDefinition(dbPath));
+ }
+
+ @Override
+ public Table<String, ChunkInfoList> getDeletedBlocksTable() {
+ // Return a wrapper around the deleted blocks table to handle prefixes
+ // when all data is stored in a single table.
+ return new SchemaOneDeletedBlocksTable(super.getDeletedBlocksTable());
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaTwoImpl.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaTwoImpl.java
new file mode 100644
index 0000000..fd8e4fa
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaTwoImpl.java
@@ -0,0 +1,44 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+
+import java.io.IOException;
+
+/**
+ * Constructs a datanode store in accordance with schema version 2, which uses
+ * three column families/tables:
+ * 1. A block data table.
+ * 2. A metadata table.
+ * 3. A deleted blocks table.
+ */
+public class DatanodeStoreSchemaTwoImpl extends AbstractDatanodeStore {
+
+ /**
+ * Constructs the datanode store and starts the DB Services.
+ *
+ * @param config - Ozone Configuration.
+ * @throws IOException - on Failure.
+ */
+ public DatanodeStoreSchemaTwoImpl(ConfigurationSource config,
+ long containerID, String dbPath)
+ throws IOException {
+ super(config, containerID, new DatanodeSchemaTwoDBDefinition(dbPath));
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeTable.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeTable.java
new file mode 100644
index 0000000..2fc5576
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeTable.java
@@ -0,0 +1,130 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Wrapper class to represent a table in a datanode RocksDB instance.
+ * This class can wrap any existing {@link Table} instance, but will throw
+ * {@link UnsupportedOperationException} for {@link Table#iterator}.
+ * This is because differing schema versions used in datanode DB layouts may
+ * have differing underlying table structures, so iterating a table instance
+ * directly, without taking into account key prefixes, may yield unexpected
+ * results.
+ */
+public class DatanodeTable<KEY, VALUE> implements Table<KEY, VALUE> {
+
+ private final Table<KEY, VALUE> table;
+
+ public DatanodeTable(Table<KEY, VALUE> table) {
+ this.table = table;
+ }
+
+ @Override
+ public void put(KEY key, VALUE value) throws IOException {
+ table.put(key, value);
+ }
+
+ @Override
+ public void putWithBatch(BatchOperation batch, KEY key,
+ VALUE value) throws IOException {
+ table.putWithBatch(batch, key, value);
+ }
+
+ @Override
+ public boolean isEmpty() throws IOException {
+ return table.isEmpty();
+ }
+
+ @Override
+ public void delete(KEY key) throws IOException {
+ table.delete(key);
+ }
+
+ @Override
+ public void deleteWithBatch(BatchOperation batch, KEY key)
+ throws IOException {
+ table.deleteWithBatch(batch, key);
+ }
+
+ @Override
+ public final TableIterator<KEY, ? extends KeyValue<KEY, VALUE>> iterator() {
+ throw new UnsupportedOperationException("Iterating tables directly is not" +
+ " supported for datanode containers due to differing schema " +
+ "version.");
+ }
+
+ @Override
+ public String getName() throws IOException {
+ return table.getName();
+ }
+
+ @Override
+ public long getEstimatedKeyCount() throws IOException {
+ return table.getEstimatedKeyCount();
+ }
+
+ @Override
+ public boolean isExist(KEY key) throws IOException {
+ return table.isExist(key);
+ }
+
+ @Override
+ public VALUE get(KEY key) throws IOException {
+ return table.get(key);
+ }
+
+ @Override
+ public VALUE getIfExist(KEY key) throws IOException {
+ return table.getIfExist(key);
+ }
+
+ @Override
+ public VALUE getReadCopy(KEY key) throws IOException {
+ return table.getReadCopy(key);
+ }
+
+ @Override
+ public List<? extends KeyValue<KEY, VALUE>> getRangeKVs(
+ KEY startKey, int count,
+ MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+ return table.getRangeKVs(startKey, count, filters);
+ }
+
+ @Override
+ public List<? extends KeyValue<KEY, VALUE>> getSequentialRangeKVs(
+ KEY startKey, int count,
+ MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+ return table.getSequentialRangeKVs(startKey, count, filters);
+ }
+
+ @Override
+ public void close() throws Exception {
+ table.close();
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneChunkInfoListCodec.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneChunkInfoListCodec.java
new file mode 100644
index 0000000..bca625c
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneChunkInfoListCodec.java
@@ -0,0 +1,68 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
+
+import java.io.IOException;
+
+/**
+ * Codec for parsing {@link ContainerProtos.ChunkInfoList} objects from data
+ * that may have been written using schema version one. Before upgrading
+ * schema versions, deleted block IDs were stored with a duplicate copy of
+ * their ID as the value in the database. After upgrading the code, any
+ * deletes that happen on the DB will save the chunk information with the
+ * deleted blocks instead, even if those deletes are performed on a database
+ * created with schema version one.
+ * <p>
+ * Since the resulting database may contain a mix of values that could either
+ * be block IDs or chunk information, this codec will attempt to parse bytes as
+ * chunk information, and on failure, throw an exception with a message about
+ * the old schema format.
+ * <p>
+ * Note that there is currently no where in the code (except for tests) where
+ * the values associated with deleted block IDs are used. If it is added in the
+ * future, callers should be aware that the associated chunk information may not
+ * always be present.
+ */
+public class SchemaOneChunkInfoListCodec implements Codec<ChunkInfoList> {
+ @Override
+ public byte[] toPersistedFormat(ChunkInfoList chunkList) {
+ return chunkList.getProtoBufMessage().toByteArray();
+ }
+
+ @Override
+ public ChunkInfoList fromPersistedFormat(byte[] rawData) throws IOException {
+ try {
+ return ChunkInfoList.getFromProtoBuf(
+ ContainerProtos.ChunkInfoList.parseFrom(rawData));
+ } catch (InvalidProtocolBufferException ex) {
+ throw new IOException("Invalid chunk information. " +
+ "This data may have been written using datanode " +
+ "schema version one, which did not save chunk information.", ex);
+ }
+ }
+
+ @Override
+ public ChunkInfoList copyObject(ChunkInfoList object) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneDeletedBlocksTable.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneDeletedBlocksTable.java
new file mode 100644
index 0000000..3f86cd2
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneDeletedBlocksTable.java
@@ -0,0 +1,180 @@
+/*
+ * 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.hadoop.ozone.container.metadata;
+
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * For RocksDB instances written using DB schema version 1, all data is
+ * stored in the default column family. This differs from later schema
+ * versions, which put deleted blocks in a different column family.
+ * As a result, the block IDs used as keys for deleted blocks must be
+ * prefixed in schema version 1 so that they can be differentiated from
+ * regular blocks. However, these prefixes are not necessary in later schema
+ * versions, because the deleted blocks and regular blocks are in different
+ * column families.
+ * <p>
+ * Since clients must operate independently of the underlying schema version,
+ * This class is returned to clients using {@link DatanodeStoreSchemaOneImpl}
+ * instances, allowing them to access keys as if no prefix is
+ * required, while it adds or removes the prefix when necessary.
+ * This means the client should omit the deleted prefix when putting and
+ * getting keys, regardless of the schema version.
+ */
+public class SchemaOneDeletedBlocksTable extends DatanodeTable<String,
+ ChunkInfoList> {
+ public static final String DELETED_KEY_PREFIX = "#deleted#";
+
+ public SchemaOneDeletedBlocksTable(Table<String, ChunkInfoList> table) {
+ super(table);
+ }
+
+ @Override
+ public void put(String key, ChunkInfoList value) throws IOException {
+ super.put(prefix(key), value);
+ }
+
+ @Override
+ public void putWithBatch(BatchOperation batch, String key,
+ ChunkInfoList value)
+ throws IOException {
+ super.putWithBatch(batch, prefix(key), value);
+ }
+
+ @Override
+ public void delete(String key) throws IOException {
+ super.delete(prefix(key));
+ }
+
+ @Override
+ public void deleteWithBatch(BatchOperation batch, String key)
+ throws IOException {
+ super.deleteWithBatch(batch, prefix(key));
+ }
+
+ @Override
+ public boolean isExist(String key) throws IOException {
+ return super.isExist(prefix(key));
+ }
+
+ @Override
+ public ChunkInfoList get(String key) throws IOException {
+ return super.get(prefix(key));
+ }
+
+ @Override
+ public ChunkInfoList getIfExist(String key) throws IOException {
+ return super.getIfExist(prefix(key));
+ }
+
+ @Override
+ public ChunkInfoList getReadCopy(String key) throws IOException {
+ return super.getReadCopy(prefix(key));
+ }
+
+ @Override
+ public List<? extends KeyValue<String, ChunkInfoList>> getRangeKVs(
+ String startKey, int count,
+ MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+
+ // Deleted blocks will always have the #deleted# key prefix and nothing
+ // else in this schema version. Ignore any user passed prefixes that could
+ // collide with this and return results that are not deleted blocks.
+ return unprefix(super.getRangeKVs(prefix(startKey), count,
+ getDeletedFilter()));
+ }
+
+ @Override
+ public List<? extends KeyValue<String, ChunkInfoList>> getSequentialRangeKVs(
+ String startKey, int count,
+ MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+
+ // Deleted blocks will always have the #deleted# key prefix and nothing
+ // else in this schema version. Ignore any user passed prefixes that could
+ // collide with this and return results that are not deleted blocks.
+ return unprefix(super.getSequentialRangeKVs(prefix(startKey), count,
+ getDeletedFilter()));
+ }
+
+ private static String prefix(String key) {
+ String result = null;
+ if (key != null) {
+ result = DELETED_KEY_PREFIX + key;
+ }
+
+ return result;
+ }
+
+ private static String unprefix(String key) {
+ String result = null;
+ if (key != null && key.startsWith(DELETED_KEY_PREFIX)) {
+ result = key.replaceFirst(DELETED_KEY_PREFIX, "");
+ }
+
+ return result;
+ }
+
+ private static List<KeyValue<String, ChunkInfoList>> unprefix(
+ List<? extends KeyValue<String, ChunkInfoList>> kvs) {
+
+ List<KeyValue<String, ChunkInfoList>> processedKVs = new ArrayList<>();
+ kvs.forEach(kv -> processedKVs.add(new UnprefixedKeyValue(kv)));
+
+ return processedKVs;
+ }
+
+ private static MetadataKeyFilters.KeyPrefixFilter getDeletedFilter() {
+ return (new MetadataKeyFilters.KeyPrefixFilter())
+ .addFilter(DELETED_KEY_PREFIX);
+ }
+
+ /**
+ * {@link KeyValue} implementation that removes the deleted key prefix from
+ * an existing {@link KeyValue}.
+ */
+ private static class UnprefixedKeyValue implements KeyValue<String,
+ ChunkInfoList> {
+
+ private final KeyValue<String, ChunkInfoList> prefixedKeyValue;
+
+ UnprefixedKeyValue(
+ KeyValue<String, ChunkInfoList> prefixedKeyValue) {
+ this.prefixedKeyValue = prefixedKeyValue;
+ }
+
+ @Override
+ public String getKey() throws IOException {
+ return unprefix(prefixedKeyValue.getKey());
+ }
+
+ @Override
+ public ChunkInfoList getValue() throws IOException {
+ return prefixedKeyValue.getValue();
+ }
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneKeyCodec.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneKeyCodec.java
new file mode 100644
index 0000000..588ff45
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/SchemaOneKeyCodec.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.hadoop.ozone.container.metadata;
+
+import java.io.IOException;
+
+import com.google.common.primitives.Longs;
+import org.apache.hadoop.hdds.StringUtils;
+import org.apache.hadoop.hdds.utils.db.Codec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Containers written using schema version 1 wrote unprefixed block ID keys
+ * as longs, and metadata or prefixed block IDs as Strings. This was done
+ * before codecs were introduced, so callers could serialize their data
+ * however they wanted. This codec handles this by checking for string
+ * prefixes in the data, and determining which format it should be
+ * encoded/decoded to/from.
+ */
+public class SchemaOneKeyCodec implements Codec<String> {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ SchemaOneKeyCodec.class);
+
+ @Override
+ public byte[] toPersistedFormat(String stringObject) {
+ try {
+ // If the caller's string has no prefix, it should be stored as a long
+ // to be encoded as a long to be consistent with the schema one
+ // container format.
+ long longObject = Long.parseLong(stringObject);
+ return Longs.toByteArray(longObject);
+ } catch (NumberFormatException ex) {
+ // If long parsing fails, the caller used a prefix and the data should
+ // be encoded as a String.
+ return StringUtils.string2Bytes(stringObject);
+ }
+ }
+
+ /**
+ * Determines whether the byte array was originally stored as a string or a
+ * long, decodes the data in the corresponding format, and returns it as a
+ * String. Data is determined to have been stored as a string if it matches
+ * the known format of a metadata key or a prefixed block ID key. If the
+ * data does not match one of these formats, it will be parsed as a long
+ * only if it is 8 bytes long. Otherwise, it will be parsed as a
+ * String.
+ * <p>
+ * Note that it is technically possible, although highly unlikely, that
+ * {@code rawData} was originally encoded as a long, but also happens to match
+ * the regex of a known string format when decoded. In this case this method
+ * will decode the data as a string. Log trace messages have been added to
+ * help debug these errors if necessary.
+ *
+ * @param rawData Byte array from the key/value store. Should not be null.
+ */
+ @Override
+ public String fromPersistedFormat(byte[] rawData) throws IOException {
+ final String prefixedBlockRegex = "^#[a-zA-Z]+#[0-9]+$";
+ final String metadataRegex = "^#[a-zA-Z]$";
+
+ String stringData = StringUtils.bytes2String(rawData);
+
+ if (stringData.matches(prefixedBlockRegex)
+ || stringData.matches(metadataRegex)) {
+
+ LOG.trace("Byte array {} matched the format for a string key." +
+ " It will be parsed as the string {}", rawData, stringData);
+ return stringData;
+ } else if (rawData.length == Long.BYTES) {
+ long longData = Longs.fromByteArray(rawData);
+ LOG.trace("Byte array {} did not match the format for a string key " +
+ "and has {} bytes. It will be parsed as the long {}",
+ rawData, Long.BYTES, longData);
+
+ return Long.toString(longData);
+ } else {
+ LOG.trace("Byte array {} did not match the format for a string key " +
+ "and does not have {} bytes. It will be parsed as the string {}",
+ rawData, Long.BYTES, stringData);
+
+ return stringData;
+ }
+ }
+
+ @Override
+ public String copyObject(String object) {
+ return object;
+ }
+}
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/package-info.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/package-info.java
new file mode 100644
index 0000000..d1a80f2
--- /dev/null
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.hadoop.ozone.container.metadata;
+/**
+ This package contains classes that define the structure of the datanode's
+ database, and allow reading and writing to it.
+**/
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java
index 6cfb8ea..9d6d1cb 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.container.common;
import java.io.File;
import java.io.IOException;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -27,9 +28,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.google.common.collect.Lists;
-import com.google.common.primitives.Longs;
import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.MutableConfigurationSource;
@@ -38,10 +37,12 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.utils.BackgroundService;
import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.common.Checksum;
import org.apache.hadoop.ozone.container.ContainerTestHelper;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
import org.apache.hadoop.ozone.container.common.impl.ChunkLayOutVersion;
import org.apache.hadoop.ozone.container.common.impl.ContainerData;
import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
@@ -75,8 +76,6 @@ import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_BLOCK_COUNT_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_PENDING_DELETE_BLOCK_COUNT_KEY;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -165,8 +164,8 @@ public class TestBlockDeletingService {
chunks.add(info);
}
kd.setChunks(chunks);
- metadata.getStore().put(StringUtils.string2Bytes(deleteStateName),
- kd.getProtoBufMessage().toByteArray());
+ metadata.getStore().getBlockDataTable().put(
+ deleteStateName, kd);
container.getContainerData().incrPendingDeletionBlocks(1);
}
@@ -174,12 +173,14 @@ public class TestBlockDeletingService {
container.getContainerData().setBytesUsed(
blockLength * numOfBlocksPerContainer);
// Set block count, bytes used and pending delete block count.
- metadata.getStore().put(DB_BLOCK_COUNT_KEY,
- Longs.toByteArray(numOfBlocksPerContainer));
- metadata.getStore().put(OzoneConsts.DB_CONTAINER_BYTES_USED_KEY,
- Longs.toByteArray(blockLength * numOfBlocksPerContainer));
- metadata.getStore().put(DB_PENDING_DELETE_BLOCK_COUNT_KEY,
- Longs.toByteArray(numOfBlocksPerContainer));
+ metadata.getStore().getMetadataTable().put(
+ OzoneConsts.BLOCK_COUNT, (long)numOfBlocksPerContainer);
+ metadata.getStore().getMetadataTable().put(
+ OzoneConsts.CONTAINER_BYTES_USED,
+ blockLength * numOfBlocksPerContainer);
+ metadata.getStore().getMetadataTable().put(
+ OzoneConsts.PENDING_DELETE_BLOCK_COUNT,
+ (long)numOfBlocksPerContainer);
}
}
}
@@ -200,19 +201,14 @@ public class TestBlockDeletingService {
*/
private int getUnderDeletionBlocksCount(ReferenceCountedDB meta)
throws IOException {
- List<Map.Entry<byte[], byte[]>> underDeletionBlocks =
- meta.getStore().getRangeKVs(null, 100,
- new MetadataKeyFilters.KeyPrefixFilter()
- .addFilter(OzoneConsts.DELETING_KEY_PREFIX));
- return underDeletionBlocks.size();
+ return meta.getStore().getBlockDataTable()
+ .getRangeKVs(null, 100,
+ MetadataKeyFilters.getDeletingKeyFilter()).size();
}
private int getDeletedBlocksCount(ReferenceCountedDB db) throws IOException {
- List<Map.Entry<byte[], byte[]>> underDeletionBlocks =
- db.getStore().getRangeKVs(null, 100,
- new MetadataKeyFilters.KeyPrefixFilter()
- .addFilter(OzoneConsts.DELETED_KEY_PREFIX));
- return underDeletionBlocks.size();
+ return db.getStore().getDeletedBlocksTable()
+ .getRangeKVs(null, 100).size();
}
@Test
@@ -250,8 +246,9 @@ public class TestBlockDeletingService {
// Ensure there are 3 blocks under deletion and 0 deleted blocks
Assert.assertEquals(3, getUnderDeletionBlocksCount(meta));
- Assert.assertEquals(3, Longs.fromByteArray(
- meta.getStore().get(DB_PENDING_DELETE_BLOCK_COUNT_KEY)));
+ Assert.assertEquals(3,
+ meta.getStore().getMetadataTable()
+ .get(OzoneConsts.PENDING_DELETE_BLOCK_COUNT).longValue());
Assert.assertEquals(0, getDeletedBlocksCount(meta));
// An interval will delete 1 * 2 blocks
@@ -270,10 +267,12 @@ public class TestBlockDeletingService {
// Check finally DB counters.
// Not checking bytes used, as handler is a mock call.
- Assert.assertEquals(0, Longs.fromByteArray(
- meta.getStore().get(DB_PENDING_DELETE_BLOCK_COUNT_KEY)));
- Assert.assertEquals(0, Longs.fromByteArray(
- meta.getStore().get(DB_BLOCK_COUNT_KEY)));
+ Assert.assertEquals(0,
+ meta.getStore().getMetadataTable()
+ .get(OzoneConsts.PENDING_DELETE_BLOCK_COUNT).longValue());
+ Assert.assertEquals(0,
+ meta.getStore().getMetadataTable()
+ .get(OzoneConsts.BLOCK_COUNT).longValue());
}
svc.shutdown();
@@ -470,4 +469,59 @@ public class TestBlockDeletingService {
service.shutdown();
}
}
+
+ @Test
+ public void testDeletedChunkInfo() throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setInt(OZONE_BLOCK_DELETING_CONTAINER_LIMIT_PER_INTERVAL, 10);
+ conf.setInt(OZONE_BLOCK_DELETING_LIMIT_PER_CONTAINER, 2);
+ ContainerSet containerSet = new ContainerSet();
+ createToDeleteBlocks(containerSet, conf, 1, 2, 3);
+
+ List<ContainerData> containerData = Lists.newArrayList();
+ containerSet.listContainer(0L, 1, containerData);
+
+ try(ReferenceCountedDB meta = BlockUtils.getDB(
+ (KeyValueContainerData) containerData.get(0), conf)) {
+
+ // Collect all ChunkInfo from blocks marked for deletion.
+ List<? extends Table.KeyValue<String, BlockData>> deletingBlocks =
+ meta.getStore().getBlockDataTable()
+ .getRangeKVs(null, 100,
+ MetadataKeyFilters.getDeletingKeyFilter());
+
+ // Delete all blocks marked for deletion.
+ BlockDeletingServiceTestImpl svc =
+ getBlockDeletingService(containerSet, conf);
+ svc.start();
+ GenericTestUtils.waitFor(svc::isStarted, 100, 3000);
+ deleteAndWait(svc, 1);
+ svc.shutdown();
+
+ // Get deleted blocks from their table, and check their ChunkInfo lists
+ // against those we saved for them before deletion.
+ List<? extends Table.KeyValue<String, ChunkInfoList>> deletedBlocks =
+ meta.getStore().getDeletedBlocksTable()
+ .getRangeKVs(null, 100);
+
+ Assert.assertEquals(deletingBlocks.size(), deletedBlocks.size());
+
+ Iterator<? extends Table.KeyValue<String, BlockData>>
+ deletingBlocksIter = deletingBlocks.iterator();
+ Iterator<? extends Table.KeyValue<String, ChunkInfoList>>
+ deletedBlocksIter = deletedBlocks.iterator();
+
+ while(deletingBlocksIter.hasNext() && deletedBlocksIter.hasNext()) {
+ List<ContainerProtos.ChunkInfo> deletingChunks =
+ deletingBlocksIter.next().getValue().getChunks();
+ List<ContainerProtos.ChunkInfo> deletedChunks =
+ deletedBlocksIter.next().getValue().asList();
+
+ // On each element of each list, this call uses the equals method
+ // for ChunkInfos generated by protobuf.
+ // This checks their internal fields for equality.
+ Assert.assertEquals(deletingChunks, deletedChunks);
+ }
+ }
+ }
}
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestContainerCache.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestContainerCache.java
index 2e38990..c00bab5 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestContainerCache.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestContainerCache.java
@@ -21,10 +21,11 @@ package org.apache.hadoop.ozone.container.common;
import org.apache.hadoop.fs.FileSystemTestHelper;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.utils.ContainerCache;
import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
-import org.apache.hadoop.hdds.utils.MetadataStore;
-import org.apache.hadoop.hdds.utils.MetadataStoreBuilder;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStore;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStoreSchemaTwoImpl;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -51,14 +52,14 @@ public class TestContainerCache {
private void createContainerDB(OzoneConfiguration conf, File dbFile)
throws Exception {
- MetadataStore store = MetadataStoreBuilder.newBuilder().setConf(conf)
- .setCreateIfMissing(true).setDbFile(dbFile).build();
+ DatanodeStore store = new DatanodeStoreSchemaTwoImpl(
+ conf, 1, dbFile.getAbsolutePath());
// we close since the SCM pre-creates containers.
// we will open and put Db handle into a cache when keys are being created
// in a container.
- store.close();
+ store.stop();
}
@Test
@@ -85,17 +86,17 @@ public class TestContainerCache {
// Get 2 references out of the same db and verify the objects are same.
ReferenceCountedDB db1 = cache.getDB(1, "RocksDB",
- containerDir1.getPath(), conf);
+ containerDir1.getPath(), OzoneConsts.SCHEMA_LATEST, conf);
Assert.assertEquals(1, db1.getReferenceCount());
ReferenceCountedDB db2 = cache.getDB(1, "RocksDB",
- containerDir1.getPath(), conf);
+ containerDir1.getPath(), OzoneConsts.SCHEMA_LATEST, conf);
Assert.assertEquals(2, db2.getReferenceCount());
Assert.assertEquals(2, db1.getReferenceCount());
Assert.assertEquals(db1, db2);
// add one more references to ContainerCache.
ReferenceCountedDB db3 = cache.getDB(2, "RocksDB",
- containerDir2.getPath(), conf);
+ containerDir2.getPath(), OzoneConsts.SCHEMA_LATEST, conf);
Assert.assertEquals(1, db3.getReferenceCount());
// and close the reference
@@ -105,7 +106,7 @@ public class TestContainerCache {
// add one more reference to ContainerCache and verify that it will not
// evict the least recent entry as it has reference.
ReferenceCountedDB db4 = cache.getDB(3, "RocksDB",
- containerDir3.getPath(), conf);
+ containerDir3.getPath(), OzoneConsts.SCHEMA_LATEST, conf);
Assert.assertEquals(1, db4.getReferenceCount());
Assert.assertEquals(2, cache.size());
@@ -121,7 +122,7 @@ public class TestContainerCache {
// The reference count for container1 is 0 but it is not evicted.
ReferenceCountedDB db5 = cache.getDB(1, "RocksDB",
- containerDir1.getPath(), conf);
+ containerDir1.getPath(), OzoneConsts.SCHEMA_LATEST, conf);
Assert.assertEquals(1, db5.getReferenceCount());
Assert.assertEquals(db1, db5);
db5.close();
@@ -150,7 +151,7 @@ public class TestContainerCache {
Runnable task = () -> {
try {
ReferenceCountedDB db1 = cache.getDB(1, "RocksDB",
- containerDir.getPath(), conf);
+ containerDir.getPath(), OzoneConsts.SCHEMA_LATEST, conf);
Assert.assertNotNull(db1);
} catch (IOException e) {
Assert.fail("Should get the DB instance");
@@ -168,7 +169,7 @@ public class TestContainerCache {
}
ReferenceCountedDB db = cache.getDB(1, "RocksDB",
- containerDir.getPath(), conf);
+ containerDir.getPath(), OzoneConsts.SCHEMA_LATEST, conf);
db.close();
db.close();
db.close();
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestKeyValueContainerData.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestKeyValueContainerData.java
index de80c6a..7171a81 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestKeyValueContainerData.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestKeyValueContainerData.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.container.common;
import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.impl.ChunkLayOutVersion;
import org.apache.hadoop.ozone.container.keyvalue.ChunkLayoutTestInfo;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
@@ -92,6 +93,7 @@ public class TestKeyValueContainerData {
kvData.incrWriteCount();
kvData.incrKeyCount();
kvData.incrPendingDeletionBlocks(1);
+ kvData.setSchemaVersion(OzoneConsts.SCHEMA_LATEST);
assertEquals(state, kvData.getState());
assertEquals(containerDBType, kvData.getContainerDBType());
@@ -106,6 +108,7 @@ public class TestKeyValueContainerData {
assertEquals(1, kvData.getNumPendingDeletionBlocks());
assertEquals(pipelineId.toString(), kvData.getOriginPipelineId());
assertEquals(datanodeId.toString(), kvData.getOriginNodeId());
+ assertEquals(OzoneConsts.SCHEMA_LATEST, kvData.getSchemaVersion());
}
}
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaOneBackwardsCompatibility.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaOneBackwardsCompatibility.java
new file mode 100644
index 0000000..01fa3bf
--- /dev/null
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaOneBackwardsCompatibility.java
@@ -0,0 +1,626 @@
+/*
+ * 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.hadoop.ozone.container.common;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
+import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml;
+import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
+import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
+import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
+import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer;
+import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
+import org.apache.hadoop.ozone.container.keyvalue.KeyValueHandler;
+import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
+import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerUtil;
+import org.apache.hadoop.ozone.container.metadata.DatanodeStore;
+import org.apache.hadoop.ozone.container.metadata.SchemaOneDeletedBlocksTable;
+import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
+import org.apache.hadoop.ozone.container.testutils.BlockDeletingServiceTestImpl;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.yaml.snakeyaml.Yaml;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_CONTAINER_LIMIT_PER_INTERVAL;
+import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests processing of containers written with DB schema version 1,
+ * which stores all its data in the default RocksDB column family.
+ * Newer schema version will use a different column family layout, but they
+ * should still be able to read, delete data, and update metadata for schema
+ * version 1 containers.
+ * <p>
+ * The functionality executed by these tests assumes that all containers will
+ * have to be closed before an upgrade, meaning that containers written with
+ * schema version 1 will only ever be encountered in their closed state.
+ * <p>
+ * For an example of a RocksDB instance written with schema version 1, see
+ * {@link TestDB}, which is used by these tests to load a pre created schema
+ * version 1 RocksDB instance from test resources.
+ */
+public class TestSchemaOneBackwardsCompatibility {
+ private OzoneConfiguration conf;
+
+ private File metadataDir;
+ private File containerFile;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ public void setup() throws Exception {
+ conf = new OzoneConfiguration();
+ TestDB testDB = new TestDB();
+
+ // Copy data to the temporary folder so it can be safely modified.
+ File tempMetadataDir =
+ tempFolder.newFolder(Long.toString(TestDB.CONTAINER_ID),
+ OzoneConsts.CONTAINER_META_PATH);
+
+ FileUtils.copyDirectoryToDirectory(testDB.getDBDirectory(),
+ tempMetadataDir);
+ FileUtils.copyFileToDirectory(testDB.getContainerFile(), tempMetadataDir);
+
+ metadataDir = tempMetadataDir;
+ containerFile = new File(metadataDir, TestDB.CONTAINER_FILE_NAME);
+ }
+
+ /**
+ * Because all tables in schema version one map back to the default table,
+ * directly iterating any of the table instances should be forbidden.
+ * Otherwise, the iterators for each table would read the entire default
+ * table, return all database contents, and yield unexpected results.
+ * @throws Exception
+ */
+ @Test
+ public void testDirectTableIterationDisabled() throws Exception {
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ DatanodeStore store = refCountedDB.getStore();
+
+ assertTableIteratorUnsupported(store.getMetadataTable());
+ assertTableIteratorUnsupported(store.getBlockDataTable());
+ assertTableIteratorUnsupported(store.getDeletedBlocksTable());
+ }
+ }
+
+ private void assertTableIteratorUnsupported(Table<?, ?> table) {
+ try {
+ table.iterator();
+ Assert.fail("Table iterator should have thrown " +
+ "UnsupportedOperationException.");
+ } catch (UnsupportedOperationException ex) {
+ // Exception thrown as expected.
+ }
+ }
+
+ /**
+ * Counts the number of deleted, pending delete, and regular blocks in the
+ * database, and checks that they match the expected values.
+ * Also makes sure that internal prefixes used to manage data in the schema
+ * one deleted blocks table are removed from keys in iterator results.
+ * @throws IOException
+ */
+ @Test
+ public void testBlockIteration() throws IOException {
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ assertEquals(TestDB.NUM_DELETED_BLOCKS, countDeletedBlocks(refCountedDB));
+
+ assertEquals(TestDB.NUM_PENDING_DELETION_BLOCKS,
+ countDeletingBlocks(refCountedDB));
+
+ assertEquals(TestDB.KEY_COUNT - TestDB.NUM_PENDING_DELETION_BLOCKS,
+ countUnprefixedBlocks(refCountedDB));
+
+ // Test that deleted block keys do not have a visible prefix when
+ // iterating.
+ final String prefix = SchemaOneDeletedBlocksTable.DELETED_KEY_PREFIX;
+ Table<String, ChunkInfoList> deletedBlocksTable =
+ refCountedDB.getStore().getDeletedBlocksTable();
+
+ // Test rangeKVs.
+ List<? extends Table.KeyValue<String, ChunkInfoList>> deletedBlocks =
+ deletedBlocksTable.getRangeKVs(null, 100);
+
+ for (Table.KeyValue<String, ChunkInfoList> kv: deletedBlocks) {
+ assertFalse(kv.getKey().contains(prefix));
+ }
+
+ // Test sequentialRangeKVs.
+ deletedBlocks = deletedBlocksTable.getRangeKVs(null, 100);
+
+ for (Table.KeyValue<String, ChunkInfoList> kv: deletedBlocks) {
+ assertFalse(kv.getKey().contains(prefix));
+ }
+ }
+
+ }
+
+ /**
+ * Tests reading of a container that was written in schema version 1, when
+ * the container has metadata keys present.
+ * The {@link KeyValueContainerUtil} will read these values to fill in a
+ * {@link KeyValueContainerData} object.
+ * @throws Exception
+ */
+ @Test
+ public void testReadWithMetadata() throws Exception {
+ checkContainerData(newKvData());
+ }
+
+ /**
+ * Tests reading of a container that was written in schema version 1, when
+ * the container has no metadata keys present.
+ * The {@link KeyValueContainerUtil} will scan the blocks in the database
+ * to fill these metadata values into the database and into a
+ * {@link KeyValueContainerData} object.
+ * @throws Exception
+ */
+ @Test
+ public void testReadWithoutMetadata() throws Exception {
+ // Delete metadata keys from our copy of the DB.
+ // This simulates them not being there to start with.
+ try (ReferenceCountedDB db = BlockUtils.getDB(newKvData(), conf)) {
+ Table<String, Long> metadataTable = db.getStore().getMetadataTable();
+
+ metadataTable.delete(OzoneConsts.BLOCK_COUNT);
+ assertNull(metadataTable.get(OzoneConsts.BLOCK_COUNT));
+
+ metadataTable.delete(OzoneConsts.CONTAINER_BYTES_USED);
+ assertNull(metadataTable.get(OzoneConsts.CONTAINER_BYTES_USED));
+
+ metadataTable.delete(OzoneConsts.PENDING_DELETE_BLOCK_COUNT);
+ assertNull(metadataTable.get(OzoneConsts.PENDING_DELETE_BLOCK_COUNT));
+ }
+
+ // Create a new container data object, and fill in its metadata by
+ // counting blocks from the database, since the metadata keys in the
+ // database are now gone.
+ checkContainerData(newKvData());
+ }
+
+ /**
+ * Tests reading blocks marked for deletion from a container written in
+ * schema version 1. Because the block deleting service both reads for
+ * deleted blocks and deletes them, this test will modify its copy of the
+ * database.
+ */
+ @Test
+ public void testDelete() throws Exception {
+ final long numBlocksToDelete = TestDB.NUM_PENDING_DELETION_BLOCKS;
+
+ runBlockDeletingService();
+
+ // Expected values after blocks with #deleting# prefix in original DB are
+ // deleted.
+ final long expectedDeletingBlocks =
+ TestDB.NUM_PENDING_DELETION_BLOCKS - numBlocksToDelete;
+ final long expectedDeletedBlocks =
+ TestDB.NUM_DELETED_BLOCKS + numBlocksToDelete;
+ final long expectedRegularBlocks =
+ TestDB.KEY_COUNT - numBlocksToDelete;
+
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ // Test results via block iteration.
+ assertEquals(expectedDeletingBlocks,
+ countDeletingBlocks(refCountedDB));
+ assertEquals(expectedDeletedBlocks,
+ countDeletedBlocks(refCountedDB));
+ assertEquals(expectedRegularBlocks,
+ countUnprefixedBlocks(refCountedDB));
+
+ // Test table metadata.
+ // Because the KeyValueHandler used for the block deleting service is
+ // mocked, the bytes used will not be updated by a ChunkManager in the
+ // KeyValueHandler. Therefore, this value is not checked.
+ Table<String, Long> metadataTable =
+ refCountedDB.getStore().getMetadataTable();
+ assertEquals(expectedRegularBlocks + expectedDeletingBlocks,
+ (long)metadataTable.get(OzoneConsts.BLOCK_COUNT));
+ }
+ }
+
+ /**
+ * Tests reading the chunk info saved from a block that was deleted from a
+ * database in schema version one. Blocks deleted from schema version one
+ * before the upgrade will have the block ID saved as their value. Trying
+ * to retrieve this value as a {@link ChunkInfoList} should fail. Blocks
+ * deleted from schema version one after the upgrade should have their
+ * {@link ChunkInfoList} saved as the corresponding value in the deleted
+ * blocks table. Reading these values should succeed.
+ * @throws Exception
+ */
+ @Test
+ public void testReadDeletedBlockChunkInfo() throws Exception {
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ // Read blocks that were already deleted before the upgrade.
+ List<? extends Table.KeyValue<String, ChunkInfoList>> deletedBlocks =
+ refCountedDB.getStore()
+ .getDeletedBlocksTable().getRangeKVs(null, 100);
+
+ Set<String> preUpgradeBlocks = new HashSet<>();
+
+ for(Table.KeyValue<String, ChunkInfoList> chunkListKV: deletedBlocks) {
+ preUpgradeBlocks.add(chunkListKV.getKey());
+ try {
+ chunkListKV.getValue();
+ Assert.fail("No exception thrown when trying to retrieve old " +
+ "deleted blocks values as chunk lists.");
+ } catch(IOException ex) {
+ // Exception thrown as expected.
+ }
+ }
+
+ Assert.assertEquals(TestDB.NUM_DELETED_BLOCKS, preUpgradeBlocks.size());
+
+ runBlockDeletingService();
+
+ // After the block deleting service runs, get the updated list of
+ // deleted blocks.
+ deletedBlocks = refCountedDB.getStore()
+ .getDeletedBlocksTable().getRangeKVs(null, 100);
+
+ int numPostUpgradeDeletesFound = 0;
+ for(Table.KeyValue<String, ChunkInfoList> chunkListKV: deletedBlocks) {
+ if (!preUpgradeBlocks.contains(chunkListKV.getKey())) {
+ numPostUpgradeDeletesFound++;
+ Assert.assertNotNull(chunkListKV.getValue());
+ }
+ }
+
+ // The blocks that were originally marked for deletion should now be
+ // deleted.
+ Assert.assertEquals(TestDB.NUM_PENDING_DELETION_BLOCKS,
+ numPostUpgradeDeletesFound);
+ }
+ }
+
+ @Test
+ public void testReadBlockData() throws Exception {
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ Table<String, BlockData> blockDataTable =
+ refCountedDB.getStore().getBlockDataTable();
+
+ // Test encoding keys and decoding database values.
+ for (String blockID: TestDB.BLOCK_IDS) {
+ BlockData blockData = blockDataTable.get(blockID);
+ Assert.assertEquals(Long.toString(blockData.getLocalID()), blockID);
+ }
+
+ // Test decoding keys from the database.
+ List<? extends Table.KeyValue<String, BlockData>> blockKeyValues =
+ blockDataTable.getRangeKVs(null, 100,
+ MetadataKeyFilters.getUnprefixedKeyFilter());
+
+ List<String> decodedKeys = new ArrayList<>();
+
+ for (Table.KeyValue<String, BlockData> blockDataKV:
+ blockKeyValues) {
+ decodedKeys.add(blockDataKV.getKey());
+ }
+
+ Assert.assertEquals(TestDB.BLOCK_IDS, decodedKeys);
+
+ // Test reading blocks with block iterator.
+ try(BlockIterator<BlockData> iter =
+ refCountedDB.getStore().getBlockIterator()) {
+
+ List<String> iteratorBlockIDs = new ArrayList<>();
+
+ while(iter.hasNext()) {
+ long localID = iter.nextBlock().getBlockID().getLocalID();
+ iteratorBlockIDs.add(Long.toString(localID));
+ }
+
+ Assert.assertEquals(TestDB.BLOCK_IDS, iteratorBlockIDs);
+ }
+ }
+ }
+
+ @Test
+ public void testReadDeletingBlockData() throws Exception {
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ Table<String, BlockData> blockDataTable =
+ refCountedDB.getStore().getBlockDataTable();
+
+ for (String blockID: TestDB.DELETING_BLOCK_IDS) {
+ BlockData blockData =
+ blockDataTable.get(OzoneConsts.DELETING_KEY_PREFIX + blockID);
+ Assert.assertEquals(Long.toString(blockData.getLocalID()), blockID);
+ }
+
+ // Test decoding keys from the database.
+ List<? extends Table.KeyValue<String, BlockData>> blockKeyValues =
+ blockDataTable.getRangeKVs(null, 100,
+ MetadataKeyFilters.getDeletingKeyFilter());
+
+ List<String> decodedKeys = new ArrayList<>();
+
+ for (Table.KeyValue<String, BlockData> blockDataKV:
+ blockKeyValues) {
+ decodedKeys.add(blockDataKV.getKey());
+ }
+
+ // Apply the deleting prefix to the saved block IDs so we can compare
+ // them to the retrieved keys.
+ List<String> expectedKeys = TestDB.DELETING_BLOCK_IDS.stream()
+ .map(key -> OzoneConsts.DELETING_KEY_PREFIX + key)
+ .collect(Collectors.toList());
+
+ Assert.assertEquals(expectedKeys, decodedKeys);
+
+ // Test reading deleting blocks with block iterator.
+ MetadataKeyFilters.KeyPrefixFilter filter =
+ MetadataKeyFilters.getDeletingKeyFilter();
+
+ try(BlockIterator<BlockData> iter =
+ refCountedDB.getStore().getBlockIterator(filter)) {
+
+ List<String> iteratorBlockIDs = new ArrayList<>();
+
+ while(iter.hasNext()) {
+ long localID = iter.nextBlock().getBlockID().getLocalID();
+ iteratorBlockIDs.add(Long.toString(localID));
+ }
+
+ Assert.assertEquals(TestDB.DELETING_BLOCK_IDS, iteratorBlockIDs);
+ }
+ }
+ }
+
+ @Test
+ public void testReadMetadata() throws Exception {
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ Table<String, Long> metadataTable =
+ refCountedDB.getStore().getMetadataTable();
+
+ Assert.assertEquals(TestDB.KEY_COUNT,
+ metadataTable.get(OzoneConsts.BLOCK_COUNT).longValue());
+ Assert.assertEquals(TestDB.BYTES_USED,
+ metadataTable.get(OzoneConsts.CONTAINER_BYTES_USED).longValue());
+ Assert.assertEquals(TestDB.NUM_PENDING_DELETION_BLOCKS,
+ metadataTable.get(OzoneConsts.PENDING_DELETE_BLOCK_COUNT)
+ .longValue());
+ }
+ }
+
+ @Test
+ public void testReadDeletedBlocks() throws Exception {
+ try(ReferenceCountedDB refCountedDB = BlockUtils.getDB(newKvData(), conf)) {
+ Table<String, ChunkInfoList> deletedBlocksTable =
+ refCountedDB.getStore().getDeletedBlocksTable();
+
+ for (String blockID: TestDB.DELETED_BLOCK_IDS) {
+ // Since chunk info for deleted blocks was not stored in schema
+ // version 1, there is no value to retrieve here.
+ Assert.assertTrue(deletedBlocksTable.isExist(blockID));
+ }
+
+ // Test decoding keys from the database.
+ List<? extends Table.KeyValue<String, ChunkInfoList>> chunkInfoKeyValues =
+ deletedBlocksTable.getRangeKVs(null, 100);
+
+ List<String> decodedKeys = new ArrayList<>();
+
+ for (Table.KeyValue<String, ChunkInfoList> kv:
+ chunkInfoKeyValues) {
+ decodedKeys.add(kv.getKey());
+ }
+
+ Assert.assertEquals(TestDB.DELETED_BLOCK_IDS, decodedKeys);
+ }
+ }
+
+ private void runBlockDeletingService() throws Exception {
+ conf.setInt(OZONE_BLOCK_DELETING_CONTAINER_LIMIT_PER_INTERVAL, 10);
+ conf.setInt(OzoneConfigKeys.OZONE_BLOCK_DELETING_LIMIT_PER_CONTAINER, 2);
+ conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY,
+ metadataDir.getAbsolutePath());
+
+ OzoneContainer container = makeMockOzoneContainer();
+
+ BlockDeletingServiceTestImpl service =
+ new BlockDeletingServiceTestImpl(container, 1000, conf);
+ service.start();
+ GenericTestUtils.waitFor(service::isStarted, 100, 3000);
+ service.runDeletingTasks();
+ GenericTestUtils.waitFor(() -> service.getTimesOfProcessed() == 1,
+ 100, 3000);
+ }
+
+ private ContainerSet makeContainerSet() throws Exception {
+ ContainerSet containerSet = new ContainerSet();
+ KeyValueContainer container = new KeyValueContainer(newKvData(), conf);
+ containerSet.addContainer(container);
+
+ return containerSet;
+ }
+
+ private OzoneContainer makeMockOzoneContainer() throws Exception {
+ ContainerSet containerSet = makeContainerSet();
+
+ OzoneContainer ozoneContainer = mock(OzoneContainer.class);
+ when(ozoneContainer.getContainerSet()).thenReturn(containerSet);
+ when(ozoneContainer.getWriteChannel()).thenReturn(null);
+ ContainerDispatcher dispatcher = mock(ContainerDispatcher.class);
+ when(ozoneContainer.getDispatcher()).thenReturn(dispatcher);
+ KeyValueHandler handler = mock(KeyValueHandler.class);
+ when(dispatcher.getHandler(any())).thenReturn(handler);
+
+ return ozoneContainer;
+ }
+
+ /**
+ * @return A {@link KeyValueContainerData} object that is read from
+ * {@link TestSchemaOneBackwardsCompatibility#containerFile} constructed to
+ * point to the instance of {@link TestDB} being used.
+ * @throws IOException
+ */
+ private KeyValueContainerData newKvData() throws IOException {
+ KeyValueContainerData kvData = (KeyValueContainerData)
+ ContainerDataYaml.readContainerFile(containerFile);
+
+ // Because the test DB is set up in a temp folder, we cannot know any
+ // absolute paths to container components until run time.
+ // For this reason, any path fields are omitted from the container file,
+ // and added here.
+ kvData.setMetadataPath(metadataDir.getAbsolutePath());
+ kvData.setChunksPath(metadataDir.getAbsolutePath());
+
+ // Changing the paths above affects the checksum, so it was also removed
+ // from the container file and calculated at run time.
+ Yaml yaml = ContainerDataYaml.getYamlForContainerType(
+ kvData.getContainerType());
+ kvData.computeAndSetChecksum(yaml);
+
+ KeyValueContainerUtil.parseKVContainerData(kvData, conf);
+
+ return kvData;
+ }
+
+ /**
+ * @param kvData The container data that will be tested to see if it has
+ * metadata values matching those in the database under test.
+ */
+ private void checkContainerData(KeyValueContainerData kvData) {
+ assertTrue(kvData.isClosed());
+ assertEquals(TestDB.SCHEMA_VERSION, kvData.getSchemaVersion());
+ assertEquals(TestDB.KEY_COUNT, kvData.getKeyCount());
+ assertEquals(TestDB.BYTES_USED, kvData.getBytesUsed());
+ assertEquals(TestDB.NUM_PENDING_DELETION_BLOCKS,
+ kvData.getNumPendingDeletionBlocks());
+ }
+
+ private int countDeletedBlocks(ReferenceCountedDB refCountedDB)
+ throws IOException {
+ return refCountedDB.getStore().getDeletedBlocksTable()
+ .getRangeKVs(null, 100,
+ MetadataKeyFilters.getUnprefixedKeyFilter()).size();
+ }
+
+ private int countDeletingBlocks(ReferenceCountedDB refCountedDB)
+ throws IOException {
+ return refCountedDB.getStore().getBlockDataTable()
+ .getRangeKVs(null, 100,
+ MetadataKeyFilters.getDeletingKeyFilter()).size();
+ }
+
+ private int countUnprefixedBlocks(ReferenceCountedDB refCountedDB)
+ throws IOException {
+ return refCountedDB.getStore().getBlockDataTable()
+ .getRangeKVs(null, 100,
+ MetadataKeyFilters.getUnprefixedKeyFilter()).size();
+ }
+
+ /**
+ * Holds information about the database used for testing by this class.
+ * This database was generated by an old version of the code that wrote
+ * data in schema version 1. The values are arbitrary. We only care that
+ * it has keys representing metadata, deleted blocks, pending delete blocks,
+ * and regular blocks.
+ * <p>
+ * The contents of the database are listed below. All values are present in
+ * the default column family. String values are surrounded by double
+ * quotes, protobuf objects are surrounded by angle brackets, and longs are
+ * written literally.
+ *
+ * "#BLOCKCOUNT" : 4
+ * "#BYTESUSED" : 400
+ * "#PENDINGDELETEBLOCKCOUNT" : 2
+ * "#deleted#1596029079371" : 1596029079371
+ * "#deleted#1596029079374" : 1596029079374
+ * "#deleting#1596029079378" : <block_data>
+ * "#deleting#1596029079380" : <block_data>
+ * 1596029079382 : <block_data>
+ * 1596029079385 : <block_data>
+ */
+ private static class TestDB {
+ // Non configurable properties of database.
+ public static final long CONTAINER_ID = 123;
+ public static final String CONTAINER_FILE_NAME =
+ CONTAINER_ID + OzoneConsts.CONTAINER_EXTENSION;
+ public static final String DB_NAME =
+ CONTAINER_ID + OzoneConsts.DN_CONTAINER_DB;
+
+ public static final String SCHEMA_VERSION = OzoneConsts.SCHEMA_V1;
+ public static final long KEY_COUNT = 4;
+ public static final long BYTES_USED = 400;
+ public static final long NUM_PENDING_DELETION_BLOCKS = 2;
+ public static final long NUM_DELETED_BLOCKS = 2;
+
+ // All keys are stored as strings, since the codecs reading to the
+ // database should handle conversion from long to string for keys before
+ // data is returned to the caller.
+ public static final List<String> DELETED_BLOCK_IDS = Arrays.asList(
+ "1596029079371",
+ "1596029079374");
+ public static final List<String> DELETING_BLOCK_IDS = Arrays.asList(
+ "1596029079378",
+ "1596029079380");
+ public static final List<String> BLOCK_IDS = Arrays.asList(
+ "1596029079382",
+ "1596029079385");
+
+ private final ClassLoader loader;
+
+ TestDB() {
+ loader = getClass().getClassLoader();
+ }
+
+ public File getContainerFile() {
+ return load(CONTAINER_FILE_NAME);
+ }
+
+ public File getDBDirectory() {
+ return load(DB_NAME);
+ }
+
+ private File load(String name) {
+ File file = null;
+ URL url = loader.getResource(name);
+
+ if (url != null) {
+ file = new File(url.getFile());
+ }
+
+ return file;
+ }
+ }
+}
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerDataYaml.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerDataYaml.java
index d78cf12..f4e21dc 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerDataYaml.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerDataYaml.java
@@ -85,6 +85,7 @@ public class TestContainerDataYaml {
keyValueContainerData.setMetadataPath(testRoot);
keyValueContainerData.setChunksPath(testRoot);
keyValueContainerData.updateDataScanTime(SCAN_TIME);
+ keyValueContainerData.setSchemaVersion(OzoneConsts.SCHEMA_LATEST);
File containerFile = new File(testRoot, containerPath);
@@ -127,6 +128,8 @@ public class TestContainerDataYaml {
assertEquals(SCAN_TIME, kvData.lastDataScanTime().get());
assertEquals(SCAN_TIME.toEpochMilli(),
kvData.getDataScanTimestamp().longValue());
+ assertEquals(OzoneConsts.SCHEMA_LATEST,
+ kvData.getSchemaVersion());
// Update ContainerData.
kvData.addMetadata(OzoneConsts.VOLUME, VOLUME_OWNER);
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java
index aff0528..c29e5df 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java
@@ -19,34 +19,30 @@
package org.apache.hadoop.ozone.container.keyvalue;
import java.io.File;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.UUID;
+import java.util.*;
import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.fs.FileUtil;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
import org.apache.hadoop.ozone.container.common.impl.ChunkLayOutVersion;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
import org.apache.hadoop.ozone.container.common.volume.RoundRobinVolumeChoosingPolicy;
import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
import org.apache.hadoop.test.GenericTestUtils;
-import com.google.common.primitives.Longs;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_DATANODE_DIR_KEY;
import static org.apache.hadoop.ozone.container.common.impl.ChunkLayOutVersion.FILE_PER_BLOCK;
import static org.apache.hadoop.ozone.container.common.impl.ChunkLayOutVersion.FILE_PER_CHUNK;
+
import org.junit.After;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -62,12 +58,14 @@ import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
public class TestKeyValueBlockIterator {
+ private static final long CONTAINER_ID = 105L;
+
private KeyValueContainer container;
private KeyValueContainerData containerData;
private MutableVolumeSet volumeSet;
private OzoneConfiguration conf;
private File testRoot;
-
+ private ReferenceCountedDB db;
private final ChunkLayOutVersion layout;
public TestKeyValueBlockIterator(ChunkLayOutVersion layout) {
@@ -88,146 +86,150 @@ public class TestKeyValueBlockIterator {
conf = new OzoneConfiguration();
conf.set(HDDS_DATANODE_DIR_KEY, testRoot.getAbsolutePath());
volumeSet = new MutableVolumeSet(UUID.randomUUID().toString(), conf);
+
+ containerData = new KeyValueContainerData(105L,
+ layout,
+ (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(),
+ UUID.randomUUID().toString());
+ // Init the container.
+ container = new KeyValueContainer(containerData, conf);
+ container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), UUID
+ .randomUUID().toString());
+ db = BlockUtils.getDB(containerData, conf);
}
@After
- public void tearDown() {
+ public void tearDown() throws Exception {
+ db.close();
+ db.cleanup();
volumeSet.shutdown();
FileUtil.fullyDelete(testRoot);
}
@Test
public void testKeyValueBlockIteratorWithMixedBlocks() throws Exception {
-
- long containerID = 100L;
- int deletedBlocks = 5;
+ int deletingBlocks = 5;
int normalBlocks = 5;
- createContainerWithBlocks(containerID, normalBlocks, deletedBlocks);
- String containerPath = new File(containerData.getMetadataPath())
- .getParent();
- try(KeyValueBlockIterator keyValueBlockIterator = new KeyValueBlockIterator(
- containerID, new File(containerPath))) {
+ Map<String, List<Long>> blockIDs = createContainerWithBlocks(CONTAINER_ID,
+ normalBlocks, deletingBlocks);
- int counter = 0;
+ // Default filter used is all unprefixed blocks.
+ List<Long> unprefixedBlockIDs = blockIDs.get("");
+ try(BlockIterator<BlockData> keyValueBlockIterator =
+ db.getStore().getBlockIterator()) {
+
+ Iterator<Long> blockIDIter = unprefixedBlockIDs.iterator();
while (keyValueBlockIterator.hasNext()) {
BlockData blockData = keyValueBlockIterator.nextBlock();
- assertEquals(blockData.getLocalID(), counter++);
+ assertEquals(blockData.getLocalID(), (long)blockIDIter.next());
}
-
assertFalse(keyValueBlockIterator.hasNext());
+ assertFalse(blockIDIter.hasNext());
keyValueBlockIterator.seekToFirst();
- counter = 0;
+ blockIDIter = unprefixedBlockIDs.iterator();
while (keyValueBlockIterator.hasNext()) {
BlockData blockData = keyValueBlockIterator.nextBlock();
- assertEquals(blockData.getLocalID(), counter++);
+ assertEquals(blockData.getLocalID(), (long)blockIDIter.next());
}
assertFalse(keyValueBlockIterator.hasNext());
+ assertFalse(blockIDIter.hasNext());
try {
keyValueBlockIterator.nextBlock();
} catch (NoSuchElementException ex) {
GenericTestUtils.assertExceptionContains("Block Iterator reached end " +
- "for ContainerID " + containerID, ex);
+ "for ContainerID " + CONTAINER_ID, ex);
}
}
}
@Test
public void testKeyValueBlockIteratorWithNextBlock() throws Exception {
- long containerID = 101L;
- createContainerWithBlocks(containerID, 2, 0);
- String containerPath = new File(containerData.getMetadataPath())
- .getParent();
- try(KeyValueBlockIterator keyValueBlockIterator = new KeyValueBlockIterator(
- containerID, new File(containerPath))) {
- long blockID = 0L;
- assertEquals(blockID++, keyValueBlockIterator.nextBlock().getLocalID());
- assertEquals(blockID, keyValueBlockIterator.nextBlock().getLocalID());
+ List<Long> blockIDs = createContainerWithBlocks(CONTAINER_ID, 2);
+ try(BlockIterator<BlockData> keyValueBlockIterator =
+ db.getStore().getBlockIterator()) {
+ assertEquals((long)blockIDs.get(0),
+ keyValueBlockIterator.nextBlock().getLocalID());
+ assertEquals((long)blockIDs.get(1),
+ keyValueBlockIterator.nextBlock().getLocalID());
try {
keyValueBlockIterator.nextBlock();
} catch (NoSuchElementException ex) {
GenericTestUtils.assertExceptionContains("Block Iterator reached end " +
- "for ContainerID " + containerID, ex);
+ "for ContainerID " + CONTAINER_ID, ex);
}
}
}
@Test
public void testKeyValueBlockIteratorWithHasNext() throws Exception {
- long containerID = 102L;
- createContainerWithBlocks(containerID, 2, 0);
- String containerPath = new File(containerData.getMetadataPath())
- .getParent();
- try(KeyValueBlockIterator keyValueBlockIterator = new KeyValueBlockIterator(
- containerID, new File(containerPath))) {
- long blockID = 0L;
+ List<Long> blockIDs = createContainerWithBlocks(CONTAINER_ID, 2);
+ try(BlockIterator<BlockData> blockIter =
+ db.getStore().getBlockIterator()) {
// Even calling multiple times hasNext() should not move entry forward.
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertEquals(blockID++, keyValueBlockIterator.nextBlock().getLocalID());
-
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertTrue(keyValueBlockIterator.hasNext());
- assertEquals(blockID, keyValueBlockIterator.nextBlock().getLocalID());
-
- keyValueBlockIterator.seekToLast();
- assertTrue(keyValueBlockIterator.hasNext());
- assertEquals(blockID, keyValueBlockIterator.nextBlock().getLocalID());
-
- keyValueBlockIterator.seekToFirst();
- blockID = 0L;
- assertEquals(blockID++, keyValueBlockIterator.nextBlock().getLocalID());
- assertEquals(blockID, keyValueBlockIterator.nextBlock().getLocalID());
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertEquals((long)blockIDs.get(0),
+ blockIter.nextBlock().getLocalID());
+
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertTrue(blockIter.hasNext());
+ assertEquals((long)blockIDs.get(1), blockIter.nextBlock().getLocalID());
+
+ blockIter.seekToFirst();
+ assertEquals((long)blockIDs.get(0), blockIter.nextBlock().getLocalID());
+ assertEquals((long)blockIDs.get(1), blockIter.nextBlock().getLocalID());
try {
- keyValueBlockIterator.nextBlock();
+ blockIter.nextBlock();
} catch (NoSuchElementException ex) {
GenericTestUtils.assertExceptionContains("Block Iterator reached end " +
- "for ContainerID " + containerID, ex);
+ "for ContainerID " + CONTAINER_ID, ex);
}
}
}
@Test
public void testKeyValueBlockIteratorWithFilter() throws Exception {
- long containerId = 103L;
- int deletedBlocks = 10;
int normalBlocks = 5;
- createContainerWithBlocks(containerId, normalBlocks, deletedBlocks);
- String containerPath = new File(containerData.getMetadataPath())
- .getParent();
- try(KeyValueBlockIterator keyValueBlockIterator = new KeyValueBlockIterator(
- containerId, new File(containerPath), MetadataKeyFilters
- .getDeletingKeyFilter())) {
-
- int counter = 5;
+ int deletingBlocks = 5;
+ Map<String, List<Long>> blockIDs = createContainerWithBlocks(CONTAINER_ID,
+ normalBlocks, deletingBlocks);
+ try(BlockIterator<BlockData> keyValueBlockIterator =
+ db.getStore().getBlockIterator(
+ MetadataKeyFilters.getDeletingKeyFilter())) {
+ List<Long> deletingBlockIDs =
+ blockIDs.get(OzoneConsts.DELETING_KEY_PREFIX);
+ int counter = 0;
while (keyValueBlockIterator.hasNext()) {
BlockData blockData = keyValueBlockIterator.nextBlock();
- assertEquals(blockData.getLocalID(), counter++);
+ assertEquals((long)deletingBlockIDs.get(counter),
+ blockData.getLocalID());
+ counter++;
}
- assertEquals(10, counter);
+
+ assertEquals(deletingBlocks, counter);
}
}
@Test
public void testKeyValueBlockIteratorWithOnlyDeletedBlocks() throws
Exception {
- long containerId = 104L;
- createContainerWithBlocks(containerId, 0, 5);
+ createContainerWithBlocks(CONTAINER_ID, 0, 5);
String containerPath = new File(containerData.getMetadataPath())
.getParent();
- try(KeyValueBlockIterator keyValueBlockIterator = new KeyValueBlockIterator(
- containerId, new File(containerPath))) {
+ try(BlockIterator<BlockData> keyValueBlockIterator =
+ db.getStore().getBlockIterator()) {
//As all blocks are deleted blocks, blocks does not match with normal key
// filter.
assertFalse(keyValueBlockIterator.hasNext());
@@ -235,24 +237,131 @@ public class TestKeyValueBlockIterator {
}
/**
- * Creates a container with specified number of normal blocks and deleted
- * blocks. First it will insert normal blocks, and then it will insert
- * deleted blocks.
+ * Due to RocksDB internals, prefixed keys may be grouped all at the
+ * beginning or end of the key iteration, depending on the serialization
+ * used. Keys of the same prefix are grouped
+ * together. This method runs the same set of tests on the iterator first
+ * positively filtering one prefix, and then positively filtering
+ * a second prefix. If the sets of keys with prefix one, prefix
+ * two, and no prefixes are not empty, it follows that the filter will
+ * encounter both of the following cases:
+ *
+ * 1. A failing key followed by a passing key.
+ * 2. A passing key followed by a failing key.
+ *
+ * Note that with the current block data table implementation, there is
+ * only ever one type of prefix. This test adds a dummy second prefix type
+ * to ensure that the iterator will continue to work if more prefixes are
+ * added in the future.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testKeyValueBlockIteratorWithAdvancedFilter() throws
+ Exception {
+ // Block data table currently only uses one prefix type.
+ // Introduce a second prefix type to make sure the iterator functions
+ // correctly if more prefixes were to be added in the future.
+ final String secondPrefix = "#FOOBAR#";
+ Map<String, Integer> prefixCounts = new HashMap<>();
+ prefixCounts.put(OzoneConsts.DELETING_KEY_PREFIX, 3);
+ prefixCounts.put("", 3);
+ prefixCounts.put(secondPrefix, 3);
+
+ Map<String, List<Long>> blockIDs = createContainerWithBlocks(CONTAINER_ID,
+ prefixCounts);
+ // Test deleting filter.
+ testWithFilter(MetadataKeyFilters.getDeletingKeyFilter(),
+ blockIDs.get(OzoneConsts.DELETING_KEY_PREFIX));
+
+ // Test arbitrary filter.
+ MetadataKeyFilters.KeyPrefixFilter secondFilter =
+ new MetadataKeyFilters.KeyPrefixFilter()
+ .addFilter(secondPrefix);
+ testWithFilter(secondFilter, blockIDs.get(secondPrefix));
+ }
+
+ /**
+ * Helper method to run some iterator tests with a provided filter.
+ */
+ private void testWithFilter(MetadataKeyFilters.KeyPrefixFilter filter,
+ List<Long> expectedIDs) throws Exception {
+ try(BlockIterator<BlockData> iterator =
+ db.getStore().getBlockIterator(filter)) {
+ // Test seek.
+ iterator.seekToFirst();
+ long firstID = iterator.nextBlock().getLocalID();
+ assertEquals(expectedIDs.get(0).longValue(), firstID);
+ assertTrue(iterator.hasNext());
+
+ // Test atypical iteration use.
+ iterator.seekToFirst();
+ int numIDsSeen = 0;
+ for (long id: expectedIDs) {
+ assertEquals(iterator.nextBlock().getLocalID(), id);
+ numIDsSeen++;
+
+ // Test that iterator can handle sporadic hasNext() calls.
+ if (id % 2 == 0 && numIDsSeen < expectedIDs.size()) {
+ assertTrue(iterator.hasNext());
+ }
+ }
+
+ assertFalse(iterator.hasNext());
+ }
+ }
+
+ /**
+ * Creates a container with specified number of unprefixed blocks.
* @param containerId
- * @param normalBlocks
- * @param deletedBlocks
+ * @param unprefixedBlocks
+ * @return The list of block IDs of normal blocks that were created.
* @throws Exception
*/
- private void createContainerWithBlocks(long containerId, int
- normalBlocks, int deletedBlocks) throws
- Exception {
- containerData = new KeyValueContainerData(containerId,
- layout,
- (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(),
- UUID.randomUUID().toString());
- container = new KeyValueContainer(containerData, conf);
- container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), UUID
- .randomUUID().toString());
+ private List<Long> createContainerWithBlocks(long containerId,
+ int unprefixedBlocks) throws Exception {
+ return createContainerWithBlocks(containerId, unprefixedBlocks, 0).get("");
+ }
+
+ /**
+ * Creates a container with specified number of unprefixed blocks and deleted
+ * blocks.
+ * @param containerId
+ * @param unprefixedBlocks
+ * @param deletingBlocks
+ * @return Each key prefix mapped to the sorted list of block IDs of blocks
+ * created whose keys had that prefix.
+ * @throws Exception
+ */
+ private Map<String, List<Long>> createContainerWithBlocks(long containerId,
+ int unprefixedBlocks, int deletingBlocks) throws Exception {
+
+ Map<String, Integer> prefixes = new HashMap<>();
+ prefixes.put("", unprefixedBlocks);
+ prefixes.put(OzoneConsts.DELETING_KEY_PREFIX, deletingBlocks);
+ return createContainerWithBlocks(containerId, prefixes);
+ }
+
+ /**
+ * Creates a container with the provided container ID based on
+ * {@link TestKeyValueBlockIterator#conf}. For each prefix specified in
+ * {@code prefixCounts}, the specified number of keys with that prefix will
+ * be created.
+ * <p>
+ * Returns each prefix mapped to a list of the block IDs that were created
+ * for that prefix. Each block ID list will be sorted for each prefix.
+ *
+ * @param containerId The ID of the container to create.
+ * @param prefixCounts A map mapping each key prefix to the number of
+ * blocks that should be created whose keys have that prefix.
+ * @return Each key prefix mapped to the sorted list of block IDs of blocks
+ * created whose keys had that prefix.
+ * @throws Exception
+ */
+ private Map<String, List<Long>> createContainerWithBlocks(long containerId,
+ Map<String, Integer> prefixCounts) throws Exception {
+ // Create required block data.
+ Map<String, List<Long>> blockIDs = new HashMap<>();
try(ReferenceCountedDB metadataStore = BlockUtils.getDB(containerData,
conf)) {
@@ -260,24 +369,29 @@ public class TestKeyValueBlockIterator {
ChunkInfo info = new ChunkInfo("chunkfile", 0, 1024);
chunkList.add(info.getProtoBufMessage());
- for (int i = 0; i < normalBlocks; i++) {
- BlockID blockID = new BlockID(containerId, i);
- BlockData blockData = new BlockData(blockID);
- blockData.setChunks(chunkList);
- metadataStore.getStore().put(Longs.toByteArray(blockID.getLocalID()),
- blockData
- .getProtoBufMessage().toByteArray());
- }
-
- for (int i = normalBlocks; i < deletedBlocks; i++) {
- BlockID blockID = new BlockID(containerId, i);
- BlockData blockData = new BlockData(blockID);
- blockData.setChunks(chunkList);
- metadataStore.getStore().put(StringUtils.string2Bytes(OzoneConsts
- .DELETING_KEY_PREFIX + blockID.getLocalID()), blockData
- .getProtoBufMessage().toByteArray());
+ int blockIndex = 0;
+ // Maps each prefix to the block IDs of the blocks created with that
+ // prefix.
+ Table<String, BlockData> blockDataTable =
+ metadataStore.getStore().getBlockDataTable();
+
+ for (Map.Entry<String, Integer> entry: prefixCounts.entrySet()) {
+ String prefix = entry.getKey();
+ blockIDs.put(prefix, new ArrayList<>());
+ int numBlocks = entry.getValue();
+
+ for (int i = 0; i < numBlocks; i++) {
+ BlockID blockID = new BlockID(containerId, blockIndex);
+ blockIndex++;
+ BlockData blockData = new BlockData(blockID);
+ blockData.setChunks(chunkList);
+ String localID = prefix + blockID.getLocalID();
+ blockDataTable.put(localID, blockData);
+ blockIDs.get(prefix).add(blockID.getLocalID());
+ }
}
}
- }
+ return blockIDs;
+ }
}
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java
index 8c6ff61..79cf6a7 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.ozone.container.keyvalue;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -26,7 +25,7 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.container.common.helpers
.StorageContainerException;
-import org.apache.hadoop.hdds.utils.MetadataStoreBuilder;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
@@ -51,7 +50,6 @@ import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.mockito.Mockito;
-import org.rocksdb.Options;
import java.io.File;
@@ -64,11 +62,9 @@ import java.util.Map;
import java.util.List;
import java.util.UUID;
-import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.ratis.util.Preconditions.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertSame;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -123,31 +119,6 @@ public class TestKeyValueContainer {
keyValueContainer = new KeyValueContainer(keyValueContainerData, conf);
}
- @Test
- public void testBlockIterator() throws Exception{
- keyValueContainerData = new KeyValueContainerData(100L,
- layout,
- (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(),
- datanodeId.toString());
- keyValueContainer = new KeyValueContainer(
- keyValueContainerData, conf);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
- KeyValueBlockIterator blockIterator = keyValueContainer.blockIterator();
- //As no blocks created, hasNext should return false.
- assertFalse(blockIterator.hasNext());
- int blockCount = 10;
- addBlocks(blockCount);
- blockIterator = keyValueContainer.blockIterator();
- assertTrue(blockIterator.hasNext());
- BlockData blockData;
- int blockCounter = 0;
- while(blockIterator.hasNext()) {
- blockData = blockIterator.nextBlock();
- assertEquals(blockCounter++, blockData.getBlockID().getLocalID());
- }
- assertEquals(blockCount, blockCounter);
- }
-
private void addBlocks(int count) throws Exception {
long containerId = keyValueContainerData.getContainerID();
@@ -165,9 +136,8 @@ public class TestKeyValueContainer {
.getLocalID(), 0), 0, 1024);
chunkList.add(info.getProtoBufMessage());
blockData.setChunks(chunkList);
- metadataStore.getStore().put(Longs.toByteArray(blockID.getLocalID()),
- blockData
- .getProtoBufMessage().toByteArray());
+ metadataStore.getStore().getBlockDataTable()
+ .put(Long.toString(blockID.getLocalID()), blockData);
}
}
}
@@ -207,19 +177,21 @@ public class TestKeyValueContainer {
keyValueContainerData.setState(
ContainerProtos.ContainerDataProto.State.CLOSED);
- int numberOfKeysToWrite = 12;
+ long numberOfKeysToWrite = 12;
//write one few keys to check the key count after import
try(ReferenceCountedDB metadataStore =
BlockUtils.getDB(keyValueContainerData, conf)) {
- for (int i = 0; i < numberOfKeysToWrite; i++) {
- metadataStore.getStore().put(("test" + i).getBytes(UTF_8),
- "test".getBytes(UTF_8));
+ Table<String, BlockData> blockDataTable =
+ metadataStore.getStore().getBlockDataTable();
+
+ for (long i = 0; i < numberOfKeysToWrite; i++) {
+ blockDataTable.put("test" + i, new BlockData(new BlockID(i, i)));
}
// As now when we put blocks, we increment block count and update in DB.
// As for test, we are doing manually so adding key count to DB.
- metadataStore.getStore().put(OzoneConsts.DB_BLOCK_COUNT_KEY,
- Longs.toByteArray(numberOfKeysToWrite));
+ metadataStore.getStore().getMetadataTable()
+ .put(OzoneConsts.BLOCK_COUNT, numberOfKeysToWrite);
}
BlockUtils.removeDB(keyValueContainerData, conf);
@@ -411,29 +383,4 @@ public class TestKeyValueContainer {
.getResult());
}
}
-
- @Test
- public void testRocksDBCreateUsesCachedOptions() throws Exception {
- int initialSize = MetadataStoreBuilder.CACHED_OPTS.size();
-
- // Create Container 1
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
- Assert.assertTrue("Rocks DB options should be cached.",
- MetadataStoreBuilder.CACHED_OPTS.containsKey(conf));
-
- Options opts = MetadataStoreBuilder.CACHED_OPTS.get(conf);
-
- // Create Container 2
- keyValueContainerData = new KeyValueContainerData(2L,
- layout,
- (long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(),
- datanodeId.toString());
-
- keyValueContainer = new KeyValueContainer(keyValueContainerData, conf);
- keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId);
-
- assertEquals(initialSize + 1, MetadataStoreBuilder.CACHED_OPTS.size());
- Options cachedOpts = MetadataStoreBuilder.CACHED_OPTS.get(conf);
- assertSame("Cache object should not be updated.", opts, cachedOpts);
- }
}
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerCheck.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerCheck.java
index 4583a54..3362d6d 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerCheck.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerCheck.java
@@ -18,10 +18,8 @@
package org.apache.hadoop.ozone.container.keyvalue;
-import com.google.common.primitives.Longs;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.fs.FileUtil;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -31,6 +29,7 @@ import org.apache.hadoop.ozone.common.Checksum;
import org.apache.hadoop.ozone.common.ChecksumData;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext;
import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerLocationUtil;
@@ -166,8 +165,8 @@ import static org.junit.Assert.assertFalse;
containerData.setDbFile(dbFile);
try (ReferenceCountedDB ignored =
BlockUtils.getDB(containerData, conf);
- KeyValueBlockIterator kvIter = new KeyValueBlockIterator(containerID,
- new File(containerData.getContainerPath()))) {
+ BlockIterator<BlockData> kvIter =
+ ignored.getStore().getBlockIterator()) {
BlockData block = kvIter.nextBlock();
assertFalse(block.getChunks().isEmpty());
ContainerProtos.ChunkInfo c = block.getChunks().get(0);
@@ -250,16 +249,13 @@ import static org.junit.Assert.assertFalse;
}
blockData.setChunks(chunkList);
+ // normal key
+ String key = Long.toString(blockID.getLocalID());
if (i >= normalBlocks) {
// deleted key
- metadataStore.getStore().put(StringUtils.string2Bytes(
- OzoneConsts.DELETING_KEY_PREFIX + blockID.getLocalID()),
- blockData.getProtoBufMessage().toByteArray());
- } else {
- // normal key
- metadataStore.getStore().put(Longs.toByteArray(blockID.getLocalID()),
- blockData.getProtoBufMessage().toByteArray());
+ key = OzoneConsts.DELETING_KEY_PREFIX + blockID.getLocalID();
}
+ metadataStore.getStore().getBlockDataTable().put(key, blockData);
}
chunkManagerTestInfo.validateFileCount(chunksPath, totalBlocks,
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java
index 0014a75..07f1d48 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java
@@ -18,12 +18,11 @@
package org.apache.hadoop.ozone.container.ozoneimpl;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.conf.StorageUnit;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
@@ -51,8 +50,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_BLOCK_COUNT_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.DB_PENDING_DELETE_BLOCK_COUNT_KEY;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.mock;
@@ -126,22 +123,22 @@ public class TestContainerReader {
.getContainerData(), conf)) {
for (int i = 0; i < count; i++) {
- byte[] blkBytes = Longs.toByteArray(blockNames.get(i));
- byte[] blkInfo = metadataStore.getStore().get(blkBytes);
+ Table<String, BlockData> blockDataTable =
+ metadataStore.getStore().getBlockDataTable();
- byte[] deletingKeyBytes =
- StringUtils.string2Bytes(OzoneConsts.DELETING_KEY_PREFIX +
- blockNames.get(i));
+ String blk = Long.toString(blockNames.get(i));
+ BlockData blkInfo = blockDataTable.get(blk);
- metadataStore.getStore().delete(blkBytes);
- metadataStore.getStore().put(deletingKeyBytes, blkInfo);
+ blockDataTable.delete(blk);
+ blockDataTable.put(OzoneConsts.DELETING_KEY_PREFIX + blk, blkInfo);
}
if (setMetaData) {
// Pending delete blocks are still counted towards the block count
// and bytes used metadata values, so those do not change.
- metadataStore.getStore().put(DB_PENDING_DELETE_BLOCK_COUNT_KEY,
- Longs.toByteArray(count));
+ Table<String, Long> metadataTable =
+ metadataStore.getStore().getMetadataTable();
+ metadataTable.put(OzoneConsts.PENDING_DELETE_BLOCK_COUNT, (long)count);
}
}
@@ -163,21 +160,21 @@ public class TestContainerReader {
blockData.addMetadata(OzoneConsts.OWNER,
OzoneConsts.OZONE_SIMPLE_HDFS_USER);
List<ContainerProtos.ChunkInfo> chunkList = new ArrayList<>();
- ChunkInfo info = new ChunkInfo(String.format("%d.data.%d", blockID
- .getLocalID(), 0), 0, blockLen);
+ long localBlockID = blockID.getLocalID();
+ ChunkInfo info = new ChunkInfo(String.format(
+ "%d.data.%d", localBlockID, 0), 0, blockLen);
chunkList.add(info.getProtoBufMessage());
blockData.setChunks(chunkList);
- blkNames.add(blockID.getLocalID());
- metadataStore.getStore().put(Longs.toByteArray(blockID.getLocalID()),
- blockData
- .getProtoBufMessage().toByteArray());
+ blkNames.add(localBlockID);
+ metadataStore.getStore().getBlockDataTable()
+ .put(Long.toString(localBlockID), blockData);
}
if (setMetaData) {
- metadataStore.getStore().put(DB_BLOCK_COUNT_KEY,
- Longs.toByteArray(blockCount));
- metadataStore.getStore().put(OzoneConsts.DB_CONTAINER_BYTES_USED_KEY,
- Longs.toByteArray(blockCount * blockLen));
+ metadataStore.getStore().getMetadataTable()
+ .put(OzoneConsts.BLOCK_COUNT, (long)blockCount);
+ metadataStore.getStore().getMetadataTable()
+ .put(OzoneConsts.CONTAINER_BYTES_USED, blockCount * blockLen);
}
}
diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java
index 2bb52f6..bc4bc25 100644
--- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java
+++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.ozone.container.ozoneimpl;
import com.google.common.base.Preconditions;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.client.BlockID;
@@ -30,6 +29,7 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo;
@@ -264,6 +264,9 @@ public class TestOzoneContainer {
ReferenceCountedDB db = BlockUtils.getDB(container
.getContainerData(), conf);
+ Table<String, Long> metadataTable = db.getStore().getMetadataTable();
+ Table<String, BlockData> blockDataTable = db.getStore().getBlockDataTable();
+
for (int bi = 0; bi < blocks; bi++) {
// Creating BlockData
BlockID blockID = new BlockID(containerId, bi);
@@ -279,15 +282,12 @@ public class TestOzoneContainer {
chunkList.add(info.getProtoBufMessage());
}
blockData.setChunks(chunkList);
- db.getStore().put(Longs.toByteArray(blockID.getLocalID()),
- blockData.getProtoBufMessage().toByteArray());
+ blockDataTable.put(Long.toString(blockID.getLocalID()), blockData);
}
// Set Block count and used bytes.
- db.getStore().put(OzoneConsts.DB_BLOCK_COUNT_KEY,
- Longs.toByteArray(blocks));
- db.getStore().put(OzoneConsts.DB_CONTAINER_BYTES_USED_KEY,
- Longs.toByteArray(usedBytes));
+ metadataTable.put(OzoneConsts.BLOCK_COUNT, (long)blocks);
+ metadataTable.put(OzoneConsts.CONTAINER_BYTES_USED, usedBytes);
// remaining available capacity of the container
return (freeBytes - usedBytes);
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000024.sst b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000024.sst
new file mode 100644
index 0000000..5e34d90
Binary files /dev/null and b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000024.sst differ
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000026.sst b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000026.sst
new file mode 100644
index 0000000..f1fc96d
Binary files /dev/null and b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000026.sst differ
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000032.sst b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000032.sst
new file mode 100644
index 0000000..a2ac89c
Binary files /dev/null and b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000032.sst differ
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000034.log b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/000034.log
new file mode 100644
index 0000000..e69de29
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/CURRENT b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/CURRENT
new file mode 100644
index 0000000..a56825e
--- /dev/null
+++ b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/CURRENT
@@ -0,0 +1 @@
+MANIFEST-000033
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/IDENTITY b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/IDENTITY
new file mode 100644
index 0000000..cdfa488
--- /dev/null
+++ b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/IDENTITY
@@ -0,0 +1 @@
+70e7c9a0-9b64-46d6-8001-4c8da4a7226a
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/MANIFEST-000033 b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/MANIFEST-000033
new file mode 100644
index 0000000..9bd1b44
Binary files /dev/null and b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/MANIFEST-000033 differ
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/OPTIONS-000033 b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/OPTIONS-000033
new file mode 100644
index 0000000..2d422a0
--- /dev/null
+++ b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/OPTIONS-000033
@@ -0,0 +1,165 @@
+# This is a RocksDB option file.
+#
+# For detailed file format spec, please refer to the example file
+# in examples/rocksdb_option_file_example.ini
+#
+
+[Version]
+ rocksdb_version=6.8.1
+ options_file_version=1.1
+
+[DBOptions]
+ manual_wal_flush=false
+ avoid_flush_during_recovery=false
+ dump_malloc_stats=false
+ info_log_level=INFO_LEVEL
+ access_hint_on_compaction_start=NORMAL
+ write_thread_slow_yield_usec=3
+ enable_write_thread_adaptive_yield=true
+ unordered_write=false
+ log_readahead_size=0
+ enable_pipelined_write=false
+ persist_stats_to_disk=false
+ preserve_deletes=false
+ stats_persist_period_sec=600
+ wal_bytes_per_sync=0
+ bytes_per_sync=0
+ delayed_write_rate=16777216
+ WAL_ttl_seconds=0
+ writable_file_max_buffer_size=1048576
+ allow_concurrent_memtable_write=true
+ paranoid_checks=true
+ WAL_size_limit_MB=0
+ write_dbid_to_manifest=false
+ atomic_flush=false
+ max_subcompactions=1
+ wal_dir=/Users/ethanrose/Repos/hadoop-ozone/hadoop-hdds/container-service/target/test-classes/123-dn-container.db
+ avoid_flush_during_shutdown=false
+ keep_log_file_num=1000
+ table_cache_numshardbits=6
+ max_file_opening_threads=16
+ max_background_compactions=-1
+ two_write_queues=false
+ max_background_jobs=2
+ use_fsync=false
+ random_access_max_buffer_size=1048576
+ max_total_wal_size=0
+ db_write_buffer_size=0
+ allow_2pc=false
+ skip_checking_sst_file_sizes_on_db_open=false
+ max_open_files=-1
+ skip_stats_update_on_db_open=false
+ error_if_exists=false
+ manifest_preallocation_size=4194304
+ max_background_flushes=-1
+ is_fd_close_on_exec=true
+ enable_thread_tracking=false
+ stats_dump_period_sec=600
+ max_log_file_size=0
+ advise_random_on_open=true
+ create_missing_column_families=false
+ log_file_time_to_roll=0
+ use_direct_io_for_flush_and_compaction=false
+ compaction_readahead_size=0
+ max_write_batch_group_size_bytes=1048576
+ use_adaptive_mutex=false
+ create_if_missing=true
+ delete_obsolete_files_period_micros=21600000000
+ recycle_log_file_num=0
+ stats_history_buffer_size=1048576
+ strict_bytes_per_sync=false
+ skip_log_error_on_recovery=false
+ allow_mmap_reads=false
+ allow_ingest_behind=false
+ fail_if_options_file_error=false
+ avoid_unnecessary_blocking_io=false
+ allow_fallocate=true
+ max_manifest_file_size=1073741824
+ write_thread_max_yield_usec=100
+ use_direct_reads=false
+ wal_recovery_mode=kPointInTimeRecovery
+ db_log_dir=
+ new_table_reader_for_compaction_inputs=false
+ base_background_compactions=-1
+ allow_mmap_writes=false
+
+
+[CFOptions "default"]
+ sample_for_compression=0
+ compaction_style=kCompactionStyleLevel
+ merge_operator=nullptr
+ compaction_options_universal={allow_trivial_move=false;stop_style=kCompactionStopStyleTotalSize;min_merge_width=2;compression_size_percent=-1;max_size_amplification_percent=200;max_merge_width=4294967295;size_ratio=1;}
+ compaction_filter=nullptr
+ memtable_factory=SkipListFactory
+ prefix_extractor=nullptr
+ bottommost_compression=kDisableCompressionOption
+ compression_per_level=
+ compression=kSnappyCompression
+ comparator=leveldb.BytewiseComparator
+ max_write_buffer_size_to_maintain=0
+ max_sequential_skip_in_iterations=8
+ max_bytes_for_level_base=268435456
+ memtable_insert_with_hint_prefix_extractor=nullptr
+ memtable_whole_key_filtering=false
+ soft_pending_compaction_bytes_limit=68719476736
+ compaction_pri=kMinOverlappingRatio
+ memtable_huge_page_size=0
+ max_successive_merges=0
+ min_write_buffer_number_to_merge=1
+ target_file_size_multiplier=1
+ table_factory=BlockBasedTable
+ num_levels=7
+ arena_block_size=8388608
+ periodic_compaction_seconds=0
+ level0_stop_writes_trigger=36
+ level0_file_num_compaction_trigger=4
+ inplace_update_num_locks=10000
+ compaction_options_fifo={allow_compaction=false;max_table_files_size=1073741824;}
+ level_compaction_dynamic_level_bytes=false
+ level0_slowdown_writes_trigger=20
+ max_bytes_for_level_multiplier_additional=1:1:1:1:1:1:1
+ ttl=2592000
+ max_write_buffer_number=2
+ force_consistency_checks=false
+ memtable_prefix_bloom_size_ratio=0.000000
+ paranoid_file_checks=false
+ hard_pending_compaction_bytes_limit=274877906944
+ max_write_buffer_number_to_maintain=0
+ target_file_size_base=67108864
+ optimize_filters_for_hits=false
+ compaction_filter_factory=nullptr
+ max_bytes_for_level_multiplier=10.000000
+ bloom_locality=0
+ write_buffer_size=67108864
+ disable_auto_compactions=false
+ max_compaction_bytes=1677721600
+ inplace_update_support=false
+ report_bg_io_stats=false
+
+[TableOptions/BlockBasedTable "default"]
+ read_amp_bytes_per_bit=8589934592
+ format_version=2
+ verify_compression=false
+ partition_filters=false
+ enable_index_compression=true
+ checksum=kCRC32c
+ index_block_restart_interval=1
+ pin_top_level_index_and_filter=true
+ block_align=false
+ block_size=4096
+ index_type=kBinarySearch
+ filter_policy=nullptr
+ metadata_block_size=4096
+ no_block_cache=false
+ whole_key_filtering=true
+ index_shortening=kShortenSeparators
+ block_size_deviation=10
+ data_block_index_type=kDataBlockBinarySearch
+ data_block_hash_table_util_ratio=0.750000
+ cache_index_and_filter_blocks=false
+ block_restart_interval=16
+ pin_l0_filter_and_index_blocks_in_cache=false
+ hash_index_allow_collision=true
+ cache_index_and_filter_blocks_with_high_priority=true
+ flush_block_policy_factory=FlushBlockBySizePolicyFactory
+
diff --git a/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/OPTIONS-000036 b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/OPTIONS-000036
new file mode 100644
index 0000000..2d422a0
--- /dev/null
+++ b/hadoop-hdds/container-service/src/test/resources/123-dn-container.db/OPTIONS-000036
@@ -0,0 +1,165 @@
+# This is a RocksDB option file.
+#
+# For detailed file format spec, please refer to the example file
+# in examples/rocksdb_option_file_example.ini
+#
+
+[Version]
+ rocksdb_version=6.8.1
+ options_file_version=1.1
+
+[DBOptions]
+ manual_wal_flush=false
+ avoid_flush_during_recovery=false
+ dump_malloc_stats=false
+ info_log_level=INFO_LEVEL
+ access_hint_on_compaction_start=NORMAL
+ write_thread_slow_yield_usec=3
+ enable_write_thread_adaptive_yield=true
+ unordered_write=false
+ log_readahead_size=0
+ enable_pipelined_write=false
+ persist_stats_to_disk=false
+ preserve_deletes=false
+ stats_persist_period_sec=600
+ wal_bytes_per_sync=0
+ bytes_per_sync=0
+ delayed_write_rate=16777216
+ WAL_ttl_seconds=0
+ writable_file_max_buffer_size=1048576
+ allow_concurrent_memtable_write=true
+ paranoid_checks=true
+ WAL_size_limit_MB=0
+ write_dbid_to_manifest=false
+ atomic_flush=false
+ max_subcompactions=1
+ wal_dir=/Users/ethanrose/Repos/hadoop-ozone/hadoop-hdds/container-service/target/test-classes/123-dn-container.db
+ avoid_flush_during_shutdown=false
+ keep_log_file_num=1000
+ table_cache_numshardbits=6
+ max_file_opening_threads=16
+ max_background_compactions=-1
+ two_write_queues=false
+ max_background_jobs=2
+ use_fsync=false
+ random_access_max_buffer_size=1048576
+ max_total_wal_size=0
+ db_write_buffer_size=0
+ allow_2pc=false
+ skip_checking_sst_file_sizes_on_db_open=false
+ max_open_files=-1
+ skip_stats_update_on_db_open=false
+ error_if_exists=false
+ manifest_preallocation_size=4194304
+ max_background_flushes=-1
+ is_fd_close_on_exec=true
+ enable_thread_tracking=false
+ stats_dump_period_sec=600
+ max_log_file_size=0
+ advise_random_on_open=true
+ create_missing_column_families=false
+ log_file_time_to_roll=0
+ use_direct_io_for_flush_and_compaction=false
+ compaction_readahead_size=0
+ max_write_batch_group_size_bytes=1048576
+ use_adaptive_mutex=false
+ create_if_missing=true
+ delete_obsolete_files_period_micros=21600000000
+ recycle_log_file_num=0
+ stats_history_buffer_size=1048576
+ strict_bytes_per_sync=false
+ skip_log_error_on_recovery=false
+ allow_mmap_reads=false
+ allow_ingest_behind=false
+ fail_if_options_file_error=false
+ avoid_unnecessary_blocking_io=false
+ allow_fallocate=true
+ max_manifest_file_size=1073741824
+ write_thread_max_yield_usec=100
+ use_direct_reads=false
+ wal_recovery_mode=kPointInTimeRecovery
+ db_log_dir=
+ new_table_reader_for_compaction_inputs=false
+ base_background_compactions=-1
+ allow_mmap_writes=false
+
+
+[CFOptions "default"]
+ sample_for_compression=0
+ compaction_style=kCompactionStyleLevel
+ merge_operator=nullptr
+ compaction_options_universal={allow_trivial_move=false;stop_style=kCompactionStopStyleTotalSize;min_merge_width=2;compression_size_percent=-1;max_size_amplification_percent=200;max_merge_width=4294967295;size_ratio=1;}
+ compaction_filter=nullptr
+ memtable_factory=SkipListFactory
+ prefix_extractor=nullptr
+ bottommost_compression=kDisableCompressionOption
+ compression_per_level=
+ compression=kSnappyCompression
+ comparator=leveldb.BytewiseComparator
+ max_write_buffer_size_to_maintain=0
+ max_sequential_skip_in_iterations=8
+ max_bytes_for_level_base=268435456
+ memtable_insert_with_hint_prefix_extractor=nullptr
+ memtable_whole_key_filtering=false
+ soft_pending_compaction_bytes_limit=68719476736
+ compaction_pri=kMinOverlappingRatio
+ memtable_huge_page_size=0
+ max_successive_merges=0
+ min_write_buffer_number_to_merge=1
+ target_file_size_multiplier=1
+ table_factory=BlockBasedTable
+ num_levels=7
+ arena_block_size=8388608
+ periodic_compaction_seconds=0
+ level0_stop_writes_trigger=36
+ level0_file_num_compaction_trigger=4
+ inplace_update_num_locks=10000
+ compaction_options_fifo={allow_compaction=false;max_table_files_size=1073741824;}
+ level_compaction_dynamic_level_bytes=false
+ level0_slowdown_writes_trigger=20
+ max_bytes_for_level_multiplier_additional=1:1:1:1:1:1:1
+ ttl=2592000
+ max_write_buffer_number=2
+ force_consistency_checks=false
+ memtable_prefix_bloom_size_ratio=0.000000
+ paranoid_file_checks=false
+ hard_pending_compaction_bytes_limit=274877906944
+ max_write_buffer_number_to_maintain=0
+ target_file_size_base=67108864
+ optimize_filters_for_hits=false
+ compaction_filter_factory=nullptr
+ max_bytes_for_level_multiplier=10.000000
+ bloom_locality=0
+ write_buffer_size=67108864
+ disable_auto_compactions=false
+ max_compaction_bytes=1677721600
+ inplace_update_support=false
+ report_bg_io_stats=false
+
+[TableOptions/BlockBasedTable "default"]
+ read_amp_bytes_per_bit=8589934592
+ format_version=2
+ verify_compression=false
+ partition_filters=false
+ enable_index_compression=true
+ checksum=kCRC32c
+ index_block_restart_interval=1
+ pin_top_level_index_and_filter=true
+ block_align=false
+ block_size=4096
+ index_type=kBinarySearch
+ filter_policy=nullptr
+ metadata_block_size=4096
+ no_block_cache=false
+ whole_key_filtering=true
+ index_shortening=kShortenSeparators
+ block_size_deviation=10
+ data_block_index_type=kDataBlockBinarySearch
+ data_block_hash_table_util_ratio=0.750000
+ cache_index_and_filter_blocks=false
+ block_restart_interval=16
+ pin_l0_filter_and_index_blocks_in_cache=false
+ hash_index_allow_collision=true
+ cache_index_and_filter_blocks_with_high_priority=true
+ flush_block_policy_factory=FlushBlockBySizePolicyFactory
+
diff --git a/hadoop-hdds/container-service/src/test/resources/123.container b/hadoop-hdds/container-service/src/test/resources/123.container
new file mode 100644
index 0000000..a84d866
--- /dev/null
+++ b/hadoop-hdds/container-service/src/test/resources/123.container
@@ -0,0 +1,10 @@
+!<KeyValueContainerData>
+containerDBType: RocksDB
+containerID: 123
+containerType: KeyValueContainer
+layOutVersion: 2
+maxSize: 1073741824
+metadata: {}
+originNodeId: fc916013-b295-4015-b775-b132a520dc7e
+originPipelineId: bf52d0e3-6418-4713-b0ff-3846243c3541
+state: CLOSED
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/MetadataKeyFilters.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/MetadataKeyFilters.java
index 371cef6..99dfc9f 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/MetadataKeyFilters.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/MetadataKeyFilters.java
@@ -30,41 +30,25 @@ import com.google.common.base.Strings;
* An utility class to filter levelDB keys.
*/
public final class MetadataKeyFilters {
-
- private static KeyPrefixFilter deletingKeyFilter =
- new MetadataKeyFilters.KeyPrefixFilter()
- .addFilter(OzoneConsts.DELETING_KEY_PREFIX);
-
- private static KeyPrefixFilter deletedKeyFilter =
- new MetadataKeyFilters.KeyPrefixFilter()
- .addFilter(OzoneConsts.DELETED_KEY_PREFIX);
-
- private static KeyPrefixFilter normalKeyFilter =
- new MetadataKeyFilters.KeyPrefixFilter()
- .addFilter(OzoneConsts.DELETING_KEY_PREFIX, true)
- .addFilter(OzoneConsts.DELETED_KEY_PREFIX, true)
- .addFilter(OzoneConsts.DELETE_TRANSACTION_KEY_PREFIX, true)
- .addFilter(OzoneConsts.BLOCK_COMMIT_SEQUENCE_ID_PREFIX, true)
- .addFilter(OzoneConsts.BLOCK_COUNT, true)
- .addFilter(OzoneConsts.CONTAINER_BYTES_USED, true)
- .addFilter(OzoneConsts.PENDING_DELETE_BLOCK_COUNT, true);
-
- private MetadataKeyFilters() {
- }
+ private MetadataKeyFilters() { }
public static KeyPrefixFilter getDeletingKeyFilter() {
- return deletingKeyFilter;
- }
-
- public static KeyPrefixFilter getDeletedKeyFilter() {
- return deletedKeyFilter;
+ return new MetadataKeyFilters.KeyPrefixFilter()
+ .addFilter(OzoneConsts.DELETING_KEY_PREFIX);
}
- public static KeyPrefixFilter getNormalKeyFilter() {
- return normalKeyFilter;
+ /**
+ * @return A {@link KeyPrefixFilter} that ignores all keys beginning with
+ * #. This uses the convention that key prefixes are surrounded by
+ * # to ignore keys with any prefix currently used or that will be
+ * added in the future.
+ */
+ public static KeyPrefixFilter getUnprefixedKeyFilter() {
+ return new MetadataKeyFilters.KeyPrefixFilter()
+ .addFilter("#", true);
}
/**
- * Interface for levelDB key filters.
+ * Interface for RocksDB key filters.
*/
public interface MetadataKeyFilter {
/**
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBDefinition.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBDefinition.java
index 6d7d09e..6485beb 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBDefinition.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBDefinition.java
@@ -18,9 +18,12 @@
*/
package org.apache.hadoop.hdds.utils.db;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.server.ServerUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
import java.util.Arrays;
import java.util.Optional;
import java.util.function.Function;
@@ -43,7 +46,16 @@ public interface DBDefinition {
String getLocationConfigKey();
/**
- * Create a new DB store instance based on the configuration.
+ * @param conf The configuration for the DB.
+ * @return The parent directory for this definition's .db file.
+ */
+ default File getDBLocation(ConfigurationSource conf) {
+ return ServerUtils.getDirectoryFromConfig(conf,
+ getLocationConfigKey(), getName());
+ }
+
+ /**
+ * @return The column families present in the DB.
*/
DBColumnFamilyDefinition[] getColumnFamilies();
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStore.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStore.java
index 8567d03..71766bd 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStore.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStore.java
@@ -83,7 +83,13 @@ public interface DBStore extends AutoCloseable, BatchOperationHandler {
* Flush the DB buffer onto persistent storage.
* @throws IOException
*/
- void flush() throws IOException;
+ void flushDB() throws IOException;
+
+ /**
+ * Flush the outstanding I/O operations of the DB.
+ * @param sync if true will sync the outstanding I/Os to the disk.
+ */
+ void flushLog(boolean sync) throws IOException;
/**
* Compact the entire database.
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStoreBuilder.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStoreBuilder.java
index 2e18530..6a20299 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStoreBuilder.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStoreBuilder.java
@@ -36,7 +36,6 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import com.google.common.base.Preconditions;
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_DB_PROFILE;
-import static org.apache.hadoop.hdds.server.ServerUtils.getDirectoryFromConfig;
import static org.apache.hadoop.hdds.server.ServerUtils.getOzoneMetaDirPath;
import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_METADATA_STORE_ROCKSDB_STATISTICS;
import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_METADATA_STORE_ROCKSDB_STATISTICS_DEFAULT;
@@ -106,6 +105,14 @@ public final class DBStoreBuilder {
return new DBStoreBuilder(configuration, rocksDBConfiguration);
}
+ public static DBStoreBuilder newBuilder(ConfigurationSource configuration,
+ DBDefinition definition) {
+ DBStoreBuilder builder = createDBStoreBuilder(configuration, definition);
+ builder.registerTables(definition);
+
+ return builder;
+ }
+
public DBStoreBuilder setProfile(DBProfile profile) {
dbProfile = profile;
return this;
@@ -269,8 +276,7 @@ public final class DBStoreBuilder {
private static DBStoreBuilder createDBStoreBuilder(
ConfigurationSource configuration, DBDefinition definition) {
- File metadataDir = getDirectoryFromConfig(configuration,
- definition.getLocationConfigKey(), definition.getName());
+ File metadataDir = definition.getDBLocation(configuration);
if (metadataDir == null) {
@@ -293,18 +299,27 @@ public final class DBStoreBuilder {
DBDefinition definition)
throws IOException {
DBStoreBuilder builder = createDBStoreBuilder(configuration, definition);
- for (DBColumnFamilyDefinition columnTableDefinition : definition
- .getColumnFamilies()) {
- builder.registerTable(columnTableDefinition);
- }
+ builder.registerTables(definition);
+
return builder.build();
}
- private <KEY, VALUE> void registerTable(
- DBColumnFamilyDefinition<KEY, VALUE> definition) {
- addTable(definition.getName())
- .addCodec(definition.getKeyType(), definition.getKeyCodec())
- .addCodec(definition.getValueType(), definition.getValueCodec());
- }
+ public <KEY, VALUE> DBStoreBuilder registerTables(DBDefinition definition) {
+ for (DBColumnFamilyDefinition<KEY, VALUE> columnFamily :
+ definition.getColumnFamilies()) {
+
+ if (!columnFamily.getTableName().equals(DEFAULT_COLUMN_FAMILY_NAME)) {
+ // The default column family is always added.
+ // If it is present in the DB Definition, ignore it so we don't get an
+ // error about adding it twice.
+ addTable(columnFamily.getName());
+ }
+ // Add new codecs specified for the table, which may be an alias for a
+ // table that was already added with different codecs.
+ addCodec(columnFamily.getKeyType(), columnFamily.getKeyCodec());
+ addCodec(columnFamily.getValueType(), columnFamily.getValueCodec());
+ }
+ return this;
+ }
}
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBStore.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBStore.java
index 4a23541..0890a81 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBStore.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBStore.java
@@ -319,7 +319,7 @@ public class RDBStore implements DBStore {
}
@Override
- public void flush() throws IOException {
+ public void flushDB() throws IOException {
try (FlushOptions flushOptions = new FlushOptions()) {
flushOptions.setWaitForFlush(true);
db.flush(flushOptions);
@@ -329,9 +329,22 @@ public class RDBStore implements DBStore {
}
@Override
+ public void flushLog(boolean sync) throws IOException {
+ if (db != null) {
+ try {
+ // for RocksDB it is sufficient to flush the WAL as entire db can
+ // be reconstructed using it.
+ db.flushWal(sync);
+ } catch (RocksDBException e) {
+ throw toIOException("Failed to flush db", e);
+ }
+ }
+ }
+
+ @Override
public DBCheckpoint getCheckpoint(boolean flush) throws IOException {
if (flush) {
- this.flush();
+ this.flushDB();
}
return checkPointManager.createCheckpoint(checkpointsParentDir);
}
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBTable.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBTable.java
index 8f2a1f8..878a1f2 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBTable.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/RDBTable.java
@@ -21,6 +21,9 @@ package org.apache.hadoop.hdds.utils.db;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Arrays;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
import org.apache.hadoop.hdds.StringUtils;
@@ -31,6 +34,8 @@ import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.WriteOptions;
+import org.rocksdb.RocksIterator;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -245,4 +250,91 @@ class RDBTable implements Table<byte[], byte[]> {
"Failed to get estimated key count of table " + getName(), e);
}
}
+
+ @Override
+ public List<ByteArrayKeyValue> getRangeKVs(byte[] startKey,
+ int count, MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+ return getRangeKVs(startKey, count, false, filters);
+ }
+
+ @Override
+ public List<ByteArrayKeyValue> getSequentialRangeKVs(byte[] startKey,
+ int count, MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+ return getRangeKVs(startKey, count, true, filters);
+ }
+
+ private List<ByteArrayKeyValue> getRangeKVs(byte[] startKey,
+ int count, boolean sequential,
+ MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+ List<ByteArrayKeyValue> result = new ArrayList<>();
+ long start = System.currentTimeMillis();
+ if (count < 0) {
+ throw new IllegalArgumentException(
+ "Invalid count given " + count + ", count must be greater than 0");
+ }
+ try (RocksIterator it = db.newIterator(handle)) {
+ if (startKey == null) {
+ it.seekToFirst();
+ } else {
+ if (get(startKey) == null) {
+ // Key not found, return empty list
+ return result;
+ }
+ it.seek(startKey);
+ }
+ while (it.isValid() && result.size() < count) {
+ byte[] currentKey = it.key();
+ byte[] currentValue = it.value();
+
+ it.prev();
+ final byte[] prevKey = it.isValid() ? it.key() : null;
+
+ it.seek(currentKey);
+ it.next();
+ final byte[] nextKey = it.isValid() ? it.key() : null;
+
+ if (filters == null) {
+ result.add(ByteArrayKeyValue
+ .create(currentKey, currentValue));
+ } else {
+ if (Arrays.stream(filters)
+ .allMatch(entry -> entry.filterKey(prevKey,
+ currentKey, nextKey))) {
+ result.add(ByteArrayKeyValue
+ .create(currentKey, currentValue));
+ } else {
+ if (result.size() > 0 && sequential) {
+ // if the caller asks for a sequential range of results,
+ // and we met a dis-match, abort iteration from here.
+ // if result is empty, we continue to look for the first match.
+ break;
+ }
+ }
+ }
+ }
+ } finally {
+ long end = System.currentTimeMillis();
+ long timeConsumed = end - start;
+ if (LOG.isDebugEnabled()) {
+ if (filters != null) {
+ for (MetadataKeyFilters.MetadataKeyFilter filter : filters) {
+ int scanned = filter.getKeysScannedNum();
+ int hinted = filter.getKeysHintedNum();
+ if (scanned > 0 || hinted > 0) {
+ LOG.debug(
+ "getRangeKVs ({}) numOfKeysScanned={}, numOfKeysHinted={}",
+ filter.getClass().getSimpleName(), filter.getKeysScannedNum(),
+ filter.getKeysHintedNum());
+ }
+ }
+ }
+ LOG.debug("Time consumed for getRangeKVs() is {}ms,"
+ + " result length is {}.", timeConsumed, result.size());
+ }
+ }
+ return result;
+ }
}
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/Table.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/Table.java
index 0aa852f..554af5a 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/Table.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/Table.java
@@ -26,6 +26,7 @@ import java.util.Map;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.hadoop.hdds.annotation.InterfaceStability;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
/**
@@ -204,6 +205,61 @@ public interface Table<KEY, VALUE> extends AutoCloseable {
}
/**
+ * Returns a certain range of key value pairs as a list based on a
+ * startKey or count. Further a {@link MetadataKeyFilters.MetadataKeyFilter}
+ * can be added to * filter keys if necessary.
+ * To prevent race conditions while listing
+ * entries, this implementation takes a snapshot and lists the entries from
+ * the snapshot. This may, on the other hand, cause the range result slight
+ * different with actual data if data is updating concurrently.
+ * <p>
+ * If the startKey is specified and found in the table, this key and the keys
+ * after this key will be included in the result. If the startKey is null
+ * all entries will be included as long as other conditions are satisfied.
+ * If the given startKey doesn't exist and empty list will be returned.
+ * <p>
+ * The count argument is to limit number of total entries to return,
+ * the value for count must be an integer greater than 0.
+ * <p>
+ * This method allows to specify one or more
+ * {@link MetadataKeyFilters.MetadataKeyFilter}
+ * to filter keys by certain condition. Once given, only the entries
+ * whose key passes all the filters will be included in the result.
+ *
+ * @param startKey a start key.
+ * @param count max number of entries to return.
+ * @param filters customized one or more
+ * {@link MetadataKeyFilters.MetadataKeyFilter}.
+ * @return a list of entries found in the database or an empty list if the
+ * startKey is invalid.
+ * @throws IOException if there are I/O errors.
+ * @throws IllegalArgumentException if count is less than 0.
+ */
+ List<? extends KeyValue<KEY, VALUE>> getRangeKVs(KEY startKey,
+ int count, MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException;
+
+ /**
+ * This method is very similar to {@link #getRangeKVs}, the only
+ * different is this method is supposed to return a sequential range
+ * of elements based on the filters. While iterating the elements,
+ * if it met any entry that cannot pass the filter, the iterator will stop
+ * from this point without looking for next match. If no filter is given,
+ * this method behaves just like {@link #getRangeKVs}.
+ *
+ * @param startKey a start key.
+ * @param count max number of entries to return.
+ * @param filters customized one or more
+ * {@link MetadataKeyFilters.MetadataKeyFilter}.
+ * @return a list of entries found in the database.
+ * @throws IOException
+ * @throws IllegalArgumentException
+ */
+ List<? extends KeyValue<KEY, VALUE>> getSequentialRangeKVs(KEY startKey,
+ int count, MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException;
+
+ /**
* Class used to represent the key and value pair of a db entry.
*/
interface KeyValue<KEY, VALUE> {
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java
index 0d22c3f..1c88290 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java
@@ -19,12 +19,14 @@
package org.apache.hadoop.hdds.utils.db;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
+import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
import org.apache.hadoop.hdds.utils.db.cache.CacheResult;
import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
@@ -309,6 +311,50 @@ public class TypedTable<KEY, VALUE> implements Table<KEY, VALUE> {
}
@Override
+ public List<TypedKeyValue> getRangeKVs(
+ KEY startKey, int count,
+ MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+
+ // A null start key means to start from the beginning of the table.
+ // Cannot convert a null key to bytes.
+ byte[] startKeyBytes = null;
+ if (startKey != null) {
+ startKeyBytes = codecRegistry.asRawData(startKey);
+ }
+
+ List<? extends KeyValue<byte[], byte[]>> rangeKVBytes =
+ rawTable.getRangeKVs(startKeyBytes, count, filters);
+
+ List<TypedKeyValue> rangeKVs = new ArrayList<>();
+ rangeKVBytes.forEach(byteKV -> rangeKVs.add(new TypedKeyValue(byteKV)));
+
+ return rangeKVs;
+ }
+
+ @Override
+ public List<TypedKeyValue> getSequentialRangeKVs(
+ KEY startKey, int count,
+ MetadataKeyFilters.MetadataKeyFilter... filters)
+ throws IOException, IllegalArgumentException {
+
+ // A null start key means to start from the beginning of the table.
+ // Cannot convert a null key to bytes.
+ byte[] startKeyBytes = null;
+ if (startKey != null) {
+ startKeyBytes = codecRegistry.asRawData(startKey);
+ }
+
+ List<? extends KeyValue<byte[], byte[]>> rangeKVBytes =
+ rawTable.getSequentialRangeKVs(startKeyBytes, count, filters);
+
+ List<TypedKeyValue> rangeKVs = new ArrayList<>();
+ rangeKVBytes.forEach(byteKV -> rangeKVs.add(new TypedKeyValue(byteKV)));
+
+ return rangeKVs;
+ }
+
+ @Override
public void cleanupCache(List<Long> epochs) {
cache.cleanup(epochs);
}
diff --git a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
index a19d1d4..2565b5e 100644
--- a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
+++ b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto
@@ -369,6 +369,10 @@ message ChunkInfo {
required ChecksumData checksumData =5;
}
+message ChunkInfoList {
+ repeated ChunkInfo chunks = 1;
+}
+
message ChecksumData {
required ChecksumType type = 1;
required uint32 bytesPerChecksum = 2;
diff --git a/hadoop-hdds/pom.xml b/hadoop-hdds/pom.xml
index 26ca19a..e17c552 100644
--- a/hadoop-hdds/pom.xml
+++ b/hadoop-hdds/pom.xml
@@ -277,6 +277,8 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
<exclude>src/test/resources/incorrect.checksum.container</exclude>
<exclude>src/test/resources/incorrect.container</exclude>
<exclude>src/test/resources/test.db.ini</exclude>
+ <exclude>src/test/resources/123-dn-container.db/**</exclude>
+ <exclude>src/test/resources/123.container</exclude>
<exclude>src/main/resources/proto.lock</exclude>
</excludes>
</configuration>
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java
index c732c88..c67fe30 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java
@@ -21,13 +21,14 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
import org.apache.hadoop.hdds.utils.MetadataKeyFilters.KeyPrefixFilter;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
@@ -38,7 +39,6 @@ import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.primitives.Longs;
import org.apache.commons.lang3.RandomStringUtils;
import org.junit.Rule;
import org.junit.rules.Timeout;
@@ -99,13 +99,15 @@ public class TestStorageContainerManagerHelper {
ReferenceCountedDB meta = getContainerMetadata(containerID);
KeyPrefixFilter filter =
new KeyPrefixFilter().addFilter(OzoneConsts.DELETING_KEY_PREFIX);
- List<Map.Entry<byte[], byte[]>> kvs = meta.getStore()
+
+ List<? extends Table.KeyValue<String, BlockData>> kvs =
+ meta.getStore().getBlockDataTable()
.getRangeKVs(null, Integer.MAX_VALUE, filter);
- kvs.forEach(entry -> {
- String key = StringUtils.bytes2String(entry.getKey());
+
+ for (Table.KeyValue<String, BlockData> entry : kvs) {
pendingDeletionBlocks
- .add(key.replace(OzoneConsts.DELETING_KEY_PREFIX, ""));
- });
+ .add(entry.getKey().replace(OzoneConsts.DELETING_KEY_PREFIX, ""));
+ }
meta.close();
return pendingDeletionBlocks;
}
@@ -122,12 +124,15 @@ public class TestStorageContainerManagerHelper {
public List<Long> getAllBlocks(Long containeID) throws IOException {
List<Long> allBlocks = Lists.newArrayList();
ReferenceCountedDB meta = getContainerMetadata(containeID);
- List<Map.Entry<byte[], byte[]>> kvs =
- meta.getStore().getRangeKVs(null, Integer.MAX_VALUE,
- MetadataKeyFilters.getNormalKeyFilter());
- kvs.forEach(entry -> {
- allBlocks.add(Longs.fromByteArray(entry.getKey()));
- });
+
+ List<? extends Table.KeyValue<String, BlockData>> kvs =
+ meta.getStore().getBlockDataTable()
+ .getRangeKVs(null, Integer.MAX_VALUE,
+ MetadataKeyFilters.getUnprefixedKeyFilter());
+
+ for (Table.KeyValue<String, BlockData> entry : kvs) {
+ allBlocks.add(Long.valueOf(entry.getKey()));
+ }
meta.close();
return allBlocks;
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
index 97b6cf4..c9551fd 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
@@ -72,9 +72,11 @@ import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
import org.apache.hadoop.ozone.common.OzoneChecksumException;
import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
import org.apache.hadoop.ozone.container.common.interfaces.Container;
-import org.apache.hadoop.ozone.container.keyvalue.KeyValueBlockIterator;
+import org.apache.hadoop.ozone.container.common.utils.ReferenceCountedDB;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
+import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils;
import org.apache.hadoop.ozone.container.keyvalue.helpers.KeyValueContainerLocationUtil;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.OmFailoverProxyUtil;
@@ -1504,10 +1506,10 @@ public abstract class TestOzoneRpcClientAbstract {
(KeyValueContainerData)(datanodeService.getDatanodeStateMachine()
.getContainer().getContainerSet().getContainer(containerID)
.getContainerData());
- String containerPath = new File(containerData.getMetadataPath())
- .getParent();
- try(KeyValueBlockIterator keyValueBlockIterator = new KeyValueBlockIterator(
- containerID, new File(containerPath))) {
+ try (ReferenceCountedDB db = BlockUtils.getDB(containerData,
+ cluster.getConf());
+ BlockIterator<BlockData> keyValueBlockIterator =
+ db.getStore().getBlockIterator()) {
while (keyValueBlockIterator.hasNext()) {
BlockData blockData = keyValueBlockIterator.nextBlock();
if (blockData.getBlockID().getLocalID() == localID) {
@@ -1667,11 +1669,10 @@ public abstract class TestOzoneRpcClientAbstract {
// the container.
KeyValueContainerData containerData =
(KeyValueContainerData) container.getContainerData();
- String containerPath =
- new File(containerData.getMetadataPath()).getParent();
- try (KeyValueBlockIterator keyValueBlockIterator =
- new KeyValueBlockIterator(containerID, new File(containerPath))) {
-
+ try (ReferenceCountedDB db = BlockUtils.getDB(containerData,
+ cluster.getConf());
+ BlockIterator<BlockData> keyValueBlockIterator =
+ db.getStore().getBlockIterator()) {
// Find the block corresponding to the key we put. We use the localID of
// the BlockData to identify out key.
BlockData blockData = null;
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestValidateBCSIDOnRestart.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestValidateBCSIDOnRestart.java
index 1dd2bd5..84416d2 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestValidateBCSIDOnRestart.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestValidateBCSIDOnRestart.java
@@ -17,9 +17,7 @@
package org.apache.hadoop.ozone.client.rpc;
-import com.google.common.primitives.Longs;
import org.apache.hadoop.fs.FileUtil;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.client.ReplicationFactor;
import org.apache.hadoop.hdds.client.ReplicationType;
import org.apache.hadoop.hdds.conf.DatanodeRatisServerConfig;
@@ -231,13 +229,11 @@ public class TestValidateBCSIDOnRestart {
keyValueContainerData = (KeyValueContainerData) containerData;
ReferenceCountedDB db = BlockUtils.
getDB(keyValueContainerData, conf);
- byte[] blockCommitSequenceIdKey =
- StringUtils.
- string2Bytes(OzoneConsts.BLOCK_COMMIT_SEQUENCE_ID_PREFIX);
// modify the bcsid for the container in the ROCKS DB thereby inducing
// corruption
- db.getStore().put(blockCommitSequenceIdKey, Longs.toByteArray(0));
+ db.getStore().getMetadataTable()
+ .put(OzoneConsts.BLOCK_COMMIT_SEQUENCE_ID, 0L);
db.decrementReference();
// after the restart, there will be a mismatch in BCSID of what is recorded
// in the and what is there in RockSDB and hence the container would be
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestBlockDeletion.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestBlockDeletion.java
index 58c3b9d..1a18681 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestBlockDeletion.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestBlockDeletion.java
@@ -17,9 +17,7 @@
*/
package org.apache.hadoop.ozone.container.common.statemachine.commandhandler;
-import com.google.common.primitives.Longs;
import org.apache.commons.lang3.RandomStringUtils;
-import org.apache.hadoop.hdds.StringUtils;
import org.apache.hadoop.hdds.client.ReplicationFactor;
import org.apache.hadoop.hdds.client.ReplicationType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -35,6 +33,7 @@ import org.apache.hadoop.hdds.scm.block.SCMBlockDeletingService;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.ReplicationManager;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.OzoneTestUtils;
@@ -43,6 +42,8 @@ import org.apache.hadoop.ozone.client.OzoneBucket;
import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.OzoneVolume;
import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.helpers.ChunkInfoList;
import org.apache.hadoop.ozone.container.common.impl.ContainerData;
import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
@@ -70,6 +71,7 @@ import java.util.List;
import java.util.HashSet;
import java.util.ArrayList;
import java.util.UUID;
+import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -422,8 +424,9 @@ public class TestBlockDeletion {
try(ReferenceCountedDB db =
BlockUtils.getDB((KeyValueContainerData) dnContainerSet
.getContainer(blockID.getContainerID()).getContainerData(), conf)) {
- Assert.assertNotNull(db.getStore().get(
- Longs.toByteArray(blockID.getLocalID())));
+ Assert.assertNotNull(
+ db.getStore().getBlockDataTable()
+ .get(Long.toString(blockID.getLocalID())));
}
}, omKeyLocationInfoGroups);
}
@@ -437,14 +440,42 @@ public class TestBlockDeletion {
try(ReferenceCountedDB db =
BlockUtils.getDB((KeyValueContainerData) dnContainerSet
.getContainer(blockID.getContainerID()).getContainerData(), conf)) {
- Assert.assertNull(db.getStore().get(
- Longs.toByteArray(blockID.getLocalID())));
- Assert.assertNull(db.getStore().get(StringUtils.string2Bytes(
- OzoneConsts.DELETING_KEY_PREFIX + blockID.getLocalID())));
- Assert.assertNotNull(StringUtils.string2Bytes(
- OzoneConsts.DELETED_KEY_PREFIX + blockID.getLocalID()));
+
+ Table<String, BlockData> blockDataTable =
+ db.getStore().getBlockDataTable();
+ Table<String, ChunkInfoList> deletedBlocksTable =
+ db.getStore().getDeletedBlocksTable();
+
+ String blockIDString = Long.toString(blockID.getLocalID());
+
+ BlockData blockData = blockDataTable.get(blockIDString);
+ Assert.assertNotNull(blockData);
+
+ String deletingKey = OzoneConsts.DELETING_KEY_PREFIX + blockIDString;
+ Assert.assertNull(blockDataTable.get(deletingKey));
+
+ ChunkInfoList deletedBlocksChunks =
+ deletedBlocksTable.get(blockIDString);
+ Assert.assertNotNull(deletedBlocksChunks);
+
+ verifyChunksEqual(blockData.getChunks(), deletedBlocksChunks.asList());
}
containerIdsWithDeletedBlocks.add(blockID.getContainerID());
}, omKeyLocationInfoGroups);
}
+
+ private void verifyChunksEqual(List<ContainerProtos.ChunkInfo> chunks1,
+ List<ContainerProtos.ChunkInfo> chunks2) {
+ Assert.assertEquals(chunks1.size(), chunks2.size());
+
+ Iterator<ContainerProtos.ChunkInfo> iter1 = chunks1.iterator();
+ Iterator<ContainerProtos.ChunkInfo> iter2 = chunks2.iterator();
+
+ while(iter1.hasNext() && iter2.hasNext()) {
+ ContainerProtos.ChunkInfo c1 = iter1.next();
+ ContainerProtos.ChunkInfo c2 = iter2.next();
+
+ Assert.assertEquals(c1, c2);
+ }
+ }
}
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
index 93e29ee..acd6375 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
@@ -347,7 +347,7 @@ public class OzoneManagerStateMachine extends BaseStateMachine {
long lastAppliedIndex = lastTermIndex.getIndex();
snapshotInfo.updateTermIndex(lastTermIndex.getTerm(),
lastAppliedIndex);
- ozoneManager.getMetadataManager().getStore().flush();
+ ozoneManager.getMetadataManager().getStore().flushDB();
return lastAppliedIndex;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org