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

[Enhancement] auto change replication_num of system tables #51799

Merged
merged 5 commits into from
Oct 14, 2024
Merged

[Enhancement] auto change replication_num of system tables #51799

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c64306e
fix load_history & task_run_history
murphyatwork Oct 11, 2024
cea63b9
fix pipe_file_list
murphyatwork Oct 11, 2024
09f1e5a
fix statistics
murphyatwork Oct 11, 2024
56af81f
refactor
murphyatwork Oct 11, 2024
a31931c
add test
murphyatwork Oct 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
fix pipe_file_list 
Signed-off-by: Murphy <mofei@starrocks.com>
  • Loading branch information
@murphyatwork
murphyatwork committed Oct 11, 2024
commit cea63b94736499e2d7f786676b1e87d03dd9552b
12 changes: 5 additions & 7 deletions fe/fe-core/src/main/java/com/starrocks/load/pipe/filelist/FileListTableRepo.java
View file
Open in desktop
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
30 changes: 10 additions & 20 deletions fe/fe-core/src/main/java/com/starrocks/load/pipe/filelist/RepoCreator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,26 +60,24 @@ 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 expectedReplicationNum =
GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getSystemTableExpectedReplicationNum();
int replica = GlobalStateMgr.getCurrentState()
.getLocalMetastore().mayGetTable(FileListTableRepo.FILE_LIST_DB_NAME, 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();
if (replica != expectedReplicationNum) {
String sql = FileListTableRepo.SQLBuilder.buildAlterTableSql(expectedReplicationNum);
RepoExecutor.getInstance().executeDDL(sql);
} else {
LOG.info("table {} already has {} replicas, no need to alter replication_num",
FileListTableRepo.FILE_LIST_FULL_NAME, replica);
LOG.info("changed table {} replication_num from {} to {}",
FileListTableRepo.FILE_LIST_FULL_NAME, replica, expectedReplicationNum);
}
return true;
}
Expand All @@ -95,8 +89,4 @@ public boolean isDatabaseExists() {
public boolean isTableExists() {
return tableExists;
}

public boolean isTableCorrected() {
return tableCorrected;
}
}
12 changes: 7 additions & 5 deletions fe/fe-core/src/test/java/com/starrocks/load/pipe/filelist/FileListRepoTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

public class FileListRepoTest {

Expand Down Expand Up @@ -228,36 +229,37 @@ public void executeDDL(String sql) {
creator.run();
Assert.assertTrue(creator.isDatabaseExists());
Assert.assertFalse(creator.isTableExists());
Assert.assertFalse(creator.isTableCorrected());

// create with 1 replica
new MockUp<SystemInfoService>() {
@Mock
public int getTotalBackendNumber() {
public int getSystemTableExpectedReplicationNum() {
return 1;
}
};
AtomicInteger changed = new AtomicInteger(0);
new MockUp<RepoExecutor>() {
@Mock
public void executeDDL(String sql) {
changed.addAndGet(1);
}
};
creator.run();
Assert.assertTrue(creator.isTableExists());
Assert.assertFalse(creator.isTableCorrected());
Assert.assertEquals(1, changed.get());

// be corrected to 3 replicas
new MockUp<SystemInfoService>() {
@Mock
public int getTotalBackendNumber() {
public int getSystemTableExpectedReplicationNum() {
return 3;
}
};

creator.run();
Assert.assertTrue(creator.isDatabaseExists());
Assert.assertTrue(creator.isTableExists());
Assert.assertTrue(creator.isTableCorrected());
Assert.assertEquals(2, changed.get());
}

@Test
Expand Down