Added version for dataNode table cache

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java

@@ -41,6 +41,8 @@
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -53,6 +55,8 @@ public class DataNodeTableCache implements ITableCache {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeTableCache.class);
 
+  private final AtomicLong version = new AtomicLong(0);
+
   // The database is without "root"
   private final Map<String, Map<String, TsTable>> databaseTableMap = new ConcurrentHashMap<>();
 
@@ -175,6 +179,7 @@ public void commitUpdateTable(String database, final String tableName) {
           .computeIfAbsent(database, k -> new ConcurrentHashMap<>())
           .put(tableName, preUpdateTableMap.get(database).get(tableName).getLeft());
       removeTableFromPreUpdateMap(database, tableName);
+      version.incrementAndGet();
       LOGGER.info("Commit-update table {}.{} successfully", database, tableName);
     } finally {
       readWriteLock.writeLock().unlock();
@@ -188,6 +193,7 @@ public void invalid(String database) {
     try {
       databaseTableMap.remove(database);
       preUpdateTableMap.remove(database);
+      version.incrementAndGet();
     } finally {
       readWriteLock.writeLock().unlock();
     }
@@ -205,6 +211,7 @@ public void invalid(String database, final String tableName) {
       if (preUpdateTableMap.containsKey(database)) {
         preUpdateTableMap.get(database).remove(tableName);
       }
+      version.incrementAndGet();
     } finally {
       readWriteLock.writeLock().unlock();
     }
@@ -228,11 +235,16 @@ public void invalid(String database, final String tableName, final String column
         }
         tableVersionPair.setRight(tableVersionPair.getRight() + 1);
       }
+      version.incrementAndGet();
     } finally {
       readWriteLock.writeLock().unlock();
     }
   }
 
+  public long getVersion() {
+    return version.get();
+  }
+
   public TsTable getTableInWrite(final String database, final String tableName) {
     final TsTable result = getTableInCache(database, tableName);
     return Objects.nonNull(result) ? result : getTable(database, tableName);
@@ -316,6 +328,7 @@ private void updateTable(
       final Map<String, Map<String, Long>> previousVersions) {
     readWriteLock.writeLock().lock();
     try {
+      final AtomicBoolean isUpdated = new AtomicBoolean(false);
       fetchedTables.forEach(
           (database, tableInfoMap) -> {
             if (preUpdateTableMap.containsKey(database)) {
@@ -329,6 +342,7 @@ private void updateTable(
                             previousVersions.get(database).get(tableName))) {
                       return;
                     }
+                    isUpdated.set(true);
                     LOGGER.info(
                         "Update table {}.{} by table fetch, table in preUpdateMap: {}, new table: {}",
                         database,
@@ -346,6 +360,9 @@ private void updateTable(
                   });
             }
           });
+      if (isUpdated.get()) {
+        version.incrementAndGet();
+      }
     } finally {
       readWriteLock.writeLock().unlock();
     }