cast int64 and numeric to string by default

crates/configuration/src/version3/mod.rs

@@ -106,7 +106,9 @@ pub async fn introspect(
                 .introspection_options
                 .introspect_prefix_function_comparison_operators,
         )
-        .bind(serde_json::to_value(base_type_representations().0)?);
+        .bind(serde_json::to_value(
+            base_type_representations(args.metadata.type_representations).0,
+        )?);
 
     let row = connection
         .fetch_one(query)
@@ -195,8 +197,13 @@ pub async fn introspect(
     })
 }
 
-fn base_type_representations() -> database::TypeRepresentations {
-    database::TypeRepresentations(
+/// Merge the type representations currenting defined in the user's configuration with
+/// our base defaults. User configuration takes precedence.
+fn base_type_representations(
+    database::TypeRepresentations(existing_type_representations): database::TypeRepresentations,
+) -> database::TypeRepresentations {
+    // Start with the default type representations
+    let mut type_representations: database::TypeRepresentations = database::TypeRepresentations(
         [
             // Bit strings:
             //   https://www.postgresql.org/docs/current/datatype-bit.html
@@ -246,7 +253,7 @@ fn base_type_representations() -> database::TypeRepresentations {
                 // This is not what we do now and is a breaking change.
                 // This will need to be changed in the future. In the meantime, we report
                 // The type representation to be json.
-                database::TypeRepresentation::Json,
+                database::TypeRepresentation::Int64AsString,
             ),
             (
                 database::ScalarType("numeric".to_string()),
@@ -282,7 +289,14 @@ fn base_type_representations() -> database::TypeRepresentations {
             ),
         ]
         .into(),
-    )
+    );
+    // If the user already has existing type representations defined,
+    // override the default ones using `insert`.
+    // We do this to not change the behaviour for the user on update.
+    for (typ, type_rep) in existing_type_representations {
+        type_representations.0.insert(typ, type_rep);
+    }
+    type_representations
 }
 
 /// Collect all the composite types that can occur in the metadata.

crates/connectors/ndc-postgres/src/schema.rs

@@ -626,8 +626,14 @@ fn map_type_representation(
         metadata::TypeRepresentation::Float64 => models::TypeRepresentation::Float64,
         metadata::TypeRepresentation::Int16 => models::TypeRepresentation::Int16,
         metadata::TypeRepresentation::Int32 => models::TypeRepresentation::Int32,
-        metadata::TypeRepresentation::Int64 => models::TypeRepresentation::Int64,
-        metadata::TypeRepresentation::BigDecimal => models::TypeRepresentation::BigDecimal,
+        // Int64 returns a number.
+        metadata::TypeRepresentation::Int64 => models::TypeRepresentation::JSON,
+        // Int64AsString returns a string.
+        metadata::TypeRepresentation::Int64AsString => models::TypeRepresentation::Int64,
+        // BigDecimal returns a number.
+        metadata::TypeRepresentation::BigDecimal => models::TypeRepresentation::JSON,
+        // BigDecimalAsString returns a string.
+        metadata::TypeRepresentation::BigDecimalAsString => models::TypeRepresentation::BigDecimal,
         metadata::TypeRepresentation::Timestamp => models::TypeRepresentation::Timestamp,
         metadata::TypeRepresentation::Timestamptz => models::TypeRepresentation::TimestampTZ,
         metadata::TypeRepresentation::Time => models::TypeRepresentation::String,

crates/query-engine/metadata/src/metadata/database.rs

@@ -23,6 +23,16 @@ pub enum Type {
     ArrayType(Box<Type>),
 }
 
+impl Type {
+    pub fn scalar_type(&self) -> Option<&ScalarType> {
+        match self {
+            Type::ScalarType(scalar_type) => Some(scalar_type),
+            Type::CompositeType(_) => None,
+            Type::ArrayType(_) => None,
+        }
+    }
+}
+
 /// Information about a composite type. These are very similar to tables, but with the crucial
 /// difference that composite types do not support constraints (such as NOT NULL).
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
@@ -225,10 +235,14 @@ pub enum TypeRepresentation {
     Int16,
     /// int4
     Int32,
-    /// int8
+    /// int8 as integer
     Int64,
+    /// int8 as string
+    Int64AsString,
     /// numeric
     BigDecimal,
+    /// numeric as string
+    BigDecimalAsString,
     /// timestamp
     Timestamp,
     /// timestamp with timezone

crates/query-engine/translation/src/translation/helpers.rs

@@ -25,6 +25,7 @@ pub struct State {
 }
 
 #[derive(Debug)]
+/// Used for generating a unique name for intermediate tables.
 pub struct TableAliasIndex(pub u64);
 
 #[derive(Debug)]
@@ -89,7 +90,7 @@ pub enum CollectionInfo<'env> {
 }
 
 #[derive(Debug)]
-/// Metadata information about a specific collection.
+/// Metadata information about a specific collection or composite type.
 pub enum CompositeTypeInfo<'env> {
     CollectionInfo(CollectionInfo<'env>),
     CompositeTypeInfo {
@@ -114,8 +115,7 @@ impl<'request> Env<'request> {
         }
     }
 
-    /// Lookup a collection's information in the metadata.
-
+    /// Lookup collection or composite type.
     pub fn lookup_composite_type(
         &self,
         type_name: &'request str,
@@ -138,6 +138,7 @@ impl<'request> Env<'request> {
         }
     }
 
+    /// Lookup a collection's information in the metadata.
     pub fn lookup_collection(
         &self,
         collection_name: &'request str,
@@ -202,6 +203,18 @@ impl<'request> Env<'request> {
             })
     }
 
+    /// Lookup type representation of a type.
+    pub fn lookup_type_representation(
+        &self,
+        scalar_type: &metadata::ScalarType,
+    ) -> Option<metadata::TypeRepresentation> {
+        self.metadata
+            .type_representations
+            .0
+            .get(scalar_type)
+            .cloned()
+    }
+
     /// Try to get the variables table reference. This will fail if no variables were passed
     /// as part of the query request.
     pub fn get_variables_table(&self) -> Result<sql::ast::TableReference, Error> {
@@ -262,6 +275,29 @@ impl CompositeTypeInfo<'_> {
                 )),
         }
     }
+
+    pub fn fields(&self) -> Vec<(&String, &String)> {
+        match self {
+            CompositeTypeInfo::CompositeTypeInfo { name: _, info } => info
+                .fields
+                .iter()
+                .map(|(name, field)| (name, &field.name))
+                .collect::<Vec<_>>(),
+
+            CompositeTypeInfo::CollectionInfo(collection_info) => match collection_info {
+                CollectionInfo::Table { name: _, info } => info
+                    .columns
+                    .iter()
+                    .map(|(name, column)| (name, &column.name))
+                    .collect::<Vec<_>>(),
+                CollectionInfo::NativeQuery { name: _, info } => info
+                    .columns
+                    .iter()
+                    .map(|(name, column)| (name, &column.name))
+                    .collect::<Vec<_>>(),
+            },
+        }
+    }
 }
 
 impl Default for State {