Skip to content

Commit 5906611

Browse files
jcscottiiijcscottiii
committed
start
1 parent e9f2943 commit 5906611

File tree

8 files changed

+658
-69
lines changed

8 files changed

+658
-69
lines changed

‎lib/gcpspanner/client.go‎

Lines changed: 91 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ var (
6868
ErrSyncAtomicWriteFailed = errors.New("sync atomic write failed")
6969
ErrSyncBatchWriteFailed = errors.New("sync batch write failed")
7070
ErrSyncFailedToGetChildMutations = errors.New("sync failed to get child mutations")
71+
ErrSyncFailedToGetPreDeleteHooks = errors.New("sync failed to get pre delete hooks")
7172
)
7273

7374
// Client is the client for interacting with GCP Spanner.
@@ -360,6 +361,10 @@ type readAllMapper interface {
360361
SelectAll() spanner.Statement
361362
}
362363

364+
type readAllByKeysMapper[KeysContainer comparable] interface {
365+
SelectAllByKeys(KeysContainer) spanner.Statement
366+
}
367+
363368
// mergeMapper handles the logic for updating an existing entity.
364369
type mergeMapper[ExternalStruct any, SpannerStruct any] interface {
365370
Merge(ExternalStruct, SpannerStruct) SpannerStruct
@@ -435,9 +440,10 @@ type syncableEntityMapper[ExternalStruct any, SpannerStruct any, Key comparable]
435440
mergeAndCheckChangedMapper[ExternalStruct, SpannerStruct]
436441
childDeleterMapper[SpannerStruct]
437442
deleteByStructMapper[SpannerStruct]
443+
preDeleteHookMapper[SpannerStruct]
438444
}
439445

440-
type ChildDeleteKeyMutations struct {
446+
type ExtraMutationsGroup struct {
441447
tableName string
442448
mutations []*spanner.Mutation
443449
}
@@ -449,7 +455,11 @@ type childDeleterMapper[SpannerStruct any] interface {
449455
ctx context.Context,
450456
client *Client,
451457
parentsToDelete []SpannerStruct,
452-
) ([]ChildDeleteKeyMutations, error)
458+
) ([]ExtraMutationsGroup, error)
459+
}
460+
461+
type preDeleteHookMapper[SpannerStruct any] interface {
462+
PreDeleteHook(ctx context.Context, client *Client, rowsToDelete []SpannerStruct) ([]ExtraMutationsGroup, error)
453463
}
454464

455465
// --- Generic Entity Components ---
@@ -860,6 +870,57 @@ func (c *entityRemover[M, ExternalStruct, SpannerStruct, Key]) removeWithTransac
860870
return nil
861871
}
862872

873+
// allByKeysEntityReader handles the reading of a Spanner table with a set of key(s).
874+
type allByKeysEntityReader[
875+
M readAllByKeysMapper[KeysContainer],
876+
KeysContainer comparable,
877+
SpannerStruct any] struct {
878+
*Client
879+
}
880+
881+
// newAllByKeysEntityReader creates a new reader.
882+
func newAllByKeysEntityReader[
883+
M readAllByKeysMapper[KeysContainer],
884+
KeysContainer comparable,
885+
SpannerStruct any,
886+
](
887+
c *Client,
888+
) *allByKeysEntityReader[M, KeysContainer, SpannerStruct] {
889+
return &allByKeysEntityReader[M, KeysContainer, SpannerStruct]{
890+
Client: c,
891+
}
892+
}
893+
894+
func (r *allByKeysEntityReader[M, KeysContainer, SpannerStruct]) readAllByKeys(
895+
ctx context.Context,
896+
keys KeysContainer,
897+
) ([]SpannerStruct, error) {
898+
var mapper M
899+
stmt := mapper.SelectAllByKeys(keys)
900+
txn := r.Single()
901+
defer txn.Close()
902+
it := txn.Query(ctx, stmt)
903+
defer it.Stop()
904+
905+
var entities []SpannerStruct
906+
for {
907+
row, err := it.Next()
908+
if errors.Is(err, iterator.Done) {
909+
break
910+
}
911+
if err != nil {
912+
return nil, errors.Join(ErrInternalQueryFailure, err)
913+
}
914+
var entity SpannerStruct
915+
if err := row.ToStruct(&entity); err != nil {
916+
return nil, err
917+
}
918+
entities = append(entities, entity)
919+
}
920+
921+
return entities, nil
922+
}
923+
863924
// entitySynchronizer handles the synchronization of a Spanner table with a
864925
// desired state provided as a slice of entities. It determines whether to
865926
// use a single atomic transaction or a high-throughput batch write based on
@@ -874,6 +935,9 @@ type entitySynchronizer[
874935
// The number of mutations at which the synchronizer will switch from a
875936
// single atomic transaction to the non-atomic batch writer.
876937
batchWriteThreshold int
938+
// Mapper is the entity mapper that provides the necessary database operation logic.
939+
// This field should be configured before calling the Sync method.
940+
Mapper M
877941
}
878942

