How to unify data return from all sources

[suibianwanwank]

My project is implementing a query engine that uses datafusion to do cross-origin queries. Now on some basic sources, I would like to use sqlx for querying, and during development, I found that there is a lot of common code between different sources. So I would like to reduce the amount of code using some methods

/// Accessor for connecting to sqlx data sources.
///
/// Calling [`get_conn`] produces an [`Connection`] Associated type of [`Database`]
///
/// Methods [`get_sql_schema`] Get the actual schema type via the incoming rows.
///
/// After implementing [`SqlxAccessor`], you can query the results and convert the result into a RecordBatch
/// by calling the [`execute_sqlx_query`] macro.
#[async_trait]
pub trait SqlxAccessor: Sync + Send
{
    /// A database driver, for details, see sqlx [Database].
    type Database: Database;

    /// Get the Connection associated with the Database for accessing the database.
    async fn get_conn(&self) -> Result<<Self::Database as Database>::Connection>;

    /// Get the actual schema information based on the row information of the query result.
    ///
    /// Here you need to map the types in the database to the types in the arrow,
    /// and fill in the actual returned precision, nullable and other information.
    ///
    /// Note that this method is needed because there is a many-to-many mapping
    /// between calcite's type system and arrow's types, and is not solved well in calcite,
    /// so automatic type fixing is needed.
    async fn get_sql_schema(&self, row: Option<&<Self::Database as Database>::Row>) -> Result<SchemaRef>;

    /// Get the sql needed for this access.
    fn get_sql(&self) -> &str;
}

/// Macro for Executing a query and return RecordBatch results Connection
/// and type conversion via `SqlxAccessor`
///
/// Arguments:
/// `ac`: a `SqlxAccessor` type provided for accessing the database to obtain a connection
/// `rp`: the row type of the association type of `ac`
#[macro_export]
macro_rules! execute_sqlx_query {
    ($ac:expr, $rp:ty) => {{
        let mut conn = $ac.get_conn().await?;

        let rows = conn
            .fetch_all(sqlx::query($ac.get_sql()))
            .await
            .context(SqlxFetchAllSnafu {})?;

        // TODO: fix get 0
        let schema = $ac.get_sql_schema(rows.get(0)).await?;

        let array = sqlx_rows_to_array::<$rp>(&rows, schema.clone())?;

        // TODO: May need fix plan schema
        Ok((array, schema))
    }};
}

I expect to achieve this in combination with the above trait and macros， and I originally expected to unify all Vec->array conversions using a generic function like this

pub fn sqlx_rows_to_array<'a, T>(rows: &'a [T], schema: SchemaRef) -> Result<Vec<ArrayRef>>
    where
        T: Row,
        usize: sqlx::ColumnIndex<T>,
        u8: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        u16: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        u32: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        u64: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        i8: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        i16: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        i32: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        i64: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
        String: sqlx::Decode<'a, <T as Row>::Database> + sqlx::Type<<T as Row>::Database>,
{
    gen_index_type!(RowIndex);

    let mut res: Vec<ArrayRef> = Vec::new();

    let mut index = RowIndex::from(0);

    schema.fields.iter().map(|field| {
        match field.data_type() {
            DataType::Int8 => make_array_data!(Int8Builder, rows, index.into_usize_and_incr()),
            DataType::Int16 => make_array_data!(Int16Builder, rows, index.into_usize_and_incr()),
            DataType::Int32 => make_array_data!(Int32Builder, rows, index.into_usize_and_incr()),
            DataType::Int64 => make_array_data!(Int64Builder, rows, index.into_usize_and_incr()),
            DataType::UInt8 => make_array_data!(UInt8Builder, rows, index.into_usize_and_incr()),
            DataType::UInt16 => make_array_data!(UInt16Builder, rows, index.into_usize_and_incr()),
            DataType::UInt32 => make_array_data!(UInt32Builder, rows, index.into_usize_and_incr()),
            DataType::UInt64 => make_array_data!(UInt64Builder, rows, index.into_usize_and_incr()),
            DataType::Utf8 => {
                // Assumes an average of 16 bytes per item.
                let mut arr = StringBuilder::with_capacity(rows.len(), rows.len() * 16);

                rows.iter().try_for_each(|row| {
                    let val: String = row
                        .try_get(index.0)
                        .map_err(|e| TryGetSqlxRow { source: e })?;
                    arr.append_value(val);
                    Ok(())
                })?;
                Arc::new(arr.finish())
            }
            _ => {
                return Err(NotSupportArrowType {
                    tp: field.data_type().clone(),
                })
            }
        };

        Ok(())
    }).collect();

    Ok(res)
}

However, during compilation it was found that not all sources implement decode and sqlx::type, for example

= help: the trait `sqlx::Type<MySql>` is implemented for `u8`
= help: for that trait implementation, expected `MySql`, found `Postgres` 

I'd like to ask if there are any good ideas for implementing this. thanks