Skip to content

Commit

Permalink
[Enhancement] auto change replication_num of system tables (backport #…
Browse files Browse the repository at this point in the history 
…51799) (#52766)

Signed-off-by: Murphy <mofei@starrocks.com>
  • Loading branch information
@murphyatwork
murphyatwork authored Nov 11, 2024
1 parent abffdca commit d48afa3
Show file tree
Hide file tree
Showing 12 changed files with 186 additions and 155 deletions.
2 changes: 1 addition & 1 deletion fe/fe-core/src/main/java/com/starrocks/load/loadv2/LoadsHistorySyncer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ public class LoadsHistorySyncer extends FrontendDaemon {
private boolean firstSync = true;

private static final TableKeeper KEEPER =
new TableKeeper(LOADS_HISTORY_DB_NAME, LOADS_HISTORY_TABLE_NAME, LOADS_HISTORY_TABLE_CREATE, 3,
new TableKeeper(LOADS_HISTORY_DB_NAME, LOADS_HISTORY_TABLE_NAME, LOADS_HISTORY_TABLE_CREATE,
() -> Math.max(1, Config.loads_history_retained_days));

public static TableKeeper createKeeper() {
Expand Down
12 changes: 5 additions & 7 deletions fe/fe-core/src/main/java/com/starrocks/load/pipe/filelist/FileListTableRepo.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@

import com.starrocks.catalog.CatalogUtils;
import com.starrocks.common.UserException;
import com.starrocks.common.util.AutoInferUtil;
import com.starrocks.load.pipe.PipeFileRecord;
import com.starrocks.statistic.StatsConstants;
import org.apache.commons.collections.CollectionUtils;
Expand Down Expand Up @@ -64,7 +63,7 @@ public class FileListTableRepo extends FileListRepo {
"properties('replication_num' = '%d') ";

protected static final String CORRECT_FILE_LIST_REPLICATION_NUM =
"ALTER TABLE %s SET ('replication_num'='3')";
"ALTER TABLE %s SET ('replication_num'='%d')";

protected static final String ALL_COLUMNS =
"`pipe_id`, `file_name`, `file_version`, `file_size`, `state`, `last_modified`, `staged_time`," +
Expand Down Expand Up @@ -155,15 +154,14 @@ public void destroy() {
*/
static class SQLBuilder {

public static String buildCreateTableSql() throws UserException {
int replica = AutoInferUtil.calDefaultReplicationNum();
public static String buildCreateTableSql(int replicationNum) throws UserException {
return String.format(FILE_LIST_TABLE_CREATE,
CatalogUtils.normalizeTableName(FILE_LIST_DB_NAME, FILE_LIST_TABLE_NAME), replica);
CatalogUtils.normalizeTableName(FILE_LIST_DB_NAME, FILE_LIST_TABLE_NAME), replicationNum);
}

public static String buildAlterTableSql() {
public static String buildAlterTableSql(int replicationNum) {
return String.format(CORRECT_FILE_LIST_REPLICATION_NUM,
CatalogUtils.normalizeTableName(FILE_LIST_DB_NAME, FILE_LIST_TABLE_NAME));
CatalogUtils.normalizeTableName(FILE_LIST_DB_NAME, FILE_LIST_TABLE_NAME), replicationNum);
}
}

Expand Down
35 changes: 6 additions & 29 deletions fe/fe-core/src/main/java/com/starrocks/load/pipe/filelist/RepoCreator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,9 @@

package com.starrocks.load.pipe.filelist;

import com.starrocks.catalog.OlapTable;
import com.starrocks.common.UserException;
import com.starrocks.server.GlobalStateMgr;
import com.starrocks.statistic.StatisticUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

Expand All @@ -30,7 +30,6 @@ public class RepoCreator {

private static boolean databaseExists = false;
private static boolean tableExists = false;
private static boolean tableCorrected = false;

public static RepoCreator getInstance() {
return INSTANCE;
Expand All @@ -50,10 +49,7 @@ public void run() {
LOG.info("table created: " + FileListTableRepo.FILE_LIST_TABLE_NAME);
tableExists = true;
}
if (!tableCorrected && correctTable()) {
LOG.info("table corrected: " + FileListTableRepo.FILE_LIST_TABLE_NAME);
tableCorrected = true;
}
correctTable();
} catch (Exception e) {
LOG.error("error happens in RepoCreator: ", e);
}
Expand All @@ -64,29 +60,14 @@ public boolean checkDatabaseExists() {
}

public static void createTable() throws UserException {
String sql = FileListTableRepo.SQLBuilder.buildCreateTableSql();
int expectedReplicationNum =
GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getSystemTableExpectedReplicationNum();
String sql = FileListTableRepo.SQLBuilder.buildCreateTableSql(expectedReplicationNum);
RepoExecutor.getInstance().executeDDL(sql);
}

public static boolean correctTable() {
int numBackends = GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getTotalBackendNumber();
int replica = GlobalStateMgr.getCurrentState()
.mayGetDb(FileListTableRepo.FILE_LIST_DB_NAME)
.flatMap(db -> db.mayGetTable(FileListTableRepo.FILE_LIST_TABLE_NAME))
.map(tbl -> ((OlapTable) tbl).getPartitionInfo().getMinReplicationNum())
.orElse((short) 1);
if (numBackends < 3) {
LOG.info("not enough backends in the cluster, expected 3 but got {}", numBackends);
return false;
}
if (replica < 3) {
String sql = FileListTableRepo.SQLBuilder.buildAlterTableSql();
RepoExecutor.getInstance().executeDDL(sql);
} else {
LOG.info("table {} already has {} replicas, no need to alter replication_num",
FileListTableRepo.FILE_LIST_FULL_NAME, replica);
}
return true;
return StatisticUtils.alterSystemTableReplicationNumIfNecessary(FileListTableRepo.FILE_LIST_TABLE_NAME);
}

public boolean isDatabaseExists() {
Expand All @@ -96,8 +77,4 @@ public boolean isDatabaseExists() {
public boolean isTableExists() {
return tableExists;
}

public boolean isTableCorrected() {
return tableCorrected;
}
}
39 changes: 11 additions & 28 deletions fe/fe-core/src/main/java/com/starrocks/scheduler/history/TableKeeper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,6 @@ public class TableKeeper {
private final String databaseName;
private final String tableName;
private final String createTableSql;
private final int tableReplicas;

private boolean databaseExisted = false;
private boolean tableExisted = false;
Expand All @@ -53,12 +52,10 @@ public class TableKeeper {
public TableKeeper(String database,
String table,
String createTable,
int expectedReplicas,
Supplier<Integer> ttlSupplier) {
this.databaseName = database;
this.tableName = table;
this.createTableSql = createTable;
this.tableReplicas = expectedReplicas;
this.ttlSupplier = ttlSupplier;
}

Expand Down Expand Up @@ -101,23 +98,21 @@ public void createTable() throws UserException {
}

public void correctTable() {
int numBackends = GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getTotalBackendNumber();
int expectedReplicationNum =
GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getSystemTableExpectedReplicationNum();
int replica = GlobalStateMgr.getCurrentState()
.mayGetDb(databaseName)
.flatMap(db -> db.mayGetTable(tableName))
.map(tbl -> ((OlapTable) tbl).getPartitionInfo().getMinReplicationNum())
.orElse((short) 1);
if (numBackends < tableReplicas) {
LOG.info("not enough backends in the cluster, expected {} but got {}",
tableReplicas, numBackends);
return;
}
if (replica < tableReplicas) {
String sql = alterTableReplicas();

if (replica != expectedReplicationNum) {
String sql = alterTableReplicas(expectedReplicationNum);
if (StringUtils.isNotEmpty(sql)) {
RepoExecutor.getInstance().executeDDL(sql);
}
LOG.info("changed replication_number of table {} to {}", tableName, replica);
LOG.info("changed replication_number of table {} from {} to {}",
tableName, replica, expectedReplicationNum);
}
}

Expand Down Expand Up @@ -149,21 +144,21 @@ private Optional<OlapTable> mayGetTable() {
.flatMap(x -> Optional.of((OlapTable) x));
}

private String alterTableReplicas() {
private String alterTableReplicas(int replicationNum) {
Optional<OlapTable> table = mayGetTable();
if (table.isEmpty()) {
return "";
}
PartitionInfo partitionInfo = table.get().getPartitionInfo();
if (partitionInfo.isRangePartition()) {
String sql1 = String.format("ALTER TABLE %s.%s MODIFY PARTITION(*) SET ('replication_num'='%d');",
databaseName, tableName, tableReplicas);
databaseName, tableName, replicationNum);
String sql2 = String.format("ALTER TABLE %s.%s SET ('default.replication_num'='%d');",
databaseName, tableName, tableReplicas);
databaseName, tableName, replicationNum);
return sql1 + sql2;
} else {
return String.format("ALTER TABLE %s.%s SET ('replication_num'='%d')",
databaseName, tableName, tableReplicas);
databaseName, tableName, replicationNum);
}
}

Expand All @@ -183,10 +178,6 @@ public String getCreateTableSql() {
return createTableSql;
}

public int getTableReplicas() {
return tableReplicas;
}

public boolean isDatabaseExisted() {
return databaseExisted;
}
Expand All @@ -203,14 +194,6 @@ public void setDatabaseExisted(boolean databaseExisted) {
this.databaseExisted = databaseExisted;
}

public void setTableExisted(boolean tableExisted) {
this.tableExisted = tableExisted;
}

public void setTableCorrected(boolean tableCorrected) {
this.tableCorrected = tableCorrected;
}

public static TableKeeperDaemon startDaemon() {
TableKeeperDaemon daemon = TableKeeperDaemon.getInstance();
daemon.start();
Expand Down
3 changes: 1 addition & 2 deletions fe/fe-core/src/main/java/com/starrocks/scheduler/history/TaskRunHistoryTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@ public class TaskRunHistoryTable {
public static final String DATABASE_NAME = StatsConstants.STATISTICS_DB_NAME;
public static final String TABLE_NAME = "task_run_history";
public static final String TABLE_FULL_NAME = DATABASE_NAME + "." + TABLE_NAME;
public static final int TABLE_REPLICAS = 3;
public static final String CREATE_TABLE =
String.format("CREATE TABLE IF NOT EXISTS %s (" +
// identifiers
Expand Down Expand Up @@ -93,7 +92,7 @@ public class TaskRunHistoryTable {
"SELECT history_content_json " + "FROM " + TABLE_FULL_NAME + " WHERE ";

private static final TableKeeper KEEPER =
new TableKeeper(DATABASE_NAME, TABLE_NAME, CREATE_TABLE, TABLE_REPLICAS,
new TableKeeper(DATABASE_NAME, TABLE_NAME, CREATE_TABLE,
() -> Math.max(1, Config.task_runs_ttl_second / 3600 / 24));

public static TableKeeper createKeeper() {
Expand Down
8 changes: 8 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2719,6 +2719,14 @@ public void replayBatchDeleteReplica(BatchDeleteReplicaInfo info) {
}
}

public Optional<Table> mayGetTable(long dbId, long tableId) {
return Optional.ofNullable(getTable(dbId, tableId));
}

public Optional<Table> mayGetTable(String dbName, String tableName) {
return Optional.ofNullable(getTable(dbName, tableName));
}

public Table getTable(long dbId, long tableId) {
Database db = idToDb.get(dbId);
if (db == null) {
Expand Down
28 changes: 28 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/statistic/StatisticUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import com.starrocks.connector.ConnectorPartitionTraits;
import com.starrocks.load.EtlStatus;
import com.starrocks.load.loadv2.LoadJobFinalOperation;
import com.starrocks.load.pipe.filelist.RepoExecutor;
import com.starrocks.load.streamload.StreamLoadTxnCommitAttachment;
import com.starrocks.privilege.PrivilegeBuiltinConstants;
import com.starrocks.qe.ConnectContext;
Expand All @@ -61,6 +62,7 @@
import com.starrocks.transaction.TransactionState;
import com.starrocks.transaction.TxnCommitAttachment;
import com.starrocks.warehouse.Warehouse;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

Expand Down Expand Up @@ -475,6 +477,32 @@ public static void dropStatisticsAfterDropTable(Table table) {
GlobalStateMgr.getCurrentState().getStatisticStorage().expireConnectorTableColumnStatistics(table, columns);
}

/**
* Change the replication_num of system table according to cluster status
* 1. When scale-out to greater than 3 nodes, change the replication_num to 3
* 3. When scale-in to less than 3 node, change it to retainedBackendNum
*/
public static boolean alterSystemTableReplicationNumIfNecessary(String tableName) {
int expectedReplicationNum =
GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getSystemTableExpectedReplicationNum();
int replica = GlobalStateMgr.getCurrentState()
.getLocalMetastore().mayGetTable(StatsConstants.STATISTICS_DB_NAME, tableName)
.map(tbl -> ((OlapTable) tbl).getPartitionInfo().getMinReplicationNum())
.orElse((short) 1);

if (replica != expectedReplicationNum) {
String sql = String.format("ALTER TABLE %s.%s SET ('replication_num'='%d')",
StatsConstants.STATISTICS_DB_NAME, tableName, expectedReplicationNum);
if (StringUtils.isNotEmpty(sql)) {
RepoExecutor.getInstance().executeDDL(sql);
}
LOG.info("changed replication_number of table {} from {} to {}",
tableName, replica, expectedReplicationNum);
return true;
}
return false;
}

// only support collect statistics for slotRef and subfield expr
public static String getColumnName(Table table, Expr column) {
String colName;
Expand Down
Loading

0 comments on commit d48afa3

Please sign in to comment.