Dynamic Mapping support for Pluggable Data formats

Signed-off-by: rayshrey <rayshrey@amazon.com>

Author: rayshrey

libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java

@@ -31,7 +31,6 @@ public final class ConcurrentQueue<T> {
     private final int concurrency;
     private final Lock[] locks;
     private final Queue<T>[] queues;
-    private final Supplier<Queue<T>> queueSupplier;
 
     ConcurrentQueue(Supplier<Queue<T>> queueSupplier, int concurrency) {
         if (concurrency < MIN_CONCURRENCY || concurrency > MAX_CONCURRENCY) {
@@ -40,7 +39,6 @@ public final class ConcurrentQueue<T> {
             );
         }
         this.concurrency = concurrency;
-        this.queueSupplier = queueSupplier;
         locks = new Lock[concurrency];
         @SuppressWarnings({ "rawtypes", "unchecked" })
         Queue<T>[] queues = new Queue[concurrency];
@@ -81,21 +79,19 @@ void add(T entry) {
     }
 
     T poll(Predicate<T> predicate) {
+        return pollAndDropIncompatible(e -> true, predicate);
+    }
+
+    T pollAndDropIncompatible(Predicate<T> isCompatible, Predicate<T> predicate) {
         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 {
-                    Iterator<T> it = queue.iterator();
-                    while (it.hasNext()) {
-                        T entry = it.next();
-                        if (predicate.test(entry)) {
-                            it.remove();
-                            return entry;
-                        }
-                    }
+                    T matched = scanAndDropIncompatible(queue, isCompatible, predicate);
+                    if (matched != null) return matched;
                 } finally {
                     lock.unlock();
                 }
@@ -107,14 +103,8 @@ T poll(Predicate<T> predicate) {
             final Queue<T> queue = queues[index];
             lock.lock();
             try {
-                Iterator<T> it = queue.iterator();
-                while (it.hasNext()) {
-                    T entry = it.next();
-                    if (predicate.test(entry)) {
-                        it.remove();
-                        return entry;
-                    }
-                }
+                T matched = scanAndDropIncompatible(queue, isCompatible, predicate);
+                if (matched != null) return matched;
             } finally {
                 lock.unlock();
             }
@@ -137,4 +127,18 @@ boolean remove(T entry) {
         }
         return false;
     }
+
+    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;
+            }
+        }
+        return null;
+    }
 }

libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockableConcurrentQueue.java

@@ -10,6 +10,7 @@
 
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 /**
@@ -56,6 +57,26 @@ public T lockAndPoll() {
         return null;
     }
 
+    /**
+     * Poll with compatibility filter. Scans the queue for a compatible entry that can be locked.
+     * Incompatible entries are removed from the queue.
+     * Compatible entries that cannot be locked (held by another thread) are skipped.
+     *
+     * @param isCompatible predicate to test compatibility — failing entries are removed
+     * @return the locked matched entry, or null if none found
+     */
+    public T lockAndPollWithRejects(Predicate<T> isCompatible) {
+        int addAndUnlockCount;
+        do {
+            addAndUnlockCount = addAndUnlockCounter.get();
+            T entry = queue.pollAndDropIncompatible(isCompatible, Lockable::tryLock);
+            if (entry != null) {
+                return entry;
+            }
+        } while (addAndUnlockCount != addAndUnlockCounter.get());
+        return null;
+    }
+
     /**
      * Remove an entry from the queue.
      *

libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockablePool.java

@@ -18,14 +18,20 @@
 import java.util.Objects;
 import java.util.Queue;
 import java.util.Set;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 /**
  * A thread-safe pool of {@link Lockable} items backed by a {@link LockableConcurrentQueue}.
  * Items are locked on checkout and unlocked on release, ensuring safe reuse across threads.
  * <p>
- * The pool is created with a supplier that produces new items on demand when the pool
- * is empty. Items are tracked in a set for registration checks and iteration.
+ * The pool supports two modes of checkout:
+ * <ul>
+ *   <li>{@link #getAndLock()} — returns any available item, or creates a new one</li>
+ *   <li>{@link #getAndLock(Predicate)} — returns a compatible item (per predicate),
+ *       rejecting incompatible ones. Rejected items are removed from the available queue
+ *       but remain tracked by the pool and included in {@link #checkoutAll()}.</li>
+ * </ul>
  *
  * @param <T> the pooled item type, must implement {@link Lockable}
  */
