dbt-adapter: Hide the dependency on try_format_type behind a dyn trait (#5783)

* dbt-adapter: Hide the dependency on try_format_type behind a dyn trait

This means concrete adapter implementations don't depend directly on
`sdf_frontend` for the type formatting function anymore.

- Introduce the `TypeFormatter` trait to dbt-fusion-adapter and
  provide an implementation for it in dbt-adapter
- Inject the `TypeFormatter` from the factory implementation all the way
  to the `SqlEngine`
- Remove `convert_type_inner` and implement `convert_type` fully by
  delegating to the `TypeFormatter` trait
- Introduce the `ReplayAdapter` trait that is implemented by
  `DbtReplayAdapter`
- Change `is_replay() -> bool` to `as_replay() -> Option<&dyn ReplayAdapter>`
- Add a `NaiveTypeFormatterImpl` so it can be passed in the construction
  of `ParseAdapter`

* cargo fmt

GitOrigin-RevId: 8faadd690d6f037501267bcbab834c15a1bd47f3

Author: felipecrv

crates/dbt-fusion-adapter/src/bridge_adapter.rs

@@ -500,7 +500,8 @@ impl BaseAdapter for BridgeAdapter {
     #[tracing::instrument(skip(self, state), level = "trace")]
     fn rename_relation(&self, state: &State, args: &[Value]) -> Result<Value, MinijinjaError> {
         self.cache_renamed(state, args)?;
-        if self.typed_adapter.is_replay() {
+        if self.typed_adapter.as_replay().is_some() {
+            // TODO: move this logic to the [ReplayAdapter]
             let iter = ArgsIter::new(
                 current_function_name!(),
                 &["from_relation", "to_relation"],
@@ -714,7 +715,8 @@ impl BaseAdapter for BridgeAdapter {
         identifier: &str,
     ) -> Result<Value, MinijinjaError> {
         // Skip cache in replay mode
-        if !self.typed_adapter.is_replay() {
+        let is_replay = self.typed_adapter.as_replay().is_some();
+        if !is_replay {
             let temp_relation = relation_object::create_relation(
                 self.typed_adapter.adapter_type(),
                 database.to_string(),
@@ -811,7 +813,8 @@ impl BaseAdapter for BridgeAdapter {
         let relation = parser.get::<Value>("relation")?;
         let relation = downcast_value_to_dyn_base_relation(&relation)?;
 
-        if self.typed_adapter.is_replay() {
+        if self.typed_adapter.as_replay().is_some() {
+            // TODO: move this logic to the [ReplayAdapter]
             return match self.typed_adapter.get_columns_in_relation(state, relation) {
                 Ok(result) => Ok(Value::from(result)),
                 Err(e) => Err(MinijinjaError::new(
@@ -899,7 +902,8 @@ impl BaseAdapter for BridgeAdapter {
         let schema = parser.get::<String>("schema")?;
 
         // Replay fast-path: consult trace-derived cache if available
-        if self.typed_adapter.is_replay() {
+        if self.typed_adapter.as_replay().is_some() {
+            // TODO: move this logic to the [ReplayAdapter]
             if let Some(exists) = self
                 .typed_adapter
                 .schema_exists_from_trace(&database, &schema)

crates/dbt-fusion-adapter/src/parse/adapter.rs

@@ -9,6 +9,7 @@ use crate::parse::relation::EmptyRelation;
 use crate::query_comment::QueryCommentConfig;
 use crate::relation_object::{RelationObject, create_relation};
 use crate::response::AdapterResponse;
+use crate::sql_types::TypeFormatter;
 use crate::stmt_splitter::NaiveStmtSplitter;
 use crate::typed_adapter::TypedBaseAdapter;
 use crate::{AdapterResult, SqlEngine};
@@ -98,6 +99,7 @@ impl ParseAdapter {
         adapter_type: AdapterType,
         config: dbt_serde_yaml::Mapping,
         package_quoting: DbtQuoting,
+        type_formatter: Box<dyn TypeFormatter>,
         token: CancellationToken,
     ) -> Self {
         let backend = backend_of(adapter_type);
@@ -114,6 +116,7 @@ impl ParseAdapter {
             adapter_factory,
             stmt_splitter,
             query_comment,
+            type_formatter,
             token.clone(),
         );
 

crates/dbt-fusion-adapter/src/record_and_replay.rs

@@ -3,6 +3,7 @@ use crate::config::AdapterConfig;
 use crate::errors::AdapterResult;
 use crate::query_comment::QueryCommentConfig;
 use crate::sql_engine::SqlEngine;
+use crate::sql_types::TypeFormatter;
 use crate::stmt_splitter::StmtSplitter;
 
 use adbc_core::error::{Error as AdbcError, Result as AdbcResult, Status as AdbcStatus};
@@ -134,6 +135,10 @@ impl RecordEngine {
         self.0.engine.query_comment()
     }
 
+    pub(crate) fn type_formatter(&self) -> &dyn TypeFormatter {
+        self.0.engine.type_formatter()
+    }
+
     pub fn cancellation_token(&self) -> CancellationToken {
         self.0.engine.cancellation_token()
     }
@@ -368,6 +373,7 @@ struct ReplayEngineInner {
     adapter_factory: Arc<dyn AdapterFactory>,
     stmt_splitter: Arc<dyn StmtSplitter>,
     query_comment: QueryCommentConfig,
+    type_formatter: Box<dyn TypeFormatter>,
     /// Global CLI cancellation token
     cancellation_token: CancellationToken,
 }
@@ -382,13 +388,15 @@ impl ReplayEngineInner {
 pub struct ReplayEngine(Arc<ReplayEngineInner>);
 
 impl ReplayEngine {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         backend: Backend,
         path: PathBuf,
         config: AdapterConfig,
         adapter_factory: Arc<dyn AdapterFactory>,
         stmt_splitter: Arc<dyn StmtSplitter>,
         query_comment: QueryCommentConfig,
+        type_formatter: Box<dyn TypeFormatter>,
         token: CancellationToken,
     ) -> Self {
         let inner = ReplayEngineInner {
@@ -398,6 +406,7 @@ impl ReplayEngine {
             adapter_factory,
             stmt_splitter,
             query_comment,
+            type_formatter,
             cancellation_token: token,
         };
         ReplayEngine(Arc::new(inner))
@@ -428,6 +437,10 @@ impl ReplayEngine {
         &self.0.query_comment
     }
 
+    pub(crate) fn type_formatter(&self) -> &dyn TypeFormatter {
+        self.0.type_formatter.as_ref()
+    }
+
     pub fn cancellation_token(&self) -> CancellationToken {
         self.0.cancellation_token.clone()
     }

crates/dbt-fusion-adapter/src/sql_engine.rs

@@ -4,6 +4,7 @@ use crate::base_adapter::{AdapterFactory, backend_of};
 use crate::config::AdapterConfig;
 use crate::errors::{AdapterError, AdapterErrorKind, AdapterResult};
 use crate::query_comment::{EMPTY_CONFIG, QueryCommentConfig};
+use crate::sql_types::{NaiveTypeFormatterImpl, TypeFormatter};
 use crate::stmt_splitter::StmtSplitter;
 
 use adbc_core::options::{OptionStatement, OptionValue};
@@ -42,6 +43,13 @@ type Options = Vec<(String, OptionValue)>;
 static NAIVE_STMT_SPLITTER: LazyLock<Arc<dyn StmtSplitter>> =
     LazyLock::new(|| Arc::new(crate::stmt_splitter::NaiveStmtSplitter));
 
+/// Naive type formatter used in the MockAdapter
+///
+/// IMPORTANT: not suitable for production use. DEFAULTS TO SNOWFLAKE ALSO.
+/// TODO: remove when the full formatter is available to this crate.
+static NAIVE_TYPE_FORMATTER: LazyLock<Box<dyn TypeFormatter>> =
+    LazyLock::new(|| Box::new(NaiveTypeFormatterImpl::new(AdapterType::Snowflake)));
+
 #[derive(Default)]
 struct IdentityHasher {
     hash: u64,
@@ -96,6 +104,8 @@ pub struct ActualEngine {
     splitter: Arc<dyn StmtSplitter>,
     /// Query comment config
     query_comment: QueryCommentConfig,
+    /// Type formatter for the dilect this engine is for
+    pub type_formatter: Box<dyn TypeFormatter>,
     /// Global CLI cancellation token
     cancellation_token: CancellationToken,
 }
@@ -107,6 +117,7 @@ impl ActualEngine {
         adapter_factory: Arc<dyn AdapterFactory>,
         splitter: Arc<dyn StmtSplitter>,
         query_comment: QueryCommentConfig,
+        type_formatter: Box<dyn TypeFormatter>,
         token: CancellationToken,
     ) -> Self {
         let threads = config
@@ -127,7 +138,9 @@ impl ActualEngine {
             semaphore: Arc::new(Semaphore::new(permits)),
             adapter_factory,
             splitter,
+            type_formatter,
             query_comment,
+
             cancellation_token: token,
         }
     }
@@ -216,6 +229,7 @@ impl SqlEngine {
         adapter_factory: Arc<dyn AdapterFactory>,
         stmt_splitter: Arc<dyn StmtSplitter>,
         query_comment: QueryCommentConfig,
+        type_formatter: Box<dyn TypeFormatter>,
         token: CancellationToken,
     ) -> Arc<Self> {
         let engine = ActualEngine::new(
@@ -224,19 +238,22 @@ impl SqlEngine {
             adapter_factory,
             stmt_splitter,
             query_comment,
+            type_formatter,
             token,
         );
         Arc::new(SqlEngine::Warehouse(Arc::new(engine)))
     }
 
     /// Create a new [`SqlEngine::Replay`] based on the given path and adapter type.
+    #[allow(clippy::too_many_arguments)]
     pub fn new_for_replaying(
         backend: Backend,
         path: PathBuf,
         config: AdapterConfig,
         adapter_factory: Arc<dyn AdapterFactory>,
         stmt_splitter: Arc<dyn StmtSplitter>,
         query_comment: QueryCommentConfig,
+        type_formatter: Box<dyn TypeFormatter>,
         token: CancellationToken,
     ) -> Arc<Self> {
         let engine = ReplayEngine::new(
@@ -246,6 +263,7 @@ impl SqlEngine {
             adapter_factory,
             stmt_splitter,
             query_comment,
+            type_formatter,
             token,
         );
         Arc::new(SqlEngine::Replay(engine))
@@ -267,6 +285,15 @@ impl SqlEngine {
         }
     }
 
+    pub fn type_formatter(&self) -> &dyn TypeFormatter {
+        match self {
+            SqlEngine::Warehouse(engine) => engine.type_formatter.as_ref(),
+            SqlEngine::Record(engine) => engine.type_formatter(),
+            SqlEngine::Replay(engine) => engine.type_formatter(),
+            SqlEngine::Mock(_adapter_type) => NAIVE_TYPE_FORMATTER.as_ref(),
+        }
+    }
+
     /// Split SQL statements using the provided dialect
     ///
     /// This method handles the splitting of SQL statements based on the dialect's rules.

crates/dbt-fusion-adapter/src/sql_types.rs

@@ -1,8 +1,59 @@
 use std::borrow::Cow;
 
+use crate::AdapterResult;
+use crate::base_adapter::backend_of;
 use crate::errors::{AdapterError, AdapterErrorKind};
 use arrow_schema::DataType;
 use dbt_common::adapter::AdapterType;
+use dbt_xdbc::sql::types::SqlType;
+
+pub trait TypeFormatter: Send + Sync {
+    /// Picks a SQL type for a given Arrow DataType and renders it as SQL.
+    ///
+    /// The implementation is dialect-specific.
+    fn format_arrow_type_as_sql(&self, data_type: &DataType, out: &mut String)
+    -> AdapterResult<()>;
+
+    /// Renders a given SqlType as SQL.
+    ///
+    /// The implementation is dialect-specific.
+    fn format_sql_type(&self, sql_type: SqlType, out: &mut String) -> AdapterResult<()>;
+}
+
+pub struct NaiveTypeFormatterImpl(AdapterType, dbt_xdbc::Backend);
+
+impl NaiveTypeFormatterImpl {
+    pub fn new(adapter_type: AdapterType) -> Self {
+        let backend = backend_of(adapter_type);
+        Self(adapter_type, backend)
+    }
+}
+
+impl TypeFormatter for NaiveTypeFormatterImpl {
+    fn format_arrow_type_as_sql(
+        &self,
+        data_type: &DataType,
+        out: &mut String,
+    ) -> AdapterResult<()> {
+        let adapter_type = self.0;
+        let hint: SqlTypeHint = data_type.try_into()?;
+        // TODO: handle has_decimal_places correctly
+        let has_decimal_places = false;
+        let res = sql_type_hint_to_str(hint, has_decimal_places, adapter_type);
+        out.push_str(res.as_ref());
+        Ok(())
+    }
+
+    fn format_sql_type(&self, sql_type: SqlType, out: &mut String) -> AdapterResult<()> {
+        let backend = self.1;
+        sql_type.write(backend, out).map_err(|e| {
+            AdapterError::new(
+                AdapterErrorKind::NotSupported,
+                format!("Failed to convert SQL type {sql_type:?}. Error: {e}"),
+            )
+        })
+    }
+}
 
 pub enum SqlTypeHint {
     Integer,

crates/dbt-fusion-adapter/src/typed_adapter.rs

@@ -364,15 +364,16 @@ pub trait TypedBaseAdapter: fmt::Debug + Send + Sync + AdapterTyping {
         }
     }
 
-    fn convert_type_inner(&self, _state: &State, data_type: &DataType) -> AdapterResult<String>;
-
     /// Convert type.
     fn convert_type(
         &self,
         state: &State,
         table: Arc<AgateTable>,
         col_idx: i64,
     ) -> AdapterResult<String> {
+        // XXX: Core uses the flattened agate table types. Here we use the original arrow
+        // schema containing the original table types including nested types. This might
+        // be what Core developers expected to get from Python agate types as well. (?)
         let schema = table.original_record_batch().schema();
         let data_type = schema.field(col_idx as usize).data_type();
 
@@ -387,7 +388,16 @@ pub trait TypedBaseAdapter: fmt::Debug + Send + Sync + AdapterTyping {
             data_type
         };
 
-        self.convert_type_inner(state, data_type)
+        if let Some(replay_adapter) = self.as_replay() {
+            // XXX: isn't the point of replay adapter to compare what it does against the actual code?
+            return replay_adapter.replay_convert_type(state, data_type);
+        }
+
+        let mut out = String::new();
+        self.engine()
+            .type_formatter()
+            .format_arrow_type_as_sql(data_type, &mut out)?;
+        Ok(out)
     }
 
     /// Expand the to_relation table's column types to match the schema of from_relation
@@ -948,9 +958,9 @@ pub trait TypedBaseAdapter: fmt::Debug + Send + Sync + AdapterTyping {
         }
     }
 
-    /// Convenience to check if this [TypedBaseAdapter] implementer is used for replaying recordings
-    fn is_replay(&self) -> bool {
-        false
+    /// This adapter as the replay adapter if it is one, None otherwise.
+    fn as_replay(&self) -> Option<&dyn ReplayAdapter> {
+        None
     }
 
     /// Optional fast-path for replay adapters: return schema existence from the trace
@@ -959,3 +969,10 @@ pub trait TypedBaseAdapter: fmt::Debug + Send + Sync + AdapterTyping {
         None
     }
 }
+
+/// Abstract interface for the concrete replay adapter implementation.
+///
+/// NOTE: this is a growing interface that is currently growing.
+pub trait ReplayAdapter: TypedBaseAdapter {
+    fn replay_convert_type(&self, state: &State, data_type: &DataType) -> AdapterResult<String>;
+}

crates/dbt-jinja-utils/src/environment_builder.rs

@@ -490,6 +490,7 @@ mod tests {
     use dbt_common::adapter::AdapterType;
     use dbt_common::cancellation::never_cancels;
     use dbt_fusion_adapter::ParseAdapter;
+    use dbt_fusion_adapter::sql_types::NaiveTypeFormatterImpl;
     use dbt_schemas::schemas::relations::DEFAULT_DBT_QUOTING;
     use minijinja::{
         constants::MACRO_DISPATCH_ORDER, context, dispatch_object::THREAD_LOCAL_DEPENDENCIES,
@@ -603,6 +604,7 @@ all okay!");
             AdapterType::Postgres,
             dbt_serde_yaml::Mapping::default(),
             DEFAULT_DBT_QUOTING,
+            Box::new(NaiveTypeFormatterImpl::new(AdapterType::Postgres)),
             never_cancels(),
         );
         let builder: JinjaEnvBuilder = JinjaEnvBuilder::new()
@@ -689,6 +691,7 @@ all okay!");
             AdapterType::Postgres,
             dbt_serde_yaml::Mapping::default(),
             DEFAULT_DBT_QUOTING,
+            Box::new(NaiveTypeFormatterImpl::new(AdapterType::Postgres)),
             never_cancels(),
         );
         let builder: JinjaEnvBuilder = JinjaEnvBuilder::new()
@@ -750,6 +753,7 @@ all okay!");
             AdapterType::Postgres,
             dbt_serde_yaml::Mapping::default(),
             DEFAULT_DBT_QUOTING,
+            Box::new(NaiveTypeFormatterImpl::new(AdapterType::Postgres)),
             never_cancels(),
         );
         let env = JinjaEnvBuilder::new()
@@ -864,6 +868,7 @@ all okay!");
             AdapterType::Postgres,
             dbt_serde_yaml::Mapping::default(),
             DEFAULT_DBT_QUOTING,
+            Box::new(NaiveTypeFormatterImpl::new(AdapterType::Postgres)),
             never_cancels(),
         );