fix(MergeFeedsJob): Add state for headers written; preserve calendar dates.

Author: binh-dam-ibigroup

src/main/java/com/conveyal/datatools/manager/jobs/MergeFeedsJob.java

@@ -432,13 +432,13 @@ private void determineMergeStrategy() {
             // in both active/future feeds and that have consistent signature.
             // These trips will be linked to the new service_ids.
             serviceIdsToCloneRenameAndExtend.addAll(
-                getActiveServiceIds(this.sharedTripIdsWithConsistentSignature)
+                feedMergeContext.active.getServiceIds(this.sharedTripIdsWithConsistentSignature)
             );
 
             // Build the set of calendars to be shortened to the day before the future feed start date
             // from trips in the active feed but not in the future feed.
             serviceIdsFromActiveFeedToTerminateEarly.addAll(
-                getActiveServiceIds(feedMergeContext.getActiveTripIdsNotInFutureFeed())
+                feedMergeContext.active.getServiceIds(feedMergeContext.getActiveTripIdsNotInFutureFeed())
             );
 
 
@@ -450,24 +450,6 @@ private void determineMergeStrategy() {
         }
     }
 
-    /**
-     * Obtains the service ids corresponding to the provided trip ids.
-     */
-    private Set<String> getActiveServiceIds(Set<String> tripIds) {
-        return tripIds.stream()
-            .map(tripId -> feedMergeContext.active.feed.trips.get(tripId).service_id)
-            .collect(Collectors.toSet());
-    }
-
-    /**
-     * Obtains the service ids corresponding to the provided trip ids.
-     */
-    private Set<String> getFutureServiceIds(Set<String> tripIds) {
-        return tripIds.stream()
-            .map(tripId -> feedMergeContext.future.feed.trips.get(tripId).service_id)
-            .collect(Collectors.toSet());
-    }
-
     /**
      * Compare stop times for the given tripId between the future and active feeds. The comparison will inform whether
      * trip and/or service IDs should be modified in the output merged feed.

src/main/java/com/conveyal/datatools/manager/jobs/feedmerge/CalendarDatesMergeLineContext.java

@@ -29,6 +29,14 @@ public boolean checkFieldsForMergeConflicts(Set<NewGTFSError> idErrors, FieldCon
         return checkCalendarDatesIds(fieldContext);
     }
 
+    @Override
+    public void afterRowWrite() throws IOException {
+        // If the current row is for a calendar service_id that is marked for cloning/renaming, clone the
+        // values, change the ID, extend the start/end dates to the feed's full range, and write the
+        // additional line to the file.
+        addClonedServiceId();
+    }
+
     @Override
     public void startNewFeed(int feedIndex) throws IOException {
         super.startNewFeed(feedIndex);
@@ -37,6 +45,7 @@ public void startNewFeed(int feedIndex) throws IOException {
 
     private boolean checkCalendarDatesIds(FieldContext fieldContext) throws IOException {
         boolean shouldSkipRecord = false;
+        String key = getTableScopedValue(table, getIdScope(), keyValue);
         // Drop any calendar_dates.txt records from the existing feed for dates that are
         // not before the first date of the future feed.
         LocalDate date = getCsvDate("date");
@@ -45,11 +54,37 @@ private boolean checkCalendarDatesIds(FieldContext fieldContext) throws IOExcept
                 "Skipping calendar_dates entry {} because it operates in the time span of future feed (i.e., after or on {}).",
                 keyValue,
                 futureFeedFirstDateForCalendarValidity);
-            String key = getTableScopedValue(table, getIdScope(), keyValue);
             mergeFeedsResult.skippedIds.add(key);
             shouldSkipRecord = true;
         }
-        // Track service ID because we want to avoid removing trips that may reference this
+
+        if (job.mergeType.equals(SERVICE_PERIOD)) {
+            if (isHandlingActiveFeed()) {
+                // Remove calendar entries that are no longer used.
+                if (feedMergeContext.active.getServiceIdsToRemove().contains(keyValue)) {
+                    LOG.warn(
+                        "Skipping active calendar entry {} because it will become unused in the merged feed.",
+                        keyValue);
+                    mergeFeedsResult.skippedIds.add(key);
+                    shouldSkipRecord = true;
+                }
+            } else {
+                // If handling the future feed, the MTC revised feed merge logic is as follows:
+                // - Calendar entries from the future feed will be inserted as is in the merged feed.
+                // so no additional processing needed here, unless the calendar entry is no longer used,
+                // in that case we drop the calendar entry.
+                if (feedMergeContext.future.getServiceIdsToRemove().contains(keyValue)) {
+                    LOG.warn(
+                        "Skipping future calendar entry {} because it will become unused in the merged feed.",
+                        keyValue);
+                    mergeFeedsResult.skippedIds.add(key);
+                    shouldSkipRecord = true;
+                }
+            }
+        }
+
+
+    // Track service ID because we want to avoid removing trips that may reference this
         // service_id when the service_id is used by calendar.txt records that operate in
         // the valid date range, i.e., before the future feed's first date.
         if (!shouldSkipRecord && fieldContext.nameEquals(SERVICE_ID)) mergeFeedsResult.serviceIds.add(fieldContext.getValueToWrite());
@@ -75,4 +110,29 @@ private LocalDate getFutureFeedFirstDateForCheckingCalendarValidity() {
         }
         return futureFeedFirstDate;
     }
-}
+
+    /**
+     * Adds a cloned service id for trips with the same signature in both the active & future feeds.
+     * The cloned service id spans from the start date in the active feed until the end date in the future feed.
+     * @throws IOException
+     */
+    public void addClonedServiceId() throws IOException {
+        if (isHandlingFutureFeed() && job.mergeType.equals(SERVICE_PERIOD)) {
+            String originalServiceId = keyValue;
+            if (job.serviceIdsToCloneRenameAndExtend.contains(originalServiceId)) {
+                String[] clonedValues = getOriginalRowValues().clone();
+                String newServiceId = clonedValues[keyFieldIndex] = String.join(":", getIdScope(), originalServiceId);
+
+                referenceTracker.checkReferencesAndUniqueness(
+                    keyValue,
+                    getLineNumber(),
+                    table.fields[0],
+                    newServiceId,
+                    table,
+                    keyField,
+                    table.getOrderFieldName()
+                );
+                writeValuesToTable(clonedValues, true);
+            }
+        }
+    }}

