Skip to content

Commit

Permalink
throw error when increment column doesn't exist and fix additional ; (#…
Browse files Browse the repository at this point in the history 
…1510)
  • Loading branch information
@PanternBao @yanhuqing666
PanternBao authored and yanhuqing666 committed Nov 20, 2019
1 parent 8ea7e5a commit 211c3e5
Show file tree
Hide file tree
Showing 6 changed files with 40 additions and 62 deletions.
9 changes: 6 additions & 3 deletions src/main/java/com/actiontech/dble/manager/dump/DumpFileContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,11 +49,11 @@ public void setSchema(String schema) {
this.schema = schema;
}

public void skipCurrentContext() {
this.isSkip = true;
public void setSkipContext(boolean skip) {
this.isSkip = skip;
}

public boolean isSkip() {
public boolean isSkipContext() {
return this.isSkip;
}

Expand Down Expand Up @@ -93,6 +93,9 @@ public void setTable(String table) throws DumpException {
if (this.tableConfig == null && this.defaultDataNode == null) {
throw new DumpException("schema " + schema + " has no default node.");
}
if (this.tableConfig != null && this.tableConfig.getParentTC() != null) {
throw new DumpException("can't process child table, skip.");
}
}

public boolean isPushDown() {
Expand Down
7 changes: 4 additions & 3 deletions src/main/java/com/actiontech/dble/manager/dump/DumpFileExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,10 @@ public void run() {
if (ServerParse.DDL == type || ServerParse.CREATE_DATABASE == type || ServerParse.USE == (0xff & type)) {
stmt = stmt.replace("/*!", "/*#");
statement = RouteStrategyFactory.getRouteStrategy().parserSQL(stmt);
context.setSkipContext(false);
}
// if ddl is wrong，the following statement is skip.
if (context.isSkip()) {
if (context.isSkipContext()) {
continue;
}
if (ServerParse.INSERT == type && !context.isPushDown()) {
Expand All @@ -89,7 +90,7 @@ public void run() {
handler.handle(context, statement);
} catch (DumpException | SQLSyntaxErrorException e) {
String currentStmt = context.getStmt().length() <= 1024 ? context.getStmt() : context.getStmt().substring(0, 1024);
context.skipCurrentContext();
context.setSkipContext(true);
LOGGER.warn("current stmt[" + currentStmt + "] error,because:" + e.getMessage());
context.addError("current stmt[" + currentStmt + "] error,because:" + e.getMessage());
} catch (InterruptedException ie) {
Expand Down Expand Up @@ -119,7 +120,7 @@ private boolean preHandle(DumpFileWriter writer, int type, String stmt) throws I
}
// skip view
if ((ServerParse.MYSQL_CMD_COMMENT == type || ServerParse.MYSQL_COMMENT == type) && skipView(stmt)) {
context.skipCurrentContext();
context.setSkipContext(true);
return true;
}
// footer
Expand Down
30 changes: 11 additions & 19 deletions src/main/java/com/actiontech/dble/manager/dump/handler/CreateTableHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,6 @@
import com.actiontech.dble.util.StringUtil;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.expr.SQLNullExpr;
import com.alibaba.druid.sql.ast.statement.SQLCharacterDataType;
import com.alibaba.druid.sql.ast.statement.SQLColumnDefinition;
import com.alibaba.druid.sql.ast.statement.SQLTableElement;
Expand All @@ -34,34 +32,28 @@ public boolean preHandle(DumpFileContext context, SQLStatement sqlStatement) thr
checkColumns(context, columns);
// partition column check
if (tableConfig.getPartitionColumn() != null && context.getPartitionColumnIndex() == -1) {
throw new DumpException("table[" + context.getTable() + "] can't find partition column in create.");
throw new DumpException("can't find partition column in create.");
}
// increment column check
if (tableConfig.isAutoIncrement() && context.getIncrementColumnIndex() == -1) {
throw new DumpException("can't find increment column in create.");
}
}
return false;
}

@Override
public void handle(DumpFileContext context, SQLStatement sqlStatement) throws InterruptedException {
public void handle(DumpFileContext context, SQLStatement sqlStatement) throws DumpException, InterruptedException {
boolean isChanged = false;
List<SQLTableElement> columns = ((MySqlCreateTableStatement) sqlStatement).getTableElementList();
TableConfig tableConfig = context.getTableConfig();
if (tableConfig.isAutoIncrement()) {
// add increment column if not exists
if (context.getIncrementColumnIndex() == -1) {
SQLColumnDefinition column = new SQLColumnDefinition();
if (context.getIncrementColumnIndex() != -1) {
// check data type of increment column
SQLColumnDefinition column = (SQLColumnDefinition) columns.get(context.getIncrementColumnIndex());
if (!column.getDataType().getName().equals("bigint")) {
context.addError("data type of increment column isn't bigint, dble replaced it by itself.");
column.setDataType(new SQLCharacterDataType("bigint"));
column.setDefaultExpr(new SQLNullExpr());
column.setName(new SQLIdentifierExpr(tableConfig.getTrueIncrementColumn()));
columns.add(column);
context.setPartitionColumnIndex(columns.size());
isChanged = true;
} else {
SQLColumnDefinition column = (SQLColumnDefinition) columns.get(context.getIncrementColumnIndex());
if (!column.getDataType().getName().equals("bigint")) {
context.addError("data type of increment column isn't bigint, dble replaced it by itself.");
column.setDataType(new SQLCharacterDataType("bigint"));
isChanged = true;
}
}
}

Expand Down
17 changes: 0 additions & 17 deletions src/main/java/com/actiontech/dble/manager/dump/handler/GlobalTableInsertHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,27 +11,10 @@ public class GlobalTableInsertHandler extends InsertHandler {

private long time = new Date().getTime();

@Override
public void preProcess(DumpFileContext context) throws InterruptedException {
if (insertHeader == null) {
return;
}
for (String dataNode : context.getTableConfig().getDataNodes()) {
context.getWriter().write(dataNode, insertHeader.toString(), true, false);
}
}

@Override
public void process(DumpFileContext context, SQLInsertStatement.ValuesClause valueClause, boolean isFirst) throws InterruptedException, SQLNonTransientException {
valueClause.addValue(new SQLIntegerExpr(time));
super.process(context, valueClause, isFirst);
}

@Override
public void postProcess(DumpFileContext context) throws InterruptedException {
for (String dataNode : context.getTableConfig().getDataNodes()) {
context.getWriter().write(dataNode, ";", false, false);
}
super.postProcess(context);
}
}
36 changes: 17 additions & 19 deletions src/main/java/com/actiontech/dble/manager/dump/handler/InsertHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLCharExpr;
import com.alibaba.druid.sql.ast.expr.SQLIntegerExpr;
import com.alibaba.druid.sql.ast.statement.SQLInsertStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlInsertStatement;
Expand All @@ -22,7 +21,7 @@ public class InsertHandler extends DefaultHandler {
protected StringBuilder insertHeader;

@Override
public boolean preHandle(DumpFileContext context, SQLStatement sqlStatement) throws DumpException, InterruptedException {
public boolean preHandle(DumpFileContext context, SQLStatement sqlStatement) throws DumpException {
MySqlInsertStatement insert = (MySqlInsertStatement) sqlStatement;
// check columns from insert columns
checkColumns(context, insert.getColumns());
Expand Down Expand Up @@ -64,11 +63,6 @@ private void handleIncrementColumn(DumpFileContext context, List<SQLExpr> values

String tableKey = StringUtil.getFullName(context.getSchema(), context.getTable());
long val = SequenceManager.getHandler().nextId(tableKey);
if (incrementIndex == values.size()) {
values.add(new SQLIntegerExpr(val));
return;
}

SQLExpr value = values.get(incrementIndex);
if (!StringUtil.isEmpty(SQLUtils.toMySqlString(value)) && !context.isNeedSkipError()) {
context.addError("For table using global sequence, dble has set increment column values for you.");
Expand Down Expand Up @@ -96,22 +90,17 @@ private void checkColumns(DumpFileContext context, List<SQLExpr> columns) throws
partitionColumnIndex = i;
}
}
if (tableConfig.isAutoIncrement() && incrementColumnIndex == -1) {
// add increment column
columns.add(new SQLCharExpr(tableConfig.getTrueIncrementColumn()));
incrementColumnIndex = columns.size();
// if increment column is same with partition column
if (tableConfig.getTrueIncrementColumn().equalsIgnoreCase(tableConfig.getPartitionColumn())) {
partitionColumnIndex = incrementColumnIndex;
}
}

// partition column check
if (tableConfig.getPartitionColumn() != null && partitionColumnIndex == -1) {
throw new DumpException("can't find partition column in insert.");
}
// increment column check
if (tableConfig.isAutoIncrement() && incrementColumnIndex == -1) {
throw new DumpException("can't find increment column in insert.");
}
context.setIncrementColumnIndex(incrementColumnIndex);
context.setPartitionColumnIndex(partitionColumnIndex);
}
context.setIncrementColumnIndex(incrementColumnIndex);
context.setPartitionColumnIndex(partitionColumnIndex);
}

protected String toString(List<SQLExpr> values, boolean isFirst) {
Expand All @@ -131,9 +120,18 @@ protected String toString(List<SQLExpr> values, boolean isFirst) {
}

public void preProcess(DumpFileContext context) throws InterruptedException {
if (insertHeader == null) {
return;
}
for (String dataNode : context.getTableConfig().getDataNodes()) {
context.getWriter().write(dataNode, insertHeader.toString(), true, false);
}
}

public void postProcess(DumpFileContext context) throws InterruptedException {
for (String dataNode : context.getTableConfig().getDataNodes()) {
context.getWriter().write(dataNode, ";", false, false);
}
insertHeader = null;
}

Expand Down
3 changes: 2 additions & 1 deletion src/main/java/com/actiontech/dble/manager/dump/handler/ShardingTableInsertHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ public class ShardingTableInsertHandler extends InsertHandler {

private Map<String, LongPtr> dataNodes = new HashMap<>(64);

@Override
public void preProcess(DumpFileContext context) {
if (!dataNodes.isEmpty()) {
dataNodes.clear();
Expand All @@ -29,7 +30,7 @@ public void postProcess(DumpFileContext context) throws InterruptedException {
for (String dataNode : dataNodes.keySet()) {
context.getWriter().write(dataNode, ";", false, false);
}
super.postProcess(context);
insertHeader = null;
}

@Override
Expand Down

0 comments on commit 211c3e5

Please sign in to comment.