From 85baeefdb5519be919c0a31b7be2a94d3d9f164b Mon Sep 17 00:00:00 2001
From: Joy Gao <17896160+jgao54@users.noreply.github.com>
Date: Thu, 25 Jun 2026 14:49:23 -1000
Subject: [PATCH 1/4] treat string partition ranges as sensitive data

---
 flow/activities/flowable.go                   |  14 ++
 flow/connectors/external_metadata/store.go    | 150 +++++++++++++++++-
 .../external_metadata/store_test.go           |  75 +++++++++
 flow/e2e/clickhouse_mysql_test.go             | 103 +++++++++++-
 .../migrations/V54__qrep_partition_ranges.sql |  15 ++
 5 files changed, 347 insertions(+), 10 deletions(-)
 create mode 100644 flow/connectors/external_metadata/store_test.go
 create mode 100644 nexus/catalog/migrations/V54__qrep_partition_ranges.sql

diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
index d1d368c0c..177932fb9 100644
--- a/flow/activities/flowable.go
+++ b/flow/activities/flowable.go
@@ -596,6 +596,12 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context,
 		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, shared.WrapError("failed to get partitions from source", err))
 	}
 	if len(partitions) > 0 {
+		// run before monitoring.InitializeQRepRun to avoid logging sensitive fields to peerdb_stats.qrep_partitions as well
+		if err := connmetadata.OffloadSensitivePartitionRanges(
+			ctx, a.CatalogPool, config.ParentMirrorName, runUUID, partitions,
+		); err != nil {
+			return nil, fmt.Errorf("failed to offload partition ranges: %w", err)
+		}
 		if err := monitoring.InitializeQRepRun(
 			ctx,
 			logger,
@@ -663,6 +669,10 @@ func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
 		return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
 	}
 
+	if err := connmetadata.RestoreSensitivePartitionRanges(ctx, a.CatalogPool, runUUID, partitions.Partitions); err != nil {
+		return fmt.Errorf("failed to rehydrate partition ranges: %w", err)
+	}
+
 	for i, partition := range partitions.Partitions {
 		partLogger := log.With(logger,
 			slog.Int64("batchID", int64(partitions.BatchId)),
@@ -1623,6 +1633,10 @@ func (a *FlowableActivity) QRepHasNewRows(ctx context.Context,
 		if maxValue.(time.Time).After(x.TimestampRange.End.AsTime()) {
 			return true, nil
 		}
+	case *protos.PartitionRange_StringRange:
+		// checking for new rows is only possible for standalone QRepFlowWorkflow;
+		// this is a legacy feature and string partitioning is not supported
+		return false, errors.New("checking for new rows by a string partition range is not supported")
 	default:
 		return false, fmt.Errorf("unknown range type: %v", x)
 	}
diff --git a/flow/connectors/external_metadata/store.go b/flow/connectors/external_metadata/store.go
index 23a4747d6..77ce24c85 100644
--- a/flow/connectors/external_metadata/store.go
+++ b/flow/connectors/external_metadata/store.go
@@ -14,6 +14,7 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 	"go.temporal.io/sdk/log"
 	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/PeerDB-io/peerdb/flow/alerting"
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils/monitoring"
@@ -24,8 +25,9 @@ import (
 )
 
 const (
-	lastSyncStateTableName = "metadata_last_sync_state"
-	qrepTableName          = "metadata_qrep_partitions"
+	lastSyncStateTableName       = "metadata_last_sync_state"
+	qrepTableName                = "metadata_qrep_partitions"
+	qrepPartitionRangesTableName = "metadata_qrep_partition_ranges"
 )
 
 type PostgresMetadata struct {
@@ -286,6 +288,146 @@ func (p *PostgresMetadata) IsQRepPartitionSynced(ctx context.Context, req *proto
 	return exists, nil
 }
 
+// OffloadSensitivePartitionRanges encrypts sensitive partition ranges, persists them to
+// the catalog, and replaces each partition's range in-place with an empty range.
+// Used in conjunction with RestoreSensitivePartitionRanges.
+func OffloadSensitivePartitionRanges(
+	ctx context.Context,
+	pool shared.CatalogPool,
+	parentMirrorName string,
+	runUUID string,
+	partitions []*protos.QRepPartition,
+) error {
+	if !partitionsContainSensitiveRanges(partitions) {
+		return nil
+	}
+
+	key, err := internal.PeerDBCurrentEncKey(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to load current encryption key: %w", err)
+	}
+
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer shared.RollbackTx(tx, internal.LoggerFromCtx(ctx))
+
+	// delete any orphaned rows from previously failed attempt
+	if _, err := tx.Exec(ctx,
+		`DELETE FROM `+qrepPartitionRangesTableName+` WHERE run_uuid=$1`, runUUID,
+	); err != nil {
+		return fmt.Errorf("failed to clear existing partition ranges: %w", err)
+	}
+
+	for _, partition := range partitions {
+		payload, err := proto.Marshal(partition.Range)
+		if err != nil {
+			return fmt.Errorf("failed to marshal partition range: %w", err)
+		}
+		encrypted, err := key.Encrypt(payload)
+		if err != nil {
+			return fmt.Errorf("failed to encrypt partition range: %w", err)
+		}
+
+		if _, err := tx.Exec(ctx,
+			`INSERT INTO `+qrepPartitionRangesTableName+
+				`(parent_mirror_name,run_uuid,partition_uuid,enc_key_id,range_payload) VALUES($1,$2,$3,$4,$5)`,
+			parentMirrorName, runUUID, partition.PartitionId, key.ID, encrypted,
+		); err != nil {
+			return fmt.Errorf("failed to persist encrypted partition range: %w", err)
+		}
+
+		partition.Range = &protos.PartitionRange{
+			Range: &protos.PartitionRange_StringRange{StringRange: &protos.StringPartitionRange{}},
+		}
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("failed to commit partition ranges: %w", err)
+	}
+	return nil
+}
+
+// RestoreSensitivePartitionRanges hydrates sensitive partitions in-place with the decrypted
+// values fetched from the catalog. Used in conjunction with OffloadSensitivePartitionRanges
+func RestoreSensitivePartitionRanges(
+	ctx context.Context,
+	pool shared.CatalogPool,
+	runUUID string,
+	partitions []*protos.QRepPartition,
+) error {
+	if !partitionsContainSensitiveRanges(partitions) {
+		return nil
+	}
+
+	partitionIDs := make([]string, 0, len(partitions))
+	partitionByIDs := make(map[string]*protos.QRepPartition, len(partitions))
+	for _, partition := range partitions {
+		partitionIDs = append(partitionIDs, partition.PartitionId)
+		partitionByIDs[partition.PartitionId] = partition
+	}
+
+	rows, err := pool.Query(ctx,
+		`SELECT partition_uuid,enc_key_id,range_payload FROM `+qrepPartitionRangesTableName+` `+
+			`WHERE run_uuid=$1 AND partition_uuid=ANY($2)`,
+		runUUID, partitionIDs,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to query partition ranges: %w", err)
+	}
+	results, err := pgx.CollectRows(rows, pgx.RowToStructByPos[struct {
+		PartitionID string
+		EncKeyID    string
+		Payload     []byte
+	}])
+	if err != nil {
+		return fmt.Errorf("failed to read encrypted partition ranges: %w", err)
+	}
+
+	for _, res := range results {
+		partition, ok := partitionByIDs[res.PartitionID]
+		if !ok {
+			return fmt.Errorf("received unexpected partition range for partition %s", res.PartitionID)
+		}
+
+		payload, err := internal.Decrypt(ctx, res.EncKeyID, res.Payload)
+		if err != nil {
+			return fmt.Errorf("failed to decrypt partition range for %s: %w", res.PartitionID, err)
+		}
+
+		var partitionRange protos.PartitionRange
+		if err := proto.Unmarshal(payload, &partitionRange); err != nil {
+			return fmt.Errorf("failed to unmarshal partition range for %s: %w", res.PartitionID, err)
+		}
+		partition.Range = &partitionRange
+		delete(partitionByIDs, res.PartitionID)
+	}
+
+	if len(partitionByIDs) > 0 {
+		missing := make([]string, 0, len(partitionByIDs))
+		for id := range partitionByIDs {
+			missing = append(missing, id)
+		}
+		return fmt.Errorf("missing encrypted partition ranges for partitions: %v", missing)
+	}
+
+	return nil
+}
+
+func partitionsContainSensitiveRanges(partitions []*protos.QRepPartition) bool {
+	for _, partition := range partitions {
+		if partition == nil || partition.Range == nil {
+			return false
+		}
+		// String ranges may contain PII data like email, so deem it as sensitive.
+		if _, ok := partition.Range.Range.(*protos.PartitionRange_StringRange); ok {
+			return true
+		}
+	}
+	return false
+}
+
 func (p *PostgresMetadata) SyncFlowCleanup(ctx context.Context, jobName string) error {
 	tx, err := p.pool.Begin(ctx)
 	if err != nil {
@@ -313,5 +455,9 @@ func SyncFlowCleanupInTx(ctx context.Context, tx pgx.Tx, jobName string) error {
 		return err
 	}
 
+	if _, err := tx.Exec(ctx, `DELETE FROM `+qrepPartitionRangesTableName+` WHERE parent_mirror_name = $1`, jobName); err != nil {
+		return err
+	}
+
 	return nil
 }
diff --git a/flow/connectors/external_metadata/store_test.go b/flow/connectors/external_metadata/store_test.go
new file mode 100644
index 000000000..d8ef2258b
--- /dev/null
+++ b/flow/connectors/external_metadata/store_test.go
@@ -0,0 +1,75 @@
+package connmetadata
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/require"
+
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/internal"
+)
+
+func TestOffloadRestoreSensitivePartitionRanges(t *testing.T) {
+	const encKeyID = "test_enc_key"
+	t.Setenv("PEERDB_CURRENT_ENC_KEY_ID", encKeyID)
+	t.Setenv("PEERDB_ENC_KEYS", `[{"id":"`+encKeyID+`","value":"cGVlcmRiX2NpX3Rlc3RfZW5jX2tleV8zMl9ieXRlcyE="}]`)
+
+	ctx := t.Context()
+	pool, err := internal.GetCatalogConnectionPoolFromEnv(ctx)
+	require.NoError(t, err)
+
+	parentMirrorName := "test_offload_restore_sensitive_partition_ranges"
+	runUUID := uuid.NewString()
+	t.Cleanup(func() {
+		_, _ = pool.Exec(ctx, `DELETE FROM `+qrepPartitionRangesTableName+` WHERE parent_mirror_name=$1`, parentMirrorName)
+	})
+
+	p1Start, p1End, p2Start, p2End := "bar", "foo", "hello", "world"
+	partitions := []*protos.QRepPartition{
+		{
+			PartitionId: uuid.NewString(),
+			Range: &protos.PartitionRange{Range: &protos.PartitionRange_StringRange{
+				StringRange: &protos.StringPartitionRange{Start: p1Start, End: p1End},
+			}},
+		},
+		{
+			PartitionId: uuid.NewString(),
+			Range: &protos.PartitionRange{Range: &protos.PartitionRange_StringRange{
+				StringRange: &protos.StringPartitionRange{Start: p2Start, End: p2End, EndInclusive: true},
+			}},
+		},
+	}
+
+	require.NoError(t, OffloadSensitivePartitionRanges(ctx, pool, parentMirrorName, runUUID, partitions))
+	for _, p := range partitions {
+		require.Empty(t, p.Range.GetStringRange().Start)
+		require.Empty(t, p.Range.GetStringRange().End)
+	}
+
+	rows, err := pool.Query(ctx,
+		`SELECT enc_key_id, range_payload FROM `+qrepPartitionRangesTableName+` WHERE parent_mirror_name=$1`,
+		parentMirrorName)
+	require.NoError(t, err)
+	results, err := pgx.CollectRows(rows, pgx.RowToStructByPos[struct {
+		KeyID   string
+		Payload []byte
+	}])
+	require.NoError(t, err)
+	require.Len(t, results, len(partitions))
+	for _, res := range results {
+		require.Equal(t, encKeyID, res.KeyID)
+		for _, plaintext := range []string{p1Start, p1End, p2Start, p2End} {
+			require.NotContains(t, string(res.Payload), plaintext)
+		}
+	}
+
+	require.NoError(t, RestoreSensitivePartitionRanges(ctx, pool, runUUID, partitions))
+	require.Equal(t, p1Start, partitions[0].Range.GetStringRange().Start)
+	require.Equal(t, p1End, partitions[0].Range.GetStringRange().End)
+	require.False(t, partitions[0].Range.GetStringRange().EndInclusive)
+	require.Equal(t, p2Start, partitions[1].Range.GetStringRange().Start)
+	require.Equal(t, p2End, partitions[1].Range.GetStringRange().End)
+	require.True(t, partitions[1].Range.GetStringRange().EndInclusive)
+}
diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go
index 1ff4bae12..bb2403b41 100644
--- a/flow/e2e/clickhouse_mysql_test.go
+++ b/flow/e2e/clickhouse_mysql_test.go
@@ -1760,7 +1760,7 @@ func (s ClickHouseSuite) Test_MySQL_Default_Partition_Key_Parallel_Snapshot() {
 	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,created_at")
 
 	partitionRows, err := s.catalog.Query(s.t.Context(),
-		`SELECT partition_start, partition_end, COALESCE(rows_in_partition, 0)
+		`SELECT partition_start, partition_end, rows_in_partition
 		 FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
 		flowConnConfig.FlowJobName)
 	require.NoError(s.t, err)
@@ -1827,8 +1827,7 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_UUID_Parallel_Snapshot(
 	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,val")
 
 	partitionRows, err := s.catalog.Query(s.t.Context(),
-		`SELECT partition_start, partition_end, COALESCE(rows_in_partition, 0)
-		 FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
+		`SELECT rows_in_partition FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
 		flowConnConfig.FlowJobName)
 	require.NoError(s.t, err)
 	defer partitionRows.Close()
@@ -1836,11 +1835,8 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_UUID_Parallel_Snapshot(
 	var partitionCount int32
 	var totalRows int64
 	for partitionRows.Next() {
-		var partitionStart, partitionEnd *string
 		var rowsInPartition int64
-		require.NoError(s.t, partitionRows.Scan(&partitionStart, &partitionEnd, &rowsInPartition))
-		require.NotNil(s.t, partitionStart)
-		require.NotNil(s.t, partitionEnd)
+		require.NoError(s.t, partitionRows.Scan(&rowsInPartition))
 		totalRows += rowsInPartition
 		partitionCount++
 	}
@@ -1894,7 +1890,7 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Arbitrary_FullTable() {
 	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,val")
 
 	partitionRows, err := s.catalog.Query(s.t.Context(),
-		`SELECT partition_start, partition_end, COALESCE(rows_in_partition, 0)
+		`SELECT partition_start, partition_end, rows_in_partition
 		 FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
 		flowConnConfig.FlowJobName)
 	require.NoError(s.t, err)
@@ -1918,3 +1914,94 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Arbitrary_FullTable() {
 	env.Cancel(s.t.Context())
 	RequireEnvCanceled(s.t, env)
 }
+
+func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Sensitive_Range_Handling() {
+	if _, ok := s.source.(*MySqlSource); !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	srcTable := "test_string_pk_sensitive"
+	srcFullName := s.attachSchemaSuffix(srcTable)
+	dstTable := "test_string_pk_sensitive_dst"
+
+	const numRows = 8
+	const numPartitions = 4
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf("CREATE TABLE %s (id CHAR(36) PRIMARY KEY, val INT NOT NULL)", srcFullName)))
+	for i := 1; i <= numRows; i++ {
+		require.NoError(s.t, s.source.Exec(s.t.Context(),
+			fmt.Sprintf("INSERT INTO %s (id, val) VALUES ('%s', %d)", srcFullName, uuid.NewString(), i)))
+	}
+
+	tableMappings := TableMappings(s, srcTable, dstTable)
+	// a String column can't be used directly as a Distributed sharding key (ClickHouse
+	// requires an integer-typed sharding expression), so hash it for the cluster suite
+	for _, tm := range tableMappings {
+		tm.ShardingKey = "cityHash64(id)"
+	}
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:   s.attachSuffix("string_pk_sensitive"),
+		TableMappings: tableMappings,
+		Destination:   s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+	flowConnConfig.SnapshotMaxParallelWorkers = 4
+	flowConnConfig.SnapshotNumPartitionsOverride = numPartitions
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,val")
+
+	// verify partition start/end are stored in metadata_qrep_partition_ranges
+	rangeRows, err := s.catalog.Query(s.t.Context(),
+		`SELECT enc_key_id, range_payload FROM metadata_qrep_partition_ranges WHERE parent_mirror_name = $1`,
+		flowConnConfig.FlowJobName)
+	require.NoError(s.t, err)
+	defer rangeRows.Close()
+	var offloadedCount int32
+	for rangeRows.Next() {
+		var encKeyID string
+		var rangePayload []byte
+		require.NoError(s.t, rangeRows.Scan(&encKeyID, &rangePayload))
+		// no encryption key is configured for CI so enc_key_id is empty
+		require.Empty(s.t, encKeyID)
+		require.NotEmpty(s.t, rangePayload)
+		offloadedCount++
+	}
+	require.NoError(s.t, rangeRows.Err())
+	require.EqualValues(s.t, numPartitions, offloadedCount)
+
+	// verify partition start/end are stripped from peerdb_stats.qrep_partitions
+	partitionRows, err := s.catalog.Query(s.t.Context(),
+		`SELECT partition_start, partition_end FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
+		flowConnConfig.FlowJobName)
+	require.NoError(s.t, err)
+	defer partitionRows.Close()
+
+	var partitionCount int32
+	for partitionRows.Next() {
+		var partitionStart, partitionEnd *string
+		require.NoError(s.t, partitionRows.Scan(&partitionStart, &partitionEnd))
+		require.Empty(s.t, partitionStart)
+		require.Empty(s.t, partitionEnd)
+		partitionCount++
+	}
+	require.NoError(s.t, partitionRows.Err())
+	require.EqualValues(s.t, numPartitions, partitionCount)
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+
+	// dropping the mirror should delete metadata_qrep_partition_ranges entries from the catalog for this mirror.
+	dropEnv := ExecuteDropFlow(s.t.Context(), tc, flowConnConfig, 0)
+	EnvWaitForFinished(s.t, dropEnv, 3*time.Minute)
+
+	var remainingRanges int64
+	require.NoError(s.t, s.catalog.QueryRow(s.t.Context(),
+		`SELECT COUNT(*) FROM metadata_qrep_partition_ranges WHERE parent_mirror_name = $1`,
+		flowConnConfig.FlowJobName).Scan(&remainingRanges))
+	require.Zero(s.t, remainingRanges)
+}
diff --git a/nexus/catalog/migrations/V54__qrep_partition_ranges.sql b/nexus/catalog/migrations/V54__qrep_partition_ranges.sql
new file mode 100644
index 000000000..8802ccb17
--- /dev/null
+++ b/nexus/catalog/migrations/V54__qrep_partition_ranges.sql
@@ -0,0 +1,15 @@
+-- Stores QRep partition ranges offloaded to the catalog instead of being passed
+-- through Temporal. Today this is used only for sensitive partition ranges.
+-- The table name and columns are deliberately generic so a future change can
+-- offload all partition ranges to catalog to stay under Temporal's payload limit.
+CREATE TABLE IF NOT EXISTS metadata_qrep_partition_ranges (
+    parent_mirror_name TEXT NOT NULL,
+    run_uuid TEXT NOT NULL,
+    partition_uuid TEXT NOT NULL,
+    enc_key_id TEXT NOT NULL,
+    range_payload BYTEA NOT NULL,
+    PRIMARY KEY (run_uuid, partition_uuid)
+);
+
+CREATE INDEX IF NOT EXISTS idx_metadata_qrep_partition_ranges_parent_mirror_name
+    ON metadata_qrep_partition_ranges (parent_mirror_name);

From b0182322e4fc8397cdfd1f51c6c8c465050b248b Mon Sep 17 00:00:00 2001
From: Joy Gao <17896160+jgao54@users.noreply.github.com>
Date: Thu, 25 Jun 2026 22:26:45 -1000
Subject: [PATCH 2/4] bug fix + batch insert

---
 flow/connectors/external_metadata/store.go | 20 +++++++++++---------
 1 file changed, 11 insertions(+), 9 deletions(-)

diff --git a/flow/connectors/external_metadata/store.go b/flow/connectors/external_metadata/store.go
index 77ce24c85..efdf66b54 100644
--- a/flow/connectors/external_metadata/store.go
+++ b/flow/connectors/external_metadata/store.go
@@ -320,6 +320,7 @@ func OffloadSensitivePartitionRanges(
 		return fmt.Errorf("failed to clear existing partition ranges: %w", err)
 	}
 
+	insertRows := make([][]any, 0, len(partitions))
 	for _, partition := range partitions {
 		payload, err := proto.Marshal(partition.Range)
 		if err != nil {
@@ -329,20 +330,21 @@ func OffloadSensitivePartitionRanges(
 		if err != nil {
 			return fmt.Errorf("failed to encrypt partition range: %w", err)
 		}
-
-		if _, err := tx.Exec(ctx,
-			`INSERT INTO `+qrepPartitionRangesTableName+
-				`(parent_mirror_name,run_uuid,partition_uuid,enc_key_id,range_payload) VALUES($1,$2,$3,$4,$5)`,
-			parentMirrorName, runUUID, partition.PartitionId, key.ID, encrypted,
-		); err != nil {
-			return fmt.Errorf("failed to persist encrypted partition range: %w", err)
-		}
+		insertRows = append(insertRows, []any{parentMirrorName, runUUID, partition.PartitionId, key.ID, encrypted})
 
 		partition.Range = &protos.PartitionRange{
 			Range: &protos.PartitionRange_StringRange{StringRange: &protos.StringPartitionRange{}},
 		}
 	}
 
+	if _, err := tx.CopyFrom(ctx,
+		pgx.Identifier{qrepPartitionRangesTableName},
+		[]string{"parent_mirror_name", "run_uuid", "partition_uuid", "enc_key_id", "range_payload"},
+		pgx.CopyFromRows(insertRows),
+	); err != nil {
+		return fmt.Errorf("failed to persist encrypted partition ranges: %w", err)
+	}
+
 	if err := tx.Commit(ctx); err != nil {
 		return fmt.Errorf("failed to commit partition ranges: %w", err)
 	}
@@ -418,7 +420,7 @@ func RestoreSensitivePartitionRanges(
 func partitionsContainSensitiveRanges(partitions []*protos.QRepPartition) bool {
 	for _, partition := range partitions {
 		if partition == nil || partition.Range == nil {
-			return false
+			continue
 		}
 		// String ranges may contain PII data like email, so deem it as sensitive.
 		if _, ok := partition.Range.Range.(*protos.PartitionRange_StringRange); ok {

From 9bab62a02c7bf9e9fa315c4be21ae3d735149426 Mon Sep 17 00:00:00 2001
From: Joy Gao <17896160+jgao54@users.noreply.github.com>
Date: Mon, 29 Jun 2026 17:38:58 -1000
Subject: [PATCH 3/4] review feedbacks: - decouple offload and restore (so
 restore derive based on range_offloaded flag) - always encrypt non-nil ranges
 regardless of type - update schema to include inserted_at timestamp - make
 sure metadata_qrep_partitions also does not store serialized partition in
 plaintext - add dynconf PEERDB_OFFLOAD_PARTITION_RANGES to allow disabling
 offload/encryption behind a FF - update tests

---
 flow/activities/flowable.go                   |  17 ++-
 flow/connectors/external_metadata/store.go    |  66 +++++-----
 .../external_metadata/store_test.go           | 124 +++++++++++-------
 .../connectors/utils/monitoring/monitoring.go |   8 +-
 flow/e2e/clickhouse_mysql_test.go             | 107 +--------------
 flow/e2e/clickhouse_test.go                   | 109 +++++++++++++++
 flow/internal/dynamicconf.go                  |  12 ++
 .../V54__qrep_offloaded_partition_ranges.sql  |  12 ++
 .../migrations/V54__qrep_partition_ranges.sql |  15 ---
 protos/flow.proto                             |   2 +
 10 files changed, 266 insertions(+), 206 deletions(-)
 create mode 100644 nexus/catalog/migrations/V54__qrep_offloaded_partition_ranges.sql
 delete mode 100644 nexus/catalog/migrations/V54__qrep_partition_ranges.sql

diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
index 177932fb9..0a24f3d30 100644
--- a/flow/activities/flowable.go
+++ b/flow/activities/flowable.go
@@ -596,11 +596,16 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context,
 		return nil, a.Alerter.LogFlowError(ctx, config.FlowJobName, shared.WrapError("failed to get partitions from source", err))
 	}
 	if len(partitions) > 0 {
-		// run before monitoring.InitializeQRepRun to avoid logging sensitive fields to peerdb_stats.qrep_partitions as well
-		if err := connmetadata.OffloadSensitivePartitionRanges(
-			ctx, a.CatalogPool, config.ParentMirrorName, runUUID, partitions,
-		); err != nil {
-			return nil, fmt.Errorf("failed to offload partition ranges: %w", err)
+		shouldOffload, err := internal.PeerDBOffloadPartitionRanges(ctx, config.Env)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read offload partition ranges setting: %w", err)
+		}
+		if shouldOffload && config.InitialCopyOnly {
+			if err := connmetadata.OffloadPartitionRanges(
+				ctx, a.CatalogPool, config.ParentMirrorName, runUUID, partitions,
+			); err != nil {
+				return nil, fmt.Errorf("failed to offload partition ranges: %w", err)
+			}
 		}
 		if err := monitoring.InitializeQRepRun(
 			ctx,
@@ -669,7 +674,7 @@ func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
 		return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
 	}
 
-	if err := connmetadata.RestoreSensitivePartitionRanges(ctx, a.CatalogPool, runUUID, partitions.Partitions); err != nil {
+	if err := connmetadata.RestoreOffloadedPartitionRanges(ctx, a.CatalogPool, runUUID, partitions.Partitions); err != nil {
 		return fmt.Errorf("failed to rehydrate partition ranges: %w", err)
 	}
 
diff --git a/flow/connectors/external_metadata/store.go b/flow/connectors/external_metadata/store.go
index efdf66b54..41c82f2df 100644
--- a/flow/connectors/external_metadata/store.go
+++ b/flow/connectors/external_metadata/store.go
@@ -6,6 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"maps"
+	"slices"
 	"strconv"
 	"time"
 
@@ -27,7 +29,7 @@ import (
 const (
 	lastSyncStateTableName       = "metadata_last_sync_state"
 	qrepTableName                = "metadata_qrep_partitions"
-	qrepPartitionRangesTableName = "metadata_qrep_partition_ranges"
+	qrepPartitionRangesTableName = "metadata_qrep_offloaded_partition_ranges"
 )
 
 type PostgresMetadata struct {
@@ -262,6 +264,15 @@ func (p *PostgresMetadata) FinishQRepPartition(
 	jobName string,
 	startTime time.Time,
 ) error {
+	if partition.RangeOffloaded {
+		// The range was offloaded but has since been rehydrated, so drop it before persisting
+		// to metadata_qrep_partitions.
+		// Clone rather than mutate in place because this partition is shared with the pull
+		// goroutine, cloning keeps us from relying on implicit ordering to avoid a write race.
+		// TODO: drop the sync_partition column entirely; nothing reads it back.
+		partition = proto.Clone(partition).(*protos.QRepPartition)
+		partition.Range = nil
+	}
 	pbytes, err := protojson.Marshal(partition)
 	if err != nil {
 		return fmt.Errorf("failed to marshal partition to json: %w", err)
@@ -288,20 +299,16 @@ func (p *PostgresMetadata) IsQRepPartitionSynced(ctx context.Context, req *proto
 	return exists, nil
 }
 
-// OffloadSensitivePartitionRanges encrypts sensitive partition ranges, persists them to
-// the catalog, and replaces each partition's range in-place with an empty range.
-// Used in conjunction with RestoreSensitivePartitionRanges.
-func OffloadSensitivePartitionRanges(
+// OffloadPartitionRanges encrypts partition ranges and persists them to the catalog,
+// then clears each partition's range in-place and sets RangeOffloaded so the range can
+// be rehydrated later. Used in conjunction with RestoreOffloadedPartitionRanges.
+func OffloadPartitionRanges(
 	ctx context.Context,
 	pool shared.CatalogPool,
 	parentMirrorName string,
 	runUUID string,
 	partitions []*protos.QRepPartition,
 ) error {
-	if !partitionsContainSensitiveRanges(partitions) {
-		return nil
-	}
-
 	key, err := internal.PeerDBCurrentEncKey(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to load current encryption key: %w", err)
@@ -322,6 +329,9 @@ func OffloadSensitivePartitionRanges(
 
 	insertRows := make([][]any, 0, len(partitions))
 	for _, partition := range partitions {
+		if partition.Range == nil {
+			continue
+		}
 		payload, err := proto.Marshal(partition.Range)
 		if err != nil {
 			return fmt.Errorf("failed to marshal partition range: %w", err)
@@ -332,9 +342,8 @@ func OffloadSensitivePartitionRanges(
 		}
 		insertRows = append(insertRows, []any{parentMirrorName, runUUID, partition.PartitionId, key.ID, encrypted})
 
-		partition.Range = &protos.PartitionRange{
-			Range: &protos.PartitionRange_StringRange{StringRange: &protos.StringPartitionRange{}},
-		}
+		partition.Range = nil
+		partition.RangeOffloaded = true
 	}
 
 	if _, err := tx.CopyFrom(ctx,
@@ -351,29 +360,29 @@ func OffloadSensitivePartitionRanges(
 	return nil
 }
 
-// RestoreSensitivePartitionRanges hydrates sensitive partitions in-place with the decrypted
-// values fetched from the catalog. Used in conjunction with OffloadSensitivePartitionRanges
-func RestoreSensitivePartitionRanges(
+// RestoreOffloadedPartitionRanges hydrates offloaded partition ranges in-place with the decrypted
+// values fetched from the catalog. Used in conjunction with OffloadPartitionRanges.
+func RestoreOffloadedPartitionRanges(
 	ctx context.Context,
 	pool shared.CatalogPool,
 	runUUID string,
 	partitions []*protos.QRepPartition,
 ) error {
-	if !partitionsContainSensitiveRanges(partitions) {
-		return nil
-	}
-
-	partitionIDs := make([]string, 0, len(partitions))
 	partitionByIDs := make(map[string]*protos.QRepPartition, len(partitions))
 	for _, partition := range partitions {
-		partitionIDs = append(partitionIDs, partition.PartitionId)
+		if !partition.RangeOffloaded {
+			continue
+		}
 		partitionByIDs[partition.PartitionId] = partition
 	}
+	if len(partitionByIDs) == 0 {
+		return nil
+	}
 
 	rows, err := pool.Query(ctx,
 		`SELECT partition_uuid,enc_key_id,range_payload FROM `+qrepPartitionRangesTableName+` `+
 			`WHERE run_uuid=$1 AND partition_uuid=ANY($2)`,
-		runUUID, partitionIDs,
+		runUUID, slices.Collect(maps.Keys(partitionByIDs)),
 	)
 	if err != nil {
 		return fmt.Errorf("failed to query partition ranges: %w", err)
@@ -417,19 +426,6 @@ func RestoreSensitivePartitionRanges(
 	return nil
 }
 
-func partitionsContainSensitiveRanges(partitions []*protos.QRepPartition) bool {
-	for _, partition := range partitions {
-		if partition == nil || partition.Range == nil {
-			continue
-		}
-		// String ranges may contain PII data like email, so deem it as sensitive.
-		if _, ok := partition.Range.Range.(*protos.PartitionRange_StringRange); ok {
-			return true
-		}
-	}
-	return false
-}
-
 func (p *PostgresMetadata) SyncFlowCleanup(ctx context.Context, jobName string) error {
 	tx, err := p.pool.Begin(ctx)
 	if err != nil {
diff --git a/flow/connectors/external_metadata/store_test.go b/flow/connectors/external_metadata/store_test.go
index d8ef2258b..c97151847 100644
--- a/flow/connectors/external_metadata/store_test.go
+++ b/flow/connectors/external_metadata/store_test.go
@@ -2,10 +2,13 @@ package connmetadata
 
 import (
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
@@ -20,56 +23,87 @@ func TestOffloadRestoreSensitivePartitionRanges(t *testing.T) {
 	pool, err := internal.GetCatalogConnectionPoolFromEnv(ctx)
 	require.NoError(t, err)
 
-	parentMirrorName := "test_offload_restore_sensitive_partition_ranges"
-	runUUID := uuid.NewString()
-	t.Cleanup(func() {
-		_, _ = pool.Exec(ctx, `DELETE FROM `+qrepPartitionRangesTableName+` WHERE parent_mirror_name=$1`, parentMirrorName)
-	})
-
-	p1Start, p1End, p2Start, p2End := "bar", "foo", "hello", "world"
-	partitions := []*protos.QRepPartition{
+	for _, tc := range []struct {
+		name   string
+		ranges []*protos.PartitionRange
+	}{
 		{
-			PartitionId: uuid.NewString(),
-			Range: &protos.PartitionRange{Range: &protos.PartitionRange_StringRange{
-				StringRange: &protos.StringPartitionRange{Start: p1Start, End: p1End},
-			}},
+			name: "int",
+			ranges: []*protos.PartitionRange{
+				{Range: &protos.PartitionRange_IntRange{IntRange: &protos.IntPartitionRange{Start: 1, End: 100}}},
+				{Range: &protos.PartitionRange_IntRange{IntRange: &protos.IntPartitionRange{Start: 101, End: 200}}},
+			},
 		},
 		{
-			PartitionId: uuid.NewString(),
-			Range: &protos.PartitionRange{Range: &protos.PartitionRange_StringRange{
-				StringRange: &protos.StringPartitionRange{Start: p2Start, End: p2End, EndInclusive: true},
-			}},
+			name: "timestamp",
+			ranges: []*protos.PartitionRange{
+				{Range: &protos.PartitionRange_TimestampRange{TimestampRange: &protos.TimestampPartitionRange{
+					Start: timestamppb.New(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)),
+					End:   timestamppb.New(time.Date(2026, 6, 30, 23, 59, 59, 0, time.UTC)),
+				}}},
+				{Range: &protos.PartitionRange_TimestampRange{TimestampRange: &protos.TimestampPartitionRange{
+					Start: timestamppb.New(time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)),
+					End:   timestamppb.New(time.Date(2026, 12, 31, 23, 59, 59, 0, time.UTC)),
+				}}},
+			},
 		},
-	}
+		{
+			name: "string_uuid",
+			ranges: []*protos.PartitionRange{
+				{Range: &protos.PartitionRange_StringRange{
+					StringRange: &protos.StringPartitionRange{Start: uuid.NewString(), End: uuid.NewString()},
+				}},
+				{Range: &protos.PartitionRange_StringRange{
+					StringRange: &protos.StringPartitionRange{Start: uuid.NewString(), End: uuid.NewString(), EndInclusive: true},
+				}},
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			parentMirrorName := "test_offload_restore_partition_ranges_" + tc.name
+			runUUID := uuid.NewString()
+			t.Cleanup(func() {
+				_, _ = pool.Exec(ctx, `DELETE FROM `+qrepPartitionRangesTableName+` WHERE parent_mirror_name=$1`, parentMirrorName)
+			})
 
-	require.NoError(t, OffloadSensitivePartitionRanges(ctx, pool, parentMirrorName, runUUID, partitions))
-	for _, p := range partitions {
-		require.Empty(t, p.Range.GetStringRange().Start)
-		require.Empty(t, p.Range.GetStringRange().End)
-	}
+			partitions := make([]*protos.QRepPartition, len(tc.ranges))
+			expectedRanges := make([]*protos.PartitionRange, len(tc.ranges))
+			for i, r := range tc.ranges {
+				partitions[i] = &protos.QRepPartition{PartitionId: uuid.NewString(), Range: r}
+				expectedRanges[i] = proto.Clone(r).(*protos.PartitionRange)
+			}
 
-	rows, err := pool.Query(ctx,
-		`SELECT enc_key_id, range_payload FROM `+qrepPartitionRangesTableName+` WHERE parent_mirror_name=$1`,
-		parentMirrorName)
-	require.NoError(t, err)
-	results, err := pgx.CollectRows(rows, pgx.RowToStructByPos[struct {
-		KeyID   string
-		Payload []byte
-	}])
-	require.NoError(t, err)
-	require.Len(t, results, len(partitions))
-	for _, res := range results {
-		require.Equal(t, encKeyID, res.KeyID)
-		for _, plaintext := range []string{p1Start, p1End, p2Start, p2End} {
-			require.NotContains(t, string(res.Payload), plaintext)
-		}
-	}
+			require.NoError(t, OffloadPartitionRanges(ctx, pool, parentMirrorName, runUUID, partitions))
+			for _, p := range partitions {
+				require.Nil(t, p.Range)
+				require.True(t, p.RangeOffloaded)
+			}
 
-	require.NoError(t, RestoreSensitivePartitionRanges(ctx, pool, runUUID, partitions))
-	require.Equal(t, p1Start, partitions[0].Range.GetStringRange().Start)
-	require.Equal(t, p1End, partitions[0].Range.GetStringRange().End)
-	require.False(t, partitions[0].Range.GetStringRange().EndInclusive)
-	require.Equal(t, p2Start, partitions[1].Range.GetStringRange().Start)
-	require.Equal(t, p2End, partitions[1].Range.GetStringRange().End)
-	require.True(t, partitions[1].Range.GetStringRange().EndInclusive)
+			rows, err := pool.Query(ctx,
+				`SELECT enc_key_id, range_payload FROM `+qrepPartitionRangesTableName+` WHERE parent_mirror_name=$1`,
+				parentMirrorName)
+			require.NoError(t, err)
+			results, err := pgx.CollectRows(rows, pgx.RowToStructByPos[struct {
+				KeyID   string
+				Payload []byte
+			}])
+			require.NoError(t, err)
+			require.Len(t, results, len(partitions))
+			for _, res := range results {
+				require.Equal(t, encKeyID, res.KeyID)
+				require.NotEmpty(t, res.Payload)
+				for _, r := range expectedRanges {
+					unencrypted, err := proto.Marshal(r)
+					require.NoError(t, err)
+					require.NotContains(t, string(res.Payload), string(unencrypted))
+				}
+			}
+
+			require.NoError(t, RestoreOffloadedPartitionRanges(ctx, pool, runUUID, partitions))
+			for i, p := range partitions {
+				require.Truef(t, proto.Equal(expectedRanges[i], p.Range),
+					"partition %d range mismatch after restore: want %v, got %v", i, expectedRanges[i], p.Range)
+			}
+		})
+	}
 }
diff --git a/flow/connectors/utils/monitoring/monitoring.go b/flow/connectors/utils/monitoring/monitoring.go
index 98fa75e6a..fd096ce7a 100644
--- a/flow/connectors/utils/monitoring/monitoring.go
+++ b/flow/connectors/utils/monitoring/monitoring.go
@@ -332,7 +332,11 @@ func addPartitionToQRepRun(ctx context.Context, tx pgx.Tx, flowJobName string,
 	}
 
 	var rangeStart, rangeEnd *string
-	if partition.Range != nil {
+	if partition.RangeOffloaded {
+		internal.LoggerFromCtx(ctx).Info(
+			"omitting partition_start/partition_end from qrep_partitions because it's offloaded",
+			slog.String("partitionId", partition.PartitionId))
+	} else if partition.Range != nil {
 		switch x := partition.Range.Range.(type) {
 		case *protos.PartitionRange_IntRange:
 			rangeStart, rangeEnd = new(strconv.FormatInt(x.IntRange.Start, 10)), new(strconv.FormatInt(x.IntRange.End, 10))
@@ -369,7 +373,7 @@ func addPartitionToQRepRun(ctx context.Context, tx pgx.Tx, flowJobName string,
 		default:
 			return fmt.Errorf("unknown range type: %v", x)
 		}
-	} else if !partition.FullTablePartition {
+	} else if !partition.FullTablePartition && len(partition.ChildTableRanges) == 0 {
 		internal.LoggerFromCtx(ctx).Warn("[monitoring]: partition "+partition.PartitionId+" has nil range",
 			slog.String(string(shared.FlowNameKey), parentMirrorName))
 	}
diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go
index bb2403b41..7d7a45bc8 100644
--- a/flow/e2e/clickhouse_mysql_test.go
+++ b/flow/e2e/clickhouse_mysql_test.go
@@ -1760,8 +1760,7 @@ func (s ClickHouseSuite) Test_MySQL_Default_Partition_Key_Parallel_Snapshot() {
 	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,created_at")
 
 	partitionRows, err := s.catalog.Query(s.t.Context(),
-		`SELECT partition_start, partition_end, rows_in_partition
-		 FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
+		`SELECT rows_in_partition FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
 		flowConnConfig.FlowJobName)
 	require.NoError(s.t, err)
 	defer partitionRows.Close()
@@ -1769,11 +1768,8 @@ func (s ClickHouseSuite) Test_MySQL_Default_Partition_Key_Parallel_Snapshot() {
 	var partitionCount int32
 	var totalRows int64
 	for partitionRows.Next() {
-		var partitionStart, partitionEnd *string
 		var rowsInPartition int64
-		require.NoError(s.t, partitionRows.Scan(&partitionStart, &partitionEnd, &rowsInPartition))
-		require.NotNil(s.t, partitionStart)
-		require.NotNil(s.t, partitionEnd)
+		require.NoError(s.t, partitionRows.Scan(&rowsInPartition))
 		totalRows += rowsInPartition
 		partitionCount++
 	}
@@ -1890,8 +1886,7 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Arbitrary_FullTable() {
 	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,val")
 
 	partitionRows, err := s.catalog.Query(s.t.Context(),
-		`SELECT partition_start, partition_end, rows_in_partition
-		 FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
+		`SELECT rows_in_partition FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
 		flowConnConfig.FlowJobName)
 	require.NoError(s.t, err)
 	defer partitionRows.Close()
@@ -1899,11 +1894,8 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Arbitrary_FullTable() {
 	var partitionCount int32
 	var totalRows int64
 	for partitionRows.Next() {
-		var partitionStart, partitionEnd *string
 		var rowsInPartition int64
-		require.NoError(s.t, partitionRows.Scan(&partitionStart, &partitionEnd, &rowsInPartition))
-		require.Nil(s.t, partitionStart)
-		require.Nil(s.t, partitionEnd)
+		require.NoError(s.t, partitionRows.Scan(&rowsInPartition))
 		totalRows += rowsInPartition
 		partitionCount++
 	}
@@ -1914,94 +1906,3 @@ func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Arbitrary_FullTable() {
 	env.Cancel(s.t.Context())
 	RequireEnvCanceled(s.t, env)
 }
-
-func (s ClickHouseSuite) Test_MySQL_String_Partition_Key_Sensitive_Range_Handling() {
-	if _, ok := s.source.(*MySqlSource); !ok {
-		s.t.Skip("only applies to mysql")
-	}
-
-	srcTable := "test_string_pk_sensitive"
-	srcFullName := s.attachSchemaSuffix(srcTable)
-	dstTable := "test_string_pk_sensitive_dst"
-
-	const numRows = 8
-	const numPartitions = 4
-	require.NoError(s.t, s.source.Exec(s.t.Context(),
-		fmt.Sprintf("CREATE TABLE %s (id CHAR(36) PRIMARY KEY, val INT NOT NULL)", srcFullName)))
-	for i := 1; i <= numRows; i++ {
-		require.NoError(s.t, s.source.Exec(s.t.Context(),
-			fmt.Sprintf("INSERT INTO %s (id, val) VALUES ('%s', %d)", srcFullName, uuid.NewString(), i)))
-	}
-
-	tableMappings := TableMappings(s, srcTable, dstTable)
-	// a String column can't be used directly as a Distributed sharding key (ClickHouse
-	// requires an integer-typed sharding expression), so hash it for the cluster suite
-	for _, tm := range tableMappings {
-		tm.ShardingKey = "cityHash64(id)"
-	}
-	connectionGen := FlowConnectionGenerationConfig{
-		FlowJobName:   s.attachSuffix("string_pk_sensitive"),
-		TableMappings: tableMappings,
-		Destination:   s.Peer().Name,
-	}
-	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
-	flowConnConfig.DoInitialSnapshot = true
-	flowConnConfig.SnapshotMaxParallelWorkers = 4
-	flowConnConfig.SnapshotNumPartitionsOverride = numPartitions
-
-	tc := NewTemporalClient(s.t)
-	env := ExecutePeerflow(s.t, tc, flowConnConfig)
-	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
-
-	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,val")
-
-	// verify partition start/end are stored in metadata_qrep_partition_ranges
-	rangeRows, err := s.catalog.Query(s.t.Context(),
-		`SELECT enc_key_id, range_payload FROM metadata_qrep_partition_ranges WHERE parent_mirror_name = $1`,
-		flowConnConfig.FlowJobName)
-	require.NoError(s.t, err)
-	defer rangeRows.Close()
-	var offloadedCount int32
-	for rangeRows.Next() {
-		var encKeyID string
-		var rangePayload []byte
-		require.NoError(s.t, rangeRows.Scan(&encKeyID, &rangePayload))
-		// no encryption key is configured for CI so enc_key_id is empty
-		require.Empty(s.t, encKeyID)
-		require.NotEmpty(s.t, rangePayload)
-		offloadedCount++
-	}
-	require.NoError(s.t, rangeRows.Err())
-	require.EqualValues(s.t, numPartitions, offloadedCount)
-
-	// verify partition start/end are stripped from peerdb_stats.qrep_partitions
-	partitionRows, err := s.catalog.Query(s.t.Context(),
-		`SELECT partition_start, partition_end FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
-		flowConnConfig.FlowJobName)
-	require.NoError(s.t, err)
-	defer partitionRows.Close()
-
-	var partitionCount int32
-	for partitionRows.Next() {
-		var partitionStart, partitionEnd *string
-		require.NoError(s.t, partitionRows.Scan(&partitionStart, &partitionEnd))
-		require.Empty(s.t, partitionStart)
-		require.Empty(s.t, partitionEnd)
-		partitionCount++
-	}
-	require.NoError(s.t, partitionRows.Err())
-	require.EqualValues(s.t, numPartitions, partitionCount)
-
-	env.Cancel(s.t.Context())
-	RequireEnvCanceled(s.t, env)
-
-	// dropping the mirror should delete metadata_qrep_partition_ranges entries from the catalog for this mirror.
-	dropEnv := ExecuteDropFlow(s.t.Context(), tc, flowConnConfig, 0)
-	EnvWaitForFinished(s.t, dropEnv, 3*time.Minute)
-
-	var remainingRanges int64
-	require.NoError(s.t, s.catalog.QueryRow(s.t.Context(),
-		`SELECT COUNT(*) FROM metadata_qrep_partition_ranges WHERE parent_mirror_name = $1`,
-		flowConnConfig.FlowJobName).Scan(&remainingRanges))
-	require.Zero(s.t, remainingRanges)
-}
diff --git a/flow/e2e/clickhouse_test.go b/flow/e2e/clickhouse_test.go
index b01d7a7e4..87297de7c 100644
--- a/flow/e2e/clickhouse_test.go
+++ b/flow/e2e/clickhouse_test.go
@@ -19,6 +19,7 @@ import (
 	"github.com/shopspring/decimal"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	protojson "google.golang.org/protobuf/encoding/protojson"
 
 	connclickhouse "github.com/PeerDB-io/peerdb/flow/connectors/clickhouse"
 	connpostgres "github.com/PeerDB-io/peerdb/flow/connectors/postgres"
@@ -3241,6 +3242,9 @@ func (s ClickHouseSuite) Test_Partition_By_CTID_With_Num_Partitions_Override() {
 	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
 	flowConnConfig.DoInitialSnapshot = true
 	flowConnConfig.SnapshotNumPartitionsOverride = 3
+	// disable range offloading so the TID block ranges remain queryable from peerdb_stats.qrep_partitions;
+	// offloading itself is covered by Test_Offload_Partition_Ranges
+	flowConnConfig.Env = map[string]string{"PEERDB_OFFLOAD_PARTITION_RANGES": "false"}
 
 	tc := NewTemporalClient(s.t)
 	env := ExecutePeerflow(s.t, tc, flowConnConfig)
@@ -3671,3 +3675,108 @@ func (s ClickHouseSuite) Test_Composite_PKey() {
 	env.Cancel(s.t.Context())
 	RequireEnvCanceled(s.t, env)
 }
+
+func (s ClickHouseSuite) Test_Offload_Partition_Ranges() {
+	srcTable := "test_offload_partition_ranges"
+	srcFullName := s.attachSchemaSuffix(srcTable)
+	dstTable := "test_offload_partition_ranges_dst"
+
+	const numRows = 100
+	const numPartitions = 8
+	require.NoError(s.t, s.source.Exec(s.t.Context(),
+		fmt.Sprintf("CREATE TABLE %s (id BIGINT PRIMARY KEY, val INT NOT NULL)", srcFullName)))
+	for i := 1; i <= numRows; i++ {
+		require.NoError(s.t, s.source.Exec(s.t.Context(),
+			fmt.Sprintf("INSERT INTO %s (id, val) VALUES (%d, %d)", srcFullName, i, i)))
+	}
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName: s.attachSuffix("offload_partition_ranges"),
+		TableMappings: []*protos.TableMapping{{
+			SourceTableIdentifier:      srcFullName,
+			DestinationTableIdentifier: s.DestinationTable(dstTable),
+			PartitionKey:               "id",
+			ShardingKey:                "id",
+		}},
+		Destination: s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+	flowConnConfig.SnapshotMaxParallelWorkers = 4
+	flowConnConfig.SnapshotNumPartitionsOverride = numPartitions
+	flowConnConfig.Env = map[string]string{"PEERDB_OFFLOAD_PARTITION_RANGES": "true"}
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTable, dstTable, "id,val")
+
+	// the offloaded ranges must be persisted into metadata_qrep_offloaded_partition_ranges
+	rangeRows, err := s.catalog.Query(s.t.Context(),
+		`SELECT enc_key_id, range_payload FROM metadata_qrep_offloaded_partition_ranges WHERE parent_mirror_name = $1`,
+		flowConnConfig.FlowJobName)
+	require.NoError(s.t, err)
+	defer rangeRows.Close()
+	var offloadedNumPartitions int32
+	for rangeRows.Next() {
+		var encKeyID string
+		var rangePayload []byte
+		require.NoError(s.t, rangeRows.Scan(&encKeyID, &rangePayload))
+		// no encryption key is configured for CI so enc_key_id is empty
+		require.Empty(s.t, encKeyID)
+		require.NotEmpty(s.t, rangePayload)
+		offloadedNumPartitions++
+	}
+	require.NoError(s.t, rangeRows.Err())
+	require.EqualValues(s.t, numPartitions, offloadedNumPartitions)
+
+	// the offloaded ranges must not be persisted into metadata_qrep_partitions
+	res, err := s.catalog.Query(s.t.Context(),
+		`SELECT sync_partition FROM metadata_qrep_partitions WHERE job_name LIKE '%' || $1 || '%'`,
+		flowConnConfig.FlowJobName)
+	require.NoError(s.t, err)
+	defer res.Close()
+	var metaCount int
+	for res.Next() {
+		var syncPartition []byte
+		require.NoError(s.t, res.Scan(&syncPartition))
+		var partition protos.QRepPartition
+		require.NoError(s.t, protojson.Unmarshal(syncPartition, &partition))
+		require.Nil(s.t, partition.Range)
+		require.True(s.t, partition.RangeOffloaded)
+		metaCount++
+	}
+	require.NoError(s.t, res.Err())
+	require.NotZero(s.t, metaCount)
+
+	// the offloaded ranges must not be persisted into peerdb_stats.qrep_partitions
+	statsRows, err := s.catalog.Query(s.t.Context(),
+		`SELECT partition_start, partition_end FROM peerdb_stats.qrep_partitions WHERE parent_mirror_name = $1`,
+		flowConnConfig.FlowJobName)
+	require.NoError(s.t, err)
+	defer statsRows.Close()
+	var statsCount int
+	for statsRows.Next() {
+		var partitionStart, partitionEnd *string
+		require.NoError(s.t, statsRows.Scan(&partitionStart, &partitionEnd))
+		require.Nil(s.t, partitionStart)
+		require.Nil(s.t, partitionEnd)
+		statsCount++
+	}
+	require.NoError(s.t, statsRows.Err())
+	require.NotZero(s.t, statsCount)
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+
+	// dropping the mirror should delete metadata_qrep_offloaded_partition_ranges entries for this mirror
+	dropEnv := ExecuteDropFlow(s.t.Context(), tc, flowConnConfig, 0)
+	EnvWaitForFinished(s.t, dropEnv, 3*time.Minute)
+
+	var remainingRanges int64
+	require.NoError(s.t, s.catalog.QueryRow(s.t.Context(),
+		`SELECT COUNT(*) FROM metadata_qrep_offloaded_partition_ranges WHERE parent_mirror_name = $1`,
+		flowConnConfig.FlowJobName).Scan(&remainingRanges))
+	require.Zero(s.t, remainingRanges)
+}
diff --git a/flow/internal/dynamicconf.go b/flow/internal/dynamicconf.go
index cbe54ce57..bff9a296f 100644
--- a/flow/internal/dynamicconf.go
+++ b/flow/internal/dynamicconf.go
@@ -476,6 +476,14 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_NEW_MIRROR,
 		TargetForSetting: protos.DynconfTarget_ALL,
 	},
+	{
+		Name:             "PEERDB_OFFLOAD_PARTITION_RANGES",
+		Description:      "Encrypt QRep partition ranges and offload them to the catalog instead of passing them through Temporal",
+		DefaultValue:     "true",
+		ValueType:        protos.DynconfValueType_BOOL,
+		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_NEW_MIRROR,
+		TargetForSetting: protos.DynconfTarget_ALL,
+	},
 	{
 		Name: "PEERDB_PG_AUTOMATED_SCHEMA_DUMP",
 		Description: "For PG-to-PG mirrors, run pg_dump --schema-only from source into psql on destination " +
@@ -870,6 +878,10 @@ func PeerDBMySQLDefaultPartitionKeyEnabled(ctx context.Context, env map[string]s
 	return dynamicConfBool(ctx, env, "PEERDB_MYSQL_DEFAULT_PARTITION_KEY_ENABLED")
 }
 
+func PeerDBOffloadPartitionRanges(ctx context.Context, env map[string]string) (bool, error) {
+	return dynamicConfBool(ctx, env, "PEERDB_OFFLOAD_PARTITION_RANGES")
+}
+
 func PeerDBPGAutomatedSchemaDump(ctx context.Context, env map[string]string) (bool, error) {
 	return dynamicConfBool(ctx, env, "PEERDB_PG_AUTOMATED_SCHEMA_DUMP")
 }
diff --git a/nexus/catalog/migrations/V54__qrep_offloaded_partition_ranges.sql b/nexus/catalog/migrations/V54__qrep_offloaded_partition_ranges.sql
new file mode 100644
index 000000000..bd038d4bb
--- /dev/null
+++ b/nexus/catalog/migrations/V54__qrep_offloaded_partition_ranges.sql
@@ -0,0 +1,12 @@
+CREATE TABLE IF NOT EXISTS metadata_qrep_offloaded_partition_ranges (
+    parent_mirror_name TEXT NOT NULL,
+    run_uuid TEXT NOT NULL,
+    partition_uuid TEXT NOT NULL,
+    enc_key_id TEXT NOT NULL,
+    range_payload BYTEA NOT NULL,
+    inserted_at TIMESTAMP NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (run_uuid, partition_uuid)
+);
+
+CREATE INDEX IF NOT EXISTS idx_metadata_qrep_offloaded_partition_ranges_parent_mirror_name
+    ON metadata_qrep_offloaded_partition_ranges (parent_mirror_name);
diff --git a/nexus/catalog/migrations/V54__qrep_partition_ranges.sql b/nexus/catalog/migrations/V54__qrep_partition_ranges.sql
deleted file mode 100644
index 8802ccb17..000000000
--- a/nexus/catalog/migrations/V54__qrep_partition_ranges.sql
+++ /dev/null
@@ -1,15 +0,0 @@
--- Stores QRep partition ranges offloaded to the catalog instead of being passed
--- through Temporal. Today this is used only for sensitive partition ranges.
--- The table name and columns are deliberately generic so a future change can
--- offload all partition ranges to catalog to stay under Temporal's payload limit.
-CREATE TABLE IF NOT EXISTS metadata_qrep_partition_ranges (
-    parent_mirror_name TEXT NOT NULL,
-    run_uuid TEXT NOT NULL,
-    partition_uuid TEXT NOT NULL,
-    enc_key_id TEXT NOT NULL,
-    range_payload BYTEA NOT NULL,
-    PRIMARY KEY (run_uuid, partition_uuid)
-);
-
-CREATE INDEX IF NOT EXISTS idx_metadata_qrep_partition_ranges_parent_mirror_name
-    ON metadata_qrep_partition_ranges (parent_mirror_name);
diff --git a/protos/flow.proto b/protos/flow.proto
index af3d5f155..fcf94bfdb 100644
--- a/protos/flow.proto
+++ b/protos/flow.proto
@@ -453,6 +453,8 @@ message QRepPartition {
   // child table and block range to query instead of the parent watermark table.
   // When populated, the top-level `range` field is unused.
   repeated ChildTableRange child_table_ranges = 5;
+  // When true, PartitionRange get offloaded and encrypted to the catalog to avoid passing through Temporal
+  bool range_offloaded = 6;
 }
 
 message QRepPartitionBatch {

From 38ff3aa9db70cf2ba169685b4b14f8c4bf6b51f3 Mon Sep 17 00:00:00 2001
From: Joy Gao <17896160+jgao54@users.noreply.github.com>
Date: Wed, 1 Jul 2026 17:44:52 -1000
Subject: [PATCH 4/4] review

---
 flow/connectors/external_metadata/store.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flow/connectors/external_metadata/store.go b/flow/connectors/external_metadata/store.go
index 41c82f2df..b13ff6d1e 100644
--- a/flow/connectors/external_metadata/store.go
+++ b/flow/connectors/external_metadata/store.go
@@ -270,7 +270,7 @@ func (p *PostgresMetadata) FinishQRepPartition(
 		// Clone rather than mutate in place because this partition is shared with the pull
 		// goroutine, cloning keeps us from relying on implicit ordering to avoid a write race.
 		// TODO: drop the sync_partition column entirely; nothing reads it back.
-		partition = proto.Clone(partition).(*protos.QRepPartition)
+		partition = proto.CloneOf(partition)
 		partition.Range = nil
 	}
 	pbytes, err := protojson.Marshal(partition)