IGNITE-24135 BPlusTree reuses lastRow from rowFilter closure to fill buffer (#11816)

Author: timoninmaxim

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/ExpiredEntriesIntegrationTest.java

@@ -23,6 +23,9 @@
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.cache.query.annotations.QuerySqlField;
 import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.metric.IoStatisticsHolder;
+import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.junit.Test;
 
@@ -35,23 +38,29 @@
  */
 public class ExpiredEntriesIntegrationTest extends AbstractBasicIntegrationTest {
     /** */
-    @Test
-    public void testExpiration() throws Exception {
+    @Override protected void beforeTest() throws Exception {
         CacheConfiguration<Integer, Developer> cacheCfg = new CacheConfiguration<Integer, Developer>()
             .setIndexedTypes(Integer.class, Developer.class)
             .setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(new Duration(TimeUnit.MILLISECONDS, 1)));
 
-        IgniteCache<Integer, Developer> cache1 = client.getOrCreateCache(new CacheConfiguration<>(cacheCfg)
+        client.createCache(new CacheConfiguration<>(cacheCfg)
             .setName("CACHE1")
             .setEagerTtl(false)
         );
 
-        IgniteCache<Integer, Developer> cache2 = client.getOrCreateCache(new CacheConfiguration<>(cacheCfg)
+        client.createCache(new CacheConfiguration<>(cacheCfg)
             .setName("CACHE2")
             .setEagerTtl(true)
         );
 
         awaitPartitionMapExchange();
+    }
+
+    /** */
+    @Test
+    public void testExpiration() throws Exception {
+        IgniteCache<Integer, Developer> cache1 = client.cache("CACHE1");
+        IgniteCache<Integer, Developer> cache2 = client.cache("CACHE2");
 
         for (int i = 0; i < 100; i++) {
             cache1.put(i, new Developer("dev" + i, i));
@@ -71,6 +80,30 @@ public void testExpiration() throws Exception {
         checkExpiration("CACHE2", true);
     }
 
+    /** Validate that check of expiry policy doesn't require additional page read. */
+    @Test
+    public void testIndexScanReadPageOnce() {
+        for (String cacheName: F.asList("CACHE1", "CACHE2")) {
+            IgniteCache<Integer, Developer> cache = client.cache(cacheName);
+
+            ExpiryPolicy expPlc = new CreatedExpiryPolicy(new Duration(TimeUnit.DAYS, 1));
+
+            int key = primaryKey(grid(0).cache(cacheName));
+
+            cache.withExpiryPolicy(expPlc).put(key, new Developer("dev0", 0));
+
+            IoStatisticsHolder statHld = grid(0).cachex(cacheName).context().group().statisticsHolderData();
+
+            long before = statHld.logicalReads();
+
+            assertQuery("SELECT /*+ FORCE_INDEX(DEVELOPER_DEPID_IDX) */ name, depId FROM " + cacheName + ".DEVELOPER where depId=0")
+                .matches(QueryChecker.containsIndexScan(cacheName, "DEVELOPER", "DEVELOPER_DEPID_IDX"))
+                .check();
+
+            assertEquals(1, statHld.logicalReads() - before);
+        }
+    }
+
     /** */
     private void checkExpiration(String schema, boolean eagerTtl) {
         assertQuery("SELECT depId, name FROM " + schema + ".DEVELOPER WHERE name IS NOT NULL")

modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/sorted/inline/InlineTreeFilterClosure.java

@@ -25,6 +25,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.BPlusIO;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.spi.indexing.IndexingQueryCacheFilter;
+import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.internal.pagemem.PageIdUtils.pageId;
 
@@ -38,6 +39,9 @@ public class InlineTreeFilterClosure implements BPlusTree.TreeRowClosure<IndexRo
     /** */
     private final BPlusTree.TreeRowClosure<IndexRow, IndexRow> rowFilter;
 
+    /** Last index row analyzed by {@link #rowFilter}. */
+    private @Nullable IndexRow lastRow;
+
     /** Constructor. */
     public InlineTreeFilterClosure(IndexingQueryCacheFilter cacheFilter,
         BPlusTree.TreeRowClosure<IndexRow, IndexRow> rowFilter) {
@@ -53,15 +57,22 @@ public InlineTreeFilterClosure(IndexingQueryCacheFilter cacheFilter,
 
         boolean val = cacheFilter == null || applyFilter((InlineIO)io, pageAddr, idx);
 
-        if (!val)
-            return false;
-
-        if (rowFilter != null)
+        if (val && rowFilter != null) {
             val = rowFilter.apply(tree, io, pageAddr, idx);
 
+            lastRow = rowFilter.lastRow();
+        }
+        else
+            lastRow = null;
+
         return val;
     }
 
+    /** {@inheritDoc} */
+    @Override public @Nullable IndexRow lastRow() {
+        return lastRow;
+    }
+
     /**
      * @param io Row IO.
      * @param pageAddr Page address.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java

@@ -5724,8 +5724,19 @@ private final class ForwardCursor extends AbstractForwardCursor implements GridC
 
             for (int idx = startIdx; idx < cnt; idx++) {
                 if (c == null || c.apply(BPlusTree.this, io, pageAddr, idx)) {
-                    rows = GridArrays.set(rows, resCnt++, rowFactory == null ? getRow(io, pageAddr, idx, x) :
-                        rowFactory.create(BPlusTree.this, io, pageAddr, idx));
+                    T row = null;
+
+                    if (rowFactory != null)
+                        row = rowFactory.create(BPlusTree.this, io, pageAddr, idx);
+                    else {
+                        if (c != null)
+                            row = c.lastRow();
+
+                        if (row == null)
+                            row = getRow(io, pageAddr, idx, x);
+                    }
+
+                    rows = GridArrays.set(rows, resCnt++, row);
                 }
             }
 
@@ -5938,7 +5949,7 @@ public boolean apply(BPlusTree<L, T> tree, BPlusIO<L> io, long pageAddr, int idx
          *
          * @return Last row that was analyzed or {@code null}.
          */
-        public default L lastRow() {
+        public default T lastRow() {
             return null;
         }
     }