Add Lucene engine impl for pluggable data formats

[mgodwan]

Description

Add Lucene engine impl for pluggable data formats

Related Issues

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[mgodwan]

All of this will be replaced in upcoming PRs with a full fledged support registry

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 908328c)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

📝 TODO sections sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneCommitter.java [172]: Will return safe commit info once the index deleter is wired in. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [745]: Delegate to IndexingExecutionEngine's Merger when merge scheduling is implemented.

🔀 Multiple PR themes Sub-PR theme: Lucene per-generation writer and document input implementation Relevant files: sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriter.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriterCodec.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneDocumentInput.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneFieldFactoryRegistry.java sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneWriterTests.java Sub-PR theme: Lucene indexing execution engine, data format, and plugin wiring Relevant files: sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneDataFormat.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LucenePlugin.java sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/LuceneSearchBackEnd.java sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngineTests.java sandbox/plugins/analytics-backend-lucene/src/test/java/org/opensearch/be/lucene/LuceneReaderManagerTests.java Sub-PR theme: Composite engine and server-side API changes for pluggable data formats Relevant files: sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeParquetIndexIT.java server/src/main/java/org/opensearch/index/engine/exec/commit/IndexStoreProvider.java server/src/main/java/org/opensearch/index/engine/dataformat/DocumentInput.java server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java

