fix(clickhouse): quorum writes on replicated clusters to stop normalize dropping rows

flow/connectors/clickhouse/clickhouse.go

@@ -213,6 +213,25 @@ func Connect(ctx context.Context, env map[string]string, config *protos.Clickhou
 	if config.Cluster != "" {
 		settings["insert_distributed_sync"] = uint64(1)
 	}
+	if config.Replicated {
+		// On a multi-replica ReplicatedMergeTree cluster, normalize populates the raw table and then
+		// immediately reads it back via INSERT ... SELECT. Those two statements may run on different
+		// connections/replicas (e.g. behind a load-balanced host), so the read can hit a replica that
+		// has not yet replicated the just-written parts. select_sequential_consistency (set above) only
+		// guarantees visibility of quorum-inserted blocks, so without quorum writes it enforces nothing
+		// and normalize silently drops rows it marks as processed. Writing with quorum closes the race:
+		// insert_quorum_parallel=0 is required for sequential-consistency reads to be honored.
+		if quorum, err := internal.PeerDBClickHouseEnableReplicatedQuorum(ctx, env); err != nil {
+			return nil, fmt.Errorf("failed to load replicated quorum config: %w", err)
+		} else if quorum {
+			// Supported by ClickHouse 22.9+
+			settings["insert_quorum"] = "auto"
+			settings["insert_quorum_parallel"] = uint64(0)
+			// Explicitly pin the read side of the read-your-writes contract alongside the quorum
+			// writes so it stays paired with them regardless of the global default.
+			settings["select_sequential_consistency"] = uint64(1)
+		}
+	}
 	clientName, err := internal.PeerDBClickHouseClientName(ctx, env)
 	if err != nil {
 		return nil, fmt.Errorf("failed to load ClickHouse client name: %w", err)

flow/internal/dynamicconf.go

@@ -305,6 +305,17 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
+	{
+		Name: "PEERDB_CLICKHOUSE_ENABLE_REPLICATED_QUORUM",
+		Description: "On Replicated ClickHouse clusters, write raw/normalize inserts with quorum " +
+			"(insert_quorum=auto, insert_quorum_parallel=0) so that select_sequential_consistency reads " +
+			"see them regardless of which replica serves the read. Prevents normalize silently dropping " +
+			"rows when the read hits a replica that has not yet replicated the just-written raw parts.",
+		DefaultValue:     "false",
+		ValueType:        protos.DynconfValueType_BOOL,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
+		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
+	},
 	{
 		Name:             "PEERDB_CLICKHOUSE_UNBOUNDED_NUMERIC_AS_STRING",
 		Description:      "Map unbounded numerics in Postgres to String in ClickHouse to preserve precision and scale",
@@ -785,6 +796,10 @@ func PeerDBClickHouseParallelNormalize(ctx context.Context, env map[string]strin
 	return dynamicConfSigned[int](ctx, env, "PEERDB_CLICKHOUSE_PARALLEL_NORMALIZE")
 }
 
+func PeerDBClickHouseEnableReplicatedQuorum(ctx context.Context, env map[string]string) (bool, error) {
+	return dynamicConfBool(ctx, env, "PEERDB_CLICKHOUSE_ENABLE_REPLICATED_QUORUM")
+}
+
 func PeerDBEnableClickHouseNumericAsString(ctx context.Context, env map[string]string) (bool, error) {
 	return dynamicConfBool(ctx, env, "PEERDB_CLICKHOUSE_UNBOUNDED_NUMERIC_AS_STRING")
 }