Skip to content

Hive: Convert the CREATE TABLE ... PARTITIONED BY to Iceberg identity partitions #1917

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 4 commits into from
Dec 18, 2020
Merged

Hive: Convert the CREATE TABLE ... PARTITIONED BY to Iceberg identity partitions #1917

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f9ccf40
Hive: Convert the CREATE TABLE ... PARTITIONED BY to Iceberg identity…
Dec 11, 2020
04073e2
Review comments
Dec 12, 2020
b362083
Refactored test so we try out the insert path for multi level partiti…
Dec 12, 2020
ff585b4
Inceased heap size
Dec 12, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions hive-metastore/src/main/java/org/apache/iceberg/hive/HiveSchemaUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
Expand Down Expand Up @@ -63,6 +64,18 @@ public static Schema convert(List<FieldSchema> fieldSchemas) {
return HiveSchemaConverter.convert(names, typeInfos);
}

/**
* Converts the Hive partition columns to Iceberg identity partition specification.
* @param schema The Iceberg schema
* @param fieldSchemas The partition column specification
* @return The Iceberg partition specification
*/
public static PartitionSpec spec(Schema schema, List<FieldSchema> fieldSchemas) {
PartitionSpec.Builder builder = PartitionSpec.builderFor(schema);
fieldSchemas.forEach(fieldSchema -> builder.identity(fieldSchema.getName()));
return builder.build();
}

