Introduce DynamicFilterSource and DynamicPhysicalExpr #15568

adriangb · 2025-04-03T18:47:58Z

Work towards [Epic] A collection of dynamic filtering related items #15512

This PR introduces:

PhysicalExpr::snapshot to capture the current state of dynamic PhyscalExpr's for feeding into systems that match on concrete trait impls (namely PruningPredicate and protobuf serialization)
Internal (private) helpers DynamicFilterSource and DynamicPhysicalExpr to make it easier to build dynamic PhysicalExpr, and to test and justify the snapshotting.

adriangb · 2025-04-06T14:40:46Z

datafusion/physical-expr-common/src/physical_expr.rs

+    fn snapshot(&self) -> Result<Option<Arc<dyn PhysicalExpr>>> {
+        // By default, we return None to indicate that this PhysicalExpr does not
+        // have any dynamic references or state.
+        // This is a safe default behavior.
+        Ok(None)
+    }


I'm wondering if it would be reasonable to inject the schema here so that this can be used for the use case in #15057 as well: we could model a predicate like variant_get(col, 'foo') = 5 by making variant_get(col, 'foo') a dynamic physicalexpr that snapshots itself to "col.typed_value.foo.typed_value if the schema/types permit otherwise it snapshots itself to _variant_get_value(col.metadata, col.value).

What do you mean "inject" the schema?

Like make this something like?

