feat(bindings/java): add OperatorReader and ReaderOptions

bindings/java/README.md

@@ -92,6 +92,25 @@ public class Main {
 Use the synchronous `Operator` for blocking calls, or `AsyncOperator` for
 `CompletableFuture`-based calls.
 
+## Reuse a reader
+
+`Operator.createReader(path, readerOptions)` creates an `OperatorReader` for repeated
+reads. `ReaderOptions` selects versions, conditions, and execution controls;
+each call selects its own byte range.
+
+```java
+ReaderOptions options = ReaderOptions.builder().chunk(8 * 1024 * 1024L).build();
+try (OperatorReader reader = op.createReader("large.bin", options)) {
+    byte[] first = reader.read(0, 1024);
+    byte[] next = reader.read(1024, 1024);
+}
+```
+
+The [Java task guide](../../website/docs/20-bindings/java/04-tasks.md#read-part-of-a-file)
+covers streams and resource lifetimes.
+See [ReaderOptions](src/main/java/org/apache/opendal/ReaderOptions.java) for all
+supported options, defaults, and constraints.
+
 ## Documentation
 
 The full user guide — getting started, connecting to services, common tasks, and

bindings/java/src/lib.rs

@@ -39,6 +39,7 @@ mod layer;
 mod operator;
 mod operator_input_stream;
 mod operator_output_stream;
+mod operator_reader;
 mod utility;
 
 pub(crate) type Result<T> = std::result::Result<T, error::Error>;
@@ -292,9 +293,64 @@ fn make_read_options<'a>(
     })
 }
 
-fn make_reader_options<'a>(
-    _: &mut Env<'a>,
-    _: &JObject,
+fn make_reader_options(
+    env: &mut Env,
+    options: &JObject,
 ) -> Result<opendal::options::ReaderOptions> {
-    Ok(opendal::options::ReaderOptions::default())
+    let concurrent = convert::read_int_field(env, options, "concurrent")?;
+    let chunk = convert::read_int64_field(env, options, "chunk")?;
+    let prefetch = convert::read_int_field(env, options, "prefetch")?;
+    let content_length_hint = convert::read_int64_field(env, options, "contentLengthHint")?;
+
+    if concurrent <= 0 {
+        return Err(Error::new(ErrorKind::ConfigInvalid, "concurrent must be positive").into());
+    }
+    let concurrent = usize::try_from(concurrent)
+        .map_err(|_| Error::new(ErrorKind::ConfigInvalid, "concurrent is too large"))?;
+    let chunk = match chunk {
+        -1 => None,
+        value if value > 0 => Some(
+            usize::try_from(value)
+                .map_err(|_| Error::new(ErrorKind::ConfigInvalid, "chunk is too large"))?,
+        ),
+        _ => {
+            return Err(
+                Error::new(ErrorKind::ConfigInvalid, "chunk must be -1 or positive").into(),
+            );
+        }
+    };
+    let prefetch = usize::try_from(prefetch)
+        .map_err(|_| Error::new(ErrorKind::ConfigInvalid, "prefetch must be non-negative"))?;
+    let content_length_hint = match content_length_hint {
+        -1 => None,
+        value => Some(u64::try_from(value).map_err(|_| {
+            Error::new(
+                ErrorKind::ConfigInvalid,
+                "contentLengthHint must be -1 or non-negative",
+            )
+        })?),
+    };
+
+    Ok(opendal::options::ReaderOptions {
+        version: convert::read_string_field(env, options, "version")?,
+        if_match: convert::read_string_field(env, options, "ifMatch")?,
+        if_none_match: convert::read_string_field(env, options, "ifNoneMatch")?,
+        if_version_match: convert::read_string_field(env, options, "ifVersionMatch")?,
+        if_version_not_match: convert::read_string_field(env, options, "ifVersionNotMatch")?,
+        if_modified_since: convert::read_instant_field_to_timestamp(
+            env,
+            options,
+            "ifModifiedSince",
+        )?,
+        if_unmodified_since: convert::read_instant_field_to_timestamp(
+            env,
+            options,
+            "ifUnmodifiedSince",
+        )?,
+        concurrent,
+        chunk,
+        prefetch,
+        content_length_hint,
+        ..Default::default()
+    })
 }

bindings/java/src/main/java/org/apache/opendal/Operator.java

@@ -23,6 +23,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Operator represents an underneath OpenDAL operator that accesses data
@@ -114,8 +115,46 @@ public byte[] read(String path, ReadOptions options) {
         return read(nativeHandle, path, options);
     }
 
