IGNITE-26683 IndexQuery can't use an index until data population completes (#12412)

Author: McAshesha

modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexName.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.cache.query.index;
 
+import java.util.Objects;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -90,4 +91,20 @@ public String cacheName() {
     @Override public String toString() {
         return fullName();
     }
+
+    /** */
+    @Override public boolean equals(Object o) {
+        if (!(o instanceof IndexName))
+            return false;
+
+        IndexName name = (IndexName)o;
+
+        return Objects.equals(schemaName, name.schemaName) && Objects.equals(tableName, name.tableName)
+            && Objects.equals(cacheName, name.cacheName) && Objects.equals(idxName, name.idxName);
+    }
+
+    /** */
+    @Override public int hashCode() {
+        return Objects.hash(schemaName, tableName, cacheName, idxName);
+    }
 }

modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexProcessor.java

@@ -21,8 +21,10 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteLogger;
@@ -115,6 +117,9 @@ public static void registerIO() {
     /** Exclusive lock for DDL operations. */
     private final ReentrantReadWriteLock ddlLock = new ReentrantReadWriteLock();
 
+    /** Set of index names currently in initial population (null if none). */
+    private @Nullable Set<IndexName> fillingIdxs;
+
     /**
      * @param ctx Kernal context.
      */
@@ -211,15 +216,45 @@ public Index createIndexDynamically(
         IndexDefinition definition,
         SchemaIndexCacheVisitor cacheVisitor
     ) {
-        Index idx = createIndex(cctx, factory, definition);
+        IndexFactory dynamicFactory = (gcctx, indexDefinition) -> {
+            Index idx = factory.createIndex(gcctx, indexDefinition);
+
+            assert ddlLock.isWriteLockedByCurrentThread();
+
+            if (fillingIdxs == null)
+                fillingIdxs = new HashSet<>();
+
+            fillingIdxs.add(indexDefinition.idxName());
+
+            return idx;
+        };
+
+        try {
+            Index idx = createIndex(cctx, dynamicFactory, definition);
 
-        // Populate index with cache rows.
-        cacheVisitor.visit(row -> {
-            if (idx.canHandle(row))
-                idx.onUpdate(null, row, false);
-        });
+            // Populate index with cache rows.
+            cacheVisitor.visit(row -> {
+                if (idx.canHandle(row))
+                    idx.onUpdate(null, row, false);
+            });
 
-        return idx;
+            return idx;
+        }
+        finally {
+            ddlLock.writeLock().lock();
+
+            try {
+                if (fillingIdxs != null) {
+                    fillingIdxs.remove(definition.idxName());
+
+                    if (fillingIdxs.isEmpty())
+                        fillingIdxs = null;
+                }
+            }
+            finally {
+                ddlLock.writeLock().unlock();
+            }
+        }
     }
 
     /**
@@ -272,8 +307,18 @@ public void removeIndex(IndexName idxName, boolean softDelete) {
 
             Index idx = idxs.remove(idxName.fullName());
 
-            if (idx != null)
+            if (idx != null) {
                 idx.destroy(softDelete);
+
+                if (fillingIdxs != null) {
+                    fillingIdxs.remove(idxName);
+
+                    if (fillingIdxs.isEmpty())
+                        fillingIdxs = null;
+                }
+
+            }
+
         }
         finally {
             ddlLock.writeLock().unlock();
@@ -371,16 +416,33 @@ public IndexesRebuildTask idxRebuild() {
      * @return Collection of indexes for specified cache.
      */
     public Collection<Index> indexes(String cacheName) {
+        return indexes(cacheName, false);
+    }
+
+    /**
+     * Returns collection of indexes for specified cache.
+     *
+     * @param cacheName Cache name.
+     * @param skipFilling If {@code true}, indexes that are currently being initially populated are excluded
+     *     from the result; if {@code false}, all known indexes for the cache are returned.
+     * @return Collection of indexes for specified cache.
+     */
+
+    public Collection<Index> indexes(String cacheName, boolean skipFilling) {
         ddlLock.readLock().lock();
 
         try {
-            Map<String, Index> idxs = cacheToIdx.get(cacheName);
+            Map<String, Index> idxMap = cacheToIdx.get(cacheName);
 
-            if (idxs == null)
+            if (idxMap == null)
                 return Collections.emptyList();
 
-            return idxs.values();
+            List<Index> idxs = new ArrayList<>(idxMap.values());
 
+            if (skipFilling && fillingIdxs != null)
+                idxs.removeIf(idx -> fillingIdxs.contains(idx.indexDefinition().idxName()));
+
+            return idxs;
         }
         finally {
             ddlLock.readLock().unlock();
@@ -394,9 +456,26 @@ public Collection<Index> indexes(String cacheName) {
      * @return Index for specified index name or {@code null} if not found.
      */
     public @Nullable Index index(IndexName idxName) {
+        return index(idxName, false);
+    }
+
+    /**
+     * Returns index for specified name.
+     *
+     * @param idxName Index name.
+     * @param skipFilling If {@code true}, returns {@code null} when the index is currently being initially
+     *     populated and therefore should be skipped; if {@code false}, returns the index if present.
+     * @return Index for specified index name or {@code null} if not found (or skipped due to population when
+     *     {@code skipFilling} is {@code true}).
+     */
+
+    public @Nullable Index index(IndexName idxName, boolean skipFilling) {
         ddlLock.readLock().lock();
 
         try {
+            if (skipFilling && fillingIdxs != null && fillingIdxs.contains(idxName))
+                return null;
+
             Map<String, Index> idxs = cacheToIdx.get(idxName.cacheName());
 
             if (idxs == null)

modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexQueryProcessor.java

@@ -217,14 +217,14 @@ private SortedSegmentedIndex findSortedIndex(GridCacheContext<?, ?> cctx, IndexQ
      * @throws IgniteCheckedException If index not found or specified index doesn't match query criteria.
      */
     private SortedSegmentedIndex indexByName(IndexName idxName, final Map<String, String> criteriaFlds) throws IgniteCheckedException {
-        SortedSegmentedIndex idx = assertSortedIndex(idxProc.index(idxName));
+        SortedSegmentedIndex idx = assertSortedIndex(idxProc.index(idxName, true));
 
         if (idx == null && !QueryUtils.PRIMARY_KEY_INDEX.equals(idxName.idxName())) {
             String normIdxName = QueryUtils.normalizeObjectName(idxName.idxName(), false);
 
             idxName = new IndexName(idxName.cacheName(), idxName.schemaName(), idxName.tableName(), normIdxName);
 
-            idx = assertSortedIndex(idxProc.index(idxName));
+            idx = assertSortedIndex(idxProc.index(idxName, true));
         }
 
         if (idx == null)
@@ -245,7 +245,7 @@ private SortedSegmentedIndex indexByCriteria(
         final Map<String, String> criteriaFlds,
         String tableName
     ) throws IgniteCheckedException {
-        Collection<Index> idxs = idxProc.indexes(cctx.name());
+        Collection<Index> idxs = idxProc.indexes(cctx.name(), true);
 
         for (Index idx: idxs) {
             SortedSegmentedIndex sortedIdx = assertSortedIndex(idx);