Changes for Reader to work

Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>

Author: alchemist51

plugins/engine-datafusion/jni/src/lib.rs

@@ -209,6 +209,8 @@ pub extern "system" fn Java_org_opensearch_datafusion_DataFusionQueryJNI_execute
     let table_path = shard_view.table_path();
     let files_meta = shard_view.files_meta();
 
+    println!("Table path: {}", table_path);
+    println!("Files: {:?}", files_meta);
 
     let list_file_cache = Arc::new(DefaultListFilesCache::default());
     list_file_cache.put(table_path.prefix(), files_meta);
@@ -252,7 +254,7 @@ pub extern "system" fn Java_org_opensearch_datafusion_DataFusionQueryJNI_execute
         // Create a new TableProvider
         let provider = Arc::new(ListingTable::try_new(config).unwrap());
         let shard_id = table_path.prefix().filename().expect("error in fetching Path");
-        ctx.register_table("logs", provider)
+        ctx.register_table("hits", provider)
             .expect("Failed to attach the Table");
 
     });

plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DatafusionEngine.java

@@ -56,7 +56,7 @@ public class DatafusionEngine extends SearchExecEngine<DatafusionContext, Datafu
 
     public DatafusionEngine(DataFormat dataFormat, Collection<FileMetadata> formatCatalogSnapshot, DataFusionService dataFusionService) throws IOException {
         this.dataFormat = dataFormat;
-        this.datafusionReaderManager = new DatafusionReaderManager("TODO://FigureOutPath", formatCatalogSnapshot);
+        this.datafusionReaderManager = new DatafusionReaderManager("/Users/abandeji/Public/workplace/OpenSearch/build/testclusters/runTask-0/data/nodes/0/indices/7xU89OS-Tn2_nO7CboVqMg/0/parquet", formatCatalogSnapshot);
         this.datafusionService = dataFusionService;
     }
 

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

@@ -31,11 +31,16 @@ public class DatafusionReader implements Closeable {
     public DatafusionReader(String directoryPath, Collection<FileMetadata> files) {
         this.directoryPath = directoryPath;
         this.files = files;
-        String[] fileNames = Objects.isNull(files) ? new String[]{} : files.stream().map(FileMetadata::fileName).toArray(String[]::new);
+        String[] fileNames = Objects.isNull(files) ? new String[]{"hits_data.parquet"} : files.stream().map(FileMetadata::fileName).toArray(String[]::new);
         this.cachePtr = DataFusionQueryJNI.createDatafusionReader(directoryPath, fileNames);
         incRef();
     }
 
+    /**
+     * Gets the cache pointer.
+     *
+     * @return the cache pointer
+     */
     public long getCachePtr() {
         return cachePtr;
     }
@@ -44,6 +49,11 @@ public void incRef() {
         refCount.getAndIncrement();
     }
 
+    /**
+     * Decrements the reference count.
+     *
+     * @throws IOException if an I/O error occurs
+     */
     public void decRef() throws IOException {
         if(refCount.get() == 0) {
             throw new IllegalStateException("Listing table has been already closed");

plugins/engine-datafusion/src/test/java/org/opensearch/datafusion/DataFusionServiceTests.java

@@ -101,7 +101,7 @@ public void testQueryPhaseExecutor() throws IOException {
         Map<String, Object[]> finalRes = new HashMap<>();
         DatafusionSearcher datafusionSearcher = null;
         try {
-            DatafusionEngine engine = new DatafusionEngine(DataFormat.CSV, List.of(new FileMetadata(new TextDF(), "generation-1-optimized.parquet")), service);
+            DatafusionEngine engine = new DatafusionEngine(DataFormat.CSV, List.of(new FileMetadata(new TextDF(), "hits_data.parquet")), service);
             datafusionSearcher = engine.acquireSearcher("Search");
 
 

server/src/main/java/org/opensearch/index/engine/exec/composite/CompositeIndexingExecutionEngine.java

@@ -54,7 +54,7 @@ public CompositeIndexingExecutionEngine(PluginsService pluginsService, Any dataf
     public CompositeIndexingExecutionEngine(PluginsService pluginsService) {
      try {
         DataSourcePlugin plugin = pluginsService.filterPlugins(DataSourcePlugin.class).stream()
-            .findFirst()
+            .findAny()
             .orElseThrow(() -> new IllegalArgumentException("dataformat [" + DataFormat.TEXT + "] is not registered."));
          delegates.add(plugin.indexingEngine());
      } catch (NullPointerException e) {

server/src/main/java/org/opensearch/index/engine/exec/coord/CompositeEngine.java

@@ -57,7 +57,9 @@ public CompositeEngine(MapperService mapperService, PluginsService pluginsServic
             for(org.opensearch.vectorized.execution.search.DataFormat dataFormat : searchEnginePlugin.getSupportedFormats()) {
                 SearchExecEngine<?,?,?,?> searchExecEngine = searchEnginePlugin.createEngine(dataFormat,
                     catalogSnapshot.getSearchableFiles(dataFormat.toString()));
-                readEngines.getOrDefault(dataFormat, new ArrayList<>()).add(searchExecEngine);
+                List<SearchExecEngine<?, ?, ?, ?>> readEngine = readEngines.getOrDefault(dataFormat, new ArrayList<>());
+                readEngine.add(searchExecEngine);
+                readEngines.put(dataFormat, readEngine);
                 // TODO : figure out how to do internal and external refresh listeners
                 // Maybe external refresh should be managed in opensearch core and plugins should always give
                 // internal refresh managers