You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by jl...@apache.org on 2019/04/15 06:37:45 UTC
[incubator-pinot] branch use-trie-in-dictionary created (now
a980adf)
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a change to branch use-trie-in-dictionary
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
at a980adf Add OnHeapTrieBasedStringDictionary
This branch includes the following new commits:
new a980adf Add OnHeapTrieBasedStringDictionary
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/01: Add OnHeapTrieBasedStringDictionary
Posted by jl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a commit to branch use-trie-in-dictionary
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit a980adf2cab6af143890bbb224d671e05df669a9
Author: jackjlli <jl...@linkedin.com>
AuthorDate: Sun Apr 14 23:37:17 2019 -0700
Add OnHeapTrieBasedStringDictionary
---
.../apache/pinot/common/config/IndexingConfig.java | 13 ++++
.../apache/pinot/common/config/TableConfig.java | 6 ++
.../pinot/common/config/IndexingConfigTest.java | 2 +
pinot-core/pom.xml | 4 +
.../index/column/PhysicalColumnIndexContainer.java | 16 +++-
.../segment/index/loader/IndexLoadingConfig.java | 16 ++++
.../index/readers/OnHeapTrieBasedDictionary.java | 33 ++++++++
.../readers/OnHeapTrieBasedStringDictionary.java | 88 ++++++++++++++++++++++
.../readers/ImmutableDictionaryReaderTest.java | 10 +++
.../v1/creator/OnHeapDictionariesTest.java | 2 +
pom.xml | 5 ++
11 files changed, 191 insertions(+), 4 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/IndexingConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/config/IndexingConfig.java
index 6c0fb65..387bbef 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/config/IndexingConfig.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/IndexingConfig.java
@@ -75,6 +75,9 @@ public class IndexingConfig {
@ConfigKey("onHeapDictionaryColumns")
private List<String> _onHeapDictionaryColumns;
+ @ConfigKey("onHeapDictionaryColumns")
+ private List<String> _onHeapTrieBasedDictionaryColumns;
+
@ConfigKey("starTreeIndexSpec")
private StarTreeIndexSpec _starTreeIndexSpec;
@@ -179,6 +182,10 @@ public class IndexingConfig {
return _onHeapDictionaryColumns;
}
+ public List<String> getOnHeapTrieBasedDictionaryColumns() {
+ return _onHeapTrieBasedDictionaryColumns;
+ }
+
public void setNoDictionaryColumns(List<String> noDictionaryColumns) {
_noDictionaryColumns = noDictionaryColumns;
}
@@ -191,6 +198,10 @@ public class IndexingConfig {
_onHeapDictionaryColumns = onHeapDictionaryColumns;
}
+ public void setOnHeapTrieBasedDictionaryColumns(List<String> onHeapTrieBasedDictionaryColumns) {
+ _onHeapTrieBasedDictionaryColumns = onHeapTrieBasedDictionaryColumns;
+ }
+
public void setStarTreeIndexSpec(StarTreeIndexSpec starTreeIndexSpec) {
_starTreeIndexSpec = starTreeIndexSpec;
}
@@ -277,6 +288,7 @@ public class IndexingConfig {
.isEqual(_noDictionaryColumns, that._noDictionaryColumns) && EqualityUtils
.isEqual(_noDictionaryConfig, that._noDictionaryConfig) && EqualityUtils
.isEqual(_onHeapDictionaryColumns, that._onHeapDictionaryColumns) && EqualityUtils
+ .isEqual(_onHeapTrieBasedDictionaryColumns, that._onHeapTrieBasedDictionaryColumns) && EqualityUtils
.isEqual(_starTreeIndexSpec, that._starTreeIndexSpec) && EqualityUtils
.isEqual(_segmentPartitionConfig, that._segmentPartitionConfig) && EqualityUtils
.isEqual(_bloomFilterColumns, that._bloomFilterColumns);
@@ -295,6 +307,7 @@ public class IndexingConfig {
result = EqualityUtils.hashCodeOf(result, _noDictionaryColumns);
result = EqualityUtils.hashCodeOf(result, _noDictionaryConfig);
result = EqualityUtils.hashCodeOf(result, _onHeapDictionaryColumns);
+ result = EqualityUtils.hashCodeOf(result, _onHeapTrieBasedDictionaryColumns);
result = EqualityUtils.hashCodeOf(result, _starTreeIndexSpec);
result = EqualityUtils.hashCodeOf(result, _segmentPartitionConfig);
result = EqualityUtils.hashCodeOf(result, _bloomFilterColumns);
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java b/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java
index 2f4ee1c..bb947b3 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/config/TableConfig.java
@@ -448,6 +448,7 @@ public class TableConfig {
private List<String> _invertedIndexColumns;
private List<String> _noDictionaryColumns;
private List<String> _onHeapDictionaryColumns;
+ private List<String> _onHeapTrieBasedDictionaryColumns;
private List<String> _bloomFilterColumns;
private Map<String, String> _streamConfigs;
private String _streamPartitionAssignmentStrategy = DEFAULT_STREAM_PARTITION_ASSIGNMENT_STRATEGY;
@@ -578,6 +579,11 @@ public class TableConfig {
return this;
}
+ public Builder setOnHeapTrieBasedDictionaryColumns(List<String> onHeapTrieBasedDictionaryColumns) {
+ _onHeapTrieBasedDictionaryColumns = onHeapTrieBasedDictionaryColumns;
+ return this;
+ }
+
public Builder setStreamPartitionAssignmentStrategy(String streamPartitionAssignmentStrategy) {
_streamPartitionAssignmentStrategy = streamPartitionAssignmentStrategy;
return this;
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/config/IndexingConfigTest.java b/pinot-common/src/test/java/org/apache/pinot/common/config/IndexingConfigTest.java
index e3d31d5..af964d8 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/config/IndexingConfigTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/config/IndexingConfigTest.java
@@ -48,6 +48,8 @@ public class IndexingConfigTest {
indexingConfig.setSortedColumn(sortedColumn);
List<String> onHeapDictionaryColumns = Arrays.asList("x", "y", "z");
indexingConfig.setOnHeapDictionaryColumns(onHeapDictionaryColumns);
+ List<String> onHeapTrieBasedDictionaryColumns = Arrays.asList("x", "y");
+ indexingConfig.setOnHeapTrieBasedDictionaryColumns(onHeapTrieBasedDictionaryColumns);
List<String> bloomFilterColumns = Arrays.asList("a", "b");
indexingConfig.setBloomFilterColumns(bloomFilterColumns);
Map<String, String> noDictionaryConfig = new HashMap<>();
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index 0810ec9..ee7df05 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -172,6 +172,10 @@
<groupId>net.sf.jopt-simple</groupId>
<artifactId>jopt-simple</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-collections4</artifactId>
+ </dependency>
<!-- Kafka -->
<dependency>
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/index/column/PhysicalColumnIndexContainer.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/column/PhysicalColumnIndexContainer.java
index df83533..459c457 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/index/column/PhysicalColumnIndexContainer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/column/PhysicalColumnIndexContainer.java
@@ -44,6 +44,7 @@ import org.apache.pinot.core.segment.index.readers.OnHeapFloatDictionary;
import org.apache.pinot.core.segment.index.readers.OnHeapIntDictionary;
import org.apache.pinot.core.segment.index.readers.OnHeapLongDictionary;
import org.apache.pinot.core.segment.index.readers.OnHeapStringDictionary;
+import org.apache.pinot.core.segment.index.readers.OnHeapTrieBasedStringDictionary;
import org.apache.pinot.core.segment.index.readers.StringDictionary;
import org.apache.pinot.core.segment.memory.PinotDataBuffer;
import org.apache.pinot.core.segment.store.ColumnIndexType;
@@ -66,10 +67,12 @@ public final class PhysicalColumnIndexContainer implements ColumnIndexContainer
String columnName = metadata.getColumnName();
boolean loadInvertedIndex = false;
boolean loadOnHeapDictionary = false;
+ boolean loadOnHeapTrieBasedDictionary = false;
boolean loadBloomFilter = false;
if (indexLoadingConfig != null) {
loadInvertedIndex = indexLoadingConfig.getInvertedIndexColumns().contains(columnName);
loadOnHeapDictionary = indexLoadingConfig.getOnHeapDictionaryColumns().contains(columnName);
+ loadOnHeapTrieBasedDictionary = indexLoadingConfig.getOnHeapTrieBasedDictionaryColumns().contains(columnName);
loadBloomFilter = indexLoadingConfig.getBloomFilterColumns().contains(columnName);
}
PinotDataBuffer fwdIndexBuffer = segmentReader.getIndexFor(columnName, ColumnIndexType.FORWARD_INDEX);
@@ -84,7 +87,7 @@ public final class PhysicalColumnIndexContainer implements ColumnIndexContainer
}
// Dictionary-based index
_dictionary = loadDictionary(segmentReader.getIndexFor(columnName, ColumnIndexType.DICTIONARY), metadata,
- loadOnHeapDictionary);
+ loadOnHeapDictionary, loadOnHeapTrieBasedDictionary);
if (metadata.isSingleValue()) {
// Single-value
if (metadata.isSorted()) {
@@ -141,7 +144,7 @@ public final class PhysicalColumnIndexContainer implements ColumnIndexContainer
}
private static ImmutableDictionaryReader loadDictionary(PinotDataBuffer dictionaryBuffer, ColumnMetadata metadata,
- boolean loadOnHeap) {
+ boolean loadOnHeap, boolean loadOnHeapTrieBasedDictionary) {
FieldSpec.DataType dataType = metadata.getDataType();
if (loadOnHeap) {
String columnName = metadata.getColumnName();
@@ -169,8 +172,13 @@ public final class PhysicalColumnIndexContainer implements ColumnIndexContainer
case STRING:
int numBytesPerValue = metadata.getColumnMaxLength();
byte paddingByte = (byte) metadata.getPaddingCharacter();
- return loadOnHeap ? new OnHeapStringDictionary(dictionaryBuffer, length, numBytesPerValue, paddingByte)
- : new StringDictionary(dictionaryBuffer, length, numBytesPerValue, paddingByte);
+ if (loadOnHeapTrieBasedDictionary) {
+ return new OnHeapTrieBasedStringDictionary(dictionaryBuffer, length, numBytesPerValue, paddingByte);
+ } else if (loadOnHeap) {
+ return new OnHeapStringDictionary(dictionaryBuffer, length, numBytesPerValue, paddingByte);
+ } else {
+ return new StringDictionary(dictionaryBuffer, length, numBytesPerValue, paddingByte);
+ }
case BYTES:
numBytesPerValue = metadata.getColumnMaxLength();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/index/loader/IndexLoadingConfig.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/loader/IndexLoadingConfig.java
index 7560ec9..b8d3ed3 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/index/loader/IndexLoadingConfig.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/loader/IndexLoadingConfig.java
@@ -47,6 +47,7 @@ public class IndexLoadingConfig {
private Set<String> _noDictionaryColumns = new HashSet<>(); // TODO: replace this by _noDictionaryConfig.
private Map<String, String> _noDictionaryConfig = new HashMap<>();
private Set<String> _onHeapDictionaryColumns = new HashSet<>();
+ private Set<String> _onHeapTrieBasedDictionaryColumns = new HashSet<>();
private Set<String> _bloomFilterColumns = new HashSet<>();
private SegmentVersion _segmentVersion;
@@ -100,6 +101,11 @@ public class IndexLoadingConfig {
_onHeapDictionaryColumns.addAll(onHeapDictionaryColumns);
}
+ List<String> onHeapTrieBasedDictionaryColumns = indexingConfig.getOnHeapTrieBasedDictionaryColumns();
+ if (onHeapTrieBasedDictionaryColumns != null) {
+ _onHeapTrieBasedDictionaryColumns.addAll(onHeapTrieBasedDictionaryColumns);
+ }
+
String tableSegmentVersion = indexingConfig.getSegmentFormatVersion();
if (tableSegmentVersion != null) {
_segmentVersion = SegmentVersion.valueOf(tableSegmentVersion.toLowerCase());
@@ -181,6 +187,11 @@ public class IndexLoadingConfig {
_onHeapDictionaryColumns = onHeapDictionaryColumns;
}
+ @VisibleForTesting
+ public void setOnHeapTrieBasedDictionaryColumns(@Nonnull Set<String> onHeapTrieBasedDictionaryColumns) {
+ _onHeapTrieBasedDictionaryColumns = onHeapTrieBasedDictionaryColumns;
+ }
+
@Nonnull
public Set<String> getNoDictionaryColumns() {
return _noDictionaryColumns;
@@ -196,6 +207,11 @@ public class IndexLoadingConfig {
return _onHeapDictionaryColumns;
}
+ @Nonnull
+ public Set<String> getOnHeapTrieBasedDictionaryColumns() {
+ return _onHeapTrieBasedDictionaryColumns;
+ }
+
public Set<String> getBloomFilterColumns() {
return _bloomFilterColumns;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/index/readers/OnHeapTrieBasedDictionary.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/readers/OnHeapTrieBasedDictionary.java
new file mode 100644
index 0000000..a3bcbff
--- /dev/null
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/readers/OnHeapTrieBasedDictionary.java
@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.segment.index.readers;
+
+import org.apache.pinot.core.segment.memory.PinotDataBuffer;
+
+
+/**
+ * Abstract base class for all on-heap trie dictionary implementations.
+ *
+ */
+public abstract class OnHeapTrieBasedDictionary extends ImmutableDictionaryReader {
+
+ protected OnHeapTrieBasedDictionary(PinotDataBuffer dataBuffer, int length, int numBytesPerValue, byte paddingByte) {
+ super(dataBuffer, length, numBytesPerValue, paddingByte);
+ }
+}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/index/readers/OnHeapTrieBasedStringDictionary.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/readers/OnHeapTrieBasedStringDictionary.java
new file mode 100644
index 0000000..dbc0e41
--- /dev/null
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/index/readers/OnHeapTrieBasedStringDictionary.java
@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.segment.index.readers;
+
+import java.util.Arrays;
+import org.apache.commons.collections4.trie.PatriciaTrie;
+import org.apache.pinot.core.segment.memory.PinotDataBuffer;
+
+
+public class OnHeapTrieBasedStringDictionary extends OnHeapTrieBasedDictionary {
+ private final byte _paddingByte;
+ private final String[] _unpaddedStrings;
+ private final String[] _paddedStrings;
+ private final PatriciaTrie<Integer> _paddedStringToIdTrie;
+ private final PatriciaTrie<Integer> _unpaddedStringToIdTrie;
+
+ public OnHeapTrieBasedStringDictionary(PinotDataBuffer dataBuffer, int length, int numBytesPerValue,
+ byte paddingByte) {
+ super(dataBuffer, length, numBytesPerValue, paddingByte);
+
+ _paddingByte = paddingByte;
+ byte[] buffer = new byte[numBytesPerValue];
+ _unpaddedStrings = new String[length];
+ _unpaddedStringToIdTrie = new PatriciaTrie<>();
+
+ for (int i = 0; i < length; i++) {
+ _unpaddedStrings[i] = getUnpaddedString(i, buffer);
+ _unpaddedStringToIdTrie.put(_unpaddedStrings[i], i);
+ }
+
+ if (_paddingByte == 0) {
+ _paddedStrings = null;
+ _paddedStringToIdTrie = null;
+ } else {
+ _paddedStrings = new String[length];
+ _paddedStringToIdTrie = new PatriciaTrie<>();
+
+ for (int i = 0; i < length; i++) {
+ _paddedStrings[i] = getPaddedString(i, buffer);
+ _paddedStringToIdTrie.put(_paddedStrings[i], i);
+ }
+ }
+ }
+
+ @Override
+ public int indexOf(Object rawValue) {
+ PatriciaTrie<Integer> stringToIdTrie = (_paddingByte == 0) ? _unpaddedStringToIdTrie : _paddedStringToIdTrie;
+ Integer index = stringToIdTrie.get(rawValue);
+ return (index != null) ? index : -1;
+ }
+
+ @Override
+ public int insertionIndexOf(Object rawValue) {
+ if (_paddingByte == 0) {
+ Integer id = _unpaddedStringToIdTrie.get(rawValue);
+ return (id != null) ? id : Arrays.binarySearch(_unpaddedStrings, rawValue);
+ } else {
+ String paddedValue = padString((String) rawValue);
+ return Arrays.binarySearch(_paddedStrings, paddedValue);
+ }
+ }
+
+ @Override
+ public String get(int dictId) {
+ return _unpaddedStrings[dictId];
+ }
+
+ @Override
+ public String getStringValue(int dictId) {
+ return _unpaddedStrings[dictId];
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/segment/index/readers/ImmutableDictionaryReaderTest.java b/pinot-core/src/test/java/org/apache/pinot/core/segment/index/readers/ImmutableDictionaryReaderTest.java
index 4f007bb..0954410 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/segment/index/readers/ImmutableDictionaryReaderTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/segment/index/readers/ImmutableDictionaryReaderTest.java
@@ -23,6 +23,7 @@ import it.unimi.dsi.fastutil.floats.FloatOpenHashSet;
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import java.io.File;
+import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Random;
@@ -254,6 +255,15 @@ public class ImmutableDictionaryReaderTest {
}
}
+ @Test
+ public void testOnHeapTrieBasedStringDictionary()
+ throws IOException {
+ try (OnHeapTrieBasedDictionary onHeapTrieBasedStringDictionary = new OnHeapTrieBasedStringDictionary(PinotDataBuffer.mapReadOnlyBigEndianFile(new File(TEMP_DIR, STRING_COLUMN_NAME + V1Constants.Dict.FILE_EXTENSION)), NUM_VALUES,
+ _numBytesPerStringValue, (byte) 0)) {
+ testStringDictionary(onHeapTrieBasedStringDictionary);
+ }
+ }
+
private void testStringDictionary(ImmutableDictionaryReader stringDictionary) {
for (int i = 0; i < NUM_VALUES; i++) {
assertEquals(stringDictionary.get(i), _stringValues[i]);
diff --git a/pinot-core/src/test/java/org/apache/pinot/segments/v1/creator/OnHeapDictionariesTest.java b/pinot-core/src/test/java/org/apache/pinot/segments/v1/creator/OnHeapDictionariesTest.java
index 2d3d9dc..8138231 100644
--- a/pinot-core/src/test/java/org/apache/pinot/segments/v1/creator/OnHeapDictionariesTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/segments/v1/creator/OnHeapDictionariesTest.java
@@ -22,6 +22,7 @@ import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -84,6 +85,7 @@ public class OnHeapDictionariesTest {
loadingConfig.setOnHeapDictionaryColumns(new HashSet<>(
Arrays.asList(new String[]{INT_COLUMN, LONG_COLUMN, FLOAT_COLUMN, DOUBLE_COLUMN, STRING_COLUMN})));
+ loadingConfig.setOnHeapTrieBasedDictionaryColumns(new HashSet<>(Collections.singletonList(STRING_COLUMN)));
_onHeapSegment = ImmutableSegmentLoader.load(new File(SEGMENT_DIR_NAME, SEGMENT_NAME), loadingConfig);
}
diff --git a/pom.xml b/pom.xml
index 52abd08..ced7c84 100644
--- a/pom.xml
+++ b/pom.xml
@@ -805,6 +805,11 @@
<artifactId>vavr</artifactId>
<version>0.9.2</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-collections4</artifactId>
+ <version>4.3</version>
+ </dependency>
</dependencies>
</dependencyManagement>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org