fix: Handle reserved words in db, schema and table names with the normalize_name dialect helper

Author: edgarrmondragon

Diff for: target_snowflake/connector.py

@@ -634,3 +634,50 @@ def _adapt_column_type(
                 sql_type,
             )
             raise
+
+    def get_fully_qualified_name(
+        self,
+        table_name: str | None = None,
+        schema_name: str | None = None,
+        db_name: str | None = None,
+        delimiter: str = ".",
+    ) -> str:
+        """Concatenates a fully qualified name from the parts.
+
+        Snowflake-specific implementation.
+
+        Args:
+            table_name: The name of the table.
+            schema_name: The name of the schema. Defaults to None.
+            db_name: The name of the database. Defaults to None.
+            delimiter: Generally: '.' for SQL names and '-' for Singer names.
+
+        Raises:
+            ValueError: If all 3 name parts not supplied.
+
+        Returns:
+            The fully qualified name as a string.
+        """
+        parts = []
+        dialect: SnowflakeDialect = self._engine.dialect
+
+        if db_name:
+            parts.append(dialect.normalize_name(db_name))
+        if schema_name:
+            parts.append(dialect.normalize_name(schema_name))
+        if table_name:
+            parts.append(dialect.normalize_name(table_name))
+
+        if not parts:
+            raise ValueError(
+                "Could not generate fully qualified name: "
+                + ":".join(
+                    [
+                        db_name or "(unknown-db)",
+                        schema_name or "(unknown-schema)",
+                        table_name or "(unknown-table-name)",
+                    ],
+                ),
+            )
+
+        return delimiter.join(parts)

Diff for: tests/core.py

@@ -458,6 +458,61 @@ def setup(self) -> None:
         )
 
 
+class SnowflakeTargetExistingReservedNameTableAlter(TargetFileTestTemplate):
+    name = "existing_reserved_name_table_alter"
+    # This sends a schema that will request altering from TIMESTAMP_NTZ to VARCHAR
+
+    @property
+    def singer_filepath(self) -> Path:
+        current_dir = Path(__file__).resolve().parent
+        return current_dir / "target_test_streams" / "reserved_words_in_table.singer"
+
+    def setup(self) -> None:
+        connector = self.target.default_sink_class.connector_class(self.target.config)
+        table = f'{self.target.config['database']}.{self.target.config['default_target_schema']}."ORDER"'.upper()
+        connector.connection.execute(
+            f"""
+            CREATE OR REPLACE TABLE {table} (
+                ID VARCHAR(16777216),
+                COL_STR VARCHAR(16777216),
+                COL_TS TIMESTAMP_NTZ(9),
+                COL_INT STRING,
+                COL_BOOL BOOLEAN,
+                COL_VARIANT VARIANT,
+                _SDC_BATCHED_AT TIMESTAMP_NTZ(9),
+                _SDC_DELETED_AT VARCHAR(16777216),
+                _SDC_EXTRACTED_AT TIMESTAMP_NTZ(9),
+                _SDC_RECEIVED_AT TIMESTAMP_NTZ(9),
+                _SDC_SEQUENCE NUMBER(38,0),
+                _SDC_TABLE_VERSION NUMBER(38,0),
+                PRIMARY KEY (ID)
+            )
+            """,
+        )
+
+
+class SnowflakeTargetReservedWordsInTable(TargetFileTestTemplate):
+    # Contains reserved words from
+    # https://docs.snowflake.com/en/sql-reference/reserved-keywords
+    # Syncs records then alters schema by adding a non-reserved word column.
+    name = "reserved_words_in_table"
+
+    @property
+    def singer_filepath(self) -> Path:
+        current_dir = Path(__file__).resolve().parent
+        return current_dir / "target_test_streams" / "reserved_words_in_table.singer"
+
+    def validate(self) -> None:
+        connector = self.target.default_sink_class.connector_class(self.target.config)
+        table = f'{self.target.config['database']}.{self.target.config['default_target_schema']}."ORDER"'.upper()
+        result = connector.connection.execute(
+            f"select * from {table}",
+        )
+        assert result.rowcount == 1
+        row = result.first()
+        assert len(row) == 13, f"Row has unexpected length {len(row)}"
+
+
 class SnowflakeTargetTypeEdgeCasesTest(TargetFileTestTemplate):
     name = "type_edge_cases"
 
@@ -540,6 +595,8 @@ def singer_filepath(self) -> Path:
         SnowflakeTargetColonsInColName,
         SnowflakeTargetExistingTable,
         SnowflakeTargetExistingTableAlter,
+        SnowflakeTargetExistingReservedNameTableAlter,
+        SnowflakeTargetReservedWordsInTable,
         SnowflakeTargetTypeEdgeCasesTest,
         SnowflakeTargetColumnOrderMismatch,
     ],

Diff for: tests/target_test_streams/reserved_words_in_table.singer

@@ -0,0 +1,2 @@
+{ "type": "SCHEMA", "stream": "ORDER", "schema": { "properties": { "id": { "type": [ "string", "null" ] }, "col_str": { "type": [ "string", "null" ] }, "col_ts": { "format": "date-time", "type": [ "string", "null" ] }, "col_int": { "type": "integer" }, "col_bool": { "type": [ "boolean", "null" ] }, "col_variant": {"type": "object"} }, "type": "object" }, "key_properties": [ "id" ], "bookmark_properties": [ "col_ts" ] }
+{ "type": "RECORD", "stream": "ORDER", "record": { "id": "123", "col_str": "foo", "col_ts": "2023-06-13 11:50:04.072", "col_int": 5, "col_bool": true, "col_variant": {"key": "val"} }, "time_extracted": "2023-06-14T18:08:23.074716+00:00" }