IGNITE-22776 : SQL table statistic is updated only once.

modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java

@@ -2747,6 +2747,8 @@ public void store(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDat
 
         if (idx != null)
             idx.store(cctx, desc, newRow, prevRow, prevRowAvailable);
+
+        statsMgr.onRowUpdated(desc.schemaName(), desc.tableName(), newRow.partition(), key.valueBytes(coctx));
     }
 
     /**
@@ -3614,6 +3616,8 @@ public void remove(GridCacheContext cctx, CacheDataRow row)
 
         if (indexingEnabled())
             idx.remove(cctx, desc, row);
+
+        statsMgr.onRowUpdated(desc.schemaName(), desc.tableName(), row.partition(), row.key().valueBytes(cctx.cacheObjectContext()));
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/query/stat/IgniteStatisticsManagerImpl.java

@@ -111,7 +111,7 @@ public class IgniteStatisticsManagerImpl implements IgniteStatisticsManager {
     private volatile boolean started;
 
     /** Schedule to process obsolescence statistics. */
-    private GridTimeoutProcessor.CancelableTask obsolescenceSchedule;
+    private volatile GridTimeoutProcessor.CancelableTask obsolescenceSchedule;
 
     /** Exchange listener. */
     private final PartitionsExchangeAware exchAwareLsnr = new PartitionsExchangeAware() {
@@ -236,16 +236,23 @@ else if (db == null)
 
         tryStart();
 
-        if (serverNode) {
-            // Use mgmt pool to work with statistics repository in busy lock to schedule some tasks.
-            obsolescenceSchedule = ctx.timeout().schedule(() -> {
-                obsolescenceBusyExecutor.execute(() -> processObsolescence());
-            }, OBSOLESCENCE_INTERVAL * 1000, OBSOLESCENCE_INTERVAL * 1000);
-        }
+        if (serverNode)
+            scheduleObsolescence(OBSOLESCENCE_INTERVAL);
 
         ctx.cache().context().exchange().registerExchangeAwareComponent(exchAwareLsnr);
     }
 
+    /** */
+    void scheduleObsolescence(int seconds) {
+        assert seconds >= 1;
+
+        if (obsolescenceSchedule != null)
+            obsolescenceSchedule.close();
+
+        obsolescenceSchedule = ctx.timeout().schedule(() -> obsolescenceBusyExecutor.execute(this::processObsolescence),
+            seconds * 1000, seconds * 1000);
+    }
+
     /**
      * Check all preconditions and stop if started and have reason to stop.
      */

modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/stat/StatisticsAbstractTest.java

