Merge branch 'pr/postgres-migrations' of https://github.com/yogesh1801/peerdb into pr/postgres-migrations

Author: yogesh1801

flow/activities/flowable.go

@@ -319,7 +319,7 @@ func (a *FlowableActivity) SyncFlow(
 		return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
 	}
 
-	if err := srcConn.SetupReplConn(ctx); err != nil {
+	if err := srcConn.SetupReplConn(ctx, config.Env); err != nil {
 		srcClose(ctx)
 		return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
 	}
@@ -612,7 +612,7 @@ func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
 		}
 
 		return func(partition *protos.QRepPartition) error {
-			stream := model.NewQRecordStream(shared.FetchAndChannelSize)
+			stream := model.NewQRecordStream(shared.QRepChannelSize)
 			outstream := stream
 
 			if luaScript != nil {
@@ -641,7 +641,7 @@ func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
 		}
 
 		return func(partition *protos.QRepPartition) error {
-			stream := model.NewQObjectStream(shared.FetchAndChannelSize)
+			stream := model.NewQObjectStream(shared.QRepChannelSize)
 
 			return replicateQRepPartition(ctx, a, srcConn, destConn, dstPeer.Type, config, partition, runUUID, stream, stream,
 				connectors.QRepPullObjectsConnector.PullQRepObjects,
@@ -1534,7 +1534,7 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
 
 	switch config.System {
 	case protos.TypeSystem_Q:
-		stream := model.NewQRecordStream(shared.FetchAndChannelSize)
+		stream := model.NewQRecordStream(shared.QRepChannelSize)
 		return replicateXminPartition(ctx, a, config, partition, runUUID,
 			stream, stream,
 			(*connpostgres.PostgresConnector).PullXminRecordStream,

flow/activities/flowable_core.go

@@ -234,7 +234,9 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		defer dstClose(ctx)
 
 		syncState.Store(shared.Ptr("updating schema"))
-		if err := dstConn.ReplayTableSchemaDeltas(ctx, config.Env, flowName, options.TableMappings, recordBatchSync.SchemaDeltas); err != nil {
+		if err := dstConn.ReplayTableSchemaDeltas(
+			ctx, config.Env, flowName, options.TableMappings, recordBatchSync.SchemaDeltas, config.Version,
+		); err != nil {
 			return nil, fmt.Errorf("failed to sync schema: %w", err)
 		}
 

flow/alerting/classifier.go

@@ -497,6 +497,11 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 				return ErrorNotifyPostgresSlotMemalloc, pgErrorInfo
 			}
 
+			// Usually a single occurrence then reconnect immediately helps
+			if strings.Contains(pgErr.Message, "pfree called with invalid pointer") {
+				return ErrorRetryRecoverable, pgErrorInfo
+			}
+
 			// Fall through for other internal errors
 			return ErrorOther, pgErrorInfo
 
@@ -608,8 +613,10 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 			return ErrorOther, myErrorInfo
 		case 1146: // ER_NO_SUCH_TABLE
 			return ErrorNotifySourceTableMissing, myErrorInfo
-		case 1943:
+		case 1943: // ER_DUPLICATE_GTID_DOMAIN (MariaDB)
 			return ErrorNotifyBadGTIDSetup, myErrorInfo
+		case 1317: // ER_QUERY_INTERRUPTED
+			return ErrorRetryRecoverable, myErrorInfo
 		default:
 			return ErrorOther, myErrorInfo
 		}

flow/alerting/classifier_test.go

@@ -170,6 +170,23 @@ func TestPostgresMemoryAllocErrorShouldBeSlotMemalloc(t *testing.T) {
 	}, errInfo, "Unexpected error info")
 }
 
+func TestPostgresPfreeInvalidPointerErrorShouldBeRecoverable(t *testing.T) {
+	// Simulate a Postgres pfree invalid pointer error
+	err := &exceptions.PostgresWalError{
+		Msg: &pgproto3.ErrorResponse{
+			Severity: "ERROR",
+			Code:     pgerrcode.InternalError,
+			Message:  "pfree called with invalid pointer 0x400720764ed0 (header 0x0000400720825ae0) ",
+		},
+	}
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("error in WAL: %w", err))
+	assert.Equal(t, ErrorRetryRecoverable, errorClass, "Unexpected error class")
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourcePostgres,
+		Code:   pgerrcode.InternalError,
+	}, errInfo, "Unexpected error info")
+}
+
 func TestClickHouseAccessEntityNotFoundErrorShouldBeRecoverable(t *testing.T) {
 	// Simulate a ClickHouse access entity not found error
 	for idx, msg := range []string{

flow/connectors/bigquery/bigquery.go

@@ -239,6 +239,7 @@ func (c *BigQueryConnector) ReplayTableSchemaDeltas(
 	flowJobName string,
 	_ []*protos.TableMapping,
 	schemaDeltas []*protos.TableSchemaDelta,
+	_ uint32,
 ) error {
 	for _, schemaDelta := range schemaDeltas {
 		if schemaDelta == nil || len(schemaDelta.AddedColumns) == 0 {

flow/connectors/bigquery/gcp.go

@@ -71,8 +71,13 @@ func (p *gcsPath) Bucket() string {
 // It is used as the prefix to query objects in the bucket.
 // For example, if the path is "folder/subfolder", it returns "folder/subfolder/".
 // If the path is "folder/subfolder/", it also returns "folder/subfolder/".
+// If the path is empty or just "/", it returns an empty string.
 func (p *gcsPath) QueryPrefix() string {
-	return strings.TrimPrefix(p.Path, "/") + "/"
+	path := strings.TrimPrefix(p.Path, "/")
+	if path == "" {
+		return ""
+	}
+	return path + "/"
 }
 
 func (p *gcsPath) JoinPath(elem ...string) *gcsPath {

flow/connectors/bigquery/qrep.go

@@ -87,7 +87,7 @@ func (c *BigQueryConnector) replayTableSchemaDeltasQRep(
 	}
 
 	if err := c.ReplayTableSchemaDeltas(
-		ctx, config.Env, config.FlowJobName, nil, []*protos.TableSchemaDelta{tableSchemaDelta},
+		ctx, config.Env, config.FlowJobName, nil, []*protos.TableSchemaDelta{tableSchemaDelta}, config.Version,
 	); err != nil {
 		return nil, fmt.Errorf("failed to add columns to destination table: %w", err)
 	}

flow/connectors/bigquery/qrep_avro_sync.go

@@ -98,7 +98,9 @@ func (s *QRepAvroSyncMethod) SyncRecords(
 		slog.String(string(shared.FlowNameKey), req.FlowJobName),
 		slog.String("dstTableName", rawTableName))
 
-	if err := s.connector.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil {
+	if err := s.connector.ReplayTableSchemaDeltas(
+		ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas, req.Version,
+	); err != nil {
 		return nil, fmt.Errorf("failed to sync schema changes: %w", err)
 	}
 

flow/connectors/bigquery/qrep_object_pull.go

@@ -397,22 +397,22 @@ func (c *BigQueryConnector) FinishExport(v any) error {
 	return nil
 }
 
-func (c *BigQueryConnector) SetupReplConn(_ context.Context) error {
+func (c *BigQueryConnector) SetupReplConn(context.Context, map[string]string) error {
 	return nil
 }
 
-func (c *BigQueryConnector) ReplPing(_ context.Context) error {
+func (c *BigQueryConnector) ReplPing(context.Context) error {
 	return nil
 }
 
-func (c *BigQueryConnector) UpdateReplStateLastOffset(_ context.Context, _ model.CdcCheckpoint) error {
+func (c *BigQueryConnector) UpdateReplStateLastOffset(context.Context, model.CdcCheckpoint) error {
 	return nil
 }
 
-func (c *BigQueryConnector) PullFlowCleanup(_ context.Context, _ string) error {
+func (c *BigQueryConnector) PullFlowCleanup(context.Context, string) error {
 	return nil
 }
 
-func (c *BigQueryConnector) SetupReplication(_ context.Context, _ *protos.SetupReplicationInput) (model.SetupReplicationResult, error) {
+func (c *BigQueryConnector) SetupReplication(context.Context, *protos.SetupReplicationInput) (model.SetupReplicationResult, error) {
 	return model.SetupReplicationResult{}, nil
 }

flow/connectors/clickhouse/cdc.go

@@ -131,7 +131,7 @@ func (c *ClickHouseConnector) syncRecordsViaAvro(
 	}
 	warnings := numericTruncator.Warnings()
 
-	if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil {
+	if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas, req.Version); err != nil {
 		return nil, fmt.Errorf("failed to sync schema changes: %w", err)
 	}
 
@@ -165,6 +165,7 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas(
 	flowJobName string,
 	tableMappings []*protos.TableMapping,
 	schemaDeltas []*protos.TableSchemaDelta,
+	internalVersion uint32,
 ) error {
 	if len(schemaDeltas) == 0 {
 		return nil
@@ -188,7 +189,7 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas(
 		for _, addedColumn := range schemaDelta.AddedColumns {
 			qvKind := types.QValueKind(addedColumn.Type)
 			clickHouseColType, err := qvalue.ToDWHColumnType(
-				ctx, qvKind, env, protos.DBType_CLICKHOUSE, c.chVersion, addedColumn, schemaDelta.NullableEnabled,
+				ctx, qvKind, env, protos.DBType_CLICKHOUSE, c.chVersion, addedColumn, schemaDelta.NullableEnabled, internalVersion,
 			)
 			if err != nil {
 				return fmt.Errorf("failed to convert column type %s to ClickHouse type: %w", addedColumn.Type, err)