You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by "ayushtkn (via GitHub)" <gi...@apache.org> on 2023/04/10 12:37:12 UTC
[GitHub] [hive] ayushtkn commented on a diff in pull request #4131: HIVE-27158: Store hive columns stats in puffin files for iceberg tables
ayushtkn commented on code in PR #4131:
URL: https://github.com/apache/hive/pull/4131#discussion_r1161623163
##########
common/src/java/org/apache/hadoop/hive/conf/HiveConf.java:
##########
@@ -2205,9 +2205,8 @@ public static enum ConfVars {
"padding tolerance config (hive.exec.orc.block.padding.tolerance)."),
HIVE_ORC_CODEC_POOL("hive.use.orc.codec.pool", false,
"Whether to use codec pool in ORC. Disable if there are bugs with codec reuse."),
- HIVE_USE_STATS_FROM("hive.use.stats.from","iceberg","Use stats from iceberg table snapshot for query " +
- "planning. This has three values metastore, puffin and iceberg"),
-
+ HIVE_ICEBERG_STATS_SOURCE("hive.iceberg.stats.source","iceberg","Use stats from iceberg table snapshot for query " +
+ "planning. This has three values metastore and iceberg"),
Review Comment:
> This has three values metastore and iceberg
what is the third value,?
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
Review Comment:
can use ```canSetColStatistics()```
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ String statsPath = getStatsPath(table).toString();
+ LOG.info("Using stats from puffin file at:" + statsPath);
Review Comment:
Logger format:
```
LOG.info("Using stats from puffin file at: {}", statsPath);
```
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ String statsPath = getStatsPath(table).toString();
+ LOG.info("Using stats from puffin file at:" + statsPath);
+ try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath)).build()) {
+ List<BlobMetadata> blobMetadata = reader.fileMetadata().blobs();
+ Map<BlobMetadata, List<ColumnStatistics>> collect =
+ Streams.stream(reader.readAll(blobMetadata)).collect(Collectors.toMap(Pair::first,
+ blobMetadataByteBufferPair -> SerializationUtils.deserialize(
+ ByteBuffers.toByteArray(blobMetadataByteBufferPair.second()))));
+ return collect.get(blobMetadata.get(0)).get(0).getStatsObj();
+ } catch (IOException e) {
+ LOG.error(String.valueOf(e));
+ }
+ return null;
+ }
+
+
+ @Override
+ public boolean setColStatistics(org.apache.hadoop.hive.ql.metadata.Table table,
+ List<ColumnStatistics> colStats) {
+ TableDesc tableDesc = Utilities.getTableDesc(table);
+ Table tbl = Catalogs.loadTable(conf, tableDesc.getProperties());
+ String snapshotId = tbl.name() + tbl.currentSnapshot().snapshotId();
+ byte[] serializeColStats = SerializationUtils.serialize((Serializable) colStats);
+
+ try (PuffinWriter writer = Puffin.write(tbl.io().newOutputFile(getStatsPath(tbl).toString()))
+ .createdBy("Hive").build()) {
+ writer.add(
+ new Blob(
+ tbl.name() + "-" + snapshotId,
+ ImmutableList.of(1),
+ tbl.currentSnapshot().snapshotId(),
+ tbl.currentSnapshot().sequenceNumber(),
+ ByteBuffer.wrap(serializeColStats),
+ PuffinCompressionCodec.NONE,
+ ImmutableMap.of()));
+ writer.finish();
+ } catch (IOException e) {
+ LOG.error(String.valueOf(e));
+ }
+ return false;
+ }
+
+ private String getStatsSource() {
+ return HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_ICEBERG_STATS_SOURCE, "metastore").toLowerCase();
+ }
Review Comment:
I don't understand this, why the default is here ``metastore``? when the config has default set as Iceberg. Who is using that default then
```
HIVE_ICEBERG_STATS_SOURCE("hive.iceberg.stats.source","iceberg",
```
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
Review Comment:
Change to ``Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());``
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
Review Comment:
Log the trace as well, rather than just the message. Along with the table name and the stats path and the snapshot id
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ String statsPath = getStatsPath(table).toString();
+ LOG.info("Using stats from puffin file at:" + statsPath);
+ try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath)).build()) {
+ List<BlobMetadata> blobMetadata = reader.fileMetadata().blobs();
+ Map<BlobMetadata, List<ColumnStatistics>> collect =
+ Streams.stream(reader.readAll(blobMetadata)).collect(Collectors.toMap(Pair::first,
+ blobMetadataByteBufferPair -> SerializationUtils.deserialize(
+ ByteBuffers.toByteArray(blobMetadataByteBufferPair.second()))));
+ return collect.get(blobMetadata.get(0)).get(0).getStatsObj();
+ } catch (IOException e) {
+ LOG.error(String.valueOf(e));
+ }
+ return null;
+ }
+
+
+ @Override
+ public boolean setColStatistics(org.apache.hadoop.hive.ql.metadata.Table table,
+ List<ColumnStatistics> colStats) {
+ TableDesc tableDesc = Utilities.getTableDesc(table);
+ Table tbl = Catalogs.loadTable(conf, tableDesc.getProperties());
+ String snapshotId = tbl.name() + tbl.currentSnapshot().snapshotId();
Review Comment:
can currentSnapshot be ``null``? like empty table and then somebody shoots a CLI command to compute statistics?
##########
ql/src/java/org/apache/hadoop/hive/ql/stats/StatsUtils.java:
##########
@@ -1069,8 +1069,12 @@ public static List<ColStatistics> getTableColumnStats(
}
if (fetchColStats && !colStatsToRetrieve.isEmpty()) {
try {
- List<ColumnStatisticsObj> colStat = Hive.get().getTableColumnStatistics(
- dbName, tabName, colStatsToRetrieve, false);
+ List<ColumnStatisticsObj> colStat;
+ if (table != null && table.isNonNative() && table.getStorageHandler().canProvideColStatistics(table)) {
Review Comment:
table can not be `null` at this point
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
Review Comment:
Change to `` Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());``
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ String statsPath = getStatsPath(table).toString();
+ LOG.info("Using stats from puffin file at:" + statsPath);
+ try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath)).build()) {
+ List<BlobMetadata> blobMetadata = reader.fileMetadata().blobs();
+ Map<BlobMetadata, List<ColumnStatistics>> collect =
+ Streams.stream(reader.readAll(blobMetadata)).collect(Collectors.toMap(Pair::first,
+ blobMetadataByteBufferPair -> SerializationUtils.deserialize(
+ ByteBuffers.toByteArray(blobMetadataByteBufferPair.second()))));
+ return collect.get(blobMetadata.get(0)).get(0).getStatsObj();
+ } catch (IOException e) {
+ LOG.error(String.valueOf(e));
+ }
+ return null;
+ }
+
+
+ @Override
+ public boolean setColStatistics(org.apache.hadoop.hive.ql.metadata.Table table,
+ List<ColumnStatistics> colStats) {
+ TableDesc tableDesc = Utilities.getTableDesc(table);
+ Table tbl = Catalogs.loadTable(conf, tableDesc.getProperties());
+ String snapshotId = tbl.name() + tbl.currentSnapshot().snapshotId();
+ byte[] serializeColStats = SerializationUtils.serialize((Serializable) colStats);
+
+ try (PuffinWriter writer = Puffin.write(tbl.io().newOutputFile(getStatsPath(tbl).toString()))
+ .createdBy("Hive").build()) {
Review Comment:
Use constant
```
Constants.HIVE_ENGINE
```
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ String statsPath = getStatsPath(table).toString();
+ LOG.info("Using stats from puffin file at:" + statsPath);
+ try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath)).build()) {
+ List<BlobMetadata> blobMetadata = reader.fileMetadata().blobs();
+ Map<BlobMetadata, List<ColumnStatistics>> collect =
+ Streams.stream(reader.readAll(blobMetadata)).collect(Collectors.toMap(Pair::first,
+ blobMetadataByteBufferPair -> SerializationUtils.deserialize(
+ ByteBuffers.toByteArray(blobMetadataByteBufferPair.second()))));
+ return collect.get(blobMetadata.get(0)).get(0).getStatsObj();
+ } catch (IOException e) {
+ LOG.error(String.valueOf(e));
+ }
+ return null;
+ }
+
+
+ @Override
+ public boolean setColStatistics(org.apache.hadoop.hive.ql.metadata.Table table,
+ List<ColumnStatistics> colStats) {
+ TableDesc tableDesc = Utilities.getTableDesc(table);
+ Table tbl = Catalogs.loadTable(conf, tableDesc.getProperties());
Review Comment:
Use ``IcebergTableUtil`` to fetch the table, It has cache, fetching and reading the table metadata multiple times have severe performance penalties, cache the table and use it from cache, unless necessary
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -361,6 +378,83 @@ private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return table;
}
+
+ @Override
+ public boolean canSetColStatistics() {
+ return getStatsSource().equals(ICEBERG);
+ }
+
+ @Override
+ public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ if (table.currentSnapshot() != null) {
+ Path statsPath = getStatsPath(table);
+ if (getStatsSource().equals(ICEBERG)) {
+ try (FileSystem fs = statsPath.getFileSystem(conf)) {
+ if (fs.exists(statsPath)) {
+ return true;
+ }
+ } catch (IOException e) {
+ LOG.warn(e.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Table table = Catalogs.loadTable(conf, Utilities.getTableDesc(hmsTable).getProperties());
+ String statsPath = getStatsPath(table).toString();
+ LOG.info("Using stats from puffin file at:" + statsPath);
+ try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath)).build()) {
+ List<BlobMetadata> blobMetadata = reader.fileMetadata().blobs();
+ Map<BlobMetadata, List<ColumnStatistics>> collect =
+ Streams.stream(reader.readAll(blobMetadata)).collect(Collectors.toMap(Pair::first,
+ blobMetadataByteBufferPair -> SerializationUtils.deserialize(
+ ByteBuffers.toByteArray(blobMetadataByteBufferPair.second()))));
+ return collect.get(blobMetadata.get(0)).get(0).getStatsObj();
+ } catch (IOException e) {
+ LOG.error(String.valueOf(e));
+ }
+ return null;
Review Comment:
Why don't we throw exception here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org