Add DistributedPlanError::NonDistributable rule and do not distribute SHOW COLUMNS

[gabotechs]

#191.

There are certain nodes we cannot distribute, like the ones related to table introspection. This PR adds a new controlled error in the distributed planner that prompts it to not distribute the query at all in a controlled manner, and uses it for StreamingTableExec.

[adriangb]

Looks good to me! In theory I think we could partially distribute plans, e.g. if there's a join and only one subtree has one of these non distributable plans we can still distribute the other subtree. But I don't have a use case for that at the moment and it can just be a future improvement.

[gabotechs]

if there's a join and only one subtree has one of these non distributable plans we can still distribute the other subtree

What I found challenging is to send StreamingTableExec over the wire. As it contains those Vec<Arc<dyn PartitionStream>>, I'd expect the effort of serializing those to be too big to justify the work.

[NGA-TRAN]

So a non-distributed plan can still include NetworkCoalesceExec? I had assumed it was exclusive to distributed plans. Was the intended fix to ensure that plans with StreamingTableExec remain non-distributed?

[gabotechs]

Oh wait, you are right, there's something wrong here. Let me fix it

[gabotechs]

nice catch! 👍

[adriangb]

What I found challenging is to send StreamingTableExec over the wire. As it contains those Vec<Arc<dyn PartitionStream>>, I'd expect the effort of serializing those to be too big to justify the work.

Right, agreed, we shouldn't even try to serialize a StreamingTableExec over the wire.
I don't think it matters for StreamingTableExec because as you say it's only used by information schema, etc.
But since we introduced a general "unserializable" error/handling here I do think there's a world where we handle that more gracefully. E.g. imagine you have:

HashJoinExec
    RepartitionExec
        AggregateExec
             DataSourceExec
    RepartitionExec
        SortExec
           StreamingTableExec

We could distribute this as:

HashJoinExec
    NetworkRepartitionExec  <- distribute this sub-tree of the plan
        AggregateExec
             DataSourceExec
    RepartitionExec  <- keep this sub-tree 
        SortExec
           StreamingTableExec

But I don't think this should be a priority until we have some use case where there is a non-serializable node and a subtree that is expensive / benefits from distributing.