merge

Author: dtunikov

.github/workflows/flow.yml

@@ -356,9 +356,9 @@ jobs:
         id: ch-version
         run: |
           if [ "${{ matrix.db-version.ch }}" = "lts" ]; then
-            echo "ch_version=v25.8.11.66-lts" >> $GITHUB_OUTPUT
+            echo "ch_version=v25.8.28.1-lts" >> $GITHUB_OUTPUT
           elif [ "${{ matrix.db-version.ch }}" = "stable" ]; then
-            echo "ch_version=v25.12.4.35-stable" >> $GITHUB_OUTPUT
+            echo "ch_version=v26.3.16.16-lts" >> $GITHUB_OUTPUT
           elif [ "${{ matrix.db-version.ch }}" = "latest" ]; then
             # note: latest tag does not always reflect the latest version (could be an update on an lts),
             # but that is okay as we are only using it to invalidate the cache.

.gitignore

@@ -26,3 +26,7 @@ ancillary.env
 # Ephemeral volumes for docker compose
 volumes/ch_*
 
+# Renovate debug files
+final-updated-packages.txt
+renovate.out
+update-proposals.json

flow/alerting/classifier.go

@@ -146,6 +146,12 @@ var (
 	ErrorNotifyBinlogEventExceededMaxAllowedPacket = ErrorClass{
 		Class: "NOTIFY_BINLOG_EVENT_EXCEEDED_MAX_ALLOWED_PACKET", action: NotifyUser,
 	}
+	ErrorNotifyBinlogPartialRowEventUnsupported = ErrorClass{
+		Class: "NOTIFY_BINLOG_PARTIAL_ROW_EVENT_UNSUPPORTED", action: NotifyUser,
+	}
+	ErrorNotifyBinlogPartialJsonUnsupported = ErrorClass{
+		Class: "NOTIFY_BINLOG_PARTIAL_JSON_UNSUPPORTED", action: NotifyUser,
+	}
 	ErrorNotifyBinlogRowMetadataInvalid = ErrorClass{
 		Class: "NOTIFY_BINLOG_ROW_METADATA_INVALID", action: NotifyUser,
 	}
@@ -1120,6 +1126,16 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if partialJsonUnsupportedError, ok := errors.AsType[*exceptions.MySQLUnsupportedBinlogRowValueOptionsError](err); ok {
+		return ErrorNotifyBinlogPartialJsonUnsupported, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "UNSUPPORTED_PARTIAL_JSON",
+			AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+				ErrorAttributeKeyTable: fmt.Sprintf("%s.%s", partialJsonUnsupportedError.Schema, partialJsonUnsupportedError.Table),
+			},
+		}
+	}
+
 	if unsupportedDDLError, ok := errors.AsType[*exceptions.MySQLUnsupportedDDLError](err); ok {
 		return ErrorNotifyBinlogRowMetadataInvalid, ErrorInfo{
 			Source: ErrorSourceMySQL,
@@ -1162,6 +1178,16 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if partialRowEventError, ok := errors.AsType[*exceptions.MySQLUnsupportedPartialRowEventError](err); ok {
+		return ErrorNotifyBinlogPartialRowEventUnsupported, ErrorInfo{
+			Source: ErrorSourceMySQL,
+			Code:   "UNSUPPORTED_PARTIAL_ROW_EVENT",
+			AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+				ErrorAttributeKeyTable: fmt.Sprintf("%s.%s", partialRowEventError.Schema, partialRowEventError.Table),
+			},
+		}
+	}
+
 	if mysqlGeometryParseError, ok := errors.AsType[*exceptions.MySQLGeometryParseError](err); ok &&
 		strings.Contains(mysqlGeometryParseError.Error(), mysqlGeometryLinearRingNotClosedError) {
 		return ErrorUnsupportedDatatype, ErrorInfo{

flow/alerting/classifier_test.go

@@ -1255,3 +1255,33 @@ func TestMySQLBinlogIncidentErrorShouldBeNotifyBinlogInvalid(t *testing.T) {
 		Code:   "BINLOG_INCIDENT",
 	}, errInfo)
 }
+
+func TestMySQLUnsupportedPartialRowEventShouldBeNotifyPartialRowEventUnsupported(t *testing.T) {
+	t.Parallel()
+
+	err := exceptions.NewMySQLUnsupportedPartialRowEventError(172, "e2e_test", "partial_rows")
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err))
+	assert.Equal(t, ErrorNotifyBinlogPartialRowEventUnsupported, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "UNSUPPORTED_PARTIAL_ROW_EVENT",
+		AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+			ErrorAttributeKeyTable: "e2e_test.partial_rows",
+		},
+	}, errInfo)
+}
+
+func TestMySQLUnsupportedBinlogRowValueOptionsErrorShouldBeNotifyBinlogPartialJsonUnsupported(t *testing.T) {
+	t.Parallel()
+
+	err := exceptions.NewMySQLUnsupportedBinlogRowValueOptionsError("mydb", "mytable")
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err))
+	assert.Equal(t, ErrorNotifyBinlogPartialJsonUnsupported, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "UNSUPPORTED_PARTIAL_JSON",
+		AdditionalAttributes: map[AdditionalErrorAttributeKey]string{
+			ErrorAttributeKeyTable: "mydb.mytable",
+		},
+	}, errInfo)
+}

