feat(bindings/java): support duplicate operator

bindings/java/src/blocking_operator.rs

@@ -19,8 +19,8 @@ use jni::objects::JByteArray;
 use jni::objects::JClass;
 use jni::objects::JObject;
 use jni::objects::JString;
-use jni::sys::jbyteArray;
 use jni::sys::jobject;
+use jni::sys::{jbyteArray, jlong};
 use jni::JNIEnv;
 
 use opendal::BlockingOperator;
@@ -41,6 +41,19 @@ pub unsafe extern "system" fn Java_org_apache_opendal_BlockingOperator_disposeIn
     drop(Box::from_raw(op));
 }
 
+/// # Safety
+///
+/// This function should not be called before the Operator are ready.
+#[no_mangle]
+pub unsafe extern "system" fn Java_org_apache_opendal_BlockingOperator_duplicate(
+    _: JNIEnv,
+    _: JObject,
+    op: *mut BlockingOperator,
+) -> jlong {
+    let op = &mut *op;
+    Box::into_raw(Box::new(op.clone())) as jlong
+}
+
 /// # Safety
 ///
 /// This function should not be called before the Operator are ready.

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

@@ -50,6 +50,16 @@ public static BlockingOperator of(String schema, Map<String, String> map) {
         this.info = info;
     }
 
+    /**
+     * @return the cloned blocking operator.
+     *
+     * @see Operator#duplicate()
+     */
+    public BlockingOperator duplicate() {
+        final long nativeHandle = duplicate(this.nativeHandle);
+        return new BlockingOperator(nativeHandle, this.info);
+    }
+
     public void write(String path, String content) {
         write(path, content.getBytes(StandardCharsets.UTF_8));
     }
@@ -85,6 +95,8 @@ public void rename(String sourcePath, String targetPath) {
     @Override
     protected native void disposeInternal(long handle);
 
+    private static native long duplicate(long nativeHandle);
+
     private static native void write(long nativeHandle, String path, byte[] content);
 
     private static native byte[] read(long nativeHandle, String path);

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

@@ -122,6 +122,20 @@ public static Operator of(String schema, Map<String, String> map) {
         this.info = info;
     }
 
+    /**
+     * Clone a new operator that is identical to this one. The new operator has its own lifecycle.
+     *
+     * <p>Since an operator will release all its resource and "flush" on lifecycle end, this method
+     * is suitable to create a narrowed "scope" while avoiding creating a brand-new operator for each
+     * scope.
+     *
+     * @return the cloned operator.
+     */
+    public Operator duplicate() {
+        final long nativeHandle = duplicate(this.nativeHandle);
+        return new Operator(nativeHandle, this.info);
+    }
+
     public BlockingOperator blocking() {
         final long nativeHandle = makeBlockingOp(this.nativeHandle);
         final OperatorInfo info = this.info;
@@ -194,6 +208,8 @@ public CompletableFuture<Void> rename(String sourcePath, String targetPath) {
     @Override
     protected native void disposeInternal(long handle);
 
+    private static native long duplicate(long nativeHandle);
+
     private static native long constructor(String schema, Map<String, String> map);
 
     private static native long read(long nativeHandle, String path);

bindings/java/src/operator.rs

@@ -64,6 +64,19 @@ fn intern_constructor(env: &mut JNIEnv, scheme: JString, map: JObject) -> Result
     Ok(Box::into_raw(Box::new(op)) as jlong)
 }
 
+/// # Safety
+///
+/// This function should not be called before the Operator are ready.
+#[no_mangle]
+pub unsafe extern "system" fn Java_org_apache_opendal_Operator_duplicate(
+    _: JNIEnv,
+    _: JClass,
+    op: *mut Operator,
+) -> jlong {
+    let op = &mut *op;
+    Box::into_raw(Box::new(op.clone())) as jlong
+}
+
 /// # Safety
 ///
 /// This function should not be called before the Operator are ready.

bindings/java/src/test/java/org/apache/opendal/test/OperatorDuplicateTest.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.opendal.test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.opendal.BlockingOperator;
+import org.apache.opendal.Operator;
+import org.junit.jupiter.api.Test;
+
+public class OperatorDuplicateTest {
+    @Test
+    public void testDuplicateOperator() {
+        final Map<String, String> conf = new HashMap<>();
+        conf.put("root", "/opendal/");
+        try (final Operator op = Operator.of("memory", conf)) {
+            final String key = "key";
+            final byte[] v0 = "v0".getBytes(StandardCharsets.UTF_8);
+            final byte[] v1 = "v1".getBytes(StandardCharsets.UTF_8);
+
+            try (final Operator duplicatedOp = op.duplicate()) {
+                assertThat(duplicatedOp.info).isNotNull();
+                assertThat(duplicatedOp.info).isEqualTo(op.info);
+                duplicatedOp.write(key, v0).join();
+                assertThat(duplicatedOp.read(key).join()).isEqualTo(v0);
+            }
+
+            assertThat(op.read(key).join()).isEqualTo(v0);
+            op.write(key, v1).join();
+            assertThat(op.read(key).join()).isEqualTo(v1);
+        }
+    }
+
+    @Test
+    public void testDuplicateBlockingOperator() {
+        final Map<String, String> conf = new HashMap<>();
+        conf.put("root", "/opendal/");
+        try (final BlockingOperator op = BlockingOperator.of("memory", conf)) {
+            final String key = "key";
+            final byte[] v0 = "v0".getBytes(StandardCharsets.UTF_8);
+            final byte[] v1 = "v1".getBytes(StandardCharsets.UTF_8);
+
+            try (final BlockingOperator duplicatedOp = op.duplicate()) {
+                assertThat(duplicatedOp.info).isNotNull();
+                assertThat(duplicatedOp.info).isEqualTo(op.info);
+                duplicatedOp.write(key, v0);
+                assertThat(duplicatedOp.read(key)).isEqualTo(v0);
+            }
+
+            assertThat(op.read(key)).isEqualTo(v0);
+            op.write(key, v1);
+            assertThat(op.read(key)).isEqualTo(v1);
+        }
+    }
+}