remove PEERDB_ENABLE_PARALLEL_SYNC_NORMALIZE (#3354)

This has been default for some time, has never come up to turn it off,
plays vital role in PeerDB's prime directive: always read from slot

Author: serprex

flow/activities/flowable.go

@@ -42,7 +42,6 @@ type CheckMetadataTablesResult struct {
 }
 
 type NormalizeBatchRequest struct {
-	Done    chan struct{}
 	BatchID int64
 }
 

flow/activities/flowable_core.go

@@ -330,27 +330,12 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 	}
 
 	if recordBatchSync.NeedsNormalize() {
-		parallel, err := internal.PeerDBEnableParallelSyncNormalize(ctx, config.Env)
-		if err != nil {
-			return nil, err
-		}
-		var done chan struct{}
-		if !parallel {
-			done = make(chan struct{})
-		}
 		syncState.Store(shared.Ptr("normalizing"))
 		select {
-		case normRequests <- NormalizeBatchRequest{BatchID: res.CurrentSyncBatchID, Done: done}:
+		case normRequests <- NormalizeBatchRequest{BatchID: res.CurrentSyncBatchID}:
 		case <-ctx.Done():
 			return res, nil
 		}
-		if done != nil {
-			select {
-			case <-done:
-			case <-ctx.Done():
-				return res, nil
-			}
-		}
 	}
 
 	return res, nil
@@ -722,8 +707,6 @@ func (a *FlowableActivity) normalizeLoop(
 							continue retryLoop
 						}
 					}
-				} else if req.Done != nil {
-					close(req.Done)
 				}
 				a.OtelManager.Metrics.LastNormalizedBatchIdGauge.Record(ctx, req.BatchID, metric.WithAttributeSet(attribute.NewSet(
 					attribute.String(otel_metrics.FlowNameKey, config.FlowJobName),

flow/internal/dynamicconf.go

@@ -33,9 +33,8 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		TargetForSetting: protos.DynconfTarget_ALL,
 	},
 	{
-		Name: "PEERDB_NORMALIZE_CHANNEL_BUFFER_SIZE",
-		Description: "Advanced setting: changes buffer size of channel PeerDB uses for queueing normalization, " +
-			"use with PEERDB_ENABLE_PARALLEL_SYNC_NORMALIZE",
+		Name:             "PEERDB_NORMALIZE_CHANNEL_BUFFER_SIZE",
+		Description:      "Advanced setting: changes buffer size of channel PeerDB uses for queueing normalization",
 		DefaultValue:     "128",
 		ValueType:        protos.DynconfValueType_INT,
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
@@ -89,14 +88,6 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
 		TargetForSetting: protos.DynconfTarget_ALL,
 	},
-	{
-		Name:             "PEERDB_ENABLE_PARALLEL_SYNC_NORMALIZE",
-		Description:      "Enables parallel sync (moving rows to target) and normalize (updating rows in target table)",
-		DefaultValue:     "true",
-		ValueType:        protos.DynconfValueType_BOOL,
-		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
-		TargetForSetting: protos.DynconfTarget_ALL,
-	},
 	{
 		Name:             "PEERDB_RECONNECT_AFTER_BATCHES",
 		Description:      "Force peerdb to reconnect connection to source after N batches",
@@ -557,10 +548,6 @@ func PeerDBWALHeartbeatQuery(ctx context.Context, env map[string]string) (string
 	return dynLookup(ctx, env, "PEERDB_WAL_HEARTBEAT_QUERY")
 }
 
-func PeerDBEnableParallelSyncNormalize(ctx context.Context, env map[string]string) (bool, error) {
-	return dynamicConfBool(ctx, env, "PEERDB_ENABLE_PARALLEL_SYNC_NORMALIZE")
-}
-
 func PeerDBReconnectAfterBatches(ctx context.Context, env map[string]string) (int32, error) {
 	return dynamicConfSigned[int32](ctx, env, "PEERDB_RECONNECT_AFTER_BATCHES")
 }