[incubator-devlake] branch release-v0.18 updated: fix: migration bugs on postgres (#5695)

[ab...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

abeizn pushed a commit to branch release-v0.18
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git


The following commit(s) were added to refs/heads/release-v0.18 by this push:
     new b751e1f55 fix: migration bugs on postgres (#5695)
b751e1f55 is described below

commit b751e1f551473d05780dcdea007a82e326d7e2db
Author: abeizn <zi...@merico.dev>
AuthorDate: Tue Jul 18 23:26:01 2023 +0800

    fix: migration bugs on postgres (#5695)
    
    * fix: migration bugs on postgres
---
 backend/helpers/migrationhelper/migrationhelper.go |  8 +++++++-
 backend/impls/dalgorm/dalgorm.go                   | 19 +++++++++++++++++++
 backend/impls/dalgorm/dalgorm_transaction.go       | 17 +++++++++--------
 3 files changed, 35 insertions(+), 9 deletions(-)

diff --git a/backend/helpers/migrationhelper/migrationhelper.go b/backend/helpers/migrationhelper/migrationhelper.go
index 4de96c326..86cb341bf 100644
--- a/backend/helpers/migrationhelper/migrationhelper.go
+++ b/backend/helpers/migrationhelper/migrationhelper.go
@@ -263,13 +263,19 @@ func CopyTableColumns[S any, D any](
 	if err != nil {
 		return errors.Default.Wrap(err, fmt.Sprintf("failed to load data from src table [%s]", srcTableName))
 	}
-	defer cursor.Close()
+	if !reflect.ValueOf(cursor).IsNil() {
+		defer cursor.Close()
+	}
+
 	batch, err := helper.NewBatchSave(basicRes, reflect.TypeOf(new(D)), 200, dstTableName)
 	if err != nil {
 		return errors.Default.Wrap(err, fmt.Sprintf("failed to instantiate BatchSave for table [%s]", srcTableName))
 	}
 	defer batch.Close()
 
+	if reflect.ValueOf(cursor).IsNil() {
+		return nil
+	}
 	for cursor.Next() {
 		srcTable := new(S)
 		err1 := db.Fetch(cursor, srcTable)
diff --git a/backend/impls/dalgorm/dalgorm.go b/backend/impls/dalgorm/dalgorm.go
index 5acde4faa..1b5f5aa02 100644
--- a/backend/impls/dalgorm/dalgorm.go
+++ b/backend/impls/dalgorm/dalgorm.go
@@ -415,6 +415,18 @@ func (d *Dalgorm) IsDuplicationError(err error) bool {
 	return strings.Contains(strings.ToLower(err.Error()), "duplicate")
 }
 
+// IsCachedPlanError checks if the error is related to postgres cached query plan
+// This error occurs occasionally in Postgres when reusing a cached query
+// plan. It can be safely ignored since it does not actually affect results.
+func (d *Dalgorm) IsCachedPlanError(err error) bool {
+	return strings.Contains(strings.ToLower(err.Error()), "cached plan must not change result type")
+}
+
+// IsJsonOrderError checks if the error is related to postgres json ordering
+func (d *Dalgorm) IsJsonOrderError(err error) bool {
+	return strings.Contains(err.Error(), "identify an ordering operator for type json")
+}
+
 // RawCursor (Deprecated) executes raw sql query and returns a database cursor
 func (d *Dalgorm) RawCursor(query string, params ...interface{}) (*sql.Rows, errors.Error) {
 	rows, err := d.db.Raw(query, params...).Rows()
@@ -436,5 +448,12 @@ func (d *Dalgorm) convertGormError(err error) errors.Error {
 	if d.IsDuplicationError(err) {
 		return errors.BadInput.WrapRaw(err)
 	}
+	if d.IsJsonOrderError(err) {
+		return errors.BadInput.WrapRaw(err)
+	}
+	if d.IsCachedPlanError(err) {
+		return nil
+	}
+
 	panic(err)
 }
diff --git a/backend/impls/dalgorm/dalgorm_transaction.go b/backend/impls/dalgorm/dalgorm_transaction.go
index ed839e06a..9ba88c935 100644
--- a/backend/impls/dalgorm/dalgorm_transaction.go
+++ b/backend/impls/dalgorm/dalgorm_transaction.go
@@ -68,19 +68,20 @@ func (t *DalgormTransaction) LockTables(lockTables dal.LockTables) errors.Error
 		}
 		return t.Exec(fmt.Sprintf("LOCK TABLES %s", clause))
 	case "postgres":
-		clause := ""
 		for _, lockTable := range lockTables {
-			if clause != "" {
-				clause += ", "
-			}
-			clause += lockTable.TableName()
+			var clause string
 			if lockTable.Exclusive {
-				clause += "  IN EXCLUSIVE MODE"
+				clause = "EXCLUSIVE"
 			} else {
-				clause += "  IN SHARE MODE"
+				clause = "SHARE"
+			}
+			stmt := fmt.Sprintf("LOCK TABLE %s IN %s MODE;", lockTable.TableName(), clause)
+			err := t.Exec(stmt)
+			if err != nil {
+				return err
 			}
 		}
-		return t.Exec(fmt.Sprintf("LOCK TABLE %s", clause))
+		return nil
 	default:
 		panic(fmt.Errorf("unknown dialect %s", t.Dialect()))
 	}