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

Added no type conversion

This commit is contained in:
Seyed Hossein Mortazavi 2020-07-09 15:39:35 -05:00
parent 05406a0af6
commit cedcb3db7d
29 changed files with 438 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataSplitManager.java
View File

@ -101,7 +101,7 @@ public class CarbondataSplitManager
CarbondataTableReader reader)
{
super(hiveConfig, metastoreProvider, partitionManager, namenodeStats, hdfsEnvironment,
directoryLister, executorService, versionEmbedder, coercionPolicy);
directoryLister, executorService, versionEmbedder, null, coercionPolicy);
this.carbonTableReader = requireNonNull(reader, "client is null");
this.metastoreProvider = requireNonNull(metastoreProvider, "metastore is null");
this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");

9
hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/pagesource/DataCenterPageSource.java
View File

@ -22,7 +22,9 @@ import io.prestosql.client.DataCenterStatementClient;
import io.prestosql.spi.Page;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.ConnectorPageSource;
import io.prestosql.spi.dynamicfilter.BloomFilterDynamicFilter;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.dynamicfilter.HashSetDynamicFilter;
import okhttp3.OkHttpClient;
import java.io.IOException;
@ -121,7 +123,12 @@ public class DataCenterPageSource
ImmutableMap.Builder<String, DynamicFilter> builder = new ImmutableMap.Builder();
for (Map.Entry<ColumnHandle, DynamicFilter> entry : dynamicFilters.entrySet()) {
if (!appliedDynamicFilters.contains(entry.getKey().getColumnName())) {
builder.put(entry.getKey().getColumnName(), entry.getValue());
DynamicFilter df = entry.getValue();
if (df instanceof HashSetDynamicFilter) {
df = BloomFilterDynamicFilter.fromHashSetDynamicFilter((HashSetDynamicFilter) df);
((BloomFilterDynamicFilter) df).createSerializedBloomFilter();
}
builder.put(entry.getKey().getColumnName(), df);
}
}

5
hetu-state-store/pom.xml
View File

@ -48,6 +48,11 @@
<artifactId>hetu-seed-store</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.airlift</groupId>
<artifactId>slice</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>

51
hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelCastSliceSerializer.java Normal file
View File

@ -0,0 +1,51 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.hetu.core.statestore.hazelcast;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.StreamSerializer;
import io.airlift.slice.Slice;
import java.io.IOException;
import static io.airlift.slice.Slices.wrappedBuffer;
public class HazelCastSliceSerializer
implements StreamSerializer<Slice>
{
@Override
public void write(ObjectDataOutput objectDataOutput, Slice slice) throws IOException
{
objectDataOutput.writeByteArray(slice.getBytes());
}
@Override
public Slice read(ObjectDataInput objectDataInput) throws IOException
{
return wrappedBuffer(objectDataInput.readByteArray());
}
@Override
public int getTypeId()
{
return 1;
}
@Override
public void destroy()
{
}
}

7
hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java
View File

@ -23,8 +23,10 @@ import com.hazelcast.config.JoinConfig;
import com.hazelcast.config.MapConfig;
import com.hazelcast.config.MaxSizePolicy;
import com.hazelcast.config.NetworkConfig;
import com.hazelcast.config.SerializerConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import io.airlift.slice.Slice;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.statestore.CipherService;
import io.prestosql.spi.statestore.StateStoreBootstrapper;
@ -68,6 +70,11 @@ public class HazelcastStateStoreBootstrapper
Config hzConfig = new Config();
// Config hazelcast cluster name
// Add serialization for Slice
SerializerConfig sc = new SerializerConfig().setImplementation(new HazelCastSliceSerializer()).setTypeClass(Slice.class);
hzConfig.getSerializationConfig().addSerializerConfig(sc);
String clusterId = config.get(STATE_STORE_CLUSTER_CONFIG_NAME);
if (clusterId == null) {
clusterId = DEFAULT_CLUSTER_ID;

5
hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreFactory.java
View File

@ -18,8 +18,10 @@ import com.google.common.util.concurrent.UncheckedExecutionException;
import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.config.DiscoveryStrategyConfig;
import com.hazelcast.config.SerializerConfig;
import com.hazelcast.core.HazelcastInstance;
import io.airlift.log.Logger;
import io.airlift.slice.Slice;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.classloader.ThreadContextClassLoader;
import io.prestosql.spi.seedstore.Seed;
@ -84,6 +86,9 @@ public class HazelcastStateStoreFactory
}
ClientConfig clientConfig = new ClientConfig();
// Add serialization for Slice
SerializerConfig sc = new SerializerConfig().setImplementation(new HazelCastSliceSerializer()).setTypeClass(Slice.class);
clientConfig.getSerializationConfig().addSerializerConfig(sc);
clientConfig.setClusterName(clusterId);
final String discoveryMode = properties.get(DISCOVERY_MODE_CONFIG_NAME);

20
hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateMap.java
View File

@ -18,8 +18,10 @@ package io.hetu.core.statestore.hazelcast;
import com.google.common.collect.ImmutableSet;
import com.hazelcast.config.Config;
import com.hazelcast.config.NetworkConfig;
import com.hazelcast.config.SerializerConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import io.airlift.slice.Slice;
import io.prestosql.spi.statestore.StateMap;
import io.prestosql.spi.statestore.StateStore;
import io.prestosql.spi.statestore.listener.EntryAddedListener;
@ -36,6 +38,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static io.airlift.slice.Slices.utf8Slice;
import static io.prestosql.spi.statestore.StateCollection.Type;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
@ -67,6 +70,8 @@ public class TestHazelcastStateMap
private void setup()
{
Config config = new Config();
SerializerConfig sc = new SerializerConfig().setImplementation(new HazelCastSliceSerializer()).setTypeClass(Slice.class);
config.getSerializationConfig().addSerializerConfig(sc);
config.setClusterName("cluster-test-map-" + UUID.randomUUID());
// Specify ports to make sure different test cases won't connect to same cluster
NetworkConfig network = config.getNetworkConfig();
@ -164,6 +169,21 @@ public class TestHazelcastStateMap
assertNull(stateMap.get(NOT_EXIST));
}
/**
* Test Hazelcast Slice serializer
*/
@Test
public void testSliceSerializer()
{
Slice s3 = utf8Slice("test3");
Slice s1 = utf8Slice("test1");
StateMap<String, Slice> ss = (StateMap<String, Slice>) stateStore.createStateCollection("slicecheck", STATE_COLLECTION_TYPE);
ss.put("s1", s1);
Slice s2 = ss.get("s1");
assertEquals(s1, s2);
}
/**
* Test remove
*/

8
presto-hive/src/main/java/io/prestosql/plugin/hive/BackgroundHiveSplitLoader.java
View File

