fix: render Hive STRUCT syntax in table column DDL

docs/sqlalchemy.md

@@ -825,12 +825,17 @@ This generates the following SQL structure:
 
 ```sql
 CREATE TABLE users (
-    id INTEGER,
-    profile ROW(name STRING, age INTEGER, email STRING),
-    settings ROW(theme STRING, notifications ROW(email STRING, push STRING))
+    id INT,
+    profile STRUCT<name:STRING, age:INT, email:STRING>,
+    settings STRUCT<theme:STRING, notifications:STRUCT<email:STRING, push:STRING>>
 )
 ```
 
+`CREATE TABLE` renders `AthenaStruct` columns with Hive `STRUCT<name:type, ...>` syntax at every nesting depth.
+That includes top-level columns, fields of a STRUCT, STRUCT values inside MAP, and STRUCT values inside ARRAY.
+Integer fields, and integer MAP keys and values, use `INT` in that DDL.
+`CAST` and other SQL expressions keep `ROW(...)`, `MAP(...)`, and `ARRAY(...)`, and spell integers as `INTEGER`.
+
 #### Querying STRUCT data
 
 PyAthena automatically converts STRUCT data between different formats:
@@ -970,13 +975,16 @@ This generates the following SQL structure:
 
 ```sql
 CREATE TABLE products (
-    id INTEGER,
+    id INT,
     attributes MAP<STRING, STRING>,
-    metrics MAP<STRING, INTEGER>,
-    categories MAP<INTEGER, STRING>
+    metrics MAP<STRING, INT>,
+    categories MAP<INT, STRING>
 )
 ```
 
+`CREATE TABLE` renders integer MAP keys and values as `INT`.
+`CAST` still spells those integers as `INTEGER`.
+
 #### Querying MAP data
 
 PyAthena automatically converts MAP data between different formats:

pyathena/sqlalchemy/compiler.py

@@ -90,6 +90,9 @@ class AthenaTypeCompiler(GenericTypeCompiler):
     - MAP: Key-value pair collections
     - ARRAY: Ordered collections of elements
 
+    CREATE TABLE columns render STRUCT fields as Hive ``STRUCT<name:type>``.
+    Compiling a type on its own renders ``ROW(...)``.
+
     See Also:
         AWS Athena Data Types:
         https://docs.aws.amazon.com/athena/latest/ug/data-types.html
@@ -121,7 +124,7 @@ def visit_TINYINT(self, type_: types.Integer, **kw: Any) -> str:
         return "TINYINT"
 
     def visit_INTEGER(self, type_: types.Integer, **kw: Any) -> str:
-        return "INT" if kw.get("_athena_array_ddl") else "INTEGER"
+        return "INT" if kw.get("_athena_hive_ddl") else "INTEGER"
 
     def visit_SMALLINT(self, type_: types.SmallInteger, **kw: Any) -> str:
         return "SMALLINT"
@@ -199,31 +202,51 @@ def visit_tinyint(self, type_, **kw):
     def visit_enum(self, type_, **kw):
         return self.visit_string(type_, **kw)
 
+    def _enable_hive_column_ddl(self, kw: dict[str, Any]) -> bool:
+        """Enable Hive spelling for a CREATE TABLE column type.
+
+        ``get_column_specification`` passes the column as ``type_expression``.
+        ARRAY compilation sets ``_athena_hive_ddl`` so nested fields use
+        ``STRUCT<name:type>`` and ``INT``. STRUCT and MAP reuse that flag in
+        column DDL. Direct compilation and CAST leave it unset.
+
+        Args:
+            kw: Type-compiler keyword arguments. When Hive spelling applies,
+                ``_athena_hive_ddl`` is set so nested types keep it.
+
+        Returns:
+            True when the type should use Hive DDL syntax.
+        """
+        if kw.get("_athena_hive_ddl") or isinstance(kw.get("type_expression"), Column):
+            kw["_athena_hive_ddl"] = True
+            return True
+        return False
+
     def visit_struct(self, type_, **kw):
-        if isinstance(type_, AthenaStruct):
-            if type_.fields:
-                field_specs = []
-                for field_name, field_type in type_.fields.items():
-                    field_type_str = self.process(field_type, **kw)
-                    preparer = (
-                        AthenaDDLIdentifierPreparer(self.dialect)
-                        if kw.get("_athena_array_ddl")
-                        else self.dialect.identifier_preparer
-                    )
-                    name = preparer.quote(field_name)
-                    separator = ":" if kw.get("_athena_array_ddl") else " "
-                    field_specs.append(f"{name}{separator}{field_type_str}")
-                if kw.get("_athena_array_ddl"):
-                    return f"STRUCT<{', '.join(field_specs)}>"
-                return f"ROW({', '.join(field_specs)})"
+        # Empty structs keep the existing ROW() rendering in every context.
+        if not isinstance(type_, AthenaStruct) or not type_.fields:
             return "ROW()"
-        return "ROW()"
+        hive_ddl = self._enable_hive_column_ddl(kw)
+        preparer = (
+            AthenaDDLIdentifierPreparer(self.dialect)
+            if hive_ddl
+            else self.dialect.identifier_preparer
+        )
+        separator = ":" if hive_ddl else " "
+        field_specs = []
+        for field_name, field_type in type_.fields.items():
+            field_type_str = self.process(field_type, **kw)
+            field_specs.append(f"{preparer.quote(field_name)}{separator}{field_type_str}")
+        if hive_ddl:
+            return f"STRUCT<{', '.join(field_specs)}>"
+        return f"ROW({', '.join(field_specs)})"
 
     def visit_STRUCT(self, type_, **kw):
         return self.visit_struct(type_, **kw)
 
     def visit_map(self, type_, **kw):
         if isinstance(type_, AthenaMap):
