Set container_cluster on edw_service (specifically for Trino)

Won't do anything for other edw services / if container_cluster is not initialized.

PiperOrigin-RevId: 853843563

Author: hubatish

perfkitbenchmarker/benchmark_spec.py

@@ -520,6 +520,11 @@ def ConstructEdwService(self):
     self.edw_service = edw_service_class(
         self.config.edw_service
     )  # pytype: disable=not-instantiable
+    if self.container_cluster:
+      assert isinstance(
+          self.container_cluster, container_service.KubernetesCluster
+      )
+      self.edw_service.SetContainerCluster(self.container_cluster)
 
   def ConstructEdwComputeResource(self):
     """Create an edw_compute_resource object."""

perfkitbenchmarker/edw_service.py

@@ -16,12 +16,14 @@
 Classes to wrap specific backend services are in the corresponding provider
 directory as a subclass of BaseEdwService.
 """
+
 import datetime
 import os
 from typing import Any, Dict, List
 
 from absl import flags
 from absl import logging
+from perfkitbenchmarker import container_service
 from perfkitbenchmarker import resource
 
 flags.DEFINE_integer(
@@ -443,6 +445,7 @@ def __init__(self, edw_service_spec):
     # resource workflow management
     self.supports_wait_on_delete = True
     self.client_interface: EdwClientInterface
+    self.container_cluster: container_service.KubernetesCluster | None = None
 
   def GetClientInterface(self) -> EdwClientInterface:
     """Gets the active Client Interface."""
@@ -509,6 +512,12 @@ def GetDatasetLastUpdatedTime(self, dataset=None):
   def SetDestinationTable(self, dataset: str):
     pass
 
+  def SetContainerCluster(
+      self, container_cluster: container_service.KubernetesCluster
+  ):
+    """Sets the container cluster if one is applicable."""
+    self.container_cluster = container_cluster
+
   def ExtractDataset(
       self, dest_bucket, dataset=None, tables=None, dest_format='CSV'
   ):