dynamic filter refactor

[jayzhan211]

Which issue does this PR close?

• Closes #.

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

[adriangb]

I think for any alternative proposal to be viable it needs to work with the current tests without calling new methods, or making copies differently, etc. The test was crafted that way because it reflects what actually happens during execution based on #15301

[adriangb]

What is expected to call with_schema? This seems like a new method on DynamicFilterPhysicalExpr. We need reassign_predicate_columns to work, that's what gets called from within ParquetSource, etc.

[jayzhan211]

When you need reassign_predicate_columns, basically it re-project columns based on the provided schema.

[adriangb]

Right but the constraint is that we can't modify what reassign_predicate_columns and similar do internally, otherwise that's a ton of API churn. The existing design works within the confines of the existing APIs.

[jayzhan211]

We need reassign_predicate_columns to work, that's what gets called from within ParquetSource, etc

We can also change reassign_predicate_columns inside Parquet if that bring us to the better state. From my view, reassign_predicate_columns is the root cause why you ends up with_new_children that doesn't actually updating the "children".

[jayzhan211]

otherwise that's a ton of API churn.

We only use in DatafusionArrowPredicate, is there any other places we need to change?

The main point is that with_new_children in the main branch isn't doing the right thing, it should update the source filter instead, but you only update the remapped filter schema. I think the filter schema is "parameters" for remapping column indexes, it doesn't need to be part of the filter expressions at all.

[jayzhan211]

If you really want to keep reassign_predicate_columns for whatever reason, you should pass the inner of DynamicFilterPhysicalExpr instead, so you are only modifying the inner and not the whole DynamicFilterPhysicalExpr. The difference is that the with_new_children is not called on the DynamicFilterPhysicalExpr level.

[adriangb]

This means that evaluate no longer uses a version with remapped children right?

[jayzhan211]

now you evaluate based on the snapshot. snapshot is the remapped filter with your filter schema

[adriangb]

I feel like we're going in circles... this is the thing that was expected to be used to produce the snapshots... now we're using snapshots to produce it?

[jayzhan211]

I think the difference is that

Your version

1. we have source filter A
2. create dynamic filter a and b by different filter schema
3. you create snapshot a, b from them.
4. evaluate batches by snapshot
5. update source filter A to B
6. dynamic filter a, b remap based on source filter B when you call evaluate

My version

1. we have source filter A
2. create snapshot based on source filter A + filter schema A and B
3. evaluate batches by snapshot
4. update source filter A to B
5. create snapshot based on source filter B + filter schema A and B
6. evaluate batches by snapshot

[jayzhan211]

snapshot is actually an evaluated dynamic filter based on the schema.

Basically, what you have is just the filter expression. You provide the schema to remap the column indexes. You get yet another filter expression.

[adriangb]

@jayzhan211 how can this have multiple readers and a writer updating with some sort of write lock?

[jayzhan211]

I think we don't need it. Given a source filter, you create snapshot with the schema. Then you evaluate based on the remapped filter. When you need a new source filter, instead of updating it, just create a new one

[adriangb]

But how do you pipe the new filter down into other operators?

The whole point is that you can create a filter at planning time, bind it to a ParquetSource and a SortExec (for example) and then the SortExec can dynamically update it at runtime.

[jayzhan211]

The whole point is that you can create a filter at planning time, bind it to a ParquetSource and a SortExec (for example) and then the SortExec can dynamically update it at runtime.

instead of sending the filter down, the change I have is sending the filter schema down. It is used to create another filter (snapshot) in SortExec dynamically at runtime.

[adriangb]

How will writers have mutable access to this if they have to package it up in an Arc?

[jayzhan211]

#15568 (comment)

why the change is equivalent to yours in the high level idea.

1. DynamicFilterPhysicalExpr gets initialized at planning time with a known set of children but a placeholder expression (lit(true))

The same.

2. with_new_children is called making a new DynamicFilterPhysicalExpr but with the children replaced (let's ignore how that happens internally for now)

We need to replace children, and we can achieve this and get the result by snapshot

3. update is called on the original reference with an expression that references the original children. This is propagated to all references, including those with new children, because of the Arc<RwLock<...>>.

We can keep Arc<RwLock<T>> and update the inner or even create another new source filter.

4. evaluate is called on one of the references that previously had with_new_children called on it. Since update was called, which swapped out inner, the children of this new inner need to be remapped to the children that we currently expose externally.

We can call evaluate on snapshot because snapshot is already remapped. Your version need to remap for each evaluate called, but snapshot in my version done it once, and we evaluate on it without remap.

The improvement of this chanage

1. We have correct with_new_children because we update the source filter now.
2. DynamicFilterPhysicalExpr is basically filter expression: Arc<dyn PhysicalExpr>>. We have a simple interface with the same capability now.

Concern

1. reassign_predicate_columns is replaced by snapshot but you think we can't do this kind of change because of API churn. I think this is not an issue because only DatafusionArrowPredicate is used.
2. Do we need Lock for source filter? I think we can create another new DynamicFilterPhysicalExpr at all. But maybe there is some reasons we can't, we can discuss further on this.

[adriangb]

Hey @jayzhan211 thank you for putting the work into trying to clarify this.

At this point I think it would be best to wait for #15566 or a PR that replaces it to be merged so that we can work against an actual use case / implementation of these dynamic filters. Otherwise I think it's a bit hard to communicate in such abstract terms. Once we're looking at a concrete use case it will be easier to make a PR to replace this implementation.

Would it be okay with you to wait until that happens to continue this discussion?

Sorry if merging a PR with a bad implementation becomes problematic... luckily it's problematic for us, not end users, since this is all private implementations.