Prune Nested Fields for Parquet Columns

[zhenxiao]

Read necessary fields only for Parquet nested columns
Currently, Presto will read all the fields in a struct for Parquet columns.
e.g.

select s.a, s.b
from t

if it is a parquet file, with struct column s: {a int, b double, c long, d float}
current Presto will read a, b, c, d from s, and output just a and b

For columnar storage as Parquet or ORC, we could do better, by just reading the necessary fields. In the previous example, just read {a int, b double} from s. Not reading other fields to save IO.

This patch introduces an optional NestedFields in ColumnHandle. When optimizing the plan, PruneNestedColumns optimizer will visit expressions, and put candidate nested fields into ColumnHandle. When scanning parquet files, the record reader could use NestedFields to specify necessary fields only for parquet files.

This has an dependency on @jxiang 's #4714, which gives us the flexibility to specify metastore schemas differently from parquet file schemas.

@dain @martint @electrum @cberner @erichwang any comments are appreciated

[martint]

The hive connector shouldn't have a dependency on presto-main. In fact, the classes in presto-main are not guaranteed to be available to plugins (due to classloader isolation)

[zhenxiao]

get it. Really need RowType and RowField, which are in presto-main, not in spi. Is there any special reason RowType (MapType, etc.) is not in spi? Could we move RowType in spi? Or, maybe duplicate the code in presto-hive? @martint @cberner @dain

[zhenxiao]

@cberner we need to get RowType's Fields, and use the RowField name to look up the corresponding parquet schema in parquet files

[cberner]

Ya, you should be able to just use the TypeSignature all the names will be in there

[zhenxiao]

with @cberner help, get comments addressed, no longer need presto-main dependency any more
@cberner @dain @martint @electrum comments or suggestions are appreciated

[cberner]

@nezihyigitbasi do you have time to take a first look at this?

[nezihyigitbasi]

@cberner sure, will do.

[nezihyigitbasi]

@zhenxiao I gave this patch a try and I guess hit a bug. Once you fix the issue I will continue reviewing.

I first enabled the new optimizer in the config file and then created a test table

presto:nyigitbasi> create table nested_field_test as select CAST(row(1, 'a', 'b', 'c') AS ROW(f1 integer, f2 varchar, f3 varchar, f4 varchar)) as row_field;
CREATE TABLE: 1 row

presto:nyigitbasi> desc nested_field_test;
  Column   |                        Type                         | Comment
-----------+-----------------------------------------------------+---------
 row_field | row(F1 integer, F2 varchar, F3 varchar, F4 varchar) |
(1 row)

then when I query the table it fails:

presto:nyigitbasi> select row_field.F1 from nested_field_test;

Query 20160713_002306_00019_he7h3, FAILED, 1 node
http://localhost:8080/query.html?20160713_002306_00019_he7h3
Splits: 2 total, 0 done (0.00%)
CPU Time: 0.5s total,     0 rows/s,     0B/s, 37% active
Per Node: 0.0 parallelism,     0 rows/s,     0B/s
Parallelism: 0.0
1:13 [0 rows, 0B] [0 rows/s, 0B/s]

