GG-18641: Statistics obsolescence (apache#1803)

GG-18641: statistics obsolescense implementation.

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

@@ -45,7 +45,6 @@
 import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
 import org.apache.ignite.internal.processors.odbc.jdbc.JdbcParameterMeta;
 import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
-import org.apache.ignite.internal.processors.query.stat.IgniteStatisticsManager;
 import org.apache.ignite.internal.util.GridAtomicLong;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.collection.IntMap;
@@ -519,13 +518,6 @@ default TimeZone clusterTimezone() {
         return TimeZone.getDefault();
     }
 
-    /**
-     * Get statistics manager.
-     *
-     * @return Statistics manager.
-     */
-    IgniteStatisticsManager statsManager();
-
     /**
      * Defragment index partition.
      *

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

@@ -16,6 +16,7 @@
 package org.apache.ignite.internal.processors.query.stat;
 
 import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.query.stat.config.StatisticsObjectConfiguration;
 
 /**
  * Statistics manager. Coordinate statistics collection and act as source of statistics.
@@ -24,10 +25,10 @@ public interface IgniteStatisticsManager {
     /**
      * Gather object statistics.
      *
-     * @param targets Target to gather statistics by.
+     * @param targets Target to params map to gather statistics by.
      * @throws IgniteCheckedException  Throws in case of errors.
      */
-    public void collectStatistics(StatisticsTarget... targets) throws IgniteCheckedException;
+    public void collectStatistics(StatisticsObjectConfiguration... targets) throws IgniteCheckedException;
 
     /**
      * Clear object statistics.
@@ -74,4 +75,14 @@ public interface IgniteStatisticsManager {
      * @return Statistics usage state.
      */
     public StatisticsUsageState usageState();
+
+    /**
+     * To track statistics invalidation. Skip value if no statistics for the given table exists.
+     *
+     * @param schemaName Schema name.
+     * @param objName Object name.
+     * @param partId Partition id.
+     * @param keyBytes Row key bytes.
+     */
+    public void onRowUpdated(String schemaName, String objName, int partId, byte[] keyBytes);
 }

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

