{"record":{"id":"f2ed809c0a8e2fb9","repo":"apache/beam","slug":"enrichment-requests-to-vertex-ai-feature-store-should","errorCode":null,"errorMessage":"Enrichment requests to Vertex AI Feature Store should contain a field: %s in the input `beam.Row` to join the input with fetched response. This is used as the `FeatureViewDataKey` to fetch feature values corresponding to this key.","messagePattern":"Enrichment requests to Vertex AI Feature Store should contain a field: (.+?) in the input `beam\\.Row` to join the input with fetched response\\. This is used as the `FeatureViewDataKey` to fetch feature values corresponding to this key\\.","errorType":"exception","errorClass":"KeyError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/enrichment_handlers/vertex_ai_feature_store.py","lineNumber":157,"sourceCode":"    \"\"\"Connect with the Vertex AI Feature Store.\"\"\"\n    self.client = aiplatform.gapic.FeatureOnlineStoreServiceClient(\n        **self.kwargs)\n    self.feature_view_path = self.client.feature_view_path(\n        self.project,\n        self.location,\n        self.feature_store_name,\n        self.feature_view_name)\n\n  def __call__(self, request: beam.Row, *args, **kwargs):\n    \"\"\"Fetches feature value for an entity-id from Vertex AI Feature Store.\n\n    Args:\n      request: the input `beam.Row` to enrich.\n    \"\"\"\n    try:\n      entity_id = request._asdict()[self.row_key]\n    except KeyError:\n      raise KeyError(\n          \"Enrichment requests to Vertex AI Feature Store should \"\n          \"contain a field: %s in the input `beam.Row` to join \"\n          \"the input with fetched response. This is used as the \"\n          \"`FeatureViewDataKey` to fetch feature values \"\n          \"corresponding to this key.\" % self.row_key)\n    try:\n      response = self.client.fetch_feature_values(\n          request=aiplatform.gapic.FetchFeatureValuesRequest(\n              data_key=aiplatform.gapic.FeatureViewDataKey(key=entity_id),\n              feature_view=self.feature_view_path,\n              data_format=aiplatform.gapic.FeatureViewDataFormat.PROTO_STRUCT,\n          ))\n    except NotFound:\n      if self.exception_level == ExceptionLevel.WARN:\n        _LOGGER.warning(\n            _not_found_err_message(\n                self.feature_store_name, self.feature_view_name, entity_id))\n        return request, beam.Row()","sourceCodeStart":139,"sourceCodeEnd":175,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/enrichment_handlers/vertex_ai_feature_store.py#L139-L175","documentation":"Raised by VertexAIFeatureStoreEnrichment.__call__ when the input beam.Row does not contain the configured row_key field. The key is required to build the FeatureViewDataKey used to look up feature values in Vertex AI Feature Store, so enrichment cannot proceed without it.","triggerScenarios":"Calling the enrichment transform (DoFn __call__) with a beam.Row whose _asdict() has no entry for self.row_key — e.g. the Row was built without the entity-id column that was passed as row_key to the handler constructor.","commonSituations":"Renaming or dropping the key column in an upstream Select/Create, constructing beam.Row() manually and forgetting the key field, or passing a row_key string that doesn't match any column name (case/typo).","solutions":["Ensure every input beam.Row contains a field named exactly like the row_key passed to VertexAIFeatureStoreEnrichment.","Fix the row_key argument to match the actual column name in your input PCollection.","Add a Select() before enrichment to guarantee the key column is present and correctly named."],"exampleFix":"// before\nrows = pcoll | beam.Map(lambda x: beam.Row(name=x['name']))\nresult = rows | Enrichment(VertexAIFeatureStoreEnrichment('store', 'view', 'user_id'))\n// after\nrows = pcoll | beam.Map(lambda x: beam.Row(user_id=x['id'], name=x['name']))\nresult = rows | Enrichment(VertexAIFeatureStoreEnrichment('store', 'view', 'user_id'))","handlingStrategy":"validation","validationCode":"def has_key(row, key):\n    return key in row._asdict()\n# before enrichment:\n# assert all(has_key(r, 'user_id') for r in pcoll) or add Select(['user_id', ...])","typeGuard":"def key_present(row, key: str) -> bool:\n    return hasattr(row, key)","tryCatchPattern":"try:\n    result = rows | Enrichment(VertexAIFeatureStoreEnrichment('fs', 'fv', 'user_id'))\nexcept KeyError as e:\n    logging.error('Enrichment input missing key field: %s', e)","preventionTips":["Pin the input schema with Select() immediately before Enrichment.","Keep the row_key string in a shared constant used by both producers and the handler."],"tags":["keyerror","beam","enrichment","vertex-ai","missing-field"],"backgroundTag":"missing-required-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}