Update DifferPipeline to generate MCF output

[vish-cs]

No description provided.

[codacy-production]

Not up to standards ⛔

🔴 Issues 8 medium · 6 minor

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

Results:
14 new issues

Category	Results
BestPractice	1 medium
CodeStyle	6 minor
Complexity	7 medium

View in Codacy

🟢 Metrics 14 complexity

Metric	Results
Complexity	14

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 refactors the Dataflow differ pipeline to output a unified MCF diff format instead of CSV, adds a Python script to generate CSV and JSON reports from these outputs, and introduces Spanner options and metrics tracking. Key feedback highlights critical issues in the Apache Beam pipeline: the Iterable from CoGbkResult.getAll() is iterated multiple times which can cause runtime failures on production runners, and using UUID.randomUUID() introduces non-determinism. Additionally, there are integration mismatches between the Java pipeline and the Python reporting script, specifically regarding expected output filenames (differ-summary.json vs. report.json) and file structures (combined vs. split MCF files).

[gemini-code-assist]

In Apache Beam, the Iterable returned by CoGbkResult.getAll() is not guaranteed to be re-iterable (it is often backed by a lazy, single-pass iterator depending on the runner). Since currentPvs and previousPvs are iterated once to build the canonical strings (lines 152, 159) and then iterated again in iterablesToMcfString (lines 179, 189, 199, 201), this will cause runtime failures or silent data loss on production runners like Dataflow.

To fix this, copy the elements of the Iterable into a local List on the first pass, and use the list for all subsequent operations.

Suggested change

	Iterable<PropertyValues> currentPvs = input.getValue().getAll(currentTag);
	Iterable<PropertyValues> previousPvs = input.getValue().getAll(previousTag);
	List<PropertyValues> currentPvs = new ArrayList<>();
	for (PropertyValues pv : input.getValue().getAll(currentTag)) {
	currentPvs.add(pv);
	}
	List<PropertyValues> previousPvs = new ArrayList<>();
	for (PropertyValues pv : input.getValue().getAll(previousTag)) {
	previousPvs.add(pv);
	}

[gemini-code-assist]

The Python reporting script generate_csv_report.py expects the Dataflow pipeline to output a file named report.json (line 71). However, DifferUtils.java is writing the summary to differ-summary.json. This mismatch will cause the Python script to print a warning and miss the input counts in the final summary.

Please update the filename to report.json to ensure compatibility.

Suggested change

	org.apache.beam.sdk.io.fs.ResourceId resourceId =
	org.apache.beam.sdk.io.FileSystems.matchNewResource(
	outputLocation + "/differ-summary.json", false);
	org.apache.beam.sdk.io.fs.ResourceId resourceId =
	org.apache.beam.sdk.io.FileSystems.matchNewResource(
	outputLocation + "/report.json", false);

[gemini-code-assist]

There is a major discrepancy between the output of DifferPipeline.java and the input expected by generate_csv_report.py.

DifferPipeline.java flattens all diffs and writes them to a single combined file prefix diff (e.g., diff-*.mcf). However, generate_csv_report.py expects individual files categorized by mutation type and node type (e.g., obs-added-*.mcf, obs-deleted-*.mcf, schema-added-*.mcf, etc.).

As a result, running generate_csv_report.py on the output of this pipeline will find no matching files and produce empty reports.

Please either:

1. Update DifferPipeline.java to write these files individually (by partitioning/filtering the PCollections by key/diff-type).
2. Or update generate_csv_report.py to parse the combined diff-*.mcf file and group/count the nodes by the DiffType: field injected into each node.

[gemini-code-assist]

Using UUID.randomUUID() inside an Apache Beam pipeline is non-deterministic. If a bundle or work item is retried due to a transient worker failure, different UUIDs will be generated, violating the idempotency of the pipeline.

Since we already have a canonical representation of the node's properties via getCanonicalNodeValue(pv), we can generate a deterministic UUID using UUID.nameUUIDFromBytes to ensure idempotency.

Suggested change

	if (nodeId.isEmpty()) {
	nodeId = "dcid:dc/o/" + java.util.UUID.randomUUID().toString().replace("-", "");
	}
	if (nodeId.isEmpty()) {
	nodeId = "dcid:dc/o/" + java.util.UUID.nameUUIDFromBytes(getCanonicalNodeValue(pv).getBytes(java.nio.charset.StandardCharsets.UTF_8)).toString().replace("-", "");
	}