adjust column/index offset when create table with like

ddl/ddl_api.go

@@ -1008,7 +1008,8 @@ func (d *ddl) CreateTableWithLike(ctx sessionctx.Context, ident, referIdent ast.
 		return infoschema.ErrTableExists.GenWithStackByArgs(ident)
 	}
 
-	tblInfo, err := buildTableInfoWithLike(d, ident, referTbl.Meta())
+	tblInfo := buildTableInfoWithLike(ident, referTbl.Meta())
+	tblInfo.ID, err = d.genGlobalID()
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -1025,25 +1026,49 @@ func (d *ddl) CreateTableWithLike(ctx sessionctx.Context, ident, referIdent ast.
 	return errors.Trace(err)
 }
 
-func buildTableInfoWithLike(d *ddl, ident ast.Ident, referTblInfo *model.TableInfo) (model.TableInfo, error) {
-	var err error
+func buildTableInfoWithLike(ident ast.Ident, referTblInfo *model.TableInfo) model.TableInfo {
 	tblInfo := *referTblInfo
-	// No public column must in the last offset.
-	if tblInfo.Columns[len(tblInfo.Columns)-1].State != model.StatePublic {
-		tblInfo.Columns = tblInfo.Columns[0 : len(tblInfo.Columns)-1]
+	// Check non-public column and adjust column offset.
+	newColumns := make([]*model.ColumnInfo, 0, len(tblInfo.Columns))
+	offsetChanged := make(map[int]int)
+	for _, col := range tblInfo.Columns {
+		if col.State == model.StatePublic {
+			newCol := col.Clone()
+			if newCol.Offset != len(newColumns) {
+				offsetChanged[newCol.Offset] = len(newColumns)
+				newCol.Offset = len(newColumns)
+			}
+			newColumns = append(newColumns, newCol)
+		}
 	}
+	// Update index column offset if have column offset changed.
 	newIndices := make([]*model.IndexInfo, 0, len(tblInfo.Indices))
-	for _, idx := range tblInfo.Indices {
-		if idx.State == model.StatePublic {
-			newIndices = append(newIndices, idx)
+	if len(offsetChanged) > 0 {
+		for _, idx := range tblInfo.Indices {
+			if idx.State == model.StatePublic {
+				newIdx := idx.Clone()
+				for _, col := range newIdx.Columns {
+					if newOffset, ok := offsetChanged[col.Offset]; ok {
+						col.Offset = newOffset
+					}
+				}
+				newIndices = append(newIndices, newIdx)
+			}
+		}
+	} else {
+		// No need to clone.
+		for _, idx := range tblInfo.Indices {
+			if idx.State == model.StatePublic {
+				newIndices = append(newIndices, idx)
+			}
 		}
 	}
+	tblInfo.Columns = newColumns
 	tblInfo.Indices = newIndices
 	tblInfo.Name = ident.Name
 	tblInfo.AutoIncID = 0
 	tblInfo.ForeignKeys = nil
-	tblInfo.ID, err = d.genGlobalID()
-	return tblInfo, errors.Trace(err)
+	return tblInfo
 }
 
 // BuildTableInfoFromAST builds model.TableInfo from a SQL statement.

ddl/table_test.go

@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/pingcap/parser/ast"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
@@ -323,3 +324,38 @@ func (s *testTableSuite) TestTableResume(c *C) {
 	testRunInterruptedJob(c, d, job)
 	testCheckTableState(c, d, s.dbInfo, tblInfo, model.StateNone)
 }
+
+func (s *testTableSuite) TestCreateTableWithLikeBuildTableInfo(c *C) {
+	t1TblInfo := testTableInfo(c, s.d, "t1", 3)
+	t1TblInfo.Columns[1].State = model.StateDeleteOnly
+	idx1Columns := []*model.IndexColumn{{
+		Name:   t1TblInfo.Columns[1].Name,
+		Offset: t1TblInfo.Columns[1].Offset,
+		Length: types.UnspecifiedLength,
+	}}
+	idx1Info := &model.IndexInfo{
+		Name:    model.NewCIStr("idx1"),
+		Columns: idx1Columns,
+		State:   model.StateDeleteOnly,
+	}
+	idx2Columns := []*model.IndexColumn{{
+		Name:   t1TblInfo.Columns[2].Name,
+		Offset: t1TblInfo.Columns[2].Offset,
+		Length: types.UnspecifiedLength,
+	}}
+	idx2Info := &model.IndexInfo{
+		Name:    model.NewCIStr("idx2"),
+		Columns: idx2Columns,
+		State:   model.StatePublic,
+	}
+	t1TblInfo.Indices = []*model.IndexInfo{idx1Info, idx2Info}
+
+	t2TblInfo := buildTableInfoWithLike(ast.Ident{Name: model.NewCIStr("t2")}, t1TblInfo)
+	c.Assert(len(t2TblInfo.Columns), Equals, 2)
+	for i, col := range t2TblInfo.Columns {
+		c.Assert(col.Offset, Equals, i)
+	}
+	c.Assert(len(t2TblInfo.Indices), Equals, 1)
+	c.Assert(t2TblInfo.Indices[0].Name.L, Equals, "idx2")
+	c.Assert(t2TblInfo.Indices[0].Columns[0].Offset, Equals, 1)
+}