Information Schema Queries fail if distributed is enabled

[marc-pydantic]

If an information schema query (e.g. SHOW COLUMNS FROM records) is ran with datafusion-distributed enabled, the resulting query fails due to no serialization support for StreamingTableExec, a similar problem of missing serialization like the one that was fixed in #17200..

Our initial attempt to solve that in a similar manner, i.e. by adding the missing serialization quickly turned out to be a rather large change. A more pragmatic approach was to detect if an information schema query was performed --- our first implementation does so checking if there is a table scan on a table named "information_schema" --- and simply disable distributed execution in this case.

This could potentially be lowered, i.e. made part of datafusion(-distributed), to not distribute these kinds of queries in the first place.

[adriangb]

The point being that there's little point in distributing information schema queries.

The cases like SHOW COLUMNS are easy, they are specialized plans, can't be used in joins, etc.

On the other hand:

select * from information_schema.df_settings s1 JOIN information_schema.df_settings s2 using (name);

Does work.

Like Marc said, these plans generate StreamingTableExecs:

> explain analyze select * from information_schema.df_settings s1 JOIN information_schema.df_settings s2 using (name);
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type         | plan                                                                                                                                                                                                                                                                                                                                                                     |
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=109, elapsed_compute=19.333µs]                                                                                                                                                                                                                                                                         |
|                   |   HashJoinExec: mode=Partitioned, join_type=Inner, on=[(name@0, name@0)], projection=[name@0, value@1, description@2, value@4, description@5], metrics=[output_rows=109, elapsed_compute=1.778637ms, build_input_batches=12, build_input_rows=109, input_batches=12, input_rows=109, output_batches=12, build_mem_used=30205, build_time=861.457µs, join_time=185.588µs] |
|                   |     CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=109, elapsed_compute=35.626µs]                                                                                                                                                                                                                                                                     |
|                   |       RepartitionExec: partitioning=Hash([name@0], 12), input_partitions=12, metrics=[fetch_time=2.241249ms, repartition_time=75.844µs, send_time=10.84µs]                                                                                                                                                                                                               |
|                   |         RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1, metrics=[fetch_time=155.708µs, repartition_time=1ns, send_time=1.928µs]                                                                                                                                                                                                                   |
|                   |           StreamingTableExec: partition_sizes=1, projection=[name, value, description], metrics=[]                                                                                                                                                                                                                                                                       |
|                   |     CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=109, elapsed_compute=20.378µs]                                                                                                                                                                                                                                                                     |
|                   |       RepartitionExec: partitioning=Hash([name@0], 12), input_partitions=12, metrics=[fetch_time=1.258083ms, repartition_time=44.012µs, send_time=11.842µs]                                                                                                                                                                                                              |
|                   |         RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1, metrics=[fetch_time=109.708µs, repartition_time=1ns, send_time=553ns]                                                                                                                                                                                                                     |
|                   |           StreamingTableExec: partition_sizes=1, projection=[name, value, description], metrics=[]                                                                                                                                                                                                                                                                       |
|                   |                                                                                                                                                                                                                                                                                                                                                                          |
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.003 seconds.

IMO the data fusion-distributed optimizer should just skip distributing if the scan is a StreamingTableExec, or only distribute subtrees of the plan that do not have one.

[gabotechs]

If DataFusion planner decides that is a good idea to repartition StreamingTableExec, probably this project should listen to that and also distribute those queries.

It should be trivial to tweak the distributed planner so that upon encountering StreamingTableExec in a subplan, to not distribute that subplan and just assign 1 task to it, we can do that 👍.

Even if the subplan containing StreamingTableExec gets executed in one task, we'll still need to be able to [de]serialize it from protobuf, as it will need to be sent over the wire. That probably should be contributed upstream, although the StreamingTableExec.partitions field (Vec<Arc<dyn PartitionStream>>) seems not very suitable for getting distributed.

[gabotechs]

Something like this come to mind:
#195

Given that it seems like a very hard task to distribute a StreamingTableExec, probably a good first approach is to not distribute queries containing that node at all.

Let me know what you guys think.

[adriangb]

Yup that's exactly what we were thinking

[NGA-TRAN]

I’m on board with the quick fix to get the query working. That said, this feels like a strong opportunity to make our distributed system smarter.

@adriangb — correct me if I’m off, but from what I see, StreamingTableExec includes a list of PartitionStreams. If those streams can be executed independently and don’t need to be merged until a specific point, we could distribute them to accelerate the query plan.

The key is knowing when merging is required—or whether the streams can be passed all the way up. That insight could really improve how we distribute and optimize the plan.

If that’s accurate, it might be worth opening a ticket to track this as potential future work.

[adriangb]

The hard part IMO would be serializing the streams: the streams are traits, that might be linked to an open HTTP connection to some external service... no idea how we'd serialize that.

[NGA-TRAN]

the streams are traits

This is a new challenge for us to think about and wok on 🙂

[gabotechs]

Unless I missed something while looking at the core repo, the StreamingTableExec and the StreamingTable logical representation are only used in:

• tests
• information_schema queries

So right now no real query will contain a StreamingTableExec. This leads me to think that the effort of sending a StreamingTableExec over the wire would be to big to justify its benefits.

[NGA-TRAN]

Agree it is not high priority.

The PartitionStreams is trait and we do not know if someone uses it to stream their data

[adriangb]

So right now no real query will contain a StreamingTableExec. This leads me to think that the effort of sending a StreamingTableExec over the wire would be to big to justify its benefits.

Totally agreed.

[gabotechs]

Closing as information schema queries no longer fail in distributed mode. Feel free to reopen if you still see issues, or open a new one if you consider that partially distributing a query with StreamingTableExec nodes in it makes sense.