Copilot commented on code in PR #11068:
URL: https://github.com/apache/ozone/pull/11068#discussion_r3899865788
##########
hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/TypedTable.java:
##########
@@ -306,6 +317,143 @@ private VALUE getFromTable(KEY key) throws
RocksDatabaseException, CodecExceptio
}
}
+ List<VALUE> multiGetFromTable(List<KEY> keys)
+ throws RocksDatabaseException, CodecException {
+ if (supportCodecBuffer) {
+ return multiGetFromTableWithCodecBuffer(keys);
+ }
+ return multiGetFromTableWithByteArray(keys);
+ }
+
+ List<VALUE> multiGetFromTableWithByteArray(List<KEY> keys)
+ throws RocksDatabaseException, CodecException {
+ List<byte[]> keyBytesList = new ArrayList<>(keys.size());
+ for (KEY key : keys) {
+ keyBytesList.add(encodeKey(key));
+ }
+ List<byte[]> valueBytesList = rawTable.multiGetSkipCache(keyBytesList);
+ List<VALUE> values = new ArrayList<>(keys.size());
+ for (byte[] valueBytes : valueBytesList) {
+ values.add(decodeValue(valueBytes));
+ }
+ return values;
+ }
+
+ List<VALUE> multiGetFromTableWithCodecBuffer(List<KEY> keys)
+ throws RocksDatabaseException, CodecException {
+ final int n = keys.size();
+ final List<CodecBuffer> keyBuffers = new ArrayList<>(n);
+ final List<ByteBuffer> keyByteBuffers = new ArrayList<>(n);
+ final List<CodecBuffer> valueBuffers = new ArrayList<>(n);
+ final List<ByteBuffer> valueByteBuffers = new ArrayList<>(n);
+ final int initialCapacity = bufferCapacity.get();
+ for (KEY key : keys) {
+ final CodecBuffer keyBuffer = keyCodec.toDirectCodecBuffer(key);
+ keyBuffers.add(keyBuffer);
+ keyByteBuffers.add(keyBuffer.asReadOnlyByteBuffer());
+ final CodecBuffer valueBuffer =
CodecBuffer.allocateDirect(initialCapacity);
+ valueBuffers.add(valueBuffer);
+ addWritableValueByteBuffer(valueBuffer, initialCapacity,
valueByteBuffers);
+ }
+
+ try {
+ final List<VALUE> values = new ArrayList<>(Collections.nCopies(n, null));
+ final List<Integer> retryIndices = new ArrayList<>();
+ final List<Integer> retrySizes = new ArrayList<>();
+ try {
+ applyMultiGetStatuses(rawTable.multiGetSkipCache(keyByteBuffers,
valueByteBuffers),
+ valueBuffers, initialCapacity, values, retryIndices, retrySizes);
+ if (!retryIndices.isEmpty()) {
+ retryOversizedMultiGet(keyByteBuffers, retryIndices, retrySizes,
values);
+ bufferCapacity.increase(Collections.max(retrySizes));
+ }
+ return values;
+ } finally {
+ IOUtils.closeQuietly(valueBuffers);
+ }
+ } finally {
+ IOUtils.closeQuietly(keyBuffers);
+ }
+ }
+
+ void applyMultiGetStatuses(List<ByteBufferGetStatus> statuses,
List<CodecBuffer> valueBuffers,
+ int valueBufferCapacity, List<VALUE> values, List<Integer> retryIndices,
List<Integer> retrySizes)
+ throws CodecException {
+ for (int i = 0; i < statuses.size(); i++) {
+ final ByteBufferGetStatus status = statuses.get(i);
+ final CodecBuffer valueBuffer = valueBuffers.get(i);
+ if (status.value == null) {
+ continue;
+ }
+ validateMultiGetStatus(status, null);
+ if (status.requiredSize > valueBufferCapacity) {
+ retryIndices.add(i);
+ retrySizes.add(status.requiredSize);
+ } else {
+ values.set(i, decodeFromMultiGetStatus(status, valueBuffer));
+ }
+ }
+ }
+
+ void retryOversizedMultiGet(List<ByteBuffer> keyByteBuffers, List<Integer>
retryIndices,
+ List<Integer> retrySizes, List<VALUE> values) throws
RocksDatabaseException, CodecException {
+ final int retryCount = retryIndices.size();
+ final List<ByteBuffer> retryKeyByteBuffers = new ArrayList<>(retryCount);
+ final List<CodecBuffer> retryValueBuffers = new ArrayList<>(retryCount);
+ final List<ByteBuffer> retryValueByteBuffers = new ArrayList<>(retryCount);
+ for (int j = 0; j < retryCount; j++) {
+ final int requiredSize = retrySizes.get(j);
+
retryKeyByteBuffers.add(keyByteBuffers.get(retryIndices.get(j)).duplicate());
+ final CodecBuffer valueBuffer = CodecBuffer.allocateDirect(requiredSize);
+ retryValueBuffers.add(valueBuffer);
+ addWritableValueByteBuffer(valueBuffer, requiredSize,
retryValueByteBuffers);
+ }
+
+ try {
+ final List<ByteBufferGetStatus> retryStatuses =
+ rawTable.multiGetSkipCache(retryKeyByteBuffers,
retryValueByteBuffers);
+ for (int j = 0; j < retryCount; j++) {
+ final int index = retryIndices.get(j);
+ final int expectedSize = retrySizes.get(j);
+ final ByteBufferGetStatus status = retryStatuses.get(j);
+ final CodecBuffer valueBuffer = retryValueBuffers.get(j);
+ if (status.value == null) {
+ continue;
+ }
+ validateMultiGetStatus(status, expectedSize);
+ values.set(index, decodeFromMultiGetStatus(status, valueBuffer));
+ }
+ } finally {
+ IOUtils.closeQuietly(retryValueBuffers);
+ }
+ }
+
+ static void addWritableValueByteBuffer(CodecBuffer valueBuffer, int capacity,
+ List<ByteBuffer> valueByteBuffers) {
+ valueBuffer.clear();
+ final ByteBuffer out = valueBuffer.asWritableByteBuffer();
+ out.position(0).limit(capacity);
+ valueByteBuffers.add(out);
+ }
+
+ static void validateMultiGetStatus(ByteBufferGetStatus status, Integer
expectedSize) {
+ if (status.requiredSize < 0 || status.status.getCode() !=
org.rocksdb.Status.Code.Ok) {
+ throw new IllegalStateException("status = " + status.status.getCode()
+ + "; required = " + status.requiredSize + " < 0");
Review Comment:
The IllegalStateException message in validateMultiGetStatus is misleading:
it always says "< 0" even when the failure is due to a non-OK RocksDB status
code. This makes debugging multiGet failures harder.
--
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]
