Decouple DynamicFilterService and SqlQueryExecution

DynamicFilterService does not need to depend on SqlQueryExecution and
this makes the code less readable.

Author: dain

core/trino-main/src/main/java/io/trino/execution/SqlQueryExecution.java

@@ -250,7 +250,7 @@ private synchronized void registerDynamicFilteringQuery(PlanRoot plan)
             return;
         }
 
-        dynamicFilterService.registerQuery(this, plan.getRoot());
+        dynamicFilterService.registerQuery(getSession(), getQueryPlan().orElseThrow().getRoot(), plan.getRoot());
         stateMachine.setDynamicFiltersStatsSupplier(
                 () -> dynamicFilterService.getDynamicFilteringStats(stateMachine.getQueryId()));
     }

core/trino-main/src/main/java/io/trino/server/DynamicFilterService.java

@@ -29,7 +29,6 @@
 import io.airlift.units.Duration;
 import io.trino.Session;
 import io.trino.execution.DynamicFilterConfig;
-import io.trino.execution.SqlQueryExecution;
 import io.trino.execution.StageId;
 import io.trino.execution.TaskId;
 import io.trino.metadata.FunctionManager;
@@ -120,9 +119,8 @@ public DynamicFilterService(Metadata metadata, FunctionManager functionManager,
         this.smallMaxSizePerFilter = dynamicFilterConfig.getSmallMaxSizePerFilter();
     }
 
-    public void registerQuery(SqlQueryExecution sqlQueryExecution, SubPlan fragmentedPlan)
+    public void registerQuery(Session session, PlanNode queryPlan, SubPlan fragmentedPlan)
     {
-        PlanNode queryPlan = sqlQueryExecution.getQueryPlan().orElseThrow().getRoot();
         Set<DynamicFilterId> dynamicFilters = getProducedDynamicFilters(queryPlan);
         Set<DynamicFilterId> replicatedDynamicFilters = getReplicatedDynamicFilters(queryPlan);
 
@@ -133,8 +131,8 @@ public void registerQuery(SqlQueryExecution sqlQueryExecution, SubPlan fragmente
         // register query only if it contains dynamic filters
         if (!dynamicFilters.isEmpty()) {
             registerQuery(
-                    sqlQueryExecution.getQueryId(),
-                    sqlQueryExecution.getSession(),
+                    session.getQueryId(),
+                    session,
                     dynamicFilters,
                     lazyDynamicFilters,
                     replicatedDynamicFilters);