/**
* Converts the Hive list of column names and column types to an Iceberg schema.
* @param names The list of the Hive column names
Expand Down
32 changes: 26 additions & 6 deletions mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,13 @@

package org.apache.iceberg.mr.hive;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.iceberg.BaseMetastoreTableOperations;
import org.apache.iceberg.CatalogUtil;
Expand Down Expand Up @@ -93,11 +96,16 @@ public void preCreateTable(org.apache.hadoop.hive.metastore.api.Table hmsTable)
// If the table does not exist collect data for table creation
// - InputFormatConfig.TABLE_SCHEMA, InputFormatConfig.PARTITION_SPEC takes precedence so the user can override the
// Iceberg schema and specification generated by the code
// - Partitioned Hive tables are currently not allowed

Schema schema = schema(catalogProperties, hmsTable);
PartitionSpec spec = spec(schema, catalogProperties, hmsTable);

// If there are partition keys specified remove them from the HMS table and add them to the column list
if (hmsTable.isSetPartitionKeys()) {
hmsTable.getSd().getCols().addAll(hmsTable.getPartitionKeys());
hmsTable.setPartitionKeysIsSet(false);
}

catalogProperties.put(InputFormatConfig.TABLE_SCHEMA, SchemaParser.toJson(schema));
catalogProperties.put(InputFormatConfig.PARTITION_SPEC, PartitionSpecParser.toJson(spec));

Expand Down Expand Up @@ -197,20 +205,32 @@ private static Schema schema(Properties properties, org.apache.hadoop.hive.metas
if (properties.getProperty(InputFormatConfig.TABLE_SCHEMA) != null) {
return SchemaParser.fromJson(properties.getProperty(InputFormatConfig.TABLE_SCHEMA));
} else {
return HiveSchemaUtil.convert(hmsTable.getSd().getCols());
if (hmsTable.isSetPartitionKeys() && !hmsTable.getPartitionKeys().isEmpty()) {
Comment thread
pvary marked this conversation as resolved.

@rdblue rdblue Dec 17, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When would isSetPartitionKeys() be true and getPartitionKeys() empty?

@pvary pvary Jan 5, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Theoretically it could be set to an empty list. Checked this way to keep on the safe side.

// Add partitioning columns to the original column list before creating the Iceberg Schema
List<FieldSchema> cols = new ArrayList<>(hmsTable.getSd().getCols());
Comment thread
pvary marked this conversation as resolved.
cols.addAll(hmsTable.getPartitionKeys());
return HiveSchemaUtil.convert(cols);
} else {
return HiveSchemaUtil.convert(hmsTable.getSd().getCols());
}
}
}

private static PartitionSpec spec(Schema schema, Properties properties,
org.apache.hadoop.hive.metastore.api.Table hmsTable) {

Preconditions.checkArgument(hmsTable.getPartitionKeys() == null || hmsTable.getPartitionKeys().isEmpty(),
"Partitioned Hive tables are currently not supported");

if (properties.getProperty(InputFormatConfig.PARTITION_SPEC) != null) {
Preconditions.checkArgument(hmsTable.isSetPartitionKeys() || hmsTable.getPartitionKeys().isEmpty(),
Comment thread
pvary marked this conversation as resolved.
Outdated
"Provide only one of the following: Hive partition specification, or the " +
InputFormatConfig.PARTITION_SPEC + " property");
return PartitionSpecParser.fromJson(schema, properties.getProperty(InputFormatConfig.PARTITION_SPEC));
} else {
return PartitionSpec.unpartitioned();
if (hmsTable.isSetPartitionKeys() && !hmsTable.getPartitionKeys().isEmpty()) {
Comment thread
pvary marked this conversation as resolved.
// If the table is partitioned then generate the identity partition definitions for the Iceberg table
return HiveSchemaUtil.spec(schema, hmsTable.getPartitionKeys());
} else {
return PartitionSpec.unpartitioned();
}
}
}
}
100 changes: 74 additions & 26 deletions mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -666,45 +666,52 @@ public void testCreateTableWithColumnSpecification() throws IOException {
}

@Test
public void testCreateTableWithColumnSpecificationPartitioned() {
public void testCreateTableWithColumnSpecificationPartitioned() throws IOException {
TableIdentifier identifier = TableIdentifier.of("default", "customers");
runPartitionedWriteTest(identifier, createSQLForPartitionedTableBySchema(testTables, identifier));
}

@Test
public void testCreatePartitionedTableByProperty() throws IOException {
TableIdentifier identifier = TableIdentifier.of("default", "customers");
runPartitionedWriteTest(identifier, createSQLForPartitionedTableByProperty(testTables, identifier));
}

@Test
public void testCreatePartitionedTableWithPropertiesAndWithColumnSpecification() {
PartitionSpec spec = PartitionSpec.builderFor(CUSTOMER_SCHEMA).identity("last_name").build();

AssertHelpers.assertThrows("should throw exception", IllegalArgumentException.class,
"currently not supported", () -> {
"Provide only one of the following", () -> {
shell.executeStatement("CREATE EXTERNAL TABLE customers (customer_id BIGINT) " +
"PARTITIONED BY (first_name STRING) " +
"STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' " +
testTables.locationForCreateTableSQL(TableIdentifier.of("default", "customers")));
testTables.locationForCreateTableSQL(TableIdentifier.of("default", "customers")) +
" TBLPROPERTIES ('" + InputFormatConfig.PARTITION_SPEC + "'='" +
PartitionSpecParser.toJson(spec) + "')");
}
);
}

@Test
public void testCreatePartitionedTable() throws IOException {
public void testCreateTableWithColumnSpecificationMultilevelPartitioned() {
Comment thread
pvary marked this conversation as resolved.
Outdated
TableIdentifier identifier = TableIdentifier.of("default", "customers");
PartitionSpec spec = PartitionSpec.builderFor(CUSTOMER_SCHEMA).identity("first_name").build();

shell.executeStatement("CREATE EXTERNAL TABLE customers " +
final Schema expectedSchema = new Schema(
optional(1, "id", Types.LongType.get()),
optional(2, "name", Types.StringType.get()),
optional(3, "branch_id", Types.IntegerType.get()),
optional(4, "country", Types.StringType.get()));
final PartitionSpec expectedSpec =
PartitionSpec.builderFor(expectedSchema).identity("branch_id").identity("country").build();
shell.executeStatement("CREATE EXTERNAL TABLE customers (id BIGINT, name STRING) " +
"PARTITIONED BY (branch_id INT, country STRING) " +
"STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' " +
testTables.locationForCreateTableSQL(identifier) +
"TBLPROPERTIES ('" + InputFormatConfig.PARTITION_SPEC + "'='" + PartitionSpecParser.toJson(spec) + "', " +
"'" + InputFormatConfig.TABLE_SCHEMA + "'='" + SchemaParser.toJson(CUSTOMER_SCHEMA) + "')");
testTables.locationForCreateTableSQL(identifier));

// Check the Iceberg table data
org.apache.iceberg.Table icebergTable = testTables.loadTable(identifier);
Assert.assertEquals(CUSTOMER_SCHEMA.asStruct(), icebergTable.schema().asStruct());
Assert.assertEquals(spec, icebergTable.spec());

testTables.appendIcebergTable(shell.getHiveConf(), icebergTable, fileFormat, Row.of("Alice"),
Arrays.asList(CUSTOMER_RECORDS.get(0)));
testTables.appendIcebergTable(shell.getHiveConf(), icebergTable, fileFormat, Row.of("Bob"),
Arrays.asList(CUSTOMER_RECORDS.get(1)));
testTables.appendIcebergTable(shell.getHiveConf(), icebergTable, fileFormat, Row.of("Trudy"),
Arrays.asList(CUSTOMER_RECORDS.get(2)));

List<Object[]> descRows = shell.executeStatement("SELECT * FROM default.customers ORDER BY customer_id DESC");

Assert.assertEquals(3, descRows.size());
Assert.assertArrayEquals(new Object[] {2L, "Trudy", "Pink"}, descRows.get(0));
Assert.assertArrayEquals(new Object[] {1L, "Bob", "Green"}, descRows.get(1));
Assert.assertArrayEquals(new Object[] {0L, "Alice", "Brown"}, descRows.get(2));
Assert.assertEquals(expectedSchema.asStruct(), icebergTable.schema().asStruct());
Assert.assertEquals(expectedSpec, icebergTable.spec());
}

@Test
Expand Down Expand Up @@ -1027,4 +1034,45 @@ public void testStructOfStructsInTable() throws IOException {
Assert.assertEquals(expectedInnerStruct.getField("value"), queryResult.get(0)[1]);
}
}

private static String createSQLForPartitionedTableBySchema(TestTables testTables, TableIdentifier identifier) {
return "CREATE EXTERNAL TABLE " + identifier.namespace() + "." + identifier.name() +
" (customer_id BIGINT, first_name STRING) PARTITIONED BY (last_name STRING) " +
"STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' " +
testTables.locationForCreateTableSQL(identifier);
}

private static String createSQLForPartitionedTableByProperty(TestTables testTables, TableIdentifier identifier) {
PartitionSpec spec = PartitionSpec.builderFor(CUSTOMER_SCHEMA).identity("last_name").build();

return "CREATE EXTERNAL TABLE " + identifier.namespace() + "." + identifier.name() +
" STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' " +
testTables.locationForCreateTableSQL(identifier) +
"TBLPROPERTIES ('" + InputFormatConfig.PARTITION_SPEC + "'='" + PartitionSpecParser.toJson(spec) + "', " +
"'" + InputFormatConfig.TABLE_SCHEMA + "'='" + SchemaParser.toJson(CUSTOMER_SCHEMA) + "')";
}

private void runPartitionedWriteTest(TableIdentifier identifier, String createSQL) throws IOException {
PartitionSpec spec = PartitionSpec.builderFor(CUSTOMER_SCHEMA).identity("last_name").build();

shell.executeStatement(createSQL);

org.apache.iceberg.Table icebergTable = testTables.loadTable(identifier);
Assert.assertEquals(CUSTOMER_SCHEMA.asStruct(), icebergTable.schema().asStruct());
Assert.assertEquals(spec, icebergTable.spec());

testTables.appendIcebergTable(shell.getHiveConf(), icebergTable, fileFormat, Row.of("Brown"),
Arrays.asList(CUSTOMER_RECORDS.get(0)));
testTables.appendIcebergTable(shell.getHiveConf(), icebergTable, fileFormat, Row.of("Green"),
Arrays.asList(CUSTOMER_RECORDS.get(1)));
testTables.appendIcebergTable(shell.getHiveConf(), icebergTable, fileFormat, Row.of("Pink"),
Arrays.asList(CUSTOMER_RECORDS.get(2)));

List<Object[]> descRows = shell.executeStatement("SELECT * FROM default.customers ORDER BY customer_id DESC");

Assert.assertEquals(3, descRows.size());
Assert.assertArrayEquals(new Object[] {2L, "Trudy", "Pink"}, descRows.get(0));
Assert.assertArrayEquals(new Object[] {1L, "Bob", "Green"}, descRows.get(1));
Assert.assertArrayEquals(new Object[] {0L, "Alice", "Brown"}, descRows.get(2));
}
}