Pass dynamic filter predicate to ConnectorSplitSource#getNextBatch

[chenjian2664]

Description

Move dynamic filter waiting logic from individual connectors into the engine layer.

SPI changes:

• Add ConnectorDynamicFilter(TupleDomain<ColumnHandle> currentPredicate, boolean isComplete) record as a per-batch snapshot of the engine's DF state.
• Add ConnectorSplitSource#getNextBatch(int, ConnectorDynamicFilter) as the new primary entry point (default delegates to the deprecated getNextBatch(int)). The engine waits up to ConnectorSplitSource#getRequestedDynamicFilterWaitTimeoutMillis() before the first call; subsequent calls pass the current predicate without additional waiting.
• Add ConnectorSplitSource#getRequestedDynamicFilterWaitTimeoutMillis() (default 0) so each connector continues to control its own DF wait timeout — no new engine-level config needed.
• Update ConnectorSplitManager#getSplits to accept Set<ColumnHandle> dynamicFilterColumns (static scan-wide info) instead of a live DynamicFilter. The deprecated DynamicFilter overload remains for backward compatibility.

Engine changes:

• ConnectorAwareSplitSource now owns the DF wait loop: reads getRequestedDynamicFilterWaitTimeoutMillis() once at construction, waits on DynamicFilter.isBlocked() if awaitable and time remains, then passes new ConnectorDynamicFilter(predicate, isComplete) on every getNextBatch call.
• SplitManager calls the new getSplits(..., dynamicFilter.getColumnsCovered(), constraint) overload.

Connector changes:

• Iceberg: IcebergSplitSource overrides getNextBatch(int, ConnectorDynamicFilter) and getRequestedDynamicFilterWaitTimeoutMillis(); removes the internal DynamicFilter/Stopwatch fields and wait loop; removes iceberg.dynamic-filtering.wait-timeout config property (@DefunctConfig) and the dynamic_filtering_wait_timeout connector session property.
• Delta Lake: Same pattern as Iceberg.
• Hive: HiveSplitSource overrides the new getNextBatch and getRequestedDynamicFilterWaitTimeoutMillis(). The partition loader receives the covered-columns set at construction; the per-batch predicate snapshot is shared via a CompletableFuture updated on each getNextBatch call.
• JDBC: JdbcDynamicFilteringSplitSource / DynamicFilteringJdbcSplitSource override the new method and timeout.
• MongoDB, Hudi, LakeHouse: Implement ConnectorSplitManager#getSplits with Set<ColumnHandle>.
• Connectors that still override the deprecated getNextBatch(int) continue to work via the default bridge — no changes required.

Additional context and related issues

[ebyhr]

Could you rebase on master to resolve conflicts?

[ebyhr]

I hope @raunaqmorarka chimes in this PR.

[chenjian2664]

Updated based on the review feedback:

• ConnectorSplitSource: Renamed getDynamicFilterWaitTimeout → getRequestedDynamicFilterWaitTimeoutMillis(), returning long (not OptionalLong) with 0 meaning "no waiting". Each connector continues to provide its own timeout — no engine-level config added. Added @since 481 and @Deprecated(forRemoval = true, since = "481") javadoc as suggested.
• ConnectorSplitManager: New getSplits(..., Set<ColumnHandle> dynamicFilterColumns, ...) overload replaces the DynamicFilter overload for static scan-wide info. Old DynamicFilter overload deprecated for removal. Engine (SplitManager) calls the new overload.
• Connectors migrated to new getSplits API: Iceberg, Delta Lake, Hive, JDBC, MongoDB, Hudi, LakeHouse.
• PR description updated to reflect the current design.

CI running now. The previous run's TestDistributedFaultTolerantEngineOnlyQueries timeout was a pre-existing resource-deadlock flakiness under CI memory pressure (stage 3 held memory waiting for stage 4 output, while stage 4 couldn't get a node because stage 3 held all available memory) — unrelated to this PR. TestFaultTolerantExecutionDynamicFiltering (the DF-specific FTE test) passed 18/18.

[chenjian2664]

@losipiuk Could you please take a look at Update engine and classloader wrappers to use new split APIs this one touches FTE relate code paths , I'd appreciate having your eyes on it

[losipiuk]

@losipiuk Could you please take a look at Update engine and classloader wrappers to use new split APIs this one touches FTE relate code paths , I'd appreciate having your eyes on it

This looks fine, but it does not seem very much related to FTE, at least not directly - did I miss sth obvious?

[chenjian2664]

@losipiuk Thanks for reviewing. I noticed the FTE tests were already failing when I started adding changes to this commit — mainly the tests related to AbstractTestCoordinatorDynamicFiltering and the Memory connector refactoring. I assumed this might also be affecting the FTE tests. :)

[losipiuk]

There may be some timing issues. Can you point me to example failures?

[chenjian2664]

@losipiuk https://github.com/trinodb/trino/actions/runs/25434596125/job/74609407569 and https://github.com/trinodb/trino/actions/runs/25430147008/job/74594178701
Those two, so I modified the Memory connector 3497e93#diff-ec6f0bb035faecd81daffeef7eadfb9542ca04ccdfdb5cfd06b5e978bdf7ffadR78-R90
Now ci is green, but I am not confident if it is the right refactoring to resolve it