[shixiao-coder]

• Add ReadCandidateNodesFn to query candidate Node records by configured node_types and filter by ENTITY_TYPES (unioned with PLACE_TYPE_ORDER) when configured.
• Add DeleteExistingEmbeddingsFn to batch candidate subject_ids and delete existing NodeEmbedding rows for the target embedding_label via Partitioned DML.
• Add GenerateEmbeddingContentFn and EmbeddingContent to build top-level JSON embedding payloads and FarmHash64 content keys, reading subject_id and name from NodeRecord and batch-resolving additional edge predicates from Spanner.
• Add EmbeddingPipeline orchestrator sequencing candidate node reading, Wait.on-gated pre-deletion, and embedding content generation per EmbeddingSpec.
• Remove unused NL_STATISTICAL_VARIABLE filter type from EmbeddingSpec and nlStatVarFilePath from EmbeddingOptions, and include subject_id in default.json for base_text_embedding.
• Add unit tests for EmbeddingSpec, EmbeddingContent, ReadCandidateNodesFn, DeleteExistingEmbeddingsFn, and GenerateEmbeddingContentFn.

This is the 2nd PR to do embedding ingestion in dataflow, it uses existing spec&config implemented, and process raw Node data with (read proper DCIDs to process, pre deletion & generate content to embed on)

Will do following PRs on:
Triggering Vertex AI model
Save to embedding table
Integration with workflow
Refactor MCFGraph into Node, Edge

[codacy-production]

Not up to standards ⛔

🔴 Issues 3 medium · 11 minor

Alerts:
⚠ 14 issues (≤ 0 issues of at least minor severity)

Results:
14 new issues

Category	Results
BestPractice	2 medium
CodeStyle	11 minor
Complexity	1 medium

View in Codacy

🟢 Metrics 62 complexity

Metric	Results
Complexity	62

View in Codacy

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.}

[gemini-code-assist]

Code Review

This pull request introduces a new node embedding content generation pipeline in Apache Beam, adding classes for reading candidate nodes, deleting existing embeddings, and generating embedding content, along with corresponding unit tests. Feedback on the changes highlights three key issues: a critical bug in ReadCandidateNodesFn where the join keys do not match (subject ID vs. type name), an anti-pattern in GenerateEmbeddingContentFn where manual buffering in StartBundle/FinishBundle risks destroying windowing and timestamp metadata, and a Spanner anti-pattern in DeleteExistingEmbeddingsFn where Partitioned DML is inefficiently used for small batch deletes instead of standard DML or Mutations.

[gmechali]

Thanks Xiao! I won't get to a full review until next week, but one high-level point first:

Right now the pipeline reads every Node of the configured types and deletes and regenerates all of their embeddings on each run. We should only embed nodes touched by the current import. Can you sketch how you plan to scope this (e.g. nodes written by this import, or last_update_timestamp) before we add the Vertex calls in part 3?

Related: rather than deleting first, consider generating content and keys up front, then skipping nodes whose embedding_content_key already exists, so unchanged content is never re-embedded or temporarily missing.

Detailed comments to follow.

[gmechali]

one comment above!

[shixiao-coder]

Thanks Xiao! I won't get to a full review until next week, but one high-level point first:

Right now the pipeline reads every Node of the configured types and deletes and regenerates all of their embeddings on each run. We should only embed nodes touched by the current import. Can you sketch how you plan to scope this (e.g. nodes written by this import, or last_update_timestamp) before we add the Vertex calls in part 3?

Related: rather than deleting first, consider generating content and keys up front, then skipping nodes whose embedding_content_key already exists, so unchanged content is never re-embedded or temporarily missing.

Detailed comments to follow.

For the first one, I'll add the logic (in a separate PR) to filter the nodes by last_update_timestamp > lock_timestamp to find the nodes that get updated, updated the comment for the ReadCandidateNodesFn accordingly

For the second one, we need pre_deletion to make sure legacy embedding content get deleted for the nodes that need update since we are using embedding_content_key as key.
Suppose we change the embedding pattern from "{name}" to "{name}. {description}. {constraint property}", we need to delete the embeddings with key on "{name}" before ingesting the embedding for "{name}. {description}. {constraint property}"

Also for embedding under the same embedding_content_key, we might also update model and hyperparameter. the embedding_content_key itself is not guarantee a latest and fixed embeddings that is needed

[gmechali]

Thanks Xiao! this alsready looks a lot better! And I know the model/edge change will make it even cleaner.

Can you explain how you tested this? i see you didnt connect this to ingestion pipeline which means you might not have built this image and run it through? Could you connect it to the MCFGraphstage today behind a conditional that is always FALSE but which you can enable for a test run only.
That way you can actually rebuild this image, deploy it and test that it works.

You will also need a code review from @vish-cs on this!

[shixiao-coder]

Thanks Xiao! this alsready looks a lot better! And I know the model/edge change will make it even cleaner.

Can you explain how you tested this? i see you didnt connect this to ingestion pipeline which means you might not have built this image and run it through? Could you connect it to the MCFGraphstage today behind a conditional that is always FALSE but which you can enable for a test run only. That way you can actually rebuild this image, deploy it and test that it works.

You will also need a code review from @vish-cs on this!

This is not tested with image yet, we still need to build the function to trigger vertex AI, then the workflow code, will build an image and trigger a run on it once we have all the missing part, currently, this is on the level of unit test/ function test to verify the function we implemented

[gmechali]

approving, thanks Xiao.

Left a few more comments about cleanup. PTAL!

[vish-cs]

I have some high-level question on the design.

• Are we going to reuse the existing GraphIngestionPipeline for embedding generation or a separate dataflow pipeline?
• Will it run as a separate step in the ingestion workflow after the ingestion pipeline runs?

[shixiao-coder]

I have some high-level question on the design.

• Are we going to reuse the existing GraphIngestionPipeline for embedding generation or a separate dataflow pipeline?
• Will it run as a separate step in the ingestion workflow after the ingestion pipeline runs?

We will reuse the existing GraphIngestionPipeline, we will run Embedding deletion in parallel (we can move this to after Node ingestion if needed) with Edge/Obs deletion before ingestion, then after Node ingestion trigger the embedding ingestion.

Both Embedding deletion and ingestion will run as separate steps in the workflow and I'm thinking of controlling triggering this with a env variable or a feature flag