flow/connectors/clickhouse/validate.go

@@ -120,6 +120,15 @@ func (c *ClickHouseConnector) ValidateMirrorDestination(
 		); err != nil {
 			return err
 		}
+		if err := chvalidate.ValidateClusterShardingKey(
+			c.Config.Cluster,
+			tableMapping.ShardingKey,
+			tableMapping.SourceTableIdentifier,
+			len(processedSchema.PrimaryKeyColumns) > 0,
+			sortingKeys,
+		); err != nil {
+			return err
+		}
 
 		// if destination table does not exist, we're good
 		if _, ok := chTableColumnsMapping[dstTableName]; !ok {

flow/connectors/mongo/qrep.go

@@ -226,19 +226,18 @@ func toRangeFilter(watermarkColumn string, partitionRange *protos.PartitionRange
 				bson.E{Key: "$lte", Value: endObjectID},
 			}},
 		}, nil
-	case *protos.PartitionRange_IntRange:
-		// Numeric _id: inclusive range. PartitionHelper builds non-overlapping
-		// [start, end] integer ranges. Mongo compares int32/int64 numerically, so
-		// int64 bounds match int32-typed _ids.
+	case *protos.PartitionRange_NumericRange:
+		endOp := "$lt"
+		if r.NumericRange.EndInclusive {
+			endOp = "$lte"
+		}
 		return bson.D{
 			bson.E{Key: watermarkColumn, Value: bson.D{
-				bson.E{Key: "$gte", Value: r.IntRange.Start},
-				bson.E{Key: "$lte", Value: r.IntRange.End},
+				bson.E{Key: "$gte", Value: r.NumericRange.Start},
+				bson.E{Key: endOp, Value: r.NumericRange.End},
 			}},
 		}, nil
 	case *protos.PartitionRange_StringRange:
