[SQL Federation][Proxy][PostgreSQL] Correlated subquery triggers repeated scans per outer row (O(N) queries) leading to timeout even on small datasets

## Bug Report

### Which version of ShardingSphere did you use?

shardingsphere-proxy 5.5.2
postgres 17.6 as backend

### Expected behavior

The query should execute successfully. Federation engine should try decorrelation (semi-join/exists), or at least cache intermediate results to avoid multiple full table scanning.

### Actual behavior

t0 is a local table containing 41 rows. t9 is an empty broadcast table. t23 is an empty local table stored in different node from t0.
Despite small dataset, the query still triggers timeout exception.

```sql
testdb=> select
*
from
t0 as ref_0
where (ref_0.c2) in (select
ref_0.c6 as c_0
from
(t9 as ref_1
full outer join t23 as ref_3
on (ref_1.vkey = ref_3.vkey ))
);
ERROR: HikariPool-27 - Connection is not available, request timed out after 30000ms
```
log:
```log
[INFO ] 2025-12-19 13:56:29.371 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.371 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.387 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.388 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.388 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.388 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.405 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.405 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.405 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.405 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.422 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.422 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.422 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.422 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.439 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.439 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.440 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.440 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.457 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.457 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t9"
[INFO ] 2025-12-19 13:56:29.457 [ShardingSphere-Command-77] ShardingSphere-SQL - Logic SQL: SELECT "vkey" FROM "public"."t23"
[INFO ] 2025-12-19 13:56:29.457 [ShardingSphere-Command-77] ShardingSphere-SQL - Actual SQL: ds_1 ::: SELECT "vkey" FROM "public"."t23"
[ERROR] 2025-12-19 13:56:59.457 [ShardingSphere-Command-77] o.a.s.p.f.c.CommandExecutorTask - Exception occur:
java.sql.SQLTransientConnectionException: HikariPool-27 - Connection is not available, request timed out after 30000ms.
at com.zaxxer.hikari.pool.HikariPool.createTimeoutException(HikariPool.java:696)
at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:197)
at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:162)
at com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:128)
at org.apache.shardingsphere.proxy.backend.connector.jdbc.datasource.JDBCBackendDataSource.createConnection(JDBCBackendDataSource.java:100)
at org.apache.shardingsphere.proxy.backend.connector.jdbc.datasource.JDBCBackendDataSource.getConnections(JDBCBackendDataSource.java:71)
at org.apache.shardingsphere.proxy.backend.connector.jdbc.datasource.JDBCBackendDataSource.getConnections(JDBCBackendDataSource.java:44)
at org.apache.shardingsphere.proxy.backend.connector.ProxyDatabaseConnectionManager.createNewConnections(ProxyDatabaseConnectionManager.java:126)
at org.apache.shardingsphere.proxy.backend.connector.ProxyDatabaseConnectionManager.getConnections(ProxyDatabaseConnectionManager.java:101)
at org.apache.shardingsphere.infra.executor.sql.prepare.driver.DriverExecutionPrepareEngine.group(DriverExecutionPrepareEngine.java:94)
at org.apache.shardingsphere.infra.executor.sql.prepare.AbstractExecutionPrepareEngine.prepare(AbstractExecutionPrepareEngine.java:70)
at org.apache.shardingsphere.sqlfederation.executor.enumerable.EnumerableScanExecutor$1.enumerator(EnumerableScanExecutor.java:130)
at org.apache.calcite.linq4j.EnumerableDefaults.toLookup_(EnumerableDefaults.java:3806)
at org.apache.calcite.linq4j.EnumerableDefaults.toLookup(EnumerableDefaults.java:3799)
at org.apache.calcite.linq4j.EnumerableDefaults.toLookup(EnumerableDefaults.java:3775)
at org.apache.calcite.linq4j.DefaultEnumerable.toLookup(DefaultEnumerable.java:761)
at org.apache.calcite.linq4j.EnumerableDefaults$7.enumerator(EnumerableDefaults.java:1502)
at Baz$1$4$1.(Unknown Source)
at Baz$1$4.enumerator(Unknown Source)
at org.apache.calcite.linq4j.EnumerableDefaults.distinct(EnumerableDefaults.java:460)
at org.apache.calcite.linq4j.DefaultEnumerable.distinct(DefaultEnumerable.java:206)
at Baz$1.apply(Unknown Source)
at Baz$1.apply(Unknown Source)
at org.apache.calcite.linq4j.EnumerableDefaults$9$1.moveNext(EnumerableDefaults.java:1729)
at org.apache.shardingsphere.sqlfederation.resultset.SQLFederationResultSet.next(SQLFederationResultSet.java:104)
at org.apache.shardingsphere.infra.executor.sql.execute.result.query.impl.driver.jdbc.type.stream.JDBCStreamQueryResult.next(JDBCStreamQueryResult.java:55)
at org.apache.shardingsphere.sharding.merge.common.IteratorStreamMergedResult.next(IteratorStreamMergedResult.java:41)
at org.apache.shardingsphere.proxy.backend.connector.StandardDatabaseConnector.next(StandardDatabaseConnector.java:340)
at org.apache.shardingsphere.proxy.frontend.postgresql.command.query.simple.PostgreSQLComQueryExecutor.next(PostgreSQLComQueryExecutor.java:124)
at org.apache.shardingsphere.proxy.frontend.postgresql.command.PostgreSQLCommandExecuteEngine.writeDataPackets(PostgreSQLCommandExecuteEngine.java:109)
at org.apache.shardingsphere.proxy.frontend.postgresql.command.PostgreSQLCommandExecuteEngine.processSimpleQuery(PostgreSQLCommandExecuteEngine.java:95)
at org.apache.shardingsphere.proxy.frontend.postgresql.command.PostgreSQLCommandExecuteEngine.writeQueryData(PostgreSQLCommandExecuteEngine.java:85)
at org.apache.shardingsphere.proxy.frontend.command.CommandExecutorTask.doExecuteCommand(CommandExecutorTask.java:132)
at org.apache.shardingsphere.proxy.frontend.command.CommandExecutorTask.executeCommand(CommandExecutorTask.java:121)
at org.apache.shardingsphere.proxy.frontend.command.CommandExecutorTask.run(CommandExecutorTask.java:78)
at com.alibaba.ttl.TtlRunnable.run(TtlRunnable.java:60)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
```