+    /**
+     * Creates a reusable reader for a file with default execution options.
+     *
+     * @param path file path
+     * @return a reader that the caller must close
+     * @see #createReader(String, ReaderOptions)
+     */
+    public OperatorReader createReader(String path) {
+        return createReader(path, ReaderOptions.builder().build());
+    }
+
+    /**
+     * Creates a reusable reader for a file. Options apply to every read through the reader;
+     * each read selects its own byte range. The reader can outlive this operator.
+     * Creation does not read file contents or guarantee that the file exists.
+     * Version selection and conditions apply to all requests through the reader, including
+     * streams. All conditions must hold; a failed condition on an existing file
+     * fails with ConditionNotMatch and a missing file fails with NotFound. Depending on the
+     * service, errors may surface at creation or during reading.
+     *
+     * @param path file path
+     * @param options reader execution options
+     * @return a reader that the caller must close
+     * @throws OpenDALException if options are invalid (ConfigInvalid), the path is a directory
+     *     (IsADirectory), or the service does not support reads or a requested option (Unsupported)
+     * @throws IllegalStateException if this operator is closed
+     */
+    public OperatorReader createReader(String path, ReaderOptions options) {
+        if (isDisposed()) {
+            throw new IllegalStateException("Operator is closed");
+        }
+        Objects.requireNonNull(path, "path");
+        Objects.requireNonNull(options, "options");
+        return new OperatorReader(createReader(nativeHandle, path, options));
+    }
+
+    private static native long createReader(long operator, String path, ReaderOptions options);
+
     public OperatorInputStream createInputStream(String path) {
-        return new OperatorInputStream(this, path, ReadOptions.builder().build());
+        return createInputStream(path, ReadOptions.builder().build());
     }
 
     public OperatorInputStream createInputStream(String path, ReadOptions options) {

bindings/java/src/main/java/org/apache/opendal/OperatorInputStream.java

@@ -22,32 +22,46 @@
 import java.io.InputStream;
 import java.util.Objects;
 
+/**
+ * Reads a byte range sequentially through an {@link OperatorReader}.
+ * Each stream owns its native iterator and must be closed independently of its source reader.
+ * Reading a closed stream throws {@link IllegalStateException}.
+ */
 public class OperatorInputStream extends InputStream {
-    private static class Reader extends NativeObject {
-        private Reader(long nativeHandle) {
+    private static class NativeIterator extends NativeObject {
+        private NativeIterator(long nativeHandle) {
             super(nativeHandle);
         }
 
         @Override
         protected void disposeInternal(long handle) {
-            disposeReader(handle);
+            disposeIterator(handle);
         }
     }
 
-    private final Reader reader;
+    private final NativeIterator iterator;
 
     private int offset = 0;
     private byte[] bytes = new byte[0];
 
+    OperatorInputStream(long nativeHandle) {
+        this.iterator = new NativeIterator(nativeHandle);
+    }
+
     public OperatorInputStream(Operator operator, String path, ReadOptions options) {
-        final long op = operator.nativeHandle;
-        this.reader = new Reader(constructReader(op, path, options));
+        Objects.requireNonNull(options, "options");
+        try (OperatorReader source = operator.createReader(path)) {
+            this.iterator = new NativeIterator(source.createBytesIterator(options.offset, options.length));
+        }
     }
 
     @Override
-    public int read() {
+    public synchronized int read() {
+        if (iterator.isDisposed()) {
+            throw new IllegalStateException("OperatorInputStream is closed");
+        }
         if (bytes != null && offset >= bytes.length) {
-            bytes = readNextBytes(reader.nativeHandle);
+            bytes = readNextBytes(iterator.nativeHandle);
             offset = 0;
         }
 
@@ -59,18 +73,20 @@ public int read() {
     }
 
     @Override
-    public int read(byte[] b, int off, int len) {
+    public synchronized int read(byte[] b, int off, int len) {
         Objects.requireNonNull(b);
         if ((b.length | off | len) < 0 || len > b.length - off) {
             // Objects.checkFromIndexSize has only been available since Java 9
             throw new IndexOutOfBoundsException(
                     String.format("Range [%s, %<s + %s) out of bounds for length %s", off, len, b.length));
         }
-
+        if (iterator.isDisposed()) {
+            throw new IllegalStateException("OperatorInputStream is closed");
+        }
         int read = 0;
         while (len > 0) {
             if (bytes != null && offset >= bytes.length) {
-                bytes = readNextBytes(reader.nativeHandle);
+                bytes = readNextBytes(iterator.nativeHandle);
                 offset = 0;
             }
 
@@ -87,21 +103,19 @@ public int read(byte[] b, int off, int len) {
         }
 
         if (bytes != null && offset >= bytes.length) {
-            bytes = readNextBytes(reader.nativeHandle);
+            bytes = readNextBytes(iterator.nativeHandle);
             offset = 0;
         }
 
         return bytes != null ? read : (read != 0 ? read : -1);
     }
 
     @Override
-    public void close() {
-        reader.close();
+    public synchronized void close() {
+        iterator.close();
     }
 
-    private static native long constructReader(long op, String path, ReadOptions options);
-
-    private static native void disposeReader(long reader);
+    private static native void disposeIterator(long nativeHandle);
 
-    private static native byte[] readNextBytes(long reader);
+    private static native byte[] readNextBytes(long nativeHandle);
 }

bindings/java/src/main/java/org/apache/opendal/OperatorReader.java

@@ -0,0 +1,119 @@
+/*
+ * 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.opendal;
+
+/**
+ * Reads a file synchronously through a reusable Rust core reader.
+ * Each read selects its own range and does not advance a shared cursor.
+ * Reader options apply to every request, including streams.
+ * A reader does not snapshot the file;
+ * changes to the file may be visible to subsequent reads.
+ *
+ * <p>Close the reader when it is no longer needed, preferably with try-with-resources.
+ * Closing the operator that created it does not close the reader.
+ * Calls on one reader are serialized, including close.
+ *
+ * @see Operator#createReader(String, ReaderOptions)
+ * @see ReaderOptions
+ */
+public final class OperatorReader extends NativeObject {
+    OperatorReader(long nativeHandle) {
+        super(nativeHandle);
+    }
+
+    /**
+     * Reads the whole file into a Java byte array.
+     *
+     * @return file contents
+     * @throws OpenDALException if the file does not exist (NotFound), a condition fails
+     *     (ConditionNotMatch), or reading fails
+     * @throws IllegalStateException if this reader is closed
+     */
+    public byte[] read() {
+        return read(0, -1);
+    }
+
+    /**
+     * Reads a byte range into a Java byte array without advancing a shared cursor.
+     *
+     * @param offset non-negative starting byte offset
+     * @param length number of bytes to read, or -1 to read to the end; zero returns an empty array
+     * @return contents of the requested range
+     * @throws OpenDALException if the range is invalid (RangeNotSatisfied), the file does not
+     *     exist (NotFound), a condition fails (ConditionNotMatch), or reading fails
+     * @throws IllegalStateException if this reader is closed
+     */
+    public synchronized byte[] read(long offset, long length) {
+        if (isDisposed()) {
+            throw new IllegalStateException("OperatorReader is closed");
+        }
+        return readBytes(nativeHandle, offset, length);
+    }
+
+    /**
+     * Creates an independent stream over the whole file.
+     *
+     * @return a stream that the caller must close
+     * @see #createInputStream(long, long)
+     */
+    public OperatorInputStream createInputStream() {
+        return createInputStream(0, -1);
+    }
+
+    /**
+     * Creates a stream over a byte range using this reader's options.
+     * Each stream has its own cursor and native resources. Closing this reader does not
+     * close its streams, and closing a stream does not close this reader or its other streams.
+     *
+     * @param offset non-negative starting byte offset
+     * @param length number of bytes to read, or -1 to read to the end; zero creates an empty stream
+     * @return a stream that the caller must close
+     * @throws OpenDALException if the range is invalid (RangeNotSatisfied) or opening the stream fails
+     * @throws IllegalStateException if this reader is closed
+     */
+    public OperatorInputStream createInputStream(long offset, long length) {
+        return new OperatorInputStream(createBytesIterator(offset, length));
+    }
+
+    // The caller owns the returned iterator independently of this reader.
+    synchronized long createBytesIterator(long offset, long length) {
+        if (isDisposed()) {
+            throw new IllegalStateException("OperatorReader is closed");
+        }
+        return createBytesIterator(nativeHandle, offset, length);
+    }
+
+    /** Releases this reader's native resources. Repeated calls have no effect. */
+    @Override
+    public synchronized void close() {
+        super.close();
+    }
+
+    @Override
+    protected void disposeInternal(long handle) {
+        disposeReader(handle);
+    }
+
+    private static native byte[] readBytes(long reader, long offset, long length);
+
+    private static native long createBytesIterator(long reader, long offset, long length);
+
+    private static native void disposeReader(long reader);
+}