+            self._enable_hive_column_ddl(kw)
             key_type_str = self.process(type_.key_type, **kw)
             value_type_str = self.process(type_.value_type, **kw)
             return f"MAP<{key_type_str}, {value_type_str}>"
@@ -234,7 +257,7 @@ def visit_MAP(self, type_, **kw):
 
     def visit_array(self, type_, **kw):
         if isinstance(type_, types.ARRAY):
-            kw["_athena_array_ddl"] = True
+            kw["_athena_hive_ddl"] = True
             item_type_str = self.process(_ArrayTypeInspector.item_type(type_), **kw)
             return f"ARRAY<{item_type_str}>"
         return "ARRAY<STRING>"
@@ -1158,6 +1181,7 @@ def get_column_specification(self, column: Column[Any], **kwargs) -> str:
             # use the int keyword to represent an integer
             type_ = "INT"
         else:
+            # type_expression marks column DDL so STRUCT and MAP use Hive syntax.
             type_ = self.dialect.type_compiler.process(column.type, type_expression=column)
         text = [f"{self.preparer.format_column(column)} {type_}"]
         if column.comment:

tests/pyathena/sqlalchemy/test_base.py

@@ -3496,8 +3496,8 @@ def test_create_table_with_map_types(self, engine):
 
         # Verify MAP types are correctly compiled
         assert "attributes MAP<STRING, STRING>" in ddl_string
-        assert "metrics MAP<STRING, INTEGER>" in ddl_string
-        assert "complex_map MAP<STRING, ROW(value STRING, count INTEGER)>" in ddl_string
+        assert "metrics MAP<STRING, INT>" in ddl_string
+        assert "complex_map MAP<STRING, STRUCT<value:STRING, count:INT>>" in ddl_string
         assert "nested_map MAP<STRING, ARRAY<STRING>>" in ddl_string
 
     def test_create_table_with_struct_types(self, engine):
@@ -3539,12 +3539,12 @@ def test_create_table_with_struct_types(self, engine):
         ddl_string = str(create_ddl)
 
         # Verify STRUCT types are correctly compiled
-        assert "user_info ROW(name STRING, age INTEGER, email STRING)" in ddl_string
+        assert "user_info STRUCT<name:STRING, age:INT, email:STRING>" in ddl_string
         assert (
-            "nested_struct ROW(personal ROW(first_name STRING, last_name STRING), "
-            "preferences MAP<STRING, STRING>)" in ddl_string
+            "nested_struct STRUCT<personal:STRUCT<first_name:STRING, last_name:STRING>, "
+            "preferences:MAP<STRING, STRING>>" in ddl_string
         )
-        assert "struct_with_array ROW(tags ARRAY<STRING>, scores ARRAY<INT>)" in ddl_string
+        assert "struct_with_array STRUCT<tags:ARRAY<STRING>, scores:ARRAY<INT>>" in ddl_string
 
     def test_create_table_with_complex_nested_types(self, engine):
         """Test DDL compilation for complex nested combinations of ARRAY, MAP, and STRUCT."""
@@ -3582,6 +3582,57 @@ def test_create_table_with_complex_nested_types(self, engine):
         )
         assert expected_type in ddl_string
 
+    def test_external_parquet_struct_columns_round_trip(self, engine):
+        """Create a Parquet table of top-level and MAP-nested STRUCTs and read the fields back."""
+        _, conn = engine
+        table_name = "test_external_parquet_struct_columns"
+        location = f"{ENV.s3_staging_dir}{ENV.schema}/{table_name}/"
+        table = Table(
+            table_name,
+            MetaData(schema=ENV.schema),
+            Column(
+                "profile",
+                AthenaStruct(
+                    ("name", types.String),
+                    ("age", types.Integer),
+                    (
+                        "address",
+                        AthenaStruct(("city", types.String), ("zip", types.Integer)),
+                    ),
+                ),
+            ),
+            Column(
+                "labels",
+                AthenaMap(
+                    types.String,
+                    AthenaStruct(("value", types.String), ("count", types.Integer)),
+                ),
+            ),
+            awsathena_location=location,
+            awsathena_file_format="PARQUET",
+        )
+        ddl = str(CreateTable(table).compile(dialect=conn.dialect))
+        assert "profile STRUCT<name:STRING, age:INT, address:STRUCT<city:STRING, zip:INT>>" in ddl
+        assert "labels MAP<STRING, STRUCT<value:STRING, count:INT>>" in ddl
+        table.create(bind=conn)
+        conn.execute(
+            text(
+                f"INSERT INTO {ENV.schema}.{table_name} VALUES ("
+                "CAST(ROW('Ada', 36, ROW('London', 12345)) AS "
+                "ROW(name VARCHAR, age INTEGER, address ROW(city VARCHAR, zip INTEGER))), "
+                "MAP(ARRAY['home'], ARRAY[CAST(ROW('Lovelace', 2) AS "
+                "ROW(value VARCHAR, count INTEGER))]))"
+            )
+        )
+        row = conn.execute(
+            text(
+                "SELECT profile.name, profile.age, profile.address.city, "
+                "profile.address.zip, labels['home'].value, labels['home'].count "
+                f"FROM {ENV.schema}.{table_name}"
+            )
+        ).one()
+        assert tuple(row) == ("Ada", 36, "London", 12345, "Lovelace", 2)
+
     def test_sqlalchemy_execute_with_execution_options_callback(self, engine):
         """Test callback functionality through SQLAlchemy execution_options."""
         engine, conn = engine