### Steps to reproduce the behavior, such as: SQL to execute, sharding rule configuration, when exception occur etc.

[ss_postgres.zip](https://github.com/user-attachments/files/24228864/ss_postgres.zip)

run `docker compose up -d` in this file.

SQL to execute first:
```sql
SET DEFAULT SINGLE TABLE STORAGE UNIT = ds_0;

DROP TABLE IF EXISTS t0;

create table t0 (
vkey int4 ,
pkey int4 ,
c0 text ,
c1 int4 ,
c2 text ,
c3 text ,
c4 int4 ,
c5 text ,
c6 text ,
c7 text ,
c8 numeric
);

DROP TABLE IF EXISTS t1;

create table t1 (
vkey int4 ,
pkey int4 ,
c9 numeric ,
c10 int4 ,
c11 int4 ,
c12 text ,
c13 int4 ,
c14 int4
);

CREATE BROADCAST TABLE RULE t9;

DROP TABLE IF EXISTS t9;

create table t9 (
vkey int4 ,
pkey int4 ,
c17 text ,
c18 numeric ,
c19 text ,
c20 int4 ,
c21 text ,
c22 numeric ,
c23 numeric ,
c24 numeric
);

SET DEFAULT SINGLE TABLE STORAGE UNIT = ds_1;

DROP TABLE IF EXISTS t23;

create table t23 (
vkey int4 ,
pkey int4 ,
c29 int4 ,
c30 numeric
);

insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(833, 843000, null::text, -1876168804, '.ZXnMK+MNml,8!mVMEl24{', 0, null::text, '`A&N<-mh.Gi|)tQona4B', 'M7VYy*oE7YLO2Z3k~?DBXRIn2D7up', 313663858.6274);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(834, 844000, null::text, 457020334, '8aj3N`', '/', 1824108415, null::text, 'kB33NX{k>Xx^f5+/u&1@5n6`x4w2', 'AXo>IH ', 40.11);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(835, 845000, ',TiPvAwMu>g?H:xZ(nOQea=@J+', 1290051349, 'QAk~=yA ,+S!G#Nq/.>a}`', 'N!A/GVD^0)Gk]{<~-', -0, 'gsWqbQ4^Z>4yd7K/u', 's5/7&sN[H9$H', '@(-]Z#1mqd6gWw<5N9', -1352080436.23063);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(836, 846000, 'SCR/iQYj@[(5WJ$.uN}P5f*[2w1w]', -63, 'W0Am', 'rchpt&NV', -1231322507, '|7r?*oX.x#', '9Vn+8V{ILgQHlmM9#', 'N', -1085472975.23770);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(837, 847000, 'G d0W', -1131153310, '4iOFEis;y)u ^z.;', 'MKSnY+B$cDs>cS#', -799768197, '.@J*$^$B75', '~Al(=h]NhP#,?$IeI', 'Z.u~t*.Kkiqi@(&sdC]h*x', 1640174292.18795);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(838, 848000, 'BZupaDWU&cC+I@9t*Bg*acP', 2085372032, '4t]BC@c@x', '3mpZ>]#lE2tN2mhVnK8{v6', 0.0);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(839, 849000, 'qhG3TEzQ&@L^mX]]-FGrPxa-m6#S', 9, 'dL<:FUAE+7J7S`lAO YKX,!@', 'lNU&~(=', 1846452538, 'b)hSc-Aj', 'ojhJ7&SZg`]&68M:@4QF.^oRv', 'f6Q<2Dw0m*G5[]E3=X/`/Lg ', -1013980742, '`90K>IX1FYI=1/Cbj[:,+b', 'C}iIBvD3`Xp;q[q|:>EW(Uy;ZLh', 915430870, '<}&Y1/.Et#G+$):^C/(b<.j', 'BQ8W&#IM-47]lh<@@&AWp!XE', 'o*.b`DPu+0.k', -870055723.12975);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(845, 855000, 'O^SC+(2)', -11, '', 'VK42)Mo(!,|x:32OVJ!k', -39, '2QAPI+RF?5:XA1/1vgVws', 's -JT4!3!^?koxX7pn', null::text, -158264017.7024);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(846, 856000, 'n$Qc1xqHAGZ4', 0, 'nBOk:FI}*P7H', ']W', 4, ':]i]O!2z=aV FZjXw0/Bk!2+q', 'qlKNuSuudb*03Q2g=K', null::text, -23.99);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(847, 857000, '$4<$!if1,/B?9UKVS=(n44', -1818777086, '6-)lVP9G$` |oHG{4G$XT0L-CeF', 'b/QI', -1383517229.4637);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(848, 858000, 'l8<#Oh]ijkwI.*', 0, 'i}?', 'b]Gk1RZ]N5OCKRm!UcP/B,9H,XPa', -0, null::text, '^J91`2Iv]I7ux{', '5c0G0]!NT1i', 1495615686.21566);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(849, 859000, null::text, 1653413035, 'Bs>,`hiS/ FFG<]N)OYgnP&*==', '/OL8AAVjEFS(4Lk;pJ3', -49, ')', ']7 44Vb` V0DYSZ(', 1.80);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(852, 862000, '8XMD)e', -1032170670, 'K@mbzk0?SM', 'O*)48Y*HE;8j/4', -49, '^8V:j$V!I!fk>aN6c=#g]]', 'P8hT,h7)5Mn', null::text, -3.52);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(853, 863000, 'ht.vplIY', -1511054507, '!K3/Jv{AwGQ1(+(KGM/tO', '', 0, '}fj8', 'n', -1225862634.23140);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(854, 864000, '>km`2P1WJtZ]Q$Dg!oB5PYHj.3:?]![{', -1102029556.31169);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(855, 865000, 'Lqi.7?l!O2t:@oE9^S>>#[O', 15, 'vuWLB^^#', 'Z{UP<3', -2040269643, '8^!->((f', ']jD13][jCsa', null::text, -0.0);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(856, 866000, '2TT2F?Y5its8bod', -5667081, '>vh*V+M8!&!}UIuKN)VQ1[>', null::text, 1755945248, 'XJhIRw Ms=', '|RP)!r,1`71eTo18o:W', '&c^D;0UH>OLBx*Q35[b', 320202203.31398);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(857, 867000, 'vce+ctUHF@>o1D1yRH`5', 1002740359, 'Z+d@,b1*9TN8e`AZ8`<>y;zG0Zz', '1Mw`!k kC#!,;X`v$', null::int4, 'm/QDXD]F', 'N,W.D&>#)l nbU#;6(cp', 'd', '+Ckaq=&`4XOt', null::numeric);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(859, 869000, null::text, -0, 'fH', '4=Q245D:Y6-Uh^G', 200585011, '!LqIir+-Wc', '', 'P', 0.0);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(860, 870000, null::text, 0, '8}do>1;', '+IH==+,F`,P/u&d6Ad30/|97e~D5', 1141152548, '=H!}MGY-M$j8XzX,D>V3w', 'X{r', '&(#{2[H0FVEEiz=Z,4R[', 0.0);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(861, 871000, null::text, -1769015070, 'd(', '.', 1541090569, '6I', null::text, 'Q2WA[2Kcui2t', 3, '|5=Xx):lR5dC`W2', 'T2YJ8.~ 0kl,C', 1714955900, 'XdYn.Z<9jD?R', 'l5SA)', 'H(s3', null::numeric);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(863, 873000, '*5q', -1160278233, '/(ZmQU&H', -73.79);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(865, 875000, '*@FDWlo@+aQ3A!J:I(4:O;NU', 1294418933, '/|8VD;u)tA', 'lCuTJ/Bz-joG', null::int4, 'X', 'eek/(/amXf+M})7', 'a9-', -16895155.11031);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(866, 876000, 'rZ}p&1#s/-SU/ p=>u', -682244592, 'Z*p23J', '}cQ1[G7)Fe#r-!bE*DBW', 57, ',N};?9+Paz7?$ZfO&vUGVT{Xp^.Z]', 'F:p3RZMD', '$5aQ yBvW+#m75@B;yW].', null::int4, 'LBZ]VS', '}p?ok4#', 8, ']KR-Bo>', 'O@|3M/HjNY<{?oT>,C]JlV', null::text, -2088414375.17883);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(868, 878000, 'K=a8GbMd7G:F;Ert)Wp^C4TY', 52, 'Nq7&XK]md38O^K#A', 'JZ', -342534526.4884);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(869, 879000, '5,,F4KadEEJ/3(Wu3kUdJ49', 243213098, '~EN5CM', '=M|L', 83, ',#93e.~)-D,=]D', null::text, 'vzx31XRO]$GZai#-#>Ek`[=CIhY', null::numeric);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(870, 880000, '*:$&', -1525865902, 'QfP@+E|9T3C36S;#0KO?!', 'iN#1P,?ppv]E7UD-(-t3-EdC2Ns', -1677657313, null::text, 'o4$Jw}^g1k);.bU3@i==', ',W', 943380194.10882);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(871, 881000, '}[{yR`T5T;&]m.taBK&V4qJ1k=', -58, 'fB3x7I', '5SV(A?}E~e3ym:W8>ThPl', 1439310750, null::text, 'L@!&p{W$~..^]We8).[=Gp', null::text, -122879731.15424);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(872, 882000, ';=6:aV^j,23,!,[0', -42, '-940hR(QywOlob!SKNl', '?1RN4~z6t1IDB', -1696348395, 'QGa3h)UI*mwpV9Y', '-Wo;KnG#3i>ren9Y0]Db$G (UYTF', 92.100);
insert into t0 (vkey, pkey, c0, c1, c2, c3, c4, c5, c6, c7, c8) values
(873, 883000, 'LA', -21, '', null::text, 1504303128, '94A9@ 1[#(:RVTDYvJ[', '~sD*9YV&Z33.IdNL', 'QJ#L}v6n)p^X37u:M]?5[Lf56oH', 1887079768.8401);
```

Start by running docker compose up -d from the supplied ss_postgres.zip reproduction and execute the provided PostgreSQL setup and query. Read the stack trace around sqlfederation executor's EnumerableScanExecutor.java and inspect the repeated scans shown in the logs. Done means the query completes without repeated per-row scans or a Hikari connection-timeout exception.

Written by the indexing model from the issue text.