You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2015/11/12 20:31:34 UTC
[1/5] hive git commit: HIVE-11587 : Fix memory estimates for mapjoin
hashtable (Wei Zheng, reviewed by Sergey Shelukhin)
Repository: hive
Updated Branches:
refs/heads/branch-1 529929a49 -> 48d0ec9ee
HIVE-11587 : Fix memory estimates for mapjoin hashtable (Wei Zheng, reviewed by Sergey Shelukhin)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/fd5d1916
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/fd5d1916
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/fd5d1916
Branch: refs/heads/branch-1
Commit: fd5d191657b7f779ecaaee3fa9157eab8d0c2876
Parents: 529929a
Author: Sergey Shelukhin <se...@apache.org>
Authored: Thu Sep 10 11:57:36 2015 -0700
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Thu Nov 12 11:24:27 2015 -0800
----------------------------------------------------------------------
.../org/apache/hadoop/hive/conf/HiveConf.java | 2 +
.../hadoop/hive/ql/exec/MapJoinOperator.java | 5 ++
.../persistence/BytesBytesMultiHashMap.java | 11 +++-
.../persistence/HybridHashTableContainer.java | 68 ++++++++++++--------
.../hive/ql/exec/tez/HashTableLoader.java | 7 +-
.../apache/hadoop/hive/serde2/WriteBuffers.java | 10 ++-
6 files changed, 68 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/fd5d1916/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index a881812..60d3678 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -770,6 +770,8 @@ public class HiveConf extends Configuration {
HIVEMAPJOINUSEOPTIMIZEDTABLE("hive.mapjoin.optimized.hashtable", true,
"Whether Hive should use memory-optimized hash table for MapJoin. Only works on Tez,\n" +
"because memory-optimized hashtable cannot be serialized."),
+ HIVEMAPJOINOPTIMIZEDTABLEPROBEPERCENT("hive.mapjoin.optimized.hashtable.probe.percent",
+ (float) 0.5, "Probing space percentage of the optimized hashtable"),
HIVEUSEHYBRIDGRACEHASHJOIN("hive.mapjoin.hybridgrace.hashtable", true, "Whether to use hybrid" +
"grace hash join as the join method for mapjoin. Tez only."),
HIVEHYBRIDGRACEHASHJOINMEMCHECKFREQ("hive.mapjoin.hybridgrace.memcheckfrequency", 1024, "For " +
http://git-wip-us.apache.org/repos/asf/hive/blob/fd5d1916/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java
index aa648b0..aa51349 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java
@@ -592,6 +592,11 @@ public class MapJoinOperator extends AbstractMapJoinOperator<MapJoinDesc> implem
// Deserialize the on-disk hash table
// We're sure this part is smaller than memory limit
+ if (rowCount <= 0) {
+ rowCount = 1024 * 1024; // Since rowCount is used later to instantiate a BytesBytesMultiHashMap
+ // as the initialCapacity which cannot be 0, we provide a reasonable
+ // positive number here
+ }
BytesBytesMultiHashMap restoredHashMap = partition.getHashMapFromDisk(rowCount);
rowCount += restoredHashMap.getNumValues();
LOG.info("Hybrid Grace Hash Join: Deserializing spilled hash partition...");
http://git-wip-us.apache.org/repos/asf/hive/blob/fd5d1916/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
index 2ba622e..573a7aa 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/BytesBytesMultiHashMap.java
@@ -153,9 +153,11 @@ public final class BytesBytesMultiHashMap {
/** 8 Gb of refs is the max capacity if memory limit is not specified. If someone has 100s of
* Gbs of memory (this might happen pretty soon) we'd need to string together arrays anyway. */
private final static int DEFAULT_MAX_CAPACITY = 1024 * 1024 * 1024;
+ /** Make sure maxCapacity has a lower limit */
+ private final static int DEFAULT_MIN_MAX_CAPACITY = 16 * 1024 * 1024;
public BytesBytesMultiHashMap(int initialCapacity,
- float loadFactor, int wbSize, long memUsage) {
+ float loadFactor, int wbSize, long maxProbeSize) {
if (loadFactor < 0 || loadFactor > 1) {
throw new AssertionError("Load factor must be between (0, 1].");
}
@@ -163,8 +165,11 @@ public final class BytesBytesMultiHashMap {
initialCapacity = (Long.bitCount(initialCapacity) == 1)
? initialCapacity : nextHighestPowerOfTwo(initialCapacity);
// 8 bytes per long in the refs, assume data will be empty. This is just a sanity check.
- int maxCapacity = (memUsage <= 0) ? DEFAULT_MAX_CAPACITY
- : (int)Math.min((long)DEFAULT_MAX_CAPACITY, memUsage / 8);
+ int maxCapacity = (maxProbeSize <= 0) ? DEFAULT_MAX_CAPACITY
+ : (int)Math.min((long)DEFAULT_MAX_CAPACITY, maxProbeSize / 8);
+ if (maxCapacity < DEFAULT_MIN_MAX_CAPACITY) {
+ maxCapacity = DEFAULT_MIN_MAX_CAPACITY;
+ }
if (maxCapacity < initialCapacity || initialCapacity <= 0) {
// Either initialCapacity is too large, or nextHighestPowerOfTwo overflows
initialCapacity = (Long.bitCount(maxCapacity) == 1)
http://git-wip-us.apache.org/repos/asf/hive/blob/fd5d1916/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
index ff64f52..52c02ae 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/HybridHashTableContainer.java
@@ -76,7 +76,6 @@ public class HybridHashTableContainer
private int totalInMemRowCount = 0; // total number of small table rows in memory
private long memoryThreshold; // the max memory limit that can be allocated
private long memoryUsed; // the actual memory used
- private int writeBufferSize; // write buffer size for this HybridHashTableContainer
private final long tableRowSize; // row size of the small table
private boolean isSpilled; // whether there's any spilled partition
private int toSpillPartitionId; // the partition into which to spill the big table row;
@@ -107,7 +106,7 @@ public class HybridHashTableContainer
Path hashMapLocalPath; // Local file system path for spilled hashMap
boolean hashMapOnDisk; // Status of hashMap. true: on disk, false: in memory
boolean hashMapSpilledOnCreation; // When there's no enough memory, cannot create hashMap
- int threshold; // Used to create an empty BytesBytesMultiHashMap
+ int initialCapacity; // Used to create an empty BytesBytesMultiHashMap
float loadFactor; // Same as above
int wbSize; // Same as above
int rowsOnDisk; // How many rows saved to the on-disk hashmap (if on disk)
@@ -115,17 +114,17 @@ public class HybridHashTableContainer
/* It may happen that there's not enough memory to instantiate a hashmap for the partition.
* In that case, we don't create the hashmap, but pretend the hashmap is directly "spilled".
*/
- public HashPartition(int threshold, float loadFactor, int wbSize, long memUsage,
+ public HashPartition(int initialCapacity, float loadFactor, int wbSize, long maxProbeSize,
boolean createHashMap) {
if (createHashMap) {
- // Hash map should be at least the size of our designated wbSize
- memUsage = Math.max(memUsage, wbSize);
- hashMap = new BytesBytesMultiHashMap(threshold, loadFactor, wbSize, memUsage);
+ // Probe space should be at least equal to the size of our designated wbSize
+ maxProbeSize = Math.max(maxProbeSize, wbSize);
+ hashMap = new BytesBytesMultiHashMap(initialCapacity, loadFactor, wbSize, maxProbeSize);
} else {
hashMapSpilledOnCreation = true;
hashMapOnDisk = true;
}
- this.threshold = threshold;
+ this.initialCapacity = initialCapacity;
this.loadFactor = loadFactor;
this.wbSize = wbSize;
}
@@ -138,18 +137,18 @@ public class HybridHashTableContainer
/* Restore the hashmap from disk by deserializing it.
* Currently Kryo is used for this purpose.
*/
- public BytesBytesMultiHashMap getHashMapFromDisk(int initialCapacity)
+ public BytesBytesMultiHashMap getHashMapFromDisk(int rowCount)
throws IOException, ClassNotFoundException {
if (hashMapSpilledOnCreation) {
- return new BytesBytesMultiHashMap(Math.max(threshold, initialCapacity) , loadFactor, wbSize, -1);
+ return new BytesBytesMultiHashMap(rowCount, loadFactor, wbSize, -1);
} else {
InputStream inputStream = Files.newInputStream(hashMapLocalPath);
com.esotericsoftware.kryo.io.Input input = new com.esotericsoftware.kryo.io.Input(inputStream);
Kryo kryo = Utilities.runtimeSerializationKryo.get();
BytesBytesMultiHashMap restoredHashMap = kryo.readObject(input, BytesBytesMultiHashMap.class);
- if (initialCapacity > 0) {
- restoredHashMap.expandAndRehashToTarget(initialCapacity);
+ if (rowCount > 0) {
+ restoredHashMap.expandAndRehashToTarget(rowCount);
}
// some bookkeeping
@@ -237,7 +236,7 @@ public class HybridHashTableContainer
public HybridHashTableContainer(Configuration hconf, long keyCount, long memoryAvailable,
long estimatedTableSize, HybridHashTableConf nwayConf)
- throws SerDeException, IOException {
+ throws SerDeException, IOException {
this(HiveConf.getFloatVar(hconf, HiveConf.ConfVars.HIVEHASHTABLEKEYCOUNTADJUSTMENT),
HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVEHASHTABLETHRESHOLD),
HiveConf.getFloatVar(hconf,HiveConf.ConfVars.HIVEHASHTABLELOADFACTOR),
@@ -245,12 +244,13 @@ public class HybridHashTableContainer
HiveConf.getIntVar(hconf,HiveConf.ConfVars.HIVEHYBRIDGRACEHASHJOINMINWBSIZE),
HiveConf.getIntVar(hconf,HiveConf.ConfVars.HIVEHASHTABLEWBSIZE),
HiveConf.getIntVar(hconf,HiveConf.ConfVars.HIVEHYBRIDGRACEHASHJOINMINNUMPARTITIONS),
+ HiveConf.getFloatVar(hconf, HiveConf.ConfVars.HIVEMAPJOINOPTIMIZEDTABLEPROBEPERCENT),
estimatedTableSize, keyCount, memoryAvailable, nwayConf);
}
private HybridHashTableContainer(float keyCountAdj, int threshold, float loadFactor,
- int memCheckFreq, int minWbSize, int maxWbSize, int minNumParts, long estimatedTableSize,
- long keyCount, long memoryAvailable, HybridHashTableConf nwayConf)
+ int memCheckFreq, int minWbSize, int maxWbSize, int minNumParts, float probePercent,
+ long estimatedTableSize, long keyCount, long memoryAvailable, HybridHashTableConf nwayConf)
throws SerDeException, IOException {
directWriteHelper = new MapJoinBytesTableContainer.DirectKeyValueWriter();
@@ -262,10 +262,10 @@ public class HybridHashTableContainer
memoryCheckFrequency = memCheckFreq;
this.nwayConf = nwayConf;
+ int writeBufferSize;
int numPartitions;
if (nwayConf == null) { // binary join
- numPartitions = calcNumPartitions(memoryThreshold, estimatedTableSize, minNumParts, minWbSize,
- nwayConf);
+ numPartitions = calcNumPartitions(memoryThreshold, estimatedTableSize, minNumParts, minWbSize);
writeBufferSize = (int)(estimatedTableSize / numPartitions);
} else { // n-way join
// It has been calculated in HashTableLoader earlier, so just need to retrieve that number
@@ -302,21 +302,33 @@ public class HybridHashTableContainer
int numPartitionsSpilledOnCreation = 0;
memoryUsed = 0;
int initialCapacity = Math.max(newKeyCount / numPartitions, threshold / numPartitions);
+ // maxCapacity should be calculated based on a percentage of memoryThreshold, which is to divide
+ // row size using long size
+ float probePercentage = (float) 8 / (tableRowSize + 8); // long_size / tableRowSize + long_size
+ if (probePercentage == 1) {
+ probePercentage = probePercent;
+ }
+ int maxCapacity = (int)(memoryThreshold * probePercentage);
for (int i = 0; i < numPartitions; i++) {
if (this.nwayConf == null || // binary join
nwayConf.getLoadedContainerList().size() == 0) { // n-way join, first (biggest) small table
if (i == 0) { // We unconditionally create a hashmap for the first hash partition
- hashPartitions[i] = new HashPartition(initialCapacity, loadFactor, writeBufferSize, memoryThreshold, true);
+ hashPartitions[i] = new HashPartition(initialCapacity, loadFactor, writeBufferSize,
+ maxCapacity, true);
} else {
- hashPartitions[i] = new HashPartition(initialCapacity, loadFactor, writeBufferSize, memoryThreshold,
- memoryUsed + writeBufferSize < memoryThreshold);
+ // To check whether we have enough memory to allocate for another hash partition,
+ // we need to get the size of the first hash partition to get an idea.
+ hashPartitions[i] = new HashPartition(initialCapacity, loadFactor, writeBufferSize,
+ maxCapacity, memoryUsed + hashPartitions[0].hashMap.memorySize() < memoryThreshold);
}
- } else { // n-way join
+ } else { // n-way join, all later small tables
// For all later small tables, follow the same pattern of the previously loaded tables.
if (this.nwayConf.doSpillOnCreation(i)) {
- hashPartitions[i] = new HashPartition(threshold, loadFactor, writeBufferSize, memoryThreshold, false);
+ hashPartitions[i] = new HashPartition(initialCapacity, loadFactor, writeBufferSize,
+ maxCapacity, false);
} else {
- hashPartitions[i] = new HashPartition(threshold, loadFactor, writeBufferSize, memoryThreshold, true);
+ hashPartitions[i] = new HashPartition(initialCapacity, loadFactor, writeBufferSize,
+ maxCapacity, true);
}
}
@@ -513,7 +525,8 @@ public class HybridHashTableContainer
Path path = Files.createTempFile("partition-" + partitionId + "-", null);
OutputStream outputStream = Files.newOutputStream(path);
- com.esotericsoftware.kryo.io.Output output = new com.esotericsoftware.kryo.io.Output(outputStream);
+ com.esotericsoftware.kryo.io.Output output =
+ new com.esotericsoftware.kryo.io.Output(outputStream);
Kryo kryo = Utilities.runtimeSerializationKryo.get();
kryo.writeObject(output, partition.hashMap); // use Kryo to serialize hashmap
output.close();
@@ -545,11 +558,10 @@ public class HybridHashTableContainer
* @param dataSize total data size for the table
* @param minNumParts minimum required number of partitions
* @param minWbSize minimum required write buffer size
- * @param nwayConf the n-way join configuration
* @return number of partitions needed
*/
public static int calcNumPartitions(long memoryThreshold, long dataSize, int minNumParts,
- int minWbSize, HybridHashTableConf nwayConf) throws IOException {
+ int minWbSize) throws IOException {
int numPartitions = minNumParts;
if (memoryThreshold < minNumParts * minWbSize) {
@@ -803,7 +815,8 @@ public class HybridHashTableContainer
return JoinUtil.JoinResult.SPILL;
}
else {
- aliasFilter = hashPartitions[partitionId].hashMap.getValueResult(output.getData(), 0, output.getLength(), hashMapResult);
+ aliasFilter = hashPartitions[partitionId].hashMap.getValueResult(output.getData(), 0,
+ output.getLength(), hashMapResult);
dummyRow = null;
if (hashMapResult.hasRows()) {
return JoinUtil.JoinResult.MATCH;
@@ -941,7 +954,8 @@ public class HybridHashTableContainer
return JoinUtil.JoinResult.SPILL;
}
else {
- aliasFilter = hashPartitions[partitionId].hashMap.getValueResult(bytes, offset, length, hashMapResult);
+ aliasFilter = hashPartitions[partitionId].hashMap.getValueResult(bytes, offset, length,
+ hashMapResult);
dummyRow = null;
if (hashMapResult.hasRows()) {
return JoinUtil.JoinResult.MATCH;
http://git-wip-us.apache.org/repos/asf/hive/blob/fd5d1916/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
index 4414382..8a3647c 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
@@ -92,6 +92,7 @@ public class HashTableLoader implements org.apache.hadoop.hive.ql.exec.HashTable
// Get the total available memory from memory manager
long totalMapJoinMemory = desc.getMemoryNeeded();
+ LOG.info("Memory manager allocates " + totalMapJoinMemory + " bytes for the loading hashtable.");
if (totalMapJoinMemory <= 0) {
totalMapJoinMemory = HiveConf.getLongVar(
hconf, HiveConf.ConfVars.HIVECONVERTJOINNOCONDITIONALTASKTHRESHOLD);
@@ -136,11 +137,9 @@ public class HashTableLoader implements org.apache.hadoop.hive.ql.exec.HashTable
long memory = tableMemorySizes.get(biggest);
int numPartitions = 0;
try {
- numPartitions = HybridHashTableContainer.calcNumPartitions(memory,
- maxSize,
+ numPartitions = HybridHashTableContainer.calcNumPartitions(memory, maxSize,
HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVEHYBRIDGRACEHASHJOINMINNUMPARTITIONS),
- HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVEHYBRIDGRACEHASHJOINMINWBSIZE),
- nwayConf);
+ HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVEHYBRIDGRACEHASHJOINMINWBSIZE));
} catch (IOException e) {
throw new HiveException(e);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/fd5d1916/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
----------------------------------------------------------------------
diff --git a/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java b/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
index 05d9359..62250ec 100644
--- a/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
+++ b/serde/src/java/org/apache/hadoop/hive/serde2/WriteBuffers.java
@@ -61,7 +61,6 @@ public final class WriteBuffers implements RandomAccessOutput {
this.offsetMask = this.wbSize - 1;
this.maxSize = maxSize;
writePos.bufferIndex = -1;
- nextBufferToWrite();
}
public int readVInt() {
@@ -207,6 +206,9 @@ public final class WriteBuffers implements RandomAccessOutput {
@Override
public void write(byte[] b, int off, int len) {
+ if (writePos.bufferIndex == -1) {
+ nextBufferToWrite();
+ }
int srcOffset = 0;
while (srcOffset < len) {
int toWrite = Math.min(len - srcOffset, wbSize - writePos.offset);
@@ -355,6 +357,9 @@ public final class WriteBuffers implements RandomAccessOutput {
public long getWritePoint() {
+ if (writePos.bufferIndex == -1) {
+ nextBufferToWrite();
+ }
return ((long)writePos.bufferIndex << wbSizeLog2) + writePos.offset;
}
@@ -498,6 +503,9 @@ public final class WriteBuffers implements RandomAccessOutput {
}
public void seal() {
+ if (writePos.bufferIndex == -1) {
+ return;
+ }
if (writePos.offset < (wbSize * 0.8)) { // arbitrary
byte[] smallerBuffer = new byte[writePos.offset];
System.arraycopy(writePos.buffer, 0, smallerBuffer, 0, writePos.offset);
[2/5] hive git commit: HIVE-11755 :Incorrect method called with
Kerberos enabled in AccumuloStorageHandler (Josh Elser via Brock Noland)
Posted by se...@apache.org.
HIVE-11755 :Incorrect method called with Kerberos enabled in AccumuloStorageHandler (Josh Elser via Brock Noland)
Signed-off-by: Ashutosh Chauhan <ha...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/dade34a5
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/dade34a5
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/dade34a5
Branch: refs/heads/branch-1
Commit: dade34a51d30b6586028df84fc740309c06f6a6c
Parents: fd5d191
Author: Josh Elser <jo...@gmail.com>
Authored: Tue Sep 8 14:46:00 2015 -0800
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Thu Nov 12 11:26:18 2015 -0800
----------------------------------------------------------------------
.../hive/accumulo/HiveAccumuloHelper.java | 55 ++++++++++---
.../mr/HiveAccumuloTableOutputFormat.java | 50 ++++++++----
.../hive/accumulo/TestHiveAccumuloHelper.java | 69 +++++++++++++++-
.../mr/TestHiveAccumuloTableOutputFormat.java | 86 +++++++++++++++++++-
4 files changed, 229 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/dade34a5/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/HiveAccumuloHelper.java
----------------------------------------------------------------------
diff --git a/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/HiveAccumuloHelper.java b/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/HiveAccumuloHelper.java
index dfc5d03..71b8b77 100644
--- a/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/HiveAccumuloHelper.java
+++ b/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/HiveAccumuloHelper.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.hive.accumulo;
import static com.google.common.base.Preconditions.checkNotNull;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collection;
@@ -238,22 +239,56 @@ public class HiveAccumuloHelper {
public void setZooKeeperInstance(JobConf jobConf, Class<?> inputOrOutputFormatClass, String
zookeepers, String instanceName, boolean useSasl) throws IOException {
try {
- Class<?> clientConfigClass = JavaUtils.loadClass(CLIENT_CONFIGURATION_CLASS_NAME);
-
- // get the ClientConfiguration
- Object clientConfig = getClientConfiguration(zookeepers, instanceName, useSasl);
-
- // AccumuloOutputFormat.setZooKeeperInstance(JobConf, ClientConfiguration) or
- // AccumuloInputFormat.setZooKeeperInstance(JobConf, ClientConfiguration)
- Method setZooKeeperMethod = inputOrOutputFormatClass.getMethod(
- SET_ZOOKEEPER_INSTANCE_METHOD_NAME, JobConf.class, clientConfigClass);
- setZooKeeperMethod.invoke(null, jobConf, clientConfig);
+ setZooKeeperInstanceWithReflection(jobConf, inputOrOutputFormatClass, zookeepers,
+ instanceName, useSasl);
+ } catch (InvocationTargetException e) {
+ Throwable cause = e.getCause();
+ if (null != cause && cause instanceof IllegalStateException) {
+ throw (IllegalStateException) cause;
+ }
+ throw new IOException("Failed to invoke setZooKeeperInstance method", e);
+ } catch (IllegalStateException e) {
+ // re-throw the ISE so the caller can work around the silly impl that throws this in the
+ // first place.
+ throw e;
} catch (Exception e) {
throw new IOException("Failed to invoke setZooKeeperInstance method", e);
}
}
/**
+ * Wrap the setZooKeeperInstance reflected-call into its own method for testing
+ *
+ * @param jobConf
+ * The JobConf
+ * @param inputOrOutputFormatClass
+ * The InputFormat or OutputFormat class
+ * @param zookeepers
+ * ZooKeeper hosts
+ * @param instanceName
+ * Accumulo instance name
+ * @param useSasl
+ * Is SASL enabled
+ * @throws IOException
+ * When invocation of the method fails
+ */
+ void setZooKeeperInstanceWithReflection(JobConf jobConf, Class<?> inputOrOutputFormatClass, String
+ zookeepers, String instanceName, boolean useSasl) throws IOException, ClassNotFoundException,
+ NoSuchMethodException, SecurityException, IllegalAccessException, IllegalArgumentException,
+ InvocationTargetException {
+ Class<?> clientConfigClass = JavaUtils.loadClass(CLIENT_CONFIGURATION_CLASS_NAME);
+
+ // get the ClientConfiguration
+ Object clientConfig = getClientConfiguration(zookeepers, instanceName, useSasl);
+
+ // AccumuloOutputFormat.setZooKeeperInstance(JobConf, ClientConfiguration) or
+ // AccumuloInputFormat.setZooKeeperInstance(JobConf, ClientConfiguration)
+ Method setZooKeeperMethod = inputOrOutputFormatClass.getMethod(
+ SET_ZOOKEEPER_INSTANCE_METHOD_NAME, JobConf.class, clientConfigClass);
+ setZooKeeperMethod.invoke(null, jobConf, clientConfig);
+ }
+
+ /**
* Wrapper around <code>ConfiguratorBase.unwrapAuthenticationToken</code> which only exists in
* 1.7.0 and new. Uses reflection to not break compat.
*
http://git-wip-us.apache.org/repos/asf/hive/blob/dade34a5/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/mr/HiveAccumuloTableOutputFormat.java
----------------------------------------------------------------------
diff --git a/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/mr/HiveAccumuloTableOutputFormat.java b/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/mr/HiveAccumuloTableOutputFormat.java
index 0189c07..3ae5431 100644
--- a/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/mr/HiveAccumuloTableOutputFormat.java
+++ b/accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/mr/HiveAccumuloTableOutputFormat.java
@@ -61,7 +61,7 @@ public class HiveAccumuloTableOutputFormat extends AccumuloOutputFormat {
}
protected void configureAccumuloOutputFormat(JobConf job) throws IOException {
- AccumuloConnectionParameters cnxnParams = new AccumuloConnectionParameters(job);
+ AccumuloConnectionParameters cnxnParams = getConnectionParams(job);
final String tableName = job.get(AccumuloSerDeParameters.TABLE_NAME);
@@ -72,35 +72,35 @@ public class HiveAccumuloTableOutputFormat extends AccumuloOutputFormat {
// Set the necessary Accumulo information
try {
if (cnxnParams.useMockInstance()) {
- setAccumuloMockInstance(job, cnxnParams.getAccumuloInstanceName());
+ setMockInstanceWithErrorChecking(job, cnxnParams.getAccumuloInstanceName());
} else {
// Accumulo instance name with ZK quorum
- setAccumuloZooKeeperInstance(job, cnxnParams.getAccumuloInstanceName(),
+ setZooKeeperInstanceWithErrorChecking(job, cnxnParams.getAccumuloInstanceName(),
cnxnParams.getZooKeepers(), cnxnParams.useSasl());
}
// Extract the delegation Token from the UGI and add it to the job
// The AccumuloOutputFormat will look for it there.
if (cnxnParams.useSasl()) {
- UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
- if (!ugi.hasKerberosCredentials()) {
- helper.addTokenFromUserToJobConf(ugi, job);
+ UserGroupInformation ugi = getCurrentUser();
+ if (!hasKerberosCredentials(ugi)) {
+ getHelper().addTokenFromUserToJobConf(ugi, job);
} else {
// Still in the local JVM, can use Kerberos credentials
try {
Connector connector = cnxnParams.getConnector();
- AuthenticationToken token = helper.getDelegationToken(connector);
+ AuthenticationToken token = getHelper().getDelegationToken(connector);
// Send the DelegationToken down to the Configuration for Accumulo to use
- setConnectorInfo(job, cnxnParams.getAccumuloUserName(), token);
+ setConnectorInfoWithErrorChecking(job, cnxnParams.getAccumuloUserName(), token);
// Convert the Accumulo token in a Hadoop token
- Token<? extends TokenIdentifier> accumuloToken = helper.getHadoopToken(token);
+ Token<? extends TokenIdentifier> accumuloToken = getHelper().getHadoopToken(token);
log.info("Adding Hadoop Token for Accumulo to Job's Credentials");
// Add the Hadoop token to the JobConf
- helper.mergeTokenIntoJobConf(job, accumuloToken);
+ getHelper().mergeTokenIntoJobConf(job, accumuloToken);
// Make sure the UGI contains the token too for good measure
if (!ugi.addToken(accumuloToken)) {
@@ -111,7 +111,7 @@ public class HiveAccumuloTableOutputFormat extends AccumuloOutputFormat {
}
}
} else {
- setAccumuloConnectorInfo(job, cnxnParams.getAccumuloUserName(),
+ setConnectorInfoWithErrorChecking(job, cnxnParams.getAccumuloUserName(),
new PasswordToken(cnxnParams.getAccumuloPassword()));
}
@@ -125,7 +125,7 @@ public class HiveAccumuloTableOutputFormat extends AccumuloOutputFormat {
// Non-static methods to wrap the static AccumuloOutputFormat methods to enable testing
- protected void setAccumuloConnectorInfo(JobConf conf, String username, AuthenticationToken token)
+ protected void setConnectorInfoWithErrorChecking(JobConf conf, String username, AuthenticationToken token)
throws AccumuloSecurityException {
try {
AccumuloOutputFormat.setConnectorInfo(conf, username, token);
@@ -136,14 +136,14 @@ public class HiveAccumuloTableOutputFormat extends AccumuloOutputFormat {
}
@SuppressWarnings("deprecation")
- protected void setAccumuloZooKeeperInstance(JobConf conf, String instanceName, String zookeepers,
+ protected void setZooKeeperInstanceWithErrorChecking(JobConf conf, String instanceName, String zookeepers,
boolean isSasl) throws IOException {
try {
if (isSasl) {
// Reflection to support Accumulo 1.5. Remove when Accumulo 1.5 support is dropped
// 1.6 works with the deprecated 1.5 method, but must use reflection for 1.7-only
// SASL support
- helper.setZooKeeperInstance(conf, AccumuloOutputFormat.class, zookeepers, instanceName,
+ getHelper().setZooKeeperInstance(conf, AccumuloOutputFormat.class, zookeepers, instanceName,
isSasl);
} else {
AccumuloOutputFormat.setZooKeeperInstance(conf, instanceName, zookeepers);
@@ -155,7 +155,7 @@ public class HiveAccumuloTableOutputFormat extends AccumuloOutputFormat {
}
}
- protected void setAccumuloMockInstance(JobConf conf, String instanceName) {
+ protected void setMockInstanceWithErrorChecking(JobConf conf, String instanceName) {
try {
AccumuloOutputFormat.setMockInstance(conf, instanceName);
} catch (IllegalStateException e) {
@@ -167,4 +167,24 @@ public class HiveAccumuloTableOutputFormat extends AccumuloOutputFormat {
protected void setDefaultAccumuloTableName(JobConf conf, String tableName) {
AccumuloOutputFormat.setDefaultTableName(conf, tableName);
}
+
+ HiveAccumuloHelper getHelper() {
+ // Allows mocking in testing.
+ return helper;
+ }
+
+ AccumuloConnectionParameters getConnectionParams(JobConf conf) {
+ // Allows mocking in testing.
+ return new AccumuloConnectionParameters(conf);
+ }
+
+ boolean hasKerberosCredentials(UserGroupInformation ugi) {
+ // Allows mocking in testing.
+ return ugi.hasKerberosCredentials();
+ }
+
+ UserGroupInformation getCurrentUser() throws IOException {
+ // Allows mocking in testing.
+ return UserGroupInformation.getCurrentUser();
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/dade34a5/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/TestHiveAccumuloHelper.java
----------------------------------------------------------------------
diff --git a/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/TestHiveAccumuloHelper.java b/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/TestHiveAccumuloHelper.java
index 88544f0..406768a 100644
--- a/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/TestHiveAccumuloHelper.java
+++ b/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/TestHiveAccumuloHelper.java
@@ -18,18 +18,23 @@ package org.apache.hadoop.hive.accumulo;
import static org.junit.Assert.assertEquals;
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collection;
+import org.apache.accumulo.core.client.mapred.AccumuloInputFormat;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
+import org.apache.log4j.Logger;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
public class TestHiveAccumuloHelper {
+ private static final Logger log = Logger.getLogger(TestHiveAccumuloHelper.class);
private HiveAccumuloHelper helper;
@@ -46,7 +51,13 @@ public class TestHiveAccumuloHelper {
Mockito.when(token.getService()).thenReturn(service);
- helper.mergeTokenIntoJobConf(jobConf, token);
+ try {
+ helper.mergeTokenIntoJobConf(jobConf, token);
+ } catch (IOException e) {
+ // Hadoop 1 doesn't support credential merging, so this will fail.
+ log.info("Ignoring exception, likely coming from Hadoop 1", e);
+ return;
+ }
Collection<Token<?>> tokens = jobConf.getCredentials().getAllTokens();
assertEquals(1, tokens.size());
@@ -66,10 +77,64 @@ public class TestHiveAccumuloHelper {
Mockito.when(token.getKind()).thenReturn(HiveAccumuloHelper.ACCUMULO_SERVICE);
Mockito.when(token.getService()).thenReturn(service);
- helper.addTokenFromUserToJobConf(ugi, jobConf);
+ try {
+ helper.addTokenFromUserToJobConf(ugi, jobConf);
+ } catch (IOException e) {
+ // Hadoop 1 doesn't support credential merging, so this will fail.
+ log.info("Ignoring exception, likely coming from Hadoop 1", e);
+ return;
+ }
Collection<Token<?>> credTokens = jobConf.getCredentials().getAllTokens();
assertEquals(1, credTokens.size());
assertEquals(service, credTokens.iterator().next().getService());
}
+
+ @Test(expected = IllegalStateException.class)
+ public void testISEIsPropagated() throws Exception {
+ final HiveAccumuloHelper helper = Mockito.mock(HiveAccumuloHelper.class);
+
+ final JobConf jobConf = Mockito.mock(JobConf.class);
+ final Class<?> inputOrOutputFormatClass = AccumuloInputFormat.class;
+ final String zookeepers = "localhost:2181";
+ final String instanceName = "accumulo_instance";
+ final boolean useSasl = false;
+
+ // Call the real "public" method
+ Mockito.doCallRealMethod().when(helper).setZooKeeperInstance(jobConf, inputOrOutputFormatClass,
+ zookeepers, instanceName, useSasl);
+
+ // Mock the private one to throw the ISE
+ Mockito.doThrow(new IllegalStateException()).when(helper).
+ setZooKeeperInstanceWithReflection(jobConf, inputOrOutputFormatClass, zookeepers,
+ instanceName, useSasl);
+
+ // Should throw an IllegalStateException
+ helper.setZooKeeperInstance(jobConf, inputOrOutputFormatClass, zookeepers, instanceName,
+ useSasl);
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testISEIsPropagatedWithReflection() throws Exception {
+ final HiveAccumuloHelper helper = Mockito.mock(HiveAccumuloHelper.class);
+
+ final JobConf jobConf = Mockito.mock(JobConf.class);
+ final Class<?> inputOrOutputFormatClass = AccumuloInputFormat.class;
+ final String zookeepers = "localhost:2181";
+ final String instanceName = "accumulo_instance";
+ final boolean useSasl = false;
+
+ // Call the real "public" method
+ Mockito.doCallRealMethod().when(helper).setZooKeeperInstance(jobConf, inputOrOutputFormatClass,
+ zookeepers, instanceName, useSasl);
+
+ // Mock the private one to throw the IAE
+ Mockito.doThrow(new InvocationTargetException(new IllegalStateException())).when(helper).
+ setZooKeeperInstanceWithReflection(jobConf, inputOrOutputFormatClass, zookeepers,
+ instanceName, useSasl);
+
+ // Should throw an IllegalStateException
+ helper.setZooKeeperInstance(jobConf, inputOrOutputFormatClass, zookeepers, instanceName,
+ useSasl);
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/dade34a5/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/mr/TestHiveAccumuloTableOutputFormat.java
----------------------------------------------------------------------
diff --git a/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/mr/TestHiveAccumuloTableOutputFormat.java b/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/mr/TestHiveAccumuloTableOutputFormat.java
index 5d3f15b..5fdab28 100644
--- a/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/mr/TestHiveAccumuloTableOutputFormat.java
+++ b/accumulo-handler/src/test/org/apache/hadoop/hive/accumulo/mr/TestHiveAccumuloTableOutputFormat.java
@@ -18,15 +18,18 @@ package org.apache.hadoop.hive.accumulo.mr;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.Properties;
+import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
@@ -36,6 +39,7 @@ import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hive.accumulo.AccumuloConnectionParameters;
+import org.apache.hadoop.hive.accumulo.HiveAccumuloHelper;
import org.apache.hadoop.hive.accumulo.columns.ColumnEncoding;
import org.apache.hadoop.hive.accumulo.serde.AccumuloRowSerializer;
import org.apache.hadoop.hive.accumulo.serde.AccumuloSerDe;
@@ -54,6 +58,10 @@ import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
@@ -87,6 +95,15 @@ public class TestHiveAccumuloTableOutputFormat {
conf.set(AccumuloConnectionParameters.INSTANCE_NAME, instanceName);
conf.set(AccumuloConnectionParameters.ZOOKEEPERS, zookeepers);
conf.set(AccumuloConnectionParameters.TABLE_NAME, outputTable);
+
+ System.setProperty("java.security.krb5.realm", "accumulo");
+ System.setProperty("java.security.krb5.kdc", "fake");
+ }
+
+ @After
+ public void cleanup() {
+ System.setProperty("java.security.krb5.realm", "");
+ System.setProperty("java.security.krb5.kdc", "");
}
@Test
@@ -94,12 +111,72 @@ public class TestHiveAccumuloTableOutputFormat {
HiveAccumuloTableOutputFormat outputFormat = Mockito.mock(HiveAccumuloTableOutputFormat.class);
Mockito.doCallRealMethod().when(outputFormat).configureAccumuloOutputFormat(conf);
+ Mockito.doCallRealMethod().when(outputFormat).getConnectionParams(conf);
+
+ outputFormat.configureAccumuloOutputFormat(conf);
+
+ Mockito.verify(outputFormat).setConnectorInfoWithErrorChecking(conf, user, new PasswordToken(password));
+ Mockito.verify(outputFormat).setZooKeeperInstanceWithErrorChecking(conf, instanceName, zookeepers, false);
+ Mockito.verify(outputFormat).setDefaultAccumuloTableName(conf, outputTable);
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ @Test
+ public void testSaslConfiguration() throws IOException, AccumuloException, AccumuloSecurityException {
+ final HiveAccumuloTableOutputFormat outputFormat = Mockito.mock(HiveAccumuloTableOutputFormat.class);
+ final AuthenticationToken authToken = Mockito.mock(AuthenticationToken.class);
+ final Token hadoopToken = Mockito.mock(Token.class);
+ final HiveAccumuloHelper helper = Mockito.mock(HiveAccumuloHelper.class);
+ final AccumuloConnectionParameters cnxnParams = Mockito.mock(AccumuloConnectionParameters.class);
+ final Connector connector = Mockito.mock(Connector.class);
+
+ // Set UGI to use Kerberos
+ // Have to use the string constant to support hadoop 1
+ conf.set("hadoop.security.authentication", "kerberos");
+ UserGroupInformation.setConfiguration(conf);
+
+ // Set the current UGI to a fake user
+ UserGroupInformation user1 = UserGroupInformation.createUserForTesting(user, new String[0]);
+ // Use that as the "current user"
+ Mockito.when(outputFormat.getCurrentUser()).thenReturn(user1);
+
+ // Turn off passwords, enable sasl and set a keytab
+ conf.unset(AccumuloConnectionParameters.USER_PASS);
+
+ // Call the real method instead of the mock
+ Mockito.doCallRealMethod().when(outputFormat).configureAccumuloOutputFormat(conf);
+
+ // Return our mocked objects
+ Mockito.when(outputFormat.getHelper()).thenReturn(helper);
+ Mockito.when(outputFormat.getConnectionParams(conf)).thenReturn(cnxnParams);
+ Mockito.when(cnxnParams.getConnector()).thenReturn(connector);
+ Mockito.when(helper.getDelegationToken(connector)).thenReturn(authToken);
+ Mockito.when(helper.getHadoopToken(authToken)).thenReturn(hadoopToken);
+ // Stub AccumuloConnectionParameters actions
+ Mockito.when(cnxnParams.useSasl()).thenReturn(true);
+ Mockito.when(cnxnParams.getAccumuloUserName()).thenReturn(user);
+ Mockito.when(cnxnParams.getAccumuloInstanceName()).thenReturn(instanceName);
+ Mockito.when(cnxnParams.getZooKeepers()).thenReturn(zookeepers);
+
+ // Stub OutputFormat actions
+ Mockito.when(outputFormat.hasKerberosCredentials(user1)).thenReturn(true);
+
+ // Invoke the method
outputFormat.configureAccumuloOutputFormat(conf);
- Mockito.verify(outputFormat).setAccumuloConnectorInfo(conf, user, new PasswordToken(password));
- Mockito.verify(outputFormat).setAccumuloZooKeeperInstance(conf, instanceName, zookeepers, false);
+ // The AccumuloInputFormat methods
+ Mockito.verify(outputFormat).setZooKeeperInstanceWithErrorChecking(conf, instanceName, zookeepers, true);
+ Mockito.verify(outputFormat).setConnectorInfoWithErrorChecking(conf, user, authToken);
Mockito.verify(outputFormat).setDefaultAccumuloTableName(conf, outputTable);
+
+ // Other methods we expect
+ Mockito.verify(helper).mergeTokenIntoJobConf(conf, hadoopToken);
+
+ // Make sure the token made it into the UGI
+ Collection<Token<? extends TokenIdentifier>> tokens = user1.getTokens();
+ Assert.assertEquals(1, tokens.size());
+ Assert.assertEquals(hadoopToken, tokens.iterator().next());
}
@Test
@@ -109,11 +186,12 @@ public class TestHiveAccumuloTableOutputFormat {
conf.unset(AccumuloConnectionParameters.ZOOKEEPERS);
Mockito.doCallRealMethod().when(outputFormat).configureAccumuloOutputFormat(conf);
+ Mockito.doCallRealMethod().when(outputFormat).getConnectionParams(conf);
outputFormat.configureAccumuloOutputFormat(conf);
- Mockito.verify(outputFormat).setAccumuloConnectorInfo(conf, user, new PasswordToken(password));
- Mockito.verify(outputFormat).setAccumuloMockInstance(conf, instanceName);
+ Mockito.verify(outputFormat).setConnectorInfoWithErrorChecking(conf, user, new PasswordToken(password));
+ Mockito.verify(outputFormat).setMockInstanceWithErrorChecking(conf, instanceName);
Mockito.verify(outputFormat).setDefaultAccumuloTableName(conf, outputTable);
}
[3/5] hive git commit: HIVE-11583 : When PTF is used over a large
partitions result could be corrupted (Illya Yalovyy via Ashutosh Chauhan)
Posted by se...@apache.org.
HIVE-11583 : When PTF is used over a large partitions result could be corrupted (Illya Yalovyy via Ashutosh Chauhan)
Signed-off-by: Ashutosh Chauhan <ha...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/9f0f8dba
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/9f0f8dba
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/9f0f8dba
Branch: refs/heads/branch-1
Commit: 9f0f8dba3edfc847c3d8c14af42b2d5e06f44ed1
Parents: dade34a
Author: Illya Yalovyy <ya...@amazon.com>
Authored: Mon Sep 14 10:18:00 2015 -0800
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Thu Nov 12 11:26:43 2015 -0800
----------------------------------------------------------------------
.../ql/exec/persistence/PTFRowContainer.java | 14 +++++----
.../hive/ql/exec/persistence/RowContainer.java | 12 +++++---
.../exec/persistence/TestPTFRowContainer.java | 31 ++++++++++++++------
3 files changed, 38 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/9f0f8dba/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/PTFRowContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/PTFRowContainer.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/PTFRowContainer.java
index d2bfea6..61cc6e8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/PTFRowContainer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/PTFRowContainer.java
@@ -81,8 +81,8 @@ import org.apache.hadoop.util.Progressable;
*/
public class PTFRowContainer<Row extends List<Object>> extends RowContainer<Row> {
- ArrayList<BlockInfo> blockInfos;
- int currentReadBlockStartRow;
+ private ArrayList<BlockInfo> blockInfos;
+ private int currentReadBlockStartRow;
public PTFRowContainer(int bs, Configuration jc, Reporter reporter
) throws HiveException {
@@ -190,14 +190,16 @@ public class PTFRowContainer<Row extends List<Object>> extends RowContainer<Row>
BlockInfo bI = blockInfos.get(blockNum);
int startSplit = bI.startingSplit;
- int endSplit = startSplit;
- if ( blockNum != blockInfos.size() - 1) {
- endSplit = blockInfos.get(blockNum+1).startingSplit;
+ int endSplit;
+ if ( blockNum != blockInfos.size() - 1 ) {
+ endSplit = blockInfos.get(blockNum + 1).startingSplit;
+ } else {
+ endSplit = getLastActualSplit();
}
try {
int readIntoOffset = 0;
- for(int i = startSplit; i <= endSplit; i++ ) {
+ for(int i = startSplit; i <= endSplit && readIntoOffset < getBlockSize(); i++ ) {
org.apache.hadoop.mapred.RecordReader rr = setReaderAtSplit(i);
if ( i == startSplit ) {
((PTFSequenceFileRecordReader)rr).seek(bI.startOffset);
http://git-wip-us.apache.org/repos/asf/hive/blob/9f0f8dba/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java
index 4252bd1..68dc482 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java
@@ -103,7 +103,7 @@ public class RowContainer<ROW extends List<Object>>
boolean firstCalled = false; // once called first, it will never be able to
// write again.
- int acutalSplitNum = 0;
+ private int actualSplitNum = 0;
int currentSplitPointer = 0;
org.apache.hadoop.mapred.RecordReader rr = null; // record reader
RecordWriter rw = null;
@@ -220,7 +220,7 @@ public class RowContainer<ROW extends List<Object>>
HiveConf.setVar(localJc, HiveConf.ConfVars.HADOOPMAPREDINPUTDIR,
org.apache.hadoop.util.StringUtils.escapeString(parentFile.getAbsolutePath()));
inputSplits = inputFormat.getSplits(localJc, 1);
- acutalSplitNum = inputSplits.length;
+ actualSplitNum = inputSplits.length;
}
currentSplitPointer = 0;
rr = inputFormat.getRecordReader(inputSplits[currentSplitPointer],
@@ -375,7 +375,7 @@ public class RowContainer<ROW extends List<Object>>
}
}
- if (nextSplit && this.currentSplitPointer < this.acutalSplitNum) {
+ if (nextSplit && this.currentSplitPointer < this.actualSplitNum) {
JobConf localJc = getLocalFSJobConfClone(jc);
// open record reader to read next split
rr = inputFormat.getRecordReader(inputSplits[currentSplitPointer], jobCloneUsingLocalFs,
@@ -421,7 +421,7 @@ public class RowContainer<ROW extends List<Object>>
addCursor = 0;
numFlushedBlocks = 0;
this.readBlockSize = 0;
- this.acutalSplitNum = 0;
+ this.actualSplitNum = 0;
this.currentSplitPointer = -1;
this.firstCalled = false;
this.inputSplits = null;
@@ -606,4 +606,8 @@ public class RowContainer<ROW extends List<Object>>
clearRows();
currentReadBlock = firstReadBlockPointer = currentWriteBlock = null;
}
+
+ protected int getLastActualSplit() {
+ return actualSplitNum - 1;
+ }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/9f0f8dba/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestPTFRowContainer.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestPTFRowContainer.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestPTFRowContainer.java
index a404ff0..0611072 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestPTFRowContainer.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/persistence/TestPTFRowContainer.java
@@ -18,12 +18,14 @@
package org.apache.hadoop.hive.ql.exec.persistence;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Random;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.hive.serde2.SerDeException;
@@ -37,11 +39,13 @@ import org.apache.hadoop.io.Text;
import org.junit.BeforeClass;
import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+
@SuppressWarnings("deprecation")
public class TestPTFRowContainer {
- private static final String COL_NAMES = "x,y,z,a,b";
- private static final String COL_TYPES = "int,string,double,int,string";
+ private static final String COL_NAMES = "x,y,z,a,b,v";
+ private static final String COL_TYPES = "int,string,double,int,string,string";
static SerDe serDe;
static Configuration cfg;
@@ -70,7 +74,7 @@ public class TestPTFRowContainer {
return rc;
}
- private void runTest(int sz, int blockSize) throws SerDeException, HiveException {
+ private void runTest(int sz, int blockSize, String value) throws SerDeException, HiveException {
List<Object> row;
PTFRowContainer<List<Object>> rc = rowContainer(blockSize);
@@ -82,16 +86,17 @@ public class TestPTFRowContainer {
row.add(new DoubleWritable(i));
row.add(new IntWritable(i));
row.add(new Text("def " + i));
+ row.add(new Text(value));
rc.addRow(row);
}
// test forward scan
- assert(rc.rowCount() == sz);
+ assertEquals(sz, rc.rowCount());
i = 0;
row = new ArrayList<Object>();
row = rc.first();
while(row != null ) {
- assert(row.get(1).toString().equals("abc " + i));
+ assertEquals("abc " + i, row.get(1).toString());
i++;
row = rc.next();
}
@@ -100,7 +105,7 @@ public class TestPTFRowContainer {
row = rc.first();
for(i = sz - 1; i >= 0; i-- ) {
row = rc.getAt(i);
- assert(row.get(1).toString().equals("abc " + i));
+ assertEquals("abc " + i, row.get(1).toString());
}
Random r = new Random(1000L);
@@ -109,20 +114,23 @@ public class TestPTFRowContainer {
for(i=0; i < 100; i++) {
int j = r.nextInt(sz);
row = rc.getAt(j);
- assert(row.get(1).toString().equals("abc " + j));
+ assertEquals("abc " + j, row.get(1).toString());
}
// intersperse getAt and next calls
for(i=0; i < 100; i++) {
int j = r.nextInt(sz);
row = rc.getAt(j);
- assert(row.get(1).toString().equals("abc " + j));
+ assertEquals("abc " + j, row.get(1).toString());
for(int k = j + 1; k < j + (blockSize/4) && k < sz; k++) {
row = rc.next();
- assert(row.get(4).toString().equals("def " + k));
+ assertEquals("def " + k, row.get(4).toString());
}
}
+ }
+ private void runTest(int sz, int blockSize) throws SerDeException, HiveException {
+ runTest(sz, blockSize, "");
}
@Test
@@ -134,4 +142,9 @@ public class TestPTFRowContainer {
public void testSmallBlockSize() throws SerDeException, HiveException {
runTest(10 * 1000, 5);
}
+
+ @Test
+ public void testBlocksLargerThanSplit() throws SerDeException, HiveException, IOException {
+ runTest(5, 2, new String(new char[(int)FileSystem.getLocal(cfg).getDefaultBlockSize()]));
+ }
}
[4/5] hive git commit: HIVE-11616 : DelegationTokenSecretManager
reuses the same objectstore,
which has concurrency issues (Cody Fu/Sergey Shelukhin,
reviewed by Chaoyu Tang)
Posted by se...@apache.org.
HIVE-11616 : DelegationTokenSecretManager reuses the same objectstore, which has concurrency issues (Cody Fu/Sergey Shelukhin, reviewed by Chaoyu Tang)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/516fd74e
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/516fd74e
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/516fd74e
Branch: refs/heads/branch-1
Commit: 516fd74ef04598bac9c30f27a2d4549125621e0a
Parents: 9f0f8db
Author: Sergey Shelukhin <se...@apache.org>
Authored: Fri Oct 30 14:28:13 2015 -0700
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Thu Nov 12 11:28:09 2015 -0800
----------------------------------------------------------------------
.../java/org/apache/hadoop/hive/thrift/TestDBTokenStore.java | 2 +-
.../java/org/apache/hadoop/hive/metastore/HiveMetaStore.java | 2 +-
.../java/org/apache/hive/service/auth/HiveAuthFactory.java | 8 +++-----
.../java/org/apache/hadoop/hive/thrift/DBTokenStore.java | 7 ++++---
.../apache/hadoop/hive/thrift/HadoopThriftAuthBridge.java | 6 +++---
.../org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java | 2 +-
6 files changed, 13 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/516fd74e/itests/hive-unit/src/test/java/org/apache/hadoop/hive/thrift/TestDBTokenStore.java
----------------------------------------------------------------------
diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/thrift/TestDBTokenStore.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/thrift/TestDBTokenStore.java
index 0b61a62..f5934ee 100644
--- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/thrift/TestDBTokenStore.java
+++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/thrift/TestDBTokenStore.java
@@ -37,7 +37,7 @@ public class TestDBTokenStore extends TestCase{
public void testDBTokenStore() throws TokenStoreException, MetaException, IOException {
DelegationTokenStore ts = new DBTokenStore();
- ts.init(new HMSHandler("Test handler").getMS(), null);
+ ts.init(new HMSHandler("Test handler"), null);
assertEquals(0, ts.getMasterKeys().length);
assertEquals(false,ts.removeMasterKey(-1));
try{
http://git-wip-us.apache.org/repos/asf/hive/blob/516fd74e/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
index fa2fab6..0ffc838 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
@@ -6075,7 +6075,7 @@ public class HiveMetaStore extends ThriftHiveMetastore {
conf.getVar(HiveConf.ConfVars.METASTORE_KERBEROS_KEYTAB_FILE),
conf.getVar(HiveConf.ConfVars.METASTORE_KERBEROS_PRINCIPAL));
// start delegation token manager
- saslServer.startDelegationTokenSecretManager(conf, baseHandler.getMS(), ServerMode.METASTORE);
+ saslServer.startDelegationTokenSecretManager(conf, baseHandler, ServerMode.METASTORE);
transFactory = saslServer.createTransportFactory(
MetaStoreUtils.getMetaStoreSaslProperties(conf));
processor = saslServer.wrapProcessor(
http://git-wip-us.apache.org/repos/asf/hive/blob/516fd74e/service/src/java/org/apache/hive/service/auth/HiveAuthFactory.java
----------------------------------------------------------------------
diff --git a/service/src/java/org/apache/hive/service/auth/HiveAuthFactory.java b/service/src/java/org/apache/hive/service/auth/HiveAuthFactory.java
index 1e6ac4f..3471f12 100644
--- a/service/src/java/org/apache/hive/service/auth/HiveAuthFactory.java
+++ b/service/src/java/org/apache/hive/service/auth/HiveAuthFactory.java
@@ -112,16 +112,14 @@ public class HiveAuthFactory {
// start delegation token manager
try {
// rawStore is only necessary for DBTokenStore
- Object rawStore = null;
+ HMSHandler baseHandler = null;
String tokenStoreClass = conf.getVar(HiveConf.ConfVars.METASTORE_CLUSTER_DELEGATION_TOKEN_STORE_CLS);
if (tokenStoreClass.equals(DBTokenStore.class.getName())) {
- HMSHandler baseHandler = new HiveMetaStore.HMSHandler(
- "new db based metaserver", conf, true);
- rawStore = baseHandler.getMS();
+ baseHandler = new HiveMetaStore.HMSHandler("new db based metaserver", conf, true);
}
- saslServer.startDelegationTokenSecretManager(conf, rawStore, ServerMode.HIVESERVER2);
+ saslServer.startDelegationTokenSecretManager(conf, baseHandler, ServerMode.HIVESERVER2);
}
catch (MetaException|IOException e) {
throw new TTransportException("Failed to start token manager", e);
http://git-wip-us.apache.org/repos/asf/hive/blob/516fd74e/shims/common/src/main/java/org/apache/hadoop/hive/thrift/DBTokenStore.java
----------------------------------------------------------------------
diff --git a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/DBTokenStore.java b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/DBTokenStore.java
index 21d18f5..de39d3d 100644
--- a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/DBTokenStore.java
+++ b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/DBTokenStore.java
@@ -132,17 +132,18 @@ public class DBTokenStore implements DelegationTokenStore {
return delTokenIdents;
}
- private Object rawStore;
+ private Object hmsHandler;
@Override
- public void init(Object rawStore, ServerMode smode) throws TokenStoreException {
- this.rawStore = rawStore;
+ public void init(Object hms, ServerMode smode) throws TokenStoreException {
+ this.hmsHandler = hms;
}
private Object invokeOnRawStore(String methName, Object[] params, Class<?> ... paramTypes)
throws TokenStoreException{
try {
+ Object rawStore = hmsHandler.getClass().getMethod("getMS").invoke(hmsHandler);
return rawStore.getClass().getMethod(methName, paramTypes).invoke(rawStore, params);
} catch (IllegalArgumentException e) {
throw new TokenStoreException(e);
http://git-wip-us.apache.org/repos/asf/hive/blob/516fd74e/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HadoopThriftAuthBridge.java
----------------------------------------------------------------------
diff --git a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HadoopThriftAuthBridge.java b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HadoopThriftAuthBridge.java
index 626be84..984679f 100644
--- a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HadoopThriftAuthBridge.java
+++ b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/HadoopThriftAuthBridge.java
@@ -426,8 +426,8 @@ public class HadoopThriftAuthBridge {
}
- public void startDelegationTokenSecretManager(Configuration conf, Object rawStore, ServerMode smode)
- throws IOException{
+ public void startDelegationTokenSecretManager(Configuration conf, Object hms, ServerMode smode)
+ throws IOException {
long secretKeyInterval =
conf.getLong(DELEGATION_KEY_UPDATE_INTERVAL_KEY,
DELEGATION_KEY_UPDATE_INTERVAL_DEFAULT);
@@ -441,7 +441,7 @@ public class HadoopThriftAuthBridge {
DELEGATION_TOKEN_GC_INTERVAL_DEFAULT);
DelegationTokenStore dts = getTokenStore(conf);
- dts.init(rawStore, smode);
+ dts.init(hms, smode);
secretManager = new TokenStoreDelegationTokenSecretManager(secretKeyInterval,
tokenMaxLifetime,
tokenRenewInterval,
http://git-wip-us.apache.org/repos/asf/hive/blob/516fd74e/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java
----------------------------------------------------------------------
diff --git a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java
index e46f293..745e467 100644
--- a/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java
+++ b/shims/common/src/main/java/org/apache/hadoop/hive/thrift/ZooKeeperTokenStore.java
@@ -429,7 +429,7 @@ public class ZooKeeperTokenStore implements DelegationTokenStore {
}
@Override
- public void init(Object objectStore, ServerMode smode) {
+ public void init(Object hmsHandler, ServerMode smode) {
this.serverMode = smode;
zkConnectString =
conf.get(HadoopThriftAuthBridge.Server.DELEGATION_TOKEN_STORE_ZK_CONNECT_STR, null);
[5/5] hive git commit: HIVE-11825 : get_json_object(col,
'$.a') is null in where clause didnt work (Cazen Lee via Ashutosh
Chauhan)
Posted by se...@apache.org.
HIVE-11825 : get_json_object(col,'$.a') is null in where clause didnt work (Cazen Lee via Ashutosh Chauhan)
Signed-off-by: Ashutosh Chauhan <ha...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/48d0ec9e
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/48d0ec9e
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/48d0ec9e
Branch: refs/heads/branch-1
Commit: 48d0ec9ee0dcfdc41dcb7eaaa3097988b56eb2ea
Parents: 516fd74
Author: Cazen Lee <Ca...@samsung.com>
Authored: Wed Sep 16 01:04:00 2015 -0800
Committer: Sergey Shelukhin <se...@apache.org>
Committed: Thu Nov 12 11:30:29 2015 -0800
----------------------------------------------------------------------
ql/src/java/org/apache/hadoop/hive/ql/udf/UDFJson.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/48d0ec9e/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFJson.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFJson.java b/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFJson.java
index e69ad68..2c42fae 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFJson.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/udf/UDFJson.java
@@ -66,6 +66,8 @@ public class UDFJson extends UDF {
static {
// Allows for unescaped ASCII control characters in JSON values
JSON_FACTORY.enable(Feature.ALLOW_UNQUOTED_CONTROL_CHARS);
+ // Enabled to accept quoting of all character backslash qooting mechanism
+ JSON_FACTORY.enable(Feature.ALLOW_BACKSLASH_ESCAPING_ANY_CHARACTER);
}
private static final ObjectMapper MAPPER = new ObjectMapper(JSON_FACTORY);
private static final JavaType MAP_TYPE = TypeFactory.fromClass(Map.class);