fn snapshot(&self, &Schema) -> Result<Option<Arc<dyn PhysicalExpr>>> {

Yes precisely.

So instead of being just dynamic on some internal state it will be dynamic on the actual physical schema of the file it is being applied to. But maybe that should be its own API:

fn with_physical_schema(&self, schema: &Schema) -> Option<Arc<dyn PhysicalExpr>>

Then if something wants a snapshot that's specialized for a specific file schema it can call both methods.

in theory the schema of the input to a PhysicaExpr shouldn't change so any expr that needs it could hold a reference 🤔

Is that true? Where would it hold onto the reference from, data_type()? Can't the same PhysicalExpr be used in multiple contexts, e.g. with the same columns but different projections / different physical column orders? Put another way, is it guaranteed that data_type() can only be called once or that evaluate() is always called with the same schema?

To be clear: I don't think this is important for this PR. After reflection in https://github.com/apache/datafusion/pull/15568/files#r2033710338 I think if we want that we can add it as a new method in #15057

Is that true? Where would it hold onto the reference from, data_type()? Can't the same PhysicalExpr be used in multiple contexts, e.g. with the same columns but different projections / different physical column orders?

Not in my understanding -- while it could in theory work this way I don't think they are typically shared.

Put another way, is it guaranteed that data_type() can only be called once or that evaluate() is always called with the same schema?

In my mind what happens is that the PhysicalExpr is created with the schema information: https://docs.rs/datafusion/latest/datafusion/physical_expr/fn.create_physical_expr.html

But it doesn't hold a reference to the schema internally 🤔

alamb

After the API in #15566 I wonder if we still need a DynamicFilterSource anymore 🤔 It seems like any nodes with dynamic filters could just pass the predicate as part of ExecutionPlan::try_pushdown_filters

I do think we'll likely still need snapshot_filter though

adriangb · 2025-04-08T13:23:48Z

My view for DynamicFilterSource is not that it's a public API in DataFusion. It's a private implementation detail / helper that internal implementations of dynamic PhysicalExpr can use to simplify implementing dynamic PhysicalExpr by making a DynamicFilterSource and wrapping it with DynamicPhysicalExpr. Maybe we expose it publically if there's enough demand for it but I think it'll be private to begin with.

I realize this PR doesn't make that clear or structure the exports correctly atm. Probably makes sense for it to be a private module in the same crate as HashJoinExec, SortExec, etc. live.

adriangb · 2025-04-08T14:15:38Z

Okay nope this PR is set up correctly: dynamic_filter is a private module in datafusion-physical-plan, so we can use it from within SortExec/TopK, HashJoinExec, etc.

So really this entire PR is just:

Introducing the public method PhysicalExpr::snapshot
Adding a private helpers to be used by various implementations of dynamic filters

alamb

Thank you @adriangb -- I really like this API

Not only will it help with topk dynamic filtering, I think it is exactly the API someone would need if they wanted to implement dynamic filters for distributed joins -- for example 'snapshot' would make / create a bloom filter to send.

ALl in all I am super excited about where this is heading.

alamb · 2025-04-08T17:12:41Z

datafusion/physical-expr-common/src/physical_expr.rs

+    /// Callers however should *not* assume anything about the returned expressions
+    /// since callers and implementers may not agree on what "simple" or "built-in"
+    /// means.
+    /// In other words, if you need to searlize a `PhysicalExpr` across the wire


alamb · 2025-04-08T17:14:04Z

datafusion/physical-expr-common/src/physical_expr.rs

+    fn snapshot(&self) -> Result<Option<Arc<dyn PhysicalExpr>>> {
+        // By default, we return None to indicate that this PhysicalExpr does not
+        // have any dynamic references or state.
+        // This is a safe default behavior.
+        Ok(None)
+    }


What do you mean "inject" the schema?

Like make this something like?

fn snapshot(&self, &Schema) -> Result<Option<Arc<dyn PhysicalExpr>>> {

datafusion/proto/src/physical_plan/to_proto.rs

alamb · 2025-04-08T17:16:40Z

datafusion/physical-expr-common/src/physical_expr.rs

@@ -283,6 +284,51 @@ pub trait PhysicalExpr: Send + Sync + Display + Debug + DynEq + DynHash {
    /// See the [`fmt_sql`] function for an example of printing `PhysicalExpr`s as SQL.
    ///
    fn fmt_sql(&self, f: &mut Formatter<'_>) -> fmt::Result;
+
+    /// Take a snapshot of this `PhysicalExpr` if it is dynamic.


Suggested change

/// Take a snapshot of this `PhysicalExpr` if it is dynamic.

/// Take a snapshot of this `PhysicalExpr`, if it is dynamic.

///

alamb · 2025-04-08T17:17:41Z

datafusion/physical-expr-common/src/physical_expr.rs

+    /// This is used to capture the current state of `PhysicalExpr`s that may contain
+    /// dynamic references to other operators in order to serialize it over the wire
+    /// or treat it via downcast matching.


Suggested change

/// This is used to capture the current state of `PhysicalExpr`s that may contain

/// dynamic references to other operators in order to serialize it over the wire

/// or treat it via downcast matching.

/// "Dynamic" in this case means containing references to structures that may change

/// during plan execution, such as hash tables.

///

/// This method is used to capture the current state of `PhysicalExpr`s that may contain

/// dynamic references to other operators in order to serialize it over the wire

/// or treat it via downcast matching.

alamb · 2025-04-08T17:22:58Z

My view for DynamicFilterSource is not that it's a public API in DataFusion. It's a private implementation detail / helper that internal implementations of dynamic PhysicalExpr can use to simplify implementing dynamic PhysicalExpr by making a DynamicFilterSource and wrapping it with DynamicPhysicalExpr. Maybe we expose it publically if there's enough demand for it but I think it'll be private to begin with.

This might be a good thing to note in the comments of DynamicFilterSource to make it clearer to future readers

adriangb · 2025-04-08T17:28:41Z

Thank you for the review, I'll address comments this afternoon

adriangb · 2025-04-08T17:48:54Z

My view for DynamicFilterSource is not that it's a public API in DataFusion. It's a private implementation detail / helper that internal implementations of dynamic PhysicalExpr can use to simplify implementing dynamic PhysicalExpr by making a DynamicFilterSource and wrapping it with DynamicPhysicalExpr. Maybe we expose it publically if there's enough demand for it but I think it'll be private to begin with.

This might be a good thing to note in the comments of DynamicFilterSource to make it clearer to future readers

Done in 75f4e40 😄

adriangb · 2025-04-08T17:50:50Z

I think it is exactly the API someone would need if they wanted to implement dynamic filters for distributed joins -- for example 'snapshot' would make / create a bloom filter to send

Yep my goal was to create a flexible interface that can be used for joins, distributed things, etc. I hope it works out in the end!

berkaysynnada

This is looking very good, thank you again @adriangb. I just need a bit help to understand some points

datafusion/physical-expr-common/src/physical_expr.rs

berkaysynnada · 2025-04-09T15:52:19Z

datafusion/physical-expr/src/expressions/dynamic_filters.rs

+/// [`PhysicalExpr::evaluate`], [`PhysicalExpr::data_type`], and [`PhysicalExpr::nullable`].
+/// It also implements [`PhysicalExpr::snapshot`] by forwarding the call to [`DynamicFilterSource::snapshot_current_filters`].
+#[derive(Debug)]
+pub struct DynamicFilterPhysicalExpr {


Is this going to be the shared object between sources and dynamic filter introducing operators?

Yes. This becomes an Arc<dyn PhysicalExpr> and is what gets pushed down into the source but holds onto a reference to the operator.

berkaysynnada · 2025-04-09T15:57:52Z

datafusion/physical-expr-common/src/physical_expr.rs

+    ///
+    /// Note for implementers: this method should *not* handle recursion.
+    /// Recursion is handled in [`snapshot_physical_expr`].
+    fn snapshot(&self) -> Result<Option<Arc<dyn PhysicalExpr>>> {


All existing expressions no need to implement this, right? Even no need another structure implementing this other than DynamicFilterPhysicalExpr, as it seems to be designed to handle all tricks

Yes I think so. No existing expressions need to implement this, unless that would be helpful to somehow simplify themselves into something that PruningPredicate or serialization can handle, but I don't know of any cases where that would make sense.

But we leave the door open for someone to implement another PhysicalExpr that does not use DynamicFilterPhysicalExpr.

berkaysynnada · 2025-04-09T16:29:48Z

datafusion/physical-expr/src/expressions/dynamic_filters.rs

+/// accessible within the crate.
+/// If you would like to use this trait in your own code, please open an issue
+/// to discuss the use case and we can consider making it public.
+pub trait DynamicFilterSource:


There will be different structs for different filter introducing streams or they will utilize a common struct?

The idea was that there would be different ones. For example TopK will have something like:

struct TopKDynamicFilterSource { ... } impl DynamicFilterSource for TopKDynamicFilterSource { ... }

But now that I think about it it will probably be more helpful to provide one concrete implementation of:

struct DynamicPhysicalExpr { inner: Arc<RwLock<Arc<dyn PhysicalExpr>>> } impl PhysicalExpr for DynamicPhysicalExpr { // similar to current }

I think that would work for at least some use cases and is a good place to start.

Wdyt @berkaysynnada ?

@berkaysynnada please take a look at bdbd438, I think it makes things simpler 😄

@berkaysynnada please take a look at bdbd438, I think it makes things simpler 😄

It seems better to me as well :D

alamb · 2025-04-10T10:16:14Z

datafusion/physical-expr/src/expressions/dynamic_filters.rs

+
+impl PartialEq for DynamicFilterPhysicalExpr {
+    fn eq(&self, other: &Self) -> bool {
+        let inner = self.current().expect("Failed to get current expression");


For equality I wonder if this would be simpler just to do a pointer check -- I think the instance has to be the same 🤔

jayzhan211 · 2025-04-10T11:57:20Z

datafusion/physical-expr/src/expressions/dynamic_filters.rs

+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(Arc::new(Self {
+            children: self.children.clone(),
+            remapped_children: Some(children),


This doesn't look correct to me.

with_new_children is the function used when we rewrite the plan with the new children.

children and other properties i.e. cache is updated with the given children: Vec<Arc<dyn PhysicalExpr>>.

~~I think you should find another way to handle the remapped_children logic. PlanContext with the custom payload may helps~~

It seems what you need is to update with new children in with_new_children then you don't need remapped_children anymore

@jayzhan211 is right, I didn't notice that. All children, remapped_children and inner fields should be updated here accordingly. Maybe we can left this as not_implemented!()

Something like

fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> { self.children.iter().collect() } fn with_new_children( self: Arc<Self>, children: Vec<Arc<dyn PhysicalExpr>>, ) -> Result<Arc<dyn PhysicalExpr>> { let current = self .inner .read() .map_err(|_| { datafusion_common::DataFusionError::Execution( "Failed to acquire read lock for inner".to_string(), ) })? .clone(); let new_cur = current .transform_up(|expr| { if let Some(pos) = self .children .iter() .position(|c| c.as_ref() == expr.as_ref()) { let new_child = Arc::clone(&children[pos]); Ok(Transformed::yes(new_child)) } else { // Otherwise, just return the expression Ok(Transformed::no(expr)) } }); Ok(Arc::new(Self { children, remapped_children: None, inner: Arc::new(RwLock::new(new_cur.data().unwrap())), data_type: Arc::clone(&self.data_type), nullable: Arc::clone(&self.nullable), })) }

and I think we don't need Arc<RwLock<T>> for inner

We would have to also keep track of remapped_children to re-apply the transformation in update / whenever the expression changes. But yes what you are suggestion is to do the update immediately instead of when curent() is called, which makes sense.

update is equivalent to with_new_children

No it is not. update() replaces the inner PhysicalExpr while preserving the outer PhysicalExpr. new_with_children is the exact opposite: it replaces the outer PhysicalExpr (because the children changed) while keeping the same inner PhysicalExpr. update() is intended to be called by the producer of the DynamicPhysicalExpr only while with_new_children is called by the consumers

Why do they need to be updated here? Children get remapped dynamically when you call current().

What if we are doing plan rewrite and want a completely different children in DynamicFilterPhysicalExpr?

impl<T: ConcreteTreeNode> TreeNode for T { fn apply_children<'n, F: FnMut(&'n Self) -> Result<TreeNodeRecursion>>( &'n self, f: F, ) -> Result<TreeNodeRecursion> { self.children().iter().apply_until_stop(f) } fn map_children<F: FnMut(Self) -> Result<Transformed<Self>>>( self, f: F, ) -> Result<Transformed<Self>> { let (new_self, children) = self.take_children(); if !children.is_empty() { let new_children = children.into_iter().map_until_stop_and_collect(f)?; // Propagate up `new_children.transformed` and `new_children.tnr` along with // the node containing transformed children. new_children.map_data(|new_children| new_self.with_new_children(new_children)) } else { Ok(Transformed::no(new_self)) } } }

with_new_children is the function called when you do the plan rewrite, but
in your code the children isn't changed but remapped_children is updated instead. IMO this seems like you rely on the with_new_children function for DynamicFilterPhysicalExpr requirement, and didn't actually have the ability to update the source of filters with the new one.

fn with_new_children( self: Arc<Self>, children: Vec<Arc<dyn PhysicalExpr>>, ) -> Result<Arc<dyn PhysicalExpr>> { Ok(Arc::new(Self { children: self.children.clone(), remapped_children: Some(children), inner: Arc::clone(&self.inner), data_type: Arc::clone(&self.data_type), nullable: Arc::clone(&self.nullable), })) }

The result of with_new_children(children: Vec<Arc<dyn PhysicalExpr>>) should be logically equivalent to DynamicFilterPhysicalExpr::new(children)

pub struct DynamicFilterPhysicalExpr { /// The source of dynamic filters. `inner` in your design. children: Arc<dyn PhysicalExpr>, /// `children` in your design. original_columns: Vec<Arc<dyn PhysicalExpr>>, /// `remapped_children` in your design. remapped_columns: Option<Vec<Arc<dyn PhysicalExpr>>>, } impl PhysicalExpr for DynamicFilterPhysicalExpr { fn with_new_children( self: Arc<Self>, children: Vec<Arc<dyn PhysicalExpr>>, ) -> Result<Arc<dyn PhysicalExpr>> { // the source is updated Ok(Arc::new(Self { children, ... })) } } impl DynamicFilterPhysicalExpr { // use this instead of `reassign_predicate_columns` to avoid the `with_new_children` fn set_remapped_children( &mut self, remapped_children: Vec<Arc<dyn PhysicalExpr>>, ) { self.remapped_children = Some(remapped_children); } fn update(new_expr: Arc<dyn PhysicalExpr>) -> Result<()> { } // Update the columns directly without transform API that call `with_new_children` fn remap_children() {} }

I must be missing something. Maybe you can make a PR that keeps the tests passing with your design?

The point is that:

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

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

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<...>>.

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.

DynamicFilterPhysicalExpr(Vec<Arc>) is enough

Again I'm sorry if I'm missing something, but how would this work to share a single updatable reference across multiple copies of the PhysicalExpr (eg because with_new_children was called)?

jayzhan211 · 2025-04-10T12:01:52Z

datafusion/physical-expr/src/expressions/dynamic_filters.rs

+            children,
+            remapped_children: None, // Initially no remapped children
+            inner: Arc::new(RwLock::new(inner)),
+            data_type: Arc::new(RwLock::new(None)),


If this is for test purpose, I think we can use with_data_type, and gated with cfg(test)

I'm not sure what you mean. Are you wanting to avoid initializing these fields if we're not in tests?

Okay will look into that in a follow up PR. My thought was that it's pretty cheap to initialize them.

adriangb · 2025-04-10T14:22:15Z

@berkaysynnada @jayzhan211 I added a test in f59577c that shows how this interacts with ParquetSource and should help with confusion in #15568 (comment)

Co-authored-by: Andrew Lamb <[email protected]>

Co-authored-by: Berkay Şahin <[email protected]>

alamb · 2025-04-10T18:09:09Z

SInce this is an internal API that is not yet hooked up, let's merge merge it in and keep iterating on main (so we can hook it up and start testing it asap). If we can find simpler ways to implement some of the functions as we go forward, such as suggested by @jayzhan211 that will be great.

alamb · 2025-04-10T18:09:13Z

Thanks everyone!

berkaysynnada · 2025-04-11T10:27:19Z

I couldn't fully follow but, do we have remaining issues known with DynamicFilterPhysicalExpr internals?

adriangb · 2025-04-11T10:41:34Z

I think @jayzhan211 still has concerns but I'm having trouble understanding what they translate to in practice

alamb · 2025-04-11T10:42:32Z

I think some of the concerns will become more clear as we connect this code all up together and see it in action. As it is currently not yet hooked up I think it is hard to reason about some of the potential concerns.

jayzhan211 · 2025-04-11T13:49:13Z

#15685

Upd: @adriangb, DynamicFilterPhysicalExpr is PhysicalExprRef, instead of carrying dynamic_filter_1 as the output of reassign_predicate_columns, we create snapshot with schema. Since we have PhysicalExprRef only in the struct, so many things are simple now.

@adriangb, the snapshot is different but I think the overall idea should be the same, while we avoid remapping each time we call evaluate() and I think the with_new_children and update are the same now. I couldn't remove Arc<RwLock> yet because we can't mutate in with_new_children, it might be possible to avoid it with some tricks

* update * Add file * fix * Add remap children test * fmt * more comments * fmt * Apply suggestions from code review Co-authored-by: Andrew Lamb <[email protected]> * Add some more comments * Update datafusion/physical-expr-common/src/physical_expr.rs Co-authored-by: Berkay Şahin <[email protected]> * Simplify trait to concrete impl * clippy * remap children in update() * better test --------- Co-authored-by: Andrew Lamb <[email protected]> Co-authored-by: Berkay Şahin <[email protected]>

github-actions bot added physical-expr Changes to the physical-expr crates optimizer Optimizer rules proto Related to proto crate labels Apr 3, 2025

adriangb mentioned this pull request Apr 3, 2025

[Epic] A collection of dynamic filtering related items #15512

Open

4 tasks

adriangb commented Apr 6, 2025

View reviewed changes

alamb mentioned this pull request Apr 8, 2025

Weekly Plan (Andrew Lamb) April 7, 2025 #15616

Closed

12 tasks

alamb reviewed Apr 8, 2025

View reviewed changes

adriangb force-pushed the dynamic-physical-expr branch from b43c39d to f13729c Compare April 8, 2025 14:15

adriangb requested a review from alamb April 8, 2025 14:16

alamb approved these changes Apr 8, 2025

View reviewed changes

berkaysynnada mentioned this pull request Apr 8, 2025

ExecutionPlan: add APIs for filter pushdown & optimizer rule to apply them #15566

Merged

berkaysynnada reviewed Apr 9, 2025

View reviewed changes

adriangb requested review from alamb and berkaysynnada April 9, 2025 22:24

alamb approved these changes Apr 10, 2025

View reviewed changes

berkaysynnada approved these changes Apr 10, 2025

View reviewed changes

jayzhan211 reviewed Apr 10, 2025

View reviewed changes

adriangb added 5 commits April 10, 2025 09:22

update

9f8ca63

Add file

c8963a2

fix

6c2b74a

Add remap children test

218de1c

fmt

57a261b

adriangb and others added 9 commits April 10, 2025 09:22

more comments

04486d4

fmt

e744c17

Apply suggestions from code review

dfe5d4d

Co-authored-by: Andrew Lamb <[email protected]>

Add some more comments

f7799fd

Update datafusion/physical-expr-common/src/physical_expr.rs

05e46f1

Co-authored-by: Berkay Şahin <[email protected]>

Simplify trait to concrete impl

461bf73

clippy

a59b648

remap children in update()

01ec0bf

better test

3cbdd2b

adriangb force-pushed the dynamic-physical-expr branch from f59577c to 3cbdd2b Compare April 10, 2025 14:22

alamb merged commit 7eb710d into apache:main Apr 10, 2025
28 of 29 checks passed

jayzhan211 mentioned this pull request Apr 12, 2025

dynamic filter refactor #15685

Draft

mbutrovich mentioned this pull request Apr 25, 2025

Push Dynamic Join Predicates into Scan ("Sideways Information Passing", etc) #7955

Open

	/// Take a snapshot of this `PhysicalExpr` if it is dynamic.
	/// Take a snapshot of this `PhysicalExpr`, if it is dynamic.
	///

Introduce DynamicFilterSource and DynamicPhysicalExpr #15568

Introduce DynamicFilterSource and DynamicPhysicalExpr #15568

Uh oh!

Conversation

adriangb commented Apr 3, 2025 • edited by alamb Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adriangb commented Apr 8, 2025

Uh oh!

adriangb commented Apr 8, 2025

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb commented Apr 8, 2025

Uh oh!

adriangb commented Apr 8, 2025

Uh oh!

adriangb commented Apr 8, 2025

Uh oh!

adriangb commented Apr 8, 2025

Uh oh!

berkaysynnada left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adriangb Apr 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jayzhan211 Apr 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

adriangb commented Apr 3, 2025 •

edited by alamb

Loading

alamb left a comment •

edited

Loading

adriangb Apr 9, 2025 •

edited

Loading

jayzhan211 Apr 10, 2025 •

edited

Loading

adriangb Apr 10, 2025 •

edited

Loading

jayzhan211 Apr 11, 2025 •

edited

Loading

jayzhan211 Apr 11, 2025 •

edited

Loading

adriangb Apr 11, 2025 •

edited

Loading

jayzhan211 Apr 10, 2025 •

edited

Loading