only lookup snapshot name when ParentMirrorName set

PeerDB-io · Oct 22, 2024 · a850f6c · a850f6c
1 parent 8c2c30d
commit a850f6c
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 17 deletions.
diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
@@ -489,13 +489,12 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context,
 	})
 	defer shutdown()
 
-	snapshotFlowName := config.ParentMirrorName
-	if snapshotFlowName == "" {
-		snapshotFlowName = config.FlowJobName
-	}
-	_, snapshotName, _, err := shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, snapshotFlowName)
-	if err != nil {
-		return nil, err
+	snapshotName := ""
+	if config.ParentMirrorName != "" {
+		_, snapshotName, _, err = shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, config.ParentMirrorName)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	partitions, err := srcConn.GetQRepPartitions(ctx, config, last, snapshotName)

diff --git a/flow/activities/flowable_core.go b/flow/activities/flowable_core.go
@@ -424,13 +424,12 @@ func replicateQRepPartition[TRead any, TWrite any, TSync connectors.QRepSyncConn
 	var rowsSynced int
 	errGroup, errCtx := errgroup.WithContext(ctx)
 	errGroup.Go(func() error {
-		snapshotFlowName := config.ParentMirrorName
-		if snapshotFlowName == "" {
-			snapshotFlowName = config.FlowJobName
-		}
-		_, snapshotName, _, err := shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, snapshotFlowName)
-		if err != nil {
-			return err
+		snapshotName := ""
+		if config.ParentMirrorName != "" {
+			_, snapshotName, _, err = shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, config.ParentMirrorName)
+			if err != nil {
+				return err
+			}
 		}
 
 		srcConn, err := connectors.GetByNameAs[TPull](ctx, config.Env, a.CatalogPool, config.SourceName)
@@ -514,9 +513,13 @@ func replicateXminPartition[TRead any, TWrite any, TSync connectors.QRepSyncConn
 	var currentSnapshotXmin int64
 	var rowsSynced int
 	errGroup.Go(func() error {
-		_, snapshotName, _, err := shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, config.FlowJobName)
-		if err != nil {
-			return err
+		snapshotName := ""
+		if config.ParentMirrorName != "" {
+			var err error
+			_, snapshotName, _, err = shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, config.ParentMirrorName)
+			if err != nil {
+				return err
+			}
 		}
 
 		srcConn, err := connectors.GetByNameAs[*connpostgres.PostgresConnector](ctx, config.Env, a.CatalogPool, config.SourceName)