Show SQL names instead of primitive types in catalog

Define a mapping of primitive types to SQL types. Add a field 'sql_name'
to the SourceFieldType message, and populate it for primitive types.
Display this value in the catalog.

Resolves: #73

arroyo-api/src/sources.rs

@@ -196,6 +196,21 @@ pub enum PrimitiveType {
     UnixMillis,
 }
 
+fn primitive_to_sql(primitive_type: &PrimitiveType) -> &'static str {
+    match primitive_type {
+        PrimitiveType::Int32 => "INTEGER",
+        PrimitiveType::Int64 => "BIGINT",
+        PrimitiveType::UInt32 => "UNSIGNED INTEGER", // not really sql
+        PrimitiveType::UInt64 => "UNSIGNED BIGINT",  // not really sql
+        PrimitiveType::F32 => "FLOAT",
+        PrimitiveType::F64 => "DOUBLE",
+        PrimitiveType::Bool => "BOOLEAN",
+        PrimitiveType::String => "VARCHAR",
+        PrimitiveType::Bytes => "BLOB",
+        PrimitiveType::UnixMillis => "TIMESTAMP",
+    }
+}
+
 #[derive(Clone, Debug)]
 pub enum SchemaFieldType {
     Primitive(PrimitiveType),
@@ -329,7 +344,7 @@ impl TryFrom<SchemaField> for SourceField {
     type Error = String;
 
     fn try_from(value: SchemaField) -> Result<Self, Self::Error> {
-        let typ = match value.typ {
+        let typ = match value.clone().typ {
             SchemaFieldType::Primitive(p) => api::source_field_type::Type::Primitive(match p {
                 PrimitiveType::Int32 => api::PrimitiveType::Int32,
                 PrimitiveType::Int64 => api::PrimitiveType::Int64,
@@ -353,9 +368,17 @@ impl TryFrom<SchemaField> for SourceField {
             }
         };
 
+        let sql_name = match value.typ {
+            SchemaFieldType::Primitive(p) => primitive_to_sql(&p),
+            _ => ""
+        };
+
         Ok(SourceField {
             field_name: value.name,
-            field_type: Some(api::SourceFieldType { r#type: Some(typ) }),
+            field_type: Some(api::SourceFieldType {
+                r#type: Some(typ),
+                sql_name: String::from(sql_name)
+            }),
             nullable: value.nullable,
         })
     }