@@ -56,9 +62,21 @@ public LockablePool(Supplier<T> itemSupplier, Supplier<Queue<T>> queueSupplier,
      * @throws IllegalStateException if the pool is closed
      */
     public T getAndLock() {
+        return getAndLock(e -> true);
+    }
+
+    /**
+     * Locks and polls a compatible item from the pool in a single pass. Items that fail
+     * the predicate are removed from the available queue but remain tracked by the pool.
+     * If no compatible item is found, a new one is created using the constructor's item supplier.
+     *
+     * @param isCompatible predicate to test each polled item
+     * @return a locked, compatible item
+     * @throws IllegalStateException if the pool is closed
+     */
+    public T getAndLock(Predicate<T> isCompatible) {
         ensureOpen();
-        T item = availableItems.lockAndPoll();
-        return Objects.requireNonNullElseGet(item, this::fetchItem);
+        return Objects.requireNonNullElseGet(availableItems.lockAndPollWithRejects(isCompatible), this::fetchItem);
     }
 
     private synchronized T fetchItem() {

libs/concurrent-queue/src/test/java/org/opensearch/common/queue/ConcurrentQueueTests.java

@@ -160,4 +160,73 @@ public void testConcurrentAddAndPoll() throws Exception {
         pollLatch.await();
         assertEquals(numThreads * itemsPerThread, totalPolled.get());
     }
+
+    // --- Tests for pollWithRejects ---
+
+    public void testPollAndDropIncompatibleReturnsCompatibleAndDropsIncompatible() {
+        ConcurrentQueue<Integer> queue = new ConcurrentQueue<>(LinkedList::new, 1);
+        queue.add(1);
+        queue.add(2);
+        queue.add(3);
+
+        // Compatible: even numbers. canSelect: always true.
+        Integer result = queue.pollAndDropIncompatible(n -> n % 2 == 0, n -> true);
+        assertEquals(Integer.valueOf(2), result);
+        // 1 was rejected (removed), 3 still in queue (after the match, not scanned)
+        assertEquals(Integer.valueOf(3), queue.poll(e -> true));
+        assertNull(queue.poll(e -> true)); // 1 was removed
+    }
+
+    public void testPollAndDropIncompatibleAllIncompatible() {
+        ConcurrentQueue<Integer> queue = new ConcurrentQueue<>(LinkedList::new, 1);
+        queue.add(1);
+        queue.add(3);
+        queue.add(5);
+
+        Integer result = queue.pollAndDropIncompatible(n -> n % 2 == 0, n -> true);
+        assertNull(result);
+        // All removed as incompatible
+        assertNull(queue.poll(e -> true));
+    }
+
+    public void testPollAndDropIncompatibleEmptyQueue() {
+        ConcurrentQueue<Integer> queue = new ConcurrentQueue<>(LinkedList::new, 1);
+        Integer result = queue.pollAndDropIncompatible(n -> true, n -> true);
+        assertNull(result);
+    }
+
+    public void testPollAndDropIncompatibleSkipsCompatibleButUnselectable() {
+        ConcurrentQueue<Integer> queue = new ConcurrentQueue<>(LinkedList::new, 1);
+        queue.add(1);  // incompatible
+        queue.add(2);  // compatible but canSelect=false
+        queue.add(4);  // compatible and canSelect=true
+
+        Integer result = queue.pollAndDropIncompatible(
+            n -> n % 2 == 0,       // compatible: even
+            n -> n > 3             // canSelect: > 3
+        );
+        assertEquals(Integer.valueOf(4), result);
+        // 1 was rejected (removed), 2 should still be in queue (compatible but not selectable)
+        assertEquals(Integer.valueOf(2), queue.poll(e -> true));
+        assertNull(queue.poll(e -> true));
+    }
+
+    public void testPollAndDropIncompatibleMultipleStripes() {
+        ConcurrentQueue<Integer> queue = new ConcurrentQueue<>(LinkedList::new, 4);
+        for (int i = 0; i < 20; i++) {
+            queue.add(i);
+        }
+
+        // Drop 0-9 as incompatible, select first compatible entry (>= 10)
+        Integer first = queue.pollAndDropIncompatible(n -> n >= 10, n -> true);
+        assertNotNull(first);
+        assertTrue(first >= 10);
+
+        // Poll all remaining entries
+        int count = 1; // counting the first result
+        while (queue.poll(e -> true) != null) {
+            count++;
+        }
+        assertEquals(10, count);
+    }
 }

libs/concurrent-queue/src/test/java/org/opensearch/common/queue/LockableConcurrentQueueTests.java

@@ -215,4 +215,36 @@ public void testConcurrentLockAndPollAndAddAndUnlock() throws Exception {
         }
         assertEquals(numEntries, remaining);
     }
