Speed up archive verification finalization · c1ebd37

Speed up archive verification finalization
Browse files

bogdan committed 6 days ago

main

1 parent 986f343

commit c1ebd37

Showing 2 changed files with 93 additions and 24 deletions

+69 -11

HealthProbe/Services/HealthKitService.swift

+                                         earliestDate: earliestDate,
+                                         latestDate: latestDate
+                                    ) {
+                    -                let verificationStartedAt = Date()
+                    -                try await archiveStore.markVerification(
+                    +                captureTimings.finalizeElapsedSeconds += try await finalizeArchiveVerification(
+                                         sampleType: sampleType,
+                    -                    verifiedAt: Date(),
+                    -                    observationID: archiveObservationID
+                    +                    typeIdentifier: typeIdentifier,
+                    +                    recordCount: unchanged.totalCount,
+                    +                    progressStarted: progressStarted,
+                    +                    processedEventCount: processedEventCount,
+                    +                    archiveObservationID: archiveObservationID,
+                    +                    progress: progress
+                    -                captureTimings.finalizeElapsedSeconds += Date().timeIntervalSince(verificationStartedAt)
+                                     progress?.updateBlockProgress(
+                                         typeIdentifier,
+                                         detail: "No HealthKit delta",
+                    -        let verificationStartedAt = Date()
+                    -        try await archiveStore.markVerification(sampleType: sampleType, verifiedAt: Date(), observationID: archiveObservationID)
+                    -        captureTimings.finalizeElapsedSeconds += Date().timeIntervalSince(verificationStartedAt)
+                    +        captureTimings.finalizeElapsedSeconds += try await finalizeArchiveVerification(
+                    +            sampleType: sampleType,
+                    +            typeIdentifier: typeIdentifier,
+                    +            recordCount: recordMap.count,
+                    +            progressStarted: progressStarted,
+                    +            processedEventCount: processedEventCount,
+                    +            archiveObservationID: archiveObservationID,
+                    +            progress: progress
+                    +        )
+                             let sortedRecordsStartedAt = Date()
+                             let sortedKeys = recordMap.keys.sorted {
+                    -        let verificationStartedAt = Date()
+                    -        try await archiveStore.markVerification(sampleType: sampleType, verifiedAt: Date(), observationID: archiveObservationID)
+                    -        captureTimings.finalizeElapsedSeconds += Date().timeIntervalSince(verificationStartedAt)
+                    +        captureTimings.finalizeElapsedSeconds += try await finalizeArchiveVerification(
+                    +            sampleType: sampleType,
+                    +            typeIdentifier: typeIdentifier,
+                    +            recordCount: recordCount,
+                    +            progressStarted: progressStarted,
+                    +            processedEventCount: processedEventCount,
+                    +            archiveObservationID: archiveObservationID,
+                    +            progress: progress
+                    +        )
+                             guard recordCount > 0 || anchor != nil else {
+                                 return SampleDistribution(
+                             return Double(processedCount) / elapsedSeconds
+                    +    private func finalizeArchiveVerification(
+                    +        sampleType: HKSampleType,
+                    +        typeIdentifier: String,
+                    +        recordCount: Int,
+                    +        progressStarted: Date,
+                    +        processedEventCount: Int,
+                    +        archiveObservationID: Int64,
+                    +        progress: SnapshotFetchProgress?
+                    +    ) async throws -> TimeInterval {
+                    +        let elapsedBeforeFinalize = Date().timeIntervalSince(progressStarted)
+                    +        progress?.updateBlockProgress(
+                    +            typeIdentifier,
+                    +            detail: "Finalizing archive aggregates",
+                    +            recordCount: recordCount,
+                    +            elapsedSeconds: elapsedBeforeFinalize,
+                    +            samplesPerSecond: Self.samplesPerSecond(
+                    +                processedCount: processedEventCount,
+                    +                elapsedSeconds: elapsedBeforeFinalize
+                    +            )
+                    +        )
+                    +        let verificationStartedAt = Date()
+                    +        try await archiveStore.markVerification(
+                    +            sampleType: sampleType,
+                    +            verifiedAt: Date(),
+                    +            observationID: archiveObservationID
+                    +        )
+                    +        let finalizeElapsed = Date().timeIntervalSince(verificationStartedAt)
+                    +        let elapsedAfterFinalize = Date().timeIntervalSince(progressStarted)
+                    +        progress?.updateBlockProgress(
+                    +            typeIdentifier,
+                    +            detail: "Archive aggregates finalized",
+                    +            recordCount: recordCount,
+                    +            elapsedSeconds: elapsedAfterFinalize,
+                    +            samplesPerSecond: Self.samplesPerSecond(
+                    +                processedCount: processedEventCount,
+                    +                elapsedSeconds: elapsedAfterFinalize
+                    +            )
+                    +        )
+                    +        return finalizeElapsed
+                    +    }
+                         private static func pageProgressDetail(
+                             operation: String,
+                             pageNumber: Int,

+24 -13

HealthProbe/Services/SQLiteHealthArchiveStore.swift

View

@@ -1712,6 +1712,7 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
         """, db: db)
         try execute("CREATE INDEX IF NOT EXISTS idx_visibility_open_ranges ON sample_visibility_ranges(last_observation_id)", db: db)
         try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_open ON sample_visibility_ranges(sample_id, last_observation_id)", db: db)
+        try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_open_version ON sample_visibility_ranges(sample_id, last_observation_id, version_id)", db: db)
         try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_version_open ON sample_visibility_ranges(sample_id, version_id, last_observation_id)", db: db)
         try execute("CREATE INDEX IF NOT EXISTS idx_visibility_point_lookup ON sample_visibility_ranges(first_observation_id, last_observation_id)", db: db)
         try execute("""
@@ -2989,6 +2990,11 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
         db: OpaquePointer?
     ) throws -> [ArchiveV2DailyAggregateRow] {
         let sql = """
+        WITH typed_samples AS (
+            SELECT id
+            FROM samples INDEXED BY idx_samples_type_id
+            WHERE sample_type_id = ?
+        )
         SELECT
             CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? AS bucket_start,
             CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? + 86400 AS bucket_end,
@@ -2996,21 +3002,20 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
             SUM(v.numeric_value),
             MAX(v.numeric_value),
             v.source_revision_id
-        FROM samples s
-        JOIN sample_visibility_ranges r
+        FROM typed_samples s
+        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 = ?
         GROUP BY bucket_start, bucket_end, v.source_revision_id
         ORDER BY bucket_start ASC, v.source_revision_id ASC
         """
         return try withStatement(sql, db: db) { statement in
-            sqlite3_bind_double(statement, 1, Double(secondsFromGMT))
+            bindInt64(sampleTypeID, to: 1, in: statement)
             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)
+            sqlite3_bind_double(statement, 5, Double(secondsFromGMT))
 
             var rows: [ArchiveV2DailyAggregateRow] = []
             while sqlite3_step(statement) == SQLITE_ROW {
@@ -3048,11 +3053,13 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
         db: OpaquePointer?
     ) throws -> (appeared: Int, disappeared: Int, representationChanged: Int) {
         let sql = """
-        SELECT event_kind, COUNT(*)
-        FROM sample_observation_events e
-        JOIN samples s ON s.id = e.sample_id
-        WHERE e.observation_id = ? AND s.sample_type_id = ?
-        GROUP BY event_kind
+        SELECT e.event_kind, COUNT(*)
+        FROM samples s INDEXED BY idx_samples_type_id
+        JOIN sample_observation_events e INDEXED BY idx_events_sample
+          ON e.sample_id = s.id
+         AND e.observation_id = ?
+        WHERE s.sample_type_id = ?
+        GROUP BY e.event_kind
         """
         return try withStatement(sql, db: db) { statement in
             bindInt64(observationID, to: 1, in: statement)
@@ -3080,13 +3087,17 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
 
     private func visibleAggregate(sampleTypeID: Int64, db: OpaquePointer?) throws -> ArchiveV2VisibleAggregate {
         let sql = """
+        WITH typed_samples AS (
+            SELECT id
+            FROM samples INDEXED BY idx_samples_type_id
+            WHERE sample_type_id = ?
+        )
         SELECT COUNT(*), MIN(v.start_date), MAX(v.end_date), SUM(v.numeric_value), MAX(v.numeric_value)
-        FROM samples s
-        JOIN sample_visibility_ranges r
+        FROM typed_samples s
+        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 = ?
         """
         return try withStatement(sql, db: db) { statement in
             bindInt64(sampleTypeID, to: 1, in: statement)


	@@ -1712,6 +1712,7 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
1712	1712	""", db: db)
1713	1713	try execute("CREATE INDEX IF NOT EXISTS idx_visibility_open_ranges ON sample_visibility_ranges(last_observation_id)", db: db)
1714	1714	try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_open ON sample_visibility_ranges(sample_id, last_observation_id)", db: db)
	1715	+ try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_open_version ON sample_visibility_ranges(sample_id, last_observation_id, version_id)", db: db)
1715	1716	try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_version_open ON sample_visibility_ranges(sample_id, version_id, last_observation_id)", db: db)
1716	1717	try execute("CREATE INDEX IF NOT EXISTS idx_visibility_point_lookup ON sample_visibility_ranges(first_observation_id, last_observation_id)", db: db)
1717	1718	try execute("""
	@@ -2989,6 +2990,11 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
2989	2990	db: OpaquePointer?
2990	2991	) throws -> [ArchiveV2DailyAggregateRow] {
2991	2992	let sql = """
	2993	+ WITH typed_samples AS (
	2994	+ SELECT id
	2995	+ FROM samples INDEXED BY idx_samples_type_id
	2996	+ WHERE sample_type_id = ?
	2997	+ )
2992	2998	SELECT
2993	2999	CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? AS bucket_start,
2994	3000	CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? + 86400 AS bucket_end,
	@@ -2996,21 +3002,20 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
2996	3002	SUM(v.numeric_value),
2997	3003	MAX(v.numeric_value),
2998	3004	v.source_revision_id
2999		- FROM samples s
3000		- JOIN sample_visibility_ranges r
	3005	+ FROM typed_samples s
	3006	+ JOIN sample_visibility_ranges r INDEXED BY idx_visibility_sample_open_version
3001	3007	ON r.sample_id = s.id
3002	3008	AND r.last_observation_id IS NULL
3003	3009	JOIN sample_versions v ON v.id = r.version_id
3004		- WHERE s.sample_type_id = ?
3005	3010	GROUP BY bucket_start, bucket_end, v.source_revision_id
3006	3011	ORDER BY bucket_start ASC, v.source_revision_id ASC
3007	3012	"""
3008	3013	return try withStatement(sql, db: db) { statement in
3009		- sqlite3_bind_double(statement, 1, Double(secondsFromGMT))
	3014	+ bindInt64(sampleTypeID, to: 1, in: statement)
3010	3015	sqlite3_bind_double(statement, 2, Double(secondsFromGMT))
3011	3016	sqlite3_bind_double(statement, 3, Double(secondsFromGMT))
3012	3017	sqlite3_bind_double(statement, 4, Double(secondsFromGMT))
3013		- bindInt64(sampleTypeID, to: 5, in: statement)
	3018	+ sqlite3_bind_double(statement, 5, Double(secondsFromGMT))
3014	3019
3015	3020	var rows: [ArchiveV2DailyAggregateRow] = []
3016	3021	while sqlite3_step(statement) == SQLITE_ROW {
	@@ -3048,11 +3053,13 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
3048	3053	db: OpaquePointer?
3049	3054	) throws -> (appeared: Int, disappeared: Int, representationChanged: Int) {
3050	3055	let sql = """
3051		- SELECT event_kind, COUNT(*)
3052		- FROM sample_observation_events e
3053		- JOIN samples s ON s.id = e.sample_id
3054		- WHERE e.observation_id = ? AND s.sample_type_id = ?
3055		- GROUP BY event_kind
	3056	+ SELECT e.event_kind, COUNT(*)
	3057	+ FROM samples s INDEXED BY idx_samples_type_id
	3058	+ JOIN sample_observation_events e INDEXED BY idx_events_sample
	3059	+ ON e.sample_id = s.id
	3060	+ AND e.observation_id = ?
	3061	+ WHERE s.sample_type_id = ?
	3062	+ GROUP BY e.event_kind
3056	3063	"""
3057	3064	return try withStatement(sql, db: db) { statement in
3058	3065	bindInt64(observationID, to: 1, in: statement)
	@@ -3080,13 +3087,17 @@ actor SQLiteHealthArchiveStore: HealthArchiveStore {
3080	3087
3081	3088	private func visibleAggregate(sampleTypeID: Int64, db: OpaquePointer?) throws -> ArchiveV2VisibleAggregate {
3082	3089	let sql = """
	3090	+ WITH typed_samples AS (
	3091	+ SELECT id
	3092	+ FROM samples INDEXED BY idx_samples_type_id
	3093	+ WHERE sample_type_id = ?
	3094	+ )
3083	3095	SELECT COUNT(*), MIN(v.start_date), MAX(v.end_date), SUM(v.numeric_value), MAX(v.numeric_value)
3084		- FROM samples s
3085		- JOIN sample_visibility_ranges r
	3096	+ FROM typed_samples s
	3097	+ JOIN sample_visibility_ranges r INDEXED BY idx_visibility_sample_open_version
3086	3098	ON r.sample_id = s.id
3087	3099	AND r.last_observation_id IS NULL
3088	3100	JOIN sample_versions v ON v.id = r.version_id
3089		- WHERE s.sample_type_id = ?
3090	3101	"""
3091	3102	return try withStatement(sql, db: db) { statement in
3092	3103	bindInt64(sampleTypeID, to: 1, in: statement)