From 2580451c98f92146b0aeba1ef613c13920fcf91f Mon Sep 17 00:00:00 2001 From: Aaron Brethorst Date: Fri, 23 Oct 2026 23:44:59 -0700 Subject: [PATCH] Remove trips consolidation leaves too short When stop consolidation drops the stop times of an absorbed stop whose canonical stop isn't stored, a trip can be left with one stop time or none. Java's bundle builder drops such a trip: TripEntriesFactory skips any trip with fewer than two stop times after StopTimeEntriesFactory has discarded those with no stop. The import now does the same for trips consolidation shortened. Those trips are deleted after their stop times stream past, so they get no stop times, frequencies or block entries. Maglev keeps one-stop trips that come straight from the feed, as before; Java drops those too, but that is a separate parity question. --- gtfsdb/db.go | 10 ++++++++++ gtfsdb/helpers.go | 7 +++++++ gtfsdb/query.sql | 5 +++++ gtfsdb/query.sql.go | 11 +++++++++++ gtfsdb/stop_consolidation.go | 13 +++++++++++++ gtfsdb/stop_consolidation_test.go | 26 +++++++++++++++++++++----- gtfsdb/stream_import.go | 20 ++++++++++++++++++-- 7 files changed, 85 insertions(+), 7 deletions(-) diff --git a/gtfsdb/db.go b/gtfsdb/db.go index 217f6f2b2..fff530e4a 100644 --- a/gtfsdb/db.go +++ b/gtfsdb/db.go @@ -132,6 +132,9 @@ func Prepare(ctx context.Context, db DBTX) (*Queries, error) { if q.createTripStmt, err = db.PrepareContext(ctx, createTrip); err != nil { return nil, fmt.Errorf("error preparing query CreateTrip: %w", err) } + if q.deleteTripStmt, err = db.PrepareContext(ctx, deleteTrip); err != nil { + return nil, fmt.Errorf("error preparing query DeleteTrip: %w", err) + } if q.getActiveBlockIDsForAgencyStmt, err = db.PrepareContext(ctx, getActiveBlockIDsForAgency); err != nil { return nil, fmt.Errorf("error preparing query GetActiveBlockIDsForAgency: %w", err) } @@ -599,6 +602,11 @@ func (q *Queries) Close() error { err = fmt.Errorf("error closing createTripStmt: %w", cerr) } } + if q.deleteTripStmt != nil { + if cerr := q.deleteTripStmt.Close(); cerr != nil { + err = fmt.Errorf("error closing deleteTripStmt: %w", cerr) + } + } if q.getActiveBlockIDsForAgencyStmt != nil { if cerr := q.getActiveBlockIDsForAgencyStmt.Close(); cerr != nil { err = fmt.Errorf("error closing getActiveBlockIDsForAgencyStmt: %w", cerr) @@ -1144,6 +1152,7 @@ type Queries struct { createStopConsolidationStmt *sql.Stmt createStopTimeStmt *sql.Stmt createTripStmt *sql.Stmt + deleteTripStmt *sql.Stmt getActiveBlockIDsForAgencyStmt *sql.Stmt getActiveLayoverBlockIDsForRouteStmt *sql.Stmt getActiveRouteIDsForStopsOnDateStmt *sql.Stmt @@ -1280,6 +1289,7 @@ func (q *Queries) WithTx(tx *sql.Tx) *Queries { createStopConsolidationStmt: q.createStopConsolidationStmt, createStopTimeStmt: q.createStopTimeStmt, createTripStmt: q.createTripStmt, + deleteTripStmt: q.deleteTripStmt, getActiveBlockIDsForAgencyStmt: q.getActiveBlockIDsForAgencyStmt, getActiveLayoverBlockIDsForRouteStmt: q.getActiveLayoverBlockIDsForRouteStmt, getActiveRouteIDsForStopsOnDateStmt: q.getActiveRouteIDsForStopsOnDateStmt, diff --git a/gtfsdb/helpers.go b/gtfsdb/helpers.go index 5e3695768..5c3c385f4 100644 --- a/gtfsdb/helpers.go +++ b/gtfsdb/helpers.go @@ -474,10 +474,17 @@ func (c *Client) StoreGtfsData(ctx context.Context, data *GtfsData) (bool, error if err := feedRows.err; err != nil { return false, fmt.Errorf("unable to read stop times: %w", err) } + removedTrips, err := removeTrips(ctx, qtx, feedRows.removedTrips) + if err != nil { + return false, err + } // Collect frequency entries from all trips var allFrequencyParams []CreateFrequencyParams for _, t := range data.Static.Trips { + if removedTrips[t.ID] { + continue + } for _, f := range t.Frequencies { params := CreateFrequencyParams{ TripID: t.ID, diff --git a/gtfsdb/query.sql b/gtfsdb/query.sql index 56a9037f1..3c871eaa8 100644 --- a/gtfsdb/query.sql +++ b/gtfsdb/query.sql @@ -1564,3 +1564,8 @@ ORDER BY -- name: ClearStopConsolidations :exec DELETE FROM stop_consolidations WHERE TRUE; + +-- name: DeleteTrip :exec +DELETE FROM trips +WHERE + id = ?; diff --git a/gtfsdb/query.sql.go b/gtfsdb/query.sql.go index e77a5d0a8..4590ced60 100644 --- a/gtfsdb/query.sql.go +++ b/gtfsdb/query.sql.go @@ -929,6 +929,17 @@ func (q *Queries) CreateTrip(ctx context.Context, arg CreateTripParams) (Trip, e return i, err } +const deleteTrip = `-- name: DeleteTrip :exec +DELETE FROM trips +WHERE + id = ? +` + +func (q *Queries) DeleteTrip(ctx context.Context, id string) error { + _, err := q.exec(ctx, q.deleteTripStmt, deleteTrip, id) + return err +} + const getActiveBlockIDsForAgency = `-- name: GetActiveBlockIDsForAgency :many SELECT COALESCE(NULLIF(trips.block_id, ''), trips.id) AS block_id FROM diff --git a/gtfsdb/stop_consolidation.go b/gtfsdb/stop_consolidation.go index 860e4812f..61243ccb8 100644 --- a/gtfsdb/stop_consolidation.go +++ b/gtfsdb/stop_consolidation.go @@ -93,3 +93,16 @@ func storeStopConsolidations(ctx context.Context, qtx *Queries, stops []gtfs.Sto } return nil } + +// removeTrips deletes trips stop consolidation left too short to keep, and +// returns their IDs as a set. They have no stop times by then. +func removeTrips(ctx context.Context, qtx *Queries, tripIDs []string) (map[string]bool, error) { + removed := make(map[string]bool, len(tripIDs)) + for _, tripID := range tripIDs { + if err := qtx.DeleteTrip(ctx, tripID); err != nil { + return nil, fmt.Errorf("unable to remove trip %s: %w", tripID, err) + } + removed[tripID] = true + } + return removed, nil +} diff --git a/gtfsdb/stop_consolidation_test.go b/gtfsdb/stop_consolidation_test.go index 4caac521a..85d2bb634 100644 --- a/gtfsdb/stop_consolidation_test.go +++ b/gtfsdb/stop_consolidation_test.go @@ -12,7 +12,9 @@ import ( // consolidationFeed has a canonical stop, an absorbed stop whose stop times // move to it, an absorbed stop whose stop times are dropped, and children of -// both absorbed stops. +// both absorbed stops. Trip t1 keeps two of its three stop times, short and +// only_dropped are left with fewer than two, and one_stop keeps its only +// stop time, which moves. func consolidationFeed(t *testing.T) *GtfsData { t.Helper() feed := zipFiles(t, map[string]string{ @@ -26,12 +28,16 @@ func consolidationFeed(t *testing.T) *GtfsData { "mid,Mid,47.65,-122.35,0,\n" + "child_of_moved,Child,47.60002,-122.3,0,moved\n" + "child_of_dropped,Child,47.70001,-122.4,0,dropped\n", - "trips.txt": "route_id,service_id,trip_id,block_id\nr,s,t1,b\nr,s,only_dropped,b\n", + "trips.txt": "route_id,service_id,trip_id,block_id\nr,s,t1,b\nr,s,short,b\nr,s,only_dropped,b\nr,s,one_stop,c\n", + "frequencies.txt": "trip_id,start_time,end_time,headway_secs\nshort,10:00:00,11:00:00,600\n", "stop_times.txt": "trip_id,arrival_time,departure_time,stop_id,stop_sequence\n" + "t1,08:00:00,08:00:00,moved,1\n" + "t1,08:10:00,08:10:00,mid,2\n" + "t1,08:20:00,08:20:00,dropped,3\n" + - "only_dropped,09:00:00,09:00:00,dropped,1\n", + "short,10:00:00,10:00:00,mid,1\n" + + "short,10:10:00,10:10:00,dropped,2\n" + + "only_dropped,09:00:00,09:00:00,dropped,1\n" + + "one_stop,11:00:00,11:00:00,moved,1\n", }) data, err := ParseGtfsStream(bytes.NewReader(feed), int64(len(feed)), FeedIdentity{Hash: strings.Repeat("a", 64), Source: "source"}) require.NoError(t, err) @@ -60,10 +66,18 @@ func TestStoreGtfsData_ConsolidatesStops(t *testing.T) { t.Run("stop times move to the canonical stop or are dropped", func(t *testing.T) { assert.Equal(t, - []string{"t1 1 canonical", "t1 2 mid"}, + []string{"one_stop 1 canonical", "t1 1 canonical", "t1 2 mid"}, queryStrings(t, client, `SELECT trip_id || ' ' || stop_sequence || ' ' || stop_id FROM stop_times ORDER BY trip_id, stop_sequence`)) }) + // Java's TripEntriesFactory drops a trip left with fewer than two stop + // times; a one-stop trip consolidation didn't shorten stays, as in + // any maglev import. + t.Run("trips consolidation leaves with fewer than two stop times are removed", func(t *testing.T) { + assert.Equal(t, []string{"one_stop", "t1"}, queryStrings(t, client, `SELECT id FROM trips ORDER BY id`)) + assert.Empty(t, queryStrings(t, client, `SELECT trip_id FROM frequencies`)) + }) + t.Run("children follow their parent", func(t *testing.T) { assert.Equal(t, []string{"child_of_dropped ", "child_of_moved canonical"}, @@ -90,10 +104,12 @@ func TestFeedRows_TripEndsUseConsolidatedStops(t *testing.T) { } require.NoError(t, rows.err) - require.Len(t, rows.tripEnds, 1, "a trip whose stop times are all dropped has no ends") + require.Len(t, rows.tripEnds, 2, "removed trips have no ends") assert.Equal(t, "t1", rows.tripEnds[0].tripID) assert.Equal(t, "canonical", rows.tripEnds[0].firstStopID) assert.Equal(t, "mid", rows.tripEnds[0].lastStopID) + assert.Equal(t, "one_stop", rows.tripEnds[1].tripID) + assert.ElementsMatch(t, []string{"short", "only_dropped"}, rows.removedTrips) } func TestStoreGtfsData_ReimportClearsStopConsolidations(t *testing.T) { diff --git a/gtfsdb/stream_import.go b/gtfsdb/stream_import.go index 51c80034d..4ba918776 100644 --- a/gtfsdb/stream_import.go +++ b/gtfsdb/stream_import.go @@ -39,6 +39,9 @@ type feedRows struct { // tripEnds holds the ends of every stored trip once stopTimes has been // read to the end. tripEnds []tripEnds + // removedTrips holds the trips stop consolidation left too short to + // keep, once stopTimes has been read to the end. + removedTrips []string // err holds the first error reading from the stream. err error } @@ -63,10 +66,12 @@ func (r *feedRows) stopTimes() iter.Seq[CreateStopTimeParams] { if !r.storedTrips[trip.ID] { return nil } - stopTimes = r.replacements.consolidate(stopTimes) - if len(stopTimes) == 0 { + consolidated := r.replacements.consolidate(stopTimes) + if isTooShortAfterConsolidation(stopTimes, consolidated) { + r.removedTrips = append(r.removedTrips, trip.ID) return nil } + stopTimes = consolidated r.tripEnds = append(r.tripEnds, newTripEnds(trip, stopTimes[0], stopTimes[len(stopTimes)-1])) for _, stopTime := range stopTimes { if !yield(stopTimeParams(trip.ID, stopTime)) { @@ -79,6 +84,17 @@ func (r *feedRows) stopTimes() iter.Seq[CreateStopTimeParams] { } } +// minStopTimesPerTrip is the fewest stop times Java's TripEntriesFactory +// keeps a trip with. +const minStopTimesPerTrip = 2 + +// isTooShortAfterConsolidation reports whether stop consolidation dropped +// enough of a trip's stop times that Java would drop the trip. A trip that +// was already short in the feed is left alone, as maglev always has. +func isTooShortAfterConsolidation(original, consolidated []gtfs.ScheduledStopTime) bool { + return len(consolidated) < len(original) && len(consolidated) < minStopTimesPerTrip +} + // shapes yields the points of every shape, one shape at a time. func (r *feedRows) shapes() iter.Seq[CreateShapeParams] { return func(yield func(CreateShapeParams) bool) {