Query 20160713_002306_00019_he7h3 failed: Error opening Hive split s3n://netflix-dataoven-prod-users/hive/warehouse/nyigitbasi.db/nested_field_test/20160713_002223_00017_he7h3_a34c4336-28ce-4a72-9253-3529044d3710 (offset=0, length=548): Schema mismatch, metastore schema for row column row_field has 4 fields but parquet schema has 1 fields
com.facebook.presto.spi.PrestoException: Error opening Hive split s3n://netflix-dataoven-prod-users/hive/warehouse/nyigitbasi.db/nested_field_test/20160713_002223_00017_he7h3_a34c4336-28ce-4a72-9253-3529044d3710 (offset=0, length=548): Schema mismatch, metastore schema for row column row_field has 4 fields but parquet schema has 1 fields
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor.createParquetRecordReader(ParquetHiveRecordCursor.java:489)
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor.<init>(ParquetHiveRecordCursor.java:254)
    at com.facebook.presto.hive.parquet.ParquetRecordCursorProvider.createHiveRecordCursor(ParquetRecordCursorProvider.java:96)
    at com.facebook.presto.hive.HivePageSourceProvider.getHiveRecordCursor(HivePageSourceProvider.java:129)
    at com.facebook.presto.hive.HivePageSourceProvider.createPageSource(HivePageSourceProvider.java:107)
    at com.facebook.presto.spi.connector.classloader.ClassLoaderSafeConnectorPageSourceProvider.createPageSource(ClassLoaderSafeConnectorPageSourceProvider.java:44)
    at com.facebook.presto.split.PageSourceManager.createPageSource(PageSourceManager.java:48)
    at com.facebook.presto.operator.ScanFilterAndProjectOperator.createSourceIfNecessary(ScanFilterAndProjectOperator.java:292)
    at com.facebook.presto.operator.ScanFilterAndProjectOperator.isFinished(ScanFilterAndProjectOperator.java:180)
    at com.facebook.presto.operator.Driver.processInternal(Driver.java:375)
    at com.facebook.presto.operator.Driver.processFor(Driver.java:301)
    at com.facebook.presto.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:618)
    at com.facebook.presto.execution.TaskExecutor$PrioritizedSplitRunner.process(TaskExecutor.java:529)
    at com.facebook.presto.execution.TaskExecutor$Runner.run(TaskExecutor.java:665)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: Schema mismatch, metastore schema for row column row_field has 4 fields but parquet schema has 1 fields
    at com.google.common.base.Preconditions.checkArgument(Preconditions.java:145)
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor$ParquetStructConverter.createConverters(ParquetHiveRecordCursor.java:864)
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor$ParquetStructConverter.<init>(ParquetHiveRecordCursor.java:946)
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor.createGroupConverter(ParquetHiveRecordCursor.java:821)
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor.access$300(ParquetHiveRecordCursor.java:125)
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor$PrestoReadSupport.<init>(ParquetHiveRecordCursor.java:534)
    at com.facebook.presto.hive.parquet.ParquetHiveRecordCursor.createParquetRecordReader(ParquetHiveRecordCursor.java:428)
    ... 16 more

But if I disable the new optimizer I can query the table fine.

presto:nyigitbasi> set session optimize_nested_columns=false;
SET SESSION
presto:nyigitbasi> select row_field.F1 from nested_field_test;
 f1
----
  1
(1 row)

[zhenxiao]

@nezihyigitbasi Thank you, get it fixed. Your comments are appreciated.

[nezihyigitbasi]

do you need this as a field? This is not used anywhere, the constructor arg typeManager is passed to the createGroupConverter() call.

[zhenxiao]

thank you, @mbasmanova
I get comments addressed.

[mbasmanova]

@zhenxiao Some more comments.

[mbasmanova]

private

[mbasmanova]

since this method returns a list, extractDereferenceExpressions might be a better name

[mbasmanova]

this variable contains a list of dereference expressions used in the predicate, but the expressions themselves are not predicates, e.g. a.b < 10 is a predicate, but a.b is not. Since this variable is used only once, it can be inlined which removes the need to come up with a better name.

[mbasmanova]

as a general rule, method names start with a verb, e.g. pushDownDereferences

[mbasmanova]

This variable is not needed. You can swap key and value in pushdownExpressions instead.

        Map<Symbol, DereferenceExpression> pushdownExpressions = expressions.entrySet().stream()
                .filter(entry -> !isCastRowType(entry.getKey()))
                .filter(entry -> outputSymbols.contains(getOnlyElement(extractAll(entry.getKey()))))
                .collect(toImmutableMap(Map.Entry::getValue, Map.Entry::getKey));
...
        ImmutableMap.Builder<Symbol, Symbol> symbolsBuilder = ImmutableMap.builder();
        pushdownExpressions.entrySet().stream()
                .forEach(entry -> symbolsBuilder.put(getOnlyElement(extractAll(entry.getValue())), entry.getKey()));
        Map<Symbol, Symbol> symbolsMap = symbolsBuilder.build();
...
        DereferenceExpression targetDereference = pushdownExpressions.get(targetSymbol);

[mbasmanova]

similar to other rules, this variable is not needed

[mbasmanova]

consider

        pushdownExpressions.entrySet().stream()
                .forEach(entry -> symbolsBuilder.put(getOnlyElement(extractAll(entry.getValue())), entry.getKey()));

[mbasmanova]

