Skip to content

Hive: Expose Snapshot Stats in HMS. #4456

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 13 commits into from
Apr 8, 2022
Merged

Hive: Expose Snapshot Stats in HMS. #4456

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
98a9b73
Core: Add Reserved Snapshot Stats Table Property and Expose in HMS.
flyrain Mar 21, 2022
5fe283c
Fix the unit test failure
flyrain Mar 31, 2022
c7b0e62
Doesn't expose the summary if its length is more than 4000 chars.
flyrain Apr 4, 2022
1d96a1b
Doesn't expose the summary if its length is more than 4000 chars.
flyrain Apr 4, 2022
8eac228
Resolve comments
flyrain Apr 6, 2022
3a11d9b
Resolve comments
flyrain Apr 6, 2022
72bd3be
Simplify the test case testSetSnapshotSummary
flyrain Apr 6, 2022
4daa05a
Allow user to config the max size
flyrain Apr 7, 2022
8f5edaa
Resolve comments
flyrain Apr 7, 2022
cd726fe
Resolve comments
flyrain Apr 7, 2022
1e0d23f
Resolve comments
flyrain Apr 7, 2022
a3bea2f
Resolve comments
flyrain Apr 7, 2022
617d665
Resolve comments
flyrain Apr 8, 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
34 changes: 33 additions & 1 deletion 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 @@ -47,6 +47,34 @@ private TableProperties() {
*/
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 Iceberg table properties list.
* <p>
Expand All @@ -55,7 +83,11 @@ private TableProperties() {
*/
public static final Set<String> RESERVED_PROPERTIES = ImmutableSet.of(
FORMAT_VERSION,
UUID
UUID,
SNAPSHOT_COUNT,
CURRENT_SNAPSHOT_ID,
CURRENT_SNAPSHOT_SUMMARY,
CURRENT_SNAPSHOT_TIMESTAMP
);

public static final String COMMIT_NUM_RETRIES = "commit.retry.num-retries";
Expand Down
20 changes: 20 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 @@ -19,6 +19,7 @@

package org.apache.iceberg.hive;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import java.net.InetAddress;
Expand Down Expand Up @@ -71,6 +72,7 @@
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.Maps;
import org.apache.iceberg.util.JsonUtil;
import org.apache.iceberg.util.Tasks;
import org.apache.thrift.TException;
import org.slf4j.Logger;
Expand Down Expand Up @@ -402,9 +404,27 @@ private void setHmsTableParameters(String newMetadataLocation, Table tbl, TableM
parameters.put(StatsSetupConst.TOTAL_SIZE, summary.get(SnapshotSummary.TOTAL_FILE_SIZE_PROP));
}

setSnapshotStats(metadata, parameters);

tbl.setParameters(parameters);
}

private void setSnapshotStats(TableMetadata metadata, Map<String, String> parameters) {
Snapshot currentSnapshot = metadata.currentSnapshot();

@szehon-ho szehon-ho Apr 5, 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 we need a way to clear the properties if they are not there anymore in current snapshot (like original code setHmsProperties). For example if the summary becomes suddenly over the length in new snapshot, then it will not override the old summary and would be wrong.

@flyrain flyrain Apr 6, 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.

Fixed in the new commit.

if (currentSnapshot != null) {
parameters.put(TableProperties.CURRENT_SNAPSHOT_ID, String.valueOf(currentSnapshot.snapshotId()));
parameters.put(TableProperties.CURRENT_SNAPSHOT_TIMESTAMP, String.valueOf(currentSnapshot.timestampMillis()));
try {
parameters.put(TableProperties.CURRENT_SNAPSHOT_SUMMARY,
JsonUtil.mapper().writeValueAsString(currentSnapshot.summary()));

@singhpk234 singhpk234 Apr 1, 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.

[question] should we take size of summary, also into consideration, As if when we should keep it or is it always optimal to keep it ?

As per my understanding and I may be wrong here, The scenario I have in mind (not sure how practical it is), If some one want's to track changed partition metrics in summary and snapshot had a lot of changed partition and we set CHANGED_PARTITION_COUNT_PROP to a high value, In this scenario the json size could be very large, which is okayish to keep in File, but in DB I am not sure. Your thoughts ?

iceberg/core/src/main/java/org/apache/iceberg/SnapshotSummary.java

Lines 183 to 190 in 86b42fe

setIf(trustPartitionMetrics, builder, CHANGED_PARTITION_COUNT_PROP, changedPartitions.size());
if (trustPartitionMetrics && changedPartitions.size() <= maxChangedPartitionsForSummaries) {
setIf(changedPartitions.size() > 0, builder, PARTITION_SUMMARY_PROP, "true");
for (String key : changedPartitions) {
setIf(key != null, builder, CHANGED_PARTITION_PREFIX + key, partitionSummary(partitionMetrics.get(key)));
}
}

@flyrain flyrain Apr 1, 2022 •
edited
Loading

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.

Good question! According to https://issues.apache.org/jira/browse/HIVE-12274 and https://issues.apache.org/jira/browse/HIVE-14145, we can at least store 4000 chars as the values before HMS3.x, and more in HMS3.x depending on the under layer RDMS, at least 32672 bytes. I'm going to cap the summary at 4000 chars.

} catch (JsonProcessingException e) {
LOG.warn("Failed to convert snapshot summary to a json string", e);

@singhpk234 singhpk234 Apr 1, 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.

[minor] Does it makes sense to capture, snapshotId as well for which we got this exception ?

@flyrain flyrain Apr 1, 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.

Make sense, will print out snapshot id in the warn message.

}
}

parameters.put(TableProperties.SNAPSHOT_COUNT, String.valueOf(metadata.snapshots().size()));
}

private StorageDescriptor storageDescriptor(TableMetadata metadata, boolean hiveEngineEnabled) {

final StorageDescriptor storageDescriptor = new StorageDescriptor();
Expand Down
51 changes: 51 additions & 0 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 @@ -21,9 +21,13 @@

import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.CachingCatalog;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
Expand Down Expand Up @@ -468,4 +472,51 @@ public void testUUIDinTableProperties() throws Exception {
catalog.dropTable(tableIdentifier);
}
}

@Test
public void testSnapshotStatsTableProperties() throws Exception {

@RussellSpitzer RussellSpitzer Apr 6, 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.

We should probably have a test with the string exceeding the property size threshold, checking to make sure the behavior is as we expect.

@flyrain flyrain Apr 6, 2022 •
edited
Loading

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 a test case for that

Schema schema = new Schema(
required(1, "id", Types.IntegerType.get(), "unique ID"),
required(2, "data", Types.StringType.get())
);
TableIdentifier tableIdentifier = TableIdentifier.of(DB_NAME, "tbl");
String location = temp.newFolder("tbl").toString();

try {
catalog.buildTable(tableIdentifier, schema)
.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();
Assert.assertEquals("0", parameters.get(TableProperties.SNAPSHOT_COUNT));

@szehon-ho szehon-ho Apr 5, 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.

Can we put some error messages in these asserts, like in other tests?

@flyrain flyrain Apr 6, 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'm not sure it is necessary. The code here can explain itself quite well.

Assert.assertNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_SUMMARY));
Assert.assertNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_ID));
Assert.assertNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_TIMESTAMP));

