Fix cleanup due to bug in Datafusion Reader not releasing catalog snapshot

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

Author: mgodwan

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

@@ -78,6 +78,9 @@ public void incRef() {
      */
     public void decRef() {
         readerHandle.close();
+        if (readerHandle.getRefCount() == 0) {
+            releaseCatalogSnapshot();
+        }
     }
 
     /**
@@ -91,6 +94,10 @@ public int getRefCount() {
     @Override
     public void close() {
         readerHandle.close();
+        releaseCatalogSnapshot();
+    }
+
+    private void releaseCatalogSnapshot() {
         try {
             if (catalogSnapshotRef != null)
                 catalogSnapshotRef.close();