@ -40,6 +40,7 @@ import io.prestosql.spi.connector.ConnectorSession;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.predicate.TupleDomain;
import io.prestosql.spi.resourcegroups.QueryType;
import io.prestosql.spi.type.TypeManager;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@ -152,6 +153,7 @@ public class BackgroundHiveSplitLoader
private volatile boolean stopped;
private Optional<QueryType> queryType;
private Map<String, Object> queryInfo;
private TypeManager typeManager;
private final Map<ColumnHandle, DynamicFilter> cachedDynamicFilters = new ConcurrentHashMap<>();
@ -170,12 +172,14 @@ public class BackgroundHiveSplitLoader
Optional<ValidWriteIdList> validWriteIds,
Supplier<Set<DynamicFilter>> dynamicFilterSupplier,
Optional<QueryType> queryType,
Map<String, Object> queryInfo)
Map<String, Object> queryInfo,
TypeManager typeManager)
{
this.table = table;
this.compactEffectivePredicate = compactEffectivePredicate;
this.tableBucketInfo = tableBucketInfo;
this.loaderConcurrency = loaderConcurrency;
this.typeManager = typeManager;
this.session = session;
this.hdfsEnvironment = hdfsEnvironment;
this.namenodeStats = namenodeStats;
@ -341,7 +345,7 @@ public class BackgroundHiveSplitLoader
if (dynamicFilterSupplier != null && isDynamicFilteringSplitFilteringEnabled(session)) {
//buildDynamicFilters(dynamicFilterSupplier.get(), cachedDynamicFilters);
if (isPartitionFiltered(partitionKeys, dynamicFilterSupplier.get())) {
if (isPartitionFiltered(partitionKeys, dynamicFilterSupplier.get(), typeManager)) {
// Avoid listing files and creating splits from a partition if it has been pruned due to dynamic filters
return COMPLETED_FUTURE;
}

10
presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSource.java
View File

@ -40,6 +40,7 @@ import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.ConnectorPageSource;
import io.prestosql.spi.connector.ConnectorSession;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.dynamicfilter.HashSetDynamicFilter;
import io.prestosql.spi.type.DecimalType;
import io.prestosql.spi.type.MapType;
import io.prestosql.spi.type.Type;
@ -650,7 +651,13 @@ public class HivePageSource
for (Map.Entry<Integer, String> column : eligibleColumns.entrySet()) {
Block block = page.getBlock(column.getKey()).getLoadedBlock();
String nativeValue = TypeUtils.readNativeValueForDynamicFilter(types[column.getKey()], block, position);
Object nativeValue;
if (dynamicFilter.get(column.getValue()) instanceof HashSetDynamicFilter) {
nativeValue = TypeUtils.readNativeValue(types[column.getKey()], block, position);
}
else {
nativeValue = TypeUtils.readNativeValueForDynamicFilter(types[column.getKey()], block, position);
}
if (nativeValue != null && dynamicFilter.get(column.getValue()) != null && !dynamicFilter.get(column.getValue()).contains(nativeValue)) {
shouldKeep = false;
@ -661,7 +668,6 @@ public class HivePageSource
ids.add(position);
}
}
return ids;
}

2
presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java
View File

@ -119,7 +119,7 @@ public class HivePageSourceProvider
Path path = new Path(hiveSplit.getPath());
// Filter out splits using partition values and dynamic filters
if (dynamicFilters != null && !dynamicFilters.isEmpty() && isPartitionFiltered(hiveSplit.getPartitionKeys(), new HashSet(dynamicFilters.values()))) {
if (dynamicFilters != null && !dynamicFilters.isEmpty() && isPartitionFiltered(hiveSplit.getPartitionKeys(), new HashSet(dynamicFilters.values()), typeManager)) {
return new FixedPageSource(ImmutableList.of());
}

19
presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitManager.java
View File

@ -42,6 +42,7 @@ import io.prestosql.spi.connector.TableNotFoundException;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.predicate.TupleDomain;
import io.prestosql.spi.resourcegroups.QueryType;
import io.prestosql.spi.type.TypeManager;
import org.weakref.jmx.Managed;
import org.weakref.jmx.Nested;
@ -94,6 +95,7 @@ public class HiveSplitManager
private final int maxSplitsPerSecond;
private final boolean recursiveDfsWalkerEnabled;
private final CounterStat highMemorySplitSourceCounter;
private final TypeManager typeManager;
@Inject
public HiveSplitManager(
@ -105,6 +107,7 @@ public class HiveSplitManager
DirectoryLister directoryLister,
@ForHive ExecutorService executorService,
VersionEmbedder versionEmbedder,
TypeManager typeManager,
CoercionPolicy coercionPolicy)
{
this(
@ -123,7 +126,8 @@ public class HiveSplitManager
hiveConfig.getMaxInitialSplits(),
hiveConfig.getSplitLoaderConcurrency(),
hiveConfig.getMaxSplitsPerSecond(),
hiveConfig.getRecursiveDirWalkerEnabled());
hiveConfig.getRecursiveDirWalkerEnabled(),
typeManager);
}
public HiveSplitManager(
@ -142,7 +146,8 @@ public class HiveSplitManager
int maxInitialSplits,
int splitLoaderConcurrency,
@Nullable Integer maxSplitsPerSecond,
boolean recursiveDfsWalkerEnabled)
boolean recursiveDfsWalkerEnabled,
TypeManager typeManager)
{
this.metastoreProvider = requireNonNull(metastoreProvider, "metastore is null");
this.partitionManager = requireNonNull(partitionManager, "partitionManager is null");
@ -161,6 +166,7 @@ public class HiveSplitManager
this.splitLoaderConcurrency = splitLoaderConcurrency;
this.maxSplitsPerSecond = firstNonNull(maxSplitsPerSecond, Integer.MAX_VALUE);
this.recursiveDfsWalkerEnabled = recursiveDfsWalkerEnabled;
this.typeManager = typeManager;
}
@Override
@ -235,7 +241,8 @@ public class HiveSplitManager
.map(validTxnWriteIdList -> validTxnWriteIdList.getTableValidWriteIdList(table.getDatabaseName() + "." + table.getTableName())),
dynamicFilterSupplier,
queryType,
queryInfo);
queryInfo,
typeManager);
HiveSplitSource splitSource;
switch (splitSchedulingStrategy) {
@ -252,7 +259,8 @@ public class HiveSplitManager
executor,
new CounterStat(),
dynamicFilterSupplier,
userDefinedCachePredicates);
userDefinedCachePredicates,
typeManager);
break;
case GROUPED_SCHEDULING:
splitSource = HiveSplitSource.bucketed(
@ -267,7 +275,8 @@ public class HiveSplitManager
executor,
new CounterStat(),
dynamicFilterSupplier,
userDefinedCachePredicates);
userDefinedCachePredicates,
typeManager);
break;
default:
throw new IllegalArgumentException("Unknown splitSchedulingStrategy: " + splitSchedulingStrategy);

21
presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitSource.java
View File

