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 6 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
42 changes: 42 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 All @@ -90,6 +92,9 @@ public class HiveTableOperations extends BaseMetastoreTableOperations {
private static final String HIVE_LOCK_CHECK_MAX_WAIT_MS = "iceberg.hive.lock-check-max-wait-ms";
private static final String HIVE_ICEBERG_METADATA_REFRESH_MAX_RETRIES = "iceberg.hive.metadata-refresh-max-retries";
private static final String HIVE_TABLE_LEVEL_LOCK_EVICT_MS = "iceberg.hive.table-level-lock-evict-ms";
private static final String HIVE_LEGACY_TABLE_PARAMETER_SIZE = "iceberg.hive.legacy.table.parameter.size";

@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.

The current name doesn't look like a boolean type. I'm open to suggestion. iceberg.hive.is.legacy.table.parameter.size?

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

hive.use.legacy.table.parameter.size?

private static final long HIVE_TABLE_PARAMETER_SIZE_MAX_DEFAULT = 32672;

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

Should we add a comment, giving some reference to when it was changed in Hive?

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

Thanks, nit: how about "iceberg.hive.max.table.parameter.size".

Also, regarding the comment, it's kind of based on backend but Hive had a global limit of 4000 regardless of database to allow it to support across different database, so maybe "For Hive versions below 2.3, max table parameter size is 4000 characters, see https://issues.apache.org/jira/browse/HIVE-12274"?

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

Made the change

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

one more nit: can the variable match the changed string

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

Actually sorry about this , I just noticed other flags, weird how they do dash after "iceberg.hive". How about HIVE_TABLE_PROPERTY_MAX_SIZE / 'iceberg.hive.table-property-max-size" (And fixes the correct term to be table-property).

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

Made the change.

private static final long HIVE_TABLE_PARAMETER_SIZE_MAX_LEGACY = 4000;
private static final long HIVE_ACQUIRE_LOCK_TIMEOUT_MS_DEFAULT = 3 * 60 * 1000; // 3 minutes
private static final long HIVE_LOCK_CHECK_MIN_WAIT_MS_DEFAULT = 50; // 50 milliseconds
private static final long HIVE_LOCK_CHECK_MAX_WAIT_MS_DEFAULT = 5 * 1000; // 5 seconds
Expand Down Expand Up @@ -148,6 +153,7 @@ private static class WaitingForLockException extends RuntimeException {
private final long lockAcquireTimeout;
private final long lockCheckMinWaitTime;
private final long lockCheckMaxWaitTime;
private final long maxHiveTableParameterSize;
private final int metadataRefreshMaxRetries;
private final FileIO fileIO;
private final ClientPool<IMetaStoreClient, TException> metaClients;
Expand All @@ -170,6 +176,8 @@ protected HiveTableOperations(Configuration conf, ClientPool metaClients, FileIO
conf.getInt(HIVE_ICEBERG_METADATA_REFRESH_MAX_RETRIES, HIVE_ICEBERG_METADATA_REFRESH_MAX_RETRIES_DEFAULT);
long tableLevelLockCacheEvictionTimeout =
conf.getLong(HIVE_TABLE_LEVEL_LOCK_EVICT_MS, HIVE_TABLE_LEVEL_LOCK_EVICT_MS_DEFAULT);
this.maxHiveTableParameterSize = conf.getBoolean(HIVE_LEGACY_TABLE_PARAMETER_SIZE, false) ?
HIVE_TABLE_PARAMETER_SIZE_MAX_LEGACY : HIVE_TABLE_PARAMETER_SIZE_MAX_DEFAULT;
initTableLevelLockCache(tableLevelLockCacheEvictionTimeout);
}

Expand Down Expand Up @@ -402,9 +410,43 @@ 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()));
setSnapshotSummary(parameters, currentSnapshot);
} else {
parameters.remove(TableProperties.CURRENT_SNAPSHOT_ID);
parameters.remove(TableProperties.CURRENT_SNAPSHOT_TIMESTAMP);
parameters.remove(TableProperties.CURRENT_SNAPSHOT_SUMMARY);
}

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