+
+    public void testLockAndPollWithRejectsReturnsCompatibleEntry() {
+        LockableConcurrentQueue<LockableEntry> queue = new LockableConcurrentQueue<>(LinkedList::new, 1);
+        LockableEntry e1 = new LockableEntry("old");
+        LockableEntry e2 = new LockableEntry("old");
+        LockableEntry e3 = new LockableEntry("current");
+        seedEntry(queue, e1);
+        seedEntry(queue, e2);
+        seedEntry(queue, e3);
+
+        LockableEntry result = queue.lockAndPollWithRejects(e -> e.id.equals("current"));
+        assertSame(e3, result);
+        assertTrue(result.isHeldByCurrentThread());
+        // e1 and e2 were incompatible and should have been dropped
+        assertNull(queue.lockAndPoll());
+        result.unlock();
+    }
+
+    public void testLockAndPollWithRejectsAllIncompatible() {
+        LockableConcurrentQueue<LockableEntry> queue = new LockableConcurrentQueue<>(LinkedList::new, 1);
+        seedEntry(queue, new LockableEntry("old"));
+        seedEntry(queue, new LockableEntry("old"));
+
+        assertNull(queue.lockAndPollWithRejects(e -> e.id.equals("new")));
+        // Queue should be empty after all incompatible entries were dropped
+        assertNull(queue.lockAndPoll());
+    }
+
+    public void testLockAndPollWithRejectsEmptyQueue() {
+        LockableConcurrentQueue<LockableEntry> queue = new LockableConcurrentQueue<>(LinkedList::new, 1);
+        assertNull(queue.lockAndPollWithRejects(e -> true));
+    }
 }

libs/concurrent-queue/src/test/java/org/opensearch/common/queue/LockablePoolTests.java

@@ -143,4 +143,57 @@ public void testClosedPoolThrowsOnCheckoutAll() throws IOException {
         IllegalStateException ex = expectThrows(IllegalStateException.class, pool::checkoutAll);
         assertEquals("LockablePool is already closed", ex.getMessage());
     }
