[improvement](multi-catalog)add last sync time for external catalog (a…

…pache#21873) which operation can update this time: 1.when refresh catalog,lastUpdateTime of catalog will be update 2.when refresh db,lastUpdateTime of db will be update 3.when reload table schema to cache,lastUpdateTime of dbtable will be update 4.when receive add/drop table event,lastUpdateTime of db will be update 5.when receive alter table event,lastUpdateTime of table will be update
xiedeyantu · Jul 21, 2023 · 0b2b1cb · 0b2b1cb
1 parent 6875ef4
commit 0b2b1cb
Show file tree

Hide file tree

Showing 19 changed files with 372 additions and 83 deletions.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowCatalogStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowCatalogStmt.java
@@ -32,6 +32,7 @@ public class ShowCatalogStmt extends ShowStmt {
  .addColumn(new Column("Type", ScalarType.createStringType()))
  .addColumn(new Column("IsCurrent", ScalarType.createStringType()))
  .addColumn(new Column("CreateTime", ScalarType.createStringType()))
+ .addColumn(new Column("LastUpdateTime", ScalarType.createStringType()))
  .addColumn(new Column("Comment", ScalarType.createStringType()))
  .build();
 

diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/DatabaseIf.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/DatabaseIf.java
@@ -235,4 +235,8 @@ default OlapTable getOlapTableOrAnalysisException(String tableName) throws Analy
  void dropTable(String tableName);
 
  CatalogIf getCatalog();
+
+ default long getLastUpdateTime() {
+ return -1L;
+ }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java
@@ -232,5 +232,9 @@ default Partition getPartition(String name) {
  default boolean isManagedTable() {
  return getType() == TableType.OLAP || getType() == TableType.MATERIALIZED_VIEW;
  }
+
+ default long getLastUpdateTime() {
+ return -1L;
+ }
 }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalDatabase.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalDatabase.java
@@ -73,6 +73,8 @@ public abstract class ExternalDatabase<T extends ExternalTable>
  protected Map<String, Long> tableNameToId = Maps.newConcurrentMap();
  @SerializedName(value = "idToTbl")
  protected Map<Long, T> idToTbl = Maps.newConcurrentMap();
+ @SerializedName(value = "lastUpdateTime")
+ protected long lastUpdateTime;
  protected final InitDatabaseLog.Type dbLogType;
  protected ExternalCatalog extCatalog;
  protected boolean invalidCacheInInit = true;
@@ -147,6 +149,7 @@ public void replayInitDb(InitDatabaseLog log, ExternalCatalog catalog) {
  }
  tableNameToId = tmpTableNameToId;
  idToTbl = tmpIdToTbl;
+ lastUpdateTime = log.getLastUpdateTime();
  initialized = true;
  }
 
@@ -179,6 +182,10 @@ protected void init() {
  tableNameToId = tmpTableNameToId;
  idToTbl = tmpIdToTbl;
  }
+
+ long currentTime = System.currentTimeMillis();
+ lastUpdateTime = currentTime;
+ initDatabaseLog.setLastUpdateTime(lastUpdateTime);
  initialized = true;
  Env.getCurrentEnv().getEditLog().logInitExternalDb(initDatabaseLog);
  }
@@ -307,6 +314,14 @@ public T getTableNullable(long tableId) {
  return idToTbl.get(tableId);
  }
 
