fix(performance): try to improove performance of inbrdv query

Author: ooemperor

pkg/builder/InbRdv.go

@@ -55,8 +55,11 @@ func BuildInbRdvSatDeleteQuery(tableName string) (string, error) {
 	}
 	script += fmt.Sprintf("UPDATE rdv.%s_sat ", tableName)
 	script += "SET delete_dts = NOW() "
-	script += fmt.Sprintf("WHERE frh NOT IN (SELECT frh FROM rdv.%s_sat_cur) ", tableName)
-	script += "AND delete_dts IS NULL;"
+	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 s.frh IS NULL;"
+	// script += fmt.Sprintf("WHERE frh NOT IN (SELECT frh FROM rdv.%s_sat_cur) ", tableName)
+	// script += "AND delete_dts IS NULL;"
 
 	return script, nil
 }
@@ -71,8 +74,9 @@ func BuildInbRdvSatInsertQuery(tableName string) (string, error) {
 	}
 
 	script += fmt.Sprintf("INSERT INTO rdv.%s_sat ", tableName)
-	script += fmt.Sprintf("SELECT * FROM rdv.%s_sat_cur ", tableName)
-	script += fmt.Sprintf("WHERE frh NOT IN (SELECT frh FROM rdv.%s_sat);", tableName)
+	script += fmt.Sprintf("SELECT sc.* FROM rdv.%s_sat_cur AS sc LEFT JOIN rdv.%s_sat AS s ON s.frh = sc.frh ", tableName, tableName)
+	script += "WHERE s.frh IS NULL;"
+	// script += fmt.Sprintf("WHERE frh NOT IN (SELECT frh FROM rdv.%s_sat);", tableName)
 
 	return script, nil
 }

pkg/builder/InbRdv_test.go

@@ -64,7 +64,11 @@ func TestBuildInbRdvSatDeleteQuery(t *testing.T) {
 		t.Fatalf("Error on BuildInbRdvSatDeleteQuery build: %v", err)
 	}
 
-	if query != "UPDATE rdv.TestTable_sat SET delete_dts = NOW() WHERE frh NOT IN (SELECT frh FROM rdv.TestTable_sat_cur) AND delete_dts IS NULL;" {
+	// if query != "UPDATE rdv.TestTable_sat SET delete_dts = NOW() WHERE frh NOT IN (SELECT frh FROM rdv.TestTable_sat_cur) AND delete_dts IS NULL;" {
+	// 	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 s.frh IS NULL;" {
 		t.Fatalf("BuildInbRdvSatDeleteQuery incorrect: %v", query)
 	}
 }
@@ -77,7 +81,11 @@ func TestBuildInbRdvSatInsertQuery(t *testing.T) {
 		t.Fatalf("Error on BuildInbRdvSatInsertQuery build: %v", err)
 	}
 
-	if query != "INSERT INTO rdv.TestTable_sat SELECT * FROM rdv.TestTable_sat_cur WHERE frh NOT IN (SELECT frh FROM rdv.TestTable_sat);" {
+	// if query != "INSERT INTO rdv.TestTable_sat SELECT * FROM rdv.TestTable_sat_cur WHERE frh NOT IN (SELECT frh FROM rdv.TestTable_sat);" {
+	// 	t.Fatalf("BuildInbRdvSatInsertQuery incorrect: %v", query)
+	// 	}
+	if query != "INSERT INTO rdv.TestTable_sat SELECT sc.* FROM rdv.TestTable_sat_cur AS sc LEFT JOIN rdv.TestTable_sat AS s ON s.frh = sc.frh WHERE s.frh IS NULL;" {
 		t.Fatalf("BuildInbRdvSatInsertQuery 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() WHERE frh NOT IN (SELECT frh FROM rdv.testTableDelete_sat_cur) AND delete_dts IS NULL;"), wantErr: false},
+		{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 s.frh IS NULL;"), wantErr: false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -175,7 +175,7 @@ func TestRdvPipeline_buildSatInsertExecutor(t *testing.T) {
 		wantErr bool
 	}{
 		{name: "SatInsertTest1", fields: fields{nil, ""}, want: nil, wantErr: true},
-		{name: "SatInsertTest2", fields: fields{nil, "testTableSatInsert"}, want: processors.NewSQLExecutor(nil, "INSERT INTO rdv.testTableSatInsert_sat SELECT * FROM rdv.testTableSatInsert_sat_cur WHERE frh NOT IN (SELECT frh FROM rdv.testTableSatInsert_sat);"), wantErr: false},
+		{name: "SatInsertTest2", fields: fields{nil, "testTableSatInsert"}, want: processors.NewSQLExecutor(nil, "INSERT INTO rdv.testTableSatInsert_sat SELECT sc.* FROM rdv.testTableSatInsert_sat_cur AS sc LEFT JOIN rdv.testTableSatInsert_sat AS s ON s.frh = sc.frh WHERE s.frh IS NULL;"), wantErr: false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {