[Refactor] Synchronize OLAP external table metadata when loading data (…

…StarRocks#24739)

Currently, StarRocks use StarRocksRepository to synchronize OLAP
external table metadata every 10 seconds, which is
inefficient, because the meta data will never be used when there is no
load job on that external table. Change the synchronization to be
triggered when there is load job on that table.

Since the synchronization is time costly, so we should release the
database lock when synchronizing meta data.

Signed-off-by: gengjun-git <gengjun@starrocks.com>

fe/fe-core/src/main/java/com/starrocks/catalog/Database.java

@@ -240,6 +240,10 @@ public void readUnlock() {
         this.rwLock.readLock().unlock();
     }
 
+    public boolean isReadLockHeldByCurrentThread() {
+        return this.rwLock.getReadHoldCount() > 0;
+    }
+
     public void writeLock() {
         long startMs = TimeUnit.MILLISECONDS.convert(System.nanoTime(), TimeUnit.NANOSECONDS);
         String threadDump = getOwnerInfo(rwLock.getOwner());

fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java

@@ -289,8 +289,7 @@ public OlapTable(long id, String tableName, List<Column> baseSchema, KeysType ke
 
     // Only Copy necessary metadata for query.
     // We don't do deep copy, because which is very expensive;
-    public OlapTable copyOnlyForQuery() {
-        OlapTable olapTable = new OlapTable();
+    public void copyOnlyForQuery(OlapTable olapTable) {
         olapTable.id = this.id;
         olapTable.name = this.name;
         olapTable.fullSchema = Lists.newArrayList(this.fullSchema);
@@ -321,7 +320,6 @@ public OlapTable copyOnlyForQuery() {
         if (this.tableProperty != null) {
             olapTable.tableProperty = this.tableProperty.copy();
         }
-        return olapTable;
     }
 
     public BinlogConfig getCurBinlogConfig() {

fe/fe-core/src/main/java/com/starrocks/external/starrocks/TableMetaSyncer.java

@@ -18,6 +18,7 @@
 import com.starrocks.catalog.ExternalOlapTable;
 import com.starrocks.common.Config;
 import com.starrocks.rpc.FrontendServiceProxy;
+import com.starrocks.sql.common.MetaNotFoundException;
 import com.starrocks.thrift.TGetTableMetaRequest;
 import com.starrocks.thrift.TGetTableMetaResponse;
 import com.starrocks.thrift.TNetworkAddress;
@@ -30,7 +31,7 @@
 public class TableMetaSyncer {
     private static final Logger LOG = LogManager.getLogger(TableMetaSyncer.class);
 
-    public void syncTable(ExternalOlapTable table) {
+    public void syncTable(ExternalOlapTable table) throws MetaNotFoundException {
         String host = table.getSourceTableHost();
         int port = table.getSourceTablePort();
         TNetworkAddress addr = new TNetworkAddress(host, port);
@@ -50,11 +51,13 @@ public void syncTable(ExternalOlapTable table) {
                     errMsg = "";
                 }
                 LOG.warn("get TableMeta failed: {}", errMsg);
+                throw new MetaNotFoundException(errMsg);
             } else {
                 table.updateMeta(request.getDb_name(), response.getTable_meta(), response.getBackends());
             }
         } catch (Exception e) {
             LOG.warn("call fe {} refreshTable rpc method failed", addr, e);
+            throw new MetaNotFoundException("get TableMeta failed from " + addr + ", error: " + e.getMessage());
         }
     }
-};
+}

fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java

@@ -1559,10 +1559,11 @@ public void handleDMLStmt(ExecPlan execPlan, DmlStmt stmt) throws Exception {
         String dbName = stmt.getTableName().getDb();
         String tableName = stmt.getTableName().getTbl();
         Database database = GlobalStateMgr.getCurrentState().getMetadataMgr().getDb(catalogName, dbName);
-        Table targetTable = GlobalStateMgr.getCurrentState().getMetadataMgr().getTable(catalogName, dbName, tableName);
-        if (isExplainAnalyze) {
-            Preconditions.checkState(targetTable instanceof OlapTable,
-                    "explain analyze only supports insert into olap native table");
+        Table targetTable;
+        if (stmt instanceof InsertStmt && ((InsertStmt) stmt).getTargetTable() != null) {
+            targetTable = ((InsertStmt) stmt).getTargetTable();
+        } else {
+            targetTable = GlobalStateMgr.getCurrentState().getMetadataMgr().getTable(catalogName, dbName, tableName);
         }
 
         if (parsedStmt instanceof InsertStmt && ((InsertStmt) parsedStmt).isOverwrite() &&

fe/fe-core/src/main/java/com/starrocks/server/GlobalStateMgr.java

@@ -142,7 +142,6 @@
 import com.starrocks.connector.hive.events.MetastoreEventsProcessor;
 import com.starrocks.consistency.ConsistencyChecker;
 import com.starrocks.credential.CredentialUtil;
-import com.starrocks.external.starrocks.StarRocksRepository;
 import com.starrocks.ha.FrontendNodeType;
 import com.starrocks.ha.HAProtocol;
 import com.starrocks.ha.LeaderInfo;
@@ -388,7 +387,6 @@ public class GlobalStateMgr {
     private Daemon replayer;
     private Daemon timePrinter;
     private EsRepository esRepository;  // it is a daemon, so add it here
-    private StarRocksRepository starRocksRepository;
     private MetastoreEventsProcessor metastoreEventsProcessor;
     private ConnectorTableMetadataProcessor connectorTableMetadataProcessor;
 
@@ -693,7 +691,6 @@ private GlobalStateMgr(boolean isCkptGlobalState) {
         this.resourceGroupMgr = new ResourceGroupMgr();
 
         this.esRepository = new EsRepository();
-        this.starRocksRepository = new StarRocksRepository();
         this.metastoreEventsProcessor = new MetastoreEventsProcessor();
         this.connectorTableMetadataProcessor = new ConnectorTableMetadataProcessor();
 
@@ -1383,7 +1380,6 @@ private void startAllNodeTypeDaemonThreads() {
         labelCleaner.start();
         // ES state store
         esRepository.start();
-        starRocksRepository.start();
 
         if (Config.enable_hms_events_incremental_sync) {
             metastoreEventsProcessor.start();
@@ -1554,7 +1550,6 @@ public void loadImage(String imageDir) throws IOException, DdlException {
                 localMetastore.recreateTabletInvertIndex();
                 // rebuild es state state
                 esRepository.loadTableFromCatalog();
-                starRocksRepository.loadTableFromCatalog();
 
                 checksum = load.loadLoadJob(dis, checksum);
                 checksum = loadAlterJob(dis, checksum);
@@ -3216,10 +3211,6 @@ public EsRepository getEsRepository() {
         return this.esRepository;
     }
 
-    public StarRocksRepository getStarRocksRepository() {
-        return this.starRocksRepository;
-    }
-
     public MetastoreEventsProcessor getMetastoreEventsProcessor() {
         return this.metastoreEventsProcessor;
     }

fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java

@@ -4840,7 +4840,6 @@ public void load(SRMetaBlockReader reader) throws IOException, SRMetaBlockExcept
 
         recreateTabletInvertIndex();
         GlobalStateMgr.getCurrentState().getEsRepository().loadTableFromCatalog();
-        GlobalStateMgr.getCurrentState().getStarRocksRepository().loadTableFromCatalog();
 
         /*
          * defaultCluster has no meaning, it is only for compatibility with

fe/fe-core/src/main/java/com/starrocks/sql/analyzer/AnalyzerUtils.java

@@ -556,7 +556,8 @@ public Void visitTable(TableRelation node, Void context) {
                 OlapTable table = (OlapTable) node.getTable();
                 olapTables.add(table);
                 // Only copy the necessary olap table meta to avoid the lock when plan query
-                OlapTable copied = table.copyOnlyForQuery();
+                OlapTable copied = new OlapTable();
+                table.copyOnlyForQuery(copied);
                 node.setTable(copied);
             }
             return null;

fe/fe-core/src/main/java/com/starrocks/sql/analyzer/InsertAnalyzer.java

@@ -23,6 +23,7 @@
 import com.starrocks.analysis.NullLiteral;
 import com.starrocks.catalog.Column;
 import com.starrocks.catalog.Database;
+import com.starrocks.catalog.ExternalOlapTable;
 import com.starrocks.catalog.HiveTable;
 import com.starrocks.catalog.IcebergTable;
 import com.starrocks.catalog.MaterializedView;
@@ -33,6 +34,7 @@
 import com.starrocks.common.ErrorCode;
 import com.starrocks.common.ErrorReport;
 import com.starrocks.planner.IcebergTableSink;
+import com.starrocks.external.starrocks.TableMetaSyncer;
 import com.starrocks.qe.ConnectContext;
 import com.starrocks.server.CatalogMgr;
 import com.starrocks.sql.ast.DefaultValueExpr;
@@ -80,6 +82,10 @@ public static void analyze(InsertStmt insertStmt, ConnectContext session) {
         Database database = MetaUtils.getDatabase(catalogName, dbName);
         Table table = MetaUtils.getTable(catalogName, dbName, tableName);
 
+        if (table instanceof ExternalOlapTable) {
+            table = getOLAPExternalTableMeta(database, (ExternalOlapTable) table);
+        }
+
         if (table instanceof MaterializedView && !insertStmt.isSystem()) {
             throw new SemanticException(
                     "The data of '%s' cannot be inserted because '%s' is a materialized view," +
@@ -310,4 +316,23 @@ private static void checkStaticKeyPartitionInsert(InsertStmt insertStmt, Table t
             }
         }
     }
+
+    private static ExternalOlapTable getOLAPExternalTableMeta(Database database, ExternalOlapTable externalOlapTable) {
+        // copy the table, and release database lock when synchronize table meta
+        ExternalOlapTable copiedTable = new ExternalOlapTable();
+        externalOlapTable.copyOnlyForQuery(copiedTable);
+        int lockTimes = 0;
+        while (database.isReadLockHeldByCurrentThread()) {
+            database.readUnlock();
+            lockTimes++;
+        }
+        try {
+            new TableMetaSyncer().syncTable(copiedTable);
+        } finally {
+            while (lockTimes-- > 0) {
+                database.readLock();
+            }
+        }
+        return copiedTable;
+    }
 }

fe/fe-core/src/main/java/com/starrocks/sql/analyzer/QueryAnalyzer.java

@@ -930,7 +930,8 @@ private Table resolveTable(TableRelation tableRelation) {
                     try {
                         // Add read lock to avoid concurrent problems.
                         database.readLock();
-                        OlapTable mvOlapTable = ((OlapTable) mvTable).copyOnlyForQuery();
+                        OlapTable mvOlapTable = new OlapTable();
+                        ((OlapTable) mvTable).copyOnlyForQuery(mvOlapTable);
                         // Copy the necessary olap table meta to avoid changing original meta;
                         mvOlapTable.setBaseIndexId(materializedIndex.second.getIndexId());
                         table = mvOlapTable;