Skip to content

Commit fe02df0

Browse files
liyafan82nealrichardson
liyafan82
authored and
nealrichardson
committed
ARROW-6863: [Java] Provide parallel searcher
For scenarios where the vector is large and the a low response time is required, we need to search the vector in parallel to improve the responsiveness. This issue tries to provide a parallel searcher for the equality semantics (the support for ordering semantics is not ready yet, as we need a way to distribute the comparator). The implementation is based on multi-threading. Closes apache#5631 from liyafan82/fly_1011_par and squashes the following commits: 25e86f4 <liyafan82> Refine Javadoc 5b21cc3 <liyafan82> Provide benchmark for the parallel searcher e735bcc <liyafan82> Provide parallel searcher Authored-by: liyafan82 <fan_li_ya@foxmail.com> Signed-off-by: Neal Richardson <neal.p.richardson@gmail.com>
1 parent 29ff4e4 commit fe02df0

4 files changed

Lines changed: 358 additions & 0 deletions

File tree

‎java/algorithm/src/main/java/org/apache/arrow/algorithm/search/ParallelSearcher.java‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.arrow.algorithm.search;
19+
20+
import java.util.concurrent.CompletableFuture;
21+
import java.util.concurrent.ExecutionException;
22+
import java.util.concurrent.ExecutorService;
23+
24+
import org.apache.arrow.algorithm.sort.VectorValueComparator;
25+
import org.apache.arrow.vector.ValueVector;
26+
import org.apache.arrow.vector.compare.Range;
27+
import org.apache.arrow.vector.compare.RangeEqualsVisitor;
28+
29+
/**
30+
* Search for a value in the vector by multiple threads.
31+
* This is often used in scenarios where the vector is large or
32+
* low response time is required.
33+
* @param <V> the vector type.
34+
*/
35+
public class ParallelSearcher<V extends ValueVector> {
36+
37+
/**
38+
* The target vector to search.
39+
*/
40+
private final V vector;
41+
42+
/**
43+
* The thread pool.
44+
*/
45+
private final ExecutorService threadPool;
46+
47+
/**
48+
* The number of threads to use.
49+
*/
50+
private final int numThreads;
51+
52+
/**
53+
* The position of the key in the target vector, if any.
54+
*/
55+
private int keyPosition = -1;
56+
57+
/**
58+
* Constructs a parallel searcher.
59+
* @param vector the vector to search.
60+
* @param threadPool the thread pool to use.
61+
* @param numThreads the number of threads to use.
62+
*/
63+
public ParallelSearcher(V vector, ExecutorService threadPool, int numThreads) {
64+
this.vector = vector;
65+
this.threadPool = threadPool;
66+
this.numThreads = numThreads;
67+
}
68+
69+
/**
70+
* Search for the key in the target vector.
71+
* @param keyVector the vector containing the search key.
72+
* @param keyIndex the index of the search key in the key vector.
73+
* @return the position of a matched value in the target vector,
74+
* or -1 if none is found. Please note that if there are multiple
75+
* matches of the key in the target vector, this method makes no
76+
* guarantees about which instance is returned.
77+
* For an alternative search implementation that always finds the first match of the key,
78+
* see {@link VectorSearcher#linearSearch(ValueVector, VectorValueComparator, ValueVector, int)}.
79+
* @throws ExecutionException if an exception occurs in a thread.
80+
* @throws InterruptedException if a thread is interrupted.
81+
*/
82+
public int search(V keyVector, int keyIndex) throws ExecutionException, InterruptedException {
83+
keyPosition = -1;
84+
final int valueCount = vector.getValueCount();
85+
final CompletableFuture<Boolean>[] futures = new CompletableFuture[numThreads];
86+
for (int i = 0; i < futures.length; i++) {
87+
futures[i] = new CompletableFuture<>();
88+
}
89+
90+
for (int i = 0; i < numThreads; i++) {
91+
final int tid = i;
92+
threadPool.submit(() -> {
93+
// convert to long to avoid overflow
94+
int start = (int) (((long) valueCount) * tid / numThreads);
95+
int end = (int) ((long) valueCount) * (tid + 1) / numThreads;
96+
97+
if (start >= end) {
98+
// no data assigned to this task.
99+
futures[tid].complete(false);
100+
return;
101+
}
102+
103+
RangeEqualsVisitor visitor = new RangeEqualsVisitor(vector, keyVector, false);
104+
Range range = new Range(0, 0, 1);
105+
for (int pos = start; pos < end; pos++) {
106+
if (keyPosition != -1) {
107+
// the key has been found by another task
108+
futures[tid].complete(false);
109+
return;
110+
}
111+
range.setLeftStart(pos).setRightStart(keyIndex);
112+
if (visitor.rangeEquals(range)) {
113+
keyPosition = pos;
114+
futures[tid].complete(true);
115+
return;
116+
}
117+
}
118+
119+
// no match value is found.
120+
futures[tid].complete(false);
121+
});
122+
}
123+
124+
CompletableFuture.allOf(futures).get();
125+
return keyPosition;
126+
}
127+
}