+ public long getLastUpdateTime() {
+ return lastUpdateTime;
+ }
+
+ public void setLastUpdateTime(long lastUpdateTime) {
+ this.lastUpdateTime = lastUpdateTime;
+ }
+
  @Override
  public void write(DataOutput out) throws IOException {
  Text.writeString(out, GsonUtils.GSON.toJson(this));

diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalTable.java
@@ -69,6 +69,8 @@ public class ExternalTable implements TableIf, Writable, GsonPostProcessable {
  protected long timestamp;
  @SerializedName(value = "dbName")
  protected String dbName;
+ @SerializedName(value = "lastUpdateTime")
+ protected long lastUpdateTime;
 
  protected boolean objectCreated;
  protected ExternalCatalog catalog;
@@ -342,6 +344,11 @@ public Optional<ColumnStatistic> getColumnStatistic(String colName) {
  *
  * @return
  */
+ public List<Column> initSchemaAndUpdateTime() {
+ lastUpdateTime = System.currentTimeMillis();
+ return initSchema();
+ }
+
  public List<Column> initSchema() {
  throw new NotImplementedException("implement in sub class");
  }

diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/CatalogsProcDir.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/CatalogsProcDir.java
@@ -20,6 +20,7 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.util.ListComparator;
+import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.datasource.CatalogIf;
 
 import com.google.common.base.Preconditions;
@@ -37,7 +38,7 @@
  */
 public class CatalogsProcDir implements ProcDirInterface {
  public static final ImmutableList<String> TITLE_NAMES = new ImmutableList.Builder<String>()
- .add("CatalogIds").add("CatalogName").add("DatabaseNum")
+ .add("CatalogIds").add("CatalogName").add("DatabaseNum").add("LastUpdateTime")
  .build();
 
  private Env env;
@@ -90,6 +91,7 @@ public ProcResult fetchResult() throws AnalysisException {
  catalogInfo.add(catalog.getId());
  catalogInfo.add(catalog.getName());
  catalogInfo.add(catalog.getDbNames().size());
+ catalogInfo.add(TimeUtils.longToTimeString(catalog.getLastUpdateTime()));
  catalogInfos.add(catalogInfo);
  }
 

diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/DbsProcDir.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/DbsProcDir.java
@@ -43,7 +43,7 @@ public class DbsProcDir implements ProcDirInterface {
  public static final ImmutableList<String> TITLE_NAMES = new ImmutableList.Builder<String>()
  .add("DbId").add("DbName").add("TableNum").add("Size").add("Quota")
  .add("LastConsistencyCheckTime").add("ReplicaCount").add("ReplicaQuota")
- .add("TransactionQuota")
+ .add("TransactionQuota").add("LastUpdateTime")
  .build();
 
  private Env env;
@@ -122,7 +122,7 @@ public ProcResult fetchResult() throws AnalysisException {
  dbInfo.add(replicaCount);
  dbInfo.add(replicaQuota);
  dbInfo.add(transactionQuota);
-
+ dbInfo.add(TimeUtils.longToTimeString(db.getLastUpdateTime()));
  } finally {
  db.readUnlock();
  }

diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/TablesProcDir.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/TablesProcDir.java
@@ -45,6 +45,7 @@ public class TablesProcDir implements ProcDirInterface {
  public static final ImmutableList<String> TITLE_NAMES = new ImmutableList.Builder<String>()
  .add("TableId").add("TableName").add("IndexNum").add("PartitionColumnName")
  .add("PartitionNum").add("State").add("Type").add("LastConsistencyCheckTime").add("ReplicaCount")
+ .add("LastUpdateTime")
  .build();
 
  private DatabaseIf db;
@@ -128,8 +129,7 @@ public ProcResult fetchResult() throws AnalysisException {
  tableInfo.add(FeConstants.null_string);
  tableInfo.add(replicaCount);
  }
-
-
+ tableInfo.add(TimeUtils.longToTimeString(table.getLastUpdateTime()));
  tableInfos.add(tableInfo);
  } finally {
  table.readUnlock();

diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java
@@ -153,6 +153,10 @@ default void onClose() {
 
  String getComment();
 
+ default long getLastUpdateTime() {
+ return -1L;
+ }
+
  default CatalogLog constructEditLog() {
  CatalogLog log = new CatalogLog();
  log.setCatalogId(getId());

diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java
@@ -43,6 +43,7 @@
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.util.PrintableMap;
+import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.persist.OperationType;
 import org.apache.doris.persist.gson.GsonPostProcessable;
@@ -378,7 +379,6 @@ public ShowResultSet showCatalogs(ShowCatalogStmt showStmt, String currentCtlg)
  matcher = PatternMatcherWrapper.createMysqlPattern(showStmt.getPattern(),
  CaseSensibility.CATALOG.getCaseSensibility());
  }
-
  for (CatalogIf catalog : nameToCatalog.values()) {
  if (Env.getCurrentEnv().getAccessManager()
  .checkCtlPriv(ConnectContext.get(), catalog.getName(), PrivPredicate.SHOW)) {
@@ -399,6 +399,7 @@ public ShowResultSet showCatalogs(ShowCatalogStmt showStmt, String currentCtlg)
  Map<String, String> props = catalog.getProperties();
  String createTime = props.getOrDefault(CreateCatalogStmt.CREATE_TIME_PROP, "UNRECORDED");
  row.add(createTime);
+ row.add(TimeUtils.longToTimeString(catalog.getLastUpdateTime()));
  row.add(catalog.getComment());
  rows.add(row);
  }
@@ -714,6 +715,7 @@ public void dropExternalTable(String dbName, String tableName, String catalogNam
  log.setCatalogId(catalog.getId());
  log.setDbId(db.getId());
  log.setTableId(table.getId());
+ log.setLastUpdateTime(System.currentTimeMillis());
  replayDropExternalTable(log);
  Env.getCurrentEnv().getEditLog().logDropExternalTable(log);
  }
@@ -739,6 +741,7 @@ public void replayDropExternalTable(ExternalObjectLog log) {
  db.writeLock();
  try {
  db.dropTable(table.getName());
+ db.setLastUpdateTime(log.getLastUpdateTime());
  } finally {
  db.writeUnlock();
  }
@@ -788,6 +791,7 @@ public void createExternalTableFromEvent(String dbName, String tableName, String
  log.setDbId(db.getId());
  log.setTableName(tableName);
  log.setTableId(Env.getCurrentEnv().getNextId());
+ log.setLastUpdateTime(System.currentTimeMillis());
  replayCreateExternalTableFromEvent(log);
  Env.getCurrentEnv().getEditLog().logCreateExternalTable(log);
  }
@@ -808,6 +812,7 @@ public void replayCreateExternalTableFromEvent(ExternalObjectLog log) {
  db.writeLock();
  try {
  db.replayCreateTableFromEvent(log.getTableName(), log.getTableId());
+ db.setLastUpdateTime(log.getLastUpdateTime());
  } finally {
  db.writeUnlock();
  }

diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
@@ -87,6 +87,8 @@ public abstract class ExternalCatalog
  private boolean initialized = false;
  @SerializedName(value = "idToDb")
  protected Map<Long, ExternalDatabase<? extends ExternalTable>> idToDb = Maps.newConcurrentMap();
+ @SerializedName(value = "lastUpdateTime")
+ protected long lastUpdateTime;
  // db name does not contains "default_cluster"
  protected Map<String, Long> dbNameToId = Maps.newConcurrentMap();
  private boolean objectCreated = false;
@@ -260,6 +262,9 @@ protected void init() {
  }
  dbNameToId = tmpDbNameToId;
  idToDb = tmpIdToDb;
+ long currentTime = System.currentTimeMillis();
+ lastUpdateTime = currentTime;
+ initCatalogLog.setLastUpdateTime(lastUpdateTime);
  Env.getCurrentEnv().getEditLog().logInitCatalog(initCatalogLog);
  }
 
@@ -282,7 +287,7 @@ public final List<Column> getSchema(String dbName, String tblName) {
  if (db.isPresent()) {
  Optional<? extends ExternalTable> table = db.get().getTable(tblName);
  if (table.isPresent()) {
- return table.get().initSchema();
+ return table.get().initSchemaAndUpdateTime();
  }
  }
  // return one column with unsupported type.
@@ -406,6 +411,14 @@ private void modifyComment(Map<String, String> props) {
  props.remove("comment");
  }
 
+ public long getLastUpdateTime() {
+ return lastUpdateTime;
+ }
+
+ public void setLastUpdateTime(long lastUpdateTime) {
+ this.lastUpdateTime = lastUpdateTime;
+ }
+
  @Override
  public void write(DataOutput out) throws IOException {
  Text.writeString(out, GsonUtils.GSON.toJson(this));
@@ -440,6 +453,7 @@ public void replayInitCatalog(InitCatalogLog log) {
  }
  dbNameToId = tmpDbNameToId;
  idToDb = tmpIdToDb;
+ lastUpdateTime = log.getLastUpdateTime();
  initialized = true;
  }
 

diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalObjectLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalObjectLog.java
@@ -56,6 +56,9 @@ public class ExternalObjectLog implements Writable {
  @SerializedName(value = "partitionNames")
  private List<String> partitionNames;
 
+ @SerializedName(value = "lastUpdateTime")
+ private long lastUpdateTime;
+
  @Override
  public void write(DataOutput out) throws IOException {
  Text.writeString(out, GsonUtils.GSON.toJson(this));

diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InitCatalogLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InitCatalogLog.java
@@ -65,6 +65,9 @@ public enum Type {
  @SerializedName(value = "type")
  private Type type;
 
+ @SerializedName(value = "lastUpdateTime")
+ private long lastUpdateTime;
+
  public InitCatalogLog() {
  refreshCount = 0;
  createCount = 0;

diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InitDatabaseLog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InitDatabaseLog.java
@@ -68,6 +68,9 @@ public enum Type {
  @SerializedName(value = "type")
  private Type type;
 
+ @SerializedName(value = "lastUpdateTime")
+ protected long lastUpdateTime;
+
  public InitDatabaseLog() {
  refreshCount = 0;
  createCount = 0;

diff --git a/fe/fe-core/src/test/java/org/apache/doris/analysis/ShowCatalogStmtTest.java b/fe/fe-core/src/test/java/org/apache/doris/analysis/ShowCatalogStmtTest.java
@@ -26,18 +26,18 @@
 public class ShowCatalogStmtTest {
  @Test
  public void testNormal() throws UserException, AnalysisException {
- final Analyzer analyzer =  AccessTestUtil.fetchBlockAnalyzer();
+ final Analyzer analyzer = AccessTestUtil.fetchBlockAnalyzer();
  ShowCatalogStmt stmt = new ShowCatalogStmt();
  stmt.analyze(analyzer);
  Assert.assertNull(stmt.getCatalogName());
- Assert.assertEquals(6, stmt.getMetaData().getColumnCount());
+ Assert.assertEquals(7, stmt.getMetaData().getColumnCount());
  Assert.assertEquals("SHOW CATALOGS", stmt.toSql());
 
  stmt = new ShowCatalogStmt(null, "%hive%");
  stmt.analyze(analyzer);
  Assert.assertNull(stmt.getCatalogName());
  Assert.assertNotNull(stmt.getPattern());
- Assert.assertEquals(6, stmt.getMetaData().getColumnCount());
+ Assert.assertEquals(7, stmt.getMetaData().getColumnCount());
  Assert.assertEquals("SHOW CATALOGS LIKE '%hive%'", stmt.toSql());
 
  stmt = new ShowCatalogStmt("testCatalog", null);