T1.33 fix 4: the declared type set by describe, inference's lines kept

The marked big integers' declared types are now set on the described
schema in describe_source, after assembly, from the catalog columns and
the preview it already holds, so sdm-schema's inference is unchanged and
the family maps' line citations into it hold.

Signed-off-by: Adi Wadaskar <awadaskar@nvidia.com>

Author: adi-kumo

rust/crates/sdm-core/src/describe.rs

@@ -110,6 +110,8 @@ pub(crate) async fn run(
     };
 
     let mut tables = Vec::with_capacity(names.len());
+    // Each table's marked big integers' declared types (big_integer_types).
+    let mut big_types = Vec::with_capacity(names.len());
     for table in &names {
         let one = || vec![table.source.clone()];
         let asked = describe_all(
@@ -176,6 +178,7 @@ pub(crate) async fn run(
         }
         // A big integer's tier, the whole column's (P1B-43).
         let preview = whole_column_tiers(&asker, &table.source, preview).await?;
+        big_types.push(big_integer_types(&preview, &declared));
         let previewed = infer::preview(
             &table.name,
             &table.source,
@@ -290,6 +293,14 @@ pub(crate) async fn run(
     for warning in warnings {
         asker.warn(warning);
     }
+    for (table, types) in description.tables.iter_mut().zip(big_types) {
+        for column in &mut table.columns {
+            column.declared_type = types
+                .iter()
+                .find(|(name, _)| *name == column.name)
+                .map(|(_, declared)| declared.clone());
+        }
+    }
     // Each table's repeated keys, as the reference's count sets them.
     if let Some(counted) = counted {
         for (table, repeats) in description.tables.iter_mut().zip(counted) {
@@ -320,6 +331,25 @@ fn falls_back(failed: &CoreError) -> bool {
     }
 }
 
+/// The catalog's declared type of each big-integer column the host layer
+/// read from an opaque type and marked ([`crate::big_integer`]), by name:
+/// what the described schema keeps (`DescribedColumn::declared_type`), so a
+/// `decimal128(38, 0)` seed is joined to such a key in the key's own type
+/// with no statement `materialize` adds (register P1B-43). Every other
+/// column keeps none, and joins as before part 3.
+fn big_integer_types(preview: &ArrowData, declared: &[CatalogColumn]) -> Vec<(String, String)> {
+    preview
+        .schema
+        .fields()
+        .iter()
+        .filter(|f| big_integer::is_marked(f))
+        .filter_map(|f| {
+            let column = declared.iter().find(|c| &c.name == f.name())?;
+            Some((column.name.clone(), column.declared_type.clone()))
+        })
+        .collect()
+}
+
 /// `preview` with each big-integer column the host layer read
 /// ([`crate::big_integer`]) in the tier of its values over the whole
 /// table, so the column is described in one tier every fetch is held to:

rust/crates/sdm-schema/src/assemble.rs

@@ -241,7 +241,7 @@ pub fn assemble(
                     stype_before_key: c.stype_before_key,
                     nullable: c.nullable,
                     unique: c.unique,
-                    declared_type: c.declared_type,
+                    declared_type: None,
                 })
                 .collect(),
             primary_key: table.primary_key.into_iter().collect(),

rust/crates/sdm-schema/src/infer.rs

@@ -19,7 +19,6 @@ use std::collections::BTreeSet;
 
 use arrow_array::RecordBatch;
 use arrow_schema::{DataType, Schema};
-use sdm_contract::data::{BIG_INTEGER_MARKED, BIG_INTEGER_MARKER};
 use sdm_contract::ops::{Dtype, Stype};
 use sdm_contract::{ColumnType, CoreError, ErrorCode, ErrorKind, Scalar};
 
@@ -77,9 +76,6 @@ pub struct InferredColumn {
     /// Whether the catalog declares a primary or unique key of this one
     /// column.
     pub unique: bool,
-    /// The catalog's declared type, for a big-integer column the host layer
-    /// read and marked only (`DescribedColumn::declared_type`).
-    pub declared_type: Option<String>,
 }
 
 /// Why inference passed a timestamp over for the time column, as the
@@ -488,9 +484,6 @@ pub fn preview(
             received: ColumnType::from_arrow(&declared.name, field.data_type())?,
             nullable: !is_primary && !declared.not_null,
             unique: declared_unique.contains(&declared.name),
-            declared_type: (field.metadata().get(BIG_INTEGER_MARKER).map(String::as_str)
-                == Some(BIG_INTEGER_MARKED))
-            .then(|| declared.declared_type.clone()),
         });
         let text_time = dtype == Dtype::String && stype == Stype::Timestamp;
         instants.push(if dtype == Dtype::Time {

rust/crates/sdm-schema/tests/assemble.rs

@@ -36,7 +36,6 @@ fn column(name: &str, dtype: Dtype, stype: Stype) -> InferredColumn {
         },
         nullable: true,
         unique: false,
-        declared_type: None,
     }
 }
 

rust/crates/sdm-schema/tests/keys.rs

@@ -25,7 +25,6 @@ fn column(name: &str, unique: bool) -> InferredColumn {
         received: ColumnType::Int64,
         nullable: true,
         unique,
-        declared_type: None,
     }
 }