fix(transactions): add commit to the wrapper for better performance

Author: ooemperor

pkg/builder/InbRdv.go

@@ -53,11 +53,12 @@ func BuildInbRdvSatDeleteQuery(tableName string) (string, error) {
 	if tableName == "" {
 		return "", fmt.Errorf("the tablename cannot be blank")
 	}
-	script += fmt.Sprintf("UPDATE rdv.%s_sat ", tableName)
+	script += fmt.Sprintf("UPDATE rdv.%s_sat s ", tableName)
 	script += "SET delete_dts = NOW() "
-	script += fmt.Sprintf("FROM rdv.%s_sat AS s ", tableName)
-	script += fmt.Sprintf("LEFT JOIN rdv.%s_sat_cur AS sc on s.frh = sc.frh ", tableName)
-	script += "WHERE sc.frh IS NULL AND s.delete_dts IS NULL;"
+	script += fmt.Sprintf("WHERE s.delete_dts IS NULL AND NOT EXISTS (SELECT 1 FROM rdv.%s_sat_cur sc WHERE sc.frh = s.frh);", tableName)
+	// script += fmt.Sprintf("FROM rdv.%s_sat AS s ", tableName)
+	// script += fmt.Sprintf("LEFT JOIN rdv.%s_sat_cur AS sc on s.frh = sc.frh ", tableName)
+	// script += "WHERE sc.frh IS NULL AND s.delete_dts IS NULL;"
 	// script += fmt.Sprintf("WHERE frh NOT IN (SELECT frh FROM rdv.%s_sat_cur) ", tableName)
 	// script += "AND delete_dts IS NULL;"
 

pkg/builder/InbRdv_test.go

@@ -68,7 +68,7 @@ func TestBuildInbRdvSatDeleteQuery(t *testing.T) {
 	// 	t.Fatalf("BuildInbRdvSatDeleteQuery incorrect: %v", query)
 	// }
 
-	if query != "UPDATE rdv.TestTable_sat SET delete_dts = NOW() FROM rdv.TestTable_sat AS s LEFT JOIN rdv.TestTable_sat_cur AS sc on s.frh = sc.frh WHERE sc.frh IS NULL AND s.delete_dts IS NULL;" {
+	if query != "UPDATE rdv.TestTable_sat s SET delete_dts = NOW() WHERE s.delete_dts IS NULL AND NOT EXISTS (SELECT 1 FROM rdv.TestTable_sat_cur sc WHERE sc.frh = s.frh);" {
 		t.Fatalf("BuildInbRdvSatDeleteQuery incorrect: %v", query)
 	}
 }

pkg/pipeline/inbrdv/RdvPipelineBuilder_test.go

@@ -140,7 +140,7 @@ func TestRdvPipeline_buildSatMarkDelete(t *testing.T) {
 		wantErr bool
 	}{
 		{name: "SatCurDeleteTest1", fields: fields{nil, ""}, want: nil, wantErr: true},
-		{name: "SatCurDeleteTest2", fields: fields{nil, "testTableDelete"}, want: processors.NewSQLExecutor(nil, "UPDATE rdv.testTableDelete_sat SET delete_dts = NOW() FROM rdv.testTableDelete_sat AS s LEFT JOIN rdv.testTableDelete_sat_cur AS sc on s.frh = sc.frh WHERE sc.frh IS NULL AND s.delete_dts IS NULL;"), wantErr: false},
+		{name: "SatCurDeleteTest2", fields: fields{nil, "testTableDelete"}, want: processors.NewSQLExecutor(nil, "UPDATE rdv.testTableDelete_sat s SET delete_dts = NOW() WHERE s.delete_dts IS NULL AND NOT EXISTS (SELECT 1 FROM rdv.testTableDelete_sat_cur sc WHERE sc.frh = s.frh);"), wantErr: false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {