From cedcb3db7dae28576cef8a072805d47285c969e7 Mon Sep 17 00:00:00 2001 From: Seyed Hossein Mortazavi Date: Thu, 9 Jul 2020 15:39:35 -0500 Subject: [PATCH] Added no type conversion --- .../carbondata/CarbondataSplitManager.java | 2 +- .../pagesource/DataCenterPageSource.java | 9 +- hetu-state-store/pom.xml | 5 ++ .../hazelcast/HazelCastSliceSerializer.java | 51 +++++++++++ .../HazelcastStateStoreBootstrapper.java | 7 ++ .../hazelcast/HazelcastStateStoreFactory.java | 5 ++ .../hazelcast/TestHazelcastStateMap.java | 20 +++++ .../hive/BackgroundHiveSplitLoader.java | 8 +- .../prestosql/plugin/hive/HivePageSource.java | 10 ++- .../plugin/hive/HivePageSourceProvider.java | 2 +- .../plugin/hive/HiveSplitManager.java | 19 ++-- .../plugin/hive/HiveSplitSource.java | 21 +++-- .../io/prestosql/plugin/hive/HiveUtil.java | 80 +++++++++++++---- .../plugin/hive/AbstractTestHive.java | 2 +- .../hive/AbstractTestHiveFileSystem.java | 2 +- .../hive/TestBackgroundHiveSplitLoader.java | 12 +-- .../plugin/hive/TestHiveSplitSource.java | 14 +-- .../prestosql/plugin/hive/TestHiveUtil.java | 20 ++--- .../io/prestosql/SystemSessionProperties.java | 2 +- .../dynamicfilter/DynamicFilterService.java | 54 ++++++------ .../operator/DynamicFilterSourceOperator.java | 86 +++++-------------- .../sql/planner/LocalDynamicFilter.java | 34 +++----- .../planner/LocalDynamicFiltersCollector.java | 26 ++---- .../sql/planner/LocalExecutionPlanner.java | 2 +- .../prestosql/utils/DynamicFilterUtils.java | 1 + ...stDynamicFilterServiceWithBloomFilter.java | 2 +- .../TestDynamicFilterSourceOperator.java | 37 +++++++- .../BloomFilterDynamicFilter.java | 46 ++++++++++ .../TestBloomFilterDynamicFilter.java | 52 +++++++++++ 29 files changed, 438 insertions(+), 193 deletions(-) create mode 100644 hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelCastSliceSerializer.java create mode 100644 presto-spi/src/test/java/io/prestosql/spi/dynamicfilter/TestBloomFilterDynamicFilter.java diff --git a/hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataSplitManager.java b/hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataSplitManager.java index bf640a69e..eed25fc10 100755 --- a/hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataSplitManager.java +++ b/hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataSplitManager.java @@ -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"); diff --git a/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/pagesource/DataCenterPageSource.java b/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/pagesource/DataCenterPageSource.java index 3114267eb..d9fcba216 100644 --- a/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/pagesource/DataCenterPageSource.java +++ b/hetu-datacenter/src/main/java/io/hetu/core/plugin/datacenter/pagesource/DataCenterPageSource.java @@ -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 builder = new ImmutableMap.Builder(); for (Map.Entry 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); } } diff --git a/hetu-state-store/pom.xml b/hetu-state-store/pom.xml index 770089ceb..4ad590f97 100644 --- a/hetu-state-store/pom.xml +++ b/hetu-state-store/pom.xml @@ -48,6 +48,11 @@ hetu-seed-store test + + io.airlift + slice + provided + org.mockito mockito-all diff --git a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelCastSliceSerializer.java b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelCastSliceSerializer.java new file mode 100644 index 000000000..fb8e756eb --- /dev/null +++ b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelCastSliceSerializer.java @@ -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 +{ + @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() + { + } +} diff --git a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java index 8205d83f6..e7de6f2b9 100644 --- a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java +++ b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreBootstrapper.java @@ -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; diff --git a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreFactory.java b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreFactory.java index d3600bc19..84888b32d 100644 --- a/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreFactory.java +++ b/hetu-state-store/src/main/java/io/hetu/core/statestore/hazelcast/HazelcastStateStoreFactory.java @@ -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); diff --git a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateMap.java b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateMap.java index 3d5faac84..9f3dce29c 100644 --- a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateMap.java +++ b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateMap.java @@ -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 ss = (StateMap) stateStore.createStateCollection("slicecheck", STATE_COLLECTION_TYPE); + ss.put("s1", s1); + Slice s2 = ss.get("s1"); + assertEquals(s1, s2); + } + /** * Test remove */ diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/BackgroundHiveSplitLoader.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/BackgroundHiveSplitLoader.java index 2dee7d7eb..c12e70d88 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/BackgroundHiveSplitLoader.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/BackgroundHiveSplitLoader.java @@ -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; private Map queryInfo; + private TypeManager typeManager; private final Map cachedDynamicFilters = new ConcurrentHashMap<>(); @@ -170,12 +172,14 @@ public class BackgroundHiveSplitLoader Optional validWriteIds, Supplier> dynamicFilterSupplier, Optional queryType, - Map queryInfo) + Map 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; } diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSource.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSource.java index 7da3afa2d..9816f0980 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSource.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSource.java @@ -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 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; } diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java index c71707c86..74dd2136a 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java @@ -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()); } diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitManager.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitManager.java index 1e9810d1e..81b95d894 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitManager.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitManager.java @@ -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); diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitSource.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitSource.java index f91485d7c..1235cd3d7 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitSource.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveSplitSource.java @@ -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> userDefinedCachePredicates; private final boolean isSplitFilteringEnabled; + private final TypeManager typeManager; + private HiveSplitSource( ConnectorSession session, String databaseName, @@ -114,7 +117,8 @@ class HiveSplitSource AtomicReference stateReference, CounterStat highMemorySplitSourceCounter, Supplier> dynamicFilterSupplier, - Set> userDefinedCachedPredicates) + Set> 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> dynamicFilterSupplier, - Set> userDefinedCachePredicates) + Set> userDefinedCachePredicates, + TypeManager typeManager) { AtomicReference 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> dynamicFilterSupplier, - Set> userDefinedCachePredicates) + Set> userDefinedCachePredicates, + TypeManager typeManager) { AtomicReference 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()); } diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveUtil.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveUtil.java index 0f50802a3..f300f4c66 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveUtil.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveUtil.java @@ -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 partitionKeys, Set dynamicFilters) + public static boolean isPartitionFiltered(List partitionKeys, Set 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> getPageSourceIterators(List pageSources) { return pageSources.stream().map(source -> new AbstractIterator() + { + @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()); + } } diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java index 209eaa880..d9df29da5 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHive.java @@ -760,7 +760,7 @@ public abstract class AbstractTestHive hiveConfig.getMaxInitialSplits(), hiveConfig.getSplitLoaderConcurrency(), hiveConfig.getMaxSplitsPerSecond(), - false); + false, null); pageSinkProvider = new HivePageSinkProvider( getDefaultHiveFileWriterFactories(hiveConfig), hdfsEnvironment, diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java index 3f626923c..9320cfdbd 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileSystem.java @@ -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, diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestBackgroundHiveSplitLoader.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestBackgroundHiveSplitLoader.java index fc6da3065..c0e7eaa9b 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestBackgroundHiveSplitLoader.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestBackgroundHiveSplitLoader.java @@ -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 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 createPartitionMetadataWithOfflinePartitions() @@ -770,7 +770,7 @@ public class TestBackgroundHiveSplitLoader EXECUTOR, new CounterStat(), null, - null); + null, null); } private static Table table( diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveSplitSource.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveSplitSource.java index 94e38008a..4a5f37b2e 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveSplitSource.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveSplitSource.java @@ -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 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 diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveUtil.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveUtil.java index eda59757a..0336c775a 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveUtil.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveUtil.java @@ -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 dynamicFilters = new HashSet<>(); List 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) diff --git a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java b/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java index 2611d1ecc..aa80a32f0 100644 --- a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java +++ b/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java @@ -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( diff --git a/presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java b/presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java index 72081ac51..16741e004 100644 --- a/presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java +++ b/presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java @@ -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 results) { - HashSet merged = new HashSet<>(); + Set merged = new HashSet<>(); for (Object o : results) { - try { - HashSet s = (HashSet) o; - merged.addAll(s); - } - catch (NullPointerException e) { - log.error(e); - } + Set s = (Set) o; + merged.addAll(s); } return merged; } diff --git a/presto-main/src/main/java/io/prestosql/operator/DynamicFilterSourceOperator.java b/presto-main/src/main/java/io/prestosql/operator/DynamicFilterSourceOperator.java index f5b10a219..08a8d530b 100644 --- a/presto-main/src/main/java/io/prestosql/operator/DynamicFilterSourceOperator.java +++ b/presto-main/src/main/java/io/prestosql/operator/DynamicFilterSourceOperator.java @@ -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[] 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 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 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 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 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; diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFilter.java b/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFilter.java index 8fb9cb45a..10f98edd7 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFilter.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFilter.java @@ -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 result; - private SettableFuture>> bloomFilterResultFuture; + private SettableFuture> hashSetResultFuture; // Number of partitions left to be processed. private int partitionsLeft; - private Map> domainResult = new HashMap<>(); + private Map 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 tupleDomain) { if (type == DynamicFilter.Type.GLOBAL) { - Map> bloomFilterResult = new HashMap<>(); + Map 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 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> bloomFilterResult = new HashMap<>(); + Map bloomFilterResult = new HashMap<>(); if (isIncomplete) { - bloomFilterResultFuture.set(bloomFilterResult); + hashSetResultFuture.set(bloomFilterResult); return; } - for (Map.Entry> entry : domainResult.entrySet()) { + for (Map.Entry 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>> getBloomFilterResultFuture() + public ListenableFuture> getDynamicFilterResultFuture() { - return bloomFilterResultFuture; + return hashSetResultFuture; } public Consumer> getTupleDomainConsumer() diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java b/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java index 235411d59..1cca04f03 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java @@ -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 predicate; private Map localFilters = new HashMap<>(); - private Map> predicates = new HashMap<>(); + private Map predicates = new HashMap<>(); private Set 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> predicate) + synchronized void intersectDynamicFilter(Map predicate) { - for (Map.Entry> entry : predicate.entrySet()) { + for (Map.Entry 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 predicateSet = predicates.get(entry.getKey()); - Set 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 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); } diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java b/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java index 0f3e4db4b..6c6c09f7c 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java @@ -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; }); } diff --git a/presto-main/src/main/java/io/prestosql/utils/DynamicFilterUtils.java b/presto-main/src/main/java/io/prestosql/utils/DynamicFilterUtils.java index 6e4c598fa..385a97cde 100644 --- a/presto-main/src/main/java/io/prestosql/utils/DynamicFilterUtils.java +++ b/presto-main/src/main/java/io/prestosql/utils/DynamicFilterUtils.java @@ -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() { diff --git a/presto-main/src/test/java/io/prestosql/TestDynamicFilterServiceWithBloomFilter.java b/presto-main/src/test/java/io/prestosql/TestDynamicFilterServiceWithBloomFilter.java index e9be48768..dbc2e66e1 100644 --- a/presto-main/src/test/java/io/prestosql/TestDynamicFilterServiceWithBloomFilter.java +++ b/presto-main/src/test/java/io/prestosql/TestDynamicFilterServiceWithBloomFilter.java @@ -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 + "")); diff --git a/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java index 80e792e94..52c771a0d 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java @@ -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); diff --git a/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/BloomFilterDynamicFilter.java b/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/BloomFilterDynamicFilter.java index cedf1778a..714929727 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/BloomFilterDynamicFilter.java +++ b/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/BloomFilterDynamicFilter.java @@ -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; + } } diff --git a/presto-spi/src/test/java/io/prestosql/spi/dynamicfilter/TestBloomFilterDynamicFilter.java b/presto-spi/src/test/java/io/prestosql/spi/dynamicfilter/TestBloomFilterDynamicFilter.java new file mode 100644 index 000000000..87fa44acf --- /dev/null +++ b/presto-spi/src/test/java/io/prestosql/spi/dynamicfilter/TestBloomFilterDynamicFilter.java @@ -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); + } +}