`flight_serve_unbound` on a expr that has a con with a table in memory fails

[dlovell]

because the con is recreated but the data isn't sent.

the solution likely involves passing a build tgz rather than an expr, though this would require being able to build unbound exprs

[dlovell]

minimal repro [ this is actually a different problem that is addressed by #1591 ]

python -c '
import xorq.api as xo
import xorq.vendor.ibis.expr.operations as ops


batting = xo.examples.batting.fetch()
awards_players = ops.UnboundTable(name="awards_players", schema=xo.examples.awards_players.fetch().schema()).to_expr()
on = tuple(set(batting.columns).intersection(awards_players.columns))
joined = batting.select(on).join(awards_players.select(on), predicates=on)


server, do_exchange = xo.expr.relations.flight_serve_unbound(joined)
df = do_exchange(xo.examples.awards_players.fetch()).read_pandas()
' 2> >(tail -n 1) && echo success || echo failure
ValueError: Error during planning: table 'datafusion.public.ibis_read_record_batches_snqir66cunfcxcpj5pzyxlxlza' not found
success

note: the python process does not exit with non-zero status

[dlovell]

here is a minimal repro that has a memtable as opposed to the deferred read above

python -c '
> import xorq.api as xo
import xorq.vendor.ibis.expr.operations as ops


batting = xo.connect().register(xo.memtable(xo.examples.batting.fetch().execute()))
awards_players = ops.UnboundTable(name="awards_players", schema=xo.examples.awards_players.fetch().schema()).to_expr()
on = tuple(set(batting.columns).intersection(awards_players.columns))
joined = batting.select(on).join(awards_players.select(on), predicates=on)


server, do_exchange = xo.expr.relations.flight_serve_unbound(joined)
df = do_exchange(xo.examples.awards_players.fetch()).read_pandas()
> ' 2> >(tail -n 1) && echo success || echo failure
pyarrow.lib.ArrowInvalid: gRPC returned resource exhausted error, with message: received initial metadata size exceeds limit
failure