⚡ Recommended focus areas for review Double Close Risk In flush(), after force-merge and commit, both indexWriter.close() and directory.close() are called. In close(), the code checks indexWriter.isOpen() before rolling back, but if flush() was called and then close() is called (e.g., in a try-with-resources), directory.close() will be called a second time, potentially causing errors or resource leaks. The directory field has no guard against double-close. public FileInfos flush() throws IOException { if (docCount == 0) { return FileInfos.empty(); } // Force merge to exactly 1 segment to maintain 1:1 mapping with other formats. indexWriter.forceMerge(1, true); indexWriter.commit(); // Verify the invariant: exactly 1 segment with docCount documents SegmentInfos segmentInfos = SegmentInfos.readLatestCommit(directory); assert segmentInfos.size() == 1 : "Expected exactly 1 segment after force merge, got " + segmentInfos.size(); SegmentCommitInfo segmentInfo = segmentInfos.info(0); assert segmentInfo.info.maxDoc() == docCount : "Expected " + docCount + " docs in segment, got " + segmentInfo.info.maxDoc(); // Build the WriterFileSet pointing to the temp directory WriterFileSet.Builder wfsBuilder = WriterFileSet.builder() .directory(tempDirectory) .writerGeneration(writerGeneration) .addNumRows(docCount); // Add all files in the segment for (String file : directory.listAll()) { if (file.startsWith("segments") == false && file.equals("write.lock") == false) { wfsBuilder.addFile(file); } } // Since flush is once only, we can close the write post this. indexWriter.close(); directory.close(); return FileInfos.builder().putWriterFileSet(dataFormat, wfsBuilder.build()).build(); } /** * Syncs all files in the temp directory to durable storage. * * @throws IOException if the sync fails */ @Override public void sync() throws IOException { directory.sync(Arrays.asList(directory.listAll())); directory.syncMetaData(); } /** {@inheritDoc} Returns the writer generation number assigned at construction. */ @Override public long generation() { return writerGeneration; } /** Acquires the writer's reentrant lock. Used by the writer pool to serialize access. */ @Override public void lock() { lock.lock(); } /** Attempts to acquire the writer's reentrant lock without blocking. */ @Override public boolean tryLock() { return lock.tryLock(); } /** Releases the writer's reentrant lock. */ @Override public void unlock() { lock.unlock(); } /** * Closes this writer, rolling back the IndexWriter if still open, closing the directory, * and deleting the temp directory. Safe to call multiple times. * * @throws IOException if cleanup fails */ @Override public void close() throws IOException { // Close the IndexWriter and Directory if they haven't been closed by flush() try { if (indexWriter.isOpen()) { indexWriter.rollback(); } } catch (Exception e) { logger.warn("Failed to rollback IndexWriter for generation {}", writerGeneration, e); } try { directory.close(); } catch (Exception e) { logger.warn("Failed to close directory for generation {}", writerGeneration, e); } IOUtils.rm(tempDirectory); } Assertion in Production The assertion assert writerGenerations.isEmpty() : "Could not get segments from all writers" at line 272 will only fire when assertions are enabled (-ea). If a writer generation is not found in the NRT reader after addIndexes (e.g., due to a merge or codec issue), this failure will be silently swallowed in production, leading to missing segments in the catalog snapshot without any error. assert writerGenerations.isEmpty() : "Could not get segments from all writers"; Temp Dir Collision The temp directory is named lucene_gen_<writerGeneration> under baseDirectory. If a writer for the same generation is created twice (e.g., after a failure and retry), Files.createDirectory will throw FileAlreadyExistsException rather than creating a fresh directory. There is no cleanup or retry logic for this case. this.tempDirectory = baseDirectory.resolve("lucene_gen_" + writerGeneration); logger.info("Creating directory for temp lucene writer: " + tempDirectory); Files.createDirectory(tempDirectory); Duplicate Import DocumentInput is imported twice at lines 17 and 18. This is a compilation warning/error depending on the Java version and should be cleaned up. import org.opensearch.index.engine.dataformat.DocumentInput; import org.opensearch.index.mapper.MappedFieldType; Resource Leak In refresh(), HardlinkCopyDirectoryWrapper(new MMapDirectory(dirPath)) is created and added to sourceDirectories. If addIndexes throws, the finally block closes all directories. However, if an exception occurs while building sourceDirectories (before the addIndexes call), directories already added to the list may not be closed since the finally block is only inside the if (sourceDirectories.isEmpty() == false) block. sourceDirectories.add(new HardlinkCopyDirectoryWrapper(new MMapDirectory(dirPath))); writerGenerations.add(wfs.writerGeneration()); } // Single batched addIndexes call for all source directories if (sourceDirectories.isEmpty() == false) { try { sharedWriter.addIndexes(sourceDirectories.toArray(new Directory[0])); logger.debug("Incorporated {} Lucene segments into shared writer in a single addIndexes call", sourceDirectories.size()); } finally { // Close all source directories for (Directory dir : sourceDirectories) { try { dir.close(); } catch (IOException e) { logger.warn("Failed to close source directory after addIndexes", e); } } }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 908328c

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Replace assert with hard runtime check for missing segments Using assert for this invariant check means it is silently ignored when assertions are disabled (the default in production JVMs). If not all writer generations are found in the NRT reader after addIndexes, data will be silently lost. This should be a hard runtime check (e.g., throw an IllegalStateException) to ensure correctness in production. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java [272] -assert writerGenerations.isEmpty() : "Could not get segments from all writers"; +if (!writerGenerations.isEmpty()) { + throw new IllegalStateException("Could not find segments for writer generations: " + writerGenerations); +} Suggestion importance[1-10]: 7 __ Why: Using assert for a critical data-integrity invariant is problematic since assertions are disabled by default in production JVMs, meaning silent data loss could occur. Replacing it with a hard IllegalStateException is a valid and important correctness improvement.	Medium
	Prevent double-close of directory after flush The indexWriter and directory are closed inside flush(), but close() also tries to close them. If flush() is called and then close() is called (e.g., in a finally block), directory.close() in close() will attempt to close an already-closed directory, potentially causing errors. The close() method already checks indexWriter.isOpen() before rolling back, but it unconditionally calls directory.close(). You should track whether the directory has been closed (e.g., with a flag) and skip closing it in close() if already done. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriter.java [175-177] -// Since flush is once only, we can close the write post this. +// Since flush is once only, we can close the writer post this. indexWriter.close(); directory.close(); +this.directoryClosed = true; Suggestion importance[1-10]: 6 __ Why: The directory is closed in flush() but close() also calls directory.close() unconditionally. The improved_code only adds a flag assignment without the corresponding field declaration or check in close(), making it incomplete. The issue is real but the suggestion's implementation is partial.	Low
General	Guard against duplicate temp directory creation If two writers with the same writerGeneration are created concurrently or sequentially without cleanup, Files.createDirectory will throw a FileAlreadyExistsException, crashing writer creation. Consider using Files.createDirectories or checking for existence first, or ensuring uniqueness of generation numbers before this call. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriter.java [95-97] this.tempDirectory = baseDirectory.resolve("lucene_gen_" + writerGeneration); -logger.info("Creating directory for temp lucene writer: " + tempDirectory); +logger.info("Creating directory for temp lucene writer: {}", tempDirectory); +if (Files.exists(tempDirectory)) { + throw new IOException("Temp directory already exists for generation " + writerGeneration + ": " + tempDirectory); +} Files.createDirectory(tempDirectory); Suggestion importance[1-10]: 4 __ Why: The suggestion adds a guard against duplicate generation directories, which is a valid defensive check. However, the generation uniqueness is expected to be enforced by the caller (DataFormatAwareEngine), making this a minor defensive improvement rather than a critical fix.	Low

---

Previous suggestions

Suggestions up to commit 28075a4

Category	Suggestion	Impact
Possible issue	Replace assert with explicit runtime check Using assert for this validation means it is silently skipped when assertions are disabled (the default in production JVMs). If a writer generation is not found in the NRT reader after addIndexes, it indicates a data loss or corruption scenario that should always be detected. Replace the assert with an explicit check that throws an IllegalStateException. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java [268] -assert writerGenerations.isEmpty() : "Could not get segments from all writers"; +if (!writerGenerations.isEmpty()) { + throw new IllegalStateException("Could not get segments from all writers; missing generations: " + writerGenerations); +} Suggestion importance[1-10]: 7 __ Why: The assert statement is disabled in production JVMs by default, meaning a missing writer generation (indicating potential data loss) would go undetected. Replacing it with an explicit IllegalStateException ensures the error is always caught.	Medium
	Guard sync against already-closed directory Calling directory.listAll() after indexWriter.close() and directory.close() (which happens in flush()) will throw an AlreadyClosedException. The sync() method should be called before flush() closes the directory, or it should guard against the closed state. Additionally, syncing all files (including write.lock and segments_*) is unnecessary and potentially harmful; only the segment data files should be synced. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriter.java [180-183] +@Override public void sync() throws IOException { + if (!indexWriter.isOpen()) { + throw new IllegalStateException("Cannot sync after writer has been closed by flush()"); + } directory.sync(Arrays.asList(directory.listAll())); directory.syncMetaData(); } Suggestion importance[1-10]: 6 __ Why: The sync() method calls directory.listAll() which will throw AlreadyClosedException if called after flush() has closed the directory and indexWriter. Adding a guard check prevents this potential runtime error.	Low
	Prevent directory resource leak on partial failure If an exception occurs while opening one of the MMapDirectory or HardlinkCopyDirectoryWrapper instances (e.g., for a later segment), the already-opened directories in sourceDirectories will be leaked because the finally block that closes them is only reached after the loop completes. The directory should be opened inside a try-with-resources or the exception handling should ensure partial lists are closed. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java [211] -sourceDirectories.add(new HardlinkCopyDirectoryWrapper(new MMapDirectory(dirPath))); +Directory srcDir = new HardlinkCopyDirectoryWrapper(new MMapDirectory(dirPath)); +try { + sourceDirectories.add(srcDir); +} catch (Exception e) { + srcDir.close(); + throw e; +} Suggestion importance[1-10]: 5 __ Why: If an exception occurs while opening one of the directories in the loop, previously opened directories in sourceDirectories would be leaked since the finally block only runs after the loop completes. However, the improved_code doesn't actually prevent the leak for already-added directories, making the fix incomplete.	Low
General	Avoid recreating provider object on every call A new anonymous IndexStoreProvider object (with a new providers map) is created on every call to getProvider(). This is wasteful and could cause subtle issues if callers compare provider identity. The composite provider should be built once (e.g., lazily or in the constructor) and reused. sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java [332-351] +private final IndexStoreProvider compositeProvider = new IndexStoreProvider() { + private final Map<DataFormat, IndexStoreProvider> providers; + { + Map<DataFormat, IndexStoreProvider> tempProviders = new HashMap<>(); + tempProviders.put(primaryEngine.getDataFormat(), primaryEngine.getProvider()); + tempProviders.putAll( + secondaryEngines.stream() + .map(eng -> new AbstractMap.SimpleEntry<>(eng.getDataFormat(), eng.getProvider())) + .collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue)) + ); + providers = tempProviders; + } + + @Override + public FormatStore getStore(DataFormat dataFormat) { + return providers.get(dataFormat).getStore(dataFormat); + } +}; + @Override public IndexStoreProvider getProvider() { - return new IndexStoreProvider() { - private final Map<DataFormat, IndexStoreProvider> providers; - { - Map<DataFormat, IndexStoreProvider> tempProviders = new HashMap<>(); - tempProviders.put(primaryEngine.getDataFormat(), primaryEngine.getProvider()); - tempProviders.putAll( - secondaryEngines.stream() - .map(eng -> new AbstractMap.SimpleEntry<>(eng.getDataFormat(), eng.getProvider())) - .collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue)) - ); - providers = tempProviders; - } - - @Override - public FormatStore getStore(DataFormat dataFormat) { - return providers.get(dataFormat).getStore(dataFormat); - } - }; + return compositeProvider; } Suggestion importance[1-10]: 4 __ Why: Creating a new IndexStoreProvider instance with a new providers map on every getProvider() call is wasteful. Caching it as a field would be more efficient, though the functional impact is minor since the map contents would be the same each time.	Low

Suggestions up to commit 9dd0022

Category	Suggestion	Impact
Possible issue	Replace assertion with hard runtime check Java assertions are disabled by default in production JVMs (unless -ea is passed). If not all writer generations are found in the NRT reader after addIndexes, this will silently pass in production, potentially causing data loss or incorrect catalog snapshots. This should be a hard runtime check (e.g., throw an IllegalStateException) rather than an assertion. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java [268] -assert writerGenerations.isEmpty() : "Could not get segments from all writers"; +if (!writerGenerations.isEmpty()) { + throw new IllegalStateException("Could not get segments from all writers; missing generations: " + writerGenerations); +} Suggestion importance[1-10]: 7 __ Why: This is a valid and important concern: Java assertions are disabled by default in production, so a missing writer generation would silently pass, potentially causing incorrect catalog snapshots or data loss. Replacing the assertion with an IllegalStateException is a meaningful correctness improvement.	Medium
	Guard against double-close after flush After indexWriter.close() and directory.close() are called in flush(), the close() method still attempts to call indexWriter.rollback() and directory.close() again. While there are try/catch blocks, calling rollback() on an already-closed IndexWriter may throw an AlreadyClosedException. A boolean flag (e.g., flushed) should be set after flush to guard against double-close in close(). sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriter.java [167-169] // Since flush is once only, we can close the write post this. indexWriter.close(); directory.close(); +this.flushed = true; Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that close() may attempt to call indexWriter.rollback() and directory.close() after flush() has already closed them. The close() method has try/catch blocks that swallow exceptions, so this is handled gracefully, but adding a flushed flag would be cleaner and more explicit. The improved code only shows setting the flag but not using it in close(), making it incomplete.	Low
	Prevent directory resource leaks during construction The MMapDirectory instances created here are wrapped but never explicitly tracked for closure in the failure path before addIndexes is called. If addIndexes throws, the finally block closes them, but if an exception occurs while building sourceDirectories (e.g., on a later iteration), previously created MMapDirectory instances may leak. Consider using a try-with-resources or ensuring all directories are closed in a single finally block that covers the entire construction loop. sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java [211] -sourceDirectories.add(new HardlinkCopyDirectoryWrapper(new MMapDirectory(dirPath))); +Directory mmap = new MMapDirectory(dirPath); +try { + sourceDirectories.add(new HardlinkCopyDirectoryWrapper(mmap)); +} catch (Exception e) { + mmap.close(); + throw e; +} Suggestion importance[1-10]: 5 __ Why: The suggestion identifies a real but unlikely resource leak: if MMapDirectory construction succeeds but HardlinkCopyDirectoryWrapper throws, the MMapDirectory would leak. However, the finally block already handles closing all directories after addIndexes, and the scenario where HardlinkCopyDirectoryWrapper throws during construction is very rare. The improved code is valid but the risk is low.	Low