@@ -50,7 +50,7 @@ public StatisticsTarget(String schema, String obj, String... columns) {
      * @param key Statistic key.
      * @param columns Array of column names or {@code null} if target - all columns.
      */
-    public StatisticsTarget(StatisticsKey key, String[] columns) {
+    public StatisticsTarget(StatisticsKey key, String... columns) {
         this.key = key;
         this.columns = (columns == null || columns.length == 0) ? null : columns;
     }

modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/stat/config/StatisticsColumnConfiguration.java → modules/core/src/main/java/org/apache/ignite/internal/processors/query/stat/config/StatisticsColumnConfiguration.java

@@ -30,30 +30,43 @@ public class StatisticsColumnConfiguration implements Serializable {
     /** Columns name. */
     private final String name;
 
-    /** Configuration version. */
+    /** Collection version. */
     private final long ver;
 
     /** Tombstone flag: {@code true} statistic for this column is dropped, otherwise {@code false}. */
     private final boolean tombstone;
 
-    /** */
+    /**
+     * Constructor.
+     *
+     * @param name Column name.
+     */
     public StatisticsColumnConfiguration(String name) {
-        this(name, 0);
+        this(name, 1, false);
     }
 
-    /** */
-    public StatisticsColumnConfiguration(String name, long ver) {
-        this(name, ver, false);
-    }
-
-    /** */
+    /**
+     * Constructor.
+     *
+     * @param name Column name.
+     * @param ver Collection version.
+     * @param tombstone if {@code true} - object represents a tombstone of configuration,
+     *                  if {@code false} - live configuration.
+     */
     private StatisticsColumnConfiguration(String name, long ver, boolean tombstone) {
         this.name = name;
         this.ver = ver;
         this.tombstone = tombstone;
     }
 
-    /** */
+    /**
+     * Constructor.
+     *
+     * @param cfg Base statistics column configuration
+     * @param ver New collection version.
+     * @param tombstone if {@code true} - object represents a tombstone of configuration,
+     *                  if {@code false} - live configuration.
+     */
     private StatisticsColumnConfiguration(StatisticsColumnConfiguration cfg, long ver, boolean tombstone) {
         this.name = cfg.name;
         this.ver = ver;
@@ -70,9 +83,9 @@ public String name() {
     }
 
     /**
-     * Get configuration version.
+     * Get collection version.
      *
-     * @return Configuration version.
+     * @return Collection version.
      */
     public long version() {
         return ver;
@@ -96,12 +109,34 @@ public boolean tombstone() {
      */
     public static StatisticsColumnConfiguration merge(
         StatisticsColumnConfiguration oldCfg,
-        StatisticsColumnConfiguration newCfg)
-    {
+        StatisticsColumnConfiguration newCfg
+    ) {
         if (oldCfg == null)
             return newCfg;
 
-        return new StatisticsColumnConfiguration(newCfg.name, oldCfg.ver + 1);
+        if (oldCfg.collectionAwareEqual(newCfg))
+            return new StatisticsColumnConfiguration(newCfg, oldCfg.ver, oldCfg.tombstone);
+
+        return new StatisticsColumnConfiguration(newCfg, oldCfg.ver + 1, false);
+    }
+
+    /**
+     * Compare only collection or gathering related fields of config.
+     *
+     * @param o StatisticsColumnConfiguration to compare with.
+     * @return {@code true} if configurations are equal from the gathering point of view, {@code false} - otherwise.
+     */
+    public boolean collectionAwareEqual(StatisticsColumnConfiguration o) {
+        if (this == o)
+            return true;
+
+        if (o == null || getClass() != o.getClass())
+            return false;
+
+        StatisticsColumnConfiguration that = (StatisticsColumnConfiguration)o;
+
+        return ver == that.ver && tombstone == that.tombstone
+            && Objects.equals(name, that.name);
     }
 
     /**
@@ -111,7 +146,7 @@ public static StatisticsColumnConfiguration merge(
      */
     public StatisticsColumnConfiguration createTombstone()
     {
-        return new StatisticsColumnConfiguration(name, ver + 1, true);
+        return new StatisticsColumnConfiguration(this, ver + 1, true);
     }
 
     /**
@@ -134,13 +169,13 @@ public StatisticsColumnConfiguration refresh()
 
         StatisticsColumnConfiguration that = (StatisticsColumnConfiguration)o;
 
-        return ver == that.ver
+        return ver == that.ver && tombstone == that.tombstone
             && Objects.equals(name, that.name);
     }
 
     /** {@inheritDoc} */
     @Override public int hashCode() {
-        return Objects.hash(name, ver);
+        return Objects.hash(name, ver, tombstone);
     }
 
     /** {@inheritDoc} */

modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/stat/config/StatisticsObjectConfiguration.java → modules/core/src/main/java/org/apache/ignite/internal/processors/query/stat/config/StatisticsObjectConfiguration.java

@@ -16,10 +16,12 @@
 package org.apache.ignite.internal.processors.query.stat.config;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 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.Objects;
 import java.util.Set;
@@ -30,11 +32,15 @@
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.NotNull;
 
 /**
  * Describe configuration of the statistic for a database object (e.g. TABLE).
  */
 public class StatisticsObjectConfiguration implements Serializable {
+    /** Rows limit to renew partition statistics in percent. */
+    public static final byte DEFAULT_OBSOLESCENCE_MAX_PERCENT = 15;
+
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -46,28 +52,37 @@ public class StatisticsObjectConfiguration implements Serializable {
     @GridToStringInclude
     private final Map<String, StatisticsColumnConfiguration> cols;
 
-    /** */
+    /** Max percent of obsolescence row. */
+    private final byte maxPartitionObsolescencePercent;
+
+    /**
+     * Constructor.
+     *
+     * @param key Statistics key.
+     * @param cols Column statistics configuration.
+     * @param maxPartitionObsolescencePercent Maximum number of changed rows per partition.
+     */
     public StatisticsObjectConfiguration(
         StatisticsKey key,
-        Collection<StatisticsColumnConfiguration> cols
+        Collection<StatisticsColumnConfiguration> cols,
+        byte maxPartitionObsolescencePercent
     ) {
         this.key = key;
-        this.cols = cols.stream()
-            .collect(
-                Collectors.toMap(StatisticsColumnConfiguration::name, Function.identity())
-            );
+        this.cols = (cols == null) ? null : cols.stream()
+            .collect(Collectors.toMap(StatisticsColumnConfiguration::name, Function.identity()));
+        this.maxPartitionObsolescencePercent = maxPartitionObsolescencePercent;
     }
 
     /**
      * Merge configuration changes with existing configuration.
      *
-     * @param oldCfg Previous configuration. May be {@code null} when new configuration is created.
+     * @param oldCfg Previous configuration.
      * @param newCfg Contains target configuration changes.
      * @return merged configuration.
      */
     public static StatisticsObjectConfiguration merge(
-        StatisticsObjectConfiguration oldCfg,
-        StatisticsObjectConfiguration newCfg
+        @NotNull StatisticsObjectConfiguration oldCfg,
+        @NotNull StatisticsObjectConfiguration newCfg
     ) {
         assert oldCfg.key.equals(newCfg.key) : "Invalid stat config to merge: [oldKey=" + oldCfg.key
             + ", newKey=" + newCfg.key + ']';
@@ -77,7 +92,11 @@ public static StatisticsObjectConfiguration merge(
         for (StatisticsColumnConfiguration c : newCfg.cols.values())
             cols.put(c.name(), StatisticsColumnConfiguration.merge(cols.get(c.name()), c));
 
-        return new StatisticsObjectConfiguration(newCfg.key, cols.values());
+        for (StatisticsColumnConfiguration oldC : oldCfg.cols.values())
+            if (!cols.containsKey(oldC.name()))
+                cols.put(oldC.name(), oldC);
+
+        return new StatisticsObjectConfiguration(newCfg.key, cols.values(), newCfg.maxPartitionObsolescencePercent);
     }
 
     /**
@@ -97,25 +116,31 @@ public StatisticsObjectConfiguration dropColumns(Set<String> dropColNames) {
                 newCols.put(col.name(), col);
         }
 
-        return new StatisticsObjectConfiguration(key, newCols.values());
+        return new StatisticsObjectConfiguration(key, newCols.values(), maxPartitionObsolescencePercent);
     }
 
     /**
      * Creates new configuration object to refresh statistic with current configuration.
      *
+     * @param refreshCols Set of columns to refresh, if {@code null} or empty - all columns will be refreshed.
      * @return Result configuration object.
      */
-    public StatisticsObjectConfiguration refresh(Set<String> refreshColumns) {
-        Map<String, StatisticsColumnConfiguration> newCols = new HashMap<>();
-
-        for (StatisticsColumnConfiguration col : cols.values()) {
-            if (F.isEmpty(refreshColumns) || refreshColumns.contains(col.name()))
-                newCols.put(col.name(), col.refresh());
-            else
-                newCols.put(col.name(), col);
+    public StatisticsObjectConfiguration refresh(Set<String> refreshCols) {
+        List<StatisticsColumnConfiguration> newCols;
+
+        if (F.isEmpty(refreshCols))
+            newCols = new ArrayList<>(cols.values());
+        else {
+            newCols = new ArrayList<>(cols.size());
+            for (StatisticsColumnConfiguration col : cols.values()) {
+                if (refreshCols.contains(col.name()))
+                    newCols.add(col.refresh());
+                else
+                    newCols.add(col);
+            }
         }
 
-        return new StatisticsObjectConfiguration(key, newCols.values());
+        return new StatisticsObjectConfiguration(key, newCols, maxPartitionObsolescencePercent);
     }
 
     /**
@@ -170,11 +195,18 @@ public Map<String, StatisticsColumnConfiguration> columnsAll() {
      * @return Map column name to column statistics configuration.
      */
     public Map<String, StatisticsColumnConfiguration> columns() {
-        return cols.entrySet().stream()
+        return (cols == null) ? Collections.emptyMap() : cols.entrySet().stream()
             .filter(e -> !e.getValue().tombstone())
             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
 
+    /**
+     * @return Maximum number of changed rows per partition.
+     */
+    public byte maxPartitionObsolescencePercent() {
+        return maxPartitionObsolescencePercent;
+    }
+
     /** {@inheritDoc} */
     @Override public boolean equals(Object o) {
         if (this == o)
@@ -186,12 +218,13 @@ public Map<String, StatisticsColumnConfiguration> columns() {
         StatisticsObjectConfiguration that = (StatisticsObjectConfiguration)o;
 
         return Objects.equals(key, that.key)
-            && Objects.equals(cols, that.cols);
+            && Objects.equals(cols, that.cols)
+            && maxPartitionObsolescencePercent == that.maxPartitionObsolescencePercent;
     }
 
     /** {@inheritDoc} */
     @Override public int hashCode() {
-        return Objects.hash(key, cols);
+        return Objects.hash(key, cols, maxPartitionObsolescencePercent);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/processors/subscription/GridInternalSubscriptionProcessor.java

@@ -78,13 +78,6 @@ public void registerDistributedMetastorageListener(@NotNull DistributedMetastora
         distributedMetastorageListeners.add(lsnr);
     }
 
-    /** */
-    public void unregisterDistributedMetastorageListener(@NotNull DistributedMetastorageLifecycleListener lsnr) {
-        requireNonNull(lsnr, "Global metastorage subscriber should be not-null.");
-
-        distributedMetastorageListeners.remove(lsnr);
-    }
-
     /** */
     public List<DistributedMetastorageLifecycleListener> getDistributedMetastorageSubscribers() {
         return distributedMetastorageListeners;