fix(interactive): invalidate query cache after schema update (#4486)

<!--
Thanks for your contribution! please review
https://github.com/alibaba/GraphScope/blob/main/CONTRIBUTING.md before
opening an issue.
-->

## What do these changes do?
as titled.

<!-- Please give a short brief about these changes. -->

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this
change? -->

Fixes #4483

Author: shirly121

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/GraphServer.java

@@ -41,6 +41,7 @@
 import com.alibaba.graphscope.gremlin.integration.result.GraphProperties;
 import com.alibaba.graphscope.gremlin.integration.result.TestGraphFactory;
 import com.alibaba.graphscope.gremlin.service.IrGremlinServer;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.Resources;
 
@@ -56,6 +57,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.List;
 
 public class GraphServer {
     private static final Logger logger = LoggerFactory.getLogger(GraphServer.class);
@@ -65,6 +67,7 @@ public class GraphServer {
     private final GraphProperties testGraph;
     private final GraphRelOptimizer optimizer;
     private final MetricsTool metricsTool;
+    private final QueryCache queryCache;
 
     private IrGremlinServer gremlinServer;
     private CypherBootstrapper cypherBootstrapper;
@@ -74,21 +77,22 @@ public GraphServer(
             ChannelFetcher channelFetcher,
             IrMetaQueryCallback metaQueryCallback,
             GraphProperties testGraph,
-            GraphRelOptimizer optimizer) {
+            GraphRelOptimizer optimizer,
+            QueryCache queryCache) {
         this.configs = configs;
         this.channelFetcher = channelFetcher;
         this.metaQueryCallback = metaQueryCallback;
         this.testGraph = testGraph;
         this.optimizer = optimizer;
         this.metricsTool = new MetricsTool(configs);
         this.metricsTool.registerMetric(new MemoryMetric());
+        this.queryCache = queryCache;
     }
 
     public void start() throws Exception {
         ExecutionClient executionClient =
                 ExecutionClient.Factory.create(configs, channelFetcher, metricsTool);
         QueryIdGenerator idGenerator = new QueryIdGenerator(configs);
-        QueryCache queryCache = new QueryCache(configs);
         if (!FrontendConfig.GREMLIN_SERVER_DISABLED.get(configs)) {
             GraphPlanner graphPlanner =
                     new GraphPlanner(configs, new LogicalPlanFactory.Gremlin(), optimizer);
@@ -177,19 +181,22 @@ public static void main(String[] args) throws Exception {
         }
         Configs configs = Configs.Factory.create(args[0]);
         GraphRelOptimizer optimizer = new GraphRelOptimizer(configs);
+        QueryCache queryCache = new QueryCache(configs);
         IrMetaQueryCallback queryCallback =
-                new IrMetaQueryCallback(createIrMetaFetcher(configs, optimizer.getGlogueHolder()));
+                new IrMetaQueryCallback(
+                        createIrMetaFetcher(configs, ImmutableList.of(optimizer, queryCache)));
         GraphServer server =
                 new GraphServer(
                         configs,
                         getChannelFetcher(configs),
                         queryCallback,
                         getTestGraph(configs),
-                        optimizer);
+                        optimizer,
+                        queryCache);
         server.start();
     }
 
-    private static IrMetaFetcher createIrMetaFetcher(Configs configs, IrMetaTracker tracker)
+    private static IrMetaFetcher createIrMetaFetcher(Configs configs, List<IrMetaTracker> tracker)
             throws IOException {
         URI schemaUri = URI.create(GraphConfig.GRAPH_META_SCHEMA_URI.get(configs));
         if (schemaUri.getScheme() == null || schemaUri.getScheme().equals("file")) {

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/IrMetaTracker.java

@@ -20,5 +20,7 @@
 
 // This interface defines the callback to be invoked when IrMeta changes.
 public interface IrMetaTracker {
-    void onChanged(IrMeta meta);
+    void onSchemaChanged(IrMeta meta);
+
+    void onStatsChanged(IrMetaStats stats);
 }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/fetcher/DynamicIrMetaFetcher.java

@@ -26,11 +26,13 @@
 import com.alibaba.graphscope.common.ir.meta.IrMetaStats;
 import com.alibaba.graphscope.common.ir.meta.IrMetaTracker;
 import com.alibaba.graphscope.common.ir.meta.reader.IrMetaReader;
+import com.alibaba.graphscope.common.ir.meta.schema.SchemaSpec;
 import com.alibaba.graphscope.groot.common.schema.api.GraphStatistics;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -48,7 +50,8 @@ public class DynamicIrMetaFetcher extends IrMetaFetcher implements AutoCloseable
     private volatile StatsState statsState;
     private volatile Boolean statsEnabled = null;
 
-    public DynamicIrMetaFetcher(Configs configs, IrMetaReader dataReader, IrMetaTracker tracker) {
+    public DynamicIrMetaFetcher(
+            Configs configs, IrMetaReader dataReader, List<IrMetaTracker> tracker) {
         super(dataReader, tracker);
         this.scheduler = new ScheduledThreadPoolExecutor(1);
         long schemaIntervalMS = GraphConfig.GRAPH_META_SCHEMA_FETCH_INTERVAL_MS.get(configs);
@@ -93,6 +96,12 @@ private synchronized void syncMeta() {
                                 meta.getSchema(),
                                 meta.getStoredProcedures(),
                                 null);
+                if (logger.isDebugEnabled()) {
+                    logger.debug(
+                            "sync schema with {}",
+                            meta.getSchema().getSchemaSpec(SchemaSpec.Type.FLEX_IN_YAML));
+                }
+                tracker.forEach(t -> t.onSchemaChanged(this.currentState));
             }
             boolean statsEnabled = getStatsEnabled(this.currentState.getGraphId());
             if (statsEnabled && this.statsState != StatsState.SYNCED
@@ -135,7 +144,7 @@ private synchronized void syncStats() {
                                         stats);
                         if (tracker != null) {
                             logger.info("start to update the glogue");
-                            tracker.onChanged(this.currentState);
+                            tracker.forEach(t -> t.onStatsChanged(this.currentState));
                         }
                         this.statsState = StatsState.SYNCED;
                     }
@@ -149,7 +158,7 @@ private synchronized void syncStats() {
                         && tracker != null
                         && this.statsState == StatsState.INITIALIZED) {
                     logger.info("start to mock the glogue");
-                    tracker.onChanged(this.currentState);
+                    tracker.forEach(t -> t.onStatsChanged(this.currentState));
                     this.statsState = StatsState.MOCKED;
                 }
             } catch (Throwable t) {

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/fetcher/IrMetaFetcher.java

@@ -22,6 +22,7 @@
 import com.alibaba.graphscope.common.ir.meta.IrMetaTracker;
 import com.alibaba.graphscope.common.ir.meta.reader.IrMetaReader;
 
+import java.util.List;
 import java.util.Optional;
 
 /**
@@ -31,9 +32,9 @@
  */
 public abstract class IrMetaFetcher {
     protected final IrMetaReader reader;
-    protected final IrMetaTracker tracker;
+    protected final List<IrMetaTracker> tracker;
 
-    protected IrMetaFetcher(IrMetaReader reader, IrMetaTracker tracker) {
+    protected IrMetaFetcher(IrMetaReader reader, List<IrMetaTracker> tracker) {
         this.reader = reader;
         this.tracker = tracker;
     }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/fetcher/StaticIrMetaFetcher.java

@@ -29,21 +29,31 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Optional;
 
 public class StaticIrMetaFetcher extends IrMetaFetcher {
     private static final Logger logger = LoggerFactory.getLogger(StaticIrMetaFetcher.class);
     private final IrMetaStats metaStats;
 
-    public StaticIrMetaFetcher(IrMetaReader dataReader, IrMetaTracker tracker) throws IOException {
+    public StaticIrMetaFetcher(IrMetaReader dataReader, List<IrMetaTracker> tracker)
+            throws IOException {
         super(dataReader, tracker);
         IrMeta meta = this.reader.readMeta();
-        GraphStatistics stats = (meta == null) ? null : fetchStats(meta);
+        tracker.forEach(t -> t.onSchemaChanged(meta));
+        GraphStatistics stats = null;
+        if (meta != null) {
+            try {
+                stats = fetchStats(meta);
+            } catch (Exception e) {
+                logger.error("failed to fetch graph statistics, error is {}", e);
+            }
+        }
         this.metaStats =
                 new IrMetaStats(
                         meta.getSnapshotId(), meta.getSchema(), meta.getStoredProcedures(), stats);
         if (tracker != null && stats != null) {
-            tracker.onChanged(this.metaStats);
+            tracker.forEach(t -> t.onStatsChanged(this.metaStats));
         }
     }
 

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GlogueHolder.java

@@ -5,10 +5,13 @@
 import com.alibaba.graphscope.common.ir.rel.graph.*;
 import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalMultiMatch;
 import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalSingleMatch;
+import com.alibaba.graphscope.gremlin.Utils;
 
+import org.apache.calcite.plan.RelDigest;
 import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.plan.hep.HepPlanner;
 import org.apache.calcite.plan.hep.HepProgram;
+import org.apache.calcite.plan.hep.HepRelVertex;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.logical.LogicalFilter;
 import org.apache.calcite.rel.logical.LogicalJoin;
@@ -20,6 +23,7 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.IdentityHashMap;
+import java.util.Map;
 
 /**
  * Original {@code HepPlanner} skip optimizations to the nested structures, i.e. {@code RelNode} nested in the {@code CommonTableScan} or {@code RexSubQuery},
@@ -173,4 +177,12 @@ private RelNode findBestIfRoot(RelNode oldRel, RelNode newRel) {
             return oldRel == root ? findBestExpOfRoot(newRel) : newRel;
         }
     }
+
+    @Override
+    public void clear() {
+        super.clear();
+        Map<RelDigest, HepRelVertex> mapDigestToVertex =
+                Utils.getFieldValue(HepPlanner.class, this, "mapDigestToVertex");
+        mapDigestToVertex.clear();
+    }
 }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphHepPlanner.java

@@ -19,14 +19,18 @@
 import com.alibaba.graphscope.common.config.Configs;
 import com.alibaba.graphscope.common.config.PlannerConfig;
 import com.alibaba.graphscope.common.ir.meta.IrMeta;
+import com.alibaba.graphscope.common.ir.meta.IrMetaStats;
+import com.alibaba.graphscope.common.ir.meta.IrMetaTracker;
 import com.alibaba.graphscope.common.ir.meta.glogue.calcite.GraphRelMetadataQuery;
 import com.alibaba.graphscope.common.ir.meta.glogue.calcite.handler.GraphMetadataHandlerProvider;
 import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalSource;
 import com.alibaba.graphscope.common.ir.rel.graph.match.AbstractLogicalMatch;
 import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalMultiMatch;
 import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalSingleMatch;
+import com.alibaba.graphscope.common.ir.rel.metadata.glogue.Glogue;
 import com.alibaba.graphscope.common.ir.rel.metadata.glogue.GlogueQuery;
+import com.alibaba.graphscope.common.ir.rel.metadata.schema.GlogueSchema;
 import com.alibaba.graphscope.common.ir.tools.GraphBuilderFactory;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 import com.google.common.base.Preconditions;
@@ -43,28 +47,33 @@
 import org.apache.calcite.rel.metadata.RelMetadataQuery;
 import org.apache.calcite.tools.RelBuilderFactory;
 import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Optimize graph relational tree which consists of match and other relational operators
  */
-public class GraphRelOptimizer implements Closeable {
+public class GraphRelOptimizer implements Closeable, IrMetaTracker {
+    private static final Logger logger = LoggerFactory.getLogger(GraphRelOptimizer.class);
     private final PlannerConfig config;
     private final RelBuilderFactory relBuilderFactory;
-    private final GlogueHolder glogueHolder;
     private final PlannerGroupManager plannerGroupManager;
 
+    private final AtomicReference<GlogueQuery> glogueRef;
+
     public GraphRelOptimizer(Configs graphConfig, Class<? extends PlannerGroupManager> instance) {
         try {
             this.config = new PlannerConfig(graphConfig);
             this.relBuilderFactory = new GraphBuilderFactory(graphConfig);
-            this.glogueHolder = new GlogueHolder(graphConfig);
             this.plannerGroupManager =
                     instance.getDeclaredConstructor(PlannerConfig.class, RelBuilderFactory.class)
                             .newInstance(this.config, this.relBuilderFactory);
+            this.glogueRef = new AtomicReference<>();
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
@@ -74,10 +83,6 @@ public GraphRelOptimizer(Configs graphConfig) {
         this(graphConfig, PlannerGroupManager.Dynamic.class);
     }
 
-    public GlogueHolder getGlogueHolder() {
-        return glogueHolder;
-    }
-
     public RelOptPlanner getMatchPlanner() {
         PlannerGroup currentGroup = this.plannerGroupManager.getCurrentGroup();
         return currentGroup.getMatchPlanner();
@@ -90,7 +95,7 @@ public RelNode optimize(RelNode before, GraphIOProcessor ioProcessor) {
 
     public @Nullable RelMetadataQuery createMetaDataQuery(IrMeta irMeta) {
         if (config.isOn() && config.getOpt() == PlannerConfig.Opt.CBO) {
-            GlogueQuery gq = this.glogueHolder.getGlogue();
+            GlogueQuery gq = this.glogueRef.get();
             Preconditions.checkArgument(gq != null, "glogue is not ready");
             return new GraphRelMetadataQuery(
                     new GraphMetadataHandlerProvider(getMatchPlanner(), gq, this.config));
@@ -105,6 +110,21 @@ public void close() {
         }
     }
 
+    @Override
+    public void onSchemaChanged(IrMeta meta) {
+        if (this.plannerGroupManager != null) {
+            this.plannerGroupManager.clean();
+        }
+    }
+
+    @Override
+    public void onStatsChanged(IrMetaStats stats) {
+        GlogueSchema g = GlogueSchema.fromMeta(stats);
+        Glogue gl = new Glogue(g, config.getGlogueSize());
+        GlogueQuery gq = new GlogueQuery(gl);
+        this.glogueRef.compareAndSet(glogueRef.get(), gq);
+    }
+
     public static class MatchOptimizer extends GraphShuttle {
         private final GraphIOProcessor ioProcessor;
         private final RelOptPlanner matchPlanner;

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphRelOptimizer.java

@@ -188,7 +188,7 @@ public synchronized void clear() {
             this.matchPlanner.addRule(rule);
         }
         List<RelOptRule> physicalRBORules = this.physicalPlanner.getRules();
-        this.physicalPlanner.clear();
+        physicalPlanner.clear();
         for (RelOptRule rule : physicalRBORules) {
             this.physicalPlanner.addRule(rule);
         }