Error at join query between table and subquery containing "distinct"

[spozdeevvk]

Description

When trying to execute a query that joins a table with a subquery that contains the DISTINCT keyword, an error occurs.

Similar query works fine at datafusion-cli + parquet.

Query:

SELECT t.* FROM test_pg as t join (select distinct foo from test_sqlite) as a ON t.foo = a.foo

Error:

Execute the dataframe: External(External(Internal("Tried to unproject agg expr for column 'foo' that was not found in the provided Aggregate!")))

thread 'main' panicked at datafusion-federation/examples/df-csv-advanced.rs:147:21:
Execute the dataframe: External(External(Internal("Tried to unproject agg expr for column 'foo' that was not found in the provided Aggregate!")))
stack backtrace:
   0: rust_begin_unwind
             at /rustc/f6e511eec7342f59a25f7c0534f1dbea00d01b14/library/std/src/panicking.rs:662:5
   1: core::panicking::panic_fmt
             at /rustc/f6e511eec7342f59a25f7c0534f1dbea00d01b14/library/core/src/panicking.rs:74:14
   2: core::result::unwrap_failed
             at /rustc/f6e511eec7342f59a25f7c0534f1dbea00d01b14/library/core/src/result.rs:1677:5
   3: core::result::Result<T,E>::expect
             at /rustc/f6e511eec7342f59a25f7c0534f1dbea00d01b14/library/core/src/result.rs:1059:23
   4: df_csv_advanced::main::{{closure}}
             at ./datafusion-federation/examples/df-csv-advanced.rs:147:5
   5: <core::pin::Pin<P> as core::future::future::Future>::poll
             at /rustc/f6e511eec7342f59a25f7c0534f1dbea00d01b14/library/core/src/future/future.rs:123:9
   6: tokio::runtime::park::CachedParkThread::block_on::{{closure}}
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/park.rs:281:63
   7: tokio::runtime::coop::with_budget
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/coop.rs:107:5
   8: tokio::runtime::coop::budget
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/coop.rs:73:5
   9: tokio::runtime::park::CachedParkThread::block_on
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/park.rs:281:31
  10: tokio::runtime::context::blocking::BlockingRegionGuard::block_on
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/context/blocking.rs:66:9
  11: tokio::runtime::scheduler::multi_thread::MultiThread::block_on::{{closure}}
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/scheduler/multi_thread/mod.rs:87:13
  12: tokio::runtime::context::runtime::enter_runtime
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/context/runtime.rs:65:16
  13: tokio::runtime::scheduler::multi_thread::MultiThread::block_on
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/scheduler/multi_thread/mod.rs:86:9
  14: tokio::runtime::runtime::Runtime::block_on_inner
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/runtime.rs:370:45
  15: tokio::runtime::runtime::Runtime::block_on
             at /home/sp/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.1/src/runtime/runtime.rs:340:13
  16: df_csv_advanced::main
             at ./datafusion-federation/examples/df-csv-advanced.rs:147:5
  17: core::ops::function::FnOnce::call_once
             at /rustc/f6e511eec7342f59a25f7c0534f1dbea00d01b14/library/core/src/ops/function.rs:250:5
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.

To reproduce

Change the query at the df-csv-advanced example to the SELECT t.* FROM test_pg as t join (select distinct foo from test_sqlite) as a ON t.foo = a.foo.

At this line:
https://github.com/datafusion-contrib/datafusion-federation/blob/main/datafusion-federation/examples/df-csv-advanced.rs#L141