-		// String _id: half-open [start, end) for all but the last partition;
-		// the last partition is closed [start, end] (EndInclusive == true).
 		endOp := "$lt"
 		if r.StringRange.EndInclusive {
 			endOp = "$lte"

flow/connectors/mongo/qrep_partition.go

@@ -27,12 +27,10 @@ const (
 // buildPartitions reads the collection's min and max _id (cheap, leverages the
 // default _id index) and dispatches to a type-specific partitioner:
 //   - ObjectID  -> uniform division of the 12-byte ObjectID keyspace
-//   - Int32/64  -> uniform division of the numeric range (reuses PartitionHelper)
+//   - Int32/64  -> uniform division of the numeric range
 //   - String    -> sampling-based quantile boundaries
 //
-// Mixed-type _id (min and max differ in type), Double, or any other type fall back
-// to a full-table partition. The change stream resume token is captured before this
-// runs, so any writes during partitioning are replayed by CDC.
+// Mixed-type _id, Double, or any other type fall back to a full-table partition.
 func (c *MongoConnector) buildPartitions(
 	ctx context.Context,
 	collection *mongo.Collection,
@@ -127,15 +125,10 @@ func (c *MongoConnector) objectIDPartitions(
 	return partitions, nil
 }
 
-// numericPartitions divides a numeric (int32/int64) _id range uniformly, reusing
-// the shared PartitionHelper which builds non-overlapping IntPartitionRange
-// partitions with tested +1 boundary semantics covering [min, max] inclusive.
-//
-// Limitation: IntPartitionRange uses inclusive-inclusive [start, end] boundaries,
-// so documents with double _id values that fall between two adjacent integer
-// boundaries (e.g. _id=2.5 between [1,2] and [3,4]) would be missed. This is an
-// edge case for collections with mixed int/double _ids; a follow-up will introduce
-// a NumericPartitionRange with half-open [start, end) semantics to address it.
+// numericPartitions divides a numeric (int32/int64) _id range uniformly. MongoDB
+// sorts all numeric types by value, so integer min/max _ids cannot rule out
+// fractional (double/decimal) _ids in-between. Therefore we use [start, end) to
+// ensure contiguous boundaries.
 func (c *MongoConnector) numericPartitions(
 	minVal int64,
 	maxVal int64,
@@ -146,16 +139,13 @@ func (c *MongoConnector) numericPartitions(
 		return utils.FullTablePartition(), nil
 	}
 
-	c.logger.Info("[mongo] using numeric _id partitioning",
-		slog.Int64("minID", minVal),
-		slog.Int64("maxID", maxVal),
-		slog.Int64("numPartitions", numPartitions))
-
-	helper := utils.NewPartitionHelper(c.logger)
-	if err := helper.AddPartitionsWithRange(minVal, maxVal, numPartitions); err != nil {
-		return nil, fmt.Errorf("failed to build numeric partitions: %w", err)
+	ranges := utils.ComputeRanges(minVal, maxVal, numPartitions)
+	c.logger.Info("[mongo] using numeric _id partitioning", slog.Int("numPartitions", len(ranges)))
+	partitions := make([]*protos.QRepPartition, 0, len(ranges))
+	for i, rng := range ranges {
+		partitions = append(partitions, utils.CreateNumericPartition(rng[0], rng[1], i == len(ranges)-1))
 	}
-	return helper.GetPartitions(), nil
+	return partitions, nil
 }
 
 // stringPartitions builds partitions for a string _id collection. Because string
@@ -187,26 +177,12 @@ func (c *MongoConnector) stringPartitions(
 	}
 
 	c.logger.Info("[mongo] using sampled string _id partitioning",
-		slog.String("minID", minVal),
-		slog.String("maxID", maxVal),
 		slog.Int("numPartitions", len(ranges)),
 		slog.Int("sampleCount", len(samples)))
 
 	partitions := make([]*protos.QRepPartition, 0, len(ranges))
 	for i, rng := range ranges {
-		partitions = append(partitions, &protos.QRepPartition{
-			PartitionId: uuid.NewString(),
-			Range: &protos.PartitionRange{
-				Range: &protos.PartitionRange_StringRange{
-					StringRange: &protos.StringPartitionRange{
-						Start:        rng[0],
-						End:          rng[1],
-						EndInclusive: i == len(ranges)-1,
-					},
-				},
-			},
-			FullTablePartition: false,
-		})
+		partitions = append(partitions, utils.CreateStringPartition(rng[0], rng[1], i == len(ranges)-1))
 	}
 	return partitions, nil
 }
@@ -229,8 +205,8 @@ func (c *MongoConnector) sampleStringIDs(
 		{Key: "aggregate", Value: collection.Name()},
 		{Key: "pipeline", Value: bson.A{
 			bson.D{{Key: "$sample", Value: bson.D{{Key: "size", Value: int32(sampleSize)}}}},
-			bson.D{{Key: "$sort", Value: bson.D{{Key: DefaultDocumentKeyColumnName, Value: 1}}}},
 			bson.D{{Key: "$project", Value: bson.D{{Key: DefaultDocumentKeyColumnName, Value: 1}}}},
+			bson.D{{Key: "$sort", Value: bson.D{{Key: DefaultDocumentKeyColumnName, Value: 1}}}},
 		}},
 		{Key: "cursor", Value: bson.D{}},
 	}

flow/connectors/mongo/qrep_partition_test.go

@@ -6,6 +6,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.mongodb.org/mongo-driver/v2/bson"
 
+	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 )
 
@@ -117,32 +118,31 @@ func TestComputeStringBoundaries(t *testing.T) {
 	})
 }
 