879943
// newEntitySynchronizer creates a new synchronizer with a default threshold.
@@ -882,12 +946,13 @@ func newEntitySynchronizer[
882946
ExternalStruct any,
883947
SpannerStruct any,
884948
Key comparable,
885-
](
886-
c *Client,
887-
) *entitySynchronizer[M, ExternalStruct, SpannerStruct, Key] {
949+
](c *Client) *entitySynchronizer[M, ExternalStruct, SpannerStruct, Key] {
950+
var m M
951+
888952
return &entitySynchronizer[M, ExternalStruct, SpannerStruct, Key]{
889953
Client: c,
890954
batchWriteThreshold: defaultBatchSize,
955+
Mapper: m,
891956
}
892957
}
893958

@@ -897,7 +962,7 @@ func (s *entitySynchronizer[M, ExternalStruct, SpannerStruct, Key]) Sync(
897962
ctx context.Context,
898963
desiredState []ExternalStruct,
899964
) error {
900-
var mapper M
965+
mapper := s.Mapper
901966
tableName := mapper.Table()
902967

903968
// 1. READ: Fetch all existing entities from the database.
@@ -974,22 +1039,21 @@ func (s *entitySynchronizer[M, ExternalStruct, SpannerStruct, Key]) Sync(
9741039
"updates", updates,
9751040
"deletes", deletes)
9761041

977-
// 3. APPLY DELETES: Handle child and parent deletions first.
978-
if err := s.applyDeletes(ctx, entitiesToDelete, deleteMutations, mapper); err != nil {
979-
return err
980-
}
981-
982-
// 4. APPLY UPSERTS: Apply all inserts and updates together.
1042+
// 3. APPLY UPSERTS: Apply all inserts and updates together.
9831043
if len(upsertMutations) < s.batchWriteThreshold {
9841044
err = s.applyAtomic(ctx, upsertMutations, tableName)
9851045
} else {
9861046
err = s.applyNonAtomic(ctx, upsertMutations, tableName)
9871047
}
988-
9891048
if err != nil {
9901049
return err
9911050
}
9921051

1052+
// 4. APPLY DELETES: Handle child and parent deletions.
1053+
if err := s.applyDeletes(ctx, entitiesToDelete, deleteMutations, mapper); err != nil {
1054+
return err
1055+
}
1056+
9931057
slog.InfoContext(ctx, "Sync successful", "table", tableName)
9941058

9951059
return nil
@@ -1009,6 +1073,20 @@ func (s *entitySynchronizer[M, ExternalStruct, SpannerStruct, Key]) applyDeletes
10091073
}
10101074
tableName := mapper.Table()
10111075

1076+
// Handle pre delete hooks first.
1077+
mutationGroups, err := mapper.PreDeleteHook(ctx, s.Client, entitiesToDelete)
1078+
if err != nil {
1079+
return errors.Join(ErrSyncFailedToGetPreDeleteHooks, err)
1080+
}
1081+
for _, group := range mutationGroups {
1082+
slog.InfoContext(ctx, "Applying pre delete mutations via batch writer",
1083+
"count", len(group.mutations), "table", group.tableName)
1084+
err := s.applyNonAtomic(ctx, group.mutations, group.tableName)
1085+
if err != nil {
1086+
return err
1087+
}
1088+
}
1089+
10121090
// Handle manual child deletions first.
10131091
// The `ON DELETE CASCADE` constraint should be the default, but it can fail
10141092
// if a cascade exceeds Spanner's 80k mutation limit.

‎lib/gcpspanner/daily_chromium_histogram_metrics.go‎

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -136,10 +136,34 @@ func (m latestDailyChromiumHistogramMetricMapper) GetChildDeleteKeyMutations(
136136
_ context.Context,
137137
_ *Client,
138138
_ []SpannerLatestDailyChromiumHistogramMetric,
139-
) ([]ChildDeleteKeyMutations, error) {
139+
) ([]ExtraMutationsGroup, error) {
140140
return nil, nil
141141
}
142142

