You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ja...@apache.org on 2021/10/29 05:28:42 UTC
[iceberg] branch master updated: AWS: Add column parameters for
Glue catalog table schema (#3352)
This is an automated email from the ASF dual-hosted git repository.
jackye pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 7e55b0d AWS: Add column parameters for Glue catalog table schema (#3352)
7e55b0d is described below
commit 7e55b0dbc7597cc610d6eafacce3c64644a3e72d
Author: Rajarshi Sarkar <sr...@amazon.com>
AuthorDate: Fri Oct 29 10:58:29 2021 +0530
AWS: Add column parameters for Glue catalog table schema (#3352)
---
.../iceberg/aws/glue/GlueCatalogTableTest.java | 71 ++++++++++++++++
.../iceberg/aws/glue/IcebergToGlueConverter.java | 94 +++++++++++++++++----
.../aws/glue/IcebergToGlueConverterTest.java | 95 ++++++++++++++++++++++
3 files changed, 245 insertions(+), 15 deletions(-)
diff --git a/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueCatalogTableTest.java b/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueCatalogTableTest.java
index b2c1ebd..5dbf904 100644
--- a/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueCatalogTableTest.java
+++ b/aws/src/integration/java/org/apache/iceberg/aws/glue/GlueCatalogTableTest.java
@@ -36,10 +36,13 @@ import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
import org.junit.Assert;
import org.junit.Test;
+import software.amazon.awssdk.services.glue.model.Column;
import software.amazon.awssdk.services.glue.model.CreateTableRequest;
import software.amazon.awssdk.services.glue.model.EntityNotFoundException;
import software.amazon.awssdk.services.glue.model.GetTableRequest;
@@ -52,6 +55,8 @@ import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
import software.amazon.awssdk.services.s3.model.S3Object;
+import static org.apache.iceberg.expressions.Expressions.truncate;
+
public class GlueCatalogTableTest extends GlueTestBase {
@Test
@@ -287,4 +292,70 @@ public class GlueCatalogTableTest extends GlueTestBase {
1, glue.getTableVersions(GetTableVersionsRequest.builder()
.databaseName(namespace).tableName(tableName).build()).tableVersions().size());
}
+
+ @Test
+ public void testColumnCommentsAndParameters() {
+ String namespace = createNamespace();
+ String tableName = createTable(namespace);
+ Table table = glueCatalog.loadTable(TableIdentifier.of(namespace, tableName));
+ table.updateSchema().addColumn("c2", Types.StructType.of(
+ Types.NestedField.required(3, "z", Types.IntegerType.get())), "c2").commit();
+ table.updateSpec().addField(truncate("c1", 8)).commit();
+ GetTableResponse response = glue.getTable(GetTableRequest.builder()
+ .databaseName(namespace).name(tableName).build());
+ List<Column> actualColumns = response.table().storageDescriptor().columns();
+
+ List<Column> expectedColumns = ImmutableList.of(
+ Column.builder()
+ .name("c1")
+ .type("string")
+ .comment("c1")
+ .parameters(ImmutableMap.of(
+ IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.SCHEMA_COLUMN,
+ IcebergToGlueConverter.ICEBERG_FIELD_ID, "1",
+ IcebergToGlueConverter.ICEBERG_FIELD_OPTIONAL, "false",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "string",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "STRING"
+ ))
+ .build(),
+ Column.builder()
+ .name("c2")
+ .type("struct<z:int>")
+ .comment("c2")
+ .parameters(ImmutableMap.of(
+ IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.SCHEMA_COLUMN,
+ IcebergToGlueConverter.ICEBERG_FIELD_ID, "2",
+ IcebergToGlueConverter.ICEBERG_FIELD_OPTIONAL, "true",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "struct<z:int>",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "STRUCT"
+ ))
+ .build(),
+ Column.builder()
+ .name("z")
+ .type("int")
+ .parameters(ImmutableMap.of(
+ IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.SCHEMA_SUBFIELD,
+ IcebergToGlueConverter.ICEBERG_FIELD_ID, "3",
+ IcebergToGlueConverter.ICEBERG_FIELD_OPTIONAL, "false",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "int",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "INTEGER"
+ ))
+ .build(),
+ Column.builder()
+ .name("c1_trunc_8")
+ .type("string")
+ .parameters(ImmutableMap.<String, String>builder()
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.PARTITION_FIELD)
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "STRING")
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "string")
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_ID, "1000")
+ .put(IcebergToGlueConverter.ICEBERG_PARTITION_FIELD_ID, "1000")
+ .put(IcebergToGlueConverter.ICEBERG_PARTITION_SOURCE_ID, "1")
+ .put(IcebergToGlueConverter.ICEBERG_PARTITION_TRANSFORM, "truncate[8]")
+ .build()
+ )
+ .build()
+ );
+ Assert.assertEquals("Columns do not match", expectedColumns, actualColumns);
+ }
}
diff --git a/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java b/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java
index 2c8f1f9..e7725ad 100644
--- a/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java
+++ b/aws/src/main/java/org/apache/iceberg/aws/glue/IcebergToGlueConverter.java
@@ -19,18 +19,26 @@
package org.apache.iceberg.aws.glue;
+import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import org.apache.iceberg.PartitionField;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.types.Types.NestedField;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.glue.model.Column;
@@ -47,6 +55,17 @@ class IcebergToGlueConverter {
private static final Pattern GLUE_DB_PATTERN = Pattern.compile("^[a-z0-9_]{1,252}$");
private static final Pattern GLUE_TABLE_PATTERN = Pattern.compile("^[a-z0-9_]{1,255}$");
+ public static final String ICEBERG_FIELD_USAGE = "iceberg.field.usage";
+ public static final String ICEBERG_FIELD_TYPE_TYPE_ID = "iceberg.field.type.typeid";
+ public static final String ICEBERG_FIELD_TYPE_STRING = "iceberg.field.type.string";
+ public static final String ICEBERG_FIELD_ID = "iceberg.field.id";
+ public static final String ICEBERG_FIELD_OPTIONAL = "iceberg.field.optional";
+ public static final String ICEBERG_PARTITION_TRANSFORM = "iceberg.partition.transform";
+ public static final String ICEBERG_PARTITION_FIELD_ID = "iceberg.partition.field-id";
+ public static final String ICEBERG_PARTITION_SOURCE_ID = "iceberg.partition.source-id";
+ public static final String SCHEMA_COLUMN = "schema-column";
+ public static final String SCHEMA_SUBFIELD = "schema-subfield";
+ public static final String PARTITION_FIELD = "partition-field";
/**
* A Glue database name cannot be longer than 252 characters.
@@ -160,21 +179,8 @@ class IcebergToGlueConverter {
tableInputBuilder
.storageDescriptor(StorageDescriptor.builder()
.location(metadata.location())
- .columns(metadata.schema().columns().stream()
- .map(f -> Column.builder()
- .name(f.name())
- .type(toTypeString(f.type()))
- .comment("Iceberg column: { " + f + " }")
- .build())
- .collect(Collectors.toList()))
- .build())
- .partitionKeys(metadata.spec().fields().stream()
- .map(f -> Column.builder()
- .name(f.name())
- .type(toTypeString(f.transform().getResultType(metadata.schema().findField(f.sourceId()).type())))
- .comment("Iceberg partition field: {" + f + "}")
- .build())
- .collect(Collectors.toList()));
+ .columns(toColumns(metadata))
+ .build());
} catch (RuntimeException e) {
LOG.warn("Encountered unexpected exception while converting Iceberg metadata to Glue table information", e);
}
@@ -231,4 +237,62 @@ class IcebergToGlueConverter {
return type.typeId().name().toLowerCase(Locale.ENGLISH);
}
}
+
+ private static List<Column> toColumns(TableMetadata metadata) {
+ List<Column> columns = Lists.newArrayList();
+ Set<NestedField> rootColumnSet = Sets.newHashSet();
+ // Add schema-column fields
+ for (NestedField field : metadata.schema().columns()) {
+ rootColumnSet.add(field);
+ columns.add(Column.builder()
+ .name(field.name())
+ .type(toTypeString(field.type()))
+ .comment(field.doc())
+ .parameters(convertToParameters(SCHEMA_COLUMN, field))
+ .build());
+ }
+ // Add schema-subfield
+ for (NestedField field : TypeUtil.indexById(metadata.schema().asStruct()).values()) {
+ if (!rootColumnSet.contains(field)) {
+ columns.add(Column.builder()
+ .name(field.name())
+ .type(toTypeString(field.type()))
+ .comment(field.doc())
+ .parameters(convertToParameters(SCHEMA_SUBFIELD, field))
+ .build());
+ }
+ }
+ // Add partition-field
+ for (PartitionField partitionField : metadata.spec().fields()) {
+ Type type = partitionField.transform()
+ .getResultType(metadata.schema().findField(partitionField.sourceId()).type());
+ columns.add(Column.builder()
+ .name(partitionField.name())
+ .type(toTypeString(type))
+ .parameters(convertToPartitionFieldParameters(type, partitionField))
+ .build());
+ }
+ return columns;
+ }
+
+ private static Map<String, String> convertToParameters(String fieldUsage, NestedField field) {
+ return ImmutableMap.of(ICEBERG_FIELD_USAGE, fieldUsage,
+ ICEBERG_FIELD_TYPE_TYPE_ID, field.type().typeId().toString(),
+ ICEBERG_FIELD_TYPE_STRING, toTypeString(field.type()),
+ ICEBERG_FIELD_ID, Integer.toString(field.fieldId()),
+ ICEBERG_FIELD_OPTIONAL, Boolean.toString(field.isOptional())
+ );
+ }
+
+ private static Map<String, String> convertToPartitionFieldParameters(Type type, PartitionField partitionField) {
+ return ImmutableMap.<String, String>builder()
+ .put(ICEBERG_FIELD_USAGE, PARTITION_FIELD)
+ .put(ICEBERG_FIELD_TYPE_TYPE_ID, type.typeId().toString())
+ .put(ICEBERG_FIELD_TYPE_STRING, toTypeString(type))
+ .put(ICEBERG_FIELD_ID, Integer.toString(partitionField.fieldId()))
+ .put(ICEBERG_PARTITION_TRANSFORM, partitionField.transform().toString())
+ .put(ICEBERG_PARTITION_FIELD_ID, Integer.toString(partitionField.fieldId()))
+ .put(ICEBERG_PARTITION_SOURCE_ID, Integer.toString(partitionField.sourceId()))
+ .build();
+ }
}
diff --git a/aws/src/test/java/org/apache/iceberg/aws/glue/IcebergToGlueConverterTest.java b/aws/src/test/java/org/apache/iceberg/aws/glue/IcebergToGlueConverterTest.java
index d4c20b7..746f044 100644
--- a/aws/src/test/java/org/apache/iceberg/aws/glue/IcebergToGlueConverterTest.java
+++ b/aws/src/test/java/org/apache/iceberg/aws/glue/IcebergToGlueConverterTest.java
@@ -23,12 +23,21 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
import org.junit.Assert;
import org.junit.Test;
+import software.amazon.awssdk.services.glue.model.Column;
import software.amazon.awssdk.services.glue.model.DatabaseInput;
+import software.amazon.awssdk.services.glue.model.StorageDescriptor;
+import software.amazon.awssdk.services.glue.model.TableInput;
public class IcebergToGlueConverterTest {
@@ -64,4 +73,90 @@ public class IcebergToGlueConverterTest {
Namespace namespace = Namespace.of("db");
Assert.assertEquals(input, IcebergToGlueConverter.toDatabaseInput(namespace, param));
}
+
+ @Test
+ public void testSetTableInputInformation() {
+ // Actual TableInput
+ TableInput.Builder actualTableInputBuilder = TableInput.builder();
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "x", Types.StringType.get(), "comment1"),
+ Types.NestedField.required(2, "y", Types.StructType.of(
+ Types.NestedField.required(3, "z", Types.IntegerType.get())), "comment2")
+ );
+ PartitionSpec partitionSpec = PartitionSpec.builderFor(schema)
+ .identity("x")
+ .withSpecId(1000)
+ .build();
+ TableMetadata tableMetadata = TableMetadata
+ .newTableMetadata(schema, partitionSpec, "s3://test", ImmutableMap.of());
+ IcebergToGlueConverter.setTableInputInformation(actualTableInputBuilder, tableMetadata);
+ TableInput actualTableInput = actualTableInputBuilder.build();
+
+ // Expected TableInput
+ TableInput expectedTableInput = TableInput.builder().storageDescriptor(
+ StorageDescriptor.builder()
+ .location("s3://test")
+ .columns(ImmutableList.of(
+ Column.builder()
+ .name("x")
+ .type("string")
+ .comment("comment1")
+ .parameters(ImmutableMap.of(
+ IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.SCHEMA_COLUMN,
+ IcebergToGlueConverter.ICEBERG_FIELD_ID, "1",
+ IcebergToGlueConverter.ICEBERG_FIELD_OPTIONAL, "false",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "string",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "STRING"
+ ))
+ .build(),
+ Column.builder()
+ .name("y")
+ .type("struct<z:int>")
+ .comment("comment2")
+ .parameters(ImmutableMap.of(
+ IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.SCHEMA_COLUMN,
+ IcebergToGlueConverter.ICEBERG_FIELD_ID, "2",
+ IcebergToGlueConverter.ICEBERG_FIELD_OPTIONAL, "false",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "struct<z:int>",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "STRUCT"
+ ))
+ .build(),
+ Column.builder()
+ .name("z")
+ .type("int")
+ .parameters(ImmutableMap.of(
+ IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.SCHEMA_SUBFIELD,
+ IcebergToGlueConverter.ICEBERG_FIELD_ID, "3",
+ IcebergToGlueConverter.ICEBERG_FIELD_OPTIONAL, "false",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "int",
+ IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "INTEGER"
+ ))
+ .build(),
+ Column.builder()
+ .name("x")
+ .type("string")
+ .parameters(ImmutableMap.<String, String>builder()
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_USAGE, IcebergToGlueConverter.PARTITION_FIELD)
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_TYPE_TYPE_ID, "STRING")
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_TYPE_STRING, "string")
+ .put(IcebergToGlueConverter.ICEBERG_FIELD_ID, "1000")
+ .put(IcebergToGlueConverter.ICEBERG_PARTITION_FIELD_ID, "1000")
+ .put(IcebergToGlueConverter.ICEBERG_PARTITION_SOURCE_ID, "1")
+ .put(IcebergToGlueConverter.ICEBERG_PARTITION_TRANSFORM, "identity")
+ .build()
+ )
+ .build()
+ )
+ ).build())
+ .build();
+
+ Assert.assertEquals(
+ "Location do not match",
+ expectedTableInput.storageDescriptor().location(),
+ actualTableInput.storageDescriptor().location());
+ Assert.assertEquals(
+ "Columns do not match",
+ expectedTableInput.storageDescriptor().columns(),
+ actualTableInput.storageDescriptor().columns());
+ }
}