EqualJoinClause can only reference symbols; it cannot contain dereference expressions, can it?

[mbasmanova]

I think it would be clearer if dereferencePushDown returned a Result. Then this check would be simplified to if (result.isEmpty())

[zhenxiao]

thank you, @mbasmanova
I get comments addressed

[zhenxiao]

you are correct. this never happens. It is dead code. I will fix

[mbasmanova]

@zhenxiao A few questions.

[mbasmanova]

drop final

[mbasmanova]

dereferences is a set of DereferenceExpression and base is an instance of SymbolReference; will this ever return true? I don't think so. Looks like this whole if statement can be deleted and the outer loop simplified to

        while (base instanceof DereferenceExpression) {
            if (dereferences.contains(base)) {
                return true;
            }
            base = ((DereferenceExpression) base).getBase();
        }

[mbasmanova]

baseExists might be a better name

[mbasmanova]

node.getAssignments().getExpressions() is equivalent but easier to understand

[mbasmanova]

Simplify using Assignments::rewrite:

Assignments assignments = node.getAssignments().rewrite(new DereferenceReplacer(expressions));

[mbasmanova]

Reformat for readability (capitalize SQL keywords, split into multiple lines, replace t1 with t(msg)):

        assertPlan("WITH t(msg) AS (SELECT * FROM (VALUES ROW(CAST(ROW(1, 2.0) AS ROW(x BIGINT, y DOUBLE))))) " +
                        "SELECT msg.x FROM t WHERE msg.x > 10",
                output(ImmutableList.of("x"),
                    project(ImmutableMap.of("x", expression("msg.x")),
                            filter("msg.x > BIGINT '10'",
                                    values("msg")))));

However, I'd expect msg.x projection to be pushed down, e.g. I'd think the plan would look like

Output(x)
   Filter(x > 10)
      Project(x: msg.x)
         Values

Also, I removed the new rules from the optimizer and re-ran the tests. All tests but testDereferencePushdownJoin succeeded. Could you modify the tests to use queries that are affected by the new rules?

[zhenxiao]

let me rework the testcases
predicatePushDown will push down filter just on top of tablescan or values. It is OK we have dereference just on top of filter, as long as ScanFilterAndProject has dereferences, so that virtual nested columns could be generated by leveraging dereferences in connector. What do you think?

[mbasmanova]

@zhenxiao That's right. This is a subtle point that's not easy to get from reading the rules code. Perhaps, modify the rules to not do anything for project(filter(tablescan)) and project(filter(values)). It would also help to update the PR description to give examples of queries where dereference pushdown will be successful and show queries where new rules don't make a difference. For the individual rules, consider adding documentation. See com.facebook.presto.sql.planner.iterative.rule.PushProjectionThroughExchange for an example.

[zhenxiao]

thank you, @mbasmanova
Let me rework on the testcases. other comments are addressed.

[zhenxiao]

let me rework the testcases
predicatePushDown will push down filter just on top of tablescan or values. It is OK we have dereference just on top of filter, as long as ScanFilterAndProject has dereferences, so that virtual nested columns could be generated by leveraging dereferences in connector. What do you think?

[mbasmanova]

@zhenxiao Some further questions and comments.

[mbasmanova]

nit: inline this variable

[mbasmanova]

requireNonNull: this.expressions = requireNonNull(expressions, "expressions is null");

[mbasmanova]

This is a bit verbose. How do you feel about creating a top-level PushDownDereferences class and covert all these rules into inner classes?

dereferencePushDownRules = new PushDownDereferences(metadata, sqlParser).rules();

dereferencePushDownRules variable can then be inlined

[mbasmanova]

Why is it guaranteed that extractAll(entry.getKey()) returns just one element? I'm thinking of a dereference expression f(a, b).c which contains two symbols: a and b.

[mbasmanova]

What's the logic behind checking for cast to row?

[mbasmanova]

Why exit here? What if the predicate has dereferences that can be pushed down?

[mbasmanova]

