Fixed dropping of view records in the merge process. (#1314)

Author: bashir2

e2e-tests/controller-spark/controller_spark_sql_validation.sh

@@ -227,10 +227,12 @@ function check_parquet() {
 
   if [[ "${isIncremental}" == "true" ]]
   then
-    # In case of incremental run, we will have two directories
-    # assuming batch run was executed before this.
+    # In case of incremental run, we will have two directories (because a batch
+    # run was executed first); one directory is for the first batch run and
+    # the second is for the merge step of the incremental run. The second
+    # directory has one more patient, hence the new totals.
     TOTAL_TEST_PATIENTS=$((2*TOTAL_TEST_PATIENTS + 1))
-    TOTAL_VIEW_PATIENTS=108
+    TOTAL_VIEW_PATIENTS=$((2*TOTAL_VIEW_PATIENTS + 1))
     TOTAL_TEST_ENCOUNTERS=$((2*TOTAL_TEST_ENCOUNTERS))
     TOTAL_TEST_OBS=$((2*TOTAL_TEST_OBS))
   fi

pipelines/batch/src/main/java/com/google/fhir/analytics/ParquetMerger.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 Google LLC
+ * Copyright 2020-2025 Google LLC
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -342,7 +342,11 @@ public static Pipeline writeMergedViewsPipeline(
                         public void processElement(ProcessContext c) {
                           KV<String, CoGbkResult> e = c.element();
                           List<GenericRecord> lastRecords = new ArrayList<>();
-                          e.getValue().getAll(newDwh).forEach(lastRecords::add);
+                          Iterable<GenericRecord> iter = e.getValue().getAll(newDwh);
+                          if (!iter.iterator().hasNext()) {
+                            iter = e.getValue().getAll(oldDwh);
+                          }
+                          iter.forEach(lastRecords::add);
                           for (GenericRecord r : lastRecords) {
                             c.output(r);
                           }