apache/beam · error · NotImplementedError

Caching is not supported for CustomQueryConfig. Consider…

Error message

Caching is not supported for CustomQueryConfig. Consider using TableFieldsQueryConfig or TableFunctionQueryConfig instead.

What it means

get_cache_key derives cache keys from request fields, which only works for table-based configs; for CustomQueryConfig the handler raises NotImplementedError advising TableFieldsQueryConfig or TableFunctionQueryConfig when caching is used.

Solutions

1. Switch to TableFieldsQueryConfig or TableFunctionQueryConfig to use caching.
2. Disable caching (cache_size=0) if you must keep CustomQueryConfig.
3. Review handler constructor cache parameters before combining with CustomQueryConfig.

Example fix

# before
handler = CloudSQLEnrichmentHandler(query_config=CustomQueryConfig(query=...), connection_config=cfg, cache_size=100)
# after
handler = CloudSQLEnrichmentHandler(query_config=TableFieldsQueryConfig(..., where_clause_fields=["id"]), connection_config=cfg, cache_size=100)

if isinstance(query_config, CustomQueryConfig):
    cache_size = 0  # caching unsupported for CustomQueryConfig

def cache_compatible(qc) -> bool:
    return isinstance(qc, (TableFieldsQueryConfig, TableFunctionQueryConfig))

try:
    handler = CloudSQLEnrichmentHandler(query_config=qc, connection_config=cc, cache_size=size)
except NotImplementedError as e:
    log.error("Caching unavailable for this config: %s", e)
    raise

When it happens

Trigger: Constructing CloudSQLEnrichmentHandler with caching enabled (e.g. cache_size > 0) while using CustomQueryConfig; calling get_cache_key directly on a CustomQueryConfig-backed handler.

Common situations: Enabling the built-in cache for performance while keeping CustomQueryConfig; copy-pasting a handler setup that had caching on.

Related errors

• Database host cannot be empty
• Either where_clause_fields or where_clause_value_fn must be…
• Instance connection URI cannot be empty
• Parameter binding not supported for
• Unsupported database adapter

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/c26a4508456a8cfb. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql.py:615

      List of parameter names found in the template (e.g., ["id"])
    """
    return re.findall(r':(\w+)', template)

  def create_row_key(self, row: beam.Row):
    if isinstance(self._query_config, TableFunctionQueryConfig):
      return tuple(self._query_config.where_clause_value_fn(row))
    if isinstance(self._query_config, TableFieldsQueryConfig):
      row_dict = row._asdict()
      return (
          tuple(
              row_dict[where_clause_field]
              for where_clause_field in self._query_config.where_clause_fields))
    raise ValueError(
        "Either where_clause_fields or where_clause_value_fn must be specified")

  def get_cache_key(self, request: Union[beam.Row, list[beam.Row]]):
    if isinstance(self._query_config, CustomQueryConfig):
      raise NotImplementedError(
          "Caching is not supported for CustomQueryConfig. "
          "Consider using TableFieldsQueryConfig or " +
          "TableFunctionQueryConfig instead.")

    if isinstance(request, list):
      cache_keys = []
      for req in request:
        req_dict = req._asdict()
        try:
          if isinstance(self._query_config, TableFunctionQueryConfig):
            current_values = self._query_config.where_clause_value_fn(req)
          elif isinstance(self._query_config, TableFieldsQueryConfig):
            current_values = [
                req_dict[field]
                for field in self._query_config.where_clause_fields
            ]
          key = ';'.join(map(repr, current_values))
          cache_keys.append(key)

View on GitHub (pinned to 12126d8942)