Dynamic Mapping support for Pluggable Data Formats

[rayshrey]

Description

Why dynamic mapping support is needed for pluggable formats

OpenSearch supports dynamic mapping — when a document contains a field not in the index mapping, the field is automatically added. This works transparently with Lucene because Lucene is inherently schema-on-write: each document can have any fields regardless of what previous documents had. There's no upfront schema declaration.

Parquet (and columnar formats in general) are fundamentally different. A Parquet file has a fixed schema declared at creation time. Every row in the file must conform to that schema. Once the native writer is initialized with a schema, it cannot accept rows with new columns. This means dynamic mapping — where new fields can appear at any time — requires explicit handling at
the engine level.

How it works at the DataFormatAwareEngine level

The engine maintains a pool of writers. Each writer tracks:

• Whether its schema can still evolve (isSchemaMutable)
• What mapping version it was created with (mappingVersion)

Normal indexing flow (no new fields):

1. Document arrives → engine reads current mapping version
2. Engine checks out a writer from the pool (predicate: writer is mutable OR its version matches)
3. Document is written to the writer
4. Writer is returned to the pool

When a new field arrives (dynamic mapping):

1. OpenSearch's bulk action detects the new field, sends mapping update to cluster manager
2. Cluster state updates → mapping version increments → MapperService updated
3. Document is retried with the new mapping in place
4. Engine reads the new mapping version → checks out a writer:
   - If a mutable writer exists → it's selected (schema evolves dynamically inside)
   - If only immutable writers exist with old versions → they're evicted from the available queue, a new writer is created with the fresh schema
5. Document is written successfully

This design is format-agnostic. Lucene writers are always mutable (isSchemaMutable = true), so they always pass the predicate and handle any document regardless of schema changes. The version tracking and eviction logic is invisible to Lucene — it just works.

How Parquet handles it

On the Parquet side, the key insight is that the native writer initialization (which locks the schema) must be deferred until we're certain the schema is complete.

Writer lifecycle:

1. Writer is created with an initial Arrow schema from the current mapping → but the native Parquet writer is NOT initialized yet (isSchemaMutable = true)
2. Documents arrive → if a document has a field not in the current Arrow schema, the field vector is dynamically added to the in-memory batch (VSR)
3. When the batch is full (rotation) or flush is triggered:
   - The batch is frozen (no more writes)
   - The native writer is initialized with the frozen batch's schema (which includes all dynamically added fields)
   - isSchemaMutable becomes false
   - The batch is written to the native writer
4. After initialization, any document with a newer mapping version will not match this writer → a new writer is created with the updated schema

Related Issues

#21587

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 120cd49)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