@ -33,6 +33,7 @@ import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.predicate.TupleDomain;
import io.prestosql.spi.type.AbstractVariableWidthType;
import io.prestosql.spi.type.Type;
import io.prestosql.spi.type.TypeManager;
import java.io.FileNotFoundException;
import java.util.List;
@ -103,6 +104,8 @@ class HiveSplitSource
private final Set<TupleDomain<ColumnMetadata>> userDefinedCachePredicates;
private final boolean isSplitFilteringEnabled;
private final TypeManager typeManager;
private HiveSplitSource(
ConnectorSession session,
String databaseName,
@ -114,7 +117,8 @@ class HiveSplitSource
AtomicReference<State> stateReference,
CounterStat highMemorySplitSourceCounter,
Supplier<Set<DynamicFilter>> dynamicFilterSupplier,
Set<TupleDomain<ColumnMetadata>> userDefinedCachedPredicates)
Set<TupleDomain<ColumnMetadata>> userDefinedCachedPredicates,
TypeManager typeManager)
{
requireNonNull(session, "session is null");
this.queryId = session.getQueryId();
@ -133,6 +137,7 @@ class HiveSplitSource
this.dynamicFilterSupplier = dynamicFilterSupplier;
this.isSplitFilteringEnabled = isDynamicFilteringSplitFilteringEnabled(session);
this.userDefinedCachePredicates = userDefinedCachedPredicates;
this.typeManager = typeManager;
}
public static HiveSplitSource allAtOnce(
@ -147,7 +152,8 @@ class HiveSplitSource
Executor executor,
CounterStat highMemorySplitSourceCounter,
Supplier<Set<DynamicFilter>> dynamicFilterSupplier,
Set<TupleDomain<ColumnMetadata>> userDefinedCachePredicates)
Set<TupleDomain<ColumnMetadata>> userDefinedCachePredicates,
TypeManager typeManager)
{
AtomicReference<State> stateReference = new AtomicReference<>(State.initial());
return new HiveSplitSource(
@ -191,7 +197,8 @@ class HiveSplitSource
stateReference,
highMemorySplitSourceCounter,
dynamicFilterSupplier,
userDefinedCachePredicates);
userDefinedCachePredicates,
typeManager);
}
public static HiveSplitSource bucketed(
@ -206,7 +213,8 @@ class HiveSplitSource
Executor executor,
CounterStat highMemorySplitSourceCounter,
Supplier<Set<DynamicFilter>> dynamicFilterSupplier,
Set<TupleDomain<ColumnMetadata>> userDefinedCachePredicates)
Set<TupleDomain<ColumnMetadata>> userDefinedCachePredicates,
TypeManager typeManager)
{
AtomicReference<State> stateReference = new AtomicReference<>(State.initial());
return new HiveSplitSource(
@ -270,7 +278,8 @@ class HiveSplitSource
stateReference,
highMemorySplitSourceCounter,
dynamicFilterSupplier,
userDefinedCachePredicates);
userDefinedCachePredicates,
typeManager);
}
/**
@ -427,7 +436,7 @@ class HiveSplitSource
// Filter out splits if dynamic filter is available
if (dynamicFilterSupplier != null && isSplitFilteringEnabled) {
splits = splits.stream()
.filter(split -> !isPartitionFiltered(HiveSplitWrapper.getOnlyHiveSplit(split).getPartitionKeys(), dynamicFilterSupplier.get()))
.filter(split -> !isPartitionFiltered(HiveSplitWrapper.getOnlyHiveSplit(split).getPartitionKeys(), dynamicFilterSupplier.get(), typeManager))
.collect(Collectors.toList());
}

80
presto-hive/src/main/java/io/prestosql/plugin/hive/HiveUtil.java
View File

@ -24,6 +24,7 @@ import io.airlift.compress.lzo.LzopCodec;
import io.airlift.json.JsonCodec;
import io.airlift.json.JsonCodecFactory;
import io.airlift.json.ObjectMapperProvider;
import io.airlift.log.Logger;
import io.airlift.slice.Slice;
import io.airlift.slice.SliceUtf8;
import io.airlift.slice.Slices;
@ -42,12 +43,15 @@ import io.prestosql.spi.connector.ConnectorPageSource;
import io.prestosql.spi.connector.ConnectorViewDefinition;
import io.prestosql.spi.connector.RecordCursor;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.dynamicfilter.HashSetDynamicFilter;
import io.prestosql.spi.predicate.NullableValue;
import io.prestosql.spi.type.AbstractVariableWidthType;
import io.prestosql.spi.type.CharType;
import io.prestosql.spi.type.DecimalType;
import io.prestosql.spi.type.Decimals;
import io.prestosql.spi.type.StandardTypes;
import io.prestosql.spi.type.Type;
import io.prestosql.spi.type.TypeManager;
import io.prestosql.spi.type.VarcharType;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@ -150,6 +154,8 @@ import static org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Cate
public final class HiveUtil
{
public static final Logger log = Logger.get(HiveUtil.class);
public static final String PRESTO_VIEW_FLAG = "presto_view";
private static final String VIEW_PREFIX = "/* Presto View: ";
@ -982,7 +988,7 @@ public final class HiveUtil
return HiveType.toHiveTypes(schema.getProperty(IOConstants.COLUMNS_TYPES, ""));
}
public static boolean isPartitionFiltered(List<HivePartitionKey> partitionKeys, Set<DynamicFilter> dynamicFilters)
public static boolean isPartitionFiltered(List<HivePartitionKey> partitionKeys, Set<DynamicFilter> dynamicFilters, TypeManager typeManager)
{
if (partitionKeys == null || dynamicFilters == null) {
return false;
@ -1012,8 +1018,23 @@ public final class HiveUtil
continue;
}
if (!dynamicFilter.contains(partitionValue)) {
return true;
if (typeManager != null && dynamicFilter instanceof HashSetDynamicFilter) {
try {
Object realObjectValue = getValueAsType(((HiveColumnHandle) dynamicFilter.getColumnHandle())
.getColumnMetadata(typeManager).getType(), partitionValue);
if (realObjectValue != null && !dynamicFilter.contains(realObjectValue)) {
return true;
}
}
catch (PrestoException | ClassCastException e) {
log.error("cannot cast class" + e.getMessage());
return false;
}
}
else {
if (!dynamicFilter.contains(partitionValue)) {
return true;
}
}
}
return false;
@ -1031,23 +1052,23 @@ public final class HiveUtil
static List<Iterator<Page>> getPageSourceIterators(List<ConnectorPageSource> pageSources)
{
return pageSources.stream().map(source -> new AbstractIterator<Page>()
{
@Override
protected Page computeNext()
{
@Override
protected Page computeNext()
{
Page nextPage;
do {
nextPage = source.getNextPage();
if (nextPage == null) {
return endOfData();
}
Page nextPage;
do {
nextPage = source.getNextPage();
if (nextPage == null) {
return endOfData();
}
while (nextPage.getPositionCount() == 0);
nextPage = nextPage.getLoadedPage();
return nextPage;
}
}).collect(toList());
while (nextPage.getPositionCount() == 0);
nextPage = nextPage.getLoadedPage();
return nextPage;
}
}).collect(toList());
}
@VisibleForTesting
@ -1061,4 +1082,29 @@ public final class HiveUtil
}
return OptionalInt.empty();
}
private static Object getValueAsType(Type type, String value) throws ClassCastException, PrestoException
{
Class<?> javaType = type.getJavaType();
if (javaType == long.class) {
if (type.equals(BIGINT) || type.equals(INTEGER)) {
return Long.valueOf(value);
}
else {
throw new PrestoException(GENERIC_INTERNAL_ERROR,
"Unhandled type for " + javaType.getSimpleName() + ":" + type.getTypeSignature());
}
}
else if (javaType == boolean.class) {
return Boolean.valueOf(value);
}
else if (javaType == double.class) {
return Double.valueOf(value);
}
else if (type instanceof AbstractVariableWidthType || javaType == Slice.class) {
return Slices.utf8Slice(value);
}
throw new PrestoException(GENERIC_INTERNAL_ERROR,
"Unhandled type for " + javaType.getSimpleName() + ":" + type.getTypeSignature());
}
}

2
presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java
View File

@ -760,7 +760,7 @@ public abstract class AbstractTestHive
hiveConfig.getMaxInitialSplits(),
hiveConfig.getSplitLoaderConcurrency(),
hiveConfig.getMaxSplitsPerSecond(),
false);
false, null);
pageSinkProvider = new HivePageSinkProvider(
getDefaultHiveFileWriterFactories(hiveConfig),
hdfsEnvironment,

2
presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java
View File

@ -208,7 +208,7 @@ public abstract class AbstractTestHiveFileSystem
config.getMaxInitialSplits(),
config.getSplitLoaderConcurrency(),
config.getMaxSplitsPerSecond(),
config.getRecursiveDirWalkerEnabled());
config.getRecursiveDirWalkerEnabled(), null);
pageSinkProvider = new HivePageSinkProvider(
getDefaultHiveFileWriterFactories(config),
hdfsEnvironment,

12
presto-hive/src/test/java/io/prestosql/plugin/hive/TestBackgroundHiveSplitLoader.java
View File

@ -341,7 +341,7 @@ public class TestBackgroundHiveSplitLoader
Optional.empty(),
null,
Optional.empty(),
Collections.emptyMap());
Collections.emptyMap(), null);
HiveSplitSource hiveSplitSource = hiveSplitSource(backgroundHiveSplitLoader);
backgroundHiveSplitLoader.start(hiveSplitSource);
@ -565,7 +565,7 @@ public class TestBackgroundHiveSplitLoader
Optional.empty(),
createTestDynamicFilterSupplier("partitionColumn", ImmutableList.of("0", "2", "3")),
Optional.empty(),
ImmutableMap.of());
ImmutableMap.of(), null);
HiveSplitSource hiveSplitSource = hiveSplitSource(backgroundHiveSplitLoader);
backgroundHiveSplitLoader.start(hiveSplitSource);
@ -673,7 +673,7 @@ public class TestBackgroundHiveSplitLoader
validWriteIds,
null,
Optional.empty(),
Collections.emptyMap());
Collections.emptyMap(), null);
}
private static BackgroundHiveSplitLoader backgroundHiveSplitLoader(List<LocatedFileStatus> files, DirectoryLister directoryLister)
@ -702,7 +702,7 @@ public class TestBackgroundHiveSplitLoader
Optional.empty(),
null,
Optional.empty(),
Collections.emptyMap());
Collections.emptyMap(), null);
}
private static BackgroundHiveSplitLoader backgroundHiveSplitLoaderOfflinePartitions()
@ -725,7 +725,7 @@ public class TestBackgroundHiveSplitLoader
Optional.empty(),
null,
Optional.empty(),
Collections.emptyMap());
Collections.emptyMap(), null);
}
private static Iterable<HivePartitionMetadata> createPartitionMetadataWithOfflinePartitions()
@ -770,7 +770,7 @@ public class TestBackgroundHiveSplitLoader
EXECUTOR,
new CounterStat(),
null,
null);
null, null);
}
private static Table table(

14
presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveSplitSource.java
View File

@ -66,7 +66,7 @@ public class TestHiveSplitSource
Executors.newFixedThreadPool(5),
new CounterStat(),
null,
null);
null, null);
// add 10 splits
for (int i = 0; i < 10; i++) {
@ -102,7 +102,7 @@ public class TestHiveSplitSource
Executors.newFixedThreadPool(5),
new CounterStat(),
null,
null);
null, null);
// add some splits
for (int i = 0; i < 5; i++) {
@ -162,7 +162,7 @@ public class TestHiveSplitSource
Executors.newFixedThreadPool(5),
new CounterStat(),
null,
null);
null, null);
final SettableFuture<ConnectorSplit> splits = SettableFuture.create();
@ -223,7 +223,7 @@ public class TestHiveSplitSource
Executors.newFixedThreadPool(5),
new CounterStat(),
null,
null);
null, null);
int testSplitSizeInBytes = new TestSplit(0).getEstimatedSizeInBytes();
int maxSplitCount = toIntExact(maxOutstandingSplitsSize.toBytes()) / testSplitSizeInBytes;
@ -262,7 +262,7 @@ public class TestHiveSplitSource
Executors.newFixedThreadPool(5),
new CounterStat(),
null,
null);
null, null);
hiveSplitSource.addToQueue(new TestSplit(0, OptionalInt.of(2)));
hiveSplitSource.noMoreSplits();
assertEquals(getSplits(hiveSplitSource, OptionalInt.of(0), 10).size(), 0);
@ -289,7 +289,7 @@ public class TestHiveSplitSource
Executors.newFixedThreadPool(5),
new CounterStat(),
createTestDynamicFilterSupplier("pt_d", ImmutableList.of("0")),
null);
null, null);
for (int i = 0; i < 5; i++) {
hiveSplitSource.addToQueue(new TestPartitionSplit(2 * i, ImmutableList.of(new HivePartitionKey("pt_d", "0")), "pt_d=0"));
@ -322,7 +322,7 @@ public class TestHiveSplitSource
Executors.newFixedThreadPool(5),
new CounterStat(),
null,
cachePredicates);
cachePredicates, null);
int[] idPrefix = new int[] {1};
ImmutableMap

20
presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveUtil.java
View File

@ -98,30 +98,30 @@ public class TestHiveUtil
@Test
public void testIsPartitionFiltered()
{
assertFalse(isPartitionFiltered(null, null), "Should not filter partition if either partitions or dynamicFilters is null");
assertFalse(isPartitionFiltered(null, null, null), "Should not filter partition if either partitions or dynamicFilters is null");
Set<DynamicFilter> dynamicFilters = new HashSet<>();
List<HivePartitionKey> partitions = new ArrayList<>();
assertFalse(isPartitionFiltered(partitions, null), "Should not filter partition if either partitions or dynamicFilters is null");
assertFalse(isPartitionFiltered(null, dynamicFilters), "Should not filter partition if either partitions or dynamicFilters is null");
assertFalse(isPartitionFiltered(partitions, dynamicFilters), "Should not filter partition if partitions and dynamicFilters are empty");
assertFalse(isPartitionFiltered(partitions, null, null), "Should not filter partition if either partitions or dynamicFilters is null");
assertFalse(isPartitionFiltered(null, dynamicFilters, null), "Should not filter partition if either partitions or dynamicFilters is null");
assertFalse(isPartitionFiltered(partitions, dynamicFilters, null), "Should not filter partition if partitions and dynamicFilters are empty");
partitions.add(new HivePartitionKey("pt_d", "0"));
partitions.add(new HivePartitionKey("app_id", "10000"));
assertFalse(isPartitionFiltered(partitions, dynamicFilters), "Should not filter partition if dynamicFilters is empty");
assertFalse(isPartitionFiltered(partitions, dynamicFilters, null), "Should not filter partition if dynamicFilters is empty");
ColumnHandle dayColumn = new HiveColumnHandle("pt_d", HIVE_INT, parseTypeSignature(INTEGER), 0, PARTITION_KEY, Optional.empty());
BloomFilter dayFilter = BloomFilter.create(Funnels.stringFunnel(Charset.defaultCharset()), 1024 * 1024, 0.01);
dynamicFilters.add(new BloomFilterDynamicFilter("1", dayColumn, dayFilter, DynamicFilter.Type.GLOBAL));
assertTrue(isPartitionFiltered(partitions, dynamicFilters), "Should filter partition if any dynamicFilter has 0 element count");
assertTrue(isPartitionFiltered(ImmutableList.of(), dynamicFilters), "Should filter partition if any dynamicFilter has 0 element count");
assertTrue(isPartitionFiltered(partitions, dynamicFilters, null), "Should filter partition if any dynamicFilter has 0 element count");
assertTrue(isPartitionFiltered(ImmutableList.of(), dynamicFilters, null), "Should filter partition if any dynamicFilter has 0 element count");
dayFilter.put("1");
assertTrue(isPartitionFiltered(partitions, dynamicFilters), "Should filter partition if partition value not in dynamicFilter");
assertTrue(isPartitionFiltered(partitions, dynamicFilters, null), "Should filter partition if partition value not in dynamicFilter");
dayFilter.put("0");
assertFalse(isPartitionFiltered(partitions, dynamicFilters), "Should not filter partition if partition value is in dynamicFilter");
assertFalse(isPartitionFiltered(partitions, dynamicFilters, null), "Should not filter partition if partition value is in dynamicFilter");
}
@Test
@ -138,7 +138,7 @@ public class TestHiveUtil
Set nameFilter = new HashSet();
nameFilter.add("Alice");
dynamicFilters.add(new HashSetDynamicFilter("1", nameColumn, nameFilter, DynamicFilter.Type.GLOBAL));
assertFalse(isPartitionFiltered(partitions, dynamicFilters), "Should not filter partition if dynamicFilter is on non-partition column");
assertFalse(isPartitionFiltered(partitions, dynamicFilters, null), "Should not filter partition if dynamicFilter is on non-partition column");
}
private static void assertToPartitionValues(String partitionName)

2
presto-main/src/main/java/io/prestosql/SystemSessionProperties.java
View File

@ -608,7 +608,7 @@ public final class SystemSessionProperties
false),
integerProperty(
DYNAMIC_FILTERING_DATA_STRUCTURE,
"Experimental: Data structure for choosing the datas tructure of the dynamic filter (0 for BloomFilter, 1 for HashSet)",
"Experimental: Data structure for choosing the datastructure of the dynamic filter (0 for BloomFilter, 1 for HashSet)",
featuresConfig.getDynamicFilteringDataStructure(),
false),
dataSizeProperty(

54
presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java
View File

@ -75,10 +75,9 @@ import static java.util.Objects.requireNonNull;
public class DynamicFilterService
{
private static final Logger log = Logger.get(DynamicFilterService.class);
private static final double EXPECTED_FPP = 0.25;
private final ScheduledExecutorService filterMergeExecutor;
private static final int THREAD_POOL_SIZE = 3;
private static final int updateInterval = 50;
private static final int updateInterval = 20;
private ScheduledFuture<?> backgroundTask;
private boolean initialized;
@ -152,9 +151,13 @@ public class DynamicFilterService
if (type != null) {
if (type.equals(DynamicFilterUtils.BLOOMFILTERTYPEGLOBAL) || type.equals(DynamicFilterUtils.BLOOMFILTERTYPELOCAL)) {
BloomFilter mergedFilter = mergeBloomFilters(filterIterator);
if (mergedFilter.expectedFpp() > EXPECTED_FPP) {
log.info("FPP too high: " + mergedFilter.expectedFpp());
if (mergedFilter == null || mergedFilter.expectedFpp() > DynamicFilterUtils.BLOOMFILTER_EXPECTEDFPP) {
if (mergedFilter == null) {
log.error("could not merge dynamic filter");
}
else {
log.info("FPP too high: " + mergedFilter.expectedFpp());
}
clearPartialResults(filterId, queryId);
return;
}
@ -170,29 +173,31 @@ public class DynamicFilterService
((StateMap) stateStoreProvider.getStateStore().getStateCollection(DynamicFilterUtils.MERGEMAP)).put(filterKey, filter);
// remove the filter so we don't need to monitor it anymore
outerEntry.getValue().remove(filterId);
log.info("Merged dynamic filter id: " + filterId + "-" + queryId + " type: " + type + ", column: " + column + ", item count: " + mergedFilter.approximateElementCount() + ", fpp: " + mergedFilter.expectedFpp());
clearPartialResults(filterId, queryId);
log.info("Merged successfully dynamic filter id: " + filterId + "-" + queryId + " type: " + type + ", column: " + column + ", item count: " + mergedFilter.approximateElementCount() + ", fpp: " + mergedFilter.expectedFpp());
}
catch (IOException e) {
log.error(e);
}
finally {
clearPartialResults(filterId, queryId);
}
}
else if (type.equals(DynamicFilterUtils.HASHSETTYPEGLOBAL) || type.equals(DynamicFilterUtils.HASHSETTYPELOCAL)) {
Set merged = mergeHashSets(results);
if (merged == null) {
log.error("could not merge dynamic filter");
clearPartialResults(filterId, queryId);
return;
}
if (!cachedDynamicFilters.containsKey(queryId)) {
cachedDynamicFilters.put(queryId, new ConcurrentHashMap<>());
}
try {
cachedDynamicFilters.get(queryId).put(filterId, new HashSetDynamicFilter(filterKey, null, merged, DynamicFilter.Type.GLOBAL));
((StateMap) stateStoreProvider.getStateStore().getStateCollection(DynamicFilterUtils.MERGEMAP)).put(filterKey, merged);
// remove the filter so we don't need to monitor it anymore
outerEntry.getValue().remove(filterId);
log.info("Merged dynamic filter id using stringsets: " + entry.getKey() + "-" + queryId + " type: " + type + ", column: " + entry.getValue() + ", item count: " + merged.size());
clearPartialResults(entry.getKey(), queryId);
}
catch (NullPointerException e) {
log.error(e);
}
cachedDynamicFilters.get(queryId).put(filterId, new HashSetDynamicFilter(filterKey, null, merged, DynamicFilter.Type.GLOBAL));
((StateMap) stateStoreProvider.getStateStore().getStateCollection(DynamicFilterUtils.MERGEMAP)).put(filterKey, merged);
// remove the filter so we don't need to monitor it anymore
outerEntry.getValue().remove(filterId);
log.info("Merged successfully dynamic filter id using stringsets: " + entry.getKey() + "-" + queryId + " type: " + type + ", column: " + entry.getValue() + ", item count: " + merged.size());
clearPartialResults(filterId, queryId);
}
}
}
@ -218,7 +223,9 @@ public class DynamicFilterService
}
}
catch (IOException e) {
mergedFilter = null;
log.error(e);
break;
}
}
return mergedFilter;
@ -226,15 +233,10 @@ public class DynamicFilterService
private Set mergeHashSets(Collection<Object> results)
{
HashSet<String> merged = new HashSet<>();
Set merged = new HashSet<>();
for (Object o : results) {
try {
HashSet<String> s = (HashSet<String>) o;
merged.addAll(s);
}
catch (NullPointerException e) {
log.error(e);
}
Set s = (Set) o;
merged.addAll(s);
}
return merged;
}

86
presto-main/src/main/java/io/prestosql/operator/DynamicFilterSourceOperator.java
View File

@ -13,10 +13,7 @@
*/
package io.prestosql.operator;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import io.airlift.log.Logger;
import io.airlift.node.NodeInfo;
import io.airlift.units.DataSize;
@ -24,6 +21,7 @@ import io.prestosql.operator.aggregation.TypedSet;
import io.prestosql.spi.Page;
import io.prestosql.spi.block.Block;
import io.prestosql.spi.block.BlockBuilder;
import io.prestosql.spi.dynamicfilter.BloomFilterDynamicFilter;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.predicate.Domain;
import io.prestosql.spi.predicate.TupleDomain;
@ -38,9 +36,6 @@ import io.prestosql.utils.DynamicFilterUtils;
import javax.annotation.Nullable;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
@ -50,7 +45,6 @@ import java.util.function.Consumer;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.base.Verify.verify;
import static io.prestosql.spi.type.TypeUtils.readNativeValue;
import static io.prestosql.spi.type.TypeUtils.readNativeValueForDynamicFilter;
import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toSet;
@ -63,7 +57,6 @@ public class DynamicFilterSourceOperator
implements Operator
{
private static final int EXPECTED_BLOCK_BUILDER_SIZE = 8;
private static final int DEFAULT_DYNAMIC_FILTER_SIZE = 1024 * 1024;
public static final Logger log = Logger.get(DynamicFilterSourceOperator.class);
public static class Channel
@ -192,7 +185,7 @@ public class DynamicFilterSourceOperator
private final StateStoreProvider stateStoreProvider;
private long driverId;
private boolean haveRegistered;
private Set<String>[] stringValueSets;
private Set[] objectValueSets;
/**
* Constructor for the Dynamic Filter Source Operator
@ -221,9 +214,9 @@ public class DynamicFilterSourceOperator
this.blockBuilders = new BlockBuilder[channels.size()];
this.valueSets = new TypedSet[channels.size()];
this.stringValueSets = new Set[channels.size()];
for (int i = 0; i < stringValueSets.length; i++) {
stringValueSets[i] = new HashSet<>();
this.objectValueSets = new Set[channels.size()];
for (int i = 0; i < objectValueSets.length; i++) {
objectValueSets[i] = new HashSet<>();
}
for (int channelIndex = 0; channelIndex < channels.size(); ++channelIndex) {
@ -276,25 +269,13 @@ public class DynamicFilterSourceOperator
Block block = page.getBlock(channels.get(channelIndex).index);
TypedSet valueSet = valueSets[channelIndex];
Type columnType = channels.get(channelIndex).type;
for (int position = 0; position < block.getPositionCount(); ++position) {
String value = readNativeValueForDynamicFilter(columnType, block, position);
if (value == null) {
handleTooLargePredicate(); // TODO: 1/7/20 rename this method as reset bloom filter etc
break outer;
}
stringValueSets[channelIndex].add(value);
if (filterType == DynamicFilter.Type.LOCAL) {
valueSet.add(block, position);
}
valueSet.add(block, position);
}
if (filterType == DynamicFilter.Type.LOCAL) {
filterSizeInBytes += valueSet.getRetainedSizeInBytes();
filterPositionsCount += valueSet.size();
}
else {
filterPositionsCount += stringValueSets[channelIndex].size();
}
filterSizeInBytes += valueSet.getRetainedSizeInBytes();
filterPositionsCount += valueSet.size();
}
if (filterPositionsCount > maxFilterPositionsCount || filterSizeInBytes > maxFilterSizeInBytes) {
// The whole filter (summed over all columns) contains too much values or exceeds maxFilterSizeInBytes.
@ -305,7 +286,7 @@ public class DynamicFilterSourceOperator
private void handleTooLargePredicate()
{
stringValueSets = null;
objectValueSets = null;
// The resulting predicate is too large, allow all probe-side values to be read.
dynamicPredicateConsumer.accept(TupleDomain.all());
@ -334,6 +315,12 @@ public class DynamicFilterSourceOperator
return; // the predicate became too large.
}
ImmutableMap.Builder<String, Domain> domainsBuilder = new ImmutableMap.Builder<>();
for (int channelIndex = 0; channelIndex < channels.size(); ++channelIndex) {
Block block = blockBuilders[channelIndex].build();
Type type = channels.get(channelIndex).type;
domainsBuilder.put(channels.get(channelIndex).filterId, convertToDomain(type, channelIndex, block));
}
finishDynamicFilterTask();
if (filterType == DynamicFilter.Type.GLOBAL) {
valueSets = null;
@ -341,31 +328,21 @@ public class DynamicFilterSourceOperator
dynamicPredicateConsumer.accept(TupleDomain.all());
return;
}
ImmutableMap.Builder<String, Domain> domainsBuilder = new ImmutableMap.Builder<>();
if (filterType != DynamicFilter.Type.GLOBAL) {
for (int channelIndex = 0; channelIndex < channels.size(); ++channelIndex) {
Block block = blockBuilders[channelIndex].build();
Type type = channels.get(channelIndex).type;
domainsBuilder.put(channels.get(channelIndex).filterId, convertToDomain(type, block));
}
}
valueSets = null;
blockBuilders = null;
dynamicPredicateConsumer.accept(TupleDomain.withColumnDomains(domainsBuilder.build()));
}
private Domain convertToDomain(Type type, Block block)
private Domain convertToDomain(Type type, int channelIndex, Block block)
{
ImmutableList.Builder<Object> values = ImmutableList.builder();
for (int position = 0; position < block.getPositionCount(); ++position) {
Object value = readNativeValue(type, block, position);
if (value != null) {
values.add(value);
objectValueSets[channelIndex].add(value);
}
}
// Inner and right join doesn't match rows with null key column values.
return Domain.create(ValueSet.copyOf(type, values.build()), false);
return Domain.create(ValueSet.copyOf(type, objectValueSets[channelIndex]), false);
}
@Override
@ -407,39 +384,22 @@ public class DynamicFilterSourceOperator
dynamicFilterType = getSetType(typeKey);
if (dynamicFilterType.equals(DynamicFilterUtils.BLOOMFILTERTYPEGLOBAL) || dynamicFilterType.equals(DynamicFilterUtils.BLOOMFILTERTYPELOCAL)) {
log.debug("creating new bloomfilter dynamic filter for size of: " + stringValueSets[channelIndex].size() + key + " " + driverId);
byte[] finalOutput = createBloomFilter(stringValueSets[channelIndex]);
log.debug("creating new bloomfilter dynamic filter for size of: " + objectValueSets[channelIndex].size() + key + " " + driverId);
byte[] finalOutput = BloomFilterDynamicFilter.convertBloomFilterToByteArray(BloomFilterDynamicFilter.createBloomFilterFromSet(objectValueSets[channelIndex]));
if (finalOutput != null) {
((StateSet) stateStoreProvider.getStateStore().getStateCollection(key)).add(finalOutput);
}
}
else {
log.debug("creating new string set dynamic filter for size of: " + stringValueSets[channelIndex].size() + key + " " + driverId);
log.debug("creating new string set dynamic filter" + key + " " + driverId);
((StateSet) stateStoreProvider.getStateStore().getStateCollection(key))
.add(stringValueSets[channelIndex]);
.add(objectValueSets[channelIndex]);
}
((StateSet) stateStoreProvider.getStateStore().getStateCollection(DynamicFilterUtils.createKey(DynamicFilterUtils.FINISHREFIX, channel.filterId, channel.queryId))).add(driverId);
((StateSet) stateStoreProvider.getStateStore().getStateCollection(DynamicFilterUtils.createKey(DynamicFilterUtils.WORKERSPREFIX, channel.filterId, channel.queryId))).add(nodeInfo.getNodeId());
}
}
public byte[] createBloomFilter(Set<String> stringValueSet)
{
byte[] finalOutput = null;
BloomFilter bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.defaultCharset()), DEFAULT_DYNAMIC_FILTER_SIZE, 0.1);
for (String value : stringValueSet) {
bloomFilter.put(value);
}
try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
bloomFilter.writeTo(out);
finalOutput = out.toByteArray();
}
catch (IOException e) {
log.error("could not finish filter, Exception happened:" + e.getMessage());
}
return finalOutput;
}
public String getSetType(String key)
{
String type = DynamicFilterUtils.BLOOMFILTERTYPEGLOBAL;

34
presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFilter.java
View File

@ -19,7 +19,6 @@ import com.google.common.collect.MultimapBuilder;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.airlift.log.Logger;
import io.airlift.slice.Slice;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.predicate.Domain;
import io.prestosql.spi.predicate.Range;
@ -65,12 +64,12 @@ public class LocalDynamicFilter
// The resulting predicate for local dynamic filtering.
private TupleDomain<String> result;
private SettableFuture<Map<Symbol, Set<String>>> bloomFilterResultFuture;
private SettableFuture<Map<Symbol, Set>> hashSetResultFuture;
// Number of partitions left to be processed.
private int partitionsLeft;
private Map<String, Set<String>> domainResult = new HashMap<>();
private Map<String, Set> domainResult = new HashMap<>();
private final StateStoreProvider stateStoreProvider;
private final DynamicFilter.Type type;
@ -82,7 +81,7 @@ public class LocalDynamicFilter
verify(probeSymbols.keySet().equals(buildChannels.keySet()), "probeSymbols and buildChannels must have same keys");
this.resultFuture = SettableFuture.create();
this.bloomFilterResultFuture = SettableFuture.create();
this.hashSetResultFuture = SettableFuture.create();
this.result = TupleDomain.none();
this.partitionsLeft = partitionCount;
@ -96,9 +95,9 @@ public class LocalDynamicFilter
private synchronized void addPartition(TupleDomain<String> tupleDomain)
{
if (type == DynamicFilter.Type.GLOBAL) {
Map<Symbol, Set<String>> bloomFilterResult = new HashMap<>();
Map<Symbol, Set> bloomFilterResult = new HashMap<>();
if (isIncomplete) {
bloomFilterResultFuture.set(bloomFilterResult);
hashSetResultFuture.set(bloomFilterResult);
return;
}
}
@ -120,17 +119,10 @@ public class LocalDynamicFilter
if (!domainResult.containsKey(key)) {
domainResult.put(key, new HashSet<>());
}
Set<String> set = domainResult.get(key);
Set set = domainResult.get(key);
for (Range range : value.getValues().getRanges().getOrderedRanges()) {
Object obj = range.getSingleValue();
String val;
if (obj instanceof Slice) {
val = new String(((Slice) obj).getBytes());
}
else {
val = String.valueOf(obj);
}
set.add(val);
set.add(obj);
}
});
@ -140,12 +132,12 @@ public class LocalDynamicFilter
if (partitionsLeft == 0) {
// No more partitions are left to be processed.
// verify(resultFuture.set(convertTupleDomain(result)), "dynamic filter result is provided more than once");
Map<Symbol, Set<String>> bloomFilterResult = new HashMap<>();
Map<Symbol, Set> bloomFilterResult = new HashMap<>();
if (isIncomplete) {
bloomFilterResultFuture.set(bloomFilterResult);
hashSetResultFuture.set(bloomFilterResult);
return;
}
for (Map.Entry<String, Set<String>> entry : domainResult.entrySet()) {
for (Map.Entry<String, Set> entry : domainResult.entrySet()) {
for (Symbol probeSymbol : probeSymbols.get(entry.getKey())) {
if (!bloomFilterResult.containsKey(probeSymbol)) {
bloomFilterResult.put(probeSymbol, new HashSet<>());
@ -153,7 +145,7 @@ public class LocalDynamicFilter
bloomFilterResult.put(probeSymbol, entry.getValue());
}
}
bloomFilterResultFuture.set(bloomFilterResult);
hashSetResultFuture.set(bloomFilterResult);
}
}
@ -278,9 +270,9 @@ public class LocalDynamicFilter
return resultFuture;
}
public ListenableFuture<Map<Symbol, Set<String>>> getBloomFilterResultFuture()
public ListenableFuture<Map<Symbol, Set>> getDynamicFilterResultFuture()
{
return bloomFilterResultFuture;
return hashSetResultFuture;
}
public Consumer<TupleDomain<String>> getTupleDomainConsumer()

26
presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java
View File

@ -14,7 +14,6 @@
package io.prestosql.sql.planner;
import io.airlift.log.Logger;
import io.airlift.slice.Slice;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.dynamicfilter.DynamicFilter;
import io.prestosql.spi.dynamicfilter.DynamicFilterFactory;
@ -41,7 +40,7 @@ public class LocalDynamicFiltersCollector
*/
private TupleDomain<Symbol> predicate;
private Map<Symbol, DynamicFilter> localFilters = new HashMap<>();
private Map<Symbol, Set<String>> predicates = new HashMap<>();
private Map<Symbol, Set> predicates = new HashMap<>();
private Set<Symbol> globalFilters = new HashSet<>();
private StateStoreProvider stateStoreProvider;
private static final Logger LOG = Logger.get(LocalDynamicFiltersCollector.class);
@ -56,9 +55,9 @@ public class LocalDynamicFiltersCollector
this.predicate = TupleDomain.all();
}
synchronized void intersectBloomFilter(Map<Symbol, Set<String>> predicate)
synchronized void intersectDynamicFilter(Map<Symbol, Set> predicate)
{
for (Map.Entry<Symbol, Set<String>> entry : predicate.entrySet()) {
for (Map.Entry<Symbol, Set> entry : predicate.entrySet()) {
if (entry.getValue().size() == 1 && entry.getValue().contains("GLOBAL")) {
globalFilters.add(entry.getKey());
continue;
@ -69,9 +68,9 @@ public class LocalDynamicFiltersCollector
continue;
}
Set<String> predicateSet = predicates.get(entry.getKey());
Set<String> newValues = entry.getValue();
for (String value : newValues) {
Set predicateSet = predicates.get(entry.getKey());
Set newValues = entry.getValue();
for (Object value : newValues) {
predicateSet.add(value);
}
}
@ -144,18 +143,7 @@ public class LocalDynamicFiltersCollector
}
if (!readFromStateStore) {
if (!localFilters.containsKey(entry.getKey()) && predicates.containsKey(entry.getKey())) {
HashSet<String> valueSet = new HashSet<>();
for (Object value : predicates.get(entry.getKey())) {
String val;
if (value instanceof Slice) {
val = new String(((Slice) value).getBytes());
}
else {
val = String.valueOf(value);
}
valueSet.add(val);
}
DynamicFilter dynamicFilter = DynamicFilterFactory.create(filterId, entry.getValue(), valueSet, DynamicFilter.Type.LOCAL);
DynamicFilter dynamicFilter = DynamicFilterFactory.create(filterId, entry.getValue(), predicates.get(entry.getKey()), DynamicFilter.Type.LOCAL);
localFilters.put(entry.getKey(), dynamicFilter);
}

2
presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java
View File

@ -2072,7 +2072,7 @@ public class LocalExecutionPlanner
.map(filter -> {
// Intersect dynamic filters' predicates when they become ready,
// in order to support multiple join nodes in the same plan fragment.
addSuccessCallback(filter.getBloomFilterResultFuture(), collector::intersectBloomFilter);
addSuccessCallback(filter.getDynamicFilterResultFuture(), collector::intersectDynamicFilter);
return filter;
});
}

1
presto-main/src/main/java/io/prestosql/utils/DynamicFilterUtils.java
View File

@ -32,6 +32,7 @@ public class DynamicFilterUtils
public static final String HASHSETTYPEGLOBAL = "HASHSETTYPEGLOBAL";
public static final String BLOOMFILTERTYPEGLOBAL = "BLOOMFILTERTYPEGLOBAL";
public static final String DFTYPEMAP = "dftypemap";
public static final double BLOOMFILTER_EXPECTEDFPP = 0.25;
private DynamicFilterUtils()
{

2
presto-main/src/test/java/io/prestosql/TestDynamicFilterServiceWithBloomFilter.java
View File

@ -111,7 +111,7 @@ public class TestDynamicFilterServiceWithBloomFilter
Assert.assertEquals(stateStoreProvider.getStateStore()
.getStateCollection(DynamicFilterUtils.createKey(DynamicFilterUtils.REGISTERPREFIX, filterId, session.getQueryId().toString())).size(), 4);
Thread.sleep(2000);
Thread.sleep(3000);
BloomFilter bf = fetchDynamicFilter(filterId, session.getQueryId().toString());
for (int i = 1; i < 9; i++) {
Assert.assertEquals(true, bf.mightContain(i + ""));

37
presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java
View File

@ -60,6 +60,7 @@ import java.util.concurrent.ScheduledExecutorService;
import static com.google.common.base.Strings.repeat;
import static io.airlift.concurrent.Threads.daemonThreadsNamed;
import static io.airlift.slice.Slices.utf8Slice;
import static io.prestosql.SequencePageBuilder.createSequencePage;
import static io.prestosql.SessionTestUtils.TEST_SESSION;
import static io.prestosql.SystemSessionProperties.getDynamicFilteringMaxPerDriverSize;
@ -68,6 +69,7 @@ import static io.prestosql.block.BlockAssertions.createBooleansBlock;
import static io.prestosql.block.BlockAssertions.createDoublesBlock;
import static io.prestosql.block.BlockAssertions.createLongRepeatBlock;
import static io.prestosql.block.BlockAssertions.createLongsBlock;
import static io.prestosql.block.BlockAssertions.createSlicesBlock;
import static io.prestosql.block.BlockAssertions.createStringsBlock;
import static io.prestosql.operator.OperatorAssertion.toMaterializedResult;
import static io.prestosql.operator.OperatorAssertion.toPages;
@ -267,6 +269,39 @@ public class TestDynamicFilterSourceOperator
.createKey(DynamicFilterUtils.WORKERSPREFIX, filterId, TEST_SESSION.getQueryId().toString()))).size(), 1);
}
@Test
public void testGlobalDynamicFilterSourceOperatorBloomFilterSlice() throws IOException
{
String filterId = "909";
DynamicFilterSourceOperator.DynamicFilterSourceOperatorFactory operatorFactory = createOperatorFactory
(DynamicFilter.Type.GLOBAL, 0, channel(0, VARCHAR, filterId));
DynamicFilterSourceOperator op1 = createOperator(operatorFactory); // will finish before noMoreOperators()
verifyPassthrough(op1,
ImmutableList.of(VARCHAR),
new Page(createSlicesBlock(utf8Slice("test1"))),
new Page(createSlicesBlock(utf8Slice("test2"))),
new Page(createSlicesBlock(utf8Slice("test3"))));
String key = DynamicFilterUtils.createKey(DynamicFilterUtils.PARTIALPREFIX, filterId, TEST_SESSION.getQueryId().toString());
String typeKey = DynamicFilterUtils.createKey(DynamicFilterUtils.TYPEPREFIX, filterId, TEST_SESSION.getQueryId().toString());
String resultType = (String) ((StateMap) stateStoreProvider.getStateStore()
.getStateCollection(DynamicFilterUtils.DFTYPEMAP)).get(typeKey);
StateSet states = ((StateSet) stateStoreProvider.getStateStore().getStateCollection(key));
for (Object bfSerialized : states.getAll()) {
BloomFilterDynamicFilter bfdf = new BloomFilterDynamicFilter(filterId, null, (byte[]) bfSerialized, DynamicFilter.Type.GLOBAL);
String value = new String((utf8Slice("test1")).getBytes());
assertEquals(bfdf.getSize(), 3);
assertEquals(bfdf.contains(value), true);
}
assertEquals(resultType, DynamicFilterUtils.BLOOMFILTERTYPEGLOBAL);
assertEquals(((StateSet) stateStoreProvider.getStateStore().getStateCollection(DynamicFilterUtils
.createKey(DynamicFilterUtils.FINISHREFIX, filterId, TEST_SESSION.getQueryId().toString()))).size(), 1);
assertEquals(((StateSet) stateStoreProvider.getStateStore().getStateCollection(DynamicFilterUtils
.createKey(DynamicFilterUtils.WORKERSPREFIX, filterId, TEST_SESSION.getQueryId().toString()))).size(), 1);
}
@Test
public void testGlobalDynamicFilterSourceOperatorHashSet() throws IOException
{
@ -290,7 +325,7 @@ public class TestDynamicFilterSourceOperator
StateSet states = ((StateSet) stateStoreProvider.getStateStore().getStateCollection(key));
for (Object bfSerialized : states.getAll()) {
HashSetDynamicFilter bfdf = new HashSetDynamicFilter(filterId, null, (Set) bfSerialized, DynamicFilter.Type.GLOBAL);
assertEquals(bfdf.contains("22"), true);
assertEquals(bfdf.contains(22L), true);
assertEquals(bfdf.getSize(), 8);
}
assertEquals(resultType, DynamicFilterUtils.HASHSETTYPEGLOBAL);

46
presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/BloomFilterDynamicFilter.java
View File

@ -16,17 +16,26 @@ package io.prestosql.spi.dynamicfilter;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import io.airlift.log.Logger;
import io.airlift.slice.Slice;
import io.prestosql.spi.connector.ColumnHandle;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Set;
public class BloomFilterDynamicFilter
extends DynamicFilter
{
public static final Logger log = Logger.get(BloomFilterDynamicFilter.class);
private byte[] bloomFilterSerialized;
private BloomFilter bloomFilterDeserialized;
public static final double BLOOMFILTER_CREAETIONFPP = 0.1;
public static final int DEFAULT_DYNAMIC_FILTER_SIZE = 1024 * 1024;
public BloomFilterDynamicFilter(String filterId, ColumnHandle columnHandle, byte[] bloomFilterSerialized, Type type)
{
this.filterId = filterId;
@ -89,4 +98,41 @@ public class BloomFilterDynamicFilter
{
return bloomFilterDeserialized;
}
public static BloomFilterDynamicFilter fromHashSetDynamicFilter(HashSetDynamicFilter hashSetDynamicFilter)
{
BloomFilter bloomFilter = BloomFilterDynamicFilter.createBloomFilterFromSet(hashSetDynamicFilter.getSetValues());
return new BloomFilterDynamicFilter(hashSetDynamicFilter.getFilterId(), hashSetDynamicFilter.getColumnHandle(), bloomFilter, hashSetDynamicFilter.getType());
}
public byte[] createSerializedBloomFilter()
{
this.bloomFilterSerialized = convertBloomFilterToByteArray(this.bloomFilterDeserialized);
return this.bloomFilterSerialized;
}
public static BloomFilter createBloomFilterFromSet(Set stringValueSet)
{
BloomFilter bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.defaultCharset()), DEFAULT_DYNAMIC_FILTER_SIZE, BLOOMFILTER_CREAETIONFPP);
for (Object value : stringValueSet) {
if (value instanceof Slice) {
value = new String(((Slice) value).getBytes());
}
bloomFilter.put(String.valueOf(value));
}
return bloomFilter;
}
public static byte[] convertBloomFilterToByteArray(BloomFilter bloomFilter)
{
byte[] finalOutput = null;
try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
bloomFilter.writeTo(out);
finalOutput = out.toByteArray();
}
catch (IOException e) {
log.error("could not finish filter, Exception happened:" + e.getMessage());
}
return finalOutput;
}
}

