jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!1060 Heuristic Index: support using mmap to hold bloom index's filter 

Merge pull request !1060 from peiwangdb/bloom-to-disk
This commit is contained in:
i-robot 2021-08-25 19:15:12 +00:00 committed by Gitee
parent 88aad1965f 07af4821eb
commit 796b73afca
11 changed files with 582 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
hetu-docs/en/indexer/bloom.md
View File

@ -48,6 +48,16 @@ data is being filtered on the column and `phone` column has a high cardinality.
> in most usecases. If the index is too large, this value can be increased
> e.g. 0.05.
### `bloom.mmapEnabled`
> - **Type:** `Boolean`
> - **Default value:** `true`
>
> Control if Memory-Mapped File (mmap) should be used while reading the Bloom Index.
> Enabling this value will cache the Bloom Index to local disk instead of in-memory during reading.
> This will reduce memory consumption but will result in slightly lower performance.
## Examples
**Creating index:**
@ -56,6 +66,7 @@ create index idx using bloom on hive.hindex.users (id);
create index idx using bloom on hive.hindex.users (id) where regionkey=1;
create index idx using bloom on hive.hindex.users (id) where regionkey in (3, 1);
create index idx using bloom on hive.hindex.users (id) WITH ("bloom.fpp" = '0.001');
create index idx using bloom on hive.hindex.users (id) WITH ("bloom.mmapEnabled" = false);
```
* assuming users table is partitioned on `regionkey`

10
hetu-docs/zh/indexer/bloom.md
View File

@ -43,6 +43,15 @@ BloomIndex仅支持相等表达式，例如`name='monkey'`。
> 更小的FPP会提高索引的过滤能力，但是会增加索引的体积。在大多数情况下默认值就足以够用。
> 如果创建的索引太大，可以考虑增加这个值（例如，至0.05)。
### `bloom.mmapEnabled`
> - **类型:** `Boolean`
> - **默认值:** `true`
>
> 控制在读取布隆索引时是否应使用内存映射文件 (mmap)。
> 启用此值将在读取期间将 Bloom 索引缓存到本地磁盘而不是内存中。
> 这将减少内存消耗，但会导致性能略有下降。
## 用例
**创建索引:**
@ -51,6 +60,7 @@ create index idx using bloom on hive.hindex.users (id);
create index idx using bloom on hive.hindex.users (id) where regionkey=1;
create index idx using bloom on hive.hindex.users (id) where regionkey in (3, 1);
create index idx using bloom on hive.hindex.users (id) WITH ("bloom.fpp" = '0.001');
create index idx using bloom on hive.hindex.users (id) WITH ("bloom.mmapEnabled" = false);
```
* 假设表已按照`regionkey`列分区

19
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java
View File