📝 TODO sections sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/fields/ArrowSchemaBuilder.java [40]: Get the mapping version while creating the schema

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue The maybeRotateActiveVSR() call is moved to after activeVSR.setRowCount(rowIndex + 1) in addDocument. If rotation triggers and the active VSR is frozen, the next document write will fail because activeVSR still references the frozen VSR. The rotation logic does not update managedVSR to point to a new active VSR after freezing the current one. maybeRotateActiveVSR(); Possible Issue In maybeRotateActiveVSR, if frozenVSR is not null, the code calls maybeInitializeWriter(frozenVSR) and then submits a background write task. However, if maybeInitializeWriter throws an IOException, the exception is not caught, and the background task is never submitted. This leaves the frozen VSR in an inconsistent state (frozen but not written). The subsequent code assumes the write task was submitted. maybeInitializeWriter(frozenVSR); Runnable writeTask = () -> { try { try (ArrowExport export = frozenVSR.exportToArrow()) { rowCount.add(frozenVSR.getRowCount()); writer.write(export.getArrayAddress(), export.getSchemaAddress()); } } catch (IOException e) { throw new RuntimeException(e); } Possible Issue The initialized flag is set to true immediately after calling RustBridge.createWriter, but if createWriter throws an IOException, the flag remains false. However, the exception propagates, so this is not a bug. The real issue is that write checks initialized == false and throws IllegalStateException, but the error message says "Writer not initialized" without indicating whether initialization failed or was never attempted. This can confuse debugging when initialization fails silently elsewhere. if (initialized == false) { throw new IllegalStateException("Writer not initialized: " + filePath); } RustBridge.write(filePath, arrayAddress, schemaAddress); Possible Issue The writer pool predicate h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion allows reusing a writer if its mapping version is greater than or equal to the current mapping version. However, if a writer's mapping version is strictly greater than the current mapping version (e.g., due to a concurrent update that was later rolled back or a race condition), this writer may contain fields that are not in the current mapping. Writing a document with the current mapping to this writer could result in missing fields or schema mismatches. DefaultLockableHolder<Writer<?>> lockedWriter = writerPool.getAndLock( h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion ); Possible Issue In pollAndDropIncompatible, the method scans the queue and removes incompatible entries. However, if the queue is modified concurrently (e.g., another thread adds an entry after the scan starts), the scan may miss the new entry. The method then falls through to the blocking phase, where it locks each stripe and scans again. If the new entry was added to a stripe that was already scanned in the blocking phase, it will be missed entirely. This can lead to starvation if all compatible entries are added after the scan starts. final int threadHash = Thread.currentThread().hashCode() & 0xFFFF; for (int i = 0; i < concurrency; ++i) { final int index = (threadHash + i) % concurrency; final Lock lock = locks[index]; final Queue<T> queue = queues[index]; if (lock.tryLock()) { try { T matched = scanAndDropIncompatible(queue, isCompatible, predicate); if (matched != null) return matched; } finally { lock.unlock(); } } } for (int i = 0; i < concurrency; ++i) { final int index = (threadHash + i) % concurrency; final Lock lock = locks[index]; final Queue<T> queue = queues[index]; lock.lock(); try { T matched = scanAndDropIncompatible(queue, isCompatible, predicate); if (matched != null) return matched; } finally { lock.unlock(); } }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to c1b964f

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Prevent concurrent field vector creation The dynamic field addition logic lacks synchronization, creating a race condition where multiple threads could simultaneously detect a missing field and attempt to add it. Synchronize the field-check-and-add block to ensure only one thread creates the vector. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [140-147] public void addDocument(ParquetDocumentInput doc) throws IOException { ManagedVSR activeVSR = managedVSR.get(); for (FieldValuePair pair : doc.getFinalInput()) { ... FieldVector vector = activeVSR.getVector(fieldType.name()); if (vector == null) { - Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); - activeVSR.addFieldVector(field); - vsrPool.updateSchema(activeVSR.getSchema()); + synchronized (activeVSR) { + vector = activeVSR.getVector(fieldType.name()); + if (vector == null) { + Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); + activeVSR.addFieldVector(field); + vsrPool.updateSchema(activeVSR.getSchema()); + } + } } parquetField.createField(fieldType, activeVSR, pair.getValue()); } ... maybeRotateActiveVSR(); } Suggestion importance[1-10]: 9 __ Why: This is a critical race condition fix. Without synchronization, multiple threads could simultaneously detect a missing field and attempt to add it, leading to duplicate field vectors or corrupted VSR state. The double-checked locking pattern in the improved code is the correct solution.	High
	Prevent Arrow memory leak The method creates a new VectorSchemaRoot but doesn't close the old one, potentially leaking Arrow memory buffers. Close the previous VSR before replacing it to prevent memory leaks. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/ManagedVSR.java [147-159] public void addFieldVector(Field field) { if (state.get() != VSRState.ACTIVE) { throw new IllegalStateException("Cannot add field to VSR in state: " + state.get()); } FieldVector vector = field.createVector(allocator); List<FieldVector> vectors = new ArrayList<>(vsr.getFieldVectors()); vectors.add(vector); List<Field> newFields = new ArrayList<>(vsr.getSchema().getFields()); newFields.add(field); int rowCount = vsr.getRowCount(); + VectorSchemaRoot oldVsr = vsr; vsr = new VectorSchemaRoot(newFields, vectors, rowCount); + oldVsr.close(); fields.put(field.getName(), vector); } Suggestion importance[1-10]: 9 __ Why: This is a critical memory leak fix. The old VectorSchemaRoot holds references to Arrow buffers that must be explicitly closed. Failing to close it before replacing with a new instance will leak native memory, which is particularly problematic in long-running processes with frequent schema evolution.	High
	Ensure thread-safe initialization Add thread-safety to the initialize method to prevent race conditions when multiple threads attempt initialization concurrently. Use compareAndSet on an AtomicBoolean instead of checking and setting a volatile boolean separately. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/bridge/NativeParquetWriter.java [59-65] +private final AtomicBoolean initialized = new AtomicBoolean(false); + public void initialize(String indexName, long schemaAddress, ParquetSortConfig sortConfig, long writerGeneration) throws IOException { - if (initialized) { + if (!initialized.compareAndSet(false, true)) { throw new IllegalStateException("Writer already initialized: " + filePath); } RustBridge.createWriter(filePath, indexName, schemaAddress, sortConfig, writerGeneration); - initialized = true; } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a thread-safety issue in the initialize method. Using AtomicBoolean.compareAndSet instead of checking and setting a volatile boolean separately prevents race conditions where multiple threads could pass the check simultaneously and attempt initialization.	Medium
	Fix mapping version comparison logic The predicate checks mappingVersion() >= mappingVersion which will always match writers with equal versions, potentially selecting a writer that hasn't been updated yet. Change to strict inequality to ensure only writers with newer mappings are reused. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [564-569] DefaultLockableHolder<Writer<?>> lockedWriter = writerPool.getAndLock( - h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion + h -> h.get().isSchemaMutable() || h.get().mappingVersion() > mappingVersion ); try { currentWriter = lockedWriter.get(); - currentWriter.updateMappingVersion(newVersion); + currentWriter.updateMappingVersion(mappingVersion); Suggestion importance[1-10]: 7 __ Why: The suggestion identifies a subtle logic issue where >= allows selecting writers with equal mapping versions, which may not have been updated yet. However, the subsequent updateMappingVersion call should handle this case, so the issue is moderate rather than critical.	Medium

---

Previous suggestions

Suggestions up to commit 120cd49

Category	Suggestion	Impact
Possible issue	Prevent inconsistent data from dynamic fields When adding a field vector dynamically, the new vector is created but not populated with values for existing rows. This creates inconsistent data where existing rows have null/unset values for the new field. Consider either rejecting dynamic field additions when rowCount > 0, or explicitly setting null values for existing rows in the new vector to maintain data consistency. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/ManagedVSR.java [141-159] public void addFieldVector(Field field) { if (state.get() != VSRState.ACTIVE) { throw new IllegalStateException("Cannot add field to VSR in state: " + state.get()); + } + if (vsr.getRowCount() > 0) { + throw new IllegalStateException("Cannot add field to VSR with existing rows: " + vsr.getRowCount()); } FieldVector vector = field.createVector(allocator); List<FieldVector> vectors = new ArrayList<>(vsr.getFieldVectors()); vectors.add(vector); List<Field> newFields = new ArrayList<>(vsr.getSchema().getFields()); newFields.add(field); int rowCount = vsr.getRowCount(); vsr = new VectorSchemaRoot(newFields, vectors, rowCount); fields.put(field.getName(), vector); } Suggestion importance[1-10]: 8 __ Why: This is a valid concern about data consistency. Adding fields dynamically when rowCount > 0 creates rows with null values for the new field in existing rows. The suggestion to reject dynamic field additions when rows exist is reasonable and would prevent inconsistent data states.	Medium
	Prevent schema mismatch after writer initialization The dynamic field addition logic modifies the active VSR's schema during document indexing, but this happens after the native writer may have been initialized with a different schema. If maybeInitializeWriter is called with a VSR that has a different schema than subsequent VSRs, the native writer will reject batches with mismatched schemas. Consider initializing the writer lazily only after all dynamic fields for the first batch are known, or validate schema consistency before writing. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [124-156] public void addDocument(ParquetDocumentInput doc) throws IOException { ManagedVSR activeVSR = managedVSR.get(); + boolean schemaChanged = false; for (FieldValuePair pair : doc.getFinalInput()) { MappedFieldType fieldType = pair.getFieldType(); ParquetField parquetField = ArrowFieldRegistry.getParquetField(fieldType.typeName()); if (parquetField == null) { continue; } - // Dynamic field vector addition: create vector if not present in VSR FieldVector vector = activeVSR.getVector(fieldType.name()); if (vector == null) { + if (writer.isInitialized()) { + throw new IOException("Cannot add field after writer initialization: " + fieldType.name()); + } Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); activeVSR.addFieldVector(field); - // Update pool schema so future VSRs include this field vsrPool.updateSchema(activeVSR.getSchema()); + schemaChanged = true; } parquetField.createField(fieldType, activeVSR, pair.getValue()); } ... maybeRotateActiveVSR(); } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a potential issue where dynamic fields are added after the native writer is initialized, which could cause schema mismatches. However, the PR's design intentionally supports lazy initialization (writer is initialized on first write), so the suggested check may be too restrictive. The concern is valid but the solution needs refinement.	Medium
General	Validate schema consistency before writer initialization The writer initialization uses the schema from the first VSR that gets written, but if dynamic fields are added to subsequent VSRs, those fields won't be in the native writer's schema. This will cause write failures when the native writer receives batches with fields it doesn't recognize. Consider deferring writer initialization until the schema is stable, or implementing schema evolution support in the native writer. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [247-255] private void maybeInitializeWriter(ManagedVSR vsr) throws IOException { if (writer.isInitialized() == false) { + Schema currentSchema = vsr.getSchema(); + Schema poolSchema = vsrPool.getSchema(); + if (!currentSchema.equals(poolSchema)) { + throw new IOException("VSR schema mismatch with pool schema during writer initialization"); + } String indexName = indexSettings.getIndex().getName(); ParquetSortConfig sortConfig = new ParquetSortConfig(indexSettings); try (ArrowSchema schema = vsr.exportSchema()) { writer.initialize(indexName, schema.memoryAddress(), sortConfig, writerGeneration); } } } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that the writer is initialized with the first VSR's schema, which may not include fields added dynamically later. The proposed validation would catch schema mismatches early. However, the PR's design relies on vsrPool.updateSchema() to propagate schema changes, so the check may be redundant if that mechanism works correctly.	Medium
	Fix mapping version comparison logic The predicate checks if the writer's mapping version is greater than or equal to the current mapping version, but this allows using a writer with a newer mapping version than the document being indexed. This could lead to schema mismatches if the mapping was updated between writer creation and document indexing. The predicate should use strict equality or reject writers with mismatched versions. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [562-569] Writer currentWriter = null; long mappingVersion = currentMappingVersion(); DefaultLockableHolder<Writer<?>> lockedWriter = writerPool.getAndLock( - h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion + h -> h.get().isSchemaMutable() || h.get().mappingVersion() == mappingVersion ); try { currentWriter = lockedWriter.get(); + if (currentWriter.mappingVersion() != mappingVersion && !currentWriter.isSchemaMutable()) { + throw new IllegalStateException("Writer mapping version mismatch"); + } currentWriter.updateMappingVersion(mappingVersion); ... } Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid point about the predicate allowing writers with newer mapping versions. However, the >= comparison is intentional to allow reusing writers that have already been updated to newer mappings. The additional check in the improved code adds safety but may be overly strict given the updateMappingVersion call that follows.	Low

Suggestions up to commit 0cc6f72

Category	Suggestion	Impact
Possible issue	Fix writer selection predicate logic The predicate rejects writers with mappingVersion < currentMappingVersion, but this causes all existing writers to be rejected when the mapping is updated, forcing creation of new writers. This defeats the purpose of writer pooling and can lead to excessive writer churn during dynamic mapping updates. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [557-560] DefaultLockableHolder<Writer<?>> lockedWriter = writerPool.getAndLock( - h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion + h -> h.get().isSchemaMutable() && h.get().mappingVersion() < mappingVersion + ? false // Reject immutable writers with stale schema + : true // Accept mutable writers (can evolve) or writers already at current version ); Suggestion importance[1-10]: 9 __ Why: This identifies a critical logic error in the writer selection predicate. The current implementation using OR (||) incorrectly accepts writers with stale schemas as long as they're mutable, when it should reject immutable writers with outdated schemas. The suggested AND-based logic correctly ensures only compatible writers are selected.	High
	Ensure schema completeness before rotation The dynamic field addition logic modifies the active VSR's schema during document indexing, but this happens after the native writer may have already been initialized with a fixed schema. If maybeRotateActiveVSR() triggers a write before all fields are added, the native writer will be initialized with an incomplete schema, causing subsequent writes with new fields to fail. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [124-156] public void addDocument(ParquetDocumentInput doc) throws IOException { ManagedVSR activeVSR = managedVSR.get(); + // First pass: ensure all dynamic fields exist in VSR before any rotation for (FieldValuePair pair : doc.getFinalInput()) { MappedFieldType fieldType = pair.getFieldType(); ParquetField parquetField = ArrowFieldRegistry.getParquetField(fieldType.typeName()); if (parquetField == null) { continue; } - // Dynamic field vector addition: create vector if not present in VSR FieldVector vector = activeVSR.getVector(fieldType.name()); if (vector == null) { Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); activeVSR.addFieldVector(field); - // Update pool schema so future VSRs include this field vsrPool.updateSchema(activeVSR.getSchema()); } - parquetField.createField(fieldType, activeVSR, pair.getValue()); + } + // Second pass: populate field values + for (FieldValuePair pair : doc.getFinalInput()) { + MappedFieldType fieldType = pair.getFieldType(); + ParquetField parquetField = ArrowFieldRegistry.getParquetField(fieldType.typeName()); + if (parquetField != null) { + parquetField.createField(fieldType, activeVSR, pair.getValue()); + } } ... maybeRotateActiveVSR(); } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a critical timing issue where maybeRotateActiveVSR() could trigger a write with an incomplete schema if called between field additions. The two-pass approach ensures all dynamic fields are added before rotation, preventing schema mismatch errors in the native writer.	Medium
	Add synchronization for schema updates The updateSchema method updates the pool's schema without any synchronization, creating a race condition where concurrent threads may see inconsistent schema states during VSR creation. This can lead to newly created VSRs having incomplete schemas if updateSchema is called while createNewVSR is executing. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRPool.java [173-182] -public void updateSchema(Schema newSchema) { +public synchronized void updateSchema(Schema newSchema) { this.schema = newSchema; } +private synchronized ManagedVSR createNewVSR() { + String vsrId = poolId + "-vsr-" + vsrCounter.incrementAndGet(); + BufferAllocator allocator = bufferPool.createChildAllocator(vsrId); + return new ManagedVSR(vsrId, schema, allocator); +} + Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a potential race condition between updateSchema and createNewVSR. However, the severity is moderate because the schema field is volatile, providing visibility guarantees. Adding synchronization would provide stronger consistency guarantees and prevent potential issues with concurrent schema updates during VSR creation.	Medium
	Use pool schema for writer initialization The writer initialization uses the schema from the VSR at the time of first write, but if dynamic fields are added to subsequent VSRs after rotation, those fields won't be in the native writer's schema. This creates a schema mismatch between the Java-side VSR and the Rust-side writer, causing write failures or data corruption. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [247-255] private void maybeInitializeWriter(ManagedVSR vsr) throws IOException { if (writer.isInitialized() == false) { String indexName = indexSettings.getIndex().getName(); ParquetSortConfig sortConfig = new ParquetSortConfig(indexSettings); - try (ArrowSchema schema = vsr.exportSchema()) { + // Use the pool's current schema (which includes all dynamically added fields) + // rather than the VSR's schema to ensure consistency + Schema currentSchema = vsrPool.getSchema(); + try (ArrowSchema schema = ArrowSchema.allocateNew(allocator)) { + Data.exportSchema(allocator, currentSchema, null, schema); writer.initialize(indexName, schema.memoryAddress(), sortConfig, writerGeneration); } } } Suggestion importance[1-10]: 3 __ Why: While the concern about schema consistency is valid, the suggested solution introduces a dependency on vsrPool.getSchema() which doesn't exist in the current implementation. The current approach of using the VSR's schema at initialization time is correct because the pool schema is updated synchronously when fields are added to the active VSR.	Low

Suggestions up to commit 0cc6f72

Category	Suggestion	Impact
Possible issue	Ensure schema completeness before rotation The dynamic field addition logic modifies the VSR schema during document indexing, but this happens after the native writer may have been initialized with a fixed schema. If maybeRotateActiveVSR() triggers a write before all fields are added, the native writer will be initialized with an incomplete schema, causing subsequent field additions to fail or be silently dropped. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [124-156] public void addDocument(ParquetDocumentInput doc) throws IOException { ManagedVSR activeVSR = managedVSR.get(); + // First pass: ensure all fields exist in the VSR before writing for (FieldValuePair pair : doc.getFinalInput()) { MappedFieldType fieldType = pair.getFieldType(); ParquetField parquetField = ArrowFieldRegistry.getParquetField(fieldType.typeName()); if (parquetField == null) { continue; } - // Dynamic field vector addition: create vector if not present in VSR FieldVector vector = activeVSR.getVector(fieldType.name()); if (vector == null) { Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); activeVSR.addFieldVector(field); - // Update pool schema so future VSRs include this field vsrPool.updateSchema(activeVSR.getSchema()); } - parquetField.createField(fieldType, activeVSR, pair.getValue()); + } + // Second pass: populate field values + for (FieldValuePair pair : doc.getFinalInput()) { + MappedFieldType fieldType = pair.getFieldType(); + ParquetField parquetField = ArrowFieldRegistry.getParquetField(fieldType.typeName()); + if (parquetField != null) { + parquetField.createField(fieldType, activeVSR, pair.getValue()); + } } ... maybeRotateActiveVSR(); } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a critical race condition where maybeRotateActiveVSR() could trigger a write with an incomplete schema if rotation happens mid-document. The two-pass approach ensures all fields are added before values are populated, preventing schema mismatches in the native writer.	Medium
	Use pool schema for writer initialization The writer initialization uses the schema from the VSR at the time of first write, but if dynamic fields are added to subsequent documents before the first rotation, those fields won't be included in the native writer's schema. This creates a mismatch between the Java-side VSR schema and the native writer's schema. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [247-255] private void maybeInitializeWriter(ManagedVSR vsr) throws IOException { if (writer.isInitialized() == false) { + // Ensure we use the latest schema from the pool, not just the current VSR + Schema latestSchema = vsrPool.getSchema(); String indexName = indexSettings.getIndex().getName(); ParquetSortConfig sortConfig = new ParquetSortConfig(indexSettings); - try (ArrowSchema schema = vsr.exportSchema()) { + try (ArrowSchema schema = ArrowSchema.allocateNew(allocator)) { + Data.exportSchema(allocator, latestSchema, null, schema); writer.initialize(indexName, schema.memoryAddress(), sortConfig, writerGeneration); } } } Suggestion importance[1-10]: 3 __ Why: The suggestion addresses a potential schema mismatch, but the proposed solution introduces complexity by requiring VSRPool to expose its schema and manually exporting it. The current implementation using vsr.exportSchema() is simpler and the schema should be consistent if the two-pass approach from suggestion 1 is adopted.	Low
General	Handle predicate exceptions during scan The method removes incompatible entries during iteration, but if the predicate throws an exception (e.g., due to corrupted state), the queue is left in an inconsistent state with some entries removed and others not. Consider wrapping the predicate calls in exception handling to ensure atomicity. libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java [131-143] private T scanAndDropIncompatible(Queue<T> queue, Predicate<T> isCompatible, Predicate<T> predicate) { Iterator<T> it = queue.iterator(); while (it.hasNext()) { T entry = it.next(); - if (isCompatible.test(entry) == false) { - it.remove(); - } else if (predicate.test(entry)) { - it.remove(); - return entry; + try { + if (isCompatible.test(entry) == false) { + it.remove(); + } else if (predicate.test(entry)) { + it.remove(); + return entry; + } + } catch (Exception e) { + // Log and skip entry to avoid partial queue corruption + logger.warn("Predicate evaluation failed for entry, skipping", e); } } return null; } Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that predicate exceptions could leave the queue in an inconsistent state. However, silently catching and logging exceptions may hide bugs. A better approach would be to fail fast or provide a configurable exception handler. The suggestion is valid but the implementation could be improved.	Low
	Require schema mutability for version match The predicate allows writers with mappingVersion >= mappingVersion, but this creates a race condition: if a writer's mapping version equals the current version, it may have been created before the latest mapping update was applied. The pool should reject writers with stale schemas to ensure consistency. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [557-560] long mappingVersion = currentMappingVersion(); DefaultLockableHolder<Writer<?>> lockedWriter = writerPool.getAndLock( - h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion + h -> h.get().isSchemaMutable() && h.get().mappingVersion() >= mappingVersion ); Suggestion importance[1-10]: 2 __ Why: The suggestion misunderstands the predicate logic. The || operator is correct: we want writers that are either schema-mutable (can accept new fields) or have a mapping version >= current (already have the latest schema). Changing to && would incorrectly reject immutable writers with the correct schema version.	Low

Suggestions up to commit 0cc6f72

Category	Suggestion	Impact
Possible issue	Initialize new vectors with null values When adding a field vector dynamically, the new vector is created but not initialized with null values for existing rows. This creates inconsistent row counts across vectors. The new vector should be pre-filled with null values up to the current row count to maintain consistency across all columns. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/ManagedVSR.java [147-159] public void addFieldVector(Field field) { if (state.get() != VSRState.ACTIVE) { throw new IllegalStateException("Cannot add field to VSR in state: " + state.get()); } FieldVector vector = field.createVector(allocator); + int rowCount = vsr.getRowCount(); + vector.setValueCount(rowCount); List<FieldVector> vectors = new ArrayList<>(vsr.getFieldVectors()); vectors.add(vector); List<Field> newFields = new ArrayList<>(vsr.getSchema().getFields()); newFields.add(field); - int rowCount = vsr.getRowCount(); vsr = new VectorSchemaRoot(newFields, vectors, rowCount); fields.put(field.getName(), vector); } Suggestion importance[1-10]: 9 __ Why: This is a critical bug fix. When adding a field vector dynamically, the new vector must be initialized with null values for existing rows to maintain consistency across all columns. Without this, the row counts will be inconsistent, leading to data corruption.	High
	Validate writer state before schema changes The dynamic field addition logic modifies the active VSR's schema during document indexing, but this happens after the native writer may have been initialized with a different schema. If the writer is already initialized when a new field is added, the schema mismatch will cause write failures. Consider validating that the writer is not yet initialized before allowing dynamic field additions, or implement a schema evolution mechanism in the native writer. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [124-148] public void addDocument(ParquetDocumentInput doc) throws IOException { ManagedVSR activeVSR = managedVSR.get(); for (FieldValuePair pair : doc.getFinalInput()) { MappedFieldType fieldType = pair.getFieldType(); ParquetField parquetField = ArrowFieldRegistry.getParquetField(fieldType.typeName()); if (parquetField == null) { continue; } - // Dynamic field vector addition: create vector if not present in VSR FieldVector vector = activeVSR.getVector(fieldType.name()); if (vector == null) { + if (writer.isInitialized()) { + throw new IOException("Cannot add field '" + fieldType.name() + "' after writer initialization"); + } Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); activeVSR.addFieldVector(field); - // Update pool schema so future VSRs include this field vsrPool.updateSchema(activeVSR.getSchema()); } parquetField.createField(fieldType, activeVSR, pair.getValue()); } ... maybeRotateActiveVSR(); } Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a critical race condition where dynamic fields could be added after the native writer is initialized with a fixed schema, causing write failures. The improved code adds proper validation to prevent schema modifications after initialization.	Medium
	Fix mapping version comparison logic The predicate checks if the writer's mapping version is greater than or equal to the current mapping version, but this allows using a writer with a newer mapping version than the current one. This could lead to indexing documents with an outdated schema. The condition should ensure the writer's mapping version matches exactly or is older (and mutable), not newer. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [557-569] DefaultLockableHolder<Writer<?>> lockedWriter = writerPool.getAndLock( - h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion + h -> h.get().isSchemaMutable() || h.get().mappingVersion() == mappingVersion ); try { currentWriter = lockedWriter.get(); currentWriter.updateMappingVersion(mappingVersion); Suggestion importance[1-10]: 7 __ Why: The suggestion identifies a potential issue with the mapping version comparison logic. However, the current logic (>=) may be intentional to allow reusing writers with newer schemas. The suggestion to use exact equality (==) is safer but may reduce writer reuse efficiency.	Medium
General	Handle predicate exceptions during iteration The method removes incompatible entries while iterating, but if an exception occurs during isCompatible.test() or predicate.test(), the queue may be left in an inconsistent state with some entries removed. Consider wrapping the removal logic to ensure atomicity or document that callers must handle exceptions appropriately. libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java [131-143] private T scanAndDropIncompatible(Queue<T> queue, Predicate<T> isCompatible, Predicate<T> predicate) { Iterator<T> it = queue.iterator(); while (it.hasNext()) { T entry = it.next(); - if (isCompatible.test(entry) == false) { - it.remove(); - } else if (predicate.test(entry)) { - it.remove(); - return entry; + try { + if (isCompatible.test(entry) == false) { + it.remove(); + } else if (predicate.test(entry)) { + it.remove(); + return entry; + } + } catch (Exception e) { + // Re-throw to preserve exception semantics, but queue may be partially modified + throw e; } } return null; } Suggestion importance[1-10]: 4 __ Why: While the suggestion correctly identifies that exceptions during predicate evaluation could leave the queue in an inconsistent state, the proposed solution (wrapping in try-catch and re-throwing) doesn't actually solve the problem. The queue would still be partially modified. A better approach would be to document this behavior or implement transactional semantics.	Low

Suggestions up to commit 0cc6f72

Category	Suggestion	Impact
Possible issue	Prevent duplicate field additions Adding a field with a name that already exists will create a duplicate entry in the schema and vectors list, but overwrite the previous entry in the fields map. This can lead to schema inconsistencies. Check if the field already exists before adding it to prevent duplicates. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/ManagedVSR.java [147-159] public void addFieldVector(Field field) { if (state.get() != VSRState.ACTIVE) { throw new IllegalStateException("Cannot add field to VSR in state: " + state.get()); } + if (fields.containsKey(field.getName())) { + return; + } FieldVector vector = field.createVector(allocator); - List<FieldVector> vectors = new ArrayList<>(vsr.getFieldVectors()); - vectors.add(vector); - List<Field> newFields = new ArrayList<>(vsr.getSchema().getFields()); - newFields.add(field); - int rowCount = vsr.getRowCount(); - vsr = new VectorSchemaRoot(newFields, vectors, rowCount); - fields.put(field.getName(), vector); + ... } Suggestion importance[1-10]: 6 __ Why: Valid concern about duplicate field additions causing schema inconsistencies. The suggested check if (fields.containsKey(field.getName())) would prevent duplicates. However, the fix should probably throw an exception or log a warning rather than silently returning, to make the duplicate attempt visible.	Low
	Update pool schema before adding field The schema update happens after adding a field to the active VSR, but before rotation. If rotation occurs immediately after (via maybeRotateActiveVSR), the pool's schema might not reflect the newly added field when creating the next VSR. Move vsrPool.updateSchema before activeVSR.addFieldVector to ensure the pool schema is updated atomically with the field addition. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [140-146] FieldVector vector = activeVSR.getVector(fieldType.name()); if (vector == null) { Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); + vsrPool.updateSchema(activeVSR.getSchema()); activeVSR.addFieldVector(field); - // Update pool schema so future VSRs include this field - vsrPool.updateSchema(activeVSR.getSchema()); } Suggestion importance[1-10]: 3 __ Why: The suggestion identifies a potential race condition but the proposed fix is incorrect. Updating the pool schema before adding the field to the active VSR would mean the pool schema includes a field that doesn't exist in the current VSR yet. The current order is correct: add field to VSR first, then update pool schema so future VSRs include it.	Low
General	Retrieve vector after dynamic addition After dynamically adding a field vector, parquetField.createField is called immediately. However, the newly created vector might not be properly initialized with the correct row count, potentially causing index out of bounds errors when writing to positions beyond 0. Verify that the new vector's capacity matches the current VSR row count before writing. sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java [140-148] -for (FieldValuePair pair : doc.getFinalInput()) { - MappedFieldType fieldType = pair.getFieldType(); - ParquetField parquetField = ArrowFieldRegistry.getParquetField(fieldType.typeName()); - if (parquetField == null) { - continue; - } - // Dynamic field vector addition: create vector if not present in VSR - FieldVector vector = activeVSR.getVector(fieldType.name()); - if (vector == null) { - Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); - activeVSR.addFieldVector(field); - // Update pool schema so future VSRs include this field - vsrPool.updateSchema(activeVSR.getSchema()); - } - parquetField.createField(fieldType, activeVSR, pair.getValue()); +FieldVector vector = activeVSR.getVector(fieldType.name()); +if (vector == null) { + Field field = new Field(fieldType.name(), parquetField.getFieldType(), null); + vsrPool.updateSchema(activeVSR.getSchema()); + activeVSR.addFieldVector(field); + vector = activeVSR.getVector(fieldType.name()); } +parquetField.createField(fieldType, activeVSR, pair.getValue()); Suggestion importance[1-10]: 4 __ Why: The suggestion raises a valid concern about vector initialization but the proposed fix (retrieving the vector again) doesn't address the underlying issue. The newly created vector is already accessible via activeVSR.getVector() after addFieldVector(). The real concern should be whether parquetField.createField() properly handles writing to a newly created vector at the current row index.	Low
	Accept immutable writers with matching version The predicate rejects writers with immutable schemas even if their mapping version matches. This can cause unnecessary writer creation when the schema hasn't actually changed. Consider accepting immutable writers if their mapping version equals the current version, not just when it's greater. server/src/main/java/org/opensearch/index/engine/DataFormatAwareEngine.java [557-560] long mappingVersion = currentMappingVersion(); DefaultLockableHolder<Writer<?>> lockedWriter = writerPool.getAndLock( - h -> h.get().isSchemaMutable() || h.get().mappingVersion() >= mappingVersion + h -> h.get().isSchemaMutable() || h.get().mappingVersion() == mappingVersion ); Suggestion importance[1-10]: 2 __ Why: The suggestion misunderstands the predicate logic. The condition h.get().mappingVersion() >= mappingVersion already accepts writers with matching versions (when == is true). Changing >= to == would reject writers with newer mapping versions, which is incorrect. The current implementation is correct.	Low

[github-actions]

❌ Gradle check result for bb21940: 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?

[mgodwan]

1. Lets add concurrency tests.
2. Can we enable dynamic mapping indexing tests which already exists for DFAE?

[github-actions]

Persistent review updated to latest commit be2a156

[github-actions]

❌ Gradle check result for be2a156: 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?

[github-actions]

Persistent review updated to latest commit 4f80641

[github-actions]

✅ Gradle check result for 4f80641: SUCCESS

[codecov]

Codecov Report

❌ Patch coverage is 90.90909% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.45%. Comparing base (446a1c9) to head (c1b964f).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...ensearch/common/queue/LockableConcurrentQueue.java	83.33%	0 Missing and 1 partial ⚠️
...opensearch/index/engine/DataFormatAwareEngine.java	83.33%	0 Missing and 1 partial ⚠️
...arch/index/engine/dataformat/RowIdAwareWriter.java	75.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21444      +/-   ##
============================================
- Coverage     73.49%   73.45%   -0.04%     
- Complexity    74624    74646      +22     
============================================
  Files          5980     5980              
  Lines        338825   338839      +14     
  Branches      48857    48860       +3     
============================================
- Hits         249010   248890     -120     
- Misses        70041    70111      +70     
- Partials      19774    19838      +64     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

Persistent review updated to latest commit c066777

[github-actions]

❌ Gradle check result for c066777: null

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?

[github-actions]

Persistent review updated to latest commit 8b6c69d

[github-actions]

❌ Gradle check result for 8b6c69d: 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?

[github-actions]

Persistent review updated to latest commit edb8baa

[github-actions]

❌ Gradle check result for edb8baa: 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?

[github-actions]

Persistent review updated to latest commit 044ebd1

[github-actions]

❌ Gradle check result for 044ebd1: 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?

[github-actions]

Persistent review updated to latest commit d48609e

[github-actions]

Persistent review updated to latest commit 2f05b57

[github-actions]

Persistent review updated to latest commit 0cc6f72

[github-actions]

❌ Gradle check result for 0cc6f72: 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?

[github-actions]

Persistent review updated to latest commit 0cc6f72

[github-actions]

❌ Gradle check result for 0cc6f72: 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?

[github-actions]

❌ Gradle check result for 0cc6f72: 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?

[github-actions]

Persistent review updated to latest commit 120cd49

[github-actions]

❌ Gradle check result for 120cd49: 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?

[github-actions]

❕ Gradle check result for c1b964f: UNSTABLE

Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.