TriTih
/
eventhub
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update reusable driver working time projections

This commit is contained in:
trifonovt 2026-06-15 13:07:41 +02:00
parent dd5c32f44f
commit cdec89aa69
5 changed files with 160 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
README_PATCH.md
View File

@ -1,16 +1,25 @@
# Card-place aggregation regression fix
This patch changes `RuntimeEventAggregationService` so that it removes only repeated reads of the same physical source record.
# Fix: cardAbsentCoveragePercent above 100% on reused Esper runtime
## Root cause
The parity implementation performed a second reduction by canonical semantic event key. Distinct same-source support records could therefore be collapsed merely because their normalized event content was equal. File-session card-place identifiers such as `CARDPLACE-1` may also repeat in separate XML `Places` sections, so generated identifiers alone are not a safe physical-record key.
`DriverWorkingTimeReusableProjectionBuilder` pools Esper runtimes. The EPL used
`VuCardAbsentInterval#keepall` as a statement-local data window, but the runtime
cleanup did not clear that retained state before the next execution.
## Fix
When the same pooled runtime processed a second request, the previous execution's
card-absent intervals remained in the overlap calculation. New intervals were
added again, so `cardAbsentDurationSeconds` was doubled while the output listener
still reported only the newly emitted `VuCardAbsentInterval` events.
- Prefer `raw.rawRecordPath` as the physical identity for file-session records.
- Fall back to `raw.sourceRowId`, `raw.supportEventId`, `externalSourceEventId`, event UUID, then canonical key.
- Include domain, type, semantic lifecycle and timestamp so START/END points of one interval remain separate.
- Remove canonical semantic reduction from aggregation.
- Preserve all card/VU evidence for downstream mixing and all CVU/IW evidence for interval reconciliation.
- Add regression tests for repeated `CARDPLACE-*` identifiers across XML sections and semantically equal but physically distinct place records.
This is source-independent. It appeared in the DB result because that request was
executed after the file-session request on the same pooled runtime.
## Changes
- Added public named window `VuCardAbsentIntervalWindow#keepall`.
- Routed generated `VuCardAbsentInterval` events into the named window.
- Changed rest-coverage overlap calculations to read from the named window.
- Added `delete from VuCardAbsentIntervalWindow` to reusable-runtime cleanup.
- Applied the same EPL structure to the legacy/reference projection bundle.
- Added a regression test that executes the same input twice on the same builder
and verifies coverage does not double and remains at or below 100%.

1
src/main/java/at/procon/eventhub/processing/driverworkingtime/service/DriverWorkingTimeReusableProjectionBuilder.java
View File

@ -76,6 +76,7 @@ public class DriverWorkingTimeReusableProjectionBuilder {
"delete from DailyWeeklyRestCandidateEndBoundaryOdometerResolvedWindow",
"delete from DailyWeeklyRestCandidateCoverageCardResolvedIntervalWindow",
"delete from DailyWeeklyRestCandidateCoverageEmittedKeyWindow",
"delete from VuCardAbsentIntervalWindow",
"delete from PreviousSignificantDrivingInterval",
"context PerDriver delete from PreviousVehicleUsageInterval"
);

9
src/main/resources/esper/driver-working-time-derived-projections.epl
View File

