Fix condition is failing when parsing SqlAlchemy columns (#592)

* Fix issue

* Update release notes

* Add tests

Author: sfc-gh-jvasquezrojas

DESCRIPTION.md

@@ -10,7 +10,8 @@ Source code is also available at:
 
 # Release Notes
 - v1.7.4(June 2, 2025)
-  - Fix dependency on DESCRIBE TABLE columns quantity (differences in columns caused by Snowflake parameters)
+  - Fix dependency on DESCRIBE TABLE columns quantity (differences in columns caused by Snowflake parameters).
+  - Fix unnecessary condition was causing issues when parsing StructuredTypes columns.
   - Update README.md to include instructions on how to verify package signatures using cosign.
 
 - v1.7.3(January 15, 2025)

src/snowflake/sqlalchemy/snowdialect.py

@@ -511,13 +511,6 @@ def _get_schema_columns(self, connection, schema, **kw):
             )
             schema_name = self.denormalize_name(schema)
 
-            iceberg_table_names = self.get_table_names_with_prefix(
-                connection,
-                schema=schema_name,
-                prefix=CustomTablePrefix.ICEBERG.name,
-                info_cache=kw.get("info_cache", None),
-            )
-
             result = connection.execute(
                 text(
                     """
@@ -578,10 +571,7 @@ def _get_schema_columns(self, connection, schema, **kw):
                     col_type_kw["scale"] = numeric_scale
                 elif issubclass(col_type, (sqltypes.String, sqltypes.BINARY)):
                     col_type_kw["length"] = character_maximum_length
-                elif (
-                    issubclass(col_type, StructuredType)
-                    and table_name in iceberg_table_names
-                ):
+                elif issubclass(col_type, StructuredType):
                     if (schema_name, table_name) not in full_columns_descriptions:
                         full_columns_descriptions[(schema_name, table_name)] = (
                             self.table_columns_as_dict(

tests/custom_tables/__snapshots__/test_reflect_snowflake_table.ambr

@@ -21,6 +21,9 @@
     }),
   ])
 # ---
+# name: test_reflection_of_table_with_object_data_type
+  'CREATE TABLE test_snowflake_table_reflection (\tid DECIMAL(38, 0) NOT NULL, \tname OBJECT, \tCONSTRAINT demo_name PRIMARY KEY (id))'
+# ---
 # name: test_simple_reflection_of_table_as_snowflake_table
   'CREATE TABLE test_snowflake_table_reflection (\tid DECIMAL(38, 0) NOT NULL, \tname VARCHAR(16777216), \tCONSTRAINT demo_name PRIMARY KEY (id))'
 # ---

tests/custom_tables/test_reflect_snowflake_table.py

@@ -7,6 +7,36 @@
 from snowflake.sqlalchemy import SnowflakeTable
 
 
+def test_reflection_of_table_with_object_data_type(
+    engine_testaccount, db_parameters, sql_compiler, snapshot
+):
+    metadata = MetaData()
+    table_name = "test_snowflake_table_reflection"
+
+    create_table_sql = f"""
+   CREATE TABLE {table_name} (id INT primary key, name OBJECT);
+    """
+
+    with engine_testaccount.connect() as connection:
+        connection.exec_driver_sql(create_table_sql)
+
+    snowflake_test_table = Table(table_name, metadata, autoload_with=engine_testaccount)
+    constraint = snowflake_test_table.constraints.pop()
+    constraint.name = "demo_name"
+    snowflake_test_table.constraints.add(constraint)
+
+    try:
+        with engine_testaccount.connect():
+            value = CreateTable(snowflake_test_table)
+
+            actual = sql_compiler(value)
+
+            assert actual == snapshot
+
+    finally:
+        metadata.drop_all(engine_testaccount)
+
+
 def test_simple_reflection_of_table_as_sqlalchemy_table(
     engine_testaccount, db_parameters, sql_compiler, snapshot
 ):