@ -81,9 +81,12 @@ public class HeuristicIndexClient
List<IndexMetadata> indexes = new LinkedList<>();
Path indexKeyPath = Paths.get(path);
IndexRecord curIndex = null;
try {
if (indexRecordManager.lookUpIndexRecord(indexKeyPath.subpath(0, 1).toString(),
new String[] {indexKeyPath.subpath(1, 2).toString()}, indexKeyPath.subpath(2, 3).toString()) == null) {
curIndex = indexRecordManager.lookUpIndexRecord(indexKeyPath.subpath(0, 1).toString(),
new String[] {indexKeyPath.subpath(1, 2).toString()}, indexKeyPath.subpath(2, 3).toString());
if (curIndex == null) {
// Use index record file to pre-screen. If record does not contain the index, skip loading
return null;
}
@ -92,7 +95,8 @@ public class HeuristicIndexClient
// On exception, log and continue reading from disk
LOG.debug("Error reading index records: " + path);
}
for (Map.Entry<String, Index> entry : readIndexMap(path).entrySet()) {
for (Map.Entry<String, Index> entry : readIndexMap(path, curIndex).entrySet()) {
String absolutePath = entry.getKey();
Path remainder = Paths.get(absolutePath.replaceFirst(root.toString(), ""));
Path table = remainder.subpath(0, 1);
@ -321,7 +325,7 @@ public class HeuristicIndexClient
* @return an immutable mapping from all index files read to the corresponding index that was loaded
* @throws IOException
*/
private Map<String, Index> readIndexMap(String path)
private Map<String, Index> readIndexMap(String path, IndexRecord indexRecord)
throws IOException
{
ImmutableMap.Builder<String, Index> result = ImmutableMap.builder();
@ -351,6 +355,10 @@ public class HeuristicIndexClient
String indexType = filename.substring(filename.lastIndexOf('.') + 1);
Index index = HeuristicIndexFactory.createIndex(indexType);
// set property for index
index.setProperties(indexRecord.getProperties());
// deserialize from file
index.deserialize(new CloseShieldInputStream(i));
LOG.debug("Loaded %s index from %s.", index.getId(), tarFile.toAbsolutePath());
result.put(tarFile.getParent().resolve(filename).toString(), index);
@ -360,7 +368,6 @@ public class HeuristicIndexClient
}
Map<String, Index> resultMap = result.build();
return resultMap;
}
@ -404,7 +411,7 @@ public class HeuristicIndexClient
{
IndexRecord indexRecord = lookUpIndexRecord(indexName);
CreateIndexMetadata.Level createLevel;
Optional<String> createLevelString = indexRecord.properties.stream().filter(s -> s.toLowerCase(Locale.ROOT).contains("level=")).findAny();
Optional<String> createLevelString = indexRecord.propertiesAsList.stream().filter(s -> s.toLowerCase(Locale.ROOT).contains("level=")).findAny();
createLevel = CreateIndexMetadata.Level.valueOf(createLevelString.get().replaceAll(".*=", ""));
Path pathToIndex = Paths.get(root.toString(), indexRecord.qualifiedTable, indexRecord.columns[0], indexRecord.indexType);

2
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java
View File

@ -171,7 +171,7 @@ public class IndexRecordManager
else {
record.partitions.removeAll(partitionsToRemove);
IndexRecord newRecord = new IndexRecord(record.name, record.user, record.qualifiedTable, record.columns,
record.indexType, record.indexSize, record.properties, record.partitions);
record.indexType, record.indexSize, record.propertiesAsList, record.partitions);
metastore.alterTableParameter(
record.catalog,
record.schema,

139
hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/index/bloom/BloomIndex.java
View File

@ -15,6 +15,7 @@
package io.hetu.core.plugin.heuristicindex.index.bloom;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.heuristicindex.Index;
@ -23,12 +24,17 @@ import io.prestosql.spi.predicate.Domain;
import io.prestosql.spi.relation.CallExpression;
import io.prestosql.spi.util.BloomFilter;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import static io.hetu.core.heuristicindex.util.IndexServiceUtils.matchCallExpEqual;
import static io.prestosql.spi.heuristicindex.TypeUtils.getActualValue;
@ -40,13 +46,21 @@ public class BloomIndex
implements Index
{
public static final String ID = "BLOOM";
protected static final int DEFAULT_EXPECTED_NUM_OF_SIZE = 200000;
private static final String FPP_KEY = "bloom.fpp";
private static final double DEFAULT_FPP = 0.001;
private Properties properties;
private BloomFilter filter;
private int expectedNumOfEntries;
private static final String FPP_KEY = "bloom.fpp";
private static final double DEFAULT_FPP = 0.001;
private double fpp = DEFAULT_FPP;
private int expectedNumOfEntries = DEFAULT_EXPECTED_NUM_OF_SIZE;
private static final String MMAP_KEY = "bloom.mmapEnabled";
private static final Boolean DEFAULT_MMAP = true;
private Boolean mmap;
private File file;
private int mmapSizeInByte;
@Override
public String getId()
@ -54,6 +68,13 @@ public class BloomIndex
return ID;
}
@VisibleForTesting
void setMmapEnabled(Boolean mmap)
{
// use mmap file to hold the bloom filter if enabled
this.mmap = mmap;
}
@Override
public Set<CreateIndexMetadata.Level> getSupportedIndexLevels()
{
@ -67,7 +88,7 @@ public class BloomIndex
List<Object> columnIdxValue = values.get(0).getSecond();
for (Object value : columnIdxValue) {
if (value != null) {
getFilter().add(value.toString().getBytes());
getFilterFromMemory().add(value.toString().getBytes());
}
}
return true;
@ -86,24 +107,66 @@ public class BloomIndex
}
else if (expression instanceof CallExpression) {
// test ComparisonExpression matching
return matchCallExpEqual(expression, object -> filter.test(object.toString().getBytes()));
return matchCallExpEqual(expression, object -> getFilter().test(object.toString().getBytes()));
}
throw new UnsupportedOperationException("Expression not supported by " + ID + " index.");
}
private void writeToMmap(BloomFilter curFilter)
throws IOException
{
try (RandomAccessFile randomFile = new RandomAccessFile(getFile(), "rw")) {
try (FileChannel channel = randomFile.getChannel()) {
long[] bits = curFilter.getBitSet();
int numHashFunctions = curFilter.getNumHashFunctions();
int numBits = bits.length;
mmapSizeInByte = numBits * 8;
MappedByteBuffer map = channel.map(FileChannel.MapMode.READ_WRITE, 0, 2 * 4 + mmapSizeInByte);
map.putInt(numHashFunctions);
map.putInt(numBits);
for (int i = 0; i < numBits; i++) {
map.putLong(bits[i]);
}
}
}
}
private BloomFilter readFromMmap()
throws IOException
{
try (RandomAccessFile randomFile = new RandomAccessFile(getFile(), "r")) {
try (FileChannel channel = randomFile.getChannel()) {
MappedByteBuffer map = channel.map(FileChannel.MapMode.READ_ONLY, 0, 2 * 4 + mmapSizeInByte);
int numHashFunctions = map.getInt();
int numBits = map.getInt();
long[] bits = new long[numBits];
for (int i = 0; i < numBits; i++) {
bits[i] = map.getLong();
}
return new BloomFilter(bits, numHashFunctions);
}
}
}
@Override
public void serialize(OutputStream out)
throws IOException
{
getFilter().writeTo(out);
getFilterFromMemory().writeTo(out);
}
@Override
public Index deserialize(InputStream in)
throws IOException
{
filter = BloomFilter.readFrom(in);
if (isMmapEnabled()) {
// write to mmap and do not write memory
writeToMmap(BloomFilter.readFrom(in));
}
else {
// deserialize filter to memory
filter = BloomFilter.readFrom(in);
}
return this;
}
@ -121,13 +184,29 @@ public class BloomIndex
private int getExpectedNumOfEntries()
{
if (expectedNumOfEntries < 1) {
throw new IllegalArgumentException("Expected number of entries must be greater than 0");
}
return expectedNumOfEntries;
}
private File getFile() throws IOException
{
if (file == null) {
file = File.createTempFile("bloomindex", UUID.randomUUID().toString());
file.delete();
file.deleteOnExit();
}
return file;
}
@Override
public void setExpectedNumOfEntries(int expectedNumOfEntries)
{
this.expectedNumOfEntries = expectedNumOfEntries;
if (expectedNumOfEntries < 1) {
throw new IllegalArgumentException("Expected number of entries must be greater than 0");
}
}
private double getFpp()
@ -136,22 +215,56 @@ public class BloomIndex
String fppValue = getProperties().getProperty(FPP_KEY);
fpp = fppValue == null ? fpp : Double.parseDouble(fppValue);
}
return fpp;
}
private BloomFilter getFilter()
private boolean isMmapEnabled()
{
if (mmap == null) {
if (getProperties() != null) {
String mmapValue = getProperties().getProperty(MMAP_KEY);
mmap = mmapValue == null ? DEFAULT_MMAP : Boolean.parseBoolean(mmapValue);
}
else {
mmap = DEFAULT_MMAP;
}
}
return mmap;
}
private BloomFilter getFilterFromMemory()
{
if (filter == null) {
filter = new BloomFilter(getExpectedNumOfEntries(), getFpp());
}
return filter;
}
@VisibleForTesting
BloomFilter getFilter()
{
if (isMmapEnabled()) {
try {
return readFromMmap();
}
catch (IOException e) {
throw new UnsupportedOperationException("Error reading bloom filter from mmap", e);
}
}
else {
return getFilterFromMemory();
}
}
@Override
public long getMemoryUsage()
{
return getFilter().getRetainedSizeInBytes();
return filter == null ? 0 : filter.getRetainedSizeInBytes();
}
@Override
public long getDiskUsage()
{
return mmap ? 2 * 4 + mmapSizeInByte : 0;
}
}