General	Handle pre-existing temp directory on writer creation Using Files.createDirectory will throw FileAlreadyExistsException if a writer with the same generation already exists (e.g., due to a retry or crash recovery). This could cause the engine to fail unexpectedly. Consider using Files.createDirectories or checking for existence first, and handle the case where the directory already exists (e.g., clean it up or fail with a clear error). sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriter.java [96-97] -logger.info("Creating directory for temp lucene writer: " + tempDirectory); +logger.info("Creating directory for temp lucene writer: {}", tempDirectory); +if (Files.exists(tempDirectory)) { + IOUtils.rm(tempDirectory); + logger.warn("Removed existing temp directory for writer generation {}: {}", writerGeneration, tempDirectory); +} Files.createDirectory(tempDirectory); Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that Files.createDirectory will throw if the directory already exists, which could happen during crash recovery or retries. The improved code adds cleanup logic and also fixes the string concatenation in the log statement to use parameterized logging. The fix is reasonable but the scenario may be handled at a higher level.	Low

[github-actions]

✅ Gradle check result for 9dd0022: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.30%. Comparing base (255c867) to head (9dd0022).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##               main   #21299    +/-   ##
==========================================
  Coverage     73.30%   73.30%            
+ Complexity    73756    73731    -25     
==========================================
  Files          5936     5936            
  Lines        335749   335753     +4     
  Branches      48396    48397     +1     
