Presto 中Hive数据源filter算子下推代码走读-CFANZ编程社区

一. 引言

支持谓词下推是Presto中一个重要的性能优化能力，本文将通过代码走读来探索Presto在Hive数据源中是如何实现filter下推的以及以parquet格式的Hive表为例探索在数据源tableScan的时候如何应用filter信息进行优化的。

本文以带过滤条件查询“select id from t1 where id < 5 and id > 1 and name = '22'”为例进行代码走读。

二. 下推条件推给TableScan

上边的filter算子“where id < 5 and id > 1 and name = '22' " 是在PushPredicateIntoTableScan的RBO优化的时候，将filter条件压给数据源的，代码如下：

PushPredicateIntoTableScan::apply

     pushFilterIntoTableScan

          decomposedPredicate = ExpressionDomainTranslator.fromPredicate

                TableScanNode tableScan = new TableScanNode(...newDomain...)

经过上述代码优化后，TableScan的Node节点便含有了filter条件封装成的Domain信息，后续coordaintor会将带有filter的TableScan 分发给Woker进行数据扫描。

三. Hive 数据源中对filter条件的读取优化

fiter算子在Hive数据源中的读取优化在ParquetPageSourceFactory::createParquetPageSource中完成：

public static ParquetPageSource createParquetPageSource(    
        ....
        // parquetTupleDomain封装着sql的filter条件的封装
        TupleDomain<ColumnDescriptor> parquetTupleDomain = getParquetTupleDomain(descriptorsByPath, effectivePredicate);
        Predicate parquetPredicate = buildPredicate(requestedSchema, parquetTupleDomain, descriptorsByPath);
        final ParquetDataSource finalDataSource = dataSource;
        ImmutableList.Builder<BlockMetaData> blocks = ImmutableList.builder();
        for (BlockMetaData block : footerBlocks.build()) {
            // 确认是否可以过滤掉parquet 文件中的一个block， 一个block是parquet中一个RowGroup
            if (predicateMatches(parquetPredicate, block, finalDataSource, descriptorsByPath, parquetTupleDomain, failOnCorruptedParquetStatistics)) {
                blocks.add(block);
            }
        }
        ...
    }
}

上述中predicateMatches的实现如下：

public static boolean predicateMatches(...)
{
    Map<ColumnDescriptor, Statistics<?>> columnStatistics = getStatistics(block, descriptorsByPath);
    if (!parquetPredicate.matches(block.getRowCount(), columnStatistics, dataSource.getId(), failOnCorruptedParquetStatistics)) {
        return false;
    }
	....
}

public boolean matches(...)
{
    ...
    Domain domain = getDomain(effectivePredicateDomain.getType(), numberOfRows, columnStatistics, id, column.toString(), failOnCorruptedParquetStatistics);
    // 其实这里就是拿parquet文件中的，每个rowgroup的统计信息构造成一个domain和filter的domain进行碰撞，row的统计信息中有该rowgroup数据的最大值，最小值，是否有空值等，如果rowgroup和filter有交值，则保留该rowgroup进行数据scan，否则直接跳过该rowgroup的扫描
    if (effectivePredicateDomain.intersect(domain).isNone()) {
        return false;
     }
     ....
}