45
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/filter/TestHeuristicIndexFilter.java
View File

@ -17,6 +17,7 @@ package io.hetu.core.heuristicindex.filter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import io.hetu.core.common.filesystem.TempFolder;
import io.hetu.core.plugin.heuristicindex.index.bloom.BloomIndex;
import io.hetu.core.plugin.heuristicindex.index.minmax.MinMaxIndex;
import io.prestosql.expressions.LogicalRowExpressions;
@ -30,6 +31,9 @@ import io.prestosql.spi.relation.VariableReferenceExpression;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collections;
@ -51,19 +55,40 @@ public class TestHeuristicIndexFilter
public void setup()
throws IOException
{
bloomIndex1 = new BloomIndex();
bloomIndex1.setExpectedNumOfEntries(2);
bloomIndex1.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("a", "b"))));
try (TempFolder folder = new TempFolder()) {
folder.create();
File testFile = folder.newFile();
bloomIndex2 = new BloomIndex();
bloomIndex2.setExpectedNumOfEntries(2);
bloomIndex2.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("c", "d"))));
bloomIndex1 = new BloomIndex();
bloomIndex1.setExpectedNumOfEntries(2);
bloomIndex1.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("a", "b"))));
minMaxIndex1 = new MinMaxIndex();
minMaxIndex1.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(1L, 5L, 10L))));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
bloomIndex1.serialize(fo);
}
minMaxIndex2 = new MinMaxIndex();
minMaxIndex2.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(50L, 80L, 100L))));
try (FileInputStream fi = new FileInputStream(testFile)) {
bloomIndex1.deserialize(fi);
}
bloomIndex2 = new BloomIndex();
bloomIndex2.setExpectedNumOfEntries(2);
bloomIndex2.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("c", "d"))));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
bloomIndex2.serialize(fo);
}
try (FileInputStream fi = new FileInputStream(testFile)) {
bloomIndex2.deserialize(fi);
}
minMaxIndex1 = new MinMaxIndex();
minMaxIndex1.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(1L, 5L, 10L))));
minMaxIndex2 = new MinMaxIndex();
minMaxIndex2.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(50L, 80L, 100L))));
}
}
@Test

399
hetu-heuristic-index/src/test/java/io/hetu/core/plugin/heuristicindex/index/bloom/TestBloomIndex.java
View File

