Skip to content

Make Lucene indexing and search work alongside parquet #20182

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

Open
expani wants to merge 3 commits into
base: feature/datafusion
Choose a base branch
from
Open

Make Lucene indexing and search work alongside parquet #20182

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7c3a749
Initial changes to make Lucene work alongside Parquet
expani Dec 8, 2025
05dab3a
Rebased with mainline and refactoring
expani Dec 9, 2025
bb7b681
Null checks
expani Dec 9, 2025
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
2 changes: 1 addition & 1 deletion modules/mapper-extras/src/main/java/org/opensearch/index/mapper/ScaledFloatFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -498,7 +498,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
}
long scaledValue = Math.round(doubleValue * scalingFactor);

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), scaledValue);
} else {
List<Field> fields = NumberFieldMapper.NumberType.LONG.createFields(
Expand Down
2 changes: 1 addition & 1 deletion modules/mapper-extras/src/main/java/org/opensearch/index/mapper/TokenCountFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,7 +185,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
tokenCount = countPositions(analyzer, name(), value, enablePositionIncrements);
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), tokenCount);
} else {
context.doc()
Expand Down
4 changes: 2 additions & 2 deletions plugins/engine-datafusion/src/main/java/org/opensearch/datafusion/DatafusionEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,8 +126,8 @@ protected void doClose() {
}
};
} catch (Exception ex) {
logger.error("Failed to acquire searcher {}", ex.toString(), ex);
// TODO
logger.error("Failed to acquire searcher", ex);
throw new RuntimeException(ex);
}
return searcher;
}
Expand Down
2 changes: 1 addition & 1 deletion plugins/mapper-size/src/main/java/org/opensearch/index/mapper/size/SizeFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ public void postParse(ParseContext context) throws IOException {
}
final int value = context.sourceToParse().source().length();

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), value);
} else {
context.doc().addAll(NumberType.INTEGER.createFields(name(), value, true, true, false, true));
Expand Down
2 changes: 2 additions & 0 deletions server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,8 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
IndexSettings.INDEX_DERIVED_SOURCE_SETTING,
IndexSettings.INDEX_DERIVED_SOURCE_TRANSLOG_ENABLED_SETTING,

IndexSettings.OPTIMIZED_INDEX_ENABLED_SETTING,

// validate that built-in similarities don't get redefined
Setting.groupSetting("index.similarity.", (s) -> {
Map<String, Settings> groups = s.getAsGroups();
Expand Down
13 changes: 13 additions & 0 deletions server/src/main/java/org/opensearch/index/IndexSettings.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -810,6 +810,13 @@ public static IndexMergePolicy fromString(String text) {
Property.Dynamic
);

public static final Setting<Boolean> OPTIMIZED_INDEX_ENABLED_SETTING = Setting.boolSetting(
"index.optimized.enabled",
false,
Property.IndexScope,
Property.Final
);

private final Index index;
private final Version version;
private final Logger logger;
Expand Down Expand Up @@ -955,6 +962,8 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
*/
private volatile boolean isStarTreeIndexEnabled;

private final boolean isOptimizedIndex;

/**
* Returns the default search fields for this index.
*/
Expand Down Expand Up @@ -1119,6 +1128,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
setDocIdFuzzySetFalsePositiveProbability(scopedSettings.get(INDEX_DOC_ID_FUZZY_SET_FALSE_POSITIVE_PROBABILITY_SETTING));
isCompositeIndex = scopedSettings.get(StarTreeIndexSettings.IS_COMPOSITE_INDEX_SETTING);
isStarTreeIndexEnabled = scopedSettings.get(StarTreeIndexSettings.STAR_TREE_SEARCH_ENABLED_SETTING);
isOptimizedIndex = scopedSettings.get(OPTIMIZED_INDEX_ENABLED_SETTING);

scopedSettings.addSettingsUpdateConsumer(
TieredMergePolicyProvider.INDEX_COMPOUND_FORMAT_SETTING,
tieredMergePolicyProvider::setNoCFSRatio
Expand Down Expand Up @@ -2166,4 +2177,6 @@ public boolean isDerivedSourceEnabledForTranslog() {
public boolean isDerivedSourceEnabled() {
return derivedSourceEnabled;
}

public boolean isOptimizedIndex() { return isOptimizedIndex; }
}
9 changes: 9 additions & 0 deletions server/src/main/java/org/opensearch/index/engine/exec/bridge/Indexer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import org.opensearch.index.engine.EngineException;
import org.opensearch.index.engine.SafeCommitInfo;
import org.opensearch.index.engine.Segment;
import org.opensearch.index.engine.exec.composite.CompositeDataFormatWriter;
import org.opensearch.index.seqno.SequenceNumbers;
import org.opensearch.index.translog.Translog;
import org.opensearch.index.translog.TranslogManager;
Expand Down Expand Up @@ -233,6 +234,14 @@ default void maybeDie(final Logger logger, final String maybeMessage, final Thro
});
}

default CompositeDataFormatWriter.CompositeDocumentInput documentInput() {
return null;
}

default long getNativeBytesUsed() {
return 0;
}

/**
* Event listener for the engine
*
Expand Down
3 changes: 2 additions & 1 deletion server/src/main/java/org/opensearch/index/engine/exec/coord/CompositeEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -409,7 +409,8 @@ LocalCheckpointTracker getLocalCheckpointTracker() {
return readEngines.values().stream().filter(list -> !list.isEmpty()).findFirst().map(List::getFirst).orElse(null);
}

public CompositeDataFormatWriter.CompositeDocumentInput documentInput() throws IOException {
@Override
public CompositeDataFormatWriter.CompositeDocumentInput documentInput() {
return engine.createCompositeWriter().newDocumentInput();
}

Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/BinaryFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
return;
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), value);
} else {
if (stored) {
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/BooleanFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
return;
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), value);
} else {
if (indexed) {
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/DateFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -849,7 +849,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
}
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), timestamp);
} else {
if (indexed) {
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/DocCountFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
throw new IllegalArgumentException("Field [" + fieldType().name() + "] must be a positive integer.");
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), value);
} else {
final Field docCount = new NumericDocValuesField(NAME, value);
Expand Down
6 changes: 3 additions & 3 deletions server/src/main/java/org/opensearch/index/mapper/FieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,7 @@ protected final void createFieldNamesField(ParseContext context) {
FieldNamesFieldType fieldNamesFieldType = context.docMapper().metadataMapper(FieldNamesFieldMapper.class).fieldType();
if (fieldNamesFieldType != null && fieldNamesFieldType.isEnabled()) {
for (String fieldName : FieldNamesFieldMapper.extractFieldNames(fieldType().name())) {
if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldNamesFieldType, fieldName);
} else {
context.doc().add(new Field(FieldNamesFieldMapper.NAME, fieldName, FieldNamesFieldMapper.Defaults.FIELD_TYPE));
Expand All @@ -346,8 +346,8 @@ protected final void createFieldNamesField(ParseContext context) {
}
}

protected final boolean isPluggableDataFormatFeatureEnabled() {
return FeatureFlags.isEnabled(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG);
protected final boolean isPluggableDataFormatFeatureEnabled(ParseContext parseContext) {
return FeatureFlags.isEnabled(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) && parseContext.indexSettings().isOptimizedIndex();
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/IdFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,7 +298,7 @@ private IdFieldMapper(Supplier<Boolean> fieldDataEnabled) {
@Override
public void preParse(ParseContext context) {
BytesRef id = Uid.encodeId(context.sourceToParse().id());
if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), id);
} else {
context.doc().add(new Field(NAME, id, Defaults.FIELD_TYPE));
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/IgnoredFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,7 @@ private IgnoredFieldMapper() {
@Override
public void postParse(ParseContext context) {
for (String field : context.getIgnoredFields()) {
if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), field);
} else {
context.doc().add(new Field(NAME, field, Defaults.FIELD_TYPE));
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/IpFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -665,7 +665,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
}
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), address);
} else {
if (indexed && hasDocValues) {
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -862,7 +862,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
value = normalizeValue(normalizer, name(), value);
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), value);
} else {
// convert to utf8 only once before feeding postings/dv/stored fields
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/NumberFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2183,7 +2183,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
numericValue = fieldType().type.parse(value, coerce.value());
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), numericValue);
} else {
context.doc().addAll(fieldType().type.createFields(fieldType().name(), numericValue, indexed, hasDocValues, skiplist, stored));
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/RoutingFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ public boolean required() {
public void preParse(ParseContext context) {
String routing = context.sourceToParse().routing();
if (routing != null) {
if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), routing);
} else {
context.doc().add(new Field(fieldType().name(), routing, Defaults.FIELD_TYPE));
Expand Down
2 changes: 1 addition & 1 deletion server/src/main/java/org/opensearch/index/mapper/TextFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1046,7 +1046,7 @@ protected void parseCreateField(ParseContext context) throws IOException {
return;
}

if (isPluggableDataFormatFeatureEnabled()) {
if (isPluggableDataFormatFeatureEnabled(context)) {
context.compositeDocumentInput().addField(fieldType(), value);
} else {
if (fieldType.indexOptions() != IndexOptions.NONE || fieldType.stored()) {
Expand Down
Loading
Loading