schema: adopt Decimal and BigDecimal common types across sources and converters (#4358)

* schema: adopt Decimal and BigDecimal common types across CDC sources and converters

Threads benthos's new Decimal and BigDecimal common-schema types end-to-end:

- Five CDC sources (postgres, mysql, mssqlserver, oracledb, mongodb) now emit
  Decimal(p, s) when precision and scale are declared and BigDecimal when
  they are not, with values normalised to canonical decimal strings via a
  new internal/sqlutil canonicaliser.
- Four format converters (iceberg, parquet, avro, json schema) honour
  Decimal natively. BigDecimal is rejected by the bounded-format encoders
  with an actionable error and accepted by JSON Schema as a permissive
  string-with-pattern.
- ecs_avro detects logicalType: decimal in Avro specs and the
  schema_registry_decode store_schema_metadata path normalises decoded
  big.Rat values to canonical strings.
- Shared Parquet decimal-byte helpers extracted into
  internal/impl/parquet/parquetdecimal so the parquet encoder and the
  iceberg shredder no longer carry duplicate implementations.

The adoption is wired through a temporary go.mod replace directive
pointing at the local benthos checkout while a tagged release is
prepared; that directive is the one remaining follow-up before merge.

* oracledb, mssqlserver: fix decimal value-shape regressions surfaced by integration tests

Two integration-test failures pinned during PR-readiness verification:

1. Oracle bare NUMBER columns (no declared precision and scale) were
   routed through the Decimal canonicaliser because go-ora's
   *sql.ColumnType.DecimalSize() reports (precision=38, scale=255, ok=true)
   for them — 255 is the driver's "any-scale" sentinel. The snapshot
   mapper treated that as a real (p, s) and called Decimal(38, 255),
   producing "decimal value has 255 significant digits" errors. The
   oracleNumberToCommon schema mapping had the same hole. Both now treat
   scale > precision as undeclared and fall back to BigDecimal so the
   schema cache and the value mapper agree, leaving the source lossless.

2. MSSQL CDC streaming scanned DECIMAL/NUMERIC columns into *any, which
   go-mssqldb coerced to a lossy float64. The streaming iterator now
   pre-allocates *sql.NullString scan targets for DECIMAL/NUMERIC and
   MONEY/SMALLMONEY so the driver hands back the lossless text
   representation. The stream-snapshot code path in replication/snapshot.go
   was also still wrapping DECIMAL/NUMERIC values in json.Number from the
   pre-Decimal era; it now routes through sqlutil.CanonicaliseDecimal /
   CanonicaliseBigDecimal in line with the regular snapshot and streaming
   paths.

Improves the snapshot mapper's error message in oracledb to include the
column name and input text so future driver quirks are easier to spot,
and updates the streaming-block fixture for NOLEADINGZERO_COL in the
oracledb all-types integration test (previously asserted as a float64,
now a canonical BigDecimal string).

* chore: clear task lint and task test gates

Three small fixes to bring the repo to a clean lint/test state:

- internal/impl/confluent/ecs_avro.go: drop the unused ecsAvroFromBytes
  wrapper (callers were migrated to ecsAvroParseFromBytes during the
  decoder normalisation work) and replace a perfsprint-flagged
  fmt.Errorf("missing") with errors.New.
- internal/impl/postgresql/pglogicalstream/schema_test.go: simplify the
  redundant "((1 << 16) | 0) + 4" atttypmod fixture to "(1 << 16) + 4"
  per staticcheck's SA4016.
- internal/impl/tigerbeetle/integration_test.go: switch the docker
  container types import from "github.com/docker/docker/api/types/container"
  to "github.com/moby/moby/api/types/container" to match the signature
  testcontainers-go now expects (used elsewhere in this repo). The
  docker module relocated these types upstream; the existing import
  was a pre-existing typecheck failure that golangci-lint surfaces.

* Pin main for benthos (temporary)

* review: address PR feedback on ecs_avro scale default and decimal canonicaliser rounding

Three review fixes:

- internal/impl/confluent/ecs_avro.go: per the Avro spec scale is
  optional in the decimal logical type and defaults to 0 when absent.
  The reverse-direction reader was treating a missing scale as an error;
  it now returns Decimal(precision, 0) for those fields. (joseph.woodward)

- internal/sqlutil/decimal.go: replace the big.Float fallback with a
  big.Rat parse and an exact "fits at the declared scale" check.
  Previously an input like "1.56789" against a NUMBER(10, 2) column
  silently rounded to "1.57" because the big.Float path added 0.5 and
  truncated; rationals represent decimals exactly, so the check is now
  a real precision-loss test. Inputs that lose precision at the
  declared scale return an error. Scientific notation, leading +, and
  fewer-than-scale fractional digits continue to canonicalise as
  before. (joseph.woodward)

- License URLs in internal/sqlutil/decimal.go and decimal_test.go:
  drop the erroneous "/v4" segment to match the rest of the RCL
  headers in the repo. (claude[bot])

Adds tests for both the scale-default Avro spec and the
precision-loss rejection.

* address unsigned decimal in mysql

* address special NUMERIC values in PostgreSQL

* oracledb: update streaming integration fixtures for BigDecimal value shape

TestIntegrationOracleDBCDCStreaming uses tables with bare NUMBER columns
(id NUMBER GENERATED ALWAYS AS IDENTITY and val NUMBER), both of which
now fall through to BigDecimal under the new schema mapping. Their
values are emitted as canonical decimal strings rather than json.Number
integers, so the per-subtest content assertions move from
{"ID":1,"VAL":1} / {"ID":1,"VAL":2} to {"ID":"1","VAL":"1"} /
{"ID":"1","VAL":"2"}.

* update failing test

---------

Co-authored-by: Joseph Woodward <joseph.woodward@xeuse.com>

Author: Jeffail

CHANGELOG.md

@@ -23,6 +23,16 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - parquet_encode: Added `default_timestamp_unit` field (values `NANOSECOND`, `MICROSECOND`, `MILLISECOND`) controlling the precision of TIMESTAMP logical types. Default remains `NANOSECOND` for backwards compatibility. Use `MICROSECOND` when writing files for Apache Spark/Databricks, AWS Athena or DuckDB, which do not support `TIMESTAMP(NANOS)`. ([#3570](https://github.com/redpanda-data/connect/issues/3570))
+- iceberg, parquet_encode, schema_registry_encode: Added support for the new `Decimal` and `BigDecimal` benthos common-schema types in metadata-driven encoding. Iceberg / Parquet / Avro encoders emit native fixed-precision decimal types for `Decimal`; JSON Schema emits a regex-validated string. `BigDecimal` is rejected by the bounded-format encoders with a clear error and accepted by JSON Schema as a permissive string pattern. ([@Jeffail](https://github.com/Jeffail))
+- schema_registry_decode: When `store_schema_metadata` is set, Avro decimal logical-type values are now normalised to canonical decimal strings to match the schema metadata's value contract. ([@Jeffail](https://github.com/Jeffail))
+
+### Changed
+
+- postgresql: NUMERIC and DECIMAL columns now emit `Decimal(p, s)` schema metadata when precision/scale is declared, or `BigDecimal` for unparameterised `numeric` columns. Values are emitted as canonical decimal strings (right-padded to the declared scale for `Decimal`). Previously these columns surfaced as `String` with the raw Postgres text. ([@Jeffail](https://github.com/Jeffail))
+- mysql_cdc: DECIMAL and NUMERIC columns now emit `Decimal(p, s)` schema metadata parsed from the column's raw type, and values are normalised to canonical decimal strings. Previously these columns surfaced as `String` with the driver's native form. ([@Jeffail](https://github.com/Jeffail))
+- microsoft_sql_server_cdc: DECIMAL and NUMERIC columns now emit `Decimal(p, s)` schema metadata sourced from `sql.ColumnType.DecimalSize()`, and values are normalised to canonical decimal strings. MONEY and SMALLMONEY remain typed as `String`, but their wire form is now a quoted canonical decimal string instead of a raw `json.Number`. Previously DECIMAL/NUMERIC were `json.Number` typed as `String`. ([@Jeffail](https://github.com/Jeffail))
+- oracledb_cdc: NUMBER columns with declared precision and scale > 0 now emit `Decimal(p, s)` schema metadata; NUMBER without `DATA_PRECISION` emits `BigDecimal`. Decimal values flow through as canonical decimal strings; integer-width NUMBER (precision ≤ 18, scale 0) continues to emit `int64`. The previous `json.Number` wrapping for NUMBER-as-String columns is gone. ([@Jeffail](https://github.com/Jeffail))
+- mongodb_cdc: `bson.Decimal128` and `bsonType: "decimal"` validator fields now emit `BigDecimal` schema metadata. Decimal values in document bodies are emitted as canonical decimal strings instead of the previous `{"$numberDecimal": "..."}` ExtJSON wrapper. ([@Jeffail](https://github.com/Jeffail))
 
 ## 4.88.0 - 2026-04-16
 

go.mod

@@ -144,7 +144,7 @@ require (
 	github.com/rabbitmq/amqp091-go v1.10.0
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9
 	github.com/redis/go-redis/v9 v9.18.0
-	github.com/redpanda-data/benthos/v4 v4.71.0
+	github.com/redpanda-data/benthos/v4 v4.71.1-0.20260428154615-ddb2b041e022
 	github.com/redpanda-data/common-go/authz v0.2.1-0.20260319205134-242ab3c168b8
 	github.com/redpanda-data/common-go/license v0.0.0-20260318014216-2bbd72bde0a0
 	github.com/redpanda-data/common-go/redpanda-otel-exporter v0.4.0

go.sum

@@ -1553,8 +1553,8 @@ github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8A
 github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs=
 github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0=
-github.com/redpanda-data/benthos/v4 v4.71.0 h1:X6U5e6rYC2XDjU6PN2KBVs5ZkmofkMpURXnKj9hLw7s=
-github.com/redpanda-data/benthos/v4 v4.71.0/go.mod h1:if/3gnj/gIz3mKIiz2MGF7gNag/gv7ak0snVxP81BM4=
+github.com/redpanda-data/benthos/v4 v4.71.1-0.20260428154615-ddb2b041e022 h1:VCBdg5AShnNZaON2LuGqdR5WkIbw7DrBhkSDq5leKV4=
+github.com/redpanda-data/benthos/v4 v4.71.1-0.20260428154615-ddb2b041e022/go.mod h1:if/3gnj/gIz3mKIiz2MGF7gNag/gv7ak0snVxP81BM4=
 github.com/redpanda-data/common-go/authz v0.2.1-0.20260319205134-242ab3c168b8 h1:hZTIp81OUDNOTCTD0gM01b1t821pDbToU9jWnZRnd/E=
 github.com/redpanda-data/common-go/authz v0.2.1-0.20260319205134-242ab3c168b8/go.mod h1:sHhzCYf64ZYUBi7snbopQl+wQaKySbFsKCvGhmSckhk=
 github.com/redpanda-data/common-go/license v0.0.0-20260318014216-2bbd72bde0a0 h1:xL2THs63tUTZmTiBfBm/mrjFMrwQaHKduvgQ6gIizXg=

internal/impl/confluent/common_to_avro.go

@@ -76,6 +76,18 @@ func commonToAvroInner(c schema.Common, recordName, namespace string, isRoot boo
 			"type":        "long",
 			"logicalType": "timestamp-millis",
 		}, nil
+	case schema.Decimal:
+		if c.Logical == nil || c.Logical.Decimal == nil {
+			return nil, fmt.Errorf("decimal field %q missing precision/scale", c.Name)
+		}
+		return map[string]any{
+			"type":        "bytes",
+			"logicalType": "decimal",
+			"precision":   c.Logical.Decimal.Precision,
+			"scale":       c.Logical.Decimal.Scale,
+		}, nil
+	case schema.BigDecimal:
+		return nil, fmt.Errorf("field %q is BigDecimal which has no fixed precision/scale; cast or coerce upstream before schema_registry_encode (avro)", c.Name)
 	case schema.Array:
 		return commonToAvroArray(c)
 	case schema.Map:

internal/impl/confluent/common_to_avro_test.go

@@ -158,6 +158,48 @@ func TestCommonToAvroUnion(t *testing.T) {
 	assert.Equal(t, []any{"string", "int", "null"}, got)
 }
 
+func TestCommonToAvroDecimal(t *testing.T) {
+	c := schema.Common{
+		Name:    "amount",
+		Type:    schema.Decimal,
+		Logical: &schema.LogicalParams{Decimal: &schema.DecimalParams{Precision: 18, Scale: 4}},
+	}
+	got := avroUnmarshal(t, c, "", "").(map[string]any)
+	assert.Equal(t, "bytes", got["type"])
+	assert.Equal(t, "decimal", got["logicalType"])
+	// JSON numbers come back as float64; both fields are present.
+	assert.Equal(t, float64(18), got["precision"])
+	assert.Equal(t, float64(4), got["scale"])
+}
+
+func TestCommonToAvroDecimalOptionalUnion(t *testing.T) {
+	c := schema.Common{
+		Name:     "amount",
+		Type:     schema.Decimal,
+		Optional: true,
+		Logical:  &schema.LogicalParams{Decimal: &schema.DecimalParams{Precision: 9, Scale: 2}},
+	}
+	got := avroUnmarshal(t, c, "", "").([]any)
+	require.Len(t, got, 2)
+	assert.Equal(t, "null", got[0])
+	inner := got[1].(map[string]any)
+	assert.Equal(t, "decimal", inner["logicalType"])
+	assert.Equal(t, float64(9), inner["precision"])
+	assert.Equal(t, float64(2), inner["scale"])
+}
+
+func TestCommonToAvroDecimalMissingLogical(t *testing.T) {
+	_, err := commonToAvroSchema(schema.Common{Name: "amount", Type: schema.Decimal}, "", "")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "missing precision/scale")
+}
+
+func TestCommonToAvroBigDecimalRejected(t *testing.T) {
+	_, err := commonToAvroSchema(schema.Common{Name: "amount", Type: schema.BigDecimal}, "", "")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "BigDecimal")
+}
+
 func TestSanitizeAvroName(t *testing.T) {
 	tests := []struct {
 		input, want string

internal/impl/confluent/common_to_json_schema.go

@@ -59,6 +59,19 @@ func commonToJSONSchemaNode(c schema.Common) (map[string]any, error) {
 		return commonToJSONSchemaUnion(c)
 	case schema.Timestamp:
 		return map[string]any{"type": "string", "format": "date-time"}, nil
+	case schema.Decimal:
+		if c.Logical == nil || c.Logical.Decimal == nil {
+			return nil, fmt.Errorf("decimal field %q missing precision/scale", c.Name)
+		}
+		return map[string]any{
+			"type":    "string",
+			"pattern": decimalPattern(c.Logical.Decimal.Precision, c.Logical.Decimal.Scale),
+		}, nil
+	case schema.BigDecimal:
+		return map[string]any{
+			"type":    "string",
+			"pattern": `^-?(0|[1-9][0-9]*)(\.[0-9]+)?$`,
+		}, nil
 	case schema.Any:
 		return map[string]any{}, nil
 	default:
@@ -128,3 +141,21 @@ func commonToJSONSchemaUnion(c schema.Common) (map[string]any, error) {
 	}
 	return map[string]any{"oneOf": oneOf}, nil
 }
+
+// decimalPattern returns a JSON Schema regex matching the canonical decimal
+// string form for Decimal(p, s). The pattern enforces no leading zeros (except
+// a single 0 before the decimal point), no scientific notation, optional
+// leading minus, and exactly s fractional digits when s > 0.
+func decimalPattern(precision, scale int32) string {
+	m := precision - scale
+	switch {
+	case scale == 0:
+		// Integer-only: up to precision digits.
+		return fmt.Sprintf(`^-?(0|[1-9][0-9]{0,%d})$`, m-1)
+	case m == 0:
+		// Fractional-only (e.g. Decimal(4,4)): integer part can only be 0.
+		return fmt.Sprintf(`^-?0\.[0-9]{%d}$`, scale)
+	default:
+		return fmt.Sprintf(`^-?(0|[1-9][0-9]{0,%d})\.[0-9]{%d}$`, m-1, scale)
+	}
+}

internal/impl/confluent/common_to_json_schema_test.go

@@ -16,6 +16,7 @@ package confluent
 
 import (
 	"encoding/json"
+	"regexp"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -132,3 +133,78 @@ func TestCommonToJSONSchemaUnion(t *testing.T) {
 	assert.Equal(t, "string", oneOf[0].(map[string]any)["type"])
 	assert.Equal(t, "integer", oneOf[1].(map[string]any)["type"])
 }
+
+func TestCommonToJSONSchemaDecimal(t *testing.T) {
+	cases := []struct {
+		name      string
+		precision int32
+		scale     int32
+		matches   []string
+		rejects   []string
+	}{
+		{
+			name:      "scale 0 integer-only",
+			precision: 5,
+			scale:     0,
+			matches:   []string{"0", "1", "12345", "-12345"},
+			rejects:   []string{"123456", "1.5", "01", "+1", "1e2", ""},
+		},
+		{
+			name:      "scale 4 mixed",
+			precision: 18,
+			scale:     4,
+			matches:   []string{"0.0000", "1.5000", "-12345.6789", "12345678901234.5678"},
+			rejects:   []string{"1.5", "1.50000", "1", "01.5000", "+1.5000", "1.5e2", ""},
+		},
+		{
+			name:      "scale equals precision fractional only",
+			precision: 4,
+			scale:     4,
+			matches:   []string{"0.0000", "0.1234", "-0.9999"},
+			rejects:   []string{"1.0000", "0.123", "0.12345", "1"},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			c := schema.Common{
+				Name:    "amount",
+				Type:    schema.Decimal,
+				Logical: &schema.LogicalParams{Decimal: &schema.DecimalParams{Precision: tc.precision, Scale: tc.scale}},
+			}
+			got := jsonSchemaUnmarshal(t, c)
+			assert.Equal(t, "string", got["type"])
+			pattern, ok := got["pattern"].(string)
+			require.True(t, ok, "pattern field must be a string")
+
+			re := regexp.MustCompile(pattern)
+			for _, ok := range tc.matches {
+				assert.True(t, re.MatchString(ok), "pattern %q should match %q", pattern, ok)
+			}
+			for _, bad := range tc.rejects {
+				assert.False(t, re.MatchString(bad), "pattern %q should NOT match %q", pattern, bad)
+			}
+		})
+	}
+}
+
+func TestCommonToJSONSchemaDecimalMissingLogical(t *testing.T) {
+	_, err := commonToJSONSchema(schema.Common{Name: "amount", Type: schema.Decimal})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "missing precision/scale")
+}
+
+func TestCommonToJSONSchemaBigDecimal(t *testing.T) {
+	c := schema.Common{Name: "amount", Type: schema.BigDecimal}
+	got := jsonSchemaUnmarshal(t, c)
+	assert.Equal(t, "string", got["type"])
+	pattern, ok := got["pattern"].(string)
+	require.True(t, ok)
+	re := regexp.MustCompile(pattern)
+	for _, s := range []string{"0", "1.5", "-1.5", "12345.67890", "1"} {
+		assert.True(t, re.MatchString(s), "permissive pattern should match %q", s)
+	}
+	for _, s := range []string{"1.5e2", "+1.5", "01", "", "1.2.3"} {
+		assert.False(t, re.MatchString(s), "permissive pattern should NOT match %q", s)
+	}
+}