Skip to content

Hive: Expose default partition spec and sort order in HMS. #4546

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
szehon-ho merged 12 commits into from
Apr 26, 2022
Merged

Hive: Expose default partition spec and sort order in HMS. #4546

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
e5cf435
Expose default partition spec and sort order in HMS.
flyrain Apr 12, 2022
409872b
Resolve comments.
flyrain Apr 14, 2022
9bf2a0f
Add source name in spec json string
flyrain Apr 21, 2022
8016618
Add check for thread hold
flyrain Apr 21, 2022
65cd12c
Resolve comments.
flyrain Apr 22, 2022
cceba61
Fix the test failure.
flyrain Apr 22, 2022
88f8c7f
Resolve comments.
flyrain Apr 22, 2022
7c9f0d0
Resolve comments.
flyrain Apr 22, 2022
49796e7
Resolve comments.
flyrain Apr 25, 2022
fcb5d48
Resolve comments
flyrain Apr 25, 2022
72dc061
Remove the spy in tests.
flyrain Apr 25, 2022
0c178a7
Remove the spy in tests.
flyrain Apr 25, 2022
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
51 changes: 35 additions & 16 deletions core/src/main/java/org/apache/iceberg/PartitionSpecParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,42 +39,39 @@ private PartitionSpecParser() {
private static final String SPEC_ID = "spec-id";
private static final String FIELDS = "fields";
private static final String SOURCE_ID = "source-id";
private static final String SOURCE_NAME = "source-name";
private static final String FIELD_ID = "field-id";
private static final String TRANSFORM = "transform";
private static final String NAME = "name";

public static void toJson(PartitionSpec spec, JsonGenerator generator) throws IOException {
toJson(spec.toUnbound(), generator);
public static String toJsonWithSourceName(PartitionSpec spec) {
return toJson(spec.toUnbound(), spec.schema(), false);
}

public static String toJson(PartitionSpec spec) {
return toJson(spec, false);
}

public static String toJson(PartitionSpec spec, boolean pretty) {
return toJson(spec.toUnbound(), pretty);
public static String toJson(UnboundPartitionSpec spec) {
return toJson(spec, false);
}

public static void toJson(UnboundPartitionSpec spec, JsonGenerator generator) throws IOException {
generator.writeStartObject();
generator.writeNumberField(SPEC_ID, spec.specId());
generator.writeFieldName(FIELDS);
toJsonFields(spec, generator);
generator.writeEndObject();
public static String toJson(PartitionSpec spec, boolean pretty) {
return toJson(spec.toUnbound(), null, pretty);
}

public static String toJson(UnboundPartitionSpec spec) {
return toJson(spec, false);
public static String toJson(UnboundPartitionSpec spec, boolean pretty) {
return toJson(spec, null, pretty);
}

public static String toJson(UnboundPartitionSpec spec, boolean pretty) {
private static String toJson(UnboundPartitionSpec spec, Schema schema, boolean pretty) {
try {
StringWriter writer = new StringWriter();
JsonGenerator generator = JsonUtil.factory().createGenerator(writer);
if (pretty) {
generator.useDefaultPrettyPrinter();
}
toJson(spec, generator);
toJson(spec, generator, schema);
generator.flush();
return writer.toString();

Expand All @@ -83,6 +80,22 @@ public static String toJson(UnboundPartitionSpec spec, boolean pretty) {
}
}

public static void toJson(PartitionSpec spec, JsonGenerator generator) throws IOException {
toJson(spec.toUnbound(), generator);
}

public static void toJson(UnboundPartitionSpec spec, JsonGenerator generator) throws IOException {
toJson(spec, generator, null);
}

private static void toJson(UnboundPartitionSpec spec, JsonGenerator generator, Schema schema) throws IOException {
generator.writeStartObject();
generator.writeNumberField(SPEC_ID, spec.specId());
generator.writeFieldName(FIELDS);
toJsonFields(spec, generator, schema);
generator.writeEndObject();
}

Comment thread
flyrain marked this conversation as resolved.
Outdated
public static PartitionSpec fromJson(Schema schema, JsonNode json) {
return fromJson(json).bind(schema);
}
Expand Down Expand Up @@ -112,16 +125,22 @@ public static PartitionSpec fromJson(Schema schema, String json) {
}

static void toJsonFields(PartitionSpec spec, JsonGenerator generator) throws IOException {
toJsonFields(spec.toUnbound(), generator);
toJsonFields(spec.toUnbound(), generator, null);
}

static void toJsonFields(UnboundPartitionSpec spec, JsonGenerator generator) throws IOException {
static void toJsonFields(UnboundPartitionSpec spec, JsonGenerator generator, Schema schema) throws IOException {
generator.writeStartArray();
for (UnboundPartitionSpec.UnboundPartitionField field : spec.fields()) {
generator.writeStartObject();
generator.writeStringField(NAME, field.name());
generator.writeStringField(TRANSFORM, field.transformAsString());
generator.writeNumberField(SOURCE_ID, field.sourceId());
if (schema != null) {
Types.NestedField nestedField = schema.findField(field.sourceId());
if (nestedField != null) {
generator.writeStringField(SOURCE_NAME, nestedField.name());
}
}
generator.writeNumberField(FIELD_ID, field.partitionId());
generator.writeEndObject();
}
Expand Down
26 changes: 14 additions & 12 deletions core/src/main/java/org/apache/iceberg/TableProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,40 +41,40 @@ private TableProperties() {
public static final String FORMAT_VERSION = "format-version";

/**
* Reserved table property for UUID.
* <p>
* This reserved property is used to store the UUID of the table.
* Reserved table property for table UUID.
*/
public static final String UUID = "uuid";

/**
* Reserved table property for the total number of snapshots.
* <p>
* This reserved property is used to store the total number of snapshots.
*/
public static final String SNAPSHOT_COUNT = "snapshot-count";

/**
* Reserved table property for current snapshot summary.
* <p>
* This reserved property is used to store the current snapshot summary.
*/
public static final String CURRENT_SNAPSHOT_SUMMARY = "current-snapshot-summary";

/**
* Reserved table property for current snapshot id.
* <p>
* This reserved property is used to store the current snapshot id.
*/
public static final String CURRENT_SNAPSHOT_ID = "current-snapshot-id";

/**
* Reserved table property for current snapshot timestamp.
* <p>
* This reserved property is used to store the current snapshot timestamp.
*/
public static final String CURRENT_SNAPSHOT_TIMESTAMP = "current-snapshot-timestamp-ms";

/**
* Reserved table property for the JSON representation of current partition spec.
*/
public static final String DEFAULT_PARTITION_SPEC = "default-partition-spec";

@kbendick kbendick Apr 14, 2022

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.

It might be helpful to clarify what is meant by default for both of these.

Each javadoc states that it’s for the “default” spec / sort order, but I’m still not entirely sure what that means.

Is it the current sort order / partition spec that would be used if the user doesn’t override it for an individual query ?

@kbendick kbendick Apr 14, 2022

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.

Maybe something like “JSON representation of the table’s current configured partition spec, which will be used if not overridden for individual writes”. Kind of wordy but something along those lines would be helpful for me if quickly looking through the JavaDocs etc. Will leave that decision to you though.

@flyrain flyrain Apr 14, 2022

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.

Yes, it is for the current partition spec and sort order. Make sense to me. Will make the change.

@szehon-ho szehon-ho Apr 22, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I think you reverted too many? Should be 'current'.

@flyrain flyrain Apr 22, 2022

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.

I was trying to be consistent with the name in metadata.json, which is default-partition-spec. The same for sort order. I changed the comments though.

@szehon-ho szehon-ho Apr 22, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

OK I see, yea I always find it confusing.

In the comment, maybe we can add that they are equivalent, otherwise the comment is even more confusing:

Reserved table property for the JSON representation of current (default) schema.

@flyrain flyrain Apr 22, 2022

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.

It is confusing. I like the current more. I keep the original name just for consistency.


/**
* Reserved table property for the JSON representation of current sort order.
*/
public static final String DEFAULT_SORT_ORDER = "default-sort-order";

/**
* Reserved Iceberg table properties list.
* <p>
Expand All @@ -87,7 +87,9 @@ private TableProperties() {
SNAPSHOT_COUNT,
CURRENT_SNAPSHOT_ID,
CURRENT_SNAPSHOT_SUMMARY,
CURRENT_SNAPSHOT_TIMESTAMP
CURRENT_SNAPSHOT_TIMESTAMP,
DEFAULT_PARTITION_SPEC,
DEFAULT_SORT_ORDER
);

public static final String COMMIT_NUM_RETRIES = "commit.retry.num-retries";
Expand Down
2 changes: 2 additions & 0 deletions core/src/test/java/org/apache/iceberg/TestPartitionSpecParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ public void testToJsonForV1Table() {
" } ]\n" +
"}";
Assert.assertEquals(expected, PartitionSpecParser.toJson(table.spec(), true));
Assert.assertTrue("Json must contain source name",
PartitionSpecParser.toJsonWithSourceName(table.spec()).contains("\"source-name\":\"data\""));

PartitionSpec spec = PartitionSpec.builderFor(table.schema())
.bucket("id", 8)
Expand Down
18 changes: 18 additions & 0 deletions hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,8 +52,10 @@
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;

@szehon-ho szehon-ho Apr 25, 2022 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Two other suggestion for this class: can we add in comment of "HIVE_TABLE_PROPERTY_MAX_SIZE" , one more sentence to let user know how to turn off feature?

// set to 0 to not expose Iceberg metadata in HMS Table properties

And also, a precondition in HiveTableOperations constructor to check if value is non-negative.

@flyrain flyrain Apr 25, 2022

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.

See #4546 (comment).

@flyrain flyrain Apr 25, 2022

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.

Added the comment. Negative is fine, right?

@szehon-ho szehon-ho Apr 26, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I think probably better to disallow negative as it makes little sense? But to me its ok either way.

@flyrain flyrain Apr 26, 2022

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.

Would throwing exception be too much in that case? May just log a warning.

import org.apache.iceberg.BaseMetastoreTableOperations;
import org.apache.iceberg.ClientPool;
import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.SortOrderParser;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.exceptions.AlreadyExistsException;
Expand Down Expand Up @@ -399,6 +401,8 @@ private void setHmsTableParameters(String newMetadataLocation, Table tbl, TableM
}

setSnapshotStats(metadata, parameters);
setPartitionSpec(metadata, parameters);
setSortOrder(metadata, parameters);

tbl.setParameters(parameters);
}
Expand Down Expand Up @@ -433,6 +437,20 @@ void setSnapshotSummary(Map<String, String> parameters, Snapshot currentSnapshot
}
}

private void setPartitionSpec(TableMetadata metadata, Map<String, String> parameters) {
parameters.remove(TableProperties.DEFAULT_PARTITION_SPEC);
if (metadata.spec() != null && metadata.spec().isPartitioned()) {

@singhpk234 singhpk234 Apr 13, 2022

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.

[question] In v1 spec when a partition is dropped it is replaced by VoidTransform, if all the transforms are void we should consider it un-partitioned (This may be beyond the scope of present PR), but presently when we call isPartitioned it will return true in this case. do we want to store partition spec in this scenario ? Your thoughts.

This is based on ticket #3014 @RussellSpitzer filed a while back.

@flyrain flyrain Apr 14, 2022

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.

Thanks @singhpk234 for pointing out. #3059 is trying to fix #3014, and it is almost ready to merge. It should be fine in that case.

parameters.put(TableProperties.DEFAULT_PARTITION_SPEC, PartitionSpecParser.toJsonWithSourceName(metadata.spec()));
}
}

private void setSortOrder(TableMetadata metadata, Map<String, String> parameters) {
parameters.remove(TableProperties.DEFAULT_SORT_ORDER);
if (metadata.sortOrder() != null && metadata.sortOrder().isSorted()) {
parameters.put(TableProperties.DEFAULT_SORT_ORDER, SortOrderParser.toJson(metadata.sortOrder()));
}
Comment on lines +471 to +476

@singhpk234 singhpk234 Apr 24, 2022 •
edited
Loading

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.

[nit] should we make this :

iceberg/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java

Lines 425 to 430 in 449a743

if (summary.length() <= maxHiveTablePropertySize) {
parameters.put(TableProperties.CURRENT_SNAPSHOT_SUMMARY, summary);
} else {
LOG.warn("Not exposing the current snapshot({}) summary in HMS since it exceeds {} characters",
currentSnapshot.snapshotId(), maxHiveTablePropertySize);
}

also use this setter

@flyrain flyrain Apr 25, 2022

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.

Was trying to do that, but the warn message needs snapshot id here. But it requires changes for method setFiled() like the below, and changes for all other callers. It's like removing one duplication, but adding a few complication. I'd suggest to keep it as is.

setField(Map<String, String> parameters, String key, String value, String warnMessage)

}

private StorageDescriptor storageDescriptor(TableMetadata metadata, boolean hiveEngineEnabled) {

final StorageDescriptor storageDescriptor = new StorageDescriptor();
Expand Down
56 changes: 40 additions & 16 deletions hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,11 @@
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.SortOrderParser;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.Transaction;
Expand All @@ -60,6 +62,8 @@

import static org.apache.iceberg.NullOrder.NULLS_FIRST;
import static org.apache.iceberg.SortDirection.ASC;
import static org.apache.iceberg.TableProperties.DEFAULT_SORT_ORDER;
import static org.apache.iceberg.expressions.Expressions.bucket;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
Expand Down Expand Up @@ -232,7 +236,7 @@ public void testReplaceTxnBuilder() throws Exception {
}

@Test
public void testCreateTableDefaultSortOrder() {
public void testCreateTableDefaultSortOrder() throws Exception {
Schema schema = new Schema(
required(1, "id", Types.IntegerType.get(), "unique ID"),
required(2, "data", Types.StringType.get())
Expand All @@ -246,13 +250,16 @@ public void testCreateTableDefaultSortOrder() {
Table table = catalog.createTable(tableIdent, schema, spec);
Assert.assertEquals("Order ID must match", 0, table.sortOrder().orderId());
Assert.assertTrue("Order must unsorted", table.sortOrder().isUnsorted());

Assert.assertFalse("Must not have default sort order in catalog",
hmsTableParameters().containsKey(DEFAULT_SORT_ORDER));
} finally {
catalog.dropTable(tableIdent);
}
}

@Test
public void testCreateTableCustomSortOrder() {
public void testCreateTableCustomSortOrder() throws Exception {
Schema schema = new Schema(
required(1, "id", Types.IntegerType.get(), "unique ID"),
required(2, "data", Types.StringType.get())
Expand All @@ -277,6 +284,8 @@ public void testCreateTableCustomSortOrder() {
Assert.assertEquals("Null order must match ", NULLS_FIRST, sortOrder.fields().get(0).nullOrder());
Transform<?, ?> transform = Transforms.identity(Types.IntegerType.get());
Assert.assertEquals("Transform must match", transform, sortOrder.fields().get(0).transform());

Assert.assertEquals(SortOrderParser.toJson(table.sortOrder()), hmsTableParameters().get(DEFAULT_SORT_ORDER));
} finally {
catalog.dropTable(tableIdent);
}
Expand Down Expand Up @@ -469,13 +478,7 @@ public void testUUIDinTableProperties() throws Exception {
.withLocation(location)
.create();

String tableName = tableIdentifier.name();
org.apache.hadoop.hive.metastore.api.Table hmsTable =
metastoreClient.getTable(tableIdentifier.namespace().level(0), tableName);

// check parameters are in expected state
Map<String, String> parameters = hmsTable.getParameters();
Assert.assertNotNull(parameters.get(TableProperties.UUID));
Assert.assertNotNull(hmsTableParameters().get(TableProperties.UUID));
} finally {
catalog.dropTable(tableIdentifier);
}
Expand All @@ -495,12 +498,8 @@ public void testSnapshotStatsTableProperties() throws Exception {
.withLocation(location)
.create();

String tableName = tableIdentifier.name();
org.apache.hadoop.hive.metastore.api.Table hmsTable =
metastoreClient.getTable(tableIdentifier.namespace().level(0), tableName);

// check whether parameters are in expected state
Map<String, String> parameters = hmsTable.getParameters();
Map<String, String> parameters = hmsTableParameters();
Assert.assertEquals("0", parameters.get(TableProperties.SNAPSHOT_COUNT));
Assert.assertNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_SUMMARY));
Assert.assertNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_ID));
Expand All @@ -517,8 +516,7 @@ public void testSnapshotStatsTableProperties() throws Exception {
icebergTable.newFastAppend().appendFile(file).commit();

// check whether parameters are in expected state
hmsTable = metastoreClient.getTable(tableIdentifier.namespace().level(0), tableName);
parameters = hmsTable.getParameters();
parameters = hmsTableParameters();
Assert.assertEquals("1", parameters.get(TableProperties.SNAPSHOT_COUNT));
String summary = JsonUtil.mapper().writeValueAsString(icebergTable.currentSnapshot().summary());
Assert.assertEquals(summary, parameters.get(TableProperties.CURRENT_SNAPSHOT_SUMMARY));
Expand Down Expand Up @@ -562,6 +560,32 @@ public void testSetSnapshotSummary() throws Exception {
Assert.assertEquals("The snapshot summary must not be in parameters due to the size limit", 0, parameters.size());
}

@Test
public void testSetDefaultPartitionSpec() throws Exception {
Schema schema = new Schema(
required(1, "id", Types.IntegerType.get(), "unique ID"),
required(2, "data", Types.StringType.get())
);
TableIdentifier tableIdent = TableIdentifier.of(DB_NAME, "tbl");

try {
Table table = catalog.buildTable(tableIdent, schema).create();
Assert.assertFalse("Must not have default partition spec",
hmsTableParameters().containsKey(TableProperties.DEFAULT_PARTITION_SPEC));

table.updateSpec().addField(bucket("data", 16)).commit();
Assert.assertEquals(PartitionSpecParser.toJsonWithSourceName(table.spec()),
hmsTableParameters().get(TableProperties.DEFAULT_PARTITION_SPEC));
} finally {
catalog.dropTable(tableIdent);
}
}

private Map<String, String> hmsTableParameters() throws TException {
org.apache.hadoop.hive.metastore.api.Table hmsTable = metastoreClient.getTable(DB_NAME, "tbl");
return hmsTable.getParameters();
}

@Test
public void testConstructorWarehousePathWithEndSlash() {
HiveCatalog catalogWithSlash = new HiveCatalog();
Expand Down