HDDS-9904. Add Ratis-backed streaming reads [POC]

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java

@@ -130,6 +130,22 @@ public class OzoneClientConfig {
       tags = ConfigTag.CLIENT)
   private boolean streamReadBlock = false;
 
+  @Config(key = "ozone.client.ratis.stream.readblock.enable",
+      defaultValue = "false",
+      type = ConfigType.BOOLEAN,
+      description = "Allow ReadBlock to use Ratis data stream read-only requests.",
+      tags = ConfigTag.CLIENT)
+  private boolean ratisStreamReadBlock = false;
+
+  @Config(key = "ozone.client.ratis.stream.read.window-size",
+      defaultValue = "268435456",
+      type = ConfigType.LONG,
+      tags = {ConfigTag.CLIENT},
+      description = "Target request window for sequential ReadBlock requests "
+          + "over Ratis data stream. Larger windows reduce data stream setup "
+          + "overhead for large reads.")
+  private long ratisStreamReadWindowSize = 256L << 20;
+
   @Config(key = "ozone.client.max.retries",
       defaultValue = "5",
       description = "Maximum number of retries by Ozone Client on "
@@ -382,6 +398,13 @@ public void validate() {
       streamReadPreReadSize = 32L << 20; // 32MB
     }
 
+    if (ratisStreamReadWindowSize < 0) {
+      LOG.warn("Invalid ozone.client.ratis.stream.read.window-size = {}. " +
+              "Resetting to default 256MB.",
+          ratisStreamReadWindowSize);
+      ratisStreamReadWindowSize = 256L << 20; // 256MB
+    }
+
     // Ensure response data size is positive.
     if (streamReadResponseDataSize <= 0) {
       LOG.warn("Invalid ozone.client.stream.read.response-data-size = {}. " +
@@ -622,6 +645,14 @@ public void setStreamReadBlock(boolean streamReadBlock) {
     this.streamReadBlock = streamReadBlock;
   }
 
+  public boolean isRatisStreamReadBlock() {
+    return ratisStreamReadBlock;
+  }
+
+  public void setRatisStreamReadBlock(boolean ratisStreamReadBlock) {
+    this.ratisStreamReadBlock = ratisStreamReadBlock;
+  }
+
   public long getStreamReadPreReadSize() {
     return streamReadPreReadSize;
   }
@@ -630,6 +661,10 @@ public int getStreamReadResponseDataSize() {
     return streamReadResponseDataSize;
   }
 
+  public long getRatisStreamReadWindowSize() {
+    return ratisStreamReadWindowSize;
+  }
+
   public Duration getStreamReadTimeout() {
     return streamReadTimeout;
   }
@@ -642,6 +677,10 @@ public void setStreamReadResponseDataSize(int streamReadResponseDataSize) {
     this.streamReadResponseDataSize = streamReadResponseDataSize;
   }
 
+  public void setRatisStreamReadWindowSize(long ratisStreamReadWindowSize) {
+    this.ratisStreamReadWindowSize = ratisStreamReadWindowSize;
+  }
+
   public void setStreamReadTimeout(Duration streamReadTimeout) {
     this.streamReadTimeout = streamReadTimeout;
   }

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java

@@ -66,7 +66,9 @@ public MultipartInputStream(String keyName,
 
     this.key = keyName;
     this.partStreams = Collections.unmodifiableList(inputStreams);
-    this.isStreamBlockInputStream = !inputStreams.isEmpty() && inputStreams.get(0) instanceof StreamBlockInputStream;
+    this.isStreamBlockInputStream = !inputStreams.isEmpty()
+        && (inputStreams.get(0) instanceof StreamBlockInputStream
+        || inputStreams.get(0) instanceof RatisDataStreamBlockInputStream);
 
     // Calculate and update the partOffsets
     this.partOffsets = new long[inputStreams.size()];
@@ -75,7 +77,9 @@ public MultipartInputStream(String keyName,
     for (PartInputStream partInputStream : inputStreams) {
       this.partOffsets[i++] = streamLength;
       if (isStreamBlockInputStream) {
-        Preconditions.assertInstanceOf(partInputStream, StreamBlockInputStream.class);
+        Preconditions.assertTrue(partInputStream instanceof StreamBlockInputStream
+            || partInputStream instanceof RatisDataStreamBlockInputStream,
+            () -> "Unexpected stream block input stream " + partInputStream);
       }
       streamLength += partInputStream.getLength();
     }
@@ -199,7 +203,12 @@ public boolean readFully(long position, ByteBuffer buffer) throws IOException {
       read(new ByteBufferReader(buffer) {
         @Override
         int readImpl(InputStream inputStream) throws IOException {
-          return Preconditions.assertInstanceOf(inputStream, StreamBlockInputStream.class)
+          if (inputStream instanceof StreamBlockInputStream) {
+            return ((StreamBlockInputStream) inputStream)
+                .readFully(getBuffer(), false);
+          }
+          return Preconditions.assertInstanceOf(inputStream,
+              RatisDataStreamBlockInputStream.class)
               .readFully(getBuffer(), false);
         }
       });