@ -304,6 +304,8 @@ create schema VuCardAbsentInterval(
nextVehicleKey string
);
@public create window VuCardAbsentIntervalWindow#keepall as VuCardAbsentInterval;
create schema PotentialHomeOvernightStayInterval(
sessionId java.util.UUID,
driverKey string,
@ -562,7 +564,7 @@ select
c.previousVehicleKey as previousVehicleKey,
c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateInterval as c unidirectional,
VuCardAbsentInterval#keepall as u
VuCardAbsentIntervalWindow as u
where u.driverKey = c.driverKey
and u.startedAtEpochSecond < c.endedAtEpochSecond
and u.endedAtEpochSecond > c.startedAtEpochSecond
@ -595,7 +597,7 @@ select
c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateInterval as c
where not exists (
select * from VuCardAbsentInterval#keepall as u
select * from VuCardAbsentIntervalWindow as u
where u.driverKey = c.driverKey
and u.startedAtEpochSecond < c.endedAtEpochSecond
and u.endedAtEpochSecond > c.startedAtEpochSecond
@ -1776,6 +1778,9 @@ select * from DailyWeeklyRestCandidateCoverageInterval;
@name('drivingInterruptionVehicleChangeIntervals')
select * from DrivingInterruptionVehicleChangeInterval;
insert into VuCardAbsentIntervalWindow
select * from VuCardAbsentInterval;
@name('vuCardAbsentIntervals')
select * from VuCardAbsentInterval;

9
src/main/resources/esper/tachograph-driving-derived-projection-bundle.epl
View File

@ -296,6 +296,8 @@ create schema VuCardAbsentInterval(
nextVehicleKey string
);
@public create window VuCardAbsentIntervalWindow#keepall as VuCardAbsentInterval;
create schema PotentialHomeOvernightStayInterval(
sessionId java.util.UUID,
driverKey string,
@ -554,7 +556,7 @@ select
c.previousVehicleKey as previousVehicleKey,
c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateInterval as c unidirectional,
VuCardAbsentInterval#keepall as u
VuCardAbsentIntervalWindow as u
where u.driverKey = c.driverKey
and u.startedAtEpochSecond < c.endedAtEpochSecond
and u.endedAtEpochSecond > c.startedAtEpochSecond
@ -587,7 +589,7 @@ select
c.nextVehicleKey as nextVehicleKey
from DailyWeeklyRestCandidateInterval as c
where not exists (
select * from VuCardAbsentInterval#keepall as u
select * from VuCardAbsentIntervalWindow as u
where u.driverKey = c.driverKey
and u.startedAtEpochSecond < c.endedAtEpochSecond
and u.endedAtEpochSecond > c.startedAtEpochSecond
@ -1768,6 +1770,9 @@ select * from DailyWeeklyRestCandidateCoverageInterval;
@name('drivingInterruptionVehicleChangeIntervals')
select * from DrivingInterruptionVehicleChangeInterval;
insert into VuCardAbsentIntervalWindow
select * from VuCardAbsentInterval;
@name('vuCardAbsentIntervals')
select * from VuCardAbsentInterval;

125
src/test/java/at/procon/eventhub/processing/driverworkingtime/service/DriverWorkingTimeReusableProjectionBuilderTest.java
View File

@ -112,4 +112,129 @@ class DriverWorkingTimeReusableProjectionBuilderTest {
assertThat(second).isEqualTo(first);
assertThat(second.drivingInterruptionIntervals()).hasSize(1);
}
@Test
void clearsVuCardAbsentWindowWhenRuntimeIsReused() {
DriverWorkingTimeReusableProjectionBuilder builder =
new DriverWorkingTimeReusableProjectionBuilder(new EventHubProperties());
UUID sessionId = UUID.randomUUID();
OffsetDateTime from = OffsetDateTime.parse("2026-05-01T08:00:00Z");
OffsetDateTime firstDriveEnd = OffsetDateTime.parse("2026-05-01T09:00:00Z");
OffsetDateTime secondDriveStart = OffsetDateTime.parse("2026-05-01T12:00:00Z");
OffsetDateTime to = OffsetDateTime.parse("2026-05-01T13:00:00Z");
DriverWorkingTimeProcessingInput input = new DriverWorkingTimeProcessingInput(
sessionId,
"12:123",
"DRIVER_CARD",
from,
to,
from,
to,
15,
30,
List.of(
new DriverWorkingTimeActivityInterval(
sessionId,
"12:123",
"ACT-1",
"DRIVE",
"DRIVER",
"INSERTED",
"SINGLE",
"12:REG-1",
"VIN-1",
"DRIVER_CARD",
"ACT-1",
"ACT-1",
from,
firstDriveEnd,
from.toEpochSecond(),
firstDriveEnd.toEpochSecond(),
firstDriveEnd.toEpochSecond() - from.toEpochSecond(),
List.of("ACT-1"),
false,
false,
"RAW_INTERVAL"
),
new DriverWorkingTimeActivityInterval(
sessionId,
"12:123",
"ACT-2",
"DRIVE",
"DRIVER",
"INSERTED",
"SINGLE",
"12:REG-1",
"VIN-1",
"DRIVER_CARD",
"ACT-2",
"ACT-2",
secondDriveStart,
to,
secondDriveStart.toEpochSecond(),
to.toEpochSecond(),
to.toEpochSecond() - secondDriveStart.toEpochSecond(),
List.of("ACT-2"),
false,
false,
"RAW_INTERVAL"
)
),
List.of(
new DriverWorkingTimeVehicleUsageInterval(
sessionId,
"12:123",
"VU-1",
"VU-1",
"VU-1",
from,
firstDriveEnd,
from.toEpochSecond(),
firstDriveEnd.toEpochSecond(),
firstDriveEnd.toEpochSecond() - from.toEpochSecond(),
100L,
150L,
"12:REG-1",
"VIN-1",
"DRIVER_CARD",
List.of("VU-1")
),
new DriverWorkingTimeVehicleUsageInterval(
sessionId,
"12:123",
"VU-2",
"VU-2",
"VU-2",
secondDriveStart,
to,
secondDriveStart.toEpochSecond(),
to.toEpochSecond(),
to.toEpochSecond() - secondDriveStart.toEpochSecond(),
150L,
200L,
"12:REG-1",
"VIN-1",
"DRIVER_CARD",
List.of("VU-2")
)
),
List.of(),
List.of()
);
DriverWorkingTimeDerivedProjectionBundle first = builder.buildDerivedProjectionBundle(input);
DriverWorkingTimeDerivedProjectionBundle second = builder.buildDerivedProjectionBundle(input);
assertThat(first.vuCardAbsentIntervals()).hasSize(1);
assertThat(second.vuCardAbsentIntervals()).hasSize(1);
assertThat(first.dailyWeeklyRestCandidateCoverageIntervals()).hasSize(1);
assertThat(second.dailyWeeklyRestCandidateCoverageIntervals()).hasSize(1);
assertThat(second.dailyWeeklyRestCandidateCoverageIntervals().get(0).cardAbsentDurationSeconds())
.isEqualTo(first.dailyWeeklyRestCandidateCoverageIntervals().get(0).cardAbsentDurationSeconds());
assertThat(second.dailyWeeklyRestCandidateCoverageIntervals().get(0).cardAbsentCoveragePercent())
.isEqualTo(first.dailyWeeklyRestCandidateCoverageIntervals().get(0).cardAbsentCoveragePercent())
.isLessThanOrEqualTo(100.0d);
}
}