Skip to content

datafusion reader manager tests fix for catalog snapshot supplier #20339

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bharath-techie merged 1 commit into from
Dec 30, 2025
Merged

datafusion reader manager tests fix for catalog snapshot supplier #20339

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 11 additions & 12 deletions ...gine-datafusion/src/test/java/org/opensearch/datafusion/DataFusionReaderManagerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
import java.io.InputStream;
import java.nio.file.Path;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;


Expand Down Expand Up @@ -111,7 +110,7 @@ public void testInitialReaderCreation() throws IOException {
segment.addSearchableFiles(getMockDataFormat().name(), writerFileSet);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));

DatafusionSearcher searcher = engine.acquireSearcher("test");
DatafusionReader reader = searcher.getReader();
Expand All @@ -125,7 +124,7 @@ public void testInitialReaderCreation() throws IOException {
// Assert RefCount 1 -> 1 for latest catalogSnapshot holder
assertEquals(1, getRefCount(reader));
reader.close();
// assertEquals(-1, reader.getReaderPtr());
// assertEquals(-1, reader.getReaderPtr());
}

/** Test that multiple searchers share the same reader instance for efficiency */
Expand All @@ -141,7 +140,7 @@ public void testMultipleSearchersShareSameReader() throws IOException {
segment.addSearchableFiles(getMockDataFormat().name(), writerFileSet);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));

DatafusionSearcher searcher1 = engine.acquireSearcher("test1");
DatafusionSearcher searcher2 = engine.acquireSearcher("test2");
Expand Down Expand Up @@ -172,7 +171,7 @@ public void testReaderSurvivesPartialSearcherClose() throws IOException {
segment.addSearchableFiles(getMockDataFormat().name(), writerFileSet);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));

DatafusionSearcher searcher1 = engine.acquireSearcher("test1");
DatafusionSearcher searcher2 = engine.acquireSearcher("test2");
Expand Down Expand Up @@ -205,7 +204,7 @@ public void testRefreshCreatesNewReader() throws IOException {
segment1.addSearchableFiles(getMockDataFormat().name(), writerFileSet1);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment1), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment1), new HashMap<>(), noOpFileDeleterSupplier)));

DatafusionSearcher searcher1 = engine.acquireSearcher("test1");
DatafusionReader reader1 = searcher1.getReader();
Expand All @@ -220,7 +219,7 @@ public void testRefreshCreatesNewReader() throws IOException {
segment2.addSearchableFiles(getMockDataFormat().name(), writerFileSet2);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(2, 2, List.of(segment2), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(2, 2, List.of(segment2), new HashMap<>(), noOpFileDeleterSupplier)));

DatafusionSearcher searcher2 = engine.acquireSearcher("test2");
DatafusionReader reader2 = searcher2.getReader();
Expand Down Expand Up @@ -253,7 +252,7 @@ public void testDecRefAfterCloseThrowsException() throws IOException {
segment.addSearchableFiles(getMockDataFormat().name(), writerFileSet);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));

DatafusionSearcher searcher = engine.acquireSearcher("test");
DatafusionReader reader = searcher.getReader();
Expand Down Expand Up @@ -284,7 +283,7 @@ public void testReaderClosesAfterSearchRelease() throws IOException {
segment.addSearchableFiles(getMockDataFormat().name(), writerFileSet);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment), new HashMap<>(), noOpFileDeleterSupplier)));

// DatafusionReader readerR1 = readerManager.acquire();
DatafusionSearcher datafusionSearcherS1 = engine.acquireSearcher("Search");
Expand All @@ -307,7 +306,7 @@ public void testReaderClosesAfterSearchRelease() throws IOException {
segment2.addSearchableFiles(getMockDataFormat().name(), writerFileSet2);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(2, 2, List.of(segment2), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(2, 2, List.of(segment2), new HashMap<>(), noOpFileDeleterSupplier)));

// now check if new Reader is created with F2, F3
// DatafusionReader readerR2 = readerManager.acquire();
Expand Down Expand Up @@ -352,7 +351,7 @@ public void testSearch() throws Exception {
segment1.addSearchableFiles(getMockDataFormat().name(), writerFileSet1);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment1), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(1, 1, List.of(segment1), new HashMap<>(), noOpFileDeleterSupplier)));

DatafusionSearcher searcher1 = engine.acquireSearcher("search");
DatafusionReader reader1 = searcher1.getReader();
Expand Down Expand Up @@ -382,7 +381,7 @@ public void testSearch() throws Exception {
segment2.addSearchableFiles(getMockDataFormat().name(), writerFileSet2);

readerManager.afterRefresh(true,
getCatalogSnapshotRef(new CatalogSnapshot(2, 1, List.of(segment2), new HashMap<>(), noOpFileDeleterSupplier)));
() -> getCatalogSnapshotRef(new CatalogSnapshot(2, 1, List.of(segment2), new HashMap<>(), noOpFileDeleterSupplier)));

expectedResults = new HashMap<>();
expectedResults.put("min", 3L);
Expand Down
Loading