sql/raw: consolidate error logging

Author: rockwotj

internal/impl/sql/output_sql_raw.go

@@ -281,6 +281,7 @@ func (s *sqlRawOutput) WriteBatch(ctx context.Context, batch service.MessageBatc
 			}
 			defer func() {
 				if err != nil {
+					s.logger.Debugf("%v", err)
 					if rerr := tx.Rollback(); rerr != nil {
 						s.logger.Debugf("Failed to rollback transaction: %v", rerr)
 					}
@@ -298,20 +299,17 @@ func (s *sqlRawOutput) WriteBatch(ctx context.Context, batch service.MessageBatc
 				var resMsg *service.Message
 				resMsg, err = argsExec[j].Query(i)
 				if err != nil {
-					s.logger.Debugf("Arguments mapping failed: %v", err)
-					return err
+					return fmt.Errorf("arguments mapping failed: %w", err)
 				}
 
 				var iargs any
 				iargs, err = resMsg.AsStructured()
 				if err != nil {
-					s.logger.Debugf("Mapping returned non-structured result: %v", err)
 					return fmt.Errorf("mapping returned non-structured result: %w", err)
 				}
 
 				var ok bool
 				if args, ok = iargs.([]any); !ok {
-					s.logger.Debugf("Mapping returned non-array result: %T", iargs)
 					return fmt.Errorf("mapping returned non-array result: %T", iargs)
 				}
 				args = s.argsConverter(args)
@@ -320,7 +318,6 @@ func (s *sqlRawOutput) WriteBatch(ctx context.Context, batch service.MessageBatc
 			queryStr := query.static
 			if query.dynamic != nil {
 				if queryStr, err = dynQueries[j].TryString(i); err != nil {
-					s.logger.Errorf("Query interoplation error: %v", err)
 					return fmt.Errorf("query interpolation error: %w", err)
 				}
 			}
@@ -331,8 +328,7 @@ func (s *sqlRawOutput) WriteBatch(ctx context.Context, batch service.MessageBatc
 				_, err = tx.ExecContext(ctx, queryStr, args...)
 			}
 			if err != nil {
-				s.logger.Debugf("Failed to run query: %v", err)
-				return err
+				return fmt.Errorf("Failed to run query: %w", err)
 			}
 		}
 		return nil

internal/impl/sql/processor_sql_raw.go

@@ -297,22 +297,19 @@ func (s *sqlRawProcessor) ProcessBatch(ctx context.Context, batch service.Messag
 					resMsg, err = argsExec[j].Query(i)
 				}
 				if err != nil {
-					s.logger.Debugf("Arguments mapping failed: %v", err)
 					err = fmt.Errorf("arguments mapping failed: %v", err)
 					break
 				}
 
 				var iargs any
 				iargs, err = resMsg.AsStructured()
 				if err != nil {
-					s.logger.Debugf("Mapping returned non-structured result: %v", err)
 					err = fmt.Errorf("mapping returned non-structured result: %w", err)
 					break
 				}
 
 				var ok bool
 				if args, ok = iargs.([]any); !ok {
-					s.logger.Debugf("Mapping returned non-array result: %T", iargs)
 					err = fmt.Errorf("mapping returned non-array result: %T", iargs)
 					break
 				}
@@ -322,7 +319,6 @@ func (s *sqlRawProcessor) ProcessBatch(ctx context.Context, batch service.Messag
 			queryStr := query.static
 			if query.dynamic != nil {
 				if queryStr, err = dynQueries[j].TryString(i); err != nil {
-					s.logger.Errorf("Query interoplation error: %v", err)
 					err = fmt.Errorf("query interpolation error: %w", err)
 					break
 				}
@@ -335,7 +331,7 @@ func (s *sqlRawProcessor) ProcessBatch(ctx context.Context, batch service.Messag
 					_, err = tx.ExecContext(ctx, queryStr, args...)
 				}
 				if err != nil {
-					s.logger.Debugf("Failed to run query: %v", err)
+					err = fmt.Errorf("failed to run query: %w", err)
 					break
 				}
 			} else {
@@ -346,13 +342,13 @@ func (s *sqlRawProcessor) ProcessBatch(ctx context.Context, batch service.Messag
 					rows, err = tx.QueryContext(ctx, queryStr, args...)
 				}
 				if err != nil {
-					s.logger.Debugf("Failed to run query: %v", err)
+					err = fmt.Errorf("failed to run query: %w", err)
 					break
 				}
 
 				var jArray []any
 				if jArray, err = sqlRowsToArray(rows); err != nil {
-					s.logger.Debugf("Failed to convert rows: %v", err)
+					err = fmt.Errorf("failed to convert rows: %w", err)
 					break
 				}
 
@@ -361,6 +357,7 @@ func (s *sqlRawProcessor) ProcessBatch(ctx context.Context, batch service.Messag
 			}
 		}
 		if err != nil {
+			s.logger.Debugf("%v", err)
 			msg.SetError(err)
 		}
 		if tx != nil {