feat(parquet): Add adaptive keyword promotion

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetIndexCreationValidator.java

@@ -11,6 +11,8 @@
 import org.apache.arrow.vector.types.pojo.Schema;
 import org.opensearch.index.IndexCreationValidator;
 import org.opensearch.index.IndexSettings;
+import org.opensearch.index.IndexSortConfig;
+import org.opensearch.index.mapper.MappedFieldType;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.parquet.fields.ArrowSchemaBuilder;
 
@@ -43,18 +45,49 @@ public void validate(MapperService mapperService, IndexSettings indexSettings) {
             );
         }
 
-        if (!isParquetIndex || !hasParquetSettings) {
+        if (!isParquetIndex) {
             return;
         }
 
+        validateSortFieldsAreSingleValued(mapperService, indexSettings);
+
+        // Building the schema validates the mapping's `multi_value` declarations: getSchema throws
+        // for a field whose type has no list support, turning what would otherwise be a
+        // per-document indexing failure into an immediate error at creation time.
         Schema schema = ArrowSchemaBuilder.getSchema(mapperService);
-        ParquetSettings.validateFieldConfigurations(
-            fieldEncodings,
-            fieldCompressions,
-            fieldBloomFilterEnabled,
-            lowCardinalityEnabledFields,
-            schema,
-            mapperService
-        );
+        if (hasParquetSettings) {
+            ParquetSettings.validateFieldConfigurations(
+                fieldEncodings,
+                fieldCompressions,
+                fieldBloomFilterEnabled,
+                lowCardinalityEnabledFields,
+                schema,
+                mapperService
+            );
+        }
+    }
+}
+
+    /**
+     * Rejects {@code index.sort.field} entries that are mapped {@code multi_value: true}.
+     * <p>
+     * An index sort needs one total order over rows, but a multi-valued cell has no canonical
+     * scalar to order by — any of min/max/lexicographic would be a silent choice the user never
+     * made. Lucene rejects index sorting on multi-valued fields for the same reason; failing here
+     * keeps parity and turns what would otherwise be a merge-time failure (the native k-way merge
+     * cannot compare LIST sort keys) into an immediate, actionable error at creation time.
+     */
+    private static void validateSortFieldsAreSingleValued(MapperService mapperService, IndexSettings indexSettings) {
+        for (String sortField : IndexSortConfig.INDEX_SORT_FIELD_SETTING.get(indexSettings.getSettings())) {
+            MappedFieldType fieldType = mapperService.fieldType(sortField);
+            if (fieldType != null && fieldType.isMultiValued()) {
+                throw new IllegalArgumentException(
+                    "Cannot use field ["
+                        + sortField
+                        + "] in [index.sort.field]: the field is mapped [multi_value: true] and a "
+                        + "multi-valued field has no single value to sort on"
+                );
+            }
+        }
     }
 }

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java

@@ -834,6 +834,18 @@ private static <T> Map<String, T> extractConfigMap(Settings groupSettings, Strin
         return result;
     }
 
