Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Refactor] Synchronize OLAP external table metadata when loading data #24739

Merged
merged 3 commits into from
Jun 12, 2023
Merged

[Refactor] Synchronize OLAP external table metadata when loading data #24739

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
90b2751
fix
gengjun-git May 30, 2023
3ba32b2
fix compile error
gengjun-git Jun 6, 2023
0a662dc
fix
gengjun-git Jun 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/catalog/Database.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
408 changes: 198 additions & 210 deletions fe/fe-core/src/main/java/com/starrocks/catalog/ExternalOlapTable.java
View file
Open in desktop

Large diffs are not rendered by default.

4 changes: 1 addition & 3 deletions fe/fe-core/src/main/java/com/starrocks/catalog/OlapTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -284,8 +284,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);
Expand Down Expand Up @@ -315,7 +314,6 @@ public OlapTable copyOnlyForQuery() {
if (this.tableProperty != null) {
olapTable.tableProperty = this.tableProperty.copy();
}
return olapTable;
}

public BinlogConfig getCurBinlogConfig() {
Expand Down
113 changes: 0 additions & 113 deletions fe/fe-core/src/main/java/com/starrocks/external/starrocks/StarRocksRepository.java
View file
Open in desktop

This file was deleted.

7 changes: 5 additions & 2 deletions fe/fe-core/src/main/java/com/starrocks/external/starrocks/TableMetaSyncer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
Expand All @@ -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());
}
}
};
}
7 changes: 6 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/qe/StmtExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1419,7 +1419,12 @@ 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);
gengjun-git marked this conversation as resolved.
Show resolved Hide resolved
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() &&
!((InsertStmt) parsedStmt).hasOverwriteJob() &&
Expand Down
9 changes: 0 additions & 9 deletions fe/fe-core/src/main/java/com/starrocks/server/GlobalStateMgr.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,6 @@
import com.starrocks.connector.hive.events.MetastoreEventsProcessor;
import com.starrocks.consistency.ConsistencyChecker;
import com.starrocks.credential.CloudCredentialUtil;
import com.starrocks.external.starrocks.StarRocksRepository;
import com.starrocks.ha.FrontendNodeType;
import com.starrocks.ha.HAProtocol;
import com.starrocks.ha.LeaderInfo;
Expand Down Expand Up @@ -380,7 +379,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;

Expand Down Expand Up @@ -683,7 +681,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();

Expand Down Expand Up @@ -1360,7 +1357,6 @@ private void startAllNodeTypeDaemonThreads() {
labelCleaner.start();
// ES state store
esRepository.start();
starRocksRepository.start();

if (Config.enable_hms_events_incremental_sync) {
metastoreEventsProcessor.start();
Expand Down Expand Up @@ -1515,7 +1511,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);
Expand Down Expand Up @@ -3143,10 +3138,6 @@ public EsRepository getEsRepository() {
return this.esRepository;
}

public StarRocksRepository getStarRocksRepository() {
return this.starRocksRepository;
}

public MetastoreEventsProcessor getMetastoreEventsProcessor() {
return this.metastoreEventsProcessor;
}
Expand Down
1 change: 0 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4731,7 +4731,6 @@ public void load(SRMetaBlockReader reader) throws IOException, SRMetaBlockExcept

recreateTabletInvertIndex();
GlobalStateMgr.getCurrentState().getEsRepository().loadTableFromCatalog();
GlobalStateMgr.getCurrentState().getStarRocksRepository().loadTableFromCatalog();

defaultCluster = new Cluster(SystemInfoService.DEFAULT_CLUSTER, 0);
}
Expand Down
3 changes: 2 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/sql/analyzer/AnalyzerUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -530,7 +530,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;
Expand Down
25 changes: 25 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/sql/analyzer/InsertAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import com.starrocks.analysis.LiteralExpr;
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;
Expand All @@ -31,6 +32,7 @@
import com.starrocks.common.AnalysisException;
import com.starrocks.common.ErrorCode;
import com.starrocks.common.ErrorReport;
import com.starrocks.external.starrocks.TableMetaSyncer;
import com.starrocks.qe.ConnectContext;
import com.starrocks.server.CatalogMgr;
import com.starrocks.sql.ast.DefaultValueExpr;
Expand Down Expand Up @@ -78,6 +80,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," +
Expand Down Expand Up @@ -282,4 +288,23 @@ public static void analyze(InsertStmt insertStmt, ConnectContext session) {
insertStmt.setTargetColumns(targetColumns);
session.getDumpInfo().addTable(database.getFullName(), table);
}

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);
nshangyiming marked this conversation as resolved.
Show resolved Hide resolved
int lockTimes = 0;
while (database.isReadLockHeldByCurrentThread()) {
database.readUnlock();
nshangyiming marked this conversation as resolved.
Show resolved Hide resolved
lockTimes++;
}
try {
new TableMetaSyncer().syncTable(copiedTable);
} finally {
while (lockTimes-- > 0) {
database.readLock();
}
}
return copiedTable;
}
}
3 changes: 2 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/sql/analyzer/QueryAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -920,7 +920,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.getId());
table = mvOlapTable;
Expand Down