@ -23,6 +23,7 @@ import io.prestosql.spi.predicate.ValueSet;
import io.prestosql.spi.relation.CallExpression;
import io.prestosql.spi.relation.RowExpression;
import io.prestosql.spi.type.Type;
import io.prestosql.spi.util.BloomFilter;
import org.testng.annotations.Test;
import java.io.File;
@ -30,12 +31,17 @@ import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.Random;
import java.util.UUID;
import static io.hetu.core.HeuristicIndexTestUtils.simplePredicate;
import static io.prestosql.spi.type.BigintType.BIGINT;
import static io.prestosql.spi.type.DoubleType.DOUBLE;
import static io.prestosql.spi.type.IntegerType.INTEGER;
import static io.prestosql.spi.type.VarcharType.VARCHAR;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@ -56,90 +62,150 @@ public class TestBloomIndex
@Test
public void testMatches()
throws IOException
{
BloomIndex bloomIndex = new BloomIndex();
List<Object> bloomValues = ImmutableList.of("a", "b", "c", "d");
bloomIndex.setExpectedNumOfEntries(bloomValues.size());
bloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", bloomValues)));
try (TempFolder folder = new TempFolder()) {
folder.create();
File testFile = folder.newFile();
RowExpression expression1 = simplePredicate(OperatorType.EQUAL, "testColumn", VARCHAR, "a");
RowExpression expression2 = simplePredicate(OperatorType.EQUAL, "testColumn", VARCHAR, "e");
BloomIndex bloomIndex = new BloomIndex();
List<Object> bloomValues = ImmutableList.of("a", "b", "c", "d");
bloomIndex.setExpectedNumOfEntries(bloomValues.size());
bloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", bloomValues)));
assertTrue(bloomIndex.matches(expression1));
assertFalse(bloomIndex.matches(expression2));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
bloomIndex.serialize(fo);
}
try (FileInputStream fi = new FileInputStream(testFile)) {
bloomIndex.deserialize(fi);
}
RowExpression expression1 = simplePredicate(OperatorType.EQUAL, "testColumn", VARCHAR, "a");
RowExpression expression2 = simplePredicate(OperatorType.EQUAL, "testColumn", VARCHAR, "e");
assertTrue(bloomIndex.matches(expression1));
assertFalse(bloomIndex.matches(expression2));
}
}
@Test
public void testDomainMatching()
throws IOException
{
BloomIndex stringBloomIndex = new BloomIndex();
List<Object> testValues = ImmutableList.of("a", "ab", "测试", "\n", "%#!", ":dfs");
stringBloomIndex.setExpectedNumOfEntries(testValues.size());
stringBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
try (TempFolder folder = new TempFolder()) {
folder.create();
File testFile = folder.newFile();
ValueSet valueSet = mock(ValueSet.class);
when(valueSet.isSingleValue()).thenReturn(true);
when(valueSet.getType()).thenReturn(VARCHAR);
BloomIndex stringBloomIndex = new BloomIndex();
List<Object> testValues = ImmutableList.of("a", "ab", "测试", "\n", "%#!", ":dfs");
stringBloomIndex.setExpectedNumOfEntries(testValues.size());
stringBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
when(valueSet.getSingleValue()).thenReturn("a");
assertTrue(stringBloomIndex.matches(Domain.create(valueSet, false)));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
stringBloomIndex.serialize(fo);
}
when(valueSet.getSingleValue()).thenReturn("%#!");
assertTrue(stringBloomIndex.matches(Domain.create(valueSet, false)));
try (FileInputStream fi = new FileInputStream(testFile)) {
stringBloomIndex.deserialize(fi);
}
when(valueSet.getSingleValue()).thenReturn("bb");
assertFalse(stringBloomIndex.matches(Domain.create(valueSet, false)));
ValueSet valueSet = mock(ValueSet.class);
when(valueSet.isSingleValue()).thenReturn(true);
when(valueSet.getType()).thenReturn(VARCHAR);
when(valueSet.getSingleValue()).thenReturn("a");
assertTrue(stringBloomIndex.matches(Domain.create(valueSet, false)));
when(valueSet.getSingleValue()).thenReturn("%#!");
assertTrue(stringBloomIndex.matches(Domain.create(valueSet, false)));
when(valueSet.getSingleValue()).thenReturn("bb");
assertFalse(stringBloomIndex.matches(Domain.create(valueSet, false)));
}
}
@Test
public void testMatching()
throws IOException
{
// Test String bloom indexer
BloomIndex stringBloomIndex = new BloomIndex();
List<Object> testValues = ImmutableList.of("a", "ab", "测试", "\n", "%#!", ":dfs");
stringBloomIndex.setExpectedNumOfEntries(testValues.size());
stringBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
try (TempFolder folder = new TempFolder()) {
folder.create();
File testFile = folder.newFile();
assertTrue(mightContain(stringBloomIndex, VARCHAR, "a"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "ab"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "测试"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "\n"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "%#!"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, ":dfs"));
assertFalse(mightContain(stringBloomIndex, VARCHAR, "random"));
assertFalse(mightContain(stringBloomIndex, VARCHAR, "abc"));
// Test String bloom indexer
BloomIndex stringBloomIndex = new BloomIndex();
List<Object> testValues = ImmutableList.of("a", "ab", "测试", "\n", "%#!", ":dfs");
stringBloomIndex.setExpectedNumOfEntries(testValues.size());
stringBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
// Test with the generic type to be Object
BloomIndex objectBloomIndex = new BloomIndex();
testValues = ImmutableList.of("a", "ab", "测试", "\n", "%#!", ":dfs");
objectBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
stringBloomIndex.serialize(fo);
}
assertTrue(mightContain(objectBloomIndex, VARCHAR, "a"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "ab"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "测试"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "\n"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "%#!"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, ":dfs"));
assertFalse(mightContain(objectBloomIndex, VARCHAR, "random"));
assertFalse(mightContain(objectBloomIndex, VARCHAR, "abc"));
try (FileInputStream fi = new FileInputStream(testFile)) {
stringBloomIndex.deserialize(fi);
}
// Test single insertion
BloomIndex simpleBloomIndex = new BloomIndex();
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("a"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("ab"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("测试"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("\n"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("%#!"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(":dfs"))));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "a"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "ab"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "测试"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "\n"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, "%#!"));
assertTrue(mightContain(stringBloomIndex, VARCHAR, ":dfs"));
assertFalse(mightContain(stringBloomIndex, VARCHAR, "random"));
assertFalse(mightContain(stringBloomIndex, VARCHAR, "abc"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "a"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "ab"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "测试"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "\n"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "%#!"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, ":dfs"));
assertFalse(mightContain(simpleBloomIndex, VARCHAR, "random"));
assertFalse(mightContain(simpleBloomIndex, VARCHAR, "abc"));
// Test with the generic type to be Object
BloomIndex objectBloomIndex = new BloomIndex();
testValues = ImmutableList.of("a", "ab", "测试", "\n", "%#!", ":dfs");
objectBloomIndex.setExpectedNumOfEntries(testValues.size());
objectBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
objectBloomIndex.serialize(fo);
}
try (FileInputStream fi = new FileInputStream(testFile)) {
objectBloomIndex.deserialize(fi);
}
assertTrue(mightContain(objectBloomIndex, VARCHAR, "a"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "ab"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "测试"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "\n"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, "%#!"));
assertTrue(mightContain(objectBloomIndex, VARCHAR, ":dfs"));
assertFalse(mightContain(objectBloomIndex, VARCHAR, "random"));
assertFalse(mightContain(objectBloomIndex, VARCHAR, "abc"));
// Test single insertion
BloomIndex simpleBloomIndex = new BloomIndex();
simpleBloomIndex.setExpectedNumOfEntries(6);
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("a"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("ab"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("测试"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("\n"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of("%#!"))));
simpleBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(":dfs"))));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
simpleBloomIndex.serialize(fo);
}
try (FileInputStream fi = new FileInputStream(testFile)) {
simpleBloomIndex.deserialize(fi);
}
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "a"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "ab"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "测试"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "\n"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, "%#!"));
assertTrue(mightContain(simpleBloomIndex, VARCHAR, ":dfs"));
assertFalse(mightContain(simpleBloomIndex, VARCHAR, "random"));
assertFalse(mightContain(simpleBloomIndex, VARCHAR, "abc"));
}
}
@Test
@ -152,6 +218,7 @@ public class TestBloomIndex
BloomIndex objectBloomIndex = new BloomIndex();
List<Object> testValues = ImmutableList.of("%#!", ":dfs", "测试", "\n", "ab", "a");
objectBloomIndex.setExpectedNumOfEntries(testValues.size());
objectBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
@ -189,6 +256,7 @@ public class TestBloomIndex
// Persist it using one object
BloomIndex objectBloomIndex = new BloomIndex();
List<Object> testValues = ImmutableList.of("a", "ab", "测试", "\n", "%#!", ":dfs");
objectBloomIndex.setExpectedNumOfEntries(testValues.size());
objectBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", testValues)));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
objectBloomIndex.serialize(fo);
@ -245,6 +313,7 @@ public class TestBloomIndex
{
// adding 3 values to default size should pass
BloomIndex defaultSizedIndex = new BloomIndex();
defaultSizedIndex.setExpectedNumOfEntries(3);
defaultSizedIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(1f, 2f, 3f))));
// adding 2 values to an index of size 2 should pass
@ -262,6 +331,7 @@ public class TestBloomIndex
public void testMemorySize()
{
BloomIndex index = new BloomIndex();
index.setExpectedNumOfEntries(3);
index.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(1f, 2f, 3f))));
assertTrue(index.getMemoryUsage() > 0);
@ -272,4 +342,209 @@ public class TestBloomIndex
CallExpression expression = simplePredicate(OperatorType.EQUAL, "testColumn", type, value);
return index.matches(expression);
}
@Test
public void testMmapUse()
throws IOException
{
// experiment test to understand the performance of using mmap
try (TempFolder folder = new TempFolder()) {
folder.create();
int dataEntryNum = 2000000;
int queryNum = 10000;
// compare the performance on int data with 2000000 values
File testFile = folder.newFile("int");
BloomIndex objectBloomIndex = new BloomIndex();
objectBloomIndex.setExpectedNumOfEntries(dataEntryNum);
Random rd = new Random();
List<Integer> arr = new ArrayList<>();
for (int i = 0; i < dataEntryNum; i++) {
arr.add(rd.nextInt());
}
objectBloomIndex.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(arr))));
try (FileOutputStream fo = new FileOutputStream(testFile)) {
objectBloomIndex.serialize(fo);
}
BloomIndex bloomIndexMemory = new BloomIndex();
bloomIndexMemory.setMmapEnabled(false);
bloomIndexMemory.setExpectedNumOfEntries(dataEntryNum);
try (FileInputStream fi = new FileInputStream(testFile)) {
bloomIndexMemory.deserialize(fi);
}
BloomIndex bloomIndexMmap = new BloomIndex();
bloomIndexMmap.setMmapEnabled(true);
bloomIndexMmap.setExpectedNumOfEntries(dataEntryNum);
try (FileInputStream fi = new FileInputStream(testFile)) {
bloomIndexMmap.deserialize(fi);
}
System.out.println(testFile);
Random rdTest = new Random();
// get query time using memory
long startTime = System.currentTimeMillis();
for (int i = 0; i < queryNum; i++) {
int testNum = rdTest.nextInt();
RowExpression expression = simplePredicate(OperatorType.EQUAL, "testColumn", INTEGER, testNum);
bloomIndexMemory.matches(expression);
}
long stopTime = System.currentTimeMillis();
long elapsedTime = stopTime - startTime;
System.out.println(elapsedTime);
// get query time using mmap
startTime = System.currentTimeMillis();
for (int i = 0; i < queryNum; i++) {
int testNum = rdTest.nextInt();
RowExpression expression = simplePredicate(OperatorType.EQUAL, "testColumn", INTEGER, testNum);
bloomIndexMmap.matches(expression);
}
stopTime = System.currentTimeMillis();
elapsedTime = stopTime - startTime;
System.out.println(elapsedTime);
BloomFilter memoryFilter = bloomIndexMemory.getFilter();
BloomFilter mmapFilter = bloomIndexMmap.getFilter();
assertEquals(mmapFilter, memoryFilter);
long usage1 = bloomIndexMemory.getMemoryUsage();
long usage2 = bloomIndexMmap.getMemoryUsage();
assertTrue(usage1 > usage2, "mmap should use less memory.");
long fileUsage1 = bloomIndexMemory.getDiskUsage();
long fileUsage2 = bloomIndexMmap.getDiskUsage();
assertTrue(fileUsage1 < fileUsage2, "mmap should use file space.");
// compare the performance on double data with 2000000 entries
File testFileDouble = folder.newFile("double");
BloomIndex objectBloomIndexDouble = new BloomIndex();
objectBloomIndexDouble.setExpectedNumOfEntries(dataEntryNum);
Random rdDouble = new Random();
List<Double> arrDouble = new ArrayList<>();
for (int i = 0; i < dataEntryNum; i++) {
arrDouble.add(rdDouble.nextDouble());
}
objectBloomIndexDouble.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(arrDouble))));
try (FileOutputStream fo = new FileOutputStream(testFileDouble)) {
objectBloomIndexDouble.serialize(fo);
}
BloomIndex bloomIndexMemoryDouble = new BloomIndex();
bloomIndexMemoryDouble.setMmapEnabled(false);
bloomIndexMemoryDouble.setExpectedNumOfEntries(dataEntryNum);
try (FileInputStream fi = new FileInputStream(testFileDouble)) {
bloomIndexMemoryDouble.deserialize(fi);
}
BloomIndex bloomIndexMmapDouble = new BloomIndex();
bloomIndexMmapDouble.setMmapEnabled(true);
bloomIndexMmapDouble.setExpectedNumOfEntries(dataEntryNum);
try (FileInputStream fi = new FileInputStream(testFileDouble)) {
bloomIndexMmapDouble.deserialize(fi);
}
System.out.println(testFileDouble);
Random rdTestDouble = new Random();
// get query time using memory
startTime = System.currentTimeMillis();
for (int i = 0; i < queryNum; i++) {
double testDouble = rdTestDouble.nextDouble();
RowExpression expression = simplePredicate(OperatorType.EQUAL, "testColumn", DOUBLE, testDouble);
bloomIndexMemoryDouble.matches(expression);
}
stopTime = System.currentTimeMillis();
elapsedTime = stopTime - startTime;
System.out.println(elapsedTime);
// get query time using mmap
startTime = System.currentTimeMillis();
for (int i = 0; i < queryNum; i++) {
double testDouble = rdTestDouble.nextDouble();
RowExpression expression = simplePredicate(OperatorType.EQUAL, "testColumn", DOUBLE, testDouble);
bloomIndexMmapDouble.matches(expression);
}
stopTime = System.currentTimeMillis();
elapsedTime = stopTime - startTime;
System.out.println(elapsedTime);
memoryFilter = bloomIndexMemoryDouble.getFilter();
mmapFilter = bloomIndexMmapDouble.getFilter();
assertEquals(mmapFilter, memoryFilter);
usage1 = bloomIndexMemoryDouble.getMemoryUsage();
usage2 = bloomIndexMmapDouble.getMemoryUsage();
assertTrue(usage1 > usage2, "mmap should use less memory.");
fileUsage1 = bloomIndexMemoryDouble.getDiskUsage();
fileUsage2 = bloomIndexMmapDouble.getDiskUsage();
assertTrue(fileUsage1 < fileUsage2, "mmap should use file space.");
// compare the performance on UUID string with 2000000 entries
File testFileString = folder.newFile("string");
BloomIndex objectBloomIndexString = new BloomIndex();
objectBloomIndexString.setExpectedNumOfEntries(dataEntryNum);
List<String> arrString = new ArrayList<>();
for (int i = 0; i < dataEntryNum; i++) {
arrString.add(UUID.randomUUID().toString());
}
objectBloomIndexString.addValues(Collections.singletonList(new Pair<>("testColumn", ImmutableList.of(arrString))));
try (FileOutputStream fo = new FileOutputStream(testFileString)) {
objectBloomIndexString.serialize(fo);
}
BloomIndex bloomIndexMemoryString = new BloomIndex();
bloomIndexMemoryString.setMmapEnabled(false);
bloomIndexMemoryString.setExpectedNumOfEntries(dataEntryNum);
try (FileInputStream fi = new FileInputStream(testFileString)) {
bloomIndexMemoryString.deserialize(fi);
}
BloomIndex bloomIndexMmapString = new BloomIndex();
bloomIndexMmapString.setMmapEnabled(true);
bloomIndexMmapString.setExpectedNumOfEntries(dataEntryNum);
try (FileInputStream fi = new FileInputStream(testFileString)) {
bloomIndexMmapString.deserialize(fi);
}
System.out.println(testFileString);
// get query time using memory
startTime = System.currentTimeMillis();
for (int i = 0; i < queryNum; i++) {
String testString = UUID.randomUUID().toString();
RowExpression expression = simplePredicate(OperatorType.EQUAL, "testColumn", VARCHAR, testString);
bloomIndexMemoryString.matches(expression);
}
stopTime = System.currentTimeMillis();
elapsedTime = stopTime - startTime;
System.out.println(elapsedTime);
// get query time using mmap
startTime = System.currentTimeMillis();
for (int i = 0; i < queryNum; i++) {
String testString = UUID.randomUUID().toString();
RowExpression expression = simplePredicate(OperatorType.EQUAL, "testColumn", VARCHAR, testString);
bloomIndexMmapString.matches(expression);
}
stopTime = System.currentTimeMillis();
elapsedTime = stopTime - startTime;
System.out.println(elapsedTime);
memoryFilter = bloomIndexMemoryString.getFilter();
mmapFilter = bloomIndexMmapString.getFilter();
assertEquals(mmapFilter, memoryFilter);
usage1 = bloomIndexMemoryString.getMemoryUsage();
usage2 = bloomIndexMmapString.getMemoryUsage();
assertTrue(usage1 > usage2, "mmap should use less memory.");
fileUsage1 = bloomIndexMemoryString.getDiskUsage();
fileUsage2 = bloomIndexMmapString.getDiskUsage();
assertTrue(fileUsage1 < fileUsage2, "mmap should use file space.");
}
}
}

