boilerplate

flow/workflows/qrep_flow.go

@@ -33,13 +33,15 @@ type QRepPartitionFlowExecution struct {
 	runUUID         string
 }
 
+var InitialLastPartition = &protos.QRepPartition{
+	PartitionId: "not-applicable-partition",
+	Range:       nil,
+}
+
 // returns a new empty QRepFlowState
 func NewQRepFlowState() *protos.QRepFlowState {
 	return &protos.QRepFlowState{
-		LastPartition: &protos.QRepPartition{
-			PartitionId: "not-applicable-partition",
-			Range:       nil,
-		},
+		LastPartition:          InitialLastPartition,
 		NumPartitionsProcessed: 0,
 		NeedsResync:            true,
 		CurrentFlowStatus:      protos.FlowStatus_STATUS_RUNNING,
@@ -482,10 +484,11 @@ func QRepWaitForNewRowsWorkflow(ctx workflow.Context, config *protos.QRepConfig,
 	if err != nil {
 		return fmt.Errorf("error checking for new rows: %w", err)
 	}
-	hasNewRows := result.Found
 
+	hasNewRows := result.Found
+	optedForOverwrite := config.WriteMode.WriteType == protos.QRepWriteType_QREP_WRITE_MODE_OVERWRITE
 	// If no new rows are found, continue as new
-	if !hasNewRows {
+	if !hasNewRows || optedForOverwrite {
 		waitBetweenBatches := 5 * time.Second
 		if config.WaitBetweenBatchesSeconds > 0 {
 			waitBetweenBatches = time.Duration(config.WaitBetweenBatchesSeconds) * time.Second
@@ -496,6 +499,9 @@ func QRepWaitForNewRowsWorkflow(ctx workflow.Context, config *protos.QRepConfig,
 		}
 
 		logger.Info("QRepWaitForNewRowsWorkflow: continuing the loop")
+		if optedForOverwrite {
+			return nil
+		}
 		return workflow.NewContinueAsNewError(ctx, QRepWaitForNewRowsWorkflow, config, lastPartition)
 	}
 
@@ -570,15 +576,21 @@ func QRepFlowWorkflow(
 		return err
 	}
 
-	if !config.InitialCopyOnly && state.LastPartition != nil {
-		if err := q.waitForNewRows(ctx, signalChan, state.LastPartition); err != nil {
+	optedForOverwrite := config.WriteMode.WriteType == protos.QRepWriteType_QREP_WRITE_MODE_OVERWRITE
+	lastPartition := state.LastPartition
+	if optedForOverwrite {
+		lastPartition = InitialLastPartition
+	}
+
+	if !config.InitialCopyOnly && lastPartition != nil {
+		if err := q.waitForNewRows(ctx, signalChan, lastPartition); err != nil {
 			return err
 		}
 	}
 
 	if q.activeSignal != model.PauseSignal {
 		logger.Info("fetching partitions to replicate for peer flow")
-		partitions, err := q.GetPartitions(ctx, state.LastPartition)
+		partitions, err := q.GetPartitions(ctx, lastPartition)
 		if err != nil {
 			return fmt.Errorf("failed to get partitions: %w", err)
 		}
@@ -606,7 +618,7 @@ func QRepFlowWorkflow(
 		logger.Info(fmt.Sprintf("%d partitions processed", len(partitions.Partitions)))
 		state.NumPartitionsProcessed += uint64(len(partitions.Partitions))
 
-		if len(partitions.Partitions) > 0 {
+		if len(partitions.Partitions) > 0 && !optedForOverwrite {
 			state.LastPartition = partitions.Partitions[len(partitions.Partitions)-1]
 		}
 	}
@@ -621,7 +633,7 @@ func QRepFlowWorkflow(
 	}
 
 	logger.Info("Continuing as new workflow",
-		slog.Any("Last Partition", state.LastPartition),
+		slog.Any("Last Partition", lastPartition),
 		slog.Uint64("Number of Partitions Processed", state.NumPartitionsProcessed))
 
 	if q.activeSignal == model.PauseSignal {