make sure streaming query handle invalid sequence error

Author: jgao54

flow/connectors/mysql/qrep.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 	"github.com/go-mysql-org/go-mysql/mysql"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
@@ -281,7 +282,7 @@ func (c *MySqlConnector) PullQRepRecords(
 		}
 
 		if err := c.ExecuteSelectStreaming(ctx, query, &rs, onRow, onResult); err != nil {
-			return 0, 0, err
+			return 0, 0, exceptions.NewMySQLStreamingError(err)
 		}
 	} else {
 		var rangeStart string
@@ -329,7 +330,7 @@ func (c *MySqlConnector) PullQRepRecords(
 		}
 
 		if err := c.ExecuteSelectStreaming(ctx, query, &rs, onRow, onResult); err != nil {
-			return 0, 0, err
+			return 0, 0, exceptions.NewMySQLStreamingError(err)
 		}
 	}