Skip to content

Commits of plugin engine-datafusion - #271

Merged
sandeshkr419 merged 112 commits into
3.3-dffrom
feature/datafusion
Dec 1, 2025
Merged

Commits of plugin engine-datafusion#271
sandeshkr419 merged 112 commits into
3.3-dffrom
feature/datafusion

Conversation

@sandeshkr419

Copy link
Copy Markdown
Owner

Description

[Describe what this change achieves]

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

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.

mch2 and others added 30 commits November 4, 2025 10:20
Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
(cherry picked from commit cb75910)
Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
(cherry picked from commit eb01905)
…project#19399)

Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
(cherry picked from commit c9d5b17)
Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
(cherry picked from commit 98de93e)
Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
(cherry picked from commit 5fef617)
Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
(cherry picked from commit e4ebf59)
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Co-authored-by: Arpit Bandejiya <abandeji@amazon.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Co-authored-by: Arpit Bandejiya <abandeji@amazon.com>
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
* Abstracting lucene away: part 1

* initial abstractions to reduce indexing engine coupling

* Text backed engine testing

---------

Co-authored-by: Mohit Godwani <mgodwan@amazon.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
* Changes in dataformat for CSVEngine

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

* Changes for Reader to work

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

---------

Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Co-authored-by: Bharathwaj G <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
* Integrate aggregators to convert result from datafusion

Signed-off-by: expani <anijainc@amazon.com>

* Initialised bigArrays and queryCollManagers for DatafusionContext

Signed-off-by: expani <anijainc@amazon.com>

* Refactored to set agg result within utility

Signed-off-by: expani <anijainc@amazon.com>

---------

Signed-off-by: expani <anijainc@amazon.com>
Co-authored-by: Arpit Bandejiya <abandeji@amazon.com>

Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
Signed-off-by: bharath-techie <bharath78910@gmail.com>
* Feature/datafusion 4 (#46)

* Composite document writer pool initial implementation

* Committer interface and lucene based commit engine implementation

* Catalog snapshot changes to create segment view during commit

---------

Co-authored-by: Shashank Gowri <shnkgo@amazon.com>

* fix build for commit integration

Signed-off-by: bharath-techie <bharath78910@gmail.com>

---------

Signed-off-by: bharath-techie <bharath78910@gmail.com>
Co-authored-by: Shashank Gowri <shnkgo@amazon.com>
expani and others added 24 commits November 17, 2025 13:07
* Add parquet benchmarks in missing javadoc

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

* add flag to disable java doc check

---------

Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
* Single commit with all changes

* Update the e2e test workflow

* Ignore lint

* Add missing doc flag

* fix newline

* remove extra missingdoc flag in action

* Add changes to have all queries

* Run only DatafusionReaderManager test

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

* Fix 200 status code

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

---------

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

# Conflicts:
#	plugins/engine-datafusion/jni/src/lib.rs
#	plugins/engine-datafusion/jni/src/listing_table.rs
#	plugins/engine-datafusion/jni/src/row_id_optimizer.rs
#	plugins/engine-datafusion/jni/src/util.rs

# Conflicts:
#	plugins/engine-datafusion/jni/src/lib.rs
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DataFusionQueryJNI.java
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DataFusionService.java
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/core/GlobalRuntimeEnv.java
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/search/DatafusionSearcher.java
#	plugins/engine-datafusion/src/test/java/org/opensearch/datafusion/DataFusionReaderManagerTests.java

# Conflicts:
#	plugins/engine-datafusion/jni/src/lib.rs
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DataFusionPlugin.java
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DataFusionService.java
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/core/GlobalRuntimeEnv.java
#	plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/search/DatafusionSearcher.java
#	plugins/engine-datafusion/src/test/java/org/opensearch/datafusion/DataFusionReaderManagerTests.java
#	plugins/engine-datafusion/src/test/java/org/opensearch/datafusion/DataFusionServiceTests.java
#	plugins/engine-datafusion/src/test/java/org/opensearch/datafusion/TestDataFusionServiceTests.java

* create CacheManager struct

* rebase

* add file2

* fix references

* cleanup commented code

* remove unused references

* rebase1

* Add Native Handles

* moving cacheOperations to use runtimeEnv

* fix references

* runtimeHandle reference

* rename configPtr reference
…#20048)

* add stream and allocator closure post fetch phase

Signed-off-by: Marc Handalian <marc.handalian@gmail.com>

* Move stream and allocator for executeQueryPhase into correct scope

Signed-off-by: Marc Handalian <marc.handalian@gmail.com>

---------

Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…nsearch-project#20032)

* Injecting count sub agg if absent to ensure doc_count is correct

Signed-off-by: expani <anijainc@amazon.com>

* Sorting keys for merge sort to work

Signed-off-by: expani <anijainc@amazon.com>

* Went into zen mode and made code simpler

Signed-off-by: expani <anijainc@amazon.com>

---------

Signed-off-by: expani <anijainc@amazon.com>
…ject#20041)

* Single commit with all changes

* Update the e2e test workflow

* Clean unused functions

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

* remove imports

---------

Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
Signed-off-by: Dhwanil Patel <dhwanip@amazon.com>
* fixing path reference while adding to cache

* remove unnecessary logs
…s with Searcher (opensearch-project#20063)

* CatalogSnapshot Manager + postDelete hooks

* Fix tests

* Minor fixes to post delete hooks

* Fix IndexFileDeleterTests

* Add IndexFileDeleter tests in E2E

* Empty commit to re-run E2E tests
Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>
…project#20079)

These tests run 42 or 43 queries right now and match results for correctness.
all queries flagged as supported are expected to succeed for this check to pass

Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
Signed-off-by: Marc Handalian <marc.handalian@gmail.com>
…sync changes (opensearch-project#20071)

* Changes for tokio io/cpu runtime + search flow async changes

Signed-off-by: bharath-techie <bharath78910@gmail.com>

* minor fixes and todos

Signed-off-by: bharath-techie <bharath78910@gmail.com>

* Addressing comment to move to ActionListener

Signed-off-by: bharath-techie <bharath78910@gmail.com>

* addressing comments

Signed-off-by: bharath-techie <bharath78910@gmail.com>

* fix to sort numeric aggs

Signed-off-by: bharath-techie <bharath78910@gmail.com>

---------

Signed-off-by: bharath-techie <bharath78910@gmail.com>
…pensearch-project#19978)

* Adding JNI abstractions for logging in rust through RustLoggerBridge

Signed-off-by: Raghuvansh Raj <raghraaj@amazon.com>

* Adding logging for java in parquet-data-format module and cleaning up stale files

Signed-off-by: Raghuvansh Raj <raghraaj@amazon.com>

---------

Signed-off-by: Raghuvansh Raj <raghraaj@amazon.com>
* Refactor stream handling for DF execution

* Change to Buffer Allocator for DF execution

* Introduce RecordBatchIterator and integrate with collector

* Minor fixup

* Update VSRManager to update field vector atomically

* Fixup

* Fixup

* Fixup

* Test fixup

* Integrate with AsyncRecordBatchIterator

* Fix up
* Update MetadataCache Warm-up to include page-indexes

* disable dynamic metadtaa cache limit updates
@sandeshkr419
sandeshkr419 merged commit 6e7f034 into 3.3-df Dec 1, 2025
10 of 12 checks passed
@sandeshkr419
sandeshkr419 deleted the feature/datafusion branch December 1, 2025 08:49
@gemini-code-assist

Copy link
Copy Markdown

Summary of Changes

Hello @sandeshkr419, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces significant infrastructure for vectorized execution and columnar storage within OpenSearch. It adds a new parquet-data-format module, leveraging Rust via JNI for efficient Parquet file writing, and a vectorized-exec-spi library for common interfaces. Additionally, it integrates a new engine-datafusion plugin to utilize DataFusion for query execution, complete with cache management and runtime monitoring. The changes also include necessary build system updates, JVM configurations for native memory, and modifications to field mappers to support a pluggable data format.

Highlights

  • New Vectorized Execution Infrastructure: Introduced a new parquet-data-format module with Rust JNI integration for efficient Parquet file writing and a vectorized-exec-spi library for common vectorized execution interfaces.
  • DataFusion Plugin Integration: Added a new engine-datafusion plugin that leverages Rust JNI bindings to integrate the DataFusion query engine, complete with cache management and runtime monitoring capabilities.
  • Build System and JVM Configuration Updates: Modified .gitattributes and .gitignore to support Parquet files and Rust build artifacts. Updated jvm.options and opensearch.yml to include native memory tracking and DataFusion-specific memory pool settings. Gradle build scripts were also updated to include new modules and remote plugin loading.
  • Pluggable Data Format Feature for Mappers: Modified ScaledFloatFieldMapper.java and TokenCountFieldMapper.java to conditionally use a new pluggable data format feature, indicating a move towards more flexible data handling.
Ignored Files
  • Ignored by pattern: .github/workflows/** (1)
    • .github/workflows/datafusion-e2e-test.yml
Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature Command Description
Code Review /gemini review Performs a code review for the current pull request in its current state.
Pull Request Summary /gemini summary Provides a summary of the current pull request in its current state.
Comment @gemini-code-assist Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help /gemini help Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

  1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution.

@sandeshkr419
sandeshkr419 restored the feature/datafusion branch December 1, 2025 08:55

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request introduces a significant amount of new functionality by adding the engine-datafusion and parquet-data-format plugins. The changes are extensive, touching build configurations, adding new SPIs, and implementing JNI bridges to Rust code for high-performance data processing. While this is a substantial contribution, the PR in its current state has several critical issues that need to be addressed. Multiple checks (API annotations, JarHell, Javadocs) and test assertions are disabled, which is risky. There are also several bugs, such as incorrect enum values, wrong return types, and potential build-breaking configurations in the Rust code. The practice of adding new classes to dependency packages (e.g., org.apache.lucene) is also a major concern. It's recommended to address these fundamental issues and re-enable the disabled checks to ensure the stability and maintainability of the codebase.

@ExperimentalApi
public enum DataFormat {
/** CSV Format*/
CSV("parquet"),

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

The CSV enum constant is initialized with the string "parquet". This is likely a copy-paste error and will cause incorrect behavior when working with CSV data formats. It should be initialized with "csv".

Suggested change
CSV("parquet"),
CSV("csv"),

* compatible open source license.
*/

package org.apache.lucene.fields;

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

This file and others in this pull request are being added to the org.apache.lucene.* package. Adding classes to a package owned by a third-party dependency is a strong anti-pattern. It can lead to classpath issues, split-package problems with the Java Platform Module System (JPMS), and unexpected behavior if the dependency is updated. All new classes should reside within the project's own namespace, such as org.opensearch.*.

Comment on lines +194 to +252
match NativeParquetWriter::create_writer(filename, schema_address as i64) {
Ok(_) => 0,
Err(_) => -1,
}
}

#[unsafe(no_mangle)]
pub extern "system" fn Java_com_parquet_parquetdataformat_bridge_RustBridge_write(
mut env: JNIEnv,
_class: JClass,
file: JString,
array_address: jlong,
schema_address: jlong
) -> jint {
let filename: String = env.get_string(&file).expect("Couldn't get java string!").into();
match NativeParquetWriter::write_data(filename, array_address as i64, schema_address as i64) {
Ok(_) => 0,
Err(_) => -1,
}
}

#[unsafe(no_mangle)]
pub extern "system" fn Java_com_parquet_parquetdataformat_bridge_RustBridge_closeWriter(
mut env: JNIEnv,
_class: JClass,
file: JString
) -> jint {
let filename: String = env.get_string(&file).expect("Couldn't get java string!").into();
match NativeParquetWriter::close_writer(filename) {
Ok(_) => 0,
Err(_) => -1,
}
}

#[unsafe(no_mangle)]
pub extern "system" fn Java_com_parquet_parquetdataformat_bridge_RustBridge_flushToDisk(
mut env: JNIEnv,
_class: JClass,
file: JString
) -> jint {
let filename: String = env.get_string(&file).expect("Couldn't get java string!").into();
match NativeParquetWriter::flush_to_disk(filename) {
Ok(_) => 0,
Err(_) => -1,
}
}

#[unsafe(no_mangle)]
pub extern "system" fn Java_com_parquet_parquetdataformat_bridge_RustBridge_getFilteredNativeBytesUsed(
mut env: JNIEnv,
_class: JClass,
path_prefix: JString
) -> jlong {
let prefix: String = env.get_string(&path_prefix).expect("Couldn't get java string!").into();
match NativeParquetWriter::get_filtered_writer_memory_usage(prefix) {
Ok(memory_usage) => memory_usage as jlong,
Err(_) => 0,
}
}

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

The JNI functions in this file (e.g., Java_com_parquet_parquetdataformat_bridge_RustBridge_createWriter) return a jint to indicate success or failure. However, the corresponding Java native methods in RustBridge.java are declared as void and throws IOException. This is a signature mismatch. The Rust code should throw a Java exception using env.throw_new() on failure instead of returning an integer code. The current implementation will silently swallow errors on the Java side.

[package]
name = "rust"
version = "0.1.0"
edition = "2024"

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

The Rust edition is set to "2024". The 2024 edition of Rust is not yet stable and is only available on nightly toolchains. This will cause build failures for anyone using a stable Rust compiler. Please use the latest stable edition, which is 2021.

Suggested change
edition = "2024"
edition = "2021"

Comment on lines +88 to +101
// for (var element : elements) {
// validate(element);
//
// if (!checkPackage(element)) {
// continue;
// }
//
// // Skip all not-public elements
// checkPublicVisibility(null, element);
//
// if (element instanceof TypeElement) {
// process((TypeElement) element);
// }
// }

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

The entire processing loop for API annotations (@PublicApi, @ExperimentalApi, @DeprecatedApi) has been commented out. This effectively disables all API compatibility and visibility checks, which is critical for maintaining a stable public API for OpenSearch. This should be re-enabled and any resulting errors should be fixed.

Suggested change
// for (var element : elements) {
// validate(element);
//
// if (!checkPackage(element)) {
// continue;
// }
//
// // Skip all not-public elements
// checkPublicVisibility(null, element);
//
// if (element instanceof TypeElement) {
// process((TypeElement) element);
// }
// }
for (var element : elements) {
validate(element);
if (!checkPackage(element)) {
continue;
}
// Skip all not-public elements
checkPublicVisibility(null, element);
if (element instanceof TypeElement) {
process((TypeElement) element);
}
}

Comment on lines +177 to +181
def possibleCargoPaths = [
System.getenv('HOME') + '/.cargo/bin/cargo',
'/usr/local/bin/cargo',
'cargo'
]

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The logic to find the cargo executable by checking a list of hardcoded paths is brittle and may fail on systems where cargo is installed in a different location. It's better to rely on cargo being available in the system's PATH, which is the standard practice. This simplifies the build script and makes it more portable.

Comment on lines +129 to +147
public void indexDataToParquetEngine() throws IOException {
//Create Engine (take Schema as Input)
// IndexingExecutionEngine<ParquetDataFormat> indexingExecutionEngine = indexingEngine();
// //Create Writer
// ParquetWriter writer = (ParquetWriter) indexingExecutionEngine.createWriter();
// for (int i=0;i<10;i++) {
// //Get DocumentInput
// DocumentInput documentInput = writer.newDocumentInput();
// ParquetDocumentInput parquetDocumentInput = (ParquetDocumentInput) documentInput;
// //Populate data
// DummyDataUtils.populateDocumentInput(parquetDocumentInput);
// //Write document
// writer.addDoc(parquetDocumentInput);
// }
// writer.flush(null);
// writer.close();
// //refresh engine
// indexingExecutionEngine.refresh(null);
}

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

This method indexDataToParquetEngine appears to be for local testing. It's commented out, but it should be removed entirely from production code before merging.

Comment on lines +164 to +168
project(":libs:opensearch-vectorized-exec-spi"), // TODO
project(":plugins:engine-datafusion"), //TODO
project(":server"),
project(":modules:parquet-data-format"),
project(":modules:parquet-data-format:benchmarks")

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Several new projects are being excluded from Javadoc checks. While this might be acceptable during initial development, it's important to add Javadocs for public APIs before the feature is finalized to ensure maintainability. The TODO comments suggest this is temporary, which is good, but this should be tracked for completion.

-XX:+UnlockDiagnosticVMOptions

# Enabling debug logs for Allocators in Arrow
-Darrow.memory.debug.allocator=true

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The -Darrow.memory.debug.allocator=true option is being enabled by default. This is a debugging flag for Apache Arrow that can have a significant performance impact. It should not be enabled by default in a production configuration. Consider removing it or leaving it commented out with an explanation.

type.getEvictionType(clusterSettings));

NativeBridge.createCache(cacheManagerPtr, type.cacheTypeName, type.getSizeLimit(clusterSettings).getBytes(), type.getEvictionType(clusterSettings));
// clusterSettings.addSettingsUpdateConsumer(type.sizeLimitSetting,(v) -> NativeBridge.cacheManagerUpdateSizeLimitForCacheType(cacheManagerPtr, CacheType.METADATA.getCacheTypeName(),v.getBytes()));

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The line to add a settings update consumer for the cache size limit is commented out. This means that if the cache size limit is changed dynamically via the cluster settings API, the change will not be propagated to the native cache manager. This functionality should be enabled to allow for dynamic cache configuration.

@sandeshkr419
sandeshkr419 deleted the feature/datafusion branch December 1, 2025 08:59
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.