[SS-193] Fix issue where copy into doesn't respect target table's numeric precision (#37276)

### Motivation

Dennis filed an issue flagging a test failure in the iceberg sink where
we failed to output the decimal value "62912915282.360725" to an iceberg
sink where we expected precision of 3. The issue appears to be upstream,
where the data being sinked doesn't match the schema.

For parallel_workload it appears we use `COPY`.

Issue
https://linear.app/materializeinc/issue/SS-193/iceberg-failed-to-convert-row-to-recordbatch-failed-to-add-insert-row

### Description

I added some tests for `COPY` behavior that failed initially and now
pass, covering default and CSV input via STDIN and Parquet. Generally
trying to mirror existing insert behavior that ultimately calls rescale.

### Verification

I've compared the results for the basic tab-separated `COPY` command to
postgres and confirmed that postgres rounds while materialize currently
doesn't.

I've added some testdrive tests and unit tests to prevent future
regressions and explore edge cases like numbers overflowing max
precision on rescale that wouldn't have before.

Note: There are a variety of other types that may have similar issues,
i.e. timestamp max precision, which I'm thinking I'll file a separate
follow-up issue on.

###  Potential Gotchas

The fivetran sink uses the constraints from `Type::Numeric`, so would
end up creating a `DECIMAL` with no options, see
https://github.com/MaterializeInc/materialize/blob/4229f003472695da7f0fb9c8c1f2aecc3fa8e39e/src/fivetran-destination/src/utils.rs#L94.
I think DECIMAL in the SDK maps to fivetran's BIGDECIMAL and so should
be more correct by default.

Author: peterdukelarsen

src/arrow-util/src/reader.rs

@@ -30,7 +30,7 @@ use mz_ore::cast::CastFrom;
 use mz_repr::adt::date::Date;
 use mz_repr::adt::interval::Interval;
 use mz_repr::adt::jsonb::JsonbPacker;
-use mz_repr::adt::numeric::Numeric;
+use mz_repr::adt::numeric::{Numeric, rescale};
 use mz_repr::adt::range::{Range, RangeLowerBound, RangeUpperBound};
 use mz_repr::adt::timestamp::CheckedTimestamp;
 use mz_repr::{Datum, RelationDesc, Row, RowPacker, SharedRow, SqlScalarType};
@@ -188,8 +188,7 @@ fn scalar_type_and_array_to_reader(
         (SqlScalarType::Float64, DataType::Float64) => {
             Ok(ColReader::Float64(downcast_array::<Float64Array>(array)))
         }
-        // TODO(cf3): Consider the max_scale for numeric.
-        (SqlScalarType::Numeric { .. }, DataType::Decimal128(precision, scale)) => {
+        (SqlScalarType::Numeric { max_scale }, DataType::Decimal128(precision, scale)) => {
             use num_traits::Pow;
 
             let base = Numeric::from(10);
@@ -209,10 +208,10 @@ fn scalar_type_and_array_to_reader(
                 array,
                 scale_factor,
                 precision,
+                destination_max_scale: (*max_scale).map(|s| s.into_u8()),
             })
         }
-        // TODO(cf3): Consider the max_scale for numeric.
-        (SqlScalarType::Numeric { .. }, DataType::Decimal256(precision, scale)) => {
+        (SqlScalarType::Numeric { max_scale }, DataType::Decimal256(precision, scale)) => {
             use num_traits::Pow;
 
             let base = Numeric::from(10);
@@ -232,6 +231,7 @@ fn scalar_type_and_array_to_reader(
                 array,
                 scale_factor,
                 precision,
+                destination_max_scale: (*max_scale).map(|s| s.into_u8()),
             })
         }
         (SqlScalarType::Bytes, DataType::Binary) => {
@@ -489,11 +489,13 @@ enum ColReader {
         array: Decimal128Array,
         scale_factor: Numeric,
         precision: usize,
+        destination_max_scale: Option<u8>,
     },
     Decimal256 {
         array: Decimal256Array,
         scale_factor: Numeric,
         precision: usize,
+        destination_max_scale: Option<u8>,
     },
 
     Binary(arrow::array::BinaryArray),
@@ -609,21 +611,31 @@ impl ColReader {
                 array,
                 scale_factor,
                 precision,
-            } => array.is_valid(idx).then(|| array.value(idx)).map(|x| {
-                // Create a Numeric from our i128 with precision.
-                let mut ctx = dec::Context::<Numeric>::default();
-                ctx.set_precision(*precision).expect("checked before");
-                let mut num = ctx.from_i128(x);
+                destination_max_scale,
+            } => array
+                .is_valid(idx)
+                .then(|| array.value(idx))
+                .map(|x| {
+                    // Create a Numeric from our i128 with precision.
+                    let mut ctx = dec::Context::<Numeric>::default();
+                    ctx.set_precision(*precision).expect("checked before");
+                    let mut num = ctx.from_i128(x);
+
+                    // Scale the number.
+                    ctx.div(&mut num, scale_factor);
 
-                // Scale the number.
-                ctx.div(&mut num, scale_factor);
+                    if let Some(destination_max_scale) = destination_max_scale {
+                        rescale(&mut num, *destination_max_scale)?;
+                    }
 
-                Datum::Numeric(OrderedDecimal(num))
-            }),
+                    Ok::<_, anyhow::Error>(Datum::Numeric(OrderedDecimal(num)))
+                })
+                .transpose()?,
             ColReader::Decimal256 {
                 array,
                 scale_factor,
                 precision,
+                destination_max_scale,
             } => array
                 .is_valid(idx)
                 .then(|| array.value(idx))
@@ -642,6 +654,10 @@ impl ColReader {
                     // Scale the number.
                     ctx.div(&mut num, scale_factor);
 
+                    if let Some(destination_max_scale) = destination_max_scale {
+                        rescale(&mut num, *destination_max_scale)?;
+                    }
+
                     Ok::<_, anyhow::Error>(Datum::Numeric(OrderedDecimal(num)))
                 })
                 .transpose()?,
@@ -1183,6 +1199,116 @@ mod tests {
         assert_eq!(num.0, Numeric::from(100000000.009f64));
     }
 
+    /// Regression test for SS-193: when the destination column declares a
+    /// `max_scale`, the reader should round the decoded value to that scale
+    /// rather than preserving the source file's scale.
+    #[mz_ore::test]
+    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
+    fn decimal_applies_destination_max_scale() {
+        use mz_repr::adt::numeric::NumericMaxScale;
+
+        // Destination column is numeric(10, 2): scale 2.
+        let desc = RelationDesc::builder()
+            .with_column(
+                "a",
+                SqlScalarType::Numeric {
+                    max_scale: Some(NumericMaxScale::try_from(2_i64).unwrap()),
+                }
+                .nullable(true),
+            )
+            .finish();
+
+        let expected = Numeric::from(10.45f64);
+
+        // The source files carry scale-3 values (10.447) that don't round
+        // evenly to the destination's scale 2.
+        let mut dec128 = arrow::array::Decimal128Builder::new();
+        dec128 = dec128.with_precision_and_scale(12, 3).unwrap();
+        dec128.append_value(10447);
+        let dec128 = dec128.finish();
+        #[allow(clippy::as_conversions)]
+        let batch128 = StructArray::from(vec![(
+            Arc::new(Field::new("a", dec128.data_type().clone(), true)),
+            Arc::new(dec128) as arrow::array::ArrayRef,
+        )]);
+
+        let reader = ArrowReader::new(&desc, batch128).unwrap();
+        let mut rnd_row = Row::default();
+        reader.read(0, &mut rnd_row).unwrap();
+        let num = rnd_row.into_element().unwrap_numeric();
+        assert_eq!(num.0, expected, "Decimal128 did not round to max_scale");
+
+        let mut dec256 = arrow::array::Decimal256Builder::new();
+        dec256 = dec256.with_precision_and_scale(12, 3).unwrap();
+        dec256.append_value(arrow::datatypes::i256::from(10447));
+        let dec256 = dec256.finish();
+        #[allow(clippy::as_conversions)]
+        let batch256 = StructArray::from(vec![(
+            Arc::new(Field::new("a", dec256.data_type().clone(), true)),
+            Arc::new(dec256) as arrow::array::ArrayRef,
+        )]);
+
+        let reader = ArrowReader::new(&desc, batch256).unwrap();
+        let mut rnd_row = Row::default();
+        reader.read(0, &mut rnd_row).unwrap();
+        let num = rnd_row.into_element().unwrap_numeric();
+        assert_eq!(num.0, expected, "Decimal256 did not round to max_scale");
+    }
+
+    #[mz_ore::test]
+    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
+    fn decimal_rescale_to_higher_scale_overflows() {
+        use mz_repr::adt::numeric::NumericMaxScale;
+
+        let value: i128 = 10_i128.pow(33) - 1;
+
+        let build_batch = || {
+            let mut dec128 = arrow::array::Decimal128Builder::new();
+            dec128 = dec128.with_precision_and_scale(38, 0).unwrap();
+            dec128.append_value(value);
+            let dec128 = dec128.finish();
+            #[allow(clippy::as_conversions)]
+            StructArray::from(vec![(
+                Arc::new(Field::new("a", dec128.data_type().clone(), true)),
+                Arc::new(dec128) as arrow::array::ArrayRef,
+            )])
+        };
+
+        let desc_with_scale = |scale: i64| {
+            RelationDesc::builder()
+                .with_column(
+                    "a",
+                    SqlScalarType::Numeric {
+                        max_scale: Some(NumericMaxScale::try_from(scale).unwrap()),
+                    }
+                    .nullable(true),
+                )
+                .finish()
+        };
+
+        // Test working case with supportable scale then broken case.
+        let reader = ArrowReader::new(&desc_with_scale(2), build_batch()).unwrap();
+        let mut row = Row::default();
+        reader
+            .read(0, &mut row)
+            .expect("value must decode at destination scale 2");
+        let num = row.into_element().unwrap_numeric();
+        let mut expected = Numeric::try_from(value).unwrap();
+        rescale(&mut expected, 2).unwrap();
+        assert_eq!(num.0, expected, "value did not rescale to scale 2");
+
+        let reader = ArrowReader::new(&desc_with_scale(8), build_batch()).unwrap();
+        let mut row = Row::default();
+        let err = reader
+            .read(0, &mut row)
+            .expect_err("value must overflow at destination scale 8");
+
+        assert!(
+            format!("{err:#}").contains("exceed maximum precision"),
+            "unexpected error: {err:#}",
+        );
+    }
+
     /// Regression test for database-issues#11330: when a Parquet file authored
     /// by an external engine encodes a discrete range in non-canonical form
     /// (e.g. `[1,10]` for `int4range`), the reader must canonicalize it to MZ's

src/environmentd/tests/testdata/http/post

@@ -331,7 +331,7 @@ http
 {"queries":[{"query":"select $1::decimal+2 as col","params":["nan"]}]}
 ----
 200 OK
-{"results":[{"tag":"SELECT 1","rows":[["NaN"]],"desc":{"columns":[{"name":"col","type_oid":1700,"type_len":-1,"type_mod":2555947}]},"notices":[]}]}
+{"results":[{"tag":"SELECT 1","rows":[["NaN"]],"desc":{"columns":[{"name":"col","type_oid":1700,"type_len":-1,"type_mod":-1}]},"notices":[]}]}
 
 # Null string value parameters
 http

src/pgrepr/src/types.rs

@@ -1257,12 +1257,9 @@ impl From<&SqlScalarType> for Type {
             },
             SqlScalarType::Uuid => Type::Uuid,
             SqlScalarType::Numeric { max_scale } => Type::Numeric {
-                constraints: Some(NumericConstraints {
+                constraints: max_scale.map(|max_scale| NumericConstraints {
                     max_precision: i32::from(NUMERIC_DATUM_MAX_PRECISION),
-                    max_scale: match max_scale {
-                        Some(max_scale) => i32::from(max_scale.into_u8()),
-                        None => i32::from(NUMERIC_DATUM_MAX_PRECISION),
-                    },
+                    max_scale: i32::from(max_scale.into_u8()),
                 }),
             },
             SqlScalarType::RegClass => Type::RegClass,

src/pgrepr/src/value.rs

@@ -13,6 +13,7 @@ use std::{io, str};
 
 use bytes::{BufMut, BytesMut};
 use chrono::{DateTime, NaiveDateTime, NaiveTime, Utc};
+use dec::OrderedDecimal;
 use itertools::Itertools;
 use mz_ore::cast::ReinterpretCast;
 use mz_pgrepr_consts::oid::TYPE_INT2_OID;
@@ -22,6 +23,7 @@ use mz_repr::adt::char;
 use mz_repr::adt::date::Date;
 use mz_repr::adt::jsonb::JsonbRef;
 use mz_repr::adt::mz_acl_item::{AclItem, MzAclItem};
+use mz_repr::adt::numeric::{self as mz_repr_numeric, NumericMaxScale, rescale};
 use mz_repr::adt::pg_legacy_name::NAME_MAX_BYTES;
 use mz_repr::adt::range::{Range, RangeInner};
 use mz_repr::adt::timestamp::CheckedTimestamp;
@@ -30,7 +32,7 @@ use mz_repr::{Datum, RowArena, RowPacker, RowRef, SqlRelationType, SqlScalarType
 use postgres_types::{FromSql, IsNull, ToSql, Type as PgType};
 use uuid::Uuid;
 
-use crate::types::{UINT2, UINT4, UINT8};
+use crate::types::{NumericConstraints, UINT2, UINT4, UINT8};
 use crate::value::error::IntoDatumError;
 use crate::{Interval, Jsonb, Numeric, Type, UInt2, UInt4, UInt8};
 
@@ -711,7 +713,10 @@ impl Value {
                 },
             )?),
             Type::Name => Value::Name(strconv::parse_pg_legacy_name(s)),
-            Type::Numeric { .. } => Value::Numeric(Numeric(strconv::parse_numeric(s)?)),
+            Type::Numeric { constraints } => Value::Numeric(Numeric(rescale_numeric(
+                strconv::parse_numeric(s)?,
+                constraints.as_ref(),
+            )?)),
             Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
                 Value::Oid(strconv::parse_oid(s)?)
             }
@@ -815,7 +820,10 @@ impl Value {
                 })?;
             }
             Type::Name => packer.push(Datum::String(&strconv::parse_pg_legacy_name(s))),
