Incrementally refresh changed daily aggregates · 457fd80

Incrementally refresh changed daily aggregates
Browse files

bogdan committed 3 days ago

main

1 parent 4045a89

commit 457fd80

Showing 3 changed files with 283 additions and 10 deletions

+11 -5

HealthProbe/Doc/04-project/Import-Optimization-Log.md

@@ -584,6 +584,7 @@ rows exist".
 | 2026-06-03 | pending | Add capture-mode summary to diagnostics. | Repeated full-profile captures rarely produce a perfect no-delta report because at least one metric can change between manual runs. Diagnostic reports now include aggregate `CaptureModes` counts plus per-metric `captureMode`, so comparisons can separate unchanged empty-delta metrics from delta-applied metrics and full imports without manually reading every `record_import` line. Expected signal: stable checksum plus high `unchangedDelta` count and zero summed processing/insert confirms the fast path even when a few metrics changed. |
 | 2026-06-03 | pending | Add delta-event counts to diagnostics. | A full-profile follow-up completed in `47.4s` with `127/127` complete, `0` degraded, `CaptureModes: unchangedDelta=115, delta=12, initialImport=0`, `SummedProcessingElapsed: 25.9s`, `SummedInsertElapsed: 0.2s`, and `SummedFinalizeElapsed: 16.0s`. This confirms anchors work and no full import ran. Remaining cost is delta application for large metrics: Heart Rate `23.5s` total (`14.1s` processing, `8.8s` finalize), Active Energy `7.1s`, and Basal Energy `6.0s`. Diagnostics now report aggregate/per-metric `DeltaEvents` so future logs can separate true HealthKit delta size from the final visible record count. |
 | 2026-06-04 | `a676df1` | Rebuild delta compact archives without large intermediate record arrays. | Follow-up full-profile delta report completed in `42.1s` with `127/127` complete, `0` degraded, `CaptureModes: unchangedDelta=108, delta=19, initialImport=0`, and `DeltaEvents: 582`. Compared with the prior `47.4s` run, `SummedProcessingElapsed` dropped `25.9s -> 12.8s`; Heart Rate processing dropped `14.1s -> 6.4s` despite `187` delta events; Active Energy processing dropped `4.9s -> 2.5s` with `100` delta events; Basal Energy processing dropped `4.1s -> 2.0s` with `82` delta events. `SummedFinalizeElapsed` rose `16.0s -> 19.6s`, so the remaining bottleneck is now archive finalization / aggregate rebuild for changed high-volume types, not Swift archive reconstruction. |
+| 2026-06-04 | pending | Incrementally replace changed daily aggregate buckets. | The latest measured bottleneck is finalize work for changed high-volume delta types. Changed metrics now copy the previous observation's daily aggregates and recompute only the days touched by appeared, disappeared, or representation-changed events, with a bounded start-date window for the replacement query. Expected signal: lower `SummedFinalizeElapsed`, especially Heart Rate / Active Energy / Basal Energy finalize times, when `DeltaEvents` is small relative to the full visible type count. |
 
 ## Current Diagnosis
 
@@ -649,6 +650,11 @@ The likely bottleneck is per-row SQLite work:
   high-value target is `markVerification` / daily aggregate rebuild for changed
   high-volume delta types, where Heart Rate still spent `8.9s` finalizing and
   Active Energy spent `4.8s`.
+- Changed high-volume metrics should not rebuild daily aggregates for the whole
+  type when the current observation touched only a few days. The active
+  experiment is to copy previous daily aggregates and replace only affected
+  buckets; compare `DeltaEvents` and per-type finalize time before doing deeper
+  schema/index work.
 
 ## Open Issues / Observations
 
@@ -687,11 +693,11 @@ Prioritize experiments in this order:
    `CaptureModes`, and high-volume type timings. Treat stable checksum, a high
    `unchangedDelta` count, and zero processing/insert as the main unchanged-path
    signal.
