You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by vg...@apache.org on 2018/01/17 18:07:34 UTC
[2/2] hive git commit: HIVE-17495: CachedStore: prewarm improvement
(avoid multiple sql calls to read partition column stats),
refactoring and caching some aggregate stats (Vaibhav Gumashta reviewed by
Daniel Dai)
HIVE-17495: CachedStore: prewarm improvement (avoid multiple sql calls to read partition column stats), refactoring and caching some aggregate stats (Vaibhav Gumashta reviewed by Daniel Dai)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/456a6518
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/456a6518
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/456a6518
Branch: refs/heads/master
Commit: 456a65180dcb84f69f26b4c9b9265165ad16dfe4
Parents: fbb3ed1
Author: Vaibhav Gumashta <vg...@hortonworks.com>
Authored: Wed Jan 17 09:59:02 2018 -0800
Committer: Vaibhav Gumashta <vg...@hortonworks.com>
Committed: Wed Jan 17 09:59:02 2018 -0800
----------------------------------------------------------------------
.../listener/DummyRawStoreFailEvent.java | 15 +-
.../hive/metastore/MetaStoreDirectSql.java | 95 ++---
.../hadoop/hive/metastore/ObjectStore.java | 24 +-
.../apache/hadoop/hive/metastore/RawStore.java | 11 +-
.../hadoop/hive/metastore/cache/CacheUtils.java | 12 +
.../hive/metastore/cache/CachedStore.java | 392 ++++++++++++++-----
.../hive/metastore/cache/SharedCache.java | 167 +++++++-
.../aggr/BinaryColumnStatsAggregator.java | 23 +-
.../aggr/BooleanColumnStatsAggregator.java | 23 +-
.../columnstats/aggr/ColumnStatsAggregator.java | 8 +-
.../aggr/DateColumnStatsAggregator.java | 56 ++-
.../aggr/DecimalColumnStatsAggregator.java | 57 ++-
.../aggr/DoubleColumnStatsAggregator.java | 57 ++-
.../aggr/LongColumnStatsAggregator.java | 56 +--
.../aggr/StringColumnStatsAggregator.java | 59 ++-
.../hive/metastore/utils/MetaStoreUtils.java | 144 ++++---
.../DummyRawStoreControlledCommit.java | 16 +-
.../DummyRawStoreForJdoConnection.java | 16 +-
18 files changed, 805 insertions(+), 426 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/itests/hcatalog-unit/src/test/java/org/apache/hive/hcatalog/listener/DummyRawStoreFailEvent.java
----------------------------------------------------------------------
diff --git a/itests/hcatalog-unit/src/test/java/org/apache/hive/hcatalog/listener/DummyRawStoreFailEvent.java b/itests/hcatalog-unit/src/test/java/org/apache/hive/hcatalog/listener/DummyRawStoreFailEvent.java
index 1fca332..bc9ef62 100644
--- a/itests/hcatalog-unit/src/test/java/org/apache/hive/hcatalog/listener/DummyRawStoreFailEvent.java
+++ b/itests/hcatalog-unit/src/test/java/org/apache/hive/hcatalog/listener/DummyRawStoreFailEvent.java
@@ -33,7 +33,6 @@ import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.AggrStats;
import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
-import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.CurrentNotificationEventId;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
@@ -75,6 +74,7 @@ import org.apache.hadoop.hive.metastore.api.UnknownTableException;
import org.apache.hadoop.hive.metastore.api.WMMapping;
import org.apache.hadoop.hive.metastore.api.WMPool;
import org.apache.hadoop.hive.metastore.partition.spec.PartitionSpecProxy;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.apache.thrift.TException;
/**
@@ -976,12 +976,6 @@ public class DummyRawStoreFailEvent implements RawStore, Configurable {
}
@Override
- public Map<String, List<ColumnStatisticsObj>> getColStatsForTablePartitions(String dbName,
- String tableName) throws MetaException, NoSuchObjectException {
- return objectStore.getColStatsForTablePartitions(dbName, tableName);
- }
-
- @Override
public String getMetastoreDbUuid() throws MetaException {
throw new MetaException("getMetastoreDbUuid is not implemented");
}
@@ -1092,4 +1086,11 @@ public class DummyRawStoreFailEvent implements RawStore, Configurable {
String poolPath) throws NoSuchObjectException, InvalidOperationException, MetaException {
objectStore.dropWMTriggerToPoolMapping(resourcePlanName, triggerName, poolPath);
}
+
+ @Override
+ public List<ColStatsObjWithSourceInfo> getPartitionColStatsForDatabase(String dbName)
+ throws MetaException, NoSuchObjectException {
+ // TODO Auto-generated method stub
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
index 786732f..47e2e6f 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
@@ -82,6 +82,7 @@ import org.apache.hadoop.hive.metastore.parser.ExpressionTree.Operator;
import org.apache.hadoop.hive.metastore.parser.ExpressionTree.TreeNode;
import org.apache.hadoop.hive.metastore.parser.ExpressionTree.TreeVisitor;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.apache.hive.common.util.BloomFilter;
import org.datanucleus.store.rdbms.query.ForwardQueryResult;
import org.slf4j.Logger;
@@ -216,7 +217,7 @@ class MetaStoreDirectSql {
doCommit = true;
}
LinkedList<Query> initQueries = new LinkedList<>();
-
+
try {
// Force the underlying db to initialize.
initQueries.add(pm.newQuery(MDatabase.class, "name == ''"));
@@ -968,7 +969,7 @@ class MetaStoreDirectSql {
/**
* Convert a boolean value returned from the RDBMS to a Java Boolean object.
* MySQL has booleans, but e.g. Derby uses 'Y'/'N' mapping.
- *
+ *
* @param value
* column value from the database
* @return The Boolean value of the database column value, null if the column
@@ -1499,81 +1500,63 @@ class MetaStoreDirectSql {
});
}
- // Get aggregated column stats for a table per partition for all columns in the partition
- // This is primarily used to populate stats object when using CachedStore (Check CachedStore#prewarm)
- public Map<String, List<ColumnStatisticsObj>> getColStatsForTablePartitions(String dbName,
- String tblName, boolean enableBitVector) throws MetaException {
- String queryText = "select \"PARTITION_NAME\", " + getStatsList(enableBitVector) + " from "
- + " " + PART_COL_STATS + " where \"DB_NAME\" = ? and \"TABLE_NAME\" = ?"
- + " order by \"PARTITION_NAME\"";
+ public List<ColStatsObjWithSourceInfo> getColStatsForAllTablePartitions(String dbName,
+ boolean enableBitVector) throws MetaException {
+ String queryText = "select \"TABLE_NAME\", \"PARTITION_NAME\", " + getStatsList(enableBitVector)
+ + " from " + " " + PART_COL_STATS + " where \"DB_NAME\" = ?";
long start = 0;
long end = 0;
Query query = null;
boolean doTrace = LOG.isDebugEnabled();
Object qResult = null;
start = doTrace ? System.nanoTime() : 0;
- query = pm.newQuery("javax.jdo.query.SQL", queryText);
- qResult = executeWithArray(query, prepareParams(dbName, tblName,
- Collections.<String>emptyList(), Collections.<String>emptyList()), queryText);
- if (qResult == null) {
- query.closeAll();
- return Collections.emptyMap();
- }
- end = doTrace ? System.nanoTime() : 0;
- timingTrace(doTrace, queryText, start, end);
- List<Object[]> list = ensureList(qResult);
- Map<String, List<ColumnStatisticsObj>> partColStatsMap =
- new HashMap<String, List<ColumnStatisticsObj>>();
- String partNameCurrent = null;
- List<ColumnStatisticsObj> partColStatsList = new ArrayList<ColumnStatisticsObj>();
- for (Object[] row : list) {
- String partName = (String) row[0];
- if (partNameCurrent == null) {
- // Update the current partition we are working on
- partNameCurrent = partName;
- // Create a new list for this new partition
- partColStatsList = new ArrayList<ColumnStatisticsObj>();
- // Add the col stat for the current column
- partColStatsList.add(prepareCSObj(row, 1));
- } else if (!partNameCurrent.equalsIgnoreCase(partName)) {
- // Save the previous partition and its col stat list
- partColStatsMap.put(partNameCurrent, partColStatsList);
- // Update the current partition we are working on
- partNameCurrent = partName;
- // Create a new list for this new partition
- partColStatsList = new ArrayList<ColumnStatisticsObj>();
- // Add the col stat for the current column
- partColStatsList.add(prepareCSObj(row, 1));
- } else {
- partColStatsList.add(prepareCSObj(row, 1));
+ List<ColStatsObjWithSourceInfo> colStatsForDB = new ArrayList<ColStatsObjWithSourceInfo>();
+ try {
+ query = pm.newQuery("javax.jdo.query.SQL", queryText);
+ qResult = executeWithArray(query, new Object[] { dbName }, queryText);
+ if (qResult == null) {
+ query.closeAll();
+ return colStatsForDB;
}
- Deadline.checkTimeout();
+ end = doTrace ? System.nanoTime() : 0;
+ timingTrace(doTrace, queryText, start, end);
+ List<Object[]> list = ensureList(qResult);
+ for (Object[] row : list) {
+ String tblName = (String) row[0];
+ String partName = (String) row[1];
+ ColumnStatisticsObj colStatObj = prepareCSObj(row, 2);
+ colStatsForDB.add(new ColStatsObjWithSourceInfo(colStatObj, dbName, tblName, partName));
+ Deadline.checkTimeout();
+ }
+ } finally {
+ query.closeAll();
}
- query.closeAll();
- return partColStatsMap;
+ return colStatsForDB;
}
/** Should be called with the list short enough to not trip up Oracle/etc. */
private List<ColumnStatisticsObj> columnStatisticsObjForPartitionsBatch(String dbName,
String tableName, List<String> partNames, List<String> colNames, boolean areAllPartsFound,
- boolean useDensityFunctionForNDVEstimation, double ndvTuner, boolean enableBitVector) throws MetaException {
- if(enableBitVector) {
- return aggrStatsUseJava(dbName, tableName, partNames, colNames, useDensityFunctionForNDVEstimation, ndvTuner);
- }
- else {
- return aggrStatsUseDB(dbName, tableName, partNames, colNames, areAllPartsFound, useDensityFunctionForNDVEstimation, ndvTuner);
+ boolean useDensityFunctionForNDVEstimation, double ndvTuner, boolean enableBitVector)
+ throws MetaException {
+ if (enableBitVector) {
+ return aggrStatsUseJava(dbName, tableName, partNames, colNames, areAllPartsFound,
+ useDensityFunctionForNDVEstimation, ndvTuner);
+ } else {
+ return aggrStatsUseDB(dbName, tableName, partNames, colNames, areAllPartsFound,
+ useDensityFunctionForNDVEstimation, ndvTuner);
}
}
private List<ColumnStatisticsObj> aggrStatsUseJava(String dbName, String tableName,
- List<String> partNames, List<String> colNames,
+ List<String> partNames, List<String> colNames, boolean areAllPartsFound,
boolean useDensityFunctionForNDVEstimation, double ndvTuner) throws MetaException {
// 1. get all the stats for colNames in partNames;
- List<ColumnStatistics> partStats = getPartitionStats(dbName, tableName, partNames, colNames,
- true);
+ List<ColumnStatistics> partStats =
+ getPartitionStats(dbName, tableName, partNames, colNames, true);
// 2. use util function to aggr stats
return MetaStoreUtils.aggrPartitionStats(partStats, dbName, tableName, partNames, colNames,
- useDensityFunctionForNDVEstimation, ndvTuner);
+ areAllPartsFound, useDensityFunctionForNDVEstimation, ndvTuner);
}
private List<ColumnStatisticsObj> aggrStatsUseDB(String dbName,
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
index 3a1bc13..b9fecc7 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
@@ -188,6 +188,7 @@ import org.apache.hadoop.hive.metastore.tools.SQLGenerator;
import org.apache.hadoop.hive.metastore.utils.FileUtils;
import org.apache.hadoop.hive.metastore.utils.JavaUtils;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.apache.hadoop.hive.metastore.utils.ObjectPair;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
@@ -7967,24 +7968,25 @@ public class ObjectStore implements RawStore, Configurable {
}
@Override
- public Map<String, List<ColumnStatisticsObj>> getColStatsForTablePartitions(String dbName,
- String tableName) throws MetaException, NoSuchObjectException {
- final boolean enableBitVector = MetastoreConf.getBoolVar(getConf(), ConfVars.STATS_FETCH_BITVECTOR);
- return new GetHelper<Map<String, List<ColumnStatisticsObj>>>(dbName, tableName, true, false) {
+ public List<ColStatsObjWithSourceInfo> getPartitionColStatsForDatabase(String dbName)
+ throws MetaException, NoSuchObjectException {
+ final boolean enableBitVector =
+ MetastoreConf.getBoolVar(getConf(), ConfVars.STATS_FETCH_BITVECTOR);
+ return new GetHelper<List<ColStatsObjWithSourceInfo>>(dbName, null, true, false) {
@Override
- protected Map<String, List<ColumnStatisticsObj>> getSqlResult(
- GetHelper<Map<String, List<ColumnStatisticsObj>>> ctx) throws MetaException {
- return directSql.getColStatsForTablePartitions(dbName, tblName, enableBitVector);
+ protected List<ColStatsObjWithSourceInfo> getSqlResult(
+ GetHelper<List<ColStatsObjWithSourceInfo>> ctx) throws MetaException {
+ return directSql.getColStatsForAllTablePartitions(dbName, enableBitVector);
}
@Override
- protected Map<String, List<ColumnStatisticsObj>> getJdoResult(
- GetHelper<Map<String, List<ColumnStatisticsObj>>> ctx) throws MetaException,
- NoSuchObjectException {
+ protected List<ColStatsObjWithSourceInfo> getJdoResult(
+ GetHelper<List<ColStatsObjWithSourceInfo>> ctx)
+ throws MetaException, NoSuchObjectException {
// This is fast path for query optimizations, if we can find this info
// quickly using directSql, do it. No point in failing back to slow path
// here.
- throw new MetaException("Jdo path is not implemented for stats aggr.");
+ throw new MetaException("Jdo path is not implemented for getPartitionColStatsForDatabase.");
}
@Override
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
index e6c7a58..e5092ba 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/RawStore.java
@@ -75,6 +75,7 @@ import org.apache.hadoop.hive.metastore.api.UnknownTableException;
import org.apache.hadoop.hive.metastore.api.WMMapping;
import org.apache.hadoop.hive.metastore.api.WMPool;
import org.apache.hadoop.hive.metastore.partition.spec.PartitionSpecProxy;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.apache.thrift.TException;
public interface RawStore extends Configurable {
@@ -599,17 +600,15 @@ public interface RawStore extends Configurable {
List<String> partNames, List<String> colNames) throws MetaException, NoSuchObjectException;
/**
- * Get all partition column statistics for a table in a db
+ * Get column stats for all partitions of all tables in the database
*
* @param dbName
- * @param tableName
- * @return Map of partition column statistics. Key in the map is partition name. Value is a list
- * of column stat object for each column in the partition
+ * @return List of column stats objects for all partitions of all tables in the database
* @throws MetaException
* @throws NoSuchObjectException
*/
- Map<String, List<ColumnStatisticsObj>> getColStatsForTablePartitions(String dbName,
- String tableName) throws MetaException, NoSuchObjectException;
+ List<ColStatsObjWithSourceInfo> getPartitionColStatsForDatabase(String dbName)
+ throws MetaException, NoSuchObjectException;
/**
* Get the next notification event.
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CacheUtils.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CacheUtils.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CacheUtils.java
index 798ada8..f0f650d 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CacheUtils.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CacheUtils.java
@@ -34,6 +34,14 @@ import org.apache.hadoop.hive.metastore.utils.StringUtils;
public class CacheUtils {
private static final String delimit = "\u0001";
+ public static String buildKey(String dbName) {
+ return dbName;
+ }
+
+ public static String buildKeyWithDelimit(String dbName) {
+ return buildKey(dbName) + delimit;
+ }
+
public static String buildKey(String dbName, String tableName) {
return dbName + delimit + tableName;
}
@@ -79,6 +87,10 @@ public class CacheUtils {
return result;
}
+ public static Object[] splitAggrColStats(String key) {
+ return key.split(delimit);
+ }
+
static Table assemble(TableWrapper wrapper, SharedCache sharedCache) {
Table t = wrapper.getTable().deepCopy();
if (wrapper.getSdHash() != null) {
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java
index 72c4a26..e28e44a 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java
@@ -76,6 +76,9 @@ import org.apache.hadoop.hive.metastore.api.PrincipalType;
import org.apache.hadoop.hive.metastore.api.PrivilegeBag;
import org.apache.hadoop.hive.metastore.api.WMResourcePlan;
import org.apache.hadoop.hive.metastore.api.WMTrigger;
+import org.apache.hadoop.hive.metastore.cache.SharedCache.StatsType;
+import org.apache.hadoop.hive.metastore.columnstats.aggr.ColumnStatsAggregator;
+import org.apache.hadoop.hive.metastore.columnstats.aggr.ColumnStatsAggregatorFactory;
import org.apache.hadoop.hive.metastore.api.Role;
import org.apache.hadoop.hive.metastore.api.RolePrincipalGrant;
import org.apache.hadoop.hive.metastore.api.SQLForeignKey;
@@ -97,6 +100,7 @@ import org.apache.hadoop.hive.metastore.partition.spec.PartitionSpecProxy;
import org.apache.hadoop.hive.metastore.utils.FileUtils;
import org.apache.hadoop.hive.metastore.utils.JavaUtils;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.apache.hadoop.hive.metastore.utils.StringUtils;
import org.apache.thrift.TException;
import org.slf4j.Logger;
@@ -126,6 +130,9 @@ public class CachedStore implements RawStore, Configurable {
private static AtomicBoolean isTableColStatsCacheDirty = new AtomicBoolean(false);
private static ReentrantReadWriteLock partitionColStatsCacheLock = new ReentrantReadWriteLock(
true);
+ private static ReentrantReadWriteLock partitionAggrColStatsCacheLock =
+ new ReentrantReadWriteLock(true);
+ private static AtomicBoolean isPartitionAggrColStatsCacheDirty = new AtomicBoolean(false);
private static AtomicBoolean isPartitionColStatsCacheDirty = new AtomicBoolean(false);
private static List<Pattern> whitelistPatterns = null;
private static List<Pattern> blacklistPatterns = null;
@@ -257,6 +264,14 @@ public class CachedStore implements RawStore, Configurable {
SharedCache sharedCache = sharedCacheWrapper.getUnsafe();
for (int i = 0; i < dbNames.size(); i++) {
String dbName = StringUtils.normalizeIdentifier(dbNames.get(i));
+ // Cache partition column stats
+ Deadline.startTimer("getColStatsForDatabase");
+ List<ColStatsObjWithSourceInfo> colStatsForDB =
+ rawStore.getPartitionColStatsForDatabase(dbName);
+ Deadline.stopTimer();
+ if (colStatsForDB != null) {
+ sharedCache.addPartitionColStatsToCache(colStatsForDB);
+ }
LOG.info("Caching database: {}. Cached {} / {} databases so far.", dbName, i, dbNames.size());
Database db = rawStore.getDatabase(dbName);
sharedCache.addDatabaseToCache(dbName, db);
@@ -285,14 +300,6 @@ public class CachedStore implements RawStore, Configurable {
for (Partition partition : partitions) {
sharedCache.addPartitionToCache(dbName, tblName, partition);
}
- // Cache partition column stats
- Deadline.startTimer("getColStatsForTablePartitions");
- Map<String, List<ColumnStatisticsObj>> colStatsPerPartition =
- rawStore.getColStatsForTablePartitions(dbName, tblName);
- Deadline.stopTimer();
- if (colStatsPerPartition != null) {
- sharedCache.addPartitionColStatsToCache(dbName, tblName, colStatsPerPartition);
- }
}
// Cache table column stats
List<String> colNames = MetaStoreUtils.getColumnNamesForTable(table);
@@ -303,6 +310,28 @@ public class CachedStore implements RawStore, Configurable {
if ((tableColStats != null) && (tableColStats.getStatsObjSize() > 0)) {
sharedCache.addTableColStatsToCache(dbName, tblName, tableColStats.getStatsObj());
}
+ // Cache aggregate stats for all partitions of a table and for all but default partition
+ List<String> partNames = rawStore.listPartitionNames(dbName, tblName, (short) -1);
+ if ((partNames != null) && (partNames.size() > 0)) {
+ AggrStats aggrStatsAllPartitions =
+ rawStore.get_aggr_stats_for(dbName, tblName, partNames, colNames);
+ // Remove default partition from partition names and get aggregate
+ // stats again
+ List<FieldSchema> partKeys = table.getPartitionKeys();
+ String defaultPartitionValue = MetastoreConf.getVar(rawStore.getConf(), ConfVars.DEFAULTPARTITIONNAME);
+ List<String> partCols = new ArrayList<String>();
+ List<String> partVals = new ArrayList<String>();
+ for (FieldSchema fs : partKeys) {
+ partCols.add(fs.getName());
+ partVals.add(defaultPartitionValue);
+ }
+ String defaultPartitionName = FileUtils.makePartName(partCols, partVals);
+ partNames.remove(defaultPartitionName);
+ AggrStats aggrStatsAllButDefaultPartition =
+ rawStore.get_aggr_stats_for(dbName, tblName, partNames, colNames);
+ sharedCache.addAggregateStatsToCache(dbName, tblName, aggrStatsAllPartitions,
+ aggrStatsAllButDefaultPartition);
+ }
}
}
// Notify all blocked threads that prewarm is complete now
@@ -325,11 +354,10 @@ public class CachedStore implements RawStore, Configurable {
if (cacheUpdateMaster == null) {
initBlackListWhiteList(conf);
if (!MetastoreConf.getBoolVar(conf, ConfVars.HIVE_IN_TEST)) {
- cacheRefreshPeriod =
- MetastoreConf.getTimeVar(conf, ConfVars.CACHED_RAW_STORE_CACHE_UPDATE_FREQUENCY,
- TimeUnit.MILLISECONDS);
+ cacheRefreshPeriod = MetastoreConf.getTimeVar(conf,
+ ConfVars.CACHED_RAW_STORE_CACHE_UPDATE_FREQUENCY, TimeUnit.MILLISECONDS);
}
- LOG.info("CachedStore: starting cache update service (run every " + cacheRefreshPeriod + "ms");
+ LOG.info("CachedStore: starting cache update service (run every {} ms", cacheRefreshPeriod);
cacheUpdateMaster = Executors.newScheduledThreadPool(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
@@ -339,8 +367,8 @@ public class CachedStore implements RawStore, Configurable {
return t;
}
});
- cacheUpdateMaster.scheduleAtFixedRate(new CacheUpdateMasterWork(conf), 0,
- cacheRefreshPeriod, TimeUnit.MILLISECONDS);
+ cacheUpdateMaster.scheduleAtFixedRate(new CacheUpdateMasterWork(conf), 0, cacheRefreshPeriod,
+ TimeUnit.MILLISECONDS);
}
}
@@ -419,6 +447,7 @@ public class CachedStore implements RawStore, Configurable {
// Update the database in cache
updateDatabases(rawStore, dbNames);
for (String dbName : dbNames) {
+ updateDatabasePartitionColStats(rawStore, dbName);
// Update the tables in cache
updateTables(rawStore, dbName);
List<String> tblNames = getAllTablesInternal(dbName, sharedCacheWrapper.getUnsafe());
@@ -430,8 +459,8 @@ public class CachedStore implements RawStore, Configurable {
updateTablePartitions(rawStore, dbName, tblName);
// Update the table column stats for a table in cache
updateTableColStats(rawStore, dbName, tblName);
- // Update the partitions column stats for a table in cache
- updateTablePartitionColStats(rawStore, dbName, tblName);
+ // Update aggregate column stats cache
+ updateAggregateStatsCache(rawStore, dbName, tblName);
}
}
}
@@ -440,6 +469,87 @@ public class CachedStore implements RawStore, Configurable {
}
}
+ private void updateDatabasePartitionColStats(RawStore rawStore, String dbName) {
+ try {
+ Deadline.startTimer("getColStatsForDatabasePartitions");
+ List<ColStatsObjWithSourceInfo> colStatsForDB =
+ rawStore.getPartitionColStatsForDatabase(dbName);
+ Deadline.stopTimer();
+ if (colStatsForDB != null) {
+ if (partitionColStatsCacheLock.writeLock().tryLock()) {
+ // Skip background updates if we detect change
+ if (isPartitionColStatsCacheDirty.compareAndSet(true, false)) {
+ LOG.debug("Skipping partition column stats cache update; the partition column stats "
+ + "list we have is dirty.");
+ return;
+ }
+ sharedCacheWrapper.getUnsafe()
+ .refreshPartitionColStats(StringUtils.normalizeIdentifier(dbName), colStatsForDB);
+ }
+ }
+ } catch (MetaException | NoSuchObjectException e) {
+ LOG.info("Updating CachedStore: unable to read partitions column stats of database: {}",
+ dbName, e);
+ } finally {
+ if (partitionColStatsCacheLock.isWriteLockedByCurrentThread()) {
+ partitionColStatsCacheLock.writeLock().unlock();
+ }
+ }
+ }
+
+ // Update cached aggregate stats for all partitions of a table and for all
+ // but default partition
+ private void updateAggregateStatsCache(RawStore rawStore, String dbName, String tblName) {
+ try {
+ Table table = rawStore.getTable(dbName, tblName);
+ List<String> partNames = rawStore.listPartitionNames(dbName, tblName, (short) -1);
+ List<String> colNames = MetaStoreUtils.getColumnNamesForTable(table);
+ if ((partNames != null) && (partNames.size() > 0)) {
+ Deadline.startTimer("getAggregareStatsForAllPartitions");
+ AggrStats aggrStatsAllPartitions =
+ rawStore.get_aggr_stats_for(dbName, tblName, partNames, colNames);
+ Deadline.stopTimer();
+ // Remove default partition from partition names and get aggregate stats again
+ List<FieldSchema> partKeys = table.getPartitionKeys();
+ String defaultPartitionValue =
+ MetastoreConf.getVar(rawStore.getConf(), ConfVars.DEFAULTPARTITIONNAME);
+ List<String> partCols = new ArrayList<String>();
+ List<String> partVals = new ArrayList<String>();
+ for (FieldSchema fs : partKeys) {
+ partCols.add(fs.getName());
+ partVals.add(defaultPartitionValue);
+ }
+ String defaultPartitionName = FileUtils.makePartName(partCols, partVals);
+ partNames.remove(defaultPartitionName);
+ Deadline.startTimer("getAggregareStatsForAllPartitionsExceptDefault");
+ AggrStats aggrStatsAllButDefaultPartition =
+ rawStore.get_aggr_stats_for(dbName, tblName, partNames, colNames);
+ Deadline.stopTimer();
+ if ((aggrStatsAllPartitions != null) && (aggrStatsAllButDefaultPartition != null)) {
+ if (partitionAggrColStatsCacheLock.writeLock().tryLock()) {
+ // Skip background updates if we detect change
+ if (isPartitionAggrColStatsCacheDirty.compareAndSet(true, false)) {
+ LOG.debug(
+ "Skipping aggregate column stats cache update; the aggregate column stats we "
+ + "have is dirty.");
+ return;
+ }
+ sharedCacheWrapper.getUnsafe().refreshAggregateStatsCache(
+ StringUtils.normalizeIdentifier(dbName), StringUtils.normalizeIdentifier(tblName),
+ aggrStatsAllPartitions, aggrStatsAllButDefaultPartition);
+ }
+ }
+ }
+ } catch (MetaException | NoSuchObjectException e) {
+ LOG.info("Updating CachedStore: unable to read aggregate column stats of table: " + tblName,
+ e);
+ } finally {
+ if (partitionAggrColStatsCacheLock.isWriteLockedByCurrentThread()) {
+ partitionAggrColStatsCacheLock.writeLock().unlock();
+ }
+ }
+ }
+
private void updateDatabases(RawStore rawStore, List<String> dbNames) {
// Prepare the list of databases
List<Database> databases = new ArrayList<>();
@@ -555,36 +665,6 @@ public class CachedStore implements RawStore, Configurable {
}
}
}
-
- // Update the cached partition col stats for a table
- private void updateTablePartitionColStats(RawStore rawStore, String dbName, String tblName) {
- try {
- Deadline.startTimer("getColStatsForTablePartitions");
- Map<String, List<ColumnStatisticsObj>> colStatsPerPartition =
- rawStore.getColStatsForTablePartitions(dbName, tblName);
- Deadline.stopTimer();
- if (colStatsPerPartition != null) {
- if (partitionColStatsCacheLock.writeLock().tryLock()) {
- // Skip background updates if we detect change
- if (isPartitionColStatsCacheDirty.compareAndSet(true, false)) {
- LOG.debug("Skipping partition column stats cache update; the partition column stats "
- + "list we have is dirty.");
- return;
- }
- sharedCacheWrapper.getUnsafe().refreshPartitionColStats(
- StringUtils.normalizeIdentifier(dbName),
- StringUtils.normalizeIdentifier(tblName), colStatsPerPartition);
- }
- }
- } catch (MetaException | NoSuchObjectException e) {
- LOG.info("Updating CachedStore: unable to read partitions column stats of table: "
- + tblName, e);
- } finally {
- if (partitionColStatsCacheLock.isWriteLockedByCurrentThread()) {
- partitionColStatsCacheLock.writeLock().unlock();
- }
- }
- }
}
@Override
@@ -838,6 +918,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
return succ;
}
@@ -864,6 +953,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
return succ;
}
@@ -892,6 +990,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
return succ;
}
@@ -908,9 +1015,8 @@ public class CachedStore implements RawStore, Configurable {
SharedCache sharedCache = sharedCacheWrapper.get();
Partition part =
sharedCache.getPartitionFromCache(dbName, tblName, part_vals);
- if (part != null) {
- part.unsetPrivileges();
- } else {
+ if (part == null) {
+ // TODO Manage privileges
throw new NoSuchObjectException("partition values=" + part_vals.toString());
}
return part;
@@ -958,6 +1064,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionColStatsCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
return succ;
}
@@ -972,11 +1087,6 @@ public class CachedStore implements RawStore, Configurable {
}
SharedCache sharedCache = sharedCacheWrapper.get();
List<Partition> parts = sharedCache.listCachedPartitions(dbName, tblName, max);
- if (parts != null) {
- for (Partition part : parts) {
- part.unsetPrivileges();
- }
- }
return parts;
}
@@ -1046,6 +1156,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionColStatsCacheLock.readLock().unlock();
}
+ // Update aggregate partition col stats keys wherever applicable
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.alterTableInAggrPartitionColStatsCache(dbName, tblName, newTable);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
}
@@ -1226,6 +1345,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionColStatsCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
@Override
@@ -1265,6 +1393,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionColStatsCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
@Override
@@ -1340,11 +1477,6 @@ public class CachedStore implements RawStore, Configurable {
Table table = sharedCache.getTableFromCache(dbName, tblName);
boolean hasUnknownPartitions = getPartitionNamesPrunedByExprNoTxn(table, expr,
defaultPartitionName, maxParts, partNames, sharedCache);
- for (String partName : partNames) {
- Partition part = sharedCache.getPartitionFromCache(dbName, tblName, partNameToVals(partName));
- part.unsetPrivileges();
- result.add(part);
- }
return hasUnknownPartitions;
}
@@ -1803,6 +1935,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionColStatsCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
return succ;
}
@@ -1836,6 +1977,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionColStatsCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
return succ;
}
@@ -1843,53 +1993,106 @@ public class CachedStore implements RawStore, Configurable {
@Override
public AggrStats get_aggr_stats_for(String dbName, String tblName, List<String> partNames,
List<String> colNames) throws MetaException, NoSuchObjectException {
+ List<ColumnStatisticsObj> colStats;
dbName = StringUtils.normalizeIdentifier(dbName);
tblName = StringUtils.normalizeIdentifier(tblName);
if (!sharedCacheWrapper.isInitialized() || !shouldCacheTable(dbName, tblName)) {
rawStore.get_aggr_stats_for(dbName, tblName, partNames, colNames);
}
SharedCache sharedCache = sharedCacheWrapper.get();
- List<ColumnStatisticsObj> colStats =
- mergeColStatsForPartitions(StringUtils.normalizeIdentifier(dbName),
- StringUtils.normalizeIdentifier(tblName), partNames, colNames, sharedCache);
- return new AggrStats(colStats, partNames.size());
+ List<String> allPartNames = rawStore.listPartitionNames(dbName, tblName, (short) -1);
+ if (partNames.size() == allPartNames.size()) {
+ colStats = sharedCache.getAggrStatsFromCache(dbName, tblName, colNames, StatsType.ALL);
+ if (colStats != null) {
+ return new AggrStats(colStats, partNames.size());
+ }
+ } else if (partNames.size() == (allPartNames.size() - 1)) {
+ String defaultPartitionName = MetastoreConf.getVar(getConf(), ConfVars.DEFAULTPARTITIONNAME);
+ if (!partNames.contains(defaultPartitionName)) {
+ colStats =
+ sharedCache.getAggrStatsFromCache(dbName, tblName, colNames, StatsType.ALLBUTDEFAULT);
+ if (colStats != null) {
+ return new AggrStats(colStats, partNames.size());
+ }
+ }
+ }
+ LOG.debug("Didn't find aggr stats in cache. Merging them. tblName= {}, parts= {}, cols= {}",
+ tblName, partNames, colNames);
+ MergedColumnStatsForPartitions mergedColStats =
+ mergeColStatsForPartitions(dbName, tblName, partNames, colNames, sharedCache);
+ return new AggrStats(mergedColStats.getColStats(), mergedColStats.getPartsFound());
}
- private List<ColumnStatisticsObj> mergeColStatsForPartitions(String dbName, String tblName,
- List<String> partNames, List<String> colNames, SharedCache sharedCache)
- throws MetaException {
- final boolean useDensityFunctionForNDVEstimation = MetastoreConf.getBoolVar(getConf(),
- ConfVars.STATS_NDV_DENSITY_FUNCTION);
+ private MergedColumnStatsForPartitions mergeColStatsForPartitions(String dbName, String tblName,
+ List<String> partNames, List<String> colNames, SharedCache sharedCache) throws MetaException {
+ final boolean useDensityFunctionForNDVEstimation =
+ MetastoreConf.getBoolVar(getConf(), ConfVars.STATS_NDV_DENSITY_FUNCTION);
final double ndvTuner = MetastoreConf.getDoubleVar(getConf(), ConfVars.STATS_NDV_TUNER);
- Map<String, List<ColumnStatistics>> map = new HashMap<>();
-
+ Map<ColumnStatsAggregator, List<ColStatsObjWithSourceInfo>> colStatsMap =
+ new HashMap<ColumnStatsAggregator, List<ColStatsObjWithSourceInfo>>();
+ boolean areAllPartsFound = true;
+ long partsFound = 0;
for (String colName : colNames) {
- List<ColumnStatistics> colStats = new ArrayList<>();
+ long partsFoundForColumn = 0;
+ ColumnStatsAggregator colStatsAggregator = null;
+ List<ColStatsObjWithSourceInfo> colStatsWithPartInfoList =
+ new ArrayList<ColStatsObjWithSourceInfo>();
for (String partName : partNames) {
- String colStatsCacheKey = CacheUtils.buildKey(dbName, tblName, partNameToVals(partName),
- colName);
- List<ColumnStatisticsObj> colStat = new ArrayList<>();
- ColumnStatisticsObj colStatsForPart = sharedCache
- .getCachedPartitionColStats(colStatsCacheKey);
+ String colStatsCacheKey =
+ CacheUtils.buildKey(dbName, tblName, partNameToVals(partName), colName);
+ ColumnStatisticsObj colStatsForPart =
+ sharedCache.getCachedPartitionColStats(colStatsCacheKey);
if (colStatsForPart != null) {
- colStat.add(colStatsForPart);
- ColumnStatisticsDesc csDesc = new ColumnStatisticsDesc(false, dbName, tblName);
- csDesc.setPartName(partName);
- colStats.add(new ColumnStatistics(csDesc, colStat));
+ ColStatsObjWithSourceInfo colStatsWithPartInfo =
+ new ColStatsObjWithSourceInfo(colStatsForPart, dbName, tblName, partName);
+ colStatsWithPartInfoList.add(colStatsWithPartInfo);
+ if (colStatsAggregator == null) {
+ colStatsAggregator = ColumnStatsAggregatorFactory.getColumnStatsAggregator(
+ colStatsForPart.getStatsData().getSetField(), useDensityFunctionForNDVEstimation,
+ ndvTuner);
+ }
+ partsFoundForColumn++;
} else {
- LOG.debug("Stats not found in CachedStore for: dbName={} tblName={} partName={} colName={}",
- dbName, tblName,partName, colName);
+ LOG.debug(
+ "Stats not found in CachedStore for: dbName={} tblName={} partName={} colName={}",
+ dbName, tblName, partName, colName);
}
}
- map.put(colName, colStats);
+ if (colStatsWithPartInfoList.size() > 0) {
+ colStatsMap.put(colStatsAggregator, colStatsWithPartInfoList);
+ }
+ if (partsFoundForColumn == partNames.size()) {
+ partsFound++;
+ }
+ if (colStatsMap.size() < 1) {
+ LOG.debug("No stats data found for: dbName={} tblName= {} partNames= {} colNames= ", dbName,
+ tblName, partNames, colNames);
+ return new MergedColumnStatsForPartitions(new ArrayList<ColumnStatisticsObj>(), 0);
+ }
}
// Note that enableBitVector does not apply here because ColumnStatisticsObj
- // itself will tell whether
- // bitvector is null or not and aggr logic can automatically apply.
- return MetaStoreUtils.aggrPartitionStats(map, dbName, tblName, partNames, colNames,
- useDensityFunctionForNDVEstimation, ndvTuner);
+ // itself will tell whether bitvector is null or not and aggr logic can automatically apply.
+ return new MergedColumnStatsForPartitions(MetaStoreUtils.aggrPartitionStats(colStatsMap,
+ partNames, areAllPartsFound, useDensityFunctionForNDVEstimation, ndvTuner), partsFound);
}
+ class MergedColumnStatsForPartitions {
+ List<ColumnStatisticsObj> colStats = new ArrayList<ColumnStatisticsObj>();
+ long partsFound;
+
+ MergedColumnStatsForPartitions(List<ColumnStatisticsObj> colStats, long partsFound) {
+ this.colStats = colStats;
+ this.partsFound = partsFound;
+ }
+
+ List<ColumnStatisticsObj> getColStats() {
+ return colStats;
+ }
+
+ long getPartsFound() {
+ return partsFound;
+ }
+ }
@Override
public long cleanupEvents() {
@@ -1988,6 +2191,15 @@ public class CachedStore implements RawStore, Configurable {
} finally {
partitionColStatsCacheLock.readLock().unlock();
}
+ // Remove aggregate partition col stats for this table
+ try {
+ // Wait if background cache update is happening
+ partitionAggrColStatsCacheLock.readLock().lock();
+ isPartitionAggrColStatsCacheDirty.set(true);
+ sharedCache.removeAggrPartitionColStatsFromCache(dbName, tblName);
+ } finally {
+ partitionAggrColStatsCacheLock.readLock().unlock();
+ }
}
@Override
@@ -2253,9 +2465,9 @@ public class CachedStore implements RawStore, Configurable {
}
@Override
- public Map<String, List<ColumnStatisticsObj>> getColStatsForTablePartitions(String dbName,
- String tableName) throws MetaException, NoSuchObjectException {
- return rawStore.getColStatsForTablePartitions(dbName, tableName);
+ public List<ColStatsObjWithSourceInfo> getPartitionColStatsForDatabase(String dbName)
+ throws MetaException, NoSuchObjectException {
+ return rawStore.getPartitionColStatsForDatabase(dbName);
}
public RawStore getRawStore() {
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java
index b606779..32ea174 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java
@@ -29,6 +29,7 @@ import java.util.TreeMap;
import org.apache.hadoop.hive.metastore.StatObjectConverter;
import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.api.AggrStats;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
@@ -41,6 +42,7 @@ import org.apache.hadoop.hive.metastore.cache.CachedStore.PartitionWrapper;
import org.apache.hadoop.hive.metastore.cache.CachedStore.StorageDescriptorWrapper;
import org.apache.hadoop.hive.metastore.cache.CachedStore.TableWrapper;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.apache.hadoop.hive.metastore.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,8 +56,24 @@ public class SharedCache {
private Map<String, ColumnStatisticsObj> partitionColStatsCache = new TreeMap<>();
private Map<String, ColumnStatisticsObj> tableColStatsCache = new TreeMap<>();
private Map<ByteArrayWrapper, StorageDescriptorWrapper> sdCache = new HashMap<>();
+ private Map<String, List<ColumnStatisticsObj>> aggrColStatsCache =
+ new HashMap<String, List<ColumnStatisticsObj>>();
private static MessageDigest md;
+ static enum StatsType {
+ ALL(0), ALLBUTDEFAULT(1);
+
+ private final int position;
+
+ private StatsType(int position) {
+ this.position = position;
+ }
+
+ public int getPosition() {
+ return position;
+ }
+ }
+
private static final Logger LOG = LoggerFactory.getLogger(SharedCache.class);
static {
@@ -151,7 +169,11 @@ public class SharedCache {
public synchronized void removeTableColStatsFromCache(String dbName, String tblName,
String colName) {
- tableColStatsCache.remove(CacheUtils.buildKey(dbName, tblName, colName));
+ if (colName == null) {
+ removeTableColStatsFromCache(dbName, tblName);
+ } else {
+ tableColStatsCache.remove(CacheUtils.buildKey(dbName, tblName, colName));
+ }
}
public synchronized void updateTableColStatsInCache(String dbName, String tableName,
@@ -231,9 +253,9 @@ public class SharedCache {
ColumnStatisticsObj colStatObj = entry.getValue();
if (key.toLowerCase().startsWith(oldPartialPartitionKey.toLowerCase())) {
Object[] decomposedKey = CacheUtils.splitPartitionColStats(key);
- String newKey =
- CacheUtils.buildKey((String) decomposedKey[0], (String) decomposedKey[1],
- (List<String>) decomposedKey[2], (String) decomposedKey[3]);
+ // New key has the new table name
+ String newKey = CacheUtils.buildKey((String) decomposedKey[0], newTable.getTableName(),
+ (List<String>) decomposedKey[2], (String) decomposedKey[3]);
newPartitionColStats.put(newKey, colStatObj);
iterator.remove();
}
@@ -243,6 +265,31 @@ public class SharedCache {
}
}
+ public synchronized void alterTableInAggrPartitionColStatsCache(String dbName, String tblName,
+ Table newTable) {
+ if (!dbName.equals(newTable.getDbName()) || !tblName.equals(newTable.getTableName())) {
+ Map<String, List<ColumnStatisticsObj>> newAggrColStatsCache =
+ new HashMap<String, List<ColumnStatisticsObj>>();
+ String oldPartialKey = CacheUtils.buildKeyWithDelimit(dbName, tblName);
+ Iterator<Entry<String, List<ColumnStatisticsObj>>> iterator =
+ aggrColStatsCache.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<String, List<ColumnStatisticsObj>> entry = iterator.next();
+ String key = entry.getKey();
+ List<ColumnStatisticsObj> value = entry.getValue();
+ if (key.toLowerCase().startsWith(oldPartialKey.toLowerCase())) {
+ Object[] decomposedKey = CacheUtils.splitAggrColStats(key);
+ // New key has the new table name
+ String newKey = CacheUtils.buildKey((String) decomposedKey[0], newTable.getTableName(),
+ (String) decomposedKey[2]);
+ newAggrColStatsCache.put(newKey, value);
+ iterator.remove();
+ }
+ }
+ aggrColStatsCache.putAll(newAggrColStatsCache);
+ }
+ }
+
public synchronized int getCachedTableCount() {
return tableCache.size();
}
@@ -340,6 +387,20 @@ public class SharedCache {
}
}
+ // Remove cached column stats for all partitions of all tables in a db
+ public synchronized void removePartitionColStatsFromCache(String dbName) {
+ String partialKey = CacheUtils.buildKeyWithDelimit(dbName);
+ Iterator<Entry<String, ColumnStatisticsObj>> iterator =
+ partitionColStatsCache.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<String, ColumnStatisticsObj> entry = iterator.next();
+ String key = entry.getKey();
+ if (key.toLowerCase().startsWith(partialKey.toLowerCase())) {
+ iterator.remove();
+ }
+ }
+ }
+
// Remove cached column stats for all partitions of a table
public synchronized void removePartitionColStatsFromCache(String dbName, String tblName) {
String partialKey = CacheUtils.buildKeyWithDelimit(dbName, tblName);
@@ -445,28 +506,94 @@ public class SharedCache {
return partitionColStatsCache.get(key)!=null?partitionColStatsCache.get(key).deepCopy():null;
}
- public synchronized void addPartitionColStatsToCache(String dbName, String tableName,
- Map<String, List<ColumnStatisticsObj>> colStatsPerPartition) {
- for (Map.Entry<String, List<ColumnStatisticsObj>> entry : colStatsPerPartition.entrySet()) {
- String partName = entry.getKey();
+ public synchronized void addPartitionColStatsToCache(
+ List<ColStatsObjWithSourceInfo> colStatsForDB) {
+ for (ColStatsObjWithSourceInfo colStatWithSourceInfo : colStatsForDB) {
+ List<String> partVals;
try {
- List<String> partVals = Warehouse.getPartValuesFromPartName(partName);
- for (ColumnStatisticsObj colStatObj : entry.getValue()) {
- String key = CacheUtils.buildKey(dbName, tableName, partVals, colStatObj.getColName());
- partitionColStatsCache.put(key, colStatObj);
- }
+ partVals = Warehouse.getPartValuesFromPartName(colStatWithSourceInfo.getPartName());
+ ColumnStatisticsObj colStatObj = colStatWithSourceInfo.getColStatsObj();
+ String key = CacheUtils.buildKey(colStatWithSourceInfo.getDbName(),
+ colStatWithSourceInfo.getTblName(), partVals, colStatObj.getColName());
+ partitionColStatsCache.put(key, colStatObj);
} catch (MetaException e) {
- LOG.info("Unable to add partition: " + partName + " to SharedCache", e);
+ LOG.info("Unable to add partition stats for: {} to SharedCache",
+ colStatWithSourceInfo.getPartName(), e);
}
}
+
}
- public synchronized void refreshPartitionColStats(String dbName, String tableName,
- Map<String, List<ColumnStatisticsObj>> newColStatsPerPartition) {
- LOG.debug("CachedStore: updating cached partition column stats objects for database: " + dbName
- + " and table: " + tableName);
- removePartitionColStatsFromCache(dbName, tableName);
- addPartitionColStatsToCache(dbName, tableName, newColStatsPerPartition);
+ public synchronized void refreshPartitionColStats(String dbName,
+ List<ColStatsObjWithSourceInfo> colStatsForDB) {
+ LOG.debug("CachedStore: updating cached partition column stats objects for database: {}",
+ dbName);
+ removePartitionColStatsFromCache(dbName);
+ addPartitionColStatsToCache(colStatsForDB);
+ }
+
+ public synchronized void addAggregateStatsToCache(String dbName, String tblName,
+ AggrStats aggrStatsAllPartitions, AggrStats aggrStatsAllButDefaultPartition) {
+ if (aggrStatsAllPartitions != null) {
+ for (ColumnStatisticsObj colStatObj : aggrStatsAllPartitions.getColStats()) {
+ String key = CacheUtils.buildKey(dbName, tblName, colStatObj.getColName());
+ List<ColumnStatisticsObj> value = new ArrayList<ColumnStatisticsObj>();
+ value.add(StatsType.ALL.getPosition(), colStatObj);
+ aggrColStatsCache.put(key, value);
+ }
+ }
+ if (aggrStatsAllButDefaultPartition != null) {
+ for (ColumnStatisticsObj colStatObj : aggrStatsAllButDefaultPartition.getColStats()) {
+ String key = CacheUtils.buildKey(dbName, tblName, colStatObj.getColName());
+ List<ColumnStatisticsObj> value = aggrColStatsCache.get(key);
+ if ((value != null) && (value.size() > 0)) {
+ value.add(StatsType.ALLBUTDEFAULT.getPosition(), colStatObj);
+ }
+ }
+ }
+ }
+
+ public List<ColumnStatisticsObj> getAggrStatsFromCache(String dbName, String tblName,
+ List<String> colNames, StatsType statsType) {
+ List<ColumnStatisticsObj> colStats = new ArrayList<ColumnStatisticsObj>();
+ for (String colName : colNames) {
+ String key = CacheUtils.buildKey(dbName, tblName, colName);
+ List<ColumnStatisticsObj> colStatList = aggrColStatsCache.get(key);
+ // If unable to find stats for a column, return null so we can build stats
+ if (colStatList == null) {
+ return null;
+ }
+ ColumnStatisticsObj colStatObj = colStatList.get(statsType.getPosition());
+ // If unable to find stats for this StatsType, return null so we can build
+ // stats
+ if (colStatObj == null) {
+ return null;
+ }
+ colStats.add(colStatObj);
+ }
+ return colStats;
+ }
+
+ public synchronized void removeAggrPartitionColStatsFromCache(String dbName, String tblName) {
+ String partialKey = CacheUtils.buildKeyWithDelimit(dbName, tblName);
+ Iterator<Entry<String, List<ColumnStatisticsObj>>> iterator =
+ aggrColStatsCache.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<String, List<ColumnStatisticsObj>> entry = iterator.next();
+ String key = entry.getKey();
+ if (key.toLowerCase().startsWith(partialKey.toLowerCase())) {
+ iterator.remove();
+ }
+ }
+ }
+
+ public synchronized void refreshAggregateStatsCache(String dbName, String tblName,
+ AggrStats aggrStatsAllPartitions, AggrStats aggrStatsAllButDefaultPartition) {
+ LOG.debug("CachedStore: updating aggregate stats cache for database: {}, table: {}", dbName,
+ tblName);
+ removeAggrPartitionColStatsFromCache(dbName, tblName);
+ addAggregateStatsToCache(dbName, tblName, aggrStatsAllPartitions,
+ aggrStatsAllButDefaultPartition);
}
public synchronized void addTableColStatsToCache(String dbName, String tableName,
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BinaryColumnStatsAggregator.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BinaryColumnStatsAggregator.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BinaryColumnStatsAggregator.java
index 45d5d8c..c18b4c7 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BinaryColumnStatsAggregator.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BinaryColumnStatsAggregator.java
@@ -22,30 +22,27 @@ package org.apache.hadoop.hive.metastore.columnstats.aggr;
import java.util.List;
import org.apache.hadoop.hive.metastore.api.BinaryColumnStatsData;
-import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
public class BinaryColumnStatsAggregator extends ColumnStatsAggregator {
@Override
- public ColumnStatisticsObj aggregate(String colName, List<String> partNames,
- List<ColumnStatistics> css) throws MetaException {
+ public ColumnStatisticsObj aggregate(List<ColStatsObjWithSourceInfo> colStatsWithSourceInfo,
+ List<String> partNames, boolean areAllPartsFound) throws MetaException {
ColumnStatisticsObj statsObj = null;
- BinaryColumnStatsData aggregateData = null;
String colType = null;
- for (ColumnStatistics cs : css) {
- if (cs.getStatsObjSize() != 1) {
- throw new MetaException(
- "The number of columns should be exactly one in aggrStats, but found "
- + cs.getStatsObjSize());
- }
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ String colName = null;
+ BinaryColumnStatsData aggregateData = null;
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
if (statsObj == null) {
+ colName = cso.getColName();
colType = cso.getColType();
- statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType, cso
- .getStatsData().getSetField());
+ statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType,
+ cso.getStatsData().getSetField());
}
BinaryColumnStatsData newData = cso.getStatsData().getBinaryStats();
if (aggregateData == null) {
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BooleanColumnStatsAggregator.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BooleanColumnStatsAggregator.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BooleanColumnStatsAggregator.java
index 8aac0fe..7630183 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BooleanColumnStatsAggregator.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/BooleanColumnStatsAggregator.java
@@ -22,30 +22,27 @@ package org.apache.hadoop.hive.metastore.columnstats.aggr;
import java.util.List;
import org.apache.hadoop.hive.metastore.api.BooleanColumnStatsData;
-import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
public class BooleanColumnStatsAggregator extends ColumnStatsAggregator {
@Override
- public ColumnStatisticsObj aggregate(String colName, List<String> partNames,
- List<ColumnStatistics> css) throws MetaException {
+ public ColumnStatisticsObj aggregate(List<ColStatsObjWithSourceInfo> colStatsWithSourceInfo,
+ List<String> partNames, boolean areAllPartsFound) throws MetaException {
ColumnStatisticsObj statsObj = null;
- BooleanColumnStatsData aggregateData = null;
String colType = null;
- for (ColumnStatistics cs : css) {
- if (cs.getStatsObjSize() != 1) {
- throw new MetaException(
- "The number of columns should be exactly one in aggrStats, but found "
- + cs.getStatsObjSize());
- }
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ String colName = null;
+ BooleanColumnStatsData aggregateData = null;
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
if (statsObj == null) {
+ colName = cso.getColName();
colType = cso.getColType();
- statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType, cso
- .getStatsData().getSetField());
+ statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType,
+ cso.getStatsData().getSetField());
}
BooleanColumnStatsData newData = cso.getStatsData().getBooleanStats();
if (aggregateData == null) {
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/ColumnStatsAggregator.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/ColumnStatsAggregator.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/ColumnStatsAggregator.java
index cd0392d..0beaf60 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/ColumnStatsAggregator.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/ColumnStatsAggregator.java
@@ -21,13 +21,15 @@ package org.apache.hadoop.hive.metastore.columnstats.aggr;
import java.util.List;
-import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
public abstract class ColumnStatsAggregator {
public boolean useDensityFunctionForNDVEstimation;
public double ndvTuner;
- public abstract ColumnStatisticsObj aggregate(String colName, List<String> partNames,
- List<ColumnStatistics> css) throws MetaException;
+
+ public abstract ColumnStatisticsObj aggregate(
+ List<ColStatsObjWithSourceInfo> colStatsWithSourceInfo, List<String> partNames,
+ boolean areAllPartsFound) throws MetaException;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DateColumnStatsAggregator.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DateColumnStatsAggregator.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DateColumnStatsAggregator.java
index 7f29561..e8ff513 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DateColumnStatsAggregator.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DateColumnStatsAggregator.java
@@ -28,13 +28,13 @@ import java.util.Map;
import org.apache.hadoop.hive.common.ndv.NumDistinctValueEstimator;
import org.apache.hadoop.hive.common.ndv.NumDistinctValueEstimatorFactory;
-import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.Date;
import org.apache.hadoop.hive.metastore.api.DateColumnStatsData;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.columnstats.cache.DateColumnStatsDataInspector;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,27 +44,23 @@ public class DateColumnStatsAggregator extends ColumnStatsAggregator implements
private static final Logger LOG = LoggerFactory.getLogger(DateColumnStatsAggregator.class);
@Override
- public ColumnStatisticsObj aggregate(String colName, List<String> partNames,
- List<ColumnStatistics> css) throws MetaException {
+ public ColumnStatisticsObj aggregate(List<ColStatsObjWithSourceInfo> colStatsWithSourceInfo,
+ List<String> partNames, boolean areAllPartsFound) throws MetaException {
ColumnStatisticsObj statsObj = null;
-
+ String colType = null;
+ String colName = null;
// check if all the ColumnStatisticsObjs contain stats and all the ndv are
// bitvectors
- boolean doAllPartitionContainStats = partNames.size() == css.size();
- LOG.debug("doAllPartitionContainStats for " + colName + " is " + doAllPartitionContainStats);
+ boolean doAllPartitionContainStats = partNames.size() == colStatsWithSourceInfo.size();
NumDistinctValueEstimator ndvEstimator = null;
- String colType = null;
- for (ColumnStatistics cs : css) {
- if (cs.getStatsObjSize() != 1) {
- throw new MetaException(
- "The number of columns should be exactly one in aggrStats, but found "
- + cs.getStatsObjSize());
- }
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
if (statsObj == null) {
+ colName = cso.getColName();
colType = cso.getColType();
- statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType, cso
- .getStatsData().getSetField());
+ statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType,
+ cso.getStatsData().getSetField());
+ LOG.trace("doAllPartitionContainStats for column: {} is: {}", colName, doAllPartitionContainStats);
}
DateColumnStatsDataInspector dateColumnStats =
(DateColumnStatsDataInspector) cso.getStatsData().getDateStats();
@@ -92,13 +88,13 @@ public class DateColumnStatsAggregator extends ColumnStatsAggregator implements
}
LOG.debug("all of the bit vectors can merge for " + colName + " is " + (ndvEstimator != null));
ColumnStatisticsData columnStatisticsData = new ColumnStatisticsData();
- if (doAllPartitionContainStats || css.size() < 2) {
+ if (doAllPartitionContainStats || colStatsWithSourceInfo.size() < 2) {
DateColumnStatsDataInspector aggregateData = null;
long lowerBound = 0;
long higherBound = 0;
double densityAvgSum = 0.0;
- for (ColumnStatistics cs : css) {
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
DateColumnStatsDataInspector newData =
(DateColumnStatsDataInspector) cso.getStatsData().getDateStats();
lowerBound = Math.max(lowerBound, newData.getNumDVs());
@@ -157,9 +153,9 @@ public class DateColumnStatsAggregator extends ColumnStatsAggregator implements
if (ndvEstimator == null) {
// if not every partition uses bitvector for ndv, we just fall back to
// the traditional extrapolation methods.
- for (ColumnStatistics cs : css) {
- String partName = cs.getStatsDesc().getPartName();
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
+ String partName = csp.getPartName();
DateColumnStatsData newData = cso.getStatsData().getDateStats();
if (useDensityFunctionForNDVEstimation) {
densityAvgSum += diff(newData.getHighValue(), newData.getLowValue()) / newData.getNumDVs();
@@ -175,9 +171,9 @@ public class DateColumnStatsAggregator extends ColumnStatsAggregator implements
int length = 0;
int curIndex = -1;
DateColumnStatsDataInspector aggregateData = null;
- for (ColumnStatistics cs : css) {
- String partName = cs.getStatsDesc().getPartName();
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
+ String partName = csp.getPartName();
DateColumnStatsDataInspector newData =
(DateColumnStatsDataInspector) cso.getStatsData().getDateStats();
// newData.isSetBitVectors() should be true for sure because we
@@ -230,11 +226,13 @@ public class DateColumnStatsAggregator extends ColumnStatsAggregator implements
}
}
}
- extrapolate(columnStatisticsData, partNames.size(), css.size(), adjustedIndexMap,
- adjustedStatsMap, densityAvgSum / adjustedStatsMap.size());
+ extrapolate(columnStatisticsData, partNames.size(), colStatsWithSourceInfo.size(),
+ adjustedIndexMap, adjustedStatsMap, densityAvgSum / adjustedStatsMap.size());
}
- LOG.debug("Ndv estimatation for {} is {} # of partitions requested: {} # of partitions found: {}", colName,
- columnStatisticsData.getDateStats().getNumDVs(),partNames.size(), css.size());
+ LOG.debug(
+ "Ndv estimatation for {} is {} # of partitions requested: {} # of partitions found: {}",
+ colName, columnStatisticsData.getDateStats().getNumDVs(), partNames.size(),
+ colStatsWithSourceInfo.size());
statsObj.setStatsData(columnStatisticsData);
return statsObj;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DecimalColumnStatsAggregator.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DecimalColumnStatsAggregator.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DecimalColumnStatsAggregator.java
index 05c0280..2715599 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DecimalColumnStatsAggregator.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DecimalColumnStatsAggregator.java
@@ -29,13 +29,13 @@ import java.util.Map;
import org.apache.hadoop.hive.common.ndv.NumDistinctValueEstimator;
import org.apache.hadoop.hive.common.ndv.NumDistinctValueEstimatorFactory;
import org.apache.hadoop.hive.metastore.StatObjectConverter;
-import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.DecimalColumnStatsData;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.columnstats.cache.DecimalColumnStatsDataInspector;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,27 +45,24 @@ public class DecimalColumnStatsAggregator extends ColumnStatsAggregator implemen
private static final Logger LOG = LoggerFactory.getLogger(DecimalColumnStatsAggregator.class);
@Override
- public ColumnStatisticsObj aggregate(String colName, List<String> partNames,
- List<ColumnStatistics> css) throws MetaException {
+ public ColumnStatisticsObj aggregate(List<ColStatsObjWithSourceInfo> colStatsWithSourceInfo,
+ List<String> partNames, boolean areAllPartsFound) throws MetaException {
ColumnStatisticsObj statsObj = null;
-
+ String colType = null;
+ String colName = null;
// check if all the ColumnStatisticsObjs contain stats and all the ndv are
// bitvectors
- boolean doAllPartitionContainStats = partNames.size() == css.size();
- LOG.debug("doAllPartitionContainStats for " + colName + " is " + doAllPartitionContainStats);
+ boolean doAllPartitionContainStats = partNames.size() == colStatsWithSourceInfo.size();
NumDistinctValueEstimator ndvEstimator = null;
- String colType = null;
- for (ColumnStatistics cs : css) {
- if (cs.getStatsObjSize() != 1) {
- throw new MetaException(
- "The number of columns should be exactly one in aggrStats, but found "
- + cs.getStatsObjSize());
- }
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
if (statsObj == null) {
+ colName = cso.getColName();
colType = cso.getColType();
- statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType, cso
- .getStatsData().getSetField());
+ statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType,
+ cso.getStatsData().getSetField());
+ LOG.trace("doAllPartitionContainStats for column: {} is: {}", colName,
+ doAllPartitionContainStats);
}
DecimalColumnStatsDataInspector decimalColumnStatsData =
(DecimalColumnStatsDataInspector) cso.getStatsData().getDecimalStats();
@@ -93,13 +90,13 @@ public class DecimalColumnStatsAggregator extends ColumnStatsAggregator implemen
}
LOG.debug("all of the bit vectors can merge for " + colName + " is " + (ndvEstimator != null));
ColumnStatisticsData columnStatisticsData = new ColumnStatisticsData();
- if (doAllPartitionContainStats || css.size() < 2) {
+ if (doAllPartitionContainStats || colStatsWithSourceInfo.size() < 2) {
DecimalColumnStatsDataInspector aggregateData = null;
long lowerBound = 0;
long higherBound = 0;
double densityAvgSum = 0.0;
- for (ColumnStatistics cs : css) {
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
DecimalColumnStatsDataInspector newData =
(DecimalColumnStatsDataInspector) cso.getStatsData().getDecimalStats();
lowerBound = Math.max(lowerBound, newData.getNumDVs());
@@ -167,9 +164,9 @@ public class DecimalColumnStatsAggregator extends ColumnStatsAggregator implemen
if (ndvEstimator == null) {
// if not every partition uses bitvector for ndv, we just fall back to
// the traditional extrapolation methods.
- for (ColumnStatistics cs : css) {
- String partName = cs.getStatsDesc().getPartName();
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
+ String partName = csp.getPartName();
DecimalColumnStatsData newData = cso.getStatsData().getDecimalStats();
if (useDensityFunctionForNDVEstimation) {
densityAvgSum += (MetaStoreUtils.decimalToDouble(newData.getHighValue()) - MetaStoreUtils
@@ -186,9 +183,9 @@ public class DecimalColumnStatsAggregator extends ColumnStatsAggregator implemen
int length = 0;
int curIndex = -1;
DecimalColumnStatsDataInspector aggregateData = null;
- for (ColumnStatistics cs : css) {
- String partName = cs.getStatsDesc().getPartName();
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
+ String partName = csp.getPartName();
DecimalColumnStatsDataInspector newData =
(DecimalColumnStatsDataInspector) cso.getStatsData().getDecimalStats();
// newData.isSetBitVectors() should be true for sure because we
@@ -251,11 +248,13 @@ public class DecimalColumnStatsAggregator extends ColumnStatsAggregator implemen
}
}
}
- extrapolate(columnStatisticsData, partNames.size(), css.size(), adjustedIndexMap,
- adjustedStatsMap, densityAvgSum / adjustedStatsMap.size());
+ extrapolate(columnStatisticsData, partNames.size(), colStatsWithSourceInfo.size(),
+ adjustedIndexMap, adjustedStatsMap, densityAvgSum / adjustedStatsMap.size());
}
- LOG.debug("Ndv estimatation for {} is {} # of partitions requested: {} # of partitions found: {}", colName,
- columnStatisticsData.getDecimalStats().getNumDVs(),partNames.size(), css.size());
+ LOG.debug(
+ "Ndv estimatation for {} is {} # of partitions requested: {} # of partitions found: {}",
+ colName, columnStatisticsData.getDecimalStats().getNumDVs(), partNames.size(),
+ colStatsWithSourceInfo.size());
statsObj.setStatsData(columnStatisticsData);
return statsObj;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/456a6518/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DoubleColumnStatsAggregator.java
----------------------------------------------------------------------
diff --git a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DoubleColumnStatsAggregator.java b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DoubleColumnStatsAggregator.java
index faf22dc..ece77dd 100644
--- a/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DoubleColumnStatsAggregator.java
+++ b/standalone-metastore/src/main/java/org/apache/hadoop/hive/metastore/columnstats/aggr/DoubleColumnStatsAggregator.java
@@ -28,12 +28,12 @@ import java.util.Map;
import org.apache.hadoop.hive.common.ndv.NumDistinctValueEstimator;
import org.apache.hadoop.hive.common.ndv.NumDistinctValueEstimatorFactory;
-import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.DoubleColumnStatsData;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.columnstats.cache.DoubleColumnStatsDataInspector;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.ColStatsObjWithSourceInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,27 +43,24 @@ public class DoubleColumnStatsAggregator extends ColumnStatsAggregator implement
private static final Logger LOG = LoggerFactory.getLogger(LongColumnStatsAggregator.class);
@Override
- public ColumnStatisticsObj aggregate(String colName, List<String> partNames,
- List<ColumnStatistics> css) throws MetaException {
+ public ColumnStatisticsObj aggregate(List<ColStatsObjWithSourceInfo> colStatsWithSourceInfo,
+ List<String> partNames, boolean areAllPartsFound) throws MetaException {
ColumnStatisticsObj statsObj = null;
-
+ String colType = null;
+ String colName = null;
// check if all the ColumnStatisticsObjs contain stats and all the ndv are
// bitvectors
- boolean doAllPartitionContainStats = partNames.size() == css.size();
- LOG.debug("doAllPartitionContainStats for " + colName + " is " + doAllPartitionContainStats);
+ boolean doAllPartitionContainStats = partNames.size() == colStatsWithSourceInfo.size();
NumDistinctValueEstimator ndvEstimator = null;
- String colType = null;
- for (ColumnStatistics cs : css) {
- if (cs.getStatsObjSize() != 1) {
- throw new MetaException(
- "The number of columns should be exactly one in aggrStats, but found "
- + cs.getStatsObjSize());
- }
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
if (statsObj == null) {
+ colName = cso.getColName();
colType = cso.getColType();
- statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType, cso
- .getStatsData().getSetField());
+ statsObj = ColumnStatsAggregatorFactory.newColumnStaticsObj(colName, colType,
+ cso.getStatsData().getSetField());
+ LOG.trace("doAllPartitionContainStats for column: {} is: {}", colName,
+ doAllPartitionContainStats);
}
DoubleColumnStatsDataInspector doubleColumnStatsData =
(DoubleColumnStatsDataInspector) cso.getStatsData().getDoubleStats();
@@ -91,13 +88,13 @@ public class DoubleColumnStatsAggregator extends ColumnStatsAggregator implement
}
LOG.debug("all of the bit vectors can merge for " + colName + " is " + (ndvEstimator != null));
ColumnStatisticsData columnStatisticsData = new ColumnStatisticsData();
- if (doAllPartitionContainStats || css.size() < 2) {
+ if (doAllPartitionContainStats || colStatsWithSourceInfo.size() < 2) {
DoubleColumnStatsDataInspector aggregateData = null;
long lowerBound = 0;
long higherBound = 0;
double densityAvgSum = 0.0;
- for (ColumnStatistics cs : css) {
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
DoubleColumnStatsDataInspector newData =
(DoubleColumnStatsDataInspector) cso.getStatsData().getDoubleStats();
lowerBound = Math.max(lowerBound, newData.getNumDVs());
@@ -154,9 +151,9 @@ public class DoubleColumnStatsAggregator extends ColumnStatsAggregator implement
if (ndvEstimator == null) {
// if not every partition uses bitvector for ndv, we just fall back to
// the traditional extrapolation methods.
- for (ColumnStatistics cs : css) {
- String partName = cs.getStatsDesc().getPartName();
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
+ String partName = csp.getPartName();
DoubleColumnStatsData newData = cso.getStatsData().getDoubleStats();
if (useDensityFunctionForNDVEstimation) {
densityAvgSum += (newData.getHighValue() - newData.getLowValue()) / newData.getNumDVs();
@@ -172,9 +169,9 @@ public class DoubleColumnStatsAggregator extends ColumnStatsAggregator implement
int length = 0;
int curIndex = -1;
DoubleColumnStatsData aggregateData = null;
- for (ColumnStatistics cs : css) {
- String partName = cs.getStatsDesc().getPartName();
- ColumnStatisticsObj cso = cs.getStatsObjIterator().next();
+ for (ColStatsObjWithSourceInfo csp : colStatsWithSourceInfo) {
+ ColumnStatisticsObj cso = csp.getColStatsObj();
+ String partName = csp.getPartName();
DoubleColumnStatsDataInspector newData =
(DoubleColumnStatsDataInspector) cso.getStatsData().getDoubleStats();
// newData.isSetBitVectors() should be true for sure because we
@@ -226,11 +223,13 @@ public class DoubleColumnStatsAggregator extends ColumnStatsAggregator implement
}
}
}
- extrapolate(columnStatisticsData, partNames.size(), css.size(), adjustedIndexMap,
- adjustedStatsMap, densityAvgSum / adjustedStatsMap.size());
+ extrapolate(columnStatisticsData, partNames.size(), colStatsWithSourceInfo.size(),
+ adjustedIndexMap, adjustedStatsMap, densityAvgSum / adjustedStatsMap.size());
}
- LOG.debug("Ndv estimatation for {} is {}. # of partitions requested: {}. # of partitions found: {}", colName,
- columnStatisticsData.getDoubleStats().getNumDVs(),partNames.size(), css.size());
+ LOG.debug(
+ "Ndv estimatation for {} is {}. # of partitions requested: {}. # of partitions found: {}",
+ colName, columnStatisticsData.getDoubleStats().getNumDVs(), partNames.size(),
+ colStatsWithSourceInfo.size());
statsObj.setStatsData(columnStatisticsData);
return statsObj;
}