15
presto-main/src/main/java/io/prestosql/operator/UpdateIndexOperator.java
View File

@ -355,9 +355,9 @@ public class UpdateIndexOperator
this.updateIndexMetadata = updateIndexMetadata;
CreateIndexMetadata.Level createLevel;
Optional<String> createLevelString = indexRecord.properties.stream().filter(s -> s.toLowerCase(Locale.ROOT).contains("level=")).findAny();
Optional<String> createLevelString = indexRecord.propertiesAsList.stream().filter(s -> s.toLowerCase(Locale.ROOT).contains("level=")).findAny();
createLevel = CreateIndexMetadata.Level.valueOf(createLevelString.get().replaceAll(".*=", ""));
Properties updatedProperties = getPropertiesFromList(indexRecord.properties);
Properties updatedProperties = indexRecord.getProperties();
updateIndexMetadata.getProperties().forEach((key, val) -> {
if (!key.toString().toLowerCase(Locale.ROOT).equals("level")) {
@ -412,15 +412,4 @@ public class UpdateIndexOperator
return new UpdateIndexOperator.UpdateIndexOperatorFactory(operatorId, planNodeId, updateIndexMetadata, heuristicIndexerManager);
}
}
private static Properties getPropertiesFromList(List<String> listOfProperties)
{
Properties properties = new Properties();
for (String p : listOfProperties) {
String key = p.substring(0, p.indexOf("="));
String val = p.substring(p.indexOf("=") + 1);
properties.setProperty(key, val);
}
return properties;
}
}

