quote identifiers in GET_DDL queries to support special characters (#26007)

* quote identifiers in GET_DDL queries to support special characters

- Fixed get_table_ddl() to quote schema and table identifiers
- Fixed get_view_definition() to quote identifiers when schema is None
- Fixed get_stream_definition() to quote schema and stream identifiers
- Supports tables with dots and other special characters (e.g., CDB_LU_AP.DSPA_ADDRESSES_INITIAL_LOAD)

Fixes: 26006"

* Fix DDL identifier quoting for Snowflake to support schema.table format

---------

Co-authored-by: Mohit Tilala <63147650+mohittilala@users.noreply.github.com>
(cherry picked from commit ca7e1b1)

Author: 2 people

ingestion/src/metadata/ingestion/source/database/snowflake/utils.py

@@ -352,7 +352,7 @@ def get_view_definition(
     )
 
     schema = schema or self.default_schema_name
-    view_name = f'"{schema}"."{table_name}"' if schema else table_name
+    view_name = f'"{schema}"."{table_name}"' if schema else f'"{table_name}"'
     cursor = connection.execute(SNOWFLAKE_GET_VIEW_DDL.format(view_name=view_name))
     try:
         result = cursor.fetchone()
@@ -371,7 +371,7 @@ def get_stream_definition(  # pylint: disable=unused-argument
     Gets the stream definition
     """
     schema = schema or self.default_schema_name
-    stream_name = f"{schema}.{stream_name}" if schema else stream_name
+    stream_name = f'"{schema}"."{stream_name}"' if schema else f'"{stream_name}"'
     cursor = connection.execute(
         SNOWFLAKE_GET_STREAM_DEFINITION.format(stream_name=stream_name)
     )
@@ -644,7 +644,7 @@ def get_table_ddl(
     Gets the Table DDL
     """
     schema = schema or self.default_schema_name
-    table_name = f"{schema}.{table_name}" if schema else table_name
+    table_name = f'"{schema}"."{table_name}"' if schema else f'"{table_name}"'
     cursor = connection.execute(SNOWFLAKE_GET_TABLE_DDL.format(table_name=table_name))
     try:
         result = cursor.fetchone()