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

!848 [I3MZ4W] Support for Hive ORC Stripe Filtering using predicate 

Merge pull request !848 from Surya Sumanth/stripefiltering
This commit is contained in:
i-robot 2021-04-22 22:48:48 +08:00 committed by Gitee
parent 937bc37e75 eba9235039
commit e120e86739
8 changed files with 168 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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<HiveColumnHandle> predicate = TupleDomain.all();
if (dynamicFilterSupplier.isPresent() && dynamicFilters != null && !dynamicFilters.isEmpty()) {
List<HiveColumnHandle> 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<RowExpression> 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<HiveColumnHandle> getPredicate(DynamicFilter dynamicFilter, Type type, HiveColumnHandle hiveColumnHandle)
{
if (dynamicFilter instanceof CombinedDynamicFilter) {
List<DynamicFilter> filters = ((CombinedDynamicFilter) dynamicFilter).getFilters();
List<TupleDomain<HiveColumnHandle>> 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,

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

@ -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) {

84
presto-hive/src/test/java/io/prestosql/plugin/hive/TestHivePageSourceProvider.java
View File

@ -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<Object> 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<Object> 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<Object> 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<Object> 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());
}
}

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

@ -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);

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

@ -163,7 +163,7 @@ public class LocalDynamicFiltersCollector
if (predicates.containsKey(filterId)) {
Optional<RowExpression> filter = context.getFilter(filterId);
Optional<Predicate<List>> 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);
}

13
presto-main/src/main/java/io/prestosql/sql/planner/optimizations/RowExpressionPredicatePushDown.java
View File

@ -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

5
presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/DynamicFilterFactory.java
View File

@ -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<Predicate<List>> filter)
public static HashSetDynamicFilter create(String filterId, ColumnHandle columnHandle, Set values, DynamicFilter.Type type, Optional<Predicate<List>> filter, Optional<RowExpression> 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);

12
presto-spi/src/main/java/io/prestosql/spi/dynamicfilter/FilteredDynamicFilter.java
View File

@ -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<Predicate<List>> filter;
private final Optional<RowExpression> filterExpression;
public FilteredDynamicFilter(String filterId, ColumnHandle columnHandle, Set valueSet, Type type, Optional<Predicate<List>> filter)
public FilteredDynamicFilter(String filterId, ColumnHandle columnHandle, Set valueSet, Type type, Optional<Predicate<List>> filter, Optional<RowExpression> filterExpression)
{
super(filterId, columnHandle, valueSet, type);
this.filter = requireNonNull(filter, "filter is null");
this.filterExpression = requireNonNull(filterExpression, "filterExpression is null");
}
public Optional<RowExpression> 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;