Classify missing table on PG (#2906)

Co-authored-by: Kevin Biju <[email protected]>

Author: Amogh-Bharadwaj

flow/alerting/classifier.go

@@ -19,6 +19,7 @@ import (
 	"github.com/jackc/pgx/v5/pgconn"
 	"golang.org/x/crypto/ssh"
 
+	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )
 
@@ -89,6 +90,9 @@ var (
 	ErrorNotifySlotInvalid = ErrorClass{
 		Class: "NOTIFY_SLOT_INVALID", action: NotifyUser,
 	}
+	ErrorNotifySourceTableMissing = ErrorClass{
+		Class: "NOTIFY_SOURCE_TABLE_MISSING", action: NotifyUser,
+	}
 	ErrorNotifyPublicationMissing = ErrorClass{
 		Class: "NOTIFY_PUBLICATION_MISSING", action: NotifyUser,
 	}
@@ -210,6 +214,12 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if errors.Is(err, shared.ErrTableDoesNotExist) {
+		return ErrorNotifySourceTableMissing, ErrorInfo{
+			Source: ErrorSourcePostgres,
+			Code:   "TABLE_DOES_NOT_EXIST",
+		}
+	}
 	if pgErr != nil {
 		switch pgErr.Code {
 		case pgerrcode.InvalidAuthorizationSpecification,

flow/connectors/postgres/client.go

@@ -100,8 +100,6 @@ const (
 	ReplicaIdentityNothing ReplicaIdentityType = 'n'
 )
 
-var ErrSlotAlreadyExists error = errors.New("slot already exists")
-
 // getRelIDForTable returns the relation ID for a table.
 func (c *PostgresConnector) getRelIDForTable(ctx context.Context, schemaTable *utils.SchemaTable) (uint32, error) {
 	var relID pgtype.Uint32
@@ -110,6 +108,9 @@ func (c *PostgresConnector) getRelIDForTable(ctx context.Context, schemaTable *u
 		 ON n.oid = c.relnamespace WHERE n.nspname=$1 AND c.relname=$2`,
 		schemaTable.Schema, schemaTable.Table).Scan(&relID)
 	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, shared.ErrTableDoesNotExist
+		}
 		return 0, fmt.Errorf("error getting relation ID for table %s: %w", schemaTable, err)
 	}
 
@@ -440,7 +441,7 @@ func (c *PostgresConnector) createSlotAndPublication(
 		c.logger.Info(fmt.Sprintf("Replication slot '%s' already exists", slot))
 		var err error
 		if doInitialCopy {
-			err = ErrSlotAlreadyExists
+			err = shared.ErrSlotAlreadyExists
 		}
 		return model.SetupReplicationResult{SlotName: slot}, err
 	}

flow/shared/err_types.go

@@ -1,5 +1,12 @@
 package shared
 
+import "errors"
+
+var (
+	ErrSlotAlreadyExists error = errors.New("slot already exists")
+	ErrTableDoesNotExist error = errors.New("table does not exist")
+)
+
 type ErrType string
 
 const (