Optimisation: Do not open destination connection for last offset fetch unless Postgres to Postgres (#3367)

Decoupling from #3206 and addressing the [comment](#3206 (comment)) there.

This PR makes it so that we do not connect to the destination peer to get the last offset for sync flow.

---------

Co-authored-by: Philip Dubé <[email protected]>

Author: Amogh-Bharadwaj

flow/activities/flowable_core.go

@@ -21,7 +21,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors"
-	connmysql "github.com/PeerDB-io/peerdb/flow/connectors/mysql"
+	connmetadata "github.com/PeerDB-io/peerdb/flow/connectors/external_metadata"
 	connpostgres "github.com/PeerDB-io/peerdb/flow/connectors/postgres"
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils/monitoring"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -139,17 +139,20 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	}
 
 	lastOffset, err := func() (model.CdcCheckpoint, error) {
-		if myConn, isMy := any(srcConn).(*connmysql.MySqlConnector); isMy {
-			return myConn.GetLastOffset(ctx, config.FlowJobName)
-		} else {
-			dstConn, err := connectors.GetByNameAs[TSync](ctx, config.Env, a.CatalogPool, config.DestinationName)
+		// special case pg-pg replication, where offsets are stored on destination instead of catalog
+		if _, isPg := any(srcConn).(*connpostgres.PostgresConnector); isPg {
+			dstPgConn, err := connectors.GetByNameAs[*connpostgres.PostgresConnector](ctx, config.Env, a.CatalogPool, config.DestinationName)
 			if err != nil {
-				return model.CdcCheckpoint{}, fmt.Errorf("failed to get destination connector: %w", err)
+				if !errors.Is(err, errors.ErrUnsupported) {
+					return model.CdcCheckpoint{}, fmt.Errorf("failed to get destination connector to get last offset: %w", err)
+				}
+				// else fallthrough to loading from catalog
+			} else {
+				return dstPgConn.GetLastOffset(ctx, config.FlowJobName)
 			}
-			defer connectors.CloseConnector(ctx, dstConn)
-
-			return dstConn.GetLastOffset(ctx, config.FlowJobName)
 		}
+		pgMetadata := connmetadata.NewPostgresMetadataFromCatalog(logger, a.CatalogPool)
+		return pgMetadata.GetLastOffset(ctx, flowName)
 	}()
 	if err != nil {
 		return nil, a.Alerter.LogFlowError(ctx, flowName, err)

flow/connectors/core.go

@@ -167,12 +167,6 @@ type CDCSyncConnectorCore interface {
 	// SetupMetadataTables creates the metadata table [PEERDB_MIRROR_JOBS] if necessary.
 	SetupMetadataTables(ctx context.Context) error
 
-	// GetLastOffset gets the last offset from the metadata table on the destination
-	GetLastOffset(ctx context.Context, jobName string) (model.CdcCheckpoint, error)
-
-	// SetLastOffset updates the last offset on the metadata table on the destination
-	SetLastOffset(ctx context.Context, jobName string, lastOffset model.CdcCheckpoint) error
-
 	// GetLastSyncBatchID gets the last batch synced to the destination from the metadata table
 	GetLastSyncBatchID(ctx context.Context, jobName string) (int64, error)