sql: Automatically convert json.Number values to appropriate ClickHouse types

Fixes: #4192

Author: benwebber

internal/impl/sql/processor_sql_raw.go

@@ -212,9 +212,12 @@ func NewSQLRawProcessorFromConfig(conf *service.ParsedConfig, mgr *service.Resou
 	}
 
 	var argsConverter argsConverter
-	if driverStr == "postgres" {
+	switch driverStr {
+	case "postgres", "pgx":
 		argsConverter = bloblValuesToPgSQLValues
-	} else {
+	case "clickhouse":
+		argsConverter = bloblValuesToClickHouseValues
+	default:
 		argsConverter = func(v []any) []any { return v }
 	}
 

internal/impl/sql/processor_sql_select.go

@@ -108,8 +108,9 @@ type sqlSelectProcessor struct {
 	builder squirrel.SelectBuilder
 	dbMut   sync.RWMutex
 
-	where       string
-	argsMapping *bloblang.Executor
+	where         string
+	argsMapping   *bloblang.Executor
+	argsConverter argsConverter
 
 	logger  *service.Logger
 	shutSig *shutdown.Signaller
@@ -162,6 +163,13 @@ func NewSQLSelectProcessorFromConfig(conf *service.ParsedConfig, mgr *service.Re
 		s.builder = s.builder.PlaceholderFormat(squirrel.Colon)
 	}
 
+	switch driverStr {
+	case "clickhouse":
+		s.argsConverter = bloblValuesToClickHouseValues
+	default:
+		s.argsConverter = func(v []any) []any { return v }
+	}
+
 	if conf.Contains("prefix") {
 		prefixStr, err := conf.FieldString("prefix")
 		if err != nil {
@@ -233,6 +241,7 @@ func (s *sqlSelectProcessor) ProcessBatch(ctx context.Context, batch service.Mes
 				msg.SetError(fmt.Errorf("mapping returned non-array result: %T", iargs))
 				continue
 			}
+			args = s.argsConverter(args)
 		}
 
 		queryBuilder := s.builder