+
+    // --- Tests for filtered getAndLock with rejection ---
+
+    public void testFilteredGetAndLockReturnsCompatibleItem() {
+        LockablePool<LockableEntry> pool = createPool();
+        LockableEntry item = pool.getAndLock();
+        pool.releaseAndUnlock(item);
+
+        LockableEntry result = pool.getAndLock(e -> true);
+        assertSame(item, result);
+        pool.releaseAndUnlock(result);
+    }
+
+    public void testFilteredGetAndLockRejectsIncompatibleItem() {
+        LockablePool<LockableEntry> pool = createPool();
+        LockableEntry item = pool.getAndLock();
+        pool.releaseAndUnlock(item);
+
+        // Reject the existing item — predicate fails, pool creates new via supplier
+        LockableEntry result = pool.getAndLock(e -> !e.id.equals(item.id));
+        assertNotSame(item, result);
+        pool.releaseAndUnlock(result);
+    }
+
+    public void testCheckoutAllIncludesRejectedItems() {
+        LockablePool<LockableEntry> pool = createPool();
+        LockableEntry item = pool.getAndLock();
+        pool.releaseAndUnlock(item);
+
+        // Reject the existing item
+        LockableEntry fresh = pool.getAndLock(e -> !e.id.equals(item.id));
+        pool.releaseAndUnlock(fresh);
+
+        List<LockableEntry> all = pool.checkoutAll();
+        assertEquals(2, all.size());
+        assertTrue(all.contains(item));
+        assertTrue(all.contains(fresh));
+    }
+
+    public void testRejectedItemNotReturnedBySubsequentPoll() {
+        LockablePool<LockableEntry> pool = createPool();
+        LockableEntry item = pool.getAndLock();
+        pool.releaseAndUnlock(item);
+
+        // Reject it
+        LockableEntry fresh = pool.getAndLock(e -> !e.id.equals(item.id));
+        pool.releaseAndUnlock(fresh);
+
+        // Next poll should return fresh, not the rejected item
+        LockableEntry next = pool.getAndLock();
+        assertSame(fresh, next);
+        pool.releaseAndUnlock(next);
+    }
 }

sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneIndexingExecutionEngine.java

@@ -31,6 +31,7 @@
 import org.opensearch.index.engine.dataformat.RefreshInput;
 import org.opensearch.index.engine.dataformat.RefreshResult;
 import org.opensearch.index.engine.dataformat.Writer;
+import org.opensearch.index.engine.dataformat.WriterConfig;
 import org.opensearch.index.engine.exec.Segment;
 import org.opensearch.index.engine.exec.WriterFileSet;
 import org.opensearch.index.engine.exec.commit.IndexStoreProvider;
@@ -52,7 +53,7 @@
  * Lucene-specific {@link IndexingExecutionEngine} that manages per-writer Lucene segments
  * and incorporates them into the shared {@link LuceneCommitter} writer during refresh.
  *
- * Write path: Each call to {@link #createWriter(long)} creates a {@link LuceneWriter} with its own
+ * Write path: Each call to {@link #createWriter(WriterConfig)} creates a {@link LuceneWriter} with its own
  * {@link IndexWriter} in an isolated temp directory. Documents are indexed into this
  * per-writer segment. On flush, the writer force-merges to exactly 1 segment.
  *
@@ -145,17 +146,17 @@ public FormatStore getStore(DataFormat dataFormat) {
      * Creates a new {@link LuceneWriter} for the given generation in an isolated temp directory
      * under the shard's Lucene base directory.
      *
-     * @param writerGeneration the generation number for the new writer
+     * @param config the writer configuration
      * @return a new writer
      * @throws RuntimeException wrapping an {@link IOException} if writer creation fails
      */
     @Override
-    public Writer<LuceneDocumentInput> createWriter(long writerGeneration) {
+    public Writer<LuceneDocumentInput> createWriter(WriterConfig config) {
         assert sharedWriter.isOpen() : "Cannot create writer — shared IndexWriter is closed";
         try {
-            return new LuceneWriter(writerGeneration, dataFormat, baseDirectory, analyzer, codec);
+            return new LuceneWriter(config.writerGeneration(), dataFormat, baseDirectory, analyzer, codec);
         } catch (IOException e) {
-            throw new RuntimeException("Failed to create LuceneWriter for generation " + writerGeneration, e);
+            throw new RuntimeException("Failed to create LuceneWriter for generation " + config.writerGeneration(), e);
         }
     }
 

sandbox/plugins/analytics-backend-lucene/src/main/java/org/opensearch/be/lucene/index/LuceneWriter.java

@@ -196,6 +196,19 @@ public long generation() {
         return writerGeneration;
     }
 
+    @Override
+    public boolean isSchemaMutable() {
+        return true;
+    }
+
+    @Override
+    public long mappingVersion() {
+        return 0;
+    }
+
+    @Override
+    public void updateMappingVersion(long newVersion) {}
+
     /** Acquires the writer's reentrant lock. Used by the writer pool to serialize access. */
     @Override
     public void lock() {