Cleanup old entries from resource_last_import_time table. (#112438)

* Cleanup old entries from resource_last_import_time table.

* Add index for last_import_time column.

* Address review feedback.

pkg/storage/unified/sql/backend.go

@@ -16,6 +16,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"go.opentelemetry.io/otel/trace"
 	"go.opentelemetry.io/otel/trace/noop"
+	"go.uber.org/atomic"
 	"google.golang.org/protobuf/proto"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -67,6 +68,9 @@ type BackendOptions struct {
 
 	// testing
 	SimulatedNetworkLatency time.Duration // slows down the create transactions by a fixed amount
+
+	// If not zero, the backend will regularly remove times from resource_last_import_time table older than this.
+	LastImportTimeMaxAge time.Duration
 }
 
 func NewBackend(opts BackendOptions) (Backend, error) {
@@ -98,6 +102,7 @@ func NewBackend(opts BackendOptions) (Backend, error) {
 		bulkLock:                &bulkLock{running: make(map[string]bool)},
 		simulatedNetworkLatency: opts.SimulatedNetworkLatency,
 		withPruner:              opts.withPruner,
+		lastImportTimeMaxAge:    opts.LastImportTimeMaxAge,
 	}, nil
 }
 
@@ -137,6 +142,9 @@ type backend struct {
 
 	historyPruner resource.Pruner
 	withPruner    bool
+
+	lastImportTimeMaxAge       time.Duration
+	lastImportTimeDeletionTime atomic.Time
 }
 
 func (b *backend) Init(ctx context.Context) error {
@@ -965,10 +973,36 @@ func (b *backend) fetchLatestHistoryRV(ctx context.Context, x db.ContextExecer,
 	return res.ResourceVersion, nil
 }
 
+// Don't run deletion of "last import times" more often than this duration.
+const limitLastImportTimesDeletion = 1 * time.Hour
+
 func (b *backend) GetResourceLastImportTimes(ctx context.Context) iter.Seq2[resource.ResourceLastImportTime, error] {
 	ctx, span := b.tracer.Start(ctx, tracePrefix+"GetLastImportTimes")
 	defer span.End()
 
+	// Delete old entries, if configured, and if enough time has passed since last deletion.
+	if b.lastImportTimeMaxAge > 0 && time.Since(b.lastImportTimeDeletionTime.Load()) > limitLastImportTimesDeletion {
+		now := time.Now()
+
+		res, err := dbutil.Exec(ctx, b.db, sqlResourceLastImportTimeDelete, &sqlResourceLastImportTimeDeleteRequest{
+			SQLTemplate: sqltemplate.New(b.dialect),
+			Threshold:   now.Add(-b.lastImportTimeMaxAge),
+		})
+
+		if err != nil {
+			return func(yield func(resource.ResourceLastImportTime, error) bool) {
+				yield(resource.ResourceLastImportTime{}, err)
+			}
+		}
+
+		aff, err := res.RowsAffected()
+		if err == nil && aff > 0 {
+			b.log.Info("Deleted old last import times", "rows", aff)
+		}
+
+		b.lastImportTimeDeletionTime.Store(now)
+	}
+
 	rows, err := dbutil.QueryRows(ctx, b.db, sqlResourceLastImportTimeQuery, &sqlResourceLastImportTimeQueryRequest{SQLTemplate: sqltemplate.New(b.dialect)})
 	if err != nil {
 		return func(yield func(resource.ResourceLastImportTime, error) bool) {

pkg/storage/unified/sql/data/resource_last_import_time_delete.sql

@@ -0,0 +1,3 @@
+DELETE FROM {{ .Ident "resource_last_import_time" }}
+WHERE {{ .Ident "last_import_time" }} <= {{ .Arg .Threshold }}
+;

pkg/storage/unified/sql/db/migrations/resource_mig.go

@@ -116,7 +116,7 @@ func initResourceTables(mg *migrator.Migrator) string {
 		},
 	})
 
-	tables = append(tables, migrator.Table{
+	resource_last_import_time := migrator.Table{
 		Name: "resource_last_import_time",
 		Columns: []*migrator.Column{
 			{Name: "group", Type: migrator.DB_NVarchar, Length: 190, Nullable: false},
@@ -125,7 +125,8 @@ func initResourceTables(mg *migrator.Migrator) string {
 			{Name: "last_import_time", Type: migrator.DB_DateTime, Nullable: false},
 		},
 		PrimaryKeys: []string{"group", "resource", "namespace"},
-	})
+	}
+	tables = append(tables, resource_last_import_time)
 
 	// Initialize all tables
 	for t := range tables {
@@ -178,5 +179,11 @@ func initResourceTables(mg *migrator.Migrator) string {
 		Name: "IDX_resource_history_namespace_group_resource_name_generation",
 	}))
 
+	mg.AddMigration("Add UQE_resource_last_import_time_last_import_time index", migrator.NewAddIndexMigration(resource_last_import_time, &migrator.Index{
+		Cols: []string{"last_import_time"},
+		Type: migrator.IndexType,
+		Name: "UQE_resource_last_import_time_last_import_time",
+	}))
+
 	return marker
 }

pkg/storage/unified/sql/queries.go

@@ -61,6 +61,7 @@ var (
 
 	sqlResourceLastImportTimeInsert = mustTemplate("resource_last_import_time_insert.sql")
 	sqlResourceLastImportTimeQuery  = mustTemplate("resource_last_import_time_query.sql")
+	sqlResourceLastImportTimeDelete = mustTemplate("resource_last_import_time_delete.sql")
 )
 
 // TxOptions.
@@ -489,3 +490,12 @@ type sqlResourceLastImportTimeQueryRequest struct {
 func (r *sqlResourceLastImportTimeQueryRequest) Validate() error {
 	return nil
 }
+
+type sqlResourceLastImportTimeDeleteRequest struct {
+	sqltemplate.SQLTemplate
+	Threshold time.Time
+}
+
+func (r *sqlResourceLastImportTimeDeleteRequest) Validate() error {
+	return nil
+}

pkg/storage/unified/sql/queries_test.go

@@ -515,5 +515,14 @@ func TestUnifiedStorageQueries(t *testing.T) {
 					},
 				},
 			},
+			sqlResourceLastImportTimeDelete: {
+				{
+					Name: "delete",
+					Data: &sqlResourceLastImportTimeDeleteRequest{
+						SQLTemplate: mocks.NewTestingSQLTemplate(),
+						Threshold:   time.Date(2025, 10, 15, 14, 30, 05, 0, time.UTC),
+					},
+				},
+			},
 		}})
 }

