From 7d983f3f36a243ed8dbfa2de4dc9249f011d82de Mon Sep 17 00:00:00 2001 From: Varun Bansal Date: Sun, 30 Aug 2026 17:51:15 +0000 Subject: [PATCH] feat(parquet): Add adaptive keyword promotion Promote pluggable keyword mappings to multi_value after a second value or an explicit empty array. Fence Parquet writer generations on scalar-to-LIST schema changes and promote scalar rows during merge. Add mapper, writer, VSR, settings, and merge coverage. --- .../ParquetIndexCreationValidator.java | 51 ++- .../opensearch/parquet/ParquetSettings.java | 14 +- .../parquet/engine/ParquetIndexingEngine.java | 2 +- .../parquet/fields/ArrowSchemaBuilder.java | 40 +- .../parquet/fields/ParquetField.java | 103 +++++ .../core/data/text/KeywordParquetField.java | 16 +- ...ChangeRequiresWriterRotationException.java | 20 + .../opensearch/parquet/vsr/VSRManager.java | 20 +- .../parquet/writer/FieldValuePair.java | 105 ++++- .../parquet/writer/ParquetDocumentInput.java | 52 ++- .../parquet/writer/ParquetWriter.java | 15 +- .../src/main/rust/src/merge/context.rs | 57 ++- .../src/main/rust/src/merge/schema.rs | 82 ++-- .../rust/src/writer_properties_builder.rs | 210 +++++++-- .../rust/tests/merge_list_column_tests.rs | 416 ++++++++++++++++++ .../parquet/vsr/VSRManagerTests.java | 137 ++++++ .../writer/ParquetDocumentInputTests.java | 149 +++++++ .../parquet/writer/ParquetWriterTests.java | 29 ++ .../index/engine/DataFormatAwareEngine.java | 5 + .../index/mapper/DocumentParser.java | 37 ++ .../index/mapper/FilterFieldType.java | 10 + .../index/mapper/KeywordFieldMapper.java | 49 ++- .../index/mapper/MappedFieldType.java | 25 ++ .../index/mapper/KeywordFieldMapperTests.java | 61 +++ 24 files changed, 1590 insertions(+), 115 deletions(-) create mode 100644 sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/SchemaChangeRequiresWriterRotationException.java create mode 100644 sandbox/plugins/parquet-data-format/src/main/rust/tests/merge_list_column_tests.rs diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetIndexCreationValidator.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetIndexCreationValidator.java index 175b71f876704..0dbb4b44c48c3 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetIndexCreationValidator.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetIndexCreationValidator.java @@ -11,6 +11,8 @@ import org.apache.arrow.vector.types.pojo.Schema; import org.opensearch.index.IndexCreationValidator; import org.opensearch.index.IndexSettings; +import org.opensearch.index.IndexSortConfig; +import org.opensearch.index.mapper.MappedFieldType; import org.opensearch.index.mapper.MapperService; import org.opensearch.parquet.fields.ArrowSchemaBuilder; @@ -43,18 +45,49 @@ public void validate(MapperService mapperService, IndexSettings indexSettings) { ); } - if (!isParquetIndex || !hasParquetSettings) { + if (!isParquetIndex) { return; } + validateSortFieldsAreSingleValued(mapperService, indexSettings); + + // Building the schema validates the mapping's `multi_value` declarations: getSchema throws + // for a field whose type has no list support, turning what would otherwise be a + // per-document indexing failure into an immediate error at creation time. Schema schema = ArrowSchemaBuilder.getSchema(mapperService); - ParquetSettings.validateFieldConfigurations( - fieldEncodings, - fieldCompressions, - fieldBloomFilterEnabled, - lowCardinalityEnabledFields, - schema, - mapperService - ); + if (hasParquetSettings) { + ParquetSettings.validateFieldConfigurations( + fieldEncodings, + fieldCompressions, + fieldBloomFilterEnabled, + lowCardinalityEnabledFields, + schema, + mapperService + ); + } + } +} + + /** + * Rejects {@code index.sort.field} entries that are mapped {@code multi_value: true}. + *

+ * An index sort needs one total order over rows, but a multi-valued cell has no canonical + * scalar to order by — any of min/max/lexicographic would be a silent choice the user never + * made. Lucene rejects index sorting on multi-valued fields for the same reason; failing here + * keeps parity and turns what would otherwise be a merge-time failure (the native k-way merge + * cannot compare LIST sort keys) into an immediate, actionable error at creation time. + */ + private static void validateSortFieldsAreSingleValued(MapperService mapperService, IndexSettings indexSettings) { + for (String sortField : IndexSortConfig.INDEX_SORT_FIELD_SETTING.get(indexSettings.getSettings())) { + MappedFieldType fieldType = mapperService.fieldType(sortField); + if (fieldType != null && fieldType.isMultiValued()) { + throw new IllegalArgumentException( + "Cannot use field [" + + sortField + + "] in [index.sort.field]: the field is mapped [multi_value: true] and a " + + "multi-valued field has no single value to sort on" + ); + } + } } } 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..401b27edceaaf 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). + * Cardinality 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..e439d63befe26 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 @@ -14,6 +14,7 @@ import org.opensearch.index.engine.dataformat.FieldTypeCapabilities; import org.opensearch.index.engine.exec.PrimaryTermFieldType; import org.opensearch.index.mapper.IdFieldMapper; +import org.opensearch.index.mapper.KeywordFieldMapper; import org.opensearch.index.mapper.MappedFieldType; import org.opensearch.index.mapper.MapperParsingException; import org.opensearch.index.mapper.SeqNoFieldMapper; @@ -21,9 +22,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 +41,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 +60,38 @@ 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 instanceof KeywordFieldMapper.KeywordFieldType) { + existing.promoteToMultiValued(value); + return; + } 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() + + "]. Only keyword fields support automatic multi-value promotion." ); } - collectedFields.add(new FieldValuePair(fieldType, value)); + existing.addValue(value); } @Override @@ -84,13 +116,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/merge/context.rs b/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/context.rs index 03d9dec407be4..22834170bf413 100644 --- a/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/context.rs +++ b/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/context.rs @@ -6,6 +6,7 @@ * compatible open source license. */ +use std::collections::HashMap; use std::fs::File; use std::path::Path; use std::sync::Arc; @@ -52,12 +53,57 @@ pub struct MergeContext { tracked_writer_bytes: usize, } +fn merge_arrow_schemas_with_list_promotion(schemas: Vec) -> MergeResult { + let mut list_fields: HashMap> = HashMap::new(); + for schema in &schemas { + for field in schema.fields() { + if matches!(field.data_type(), ArrowDataType::List(_)) { + list_fields.insert(field.name().clone(), Arc::clone(field)); + } + } + } + + let normalized = schemas + .into_iter() + .map(|schema| { + let fields = schema + .fields() + .iter() + .map(|field| { + let Some(list_field) = list_fields.get(field.name()) else { + return field.as_ref().clone(); + }; + let ArrowDataType::List(child) = list_field.data_type() else { + return field.as_ref().clone(); + }; + if field.data_type() == child.data_type() { + list_field + .as_ref() + .clone() + .with_nullable(list_field.is_nullable() || field.is_nullable()) + } else { + field.as_ref().clone() + } + }) + .collect::>(); + ArrowSchema::new_with_metadata(fields, schema.metadata().clone()) + }) + .collect::>(); + + ArrowSchema::try_merge(normalized).map_err(|e| { + MergeError::Logic(format!( + "Failed to compute union schema across input files after scalar-to-LIST promotion: {}", + e + )) + }) +} + impl MergeContext { /// Creates a new merge context: builds union schemas, opens the output /// writer, and spawns the background IO task. pub fn new( arrow_schemas: Vec, - parquet_descriptors: &[SchemaDescriptor], + _parquet_descriptors: &[SchemaDescriptor], output_path: &str, index_name: &str, output_flush_rows: usize, @@ -75,12 +121,7 @@ impl MergeContext { } } - let union_data_schema = ArrowSchema::try_merge(arrow_schemas).map_err(|e| { - MergeError::Logic(format!( - "Failed to compute union schema across input files: {}", - e - )) - })?; + let union_data_schema = merge_arrow_schemas_with_list_promotion(arrow_schemas)?; let data_schema = Arc::new(union_data_schema); let mut output_fields: Vec = data_schema @@ -95,7 +136,7 @@ impl MergeContext { )); let output_schema = Arc::new(ArrowSchema::new(output_fields)); - let parquet_root = build_parquet_root_schema(parquet_descriptors)?; + let parquet_root = build_parquet_root_schema(output_schema.as_ref())?; let output_file = File::create(output_path)?; let throttled_writer = diff --git a/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs b/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs index 73e572f9b951f..3237d8c32f86a 100644 --- a/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs +++ b/sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs @@ -6,49 +6,26 @@ * compatible open source license. */ -use std::collections::HashSet; use std::sync::Arc; -use arrow::array::{ArrayRef, Int64Array, RecordBatch}; -use arrow::datatypes::Schema as ArrowSchema; -use parquet::basic::Repetition; +use arrow::array::{ArrayRef, Int64Array, ListArray, RecordBatch}; +use arrow::buffer::OffsetBuffer; +use arrow::datatypes::{DataType, Schema as ArrowSchema}; +use parquet::arrow::ArrowSchemaConverter; use parquet::schema::types::Type; -use super::error::MergeResult; +use super::error::{MergeError, MergeResult}; /// Reserved column name for the synthetic row identifier added during merge. pub const ROW_ID_COLUMN_NAME: &str = "__row_id__"; -/// Builds the output Parquet schema as the union of pre-read schema descriptors. +/// Builds the output Parquet schema from the canonical Arrow merge schema. /// -/// The output schema contains every column seen across all inputs, except: -/// - Any existing `__row_id__` column is removed. -/// - A fresh `__row_id__` INT64 REQUIRED column is appended at the end. -pub fn build_parquet_root_schema( - schema_descriptors: &[parquet::schema::types::SchemaDescriptor], -) -> MergeResult> { - let mut seen_names: HashSet = HashSet::new(); - let mut parquet_fields: Vec> = Vec::new(); - - for descr in schema_descriptors { - let root = descr.root_schema(); - for field in root.get_fields() { - if field.name() != ROW_ID_COLUMN_NAME && seen_names.insert(field.name().to_string()) { - parquet_fields.push(Arc::new(field.as_ref().clone())); - } - } - } - - let row_id_type = Type::primitive_type_builder(ROW_ID_COLUMN_NAME, parquet::basic::Type::INT64) - .with_repetition(Repetition::REQUIRED) - .build()?; - parquet_fields.push(Arc::new(row_id_type)); - - let parquet_root = Type::group_type_builder("schema") - .with_fields(parquet_fields) - .build()?; - - Ok(Arc::new(parquet_root)) +/// The canonical schema has already promoted compatible scalar/LIST field pairs to LIST, so +/// deriving the Parquet schema from it avoids selecting an arbitrary first input file's shape. +pub fn build_parquet_root_schema(schema: &ArrowSchema) -> MergeResult> { + let descriptor = ArrowSchemaConverter::new().convert(schema)?; + Ok(descriptor.root_schema_ptr()) } /// Returns column indices that exclude `__row_id__`, for use as a projection mask. @@ -101,7 +78,10 @@ impl ColumnMapping { for (target_idx, field) in target_schema.fields().iter().enumerate() { match source_schema.index_of(field.name()) { Ok(src_idx) => { - if is_identity && src_idx != target_idx { + if is_identity + && (src_idx != target_idx + || source_schema.field(src_idx).data_type() != field.data_type()) + { is_identity = false; } mapping.push(Some(src_idx)); @@ -130,7 +110,37 @@ impl ColumnMapping { let mut columns: Vec = Vec::with_capacity(self.mapping.len()); for (i, entry) in self.mapping.iter().enumerate() { match entry { - Some(src_idx) => columns.push(batch.column(*src_idx).clone()), + Some(src_idx) => { + let source = batch.column(*src_idx); + let target_field = &self.target_schema.fields()[i]; + if source.data_type() == target_field.data_type() { + columns.push(source.clone()); + } else if let DataType::List(child) = target_field.data_type() { + if source.data_type() != child.data_type() { + return Err(MergeError::Logic(format!( + "Cannot promote field '{}' from {:?} to {:?}", + target_field.name(), + source.data_type(), + target_field.data_type() + ))); + } + let offsets = + OffsetBuffer::new((0..=num_rows as i32).collect::>().into()); + columns.push(Arc::new(ListArray::new( + Arc::clone(child), + offsets, + source.clone(), + source.nulls().cloned(), + ))); + } else { + return Err(MergeError::Logic(format!( + "Cannot adapt field '{}' from {:?} to {:?}", + target_field.name(), + source.data_type(), + target_field.data_type() + ))); + } + } None => { let field = &self.target_schema.fields()[i]; columns.push(arrow::array::new_null_array(field.data_type(), num_rows)); 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/main/rust/tests/merge_list_column_tests.rs b/sandbox/plugins/parquet-data-format/src/main/rust/tests/merge_list_column_tests.rs new file mode 100644 index 0000000000000..2685ee887c148 --- /dev/null +++ b/sandbox/plugins/parquet-data-format/src/main/rust/tests/merge_list_column_tests.rs @@ -0,0 +1,416 @@ +/* + * 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. + */ + +//! Merge coverage for multi-valued (Arrow LIST / parquet repeated) columns. +//! +//! A LIST column breaks the "one leaf value per row" identity every flat column satisfies, so the +//! merge path needs its own coverage: the k-way merge slices batches by row, appends a fresh +//! `__row_id__`, and re-encodes each column through `compute_leaves`. All of that must keep a +//! repeated column's values grouped with their original row. + +use std::fs::File; +use std::sync::Arc; + +use arrow::array::*; +use arrow::datatypes::{DataType, Field, Schema}; +use opensearch_parquet_format::merge::{merge_sorted, merge_unsorted}; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use parquet::arrow::ArrowWriter; +use tempfile::tempdir; + +/// Schema: `id` (sort key, Int64) + `tags` (List) + `__row_id__`. +fn list_schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new( + "tags", + DataType::List(Arc::new(Field::new("element", DataType::Utf8, true))), + true, + ), + Field::new("__row_id__", DataType::Int64, false), + ])) +} + +/// Build a file where row i has `id = ids[i]` and `tags = rows[i]` (None ⇒ null list). +fn write_list_file(path: &str, ids: &[i64], rows: &[Option>]) { + assert_eq!(ids.len(), rows.len()); + let mut values: Vec> = Vec::new(); + let mut offsets: Vec = vec![0]; + let mut validity: Vec = Vec::new(); + for r in rows { + match r { + Some(v) => { + for s in v { + values.push(Some((*s).to_string())); + } + validity.push(true); + } + None => validity.push(false), + } + offsets.push(values.len() as i32); + } + let list = ListArray::new( + Arc::new(Field::new("element", DataType::Utf8, true)), + arrow::buffer::OffsetBuffer::new(offsets.into()), + Arc::new(StringArray::from(values)), + Some(arrow::buffer::NullBuffer::from(validity)), + ); + let schema = list_schema(); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Int64Array::from(ids.to_vec())) as ArrayRef, + Arc::new(list) as ArrayRef, + Arc::new(Int64Array::from((0..ids.len() as i64).collect::>())) as ArrayRef, + ], + ) + .unwrap(); + let file = File::create(path).unwrap(); + let mut w = ArrowWriter::try_new(file, schema, None).unwrap(); + w.write(&batch).unwrap(); + w.close().unwrap(); +} + +fn scalar_schema() -> Arc { + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("tags", DataType::Utf8, true), + Field::new("__row_id__", DataType::Int64, false), + ])) +} + +fn write_scalar_file(path: &str, ids: &[i64], rows: &[Option<&str>]) { + assert_eq!(ids.len(), rows.len()); + let schema = scalar_schema(); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int64Array::from(ids.to_vec())) as ArrayRef, + Arc::new(StringArray::from(rows.to_vec())) as ArrayRef, + Arc::new(Int64Array::from((0..ids.len() as i64).collect::>())) as ArrayRef, + ], + ) + .unwrap(); + let file = File::create(path).unwrap(); + let mut writer = ArrowWriter::try_new(file, schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); +} + +/// Read back `(id, tags)` pairs in file order. +fn read_pairs(path: &str) -> Vec<(i64, Option>)> { + let reader = ParquetRecordBatchReaderBuilder::try_new(File::open(path).unwrap()) + .unwrap() + .build() + .unwrap(); + let mut out = Vec::new(); + for batch in reader { + let batch = batch.unwrap(); + let ids = batch + .column(batch.schema().index_of("id").unwrap()) + .as_primitive::(); + let lists = batch + .column(batch.schema().index_of("tags").unwrap()) + .as_any() + .downcast_ref::() + .expect("tags must decode as a ListArray"); + for i in 0..batch.num_rows() { + let tags = if lists.is_null(i) { + None + } else { + let v = lists.value(i); + let s = v.as_any().downcast_ref::().unwrap(); + Some((0..s.len()).map(|j| s.value(j).to_string()).collect()) + }; + out.push((ids.value(i), tags)); + } + } + out +} + +/// Borrow an owned rows-of-strings structure as the `&str` shape `write_list_file` takes. +fn as_refs(rows: &[Option>]) -> Vec>> { + rows.iter() + .map(|r| r.as_ref().map(|v| v.iter().map(|s| s.as_str()).collect())) + .collect() +} + +fn v(items: &[&str]) -> Option> { + Some(items.iter().map(|s| (*s).to_string()).collect()) +} + +#[test] +fn unsorted_merge_preserves_list_values() { + let tmp = tempdir().unwrap(); + let a = tmp.path().join("a.parquet").to_string_lossy().to_string(); + let b = tmp.path().join("b.parquet").to_string_lossy().to_string(); + let out = tmp + .path() + .join("merged.parquet") + .to_string_lossy() + .to_string(); + + // Varying list lengths, a duplicate, an empty list, and a null list. + write_list_file( + &a, + &[1, 2, 3], + &[ + Some(vec!["beta", "alpha", "beta"]), + None, + Some(vec!["solo"]), + ], + ); + write_list_file(&b, &[4, 5], &[Some(vec![]), Some(vec!["x", "y"])]); + + merge_unsorted(&[a, b], &out, "merge-list-unsorted", 0).unwrap(); + + let pairs = read_pairs(&out); + assert_eq!(pairs.len(), 5, "row count must be preserved"); + assert_eq!(pairs[0], (1, v(&["beta", "alpha", "beta"]))); + assert_eq!(pairs[1].0, 2); + assert!(pairs[1].1.is_none() || pairs[1].1 == Some(vec![])); + assert_eq!(pairs[2], (3, v(&["solo"]))); + assert_eq!(pairs[3].0, 4); + assert_eq!(pairs[3].1, Some(vec![]), "empty list must stay empty"); + assert_eq!(pairs[4], (5, v(&["x", "y"]))); +} + +#[test] +fn unsorted_merge_promotes_scalar_values_to_singleton_lists() { + let tmp = tempdir().unwrap(); + let scalar = tmp + .path() + .join("scalar.parquet") + .to_string_lossy() + .to_string(); + let list = tmp + .path() + .join("list.parquet") + .to_string_lossy() + .to_string(); + let out = tmp + .path() + .join("merged.parquet") + .to_string_lossy() + .to_string(); + + write_scalar_file(&scalar, &[1, 2], &[Some("prod"), None]); + write_list_file( + &list, + &[3, 4], + &[Some(vec!["prod", "error"]), Some(vec!["solo"])], + ); + + merge_unsorted(&[scalar, list], &out, "merge-scalar-list", 0).unwrap(); + + assert_eq!( + read_pairs(&out), + vec![ + (1, v(&["prod"])), + (2, None), + (3, v(&["prod", "error"])), + (4, v(&["solo"])), + ] + ); +} + +#[test] +fn sorted_merge_keeps_list_values_with_their_row() { + let tmp = tempdir().unwrap(); + let a = tmp.path().join("a.parquet").to_string_lossy().to_string(); + let b = tmp.path().join("b.parquet").to_string_lossy().to_string(); + let out = tmp + .path() + .join("merged.parquet") + .to_string_lossy() + .to_string(); + + // Interleaving sort keys force the k-way merge to alternate between cursors, so a row-vs-value + // confusion in the LIST column would surface as values attached to the wrong id. + write_list_file( + &a, + &[1, 3, 5], + &[ + Some(vec!["one"]), + Some(vec!["three", "iii"]), + Some(vec!["five"]), + ], + ); + write_list_file( + &b, + &[2, 4, 6], + &[Some(vec!["two", "ii", "2"]), None, Some(vec!["six"])], + ); + + merge_sorted( + &[a, b], + &out, + "merge-list-sorted", + &["id".to_string()], + &[false], + &[false], + 0, + ) + .unwrap(); + + let pairs = read_pairs(&out); + assert_eq!(pairs.len(), 6); + // Sort order by id, and every row keeps exactly its own values. + assert_eq!(pairs[0], (1, v(&["one"]))); + assert_eq!(pairs[1], (2, v(&["two", "ii", "2"]))); + assert_eq!(pairs[2], (3, v(&["three", "iii"]))); + assert_eq!(pairs[3].0, 4); + assert!(pairs[3].1.is_none() || pairs[3].1 == Some(vec![])); + assert_eq!(pairs[4], (5, v(&["five"]))); + assert_eq!(pairs[5], (6, v(&["six"]))); +} + +#[test] +fn sorted_merge_on_list_column_as_sort_key_errors_cleanly() { + // heap.rs `get_sort_value` has no arm for List, so sorting BY a list column must surface a + // clean MergeError rather than panicking or silently producing garbage. + let tmp = tempdir().unwrap(); + let a = tmp.path().join("a.parquet").to_string_lossy().to_string(); + let b = tmp.path().join("b.parquet").to_string_lossy().to_string(); + let out = tmp + .path() + .join("merged.parquet") + .to_string_lossy() + .to_string(); + write_list_file(&a, &[1], &[Some(vec!["a"])]); + write_list_file(&b, &[2], &[Some(vec!["b"])]); + + let res = merge_sorted( + &[a, b], + &out, + "merge-list-sortkey", + &["tags".to_string()], + &[false], + &[false], + 0, + ); + match res { + Err(e) => { + let msg = e.to_string(); + println!("sorting by a LIST column errored as expected: {msg}"); + assert!( + msg.contains("Unsupported sort column type"), + "expected an explicit unsupported-sort-type error, got: {msg}" + ); + } + Ok(_) => panic!("sorting by a LIST column should not silently succeed"), + } +} + +/// Multi-batch cursors + deferred column decode. +/// +/// Two settings make this the adversarial case for a repeated column: +/// `merge_batch_size` small enough that each cursor spans several batches (so the k-way merge +/// crosses batch boundaries mid-file, exercising `advance`/`load_next_batch` and `take_slice`), and +/// `merge_deferred_column_threshold = 0` which forces the cursor's *deferred* mode — sort columns +/// and data columns are read through two independent parquet readers kept in lockstep by batch +/// index. If a LIST column's leaf desynchronised from the sort reader, the two would drift. +#[test] +fn sorted_merge_list_across_batches_in_deferred_mode() { + use opensearch_parquet_format::native_settings::NativeSettings; + use opensearch_parquet_format::writer::SETTINGS_STORE; + + let index = "merge-list-deferred"; + SETTINGS_STORE.insert( + index.to_string(), + NativeSettings { + merge_batch_size: Some(4), + merge_deferred_column_threshold: Some(0), + ..Default::default() + }, + ); + + let tmp = tempdir().unwrap(); + let a = tmp.path().join("a.parquet").to_string_lossy().to_string(); + let b = tmp.path().join("b.parquet").to_string_lossy().to_string(); + let out = tmp + .path() + .join("merged.parquet") + .to_string_lossy() + .to_string(); + + // 40 rows total, interleaved odd/even ids, with per-row list lengths that vary 0..3 so the + // value count is unrelated to the row count in every batch. + let ids_a: Vec = (0..20).map(|i| i * 2 + 1).collect(); + let ids_b: Vec = (0..20).map(|i| i * 2 + 2).collect(); + let owned_a: Vec>> = ids_a + .iter() + .enumerate() + .map(|(i, id)| match i % 4 { + 0 => None, + 1 => Some(vec![]), + 2 => Some(vec![format!("a{id}")]), + _ => Some(vec![format!("a{id}"), format!("a{id}b"), format!("a{id}c")]), + }) + .collect(); + let owned_b: Vec>> = ids_b + .iter() + .enumerate() + .map(|(i, id)| match i % 3 { + 0 => Some(vec![format!("b{id}"), format!("b{id}x")]), + 1 => None, + _ => Some(vec![format!("b{id}")]), + }) + .collect(); + write_list_file(&a, &ids_a, &as_refs(&owned_a)); + write_list_file(&b, &ids_b, &as_refs(&owned_b)); + + merge_sorted( + &[a, b], + &out, + index, + &["id".to_string()], + &[false], + &[false], + 0, + ) + .unwrap(); + + let pairs = read_pairs(&out); + assert_eq!(pairs.len(), 40, "all rows must survive the merge"); + + // Rebuild the expected (id → tags) mapping and check every row independently, so a single + // misattached value fails loudly with the offending id. + let mut expected: std::collections::HashMap>> = + std::collections::HashMap::new(); + for (id, tags) in ids_a.iter().zip(owned_a.iter()) { + expected.insert(*id, tags.clone()); + } + for (id, tags) in ids_b.iter().zip(owned_b.iter()) { + expected.insert(*id, tags.clone()); + } + + let mut prev_id = i64::MIN; + for (id, tags) in &pairs { + assert!( + *id > prev_id, + "output must be sorted by id, saw {id} after {prev_id}" + ); + prev_id = *id; + let want = expected + .remove(id) + .unwrap_or_else(|| panic!("unexpected id {id}")); + // A null list and an empty list are both legitimately readable as "no values". + let norm = |t: &Option>| t.clone().unwrap_or_default(); + assert_eq!( + norm(tags), + norm(&want), + "row id={id} lost or gained values in the merge" + ); + } + assert!( + expected.is_empty(), + "rows missing from output: {:?}", + expected.keys() + ); +} 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..897056fcbb96f 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,14 @@ 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.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 +577,138 @@ 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 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..67d36e8a0fa68 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,153 @@ 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"); + 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 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 testDerivedSourceCompanionFieldFollowsParentCardinality() { + // 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..1581ba3f7a669 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 KeywordFieldMapper keywordFieldMapper) { + if (keywordFieldMapper.fieldType().isMultiValued() == false) { + keywordFieldMapper.addMultiValueMappingUpdate(context); + } + context.documentInput().addField(keywordFieldMapper.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..9cfc563b62d85 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,16 @@ 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 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..11c0bd3f8667f 100644 --- a/server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java +++ b/server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java @@ -57,6 +57,7 @@ import org.opensearch.common.lucene.BytesRefs; import org.opensearch.common.lucene.Lucene; import org.opensearch.common.lucene.search.AutomatonQueries; +import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.Fuzziness; import org.opensearch.core.xcontent.XContentParser; import org.opensearch.index.IndexSettings; @@ -176,6 +177,16 @@ 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 = Parameter.boolParam("multi_value", true, m -> toType(m).multiValued, false) + .setMergeValidator((previous, next) -> previous == false || next); + private final IndexAnalyzers indexAnalyzers; private final boolean canConsumeRawValueForSource; @@ -209,6 +220,11 @@ Builder normalizer(String normalizerName) { return this; } + Builder multiValue(boolean multiValue) { + this.multiValue.setValue(multiValue); + return this; + } + Builder nullValue(String nullValue) { this.nullValue.setValue(nullValue); return this; @@ -236,7 +252,8 @@ protected List> getParameters() { normalizer, splitQueriesOnWhitespace, boost, - meta + meta, + multiValue ) ); parameters.addAll(pluginMappingParameters()); @@ -359,6 +376,7 @@ public KeywordFieldType(String name, FieldType fieldType, NamedAnalyzer normaliz setEagerGlobalOrdinals(builder.eagerGlobalOrdinals.getValue()); setIndexAnalyzer(normalizer); setBoost(builder.boost.getValue()); + setMultiValued(builder.multiValue.getValue()); this.ignoreAbove = builder.ignoreAbove.getValue(); this.nullValue = builder.nullValue.getValue(); this.useSimilarity = builder.useSimilarity.getValue(); @@ -846,6 +864,7 @@ private void checkToDisableCaching(QueryShardContext context) { private final boolean useSimilarity; private final String normalizerName; private final boolean splitQueriesOnWhitespace; + private final boolean multiValued; private final KeywordFieldType rawKeywordValueFieldType; private final IndexAnalyzers indexAnalyzers; @@ -874,6 +893,7 @@ protected KeywordFieldMapper( this.useSimilarity = builder.useSimilarity.getValue(); this.normalizerName = builder.normalizer.getValue(); this.splitQueriesOnWhitespace = builder.splitQueriesOnWhitespace.getValue(); + this.multiValued = builder.multiValue.getValue(); this.indexAnalyzers = builder.indexAnalyzers; this.canConsumeRawValueForSource = builder.canConsumeRawValueForSource; this.mappingPluginParameterValues = builder.pluginMappingParameterValues(); @@ -911,7 +931,18 @@ 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 cardinality or source reconstruction would lose values. + rawValueType.setMultiValued(multiValued); + return rawValueType; } return null; } @@ -957,6 +988,15 @@ protected void parseCreateField(ParseContext context) throws IOException { } } + void addMultiValueMappingUpdate(ParseContext context) { + if (fieldType().isMultiValued()) { + return; + } + Builder updateBuilder = (Builder) getMergeBuilder(); + updateBuilder.multiValue(true); + context.addDynamicMapper(updateBuilder.build(new BuilderContext(Settings.EMPTY, context.path()))); + } + @Override protected void parseCreateFieldForPluggableFormat(ParseContext context) throws IOException { String textValue = textValue(context); @@ -965,6 +1005,11 @@ protected void parseCreateFieldForPluggableFormat(ParseContext context) throws I } String value = parseKeyword(textValue); if (value != null) { + if (fieldType().isMultiValued() == false + && context.indexSettings().isPluggableDataFormatEnabled() + && context.documentInput().getFieldCount(fieldType().name()) > 0) { + addMultiValueMappingUpdate(context); + } context.documentInput().addField(fieldType(), value); } // For derived source: store raw value separately when normalizer/ignore_above alters it. 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..ec6509bd5b8de 100644 --- a/server/src/main/java/org/opensearch/index/mapper/MappedFieldType.java +++ b/server/src/main/java/org/opensearch/index/mapper/MappedFieldType.java @@ -97,6 +97,7 @@ public abstract class MappedFieldType { private float boost; private NamedAnalyzer indexAnalyzer; private boolean eagerGlobalOrdinals; + private boolean multiValued; /** * Capability map assigning each registered {@link DataFormat} to the set of capabilities it owns for this field type. @@ -478,6 +479,30 @@ 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 multiValued; + } + + /** + * Declares this field multi-valued for columnar data formats. Set during mapping build by + * mappers exposing the {@code multi_value} parameter. + * + * @opensearch.experimental + */ + @ExperimentalApi + public void setMultiValued(boolean multiValued) { + this.multiValued = multiValued; + } + @ExperimentalApi public Map> getCapabilityMap() { return capabilityMap; 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..d67cfd0604ab3 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,53 @@ public void testPluggableDataFormatDefaultKeyword() throws IOException { assertTrue("Expected keyword field captured with value 'test_value'", found); } + @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();