Unlike all other Snowflake operators, S3ToSnowflakeTransfer requires the usage of schema instead of letting it be an optional variable that when passed would be used, and if not, then connection metadata is used. 

I can be assigned on this if needed. Keeping this up here so this doesn't go unnoticed. 
airflow/providers/snowflake/transfers/s3_to_snowflake.py
)
        files = ""
        if self.s3_keys:
            files = "files=({})".format(", ".join(f"'{key}'" for key in self.s3_keys))
        
        base_sql = """
                    FROM @{stage}/{prefix}
                    {files}
                    file_format={file_format}
                """.format(
            stage=self.stage,
            prefix=(self.prefix if self.prefix else ""),
            files=files,
            file_format=self.file_format,
        )
        if self.columns_array:
            copy_query = """
                COPY INTO {schema}.{table}({columns}) {base_sql}
            """.format(
                schema=self.schema, table=self.table, columns=",".join(self.columns_array), base_sql=base_sql
            )
        else:
            copy_query = f"""
                COPY INTO {self.schema}.{self.table} {base_sql}
            """
        copy_query = "\n".join(line.strip() for line in copy_query.splitlines())
        self.log.info('Executing COPY command...')
        snowflake_hook.run(copy_query, self.autocommit)
        self.log.info("COPY command completed")