@VisibleForTesting
void setSnapshotSummary(Map<String, String> parameters, Snapshot currentSnapshot) {
try {
String summary = JsonUtil.mapper().writeValueAsString(currentSnapshot.summary());
if (summary.length() <= maxHiveTableParameterSize) {
parameters.put(TableProperties.CURRENT_SNAPSHOT_SUMMARY, summary);
} else {
parameters.remove(TableProperties.CURRENT_SNAPSHOT_SUMMARY);

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

Optional: we can actually put the parameters.remove() for all the properties above the parameters.put(), here we are doing the remove three times.

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

Good idea. Made the change.

LOG.warn("Not exposing the current snapshot({}) summary in HMS since it exceeds {} characters",
currentSnapshot.snapshotId(), maxHiveTableParameterSize);
}
} catch (JsonProcessingException e) {
parameters.remove(TableProperties.CURRENT_SNAPSHOT_SUMMARY);
LOG.warn("Failed to convert current snapshot({}) summary to a json string", currentSnapshot.snapshotId(), e);
}
}

private StorageDescriptor storageDescriptor(TableMetadata metadata, boolean hiveEngineEnabled) {

final StorageDescriptor storageDescriptor = new StorageDescriptor();
Expand Down
107 changes: 107 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 @@ -19,13 +19,20 @@

package org.apache.iceberg.hive;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.hadoop.conf.Configuration;
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.Snapshot;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
Expand All @@ -42,6 +49,7 @@
import org.apache.iceberg.transforms.Transform;
import org.apache.iceberg.transforms.Transforms;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.JsonUtil;
import org.apache.thrift.TException;
import org.junit.Assert;
import org.junit.Rule;
Expand All @@ -52,6 +60,9 @@
import static org.apache.iceberg.NullOrder.NULLS_FIRST;
import static org.apache.iceberg.SortDirection.ASC;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;

public class TestHiveCatalog extends HiveMetastoreTest {
private static ImmutableMap meta = ImmutableMap.of(
Expand Down Expand Up @@ -468,4 +479,100 @@ 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));
String summary = JsonUtil.mapper().writeValueAsString(icebergTable.currentSnapshot().summary());
Assert.assertEquals(summary, parameters.get(TableProperties.CURRENT_SNAPSHOT_SUMMARY));
long snapshotId = icebergTable.currentSnapshot().snapshotId();
Assert.assertEquals(String.valueOf(snapshotId), parameters.get(TableProperties.CURRENT_SNAPSHOT_ID));
Assert.assertEquals(String.valueOf(icebergTable.currentSnapshot().timestampMillis()),
parameters.get(TableProperties.CURRENT_SNAPSHOT_TIMESTAMP));

} finally {
catalog.dropTable(tableIdentifier);
}
}

@Test
public void testSetSnapshotSummary() throws Exception {
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 {
Table table = catalog.buildTable(tableIdentifier, schema)
.withLocation(location)
.create();

String tableName = tableIdentifier.name();
CachedClientPool spyCachedClientPool = spy(new CachedClientPool(hiveConf, Collections.emptyMap()));
Configuration conf = new Configuration();
conf.set("iceberg.hive.legacy.table.parameter.size", "true");
HiveTableOperations spyOps = spy(new HiveTableOperations(conf, spyCachedClientPool, table.io(),
catalog.name(), tableIdentifier.namespace().level(0), tableName));
Snapshot snapshot = mock(Snapshot.class);
Map<String, String> summary = Maps.newHashMap();
// create a snapshot summary whose json string size is less than the limit
for (int i = 0; i < 100; i++) {
summary.put(String.valueOf(i), "value");
}
Assert.assertTrue(JsonUtil.mapper().writeValueAsString(summary).length() < 4000);

when(snapshot.summary()).thenReturn(summary);
Map<String, String> parameter = Maps.newHashMap();
spyOps.setSnapshotSummary(parameter, snapshot);

Assert.assertEquals("The snapshot summary must be in parameters", 1, parameter.size());

// create a snapshot summary whose json string size exceeds the limit
for (int i = 0; i < 1000; i++) {
summary.put(String.valueOf(i), "value");
}
Assert.assertTrue(JsonUtil.mapper().writeValueAsString(summary).length() > 4000);
spyOps.setSnapshotSummary(parameter, snapshot);
Assert.assertEquals("The snapshot summary must not be in parameters due to the size limit", 0, parameter.size());
} 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