143+
// PreDeleteHook is a no-op for this table.
144+
func (m latestDailyChromiumHistogramMetricMapper) PreDeleteHook(
145+
_ context.Context,
146+
_ *Client,
147+
_ []SpannerLatestDailyChromiumHistogramMetric,
148+
) ([]ExtraMutationsGroup, error) {
149+
return nil, nil
150+
}
151+
152+
type latestDailyChromiumHistogramMetricByWebFeatureIDMapper struct{}
153+
154+
func (m latestDailyChromiumHistogramMetricByWebFeatureIDMapper) SelectAllByKeys(webFeatureID string) spanner.Statement {
155+
stmt := spanner.NewStatement(`
156+
SELECT
157+
*
158+
FROM LatestDailyChromiumHistogramMetrics
159+
WHERE WebFeatureID = @webFeatureID`)
160+
stmt.Params = map[string]interface{}{
161+
"webFeatureID": webFeatureID,
162+
}
163+
164+
return stmt
165+
}
166+
143167
// DeleteMutation creates a Spanner delete mutation.
144168
func (m latestDailyChromiumHistogramMetricMapper) DeleteMutation(
145169
in SpannerLatestDailyChromiumHistogramMetric) *spanner.Mutation {
@@ -203,3 +227,12 @@ func (c *Client) getDesiredLatestDailyChromiumHistogramMetrics(
203227

204228
return desiredState, nil
205229
}
230+
231+
func (c *Client) getAllLatestDailyChromiumHistogramMetricsByFeatureID(
232+
ctx context.Context, featureID string) ([]SpannerLatestDailyChromiumHistogramMetric, error) {
233+
return newAllByKeysEntityReader[
234+
latestDailyChromiumHistogramMetricByWebFeatureIDMapper,
235+
string,
236+
SpannerLatestDailyChromiumHistogramMetric,
237+
](c).readAllByKeys(ctx, featureID)
238+
}

‎lib/gcpspanner/spanneradapters/web_features_consumer.go‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,8 @@ import (
2626

2727
// WebFeatureSpannerClient expects a subset of the functionality from lib/gcpspanner that only apply to WebFeatures.
2828
type WebFeatureSpannerClient interface {
29-
SyncWebFeatures(ctx context.Context, features []gcpspanner.WebFeature) error
29+
SyncWebFeatures(ctx context.Context, features []gcpspanner.WebFeature,
30+
options ...gcpspanner.SyncWebFeaturesOption) error
3031
FetchAllWebFeatureIDsAndKeys(ctx context.Context) ([]gcpspanner.SpannerFeatureIDAndKey, error)
3132
UpsertFeatureBaselineStatus(ctx context.Context, featureID string, status gcpspanner.FeatureBaselineStatus) error
3233
UpsertBrowserFeatureAvailability(
@@ -68,9 +69,19 @@ func (c *WebFeaturesConsumer) InsertWebFeatures(
6869
allFeatures = append(allFeatures, webFeature)
6970
}
7071

72+
redirectMap := map[string]string{}
73+
for sourceKey, targetData := range data.Features.Moved {
74+
redirectMap[sourceKey] = targetData.RedirectTarget
75+
}
76+
77+
options := []gcpspanner.SyncWebFeaturesOption{}
78+
if len(redirectMap) > 0 {
79+
options = append(options, gcpspanner.WithRedirectTargets(redirectMap))
80+
}
81+
7182
// 2. Sync all features at once. This will insert, update, and delete features
7283
// to make the database match the desired state.
73-
if err := c.client.SyncWebFeatures(ctx, allFeatures); err != nil {
84+
if err := c.client.SyncWebFeatures(ctx, allFeatures, options...); err != nil {
7485
slog.ErrorContext(ctx, "failed to sync web features", "error", err)
7586

7687
return nil, err

‎lib/gcpspanner/spanneradapters/web_features_consumer_test.go‎

Lines changed: 35 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -197,9 +197,10 @@ func TestGetBaselineStatusEnum(t *testing.T) {
197197
}
198198

199199
type mockSyncWebFeaturesConfig struct {
200-
expectedInput []gcpspanner.WebFeature
201-
err error
202-
expectedCount int
200+
expectedInput []gcpspanner.WebFeature
201+
expectedOptions []gcpspanner.SyncWebFeaturesOption
202+
err error
203+
expectedCount int
203204
}
204205

205206
type mockFetchIDsAndKeysConfig struct {
@@ -294,7 +295,7 @@ func (c *mockWebFeatureSpannerClient) SyncSplitWebFeatures(
294295
}
295296

296297
func (c *mockWebFeatureSpannerClient) SyncWebFeatures(
297-
_ context.Context, features []gcpspanner.WebFeature) error {
298+
_ context.Context, features []gcpspanner.WebFeature, opts ...gcpspanner.SyncWebFeaturesOption) error {
298299
// Sort both slices for stable comparison
299300
sort.Slice(features, func(i, j int) bool {
300301
return features[i].FeatureKey < features[j].FeatureKey
@@ -306,6 +307,12 @@ func (c *mockWebFeatureSpannerClient) SyncWebFeatures(
306307
if diff := cmp.Diff(c.mockSyncWebFeaturesCfg.expectedInput, features); diff != "" {
307308
c.t.Errorf("SyncWebFeatures unexpected input (-want +got):\n%s", diff)
308309
}
310+
311+
if !reflect.DeepEqual(c.mockSyncWebFeaturesCfg.expectedOptions, opts) {
312+
c.t.Errorf("SyncWebFeatures unexpected options expected %v received %v",
313+
c.mockSyncWebFeaturesCfg.expectedOptions, opts)
314+
}
315+
309316
c.syncWebFeaturesCount++
310317

311318
return c.mockSyncWebFeaturesCfg.err
@@ -532,8 +539,9 @@ func TestInsertWebFeatures(t *testing.T) {
532539
DescriptionHTML: "<html>",
533540
},
534541
},
535-
err: nil,
536-
expectedCount: 1,
542+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
543+
err: nil,
544+
expectedCount: 1,
537545
},
538546
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
539547
output: []gcpspanner.SpannerFeatureIDAndKey{
@@ -730,8 +738,9 @@ func TestInsertWebFeatures(t *testing.T) {
730738
expectedInput: []gcpspanner.WebFeature{
731739
{FeatureKey: "feature1", Name: "Feature 1", Description: "text", DescriptionHTML: "<html>"},
732740
},
733-
err: ErrSyncWebFeaturesTest,
734-
expectedCount: 1,
741+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
742+
err: ErrSyncWebFeaturesTest,
743+
expectedCount: 1,
735744
},
736745
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
737746
output: nil,
@@ -815,8 +824,9 @@ func TestInsertWebFeatures(t *testing.T) {
815824
DescriptionHTML: "<html>",
816825
},
817826
},
818-
err: nil,
819-
expectedCount: 1,
827+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
828+
err: nil,
829+
expectedCount: 1,
820830
},
821831
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
822832
output: nil,
@@ -908,8 +918,9 @@ func TestInsertWebFeatures(t *testing.T) {
908918
DescriptionHTML: "<html>",
909919
},
910920
},
911-
err: nil,
912-
expectedCount: 1,
921+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
922+
err: nil,
923+
expectedCount: 1,
913924
},
914925
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
915926
output: nil,
@@ -1012,8 +1023,9 @@ func TestInsertWebFeatures(t *testing.T) {
10121023
DescriptionHTML: "<html>",
10131024
},
10141025
},
1015-
err: nil,
1016-
expectedCount: 1,
1026+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
1027+
err: nil,
1028+
expectedCount: 1,
10171029
},
10181030
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
10191031
output: nil,
@@ -1158,8 +1170,9 @@ func TestInsertWebFeatures(t *testing.T) {
11581170
DescriptionHTML: "<html>",
11591171
},
11601172
},
1161-
err: nil,
1162-
expectedCount: 1,
1173+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
1174+
err: nil,
1175+
expectedCount: 1,
11631176
},
11641177
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
11651178
output: nil,
@@ -1356,8 +1369,9 @@ func TestInsertWebFeatures(t *testing.T) {
13561369
DescriptionHTML: "<html>",
13571370
},
13581371
},
1359-
err: nil,
1360-
expectedCount: 1,
1372+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
1373+
err: nil,
1374+
expectedCount: 1,
13611375
},
13621376
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
13631377
output: nil,
@@ -1556,8 +1570,9 @@ func TestInsertWebFeatures(t *testing.T) {
15561570
DescriptionHTML: "<html>",
15571571
},
15581572
},
1559-
err: nil,
1560-
expectedCount: 1,
1573+
expectedOptions: []gcpspanner.SyncWebFeaturesOption{},
1574+
err: nil,
1575+
expectedCount: 1,
15611576
},
15621577
mockFetchIDsAndKeysCfg: mockFetchIDsAndKeysConfig{
15631578
output: nil,

0 commit comments

Comments
 (0)