52
presto-spi/src/test/java/io/prestosql/spi/dynamicfilter/TestBloomFilterDynamicFilter.java Normal file
View File

@ -0,0 +1,52 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.spi.dynamicfilter;
import io.airlift.slice.Slice;
import org.testng.annotations.Test;
import java.util.HashSet;
import static io.airlift.slice.Slices.utf8Slice;
import static org.testng.Assert.assertEquals;
public class TestBloomFilterDynamicFilter
{
@Test
public void testDynamicFilterTypeConversion()
{
int v1 = 1;
String v2 = "test";
long v3 = 2L;
double v4 = 0.9;
Slice v5 = utf8Slice("test2");
HashSet hs = new HashSet();
hs.add(v1);
hs.add(v2);
hs.add(v3);
hs.add(v4);
hs.add(v5);
HashSetDynamicFilter hsdf = new HashSetDynamicFilter("19", null, hs, DynamicFilter.Type.LOCAL);
BloomFilterDynamicFilter bfdf = BloomFilterDynamicFilter.fromHashSetDynamicFilter(hsdf);
assertEquals(bfdf.contains(String.valueOf(v1)), true);
assertEquals(bfdf.contains(String.valueOf(v2)), true);
assertEquals(bfdf.contains(String.valueOf(v3)), true);
assertEquals(bfdf.contains(String.valueOf(v4)), true);
assertEquals(bfdf.contains(new String(v5.getBytes())), true);
assertEquals(bfdf.contains(String.valueOf(5)), false);
}
}