@@ -261,12 +261,23 @@ protected List<List<?>> sql(String sql) {
     }
 
     /**
-     * Create SQL table with the given index.
+     * Creates SQL table with the given index and fills with small amount of data.
      *
      * @param suffix Table idx, if {@code null} - name "SMALL" without index will be used.
      * @return Table name.
      */
     protected String createSmallTable(String suffix) {
+        return createSmallTable(SMALL_SIZE, suffix);
+    }
+
+    /**
+     * Creates SQL table with the given index and fills with data of the passed amount.
+     *
+     * @param preloadCnt Records cnt to load after creation.
+     * @param suffix Table idx, if {@code null} - name "SMALL" without index will be used.
+     * @return Table name.
+     */
+    protected String createSmallTable(int preloadCnt, String suffix) {
         String tblName = "small" + ((suffix != null) ? suffix : "");
 
         sql("DROP TABLE IF EXISTS " + tblName);
@@ -279,7 +290,7 @@ protected String createSmallTable(String suffix) {
 
         sql(String.format("CREATE INDEX %s_c ON %s(c)", tblName, tblName));
 
-        for (int i = 0; i < SMALL_SIZE; i++)
+        for (int i = 0; i < preloadCnt; i++)
             sql(String.format("INSERT INTO %s(a, b, c) VALUES(%d, %d, %d)", tblName, i, i, i % 10));
 
         return tblName;

modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/stat/StatisticsObsolescenceTest.java

@@ -18,22 +18,147 @@
 package org.apache.ignite.internal.processors.query.stat;
 
 import java.util.Map;
-
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 import org.apache.ignite.Ignite;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
 import org.apache.ignite.cluster.ClusterState;
 import org.apache.ignite.configuration.DataRegionConfiguration;
 import org.apache.ignite.configuration.DataStorageConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.util.collection.IntMap;
+import org.apache.ignite.internal.util.typedef.G;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.junit.Test;
 
 import static org.apache.ignite.internal.processors.query.stat.IgniteStatisticsHelper.buildDefaultConfigurations;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
 
 /**
  * Test for statistics obsolescence.
  */
 public class StatisticsObsolescenceTest extends StatisticsAbstractTest {
+    /** */
+    @Test
+    public void testObsolescenceWithInsert() throws Exception {
+        doTestObsolescenceUnderLoad(false, 1,
+            key -> sql(String.format("insert into SMALL(A, B, C) values(%d, %d, %d)", key, key, key)));
+    }
+
+    /** */
+    @Test
+    public void testObsolescenceWithUpdate() throws Exception {
+        doTestObsolescenceUnderLoad(true, 0, key -> sql("update SMALL set B=B+1 where A=" + key));
+    }
+
+    /** */
+    @Test
+    public void testObsolescenceWithDelete() throws Exception {
+        doTestObsolescenceUnderLoad(true, -1, key -> sql("delete from SMALL where A=" + key));
+    }
+
+    /** */
+    private void doTestObsolescenceUnderLoad(boolean preload, int rowCntCmp, Consumer<Long> op) throws Exception {
+        // Keep enough data to touch every partition. The statistics collection is sensitive to a partition's empty rows num
+        // and is able to reassemble in this case. This would give false-positive result.
+        int workingRowsNum = RendezvousAffinityFunction.DFLT_PARTITION_COUNT * 10;
+        int preloadCnt = preload ? workingRowsNum : 0;
+
+        int osbInterval = 7;
+
+        CyclicBarrier barrier = new CyclicBarrier(2);
+
+        try {
+            startGridsMultiThreaded(2);
+
+            for (Ignite ig : G.allGrids())
+                ((IgniteStatisticsManagerImpl)((IgniteEx)ig).context().query().statsManager()).scheduleObsolescence(osbInterval);
+
+            createSmallTable(preloadCnt, null);
+
+            statisticsMgr(0).usageState(StatisticsUsageState.ON);
+            statisticsMgr(0).collectStatistics(buildDefaultConfigurations(SMALL_TARGET));
+
+            // Initialized statistics.
+            assertTrue(waitForCondition(() -> statisticsMgr(0).getLocalStatistics(SMALL_KEY) != null, osbInterval * 1000));
+            assertTrue(waitForCondition(() -> statisticsMgr(1).getLocalStatistics(SMALL_KEY) != null, osbInterval * 1000));
+
+            ObjectStatisticsImpl initStat1 = (ObjectStatisticsImpl)statisticsMgr(0).getLocalStatistics(SMALL_KEY);
+            ObjectStatisticsImpl initStat2 = (ObjectStatisticsImpl)statisticsMgr(1).getLocalStatistics(SMALL_KEY);
+
+            assertEquals(preloadCnt, initStat1.rowCount() + initStat2.rowCount());
+
+            GridTestUtils.runAsync(() -> {
+                AtomicLong key = new AtomicLong(1L);
+
+                long opCnt = 0;
+
+                while (!barrier.isBroken()) {
+                    op.accept(key.getAndIncrement());
+
+                    // Enough updates to trigger the statistics.
+                    if (++opCnt == workingRowsNum / 3) {
+                        opCnt = 0;
+
+                        barrier.await();
+                        barrier.await();
+                    }
+                }
+            });
+
+            barrier.await();
+
+            waitForStatsUpdates(initStat1, osbInterval * 2);
+
+            ObjectStatisticsImpl updatedStat = (ObjectStatisticsImpl)statisticsMgr(0).getLocalStatistics(SMALL_KEY);
+
+            assertTrue(rowCntCmp > 0 ? updatedStat.rowCount() > initStat1.rowCount() :
+                (rowCntCmp < 0 ? updatedStat.rowCount() < initStat1.rowCount() : updatedStat.rowCount() == initStat1.rowCount()));
+
+            barrier.await();
+            barrier.await();
+
+            // Continuing data loading, the table is being updated. Since the row count is inreasing, we must obtain a
+            // new statistics, greather than {@code firstNotEmpty}.
+            waitForStatsUpdates(updatedStat, osbInterval * 2);
+
+            ObjectStatisticsImpl finalStat = (ObjectStatisticsImpl)statisticsMgr(0).getLocalStatistics(SMALL_KEY);
+
+            assertTrue(rowCntCmp > 0 ? finalStat.rowCount() > updatedStat.rowCount() :
+                (rowCntCmp < 0 ? finalStat.rowCount() < updatedStat.rowCount() : finalStat.rowCount() == updatedStat.rowCount()));
+        }
+        finally {
+            barrier.reset();
+        }
+    }
+
+    /** */
+    private void waitForStatsUpdates(ObjectStatisticsImpl compareTo, long timeoutSec) throws IgniteInterruptedCheckedException {
+        assertTrue(waitForCondition(() -> {
+            ObjectStatisticsImpl updatedStat = (ObjectStatisticsImpl)statisticsMgr(0).getLocalStatistics(SMALL_KEY);
+
+            if (updatedStat == null)
+                return false;
+
+            AtomicBoolean passed = new AtomicBoolean(true);
+
+            updatedStat.columnsStatistics().forEach((col, stat) -> {
+                ColumnStatistics compared = compareTo.columnStatistics(col);
+
+                assert compared != null;
+
+                if (compared.createdAt() >= stat.createdAt())
+                    passed.set(false);
+            });
+
+            return passed.get();
+        }, timeoutSec * 1000));
+    }
+
     /**
      * Test statistics refreshing after significant changes of base table:
      * 1) Create and populate small table
@@ -51,7 +176,7 @@ public void testObsolescence() throws Exception {
 
         statisticsMgr(0).collectStatistics(buildDefaultConfigurations(SMALL_TARGET));
 
-        assertTrue(GridTestUtils.waitForCondition(() -> statisticsMgr(0).getLocalStatistics(SMALL_KEY) != null, TIMEOUT));
+        assertTrue(waitForCondition(() -> statisticsMgr(0).getLocalStatistics(SMALL_KEY) != null, TIMEOUT));
 
         ObjectStatisticsImpl stat1 = (ObjectStatisticsImpl)statisticsMgr(0).getLocalStatistics(SMALL_KEY);
 
@@ -62,7 +187,7 @@ public void testObsolescence() throws Exception {
 
         statisticsMgr(0).processObsolescence();
 
-        assertTrue(GridTestUtils.waitForCondition(() -> {
+        assertTrue(waitForCondition(() -> {
             ObjectStatisticsImpl stat2 = (ObjectStatisticsImpl)statisticsMgr(0).getLocalStatistics(SMALL_KEY);
 
             return stat2 != null && stat2.rowCount() > stat1.rowCount();
@@ -103,7 +228,7 @@ public void testInactiveLoad() throws Exception {
 
         ignite.cluster().state(ClusterState.ACTIVE);
 
-        assertTrue(GridTestUtils.waitForCondition(() -> statObs.get(SMALL_KEY).size() > oldSize, TIMEOUT));
+        assertTrue(waitForCondition(() -> statObs.get(SMALL_KEY).size() > oldSize, TIMEOUT));
     }
 
     /** {@inheritDoc} */