3 ms·

Computation reuse via fusion in Amazon Athena

48 points by reckel 4y ago

BenoitP 4y ago
It took me some time to get a good grasp of the power of SQL; and it really kicked in when I learned about optimization rules. A query is a program that you rewrite, just like an optimizing compiler would. You state what you want; then you have different ways to fetch and match and massage data; and you can search through this space to produce a physical plan. Hopefully you used knowledge to weight parts to be optimized (table statistics, like Java's JIT would detect hot spots). I find it fascinating to peer through database code to see what is going on. Lately, there's been new advances towards streaming databases, which bring a whole new design space. For example, now you have latency of individual new rows to optimize for, as opposed to batch it whole to optimize the latency of a dataset. Batch scanning will be benefit from better use of your CPU caches. And maybe you could have a hybrid system which reads history from a log and aggregates in a batched manner, and then switches to a stream-optimized execution plan when it reaches the end of the log. If you want to have a peek at that here are Flink's set of rules [1], generic and stream-specific ones. The names can be cryptic, but usually give a good sense of what is going on. For example: PushFilterIntoTableSourceScanRule makes the WHERE clause apply the earliest possible, to save some CPU/network bandwidth further down. PushPartitionIntoTableSourceScanRule tries to make a fan-out/shuffle happen the earliest possible, so that parallelism can be made use of. [1] https://github.com/apache/flink/blob/5f8fb304fb5d68cdb0b3e3c6203691b14be92935/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala https://github.com/apache/flink/blob/5f8fb304fb5d68cdb0b3e3c...
gavinray 4y ago
This is really neat, but I'm curious how much of this is novel? I don't know much about relational algebra and query planning/optimization. But it seemed like many of these things were "no-brainers" (completely duplicate query conditions, etc) Is the point more the general-purpose framework rather than the query outcomes/rules? I took one of the queries from the paper (the first one), and used Apache Calcite to run the same query, on the same schema https://github.com/GavinRay97/athena-paper-calcite-sandbox/blob/fffbb5b5d3fa741ebf3e97838dfc7d005d9d1283/src/main/java/org/example/Main.java#L29-L33 https://github.com/GavinRay97/athena-paper-calcite-sandbox/b... Here is the original plan for Query 1: LogicalSort(sort0=[$0], sort1=[$1], dir0=[ASC], dir1=[ASC], fetch=[100]) LogicalProject(exprs=[[$5, $33, $55]]) LogicalFilter(condition=[AND(=($51, $53), <=($55, *(0.1:DECIMAL(2, 1), $52)), =($0, $53), =($29, $54))]) LogicalJoin(condition=[true], joinType=[inner]) LogicalJoin(condition=[true], joinType=[inner]) LogicalJoin(condition=[true], joinType=[inner]) LogicalTableScan(table=[[tpc_ds, STORE]]) LogicalTableScan(table=[[tpc_ds, ITEM]]) LogicalAggregate(group=[{0}], AVE=[AVG($1)]) LogicalProject(inputs=[0], exprs=[[$2]]) LogicalAggregate(group=[{0, 1}], REVENUE=[SUM($2)]) LogicalProject(exprs=[[$7, $2, $13]]) LogicalFilter(condition=[AND(=($0, $23), >=($26, 1212), <=($26, 1247))]) LogicalJoin(condition=[true], joinType=[inner]) LogicalTableScan(table=[[tpc_ds, STORE_SALES]]) LogicalTableScan(table=[[tpc_ds, DATE_DIM]]) LogicalAggregate(group=[{0, 1}], REVENUE=[SUM($2)]) LogicalProject(exprs=[[$7, $2, $13]]) LogicalFilter(condition=[AND(=($0, $23), >=($26, 1212), <=($26, 1247))]) LogicalJoin(condition=[true], joinType=[inner]) LogicalTableScan(table=[[tpc_ds, STORE_SALES]]) LogicalTableScan(table=[[tpc_ds, DATE_DIM]]) After using the default query planner rules, it gives this: EnumerableProject(S_STORE_NAME=[$8], I_ITEM_DESC=[$36], REVENUE=[$2]) EnumerableLimit(fetch=[100]) EnumerableSort(sort0=[$8], sort1=[$36], dir0=[ASC], dir1=[ASC]) EnumerableMergeJoin(condition=[AND(=($0, $54), <=($2, *(0.1:DECIMAL(2, 1), $55)))], joinType=[inner]) EnumerableMergeJoin(condition=[AND(=($0, $3), =($1, $32))], joinType=[inner]) EnumerableSort(sort0=[$0], sort1=[$1], dir0=[ASC], dir1=[ASC]) EnumerableProject(SS_STORE_SK=[$1], SS_ITEM_SK=[$0], REVENUE=[$2]) EnumerableAggregate(group=[{30, 35}], REVENUE=[$SUM0($41)]) EnumerableHashJoin(condition=[=($0, $28)], joinType=[inner]) EnumerableFilter(condition=[SEARCH($3, Sarg[[1212..1247]])]) EnumerableTableScan(table=[[tpc_ds, DATE_DIM]]) EnumerableTableScan(table=[[tpc_ds, STORE_SALES]]) EnumerableSort(sort0=[$0], sort1=[$29], dir0=[ASC], dir1=[ASC]) EnumerableNestedLoopJoin(condition=[true], joinType=[inner]) EnumerableTableScan(table=[[tpc_ds, STORE]]) EnumerableTableScan(table=[[tpc_ds, ITEM]]) EnumerableSort(sort0=[$0], dir0=[ASC]) EnumerableProject(SS_STORE_SK=[$0], AVE=[/($1, $2)]) EnumerableAggregate(group=[{1}], agg#0=[$SUM0($2)], agg#1=[COUNT()]) EnumerableAggregate(group=[{30, 35}], REVENUE=[$SUM0($41)]) EnumerableHashJoin(condition=[=($0, $28)], joinType=[inner]) EnumerableFilter(condition=[SEARCH($3, Sarg[[1212..1247]])]) EnumerableTableScan(table=[[tpc_ds, DATE_DIM]]) EnumerableTableScan(table=[[tpc_ds, STORE_SALES]])
- aseipp 4y ago
  The paper partially addresses this but Athena is a streaming engine, so it doesn't work in the same way as you might expect. The query planner has to take this into account. So a normal planner might see the CTE and then "materialize" it by pushing rows somewhere in an "intermediate place", like memory so the tuples can then be used by the outer queries, sharing the single result. But Athena does not do this; it doesn't materialize the CTE in memory or anywhere first, it streams from producer operators to consumer operators directly, and can begin evaluating parts of the plan immediately. So a hash join is out completely. Just as a more general thought experiment, in a distributed-first database engine, a good question is "where do you materialize intermediate results" -- in memory, in storage somewhere? -- and it's complicated to answer. So it's not enough to just stick a query plan into some compiler and see the result. The query planner is in servitude to the actual storage architecture, which is much more important to understand for realizing practical optimizations. A good rule of thumb is that a lot of optimizations in any compiler, like these, on their own aren't very novel. You can sketch them on paper and even implement them well in an ideal system. But getting them to work reliably under the design constraints of your system today is the novel part.
  - gavinray 4y ago
    > "... like memory so the tuples can then be used by the outer queries, sharing the single result. But Athena does not do this; it doesn't materialize the CTE in memory or anywhere first, it streams from producer operators to consumer operators directly, and can begin evaluating parts of the plan immediately." That is a little mind-bending. Very cool, this is the equivalent of having something like a "Consumer<T>" and "Producer<T>" and the producer emitting tuples? > "The query planner is in servitude to the actual storage architecture, which is much more important to understand for realizing practical optimizations." > "... But getting them to work reliably under the design constraints of your system today is the novel part." Great points, especially the first one which I didn't consider at all.

48 points by reckel 4y ago

It took me some time to get a good grasp of the power of SQL; and it really kicked in when I learned about optimization rules. A query is a program that you rewrite, just like an optimizing compiler would. You state what you want; then you have different ways to fetch and match and massage data; and you can search through this space to produce a physical plan. Hopefully you used knowledge to weight parts to be optimized (table statistics, like Java's JIT would detect hot spots). I find it fascinating to peer through database code to see what is going on. Lately, there's been new advances towards streaming databases, which bring a whole new design space. For example, now you have latency of individual new rows to optimize for, as opposed to batch it whole to optimize the latency of a dataset. Batch scanning will be benefit from better use of your CPU caches. And maybe you could have a hybrid system which reads history from a log and aggregates in a batched manner, and then switches to a stream-optimized execution plan when it reaches the end of the log. If you want to have a peek at that here are Flink's set of rules [1], generic and stream-specific ones. The names can be cryptic, but usually give a good sense of what is going on. For example: PushFilterIntoTableSourceScanRule makes the WHERE clause apply the earliest possible, to save some CPU/network bandwidth further down. PushPartitionIntoTableSourceScanRule tries to make a fan-out/shuffle happen the earliest possible, so that parallelism can be made use of. [1] https://github.com/apache/flink/blob/5f8fb304fb5d68cdb0b3e3c6203691b14be92935/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala https://github.com/apache/flink/blob/5f8fb304fb5d68cdb0b3e3c...

gavinray 4y ago

This is really neat, but I'm curious how much of this is novel? I don't know much about relational algebra and query planning/optimization. But it seemed like many of these things were "no-brainers" (completely duplicate query conditions, etc) Is the point more the general-purpose framework rather than the query outcomes/rules? I took one of the queries from the paper (the first one), and used Apache Calcite to run the same query, on the same schema https://github.com/GavinRay97/athena-paper-calcite-sandbox/blob/fffbb5b5d3fa741ebf3e97838dfc7d005d9d1283/src/main/java/org/example/Main.java#L29-L33 https://github.com/GavinRay97/athena-paper-calcite-sandbox/b... Here is the original plan for Query 1: LogicalSort(sort0=[$0], sort1=[$1], dir0=[ASC], dir1=[ASC], fetch=[100]) LogicalProject(exprs=[[$5, $33, $55]]) LogicalFilter(condition=[AND(=($51, $53), <=($55, *(0.1:DECIMAL(2, 1), $52)), =($0, $53), =($29, $54))]) LogicalJoin(condition=[true], joinType=[inner]) LogicalJoin(condition=[true], joinType=[inner]) LogicalJoin(condition=[true], joinType=[inner]) LogicalTableScan(table=[[tpc_ds, STORE]]) LogicalTableScan(table=[[tpc_ds, ITEM]]) LogicalAggregate(group=[{0}], AVE=[AVG($1)]) LogicalProject(inputs=[0], exprs=[[$2]]) LogicalAggregate(group=[{0, 1}], REVENUE=[SUM($2)]) LogicalProject(exprs=[[$7, $2, $13]]) LogicalFilter(condition=[AND(=($0, $23), >=($26, 1212), <=($26, 1247))]) LogicalJoin(condition=[true], joinType=[inner]) LogicalTableScan(table=[[tpc_ds, STORE_SALES]]) LogicalTableScan(table=[[tpc_ds, DATE_DIM]]) LogicalAggregate(group=[{0, 1}], REVENUE=[SUM($2)]) LogicalProject(exprs=[[$7, $2, $13]]) LogicalFilter(condition=[AND(=($0, $23), >=($26, 1212), <=($26, 1247))]) LogicalJoin(condition=[true], joinType=[inner]) LogicalTableScan(table=[[tpc_ds, STORE_SALES]]) LogicalTableScan(table=[[tpc_ds, DATE_DIM]]) After using the default query planner rules, it gives this: EnumerableProject(S_STORE_NAME=[$8], I_ITEM_DESC=[$36], REVENUE=[$2]) EnumerableLimit(fetch=[100]) EnumerableSort(sort0=[$8], sort1=[$36], dir0=[ASC], dir1=[ASC]) EnumerableMergeJoin(condition=[AND(=($0, $54), <=($2, *(0.1:DECIMAL(2, 1), $55)))], joinType=[inner]) EnumerableMergeJoin(condition=[AND(=($0, $3), =($1, $32))], joinType=[inner]) EnumerableSort(sort0=[$0], sort1=[$1], dir0=[ASC], dir1=[ASC]) EnumerableProject(SS_STORE_SK=[$1], SS_ITEM_SK=[$0], REVENUE=[$2]) EnumerableAggregate(group=[{30, 35}], REVENUE=[$SUM0($41)]) EnumerableHashJoin(condition=[=($0, $28)], joinType=[inner]) EnumerableFilter(condition=[SEARCH($3, Sarg[[1212..1247]])]) EnumerableTableScan(table=[[tpc_ds, DATE_DIM]]) EnumerableTableScan(table=[[tpc_ds, STORE_SALES]]) EnumerableSort(sort0=[$0], sort1=[$29], dir0=[ASC], dir1=[ASC]) EnumerableNestedLoopJoin(condition=[true], joinType=[inner]) EnumerableTableScan(table=[[tpc_ds, STORE]]) EnumerableTableScan(table=[[tpc_ds, ITEM]]) EnumerableSort(sort0=[$0], dir0=[ASC]) EnumerableProject(SS_STORE_SK=[$0], AVE=[/($1, $2)]) EnumerableAggregate(group=[{1}], agg#0=[$SUM0($2)], agg#1=[COUNT()]) EnumerableAggregate(group=[{30, 35}], REVENUE=[$SUM0($41)]) EnumerableHashJoin(condition=[=($0, $28)], joinType=[inner]) EnumerableFilter(condition=[SEARCH($3, Sarg[[1212..1247]])]) EnumerableTableScan(table=[[tpc_ds, DATE_DIM]]) EnumerableTableScan(table=[[tpc_ds, STORE_SALES]])

aseipp 4y ago
The paper partially addresses this but Athena is a streaming engine, so it doesn't work in the same way as you might expect. The query planner has to take this into account. So a normal planner might see the CTE and then "materialize" it by pushing rows somewhere in an "intermediate place", like memory so the tuples can then be used by the outer queries, sharing the single result. But Athena does not do this; it doesn't materialize the CTE in memory or anywhere first, it streams from producer operators to consumer operators directly, and can begin evaluating parts of the plan immediately. So a hash join is out completely. Just as a more general thought experiment, in a distributed-first database engine, a good question is "where do you materialize intermediate results" -- in memory, in storage somewhere? -- and it's complicated to answer. So it's not enough to just stick a query plan into some compiler and see the result. The query planner is in servitude to the actual storage architecture, which is much more important to understand for realizing practical optimizations. A good rule of thumb is that a lot of optimizations in any compiler, like these, on their own aren't very novel. You can sketch them on paper and even implement them well in an ideal system. But getting them to work reliably under the design constraints of your system today is the novel part.
- gavinray 4y ago
  > "... like memory so the tuples can then be used by the outer queries, sharing the single result. But Athena does not do this; it doesn't materialize the CTE in memory or anywhere first, it streams from producer operators to consumer operators directly, and can begin evaluating parts of the plan immediately." That is a little mind-bending. Very cool, this is the equivalent of having something like a "Consumer<T>" and "Producer<T>" and the producer emitting tuples? > "The query planner is in servitude to the actual storage architecture, which is much more important to understand for realizing practical optimizations." > "... But getting them to work reliably under the design constraints of your system today is the novel part." Great points, especially the first one which I didn't consider at all.