Repository: hive Updated Branches: refs/heads/master 974708336 -> 71dfd1d11
http://git-wip-us.apache.org/repos/asf/hive/blob/71dfd1d1/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java ---------------------------------------------------------------------- diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java index c0b1d87..b43fb5e 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java @@ -3521,6 +3521,29 @@ public class HiveMetaStore extends ThriftHiveMetastore { new AddPartitionEvent(tbl, newParts, true, this)); } + if (!listeners.isEmpty()) { + MetaStoreListenerNotifier.notifyEvent(listeners, + EventType.ADD_PARTITION, + new AddPartitionEvent(tbl, newParts, true, this), + null, + transactionalListenerResponses, ms); + + if (!existingParts.isEmpty()) { + // The request has succeeded but we failed to add these partitions. + MetaStoreListenerNotifier.notifyEvent(listeners, + EventType.ADD_PARTITION, + new AddPartitionEvent(tbl, existingParts, false, this), + null, null, ms); + } + } + + // Update partition column statistics if available + for (Partition newPart : newParts) { + if (newPart.isSetColStats()) { + updatePartitonColStatsInternal(tbl, newPart.getColStats(), null, newPart.getWriteId()); + } + } + success = ms.commitTransaction(); } finally { if (!success) { @@ -3533,24 +3556,9 @@ public class HiveMetaStore extends ThriftHiveMetastore { new AddPartitionEvent(tbl, parts, false, this), null, null, ms); } - } else { - if (!listeners.isEmpty()) { - MetaStoreListenerNotifier.notifyEvent(listeners, - EventType.ADD_PARTITION, - new AddPartitionEvent(tbl, newParts, true, this), - null, - transactionalListenerResponses, ms); - - if (!existingParts.isEmpty()) { - // The request has succeeded but we failed to add these partitions. - MetaStoreListenerNotifier.notifyEvent(listeners, - EventType.ADD_PARTITION, - new AddPartitionEvent(tbl, existingParts, false, this), - null, null, ms); - } - } } } + return newParts; } @@ -6047,12 +6055,14 @@ public class HiveMetaStore extends ThriftHiveMetastore { if (transactionalListeners != null && !transactionalListeners.isEmpty()) { MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventType.UPDATE_PARTITION_COLUMN_STAT, - new UpdatePartitionColumnStatEvent(colStats, partVals, parameters, validWriteIds, writeId, this)); + new UpdatePartitionColumnStatEvent(colStats, partVals, parameters, tbl, validWriteIds, + writeId, this)); } if (!listeners.isEmpty()) { MetaStoreListenerNotifier.notifyEvent(listeners, EventType.UPDATE_PARTITION_COLUMN_STAT, - new UpdatePartitionColumnStatEvent(colStats, partVals, parameters, validWriteIds, writeId, this)); + new UpdatePartitionColumnStatEvent(colStats, partVals, parameters, tbl, validWriteIds, + writeId, this)); } } committed = getMS().commitTransaction(); @@ -6336,25 +6346,66 @@ public class HiveMetaStore extends ThriftHiveMetastore { @Override public List<Partition> get_partitions_by_names(final String dbName, final String tblName, final List<String> partNames) throws TException { + return get_partitions_by_names(dbName, tblName, partNames, false); + } - String[] parsedDbName = parseDbName(dbName, conf); + @Override + public GetPartitionsByNamesResult get_partitions_by_names_req(GetPartitionsByNamesRequest gpbnr) + throws TException { + List<Partition> partitions = get_partitions_by_names(gpbnr.getDb_name(), + gpbnr.getTbl_name(), gpbnr.getNames(), + gpbnr.isSetGet_col_stats() && gpbnr.isGet_col_stats()); + return new GetPartitionsByNamesResult(partitions); + } + + public List<Partition> get_partitions_by_names(final String dbName, final String tblName, + final List<String> partNames, boolean getColStats) throws TException { + + String[] dbNameParts = parseDbName(dbName, conf); + String parsedCatName = dbNameParts[CAT_NAME]; + String parsedDbName = dbNameParts[DB_NAME]; List<Partition> ret = null; Exception ex = null; - - startTableFunction("get_partitions_by_names", parsedDbName[CAT_NAME], parsedDbName[DB_NAME], - tblName); + boolean success = false; + startTableFunction("get_partitions_by_names", parsedCatName, parsedDbName, + tblName); try { - authorizeTableForPartitionMetadata(parsedDbName[CAT_NAME], parsedDbName[DB_NAME], tblName); + getMS().openTransaction(); + authorizeTableForPartitionMetadata(parsedCatName, parsedDbName, tblName); - fireReadTablePreEvent(parsedDbName[CAT_NAME], parsedDbName[DB_NAME], tblName); + fireReadTablePreEvent(parsedCatName, parsedDbName, tblName); - ret = getMS().getPartitionsByNames(parsedDbName[CAT_NAME], parsedDbName[DB_NAME], tblName, - partNames); + ret = getMS().getPartitionsByNames(parsedCatName, parsedDbName, tblName, partNames); ret = FilterUtils.filterPartitionsIfEnabled(isServerFilterEnabled, filterHook, ret); + + // If requested add column statistics in each of the partition objects + if (getColStats) { + Table table = getTable(parsedCatName, parsedDbName, tblName); + // Since each partition may have stats collected for different set of columns, we + // request them separately. + for (Partition part: ret) { + String partName = Warehouse.makePartName(table.getPartitionKeys(), part.getValues()); + List<ColumnStatistics> partColStatsList = + getMS().getPartitionColumnStatistics(parsedCatName, parsedDbName, tblName, + Collections.singletonList(partName), + StatsSetupConst.getColumnsHavingStats(part.getParameters())); + if (partColStatsList != null && !partColStatsList.isEmpty()) { + ColumnStatistics partColStats = partColStatsList.get(0); + if (partColStats != null) { + part.setColStats(partColStats); + } + } + } + } + + success = getMS().commitTransaction(); } catch (Exception e) { ex = e; rethrowException(e); } finally { + if (!success) { + getMS().rollbackTransaction(); + } endFunction("get_partitions_by_names", ret != null, ex, tblName); } return ret; http://git-wip-us.apache.org/repos/asf/hive/blob/71dfd1d1/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/events/UpdatePartitionColumnStatEvent.java ---------------------------------------------------------------------- diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/events/UpdatePartitionColumnStatEvent.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/events/UpdatePartitionColumnStatEvent.java index a61b98c..094f799 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/events/UpdatePartitionColumnStatEvent.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/events/UpdatePartitionColumnStatEvent.java @@ -22,6 +22,7 @@ import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.hive.metastore.IHMSHandler; import org.apache.hadoop.hive.metastore.api.ColumnStatistics; +import org.apache.hadoop.hive.metastore.api.Table; import java.util.List; import java.util.Map; @@ -38,6 +39,7 @@ public class UpdatePartitionColumnStatEvent extends ListenerEvent { private long writeId; private Map<String, String> parameters; private List<String> partVals; + private Table tableObj; /** * @param statsObj Columns statistics Info. @@ -48,13 +50,15 @@ public class UpdatePartitionColumnStatEvent extends ListenerEvent { * @param handler handler that is firing the event */ public UpdatePartitionColumnStatEvent(ColumnStatistics statsObj, List<String> partVals, Map<String, String> parameters, - String validWriteIds, long writeId, IHMSHandler handler) { + Table tableObj, String validWriteIds, long writeId, + IHMSHandler handler) { super(true, handler); this.partColStats = statsObj; this.validWriteIds = validWriteIds; this.writeId = writeId; this.parameters = parameters; this.partVals = partVals; + this.tableObj = tableObj; } /** @@ -62,13 +66,15 @@ public class UpdatePartitionColumnStatEvent extends ListenerEvent { * @param partVals partition names * @param handler handler that is firing the event */ - public UpdatePartitionColumnStatEvent(ColumnStatistics statsObj, List<String> partVals, IHMSHandler handler) { + public UpdatePartitionColumnStatEvent(ColumnStatistics statsObj, List<String> partVals, + Table tableObj, IHMSHandler handler) { super(true, handler); this.partColStats = statsObj; this.partVals = partVals; this.validWriteIds = null; this.writeId = 0; this.parameters = null; + this.tableObj = tableObj; } public ColumnStatistics getPartColStats() { @@ -90,4 +96,6 @@ public class UpdatePartitionColumnStatEvent extends ListenerEvent { public List<String> getPartVals() { return partVals; } + + public Table getTableObj() { return tableObj; } } http://git-wip-us.apache.org/repos/asf/hive/blob/71dfd1d1/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageBuilder.java ---------------------------------------------------------------------- diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageBuilder.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageBuilder.java index e3a91f9..10c6b44 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageBuilder.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/MessageBuilder.java @@ -300,9 +300,9 @@ public class MessageBuilder { public JSONUpdatePartitionColumnStatMessage buildUpdatePartitionColumnStatMessage(ColumnStatistics colStats, List<String> partVals, Map<String, String> parameters, - String validWriteIds, long writeId) { + Table tableObj, String validWriteIds, long writeId) { return new JSONUpdatePartitionColumnStatMessage(MS_SERVER_URL, MS_SERVICE_PRINCIPAL, now(), colStats, partVals, - parameters, validWriteIds, writeId); + parameters, tableObj, validWriteIds, writeId); } public JSONDeletePartitionColumnStatMessage buildDeletePartitionColumnStatMessage(String dbName, String colName, http://git-wip-us.apache.org/repos/asf/hive/blob/71dfd1d1/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/UpdatePartitionColumnStatMessage.java ---------------------------------------------------------------------- diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/UpdatePartitionColumnStatMessage.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/UpdatePartitionColumnStatMessage.java index d15c6e6..7eb6c07 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/UpdatePartitionColumnStatMessage.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/UpdatePartitionColumnStatMessage.java @@ -18,6 +18,8 @@ package org.apache.hadoop.hive.metastore.messaging; import org.apache.hadoop.hive.metastore.api.ColumnStatistics; +import org.apache.hadoop.hive.metastore.api.Table; + import java.util.List; import java.util.Map; @@ -39,4 +41,6 @@ public abstract class UpdatePartitionColumnStatMessage extends EventMessage { public abstract Map<String, String> getParameters(); public abstract List<String> getPartVals(); + + public abstract Table getTableObject() throws Exception; } http://git-wip-us.apache.org/repos/asf/hive/blob/71dfd1d1/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONUpdatePartitionColumnStatMessage.java ---------------------------------------------------------------------- diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONUpdatePartitionColumnStatMessage.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONUpdatePartitionColumnStatMessage.java index 3f20091..1b35df5 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONUpdatePartitionColumnStatMessage.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/messaging/json/JSONUpdatePartitionColumnStatMessage.java @@ -21,6 +21,7 @@ package org.apache.hadoop.hive.metastore.messaging.json; import com.fasterxml.jackson.annotation.JsonProperty; import org.apache.hadoop.hive.metastore.api.ColumnStatistics; +import org.apache.hadoop.hive.metastore.api.Table; import org.apache.hadoop.hive.metastore.messaging.MessageBuilder; import org.apache.hadoop.hive.metastore.messaging.UpdatePartitionColumnStatMessage; import org.apache.thrift.TException; @@ -48,6 +49,9 @@ public class JSONUpdatePartitionColumnStatMessage extends UpdatePartitionColumnS @JsonProperty private List<String> partVals; + @JsonProperty + private String tableObjJson; + /** * Default constructor, needed for Jackson. */ @@ -56,7 +60,8 @@ public class JSONUpdatePartitionColumnStatMessage extends UpdatePartitionColumnS public JSONUpdatePartitionColumnStatMessage(String server, String servicePrincipal, Long timestamp, ColumnStatistics colStats, List<String> partVals, - Map<String, String> parameters, String validWriteIds, long writeId) { + Map<String, String> parameters, + Table tableObj, String validWriteIds, long writeId) { this.timestamp = timestamp; this.server = server; this.servicePrincipal = servicePrincipal; @@ -66,6 +71,7 @@ public class JSONUpdatePartitionColumnStatMessage extends UpdatePartitionColumnS this.partVals = partVals; try { this.colStatsJson = MessageBuilder.createTableColumnStatJson(colStats); + this.tableObjJson = MessageBuilder.createTableObjJson(tableObj); } catch (TException e) { throw new IllegalArgumentException("Could not serialize JSONUpdatePartitionColumnStatMessage : ", e); } @@ -122,6 +128,11 @@ public class JSONUpdatePartitionColumnStatMessage extends UpdatePartitionColumnS } @Override + public Table getTableObject() throws Exception { + return (Table) MessageBuilder.getTObj(tableObjJson, Table.class); + } + + @Override public String toString() { try { return JSONMessageDeserializer.mapper.writeValueAsString(this); http://git-wip-us.apache.org/repos/asf/hive/blob/71dfd1d1/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java ---------------------------------------------------------------------- diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java index aafdef9..481abbc 100644 --- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java @@ -1389,8 +1389,18 @@ public class HiveMetaStoreClientPreCatalog implements IMetaStoreClient, AutoClos @Override public List<Partition> getPartitionsByNames(String db_name, String tbl_name, - List<String> part_names) throws NoSuchObjectException, MetaException, TException { - List<Partition> parts = client.get_partitions_by_names(db_name, tbl_name, part_names); + List<String> part_names) throws NoSuchObjectException, MetaException, TException { + return getPartitionsByNames(db_name, tbl_name, part_names, false); + } + + @Override + public List<Partition> getPartitionsByNames(String db_name, String tbl_name, + List<String> part_names, boolean get_col_stats) + throws NoSuchObjectException, MetaException, TException { + GetPartitionsByNamesRequest gpbnr = new GetPartitionsByNamesRequest(db_name, tbl_name); + gpbnr.setNames(part_names); + gpbnr.setGet_col_stats(get_col_stats); + List<Partition> parts = client.get_partitions_by_names_req(gpbnr).getPartitions(); return fastpath ? parts : deepCopyPartitions(filterHook.filterPartitions(parts)); } @@ -3262,6 +3272,13 @@ public class HiveMetaStoreClientPreCatalog implements IMetaStoreClient, AutoClos } @Override + public List<Partition> getPartitionsByNames(String catName, String db_name, String tbl_name, + List<String> part_names, boolean getColStats) + throws NoSuchObjectException, MetaException, TException { + throw new UnsupportedOperationException(); + } + + @Override public List<Partition> listPartitionsWithAuthInfo(String catName, String dbName, String tableName, List<String> partialPvals, int maxParts, String userName, List<String> groupNames) throws http://git-wip-us.apache.org/repos/asf/hive/blob/71dfd1d1/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestStats.java ---------------------------------------------------------------------- diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestStats.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestStats.java index 5ba768f..03378ba 100644 --- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestStats.java +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestStats.java @@ -56,6 +56,7 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -245,6 +246,27 @@ public class TestStats { Assert.assertEquals(partNames.size(), aggr.getPartsFound()); Assert.assertEquals(colMap.size(), aggr.getColStatsSize()); aggr.getColStats().forEach(cso -> colMap.get(cso.getColName()).compareAggr(cso)); + + // Test column stats obtained through getPartitions call + for (int i = 0; i < partNames.size(); i++) { + String partName = partNames.get(i); + List<Partition> partitions = catName.equals(NO_CAT) ? + client.getPartitionsByNames(dbName, tableName, Collections.singletonList(partName), + true) : + client.getPartitionsByNames(catName, dbName, tableName, + Collections.singletonList(partName), true); + Partition partition = partitions.get(0); + compareStatsForOneTableOrPartition(partition.getColStats().getStatsObj(), i, colMap); + + // Also test that we do not get statistics when not requested + partitions = catName.equals(NO_CAT) ? + client.getPartitionsByNames(dbName, tableName, Collections.singletonList(partName), + true) : + client.getPartitionsByNames(catName, dbName, tableName, + Collections.singletonList(partName), true); + partition = partitions.get(0); + Assert.assertFalse(partition.isSetColStats()); + } } private void compareStatsForOneTableOrPartition(List<ColumnStatisticsObj> objs,
