Merge pull request #179 from xataio/pg-writer-add-syntax-error

[PostgreSQL processor] Add handling of syntax error

Author: eminano

internal/postgres/errors.go

@@ -32,6 +32,14 @@ func (e *ErrConstraintViolation) Error() string {
 	return fmt.Sprintf("constraint violation: %s", e.Details)
 }
 
+type ErrSyntaxError struct {
+	Details string
+}
+
+func (e *ErrSyntaxError) Error() string {
+	return fmt.Sprintf("syntax error: %s", e.Details)
+}
+
 func mapError(err error) error {
 	if pgconn.Timeout(err) {
 		return ErrConnTimeout
@@ -48,6 +56,11 @@ func mapError(err error) error {
 				Details: pgErr.Message,
 			}
 		}
+		if pgErr.Code == "42601" {
+			return &ErrSyntaxError{
+				Details: pgErr.Message,
+			}
+		}
 		// Class 23 — Integrity Constraint Violation
 		if strings.HasPrefix(pgErr.Code, "23") {
 			return &ErrConstraintViolation{

pkg/stream/stream_run.go

@@ -211,6 +211,7 @@ func Run(ctx context.Context, logger loglib.Logger, config *Config, instrumentat
 			return err
 		}
 		defer pgBatchWriter.Close()
+		logger.Info("starting postgres batch writer...")
 
 		processor = pgBatchWriter
 
@@ -222,6 +223,7 @@ func Run(ctx context.Context, logger loglib.Logger, config *Config, instrumentat
 		logger.Info("adding transformation layer to processor...")
 		transformer, err := transformer.New(config.Processor.Transformer, processor, transformer.WithLogger(logger))
 		if err != nil {
+			logger.Error(err, "creating transformer layer")
 			return err
 		}
 		processor = transformer

pkg/wal/processor/postgres/postgres_batch_writer.go

@@ -242,9 +242,11 @@ func (w *BatchWriter) execQueries(ctx context.Context, queries []*query) ([]*que
 func (w *BatchWriter) isInternalError(err error) bool {
 	var errRelationDoesNotExist *pglib.ErrRelationDoesNotExist
 	var errConstraintViolation *pglib.ErrConstraintViolation
+	var errSyntaxError *pglib.ErrSyntaxError
 	switch {
 	case errors.As(err, &errRelationDoesNotExist),
-		errors.As(err, &errConstraintViolation):
+		errors.As(err, &errConstraintViolation),
+		errors.As(err, &errSyntaxError):
 		return false
 	default:
 		return true