Extend payload join optimization for cases without map columns; fix b…

[mlyublena]

A collection of bug fixes for the payload join optimization, including the following:

• fixed cases where not all collected join keys belong to the left-most table
• abort optimization if the rewrite hides required columns
• abort rewrite for operators other than Select/Project/Scan/Join. This fixes wrong result bugs when the query contains aggregates and other operators that may change the cardinality of intermediate levels
• remove restriction for the fact table to have map or array columns. In the future we'll incorporate HBO to make a cost-based decision when to apply this optimization

[rschlussel]

Still reviewing the second commit. Can you add more context to the commit message about what the problem is/why it's a problem?

Also, can you make sure all of your commit messages follow the guidelines here: https://cbea.ms/git-commit/

[rschlussel]

I find the flow with all the context setting a bit confusing (even before this change), so I have a hard time being confident about the change.

It would be good to have more tests for cases that we don't expect the optimizer to kick in (e.g. left join with nested right or inner joins). Also, looks like the tests for correctness just validate the row count, but not the values or even column types. Would be good to have tests do a regular assertQuery correctness test for all the test queries to fully validate the results.

[mlyublena]

I find the flow with all the context setting a bit confusing (even before this change), so I have a hard time being confident about the change.

It would be good to have more tests for cases that we don't expect the optimizer to kick in (e.g. left join with nested right or inner joins). Also, looks like the tests for correctness just validate the row count, but not the values or even column types. Would be good to have tests do a regular assertQuery correctness test for all the test queries to fully validate the results.

The logic is basically as follows:

• when visiting nodes on the way down I'm collecting join keys in the join context until I hit a supported leaf node (SelectProjectScan node)
• if all join keys are contained are contained in the leaf node, I add an Aggregate on top to compute only the join keys without the payload, and add the expression to reattach back the payload in the context
• some of the intermediate nodes may require columns from the payload table that are not join keys (and the Aggregate hid them): if that is the case I abort the optimization and try from scratch with a new context. I refactored this logic to use a utility function usedColumns that returns the necessary columns for each node type, and use that in validateUsedColumnsForUnaryNode
• when I reach back the root of the join stack (isTopJoin = true) I reattach the payload using the IsNotDistinctFrom join. I also reset the context in that case to mark that the payload part has been consumed and doesn't need to get reattached

Regarding testing, I'd prefer to create a verifier suite (like this one that I used to debug the remaining issues: https://www.internalfb.com/intern/presto/verifier/results/?suite=test_payload_0808) as it gets harder and harder to enumerate all combinations of joins and unary operators

[mlyublena]

Note: will squash the commits after review and will update the commit description

[rschlussel]

some of the intermediate nodes may require columns from the payload table that are not join keys (and the Aggregate hid them): if that is the case I abort the optimization and try from scratch with a new context. I refactored this logic to use a utility function usedColumns that returns the necessary columns for each node type, and use that in validateUsedColumnsForUnaryNode

I wonder if it's possible to have the initial rewrite take into account the required columns to begin with. It seems not ideal that the rewrite can produce a "wrong" plan that we need to undo afterwards.

Regarding testing, I'd prefer to create a verifier suite (like this one that I used to debug the remaining issues: https://www.internalfb.com/intern/presto/verifier/results/?suite=test_payload_0808) as it gets harder and harder to enumerate all combinations of joins and unary operators

The verifier suite is great, but it's only available internally at Meta, and also doesn't run automatically with every PR as someone develops. Of course it's impossible to be 100% comprehensive, but we do also need sufficient correctness tests in the codebase itself (of both the plan and the results).

[mlyublena]

some of the intermediate nodes may require columns from the payload table that are not join keys (and the Aggregate hid them): if that is the case I abort the optimization and try from scratch with a new context. I refactored this logic to use a utility function usedColumns that returns the necessary columns for each node type, and use that in validateUsedColumnsForUnaryNode

I wonder if it's possible to have the initial rewrite take into account the required columns to begin with. It seems not ideal that the rewrite can produce a "wrong" plan that we need to undo afterwards.

during top-down traversal when we explore a node we might not have encountered all the join keys yet. So we won't know if a column used in an operator is a join key from below, or is not a join key and will be hidden by the rewrite.
Consider for example the following query: Join(Filter A=5 (join(join(R,S)), T)
When we explore the filter node we know what the top level join keys are but not what the join keys of the children are as we haven't traversed them yet. So we won't know if A is a key or not.

[rschlussel]

during top-down traversal when we explore a node we might not have encountered all the join keys yet. So we won't know if a column used in an operator is a join key from below, or is not a join key and will be hidden by the rewrite.
Consider for example the following query: Join(Filter A=5 (join(join(R,S)), T)
When we explore the filter node we know what the top level join keys are but not what the join keys of the children are as we haven't traversed them yet. So we won't know if A is a key or not.

But we'll know when we explore the child joins that A is a needed output column for a later node, so we should be able to choose not to rewrite at that point.

[mlyublena]

during top-down traversal when we explore a node we might not have encountered all the join keys yet. So we won't know if a column used in an operator is a join key from below, or is not a join key and will be hidden by the rewrite.
Consider for example the following query: Join(Filter A=5 (join(join(R,S)), T)
When we explore the filter node we know what the top level join keys are but not what the join keys of the children are as we haven't traversed them yet. So we won't know if A is a key or not.

But we'll know when we explore the child joins that A is a needed output column for a later node, so we should be able to choose not to rewrite at that point.

That's a great point - I ran into this in another optimization where I wanted to know the required columns from the top node (and that one was a rule-based optimization so things were even harder). I think it's worth having built-in support for passing required properties (such as required columns) from parent operators and making it a first-class citizen in the optimizer.

[mlyublena]

It would be good to have more tests for cases that we don't expect the optimizer to kick in (e.g. left join with nested right or inner joins). Also, looks like the tests for correctness just validate the row count, but not the values or even column types. Would be good to have tests do a regular assertQuery correctness test for all the test queries to fully validate the results.

Added more tests to cover for the various operators where we abort the transformation. Also added assertQuery tests to check for correctness of the result

[pranjalssh]

The comment says for binary and n-ary, but the code would do the same for unary operators as well.

[mlyublena]

recently fixed that, forgot to update/remove the comment

[mlyublena]

fixed

[pranjalssh]

It seems this is only called when plan is FilterNode. Why do we need to handle other cases?

[mlyublena]

I narrowed down the supported cases to only filter/project/scan + join, will clean up the implementation

[mlyublena]

fixed

[pranjalssh]

Can calling rewriter multiple times from a node lead to high time complexity?

[mlyublena]

ideally, we want to do this upfront and not have to discard plans after they have been computed - I'd prefer to do this in a separate PR as this one was meant to fix a large number of the original bugs with the optimization (and the optimization is disabled by default so no latency impact on production workloads at the moment)

[pranjalssh]

LGTM

[pranjalssh]

/s/cource/source