[WIP] AWS: Add partition info to Glue

aws/src/integration/java/org/apache/iceberg/aws/glue/GlueTestBase.java

@@ -65,7 +65,7 @@ public class GlueTestBase {
   static GlueCatalog glueCatalogWithSkip;
 
   static Schema schema = new Schema(Types.NestedField.required(1, "c1", Types.StringType.get(), "c1"));
-  static PartitionSpec partitionSpec = PartitionSpec.builderFor(schema).build();
+  static PartitionSpec partitionSpec = PartitionSpec.builderFor(schema).identity("c1").build();
 
   // table location properties
   static final Map<String, String> tableLocationProperties = ImmutableMap.of(

aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogTable.java

@@ -78,6 +78,7 @@ public void testCreateTable() {
     Assert.assertEquals("additionalLocations should match",
         tableLocationProperties.values().stream().sorted().collect(Collectors.toList()),
         response.table().storageDescriptor().additionalLocations().stream().sorted().collect(Collectors.toList()));
+    Assert.assertEquals(partitionSpec.fields().size(), response.table().partitionKeys().size());
     // verify metadata file exists in S3
     String metaLocation = response.table().parameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP);
     String key = metaLocation.split(testBucketName, -1)[1].substring(1);

aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java

@@ -209,7 +209,8 @@ static void setTableInputInformation(TableInput.Builder tableInputBuilder, Table
           .storageDescriptor(storageDescriptor
               .location(metadata.location())
               .columns(toColumns(metadata))
-              .build());
+              .build())
+          .partitionKeys(toPartitionColumns(metadata));
     } catch (RuntimeException e) {
       LOG.warn("Encountered unexpected exception while converting Iceberg metadata to Glue table information", e);
     }
@@ -302,4 +303,14 @@ private static void addColumnWithDedupe(List<Column> columns, Set<String> dedupe
       dedupe.add(field.name());
     }
   }
+
+  private static List<Column> toPartitionColumns(TableMetadata metadata) {
+    return metadata.spec().fields().stream()
+        .map(f -> Column.builder()
+            .name(f.transform() + "(" + f.name() + ")")
+            .type(toTypeString(f.transform().getResultType(metadata.schema().findField(f.sourceId()).type())))
+            .comment("fieldId: " + f.fieldId() + ", sourceId: " + f.sourceId() + ", transform: " + f.transform())
+            .build())
+        .collect(Collectors.toList());
+  }
 }

aws/src/test/java/org/apache/iceberg/aws/glue/TestIcebergToGlueConverter.java

@@ -129,6 +129,7 @@ public void testSetTableInputInformation() {
         .identity("x")
         .withSpecId(1000)
         .build();
+
     TableMetadata tableMetadata = TableMetadata
         .newTableMetadata(schema, partitionSpec, "s3://test", tableLocationProperties);
     IcebergToGlueConverter.setTableInputInformation(actualTableInputBuilder, tableMetadata);
@@ -161,6 +162,13 @@ public void testSetTableInputInformation() {
                     ))
                     .build()))
             .build())
+        .partitionKeys(ImmutableList.of(
+            Column.builder()
+                .name("identity(x)")
+                .type("string")
+                .comment("fieldId: 1000, sourceId: 1, transform: identity")
+                .build()
+        ))
         .build();
 
     Assert.assertEquals(
@@ -175,6 +183,10 @@ public void testSetTableInputInformation() {
         "Columns should match",
         expectedTableInput.storageDescriptor().columns(),
         actualTableInput.storageDescriptor().columns());
+    Assert.assertEquals(
+        "Partition keys should match",
+        expectedTableInput.partitionKeys(),
+        actualTableInput.partitionKeys());
   }
 
   @Test