2
presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java
View File

@ -914,7 +914,7 @@ final class ShowQueriesRewrite
new StringLiteral(DataSize.succinctBytes(v.indexSize).toString()),
new StringLiteral(indexStatus),
new StringLiteral(partitionsStrToDisplay.toString()),
new StringLiteral(String.join(",", v.properties) + inProgressHint),
new StringLiteral(String.join(",", v.propertiesAsList) + inProgressHint),
TRUE_LITERAL));
}

40
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java
View File

@ -26,6 +26,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import static io.prestosql.spi.connector.CreateIndexMetadata.AUTOLOAD_PROP_KEY;
import static java.util.Objects.requireNonNull;
@ -47,11 +48,12 @@ public class IndexRecord
public final String[] columns;
public final String indexType;
public final long indexSize;
public final List<String> properties;
public final List<String> propertiesAsList;
public final Properties propertiesAsProperties;
public final List<String> partitions;
public final long lastModifiedTime;
public IndexRecord(String name, String user, String qualifiedTable, String[] columns, String indexType, long indexSize, List<String> properties, List<String> partitions)
public IndexRecord(String name, String user, String qualifiedTable, String[] columns, String indexType, long indexSize, List<String> propertiesAsList, List<String> partitions)
{
this.name = name;
this.user = user == null ? "" : user;
@ -66,9 +68,21 @@ public class IndexRecord
this.columns = Arrays.stream(columns).map(String::toLowerCase).toArray(String[]::new);
this.indexType = indexType.toUpperCase(Locale.ENGLISH);
this.indexSize = indexSize;
this.properties = properties;
this.partitions = partitions;
this.lastModifiedTime = System.currentTimeMillis();
this.propertiesAsList = propertiesAsList;
this.propertiesAsProperties = stringToProperties(propertiesAsList);
}
private Properties stringToProperties(List<String> properties)
{
Properties curProperties = new Properties();
for (String prop : properties) {
String key = prop.substring(0, prop.indexOf("="));
String val = prop.substring(prop.indexOf("=") + 1);
curProperties.setProperty(key, val);
}
return curProperties;
}
/**
@ -96,8 +110,9 @@ public class IndexRecord
this.name = requireNonNull(values.get("name"), "attribute 'name' should not be null. Index is in an invalid state, recreate the index to resolve the issue.").getAsString();
this.user = requireNonNull(values.get("user"), "attribute 'user' should not be null. Index is in an invalid state, recreate the index to resolve the issue.").getAsString();
this.indexSize = values.has("indexSize") ? values.get("indexSize").getAsLong() : 0;
this.properties = new ArrayList<>();
requireNonNull(values.get("properties"), "attribute 'properties' should not be null. Index is in an invalid state, recreate the index to resolve the issue.").getAsJsonArray().forEach(e -> this.properties.add(e.getAsString()));
this.propertiesAsList = new ArrayList<>();
requireNonNull(values.get("properties"), "attribute 'properties' should not be null. Index is in an invalid state, recreate the index to resolve the issue.").getAsJsonArray().forEach(e -> this.propertiesAsList.add(e.getAsString()));
this.propertiesAsProperties = stringToProperties(this.propertiesAsList);
this.partitions = new ArrayList<>();
requireNonNull(values.get("partitions"), "attribute 'partitions' should not be null. Index is in an invalid state, recreate the index to resolve the issue.").getAsJsonArray().forEach(e -> this.partitions.add(e.getAsString()));
this.lastModifiedTime = requireNonNull(values.get("lastModifiedTime"), "attribute 'lastModifiedTime' should not be null. Index is in an invalid state, recreate the index to resolve the issue.").getAsLong();
@ -115,7 +130,7 @@ public class IndexRecord
content.put("name", name);
content.put("user", user);
content.put("indexSize", indexSize);
content.put("properties", properties);
content.put("properties", propertiesAsList);
content.put("partitions", partitions);
content.put("lastModifiedTime", lastModifiedTime);
@ -124,12 +139,17 @@ public class IndexRecord
public boolean isInProgressRecord()
{
return this.properties.stream().anyMatch(property -> property.startsWith(INPROGRESS_PROPERTY_KEY));
return this.propertiesAsProperties.containsKey(INPROGRESS_PROPERTY_KEY);
}
public Properties getProperties()
{
return propertiesAsProperties;
}
public String getProperty(String key)
{
for (String property : properties) {
for (String property : this.propertiesAsList) {
if (property.toLowerCase(Locale.ROOT).startsWith(key.toLowerCase(Locale.ROOT))) {
String[] entry = property.split("=");
return entry[1];
@ -164,7 +184,7 @@ public class IndexRecord
@Override
public int hashCode()
{
int result = Objects.hash(name, user, catalog, schema, table, qualifiedTable, indexType, indexSize, properties, partitions, lastModifiedTime);
int result = Objects.hash(name, user, catalog, schema, table, qualifiedTable, indexType, indexSize, propertiesAsList, partitions, lastModifiedTime);
result = 31 * result + Arrays.hashCode(columns);
return result;
}
@ -182,7 +202,7 @@ public class IndexRecord
", columns=" + Arrays.toString(columns) +
", indexType='" + indexType + '\'' +
", indexSize=" + indexSize +
", properties=" + properties +
", properties=" + propertiesAsList +
", partitions=" + partitions +
", lastModifiedTime=" + lastModifiedTime +
'}';

16
presto-spi/src/main/java/io/prestosql/spi/util/BloomFilter.java
View File

@ -27,6 +27,7 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.math.RoundingMode;
import java.util.Arrays;
import static com.google.common.base.MoreObjects.toStringHelper;
import static com.google.common.base.Preconditions.checkArgument;
@ -397,6 +398,21 @@ public class BloomFilter
}
}
@Override
public boolean equals(Object other)
{
return (other != null) &&
(other.getClass() == getClass()) &&
Arrays.equals(this.getData(), ((BitSet) other).getData()) &&
(this.bitCount() == ((BitSet) other).bitCount());
}
@Override
public int hashCode()
{
return Arrays.hashCode(this.data) + (int) this.bitCount * 5;
}
/**
* Merge bitsets
*