From eba92350395f9e10fb2d4e976c354f9bec42998d Mon Sep 17 00:00:00 2001 From: SURYA SUMANTH N Date: Mon, 19 Apr 2021 11:00:10 +0530 Subject: [PATCH] Support for Hive ORC Stripe Filtering using predicate --- .../plugin/hive/HivePageSourceProvider.java | 68 ++++++++++++++- ...ibutedJoinQueriesWithDynamicFiltering.java | 2 +- .../hive/TestHivePageSourceProvider.java | 84 +++++++++++++++++++ .../dynamicfilter/DynamicFilterService.java | 2 +- .../planner/LocalDynamicFiltersCollector.java | 2 +- .../RowExpressionPredicatePushDown.java | 13 +-- .../dynamicfilter/DynamicFilterFactory.java | 5 +- .../dynamicfilter/FilteredDynamicFilter.java | 12 ++- 8 files changed, 168 insertions(+), 20 deletions(-) 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 b42b8deda..5c422d116 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 @@ -14,6 +14,7 @@ package io.prestosql.plugin.hive; import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import io.airlift.slice.Slice; import io.airlift.slice.Slices; @@ -31,11 +32,20 @@ import io.prestosql.spi.connector.ConnectorTransactionHandle; import io.prestosql.spi.connector.FixedPageSource; import io.prestosql.spi.connector.RecordCursor; import io.prestosql.spi.connector.RecordPageSource; +import io.prestosql.spi.dynamicfilter.CombinedDynamicFilter; import io.prestosql.spi.dynamicfilter.DynamicFilter; import io.prestosql.spi.dynamicfilter.DynamicFilterSupplier; +import io.prestosql.spi.dynamicfilter.FilteredDynamicFilter; +import io.prestosql.spi.function.BuiltInFunctionHandle; +import io.prestosql.spi.function.Signature; import io.prestosql.spi.heuristicindex.IndexMetadata; import io.prestosql.spi.heuristicindex.SplitMetadata; +import io.prestosql.spi.predicate.Domain; +import io.prestosql.spi.predicate.Range; import io.prestosql.spi.predicate.TupleDomain; +import io.prestosql.spi.predicate.ValueSet; +import io.prestosql.spi.relation.CallExpression; +import io.prestosql.spi.relation.RowExpression; import io.prestosql.spi.type.Type; import io.prestosql.spi.type.TypeManager; import org.apache.hadoop.conf.Configuration; @@ -176,6 +186,17 @@ public class HivePageSourceProvider URI splitUri = URI.create(URIUtil.encodePath(hiveSplit.getPath())); SplitMetadata splitMetadata = new SplitMetadata(splitUri.getRawPath(), hiveSplit.getLastModifiedTime()); + TupleDomain predicate = TupleDomain.all(); + if (dynamicFilterSupplier.isPresent() && dynamicFilters != null && !dynamicFilters.isEmpty()) { + List filteredHiveColumnHandles = hiveColumns.stream().filter(column -> dynamicFilters.containsKey(column)).collect(toList()); + HiveColumnHandle hiveColumnHandle = filteredHiveColumnHandles.get(0); + Type type = hiveColumnHandle.getColumnMetadata(typeManager).getType(); + predicate = getPredicate(dynamicFilters.get(hiveColumnHandle), type, hiveColumnHandle); + if (predicate.isNone()) { + predicate = TupleDomain.all(); + } + } + /** * This is main logical division point to process filter pushdown enabled case (aka as selective read flow). * If user configuration orc_predicate_pushdown_enabled is true and if all clause of query can be handled by hive @@ -207,7 +228,7 @@ public class HivePageSourceProvider hiveSplit.getLength(), hiveSplit.getFileSize(), hiveSplit.getSchema(), - hiveTable.getCompactEffectivePredicate(), + hiveTable.getCompactEffectivePredicate().intersect(predicate), hiveColumns, hiveSplit.getPartitionKeys(), hiveStorageTimeZone, @@ -701,6 +722,51 @@ public class HivePageSourceProvider return partitionValue; } + protected static Domain modifyDomain(Domain domain, Optional filter) + { + Range range = domain.getValues().getRanges().getSpan(); + if (filter.isPresent() && filter.get() instanceof CallExpression) { + CallExpression call = (CallExpression) filter.get(); + BuiltInFunctionHandle builtInFunctionHandle = (BuiltInFunctionHandle) call.getFunctionHandle(); + String name = builtInFunctionHandle.getSignature().getNameSuffix(); + if (name.contains("$operator$") && Signature.unmangleOperator(name).isComparisonOperator()) { + switch (Signature.unmangleOperator(name)) { + case LESS_THAN: + range = Range.lessThan(domain.getType(), range.getHigh().getValue()); + break; + case GREATER_THAN: + range = Range.greaterThan(domain.getType(), range.getLow().getValue()); + break; + case LESS_THAN_OR_EQUAL: + range = Range.lessThanOrEqual(domain.getType(), range.getHigh().getValue()); + break; + case GREATER_THAN_OR_EQUAL: + range = Range.greaterThanOrEqual(domain.getType(), range.getLow().getValue()); + break; + default: + return domain; + } + domain = Domain.create(ValueSet.ofRanges(range), false); + } + } + return domain; + } + + private static TupleDomain getPredicate(DynamicFilter dynamicFilter, Type type, HiveColumnHandle hiveColumnHandle) + { + if (dynamicFilter instanceof CombinedDynamicFilter) { + List filters = ((CombinedDynamicFilter) dynamicFilter).getFilters(); + List> predicates = filters.stream().map(filter -> getPredicate(filter, type, hiveColumnHandle)).collect(toList()); + return predicates.stream().reduce(TupleDomain.all(), TupleDomain::intersect); + } + if (dynamicFilter instanceof FilteredDynamicFilter && !((FilteredDynamicFilter) dynamicFilter).getSetValues().isEmpty()) { + Domain domain = Domain.create(ValueSet.copyOf(type, ((FilteredDynamicFilter) dynamicFilter).getSetValues()), false); + domain = modifyDomain(domain, ((FilteredDynamicFilter) dynamicFilter).getFilterExpression()); + return TupleDomain.withColumnDomains(ImmutableMap.of(hiveColumnHandle, domain)); + } + return TupleDomain.all(); + } + public enum ColumnMappingKind { REGULAR, diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveDistributedJoinQueriesWithDynamicFiltering.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveDistributedJoinQueriesWithDynamicFiltering.java index db65d6cef..d7311a38e 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveDistributedJoinQueriesWithDynamicFiltering.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveDistributedJoinQueriesWithDynamicFiltering.java @@ -167,7 +167,7 @@ public class TestHiveDistributedJoinQueriesWithDynamicFiltering } try { - ConnectorPageSource result = provider.createPageSource(transaction, session, split, table, ImmutableList.of(testColumnHandle3), dynamicFilterSupplier4); + ConnectorPageSource result = provider.createPageSource(transaction, session, split, table, ImmutableList.of(testColumnHandle4), dynamicFilterSupplier4); assertFalse(result instanceof FixedPageSource); } catch (Exception e) { diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHivePageSourceProvider.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHivePageSourceProvider.java index c41da77f2..fdc63354d 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHivePageSourceProvider.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHivePageSourceProvider.java @@ -14,11 +14,27 @@ */ package io.prestosql.plugin.hive; +import com.google.common.collect.ImmutableList; +import io.prestosql.spi.connector.QualifiedObjectName; +import io.prestosql.spi.function.BuiltInFunctionHandle; +import io.prestosql.spi.function.FunctionKind; +import io.prestosql.spi.function.Signature; +import io.prestosql.spi.predicate.Domain; +import io.prestosql.spi.predicate.ValueSet; +import io.prestosql.spi.relation.CallExpression; +import io.prestosql.spi.relation.VariableReferenceExpression; +import io.prestosql.spi.type.TypeSignature; import org.eclipse.jetty.util.URIUtil; import org.testng.annotations.Test; import java.net.URI; +import java.util.Collection; +import java.util.HashSet; +import java.util.Optional; +import static io.prestosql.plugin.hive.HivePageSourceProvider.modifyDomain; +import static io.prestosql.spi.type.BigintType.BIGINT; +import static io.prestosql.spi.type.BooleanType.BOOLEAN; import static org.testng.Assert.assertEquals; public class TestHivePageSourceProvider @@ -93,4 +109,72 @@ public class TestHivePageSourceProvider URI splitUri32 = URI.create(URIUtil.encodePath("hdfs://localhost:9000/user/hive/warehouse/part=part_1/20200405`12345`abcdefgh")); assertEquals("/user/hive/warehouse/part=part_1/20200405%6012345%60abcdefgh", splitUri32.getRawPath()); } + + @Test + public void testModifyDomainGreaterThanOrEqual() + { + Collection valueSet = new HashSet<>(); + valueSet.add(Long.valueOf(40)); + VariableReferenceExpression argument1 = new VariableReferenceExpression("arg_1", BIGINT); + VariableReferenceExpression argument2 = new VariableReferenceExpression("arg_2", BIGINT); + QualifiedObjectName objectName = new QualifiedObjectName("presto", "default", "$operator$greater_than_or_equal"); + + BuiltInFunctionHandle functionHandle = new BuiltInFunctionHandle(new Signature(objectName, FunctionKind.SCALAR, ImmutableList.of(), ImmutableList.of(), new TypeSignature("boolean"), ImmutableList.of(new TypeSignature("bigint"), new TypeSignature("bigint")), false)); + CallExpression filter = new CallExpression("GREATER_THAN_OR_EQUAL", functionHandle, BOOLEAN, ImmutableList.of(argument1, argument2)); + Domain domain = Domain.create(ValueSet.copyOf(BIGINT, valueSet), false); + domain = modifyDomain(domain, Optional.of(filter)); + assertEquals(domain.getValues().getRanges().getSpan().getHigh().getValueBlock(), Optional.empty()); + assertEquals(domain.getValues().getRanges().getSpan().getLow().getValue(), Long.valueOf(40)); + } + + @Test + public void testModifyDomainGreaterThan() + { + Collection valueSet = new HashSet<>(); + valueSet.add(Long.valueOf(40)); + VariableReferenceExpression argument1 = new VariableReferenceExpression("arg_1", BIGINT); + VariableReferenceExpression argument2 = new VariableReferenceExpression("arg_2", BIGINT); + QualifiedObjectName objectName = new QualifiedObjectName("presto", "default", "$operator$greater_than"); + + BuiltInFunctionHandle functionHandle = new BuiltInFunctionHandle(new Signature(objectName, FunctionKind.SCALAR, ImmutableList.of(), ImmutableList.of(), new TypeSignature("boolean"), ImmutableList.of(new TypeSignature("bigint"), new TypeSignature("bigint")), false)); + CallExpression filter = new CallExpression("GREATER_THAN", functionHandle, BOOLEAN, ImmutableList.of(argument1, argument2)); + Domain domain = Domain.create(ValueSet.copyOf(BIGINT, valueSet), false); + domain = modifyDomain(domain, Optional.of(filter)); + assertEquals(domain.getValues().getRanges().getSpan().getHigh().getValueBlock(), Optional.empty()); + assertEquals(domain.getValues().getRanges().getSpan().getLow().getValue(), Long.valueOf(40)); + } + + @Test + public void testModifyDomainLessThanOrEqual() + { + Collection valueSet = new HashSet<>(); + valueSet.add(Long.valueOf(40)); + VariableReferenceExpression argument1 = new VariableReferenceExpression("arg_1", BIGINT); + VariableReferenceExpression argument2 = new VariableReferenceExpression("arg_2", BIGINT); + QualifiedObjectName objectName = new QualifiedObjectName("presto", "default", "$operator$less_than_or_equal"); + + BuiltInFunctionHandle functionHandle = new BuiltInFunctionHandle(new Signature(objectName, FunctionKind.SCALAR, ImmutableList.of(), ImmutableList.of(), new TypeSignature("boolean"), ImmutableList.of(new TypeSignature("bigint"), new TypeSignature("bigint")), false)); + CallExpression filter = new CallExpression("LESS_THAN", functionHandle, BOOLEAN, ImmutableList.of(argument1, argument2)); + Domain domain = Domain.create(ValueSet.copyOf(BIGINT, valueSet), false); + domain = modifyDomain(domain, Optional.of(filter)); + assertEquals(domain.getValues().getRanges().getSpan().getHigh().getValue(), Long.valueOf(40)); + assertEquals(domain.getValues().getRanges().getSpan().getLow().getValueBlock(), Optional.empty()); + } + + @Test + public void testModifyDomainLessThan() + { + Collection valueSet = new HashSet<>(); + valueSet.add(Long.valueOf(40)); + VariableReferenceExpression argument1 = new VariableReferenceExpression("arg_1", BIGINT); + VariableReferenceExpression argument2 = new VariableReferenceExpression("arg_2", BIGINT); + QualifiedObjectName objectName = new QualifiedObjectName("presto", "default", "$operator$less_than"); + + BuiltInFunctionHandle functionHandle = new BuiltInFunctionHandle(new Signature(objectName, FunctionKind.SCALAR, ImmutableList.of(), ImmutableList.of(), new TypeSignature("boolean"), ImmutableList.of(new TypeSignature("bigint"), new TypeSignature("bigint")), false)); + CallExpression filter = new CallExpression("LESS_THAN_OR_EQUAL", functionHandle, BOOLEAN, ImmutableList.of(argument1, argument2)); + Domain domain = Domain.create(ValueSet.copyOf(BIGINT, valueSet), false); + domain = modifyDomain(domain, Optional.of(filter)); + assertEquals(domain.getValues().getRanges().getSpan().getHigh().getValue(), Long.valueOf(40)); + assertEquals(domain.getValues().getRanges().getSpan().getLow().getValueBlock(), Optional.empty()); + } } 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 c71d5595f..3010a4d0a 100644 --- a/presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java +++ b/presto-main/src/main/java/io/prestosql/dynamicfilter/DynamicFilterService.java @@ -195,7 +195,7 @@ public class DynamicFilterService } else if (filterDataType == HASHSET) { Set mergedSet = mergeHashSets(results); - mergedFilter = DynamicFilterFactory.create(filterKey, null, mergedSet, filterType, dfFilter); + mergedFilter = DynamicFilterFactory.create(filterKey, null, mergedSet, filterType, dfFilter, Optional.empty()); if (filterType == GLOBAL) { mergedDynamicFilters.put(filterKey, mergedSet); 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 69981ddd6..64cbc47ed 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 @@ -163,7 +163,7 @@ public class LocalDynamicFiltersCollector if (predicates.containsKey(filterId)) { Optional filter = context.getFilter(filterId); Optional> filterPredicate = DynamicFilters.createDynamicFilterPredicate(filter); - DynamicFilter dynamicFilter = DynamicFilterFactory.create(filterId, columnHandle, predicates.get(filterId), LOCAL, filterPredicate); + DynamicFilter dynamicFilter = DynamicFilterFactory.create(filterId, columnHandle, predicates.get(filterId), LOCAL, filterPredicate, filter); cachedDynamicFilters.put(filterId, dynamicFilter); result.put(columnHandle, dynamicFilter); } diff --git a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/RowExpressionPredicatePushDown.java b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/RowExpressionPredicatePushDown.java index be76d9db1..6c59a5ea5 100644 --- a/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/RowExpressionPredicatePushDown.java +++ b/presto-main/src/main/java/io/prestosql/sql/planner/optimizations/RowExpressionPredicatePushDown.java @@ -46,7 +46,6 @@ import io.prestosql.spi.relation.ConstantExpression; import io.prestosql.spi.relation.RowExpression; import io.prestosql.spi.relation.VariableReferenceExpression; import io.prestosql.spi.type.TypeManager; -import io.prestosql.sql.DynamicFilters; import io.prestosql.sql.planner.PlanSymbolAllocator; import io.prestosql.sql.planner.RowExpressionEqualityInference; import io.prestosql.sql.planner.RowExpressionInterpreter; @@ -80,7 +79,6 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import static com.google.common.base.Preconditions.checkArgument; @@ -329,17 +327,8 @@ public class RowExpressionPredicatePushDown .filter(childOutputSet::contains) .collect(Collectors.groupingBy(identity(), Collectors.counting())); - AtomicInteger maxOccurance = new AtomicInteger(1); - if (expression instanceof CallExpression) { - CallExpression callExpression = (CallExpression) expression; - if (callExpression.getDisplayName().equals(DynamicFilters.Function.NAME) - && callExpression.getFilter().isPresent()) { - maxOccurance.set(3); - } - } - return dependencies.entrySet().stream() - .allMatch(entry -> entry.getValue() == maxOccurance.get() || node.getAssignments().get(toSymbol(entry.getKey())) instanceof ConstantExpression); + .allMatch(entry -> entry.getValue() == 1 || node.getAssignments().get(toSymbol(entry.getKey())) instanceof ConstantExpression); } @Override diff --git a/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/DynamicFilterFactory.java b/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/DynamicFilterFactory.java index c2a44221c..075a015ab 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/DynamicFilterFactory.java +++ b/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/DynamicFilterFactory.java @@ -15,6 +15,7 @@ package io.prestosql.spi.dynamicfilter; import io.prestosql.spi.connector.ColumnHandle; +import io.prestosql.spi.relation.RowExpression; import java.util.List; import java.util.Optional; @@ -37,10 +38,10 @@ public class DynamicFilterFactory return new HashSetDynamicFilter(filterId, columnHandle, values, type); } - public static HashSetDynamicFilter create(String filterId, ColumnHandle columnHandle, Set values, DynamicFilter.Type type, Optional> filter) + public static HashSetDynamicFilter create(String filterId, ColumnHandle columnHandle, Set values, DynamicFilter.Type type, Optional> filter, Optional filterExpression) { if (filter.isPresent()) { - return new FilteredDynamicFilter(filterId, columnHandle, values, type, filter); + return new FilteredDynamicFilter(filterId, columnHandle, values, type, filter, filterExpression); } else { return create(filterId, columnHandle, values, type); diff --git a/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/FilteredDynamicFilter.java b/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/FilteredDynamicFilter.java index 67f2ec20d..e192c731a 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/FilteredDynamicFilter.java +++ b/presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/FilteredDynamicFilter.java @@ -16,6 +16,7 @@ package io.prestosql.spi.dynamicfilter; import com.google.common.collect.ImmutableList; import io.prestosql.spi.connector.ColumnHandle; +import io.prestosql.spi.relation.RowExpression; import java.util.List; import java.util.Optional; @@ -28,11 +29,18 @@ public class FilteredDynamicFilter extends HashSetDynamicFilter { private final Optional> filter; + private final Optional filterExpression; - public FilteredDynamicFilter(String filterId, ColumnHandle columnHandle, Set valueSet, Type type, Optional> filter) + public FilteredDynamicFilter(String filterId, ColumnHandle columnHandle, Set valueSet, Type type, Optional> filter, Optional filterExpression) { super(filterId, columnHandle, valueSet, type); this.filter = requireNonNull(filter, "filter is null"); + this.filterExpression = requireNonNull(filterExpression, "filterExpression is null"); + } + + public Optional getFilterExpression() + { + return filterExpression; } @Override @@ -52,7 +60,7 @@ public class FilteredDynamicFilter @Override public DynamicFilter clone() { - FilteredDynamicFilter filteredDynamicFilter = new FilteredDynamicFilter(filterId, columnHandle, valueSet, type, filter); + FilteredDynamicFilter filteredDynamicFilter = new FilteredDynamicFilter(filterId, columnHandle, valueSet, type, filter, filterExpression); filteredDynamicFilter.setMin(min); filteredDynamicFilter.setMax(max); return filteredDynamicFilter;