diff --git a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeDynamicMappingIT.java b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeDynamicMappingIT.java index eaea52d0e898e..46221130ab987 100644 --- a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeDynamicMappingIT.java +++ b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeDynamicMappingIT.java @@ -235,6 +235,129 @@ public void testConcurrentDynamicUpdatesWithLuceneSecondary() throws Throwable { ensureNoActiveMerges(indexName); } + /** + * Verifies the complete AUTO promotion path: the first scalar row is committed with a scalar + * schema, a later array publishes {@code multi_value: true} into IndexMetadata, and the same + * indexing request is retried and persisted as a LIST row. + */ + public void testAdaptiveKeywordPromotionUpdatesClusterStateAndRetriesDocument() throws Exception { + String indexName = "test-adaptive-keyword"; + CreateIndexResponse createResponse = client().admin() + .indices() + .prepareCreate(indexName) + .setSettings(parquetPrimaryLuceneSecondarySettings()) + .setMapping("tags", "type=keyword") + .get(); + assertTrue(createResponse.isAcknowledged()); + ensureGreen(indexName); + + Map initialFieldMapping = clusterStateFieldMapping(indexName, "tags"); + assertFalse(initialFieldMapping.containsKey("multi_value")); + long initialMappingVersion = getClusterState().metadata().index(indexName).getMappingVersion(); + + IndexResponse scalarResponse = client().prepareIndex(indexName).setSource("tags", "solo").get(); + assertEquals(RestStatus.CREATED, scalarResponse.status()); + refreshAndFlush(indexName); + + IndexResponse listResponse = client().prepareIndex(indexName).setSource("tags", List.of("prod", "error", "prod")).get(); + assertEquals(RestStatus.CREATED, listResponse.status()); + + assertBusy(() -> { + assertEquals(Boolean.TRUE, clusterStateFieldMapping(indexName, "tags").get("multi_value")); + assertTrue(getClusterState().metadata().index(indexName).getMappingVersion() > initialMappingVersion); + }); + + List> rows = refreshFlushAndReadParquetRows(indexName); + assertEquals(2, rows.size()); + assertTrue(rows.stream().anyMatch(row -> "solo".equals(row.get("tags")))); + assertTrue(rows.stream().anyMatch(row -> isListColumnPlaceholder(row.get("tags")))); + } + + /** Verifies that explicit SCALAR state rejects an array without publishing a mapping update. */ + public void testExplicitScalarKeywordRejectsArrayWithoutUpdatingClusterState() throws Exception { + String indexName = "test-scalar-keyword"; + CreateIndexResponse createResponse = client().admin() + .indices() + .prepareCreate(indexName) + .setSettings(parquetPrimaryLuceneSecondarySettings()) + .setMapping("tags", "type=keyword,multi_value=false") + .get(); + assertTrue(createResponse.isAcknowledged()); + ensureGreen(indexName); + + assertEquals(Boolean.FALSE, clusterStateFieldMapping(indexName, "tags").get("multi_value")); + long mappingVersion = getClusterState().metadata().index(indexName).getMappingVersion(); + + IndexResponse scalarResponse = client().prepareIndex(indexName).setSource("tags", "solo").get(); + assertEquals(RestStatus.CREATED, scalarResponse.status()); + + Exception error = expectThrows( + Exception.class, + () -> client().prepareIndex(indexName).setSource("tags", List.of("one", "two")).get() + ); + assertThat( + org.opensearch.ExceptionsHelper.stackTrace(error), + org.hamcrest.Matchers.containsString("locked scalar by [multi_value: false]") + ); + assertEquals(mappingVersion, getClusterState().metadata().index(indexName).getMappingVersion()); + assertEquals(Boolean.FALSE, clusterStateFieldMapping(indexName, "tags").get("multi_value")); + + List> rows = refreshFlushAndReadParquetRows(indexName); + assertEquals(1, rows.size()); + assertEquals("solo", rows.get(0).get("tags")); + } + + /** Verifies that explicit LIST state writes scalar input as a singleton list and arrays unchanged. */ + public void testExplicitListKeywordPersistsListShapeForEveryDocument() throws Exception { + String indexName = "test-list-keyword"; + CreateIndexResponse createResponse = client().admin() + .indices() + .prepareCreate(indexName) + .setSettings(parquetPrimaryLuceneSecondarySettings()) + .setMapping("tags", "type=keyword,multi_value=true") + .get(); + assertTrue(createResponse.isAcknowledged()); + ensureGreen(indexName); + + assertEquals(Boolean.TRUE, clusterStateFieldMapping(indexName, "tags").get("multi_value")); + assertEquals(RestStatus.CREATED, client().prepareIndex(indexName).setSource("tags", "solo").get().status()); + assertEquals(RestStatus.CREATED, client().prepareIndex(indexName).setSource("tags", List.of("one", "two", "one")).get().status()); + + List> rows = refreshFlushAndReadParquetRows(indexName); + assertEquals(2, rows.size()); + assertTrue(rows.stream().allMatch(row -> isListColumnPlaceholder(row.get("tags")))); + } + + /** + * RustBridge's test-only JSON renderer decodes primitive columns and emits this marker for + * nested columns. Matching it verifies that the physical Parquet column is LIST; element-value + * preservation is covered by the lower-level VSR and ParquetDocumentInput tests. + */ + private boolean isListColumnPlaceholder(Object value) { + return value instanceof String text && text.startsWith(" clusterStateFieldMapping(String indexName, String fieldName) { + Map mappingSource = getClusterState().metadata().index(indexName).mapping().sourceAsMap(); + Map properties = (Map) mappingSource.get("properties"); + return (Map) properties.get(fieldName); + } + + private void refreshAndFlush(String indexName) { + client().admin().indices().prepareRefresh(indexName).get(); + client().admin().indices().prepareFlush(indexName).setForce(true).setWaitIfOngoing(true).get(); + } + + private List> refreshFlushAndReadParquetRows(String indexName) throws IOException { + refreshAndFlush(indexName); + IndexShard shard = getIndexShard(indexName); + Path parquetDir = shard.shardPath().getDataPath().resolve("parquet"); + try (GatedCloseable> parquetFilesRef = listParquetFiles(parquetDir, shard)) { + return readAllParquetRows(parquetFilesRef.get()); + } + } + // ══════════════════════════════════════════════════════════════════════ // Private helpers: index settings // ══════════════════════════════════════════════════════════════════════ diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java index dd8bf0800b53a..d0e3f186f2ef0 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetSettings.java @@ -834,6 +834,18 @@ private static Map 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 arrowTypes = new HashMap<>(); for (Field field : schema.getFields()) { - arrowTypes.put(field.getName(), field.getType()); + arrowTypes.put(field.getName(), elementTypeOf(field)); } // Validate encoding configurations diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/engine/ParquetIndexingEngine.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/engine/ParquetIndexingEngine.java index 612f751dce25a..d4497899ca73d 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/engine/ParquetIndexingEngine.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/engine/ParquetIndexingEngine.java @@ -257,7 +257,7 @@ public Writer createWriter(WriterConfig config) { return new ParquetWriter( filePath.toString(), config.writerGeneration(), - 0L, + mappingVersion, dataFormat, schema, this::getOrBuildSchema, diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ArrowSchemaBuilder.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ArrowSchemaBuilder.java index f9718af73f820..ace8e6d671e21 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ArrowSchemaBuilder.java +++ b/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 + * + *

A field whose mapper declares {@code multi_value: true} + * ({@link MappedFieldType#isMultiValued()}) is emitted as a {@code LIST} 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 fields, ParquetField parquetField) { + private static void handleNormalizedField( + Mapper mapper, + DocumentMapper documentMapper, + List 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 diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ParquetField.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ParquetField.java index 099c303c8282f..d2b9a09add377 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ParquetField.java +++ b/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 .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. + *

+ * 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. + *

+ * 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. + *

+ * When {@code multiValue} is true the result is a {@code LIST} 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. + *

+ * 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. diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/core/data/text/KeywordParquetField.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/core/data/text/KeywordParquetField.java index eff1e5ffdaa8e..6fedf90ab360c 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/core/data/text/KeywordParquetField.java +++ b/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 diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/SchemaChangeRequiresWriterRotationException.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/SchemaChangeRequiresWriterRotationException.java new file mode 100644 index 0000000000000..a783ac01faeef --- /dev/null +++ b/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 + "]"); + } +} diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java index 6e5e2968d0117..2d036788475f0 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java @@ -192,9 +192,9 @@ public VSRManager( * Transfers collected fields from the document input into the active VSR * using the ArrowFieldRegistry to resolve typed vector writes. *

- * Single-value semantics are enforced at the {@link ParquetDocumentInput} layer: - * if an array field produces multiple values for the same field type, only the - * last value is retained (last-value-wins). + * Field Shape is decided at the {@link ParquetDocumentInput} layer: fields mapped with + * {@code multi_value: true} accumulate every value into one list-valued pair, + * and all other fields still reject a second value. * * @param doc the document input containing field-value pairs */ @@ -366,10 +366,18 @@ public boolean reconcileSchema(Schema newSchema) { ManagedVSR activeVSR = managedVSR.get(); boolean changed = false; for (Field schemaField : newSchema.getFields()) { - if (activeVSR.getVector(schemaField.getName()) == null) { - Field field = new Field(schemaField.getName(), schemaField.getFieldType(), null); - activeVSR.addFieldVector(field); + FieldVector existingVector = activeVSR.getVector(schemaField.getName()); + if (existingVector == null) { + // Pass the schema field through as-is: rebuilding it from name + FieldType alone + // would drop getChildren(), leaving a LIST column with no element vector. + activeVSR.addFieldVector(schemaField); changed = true; + } else if (existingVector.getField().equals(schemaField) == false) { + throw new SchemaChangeRequiresWriterRotationException( + schemaField.getName(), + existingVector.getField().getType(), + schemaField.getType() + ); } } if (changed) { diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/FieldValuePair.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/FieldValuePair.java index 1444b89b70e10..8e06c9d08751b 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/FieldValuePair.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/FieldValuePair.java @@ -10,24 +10,40 @@ import org.opensearch.index.mapper.MappedFieldType; +import java.util.ArrayList; +import java.util.List; + /** - * Immutable pair of an OpenSearch {@link MappedFieldType} and its parsed value. + * Pair of an OpenSearch {@link MappedFieldType} and the value(s) parsed for it. * *

Represents a single field entry collected by {@link ParquetDocumentInput} during * document indexing. The field type is used to resolve the corresponding Arrow vector * type via {@link org.opensearch.parquet.fields.ArrowFieldRegistry}, and the value is * written into that vector during document transfer to the VSR. * - *

The field type must not be null (enforced by constructor); the value may be null + *

Scalar pairs are immutable and hold exactly one value. Pairs created via + * {@link #multiValued} back a Parquet LIST column and are mutable: {@link #addValue} + * appends as the document parser reports each array element, so {@link #getValue()} returns a + * {@code List} for those — including a single-element list when the document supplied one value, or + * an empty list for an explicit empty array via {@link #emptyMultiValued}. + * + *

Because multi-valued pairs grow during parsing, a reference must not be read until the + * document is finalized. The sole consumer, {@code VSRManager#addDocument}, reads only through + * {@link ParquetDocumentInput#getFinalInput()} after the whole document has been parsed, so no + * caller observes a partially populated list. Do not cache or hand a pair across threads before + * then. + * + *

The field type must not be null (enforced by constructor); values may be null * for nullable fields. */ public class FieldValuePair { private final MappedFieldType fieldType; - private final Object value; + private Object value; + private List values; /** - * Creates a new FieldValuePair. + * Creates a single-valued FieldValuePair. * * @param fieldType the mapped field type * @param value the parsed field value @@ -38,6 +54,80 @@ public FieldValuePair(MappedFieldType fieldType, Object value) { } this.fieldType = fieldType; this.value = value; + this.values = null; + } + + private FieldValuePair(MappedFieldType fieldType, List values) { + if (fieldType == null) { + throw new IllegalArgumentException("fieldType cannot be null"); + } + this.fieldType = fieldType; + this.value = null; + this.values = values; + } + + /** + * Creates a multi-valued FieldValuePair seeded with its first value. Further values are + * appended via {@link #addValue}, preserving document order and any duplicates. + * + * @param fieldType the mapped field type + * @param firstValue the first parsed value + * @return a multi-valued pair + */ + public static FieldValuePair multiValued(MappedFieldType fieldType, Object firstValue) { + List values = new ArrayList<>(1); + values.add(firstValue); + return new FieldValuePair(fieldType, values); + } + + /** + * Creates a multi-valued FieldValuePair holding zero values, representing an explicit empty + * array ({@code "field": []}). It backs a zero-length, non-null LIST cell, which reads back as + * {@code []} and so stays distinct from an absent field (a null cell). + * + * @param fieldType the mapped field type + * @return an empty multi-valued pair + */ + public static FieldValuePair emptyMultiValued(MappedFieldType fieldType) { + return new FieldValuePair(fieldType, new ArrayList<>(0)); + } + + /** + * Appends another value. Only valid on a multi-valued pair. + * + * @param nextValue the value to append + */ + public void addValue(Object nextValue) { + if (values == null) { + throw new IllegalStateException("Cannot add a value to a single-valued FieldValuePair for [" + fieldType.name() + "]"); + } + values.add(nextValue); + } + + /** + * Converts a scalar pair into a multi-valued pair and appends the triggering value. + * The first parse attempt is discarded while its dynamic mapping update is published; + * the retried document is parsed against the LIST mapping. + */ + public void promoteToMultiValued(Object nextValue) { + if (values != null) { + values.add(nextValue); + return; + } + values = new ArrayList<>(2); + values.add(value); + values.add(nextValue); + value = null; + } + + /** Returns whether this pair accumulates multiple values into a list column. */ + public boolean isMultiValued() { + return values != null; + } + + /** Returns the number of values held: always 1 for a scalar pair. */ + public int valueCount() { + return values == null ? 1 : values.size(); } /** @@ -50,11 +140,12 @@ public MappedFieldType getFieldType() { } /** - * Returns the value. + * Returns the value: the single parsed value, or the {@code List} of values for a + * multi-valued pair. * - * @return the parsed field value + * @return the parsed field value(s) */ public Object getValue() { - return value; + return values != null ? values : value; } } diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetDocumentInput.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetDocumentInput.java index 3429c030266fc..2c2c2ca8139aa 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetDocumentInput.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetDocumentInput.java @@ -21,9 +21,9 @@ import org.opensearch.parquet.ParquetDataFormatPlugin; import java.util.ArrayList; -import java.util.Collections; -import java.util.IdentityHashMap; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Set; /** @@ -40,7 +40,12 @@ public class ParquetDocumentInput implements DocumentInput> private static final Logger logger = LogManager.getLogger(ParquetDocumentInput.class); private final List collectedFields = new ArrayList<>(); - private final Set dedup = Collections.newSetFromMap(new IdentityHashMap<>()); + // Keyed by field name, not field-type identity: within a single document parse each logical + // field (including the derived-source `_ignored_source.*` companion) has a unique name, while + // identity would silently miss a match if the parser ever handed back a fresh wrapper per array + // element — degrading a multi_value field to last-value-wins or bypassing the scalar duplicate + // guard. Name keying makes accumulation robust to that. + private final Map seen = new HashMap<>(); private long rowId = -1; private boolean isClosed = false; @@ -54,12 +59,37 @@ public void addField(MappedFieldType fieldType, Object value) { logger.trace("Ignored to add field: {} {}", fieldType.name(), fieldType.getCapabilityMap()); return; } - if (dedup.add(fieldType) == false) { + FieldValuePair existing = seen.get(fieldType.name()); + if (existing == null) { + // Fields declared `multi_value: true` in the mapping start out as a list of one so the + // value shape reaching the VSR is the same whether the document had one value or several. + // An explicit empty array (`"field": []`) is signalled by an empty List and seeds a + // zero-value pair, so its LIST cell is written empty-but-non-null rather than null. + final FieldValuePair pair; + if (fieldType.isMultiValued()) { + pair = value instanceof List list && list.isEmpty() + ? FieldValuePair.emptyMultiValued(fieldType) + : FieldValuePair.multiValued(fieldType, value); + } else { + pair = new FieldValuePair(fieldType, value); + } + seen.put(fieldType.name(), pair); + collectedFields.add(pair); + return; + } + if (existing.isMultiValued() == false) { + if (fieldType.isMultiValueSupported() && fieldType.isMultiValueAutoPromotionEnabled()) { + existing.promoteToMultiValued(value); + return; + } + String reason = fieldType.isMultiValueSupported() + ? "the field is locked scalar by [multi_value: false]" + : "the field type does not support automatic multi-value promotion"; throw new MapperParsingException( - "Cannot accept multiple values for field: [" + fieldType.name() + "] of type: [" + fieldType.typeName() + "]." + "Cannot accept multiple values for field: [" + fieldType.name() + "] of type: [" + fieldType.typeName() + "]: " + reason ); } - collectedFields.add(new FieldValuePair(fieldType, value)); + existing.addValue(value); } @Override @@ -84,13 +114,19 @@ public List getFinalInput() { @Override public long getFieldCount(String fieldName) { - return collectedFields.stream().filter(fvp -> fvp.getFieldType().name().equals(fieldName)).count(); + // Counts values, not entries: a multi-valued field is one entry holding N values, and + // callers (single-value assertions below, the data-stream @timestamp check) mean values. + return collectedFields.stream() + .filter(fvp -> fvp.getFieldType().name().equals(fieldName)) + .mapToLong(FieldValuePair::valueCount) + .sum(); } @Override public void close() { isClosed = true; collectedFields.clear(); + seen.clear(); rowId = -1; } diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetWriter.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetWriter.java index 4f74e4f07374d..542b7c875ca59 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetWriter.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetWriter.java @@ -27,6 +27,7 @@ import org.opensearch.parquet.engine.ParquetDataFormat; import org.opensearch.parquet.memory.ArrowBufferPool; import org.opensearch.parquet.stats.ParquetShardStatsTracker; +import org.opensearch.parquet.vsr.SchemaChangeRequiresWriterRotationException; import org.opensearch.parquet.vsr.VSRManager; import org.opensearch.plugin.stats.StatsRecorder; import org.opensearch.threadpool.ThreadPool; @@ -244,8 +245,18 @@ public void updateMappingVersion(long newVersion) { schema.getFields().size(), schema.getFields().stream().map(f -> f.getName()).collect(java.util.stream.Collectors.joining(", ")) ); - boolean updated = vsrManager.reconcileSchema(schema); - logger.debug("updateMappingVersion: reconcileSchema returned updated={}", updated); + try { + boolean updated = vsrManager.reconcileSchema(schema); + logger.debug("updateMappingVersion: reconcileSchema returned updated={}", updated); + } catch (SchemaChangeRequiresWriterRotationException e) { + state = WriterState.RETIRED_FLUSHABLE; + logger.debug( + "[Gen: {}] mapping version {} requires a new Parquet writer: {}", + writerGeneration, + newVersion, + e.getMessage() + ); + } } else { logger.trace( "[Gen: {}] updateMappingVersion: no-op, newVersion={} <= current mappingVersion={}", diff --git a/sandbox/plugins/parquet-data-format/src/main/rust/src/writer_properties_builder.rs b/sandbox/plugins/parquet-data-format/src/main/rust/src/writer_properties_builder.rs index 1e8b29de8463c..e5d34eaa73104 100644 --- a/sandbox/plugins/parquet-data-format/src/main/rust/src/writer_properties_builder.rs +++ b/sandbox/plugins/parquet-data-format/src/main/rust/src/writer_properties_builder.rs @@ -66,6 +66,34 @@ pub fn read_format_version(metadata: &FileMetaData) -> String { /// - **Dependency Inversion**: Depends on NativeSettings abstraction pub struct WriterPropertiesBuilder; +/// Returns the type that column properties apply to. Parquet encodes a LIST column's leaf, not +/// the list wrapper, so encoding/compression decisions must be made on the element type. +fn effective_type(dt: &arrow::datatypes::DataType) -> &arrow::datatypes::DataType { + use arrow::datatypes::DataType::*; + match dt { + List(f) | LargeList(f) | FixedSizeList(f, _) => effective_type(f.data_type()), + other => other, + } +} + +/// Builds the parquet `ColumnPath` for a top-level Arrow field. +/// +/// `ColumnPath::from(String)` yields a single-segment path, which only matches primitive columns. +/// A LIST column's leaf lives at `.list.element`, so per-column encoding, compression, and +/// bloom-filter settings silently no-op for lists unless the full path is spelled out. The +/// `list`/`element` segment names match what the arrow-rs writer emits for `DataType::List`. +fn column_path_for(field: &arrow::datatypes::Field) -> parquet::schema::types::ColumnPath { + use arrow::datatypes::DataType::*; + let mut parts = vec![field.name().clone()]; + let mut current = field.data_type(); + while let List(child) | LargeList(child) | FixedSizeList(child, _) = current { + parts.push("list".to_string()); + parts.push(child.name().clone()); + current = child.data_type(); + } + parquet::schema::types::ColumnPath::new(parts) +} + /// Maps an Arrow DataType to a lowercase string key used for cluster-level type config lookups. fn arrow_type_key(dt: &arrow::datatypes::DataType) -> String { use arrow::datatypes::DataType::*; @@ -196,17 +224,28 @@ impl WriterPropertiesBuilder { config: &NativeSettings, schema: &ArrowSchema, ) -> Result { - let type_map: std::collections::HashMap<&str, (&arrow::datatypes::DataType, String)> = - schema - .fields() - .iter() - .map(|f| { + // Key config decisions off the element type and address the column by its full leaf path, + // so a LIST column is configured like the scalar column of its element type. + type FieldEntry<'a> = ( + &'a arrow::datatypes::DataType, + String, + parquet::schema::types::ColumnPath, + ); + let type_map: std::collections::HashMap<&str, FieldEntry<'_>> = schema + .fields() + .iter() + .map(|f| { + let element_type = effective_type(f.data_type()); + ( + f.name().as_str(), ( - f.name().as_str(), - (f.data_type(), arrow_type_key(f.data_type())), - ) - }) - .collect(); + element_type, + arrow_type_key(element_type), + column_path_for(f), + ), + ) + }) + .collect(); let mut field_names: std::collections::HashSet<&str> = std::collections::HashSet::new(); if let Some(fc) = &config.field_configs { @@ -224,8 +263,8 @@ impl WriterPropertiesBuilder { .as_ref() .and_then(|m| m.get(field_name)); - let (arrow_type, type_key) = match type_map.get(field_name) { - Some(v) => (v.0, Some(v.1.as_str())), + let (arrow_type, type_key, column_path) = match type_map.get(field_name) { + Some(v) => (v.0, Some(v.1.as_str()), v.2.clone()), None => { return Err(format!( "Field '{}' in field_configs does not exist in schema", @@ -283,18 +322,15 @@ impl WriterPropertiesBuilder { | Encoding::BYTE_STREAM_SPLIT | Encoding::RLE ) { - builder = - builder.set_column_dictionary_enabled(field_name.to_string().into(), false); - builder = builder.set_column_encoding(field_name.to_string().into(), enc); + builder = builder.set_column_dictionary_enabled(column_path.clone(), false); + builder = builder.set_column_encoding(column_path.clone(), enc); } else if matches!(enc, Encoding::RLE_DICTIONARY) { // RLE_DICTIONARY means use dictionary encoding - just ensure it's enabled - builder = - builder.set_column_dictionary_enabled(field_name.to_string().into(), true); + builder = builder.set_column_dictionary_enabled(column_path.clone(), true); } else { // PLAIN: explicitly disable dictionary so the writer uses plain encoding - builder = - builder.set_column_dictionary_enabled(field_name.to_string().into(), false); - builder = builder.set_column_encoding(field_name.to_string().into(), enc); + builder = builder.set_column_dictionary_enabled(column_path.clone(), false); + builder = builder.set_column_encoding(column_path.clone(), enc); } } @@ -326,7 +362,7 @@ impl WriterPropertiesBuilder { }; if let Some(comp) = compression { - builder = builder.set_column_compression(field_name.to_string().into(), comp); + builder = builder.set_column_compression(column_path.clone(), comp); } // Bloom filter: field-level > type-level > global. Applied per-column. @@ -338,8 +374,7 @@ impl WriterPropertiesBuilder { }) .unwrap_or(config.get_bloom_filter_enabled()); if bf_enabled { - builder = - builder.set_column_bloom_filter_enabled(field_name.to_string().into(), true); + builder = builder.set_column_bloom_filter_enabled(column_path.clone(), true); let bf_fpp = index_cfg .and_then(|fc| fc.bloom_filter_fpp) .or_else(|| { @@ -347,8 +382,7 @@ impl WriterPropertiesBuilder { .and_then(|t| config.type_bloom_filter_fpp.as_ref()?.get(t).copied()) }) .unwrap_or(config.get_bloom_filter_fpp()); - builder = - builder.set_column_bloom_filter_fpp(field_name.to_string().into(), bf_fpp); + builder = builder.set_column_bloom_filter_fpp(column_path.clone(), bf_fpp); let bf_ndv = index_cfg .and_then(|fc| fc.bloom_filter_ndv) .or_else(|| { @@ -356,8 +390,7 @@ impl WriterPropertiesBuilder { .and_then(|t| config.type_bloom_filter_ndv.as_ref()?.get(t).copied()) }) .unwrap_or(config.get_bloom_filter_ndv()); - builder = - builder.set_column_bloom_filter_ndv(field_name.to_string().into(), bf_ndv); + builder = builder.set_column_bloom_filter_ndv(column_path.clone(), bf_ndv); } } Ok(builder) @@ -1371,4 +1404,127 @@ mod tests { assert!(has_format, "format_version stamp missing"); assert!(has_gen, "writer_generation stamp missing"); } + + /// A LIST column's leaf lives at `.list.element`; a single-segment path would miss it + /// and every per-column setting would silently fall back to defaults. + fn list_schema(name: &str, element: ArrowDataType) -> ArrowSchema { + let child = std::sync::Arc::new(Field::new("element", element, true)); + ArrowSchema::new(vec![Field::new(name, ArrowDataType::List(child), true)]) + } + + fn list_leaf_path(name: &str) -> parquet::schema::types::ColumnPath { + parquet::schema::types::ColumnPath::new(vec![ + name.to_string(), + "list".to_string(), + "element".to_string(), + ]) + } + + #[test] + fn test_list_column_uses_leaf_column_path_for_encoding() { + let mut field_configs = HashMap::new(); + field_configs.insert( + "tags".to_string(), + FieldConfig { + encoding_type: Some("DELTA_BYTE_ARRAY".to_string()), + ..Default::default() + }, + ); + let config = NativeSettings { + field_configs: Some(field_configs), + ..Default::default() + }; + let schema = list_schema("tags", ArrowDataType::Utf8); + let props = WriterPropertiesBuilder::build(&config, &schema).unwrap(); + + assert_eq!( + props.encoding(&list_leaf_path("tags")), + Some(Encoding::DELTA_BYTE_ARRAY), + "encoding must be set on the list leaf path" + ); + // The single-segment path must NOT carry the setting — that was the silent-no-op bug. + assert_eq!( + props.encoding(&parquet::schema::types::ColumnPath::from("tags")), + None + ); + } + + #[test] + fn test_list_column_encoding_validated_against_element_type() { + // DELTA_BYTE_ARRAY is valid for Utf8 but not Int64. Validation must look through the list + // wrapper at the element, otherwise every encoding would be accepted for a list column. + let mut field_configs = HashMap::new(); + field_configs.insert( + "nums".to_string(), + FieldConfig { + encoding_type: Some("DELTA_BYTE_ARRAY".to_string()), + ..Default::default() + }, + ); + let config = NativeSettings { + field_configs: Some(field_configs), + ..Default::default() + }; + let schema = list_schema("nums", ArrowDataType::Int64); + let err = WriterPropertiesBuilder::build(&config, &schema) + .expect_err("DELTA_BYTE_ARRAY must be rejected for a list of Int64"); + assert!(err.contains("not supported"), "unexpected error: {}", err); + } + + #[test] + fn test_list_column_resolves_type_level_config_by_element_type() { + // A cluster-level `utf8` rule must apply to LIST: the type key is derived from the + // element, not from the Debug string of the list wrapper. + let mut type_compression = HashMap::new(); + type_compression.insert("utf8".to_string(), "SNAPPY".to_string()); + let config = NativeSettings { + type_compression_configs: Some(type_compression), + ..Default::default() + }; + let schema = list_schema("tags", ArrowDataType::Utf8); + let props = WriterPropertiesBuilder::build(&config, &schema).unwrap(); + assert!(matches!( + props.compression(&list_leaf_path("tags")), + Compression::SNAPPY + )); + } + + #[test] + fn test_list_column_bloom_filter_on_leaf_path() { + let config = NativeSettings { + bloom_filter_enabled: Some(true), + ..Default::default() + }; + let schema = list_schema("tags", ArrowDataType::Utf8); + let props = WriterPropertiesBuilder::build(&config, &schema).unwrap(); + assert!( + props + .bloom_filter_properties(&list_leaf_path("tags")) + .is_some(), + "bloom filter must be enabled on the list leaf path" + ); + } + + #[test] + fn test_scalar_column_path_unchanged() { + // Regression guard: non-list columns must still be addressed by bare name. + let mut field_configs = HashMap::new(); + field_configs.insert( + "name".to_string(), + FieldConfig { + encoding_type: Some("DELTA_BYTE_ARRAY".to_string()), + ..Default::default() + }, + ); + let config = NativeSettings { + field_configs: Some(field_configs), + ..Default::default() + }; + let schema = schema_with(vec![("name", ArrowDataType::Utf8)]); + let props = WriterPropertiesBuilder::build(&config, &schema).unwrap(); + assert_eq!( + props.encoding(&parquet::schema::types::ColumnPath::from("name")), + Some(Encoding::DELTA_BYTE_ARRAY) + ); + } } diff --git a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/vsr/VSRManagerTests.java b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/vsr/VSRManagerTests.java index d47fbd31f3ed7..177fea0f6f95f 100644 --- a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/vsr/VSRManagerTests.java +++ b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/vsr/VSRManagerTests.java @@ -9,6 +9,8 @@ package org.opensearch.parquet.vsr; import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.VarCharVector; +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; @@ -28,11 +30,15 @@ import org.opensearch.parquet.bridge.ParquetFileMetadata; import org.opensearch.parquet.bridge.RustBridge; import org.opensearch.parquet.engine.ParquetDataFormat; +import org.opensearch.parquet.fields.ParquetField; +import org.opensearch.parquet.fields.core.data.number.IntegerParquetField; +import org.opensearch.parquet.fields.core.data.text.KeywordParquetField; import org.opensearch.parquet.memory.ArrowBufferPool; import org.opensearch.parquet.writer.ParquetDocumentInput; import org.opensearch.threadpool.FixedExecutorBuilder; import org.opensearch.threadpool.ThreadPool; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Future; @@ -572,6 +578,176 @@ public void testRowIdColumnIsSortedAndContiguous() throws Exception { } } + public void testMultiValueFieldWritesListColumn() throws Exception { + String filePath = createTempDir().resolve("multi-value.parquet").toString(); + VSRManager manager = new VSRManager(filePath, indexSettings, schema, bufferPool, 100, threadPool, 0L); + try { + // Mapping update introduces a keyword field declared multi-valued, so it arrives as a + // LIST rather than a flat Utf8 column. + manager.reconcileSchema(schemaWithMultiValue("tags")); + + KeywordFieldMapper.KeywordFieldType tags = new KeywordFieldMapper.KeywordFieldType("tags"); + tags.setMultiValued(true); + assignTestCapabilities(tags, PARQUET_FORMAT); + NumberFieldMapper.NumberFieldType valField = new NumberFieldMapper.NumberFieldType("val", NumberFieldMapper.NumberType.INTEGER); + assignTestCapabilities(valField, PARQUET_FORMAT); + + // Row 0: three values including a duplicate. Row 1: field absent entirely. + // Row 2: a single value. Covers the three cardinalities in one file. + ParquetDocumentInput doc0 = new ParquetDocumentInput(); + populateMetadataFields(doc0); + doc0.setRowId(DocumentInput.ROW_ID_FIELD, 0); + doc0.addField(valField, 1); + doc0.addField(tags, "b"); + doc0.addField(tags, "a"); + doc0.addField(tags, "b"); + manager.addDocument(doc0); + + ParquetDocumentInput doc1 = new ParquetDocumentInput(); + populateMetadataFields(doc1); + doc1.setRowId(DocumentInput.ROW_ID_FIELD, 1); + doc1.addField(valField, 2); + manager.addDocument(doc1); + + ParquetDocumentInput doc2 = new ParquetDocumentInput(); + populateMetadataFields(doc2); + doc2.setRowId(DocumentInput.ROW_ID_FIELD, 2); + doc2.addField(valField, 3); + doc2.addField(tags, "solo"); + manager.addDocument(doc2); + + ListVector listVector = (ListVector) manager.getActiveManagedVSR().getVector("tags"); + assertEquals(List.of("b", "a", "b"), listElements(listVector, 0)); + assertTrue("absent field must read back as a null list", listVector.isNull(1)); + assertEquals(List.of("solo"), listElements(listVector, 2)); + + ParquetFileMetadata metadata = manager.flush(); + assertNotNull(metadata); + assertEquals(3, metadata.numRows()); + } finally { + manager.close(); + } + } + + public void testMultiValueNumericFieldWritesListColumnAndFlushes() throws Exception { + String filePath = createTempDir().resolve("multi-value-numeric.parquet").toString(); + VSRManager manager = new VSRManager(filePath, indexSettings, schema, bufferPool, 100, threadPool, 0L); + try { + List fields = new ArrayList<>(schema.getFields()); + fields.addAll(metadataFields()); + fields.add(new IntegerParquetField().toArrowField("numbers", true)); + manager.reconcileSchema(new Schema(fields)); + + NumberFieldMapper.NumberFieldType numbers = new NumberFieldMapper.NumberFieldType( + "numbers", + NumberFieldMapper.NumberType.INTEGER + ); + numbers.setMultiValued(true); + numbers.setMultiValueSupported(true); + assignTestCapabilities(numbers, PARQUET_FORMAT); + + ParquetDocumentInput doc = new ParquetDocumentInput(); + populateMetadataFields(doc); + doc.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + doc.addField(numbers, 10); + doc.addField(numbers, 20); + manager.addDocument(doc); + + ListVector vector = (ListVector) manager.getActiveManagedVSR().getVector("numbers"); + assertEquals(2, vector.getInnerValueCountAt(0)); + IntVector elements = (IntVector) vector.getDataVector(); + assertEquals(10, elements.get(0)); + assertEquals(20, elements.get(1)); + + ParquetFileMetadata metadata = manager.flush(); + assertNotNull(metadata); + assertEquals(1, metadata.numRows()); + } finally { + manager.close(); + } + } + + public void testMultiValueFieldWritesEmptyListDistinctFromAbsent() throws Exception { + String filePath = createTempDir().resolve("multi-value-empty.parquet").toString(); + VSRManager manager = new VSRManager(filePath, indexSettings, schema, bufferPool, 100, threadPool, 0L); + try { + manager.reconcileSchema(schemaWithMultiValue("tags")); + NumberFieldMapper.NumberFieldType valField = new NumberFieldMapper.NumberFieldType("val", NumberFieldMapper.NumberType.INTEGER); + assignTestCapabilities(valField, PARQUET_FORMAT); + + // An explicit "tags": [] parses to zero addField calls, so the writer never sees the + // field and the row is null — same as absent. Documented here so the distinction + // between [] and absent is a deliberate, tested choice rather than an accident. + ParquetDocumentInput doc = new ParquetDocumentInput(); + populateMetadataFields(doc); + doc.setRowId(DocumentInput.ROW_ID_FIELD, 0); + doc.addField(valField, 1); + manager.addDocument(doc); + + ListVector listVector = (ListVector) manager.getActiveManagedVSR().getVector("tags"); + assertTrue(listVector.isNull(0)); + assertEquals(1, manager.flush().numRows()); + } finally { + manager.close(); + } + } + + public void testReconcileSchemaPreservesListChildren() throws Exception { + String filePath = createTempDir().resolve("reconcile-children.parquet").toString(); + VSRManager manager = new VSRManager(filePath, indexSettings, schema, bufferPool, 100, threadPool, 0L); + try { + assertTrue(manager.reconcileSchema(schemaWithMultiValue("tags"))); + + // reconcileSchema used to rebuild each Field from name + FieldType, dropping children + // and leaving a ListVector with no element vector to write into. + ListVector listVector = (ListVector) manager.getActiveManagedVSR().getVector("tags"); + assertNotNull(listVector); + // A dropped child would leave the data vector as an uninitialized ZeroVector. + assertTrue( + "list vector must have a typed element vector, got " + listVector.getDataVector().getClass().getSimpleName(), + listVector.getDataVector() instanceof VarCharVector + ); + + // Assert on the VSR *schema* rather than the vector's own field: Arrow Java renames a + // list vector's child to "$data$" internally, but the schema — which is what + // exportSchema hands to the native writer, and therefore what determines the Parquet + // leaf path "tags.list.element" — keeps the declared name. + Field tagsField = manager.getActiveManagedVSR() + .getSchema() + .getFields() + .stream() + .filter(f -> f.getName().equals("tags")) + .findFirst() + .orElseThrow(); + assertEquals(ArrowType.List.INSTANCE, tagsField.getType()); + assertEquals(1, tagsField.getChildren().size()); + assertEquals(ParquetField.LIST_ELEMENT_NAME, tagsField.getChildren().get(0).getName()); + assertEquals(new ArrowType.Utf8(), tagsField.getChildren().get(0).getType()); + } finally { + manager.close(); + } + } + + /** Reads back the elements of one row of a list vector. */ + private static List listElements(ListVector listVector, int row) { + int start = listVector.getOffsetBuffer().getInt((long) row * 4); + int end = listVector.getOffsetBuffer().getInt((long) (row + 1) * 4); + VarCharVector data = (VarCharVector) listVector.getDataVector(); + List values = new ArrayList<>(end - start); + for (int i = start; i < end; i++) { + values.add(new String(data.get(i), StandardCharsets.UTF_8)); + } + return values; + } + + /** Test schema plus metadata fields plus a keyword field declared multi-valued. */ + private Schema schemaWithMultiValue(String name) { + List fields = new ArrayList<>(schema.getFields()); + fields.addAll(metadataFields()); + fields.add(new KeywordParquetField().toArrowField(name, true)); + return new Schema(fields); + } + /** Returns a copy of the test schema with one extra field appended (alongside metadata fields). */ private Schema schemaWith(String name, ArrowType type) { List fields = new ArrayList<>(schema.getFields()); diff --git a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetDocumentInputTests.java b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetDocumentInputTests.java index 0fd5ec545326b..d342272fac181 100644 --- a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetDocumentInputTests.java +++ b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetDocumentInputTests.java @@ -86,4 +86,184 @@ public void testRejectsDuplicateFieldInSingleDocument() throws Exception { input.addField(valField, 10); expectThrows(MapperParsingException.class, () -> input.addField(valField, 20)); } + + public void testDeclaredMultiValueFieldAccumulatesValuesInOrder() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType tags = new KeywordFieldMapper.KeywordFieldType("tags"); + tags.setMultiValued(true); + assignTestCapabilities(tags, PARQUET_FORMAT); + + // The document parser reports one addField call per array element. + input.addField(tags, "b"); + input.addField(tags, "a"); + input.addField(tags, "b"); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + FieldValuePair pair = findPair(input, "tags"); + assertTrue(pair.isMultiValued()); + // Document order and duplicates are preserved: the values are the source of truth for + // derived _source, so they must not be sorted or deduplicated. + assertEquals(List.of("b", "a", "b"), pair.getValue()); + assertEquals(3, pair.valueCount()); + assertEquals(3L, input.getFieldCount("tags")); + } + + public void testDeclaredMultiValueFieldWithSingleValueIsStillAList() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType tags = new KeywordFieldMapper.KeywordFieldType("tags"); + tags.setMultiValued(true); + assignTestCapabilities(tags, PARQUET_FORMAT); + + input.addField(tags, "solo"); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + // A scalar JSON value on a declared list column still writes a one-element list, so the + // column type stays consistent across documents. + FieldValuePair pair = findPair(input, "tags"); + assertTrue(pair.isMultiValued()); + assertEquals(List.of("solo"), pair.getValue()); + } + + public void testDeclaredMultiValueFieldWithEmptyArrayIsPresentEmptyList() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType tags = new KeywordFieldMapper.KeywordFieldType("tags"); + tags.setMultiValued(true); + assignTestCapabilities(tags, PARQUET_FORMAT); + + // The parser signals an explicit empty array ("tags": []) with an empty List value. It must + // seed a present, zero-value list (written as an empty-but-non-null LIST cell) rather than + // being dropped, so an empty array stays distinct from an absent field in reconstructed + // _source. + input.addField(tags, List.of()); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + FieldValuePair pair = findPair(input, "tags"); + assertTrue(pair.isMultiValued()); + assertEquals(List.of(), pair.getValue()); + assertEquals(0, pair.valueCount()); + assertEquals(0L, input.getFieldCount("tags")); + } + + public void testMultiValueAccumulationKeyedByNameNotInstanceIdentity() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + // Two DISTINCT field-type instances that share a name, as could arise if the parse path ever + // handed back a fresh wrapper per array element. Accumulation keys on the name, so both + // elements must land in the SAME list rather than the second creating a new pair (which + // would silently degrade multi_value to last-value-wins). + MappedFieldType first = new KeywordFieldMapper.KeywordFieldType("tags"); + first.setMultiValued(true); + assignTestCapabilities(first, PARQUET_FORMAT); + MappedFieldType second = new KeywordFieldMapper.KeywordFieldType("tags"); + second.setMultiValued(true); + assignTestCapabilities(second, PARQUET_FORMAT); + assertNotSame(first, second); + + input.addField(first, "a"); + input.addField(second, "b"); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + assertEquals( + "both elements must accumulate into one pair", + 1, + input.getFinalInput().stream().filter(p -> p.getFieldType().name().equals("tags")).count() + ); + FieldValuePair pair = findPair(input, "tags"); + assertEquals(List.of("a", "b"), pair.getValue()); + } + + public void testUndeclaredKeywordPromotesWhenSecondValueArrives() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType other = new KeywordFieldMapper.KeywordFieldType("other"); + other.setMultiValueSupported(true); + assignTestCapabilities(other, PARQUET_FORMAT); + + input.addField(other, "one"); + input.addField(other, "two"); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + FieldValuePair pair = findPair(input, "other"); + assertTrue(pair.isMultiValued()); + assertEquals(List.of("one", "two"), pair.getValue()); + assertEquals(2L, input.getFieldCount("other")); + } + + public void testExplicitSingleFieldRejectsSecondValue() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType number = new NumberFieldMapper.NumberFieldType("number", NumberFieldMapper.NumberType.INTEGER); + number.setMultiValueSupported(true); + number.setMultiValueState(MappedFieldType.MultiValueState.SCALAR); + assignTestCapabilities(number, PARQUET_FORMAT); + + input.addField(number, 10); + MapperParsingException error = expectThrows(MapperParsingException.class, () -> input.addField(number, 20)); + assertThat(error.getMessage(), org.hamcrest.Matchers.containsString("locked scalar by [multi_value: false]")); + } + + public void testUndeclaredNumericFieldPromotesWhenSecondValueArrives() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType number = new NumberFieldMapper.NumberFieldType("number", NumberFieldMapper.NumberType.INTEGER); + number.setMultiValueSupported(true); + assignTestCapabilities(number, PARQUET_FORMAT); + + input.addField(number, 10); + input.addField(number, 20); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + FieldValuePair pair = findPair(input, "number"); + assertTrue(pair.isMultiValued()); + assertEquals(List.of(10, 20), pair.getValue()); + assertEquals(2L, input.getFieldCount("number")); + } + + public void testMultiValueFieldCountIsValueCountNotEntryCount() { + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType tags = new KeywordFieldMapper.KeywordFieldType("tags"); + tags.setMultiValued(true); + assignTestCapabilities(tags, PARQUET_FORMAT); + input.addField(tags, "x"); + input.addField(tags, "y"); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + // Metadata fields must still count as exactly one so getFinalInput's assertions hold. + assertEquals(1L, input.getFieldCount(org.opensearch.index.mapper.IdFieldMapper.NAME)); + assertEquals(2L, input.getFieldCount("tags")); + // One collected entry holding two values. + assertEquals(5, input.getFinalInput().size()); + } + + public void testDerivedSourceCompanionFieldFollowsParentState() { + // KeywordFieldMapper emits "_ignored_source." alongside the parent when a normalizer + // or ignore_above alters the value, and buildRawKeywordValueFieldType copies the parent's + // multi_value flag onto it, so the document input must accumulate its values too — + // otherwise it would reject the second value while its own column expects a list. + ParquetDocumentInput input = new ParquetDocumentInput(); + populateMetadataFields(input); + MappedFieldType rawValue = new KeywordFieldMapper.KeywordFieldType("_ignored_source.tags"); + rawValue.setMultiValued(true); + assignTestCapabilities(rawValue, PARQUET_FORMAT); + + input.addField(rawValue, "RAW-ONE"); + input.addField(rawValue, "RAW-TWO"); + input.setRowId(DocumentInput.ROW_ID_FIELD, 0L); + + FieldValuePair pair = findPair(input, "_ignored_source.tags"); + assertTrue(pair.isMultiValued()); + assertEquals(List.of("RAW-ONE", "RAW-TWO"), pair.getValue()); + } + + private static FieldValuePair findPair(ParquetDocumentInput input, String fieldName) { + return input.getFinalInput() + .stream() + .filter(p -> p.getFieldType().name().equals(fieldName)) + .findFirst() + .orElseThrow(() -> new AssertionError("no collected field named " + fieldName)); + } } diff --git a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetWriterTests.java b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetWriterTests.java index 4d8e6fe7dc24c..f2ab949a8f672 100644 --- a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetWriterTests.java +++ b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/writer/ParquetWriterTests.java @@ -21,6 +21,7 @@ import org.opensearch.index.engine.dataformat.FileInfos; import org.opensearch.index.engine.dataformat.FlushInput; import org.opensearch.index.engine.dataformat.WriteResult; +import org.opensearch.index.engine.dataformat.WriterState; import org.opensearch.index.mapper.KeywordFieldMapper; import org.opensearch.index.mapper.MappedFieldType; import org.opensearch.index.mapper.NumberFieldMapper; @@ -200,6 +201,34 @@ public void testFlushWithNoDocuments() throws Exception { assertEquals(FileInfos.empty(), writer.flush(FlushInput.EMPTY)); } + public void testMappingTypeChangeRetiresWriterForSchemaFence() throws Exception { + String filePath = createTempDir().resolve("schema-fence.parquet").toString(); + ParquetField keyword = ArrowFieldRegistry.getParquetField(nameField.typeName()); + List promotedFields = new ArrayList<>(); + promotedFields.add(ArrowFieldRegistry.getParquetField(idField.typeName()).toArrowField(idField.name(), false)); + promotedFields.add(keyword.toArrowField(nameField.name(), true)); + promotedFields.add(ArrowFieldRegistry.getParquetField(scoreField.typeName()).toArrowField(scoreField.name(), false)); + promotedFields.addAll(metadataFields()); + Schema promotedSchema = new Schema(promotedFields); + ParquetWriter writer = new ParquetWriter( + filePath, + 1L, + 1L, + new ParquetDataFormat(), + schema, + () -> promotedSchema, + bufferPool, + indexSettings, + threadPool, + null + ); + + writer.updateMappingVersion(2L); + + assertEquals(WriterState.RETIRED_FLUSHABLE, writer.state()); + assertEquals(FileInfos.empty(), writer.flush(FlushInput.EMPTY)); + } + public void testAddDocReturnsFailureOnOutOfMemory() throws Exception { String filePath = createTempDir().resolve("oom.parquet").toString(); ParquetWriter writer = new ParquetWriter( diff --git a/server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java b/server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java index 7938f539e368e..cdf3055e9aada 100644 --- a/server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java +++ b/server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java @@ -708,6 +708,11 @@ private Engine.IndexResult indexIntoEngine(Engine.Index index, IndexingStrategy }); Writer currentWriter = lockedWriter.get(); currentWriter.updateMappingVersion(mappingVersion); + if (currentWriter.state() != WriterState.ACTIVE) { + writerCheckedOut = retireWriterIfNeeded(lockedWriter); + lockedWriter = null; + return indexIntoEngine(index, plan); + } // Writer pool must never return null — it creates on demand via the supplier assert index.seqNo() >= 0 : "seqNo must be assigned before writing but was: " + index.seqNo(); assert index.primaryTerm() > 0 : "primaryTerm must be positive but was: " + index.primaryTerm(); diff --git a/server/src/main/java/org/opensearch/index/mapper/DocumentParser.java b/server/src/main/java/org/opensearch/index/mapper/DocumentParser.java index 843a7015b7861..7c01e43fda8f5 100644 --- a/server/src/main/java/org/opensearch/index/mapper/DocumentParser.java +++ b/server/src/main/java/org/opensearch/index/mapper/DocumentParser.java @@ -1522,7 +1522,9 @@ private static void parseNonDynamicArray(ParseContext context, ObjectMapper mapp ); } final String[] paths = resolvePathForParsing(mapper, lastFieldName); + boolean sawElement = false; while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) { + sawElement = true; if (token == XContentParser.Token.START_OBJECT) { parseObject(context, mapper, lastFieldName, paths); } else if (token == XContentParser.Token.START_ARRAY) { @@ -1542,6 +1544,41 @@ private static void parseNonDynamicArray(ParseContext context, ObjectMapper mapp parseValue(context, mapper, lastFieldName, token, paths); } } + if (sawElement == false) { + registerEmptyMultiValueArray(context, mapper, lastFieldName, paths); + } + } + + /** + * Records an empty array ({@code "field": []}) for a pluggable-data-format field mapped with + * {@code multi_value: true}. The element loop above never fires for an empty array, so without + * this the field would be absent from the document input and its LIST column cell would be + * written null — collapsing the distinction between {@code []} and a missing field when + * {@code _source} is later reconstructed from the columns. Registering an empty list lets the + * writer emit a zero-length, non-null list instead. + * + *

Strictly gated: no-op unless the pluggable data format is enabled and the resolved leaf is + * a {@code multi_value} {@link FieldMapper}, so stock indexing is unaffected. + * + *

Reached from every scalar-leaf array route — top-level, nested, and disable_objects arrays + * all funnel through {@link #parseNonDynamicArray}. The only array route that bypasses it is a + * mapper with {@link FieldMapper#parsesArrayValue()} true (geo/completion), which no + * {@code multi_value} type currently is; if that ever changes, that route needs equivalent + * empty-array handling or {@code []} would collapse to an absent field there. + */ + private static void registerEmptyMultiValueArray(ParseContext context, ObjectMapper mapper, String lastFieldName, String[] paths) { + if (context.indexSettings().isPluggableDataFormatEnabled() == false) { + return; + } + Mapper leaf = getMapper(context, mapper, lastFieldName, paths); + if (leaf instanceof ParametrizedFieldMapper fieldMapper && fieldMapper.fieldType().isMultiValueSupported()) { + if (fieldMapper.fieldType().isMultiValued() == false) { + fieldMapper.addMultiValueMappingUpdate(context); + } + context.documentInput().addField(fieldMapper.fieldType(), List.of()); + } else if (leaf instanceof FieldMapper fieldMapper && fieldMapper.fieldType().isMultiValued()) { + context.documentInput().addField(fieldMapper.fieldType(), List.of()); + } } private static void parseValue( diff --git a/server/src/main/java/org/opensearch/index/mapper/FilterFieldType.java b/server/src/main/java/org/opensearch/index/mapper/FilterFieldType.java index ecec066151f71..d6b36c5435bf4 100644 --- a/server/src/main/java/org/opensearch/index/mapper/FilterFieldType.java +++ b/server/src/main/java/org/opensearch/index/mapper/FilterFieldType.java @@ -269,6 +269,41 @@ public void setEagerGlobalOrdinals(boolean eagerGlobalOrdinals) { delegate.setEagerGlobalOrdinals(eagerGlobalOrdinals); } + @Override + public boolean isMultiValued() { + return delegate.isMultiValued(); + } + + @Override + public void setMultiValued(boolean multiValued) { + delegate.setMultiValued(multiValued); + } + + @Override + public MultiValueState multiValueState() { + return delegate.multiValueState(); + } + + @Override + public void setMultiValueState(MultiValueState multiValueState) { + delegate.setMultiValueState(multiValueState); + } + + @Override + public boolean isMultiValueAutoPromotionEnabled() { + return delegate.isMultiValueAutoPromotionEnabled(); + } + + @Override + public boolean isMultiValueSupported() { + return delegate.isMultiValueSupported(); + } + + @Override + public void setMultiValueSupported(boolean multiValueSupported) { + delegate.setMultiValueSupported(multiValueSupported); + } + @Override public DocValueFormat docValueFormat(String format, ZoneId timeZone) { return delegate.docValueFormat(format, timeZone); diff --git a/server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java b/server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java index 79c74fe5f3bec..ed8b3fd68fc21 100644 --- a/server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java +++ b/server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java @@ -176,6 +176,15 @@ public static class Builder extends ParametrizedFieldMapper.Builder { private final Parameter> meta = Parameter.metaParam(); private final Parameter boost = Parameter.boostParam(); + /** + * Declares this field multi-valued for columnar data formats. Lucene is inherently + * multi-valued, so the flag matters only to pluggable formats whose column type is + * fixed per file. A scalar field can promote to multi-valued when indexing first + * encounters a second stored value. The transition is one-way because existing LIST + * files cannot be interpreted as scalar columns. + */ + private final Parameter multiValue = multiValueParameter(); + private final IndexAnalyzers indexAnalyzers; private final boolean canConsumeRawValueForSource; @@ -236,7 +245,8 @@ protected List> getParameters() { normalizer, splitQueriesOnWhitespace, boost, - meta + meta, + multiValue ) ); parameters.addAll(pluginMappingParameters()); @@ -359,6 +369,8 @@ public KeywordFieldType(String name, FieldType fieldType, NamedAnalyzer normaliz setEagerGlobalOrdinals(builder.eagerGlobalOrdinals.getValue()); setIndexAnalyzer(normalizer); setBoost(builder.boost.getValue()); + setMultiValueState(builder.multiValue.getValue()); + setMultiValueSupported(true); this.ignoreAbove = builder.ignoreAbove.getValue(); this.nullValue = builder.nullValue.getValue(); this.useSimilarity = builder.useSimilarity.getValue(); @@ -846,6 +858,7 @@ private void checkToDisableCaching(QueryShardContext context) { private final boolean useSimilarity; private final String normalizerName; private final boolean splitQueriesOnWhitespace; + private final MappedFieldType.MultiValueState multiValueState; private final KeywordFieldType rawKeywordValueFieldType; private final IndexAnalyzers indexAnalyzers; @@ -874,6 +887,7 @@ protected KeywordFieldMapper( this.useSimilarity = builder.useSimilarity.getValue(); this.normalizerName = builder.normalizer.getValue(); this.splitQueriesOnWhitespace = builder.splitQueriesOnWhitespace.getValue(); + this.multiValueState = builder.multiValue.getValue(); this.indexAnalyzers = builder.indexAnalyzers; this.canConsumeRawValueForSource = builder.canConsumeRawValueForSource; this.mappingPluginParameterValues = builder.pluginMappingParameterValues(); @@ -911,7 +925,19 @@ public String normalizerName() { */ private KeywordFieldType buildRawKeywordValueFieldType() { if (isIneligibleForGeneratingSource() && canConsumeRawValueForSource) { - return new KeywordFieldType("_ignored_source." + fieldType().name(), false, true, false, false, fieldType().meta()); + KeywordFieldType rawValueType = new KeywordFieldType( + "_ignored_source." + fieldType().name(), + false, + true, + false, + false, + fieldType().meta() + ); + // The companion carries the pre-normalization values for derived source, so it must + // mirror the parent's multi-value state or source reconstruction would lose values. + rawValueType.setMultiValueState(multiValueState); + rawValueType.setMultiValueSupported(true); + return rawValueType; } return null; } @@ -965,7 +991,7 @@ protected void parseCreateFieldForPluggableFormat(ParseContext context) throws I } String value = parseKeyword(textValue); if (value != null) { - context.documentInput().addField(fieldType(), value); + addFieldForPluggableFormat(context, value); } // For derived source: store raw value separately when normalizer/ignore_above alters it. // Skip for multi-field sub-fields (name contains dot) — parent field stores the raw source. diff --git a/server/src/main/java/org/opensearch/index/mapper/MappedFieldType.java b/server/src/main/java/org/opensearch/index/mapper/MappedFieldType.java index 6310a7f270fb5..c38f664f7a529 100644 --- a/server/src/main/java/org/opensearch/index/mapper/MappedFieldType.java +++ b/server/src/main/java/org/opensearch/index/mapper/MappedFieldType.java @@ -88,6 +88,17 @@ @PublicApi(since = "1.0.0") public abstract class MappedFieldType { + /** + * Multi-value state for columnar mappings: omitted {@code multi_value} is AUTO, explicit + * {@code false} locks the field in SCALAR state, and explicit {@code true} selects LIST state. + */ + @ExperimentalApi + public enum MultiValueState { + AUTO, + SCALAR, + LIST + } + private final String name; private final boolean docValues; private final boolean isIndexed; @@ -97,6 +108,8 @@ public abstract class MappedFieldType { private float boost; private NamedAnalyzer indexAnalyzer; private boolean eagerGlobalOrdinals; + private MultiValueState multiValueState = MultiValueState.AUTO; + private boolean multiValueSupported; /** * Capability map assigning each registered {@link DataFormat} to the set of capabilities it owns for this field type. @@ -478,6 +491,65 @@ public void setEagerGlobalOrdinals(boolean eagerGlobalOrdinals) { this.eagerGlobalOrdinals = eagerGlobalOrdinals; } + /** + * Whether this field is declared to hold multiple values per document in columnar data + * formats ({@code multi_value} mapping parameter). Lucene is inherently multi-valued, so + * this flag only matters to pluggable formats whose column type is fixed per file (e.g. + * Parquet, where a declared field is stored as {@code LIST}). + * + * @opensearch.experimental + */ + @ExperimentalApi + public boolean isMultiValued() { + return multiValueState == MultiValueState.LIST; + } + + /** + * Compatibility setter for callers that only know scalar versus multi-valued. Setting false + * restores AUTO; mapping builders use {@link #setMultiValueState(MultiValueState)} to preserve + * explicit SCALAR intent. + */ + @ExperimentalApi + public void setMultiValued(boolean multiValued) { + this.multiValueState = multiValued ? MultiValueState.LIST : MultiValueState.AUTO; + } + + /** Returns the durable multi-value state. */ + @ExperimentalApi + public MultiValueState multiValueState() { + return multiValueState; + } + + /** Sets the durable multi-value state. */ + @ExperimentalApi + public void setMultiValueState(MultiValueState multiValueState) { + this.multiValueState = Objects.requireNonNull(multiValueState); + } + + /** Whether an additional value may trigger an automatic mapping promotion. */ + @ExperimentalApi + public boolean isMultiValueAutoPromotionEnabled() { + return multiValueState == MultiValueState.AUTO; + } + + /** + * Whether this field type can represent multiple scalar values in a pluggable data format. + * This is a mapper capability, distinct from {@link #isMultiValued()}, which is the current + * one-way mapping state. + * + * @opensearch.experimental + */ + @ExperimentalApi + public boolean isMultiValueSupported() { + return multiValueSupported; + } + + /** Sets whether this field type supports automatic scalar-to-multi-value promotion. */ + @ExperimentalApi + public void setMultiValueSupported(boolean multiValueSupported) { + this.multiValueSupported = multiValueSupported; + } + @ExperimentalApi public Map> getCapabilityMap() { return capabilityMap; diff --git a/server/src/main/java/org/opensearch/index/mapper/ParametrizedFieldMapper.java b/server/src/main/java/org/opensearch/index/mapper/ParametrizedFieldMapper.java index 828cac19467df..2ceee6b87ac37 100644 --- a/server/src/main/java/org/opensearch/index/mapper/ParametrizedFieldMapper.java +++ b/server/src/main/java/org/opensearch/index/mapper/ParametrizedFieldMapper.java @@ -128,6 +128,71 @@ public Map mappingPluginParameterValues() { public abstract ParametrizedFieldMapper.Builder getMergeBuilder(); + /** Creates the shared tri-state {@code multi_value} mapping parameter for scalar leaf mappers. */ + protected static Parameter multiValueParameter() { + return new Parameter<>( + "multi_value", + true, + () -> MappedFieldType.MultiValueState.AUTO, + (name, context, value) -> XContentMapValues.nodeBooleanValue(value) + ? MappedFieldType.MultiValueState.LIST + : MappedFieldType.MultiValueState.SCALAR, + mapper -> mapper.fieldType().multiValueState() + ).setSerializer((builder, name, mode) -> builder.field(name, mode == MappedFieldType.MultiValueState.LIST), mode -> switch (mode) { + case AUTO -> "auto"; + case SCALAR -> "false"; + case LIST -> "true"; + }) + .setSerializerCheck((includeDefaults, configured, mode) -> mode != MappedFieldType.MultiValueState.AUTO) + .setMergeValueNormalizer((current, incoming) -> incoming == MappedFieldType.MultiValueState.AUTO ? current : incoming) + .setMergeValidator((previous, next) -> previous == MappedFieldType.MultiValueState.AUTO || previous == next); + } + + /** + * Adds one successfully parsed scalar value to the pluggable document input and requests a + * mapping promotion when this is the second value for a supported field. + */ + protected final void addFieldForPluggableFormat(ParseContext context, Object value) { + MappedFieldType fieldType = fieldType(); + if (fieldType.isMultiValued() == false + && fieldType.isMultiValueSupported() + && context.documentInput().getFieldCount(fieldType.name()) > 0) { + if (fieldType.isMultiValueAutoPromotionEnabled() == false) { + throw new MapperParsingException( + "Field [" + fieldType.name() + "] is locked scalar by [multi_value: false] and cannot accept multiple values" + ); + } + addMultiValueMappingUpdate(context); + } + context.documentInput().addField(fieldType, value); + } + + /** Publishes the idempotent scalar-to-multi-value mapping update for this mapper. */ + final void addMultiValueMappingUpdate(ParseContext context) { + if (fieldType().isMultiValued()) { + return; + } + if (fieldType().isMultiValueSupported() == false) { + throw new MapperParsingException( + "Field [" + fieldType().name() + "] of type [" + fieldType().typeName() + "] does not support [multi_value]" + ); + } + if (fieldType().isMultiValueAutoPromotionEnabled() == false) { + throw new MapperParsingException( + "Field [" + fieldType().name() + "] is locked scalar by [multi_value: false] and cannot promote" + ); + } + Builder updateBuilder = getMergeBuilder(); + updateBuilder.setParameterValue("multi_value", MappedFieldType.MultiValueState.LIST); + ParametrizedFieldMapper update = updateBuilder.build(new BuilderContext(Settings.EMPTY, context.path())); + if (update.fieldType().isMultiValued() == false) { + throw new IllegalStateException( + "Mapper [" + fieldType().name() + "] advertises multi-value support but did not apply [multi_value]" + ); + } + context.addDynamicMapper(update); + } + @Override public ParametrizedFieldMapper merge(Mapper mergeWith) { @@ -234,6 +299,7 @@ public static sealed class Parameter implements Supplier permits SideEffec private SerializerCheck serializerCheck = (includeDefaults, isConfigured, value) -> includeDefaults || isConfigured; private Function conflictSerializer = Objects::toString; private BiPredicate mergeValidator; + private BiFunction mergeValueNormalizer = (current, incoming) -> incoming; private T value; private boolean isSet; @@ -360,6 +426,16 @@ public Parameter setMergeValidator(BiPredicate mergeValidator) { return this; } + /** + * Normalizes an incoming mapping-update value before merge validation. This is useful for + * tri-state parameters where the default value means "unspecified" and must preserve the + * current state rather than overwrite it. + */ + public Parameter setMergeValueNormalizer(BiFunction mergeValueNormalizer) { + this.mergeValueNormalizer = Objects.requireNonNull(mergeValueNormalizer); + return this; + } + private void validate() { if (validator != null) { validator.accept(getValue()); @@ -375,12 +451,13 @@ private void parse(String field, ParserContext context, Object in) { } private void merge(FieldMapper toMerge, Conflicts conflicts) { - T value = initializer.apply(toMerge); + T incoming = initializer.apply(toMerge); T current = getValue(); - if (mergeValidator.test(current, value)) { - setValue(value); + T merged = mergeValueNormalizer.apply(current, incoming); + if (mergeValidator.test(current, merged)) { + setValue(merged); } else { - conflicts.addConflict(name, conflictSerializer.apply(current), conflictSerializer.apply(value)); + conflicts.addConflict(name, conflictSerializer.apply(current), conflictSerializer.apply(incoming)); } } diff --git a/server/src/test/java/org/opensearch/index/mapper/KeywordFieldMapperTests.java b/server/src/test/java/org/opensearch/index/mapper/KeywordFieldMapperTests.java index 993a8af11075f..71de8950cb586 100644 --- a/server/src/test/java/org/opensearch/index/mapper/KeywordFieldMapperTests.java +++ b/server/src/test/java/org/opensearch/index/mapper/KeywordFieldMapperTests.java @@ -197,6 +197,7 @@ protected void registerParameters(ParameterChecker checker) throws IOException { checker.registerConflictCheck("null_value", b -> b.field("null_value", "foo")); checker.registerConflictCheck("similarity", b -> b.field("similarity", "boolean")); checker.registerConflictCheck("normalizer", b -> b.field("normalizer", "lowercase")); + checker.registerUpdateCheck(b -> b.field("multi_value", true), m -> assertTrue(m.fieldType().isMultiValued())); checker.registerUpdateCheck(b -> b.field("eager_global_ordinals", true), m -> assertTrue(m.fieldType().eagerGlobalOrdinals())); checker.registerUpdateCheck(b -> b.field("ignore_above", 256), m -> assertEquals(256, ((KeywordFieldMapper) m).ignoreAbove())); @@ -338,6 +339,19 @@ public void testEnableNorms() throws IOException { assertEquals(0, fieldNamesFields.length); } + public void testMultiValueCanPromoteButCannotDowngrade() throws IOException { + MapperService mapperService = createMapperService(fieldMapping(b -> b.field("type", "keyword"))); + + merge(mapperService, fieldMapping(b -> b.field("type", "keyword").field("multi_value", true))); + assertTrue(mapperService.fieldType("field").isMultiValued()); + + IllegalArgumentException error = expectThrows( + IllegalArgumentException.class, + () -> merge(mapperService, fieldMapping(b -> b.field("type", "keyword").field("multi_value", false))) + ); + assertThat(error.getMessage(), containsString("Cannot update parameter [multi_value] from [true] to [false]")); + } + public void testConfigureSimilarity() throws IOException { MapperService mapperService = createMapperService(fieldMapping(b -> b.field("type", "keyword").field("similarity", "boolean"))); MappedFieldType ft = mapperService.documentMapper().fieldTypes().get("field"); @@ -602,6 +616,66 @@ public void testPluggableDataFormatDefaultKeyword() throws IOException { assertTrue("Expected keyword field captured with value 'test_value'", found); } + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testRawSourceCompanionSupportsDynamicPromotion() throws IOException { + Settings settings = Settings.builder().put(getIndexSettings()).put("index.pluggable.dataformat.enabled", true).build(); + DocumentMapper mapper = createDocumentMapper( + settings, + mapping(b -> b.startObject("field").field("type", "keyword").field("normalizer", "lowercase").endObject()) + ); + KeywordFieldMapper fieldMapper = (KeywordFieldMapper) mapper.mappers().getMapper("field"); + + assertNotNull(fieldMapper.getRawValueFieldType()); + assertTrue(fieldMapper.getRawValueFieldType().isMultiValueSupported()); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testPluggableDataFormatSingletonArrayRemainsScalar() throws IOException { + Settings pluggableSettings = Settings.builder().put(getIndexSettings()).put("index.pluggable.dataformat.enabled", true).build(); + DocumentMapper mapper = createDocumentMapper( + pluggableSettings, + mapping(b -> b.startObject("field").field("type", "keyword").endObject()) + ); + CapturingDocumentInput docInput = new CapturingDocumentInput(); + ParsedDocument parsed = mapper.parse(source(b -> b.array("field", "one")), docInput); + + assertEquals(1L, docInput.getFieldCount("field")); + assertNull(parsed.dynamicMappingsUpdate()); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testPluggableDataFormatEmptyArrayPromotesKeyword() throws IOException { + Settings pluggableSettings = Settings.builder().put(getIndexSettings()).put("index.pluggable.dataformat.enabled", true).build(); + DocumentMapper mapper = createDocumentMapper( + pluggableSettings, + mapping(b -> b.startObject("field").field("type", "keyword").endObject()) + ); + CapturingDocumentInput docInput = new CapturingDocumentInput(); + ParsedDocument parsed = mapper.parse(source(b -> b.startArray("field").endArray()), docInput); + + assertNotNull(parsed.dynamicMappingsUpdate()); + Mapper update = parsed.dynamicMappingsUpdate().root().getMapper("field"); + assertThat(update, instanceOf(KeywordFieldMapper.class)); + assertTrue(((KeywordFieldMapper) update).fieldType().isMultiValued()); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testPluggableDataFormatPromotesKeywordOnSecondValue() throws IOException { + Settings pluggableSettings = Settings.builder().put(getIndexSettings()).put("index.pluggable.dataformat.enabled", true).build(); + DocumentMapper mapper = createDocumentMapper( + pluggableSettings, + mapping(b -> b.startObject("field").field("type", "keyword").endObject()) + ); + CapturingDocumentInput docInput = new CapturingDocumentInput(); + ParsedDocument parsed = mapper.parse(source(b -> b.array("field", "one", "two")), docInput); + + assertEquals(2L, docInput.getFieldCount("field")); + assertNotNull(parsed.dynamicMappingsUpdate()); + Mapper update = parsed.dynamicMappingsUpdate().root().getMapper("field"); + assertThat(update, instanceOf(KeywordFieldMapper.class)); + assertTrue(((KeywordFieldMapper) update).fieldType().isMultiValued()); + } + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) public void testPluggableDataFormatNullValueSkipped() throws IOException { Settings pluggableSettings = Settings.builder().put(getIndexSettings()).put("index.pluggable.dataformat.enabled", true).build(); diff --git a/server/src/test/java/org/opensearch/index/mapper/MultiValueFieldMapperTests.java b/server/src/test/java/org/opensearch/index/mapper/MultiValueFieldMapperTests.java new file mode 100644 index 0000000000000..ea366b2128047 --- /dev/null +++ b/server/src/test/java/org/opensearch/index/mapper/MultiValueFieldMapperTests.java @@ -0,0 +1,212 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.index.mapper; + +import org.opensearch.common.settings.Settings; +import org.opensearch.common.util.FeatureFlags; + +import java.io.IOException; +import java.util.List; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.not; + +public class MultiValueFieldMapperTests extends MapperServiceTestCase { + + private record FieldCase(String type, Object first, Object second) { + } + + private static List supportedScalarFields() { + return List.of( + new FieldCase("keyword", "prod", "error"), + new FieldCase("text", "first message", "second message"), + new FieldCase("match_only_text", "first message", "second message"), + new FieldCase("byte", 1, 2), + new FieldCase("short", 10, 20), + new FieldCase("integer", 100, 200), + new FieldCase("long", 1_000L, 2_000L), + new FieldCase("half_float", 1.5f, 2.5f), + new FieldCase("float", 1.25f, 2.25f), + new FieldCase("double", 1.125d, 2.125d), + new FieldCase("unsigned_long", 10L, 20L), + new FieldCase("boolean", true, false), + new FieldCase("date", "2026-01-01T00:00:00Z", "2026-01-02T00:00:00Z"), + new FieldCase("date_nanos", "2026-01-01T00:00:00.000000001Z", "2026-01-01T00:00:00.000000002Z"), + new FieldCase("ip", "10.0.0.1", "10.0.0.2"), + new FieldCase("binary", "AQI=", "AwQ=") + ); + } + + private Settings pluggableSettings() { + return Settings.builder().put(getIndexSettings()).put("index.pluggable.dataformat.enabled", true).build(); + } + + private DocumentMapper mapper(String type) throws IOException { + return mapper(type, null); + } + + private DocumentMapper mapper(String type, Boolean multiValue) throws IOException { + return createDocumentMapper(pluggableSettings(), mapping(b -> { + b.startObject("field").field("type", type); + if (multiValue != null) { + b.field("multi_value", multiValue); + } + if ("binary".equals(type)) { + b.field("store", true); + } + b.endObject(); + })); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testSecondValuePromotesEverySupportedScalarFamily() throws IOException { + for (FieldCase fieldCase : supportedScalarFields()) { + DocumentMapper mapper = mapper(fieldCase.type()); + CapturingDocumentInput input = new CapturingDocumentInput(); + ParsedDocument parsed = mapper.parse( + source(b -> b.startArray("field").value(fieldCase.first()).value(fieldCase.second()).endArray()), + input + ); + + assertEquals(fieldCase.type(), 2L, input.getFieldCount("field")); + assertNotNull(fieldCase.type(), parsed.dynamicMappingsUpdate()); + Mapper update = parsed.dynamicMappingsUpdate().root().getMapper("field"); + assertThat(fieldCase.type(), update, instanceOf(ParametrizedFieldMapper.class)); + assertTrue(fieldCase.type(), ((FieldMapper) update).fieldType().isMultiValued()); + assertTrue(fieldCase.type(), ((FieldMapper) update).fieldType().isMultiValueSupported()); + } + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testSingletonArrayStaysScalarEverySupportedScalarFamily() throws IOException { + for (FieldCase fieldCase : supportedScalarFields()) { + DocumentMapper mapper = mapper(fieldCase.type()); + CapturingDocumentInput input = new CapturingDocumentInput(); + ParsedDocument parsed = mapper.parse(source(b -> b.startArray("field").value(fieldCase.first()).endArray()), input); + + assertEquals(fieldCase.type(), 1L, input.getFieldCount("field")); + assertNull(fieldCase.type(), parsed.dynamicMappingsUpdate()); + } + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testIgnoredNullElementDoesNotPromoteNumericField() throws IOException { + DocumentMapper mapper = mapper("integer"); + CapturingDocumentInput input = new CapturingDocumentInput(); + ParsedDocument parsed = mapper.parse(source(b -> b.startArray("field").value(1).nullValue().endArray()), input); + + assertEquals(1L, input.getFieldCount("field")); + assertNull(parsed.dynamicMappingsUpdate()); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testEmptyArrayPromotesEverySupportedScalarFamily() throws IOException { + for (FieldCase fieldCase : supportedScalarFields()) { + DocumentMapper mapper = mapper(fieldCase.type()); + ParsedDocument parsed = mapper.parse(source(b -> b.startArray("field").endArray()), new CapturingDocumentInput()); + + assertNotNull(fieldCase.type(), parsed.dynamicMappingsUpdate()); + FieldMapper update = (FieldMapper) parsed.dynamicMappingsUpdate().root().getMapper("field"); + assertTrue(fieldCase.type(), update.fieldType().isMultiValued()); + } + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testExplicitMultiValueMappingEverySupportedScalarFamily() throws IOException { + for (FieldCase fieldCase : supportedScalarFields()) { + DocumentMapper mapper = createDocumentMapper(pluggableSettings(), mapping(b -> { + b.startObject("field").field("type", fieldCase.type()).field("multi_value", true); + if ("binary".equals(fieldCase.type())) { + b.field("store", true); + } + b.endObject(); + })); + FieldMapper fieldMapper = (FieldMapper) mapper.mappers().getMapper("field"); + assertTrue(fieldCase.type(), fieldMapper.fieldType().isMultiValued()); + assertTrue(fieldCase.type(), fieldMapper.fieldType().isMultiValueSupported()); + } + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testExplicitFalseLocksEverySupportedScalarFamily() throws IOException { + for (FieldCase fieldCase : supportedScalarFields()) { + DocumentMapper mapper = mapper(fieldCase.type(), false); + FieldMapper fieldMapper = (FieldMapper) mapper.mappers().getMapper("field"); + assertEquals(fieldCase.type(), MappedFieldType.MultiValueState.SCALAR, fieldMapper.fieldType().multiValueState()); + assertThat(fieldCase.type(), mapper.mappingSource().string(), containsString("\"multi_value\":false")); + + ParsedDocument singleton = mapper.parse( + source(b -> b.startArray("field").value(fieldCase.first()).endArray()), + new CapturingDocumentInput() + ); + assertNull(fieldCase.type(), singleton.dynamicMappingsUpdate()); + + MapperParsingException error = expectThrows( + MapperParsingException.class, + () -> mapper.parse( + source(b -> b.startArray("field").value(fieldCase.first()).value(fieldCase.second()).endArray()), + new CapturingDocumentInput() + ) + ); + assertNotNull(fieldCase.type(), error.getCause()); + assertThat(fieldCase.type(), error.getCause().getMessage(), containsString("locked scalar by [multi_value: false]")); + } + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testExplicitFalseRejectsEmptyArray() throws IOException { + DocumentMapper mapper = mapper("integer", false); + MapperParsingException error = expectThrows( + MapperParsingException.class, + () -> mapper.parse(source(b -> b.startArray("field").endArray()), new CapturingDocumentInput()) + ); + assertThat(error.getMessage(), containsString("locked scalar by [multi_value: false]")); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testOmittedParameterRemainsAutoAndIsNotSerialized() throws IOException { + DocumentMapper mapper = mapper("integer"); + FieldMapper fieldMapper = (FieldMapper) mapper.mappers().getMapper("field"); + assertEquals(MappedFieldType.MultiValueState.AUTO, fieldMapper.fieldType().multiValueState()); + assertThat(mapper.mappingSource().string(), not(containsString("multi_value"))); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testExplicitFalseCannotBeUpdatedToTrueAndUnspecifiedUpdatePreservesLock() throws IOException { + MapperService mapperService = createMapperService( + pluggableSettings(), + mapping(b -> b.startObject("field").field("type", "integer").field("multi_value", false).endObject()) + ); + + merge(mapperService, mapping(b -> b.startObject("field").field("type", "integer").endObject())); + assertEquals(MappedFieldType.MultiValueState.SCALAR, mapperService.fieldType("field").multiValueState()); + + IllegalArgumentException error = expectThrows( + IllegalArgumentException.class, + () -> merge(mapperService, mapping(b -> b.startObject("field").field("type", "integer").field("multi_value", true).endObject())) + ); + assertThat(error.getMessage(), containsString("Cannot update parameter [multi_value] from [false] to [true]")); + } + + @LockFeatureFlag(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG) + public void testIndexSortFieldCanPromote() throws IOException { + Settings settings = Settings.builder() + .put(pluggableSettings()) + .putList("index.sort.field", "field") + .putList("index.sort.order", "asc") + .build(); + DocumentMapper mapper = createDocumentMapper(settings, mapping(b -> b.startObject("field").field("type", "integer").endObject())); + + ParsedDocument parsed = mapper.parse(source(b -> b.array("field", 2, 1)), new CapturingDocumentInput()); + assertNotNull(parsed.dynamicMappingsUpdate()); + FieldMapper update = (FieldMapper) parsed.dynamicMappingsUpdate().root().getMapper("field"); + assertEquals(MappedFieldType.MultiValueState.LIST, update.fieldType().multiValueState()); + } +}