+    /**
+     * Returns the type that encoding and compression settings apply to for a field. For a LIST
+     * column that is the element type, because Parquet encodes the leaf, not the list wrapper —
+     * validating against {@code ArrowType.List} would wrongly accept every encoding.
+     */
+    private static ArrowType elementTypeOf(Field field) {
+        if (field.getType() instanceof ArrowType.List && field.getChildren().isEmpty() == false) {
+            return field.getChildren().get(0).getType();
+        }
+        return field.getType();
+    }
+
     /**
      * Validates that field-level configurations are compatible with their Arrow types in the schema.
      * Only keyword and text field types are accepted for low_cardinality_enable.
@@ -848,7 +860,7 @@ public static void validateFieldConfigurations(
     ) {
         Map<String, ArrowType> arrowTypes = new HashMap<>();
         for (Field field : schema.getFields()) {
-            arrowTypes.put(field.getName(), field.getType());
+            arrowTypes.put(field.getName(), elementTypeOf(field));
         }
 
         // Validate encoding configurations

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/engine/ParquetIndexingEngine.java

@@ -257,7 +257,7 @@ public Writer<ParquetDocumentInput> createWriter(WriterConfig config) {
         return new ParquetWriter(
             filePath.toString(),
             config.writerGeneration(),
-            0L,
+            mappingVersion,
             dataFormat,
             schema,
             this::getOrBuildSchema,

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ArrowSchemaBuilder.java

@@ -14,9 +14,11 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.index.engine.dataformat.DocumentInput;
 import org.opensearch.index.mapper.DocumentMapper;
+import org.opensearch.index.mapper.FieldMapper;
 import org.opensearch.index.mapper.FieldNamesFieldMapper;
 import org.opensearch.index.mapper.IndexFieldMapper;
 import org.opensearch.index.mapper.KeywordFieldMapper;
+import org.opensearch.index.mapper.MappedFieldType;
 import org.opensearch.index.mapper.Mapper;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.mapper.NestedPathFieldMapper;
@@ -39,8 +41,13 @@ private ArrowSchemaBuilder() {}
 
     /**
      * Creates an Arrow Schema from the MapperService.
-     * @param mapperService the mapper service containing field mappings
+     *
+     * <p>A field whose mapper declares {@code multi_value: true}
+     * ({@link MappedFieldType#isMultiValued()}) is emitted as a {@code LIST<element>} column;
+     * every other field keeps its scalar column.
      * TODO - Get the mapping version while creating the schema
+     *
+     * @param mapperService the mapper service containing field mappings
      */
     public static Schema getSchema(MapperService mapperService) {
         Objects.requireNonNull(mapperService, "MapperService cannot be null");
@@ -55,8 +62,18 @@ public static Schema getSchema(MapperService mapperService) {
 
                 ParquetField parquetField = ArrowFieldRegistry.getParquetField(mapper.typeName());
                 if (parquetField != null) {
-                    fields.add(new Field(mapper.name(), parquetField.getFieldType(), null));
-                    handleNormalizedField(mapper, documentMapper, fields, parquetField);
+                    boolean multiValue = isMultiValued(mapper);
+                    if (multiValue && parquetField.supportsMultiValue() == false) {
+                        throw new IllegalArgumentException(
+                            "Field ["
+                                + mapper.name()
+                                + "] of type ["
+                                + mapper.typeName()
+                                + "] does not support [multi_value] storage in the parquet data format"
+                        );
+                    }
+                    fields.add(parquetField.toArrowField(mapper.name(), multiValue));
+                    handleNormalizedField(mapper, documentMapper, fields, parquetField, multiValue);
                 } else {
                     logger.debug("No ParquetField registered for field: [{}] of type [{}]", mapper.name(), mapper.typeName());
                 }
@@ -69,15 +86,28 @@ public static Schema getSchema(MapperService mapperService) {
         return new Schema(fields);
     }
 
-    private static void handleNormalizedField(Mapper mapper, DocumentMapper documentMapper, List<Field> fields, ParquetField parquetField) {
+    private static void handleNormalizedField(
+        Mapper mapper,
+        DocumentMapper documentMapper,
+        List<Field> fields,
+        ParquetField parquetField,
+        boolean multiValue
+    ) {
         if (mapper instanceof KeywordFieldMapper keywordFieldMapper) {
             if (!documentMapper.mappers().isMultiField(mapper.name()) && keywordFieldMapper.getRawValueFieldType() != null) {
                 KeywordFieldMapper.KeywordFieldType rawValueField = keywordFieldMapper.getRawValueFieldType();
-                fields.add(new Field(rawValueField.name(), parquetField.getFieldType(), null));
+                // The raw-value companion holds the pre-normalization source for derived source, so
+                // it must mirror the parent's cardinality or source reconstruction would lose values.
+                fields.add(parquetField.toArrowField(rawValueField.name(), multiValue));
             }
         }
     }
 
+    /** Reads the {@code multi_value} declaration from the mapper's field type. */
+    private static boolean isMultiValued(Mapper mapper) {
+        return mapper instanceof FieldMapper fieldMapper && fieldMapper.fieldType().isMultiValued();
+    }
+
     private static boolean isUnsupportedMetadataField(Mapper mapper) {
         return mapper instanceof SourceFieldMapper
             || mapper instanceof FieldNamesFieldMapper

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ParquetField.java

@@ -8,12 +8,16 @@
 
 package org.opensearch.parquet.fields;
 
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.ListVector;
 import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.FieldType;
 import org.opensearch.index.engine.dataformat.FieldTypeCapabilities;
 import org.opensearch.index.mapper.MappedFieldType;
 import org.opensearch.parquet.vsr.ManagedVSR;
 
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -22,6 +26,12 @@
  */
 public abstract class ParquetField {
 
+    /**
+     * Name of the child field inside a LIST column. Matches the parquet-rs convention
+     * ({@code PARQUET_LIST_ELEMENT_NAME}) so the leaf path is {@code <field>.list.element}.
+     */
+    public static final String LIST_ELEMENT_NAME = "element";
+
     /** Creates a new ParquetField. */
     public ParquetField() {}
 
@@ -33,6 +43,67 @@ public ParquetField() {}
      */
     protected abstract void addToGroup(MappedFieldType fieldType, ManagedVSR managedVSR, Object parseValue);
 
+    /**
+     * Writes a single parsed value at an explicit index in the given vector.
+     * <p>
+     * Scalar columns write at the row index, so {@link #addToGroup} can derive the position from
+     * the VSR's row count. List columns write several values per row at positions in the child
+     * vector that have nothing to do with the row number, so multi-value writes need this
+     * index-explicit form instead.
+     * <p>
+     * Subclasses must override this to support being declared multi-valued; the default throws.
+     * When overridden, {@link #addToGroup} should delegate to it so the scalar and list paths
+     * share one value-coercion implementation.
+     *
+     * @param vector the target vector (the child data vector when writing into a list)
+     * @param index the position to write at
+     * @param parseValue the parsed value to write
+     */
+    protected void addToVector(FieldVector vector, int index, Object parseValue) {
+        throw new UnsupportedOperationException(
+            "Field type [" + getClass().getSimpleName() + "] does not support multi-valued (list) storage"
+        );
+    }
+
+    /**
+     * Returns whether this field can be stored as a Parquet LIST column, i.e. whether it
+     * implements {@link #addToVector}.
+     *
+     * @return true if multi-valued storage is supported
+     */
+    public boolean supportsMultiValue() {
+        return false;
+    }
+
+    /**
+     * Builds the Arrow field describing this column, including any child fields.
+     * <p>
+     * When {@code multiValue} is true the result is a {@code LIST<element>} whose child carries
+     * this field's element type, so the same {@link ParquetField} describes both shapes.
+     *
+     * @param name the Arrow field name
+     * @param multiValue whether to wrap the element type in a list
+     * @return the Arrow field
+     */
+    public final Field toArrowField(String name, boolean multiValue) {
+        if (multiValue == false) {
+            return new Field(name, getFieldType(), null);
+        }
+        if (supportsMultiValue() == false) {
+            throw new IllegalArgumentException(
+                "Field ["
+                    + name
+                    + "] cannot be stored as multi-valued: type ["
+                    + getClass().getSimpleName()
+                    + "] does not support list storage"
+            );
+        }
+        // The element is always nullable: a null inside an array (e.g. ["a", null]) is a legal
+        // document even when the column itself is declared non-nullable.
+        Field element = new Field(LIST_ELEMENT_NAME, FieldType.nullable(getArrowType()), null);
+        return new Field(name, FieldType.nullable(ArrowType.List.INSTANCE), List.of(element));
+    }
+
     /**
      * Creates and processes a field entry. Throws if vector not present in VSR.
      * @param fieldType the mapped field type
@@ -42,9 +113,41 @@ public ParquetField() {}
     public final void createField(MappedFieldType fieldType, ManagedVSR managedVSR, Object parseValue) {
         assert fieldType != null : "MappedFieldType cannot be null";
         assert managedVSR != null : "ManagedVSR cannot be null";
+        FieldVector vector = managedVSR.getVector(fieldType.name());
+        if (vector instanceof ListVector listVector) {
+            writeList(fieldType, managedVSR, listVector, parseValue);
+            return;
+        }
         addToGroup(fieldType, managedVSR, parseValue);
     }
 
+    /**
+     * Writes all values collected for one document into a list column at the current row.
+     * <p>
+     * A null {@code parseValue} is written as a null list, which is how an absent field is
+     * represented. An empty list is written as a zero-length, non-null list, preserving the
+     * distinction between {@code "tags": []} and no {@code tags} at all.
+     */
+    private void writeList(MappedFieldType fieldType, ManagedVSR managedVSR, ListVector listVector, Object parseValue) {
+        int row = managedVSR.getRowCount();
+        if (parseValue == null) {
+            listVector.setNull(row);
+            return;
+        }
+        List<?> values = parseValue instanceof List<?> list ? list : List.of(parseValue);
+        int start = listVector.startNewValue(row);
+        FieldVector dataVector = listVector.getDataVector();
+        for (int i = 0; i < values.size(); i++) {
+            Object value = values.get(i);
+            if (value == null) {
+                dataVector.setNull(start + i);
+            } else {
+                addToVector(dataVector, start + i, value);
+            }
+        }
+        listVector.endValue(row, values.size());
+    }
+
     /**
      * Returns the set of capabilities supported by this field type.
      * Subclasses may override to declare different capabilities.

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/core/data/text/KeywordParquetField.java

@@ -8,6 +8,7 @@
 
 package org.opensearch.parquet.fields.core.data.text;
 
+import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VarCharVector;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.FieldType;
@@ -27,10 +28,17 @@ public KeywordParquetField() {}
 
     @Override
     protected void addToGroup(MappedFieldType mappedFieldType, ManagedVSR managedVSR, Object parseValue) {
-        ((VarCharVector) managedVSR.getVector(mappedFieldType.name())).setSafe(
-            managedVSR.getRowCount(),
-            parseValue.toString().getBytes(StandardCharsets.UTF_8)
-        );
+        addToVector(managedVSR.getVector(mappedFieldType.name()), managedVSR.getRowCount(), parseValue);
+    }
+
+    @Override
+    protected void addToVector(FieldVector vector, int index, Object parseValue) {
+        ((VarCharVector) vector).setSafe(index, parseValue.toString().getBytes(StandardCharsets.UTF_8));
+    }
+
+    @Override
+    public boolean supportsMultiValue() {
+        return true;
     }
 
     @Override

sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/SchemaChangeRequiresWriterRotationException.java

@@ -0,0 +1,20 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file to be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.parquet.vsr;
+
+/**
+ * Signals that an existing Arrow vector changed type and the current Parquet file must be
+ * finalized before indexing can continue with the new schema.
+ */
+public class SchemaChangeRequiresWriterRotationException extends RuntimeException {
+
+    public SchemaChangeRequiresWriterRotationException(String fieldName, Object previousType, Object nextType) {
+        super("Field [" + fieldName + "] changed Parquet type from [" + previousType + "] to [" + nextType + "]");
+    }
+}