> But please remember that there is usually no good plan for a federated join. There are a few algorithms (e.g. using bloom filters) that may help a little, but if each query ends up shipping a significant fraction of your table over the WAN, you’ve lost. Just replicate one of the tables.
Sure, fair enough. Not sure if you've heard of Apache Teiid before, but it was a Data Virtualization tool that provided some of the Query Engine capabilities of Calcite: (You might call it a sort of spiritual predecessor to Presto and the likes) teiid/teiid: Teiid is a data virtualization system that allows applications to use data from multiple, heterogenous data stores. <https://github.com/teiid/teiid> I had the opportunity to speak with Steven and Ramesh who had been the primary authors back in 2022, and Steven had quite a bit to say on this particular subject. (Honestly, I probably ought to aggregate and publish the conversations we had somewhere for other folks who might find it useful on the internet) In response to my question of: *"Are there any pieces of advice or gotchas you have around the general concept of federated queries/joins?"* *Unfortunately it's a lot of gotchas. * > *The more that the joins involve simple equi-join predicates the better. > More complicated predicates can cause nested loop processing. Any problem > in general that applies to a difference in behavior when pushing down vs > not pushing down will apply - and it's more likely to occur when something > is going across sources. For example I don't think we fully inhibited case > insensitive comparisons from pushing down - that helps make single source > q**ueries > behave more closely to directly querying the source. However that becomes > problematic when doing something like inner join a.col = > b.case_insensitive_col - if you push the keyset one direction it will > compare in case insensitive manner, but not the other direction.* *For Teiid to process cross-source sort-merge joins, the sort order must be > exact across sources - that may mean that actual source support for sorting > is not used. Teiid does support multi-pass sort and uses Java's support > for parallelizing the in-memory portion, but even with that processing > sorts in Teiid over large relations can be quite expensive. There are > several variants of the sort-merge join processing that help in specific > situations - which can make it more complicated from just the query plan to > tell what processing has occurred, because some of the decisions are > deferred until processing time. Only by getting the query plan after > processing are you able to see the decisions that were actually made.* *Without statistics, or sometimes even with, hints are still needed to help > determine what direction should be the dependent side of the join. > Processing the join in the wrong direction is very costly - there are > several circuit breakers, but that's little consolation to the user that's > waiting for results. We toyed several times not only with providing an > operational repository and a stats update routine, but also with adaptive > optimization in general - ideally as you process source queries there are > many times you could infer the stats or at least use sampling methods.* *Related to joins are subqueries. Many queries involving subqueries we > attempt to rewrite into a join form (sometimes as an internal semi or > anti-semi join type) so that they can be processed using the more optimized > subquery logic - however that is not always possible and you can easily end > up in with an n-many cross-source processing plan where for each row you > have to re-execute the subplan.* *Full dependent join pushdown also called data shipment joins unfortunately > received only limited use. The idea there is that rather than processing > the join in Teiid, you create a temporary table on one side of the join and > push the relation from the other down to the source for processing there. > It should be beneficial in situations where you have a "primary source".* It seems like using Calcite as a generic frontend to multiple datasources for federated querying is not a common usecase amongst its user base. I've spent most of my career working on tools in this space -- I think it's a fascinating problem. Having a rule to push down cross-join JOIN's to IN even if it supported only the "easy" cases seems like it'd unlock some neat usecases. Not that I'm asking/petitioning anyone to write it, more just thinking out loud. On Fri, May 16, 2025 at 10:27 AM Julian Hyde <jhyde.apa...@gmail.com> wrote: > The short answer to your “why?” is that no one ever did the work to make > this happen. > > The work would be a planner rule plus a cost model. > > But please remember that there is usually no good plan for a federated > join. There are a few algorithms (e.g. using bloom filters) that may help a > little, but if each query ends up shipping a significant fraction of your > table over the WAN, you’ve lost. Just replicate one of the tables. > > Why, then, does Calcite make it possible to write federated queries? > Because you can’t answer a question unless you can ask it. The best way to > operationalize a federated query is to design a data replication strategy: > replicas, other materialized views, and a process to keep them up to date. > > Julian > > > On May 16, 2025, at 06:53, Gavin Ray <ray.gavi...@gmail.com> wrote: > > > > Suppose I have two Calcite schemas from JDBC sources. > > > > One contains `users` and the other `user_todos`. > > > > If I run a query like: > > > > SELECT u.name, t.todo > > FROM USERSDB.users u > > LEFT JOIN TODOSDB.user_todos t ON u.id = t.user_id > > GROUP BY u.id, u.name, t.todo > > > > What seems like the most logical thing to do would be to run: > > > > SELECT u.name > > FROM USERSDB.users u > > > > And then consecutively: > > > > SELECT t.todo > > FROM TODOSDB.user_todos > > WHERE t.user_id IN (...) > > > > But what happens is that the entire `user_todos` is scanned: > > > > /**/stat2.execute("SELECT \"USER_ID\", \"TODO\"\nFROM > > \"USER_TODOS\"\nORDER BY \"USER_ID\" NULLS LAST"); > > 2025-05-16 09:17:54.191653-04:00 jdbc[10]: Plan : calculate cost > > for plan [PUBLIC.USER_TODOS] > > 2025-05-16 09:17:54.191766-04:00 jdbc[10]: Plan : for table > > filter PUBLIC.USER_TODOS > > 2025-05-16 09:17:54.192225-04:00 jdbc[10]: Table : potential > > plan item cost 10,230 index PUBLIC.USER_TODOS.tableScan > > 2025-05-16 09:17:54.192361-04:00 jdbc[10]: Table : potential > > plan item cost 12,240 index PUBLIC.PRIMARY_KEY_C > > 2025-05-16 09:17:54.192465-04:00 jdbc[10]: Plan : best plan > > item cost 10,230 index PUBLIC.USER_TODOS.tableScan > > > > Is this intentional? > > Why not push down JOIN predicates or correlative predicates to the JDBC > > source? >