-4. Optimize finalization for changed high-volume delta types. The latest
-   full-profile report showed `SummedFinalizeElapsed: 19.6s`, with Heart Rate
-   `8.9s`, Active Energy `4.8s`, and Basal Energy `1.7s`. Investigate whether
-   daily aggregates can be updated incrementally from `DeltaEvents` instead of
-   rebuilding all visible aggregates for the type.
+4. Run a full-profile repeated capture after incremental changed-bucket
+   aggregate replacement. Compare `SummedFinalizeElapsed`, Heart Rate / Active
+   Energy / Basal Energy finalize times, and `DeltaEvents`. If finalize remains
+   high with small `DeltaEvents`, inspect the bounded replacement SQL plan before
+   adding new write-time indexes.
 5. Keep using `DeltaEvents` to quantify changed high-volume metrics, especially
    Heart Rate, Active Energy, and Basal Energy. If delta events are small while
    finalize remains large, optimize aggregate rebuild/finalization rather than


+217 -5

HealthProbe/Services/SQLiteHealthArchiveStore.swift

View

+                                 lookupCache: lookupCache
+                             let counts = try eventCounts(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
+                    -        if counts.appeared == 0,
+                    -           counts.disappeared == 0,
+                    -           counts.representationChanged == 0,
+                    -           let previousSummary = try previousTypeSummary(
+                    +        let previousSummary = try previousTypeSummary(
+                                 sampleTypeID: sampleTypeID,
+                                 beforeObservationID: observationID,
+                                 db: db,
+                                 statementCache: statementCache
+                    -           ) {
+                    +        )
+                    +        if counts.appeared == 0,
+                    +           counts.disappeared == 0,
+                    +           counts.representationChanged == 0,
+                    +           let previousSummary {
+                                 try insertObservationTypeRun(
+                                     observationID: observationID,
+                                     sampleTypeID: sampleTypeID,
+                                 aggregate: aggregate,
+                                 db: db
+                    +        if let previousSummary,
+                    +           try replaceChangedDailyAggregates(
+                    +            fromObservationID: previousSummary.observationID,
+                    +            toObservationID: observationID,
+                    +            sampleTypeID: sampleTypeID,
+                    +            observedAt: verifiedAt,
+                    +            db: db
+                    +           ) {
+                    +            return
+                    +        }
+                             try rebuildDailyAggregates(
+                                 observationID: observationID,
+                                 sampleTypeID: sampleTypeID,
+                    +    private func replaceChangedDailyAggregates(
+                    +        fromObservationID: Int64,
+                    +        toObservationID: Int64,
+                    +        sampleTypeID: Int64,
+                    +        observedAt: Date,
+                    +        db: OpaquePointer?
+                    +    ) throws -> Bool {
+                    +        let secondsFromGMT = TimeZone.current.secondsFromGMT(for: observedAt)
+                    +        let affectedBuckets = try changedDailyAggregateBuckets(
+                    +            observationID: toObservationID,
+                    +            sampleTypeID: sampleTypeID,
+                    +            secondsFromGMT: secondsFromGMT,
+                    +            db: db
+                    +        )
+                    +        guard !affectedBuckets.isEmpty else { return false }
+                    +        try copyDailyAggregates(
+                    +            fromObservationID: fromObservationID,
+                    +            toObservationID: toObservationID,
+                    +            sampleTypeID: sampleTypeID,
+                    +            db: db
+                    +        )
+                    +        try deleteDailyAggregateRows(
+                    +            observationID: toObservationID,
+                    +            sampleTypeID: sampleTypeID,
+                    +            bucketStarts: affectedBuckets,
+                    +            db: db
+                    +        )
+                    +        let replacementRows = try dailyAggregateRows(
+                    +            sampleTypeID: sampleTypeID,
+                    +            secondsFromGMT: secondsFromGMT,
+                    +            bucketStarts: affectedBuckets,
+                    +            db: db
+                    +        )
+                    +        try insertDailyAggregateRows(
+                    +            replacementRows,
+                    +            observationID: toObservationID,
+                    +            sampleTypeID: sampleTypeID,
+                    +            db: db
+                    +        )
+                    +        return true
+                    +    }
+                    +    private func deleteDailyAggregateRows(
+                    +        observationID: Int64,
+                    +        sampleTypeID: Int64,
+                    +        bucketStarts: [Double],
+                    +        db: OpaquePointer?
+                    +    ) throws {
+                    +        guard !bucketStarts.isEmpty else { return }
+                    +        let placeholders = Array(repeating: "?", count: bucketStarts.count).joined(separator: ", ")
+                    +        try withStatement(
+                    +            """
+                    +            DELETE FROM daily_type_aggregates
+                    +            WHERE observation_id = ?
+                    +              AND sample_type_id = ?
+                    +              AND bucket_start IN (\(placeholders))
+                    +            """,
+                    +            db: db
+                    +        ) { statement in
+                    +            bindInt64(observationID, to: 1, in: statement)
+                    +            bindInt64(sampleTypeID, to: 2, in: statement)
+                    +            for (index, bucketStart) in bucketStarts.enumerated() {
+                    +                sqlite3_bind_double(statement, Int32(index + 3), bucketStart)
+                    +            }
+                    +            guard sqlite3_step(statement) == SQLITE_DONE else {
+                    +                throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
+                    +            }
+                    +        }
+                    +    }
+                         private func insertDailyAggregateRows(
+                             _ rows: [ArchiveV2DailyAggregateRow],
+                             observationID: Int64,
+                    +    private func changedDailyAggregateBuckets(
+                    +        observationID: Int64,
+                    +        sampleTypeID: Int64,
+                    +        secondsFromGMT: Int,
+                    +        db: OpaquePointer?
+                    +    ) throws -> [Double] {
+                    +        let sql = """
+                    +        WITH affected_samples AS (
+                    +            SELECT DISTINCT e.sample_id
+                    +            FROM sample_observation_events e
+                    +            JOIN samples s ON s.id = e.sample_id
+                    +            WHERE e.observation_id = ?
+                    +              AND s.sample_type_id = ?
+                    +        ),
+                    +        candidate_dates AS (
+                    +            SELECT v.start_date
+                    +            FROM sample_observation_events e
+                    +            JOIN samples s ON s.id = e.sample_id
+                    +            JOIN sample_versions v ON v.id = e.version_id
+                    +            WHERE e.observation_id = ?
+                    +              AND s.sample_type_id = ?
+                    +              AND e.version_id IS NOT NULL
+                    +            UNION
+                    +            SELECT v.start_date
+                    +            FROM affected_samples a
+                    +            JOIN sample_visibility_ranges r
+                    +              ON r.sample_id = a.sample_id
+                    +             AND r.last_observation_id IS NULL
+                    +            JOIN sample_versions v ON v.id = r.version_id
+                    +            UNION
+                    +            SELECT v.start_date
+                    +            FROM affected_samples a
+                    +            JOIN sample_visibility_ranges r
+                    +              ON r.sample_id = a.sample_id
+                    +             AND r.last_observation_id = ?
+                    +            JOIN sample_versions v ON v.id = r.version_id
+                    +        )
+                    +        SELECT DISTINCT CAST(((start_date + ?) / 86400) AS INTEGER) * 86400 - ? AS bucket_start
+                    +        FROM candidate_dates
+                    +        ORDER BY bucket_start ASC
+                    +        """
+                    +        return try withStatement(sql, db: db) { statement in
+                    +            bindInt64(observationID, to: 1, in: statement)
+                    +            bindInt64(sampleTypeID, to: 2, in: statement)
+                    +            bindInt64(observationID, to: 3, in: statement)
+                    +            bindInt64(sampleTypeID, to: 4, in: statement)
+                    +            bindInt64(observationID, to: 5, in: statement)
+                    +            sqlite3_bind_double(statement, 6, Double(secondsFromGMT))
+                    +            sqlite3_bind_double(statement, 7, Double(secondsFromGMT))
+                    +            var buckets: [Double] = []
+                    +            while sqlite3_step(statement) == SQLITE_ROW {
+                    +                buckets.append(sqlite3_column_double(statement, 0))
+                    +            }
+                    +            return buckets
+                    +        }
+                    +    }
+                    +    private func dailyAggregateRows(
+                    +        sampleTypeID: Int64,
+                    +        secondsFromGMT: Int,
+                    +        bucketStarts: [Double],
+                    +        db: OpaquePointer?
+                    +    ) throws -> [ArchiveV2DailyAggregateRow] {
+                    +        guard !bucketStarts.isEmpty else { return [] }
+                    +        let lowerStartDate = bucketStarts.min() ?? 0
+                    +        let upperStartDate = (bucketStarts.max() ?? lowerStartDate) + 86_400
+                    +        let placeholders = Array(repeating: "?", count: bucketStarts.count).joined(separator: ", ")
+                    +        let sql = """
+                    +        WITH visible_rows AS (
+                    +            SELECT
+                    +                CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? AS bucket_start,
+                    +                CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? + 86400 AS bucket_end,
+                    +                v.numeric_value,
+                    +                v.source_revision_id
+                    +            FROM samples s INDEXED BY idx_samples_type_id
+                    +            JOIN sample_visibility_ranges r INDEXED BY idx_visibility_sample_open_version
+                    +              ON r.sample_id = s.id
+                    +             AND r.last_observation_id IS NULL
+                    +            JOIN sample_versions v ON v.id = r.version_id
+                    +            WHERE s.sample_type_id = ?
+                    +              AND v.start_date >= ?
+                    +              AND v.start_date < ?
+                    +        )
+                    +        SELECT
+                    +            bucket_start,
+                    +            bucket_end,
+                    +            COUNT(*),
+                    +            SUM(numeric_value),
+                    +            MAX(numeric_value),
+                    +            source_revision_id
+                    +        FROM visible_rows
+                    +        WHERE bucket_start IN (\(placeholders))
+                    +        GROUP BY bucket_start, bucket_end, source_revision_id
+                    +        ORDER BY bucket_start ASC, source_revision_id ASC
+                    +        """
+                    +        return try withStatement(sql, db: db) { statement in
+                    +            sqlite3_bind_double(statement, 1, Double(secondsFromGMT))
+                    +            sqlite3_bind_double(statement, 2, Double(secondsFromGMT))
+                    +            sqlite3_bind_double(statement, 3, Double(secondsFromGMT))
+                    +            sqlite3_bind_double(statement, 4, Double(secondsFromGMT))
+                    +            bindInt64(sampleTypeID, to: 5, in: statement)
+                    +            bindDouble(lowerStartDate, to: 6, in: statement)
+                    +            bindDouble(upperStartDate, to: 7, in: statement)
+                    +            for (index, bucketStart) in bucketStarts.enumerated() {
+                    +                sqlite3_bind_double(statement, Int32(index + 8), bucketStart)
+                    +            }
+                    +            var rows: [ArchiveV2DailyAggregateRow] = []
+                    +            while sqlite3_step(statement) == SQLITE_ROW {
+                    +                rows.append(ArchiveV2DailyAggregateRow(
+                    +                    bucketStart: sqlite3_column_double(statement, 0),
+                    +                    bucketEnd: sqlite3_column_double(statement, 1),
+                    +                    visibleRecordCount: columnInt(statement, 2) ?? 0,
+                    +                    valueSum: columnDouble(statement, 3),
+                    +                    valueMax: columnDouble(statement, 4),
+                    +                    sourceRevisionID: columnInt64(statement, 5)
+                    +                ))
+                    +            }
+                    +            return rows
+                    +        }
+                    +    }
+                         private func typeSummary(observationID: Int64, sampleTypeID: Int64, db: OpaquePointer?) throws -> ArchiveV2TypeSummary {
+                             let counts = try eventCounts(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
+                             let aggregate = try visibleAggregate(sampleTypeID: sampleTypeID, db: db)

+55 -0

HealthProbeTests/SQLiteHealthArchiveStoreTests.swift

View

+                             XCTAssertFalse(try tableExists("archive_samples", at: url))
+                    +    func testChangedVerificationReplacesOnlyAffectedDailyAggregateBuckets() async throws {
+                    +        let url = databaseURL()
+                    +        let store = SQLiteHealthArchiveStore(databaseURL: url)
+                    +        let firstDaySample = makeStepCountSample(value: 10, start: 1_000)
+                    +        let secondDaySample = makeStepCountSample(value: 20, start: 90_000)
+                    +        let secondDayDeltaSample = makeStepCountSample(value: 30, start: 91_000)
+                    +        _ = try await store.upsertSamples(
+                    +            [firstDaySample, secondDaySample],
+                    +            observedAt: Date(timeIntervalSince1970: 95_000)
+                    +        )
+                    +        try await store.markVerification(
+                    +            sampleType: firstDaySample.sampleType,
+                    +            verifiedAt: Date(timeIntervalSince1970: 95_060)
+                    +        )
+                    +        let changedObservationID = try await store.beginObservation(
+                    +            observedAt: Date(timeIntervalSince1970: 96_000),
+                    +            triggerReason: "manual",
+                    +            selectedTypeSetHash: "selected-types"
+                    +        )
+                    +        _ = try await store.upsertSamples(
+                    +            [secondDayDeltaSample],
+                    +            observedAt: Date(timeIntervalSince1970: 96_000),
+                    +            observationID: changedObservationID
+                    +        )
+                    +        try await store.markVerification(
+                    +            sampleType: firstDaySample.sampleType,
+                    +            verifiedAt: Date(timeIntervalSince1970: 96_000),
+                    +            observationID: changedObservationID
+                    +        )
+                    +        try await store.finishObservation(
+                    +            observationID: changedObservationID,
+                    +            status: "completed",
+                    +            endedAt: Date(timeIntervalSince1970: 96_010)
+                    +        )
+                    +        XCTAssertEqual(
+                    +            try countRows(in: "daily_type_aggregates WHERE observation_id = \(changedObservationID)", at: url),
+                    +            2
+                    +        )
+                    +        XCTAssertEqual(
+                    +            try countRows(in: "daily_type_aggregates WHERE observation_id = \(changedObservationID) AND visible_record_count = 1", at: url),
+                    +            1
+                    +        )
+                    +        XCTAssertEqual(
+                    +            try countRows(in: "daily_type_aggregates WHERE observation_id = \(changedObservationID) AND visible_record_count = 2", at: url),
+                    +            1
+                    +        )
+                    +        XCTAssertEqual(
+                    +            try countRows(in: "observation_type_summaries WHERE observation_id = \(changedObservationID) AND visible_record_count = 3 AND appeared_count = 1", at: url),
+                    +            1
+                    +        )
+                    +    }
+                         func testDiffSummaryAndRecordsBetweenObservationsUseSQLVisibility() async throws {
+                             let url = databaseURL()
+                             let store = SQLiteHealthArchiveStore(databaseURL: url)

	@@ -584,6 +584,7 @@ rows exist".
584	584	\| 2026-06-03 \| pending \| Add capture-mode summary to diagnostics. \| Repeated full-profile captures rarely produce a perfect no-delta report because at least one metric can change between manual runs. Diagnostic reports now include aggregate `CaptureModes` counts plus per-metric `captureMode`, so comparisons can separate unchanged empty-delta metrics from delta-applied metrics and full imports without manually reading every `record_import` line. Expected signal: stable checksum plus high `unchangedDelta` count and zero summed processing/insert confirms the fast path even when a few metrics changed. \|
585	585	\| 2026-06-03 \| pending \| Add delta-event counts to diagnostics. \| A full-profile follow-up completed in `47.4s` with `127/127` complete, `0` degraded, `CaptureModes: unchangedDelta=115, delta=12, initialImport=0`, `SummedProcessingElapsed: 25.9s`, `SummedInsertElapsed: 0.2s`, and `SummedFinalizeElapsed: 16.0s`. This confirms anchors work and no full import ran. Remaining cost is delta application for large metrics: Heart Rate `23.5s` total (`14.1s` processing, `8.8s` finalize), Active Energy `7.1s`, and Basal Energy `6.0s`. Diagnostics now report aggregate/per-metric `DeltaEvents` so future logs can separate true HealthKit delta size from the final visible record count. \|
586	586	\| 2026-06-04 \| `a676df1` \| Rebuild delta compact archives without large intermediate record arrays. \| Follow-up full-profile delta report completed in `42.1s` with `127/127` complete, `0` degraded, `CaptureModes: unchangedDelta=108, delta=19, initialImport=0`, and `DeltaEvents: 582`. Compared with the prior `47.4s` run, `SummedProcessingElapsed` dropped `25.9s -> 12.8s`; Heart Rate processing dropped `14.1s -> 6.4s` despite `187` delta events; Active Energy processing dropped `4.9s -> 2.5s` with `100` delta events; Basal Energy processing dropped `4.1s -> 2.0s` with `82` delta events. `SummedFinalizeElapsed` rose `16.0s -> 19.6s`, so the remaining bottleneck is now archive finalization / aggregate rebuild for changed high-volume types, not Swift archive reconstruction. \|
	587	+\| 2026-06-04 \| pending \| Incrementally replace changed daily aggregate buckets. \| The latest measured bottleneck is finalize work for changed high-volume delta types. Changed metrics now copy the previous observation's daily aggregates and recompute only the days touched by appeared, disappeared, or representation-changed events, with a bounded start-date window for the replacement query. Expected signal: lower `SummedFinalizeElapsed`, especially Heart Rate / Active Energy / Basal Energy finalize times, when `DeltaEvents` is small relative to the full visible type count. \|
587	588
588	589	## Current Diagnosis
589	590
	@@ -649,6 +650,11 @@ The likely bottleneck is per-row SQLite work:
649	650	high-value target is `markVerification` / daily aggregate rebuild for changed
650	651	high-volume delta types, where Heart Rate still spent `8.9s` finalizing and
651	652	Active Energy spent `4.8s`.
	653	+- Changed high-volume metrics should not rebuild daily aggregates for the whole
	654	+ type when the current observation touched only a few days. The active
	655	+ experiment is to copy previous daily aggregates and replace only affected
	656	+ buckets; compare `DeltaEvents` and per-type finalize time before doing deeper
	657	+ schema/index work.
652	658
653	659	## Open Issues / Observations
654	660
	@@ -687,11 +693,11 @@ Prioritize experiments in this order:
687	693	`CaptureModes`, and high-volume type timings. Treat stable checksum, a high
688	694	`unchangedDelta` count, and zero processing/insert as the main unchanged-path
689	695	signal.
690		-4. Optimize finalization for changed high-volume delta types. The latest
691		- full-profile report showed `SummedFinalizeElapsed: 19.6s`, with Heart Rate
692		- `8.9s`, Active Energy `4.8s`, and Basal Energy `1.7s`. Investigate whether
693		- daily aggregates can be updated incrementally from `DeltaEvents` instead of
694		- rebuilding all visible aggregates for the type.
	696	+4. Run a full-profile repeated capture after incremental changed-bucket
	697	+ aggregate replacement. Compare `SummedFinalizeElapsed`, Heart Rate / Active
	698	+ Energy / Basal Energy finalize times, and `DeltaEvents`. If finalize remains
	699	+ high with small `DeltaEvents`, inspect the bounded replacement SQL plan before
	700	+ adding new write-time indexes.
695	701	5. Keep using `DeltaEvents` to quantify changed high-volume metrics, especially
696	702	Heart Rate, Active Energy, and Basal Energy. If delta events are small while
697	703	finalize remains large, optimize aggregate rebuild/finalization rather than