pkg/storage/unified/sql/server.go

@@ -102,12 +102,13 @@ func NewResourceServer(opts ServerOptions) (resource.ResourceServer, error) {
 		withPruner := opts.Features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageHistoryPruner)
 
 		backend, err := NewBackend(BackendOptions{
-			DBProvider:     eDB,
-			Tracer:         opts.Tracer,
-			Reg:            opts.Reg,
-			IsHA:           isHA,
-			withPruner:     withPruner,
-			storageMetrics: opts.StorageMetrics,
+			DBProvider:           eDB,
+			Tracer:               opts.Tracer,
+			Reg:                  opts.Reg,
+			IsHA:                 isHA,
+			withPruner:           withPruner,
+			storageMetrics:       opts.StorageMetrics,
+			LastImportTimeMaxAge: opts.SearchOptions.MaxIndexAge, // No need to keep last_import_times older than max index age.
 		})
 		if err != nil {
 			return nil, err

pkg/storage/unified/sql/test/integration_test.go

@@ -53,6 +53,7 @@ func newTestBackend(t *testing.T, isHA bool, simulatedNetworkLatency time.Durati
 		DBProvider:              eDB,
 		IsHA:                    isHA,
 		SimulatedNetworkLatency: simulatedNetworkLatency,
+		LastImportTimeMaxAge:    24 * time.Hour,
 	})
 	require.NoError(t, err)
 	require.NotNil(t, backend)

pkg/storage/unified/sql/testdata/mysql--resource_last_import_time_delete-delete.sql

@@ -0,0 +1,3 @@
+DELETE FROM `resource_last_import_time`
+WHERE `last_import_time` <= '2025-10-15 14:30:05 +0000 UTC'
+;

pkg/storage/unified/sql/testdata/postgres--resource_last_import_time_delete-delete.sql

@@ -0,0 +1,3 @@
+DELETE FROM "resource_last_import_time"
+WHERE "last_import_time" <= '2025-10-15 14:30:05 +0000 UTC'
+;

pkg/storage/unified/sql/testdata/sqlite--resource_last_import_time_delete-delete.sql

@@ -0,0 +1,3 @@
+DELETE FROM "resource_last_import_time"
+WHERE "last_import_time" <= '2025-10-15 14:30:05 +0000 UTC'
+;