Transform op - database is not picked from metadata

[utkarsharma2]

Describe the bug

with DAG(dag_id="test", schedule=None, start_date=datetime(1970, 1, 1)) as dag:
    @aql.transform
    def select(input_table: Table):
        return "SELECT * FROM my_db.my_schema.my_table LIMIT 4;"  # {{ input_table }} was also failing
    select(
        input_table=Table(metadata={"schema": "my_schema", "database": "db"}, name="my_table", conn_id="my_conn_id"),
        output_table=Table(name="fritz_test_1234")
    )

snowflake.connector.errors.ProgrammingError: 090105 (22000): 01a77af0-0605-96b0-0000-682112f6e8ae: Cannot perform SELECT. This session does not have a current database. Call 'USE DATABASE', or use a qualified name.
ref: https://astronomer.slack.com/archives/C02B8SPT93K/p1665414029917169

Version

• Astro: 1.1.1

To Reproduce
Try the above code.

Expected behavior
If the DB is not specified in the connection we should be able to pick from metadata.

[pankajastro]

This looks like a problem when using snowflake. I quickly tested the bigquery worked fine

[pankajastro]

For the hook creation in astro-sdk we tightly depends on the airflow connection
for example

 @property
    def hook(self) -> BigQueryHook:
        """Retrieve Airflow hook to interface with the BigQuery database."""
        return BigQueryHook(gcp_conn_id=self.conn_id, use_legacy_sql=False)

Snowflake hook expects the database, schema, etc., so the ideal place to set these values is a connection only. Even in our case, we do

 @property
    def hook(self) -> SnowflakeHook:
        """Retrieve Airflow hook to interface with the Snowflake database."""
        return SnowflakeHook(snowflake_conn_id=self.conn_id)

If a user is setting up the schema, and database in the input and output table in the transform operator which value should we use from input_table or output_table or will we create multiple hooks? it looks a little bit complicated to me in the current design.

For now, I have modified the create_database method to accept conn_id as well as a table and I'm passing the output table from the base database class and setting the schema and database in the snowflake hook if the table metadata has these values.
I'll fix the CI if people agree with the approach. PR #1117