-            Type::Numeric { .. } => packer.push(Datum::Numeric(strconv::parse_numeric(s)?)),
+            Type::Numeric { constraints } => packer.push(Datum::Numeric(rescale_numeric(
+                strconv::parse_numeric(s)?,
+                constraints.as_ref(),
+            )?)),
             Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
                 packer.push(Datum::UInt32(strconv::parse_oid(s)?))
             }
@@ -887,7 +895,13 @@ impl Value {
                 }
                 Ok(Value::Name(s))
             }
-            Type::Numeric { .. } => Numeric::from_sql(ty.inner(), raw).map(Value::Numeric),
+            Type::Numeric { constraints } => {
+                let n = Numeric::from_sql(ty.inner(), raw)?;
+                Ok(Value::Numeric(Numeric(rescale_numeric(
+                    n.0,
+                    constraints.as_ref(),
+                )?)))
+            }
             Type::Oid | Type::RegClass | Type::RegProc | Type::RegType => {
                 u32::from_sql(ty.inner(), raw).map(Value::Oid)
             }
@@ -923,6 +937,20 @@ impl Value {
     }
 }
 
+/// Rescales `n` to the scale required by `constraints`, if any.
+fn rescale_numeric(
+    mut n: OrderedDecimal<mz_repr_numeric::Numeric>,
+    constraints: Option<&NumericConstraints>,
+) -> Result<OrderedDecimal<mz_repr_numeric::Numeric>, Box<dyn Error + Sync + Send>> {
+    if let Some(constraints) = constraints {
+        rescale(
+            &mut n.0,
+            NumericMaxScale::try_from(i64::from(constraints.max_scale()))?.into_u8(),
+        )?;
+    }
+    Ok(n)
+}
+
 fn encode_element(buf: &mut BytesMut, elem: Option<&Value>, ty: &Type) -> Result<(), io::Error> {
     match elem {
         None => buf.put_i32(-1),
@@ -1020,4 +1048,37 @@ mod tests {
             "invalid input syntax for type array: Specifying array lower bounds is not supported: \"[0:0]={t}\"".to_string()
         );
     }
+
+    /// Decoding a numeric must round it to the destination's declared scale,
+    /// and the text and binary paths must agree. `COPY ... FROM` relies on the
+    /// text/CSV side of this (SS-193); binary parameters rely on the binary
+    /// side. `COPY ... FORMAT BINARY` is unsupported, so the binary path is
+    /// exercised here rather than via mzcompose.
+    #[mz_ore::test]
+    #[cfg_attr(miri, ignore)] // numeric/decimal contexts unsupported under miri
+    fn decode_numeric_applies_destination_scale() {
+        // A `numeric(10, 2)` destination: scale 2.
+        let ty = Type::from(&SqlScalarType::Numeric {
+            max_scale: Some(NumericMaxScale::try_from(2_i64).unwrap()),
+        });
+        let expected = strconv::parse_numeric("10.45").unwrap();
+
+        // Encode an over-scale value (10.447, scale 3) to binary, then decode
+        // it back through the scale-2 type.
+        let input = Value::Numeric(Numeric(strconv::parse_numeric("10.447").unwrap()));
+        let mut buf = BytesMut::new();
+        input
+            .encode_binary(&ty, &mut buf)
+            .expect("encoding 10.447 as numeric must succeed");
+        let Value::Numeric(Numeric(binary)) = Value::decode_binary(&ty, &buf).unwrap() else {
+            panic!("decode_binary of a numeric must yield Value::Numeric");
+        };
+        assert_eq!(binary, expected, "binary decode did not rescale to scale 2");
+
+        // The text path must agree with the binary path.
+        let Value::Numeric(Numeric(text)) = Value::decode_text(&ty, b"10.447").unwrap() else {
+            panic!("decode_text of a numeric must yield Value::Numeric");
+        };
+        assert_eq!(text, expected, "text decode did not rescale to scale 2");
+    }
 }