feat(parquet): Add adaptive keyword promotion

[linuxpi]

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.

Description

[Describe what this change achieves]

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 7d983f3)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Fix per-column WriterProperties path for LIST columns Relevant files: sandbox/plugins/parquet-data-format/src/main/rust/src/writer_properties_builder.rs Sub-PR theme: Merge-time scalar→LIST promotion and schema union Relevant files: sandbox/plugins/parquet-data-format/src/main/rust/src/merge/context.rs sandbox/plugins/parquet-data-format/src/main/rust/src/merge/schema.rs sandbox/plugins/parquet-data-format/src/main/rust/tests/merge_list_column_tests.rs Sub-PR theme: Server-side multi_value mapping parameter and adaptive promotion Relevant files: server/src/main/java/org/opensearch/index/mapper/KeywordFieldMapper.java server/src/main/java/org/opensearch/index/mapper/MappedFieldType.java server/src/main/java/org/opensearch/index/mapper/FilterFieldType.java server/src/main/java/org/opensearch/index/mapper/DocumentParser.java server/src/test/java/org/opensearch/index/mapper/KeywordFieldMapperTests.java

⚡ Recommended focus areas for review Syntax Error / Extra Closing Brace The validate method now has an extra closing brace before validateSortFieldsAreSingleValued is defined. The class appears to close with }} after validateFieldConfigurations(...) and then declares private static void validateSortFieldsAreSingleValued(...) outside any class. This will not compile. The validateSortFieldsAreSingleValued method needs to live inside the class body, not after its closing brace. } } /** * Rejects {@code index.sort.field} entries that are mapped {@code multi_value: true}. * <p> * An index sort needs one total order over rows, but a multi-valued cell has no canonical * scalar to order by — any of min/max/lexicographic would be a silent choice the user never * made. Lucene rejects index sorting on multi-valued fields for the same reason; failing here * keeps parity and turns what would otherwise be a merge-time failure (the native k-way merge * cannot compare LIST sort keys) into an immediate, actionable error at creation time. */ private static void validateSortFieldsAreSingleValued(MapperService mapperService, IndexSettings indexSettings) { for (String sortField : IndexSortConfig.INDEX_SORT_FIELD_SETTING.get(indexSettings.getSettings())) { MappedFieldType fieldType = mapperService.fieldType(sortField); if (fieldType != null && fieldType.isMultiValued()) { throw new IllegalArgumentException( "Cannot use field [" + sortField + "] in [index.sort.field]: the field is mapped [multi_value: true] and a " + "multi-valued field has no single value to sort on" ); } } } } Possible Issue registerEmptyMultiValueArray calls keywordFieldMapper.addMultiValueMappingUpdate(context) and then immediately calls context.documentInput().addField(keywordFieldMapper.fieldType(), List.of()). Since the mapping update is dynamic and only applied on retry, the current fieldType() still has isMultiValued() == false, meaning ParquetDocumentInput.addField may create a scalar FieldValuePair with value List.of() instead of an empty multi-valued pair. Verify that dynamic mapping update semantics discard the current parse (like the second-value promotion path) so the empty-list registration is only observed on the retry after the mapping is applied. 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()); } } Possible Issue When fieldType.isMultiValued() is true and value is an empty List, the pair is seeded via emptyMultiValued. But if a later addField call is made for the same field with a scalar value (e.g. mixed sources), the code path takes existing.addValue(value) which appends the scalar to the empty list — probably intended. However, if the first call arrives with a non-empty List value (rather than a scalar), it becomes a single "value" that IS the list, and subsequent adds also append raw values, causing a nested/mixed structure. Confirm the parser always calls addField once per element rather than passing a List for multi-valued fields. 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() + "]. Only keyword fields support automatic multi-value promotion." ); } existing.addValue(value); State Transition Concern On SchemaChangeRequiresWriterRotationException, the writer state is set to RETIRED_FLUSHABLE but mappingVersion was already bumped to newVersion before the exception. If the caller re-reads mappingVersion for routing/rotation logic, the retired writer will report the new mapping version despite not having reconciled its schema. Consider bumping mappingVersion only after reconcileSchema succeeds, or documenting this behavior. this.mappingVersion = newVersion; Schema schema = schemaSupplier.get(); logger.trace( "[Gen: {}] updateMappingVersion: schema from supplier has {} fields: {}", writerGeneration, schema.getFields().size(), schema.getFields().stream().map(f -> f.getName()).collect(java.util.stream.Collectors.joining(", ")) ); 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() ); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 7d983f3
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix stray closing brace breaking compilation The class body is closed with } immediately before the new method validateSortFieldsAreSingleValued, leaving that method and the trailing } outside any class. This will fail to compile. Remove the premature closing brace so the new method lives inside the class. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetIndexCreationValidator.java [68-71] schema, mapperService ); } } -} /** * Rejects {@code index.sort.field} entries that are mapped {@code multi_value: true}. Suggestion importance[1-10]: 10 __ Why: The diff clearly shows an extra } at line 69 that closes the class before the new validateSortFieldsAreSingleValued method, which would cause a compilation failure. This is a critical correctness issue.	High
	Prevent mixed scalar/list rows in same file Auto-promotion on the second value writes prior document values as a scalar column, but any subsequent document (in the same writer generation) then arrives with the field declared multi_value: true and expects a LIST cell — mixing scalar-written rows and list-written rows in the same Parquet file. Ensure promotion also triggers a writer rotation/schema fence, or restrict promotion to the very first value written into a fresh writer. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetDocumentInput.java [82-92] if (fieldType instanceof KeywordFieldMapper.KeywordFieldType) { + // NOTE: promotion mid-file requires the writer to rotate on the mapping update; + // otherwise earlier rows are scalar and later rows are LIST in the same file. existing.promoteToMultiValued(value); return; } throw new MapperParsingException( "Cannot accept multiple values for field: [" + fieldType.name() + "] of type: [" + fieldType.typeName() + "]. Only keyword fields support automatic multi-value promotion." ); Suggestion importance[1-10]: 6 __ Why: Valid concern about mid-file promotion creating mixed scalar/LIST rows in the same Parquet file, though the PR does add a schema-fence mechanism (SchemaChangeRequiresWriterRotationException) that may handle this via reconcileSchema. The suggested code only adds a comment, so the impact of the change itself is minimal.	Low
	Bound the recursive retry to avoid stack overflow The recursive re-entry into indexIntoEngine after retiring the writer has no depth bound. If updateMappingVersion keeps forcing retirement (e.g. persistent schema-fence exception on every new writer under the same mapping version), this recurses until StackOverflowError. Convert to an explicit retry loop with a bounded number of attempts, and fail the op if the bound is exceeded. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [711-715] Writer currentWriter = lockedWriter.get(); currentWriter.updateMappingVersion(mappingVersion); if (currentWriter.state() != WriterState.ACTIVE) { writerCheckedOut = retireWriterIfNeeded(lockedWriter); lockedWriter = null; - return indexIntoEngine(index, plan); + // Bounded retry — see indexIntoEngine caller for the attempt counter. + return indexIntoEngineWithRetry(index, plan, /*attempt*/ 1); } Suggestion importance[1-10]: 6 __ Why: Legitimate concern about unbounded recursion if writer retirement keeps repeating, which could cause StackOverflowError in pathological cases. However, in practice a fresh writer would typically match the current mapping version, making this an edge case.	Low
