Add a config to disable CDC store for ClickHouse (#4438)

Building on #4366
- Introduce `PEERDB_CLICKHOUSE_CDC_STORE_ENABLED` to allow separating
handling of cdc store enablement for ClickHouse (since other
destinations like PG-to-PG and PG-to-Snowflake requires this to be
enabled).
- Still honor existing services with CDC store disabled, to avoid OOM
issues resurfacing
- Currently `PEERDB_CLICKHOUSE_CDC_STORE_ENABLED` is default to `true`,
can be updated to `false` once we are ready to flip the switch (PR is
currently a no-op)

---------

Co-authored-by: Joy Gao <17896160+jgao54@users.noreply.github.com>

Author: ilidemi

flow/activities/flowable.go

@@ -71,15 +71,11 @@ func getInitialNormalizeBatchID(
 	catalogPool shared.CatalogPool,
 	env map[string]string,
 	destinationName string,
+	destinationType protos.DBType,
 	flowName string,
 ) (int64, error) {
-	dstPeer, err := connectors.LoadPeer(ctx, catalogPool, destinationName)
-	if err != nil {
-		return 0, fmt.Errorf("failed to load destination peer for normalize state: %w", err)
-	}
-
 	// Postgres keeps normalize progress in destination-local metadata.
-	if dstPeer.Type == protos.DBType_POSTGRES {
+	if destinationType == protos.DBType_POSTGRES {
 		dstPgConn, dstClose, err := connectors.GetPostgresConnectorByName(ctx, env, catalogPool, destinationName)
 		if err != nil {
 			return 0, fmt.Errorf("failed to get postgres destination connector for normalize state: %w", err)
@@ -360,7 +356,14 @@ func (a *FlowableActivity) SyncFlow(
 	ctx = internal.WithOperationContext(ctx, protos.FlowOperation_FLOW_OPERATION_SYNC)
 	logger := internal.LoggerFromCtx(ctx)
 
-	srcConn, srcClose, err := connectors.GetByNameAs[connectors.CDCPullConnectorCore](ctx, config.Env, a.CatalogPool, config.SourceName)
+	destinationType, err := connectors.LoadPeerType(ctx, a.CatalogPool, config.DestinationName)
+	if err != nil {
+		return a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to load destination peer type: %w", err))
+	}
+
+	srcConn, srcClose, err := connectors.GetByNameWithCDCDestinationTypeAs[connectors.CDCPullConnectorCore](
+		ctx, config.Env, a.CatalogPool, config.SourceName, destinationType,
+	)
 	if err != nil {
 		return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
 	}
@@ -383,7 +386,7 @@ func (a *FlowableActivity) SyncFlow(
 	normResponses := concurrency.NewLastChan()
 
 	lastNormBatchID, err := getInitialNormalizeBatchID(
-		ctx, logger, a.CatalogPool, config.Env, config.DestinationName, config.FlowJobName,
+		ctx, logger, a.CatalogPool, config.Env, config.DestinationName, destinationType, config.FlowJobName,
 	)
 	if err != nil {
 		return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
@@ -401,7 +404,7 @@ func (a *FlowableActivity) SyncFlow(
 	// under normal steady operation where the batch hits idle timeout every time it will match the hours very closely
 	// effective hours will be longer if pull is idling, or there are waits on big transactions,
 	// or the sync interval is so small that start/stop overhead starts being visible
-	// will be shorter if the batches hit the size limit rather rather than idle timeout
+	// will be shorter if the batches hit the size limit rather than idle timeout
 	normBufferSize := normBufferHours * 3600 / int64(idleTimeout.Seconds())
 	// Normalize is always 1 batch behind, allow 2 to still run in parallel with pull-sync
 	normBufferSize = max(normBufferSize, 2)

flow/connectors/core.go

@@ -546,9 +546,13 @@ func LoadPeer(ctx context.Context, catalogPool shared.CatalogPool, peerName stri
 }
 
 func GetConnector(ctx context.Context, env map[string]string, config *protos.Peer) (Connector, error) {
+	return getConnector(ctx, env, config, protos.DBType_DBTYPE_UNKNOWN)
+}
+
+func getConnector(ctx context.Context, env map[string]string, config *protos.Peer, cdcDestinationType protos.DBType) (Connector, error) {
 	switch inner := config.Config.(type) {
 	case *protos.Peer_PostgresConfig:
-		return connpostgres.NewPostgresConnector(ctx, env, inner.PostgresConfig)
+		return connpostgres.NewPostgresConnectorWithCDCDestination(ctx, env, inner.PostgresConfig, cdcDestinationType)
 	case *protos.Peer_BigqueryConfig:
 		return connbigquery.NewBigQueryConnector(ctx, inner.BigqueryConfig)
 	case *protos.Peer_SnowflakeConfig:
@@ -579,7 +583,7 @@ var noopClose = func(context.Context) {}
 // Gets typed connector by config. Returns a close function to recruit the compiler into helping us avoid connection leaks.
 func GetAs[T Connector](ctx context.Context, env map[string]string, config *protos.Peer) (T, func(context.Context), error) {
 	var none T
-	conn, err := GetConnector(ctx, env, config)
+	conn, err := getConnector(ctx, env, config, protos.DBType_DBTYPE_UNKNOWN)
 	if err != nil {
 		return none, noopClose, exceptions.NewPeerCreateError(err)
 	}
@@ -621,6 +625,34 @@ func GetByNameAs[T Connector](
 	return conn, connClose, err
 }
 
+// Gets connector by name, aware of CDC destination connector type.
+// Returns a close function to recruit the compiler into helping us avoid connection leaks.
+func GetByNameWithCDCDestinationTypeAs[T Connector](
+	ctx context.Context, env map[string]string, catalogPool shared.CatalogPool, name string, destinationType protos.DBType,
+) (T, func(context.Context), error) {
+	var none T
+	peer, err := LoadPeer(ctx, catalogPool, name)
+	if err != nil {
+		return none, noopClose, err
+	}
+	conn, err := getConnector(ctx, env, peer, destinationType)
+	if err != nil {
+		return none, noopClose, exceptions.NewPeerCreateError(err)
+	}
+
+	if tconn, ok := conn.(T); ok {
+		connClose := func(closeCtx context.Context) {
+			if err := conn.Close(); err != nil {
+				internal.LoggerFromCtx(closeCtx).Error("error closing connector", slog.Any("error", err))
+			}
+		}
+		return tconn, connClose, nil
+	} else {
+		conn.Close()
+		return none, noopClose, errors.ErrUnsupported
+	}
+}
+
 // Gets Postgres connector by name. Returns a close function to recruit the compiler into helping us avoid connection leaks.
 func GetPostgresConnectorByName(
 	ctx context.Context,

flow/connectors/postgres/cdc.go

@@ -525,13 +525,9 @@ func PullCdcRecords[Items model.Items](
 		}
 	}
 
-	// Remove exceptions.PrimaryKeyModifiedError and its classification when cdc store is removed
-	cdcStoreEnabled, err := internal.PeerDBCDCStoreEnabled(ctx, req.Env)
-	if err != nil {
-		return err
-	}
 	var cdcRecordsStorage *utils.CDCStore[Items]
-	if cdcStoreEnabled {
+	if p.cdcStoreEnabled {
+		var err error
 		cdcRecordsStorage, err = utils.NewCDCStore[Items](ctx, req.Env, p.flowJobName)
 		if err != nil {
 			return err

flow/connectors/postgres/postgres.go

@@ -51,9 +51,25 @@ type PostgresConnector struct {
 	replLock               sync.Mutex
 	closeLock              sync.Mutex
 	pgVersion              shared.PGVersion
+
+	cdcStoreEnabled bool
+}
+
+func NewPostgresConnectorWithCDCDestination(
+	ctx context.Context, env map[string]string, pgConfig *protos.PostgresConfig, destinationType protos.DBType,
+) (*PostgresConnector, error) {
+	return newPostgresConnector(ctx, env, pgConfig, destinationType)
 }
 
-func NewPostgresConnector(ctx context.Context, env map[string]string, pgConfig *protos.PostgresConfig) (*PostgresConnector, error) {
+func NewPostgresConnector(
+	ctx context.Context, env map[string]string, pgConfig *protos.PostgresConfig,
+) (*PostgresConnector, error) {
+	return newPostgresConnector(ctx, env, pgConfig, protos.DBType_DBTYPE_UNKNOWN)
+}
+
+func newPostgresConnector(
+	ctx context.Context, env map[string]string, pgConfig *protos.PostgresConfig, destinationType protos.DBType,
+) (*PostgresConnector, error) {
 	logger := internal.LoggerFromCtx(ctx)
 	flowNameInApplicationName, err := internal.PeerDBApplicationNamePerMirrorName(ctx, nil)
 	if err != nil {
@@ -114,6 +130,17 @@ func NewPostgresConnector(ctx context.Context, env map[string]string, pgConfig *
 		metadataSchema = *pgConfig.MetadataSchema
 	}
 
+	cdcStoreEnabled, err := internal.PeerDBCDCStoreEnabled(ctx, env)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch CDC Store setting: %w", err)
+	}
+	if cdcStoreEnabled && destinationType == protos.DBType_CLICKHOUSE {
+		cdcStoreEnabled, err = internal.PeerDBClickHouseCDCStoreEnabled(ctx, env)
+		if err != nil {
+			return nil, fmt.Errorf("failed to fetch ClickHouse CDC Store setting: %w", err)
+		}
+	}
+
 	connector := &PostgresConnector{
 		logger:                 logger,
 		Config:                 pgConfig,
@@ -130,6 +157,7 @@ func NewPostgresConnector(ctx context.Context, env map[string]string, pgConfig *
 		pgVersion:              0,
 		typeMap:                pgtype.NewMap(),
 		rdsAuth:                rdsAuth,
+		cdcStoreEnabled:        cdcStoreEnabled,
 	}
 
 	tunnel.StartKeepalive(context.Background(), func() {

flow/internal/dynamicconf.go

@@ -68,9 +68,17 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		Description:      "Controls whether to enable the store for recovering unchanged Postgres TOAST values within a CDC batch",
 		DefaultValue:     "true",
 		ValueType:        protos.DynconfValueType_BOOL,
-		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
 		TargetForSetting: protos.DynconfTarget_ALL,
 	},
+	{
+		Name:             "PEERDB_CLICKHOUSE_CDC_STORE_ENABLED",
+		Description:      "Override PEERDB_CDC_STORE_ENABLED when destination is ClickHouse",
+		DefaultValue:     "true",
+		ValueType:        protos.DynconfValueType_BOOL,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
+		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
+	},
 	{
 		Name:             "PEERDB_CDC_DISK_SPILL_RECORDS_THRESHOLD",
 		Description:      "CDC: number of records beyond which records are written to disk instead",
@@ -649,6 +657,10 @@ func PeerDBCDCStoreEnabled(ctx context.Context, env map[string]string) (bool, er
 	return dynamicConfBool(ctx, env, "PEERDB_CDC_STORE_ENABLED")
 }
 
+func PeerDBClickHouseCDCStoreEnabled(ctx context.Context, env map[string]string) (bool, error) {
+	return dynamicConfBool(ctx, env, "PEERDB_CLICKHOUSE_CDC_STORE_ENABLED")
+}
+
 func PeerDBCDCDiskSpillRecordsThreshold(ctx context.Context, env map[string]string) (int64, error) {
 	return dynamicConfSigned[int64](ctx, env, "PEERDB_CDC_DISK_SPILL_RECORDS_THRESHOLD")
 }