This line is too long. It might be easier to read if rewritten to

        return Result.ofPlanNode(
                new JoinNode(
                    context.getIdAllocator().getNextId(),
                    joinNode.getType(),
                    leftChild,
                    rightChild,
                    joinNode.getCriteria(),
                    ImmutableList.<Symbol>builder()
                            .addAll(leftChild.getOutputSymbols())
                            .addAll(rightChild.getOutputSymbols())
                            .build(),
                    joinNode.getFilter().map(expression -> replaceDereferences(expression, expressions)),
                    joinNode.getLeftHashSymbol(),
                    joinNode.getRightHashSymbol(),
                    joinNode.getDistributionType()));

[mbasmanova]

this method is not used

[mbasmanova]

inline this variable and rightOutputs

        Assignments.Builder leftBuilder = Assignments.builder();
        leftBuilder.putIdentities(left.getOutputSymbols().stream()
                .filter(symbol -> !symbolsMap.containsKey(symbol))
                .collect(toImmutableList()));

        Assignments.Builder rightBuilder = Assignments.builder();
        rightBuilder.putIdentities(right.getOutputSymbols().stream()
                .filter(symbol -> !symbolsMap.containsKey(symbol))
                .collect(toImmutableList()));

[mbasmanova]

This can be simplified as

        Map<Symbol, Symbol> symbolsMap = pushdownExpressions.entrySet().stream()
                .collect(toImmutableMap(entry -> getOnlyElement(extractAll(entry.getValue())), Map.Entry::getKey));

[zhenxiao]

thank you, @mbasmanova
I get comments addressed

A few updates:

1. we need dereference pushdown for Project(Join). Or joins would output whole struct.
2. we need dereference pushdown for Project(Project), Project(Sort), so that dereferences could pass through Project and Sort, and down to Joins, if there is any.
3. ScanFilterAndProject will not change, as projections belongs to the same table are merged into ScanFilterAndProject
4. do not need Project(Filter), Project(Aggregation). Project either already happens at lower level, or predicate pushdown will guarantee Project(Filter(TableScan))), which is OK, as ScanFilterAndProject has all the projections, we could leverage it for following steps, e.g. getNestedColumnHandles.

[mbasmanova]

@zhenxiao Some further questions and comments.

[mbasmanova]

inline this variable

[mbasmanova]

wrong indentation

[mbasmanova]

private

[mbasmanova]

This is confusing because join's equi-clause can't contain expressions. The following might be clearer:

     * Transforms:
     * <pre>
     *  Project(a_x := a.msg.x)
     *    Join(a_y = b_y) => [a]
     *      Project(a_y := a.msg.y)
     *          Source(a)
     *      Project(b_y := b.msg.y)
     *          Source(b)
     *  </pre>
     * to:
     * <pre>
     *  Join(a_y = b_y) => [a_x]
     *    Project(a_x := a.msg.x, a_y := a.msg.y)
     *      Source(a)
     *    Project(b_y := b.msg.y)
     *      Source(b)
     * </pre>

[mbasmanova]

PushDownDereferenceThroughJoin might be a better name

[mbasmanova]

private

[mbasmanova]

private

[mbasmanova]

private

[mbasmanova]

field, field1, left and right names and not easy to follow. How about msg, a_y, b_y, etc.?

        assertPlan("WITH t(msg) AS (SELECT * FROM (VALUES ROW(CAST(ROW(1, 2.0) AS ROW(x BIGINT, y DOUBLE))))) " +
                        "SELECT b.msg.x FROM t a, t b WHERE a.msg.y = b.msg.y",
                output(ImmutableList.of("b_x"),
                        join(INNER, ImmutableList.of(equiJoinClause("a_y", "b_y")),
                                anyTree(
                                        project(ImmutableMap.of("a_y", expression("msg.y")),
                                                values("msg"))
                                ), anyTree(
                                        project(ImmutableMap.of("b_y", expression("msg.y"), "b_x", expression("msg.x")),
                                                values("msg"))))));

[zhenxiao]

thank you, @mbasmanova
I get comments addressed
will add more testcase coverage. could you please review when u are free?

[zhenxiao]

no need to pushdown (Cast As Row).field dereference, since lower level does not have this as Row.
Fix me if my understanding is wrong

[zhenxiao]

my bad. trying to mean key is a primitive type, but msg is a struct type, only pushdown dereferences. Will fix it

[zhenxiao]

my bad, it is push dereference through sort. Will fix and add unnest

[mbasmanova]

@zhenxiao Some further questions.

[mbasmanova]

