Snowflake Normalization doesn't work with table prefix starting with numbers

[marcosmarxm]

Environment

• Airbyte version: 0.35.2-alha
• OS Version / Instance: GCP
• Deployment: Docker
• Source Connector and version: All Sources
• Destination Connector and version: Snowflake
• Severity: Very Low / Low / Medium / High / Critical
• Step where error happened: Sync Job in Normalization Step

Current Behavior

With incremental normalization running Snowflake destination with table prefix starting with numbers break the code.
This is caused because:

airbyte/airbyte-integrations/bases/base-normalization/normalization/transform_catalog/stream_processor.py

Lines 1000 to 1012 in dfebddc

	if self.is_incremental_mode(self.destination_sync_mode):
	if suffix == "scd":
	stg_schema = self.get_schema(True)
	stg_table = self.tables_registry.get_file_name(schema, self.json_path, self.stream_name, "stg", truncate_name)
	if self.destination_type.value == DestinationType.POSTGRES.value:
	# Keep only rows with the max emitted_at to keep incremental behavior
	config["post_hook"] = (
	f"['delete from {stg_schema}.{stg_table} "
	+ f"where {self.airbyte_emitted_at} != (select max({self.airbyte_emitted_at}) "
	+ f"from {stg_schema}.{stg_table})']"
	)
	else:
	config["post_hook"] = f"['drop view {stg_schema}.{stg_table}']"

When creating a connection with prefix, eg, 1_stg_table the code will generate a post-hook for Snowflake as:

drop view schema.1_stg_table'

which is invalid.

Expected Behavior

Correct the normalization to work with number table name prefix.

Logs

If applicable, please upload the logs from the failing operation.
For sync jobs, you can download the full logs from the UI by going to the sync attempt page and
clicking the download logs button at the top right of the logs display window.

LOG

replace this with
your long log
output here

Steps to Reproduce

Are you willing to submit a PR?

Remove this with your answer.