‎java/algorithm/src/test/java/org/apache/arrow/algorithm/search/TestParallelSearcher.java‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.arrow.algorithm.search;
19+
20+
import static org.junit.jupiter.api.Assertions.assertEquals;
21+
22+
import java.util.concurrent.ExecutionException;
23+
import java.util.concurrent.ExecutorService;
24+
import java.util.concurrent.Executors;
25+
26+
import org.apache.arrow.memory.BufferAllocator;
27+
import org.apache.arrow.memory.RootAllocator;
28+
import org.apache.arrow.vector.IntVector;
29+
import org.apache.arrow.vector.VarCharVector;
30+
import org.junit.After;
31+
import org.junit.Before;
32+
import org.junit.Test;
33+
34+
/**
35+
* Test cases for {@link ParallelSearcher}.
36+
*/
37+
public class TestParallelSearcher {
38+
39+
private static final int THREAD_COUNT = 10;
40+
41+
private static final int VECTOR_LENGTH = 10000;
42+
43+
private BufferAllocator allocator;
44+
45+
private ExecutorService threadPool;
46+
47+
@Before
48+
public void prepare() {
49+
allocator = new RootAllocator(1024 * 1024);
50+
threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
51+
}
52+
53+
@After
54+
public void shutdown() {
55+
allocator.close();
56+
threadPool.shutdown();
57+
}
58+
59+
@Test
60+
public void testParallelIntSearch() throws ExecutionException, InterruptedException {
61+
try (IntVector targetVector = new IntVector("targetVector", allocator);
62+
IntVector keyVector = new IntVector("keyVector", allocator)) {
63+
targetVector.allocateNew(VECTOR_LENGTH);
64+
keyVector.allocateNew(VECTOR_LENGTH);
65+
66+
for (int i = 0; i < VECTOR_LENGTH; i++) {
67+
targetVector.set(i, i);
68+
keyVector.set(i, i * 2);
69+
}
70+
targetVector.setValueCount(VECTOR_LENGTH);
71+
keyVector.setValueCount(VECTOR_LENGTH);
72+
73+
ParallelSearcher<IntVector> searcher = new ParallelSearcher<>(targetVector, threadPool, THREAD_COUNT);
74+
for (int i = 0; i < VECTOR_LENGTH; i++) {
75+
int pos = searcher.search(keyVector, i);
76+
if (i * 2 < VECTOR_LENGTH) {
77+
assertEquals(i * 2, pos);
78+
} else {
79+
assertEquals(-1, pos);
80+
}
81+
}
82+
}
83+
}
84+
85+
@Test
86+
public void testParallelStringSearch() throws ExecutionException, InterruptedException {
87+
try (VarCharVector targetVector = new VarCharVector("targetVector", allocator);
88+
VarCharVector keyVector = new VarCharVector("keyVector", allocator)) {
89+
targetVector.allocateNew(VECTOR_LENGTH);
90+
keyVector.allocateNew(VECTOR_LENGTH);
91+
92+
for (int i = 0; i < VECTOR_LENGTH; i++) {
93+
targetVector.setSafe(i, String.valueOf(i).getBytes());
94+
keyVector.setSafe(i, String.valueOf(i * 2).getBytes());
95+
}
96+
targetVector.setValueCount(VECTOR_LENGTH);
97+
keyVector.setValueCount(VECTOR_LENGTH);
98+
99+
ParallelSearcher<VarCharVector> searcher = new ParallelSearcher<>(targetVector, threadPool, THREAD_COUNT);
100+
for (int i = 0; i < VECTOR_LENGTH; i++) {
101+
int pos = searcher.search(keyVector, i);
102+
if (i * 2 < VECTOR_LENGTH) {
103+
assertEquals(i * 2, pos);
104+
} else {
105+
assertEquals(-1, pos);
106+
}
107+
}
108+
}
109+
}
110+
}