General	Use null-tolerant singleton list wrapper List.of(parseValue) throws NullPointerException if parseValue is a non-List null slipped in via a different code path (guarded here only against being outer-null). More importantly, List.of disallows null elements — safe currently, but brittle. Use Collections.singletonList (nullable-safe) to make the fallback robust. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ParquetField.java [137-148] - List<?> values = parseValue instanceof List<?> list ? list : List.of(parseValue); + List<?> values = parseValue instanceof List<?> list ? list : java.util.Collections.singletonList(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()); Suggestion importance[1-10]: 3 __ Why: Minor robustness improvement. The current List.of(parseValue) path would throw NPE for a scalar null value, though the outer null check handles the primary case. Low impact defensive change.	Low

---

Previous suggestions

Suggestions up to commit d3994e5

Category	Suggestion	Impact
Possible issue	Avoid registering empty array on scalar field type When an empty array triggers addMultiValueMappingUpdate, the scalar field type is passed to addField with List.of(). Since the field type is still scalar (isMultiValued() returns false), ParquetDocumentInput.addField will wrap the empty List as a scalar value rather than seeding an empty multi-valued pair. The mapping update takes effect only on retry; the current document should either be retried or the mapping-update path should ensure the empty-list is registered against the promoted field type. server/src/main/java/org/opensearch/index/mapper/DocumentParser.java [1249-1256] if (leaf instanceof KeywordFieldMapper keywordFieldMapper) { if (keywordFieldMapper.fieldType().isMultiValued() == false) { keywordFieldMapper.addMultiValueMappingUpdate(context); + // Field type is still scalar this pass; document will be reparsed after the mapping update publishes. + return; } context.documentInput().addField(keywordFieldMapper.fieldType(), List.of()); } else if (leaf instanceof FieldMapper fieldMapper && fieldMapper.fieldType().isMultiValued()) { context.documentInput().addField(fieldMapper.fieldType(), List.of()); } Suggestion importance[1-10]: 6 __ Why: Valid observation: when a mapping update is issued via addMultiValueMappingUpdate, the field type in hand is still scalar, so addField with List.of() will treat it as a scalar value. However, the actual behavior depends on the retry mechanism, and the suggested fix returns early which may not be entirely correct either. Moderate impact.	Low
General	Flatten List values for multi-valued fields For a declared multi-valued field, a non-empty List value (e.g., from DocumentParser.registerEmptyMultiValueArray in future variations, or any caller passing a pre-built list) will be stored as a single list-typed element inside the list rather than being flattened. Consider unwrapping a List value into individual elements to keep semantics consistent regardless of how the parser delivers values. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/writer/ParquetDocumentInput.java [70-76] if (fieldType.isMultiValued()) { - pair = value instanceof List<?> list && list.isEmpty() - ? FieldValuePair.emptyMultiValued(fieldType) - : FieldValuePair.multiValued(fieldType, value); + if (value instanceof List<?> list) { + if (list.isEmpty()) { + pair = FieldValuePair.emptyMultiValued(fieldType); + } else { + pair = FieldValuePair.multiValued(fieldType, list.get(0)); + for (int i = 1; i < list.size(); i++) { + pair.addValue(list.get(i)); + } + } + } else { + pair = FieldValuePair.multiValued(fieldType, value); + } } else { pair = new FieldValuePair(fieldType, value); } Suggestion importance[1-10]: 3 __ Why: The current parser reports one addField call per array element, so a pre-built List is not the expected input format. The suggestion is speculative and may not reflect an actual bug in the current code path.	Low
	Consider validating unmapped sort fields explicitly IndexSortConfig.INDEX_SORT_FIELD_SETTING returns a List per current API; ensure the iteration handles the case where the setting is unset (empty list) and where mapperService.fieldType(sortField) returns null (unmapped field) — the current code correctly null-checks, but consider raising an explicit error for unmapped sort fields rather than silently accepting them, since they would fail later anyway. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetIndexCreationValidator.java [43-47] +if (!isParquetIndex) { + return; +} +validateSortFieldsAreSingleValued(mapperService, indexSettings); Suggestion importance[1-10]: 2 __ Why: The suggestion's improved_code is identical to the existing_code — it only asks the author to consider a change without providing an actual improvement.	Low

[github-actions]

Persistent review updated to latest commit 7d983f3

[github-actions]

❌ Gradle check result for 7d983f3: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?