-func intPartition(start, end int64) *protos.PartitionRange {
-	return &protos.PartitionRange{Range: &protos.PartitionRange_IntRange{
-		IntRange: &protos.IntPartitionRange{Start: start, End: end},
-	}}
-}
-
-func stringPartition(start, end string, endInclusive bool) *protos.PartitionRange {
-	return &protos.PartitionRange{Range: &protos.PartitionRange_StringRange{
-		StringRange: &protos.StringPartitionRange{Start: start, End: end, EndInclusive: endInclusive},
-	}}
-}
-
 func TestToRangeFilter(t *testing.T) {
-	t.Run("int range is inclusive on both ends", func(t *testing.T) {
-		filter, err := toRangeFilter(DefaultDocumentKeyColumnName, intPartition(10, 20))
+	t.Run("numeric range is half-open", func(t *testing.T) {
+		filter, err := toRangeFilter(DefaultDocumentKeyColumnName, utils.CreateNumericPartition(10, 20, false).Range)
 		require.NoError(t, err)
 		require.Equal(t, bson.D{
 			bson.E{Key: DefaultDocumentKeyColumnName, Value: bson.D{
 				bson.E{Key: "$gte", Value: int64(10)},
-				bson.E{Key: "$lte", Value: int64(20)},
+				bson.E{Key: "$lt", Value: int64(20)},
+			}},
+		}, filter)
+	})
+
+	t.Run("last numeric range is closed", func(t *testing.T) {
+		filter, err := toRangeFilter(DefaultDocumentKeyColumnName, utils.CreateNumericPartition(20, 30, true).Range)
+		require.NoError(t, err)
+		require.Equal(t, bson.D{
+			bson.E{Key: DefaultDocumentKeyColumnName, Value: bson.D{
+				bson.E{Key: "$gte", Value: int64(20)},
+				bson.E{Key: "$lte", Value: int64(30)},
 			}},
 		}, filter)
 	})
 
 	t.Run("string range is half-open", func(t *testing.T) {
-		filter, err := toRangeFilter(DefaultDocumentKeyColumnName, stringPartition("com.a", "com.m", false))
+		filter, err := toRangeFilter(DefaultDocumentKeyColumnName, utils.CreateStringPartition("com.a", "com.m", false).Range)
 		require.NoError(t, err)
 		require.Equal(t, bson.D{
 			bson.E{Key: DefaultDocumentKeyColumnName, Value: bson.D{
@@ -153,7 +153,7 @@ func TestToRangeFilter(t *testing.T) {
 	})
 
 	t.Run("last string range is closed", func(t *testing.T) {
-		filter, err := toRangeFilter(DefaultDocumentKeyColumnName, stringPartition("com.m", "org.z", true))
+		filter, err := toRangeFilter(DefaultDocumentKeyColumnName, utils.CreateStringPartition("com.m", "org.z", true).Range)
 		require.NoError(t, err)
 		require.Equal(t, bson.D{
 			bson.E{Key: DefaultDocumentKeyColumnName, Value: bson.D{