‎java/performance/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,12 @@
7777
<artifactId>arrow-jdbc</artifactId>
7878
<version>${project.version}</version>
7979
</dependency>
80+
<dependency>
81+
<groupId>org.apache.arrow</groupId>
82+
<artifactId>arrow-algorithm</artifactId>
83+
<version>1.0.0-SNAPSHOT</version>
84+
<scope>test</scope>
85+
</dependency>
8086
</dependencies>
8187

8288
<properties>

‎java/performance/src/test/java/org/apache/arrow/algorithm/search/ParallelSearcherBenchmarks.java‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.arrow.algorithm.search;
19+
20+
import java.util.concurrent.ExecutorService;
21+
import java.util.concurrent.Executors;
22+
import java.util.concurrent.TimeUnit;
23+
24+
import org.apache.arrow.memory.BaseAllocator;
25+
import org.apache.arrow.memory.RootAllocator;
26+
import org.apache.arrow.vector.IntVector;
27+
import org.openjdk.jmh.annotations.Benchmark;
28+
import org.openjdk.jmh.annotations.BenchmarkMode;
29+
import org.openjdk.jmh.annotations.Level;
30+
import org.openjdk.jmh.annotations.Mode;
31+
import org.openjdk.jmh.annotations.OutputTimeUnit;
32+
import org.openjdk.jmh.annotations.Param;
33+
import org.openjdk.jmh.annotations.Scope;
34+
import org.openjdk.jmh.annotations.Setup;
35+
import org.openjdk.jmh.annotations.State;
36+
import org.openjdk.jmh.annotations.TearDown;
37+
import org.openjdk.jmh.runner.Runner;
38+
import org.openjdk.jmh.runner.RunnerException;
39+
import org.openjdk.jmh.runner.options.Options;
40+
import org.openjdk.jmh.runner.options.OptionsBuilder;
41+
42+
/**
43+
* Benchmarks for {@link ParallelSearcher}.
44+
*/
45+
public class ParallelSearcherBenchmarks {
46+
47+
private static final int VECTOR_LENGTH = 1024 * 1024;
48+
49+
/**
50+
* State object for the benchmarks.
51+
*/
52+
@State(Scope.Benchmark)
53+
public static class SearchState {
54+
55+
@Param({"1", "2", "5", "10", "20", "50", "100"})
56+
int numThreads;
57+
58+
BaseAllocator allocator;
59+
60+
ExecutorService threadPool;
61+
62+
IntVector targetVector;
63+
64+
IntVector keyVector;
65+
66+
ParallelSearcher<IntVector> searcher;
67+
68+
@Setup(Level.Trial)
69+
public void prepare() {
70+
allocator = new RootAllocator(Integer.MAX_VALUE);
71+
targetVector = new IntVector("target vector", allocator);
72+
targetVector.allocateNew(VECTOR_LENGTH);
73+
keyVector = new IntVector("key vector", allocator);
74+
keyVector.allocateNew(1);
75+
threadPool = Executors.newFixedThreadPool(numThreads);
76+
77+
for (int i = 0; i < VECTOR_LENGTH; i++) {
78+
targetVector.set(i, i);
79+
}
80+
targetVector.setValueCount(VECTOR_LENGTH);
81+
82+
keyVector.set(0, VECTOR_LENGTH / 3);
83+
keyVector.setValueCount(1);
84+
}
85+
86+
@Setup(Level.Invocation)
87+
public void prepareInvoke() {
88+
searcher = new ParallelSearcher<>(targetVector, threadPool, numThreads);
89+
}
90+
91+
@TearDown(Level.Trial)
92+
public void tearDownState() {
93+
targetVector.close();
94+
keyVector.close();
95+
allocator.close();
96+
threadPool.shutdown();
97+
}
98+
}
99+
100+
@Benchmark
101+
@BenchmarkMode(Mode.AverageTime)
102+
@OutputTimeUnit(TimeUnit.MICROSECONDS)
103+
public void searchBenchmark(SearchState state) throws Exception {
104+
state.searcher.search(state.keyVector, 0);
105+
}
106+
107+
public static void main(String[] args) throws RunnerException {
108+
Options opt = new OptionsBuilder()
109+
.include(ParallelSearcherBenchmarks.class.getSimpleName())
110+
.forks(1)
111+
.build();
112+
113+
new Runner(opt).run();
114+
}
115+
}

0 commit comments

Comments
 (0)