==========================================
+ Hits         246106   246109     +3     
+ Misses        70031    69926   -105     
- Partials      19612    19718   +106     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

Persistent review updated to latest commit 28075a4

[github-actions]

❌ Gradle check result for 28075a4: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[Bukhtawar]

Overall looks good, minor suggestions raised

[github-actions]

Persistent review updated to latest commit 908328c

[github-actions]

❌ Gradle check result for 908328c: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit 59d19c2.

Path	Line	Severity	Description
sandbox/plugins/parquet-data-format/build.gradle	28	high	Dependency version bump: 'org.checkerframework:checker-qual' changed from 3.42.0 to 3.43.0. Per mandatory rule, all dependency version changes must be flagged. Maintainers must verify the artifact at version 3.43.0 is authentic and not a supply chain substitution. The corresponding SHA1 file was also replaced (old: 638ec33f363a94d41a4f03c3e7d3dcfba64e402d → new: 9425eee39e56b116d2b998b7c2cebcbd11a3c98b).
sandbox/plugins/composite-engine/build.gradle	46	high	New dependency added: 'sandbox:plugins:analytics-backend-datafusion' added as both testImplementation and internalClusterTestImplementation. Per mandatory rule, all new dependency additions must be flagged regardless of how legitimate the name appears. Maintainers must verify this artifact's origin, contents, and that the project ':sandbox:plugins:analytics-backend-datafusion' is an expected internal module.

The table above displays the top 10 most important findings.

Total: 2 | Critical: 0 | High: 2 | Medium: 0 | Low: 0

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.