`ViewTable::scan` ignores filters and limits

[andygrove]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
We are missing an optimization opportunity when querying views. We should apply the supplied filter and limit to the logical plan before creating the physical plan in ViewTable::scan so that the query gets optimized with push down.

Describe the solution you'd like
As above.

Describe alternatives you've considered
None

Additional context
None

[HaoYang670]

Hi Andy. After learning and contributing to arrow-rs for several months, I'd like to focus on Datafusion from now to find more interesting things to learn. As this ticket is marked as good first issue, I guess this could be a good start for me. So I'd like to have a try!

[avantgardnerio]

Good luck @HaoYang670 , welcome, and please feel free to submit a draft PR or ask questions here if you run into issues!

[alamb]

I have a suggestion on how to implement this functionality, which is to "inline" the view definition so it is visible to the existing query optimizers rather than trying to do extra pushdown.

This might also be described as "replace view references with subqueries" -- in general I would expect the plan for a query that references to to be the same as the equivalent query if it had the view definitions inlined as subqueries.

For example

create table t as select * from (values (1), (2), (3)) as sq;

-- Create a view with a single predicate
create view v as select * from t where column1 != 1;

-- Run a query with a second predicate
explain select * from v where column1 != 2;

The actual plan is (note there are two Filter nodes 👎 ):

-- +---------------+---------------------------------------------------------------------+
-- | plan_type     | plan                                                                |
-- +---------------+---------------------------------------------------------------------+
-- | logical_plan  | Projection: #v.column1                                              |
-- |               |   Filter: #v.column1 != Int64(2)                                    |
-- |               |     TableScan: v projection=[column1]                               |
-- | physical_plan | ProjectionExec: expr=[column1@0 as column1]                         |
-- |               |   CoalesceBatchesExec: target_batch_size=4096                       |
-- |               |     FilterExec: column1@0 != 2                                      |
-- |               |       ProjectionExec: expr=[column1@0 as column1]                   |
-- |               |         CoalesceBatchesExec: target_batch_size=4096                 |
-- |               |           CoalesceBatchesExec: target_batch_size=4096               |
-- |               |             FilterExec: column1@0 != 1                              |
-- |               |               CoalesceBatchesExec: target_batch_size=4096           |
-- |               |                 RepartitionExec: partitioning=RoundRobinBatch(16)   |
-- |               |                   RepartitionExec: partitioning=RoundRobinBatch(16) |
-- |               |                     MemoryExec: partitions=1, partition_sizes=[1]   |
-- |               |                                                                     |
-- +---------------+---------------------------------------------------------------------+
-- 2 rows in set. Query took 0.005 seconds.

I expect a plan with a single filter with a predicate of column1 != 1 AND column1 != 2, which is exactly what happens when the view's definition is inlined as a subquery:

-- ❯ explain select * from (select * from t where column1 != 1) as sq where column1 != 2;
-- +---------------+---------------------------------------------------------------+
-- | plan_type     | plan                                                          |
-- +---------------+---------------------------------------------------------------+
-- | logical_plan  | Projection: #sq.column1                                       |
-- |               |   Projection: #t.column1, alias=sq                            |
-- |               |     Filter: #t.column1 != Int64(2) AND #t.column1 != Int64(1) |
-- |               |       TableScan: t projection=[column1]                       |
-- | physical_plan | ProjectionExec: expr=[column1@0 as column1]                   |
-- |               |   ProjectionExec: expr=[column1@0 as column1]                 |
-- |               |     CoalesceBatchesExec: target_batch_size=4096               |
-- |               |       FilterExec: column1@0 != 2 AND column1@0 != 1           |
-- |               |         RepartitionExec: partitioning=RoundRobinBatch(16)     |
-- |               |           MemoryExec: partitions=1, partition_sizes=[1]       |
-- |               |                                                               |
-- +---------------+---------------------------------------------------------------+
-- 2 rows in set. Query took 0.006 seconds.

[alamb]

One suggested implementation would be:

1. Add an optimizer pass (like InlineViewDefinitions or something) that special cases View references as the very first pass)
2. Remove ViewTable (or at least make scan throw an error)

[HaoYang670]

Hi @alamb, I am a little confused. Do you mean the behaviour of create view should be equivalent of subquery?

And create view is just to give a name binding to a query expression. And we always unfold (or inline) the view in the logical plan as a subquery, so that we could do different optimizations in different contexts. Is my understanding correct?

And this should be an optimization of logical plan, right?

[HaoYang670]

Another question is that should we treat treat view as a lazy value or an expression?
Let's say we have the following queries that use a view v multiple times

create table t as select * from (values (1), (2), (3)) as sq;

-- Create a view with a single predicate
create view v as select * from t where column1 != 1;

-- the first query using the view `v`
explain select * from v where column1 != 2;

-- the second query using the view `v`
explain select * from v where column1 != 3;

If the v is inlined in each query, we have to recompute it each time when using it. But as v is binded to an immutable dataframe, I'd like to access (or scan) it directly when it has been computed.

I guess the expected behaviour here is similar to lazy value, which we just inline the view at the first time and remember its value, and in the following queries, we can scan the view directly.

(BTW, actually I am not very clear about the difference between create table and create view)

[alamb]

Hi @alamb, I am a little confused. Do you mean the behaviour of create view should be equivalent of subquery?

I think I would say it like "create view" makes a entry in the catalog and then when querying if there is a view reference for that query the view should be be treated like a subquery

And create view is just to give a name binding to a query expression. And we always unfold (or inline) the view in the logical plan as a subquery, so that we could do different optimizations in different contexts. Is my understanding correct?

Yes, I think that is the ideal behavior for views

And this should be an optimization of logical plan, right?

I think that would be a good implementation, as it would work for both SQL and DataFrame interfaces.

Another question is that should we treat treat view as a lazy value or an expression?

I think the answer is expression (even though that would mean v is recomputed each query in your example). This is the core difference between views and tables. If a user wants to calculate the result once, they can use a table (or CREATE TABLE AS SELECT.... If they want a convenient alias that is expanded at query time, they use a view.

Trying to save the output of a view has at least two potential challenges:

1. When the the data in t changes (a new row got added, say) then the output needs to be updated (like some sort of cache invalidation)
2. The size of the view v may be very large and so caching it might be infeasible.

As an example, in TPCH schema maybe you don't want to have to explicitly write the joins from lineitem to order each time so you could define a view like

create view lineitems_detailed as 
select * from lineitems l JOIN orders o on l.o_key = o.o_key;

And then let's say you write queries with predicates like

select count(*) from lineitems_detailed where order_type = 'foo' AND o_date between '1/1/2022' AND '1/1/2023';

So in this case the actual values of the view are never "materialized"

[alamb]

Maybe we could use the https://docs.rs/datafusion/11.0.0/datafusion/datasource/datasource/trait.TableProvider.html#method.get_table_definition method to find the corresponding view definition and parse / expand it out?

[Dandandan]

I need to have a look at the code more, but I wonder whether it even has to be an optimization pass (e.g. inline view)? The current implementation of CTEs is also "just" using the definition as a subquery, I think the same could be done for the view definition.

[alamb]

I need to have a look at the code more, but I wonder whether it even has to be an optimization pass (e.g. inline view)? The current implementation of CTEs is also "just" using the definition as a subquery, I think the same could be done for the view definition.

I agree the view expansion could be done in the SQL planner (maybe that is the best place 🤷 ).

However, it would not be applied to the DataFrame API