// create a snapshot
Table icebergTable = catalog.loadTable(tableIdentifier);
String fileName = UUID.randomUUID().toString();
DataFile file = DataFiles.builder(icebergTable.spec())
.withPath(FileFormat.PARQUET.addExtension(fileName))
.withRecordCount(2)
.withFileSizeInBytes(0)
.build();
icebergTable.newFastAppend().appendFile(file).commit();

// check whether parameters are in expected state
hmsTable = metastoreClient.getTable(tableIdentifier.namespace().level(0), tableName);
parameters = hmsTable.getParameters();
Assert.assertEquals("1", parameters.get(TableProperties.SNAPSHOT_COUNT));
Assert.assertNotNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_SUMMARY));
Assert.assertNotNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_ID));
Assert.assertNotNull(parameters.get(TableProperties.CURRENT_SNAPSHOT_TIMESTAMP));
} finally {
catalog.dropTable(tableIdentifier);
}
}
}
4 changes: 2 additions & 2 deletions mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerNoScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -624,7 +624,7 @@ public void testIcebergAndHmsTableProperties() throws Exception {
Assert.assertEquals(expectedIcebergProperties, icebergTable.properties());

if (Catalogs.hiveCatalog(shell.getHiveConf(), tableProperties)) {
Assert.assertEquals(11, hmsParams.size());
Assert.assertEquals(12, hmsParams.size());
Comment thread
flyrain marked this conversation as resolved.
Assert.assertEquals("initial_val", hmsParams.get("custom_property"));
Assert.assertEquals("TRUE", hmsParams.get(InputFormatConfig.EXTERNAL_TABLE_PURGE));
Assert.assertEquals("TRUE", hmsParams.get("EXTERNAL"));
Expand Down Expand Up @@ -662,7 +662,7 @@ public void testIcebergAndHmsTableProperties() throws Exception {
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

if (Catalogs.hiveCatalog(shell.getHiveConf(), tableProperties)) {
Assert.assertEquals(14, hmsParams.size()); // 2 newly-added properties + previous_metadata_location prop
Assert.assertEquals(15, hmsParams.size()); // 2 newly-added properties + previous_metadata_location prop
Assert.assertEquals("true", hmsParams.get("new_prop_1"));
Assert.assertEquals("false", hmsParams.get("new_prop_2"));
Assert.assertEquals("new_val", hmsParams.get("custom_property"));
Expand Down