You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by GitBox <gi...@apache.org> on 2020/07/20 07:53:16 UTC
[GitHub] [hive] abstractdog opened a new pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
abstractdog opened a new pull request #1280:
URL: https://github.com/apache/hive/pull/1280
…AFBloomFilterMerge
Change-Id: I235248ad327b0cea91e637e74a0c67720710737e
## NOTICE
Please create an issue in ASF JIRA before opening a pull request,
and you need to set the title of the pull request which starts with
the corresponding JIRA issue number. (e.g. HIVE-XXXXX: Fix a typo in YYY)
For more details, please see https://cwiki.apache.org/confluence/display/Hive/HowToContribute
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458683069
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -506,18 +505,19 @@ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, in
}
private static class BloomFilterMergeWorker implements Runnable {
- Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ ArrayBlockingQueue<ElementWrapper> queue;
Review comment:
agreed, I'll change it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUDAFBloomFilterMerge
Posted by GitBox <gi...@apache.org>.
abstractdog commented on pull request #1280:
URL: https://github.com/apache/hive/pull/1280#issuecomment-680846635
all checks passed, pushing this to master, thanks for the review for all you guys
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] mustafaiman commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
mustafaiman commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r470310851
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -1126,6 +1137,7 @@ protected void initializeOp(Configuration hconf) throws HiveException {
VectorAggregateExpression vecAggrExpr = null;
try {
vecAggrExpr = ctor.newInstance(vecAggrDesc);
+ vecAggrExpr.withConf(hconf);
Review comment:
I think making `VectorUDAFBloomFilterMerge` construction a special case and supplying the single int to that constructor is much cleaner. While trying to avoid that specialization, you are injecting the conf object to all the other classes.
I specifically despise passing conf object around in Hive as it is abused so much in every part of the codebase. I'd prefer the other way but I won't insist on it. It is not a big deal for this patch.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458771841
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -362,16 +378,178 @@ public static void mergeBloomFilterBytes(
// Just bitwise-OR the bits together - size/# functions should be the same,
// rest of the data is serialized long values for the bitset which are supposed to be bitwise-ORed.
- for (int idx = START_OF_SERIALIZED_LONGS; idx < bf1Length; ++idx) {
+ for (int idx = mergeStart; idx < mergeEnd; ++idx) {
bf1Bytes[bf1Start + idx] |= bf2Bytes[bf2Start + idx];
}
}
+ public static void mergeBloomFilterBytesFromInputColumn(
+ byte[] bf1Bytes, int bf1Start, int bf1Length, long bf1ExpectedEntries,
+ BytesColumnVector inputColumn, int batchSize, boolean selectedInUse, int[] selected, int numThreads) {
Review comment:
batchSize is I assume bfSize? maybe rename to something more explicit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] mustafaiman commented on pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
mustafaiman commented on pull request #1280:
URL: https://github.com/apache/hive/pull/1280#issuecomment-673894751
@abstractdog I missed that call. I think that covers it.
Good work.
+1
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469099535
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -1126,6 +1137,7 @@ protected void initializeOp(Configuration hconf) throws HiveException {
VectorAggregateExpression vecAggrExpr = null;
try {
vecAggrExpr = ctor.newInstance(vecAggrDesc);
+ vecAggrExpr.withConf(hconf);
Review comment:
1. constructor: first I tried to pass it to constructor, but that breaks compatibility with every other subclasses of VectorAggregateExpression, if I use ctor.newInstance(vecAggrDesc, hconf), I need to add that constructor to all subclasses, because they don't inherit this ctor from VectorAggregateExpression...withConf can solve this, let me know about better ways
2. single int: this config is specific to VectorUDAFBloomFilterMerge, so I believe I should not pass it through a constructor to every VectorAggregateExpression, and I didn't want to go for an instanceof hack for a cast + specific call
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r470419947
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -1126,6 +1137,7 @@ protected void initializeOp(Configuration hconf) throws HiveException {
VectorAggregateExpression vecAggrExpr = null;
try {
vecAggrExpr = ctor.newInstance(vecAggrDesc);
+ vecAggrExpr.withConf(hconf);
Review comment:
Sadly, I need to agree with conf abusing in (hive) codebase :) somehow I don't really like instanceof stuff here, only for a single expression, moreover, I wanted to find a general way to provide some configuration to expressions, as this patch showed that they might need that (in the future). On the other hand, explicitly calling a specific constructor for different types could be a kind of documentation in one place about "how to instantiate" these expressions. I'm about to refactor this logic to a separate method in VectorGroupByOperator and let this patch go!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469761824
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -252,6 +258,13 @@ protected VectorAggregationBufferRow allocateAggregationBuffer() throws HiveExce
return bufferSet;
}
+ protected void finishAggregators(boolean aborted) {
Review comment:
I'll take care of this in next patch
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458761137
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -36,7 +44,7 @@
*
* This implementation has much lesser L1 data cache misses than {@link BloomFilter}.
Review comment:
I am afraid changing the logic of the implementation will also affect the cache effectiveness/behaviour -- would it make sense to add a JMH test here as well?
We could even save the bench-results in a log to keep track of whats happening
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458678926
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -506,18 +505,19 @@ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, in
}
private static class BloomFilterMergeWorker implements Runnable {
- Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ ArrayBlockingQueue<ElementWrapper> queue;
private ExecutorService executor;
private byte[] bfAggregation;
private int bfAggregationStart;
private int bfAggregationLength;
- public BloomFilterMergeWorker(ExecutorService executor, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
+ public BloomFilterMergeWorker(ExecutorService executor, int batchSize, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
this.executor = executor;
Review comment:
@belugabehr : sorry I cannot get your point, what do you mean?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] mustafaiman commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
mustafaiman commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r468748810
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
Review comment:
I see this method is used only for logging. What is the benefit of having that log? I am asking because if we get rid of this method, we can get rid of isMerging atomic variable.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469300187
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
Review comment:
Its just not clear to me what this number depends on -- any chance this could greatly vary at a non-test scenario?
* If not VectorizedRowBatch.DEFAULT_SIZE * 2 is fine (we should just add a comment why this number was picked)
* If yes we could even have an unbounded queue here right?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458771841
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -362,16 +378,178 @@ public static void mergeBloomFilterBytes(
// Just bitwise-OR the bits together - size/# functions should be the same,
// rest of the data is serialized long values for the bitset which are supposed to be bitwise-ORed.
- for (int idx = START_OF_SERIALIZED_LONGS; idx < bf1Length; ++idx) {
+ for (int idx = mergeStart; idx < mergeEnd; ++idx) {
bf1Bytes[bf1Start + idx] |= bf2Bytes[bf2Start + idx];
}
}
+ public static void mergeBloomFilterBytesFromInputColumn(
+ byte[] bf1Bytes, int bf1Start, int bf1Length, long bf1ExpectedEntries,
+ BytesColumnVector inputColumn, int batchSize, boolean selectedInUse, int[] selected, int numThreads) {
Review comment:
batchSize is I assume bfSize?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469109686
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorAggregateExpression.java
##########
@@ -20,24 +20,25 @@
import java.io.Serializable;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.type.DataTypePhysicalVariation;
import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorAggregationBufferRow;
import org.apache.hadoop.hive.ql.exec.vector.VectorAggregationDesc;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression;
import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.plan.AggregationDesc;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
/**
* Base class for aggregation expressions.
*/
public abstract class VectorAggregateExpression implements Serializable {
-
+ protected final Logger LOG = LoggerFactory.getLogger(getClass().getName());
Review comment:
personally, I don't really like protected static Logger, because subclasses won't show the actual class name (only the parent), but in this case, this LOG is not used in VectorUDAFBloomFilterMerge at all, it's useless leftover, I'm going to remove it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] belugabehr commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
belugabehr commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r459437391
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -45,6 +48,8 @@
* This implementation has much lesser L1 data cache misses than {@link BloomFilter}.
*/
public class BloomKFilter {
+ private static final Logger LOG = LoggerFactory.getLogger(BloomKFilter.class.getName());
Review comment:
Nit:: Does not require `.getName()`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458754144
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -4285,6 +4285,8 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
"Bloom filter should be of at max certain size to be effective"),
TEZ_BLOOM_FILTER_FACTOR("hive.tez.bloom.filter.factor", (float) 1.0,
"Bloom filter should be a multiple of this factor with nDV"),
+ TEZ_BLOOM_FILTER_MERGE_THREADS("hive.tez.bloom.filter.merge.threads", 8,
Review comment:
Any chance we can get rid of the extra configuration?
I really hope one day we will starting removing configuration instead of adding new :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469110972
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -4330,6 +4330,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
"Bloom filter should be of at max certain size to be effective"),
TEZ_BLOOM_FILTER_FACTOR("hive.tez.bloom.filter.factor", (float) 1.0,
"Bloom filter should be a multiple of this factor with nDV"),
+ TEZ_BLOOM_FILTER_MERGE_THREADS("hive.tez.bloom.filter.merge.threads", 1,
+ "How many threads are used for merging bloom filters?\n"
Review comment:
agree, adding that
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458688040
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -527,20 +527,19 @@ public void add(ElementWrapper wrapper) {
@Override
public void run() {
while (!executor.isTerminated() && !queue.isEmpty()) {
Review comment:
thanks, your comment regarding isTerminated check seems valid, let me think that over
queue.isEmpty() check is needed I think, please a take a look at the implementation above, every thread uses its own queue (which is filled in advance), there is no concurrency between them, so queue.isEmpty() is a simple check for every thread that "whether I have further things to do, or I can quit", this way every thread will quit once they're ready, so executor.awaitTermination will return and we're done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog closed pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUDAFBloomFilterMerge
Posted by GitBox <gi...@apache.org>.
abstractdog closed pull request #1280:
URL: https://github.com/apache/hive/pull/1280
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469227267
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
Review comment:
I'm afraid it's not available in advance, bloom filters comes one by one, in my testing scenario it is ~1200 row batches, and every row batch contains 1 bloom filter
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469765404
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
Review comment:
if there are 1000 upstream mapper tasks (creating bloom filters), there will be 1000 rowbatches (=1000 bloom filters), for example on TPCDS 30GB there were 1000<x<2000...anyway, you're absolutely right, I don't want to take care of correct bounds, which is unpredictable, I've just chosen a wrong implementation...I'm going to change this to LinkedBlockingDeque and letting this size confusion go
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469107566
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
Review comment:
not really, workers.len is horizontal, which is related to how many parts a bloom filter is split into, but queue.length needs to hold the same portion of the incoming bloom filters, so 1000 incoming bloom filters means 1000 items in the queue, so it's vertical (however, hopefully, ~1000 items won't be present in the queue if it's busy working, but I needed to find a safe value)...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] belugabehr commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
belugabehr commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r457496282
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -527,20 +527,19 @@ public void add(ElementWrapper wrapper) {
@Override
public void run() {
while (!executor.isTerminated() && !queue.isEmpty()) {
Review comment:
A bit unrelated, but since you're touching this code. This check is completely useless:
```
while (!executor.isTerminated() && !queue.isEmpty()) {
...
}
```
I cannot think of many scenarios where the thread needs to check the state of its own `ExecutorService`. If the `ExecutorService` is terminated, it will Interrupt every thread in the pool and that should cause it to cease to run. Also, checking if the `Queue` is empty is improper. You will have two threads that check the state of the Queue (size = 1), see the same non-empty queue, and both try to read, even if there is only one item left. Both should just try to `take` and one will succeed and the other will fail.
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -527,20 +527,19 @@ public void add(ElementWrapper wrapper) {
@Override
public void run() {
while (!executor.isTerminated() && !queue.isEmpty()) {
- ElementWrapper currentBf = queue.poll();
+ ElementWrapper currentBf = null;
+ try {
+ currentBf = queue.take();
+ } catch (InterruptedException e) {
Review comment:
Do not ignore. An Interrupt means that it's time to exit.
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -506,18 +505,19 @@ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, in
}
private static class BloomFilterMergeWorker implements Runnable {
- Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ ArrayBlockingQueue<ElementWrapper> queue;
Review comment:
Use the generic `BlockingQueue` here.
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -506,18 +505,19 @@ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, in
}
private static class BloomFilterMergeWorker implements Runnable {
- Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ ArrayBlockingQueue<ElementWrapper> queue;
private ExecutorService executor;
private byte[] bfAggregation;
private int bfAggregationStart;
private int bfAggregationLength;
- public BloomFilterMergeWorker(ExecutorService executor, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
+ public BloomFilterMergeWorker(ExecutorService executor, int batchSize, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
this.executor = executor;
Review comment:
Do not capture this value.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469184919
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
Review comment:
fortunately we won't need this, I've eliminated with boolean return hack according to another PR comment
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469099535
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -1126,6 +1137,7 @@ protected void initializeOp(Configuration hconf) throws HiveException {
VectorAggregateExpression vecAggrExpr = null;
try {
vecAggrExpr = ctor.newInstance(vecAggrDesc);
+ vecAggrExpr.withConf(hconf);
Review comment:
1. constructor: first I tried to pass it to constructor, but that breaks compatibility with every other subclasses of VectorAggregateExpression, if I use ctor.newInstance(vecAggrDesc, hconf), I need to add that constructor to all subclasses, because they don't inherit this ctor from VectorAggregateExpression...withConf can solve this, let me know about better ways
2. single int: this config is specific to VectorUDAFBloomFilterMerge, I think I should pass it through a constructor to every VectorAggregateExpressio, and I didn't want to go for an instanceof hack for a cast + specific call
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on pull request #1280:
URL: https://github.com/apache/hive/pull/1280#issuecomment-669038358
@zabetak : let me grab the opportunity to thank you for your [JMH benchmarks](https://issues.apache.org/jira/browse/HIVE-23880?focusedCommentId=17163111&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-17163111)! it helped a lot, some of my findings from the last 2 weeks:
1. on cluster, JDK11 is better, in every scenario, we'll have to switch to that in LLAP daemons
2. more threads doesn't make any serious improvement <- that's the most important what I've found in the last two weeks...basically, my implementation was wrong, and the results got distorted by the improper usage of executor service (that's what is fixed in the new, squashed commit), so now, on the cluster I can see results which are in line with your JMH findings
3. removed automatic thread calculation: performance tests revealed that 1 thread is the most optimal, and can lead to serious improvements, this is something that cannot be measured from JMH easily because the advantage of 1 thread (which is the main task thread + 1 thread) is to decouple from the main thread, and let it handle other, probably CPU heavy stuff (waiting for inputs one by one, build vectorized row batches one by one, etc.), by this I reduced the task runtime by 50-60 seconds (170s -> 110s)
4. as agreed with @ashutoshc, I've left the support of multiple threads in the code, because we don't know if we can have the advantage of it later, and the split logic doesn't consume significant amount of resources...but I've set default 1 thread in HiveConf in order to let the user know that this is the recommended, optimal usage
cc: @pgaref , @ashutoshc
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r468513649
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -4330,6 +4330,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
"Bloom filter should be of at max certain size to be effective"),
TEZ_BLOOM_FILTER_FACTOR("hive.tez.bloom.filter.factor", (float) 1.0,
"Bloom filter should be a multiple of this factor with nDV"),
+ TEZ_BLOOM_FILTER_MERGE_THREADS("hive.tez.bloom.filter.merge.threads", 1,
+ "How many threads are used for merging bloom filters?\n"
Review comment:
The number of threads used variable is actually in **addition to tasks main threads** -- I would make this a bit clearer
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorAggregateExpression.java
##########
@@ -20,24 +20,25 @@
import java.io.Serializable;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.type.DataTypePhysicalVariation;
import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorAggregationBufferRow;
import org.apache.hadoop.hive.ql.exec.vector.VectorAggregationDesc;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression;
import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.plan.AggregationDesc;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
/**
* Base class for aggregation expressions.
*/
public abstract class VectorAggregateExpression implements Serializable {
-
+ protected final Logger LOG = LoggerFactory.getLogger(getClass().getName());
Review comment:
Should we make this static? Do we really want an instance per Expr?
PS: it also seems that we dont use it all below..
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -4330,6 +4330,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
"Bloom filter should be of at max certain size to be effective"),
TEZ_BLOOM_FILTER_FACTOR("hive.tez.bloom.filter.factor", (float) 1.0,
"Bloom filter should be a multiple of this factor with nDV"),
+ TEZ_BLOOM_FILTER_MERGE_THREADS("hive.tez.bloom.filter.merge.threads", 1,
+ "How many threads are used for merging bloom filters?\n"
+ + "-1: sanity check, it will fail if execution hits bloom filter merge codepath\n"
+ + " 0: feature is disabled\n"
Review comment:
feature disabled -- use only task main thread for BF merging
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
Review comment:
Could add a comment describing the return boolean value
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
Review comment:
Could also follow an alternative approach -- where we have an external isSelected Condition and we check for isNull in internal ones (similar to what we do elsewhere in the code)
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
Review comment:
Why VectorizedRowBatch.DEFAULT_SIZE * 2 aka 2048 ?
maybe workers.len * 2?
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
Review comment:
!inputColumn.isNull[j] can be merged with the above condition
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
Review comment:
Combine in a single loop?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469762209
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
+ }
+
+ public void add(ElementWrapper wrapper) {
+ queue.add(wrapper);
+ }
+
+ @Override
+ public void run() {
+ while (true) {
+ ElementWrapper currentBf = null;
+ try {
+ currentBf = queue.take();
+ // at this point we have a currentBf wrapper which contains the whole byte[] of the
+ // serialized bloomfilter, but we only want to merge a modified "start -> start+length"
+ // part of it, which is pointed by modifiedStart/modifiedLength fields by ElementWrapper
+ merge(currentBf);
+ } catch (InterruptedException e) {// Executor.shutdownNow() is called
+ if (!queue.isEmpty()){
+ LOG.debug(
+ "bloom filter merge was interrupted while processing and queue is still not empty"
+ + ", this is fine in case of shutdownNow");
+ }
+ while (!queue.isEmpty()) { // time to finish work if any
Review comment:
I'm implementing abort in the next patch, I agree it's worth taking care of this
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469186380
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
Review comment:
I see, is the BF num available to use here? Maybe BF num * 2 would be safe
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469109686
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorAggregateExpression.java
##########
@@ -20,24 +20,25 @@
import java.io.Serializable;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.type.DataTypePhysicalVariation;
import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorAggregationBufferRow;
import org.apache.hadoop.hive.ql.exec.vector.VectorAggregationDesc;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression;
import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.plan.AggregationDesc;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
/**
* Base class for aggregation expressions.
*/
public abstract class VectorAggregateExpression implements Serializable {
-
+ protected final Logger LOG = LoggerFactory.getLogger(getClass().getName());
Review comment:
personally, I don't really like protected static Logger, because subclasses won't show the actual class name (only the parent)
in this case, you're right, this LOG is not used in VectorUDAFBloomFilterMerge at all, it's useless leftover, I'm going to remove it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on pull request #1280:
URL: https://github.com/apache/hive/pull/1280#issuecomment-673892379
> @abstractdog
> I am almost ok with this patch. However I still dont understand how this integrates with `ProcessingModeHashAggregate`. Since there are multiple VectorAggregationBufferRows in hash mode, I think we should `finish` each of them as we process them. Otherwise, we pass to the next operator in the pipeline without completing the bloom filter. Also, since hash mode dynamically allocates and frees VectorAggregationBufferRows these `finish`es should happen as we deallocate each of them, rather than only at the end of the operator.
Good point. I was creating this patch by focusing on finishing buffers correctly, I think I've already taken care of by this, please take a look:
https://github.com/apache/hive/pull/1280/commits/0ada66534a937b8f4492d14f508903fa98402aed#diff-07c28d3f5c72db581b9cd4fa424a0ecbR675
As you can see, I'm calling finish before every instance of writeSingleRow. I'm assuming that writeSingleRow is a point where a buffer should be finished for writing. In ProcessingModeHashAggregate, the above part is enclosed in an iteration on buffers in flush method. Are you aware of any other places where I should finish a buffer?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r457402888
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -362,16 +379,178 @@ public static void mergeBloomFilterBytes(
// Just bitwise-OR the bits together - size/# functions should be the same,
// rest of the data is serialized long values for the bitset which are supposed to be bitwise-ORed.
- for (int idx = START_OF_SERIALIZED_LONGS; idx < bf1Length; ++idx) {
+ for (int idx = mergeStart; idx < mergeEnd; ++idx) {
bf1Bytes[bf1Start + idx] |= bf2Bytes[bf2Start + idx];
}
}
+ public static void mergeBloomFilterBytesFromInputColumn(
+ byte[] bf1Bytes, int bf1Start, int bf1Length, long bf1ExpectedEntries,
+ BytesColumnVector inputColumn, int batchSize, boolean selectedInUse, int[] selected, int numThreads) {
+ if (numThreads == 0) {
+ numThreads = Runtime.getRuntime().availableProcessors();
+ }
+ if (numThreads < 0) {
+ throw new RuntimeException("invalid number of threads: " + numThreads);
+ }
+
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+
+ BloomFilterMergeWorker[] workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(executor, bf1Bytes, bf1Start, bf1Length);
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+
+ executor.shutdown();
+ try {
+ executor.awaitTermination(3600, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public static byte[] getInitialBytes(long expectedEntries) {
+ ByteArrayOutputStream bytesOut = null;
+ try {
+ bytesOut = new ByteArrayOutputStream();
+ BloomKFilter bf = new BloomKFilter(expectedEntries);
+ BloomKFilter.serialize(bytesOut, bf);
+ return bytesOut.toByteArray();
+ } catch (Exception err) {
+ throw new IllegalArgumentException("Error creating aggregation buffer", err);
+ } finally {
+ IOUtils.closeStream(bytesOut);
+ }
+ }
+
+ public static class ElementWrapper {
+ public byte[] bytes;
+ public int start;
+ public int length;
+ public int modifiedStart;
+ public int modifiedLength;
+
+ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, int modifiedLength) {
+ this.bytes = bytes;
+ this.start = start;
+ this.length = length;
+ this.modifiedStart = modifiedStart;
+ this.modifiedLength = modifiedLength;
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ private ExecutorService executor;
+
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+
+ public BloomFilterMergeWorker(ExecutorService executor, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
+ this.executor = executor;
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ }
+
+ public void add(ElementWrapper wrapper) {
+ queue.add(wrapper);
+ }
+
+ @Override
+ public void run() {
+ while (!executor.isTerminated() && !queue.isEmpty()) {
+ ElementWrapper currentBf = queue.poll();
+ if (currentBf != null) {
+ // at this point we have a currentBf wrapper which contains the whole byte[] of the
+ // serialized bloomfilter, but we only want to merge a modified "start -> start+length"
+ // part of it, which is pointed by modifiedStart/modifiedLength fields by ElementWrapper
+ BloomKFilter.mergeBloomFilterBytes(bfAggregation, bfAggregationStart, bfAggregationLength,
+ currentBf.bytes, currentBf.start, currentBf.length, currentBf.modifiedStart,
+ currentBf.modifiedStart + currentBf.modifiedLength);
+ } else {
+ try {
+ Thread.sleep(10); // relax this thread while the queue is empty
Review comment:
thanks @rbalamohan! that makes sense, I'll change and test on cluster
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469120003
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
Review comment:
yeah, I missed this cleanup while mirroring the [original logic](https://github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java#L132-L191)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog closed pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUDAFBloomFilterMerge
Posted by GitBox <gi...@apache.org>.
abstractdog closed pull request #1280:
URL: https://github.com/apache/hive/pull/1280
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458960692
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -506,18 +505,19 @@ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, in
}
private static class BloomFilterMergeWorker implements Runnable {
- Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ ArrayBlockingQueue<ElementWrapper> queue;
private ExecutorService executor;
private byte[] bfAggregation;
private int bfAggregationStart;
private int bfAggregationLength;
- public BloomFilterMergeWorker(ExecutorService executor, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
+ public BloomFilterMergeWorker(ExecutorService executor, int batchSize, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
this.executor = executor;
Review comment:
sure, I misunderstood, somehow I thought you meant batchSize...yeah, we won't need executor reference if I remove executor.isTerminated() check
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] mustafaiman commented on pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
mustafaiman commented on pull request #1280:
URL: https://github.com/apache/hive/pull/1280#issuecomment-673767435
@abstractdog
I am almost ok with this patch. However I still dont understand how this integrates with `ProcessingModeHashAggregate`. Since there are multiple VectorAggregationBufferRows in hash mode, I think we should `finish` each of them as we process them. Otherwise, we pass to the next operator in the pipeline without completing the bloom filter. Also, since hash mode dynamically allocates and frees VectorAggregationBufferRows these `finish`es should happen as we deallocate each of them, rather than only at the end of the operator.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] belugabehr commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
belugabehr commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458795827
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -506,18 +505,19 @@ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, in
}
private static class BloomFilterMergeWorker implements Runnable {
- Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ ArrayBlockingQueue<ElementWrapper> queue;
private ExecutorService executor;
private byte[] bfAggregation;
private int bfAggregationStart;
private int bfAggregationLength;
- public BloomFilterMergeWorker(ExecutorService executor, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
+ public BloomFilterMergeWorker(ExecutorService executor, int batchSize, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
this.executor = executor;
Review comment:
@abstractdog Hey, ya, of course. This just goes with what I was saying about the relationship between Runnable/Thread/Callable and ExecutorService. The thread rarely needs a reference to its own ExecutorService. The only reason this was being capture was to support `!executor.isTerminated()` which is not the correct thing to do. Please remove the `this.executor` instance variable and/or change the constructor to not include it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469185446
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
Review comment:
good catch, I'm eliminating this check by initializing parallel behavior while initializing the aggregation buffer
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469112026
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
Review comment:
cood catch, moving them to a single loop
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
Review comment:
good catch, moving them to a single loop
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] zabetak commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
zabetak commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458811039
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -36,7 +44,7 @@
*
* This implementation has much lesser L1 data cache misses than {@link BloomFilter}.
Review comment:
Actually, I was having the same thoughts :) FYI: I did some JMH benhcmarks (outside Hive) some time ago. I am adapting them slightly and I will share some JMH results later today.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469229142
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -517,6 +532,10 @@ public void close(boolean aborted) throws HiveException {
}
+ //TODO: implement finishAggregators
+ protected void finishAggregators(boolean aborted) {
Review comment:
valid point, I need to recheck and learn how aggregators and aggregation buffers are paired together for a specific mode, it seems complicated for the first time
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rbalamohan commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
rbalamohan commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r457184037
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -362,16 +379,178 @@ public static void mergeBloomFilterBytes(
// Just bitwise-OR the bits together - size/# functions should be the same,
// rest of the data is serialized long values for the bitset which are supposed to be bitwise-ORed.
- for (int idx = START_OF_SERIALIZED_LONGS; idx < bf1Length; ++idx) {
+ for (int idx = mergeStart; idx < mergeEnd; ++idx) {
bf1Bytes[bf1Start + idx] |= bf2Bytes[bf2Start + idx];
}
}
+ public static void mergeBloomFilterBytesFromInputColumn(
+ byte[] bf1Bytes, int bf1Start, int bf1Length, long bf1ExpectedEntries,
+ BytesColumnVector inputColumn, int batchSize, boolean selectedInUse, int[] selected, int numThreads) {
+ if (numThreads == 0) {
+ numThreads = Runtime.getRuntime().availableProcessors();
+ }
+ if (numThreads < 0) {
+ throw new RuntimeException("invalid number of threads: " + numThreads);
+ }
+
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+
+ BloomFilterMergeWorker[] workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(executor, bf1Bytes, bf1Start, bf1Length);
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+
+ executor.shutdown();
+ try {
+ executor.awaitTermination(3600, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public static byte[] getInitialBytes(long expectedEntries) {
+ ByteArrayOutputStream bytesOut = null;
+ try {
+ bytesOut = new ByteArrayOutputStream();
+ BloomKFilter bf = new BloomKFilter(expectedEntries);
+ BloomKFilter.serialize(bytesOut, bf);
+ return bytesOut.toByteArray();
+ } catch (Exception err) {
+ throw new IllegalArgumentException("Error creating aggregation buffer", err);
+ } finally {
+ IOUtils.closeStream(bytesOut);
+ }
+ }
+
+ public static class ElementWrapper {
+ public byte[] bytes;
+ public int start;
+ public int length;
+ public int modifiedStart;
+ public int modifiedLength;
+
+ public ElementWrapper(byte[] bytes, int start, int length, int modifiedStart, int modifiedLength) {
+ this.bytes = bytes;
+ this.start = start;
+ this.length = length;
+ this.modifiedStart = modifiedStart;
+ this.modifiedLength = modifiedLength;
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ Queue<ElementWrapper> queue = new LinkedBlockingDeque<>();
+ private ExecutorService executor;
+
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+
+ public BloomFilterMergeWorker(ExecutorService executor, byte[] bfAggregation, int bfAggregationStart, int bfAggregationLength) {
+ this.executor = executor;
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ }
+
+ public void add(ElementWrapper wrapper) {
+ queue.add(wrapper);
+ }
+
+ @Override
+ public void run() {
+ while (!executor.isTerminated() && !queue.isEmpty()) {
+ ElementWrapper currentBf = queue.poll();
+ if (currentBf != null) {
+ // at this point we have a currentBf wrapper which contains the whole byte[] of the
+ // serialized bloomfilter, but we only want to merge a modified "start -> start+length"
+ // part of it, which is pointed by modifiedStart/modifiedLength fields by ElementWrapper
+ BloomKFilter.mergeBloomFilterBytes(bfAggregation, bfAggregationStart, bfAggregationLength,
+ currentBf.bytes, currentBf.start, currentBf.length, currentBf.modifiedStart,
+ currentBf.modifiedStart + currentBf.modifiedLength);
+ } else {
+ try {
+ Thread.sleep(10); // relax this thread while the queue is empty
Review comment:
Why is this needed? "splitVectorAcrossWorkers" seems to be adding all data before submitting to threadPool. IAC, it would be good to try with "queue::take()" (with additional change) which is a blocking call, to avoid "Thread.sleep".
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469102366
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
Review comment:
yeah, only for logging, it was for validating my executor shutdown correctness...that can be misleading, I'm removing it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] mustafaiman commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
mustafaiman commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r468723475
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -252,6 +258,13 @@ protected VectorAggregationBufferRow allocateAggregationBuffer() throws HiveExce
return bufferSet;
}
+ protected void finishAggregators(boolean aborted) {
Review comment:
Instead of `finishAggregators`, can you make this method default `close` method for `ProcessingModeBase` and call `super.close(boolean)` from close methods of appropriate subclasses. That way common finalization code would be in `close` of common super class and specific finalization code would be in `close` method of each subclass.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -517,6 +532,10 @@ public void close(boolean aborted) throws HiveException {
}
+ //TODO: implement finishAggregators
+ protected void finishAggregators(boolean aborted) {
Review comment:
What about this mode? Seems not complete.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -1126,6 +1137,7 @@ protected void initializeOp(Configuration hconf) throws HiveException {
VectorAggregateExpression vecAggrExpr = null;
try {
vecAggrExpr = ctor.newInstance(vecAggrDesc);
+ vecAggrExpr.withConf(hconf);
Review comment:
Why is `withConf` a seperate method? Conf should be a parameter to VectorAggregateExpression's constructor.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
+ return; // this will leave isParallel=false
+ }
+ isParallel = true;
+ executor = Executors.newFixedThreadPool(numThreads);
+
+ workers = new BloomFilterMergeWorker[numThreads];
+ for (int f = 0; f < numThreads; f++) {
+ workers[f] = new BloomFilterMergeWorker(bfBytes, 0, bfBytes.length);
+ }
+
+ for (int f = 0; f < numThreads; f++) {
+ executor.submit(workers[f]);
+ }
+ }
+
+ public int getNumberOfWaitingMergeTasks(){
+ int size = 0;
+ for (BloomFilterMergeWorker w : workers){
+ size += w.queue.size();
+ }
+ return size;
+ }
+
+ public int getNumberOfMergingWorkers() {
+ int working = 0;
+ for (BloomFilterMergeWorker w : workers) {
+ if (w.isMerging.get()) {
+ working += 1;
+ }
+ }
+ return working;
+ }
+
+ private static void splitVectorAcrossWorkers(BloomFilterMergeWorker[] workers, byte[] bytes,
+ int start, int length) {
+ if (bytes == null || length == 0) {
+ return;
+ }
+ /*
+ * This will split a byte[] across workers as below:
+ * let's say there are 10 workers for 7813 bytes, in this case
+ * length: 7813, elementPerBatch: 781
+ * bytes assigned to workers: inclusive lower bound, exclusive upper bound
+ * 1. worker: 5 -> 786
+ * 2. worker: 786 -> 1567
+ * 3. worker: 1567 -> 2348
+ * 4. worker: 2348 -> 3129
+ * 5. worker: 3129 -> 3910
+ * 6. worker: 3910 -> 4691
+ * 7. worker: 4691 -> 5472
+ * 8. worker: 5472 -> 6253
+ * 9. worker: 6253 -> 7034
+ * 10. worker: 7034 -> 7813 (last element per batch is: 779)
+ *
+ * This way, a particular worker will be given with the same part
+ * of all bloom filters along with the shared base bloom filter,
+ * so the bitwise OR function will not be a subject of threading/sync issues.
+ */
+ int elementPerBatch =
+ (int) Math.ceil((double) (length - START_OF_SERIALIZED_LONGS) / workers.length);
+
+ for (int w = 0; w < workers.length; w++) {
+ int modifiedStart = START_OF_SERIALIZED_LONGS + w * elementPerBatch;
+ int modifiedLength = (w == workers.length - 1)
+ ? length - (START_OF_SERIALIZED_LONGS + w * elementPerBatch) : elementPerBatch;
+
+ ElementWrapper wrapper =
+ new ElementWrapper(bytes, start, length, modifiedStart, modifiedLength);
+ workers[w].add(wrapper);
+ }
+ }
+
+ public void shutdownAndWaitForMergeTasks() {
+ /**
+ * Executor.shutdownNow() is supposed to send Thread.interrupt to worker threads, and they are
+ * supposed to finish their work.
+ */
+ executor.shutdownNow();
+ try {
+ executor.awaitTermination(180, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Bloom filter merge is interrupted while waiting to finish, this is unexpected",
+ e);
+ }
+ }
+ }
+
+ private static class BloomFilterMergeWorker implements Runnable {
+ private BlockingQueue<ElementWrapper> queue;
+ private byte[] bfAggregation;
+ private int bfAggregationStart;
+ private int bfAggregationLength;
+ AtomicBoolean isMerging = new AtomicBoolean(false);
+
+ public BloomFilterMergeWorker(byte[] bfAggregation, int bfAggregationStart,
+ int bfAggregationLength) {
+ this.bfAggregation = bfAggregation;
+ this.bfAggregationStart = bfAggregationStart;
+ this.bfAggregationLength = bfAggregationLength;
+ this.queue = new ArrayBlockingQueue<>(VectorizedRowBatch.DEFAULT_SIZE * 2);
+ }
+
+ public void add(ElementWrapper wrapper) {
+ queue.add(wrapper);
+ }
+
+ @Override
+ public void run() {
+ while (true) {
+ ElementWrapper currentBf = null;
+ try {
+ currentBf = queue.take();
+ // at this point we have a currentBf wrapper which contains the whole byte[] of the
+ // serialized bloomfilter, but we only want to merge a modified "start -> start+length"
+ // part of it, which is pointed by modifiedStart/modifiedLength fields by ElementWrapper
+ merge(currentBf);
+ } catch (InterruptedException e) {// Executor.shutdownNow() is called
+ if (!queue.isEmpty()){
+ LOG.debug(
+ "bloom filter merge was interrupted while processing and queue is still not empty"
+ + ", this is fine in case of shutdownNow");
+ }
+ while (!queue.isEmpty()) { // time to finish work if any
Review comment:
What if the operator was aborted? Do we still want to continue processing in that case? I am not sure how heavy an operation this is. If it is a short operation in all cases, it is okay to not have an abort path. Otherwise, I think there should be an abort path where we do not bother completing operations.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -1126,6 +1137,7 @@ protected void initializeOp(Configuration hconf) throws HiveException {
VectorAggregateExpression vecAggrExpr = null;
try {
vecAggrExpr = ctor.newInstance(vecAggrDesc);
+ vecAggrExpr.withConf(hconf);
Review comment:
Furthermore, the conf object is used for only a single config option: TEZ_BLOOM_FILTER_MERGE_THREADS . Instead of passing the config around, we should extract the value here and just pass a single int.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
+ numThreads = conf.getInt(HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.varname,
+ HiveConf.ConfVars.TEZ_BLOOM_FILTER_MERGE_THREADS.defaultIntVal);
+ LOG.info("Number of threads used for bloom filter merge: {}", numThreads);
+
+ if (numThreads < 0) {
+ throw new RuntimeException(
+ "invalid number of threads for bloom filter merge: " + numThreads);
+ }
+ if (numThreads == 0) { // disable parallel feature
Review comment:
The same check appears in `mergeBloomFilterBytesFromInputColumn`. I feel like these checks should have happened when we initialized VectorUDAFBloomFilterMerge. We should not be checking if parallel processing was enabled every time aggregation is called.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
Review comment:
`batchSize` is unused
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469186450
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/expressions/aggregates/VectorUDAFBloomFilterMerge.java
##########
@@ -77,6 +75,211 @@ public void reset() {
// Do not change the initial bytes which contain NumHashFunctions/NumBits!
Arrays.fill(bfBytes, BloomKFilter.START_OF_SERIALIZED_LONGS, bfBytes.length, (byte) 0);
}
+
+ public boolean mergeBloomFilterBytesFromInputColumn(BytesColumnVector inputColumn,
+ int batchSize, boolean selectedInUse, int[] selected, Configuration conf) {
+ // already set in previous iterations, no need to call initExecutor again
+ if (numThreads == 0) {
+ return false;
+ }
+ if (executor == null) {
+ initExecutor(conf, batchSize);
+ if (!isParallel) {
+ return false;
+ }
+ }
+
+ // split every bloom filter (represented by a part of a byte[]) across workers
+ for (int j = 0; j < batchSize; j++) {
+ if (!selectedInUse && inputColumn.noNulls) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ } else if (!selectedInUse) {
+ if (!inputColumn.isNull[j]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[j], inputColumn.start[j],
+ inputColumn.length[j]);
+ }
+ } else if (inputColumn.noNulls) {
+ int i = selected[j];
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ } else {
+ int i = selected[j];
+ if (!inputColumn.isNull[i]) {
+ splitVectorAcrossWorkers(workers, inputColumn.vector[i], inputColumn.start[i],
+ inputColumn.length[i]);
+ }
+ }
+ }
+
+ return true;
+ }
+
+ private void initExecutor(Configuration conf, int batchSize) {
Review comment:
right, I'll remove
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469229142
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByOperator.java
##########
@@ -517,6 +532,10 @@ public void close(boolean aborted) throws HiveException {
}
+ //TODO: implement finishAggregators
+ protected void finishAggregators(boolean aborted) {
Review comment:
valid point, I need to recheck and learn how aggregators and aggretation buffers are paired together for a specific mode, it seems complicated for the first time
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r469111595
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -4330,6 +4330,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
"Bloom filter should be of at max certain size to be effective"),
TEZ_BLOOM_FILTER_FACTOR("hive.tez.bloom.filter.factor", (float) 1.0,
"Bloom filter should be a multiple of this factor with nDV"),
+ TEZ_BLOOM_FILTER_MERGE_THREADS("hive.tez.bloom.filter.merge.threads", 1,
+ "How many threads are used for merging bloom filters?\n"
+ + "-1: sanity check, it will fail if execution hits bloom filter merge codepath\n"
+ + " 0: feature is disabled\n"
Review comment:
agree, added
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458962698
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -4285,6 +4285,8 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
"Bloom filter should be of at max certain size to be effective"),
TEZ_BLOOM_FILTER_FACTOR("hive.tez.bloom.filter.factor", (float) 1.0,
"Bloom filter should be a multiple of this factor with nDV"),
+ TEZ_BLOOM_FILTER_MERGE_THREADS("hive.tez.bloom.filter.merge.threads", 8,
Review comment:
agree, you'll see that the new version of PR will default to 0 which causes automatic thread number calculation
but I would leave this option as an "expert setting"...e.g.: explicitly setting a value for testing purposes, or setting -1 for sanity check (it will fail with negative values, so you can double-check if bloom filter merge really happens in your query)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458678445
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -527,20 +527,19 @@ public void add(ElementWrapper wrapper) {
@Override
public void run() {
while (!executor.isTerminated() && !queue.isEmpty()) {
- ElementWrapper currentBf = queue.poll();
+ ElementWrapper currentBf = null;
+ try {
+ currentBf = queue.take();
+ } catch (InterruptedException e) {
Review comment:
right, I'll do a simple method return with an LOG.warn to have some trace in the logs
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] abstractdog commented on a change in pull request #1280: HIVE-23880: Bloom filters can be merged in a parallel way in VectorUD…
Posted by GitBox <gi...@apache.org>.
abstractdog commented on a change in pull request #1280:
URL: https://github.com/apache/hive/pull/1280#discussion_r458967592
##########
File path: storage-api/src/java/org/apache/hive/common/util/BloomKFilter.java
##########
@@ -362,16 +378,178 @@ public static void mergeBloomFilterBytes(
// Just bitwise-OR the bits together - size/# functions should be the same,
// rest of the data is serialized long values for the bitset which are supposed to be bitwise-ORed.
- for (int idx = START_OF_SERIALIZED_LONGS; idx < bf1Length; ++idx) {
+ for (int idx = mergeStart; idx < mergeEnd; ++idx) {
bf1Bytes[bf1Start + idx] |= bf2Bytes[bf2Start + idx];
}
}
+ public static void mergeBloomFilterBytesFromInputColumn(
+ byte[] bf1Bytes, int bf1Start, int bf1Length, long bf1ExpectedEntries,
+ BytesColumnVector inputColumn, int batchSize, boolean selectedInUse, int[] selected, int numThreads) {
Review comment:
batchSize is not related to the bloom filter instances itself, it reflects the number of bloom filters in the vectorized row batch...bfSize is more reminds the byte size of a single bloom filter, which is bf1Length
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org