From eec775698ab493a2d83e108629489cc7dc20a235 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Fri, 17 Apr 2026 17:19:52 -0700 Subject: [PATCH 1/6] Return structured gRPC errors on resync table missing --- flow/cmd/api_error.go | 29 +++-------- flow/cmd/handler.go | 6 ++- flow/cmd/validate_mirror.go | 9 +++- flow/connectors/core.go | 2 + flow/connectors/mongo/validate.go | 15 +++--- flow/connectors/mysql/validate.go | 3 ++ flow/connectors/postgres/validate.go | 5 ++ flow/e2e/api_test.go | 78 ++++++++++++++++++++++++++++ flow/pkg/common/errors.go | 26 ++++++++++ flow/pkg/mongo/validation.go | 2 +- 10 files changed, 142 insertions(+), 33 deletions(-) create mode 100644 flow/pkg/common/errors.go diff --git a/flow/cmd/api_error.go b/flow/cmd/api_error.go index 1dfaa8429..0610b6cce 100644 --- a/flow/cmd/api_error.go +++ b/flow/cmd/api_error.go @@ -10,6 +10,7 @@ import ( "google.golang.org/protobuf/protoadapt" "github.com/PeerDB-io/peerdb/flow/generated/grpc_handler" + "github.com/PeerDB-io/peerdb/flow/pkg/common" ) // APIError is a strongly-typed error that must be a gRPC status error. @@ -107,33 +108,19 @@ func AsAPIError(err error) APIError { return NewInternalApiError(err) } -const ( - ErrorInfoReasonClickHousePeer = "CLICKHOUSE_PEER" - ErrorInfoReasonMirror = "MIRROR" -) - -const ( - ErrorInfoDomain = "peerdb.io" -) - -const ( - ErrorMetadataDownstreamErrorCode = "downstreamErrorCode" - ErrorMetadataOffendingField = "offendingField" -) - -func NewClickHousePeerErrorInfo(metadata map[string]string) *rpc.ErrorInfo { +func NewMirrorErrorInfo(metadata map[string]string) *rpc.ErrorInfo { return &rpc.ErrorInfo{ - Reason: ErrorInfoReasonClickHousePeer, - Domain: ErrorInfoDomain, + Reason: common.ErrorInfoReasonMirror, + Domain: common.ErrorInfoDomain, Metadata: metadata, } } -func NewMirrorErrorInfo(metadata map[string]string) *rpc.ErrorInfo { +func NewSourceTableMissingErrorInfo(table string) *rpc.ErrorInfo { return &rpc.ErrorInfo{ - Reason: ErrorInfoReasonMirror, - Domain: ErrorInfoDomain, - Metadata: metadata, + Reason: common.ErrorInfoReasonSourceTableMissing, + Domain: common.ErrorInfoDomain, + Metadata: map[string]string{common.ErrorMetadataMissingTable: table}, } } diff --git a/flow/cmd/handler.go b/flow/cmd/handler.go index 83511bace..4a37d77df 100644 --- a/flow/cmd/handler.go +++ b/flow/cmd/handler.go @@ -505,7 +505,7 @@ func (h *FlowRequestHandler) FlowStateChange( if _, err := h.ValidateCDCMirror(ctx, &protos.CreateCDCFlowRequest{ ConnectionConfigs: config, }); err != nil { - return nil, NewFailedPreconditionApiError(fmt.Errorf("invalid mirror: %w", err)) + return nil, err } changeErr = model.FlowSignalStateChange.SignalClientWorkflow(ctx, h.temporalClient, workflowID, "", req) if changeErr == nil { @@ -534,6 +534,10 @@ func (h *FlowRequestHandler) FlowStateChange( } if changeErr != nil { slog.ErrorContext(ctx, "unable to signal workflow", logs, slog.Any("error", changeErr)) + // Preserve typed API errors + if apiErr, ok := changeErr.(APIError); ok { + return nil, apiErr + } return nil, NewInternalApiError(fmt.Errorf("unable to signal workflow: %w", changeErr)) } } diff --git a/flow/cmd/validate_mirror.go b/flow/cmd/validate_mirror.go index 1b3fe2390..07c153ebe 100644 --- a/flow/cmd/validate_mirror.go +++ b/flow/cmd/validate_mirror.go @@ -12,6 +12,7 @@ import ( "github.com/PeerDB-io/peerdb/flow/generated/proto_conversions" "github.com/PeerDB-io/peerdb/flow/generated/protos" "github.com/PeerDB-io/peerdb/flow/internal" + "github.com/PeerDB-io/peerdb/flow/pkg/common" "github.com/PeerDB-io/peerdb/flow/shared" "github.com/PeerDB-io/peerdb/flow/shared/types" ) @@ -65,7 +66,7 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl( return nil, NewAlreadyExistsApiError( fmt.Errorf("mirror with name %s already exists", connectionConfigs.FlowJobName), NewMirrorErrorInfo(map[string]string{ - ErrorMetadataOffendingField: "flow_job_name", + common.ErrorMetadataOffendingField: "flow_job_name", })) } } @@ -106,6 +107,12 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl( defer srcClose(ctx) if err := srcConn.ValidateMirrorSource(ctx, connectionConfigs); err != nil { + if missing, ok := errors.AsType[*common.SourceTableMissingError](err); ok { + table := fmt.Sprintf("%s.%s", missing.Table.Namespace, missing.Table.Table) + return nil, NewFailedPreconditionApiError( + fmt.Errorf("source table %s does not exist", table), + NewSourceTableMissingErrorInfo(table)) + } return nil, NewFailedPreconditionApiError( fmt.Errorf("failed to validate source connector %s: %w", connectionConfigs.SourceName, err)) } diff --git a/flow/connectors/core.go b/flow/connectors/core.go index d8275a4c5..3cdbdb624 100644 --- a/flow/connectors/core.go +++ b/flow/connectors/core.go @@ -44,6 +44,8 @@ type ValidationConnector interface { type MirrorSourceValidationConnector interface { GetTableSchemaConnector + // ValidateMirrorSource checks that the source is ready to replicate the configured tables. + // MUST return *common.SourceTableMissingError when a mapped source table is absent. ValidateMirrorSource(context.Context, *protos.FlowConnectionConfigsCore) error } diff --git a/flow/connectors/mongo/validate.go b/flow/connectors/mongo/validate.go index 5f6a1bbec..faa6cea07 100644 --- a/flow/connectors/mongo/validate.go +++ b/flow/connectors/mongo/validate.go @@ -21,14 +21,6 @@ func (c *MongoConnector) ValidateCheck(ctx context.Context) error { } func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error { - if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly { - return nil - } - - if err := shared_mongo.ValidateOplogRetention(ctx, c.client); err != nil { - return err - } - tables := make([]*common.QualifiedTable, 0, len(cfg.TableMappings)) for _, tm := range cfg.TableMappings { t, err := common.ParseTableIdentifier(tm.SourceTableIdentifier) @@ -41,5 +33,10 @@ func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.F return err } - return nil + // no need to check oplog retention for initial-snapshot-only mirrors + if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly { + return nil + } + + return shared_mongo.ValidateOplogRetention(ctx, c.client) } diff --git a/flow/connectors/mysql/validate.go b/flow/connectors/mysql/validate.go index 3fe604c84..9a9e7e36b 100644 --- a/flow/connectors/mysql/validate.go +++ b/flow/connectors/mysql/validate.go @@ -17,6 +17,9 @@ import ( func (c *MySqlConnector) CheckSourceTables(ctx context.Context, tableNames []*common.QualifiedTable) error { for _, parsedTable := range tableNames { if _, err := c.Execute(ctx, fmt.Sprintf("SELECT * FROM %s LIMIT 0", parsedTable.MySQL())); err != nil { + if mErr, ok := errors.AsType[*mysql.MyError](err); ok && mErr.Code == mysql.ER_NO_SUCH_TABLE { + return common.NewSourceTableMissingError(*parsedTable) + } return fmt.Errorf("error checking table %s: %w", parsedTable.MySQL(), err) } } diff --git a/flow/connectors/postgres/validate.go b/flow/connectors/postgres/validate.go index 099a91a95..a6801946a 100644 --- a/flow/connectors/postgres/validate.go +++ b/flow/connectors/postgres/validate.go @@ -7,7 +7,9 @@ import ( "slices" "strings" + "github.com/jackc/pgerrcode" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/PeerDB-io/peerdb/flow/connectors/utils" "github.com/PeerDB-io/peerdb/flow/generated/protos" @@ -62,6 +64,9 @@ func (c *PostgresConnector) CheckSourceTables( if err := c.conn.QueryRow(ctx, fmt.Sprintf("SELECT %s FROM %s LIMIT 0", selectedColumnsStr, parsedTable), ).Scan(&row); err != nil && !errors.Is(err, pgx.ErrNoRows) { + if pgErr, ok := errors.AsType[*pgconn.PgError](err); ok && pgErr.Code == pgerrcode.UndefinedTable { + return common.NewSourceTableMissingError(*parsedTable) + } return fmt.Errorf("failed to select from table %s: %w", parsedTable, err) } } diff --git a/flow/e2e/api_test.go b/flow/e2e/api_test.go index 171886095..3a8c62059 100644 --- a/flow/e2e/api_test.go +++ b/flow/e2e/api_test.go @@ -20,6 +20,7 @@ import ( "go.mongodb.org/mongo-driver/v2/mongo/options" "go.temporal.io/api/enums/v1" "go.temporal.io/api/workflowservice/v1" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" @@ -1370,6 +1371,83 @@ func (s APITestSuite) TestResyncCompleted() { require.Equal(s.t, "test", tagMap[common.PipeNameTag]) } +func (s APITestSuite) TestResyncSourceTableMissing() { + tableName := "missing_src" + qualifiedSourceTable := AttachSchema(s, tableName) + var cols string + switch s.source.(type) { + case *PostgresSource, *MySqlSource: + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE TABLE %s(id int primary key, val text)", qualifiedSourceTable))) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("INSERT INTO %s(id, val) values (1,'first')", qualifiedSourceTable))) + cols = "id,val" + case *MongoSource: + res, err := s.Source().(*MongoSource).AdminClient(). + Database(Schema(s)).Collection(tableName). + InsertOne(s.t.Context(), bson.D{bson.E{Key: "id", Value: 1}, bson.E{Key: "val", Value: "first"}}, options.InsertOne()) + require.NoError(s.t, err) + require.True(s.t, res.Acknowledged) + cols = fmt.Sprintf("%s,%s", connmongo.DefaultDocumentKeyColumnName, connmongo.DefaultFullDocumentColumnName) + default: + require.Fail(s.t, fmt.Sprintf("unknown source type %T", s.source)) + } + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: "resync_missing_" + s.suffix, + TableNameMapping: map[string]string{qualifiedSourceTable: tableName}, + Destination: s.ch.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + flowConnConfig.InitialSnapshotOnly = true + + response, err := s.CreateCDCFlow(s.t.Context(), &protos.CreateCDCFlowRequest{ConnectionConfigs: flowConnConfig}) + require.NoError(s.t, err) + require.NotNil(s.t, response) + + tc := NewTemporalClient(s.t) + env, err := GetPeerflow(s.t.Context(), s.catalog, tc, flowConnConfig.FlowJobName) + require.NoError(s.t, err) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + EnvWaitForFinished(s.t, env, 3*time.Minute) + RequireEqualTables(s.ch, tableName, cols) + + // Drop the source table/collection so that the resync-time validation fails. + switch src := s.source.(type) { + case *PostgresSource, *MySqlSource: + require.NoError(s.t, s.source.Exec(s.t.Context(), "DROP TABLE "+qualifiedSourceTable)) + case *MongoSource: + require.NoError(s.t, src.AdminClient().Database(Schema(s)).Collection(tableName).Drop(s.t.Context())) + default: + require.Fail(s.t, fmt.Sprintf("unknown source type %T", s.source)) + } + + _, err = s.FlowStateChange(s.t.Context(), &protos.FlowStateChangeRequest{ + FlowJobName: flowConnConfig.FlowJobName, + RequestedFlowState: protos.FlowStatus_STATUS_RESYNC, + }) + require.Error(s.t, err) + + st, ok := status.FromError(err) + require.True(s.t, ok, "expected gRPC status error, got %T: %v", err, err) + require.Equal(s.t, codes.FailedPrecondition, st.Code(), "expected FailedPrecondition, got %s", st.Code()) + + var missing *errdetails.ErrorInfo + for _, d := range st.Details() { + info, ok := d.(*errdetails.ErrorInfo) + if !ok { + continue + } + if info.Domain == common.ErrorInfoDomain && info.Reason == common.ErrorInfoReasonSourceTableMissing { + missing = info + break + } + } + require.NotNil(s.t, missing, "expected SourceTableMissing ErrorInfo detail in status") + require.Equal(s.t, fmt.Sprintf("%s.%s", Schema(s), tableName), missing.Metadata[common.ErrorMetadataMissingTable]) +} + func (s APITestSuite) TestResyncFailed() { pgSource, ok := s.source.(*PostgresSource) if !ok { diff --git a/flow/pkg/common/errors.go b/flow/pkg/common/errors.go new file mode 100644 index 000000000..534444004 --- /dev/null +++ b/flow/pkg/common/errors.go @@ -0,0 +1,26 @@ +package common + +import "fmt" + +// gRPC ErrorInfo constants +const ( + ErrorInfoDomain = "peerdb.io" + + ErrorInfoReasonMirror = "MIRROR" + ErrorInfoReasonSourceTableMissing = "SOURCE_TABLE_MISSING" + + ErrorMetadataOffendingField = "offendingField" + ErrorMetadataMissingTable = "missingTable" +) + +type SourceTableMissingError struct { + Table QualifiedTable +} + +func NewSourceTableMissingError(table QualifiedTable) *SourceTableMissingError { + return &SourceTableMissingError{Table: table} +} + +func (e *SourceTableMissingError) Error() string { + return fmt.Sprintf("source table %s.%s does not exist", e.Table.Namespace, e.Table.Table) +} diff --git a/flow/pkg/mongo/validation.go b/flow/pkg/mongo/validation.go index 44653662a..c9ff2c3e3 100644 --- a/flow/pkg/mongo/validation.go +++ b/flow/pkg/mongo/validation.go @@ -123,7 +123,7 @@ func ValidateCollections(ctx context.Context, client *mongo.Client, tables []*co } for _, col := range collections { if !slices.Contains(allCollections, col) { - return fmt.Errorf("collection %s.%s does not exist", database, col) + return common.NewSourceTableMissingError(common.QualifiedTable{Namespace: database, Table: col}) } } } From 78fb576be108dfb4b6da6ae91e835c7dd59a5670 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Fri, 17 Apr 2026 18:00:05 -0700 Subject: [PATCH 2/6] Support BQ --- flow/connectors/bigquery/source.go | 8 ++++++++ flow/e2e/bigquery_source_test.go | 6 +++++- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/flow/connectors/bigquery/source.go b/flow/connectors/bigquery/source.go index 0ead1dd6d..3f11646a6 100644 --- a/flow/connectors/bigquery/source.go +++ b/flow/connectors/bigquery/source.go @@ -4,11 +4,13 @@ import ( "context" "errors" "fmt" + "net/http" "cloud.google.com/go/storage" "google.golang.org/api/iterator" "github.com/PeerDB-io/peerdb/flow/generated/protos" + "github.com/PeerDB-io/peerdb/flow/pkg/common" ) func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigsCore) error { @@ -25,6 +27,12 @@ func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *proto table := c.client.DatasetInProject(c.projectID, dstDatasetTable.dataset).Table(dstDatasetTable.table) if _, err := table.Metadata(ctx); err != nil { + if c.isApiErrorWithStatusCode(err, http.StatusNotFound) { + return common.NewSourceTableMissingError(common.QualifiedTable{ + Namespace: dstDatasetTable.dataset, + Table: dstDatasetTable.table, + }) + } return fmt.Errorf("failed to get metadata for table %s: %w", tableMapping.DestinationTableIdentifier, err) } } diff --git a/flow/e2e/bigquery_source_test.go b/flow/e2e/bigquery_source_test.go index 9e727dc51..33a14c607 100644 --- a/flow/e2e/bigquery_source_test.go +++ b/flow/e2e/bigquery_source_test.go @@ -19,6 +19,7 @@ import ( "github.com/PeerDB-io/peerdb/flow/e2eshared" "github.com/PeerDB-io/peerdb/flow/generated/protos" "github.com/PeerDB-io/peerdb/flow/model" + "github.com/PeerDB-io/peerdb/flow/pkg/common" "github.com/PeerDB-io/peerdb/flow/shared" "github.com/PeerDB-io/peerdb/flow/shared/types" ) @@ -302,7 +303,10 @@ func (s BigQueryClickhouseSuite) Test_BigQuery_Source_Invalid_Table_Mappings() { err := bqConn.ValidateMirrorSource(ctx, flowConfig) require.Error(t, err, "should fail with non-existent table") - require.Contains(t, err.Error(), "failed to get metadata for table", "error should mention metadata failure") + missing, ok := errors.AsType[*common.SourceTableMissingError](err) + require.True(t, ok, "expected SourceTableMissingError, got %T: %v", err, err) + require.Equal(t, source.config.DatasetId, missing.Table.Namespace) + require.Equal(t, "nonexistent_table", missing.Table.Table) } func (s BigQueryClickhouseSuite) Test_BigQuery_Source_ValidateMirrorSource_Success() { From 8f62215405a605b6bb464fdcbe2282202b3407a4 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Mon, 20 Apr 2026 14:38:21 -0700 Subject: [PATCH 3/6] Return machine-readable list of tables, add not in publication error --- flow/cmd/api_error.go | 69 +++++++--- flow/cmd/validate_mirror.go | 14 +- flow/connectors/bigquery/source.go | 7 +- flow/connectors/core.go | 2 +- flow/connectors/mysql/validate.go | 7 +- flow/connectors/postgres/validate.go | 20 +-- flow/e2e/api_test.go | 198 +++++++++++++++++++++++---- flow/e2e/bigquery_source_test.go | 7 +- flow/pkg/common/errors.go | 52 +++++-- flow/pkg/mongo/validation.go | 6 +- 10 files changed, 309 insertions(+), 73 deletions(-) diff --git a/flow/cmd/api_error.go b/flow/cmd/api_error.go index 0610b6cce..65d371101 100644 --- a/flow/cmd/api_error.go +++ b/flow/cmd/api_error.go @@ -2,9 +2,11 @@ package cmd import ( "errors" + "fmt" "log/slog" "github.com/gogo/googleapis/google/rpc" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/protoadapt" @@ -25,31 +27,31 @@ func newAPIError(s *status.Status) *apiError { return &apiError{status: s} } -func NewInvalidArgumentApiError(err error, details ...*rpc.ErrorInfo) *apiError { +func NewInvalidArgumentApiError(err error, details ...protoadapt.MessageV1) *apiError { return newAPIError(convertToStatus(codes.InvalidArgument, err, details...)) } -func NewFailedPreconditionApiError(err error, details ...*rpc.ErrorInfo) *apiError { +func NewFailedPreconditionApiError(err error, details ...protoadapt.MessageV1) *apiError { return newAPIError(convertToStatus(codes.FailedPrecondition, err, details...)) } -func NewInternalApiError(err error, details ...*rpc.ErrorInfo) *apiError { +func NewInternalApiError(err error, details ...protoadapt.MessageV1) *apiError { return newAPIError(convertToStatus(codes.Internal, err, details...)) } -func NewUnavailableApiError(err error, details ...*rpc.ErrorInfo) *apiError { +func NewUnavailableApiError(err error, details ...protoadapt.MessageV1) *apiError { return newAPIError(convertToStatus(codes.Unavailable, err, details...)) } -func NewUnimplementedApiError(err error, details ...*rpc.ErrorInfo) *apiError { +func NewUnimplementedApiError(err error, details ...protoadapt.MessageV1) *apiError { return newAPIError(convertToStatus(codes.Unimplemented, err, details...)) } -func NewAlreadyExistsApiError(err error, details ...*rpc.ErrorInfo) *apiError { +func NewAlreadyExistsApiError(err error, details ...protoadapt.MessageV1) *apiError { return newAPIError(convertToStatus(codes.AlreadyExists, err, details...)) } -func NewNotFoundApiError(err error, details ...*rpc.ErrorInfo) *apiError { +func NewNotFoundApiError(err error, details ...protoadapt.MessageV1) *apiError { return newAPIError(convertToStatus(codes.NotFound, err, details...)) } @@ -71,18 +73,14 @@ func (e *apiError) Code() codes.Code { return e.status.Code() } -func convertToStatus(code codes.Code, err error, details ...*rpc.ErrorInfo) *status.Status { +func convertToStatus(code codes.Code, err error, details ...protoadapt.MessageV1) *status.Status { errorStatus := status.New(code, err.Error()) if len(details) == 0 { return errorStatus } - convertedDetails := make([]protoadapt.MessageV1, len(details)) - for i, detail := range details { - convertedDetails[i] = detail - } - richStatus, err := errorStatus.WithDetails(convertedDetails...) + richStatus, err := errorStatus.WithDetails(details...) if err != nil { - // This cannot happen because we control all calls to convertToStatus and only pass code != OK and allow only rpc.ErrorInfo in details + // This cannot happen because we control all calls to convertToStatus and only pass code != OK and valid proto details slog.Error("Failed to convert to grpc proto error", slog.Any("error", err)) //nolint:sloglint // No context in conversion helper return errorStatus } @@ -116,12 +114,49 @@ func NewMirrorErrorInfo(metadata map[string]string) *rpc.ErrorInfo { } } -func NewSourceTableMissingErrorInfo(table string) *rpc.ErrorInfo { +func NewSourceTableMissingErrorInfo() *rpc.ErrorInfo { return &rpc.ErrorInfo{ - Reason: common.ErrorInfoReasonSourceTableMissing, + Reason: common.ErrorInfoReasonSourceTableMissing, + Domain: common.ErrorInfoDomain, + } +} + +// NewSourceTableMissingPreconditionFailure builds the per-table breakdown detail +// that accompanies the SOURCE_TABLE_MISSING ErrorInfo on FailedPrecondition. +func NewSourceTableMissingPreconditionFailure(tables []common.QualifiedTable) protoadapt.MessageV1 { + violations := make([]*errdetails.PreconditionFailure_Violation, len(tables)) + for i, t := range tables { + subject := fmt.Sprintf("%s.%s", t.Namespace, t.Table) + violations[i] = &errdetails.PreconditionFailure_Violation{ + Type: common.ErrorInfoReasonSourceTableMissing, + Subject: subject, + Description: fmt.Sprintf("source table %s does not exist", subject), + } + } + return protoadapt.MessageV1Of(&errdetails.PreconditionFailure{Violations: violations}) +} + +func NewTablesNotInPublicationErrorInfo(publication string) *rpc.ErrorInfo { + return &rpc.ErrorInfo{ + Reason: common.ErrorInfoReasonTablesNotInPublication, Domain: common.ErrorInfoDomain, - Metadata: map[string]string{common.ErrorMetadataMissingTable: table}, + Metadata: map[string]string{common.ErrorMetadataPublication: publication}, + } +} + +// NewTablesNotInPublicationPreconditionFailure builds the per-table breakdown detail +// that accompanies the TABLES_NOT_IN_PUBLICATION ErrorInfo on FailedPrecondition. +func NewTablesNotInPublicationPreconditionFailure(publication string, tables []common.QualifiedTable) protoadapt.MessageV1 { + violations := make([]*errdetails.PreconditionFailure_Violation, len(tables)) + for i, t := range tables { + subject := fmt.Sprintf("%s.%s", t.Namespace, t.Table) + violations[i] = &errdetails.PreconditionFailure_Violation{ + Type: common.ErrorInfoReasonTablesNotInPublication, + Subject: subject, + Description: fmt.Sprintf("table %s is not in publication %q", subject, publication), + } } + return protoadapt.MessageV1Of(&errdetails.PreconditionFailure{Violations: violations}) } var ErrUnderMaintenance = errors.New("PeerDB is under maintenance. Please retry in a few minutes") diff --git a/flow/cmd/validate_mirror.go b/flow/cmd/validate_mirror.go index 07c153ebe..d6882a1de 100644 --- a/flow/cmd/validate_mirror.go +++ b/flow/cmd/validate_mirror.go @@ -107,11 +107,17 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl( defer srcClose(ctx) if err := srcConn.ValidateMirrorSource(ctx, connectionConfigs); err != nil { - if missing, ok := errors.AsType[*common.SourceTableMissingError](err); ok { - table := fmt.Sprintf("%s.%s", missing.Table.Namespace, missing.Table.Table) + if missing, ok := errors.AsType[*common.SourceTablesMissingError](err); ok { return nil, NewFailedPreconditionApiError( - fmt.Errorf("source table %s does not exist", table), - NewSourceTableMissingErrorInfo(table)) + missing, + NewSourceTableMissingErrorInfo(), + NewSourceTableMissingPreconditionFailure(missing.Tables)) + } + if notInPub, ok := errors.AsType[*common.TablesNotInPublicationError](err); ok { + return nil, NewFailedPreconditionApiError( + notInPub, + NewTablesNotInPublicationErrorInfo(notInPub.Publication), + NewTablesNotInPublicationPreconditionFailure(notInPub.Publication, notInPub.Tables)) } return nil, NewFailedPreconditionApiError( fmt.Errorf("failed to validate source connector %s: %w", connectionConfigs.SourceName, err)) diff --git a/flow/connectors/bigquery/source.go b/flow/connectors/bigquery/source.go index 3f11646a6..0f1a8e329 100644 --- a/flow/connectors/bigquery/source.go +++ b/flow/connectors/bigquery/source.go @@ -18,6 +18,7 @@ func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *proto return fmt.Errorf("BigQuery source connector only supports initial snapshot flows. CDC is not supported") } + var missingTables []common.QualifiedTable for _, tableMapping := range cfg.TableMappings { dstDatasetTable, err := c.convertToDatasetTable(tableMapping.SourceTableIdentifier) if err != nil { @@ -28,14 +29,18 @@ func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *proto if _, err := table.Metadata(ctx); err != nil { if c.isApiErrorWithStatusCode(err, http.StatusNotFound) { - return common.NewSourceTableMissingError(common.QualifiedTable{ + missingTables = append(missingTables, common.QualifiedTable{ Namespace: dstDatasetTable.dataset, Table: dstDatasetTable.table, }) + continue } return fmt.Errorf("failed to get metadata for table %s: %w", tableMapping.DestinationTableIdentifier, err) } } + if len(missingTables) > 0 { + return common.NewSourceTableMissingError(missingTables) + } if cfg.SnapshotStagingPath == "" { return fmt.Errorf("snapshot bucket is required for BigQuery source connector") diff --git a/flow/connectors/core.go b/flow/connectors/core.go index 3cdbdb624..2179ba7e1 100644 --- a/flow/connectors/core.go +++ b/flow/connectors/core.go @@ -45,7 +45,7 @@ type MirrorSourceValidationConnector interface { GetTableSchemaConnector // ValidateMirrorSource checks that the source is ready to replicate the configured tables. - // MUST return *common.SourceTableMissingError when a mapped source table is absent. + // MUST return *common.SourceTablesMissingError when a mapped source table is absent. ValidateMirrorSource(context.Context, *protos.FlowConnectionConfigsCore) error } diff --git a/flow/connectors/mysql/validate.go b/flow/connectors/mysql/validate.go index 9a9e7e36b..8772afa50 100644 --- a/flow/connectors/mysql/validate.go +++ b/flow/connectors/mysql/validate.go @@ -15,14 +15,19 @@ import ( ) func (c *MySqlConnector) CheckSourceTables(ctx context.Context, tableNames []*common.QualifiedTable) error { + var missingTables []common.QualifiedTable for _, parsedTable := range tableNames { if _, err := c.Execute(ctx, fmt.Sprintf("SELECT * FROM %s LIMIT 0", parsedTable.MySQL())); err != nil { if mErr, ok := errors.AsType[*mysql.MyError](err); ok && mErr.Code == mysql.ER_NO_SUCH_TABLE { - return common.NewSourceTableMissingError(*parsedTable) + missingTables = append(missingTables, *parsedTable) + continue } return fmt.Errorf("error checking table %s: %w", parsedTable.MySQL(), err) } } + if len(missingTables) > 0 { + return common.NewSourceTableMissingError(missingTables) + } return nil } diff --git a/flow/connectors/postgres/validate.go b/flow/connectors/postgres/validate.go index a6801946a..02658450e 100644 --- a/flow/connectors/postgres/validate.go +++ b/flow/connectors/postgres/validate.go @@ -43,6 +43,7 @@ func (c *PostgresConnector) CheckSourceTables( // Check that we can select from all tables tableArr := make([]string, 0, len(tableNames)) + var missingTables []common.QualifiedTable for idx, parsedTable := range tableNames { var row pgx.Row tableArr = append(tableArr, fmt.Sprintf(`(%s::text,%s::text)`, @@ -65,11 +66,15 @@ func (c *PostgresConnector) CheckSourceTables( fmt.Sprintf("SELECT %s FROM %s LIMIT 0", selectedColumnsStr, parsedTable), ).Scan(&row); err != nil && !errors.Is(err, pgx.ErrNoRows) { if pgErr, ok := errors.AsType[*pgconn.PgError](err); ok && pgErr.Code == pgerrcode.UndefinedTable { - return common.NewSourceTableMissingError(*parsedTable) + missingTables = append(missingTables, *parsedTable) + continue } return fmt.Errorf("failed to select from table %s: %w", parsedTable, err) } } + if len(missingTables) > 0 { + return common.NewSourceTableMissingError(missingTables) + } if pubName != "" && !noCDC { // Check if publication exists @@ -98,20 +103,19 @@ func (c *PostgresConnector) CheckSourceTables( if err != nil { return err } - missing, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (string, error) { - var schema string - var table string - if err := row.Scan(&schema, &table); err != nil { - return "", err + missing, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (common.QualifiedTable, error) { + var qt common.QualifiedTable + if err := row.Scan(&qt.Namespace, &qt.Table); err != nil { + return common.QualifiedTable{}, err } - return fmt.Sprintf("%s.%s", common.QuoteIdentifier(schema), common.QuoteIdentifier(table)), nil + return qt, nil }) if err != nil { return err } if len(missing) != 0 { - return fmt.Errorf("some tables missing from publication: %s", strings.Join(missing, ", ")) + return common.NewTablesNotInPublicationError(pubName, missing) } } } diff --git a/flow/e2e/api_test.go b/flow/e2e/api_test.go index 3a8c62059..4903bcabe 100644 --- a/flow/e2e/api_test.go +++ b/flow/e2e/api_test.go @@ -1372,31 +1372,38 @@ func (s APITestSuite) TestResyncCompleted() { } func (s APITestSuite) TestResyncSourceTableMissing() { - tableName := "missing_src" - qualifiedSourceTable := AttachSchema(s, tableName) + tableNames := []string{"missing_src_a", "missing_src_b"} + qualifiedSourceTables := []string{AttachSchema(s, tableNames[0]), AttachSchema(s, tableNames[1])} var cols string switch s.source.(type) { case *PostgresSource, *MySqlSource: - require.NoError(s.t, s.source.Exec(s.t.Context(), - fmt.Sprintf("CREATE TABLE %s(id int primary key, val text)", qualifiedSourceTable))) - require.NoError(s.t, s.source.Exec(s.t.Context(), - fmt.Sprintf("INSERT INTO %s(id, val) values (1,'first')", qualifiedSourceTable))) + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE TABLE %s(id int primary key, val text)", qt))) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("INSERT INTO %s(id, val) values (1,'first')", qt))) + } cols = "id,val" case *MongoSource: - res, err := s.Source().(*MongoSource).AdminClient(). - Database(Schema(s)).Collection(tableName). - InsertOne(s.t.Context(), bson.D{bson.E{Key: "id", Value: 1}, bson.E{Key: "val", Value: "first"}}, options.InsertOne()) - require.NoError(s.t, err) - require.True(s.t, res.Acknowledged) + for _, tn := range tableNames { + res, err := s.Source().(*MongoSource).AdminClient(). + Database(Schema(s)).Collection(tn). + InsertOne(s.t.Context(), bson.D{bson.E{Key: "id", Value: 1}, bson.E{Key: "val", Value: "first"}}, options.InsertOne()) + require.NoError(s.t, err) + require.True(s.t, res.Acknowledged) + } cols = fmt.Sprintf("%s,%s", connmongo.DefaultDocumentKeyColumnName, connmongo.DefaultFullDocumentColumnName) default: require.Fail(s.t, fmt.Sprintf("unknown source type %T", s.source)) } connectionGen := FlowConnectionGenerationConfig{ - FlowJobName: "resync_missing_" + s.suffix, - TableNameMapping: map[string]string{qualifiedSourceTable: tableName}, - Destination: s.ch.Peer().Name, + FlowJobName: "resync_missing_" + s.suffix, + TableNameMapping: map[string]string{ + qualifiedSourceTables[0]: tableNames[0], + qualifiedSourceTables[1]: tableNames[1], + }, + Destination: s.ch.Peer().Name, } flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) flowConnConfig.DoInitialSnapshot = true @@ -1411,14 +1418,19 @@ func (s APITestSuite) TestResyncSourceTableMissing() { require.NoError(s.t, err) SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) EnvWaitForFinished(s.t, env, 3*time.Minute) - RequireEqualTables(s.ch, tableName, cols) + for _, tn := range tableNames { + RequireEqualTables(s.ch, tn, cols) + } - // Drop the source table/collection so that the resync-time validation fails. switch src := s.source.(type) { case *PostgresSource, *MySqlSource: - require.NoError(s.t, s.source.Exec(s.t.Context(), "DROP TABLE "+qualifiedSourceTable)) + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), "DROP TABLE "+qt)) + } case *MongoSource: - require.NoError(s.t, src.AdminClient().Database(Schema(s)).Collection(tableName).Drop(s.t.Context())) + for _, tn := range tableNames { + require.NoError(s.t, src.AdminClient().Database(Schema(s)).Collection(tn).Drop(s.t.Context())) + } default: require.Fail(s.t, fmt.Sprintf("unknown source type %T", s.source)) } @@ -1429,23 +1441,155 @@ func (s APITestSuite) TestResyncSourceTableMissing() { }) require.Error(s.t, err) + // Shape of the error on the wire (see AIP-193, https://google.aip.dev/193): + // status.Code = FailedPrecondition + // status.Details = [ + // google.rpc.ErrorInfo{ + // Domain: "peerdb.io", Reason: "SOURCE_TABLE_MISSING", + // }, + // google.rpc.PreconditionFailure{ + // Violations: [ + // {Type: "SOURCE_TABLE_MISSING", Subject: ".", Description: "..."}, + // {Type: "SOURCE_TABLE_MISSING", Subject: ".", Description: "..."}, + // ], + // }, + // ] st, ok := status.FromError(err) require.True(s.t, ok, "expected gRPC status error, got %T: %v", err, err) require.Equal(s.t, codes.FailedPrecondition, st.Code(), "expected FailedPrecondition, got %s", st.Code()) - var missing *errdetails.ErrorInfo + var hasSourceTableMissing bool + var violations []*errdetails.PreconditionFailure_Violation for _, d := range st.Details() { - info, ok := d.(*errdetails.ErrorInfo) - if !ok { - continue + switch detail := d.(type) { + case *errdetails.ErrorInfo: + if detail.Domain == common.ErrorInfoDomain && detail.Reason == common.ErrorInfoReasonSourceTableMissing { + hasSourceTableMissing = true + } + case *errdetails.PreconditionFailure: + violations = append(violations, detail.Violations...) } - if info.Domain == common.ErrorInfoDomain && info.Reason == common.ErrorInfoReasonSourceTableMissing { - missing = info - break + } + require.True(s.t, hasSourceTableMissing, "expected SourceTableMissing ErrorInfo detail in status") + require.Len(s.t, violations, len(tableNames), "expected one PreconditionFailure violation per dropped table") + gotSubjects := make([]string, len(violations)) + for i, v := range violations { + require.Equal(s.t, common.ErrorInfoReasonSourceTableMissing, v.Type) + gotSubjects[i] = v.Subject + } + wantSubjects := make([]string, len(tableNames)) + for i, tn := range tableNames { + wantSubjects[i] = fmt.Sprintf("%s.%s", Schema(s), tn) + } + require.ElementsMatch(s.t, wantSubjects, gotSubjects) +} + +func (s APITestSuite) TestResyncTablesNotInPublication() { + if _, ok := s.source.(*PostgresSource); !ok { + s.t.Skip("only for PostgreSQL (publications are PG-specific)") + } + + tableNames := []string{"resync_pub_a", "resync_pub_b"} + qualifiedSourceTables := []string{AttachSchema(s, tableNames[0]), AttachSchema(s, tableNames[1])} + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE TABLE %s(id int primary key, val text)", qt))) + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("INSERT INTO %s(id, val) values (1,'first')", qt))) + } + + pubName := "pub_resync_" + s.suffix + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("CREATE PUBLICATION %s FOR TABLE %s, %s", pubName, qualifiedSourceTables[0], qualifiedSourceTables[1]))) + s.t.Cleanup(func() { + _ = s.source.Exec(context.Background(), "DROP PUBLICATION IF EXISTS "+pubName) + }) + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: "resync_not_in_pub_" + s.suffix, + TableNameMapping: map[string]string{ + qualifiedSourceTables[0]: tableNames[0], + qualifiedSourceTables[1]: tableNames[1], + }, + Destination: s.ch.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + flowConnConfig.PublicationName = pubName + + response, err := s.CreateCDCFlow(s.t.Context(), &protos.CreateCDCFlowRequest{ConnectionConfigs: flowConnConfig}) + require.NoError(s.t, err) + require.NotNil(s.t, response) + + tc := NewTemporalClient(s.t) + env, err := GetPeerflow(s.t.Context(), s.catalog, tc, flowConnConfig.FlowJobName) + require.NoError(s.t, err) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + for _, tn := range tableNames { + EnvWaitForCount(env, s.ch, "initial snapshot", tn, "id,val", 1) + } + EnvWaitFor(s.t, env, 3*time.Minute, "flow running", func() bool { + return env.GetFlowStatus(s.t) == protos.FlowStatus_STATUS_RUNNING + }) + + for _, qt := range qualifiedSourceTables { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf("ALTER PUBLICATION %s DROP TABLE %s", pubName, qt))) + } + + _, err = s.FlowStateChange(s.t.Context(), &protos.FlowStateChangeRequest{ + FlowJobName: flowConnConfig.FlowJobName, + RequestedFlowState: protos.FlowStatus_STATUS_RESYNC, + }) + require.Error(s.t, err) + + // Shape of the error on the wire: + // status.Code = FailedPrecondition + // status.Details = [ + // google.rpc.ErrorInfo{ + // Domain: "peerdb.io", Reason: "TABLES_NOT_IN_PUBLICATION", + // Metadata: {"publication": ""}, + // }, + // google.rpc.PreconditionFailure{ + // Violations: [ + // {Type: "TABLES_NOT_IN_PUBLICATION", Subject: ".", Description: "..."}, + // {Type: "TABLES_NOT_IN_PUBLICATION", Subject: ".", Description: "..."}, + // ], + // }, + // ] + st, ok := status.FromError(err) + require.True(s.t, ok, "expected gRPC status error, got %T: %v", err, err) + require.Equal(s.t, codes.FailedPrecondition, st.Code(), "expected FailedPrecondition, got %s", st.Code()) + + var gotReason, gotPublication string + var violations []*errdetails.PreconditionFailure_Violation + for _, d := range st.Details() { + switch detail := d.(type) { + case *errdetails.ErrorInfo: + if detail.Domain == common.ErrorInfoDomain { + gotReason = detail.Reason + gotPublication = detail.Metadata[common.ErrorMetadataPublication] + } + case *errdetails.PreconditionFailure: + violations = append(violations, detail.Violations...) } } - require.NotNil(s.t, missing, "expected SourceTableMissing ErrorInfo detail in status") - require.Equal(s.t, fmt.Sprintf("%s.%s", Schema(s), tableName), missing.Metadata[common.ErrorMetadataMissingTable]) + require.Equal(s.t, common.ErrorInfoReasonTablesNotInPublication, gotReason) + require.Equal(s.t, pubName, gotPublication) + require.Len(s.t, violations, len(tableNames)) + gotSubjects := make([]string, len(violations)) + for i, v := range violations { + require.Equal(s.t, common.ErrorInfoReasonTablesNotInPublication, v.Type) + gotSubjects[i] = v.Subject + } + wantSubjects := make([]string, len(tableNames)) + for i, tn := range tableNames { + wantSubjects[i] = fmt.Sprintf("%s.%s", Schema(s), tn) + } + require.ElementsMatch(s.t, wantSubjects, gotSubjects) + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) } func (s APITestSuite) TestResyncFailed() { diff --git a/flow/e2e/bigquery_source_test.go b/flow/e2e/bigquery_source_test.go index 33a14c607..f020d77d9 100644 --- a/flow/e2e/bigquery_source_test.go +++ b/flow/e2e/bigquery_source_test.go @@ -303,10 +303,11 @@ func (s BigQueryClickhouseSuite) Test_BigQuery_Source_Invalid_Table_Mappings() { err := bqConn.ValidateMirrorSource(ctx, flowConfig) require.Error(t, err, "should fail with non-existent table") - missing, ok := errors.AsType[*common.SourceTableMissingError](err) + missing, ok := errors.AsType[*common.SourceTablesMissingError](err) require.True(t, ok, "expected SourceTableMissingError, got %T: %v", err, err) - require.Equal(t, source.config.DatasetId, missing.Table.Namespace) - require.Equal(t, "nonexistent_table", missing.Table.Table) + require.Len(t, missing.Tables, 1) + require.Equal(t, source.config.DatasetId, missing.Tables[0].Namespace) + require.Equal(t, "nonexistent_table", missing.Tables[0].Table) } func (s BigQueryClickhouseSuite) Test_BigQuery_Source_ValidateMirrorSource_Success() { diff --git a/flow/pkg/common/errors.go b/flow/pkg/common/errors.go index 534444004..fcf6eed95 100644 --- a/flow/pkg/common/errors.go +++ b/flow/pkg/common/errors.go @@ -1,26 +1,58 @@ package common -import "fmt" +import ( + "fmt" + "strings" +) // gRPC ErrorInfo constants const ( ErrorInfoDomain = "peerdb.io" - ErrorInfoReasonMirror = "MIRROR" - ErrorInfoReasonSourceTableMissing = "SOURCE_TABLE_MISSING" + ErrorInfoReasonMirror = "MIRROR" + ErrorInfoReasonSourceTableMissing = "SOURCE_TABLE_MISSING" + ErrorInfoReasonTablesNotInPublication = "TABLES_NOT_IN_PUBLICATION" ErrorMetadataOffendingField = "offendingField" - ErrorMetadataMissingTable = "missingTable" + ErrorMetadataPublication = "publication" ) -type SourceTableMissingError struct { - Table QualifiedTable +type SourceTablesMissingError struct { + Tables []QualifiedTable +} + +func NewSourceTableMissingError(tables []QualifiedTable) *SourceTablesMissingError { + return &SourceTablesMissingError{Tables: tables} +} + +func (e *SourceTablesMissingError) Error() string { + switch len(e.Tables) { + case 0: + return "some source tables do not exist" // Not expected to be hit but can't easily enforce it either + case 1: + return fmt.Sprintf("source table %s.%s does not exist", e.Tables[0].Namespace, e.Tables[0].Table) + default: + parts := make([]string, len(e.Tables)) + for i, t := range e.Tables { + parts[i] = fmt.Sprintf("%s.%s", t.Namespace, t.Table) + } + return "source tables do not exist: " + strings.Join(parts, ", ") + } +} + +type TablesNotInPublicationError struct { + Tables []QualifiedTable + Publication string } -func NewSourceTableMissingError(table QualifiedTable) *SourceTableMissingError { - return &SourceTableMissingError{Table: table} +func NewTablesNotInPublicationError(publication string, tables []QualifiedTable) *TablesNotInPublicationError { + return &TablesNotInPublicationError{Publication: publication, Tables: tables} } -func (e *SourceTableMissingError) Error() string { - return fmt.Sprintf("source table %s.%s does not exist", e.Table.Namespace, e.Table.Table) +func (e *TablesNotInPublicationError) Error() string { + parts := make([]string, len(e.Tables)) + for i, t := range e.Tables { + parts[i] = fmt.Sprintf("%s.%s", t.Namespace, t.Table) + } + return fmt.Sprintf("tables not in publication %q: %s", e.Publication, strings.Join(parts, ", ")) } diff --git a/flow/pkg/mongo/validation.go b/flow/pkg/mongo/validation.go index c9ff2c3e3..2ed5e7243 100644 --- a/flow/pkg/mongo/validation.go +++ b/flow/pkg/mongo/validation.go @@ -116,6 +116,7 @@ func ValidateCollections(ctx context.Context, client *mongo.Client, tables []*co databaseCollectionsMapping[t.Namespace] = append(databaseCollectionsMapping[t.Namespace], t.Table) } + var missingTables []common.QualifiedTable for database, collections := range databaseCollectionsMapping { allCollections, err := GetCollectionNames(ctx, client, database) if err != nil { @@ -123,10 +124,13 @@ func ValidateCollections(ctx context.Context, client *mongo.Client, tables []*co } for _, col := range collections { if !slices.Contains(allCollections, col) { - return common.NewSourceTableMissingError(common.QualifiedTable{Namespace: database, Table: col}) + missingTables = append(missingTables, common.QualifiedTable{Namespace: database, Table: col}) } } } + if len(missingTables) > 0 { + return common.NewSourceTableMissingError(missingTables) + } return nil } From ef61b3ac22d2c1dacea48603b22fb0528ba1da4c Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Mon, 20 Apr 2026 14:49:39 -0700 Subject: [PATCH 4/6] Check tables before the publication --- flow/connectors/postgres/validate.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/flow/connectors/postgres/validate.go b/flow/connectors/postgres/validate.go index 02658450e..7bf0620ab 100644 --- a/flow/connectors/postgres/validate.go +++ b/flow/connectors/postgres/validate.go @@ -250,6 +250,11 @@ func (c *PostgresConnector) ValidateMirrorSource(ctx context.Context, cfg *proto } pubName := cfg.PublicationName + // Check source tables before the publication, for better errors + if err := c.CheckSourceTables(ctx, sourceTables, cfg.TableMappings, pubName, noCDC); err != nil { + return fmt.Errorf("provided source tables invalidated: %w", err) + } + if pubName == "" && !noCDC { srcTableNames := make([]string, 0, len(sourceTables)) for _, srcTable := range sourceTables { @@ -261,10 +266,6 @@ func (c *PostgresConnector) ValidateMirrorSource(ctx context.Context, cfg *proto } } - if err := c.CheckSourceTables(ctx, sourceTables, cfg.TableMappings, pubName, noCDC); err != nil { - return fmt.Errorf("provided source tables invalidated: %w", err) - } - return nil } From d900edfc909f922f8c511cebe897ab546f1cd45e Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Mon, 20 Apr 2026 15:02:41 -0700 Subject: [PATCH 5/6] plural, simplify --- flow/pkg/common/errors.go | 17 +++++------------ flow/pkg/mongo/validation.go | 2 +- 2 files changed, 6 insertions(+), 13 deletions(-) diff --git a/flow/pkg/common/errors.go b/flow/pkg/common/errors.go index fcf6eed95..68f695527 100644 --- a/flow/pkg/common/errors.go +++ b/flow/pkg/common/errors.go @@ -21,23 +21,16 @@ type SourceTablesMissingError struct { Tables []QualifiedTable } -func NewSourceTableMissingError(tables []QualifiedTable) *SourceTablesMissingError { +func NewSourceTablesMissingError(tables []QualifiedTable) *SourceTablesMissingError { return &SourceTablesMissingError{Tables: tables} } func (e *SourceTablesMissingError) Error() string { - switch len(e.Tables) { - case 0: - return "some source tables do not exist" // Not expected to be hit but can't easily enforce it either - case 1: - return fmt.Sprintf("source table %s.%s does not exist", e.Tables[0].Namespace, e.Tables[0].Table) - default: - parts := make([]string, len(e.Tables)) - for i, t := range e.Tables { - parts[i] = fmt.Sprintf("%s.%s", t.Namespace, t.Table) - } - return "source tables do not exist: " + strings.Join(parts, ", ") + parts := make([]string, len(e.Tables)) + for i, t := range e.Tables { + parts[i] = fmt.Sprintf("%s.%s", t.Namespace, t.Table) } + return "source tables do not exist: " + strings.Join(parts, ", ") } type TablesNotInPublicationError struct { diff --git a/flow/pkg/mongo/validation.go b/flow/pkg/mongo/validation.go index 2ed5e7243..0c8852f03 100644 --- a/flow/pkg/mongo/validation.go +++ b/flow/pkg/mongo/validation.go @@ -129,7 +129,7 @@ func ValidateCollections(ctx context.Context, client *mongo.Client, tables []*co } } if len(missingTables) > 0 { - return common.NewSourceTableMissingError(missingTables) + return common.NewSourceTablesMissingError(missingTables) } return nil } From c3de9921e5e5c33b0768a24d65c442175601f711 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Mon, 20 Apr 2026 15:08:45 -0700 Subject: [PATCH 6/6] bad rename --- flow/connectors/bigquery/source.go | 2 +- flow/connectors/mysql/validate.go | 2 +- flow/connectors/postgres/validate.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/flow/connectors/bigquery/source.go b/flow/connectors/bigquery/source.go index 0f1a8e329..13eebee50 100644 --- a/flow/connectors/bigquery/source.go +++ b/flow/connectors/bigquery/source.go @@ -39,7 +39,7 @@ func (c *BigQueryConnector) ValidateMirrorSource(ctx context.Context, cfg *proto } } if len(missingTables) > 0 { - return common.NewSourceTableMissingError(missingTables) + return common.NewSourceTablesMissingError(missingTables) } if cfg.SnapshotStagingPath == "" { diff --git a/flow/connectors/mysql/validate.go b/flow/connectors/mysql/validate.go index 8772afa50..0ef83e4cc 100644 --- a/flow/connectors/mysql/validate.go +++ b/flow/connectors/mysql/validate.go @@ -26,7 +26,7 @@ func (c *MySqlConnector) CheckSourceTables(ctx context.Context, tableNames []*co } } if len(missingTables) > 0 { - return common.NewSourceTableMissingError(missingTables) + return common.NewSourceTablesMissingError(missingTables) } return nil } diff --git a/flow/connectors/postgres/validate.go b/flow/connectors/postgres/validate.go index 7bf0620ab..4d942747a 100644 --- a/flow/connectors/postgres/validate.go +++ b/flow/connectors/postgres/validate.go @@ -73,7 +73,7 @@ func (c *PostgresConnector) CheckSourceTables( } } if len(missingTables) > 0 { - return common.NewSourceTableMissingError(missingTables) + return common.NewSourceTablesMissingError(missingTables) } if pubName != "" && !noCDC {