Indeed, but the same can be said about any dereference where base is not a symbol, e.g. f(a).x. Why is it important to filter out cast, but not other functions? Could you add a test that covers these cases?

[mbasmanova]

I still don't understand this filter. entry.getKey() is a dereference expression coming from a projection above this join. extractAll(entry.getKey()) returns all symbols that are used in that dereference expression. outputSymbols are symbols produced by the join; these are inputs to the project node above. With project over join, the project can only use output symbols of the join. Hence, this check must always be true. Hence, it is not needed. Am I missing something? I commented out this check and the above check for cast and ran the tests. All passed.

[mbasmanova]

.stream().collect(toImmutableList()) is unnecessary; just change the type of the first argument of getDereferenceSymbolMap to Collection<Expression>.

[mbasmanova]

Shouldn't this filter use baseExists?

[zhenxiao]

No. This is to filter out join filters dereferences that not covered in projectExpressions.
baseExists is to filter out dereferences already exists in other dereferences base

[mbasmanova]

Could you scan through all the node types and see if there are more nodes to add here? I'm thinking about AssignUniqueId, MarkDistinct, Limit and SemiJoin.

[mbasmanova]

This project node is redundant. Why not remove it?

[zhenxiao]

there is RemoveRedundantIdentityProjections following PushDownDereferences, so that redundant project node is removed. I will remove it from comments. Or, do you think we should add RemoveRedundantIdentityProjections to PushDownDereferences?

[mbasmanova]

This project node is redundant.

[mbasmanova]

This project node is redundant.

[mbasmanova]

Other examples show an identity projection in the result, but it is missing here. At the same time the code seems to be actually generating an identity projection and there is no logic to remove it. Let's add it here or add logic to remove it to the rule.

[mbasmanova]

Tests only cover one-level deep dereferences, e.g. msg.x and msg.y. Would you add tests with more levels to test the logic of consolidating a.b.c and a.b into a.b?

[zhenxiao]

thank you, @mbasmanova
I get comments addressed
will add support for AssignUniqueId, MarkDistinct, Limit, Aggregation and SemiJoin

[zhenxiao]

No. This is to filter out join filters dereferences that not covered in projectExpressions.
baseExists is to filter out dereferences already exists in other dereferences base

[zhenxiao]

yep, should only pushdown dereferences with base as DeferenceExpression or SymbolReference. Will fix it

[zhenxiao]

there is RemoveRedundantIdentityProjections following PushDownDereferences, so that redundant project node is removed. I will remove it from comments. Or, do you think we should add RemoveRedundantIdentityProjections to PushDownDereferences?

[mbasmanova]

@zhenxiao

I get comments addressed
will add support for AssignUniqueId, MarkDistinct, Limit, Aggregation and SemiJoin

Sounds good. Ping me when the changes are ready for review.

[phd3]

@zhenxiao I was trying your patch on some examples and the following query fails with the PushDownDereferences optimizer.

with t(myint, arr) as
          (SELECT * FROM (VALUES (5, CAST(ARRAY[ROW(1,ROW(2,5)),ROW(2,ROW(3,9))] AS ARRAY(ROW(BIGINT, ROW(a BIGINT, b BIGINT)))))))
          select w.colb.a from t cross join unnest(arr) as w(cola, colb)

Plan for this query without the optimizer:

- Output[a] => [expr_16:bigint]
        a := expr_16
    - Project[] => [expr_16:bigint]
            expr_16 := "field_15".a
        - Unnest[replicate=, unnest=field_0:array(row(bigint, row(a bigint, b bigint)))] => [field_14:bigint, field_15:row(a bigint, b bigint)]
            - Values => [field_0:array(row(bigint, row(a bigint, b bigint)))]
                    (CAST($literal$array(row(integer,row(integer,integer)))(from_base64(VARCHAR(184) AwAAAFJPVw...))))

Do you know why that might be the case?

[phd3]

@zhenxiao Can you please add some tests for the unnest queries on multi-level complex data types in arrays ? for example, pushdown in cases where only field y.w.u is accessed after unnesting array(row(x BIGINT, y row(z BIGINT, w row(u BIGINT, v DOUBLE))))) ?

The testcases will help us have a better documentation of what this feature does and does not support.

[mbasmanova]

I assume #13271 superceds this one, hence, closing.