Local dynamic filter support for Iceberg

[alexjo2144]

Issue: #4115
An original PR with the same changes: #5719

I think this should wait for #9193 since the tests are fairly flaky right now, relying on the left side of the join happening fast enough for the DF to be useful. Once that PR is merged we could add a DF wait time.

[findepi]

LGTM % ensuring test determinism.

[raunaqmorarka]

We could consider adding another method dedicated for operator stats assertions

protected void assertOperatorStats(
            Session session,
            @Language("SQL") String query,
            Predicate<PlanNode> planNodeMatcher,
            Consumer<List<OperatorStats>> operatorStatsAssertion,
            Consumer<MaterializedResult> resultAssertion,
            Duration timeout)
    {
        // TODO: replace this with a simple operator stats check once we find a way to wait until all pending updates to query stats have been applied
        // (might be fixed by https://github.com/trinodb/trino/issues/5172)
        assertEventually(timeout, () -> {
            DistributedQueryRunner queryRunner = getDistributedQueryRunner();
            ResultWithQueryId<MaterializedResult> resultWithQueryId = queryRunner.executeWithQueryId(session, query);
            QueryId queryId = resultWithQueryId.getQueryId();
            Plan plan = getDistributedQueryRunner().getQueryPlan(queryId);
            PlanNodeId nodeId = PlanNodeSearcher.searchFrom(plan.getRoot())
                    .where(planNodeMatcher)
                    .findOnlyElement()
                    .getId();
            List<OperatorStats> operatorStats = getDistributedQueryRunner().getCoordinator()
                    .getQueryManager()
                    .getFullQueryInfo(queryId)
                    .getQueryStats()
                    .getOperatorSummaries()
                    .stream()
                    .filter(summary -> nodeId.equals(summary.getPlanNodeId()))
                    .collect(toImmutableList());
            operatorStatsAssertion.accept(operatorStats);
            resultAssertion.accept(resultWithQueryId.getResult());
        });
    }

[alexjo2144]

Moved the test into BaseIcebergConnectorTest and added lineitem as a table requirement there. Also added a split count assertion to make sure that no file level filtering is happening.

To make it (more) deterministic, one needs to use large probe side, to work around the fact we don't wait for anything.

It's already a select * from lineitem so I'm not sure we can go much bigger with the tpch tables we have in the connector test

[findepi]

It's already a select * from lineitem so I'm not sure we can go much bigger with the tpch tables we have in the connector test

query runner factory can provision tables from sf1 only, but the test can 'manually' use any scale factor we want (within reasonable execution time limits).

[findepi]

@alexjo2144 see ci

[alexjo2144]

Thanks @findepi, I forgot I turned off auto-optimize imports. Fixed

[findepi]

typo in cmt msg

Suppert assertQueryStats assertions that use queryId

[findepi]

however, drop this commit, see comment in next

[findepi]

you use queryId, but you don't use statistics
this looks like abuse of assertQueryStats

let's use executeWithQueryId + searchScanFilterAndProjectOperatorStats directly
let's wrap in assertEventually, linking to #5172