Indexing changes for composite engine (#20790)

* Lucene + parquet indexing changes

* Add changes for mappers

* Add secondary index setting

* Add test fix

Author: alchemist51

.gitignore

@@ -1,6 +1,8 @@
 .claude
 CLAUDE.md
 .cursor*
+.kiro*
+examples/
 
 # intellij files
 .idea/

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/ParquetDataFormatPlugin.java

@@ -23,15 +23,18 @@
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
+import org.opensearch.index.engine.EngineConfig;
 import org.opensearch.index.engine.exec.DataFormat;
+import org.opensearch.index.engine.exec.FieldAssignments;
+import org.opensearch.index.engine.exec.FieldSupportRegistry;
 import org.opensearch.index.engine.exec.IndexingExecutionEngine;
 import com.parquet.parquetdataformat.bridge.RustBridge;
 import com.parquet.parquetdataformat.engine.ParquetExecutionEngine;
+import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.shard.ShardPath;
 import org.opensearch.index.store.FormatStoreDirectory;
 import org.opensearch.index.store.GenericStoreDirectory;
 import org.opensearch.plugins.DataSourcePlugin;
-import org.opensearch.index.mapper.MapperService;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.plugins.spi.vectorized.DataSourceCodec;
 import org.opensearch.repositories.RepositoriesService;
@@ -82,8 +85,15 @@ public class ParquetDataFormatPlugin extends Plugin implements DataSourcePlugin
 
     @Override
     @SuppressWarnings("unchecked")
-    public <T extends DataFormat> IndexingExecutionEngine<T> indexingEngine(MapperService mapperService, ShardPath shardPath, IndexSettings indexSettings) {
-        return (IndexingExecutionEngine<T>) new ParquetExecutionEngine(settings, () -> ArrowSchemaBuilder.getSchema(mapperService), shardPath, indexSettings);
+    public <T extends DataFormat> IndexingExecutionEngine<T> indexingEngine(EngineConfig engineConfig, MapperService mapperService, boolean isPrimary, ShardPath shardPath, IndexSettings indexSettings, FieldAssignments fieldAssignments) {
+        ParquetExecutionEngine engine = new ParquetExecutionEngine(
+            settings,
+            isPrimary,
+            () -> ArrowSchemaBuilder.getSchema(mapperService, isPrimary),
+            shardPath,
+            indexSettings
+        );
+        return (IndexingExecutionEngine<T>) engine;
     }
 
     @Override
@@ -136,6 +146,15 @@ public BlobContainer createBlobContainer(BlobStore blobStore, BlobPath baseBlobP
         return blobStore.blobContainer(formatPath);
     }
 
+    @Override
+    public void registerFieldSupport(FieldSupportRegistry registry) {
+        DataFormat parquet = getDataFormat();
+        for (Map.Entry<String, com.parquet.parquetdataformat.fields.ParquetField> entry :
+                com.parquet.parquetdataformat.fields.ArrowFieldRegistry.getRegisteredFields().entrySet()) {
+            registry.register(entry.getKey(), parquet, entry.getValue().getFieldCapabilities());
+        }
+    }
+
     @Override
     public List<Setting<?>> getSettings() {
         return List.of(

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/engine/ParquetDataFormat.java

@@ -50,12 +50,14 @@ public void configureStore() {
 
     @Override
     public boolean equals(Object obj) {
-        return true;
+        if (this == obj) return true;
+        if (!(obj instanceof DataFormat)) return false;
+        return name().equals(((DataFormat) obj).name());
     }
 
     @Override
     public int hashCode() {
-        return 0;
+        return name().hashCode();
     }
 
     @Override

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/engine/ParquetExecutionEngine.java

@@ -14,7 +14,9 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.IndexSettings;
+import com.parquet.parquetdataformat.fields.ArrowFieldRegistry;
 import org.opensearch.index.engine.exec.DataFormat;
+import org.opensearch.index.engine.exec.EngineRole;
 import org.opensearch.index.engine.exec.IndexingExecutionEngine;
 import org.opensearch.index.engine.exec.Merger;
 import org.opensearch.index.engine.exec.RefreshInput;
@@ -75,9 +77,11 @@ public class ParquetExecutionEngine implements IndexingExecutionEngine<ParquetDa
     private final ParquetMerger parquetMerger;
     private final ArrowBufferPool arrowBufferPool;
     private final IndexSettings indexSettings;
+    private final boolean isPrimaryEngine;
 
     public ParquetExecutionEngine(
         Settings settings,
+        boolean isPrimaryEngine,
         Supplier<Schema> schema,
         ShardPath shardPath,
         IndexSettings indexSettings
@@ -87,7 +91,7 @@ public ParquetExecutionEngine(
         this.arrowBufferPool = new ArrowBufferPool(settings);
         this.indexSettings = indexSettings;
         this.parquetMerger = new ParquetMergeExecutor(CompactionStrategy.RECORD_BATCH, indexSettings.getIndex().getName());
-
+        this.isPrimaryEngine = isPrimaryEngine;
         // Push current settings to Rust store once on construction, then keep in sync on updates
         pushSettingsToRust(indexSettings);
 
@@ -143,14 +147,15 @@ public void deleteFiles(Map<String, Collection<String>> filesToDelete) {
     }
 
     @Override
-    public List<String> supportedFieldTypes() {
-        return List.of();
+    public List<String> supportedFieldTypes(boolean isPrimaryEngine) {
+        return new java.util.ArrayList<>(ArrowFieldRegistry.getRegisteredFieldNames());
     }
 
     @Override
     public Writer<ParquetDocumentInput> createWriter(long writerGeneration) {
         String fileName = Path.of(shardPath.getDataPath().toString(), getDataFormat().name(), FILE_NAME_PREFIX + "_" + writerGeneration + FILE_NAME_EXT).toString();
-        return new ParquetWriter(fileName, schema.get(), writerGeneration, arrowBufferPool, indexSettings);
+        EngineRole role = isPrimaryEngine ? EngineRole.PRIMARY : EngineRole.SECONDARY;
+        return new ParquetWriter(fileName, schema.get(), writerGeneration, arrowBufferPool, indexSettings, role);
     }
 
     @Override

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/fields/ArrowFieldRegistry.java

@@ -62,7 +62,7 @@ private static void registerCorePlugins() {
         // Register core data fields
         registerPlugin(new CoreDataFieldPlugin(), "CoreDataFields");
 
-        // REgister metadata fields
+        // Register metadata fields
         registerPlugin(new MetadataFieldPlugin(), "MetadataFields");
     }
     /**
@@ -141,6 +141,13 @@ public static ParquetField getParquetField(String fieldType) {
         return FIELD_REGISTRY.get(fieldType);
     }
 
+    /**
+     * Returns an unmodifiable view of all registered field mappings.
+     */
+    public static Map<String, ParquetField> getRegisteredFields() {
+        return Collections.unmodifiableMap(FIELD_REGISTRY);
+    }
+
     public static class RegistryStats {
         private final int totalFields;
         private final Set<String> allFieldTypes;

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/fields/ArrowSchemaBuilder.java

@@ -46,10 +46,10 @@ private ArrowSchemaBuilder() {
      * @throws IllegalArgumentException if mapperService is null
      * @throws IllegalStateException if no valid fields are found or if a field type is not supported
      */
-    public static Schema getSchema(final MapperService mapperService) {
+    public static Schema getSchema(final MapperService mapperService, boolean isPrimary) {
         Objects.requireNonNull(mapperService, "MapperService cannot be null");
 
-        final List<Field> fields = extractFieldsFromMappers(mapperService);
+        final List<Field> fields = extractFieldsFromMappers(mapperService, isPrimary);
 
         if (fields.isEmpty()) {
             throw new IllegalStateException("No valid fields found in mapper service");
@@ -62,22 +62,26 @@ public static Schema getSchema(final MapperService mapperService) {
      * Extracts Arrow fields from the mapper service, filtering out metadata fields.
      *
      * @param mapperService the mapper service to extract fields from
+     * @param isPrimary whether this is a primary engine context
      * @return a list of Arrow fields
      */
-    private static List<Field> extractFieldsFromMappers(final MapperService mapperService) {
+    private static List<Field> extractFieldsFromMappers(final MapperService mapperService, boolean isPrimary) {
         final List<Field> fields = new ArrayList<>();
 
         for (final Mapper mapper : mapperService.documentMapper().mappers()) {
             if (notSupportedMetadataField(mapper)) {
                 continue;
             }
 
-            final Field arrowField = createArrowField(mapper);
-            fields.add(arrowField);
+            final Field arrowField = createArrowField(mapper, isPrimary);
+            if (arrowField != null) {
+                fields.add(arrowField);
+            }
         }
 
-        fields.add(new Field(CompositeDataFormatWriter.ROW_ID, new LongParquetField().getFieldType(), null));
-        fields.add(new Field(SeqNoFieldMapper.PRIMARY_TERM_NAME, new LongParquetField().getFieldType(), null));
+        LongParquetField longField = new LongParquetField();
+        fields.add(new Field(CompositeDataFormatWriter.ROW_ID, longField.getFieldType(), null));
+        fields.add(new Field(SeqNoFieldMapper.PRIMARY_TERM_NAME, longField.getFieldType(), null));
 
         return fields;
     }
@@ -98,20 +102,27 @@ private static boolean notSupportedMetadataField(final Mapper mapper) {
     }
 
     /**
-     * Creates an Arrow Field from an OpenSearch Mapper.
+     * Creates an Arrow Field from an OpenSearch Mapper using the ArrowFieldRegistry.
+     * For non-primary contexts, returns null if the field type has no eligible ParquetField,
+     * allowing the caller to skip the field. For primary contexts, throws IllegalStateException
+     * if no ParquetField is found.
      *
      * @param mapper the mapper to convert
-     * @return a new Arrow Field
-     * @throws IllegalStateException if the mapper type is not supported
+     * @param isPrimary whether this is a primary engine context
+     * @return a new Arrow Field, or null if the field is not eligible for the role
+     * @throws IllegalStateException if the mapper type is not supported in primary context
      */
-    private static Field createArrowField(final Mapper mapper) {
+    private static Field createArrowField(final Mapper mapper, boolean isPrimary) {
         final ParquetField parquetField = ArrowFieldRegistry.getParquetField(mapper.typeName());
 
         if (parquetField == null) {
-            throw new IllegalStateException(
-                String.format("Unsupported field type '%s' for field '%s'",
-                    mapper.typeName(), mapper.name())
-            );
+            if (isPrimary) {
+                throw new IllegalStateException(
+                    String.format("Unsupported field type '%s' for field '%s'",
+                        mapper.typeName(), mapper.name())
+                );
+            }
+            return null;
         }
 
         return new Field(mapper.name(), parquetField.getFieldType(), null);

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/fields/ParquetField.java

@@ -11,9 +11,11 @@
 import com.parquet.parquetdataformat.vsr.ManagedVSR;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.FieldType;
+import org.opensearch.index.engine.exec.FieldCapability;
 import org.opensearch.index.mapper.MappedFieldType;
 
 import java.util.Objects;
+import java.util.Set;
 
 /**
  * Abstract base class for all Parquet field implementations that handle the conversion
@@ -37,49 +39,30 @@ public abstract class ParquetField {
 
     /**
      * Adds the parsed field value to the appropriate vector group within the managed VSR.
-     * This method is responsible for the actual data conversion and storage in the
-     * columnar format specific to each field type.
      *
-     * <p>Implementations must handle null values appropriately and ensure type safety
-     * when casting the parseValue to the expected type.</p>
-     *
-     * @param mappedFieldType the OpenSearch field type metadata containing field configuration
+     * @param fieldType  the per-field MappedFieldType carrying field name, type name, and capability flags
      * @param managedVSR the managed vector schema root for columnar data storage
      * @param parseValue the parsed field value to be stored, may be null
-     * @throws IllegalArgumentException if any parameter is invalid for this field type
-     * @throws ClassCastException if parseValue cannot be cast to the expected type
      */
-    protected abstract void addToGroup(MappedFieldType mappedFieldType, ManagedVSR managedVSR, Object parseValue);
+    protected abstract void addToGroup(MappedFieldType fieldType, ManagedVSR managedVSR, Object parseValue);
 
     /**
      * Creates and processes a field entry if the field type supports columnar storage.
-     * This method serves as the main entry point for field processing and includes
-     * validation logic to ensure only columnar fields are processed.
-     *
-     * <p>The method performs the following operations:
-     * <ol>
-     *   <li>Validates input parameters</li>
-     *   <li>Checks if the field supports columnar storage</li>
-     *   <li>Delegates to {@link #addToGroup} for actual data processing</li>
-     * </ol>
      *
-     * @param mappedFieldType the OpenSearch field type metadata, must not be null
+     * @param fieldType  the per-field MappedFieldType carrying field name, type name, and capability flags, must not be null
      * @param managedVSR the managed vector schema root, must not be null
      * @param parseValue the parsed field value to be processed, may be null
-     * @throws IllegalArgumentException if mappedFieldType or managedVSR is null
      */
-    public final void createField(final MappedFieldType mappedFieldType,
+    public final void createField(final MappedFieldType fieldType,
                                   final ManagedVSR managedVSR,
                                   final Object parseValue) {
-        Objects.requireNonNull(mappedFieldType, "MappedFieldType cannot be null");
+        Objects.requireNonNull(fieldType, "MappedFieldType cannot be null");
         Objects.requireNonNull(managedVSR, "ManagedVSR cannot be null");
 
-        if (mappedFieldType.isColumnar()) {
-            // TODO: support dynamic mapping update
-            // for now ignore the field
-            if (managedVSR.getVector(mappedFieldType.name()) != null) {
-                addToGroup(mappedFieldType, managedVSR, parseValue);
-            }
+        // TODO: support dynamic mapping update
+        // for now ignore the field
+        if (managedVSR.getVector(fieldType.name()) != null) {
+            addToGroup(fieldType, managedVSR, parseValue);
         }
     }
 
@@ -109,6 +92,12 @@ public final void createField(final MappedFieldType mappedFieldType,
      */
     public abstract FieldType getFieldType();
 
+    /**
+     * Returns the set of capabilities this field supports.
+     * The engine uses this to populate the FieldSupportRegistry.
+     */
+    public abstract Set<FieldCapability> getFieldCapabilities();
+
     /**
      * Provides a string representation of this ParquetField for debugging purposes.
      * The default implementation includes the class name and Arrow type information.

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/fields/core/data/BinaryParquetField.java

@@ -8,12 +8,16 @@
 
 package com.parquet.parquetdataformat.fields.core.data;
 
+import org.opensearch.index.engine.exec.FieldCapability;
+import org.opensearch.index.mapper.MappedFieldType;
 import com.parquet.parquetdataformat.fields.ParquetField;
 import com.parquet.parquetdataformat.vsr.ManagedVSR;
 import org.apache.arrow.vector.VarBinaryVector;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.FieldType;
-import org.opensearch.index.mapper.MappedFieldType;
+
+import java.util.EnumSet;
+import java.util.Set;
 
 /**
  * Parquet field implementation for handling binary data types in OpenSearch documents.
@@ -55,4 +59,9 @@ public ArrowType getArrowType() {
     public FieldType getFieldType() {
         return FieldType.nullable(getArrowType());
     }
+
+    @Override
+    public Set<FieldCapability> getFieldCapabilities() {
+        return EnumSet.of(FieldCapability.DOC_VALUES, FieldCapability.STORE);
+    }
 }

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/fields/core/data/BooleanParquetField.java

@@ -8,13 +8,17 @@
 
 package com.parquet.parquetdataformat.fields.core.data;
 
+import org.opensearch.index.engine.exec.FieldCapability;
+import org.opensearch.index.mapper.MappedFieldType;
 import com.parquet.parquetdataformat.fields.ArrowFieldRegistry;
 import com.parquet.parquetdataformat.fields.ParquetField;
 import com.parquet.parquetdataformat.vsr.ManagedVSR;
 import org.apache.arrow.vector.BitVector;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.FieldType;
-import org.opensearch.index.mapper.MappedFieldType;
+
+import java.util.EnumSet;
+import java.util.Set;
 
 /**
  * Parquet field implementation for handling boolean data types in OpenSearch documents.
@@ -56,4 +60,9 @@ public ArrowType getArrowType() {
     public FieldType getFieldType() {
         return FieldType.nullable(getArrowType());
     }
+
+    @Override
+    public Set<FieldCapability> getFieldCapabilities() {
+        return EnumSet.of(FieldCapability.DOC_VALUES, FieldCapability.STORE);
+    }
 }