Dynamic Mapping support for Pluggable Data Formats

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-datafusion/rust/src/indexed_executor.rs

@@ -427,12 +427,10 @@ pub async unsafe fn execute_indexed_with_context(
     // with IndexedTableProvider after plan decoding.
     ctx.deregister_table(&table_name)?;
 
-    let store = ctx
-        .state()
-        .runtime_env()
-        .object_store(&table_path)?;
+    let state = ctx.state();
+    let store = state.runtime_env().object_store(&table_path)?;
 
-    let (segments, schema) = build_segments(Arc::clone(&store), object_metas.as_ref())
+    let (segments, schema) = build_segments(&state, Arc::clone(&store), object_metas.as_ref())
         .await
         .map_err(DataFusionError::Execution)?;
     for (i, seg) in segments.iter().enumerate() {