controller: DfType for custom types' default value

If a postgres table's column has a custom type (which includes enums),
Readyset does not resolve that custom type when determining the
column's default value at lowering from MIR to Dataflow. This causes a
failure when deriving the column's default value when the column is
attributed as either having a default or being not null.

This CL passes the known `custom_types` to the existing call to
`DfType::from_sql_type()`.

Addresses: REA-5746
Release-Note-Core: Fix that allows snapshotting postgres tables where
  a column of a custom type is attributed NOT NULL but has no default
  value.
Change-Id: I4e37e44bbe0b56b120ccfdb7d65e702e5d1afd8b
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9954
Reviewed-by: Michael Zink <michael.z@readyset.io>
Tested-by: Buildkite CI

Author: jasobrown-rs

readyset-server/src/controller/mir_to_flow.rs

@@ -380,16 +380,23 @@ fn make_base_node(
                 let df: DfValue = dv
                     .try_into_dialect(mig.dialect.into())
                     .map_err(|e| internal_err!("Failed to convert default value: {}", e))?;
-                let dftype_to =
-                    DfType::from_sql_type(&cs.sql_type, mig.dialect, |_| None, collation).map_err(
-                        |e| internal_err!("Failed to convert SQL type to DfType: {}", e),
-                    )?;
+                let dftype_to = DfType::from_sql_type(
+                    &cs.sql_type,
+                    mig.dialect,
+                    |ty| custom_types.get(&ty).cloned(),
+                    collation,
+                )
+                .map_err(|e| internal_err!("Failed to convert SQL type to DfType: {}", e))?;
                 return df.coerce_to(&dftype_to, &df.infer_dataflow_type());
-            } else if cs.is_not_null() {
-                let dftype_to =
-                    DfType::from_sql_type(&cs.sql_type, mig.dialect, |_| None, collation).map_err(
-                        |e| internal_err!("Failed to convert SQL type to DfType: {}", e),
-                    )?;
+            } else if cs.is_not_null() && mig.dialect == readyset_sql::Dialect::MySQL.into() {
+                // this branch is really only to support mysql minimal row-based replication.
+                let dftype_to = DfType::from_sql_type(
+                    &cs.sql_type,
+                    mig.dialect,
+                    |ty| custom_types.get(&ty).cloned(),
+                    collation,
+                )
+                .map_err(|e| internal_err!("Failed to convert SQL type to DfType: {}", e))?;
 
                 return Ok(DfValue::implicit_default(
                     collation.unwrap_or(Collation::Utf8),