Skip to content
This repository was archived by the owner on Sep 7, 2021. It is now read-only.
This repository is currently being migrated. It's locked while the migration is in progress.

Improve sync2 #1443

Merged
merged 1 commit into from
Sep 30, 2019
Merged

Improve sync2 #1443

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
47 changes: 27 additions & 20 deletions engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -377,6 +377,32 @@ func (engine *Engine) NoAutoCondition(no ...bool) *Session {
return session.NoAutoCondition(no...)
}

func (engine *Engine) loadTableInfo(table *core.Table) error {
colSeq, cols, err := engine.dialect.GetColumns(table.Name)
if err != nil {
return err
}
for _, name := range colSeq {
table.AddColumn(cols[name])
}
indexes, err := engine.dialect.GetIndexes(table.Name)
if err != nil {
return err
}
table.Indexes = indexes

for _, index := range indexes {
for _, name := range index.Cols {
if col := table.GetColumn(name); col != nil {
col.Indexes[index.Name] = index.Type
} else {
return fmt.Errorf("Unknown col %s in index %v of table %v, columns %v", name, index.Name, table.Name, table.ColumnsSeq())
}
}
}
return nil
}

// DBMetas Retrieve all tables, columns, indexes' informations from database.
func (engine *Engine) DBMetas() ([]*core.Table, error) {
tables, err := engine.dialect.GetTables()
Expand All @@ -385,28 +411,9 @@ func (engine *Engine) DBMetas() ([]*core.Table, error) {
}

for _, table := range tables {
colSeq, cols, err := engine.dialect.GetColumns(table.Name)
if err != nil {
return nil, err
}
for _, name := range colSeq {
table.AddColumn(cols[name])
}
indexes, err := engine.dialect.GetIndexes(table.Name)
if err != nil {
if err = engine.loadTableInfo(table); err != nil {
return nil, err
}
table.Indexes = indexes

for _, index := range indexes {
for _, name := range index.Cols {
if col := table.GetColumn(name); col != nil {
col.Indexes[index.Name] = index.Type
} else {
return nil, fmt.Errorf("Unknown col %s in index %v of table %v, columns %v", name, index.Name, table.Name, table.ColumnsSeq())
}
}
}
}
return tables, nil
}
Expand Down
226 changes: 112 additions & 114 deletions session_schema.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,7 @@ func (session *Session) Sync2(beans ...interface{}) error {
defer session.Close()
}

tables, err := engine.DBMetas()
tables, err := engine.dialect.GetTables()
if err != nil {
return err
}
Expand All @@ -239,15 +239,12 @@ func (session *Session) Sync2(beans ...interface{}) error {
session.resetStatement()
}()

var structTables []*core.Table

for _, bean := range beans {
v := rValue(bean)
table, err := engine.mapType(v)
if err != nil {
return err
}
structTables = append(structTables, table)
tbName := engine.TableName(bean)
tbNameWithSchema := engine.TableName(tbName, true)

Expand All @@ -259,6 +256,7 @@ func (session *Session) Sync2(beans ...interface{}) error {
}
}

// this is a new table
if oriTable == nil {
err = session.StoreEngine(session.statement.StoreEngine).createTable(bean)
if err != nil {
Expand All @@ -274,148 +272,148 @@ func (session *Session) Sync2(beans ...interface{}) error {
if err != nil {
return err
}
} else {
for _, col := range table.Columns() {
var oriCol *core.Column
for _, col2 := range oriTable.Columns() {
if strings.EqualFold(col.Name, col2.Name) {
oriCol = col2
break
}
}
continue
}

if oriCol != nil {
expectedType := engine.dialect.SqlType(col)
curType := engine.dialect.SqlType(oriCol)
if expectedType != curType {
if expectedType == core.Text &&
strings.HasPrefix(curType, core.Varchar) {
// currently only support mysql & postgres
if engine.dialect.DBType() == core.MYSQL ||
engine.dialect.DBType() == core.POSTGRES {
engine.logger.Infof("Table %s column %s change type from %s to %s\n",
tbNameWithSchema, col.Name, curType, expectedType)
_, err = session.exec(engine.dialect.ModifyColumnSql(tbNameWithSchema, col))
} else {
engine.logger.Warnf("Table %s column %s db type is %s, struct type is %s\n",
tbNameWithSchema, col.Name, curType, expectedType)
}
} else if strings.HasPrefix(curType, core.Varchar) && strings.HasPrefix(expectedType, core.Varchar) {
if engine.dialect.DBType() == core.MYSQL {
if oriCol.Length < col.Length {
engine.logger.Infof("Table %s column %s change type from varchar(%d) to varchar(%d)\n",
tbNameWithSchema, col.Name, oriCol.Length, col.Length)
_, err = session.exec(engine.dialect.ModifyColumnSql(tbNameWithSchema, col))
}
}
} else {
if !(strings.HasPrefix(curType, expectedType) && curType[len(expectedType)] == '(') {
engine.logger.Warnf("Table %s column %s db type is %s, struct type is %s",
tbNameWithSchema, col.Name, curType, expectedType)
}
}
} else if expectedType == core.Varchar {
if engine.dialect.DBType() == core.MYSQL {
if oriCol.Length < col.Length {
engine.logger.Infof("Table %s column %s change type from varchar(%d) to varchar(%d)\n",
tbNameWithSchema, col.Name, oriCol.Length, col.Length)
_, err = session.exec(engine.dialect.ModifyColumnSql(tbNameWithSchema, col))
}
}
}
if col.Default != oriCol.Default {
engine.logger.Warnf("Table %s Column %s db default is %s, struct default is %s",
tbName, col.Name, oriCol.Default, col.Default)
}
if col.Nullable != oriCol.Nullable {
engine.logger.Warnf("Table %s Column %s db nullable is %v, struct nullable is %v",
tbName, col.Name, oriCol.Nullable, col.Nullable)
}
} else {
session.statement.RefTable = table
session.statement.tableName = tbNameWithSchema
err = session.addColumn(col.Name)
// this will modify an old table
if err = engine.loadTableInfo(oriTable); err != nil {
return err
}

// check columns
for _, col := range table.Columns() {
var oriCol *core.Column
for _, col2 := range oriTable.Columns() {
if strings.EqualFold(col.Name, col2.Name) {
oriCol = col2
break
}
if err != nil {
}

// column is not exist on table
if oriCol == nil {
session.statement.RefTable = table
session.statement.tableName = tbNameWithSchema
if err = session.addColumn(col.Name); err != nil {
return err
}
continue
}

var foundIndexNames = make(map[string]bool)
var addedNames = make(map[string]*core.Index)

for name, index := range table.Indexes {
var oriIndex *core.Index
for name2, index2 := range oriTable.Indexes {
if index.Equal(index2) {
oriIndex = index2
foundIndexNames[name2] = true
break
err = nil
expectedType := engine.dialect.SqlType(col)
curType := engine.dialect.SqlType(oriCol)
if expectedType != curType {
if expectedType == core.Text &&
strings.HasPrefix(curType, core.Varchar) {
// currently only support mysql & postgres
if engine.dialect.DBType() == core.MYSQL ||
engine.dialect.DBType() == core.POSTGRES {
engine.logger.Infof("Table %s column %s change type from %s to %s\n",
tbNameWithSchema, col.Name, curType, expectedType)
_, err = session.exec(engine.dialect.ModifyColumnSql(tbNameWithSchema, col))
} else {
engine.logger.Warnf("Table %s column %s db type is %s, struct type is %s\n",
tbNameWithSchema, col.Name, curType, expectedType)
}
}

if oriIndex != nil {
if oriIndex.Type != index.Type {
sql := engine.dialect.DropIndexSql(tbNameWithSchema, oriIndex)
_, err = session.exec(sql)
if err != nil {
return err
} else if strings.HasPrefix(curType, core.Varchar) && strings.HasPrefix(expectedType, core.Varchar) {
if engine.dialect.DBType() == core.MYSQL {
if oriCol.Length < col.Length {
engine.logger.Infof("Table %s column %s change type from varchar(%d) to varchar(%d)\n",
tbNameWithSchema, col.Name, oriCol.Length, col.Length)
_, err = session.exec(engine.dialect.ModifyColumnSql(tbNameWithSchema, col))
}
oriIndex = nil
}
} else {
if !(strings.HasPrefix(curType, expectedType) && curType[len(expectedType)] == '(') {
engine.logger.Warnf("Table %s column %s db type is %s, struct type is %s",
tbNameWithSchema, col.Name, curType, expectedType)
}
}

if oriIndex == nil {
addedNames[name] = index
} else if expectedType == core.Varchar {
if engine.dialect.DBType() == core.MYSQL {
if oriCol.Length < col.Length {
engine.logger.Infof("Table %s column %s change type from varchar(%d) to varchar(%d)\n",
tbNameWithSchema, col.Name, oriCol.Length, col.Length)
_, err = session.exec(engine.dialect.ModifyColumnSql(tbNameWithSchema, col))
}
}
}
if col.Default != oriCol.Default {
engine.logger.Warnf("Table %s Column %s db default is %s, struct default is %s",
tbName, col.Name, oriCol.Default, col.Default)
}
if col.Nullable != oriCol.Nullable {
engine.logger.Warnf("Table %s Column %s db nullable is %v, struct nullable is %v",
tbName, col.Name, oriCol.Nullable, col.Nullable)
}

if err != nil {
return err
}
}

var foundIndexNames = make(map[string]bool)
var addedNames = make(map[string]*core.Index)

for name, index := range table.Indexes {
var oriIndex *core.Index
for name2, index2 := range oriTable.Indexes {
if _, ok := foundIndexNames[name2]; !ok {
sql := engine.dialect.DropIndexSql(tbNameWithSchema, index2)
if index.Equal(index2) {
oriIndex = index2
foundIndexNames[name2] = true
break
}
}

if oriIndex != nil {
if oriIndex.Type != index.Type {
sql := engine.dialect.DropIndexSql(tbNameWithSchema, oriIndex)
_, err = session.exec(sql)
if err != nil {
return err
}
oriIndex = nil
}
}

for name, index := range addedNames {
if index.Type == core.UniqueType {
session.statement.RefTable = table
session.statement.tableName = tbNameWithSchema
err = session.addUnique(tbNameWithSchema, name)
} else if index.Type == core.IndexType {
session.statement.RefTable = table
session.statement.tableName = tbNameWithSchema
err = session.addIndex(tbNameWithSchema, name)
}
if oriIndex == nil {
addedNames[name] = index
}
}

for name2, index2 := range oriTable.Indexes {
if _, ok := foundIndexNames[name2]; !ok {
sql := engine.dialect.DropIndexSql(tbNameWithSchema, index2)
_, err = session.exec(sql)
if err != nil {
return err
}
}
}
}

for _, table := range tables {
var oriTable *core.Table
for _, structTable := range structTables {
if strings.EqualFold(table.Name, session.tbNameNoSchema(structTable)) {
oriTable = structTable
break
for name, index := range addedNames {
if index.Type == core.UniqueType {
session.statement.RefTable = table
session.statement.tableName = tbNameWithSchema
err = session.addUnique(tbNameWithSchema, name)
} else if index.Type == core.IndexType {
session.statement.RefTable = table
session.statement.tableName = tbNameWithSchema
err = session.addIndex(tbNameWithSchema, name)
}
if err != nil {
return err
}
}

if oriTable == nil {
//engine.LogWarnf("Table %s has no struct to mapping it", table.Name)
continue
}

for _, colName := range table.ColumnsSeq() {
if oriTable.GetColumn(colName) == nil {
engine.logger.Warnf("Table %s has column %s but struct has not related field", engine.TableName(table.Name, true), colName)
// check all the columns which removed from struct fields but left on database tables.
for _, colName := range oriTable.ColumnsSeq() {
if table.GetColumn(colName) == nil {
engine.logger.Warnf("Table %s has column %s but struct has not related field", engine.TableName(oriTable.Name, true), colName)
}
}
}

return nil
}