Fix cleanup due to bug in Datafusion Reader not releasing catalog snapshot (#20165)

Signed-off-by: Mohit Godwani <[email protected]>

Author: mgodwan

plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/jni/handle/ReaderHandle.java

@@ -11,17 +11,23 @@
 import org.opensearch.datafusion.jni.NativeBridge;
 import org.opensearch.vectorized.execution.jni.RefCountedNativeHandle;
 
+import java.io.Closeable;
+
 /**
  * Reference-counted handle for native reader.
  */
 public final class ReaderHandle extends RefCountedNativeHandle {
 
-    public ReaderHandle(String path, String[] files) {
+    private final Runnable onClose;
+
+    public ReaderHandle(String path, String[] files, Runnable onClose) {
         super(NativeBridge.createDatafusionReader(path, files));
+        this.onClose = onClose;
     }
 
     @Override
     protected void doClose() {
         NativeBridge.closeDatafusionReader(ptr);
+        onClose.run();
     }
 }

plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/search/DatafusionReader.java

@@ -55,7 +55,7 @@ public DatafusionReader(String directoryPath, CompositeEngine.ReleasableRef<Cata
         }
         System.out.println("File names: " + Arrays.toString(fileNames));
         System.out.println("Directory path: " + directoryPath);
-        this.readerHandle = new ReaderHandle(directoryPath, fileNames);
+        this.readerHandle = new ReaderHandle(directoryPath, fileNames, this::releaseCatalogSnapshot);
     }
 
     /**
@@ -91,6 +91,9 @@ public int getRefCount() {
     @Override
     public void close() {
         readerHandle.close();
+    }
+
+    private void releaseCatalogSnapshot() {
         try {
             if (catalogSnapshotRef != null)
                 catalogSnapshotRef.close();