src/main/java/com/conveyal/datatools/manager/jobs/feedmerge/CalendarMergeLineContext.java

@@ -12,6 +12,7 @@
 import java.util.Set;
 import java.util.zip.ZipOutputStream;
 
+import static com.conveyal.datatools.manager.jobs.feedmerge.MergeFeedsType.SERVICE_PERIOD;
 import static com.conveyal.datatools.manager.utils.MergeFeedUtils.getTableScopedValue;
 import static com.conveyal.datatools.manager.utils.MergeFeedUtils.hasDuplicateError;
 import static com.conveyal.gtfs.loader.DateField.GTFS_DATE_FORMATTER;
@@ -141,10 +142,9 @@ private boolean checkCalendarIds(Set<NewGTFSError> idErrors, FieldContext fieldC
      * @throws IOException
      */
     public void addClonedServiceId() throws IOException {
-        if (isHandlingFutureFeed()) {
+        if (isHandlingFutureFeed() && job.mergeType.equals(SERVICE_PERIOD)) {
             String originalServiceId = keyValue;
             if (job.serviceIdsToCloneRenameAndExtend.contains(originalServiceId)) {
-                // FIXME: Do we need to worry about calendar_dates?
                 String[] clonedValues = getOriginalRowValues().clone();
                 String newServiceId = clonedValues[keyFieldIndex] = String.join(":", getIdScope(), originalServiceId);
                 // Modify start date only (preserve the end date from the future calendar entry).

src/main/java/com/conveyal/datatools/manager/jobs/feedmerge/MergeLineContext.java

@@ -76,6 +76,7 @@ public class MergeLineContext {
     public FeedSource feedSource;
     public boolean skipFile;
     public int mergedLineNumber = 0;
+    private boolean headersWritten = false;
 
     public static MergeLineContext create(MergeFeedsJob job, Table table, ZipOutputStream out) throws IOException {
         switch (table.name) {
@@ -509,7 +510,7 @@ public void flushAndClose() throws IOException {
         out.closeEntry();
     }
 
-    public void writeHeaders() throws IOException {
+    private void writeHeaders() throws IOException {
         // Create entry for zip file.
         ZipEntry tableEntry = new ZipEntry(table.name + ".txt");
         out.putNextEntry(tableEntry);
@@ -518,6 +519,8 @@ public void writeHeaders() throws IOException {
             .map(f -> f.name)
             .toArray(String[]::new);
         writeValuesToTable(headers, false);
+
+        headersWritten = true;
     }
 
     /**
@@ -596,7 +599,7 @@ private void finishRowAndWriteToZip() throws IOException {
         }
 
         // Finally, handle writing lines to zip entry.
-        if (mergedLineNumber == 0) {
+        if (mergedLineNumber == 0 && !headersWritten) {
             writeHeaders();
         }
 

src/test/java/com/conveyal/datatools/manager/jobs/MergeFeedsJobTest.java

@@ -484,6 +484,12 @@ void mergeMTCShouldHandleMatchingTripIdsAndDropUnusedFutureCalendar() throws SQL
             3
         );
 
+        // The calendar_dates entry should be preserved, but remapped to a different id.
+        assertThatSqlCountQueryYieldsExpectedCount(
+            String.format("SELECT count(*) FROM %s.calendar_dates", mergedNamespace),
+            1
+        );
+
         assertNoUnusedServiceIds(mergedNamespace);
         assertNoRefIntegrityErrors(mergedNamespace);
     }