You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by nz...@apache.org on 2011/06/03 18:45:40 UTC
svn commit: r1131106 [3/11] - in /hive/trunk:
common/src/java/org/apache/hadoop/hive/conf/
contrib/src/java/org/apache/hadoop/hive/contrib/serde2/
contrib/src/java/org/apache/hadoop/hive/contrib/serde2/s3/
hbase-handler/src/java/org/apache/hadoop/hive/...
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java Fri Jun 3 16:45:37 2011
@@ -27,9 +27,9 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;
-import java.util.Map.Entry;
import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;
@@ -92,6 +92,7 @@ import org.apache.hadoop.hive.ql.metadat
import org.apache.hadoop.hive.ql.optimizer.GenMRFileSink1;
import org.apache.hadoop.hive.ql.optimizer.GenMROperator;
import org.apache.hadoop.hive.ql.optimizer.GenMRProcContext;
+import org.apache.hadoop.hive.ql.optimizer.GenMRProcContext.GenMapRedCtx;
import org.apache.hadoop.hive.ql.optimizer.GenMRRedSink1;
import org.apache.hadoop.hive.ql.optimizer.GenMRRedSink2;
import org.apache.hadoop.hive.ql.optimizer.GenMRRedSink3;
@@ -101,7 +102,6 @@ import org.apache.hadoop.hive.ql.optimiz
import org.apache.hadoop.hive.ql.optimizer.GenMapRedUtils;
import org.apache.hadoop.hive.ql.optimizer.MapJoinFactory;
import org.apache.hadoop.hive.ql.optimizer.Optimizer;
-import org.apache.hadoop.hive.ql.optimizer.GenMRProcContext.GenMapRedCtx;
import org.apache.hadoop.hive.ql.optimizer.physical.PhysicalContext;
import org.apache.hadoop.hive.ql.optimizer.physical.PhysicalOptimizer;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
@@ -122,6 +122,7 @@ import org.apache.hadoop.hive.ql.plan.Ex
import org.apache.hadoop.hive.ql.plan.FetchWork;
import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.FilterDesc;
+import org.apache.hadoop.hive.ql.plan.FilterDesc.sampleDesc;
import org.apache.hadoop.hive.ql.plan.ForwardDesc;
import org.apache.hadoop.hive.ql.plan.GroupByDesc;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
@@ -144,13 +145,12 @@ import org.apache.hadoop.hive.ql.plan.Ta
import org.apache.hadoop.hive.ql.plan.TableScanDesc;
import org.apache.hadoop.hive.ql.plan.UDTFDesc;
import org.apache.hadoop.hive.ql.plan.UnionDesc;
-import org.apache.hadoop.hive.ql.plan.FilterDesc.sampleDesc;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionState.ResourceType;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFHash;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
-import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
import org.apache.hadoop.hive.serde.Constants;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.hive.serde2.MetadataTypedColumnsetSerDe;
@@ -158,9 +158,9 @@ import org.apache.hadoop.hive.serde2.Ser
import org.apache.hadoop.hive.serde2.SerDeUtils;
import org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
@@ -6136,7 +6136,7 @@ public class SemanticAnalyzer extends Ba
// Create the root of the operator tree
TableScanDesc tsDesc = new TableScanDesc(alias, vcList);
- setupStats(tsDesc, qb.getParseInfo(), tab, alias);
+ setupStats(tsDesc, qb.getParseInfo(), tab, alias, rwsch);
top = putOpInsertMap(OperatorFactory.get(tsDesc,
new RowSchema(rwsch.getColumnInfos())), rwsch);
@@ -6289,7 +6289,7 @@ public class SemanticAnalyzer extends Ba
return output;
}
- private void setupStats(TableScanDesc tsDesc, QBParseInfo qbp, Table tab, String alias)
+ private void setupStats(TableScanDesc tsDesc, QBParseInfo qbp, Table tab, String alias, RowResolver rwsch)
throws SemanticException {
if (!qbp.isAnalyzeCommand()) {
@@ -6297,6 +6297,17 @@ public class SemanticAnalyzer extends Ba
} else {
tsDesc.setGatherStats(true);
+ // append additional virtual columns for storing statistics
+ Iterator<VirtualColumn> vcs = VirtualColumn.getStatsRegistry(conf).iterator();
+ List<VirtualColumn> vcList = new ArrayList<VirtualColumn>();
+ while (vcs.hasNext()) {
+ VirtualColumn vc = vcs.next();
+ rwsch.put(alias, vc.getName(), new ColumnInfo(vc.getName(),
+ vc.getTypeInfo(), alias, true, vc.getIsHidden()));
+ vcList.add(vc);
+ }
+ tsDesc.addVirtualCols(vcList);
+
String tblName = tab.getTableName();
tableSpec tblSpec = qbp.getTableSpec(alias);
Map<String, String> partSpec = tblSpec.getPartSpec();
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java Fri Jun 3 16:45:37 2011
@@ -113,6 +113,10 @@ public class TableScanDesc implements Se
this.virtualCols = virtualCols;
}
+ public void addVirtualCols(List<VirtualColumn> virtualCols) {
+ this.virtualCols.addAll(virtualCols);
+ }
+
public void setStatsAggPrefix(String k) {
statsAggKeyPrefix = k;
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsAggregator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsAggregator.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsAggregator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsAggregator.java Fri Jun 3 16:45:37 2011
@@ -28,7 +28,9 @@ public interface StatsAggregator {
/**
* This method connects to the temporary storage.
- * @param hconf HiveConf that contains the connection parameters.
+ *
+ * @param hconf
+ * HiveConf that contains the connection parameters.
* @return true if connection is successful, false otherwise.
*/
public boolean connect(Configuration hconf);
@@ -38,34 +40,43 @@ public interface StatsAggregator {
* After aggregation, this method also automatically removes all records
* that have been aggregated.
*
- * @param keyPrefix a prefix of the keys used in StatsPublisher to publish stats.
- * Any rows that starts with the same prefix will be aggregated. For example, if
- * the StatsPublisher uses the following compound key to publish stats:
- *
- * the output directory name (unique per FileSinkOperator) +
- * the partition specs (only for dynamic partitions) +
- * taskID (last component of task file)
+ * @param keyPrefix
+ * a prefix of the keys used in StatsPublisher to publish stats.
+ * Any rows that starts with the same prefix will be aggregated. For example, if
+ * the StatsPublisher uses the following compound key to publish stats:
+ *
+ * the output directory name (unique per FileSinkOperator) +
+ * the partition specs (only for dynamic partitions) +
+ * taskID (last component of task file)
*
- * The keyPrefix for aggregation could be first 2 components. This will aggregates stats
- * across all tasks for each partition.
+ * The keyPrefix for aggregation could be first 2 components. This will aggregates stats
+ * across all tasks for each partition.
*
- * @param statType a string noting the key to be published. Ex: "numRows".
+ * @param statType
+ * a string noting the key to be published. Ex: "numRows".
* @return a string representation of a long value, null if there are any error/exception.
*/
public String aggregateStats(String keyPrefix, String statType);
/**
* This method closes the connection to the temporary storage.
+ *
* @return true if close connection is successful, false otherwise.
*/
public boolean closeConnection();
/**
- * This method clears the temporary statistics that have been published without being aggregated.
- * Typically this happens when a job fails, or is forcibly stopped after publishing some statistics.
+ * This method is called after all statistics have been aggregated. Since we support multiple
+ * statistics, we do not perform automatic cleanup after aggregation.
+ * After this method is called, closeConnection must be called as well.
+ * This method is also used to clear the temporary statistics that have been published without
+ * being aggregated.
+ * Typically this happens when a job fails, or is forcibly stopped after publishing some
+ * statistics.
*
- * @param keyPrefix a prefix of the keys used in StatsPublisher to publish stats. It is the same
- * as the first parameter in aggregateStats().
+ * @param keyPrefix
+ * a prefix of the keys used in StatsPublisher to publish stats. It is the same
+ * as the first parameter in aggregateStats().
* @return true if cleanup is successful, false otherwise.
*/
public boolean cleanUp(String keyPrefix);
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsPublisher.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsPublisher.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsPublisher.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsPublisher.java Fri Jun 3 16:45:37 2011
@@ -18,6 +18,8 @@
package org.apache.hadoop.hive.ql.stats;
+import java.util.Map;
+
import org.apache.hadoop.conf.Configuration;
/**
@@ -47,17 +49,19 @@ public interface StatsPublisher {
/**
* This method publishes a given statistic into a disk storage, possibly HBase or MySQL.
*
- * fileID : a string identification the statistics to be published by all mappers/reducers
- * and then gathered. The statID is unique per output partition per task, e.g.,:
- * the output directory name (uniq per FileSinkOperator) +
- * the partition specs (only for dynamic partitions) +
- * taskID (last component of task file)
- *
- * statType : a string noting the key to be published. Ex: "numRows".
- *
- * value : an integer noting the value of the published key.
+ * @param fileID
+ * : a string identification the statistics to be published by all mappers/reducers
+ * and then gathered. The statID is unique per output partition per task, e.g.,:
+ * the output directory name (uniq per FileSinkOperator) +
+ * the partition specs (only for dynamic partitions) +
+ * taskID (last component of task file)
+ * @param stats
+ * : a map containing key-value pairs, where key is a string representing the statistic
+ * to be published,
+ * and value is a string representing the value for the given statistic
+ * @return true if successful, false otherwise
*/
- public boolean publishStat(String fileID, String statType, String value);
+ public boolean publishStat(String fileID, Map<String, String> stats);
/**
* This method closes the connection to the temporary storage.
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsSetupConst.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsSetupConst.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsSetupConst.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsSetupConst.java Fri Jun 3 16:45:37 2011
@@ -34,23 +34,31 @@ public class StatsSetupConst {
public static final String JDBC_IMPL_CLASS_VAL = "jdbc";
/**
- * The name of the statistic Row Count to be published or gathered.
+ * The name of the statistic Num Files to be published or gathered.
*/
- public static final String ROW_COUNT = "numRows";
+ public static final String NUM_FILES = "numFiles";
/**
- * The name of the statistic Row Count to be published or gathered.
+ * The name of the statistic Num Partitions to be published or gathered.
*/
- public static final String NUM_FILES = "numFiles";
+ public static final String NUM_PARTITIONS = "numPartitions";
/**
- * The name of the statistic Row Count to be published or gathered.
+ * The name of the statistic Total Size to be published or gathered.
*/
- public static final String NUM_PARTITIONS = "numPartitions";
+ public static final String TOTAL_SIZE = "totalSize";
+
+
+ // statistics stored in metastore
/**
* The name of the statistic Row Count to be published or gathered.
*/
- public static final String TOTAL_SIZE = "totalSize";
+ public static final String ROW_COUNT = "numRows";
+
+ /**
+ * The name of the statistic Raw Data Size to be published or gathered.
+ */
+ public static final String RAW_DATA_SIZE = "rawDataSize";
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsAggregator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsAggregator.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsAggregator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsAggregator.java Fri Jun 3 16:45:37 2011
@@ -24,7 +24,8 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLRecoverableException;
-import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Random;
import org.apache.commons.logging.Log;
@@ -33,14 +34,13 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.stats.StatsAggregator;
-import org.apache.hadoop.hive.ql.stats.StatsSetupConst;
public class JDBCStatsAggregator implements StatsAggregator {
private Connection conn;
private String connectionString;
private Configuration hiveconf;
- private PreparedStatement selStmt, delStmt;
+ private final Map<String, PreparedStatement> columnMapping;
private final Log LOG = LogFactory.getLog(this.getClass().getName());
private int timeout = 30;
private final String comment = "Hive stats aggregation: " + this.getClass().getName();
@@ -48,6 +48,7 @@ public class JDBCStatsAggregator impleme
private final Random r;
public JDBCStatsAggregator() {
+ columnMapping = new HashMap<String, PreparedStatement>();
r = new Random();
}
@@ -67,28 +68,6 @@ public class JDBCStatsAggregator impleme
return false;
}
- // prepare the SELECT/DELETE statements
- String select =
- "SELECT /* " + comment + " */ " +
- " SUM(" + JDBCStatsSetupConstants.PART_STAT_ROW_COUNT_COLUMN_NAME + ")" +
- " FROM " + JDBCStatsSetupConstants.PART_STAT_TABLE_NAME +
- " WHERE " + JDBCStatsSetupConstants.PART_STAT_ID_COLUMN_NAME +
- " LIKE ? ESCAPE ?";
-
- /* Automatic Cleaning:
- IMPORTANT: Since we publish and aggregate only 1 value (1 column) which is the row count, it
- is valid to delete the row after aggregation (automatic cleaning) because we know that there is no
- other values to aggregate.
- If ;in the future; other values are aggregated and published, then we cannot do cleaning except
- when we are sure that all values are aggregated, or we can separate the implementation of cleaning
- through a separate method which the developer has to call it manually in the code.
- */
- String delete =
- "DELETE /* " + comment + " */ " +
- " FROM " + JDBCStatsSetupConstants.PART_STAT_TABLE_NAME +
- " WHERE " + JDBCStatsSetupConstants.PART_STAT_ID_COLUMN_NAME +
- " LIKE ? ESCAPE ?";
-
// stats is non-blocking -- throw an exception when timeout
DriverManager.setLoginTimeout(timeout);
// function pointer for executeWithRetry to setQueryTimeout
@@ -101,23 +80,23 @@ public class JDBCStatsAggregator impleme
};
// retry connection and statement preparations
- for (int failures = 0; ;failures++) {
+ for (int failures = 0;; failures++) {
try {
conn = Utilities.connectWithRetry(connectionString, waitWindow, maxRetries);
- // prepare statements
- selStmt = Utilities.prepareWithRetry(conn, select, waitWindow, maxRetries);
- delStmt = Utilities.prepareWithRetry(conn, delete, waitWindow, maxRetries);
-
- // set query timeout
- Utilities.executeWithRetry(setQueryTimeout, selStmt, waitWindow, failures);
- Utilities.executeWithRetry(setQueryTimeout, delStmt, waitWindow, failures);
-
+ for (String statType : JDBCStatsUtils.getSupportedStatistics()) {
+ // prepare statements
+ PreparedStatement selStmt = Utilities.prepareWithRetry(conn,
+ JDBCStatsUtils.getSelectAggr(statType, comment), waitWindow, maxRetries);
+ columnMapping.put(statType, selStmt);
+ // set query timeout
+ Utilities.executeWithRetry(setQueryTimeout, selStmt, waitWindow, failures);
+ }
return true;
} catch (SQLRecoverableException e) {
if (failures > maxRetries) {
LOG.error("Error during JDBC connection and preparing statement: " + e);
- return false;
+ return false;
}
long waitTime = Utilities.getRandomWaitTime(waitWindow, failures, r);
try {
@@ -135,11 +114,9 @@ public class JDBCStatsAggregator impleme
@Override
public String aggregateStats(String fileID, String statType) {
- LOG.info("Stats Aggregator for key " + fileID);
-
- if (statType != StatsSetupConst.ROW_COUNT) {
- LOG.warn("Warning. Invalid statistic. Currently " +
- "row count is the only supported statistic");
+ if (!JDBCStatsUtils.isValidStatistic(statType)) {
+ LOG.warn("Warning. Invalid statistic: " + statType + ", supported stats: " +
+ JDBCStatsUtils.getSupportedStatistics());
return null;
}
@@ -150,43 +127,22 @@ public class JDBCStatsAggregator impleme
}
};
- Utilities.SQLCommand<Void> execUpdate = new Utilities.SQLCommand<Void>() {
- @Override
- public Void run(PreparedStatement stmt) throws SQLException {
- stmt.executeUpdate();
- return null;
- }
- };
-
String keyPrefix = Utilities.escapeSqlLike(fileID) + "%";
- for (int failures = 0; ; failures++) {
+ for (int failures = 0;; failures++) {
try {
long retval = 0;
+ PreparedStatement selStmt = columnMapping.get(statType);
selStmt.setString(1, keyPrefix);
selStmt.setString(2, Character.toString(Utilities.sqlEscapeChar));
+
ResultSet result = Utilities.executeWithRetry(execQuery, selStmt, waitWindow, maxRetries);
if (result.next()) {
retval = result.getLong(1);
} else {
LOG.warn("Warning. Nothing published. Nothing aggregated.");
- return "";
+ return null;
}
-
- /* Automatic Cleaning:
- IMPORTANT: Since we publish and aggregate only 1 value (1 column) which is the row count, it
- is valid to delete the row after aggregation (automatic cleaning) because we know that there is no
- other values to aggregate.
- If in the future; other values are aggregated and published, then we cannot do cleaning except
- when we are sure that all values are aggregated, or we can separate the implementation of cleaning
- through a separate method which the developer has to call it manually in the code.
- */
- delStmt.setString(1, keyPrefix);
- delStmt.setString(2, Character.toString(Utilities.sqlEscapeChar));
- Utilities.executeWithRetry(execUpdate, delStmt, waitWindow, maxRetries);
-
- LOG.info("Stats aggregator got " + retval);
-
return Long.toString(retval);
} catch (SQLRecoverableException e) {
// need to start from scratch (connection)
@@ -225,12 +181,13 @@ public class JDBCStatsAggregator impleme
try {
conn.close();
// In case of derby, explicitly close the database connection
- if(HiveConf.getVar(hiveconf, HiveConf.ConfVars.HIVESTATSDBCLASS).equalsIgnoreCase("jdbc:derby")) {
+ if (HiveConf.getVar(hiveconf, HiveConf.ConfVars.HIVESTATSDBCLASS).equalsIgnoreCase(
+ "jdbc:derby")) {
try {
- // The following closes the derby connection. It throws an exception that has to be caught and ignored.
+ // The following closes the derby connection. It throws an exception that has to be caught
+ // and ignored.
DriverManager.getConnection(connectionString + ";shutdown=true");
- }
- catch (Exception e) {
+ } catch (Exception e) {
// Do nothing because we know that an exception is thrown anyway.
}
}
@@ -243,16 +200,46 @@ public class JDBCStatsAggregator impleme
@Override
public boolean cleanUp(String rowID) {
+
+ Utilities.SQLCommand<Void> execUpdate = new Utilities.SQLCommand<Void>() {
+ @Override
+ public Void run(PreparedStatement stmt) throws SQLException {
+ stmt.executeUpdate();
+ return null;
+ }
+ };
try {
- Statement stmt = conn.createStatement();
- String delete =
- "DELETE /* " + comment + " */ " +
- " FROM " + JDBCStatsSetupConstants.PART_STAT_TABLE_NAME +
- " WHERE " + JDBCStatsSetupConstants.PART_STAT_ID_COLUMN_NAME + " LIKE '" + rowID + "%'";
- stmt.executeUpdate(delete);
- stmt.close();
- return closeConnection();
+ PreparedStatement delStmt = Utilities.prepareWithRetry(conn,
+ JDBCStatsUtils.getDeleteAggr(rowID, comment), waitWindow, maxRetries);
+ for (int failures = 0;; failures++) {
+ try {
+ Utilities.executeWithRetry(execUpdate, delStmt, waitWindow, maxRetries);
+ return true;
+ } catch (SQLRecoverableException e) {
+ // need to start from scratch (connection)
+ if (failures >= maxRetries) {
+ return false;
+ }
+ // close the current connection
+ closeConnection();
+ long waitTime = Utilities.getRandomWaitTime(waitWindow, failures, r);
+ try {
+ Thread.sleep(waitTime);
+ } catch (InterruptedException iex) {
+ }
+ // getting a new connection
+ if (!connect(hiveconf)) {
+ LOG.error("Error during clean-up. " + e);
+ return false;
+ }
+ } catch (SQLException e) {
+ // for SQLTransientException (already handled by Utilities.*WithRetries() functions
+ // and SQLNonTransientException, just declare failure.
+ LOG.error("Error during clean-up. " + e);
+ return false;
+ }
+ }
} catch (SQLException e) {
LOG.error("Error during publishing aggregation. " + e);
return false;
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsPublisher.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsPublisher.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsPublisher.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsPublisher.java Fri Jun 3 16:45:37 2011
@@ -27,6 +27,8 @@ import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.sql.SQLRecoverableException;
import java.sql.Statement;
+import java.util.List;
+import java.util.Map;
import java.util.Random;
import org.apache.commons.logging.Log;
@@ -35,7 +37,6 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.stats.StatsPublisher;
-import org.apache.hadoop.hive.ql.stats.StatsSetupConst;
public class JDBCStatsPublisher implements StatsPublisher {
@@ -70,23 +71,6 @@ public class JDBCStatsPublisher implemen
return false;
}
- // prepare the SELECT/UPDATE/INSERT statements
-
- String update =
- "UPDATE /* " + comment + " */ " + JDBCStatsSetupConstants.PART_STAT_TABLE_NAME +
- " SET " + JDBCStatsSetupConstants.PART_STAT_ROW_COUNT_COLUMN_NAME + "= ? " +
- " WHERE " + JDBCStatsSetupConstants.PART_STAT_ID_COLUMN_NAME + " = ? " +
- " AND ? > ( SELECT TEMP." + JDBCStatsSetupConstants.PART_STAT_ROW_COUNT_COLUMN_NAME
- + " FROM ( " +
- " SELECT " + JDBCStatsSetupConstants.PART_STAT_ROW_COUNT_COLUMN_NAME + " FROM "
- + JDBCStatsSetupConstants.PART_STAT_TABLE_NAME +
- " WHERE " + JDBCStatsSetupConstants.PART_STAT_ID_COLUMN_NAME
- + " = ?) TEMP )";
-
- String insert =
- "INSERT INTO /* " + comment + " */ " + JDBCStatsSetupConstants.PART_STAT_TABLE_NAME +
- " VALUES (?, ?)";
-
DriverManager.setLoginTimeout(timeout); // stats is non-blocking
// function pointer for executeWithRetry to setQueryTimeout
@@ -103,15 +87,17 @@ public class JDBCStatsPublisher implemen
conn = Utilities.connectWithRetry(connectionString, waitWindow, maxRetries);
// prepare statements
- updStmt = Utilities.prepareWithRetry(conn, update, waitWindow, maxRetries);
- insStmt = Utilities.prepareWithRetry(conn, insert, waitWindow, maxRetries);
+ updStmt = Utilities.prepareWithRetry(conn, JDBCStatsUtils.getUpdate(comment), waitWindow,
+ maxRetries);
+ insStmt = Utilities.prepareWithRetry(conn, JDBCStatsUtils.getInsert(comment), waitWindow,
+ maxRetries);
// set query timeout
Utilities.executeWithRetry(setQueryTimeout, updStmt, waitWindow, maxRetries);
Utilities.executeWithRetry(setQueryTimeout, insStmt, waitWindow, maxRetries);
- return true;
+ return true;
} catch (SQLRecoverableException e) {
if (failures >= maxRetries) {
LOG.error("Error during JDBC connection to " + connectionString + ". ", e);
@@ -131,19 +117,18 @@ public class JDBCStatsPublisher implemen
}
@Override
- public boolean publishStat(String fileID, String statType, String value) {
+ public boolean publishStat(String fileID, Map<String, String> stats) {
if (conn == null) {
LOG.error("JDBC connection is null. Cannot publish stats without JDBC connection.");
return false;
}
- if (statType != StatsSetupConst.ROW_COUNT) {
- LOG.warn("Warning. Invalid statistic. Currently " +
- "row count is the only supported statistic");
+ if (!JDBCStatsUtils.isValidStatisticSet(stats.keySet())) {
+ LOG.warn("Warning. Invalid statistic.");
return false;
}
- LOG.info("Stats publishing for key " + fileID + ". Value = " + value);
+ LOG.info("Stats publishing for key " + fileID);
Utilities.SQLCommand<Void> execUpdate = new Utilities.SQLCommand<Void>() {
@Override
@@ -153,10 +138,14 @@ public class JDBCStatsPublisher implemen
}
};
+ List<String> supportedStatistics = JDBCStatsUtils.getSupportedStatistics();
+
for (int failures = 0;; failures++) {
try {
insStmt.setString(1, fileID);
- insStmt.setString(2, value);
+ for (int i = 0; i < JDBCStatsUtils.getSupportedStatistics().size(); i++) {
+ insStmt.setString(i + 2, stats.get(supportedStatistics.get(i)));
+ }
Utilities.executeWithRetry(execUpdate, insStmt, waitWindow, maxRetries);
return true;
} catch (SQLIntegrityConstraintViolationException e) {
@@ -168,10 +157,14 @@ public class JDBCStatsPublisher implemen
for (int updateFailures = 0;; updateFailures++) {
try {
- updStmt.setString(1, value);
- updStmt.setString(2, fileID);
- updStmt.setString(3, value);
- updStmt.setString(4, fileID);
+ int i;
+ for (i = 0; i < JDBCStatsUtils.getSupportedStatistics().size(); i++) {
+ updStmt.setString(i + 1, stats.get(supportedStatistics.get(i)));
+ }
+ updStmt.setString(supportedStatistics.size() + 1, fileID);
+ updStmt.setString(supportedStatistics.size() + 2,
+ stats.get(JDBCStatsUtils.getBasicStat()));
+ updStmt.setString(supportedStatistics.size() + 3, fileID);
Utilities.executeWithRetry(execUpdate, updStmt, waitWindow, maxRetries);
return true;
} catch (SQLRecoverableException ue) {
@@ -184,6 +177,7 @@ public class JDBCStatsPublisher implemen
return false;
}
}
+
} catch (SQLRecoverableException e) {
// need to start from scratch (connection)
if (!handleSQLRecoverableException(e, failures)) {
@@ -222,12 +216,12 @@ public class JDBCStatsPublisher implemen
return true;
}
try {
- if (insStmt != null) {
- insStmt.close();
- }
if (updStmt != null) {
updStmt.close();
}
+ if (insStmt != null) {
+ insStmt.close();
+ }
conn.close();
@@ -269,18 +263,13 @@ public class JDBCStatsPublisher implemen
// Check if the table exists
DatabaseMetaData dbm = conn.getMetaData();
- ResultSet rs = dbm.getTables(null, null, JDBCStatsSetupConstants.PART_STAT_TABLE_NAME, null);
+ ResultSet rs = dbm.getTables(null, null, JDBCStatsUtils.getStatTableName(), null);
boolean tblExists = rs.next();
if (!tblExists) { // Table does not exist, create it
- String createTable =
- "CREATE TABLE " + JDBCStatsSetupConstants.PART_STAT_TABLE_NAME + " (" +
- JDBCStatsSetupConstants.PART_STAT_ID_COLUMN_NAME + " VARCHAR(255) PRIMARY KEY, " +
- JDBCStatsSetupConstants.PART_STAT_ROW_COUNT_COLUMN_NAME + " BIGINT )";
-
+ String createTable = JDBCStatsUtils.getCreate("");
stmt.executeUpdate(createTable);
stmt.close();
}
-
closeConnection();
} catch (Exception e) {
LOG.error("Error during JDBC initialization. ", e);
@@ -288,4 +277,5 @@ public class JDBCStatsPublisher implemen
}
return true;
}
+
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsSetupConstants.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsSetupConstants.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsSetupConstants.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsSetupConstants.java Fri Jun 3 16:45:37 2011
@@ -19,10 +19,14 @@ package org.apache.hadoop.hive.ql.stats.
public final class JDBCStatsSetupConstants {
- public static final String PART_STAT_TABLE_NAME = "PARTITION_STAT_TBL";
+ public static final String PART_STAT_ID_COLUMN_NAME = "ID";
- public static final String PART_STAT_ID_COLUMN_NAME = "ID";
+ public static final String PART_STAT_TABLE_NAME = "PARTITION_STAT_TBL";
- public static final String PART_STAT_ROW_COUNT_COLUMN_NAME = "ROW_COUNT";
+ // supported statistics - column names
+
+ public static final String PART_STAT_ROW_COUNT_COLUMN_NAME = "ROW_COUNT";
+
+ public static final String PART_STAT_RAW_DATA_SIZE_COLUMN_NAME = "RAW_DATA_SIZE";
}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsUtils.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsUtils.java?rev=1131106&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsUtils.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/stats/jdbc/JDBCStatsUtils.java Fri Jun 3 16:45:37 2011
@@ -0,0 +1,169 @@
+package org.apache.hadoop.hive.ql.stats.jdbc;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.ql.stats.StatsSetupConst;
+
+public class JDBCStatsUtils {
+
+ private static final List<String> supportedStats = new ArrayList<String>();
+ private static final Map<String, String> columnNameMapping = new HashMap<String, String>();
+ static {
+ // supported statistics
+ supportedStats.add(StatsSetupConst.ROW_COUNT);
+ supportedStats.add(StatsSetupConst.RAW_DATA_SIZE);
+
+ // row count statistics
+ columnNameMapping.put(StatsSetupConst.ROW_COUNT,
+ JDBCStatsSetupConstants.PART_STAT_ROW_COUNT_COLUMN_NAME);
+
+ // raw data size
+ columnNameMapping.put(StatsSetupConst.RAW_DATA_SIZE,
+ JDBCStatsSetupConstants.PART_STAT_RAW_DATA_SIZE_COLUMN_NAME);
+ }
+
+ /**
+ * Returns the set of supported statistics
+ */
+ public static List<String> getSupportedStatistics() {
+ return supportedStats;
+ }
+
+ /**
+ * Check if the set to be published is within the supported statistics.
+ * It must also contain at least the basic statistics (used for comparison)
+ *
+ * @param stats
+ * - stats to be published
+ * @return true if is a valid statistic set, false otherwise
+ */
+
+ public static boolean isValidStatisticSet(Collection<String> stats) {
+ if (!stats.contains(getBasicStat())) {
+ return false;
+ }
+ for (String stat : stats) {
+ if (!supportedStats.contains(stat)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Check if a particular statistic type is supported
+ *
+ * @param statType
+ * - statistic to be published
+ * @return true if statType is supported, false otherwise
+ */
+ public static boolean isValidStatistic(String statType) {
+ return supportedStats.contains(statType);
+ }
+
+ /**
+ * Returns the name of the column storing the key for statistics.
+ */
+ public static String getIdColumnName() {
+ return JDBCStatsSetupConstants.PART_STAT_ID_COLUMN_NAME;
+ }
+
+ public static String getStatTableName() {
+ return JDBCStatsSetupConstants.PART_STAT_TABLE_NAME;
+ }
+
+ /**
+ * Returns the column where the statistics for the given type are stored.
+ *
+ * @param statType
+ * - supported statistic.
+ * @return column name for the given statistic.
+ */
+ public static String getStatColumnName(String statType) {
+ return columnNameMapping.get(statType);
+ }
+
+ /**
+ * Returns the basic type of the supported statistics.
+ * It is used to determine which statistics are fresher.
+ */
+ public static String getBasicStat() {
+ return supportedStats.get(0);
+ }
+
+
+
+
+
+ /**
+ * Prepares CREATE TABLE query
+ */
+ public static String getCreate(String comment) {
+ String create = "CREATE TABLE /* " + comment + " */ " + JDBCStatsUtils.getStatTableName() +
+ " (" + JDBCStatsUtils.getIdColumnName() + " VARCHAR(255) PRIMARY KEY ";
+ for (int i = 0; i < supportedStats.size(); i++) {
+ create += ", " + getStatColumnName(supportedStats.get(i)) + " BIGINT ";
+ }
+ create += ")";
+ return create;
+ }
+
+ /**
+ * Prepares UPDATE statement issued when updating existing statistics
+ */
+ public static String getUpdate(String comment) {
+ String update = "UPDATE /* " + comment + " */ " + getStatTableName() + " SET ";
+ for (int i = 0; i < supportedStats.size(); i++) {
+ update += columnNameMapping.get(supportedStats.get(i)) + " = ? , ";
+ }
+ update = update.substring(0, update.length() - 2);
+ update += " WHERE " + JDBCStatsUtils.getIdColumnName() + " = ? AND ? > ( SELECT TEMP."
+ + getStatColumnName(getBasicStat()) + " FROM ( " +
+ " SELECT " + getStatColumnName(getBasicStat()) + " FROM " + getStatTableName() + " WHERE "
+ + getIdColumnName() + " = ? ) TEMP )";
+ return update;
+ }
+
+ /**
+ * Prepares INSERT statement for statistic publishing.
+ */
+ public static String getInsert(String comment) {
+ String insert = "INSERT INTO /* " + comment + " */ " + getStatTableName() + " VALUES (?, ";
+ for (int i = 0; i < supportedStats.size(); i++) {
+ insert += "? , ";
+ }
+ insert = insert.substring(0, insert.length() - 3);
+ insert += ")";
+ return insert;
+ }
+
+ /**
+ * Prepares SELECT query for statistics aggregation.
+ *
+ * @param statType
+ * - statistic type to be aggregated.
+ * @param comment
+ * @return aggregated value for the given statistic
+ */
+ public static String getSelectAggr(String statType, String comment) {
+ String select = "SELECT /* " + comment + " */ " + "SUM( "
+ + getStatColumnName(statType) + " ) " + " FROM "
+ + getStatTableName() + " WHERE " + JDBCStatsUtils.getIdColumnName() + " LIKE ? ESCAPE ?";
+ return select;
+ }
+
+ /**
+ * Prepares DELETE statement for cleanup.
+ */
+ public static String getDeleteAggr(String rowID, String comment) {
+ String delete = "DELETE /* " + comment + " */ " +
+ " FROM " + getStatTableName() + " WHERE " + JDBCStatsUtils.getIdColumnName() + " LIKE '"
+ + rowID + "%'";
+ return delete;
+ }
+
+}
Modified: hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisher.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisher.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisher.java (original)
+++ hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisher.java Fri Jun 3 16:45:37 2011
@@ -1,260 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.hadoop.hive.ql.exec;
-
-import junit.framework.TestCase;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.ql.stats.StatsAggregator;
-import org.apache.hadoop.hive.ql.stats.StatsFactory;
-import org.apache.hadoop.hive.ql.stats.StatsPublisher;
-import org.apache.hadoop.hive.ql.stats.StatsSetupConst;
-import org.apache.hadoop.mapred.JobConf;
-
-/**
- * TestOperators.
- *
- */
-public class TestStatsPublisher extends TestCase {
-
- protected Configuration conf;
- protected String statsImplementationClass;
-
- public TestStatsPublisher(String name) {
- super(name);
- }
-
- @Override
- protected void setUp() {
- System.out.println("StatPublisher Test");
- conf = new JobConf(TestStatsPublisher.class);
-
- statsImplementationClass = HiveConf.getVar(conf, HiveConf.ConfVars.HIVESTATSDBCLASS);
- StatsFactory.setImplementation(statsImplementationClass, conf);
- }
-
- public void testStatsPublisherOneStat() throws Throwable {
- try {
- System.out.println("StatsPublisher - one stat published per key - aggregating matching key");
-
- // instantiate stats publisher
- StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
- assertNotNull(statsPublisher);
- assertTrue(statsPublisher.init(conf));
- assertTrue(statsPublisher.connect(conf));
-
- // instantiate stats aggregator
- StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
- assertNotNull(statsAggregator);
- assertTrue(statsAggregator.connect(conf));
-
- // publish stats
- assertTrue(statsPublisher.publishStat("file_00000", StatsSetupConst.ROW_COUNT, "200"));
- assertTrue(statsPublisher.publishStat("file_00001", StatsSetupConst.ROW_COUNT, "400"));
-
- // aggregate existing stats for prefixes
- String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
- assertEquals("200", rows0);
- String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
- assertEquals("400", rows1);
-
- // close connections
- assertTrue(statsPublisher.closeConnection());
- assertTrue(statsAggregator.closeConnection());
-
- System.out.println("StatsPublisher - one stat published per key - aggregating matching key - OK");
- } catch (Throwable e) {
- e.printStackTrace();
- throw e;
- }
- }
-
-
- public void testStatsPublisher() throws Throwable {
- try {
- System.out.println("StatsPublisher - basic functionality");
-
- // instantiate stats publisher
- StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
- assertNotNull(statsPublisher);
- assertTrue(statsPublisher.init(conf));
- assertTrue(statsPublisher.connect(conf));
-
- // instantiate stats aggregator
- StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
- assertNotNull(statsAggregator);
- assertTrue(statsAggregator.connect(conf));
-
- // publish stats
- assertTrue(statsPublisher.publishStat("file_00000_a", StatsSetupConst.ROW_COUNT, "200"));
- assertTrue(statsPublisher.publishStat("file_00000_b", StatsSetupConst.ROW_COUNT, "300"));
- assertTrue(statsPublisher.publishStat("file_00001_a", StatsSetupConst.ROW_COUNT, "400"));
- assertTrue(statsPublisher.publishStat("file_00001_b", StatsSetupConst.ROW_COUNT, "500"));
-
- // aggregate existing stats for prefixes
- String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
- assertEquals("500", rows0);
- String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
- assertEquals("900", rows1);
-
- // aggregate non-existent stats
- String rowsX = statsAggregator.aggregateStats("file_00002", StatsSetupConst.ROW_COUNT);
- assertEquals("0", rowsX);
-
- // close connections
- assertTrue(statsPublisher.closeConnection());
- assertTrue(statsAggregator.closeConnection());
-
- System.out.println("StatsPublisher - basic functionality - OK");
- } catch (Throwable e) {
- e.printStackTrace();
- throw e;
- }
- }
-
- public void testStatsPublisherMultipleUpdates() throws Throwable {
- try {
- System.out.println("StatsPublisher - multiple updates");
-
- // instantiate stats publisher
- StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
- assertNotNull(statsPublisher);
- assertTrue(statsPublisher.init(conf));
- assertTrue(statsPublisher.connect(conf));
-
- // instantiate stats aggregator
- StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
- assertNotNull(statsAggregator);
- assertTrue(statsAggregator.connect(conf));
-
- // publish stats
- assertTrue(statsPublisher.publishStat("file_00000_a", StatsSetupConst.ROW_COUNT, "200"));
- assertTrue(statsPublisher.publishStat("file_00000_b", StatsSetupConst.ROW_COUNT, "300"));
- assertTrue(statsPublisher.publishStat("file_00001_a", StatsSetupConst.ROW_COUNT, "400"));
- assertTrue(statsPublisher.publishStat("file_00001_b", StatsSetupConst.ROW_COUNT, "500"));
-
- // repetitive update - should not change the stored value - as the published values are
- // smaller than the current ones
- assertTrue(statsPublisher.publishStat("file_00000_a", StatsSetupConst.ROW_COUNT, "100"));
- assertTrue(statsPublisher.publishStat("file_00000_b", StatsSetupConst.ROW_COUNT, "150"));
-
- // should change the stored value - the published values are greater than the current values
- assertTrue(statsPublisher.publishStat("file_00001_a", StatsSetupConst.ROW_COUNT, "500"));
- assertTrue(statsPublisher.publishStat("file_00001_b", StatsSetupConst.ROW_COUNT, "600"));
-
- // aggregate stats
- String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
- assertEquals("500", rows0);
- String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
- assertEquals("1100", rows1);
-
- // close connections
- assertTrue(statsPublisher.closeConnection());
- assertTrue(statsAggregator.closeConnection());
-
- System.out.println("StatsPublisher - multiple updates - OK");
- } catch (Throwable e) {
- e.printStackTrace();
- throw e;
- }
- }
-
- public void testStatsPublisherAfterAggregationCleanUp() throws Throwable {
- try {
- System.out.println("StatsPublisher - clean-up after aggregation");
-
- // instantiate stats publisher
- StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
- assertNotNull(statsPublisher);
- assertTrue(statsPublisher.init(conf));
- assertTrue(statsPublisher.connect(conf));
-
- // instantiate stats aggregator
- StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
- assertNotNull(statsAggregator);
- assertTrue(statsAggregator.connect(conf));
-
- // publish stats
- assertTrue(statsPublisher.publishStat("file_00000_a", StatsSetupConst.ROW_COUNT, "200"));
- assertTrue(statsPublisher.publishStat("file_00000_b", StatsSetupConst.ROW_COUNT, "300"));
- assertTrue(statsPublisher.publishStat("file_00001_a", StatsSetupConst.ROW_COUNT, "400"));
- assertTrue(statsPublisher.publishStat("file_00001_b", StatsSetupConst.ROW_COUNT, "500"));
-
- // aggregate stats
- String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
- assertEquals("500", rows0);
- String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
- assertEquals("900", rows1);
-
- // now the table should be empty
- rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
- assertEquals("0", rows0);
- rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
- assertEquals("0", rows1);
-
- // close connections
- assertTrue(statsPublisher.closeConnection());
- assertTrue(statsAggregator.closeConnection());
-
- System.out.println("StatsPublisher - clean-up after aggregation - OK");
- } catch (Throwable e) {
- e.printStackTrace();
- throw e;
- }
- }
-
- public void testStatsAggregatorCleanUp() throws Throwable {
- try {
- System.out.println("StatsAggregator - clean-up");
-
- // instantiate stats publisher
- StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
- assertNotNull(statsPublisher);
- assertTrue(statsPublisher.init(conf));
- assertTrue(statsPublisher.connect(conf));
-
- // instantiate stats aggregator
- StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
- assertNotNull(statsAggregator);
- assertTrue(statsAggregator.connect(conf));
-
- // publish stats
- assertTrue(statsPublisher.publishStat("file_00000_a", StatsSetupConst.ROW_COUNT, "200"));
- assertTrue(statsPublisher.publishStat("file_00000_b", StatsSetupConst.ROW_COUNT, "300"));
-
- // cleanUp (closes the connection)
- assertTrue(statsAggregator.cleanUp("file_00000"));
-
- // now the connection should be closed (aggregator will report an error)
- String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
- assertNull(rows0);
-
- // close connections
- assertTrue(statsPublisher.closeConnection());
- assertTrue(statsAggregator.closeConnection());
-
- System.out.println("StatsAggregator - clean-up - OK");
- } catch (Throwable e) {
- e.printStackTrace();
- throw e;
- }
- }
-
-}
Added: hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisherEnhanced.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisherEnhanced.java?rev=1131106&view=auto
==============================================================================
--- hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisherEnhanced.java (added)
+++ hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/exec/TestStatsPublisherEnhanced.java Fri Jun 3 16:45:37 2011
@@ -0,0 +1,374 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.exec;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.stats.StatsAggregator;
+import org.apache.hadoop.hive.ql.stats.StatsFactory;
+import org.apache.hadoop.hive.ql.stats.StatsPublisher;
+import org.apache.hadoop.hive.ql.stats.StatsSetupConst;
+import org.apache.hadoop.mapred.JobConf;
+
+/**
+ * TestPublisher jdbc.
+ *
+ */
+public class TestStatsPublisherEnhanced extends TestCase {
+
+ protected Configuration conf;
+ protected String statsImplementationClass;
+ protected Map<String, String> stats;
+
+ public TestStatsPublisherEnhanced(String name) {
+ super(name);
+ }
+
+ @Override
+ protected void setUp() {
+ conf = new JobConf(TestStatsPublisherEnhanced.class);
+
+ conf.set("hive.stats.dbclass", "jdbc:derby");
+
+ statsImplementationClass = HiveConf.getVar(conf, HiveConf.ConfVars.HIVESTATSDBCLASS);
+ StatsFactory.setImplementation(statsImplementationClass, conf);
+
+ stats = new HashMap<String, String>();
+ StatsAggregator sa = StatsFactory.getStatsAggregator();
+ sa.connect(conf);
+ sa.cleanUp("file_0");
+ sa.closeConnection();
+ }
+
+
+ private void fillStatMap(String numRows, String rawDataSize) {
+ stats.clear();
+ stats.put(StatsSetupConst.ROW_COUNT, numRows);
+ if (!rawDataSize.equals("")) {
+ stats.put(StatsSetupConst.RAW_DATA_SIZE, rawDataSize);
+ }
+ }
+
+ public void testStatsPublisherOneStat() throws Throwable {
+ try {
+ System.out.println("StatsPublisher - one stat published per key - aggregating matching key");
+
+ // instantiate stats publisher
+ StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
+ assertNotNull(statsPublisher);
+ assertTrue(statsPublisher.init(conf));
+ assertTrue(statsPublisher.connect(conf));
+
+ // instantiate stats aggregator
+ StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
+ assertNotNull(statsAggregator);
+ assertTrue(statsAggregator.connect(conf));
+
+ // publish stats
+ fillStatMap("200", "1000");
+ assertTrue(statsPublisher.publishStat("file_00000", stats));
+ fillStatMap("400", "3000");
+ assertTrue(statsPublisher.publishStat("file_00001", stats));
+
+
+ // aggregate existing stats
+ String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
+ assertEquals("200", rows0);
+ String usize0 = statsAggregator.aggregateStats("file_00000",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("1000", usize0);
+
+ String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
+ assertEquals("400", rows1);
+ String usize1 = statsAggregator.aggregateStats("file_00001",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("3000", usize1);
+
+ // close connections
+ assertTrue(statsPublisher.closeConnection());
+ assertTrue(statsAggregator.closeConnection());
+
+ System.out
+ .println("StatsPublisher - one stat published per key - aggregating matching key - OK");
+ } catch (Throwable e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ public void testStatsPublisher() throws Throwable {
+ try {
+ System.out.println("StatsPublisher - basic functionality");
+
+ // instantiate stats publisher
+ StatsPublisher statsPublisher = Utilities.getStatsPublisher(
+ (JobConf) conf);
+ assertNotNull(statsPublisher);
+ assertTrue(statsPublisher.init(conf));
+ assertTrue(statsPublisher.connect(conf));
+
+ // instantiate stats aggregator
+ StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
+ assertNotNull(statsAggregator);
+ assertTrue(statsAggregator.connect(conf));
+ // statsAggregator.cleanUp("file_0000");
+ // assertTrue(statsAggregator.connect(conf));
+
+ // publish stats
+ fillStatMap("200", "1000");
+ assertTrue(statsPublisher.publishStat("file_00000_a", stats));
+ fillStatMap("300", "2000");
+ assertTrue(statsPublisher.publishStat("file_00000_b", stats));
+
+ fillStatMap("400", "3000");
+ assertTrue(statsPublisher.publishStat("file_00001_a", stats));
+ fillStatMap("500", "4000");
+ assertTrue(statsPublisher.publishStat("file_00001_b", stats));
+
+ // aggregate existing stats
+ String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
+ assertEquals("500", rows0);
+ String usize0 = statsAggregator.aggregateStats("file_00000",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("3000", usize0);
+
+ String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
+ assertEquals("900", rows1);
+ String usize1 = statsAggregator.aggregateStats("file_00001",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("7000", usize1);
+
+ // aggregate non-existent stats
+ String rowsX = statsAggregator.aggregateStats("file_00002", StatsSetupConst.ROW_COUNT);
+ assertEquals("0", rowsX);
+ String usizeX = statsAggregator.aggregateStats("file_00002",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("0", usizeX);
+
+ assertTrue(statsAggregator.cleanUp("file_0000"));
+
+ // close connections
+ assertTrue(statsPublisher.closeConnection());
+ assertTrue(statsAggregator.closeConnection());
+
+ System.out.println("StatsPublisher - basic functionality - OK");
+ } catch (Throwable e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ public void testStatsPublisherMultipleUpdates() throws Throwable {
+ try {
+ System.out.println("StatsPublisher - multiple updates");
+
+ // instantiate stats publisher
+ StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
+ assertNotNull(statsPublisher);
+ assertTrue(statsPublisher.init(conf));
+ assertTrue(statsPublisher.connect(conf));
+
+ // instantiate stats aggregator
+ StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
+ assertNotNull(statsAggregator);
+ assertTrue(statsAggregator.connect(conf));
+
+ // publish stats
+ fillStatMap("200", "1000");
+ assertTrue(statsPublisher.publishStat("file_00000_a", stats));
+ fillStatMap("300", "2000");
+ assertTrue(statsPublisher.publishStat("file_00000_b", stats));
+
+ fillStatMap("400", "3000");
+ assertTrue(statsPublisher.publishStat("file_00001_a", stats));
+ fillStatMap("500", "4000");
+ assertTrue(statsPublisher.publishStat("file_00001_b", stats));
+
+ // update which should not take any effect
+ fillStatMap("190", "1000");
+ assertTrue(statsPublisher.publishStat("file_00000_a", stats));
+ fillStatMap("290", "2000");
+ assertTrue(statsPublisher.publishStat("file_00000_b", stats));
+
+ // update that should take effect
+ fillStatMap("500", "5000");
+ assertTrue(statsPublisher.publishStat("file_00001_a", stats));
+ fillStatMap("600", "6000");
+ assertTrue(statsPublisher.publishStat("file_00001_b", stats));
+
+ // aggregate existing stats
+ String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
+ assertEquals("500", rows0);
+ String usize0 = statsAggregator.aggregateStats("file_00000",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("3000", usize0);
+
+ String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
+ assertEquals("1100", rows1);
+ String usize1 = statsAggregator.aggregateStats("file_00001",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("11000", usize1);
+
+ assertTrue(statsAggregator.cleanUp("file_0000"));
+
+ // close connections
+ assertTrue(statsPublisher.closeConnection());
+ assertTrue(statsAggregator.closeConnection());
+
+ System.out.println("StatsPublisher - multiple updates - OK");
+ } catch (Throwable e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ public void testStatsPublisherMultipleUpdatesSubsetStatistics() throws Throwable {
+ try {
+ System.out
+ .println("StatsPublisher - (multiple updates + publishing subset of supported statistics)");
+
+ // instantiate stats publisher
+ StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
+ assertNotNull(statsPublisher);
+ assertTrue(statsPublisher.init(conf));
+ assertTrue(statsPublisher.connect(conf));
+
+ // instantiate stats aggregator
+ StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
+ assertNotNull(statsAggregator);
+ assertTrue(statsAggregator.connect(conf));
+
+ // publish stats
+ fillStatMap("200", "");
+ assertTrue(statsPublisher.publishStat("file_00000_a", stats));
+ fillStatMap("300", "2000");
+ assertTrue(statsPublisher.publishStat("file_00000_b", stats));
+
+
+ // aggregate existing stats
+ String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
+ assertEquals("500", rows0);
+ String usize0 = statsAggregator.aggregateStats("file_00000",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("2000", usize0);
+
+ // update which should not take any effect - plus the map published is a supset of supported
+ // stats
+ fillStatMap("190", "");
+ assertTrue(statsPublisher.publishStat("file_00000_a", stats));
+ fillStatMap("290", "");
+ assertTrue(statsPublisher.publishStat("file_00000_b", stats));
+
+ // nothing changed
+ rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
+ assertEquals("500", rows0);
+ usize0 = statsAggregator.aggregateStats("file_00000",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("2000", usize0);
+
+ fillStatMap("500", "");
+ assertTrue(statsPublisher.publishStat("file_00000_a", stats));
+ fillStatMap("500", "");
+ assertTrue(statsPublisher.publishStat("file_00000_b", stats));
+
+ // changed + the rawDataSize size was overwriten !!!
+ rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
+ assertEquals("1000", rows0);
+ usize0 = statsAggregator.aggregateStats("file_00000",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("0", usize0);
+
+ assertTrue(statsAggregator.cleanUp("file_0000"));
+
+ // close connections
+ assertTrue(statsPublisher.closeConnection());
+ assertTrue(statsAggregator.closeConnection());
+
+ System.out
+ .println("StatsPublisher - (multiple updates + publishing subset of supported statistics) - OK");
+ } catch (Throwable e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+
+
+ public void testStatsAggregatorCleanUp() throws Throwable {
+ try {
+ System.out.println("StatsAggregator - clean-up");
+
+ // instantiate stats publisher
+ StatsPublisher statsPublisher = Utilities.getStatsPublisher((JobConf) conf);
+ assertNotNull(statsPublisher);
+ assertTrue(statsPublisher.init(conf));
+ assertTrue(statsPublisher.connect(conf));
+
+ // instantiate stats aggregator
+ StatsAggregator statsAggregator = StatsFactory.getStatsAggregator();
+ assertNotNull(statsAggregator);
+ assertTrue(statsAggregator.connect(conf));
+
+ // publish stats
+ fillStatMap("200", "1000");
+ assertTrue(statsPublisher.publishStat("file_00000_a", stats));
+ fillStatMap("300", "2000");
+ assertTrue(statsPublisher.publishStat("file_00000_b", stats));
+
+ fillStatMap("400", "3000");
+ assertTrue(statsPublisher.publishStat("file_00001_a", stats));
+ fillStatMap("500", "4000");
+ assertTrue(statsPublisher.publishStat("file_00001_b", stats));
+
+ // cleanUp
+ assertTrue(statsAggregator.cleanUp("file_00000"));
+
+ // now clean-up just for one key
+ String rows0 = statsAggregator.aggregateStats("file_00000", StatsSetupConst.ROW_COUNT);
+ assertEquals("0", rows0);
+ String usize0 = statsAggregator.aggregateStats("file_00000",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("0", usize0);
+
+ // this should still be in the table
+ String rows1 = statsAggregator.aggregateStats("file_00001", StatsSetupConst.ROW_COUNT);
+ assertEquals("900", rows1);
+ String usize1 = statsAggregator.aggregateStats("file_00001",
+ StatsSetupConst.RAW_DATA_SIZE);
+ assertEquals("7000", usize1);
+
+ assertTrue(statsAggregator.cleanUp("file_0000"));
+
+ // close connections
+ assertTrue(statsPublisher.closeConnection());
+ assertTrue(statsAggregator.closeConnection());
+
+ System.out.println("StatsAggregator - clean-up - OK");
+ } catch (Throwable e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+}
Modified: hive/trunk/ql/src/test/org/apache/hadoop/hive/serde2/TestSerDe.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/org/apache/hadoop/hive/serde2/TestSerDe.java?rev=1131106&r1=1131105&r2=1131106&view=diff
==============================================================================
--- hive/trunk/ql/src/test/org/apache/hadoop/hive/serde2/TestSerDe.java (original)
+++ hive/trunk/ql/src/test/org/apache/hadoop/hive/serde2/TestSerDe.java Fri Jun 3 16:45:37 2011
@@ -29,10 +29,10 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.serde2.objectinspector.MetadataListStructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
@@ -198,4 +198,9 @@ public class TestSerDe implements SerDe
return serializeCache;
}
+ public SerDeStats getSerDeStats() {
+ // no support for statistics
+ return null;
+ }
+
}
Added: hive/trunk/ql/src/test/queries/clientpositive/stats14.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/stats14.q?rev=1131106&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/stats14.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/stats14.q Fri Jun 3 16:45:37 2011
@@ -0,0 +1,26 @@
+set datanucleus.cache.collections=false;
+
+create table stats_src like src;
+insert overwrite table stats_src select * from src;
+analyze table stats_src compute statistics;
+desc formatted stats_src;
+
+create table stats_part like srcpart;
+
+insert overwrite table stats_part partition (ds='2010-04-08', hr = '11') select key, value from src;
+insert overwrite table stats_part partition (ds='2010-04-08', hr = '12') select key, value from src;
+
+analyze table stats_part partition(ds='2010-04-08', hr='11') compute statistics;
+analyze table stats_part partition(ds='2010-04-08', hr='12') compute statistics;
+
+insert overwrite table stats_part partition (ds='2010-04-08', hr = '13') select key, value from src;
+
+desc formatted stats_part;
+desc formatted stats_part partition (ds='2010-04-08', hr = '11');
+desc formatted stats_part partition (ds='2010-04-08', hr = '12');
+
+analyze table stats_part partition(ds, hr) compute statistics;
+desc formatted stats_part;
+
+drop table stats_src;
+drop table stats_part;
\ No newline at end of file
Added: hive/trunk/ql/src/test/queries/clientpositive/stats15.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/stats15.q?rev=1131106&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/stats15.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/stats15.q Fri Jun 3 16:45:37 2011
@@ -0,0 +1,27 @@
+set datanucleus.cache.collections=false;
+set hive.stats.collect.uncompressedsize=false;
+
+create table stats_src like src;
+insert overwrite table stats_src select * from src;
+analyze table stats_src compute statistics;
+desc formatted stats_src;
+
+create table stats_part like srcpart;
+
+insert overwrite table stats_part partition (ds='2010-04-08', hr = '11') select key, value from src;
+insert overwrite table stats_part partition (ds='2010-04-08', hr = '12') select key, value from src;
+
+analyze table stats_part partition(ds='2010-04-08', hr='11') compute statistics;
+analyze table stats_part partition(ds='2010-04-08', hr='12') compute statistics;
+
+insert overwrite table stats_part partition (ds='2010-04-08', hr = '13') select key, value from src;
+
+desc formatted stats_part;
+desc formatted stats_part partition (ds='2010-04-08', hr = '11');
+desc formatted stats_part partition (ds='2010-04-08', hr = '12');
+
+analyze table stats_part partition(ds, hr) compute statistics;
+desc formatted stats_part;
+
+drop table stats_src;
+drop table stats_part;
\ No newline at end of file