jojochuang commented on code in PR #5662:
URL: https://github.com/apache/ozone/pull/5662#discussion_r1454869654
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaThreeImpl.java:
##########
@@ -224,4 +235,185 @@ public void compactionIfNeeded() throws Exception {
}
}
}
+
+ @Override
+ public BlockData getBlockByID(BlockID blockID,
+ KeyValueContainerData containerData) throws IOException {
+ String blockKey = containerData.getBlockKey(blockID.getLocalID());
+
+ // check last chunk table
+ BlockData lastChunk = getLastChunkInfoTable().get(blockKey);
+ // check block data table
+ BlockData blockData = getBlockDataTable().get(blockKey);
+
+ if (blockData == null) {
+ if (lastChunk == null) {
+ throw new StorageContainerException(
+ NO_SUCH_BLOCK_ERR_MSG + " BlockID : " + blockID, NO_SUCH_BLOCK);
+ } else {
+ LOG.debug("blockData=(null), lastChunk={}", lastChunk.getChunks());
+ return lastChunk;
+ }
+ } else {
+ if (lastChunk != null) {
+ reconcilePartialChunks(lastChunk, blockData);
+ } else {
+ LOG.debug("blockData={}, lastChunk=(null)", blockData.getChunks());
+ }
+ }
+
+ return blockData;
+ }
+
+ private static void reconcilePartialChunks(
+ BlockData lastChunk, BlockData blockData) {
+ LOG.debug("blockData={}, lastChunk={}",
+ blockData.getChunks(), lastChunk.getChunks());
+ Preconditions.checkState(lastChunk.getChunks().size() == 1);
+ ContainerProtos.ChunkInfo lastChunkInBlockData =
+ blockData.getChunks().get(blockData.getChunks().size() - 1);
+ Preconditions.checkState(
+ lastChunkInBlockData.getOffset() + lastChunkInBlockData.getLen()
+ == lastChunk.getChunks().get(0).getOffset(),
+ "chunk offset does not match");
+
+ // append last partial chunk to the block data
+ List<ContainerProtos.ChunkInfo> chunkInfos =
+ new ArrayList<>(blockData.getChunks());
+ chunkInfos.add(lastChunk.getChunks().get(0));
+ blockData.setChunks(chunkInfos);
+
+ blockData.setBlockCommitSequenceId(
+ lastChunk.getBlockCommitSequenceId());
+ }
+
+ private static boolean isPartialChunkList(BlockData data) {
+ return data.getMetadata().containsKey(INCREMENTAL_CHUNK_LIST);
+ }
+
+ private static boolean isFullChunk(ContainerProtos.ChunkInfo chunkInfo) {
+ for (ContainerProtos.KeyValue kv: chunkInfo.getMetadataList()) {
+ if (kv.getKey().equals(FULL_CHUNK)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ // if eob or if the last chunk is full,
+ private static boolean shouldAppendLastChunk(boolean endOfBlock,
+ BlockData data) {
+ if (endOfBlock || data.getChunks().isEmpty()) {
+ return true;
+ }
+ return isFullChunk(data.getChunks().get(data.getChunks().size() - 1));
+ }
+
+ public void putBlockByID(BatchOperation batch, boolean incremental,
+ long localID, BlockData data, KeyValueContainerData containerData,
+ boolean endOfBlock) throws IOException {
+ if (!incremental && !isPartialChunkList(data)) {
+ // Case (1) old client: override chunk list.
+ getBlockDataTable().putWithBatch(
+ batch, containerData.getBlockKey(localID), data);
+ } else if (shouldAppendLastChunk(endOfBlock, data)) {
+ moveLastChunkToBlockData(batch, localID, data, containerData);
+ } else {
+ // incremental chunk list,
+ // not end of block, has partial chunks
+ putBlockWithPartialChunks(batch, localID, data, containerData);
+ }
+ }
+
+ private void moveLastChunkToBlockData(BatchOperation batch, long localID,
+ BlockData data, KeyValueContainerData containerData) throws IOException {
+ // if eob or if the last chunk is full,
+ // the 'data' is full so append it to the block table's chunk info
+ // and then remove from lastChunkInfo
+ BlockData blockData = getBlockDataTable().get(
+ containerData.getBlockKey(localID));
+ if (blockData == null) {
+ // Case 2.1 if the block did not have full chunks before,
+ // the block's chunk is what received from client this time.
+ blockData = data;
+ } else {
+ // case 2.2 the block already has some full chunks
+ List<ContainerProtos.ChunkInfo> chunkInfoList = blockData.getChunks();
+ blockData.setChunks(new ArrayList<>(chunkInfoList));
+ for (ContainerProtos.ChunkInfo chunk : data.getChunks()) {
+ blockData.addChunk(chunk);
+ }
+ blockData.setBlockCommitSequenceId(data.getBlockCommitSequenceId());
+
+ /*int next = 0;
+ for (ContainerProtos.ChunkInfo info : chunkInfoList) {
+ assert info.getOffset() == next;
+ next += info.getLen();
+ }*/
+ }
+ // delete the entry from last chunk info table
+ getLastChunkInfoTable().deleteWithBatch(
+ batch, containerData.getBlockKey(localID));
+ // update block data table
+ getBlockDataTable().putWithBatch(batch,
+ containerData.getBlockKey(localID), blockData);
+ }
+
+ private void putBlockWithPartialChunks(BatchOperation batch, long localID,
+ BlockData data, KeyValueContainerData containerData) throws IOException {
+ if (data.getChunks().size() == 1) {
+ // Case (3.1) replace/update the last chunk info table
+ getLastChunkInfoTable().putWithBatch(
+ batch, containerData.getBlockKey(localID), data);
+ } else {
+ int lastChunkIndex = data.getChunks().size() - 1;
+ // received more than one chunk this time
+ List<ContainerProtos.ChunkInfo> lastChunkInfo =
+ Collections.singletonList(
+ data.getChunks().get(lastChunkIndex));
+ BlockData blockData = getBlockDataTable().get(
+ containerData.getBlockKey(localID));
+ if (blockData == null) {
+ // Case 3.2: if the block does not exist in the block data table
+ List<ContainerProtos.ChunkInfo> chunkInfos =
+ new ArrayList<>(data.getChunks());
+ chunkInfos.remove(lastChunkIndex);
+ data.setChunks(chunkInfos);
+ blockData = data;
+ LOG.debug("block {} does not have full chunks yet. Adding the " +
+ "chunks to it {}", localID, blockData);
+ } else {
+ // Case 3.3: if the block exists in the block data table,
+ // append chunks till except the last one (supposedly partial)
+ List<ContainerProtos.ChunkInfo> chunkInfos =
+ new ArrayList<>(blockData.getChunks());
+
+ LOG.debug("blockData.getChunks()={}", chunkInfos);
+ LOG.debug("data.getChunks()={}", data.getChunks());
+
+ for (int i = 0; i < lastChunkIndex; i++) {
+ chunkInfos.add(data.getChunks().get(i));
+ }
+ blockData.setChunks(chunkInfos);
+ blockData.setBlockCommitSequenceId(data.getBlockCommitSequenceId());
+
+ /*int next = 0;
+ for (ContainerProtos.ChunkInfo info : chunkInfos) {
+ if (info.getOffset() != next) {
+
+ LOG.error("blockData.getChunks()={}", chunkInfos);
+ LOG.error("data.getChunks()={}", data.getChunks());
+ }
+ assert info.getOffset() == next;
+ next += info.getLen();
+ }*/
+ }
+ getBlockDataTable().putWithBatch(batch,
+ containerData.getBlockKey(localID), blockData);
+ // update the last partial chunk
+ data.setChunks(lastChunkInfo);
+ getLastChunkInfoTable().putWithBatch(
+ batch, containerData.getBlockKey(localID), data);
Review Comment:
i think there could still be cases where the client's buffer has more than
one chunk's full of data that gets flushed out at a time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
