refactor: Handle no entities found gracefully in event cleanup

- Change deleteReadingsBySubQuery and deleteEvents to return KindEntityDoesNotExist
  instead of KindContractInvalid when no rows are found
- Include SQL statement in error messages for debugging
- Log KindEntityDoesNotExist error at debug level

Signed-off-by: FelixTing <felix@iotechsys.com>

Author: FelixTing

internal/pkg/infrastructure/postgres/event.go

@@ -236,14 +236,22 @@ func (c *Client) deleteEventsByConditions(cols []string, values pgx.NamedArgs) e
 			// select the event-ids of the specified device name from event table as the sub-query of deleting readings
 			subSqlStatement := sqlQueryEventIdFieldsByCol(cols...)
 			if err = deleteReadingsBySubQuery(ctx, tx, subSqlStatement, values); err != nil {
-				c.loggingClient.Errorf("failed delete readings with conditions '%v' '%v': %v", cols, values, err)
-				return err
+				if errors.Kind(err) == errors.KindEntityDoesNotExist {
+					c.loggingClient.Debugf("no readings found for deletion: %s", err.Error())
+				} else {
+					c.loggingClient.Errorf("failed delete readings with conditions '%v' '%v': %v", cols, values, err)
+					return err
+				}
 			}
 
 			err = deleteEvents(ctx, tx, sqlStatement, values)
 			if err != nil {
-				c.loggingClient.Errorf("failed delete event with conditions '%v' '%v': %v", cols, values, err)
-				return err
+				if errors.Kind(err) == errors.KindEntityDoesNotExist {
+					c.loggingClient.Debugf("no events found for deletion: %s", err.Error())
+				} else {
+					c.loggingClient.Errorf("failed delete event with conditions '%v' '%v': %v", cols, values, err)
+					return err
+				}
 			}
 
 			for _, deviceInfo := range deviceInfos {
@@ -297,14 +305,22 @@ func (c *Client) deleteEventsByAgeAndConditions(age int64, cols []string, values
 			// select the event ids within the origin time range from event table as the sub-query of deleting readings
 			subSqlStatement := sqlQueryEventIdFieldByTimeRangeAndConditions(originCol, cols...)
 			if err := deleteReadingsBySubQuery(ctx, tx, subSqlStatement, values); err != nil {
-				c.loggingClient.Errorf("failed delete readings by age '%d' nanoseconds: %v", age, err)
-				return err
+				if errors.Kind(err) == errors.KindEntityDoesNotExist {
+					c.loggingClient.Debugf("no readings found for deletion by age '%d' nanoseconds: %s", age, err.Error())
+				} else {
+					c.loggingClient.Errorf("failed delete readings by age '%d' nanoseconds: %v", age, err)
+					return err
+				}
 			}
 
 			err := deleteEvents(ctx, tx, sqlStatement, values)
 			if err != nil {
-				c.loggingClient.Errorf("failed delete event by age '%d' nanoseconds: %v", age, err)
-				return err
+				if errors.Kind(err) == errors.KindEntityDoesNotExist {
+					c.loggingClient.Debugf("no events found for deletion by age '%d' nanoseconds: %s", age, err.Error())
+				} else {
+					c.loggingClient.Errorf("failed delete event by age '%d' nanoseconds: %v", age, err)
+					return err
+				}
 			}
 			return nil
 		})
@@ -356,7 +372,7 @@ func deleteEvents(ctx context.Context, tx pgx.Tx, sqlStatement string, args pgx.
 		return pgClient.WrapDBError("event(s) delete failed", err)
 	}
 	if commandTag.RowsAffected() == 0 {
-		return errors.NewCommonEdgeX(errors.KindContractInvalid, "no event found", nil)
+		return errors.NewCommonEdgeX(errors.KindEntityDoesNotExist, fmt.Sprintf("no event found; SQL statement: %s", sqlStatement), nil)
 	}
 	return nil
 }

internal/pkg/infrastructure/postgres/event_test.go

@@ -0,0 +1,149 @@
+//
+// Copyright (C) 2025 IOTech Ltd
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/edgexfoundry/edgex-go/internal/pkg/infrastructure/postgres/mocks"
+	"github.com/edgexfoundry/go-mod-core-contracts/v4/errors"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestDeleteEvents(t *testing.T) {
+	ctx := context.Background()
+	sqlStatement := "DELETE FROM event WHERE id = @id"
+	args := pgx.NamedArgs{"id": "test-id"}
+
+	tests := []struct {
+		name            string
+		rowsAffected    int64
+		execError       error
+		expectError     bool
+		expectedErrKind errors.ErrKind
+		errorContains   []string
+	}{
+		{
+			name:            "No rows affected - should return KindEntityDoesNotExist",
+			rowsAffected:    0,
+			execError:       nil,
+			expectError:     true,
+			expectedErrKind: errors.KindEntityDoesNotExist,
+			errorContains:   []string{"no event found", "SQL statement:", sqlStatement},
+		},
+		{
+			name:            "Rows affected - should succeed",
+			rowsAffected:    1,
+			execError:       nil,
+			expectError:     false,
+			expectedErrKind: "",
+			errorContains:   nil,
+		},
+		{
+			name:            "Exec returns error - should return KindDatabaseError",
+			rowsAffected:    0,
+			execError:       fmt.Errorf("database connection error"),
+			expectError:     true,
+			expectedErrKind: errors.KindDatabaseError,
+			errorContains:   []string{"event(s) delete failed"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tx := new(mocks.Tx)
+			// On successful completion, a DELETE command returns a command tag of the form "DELETE count"
+			// https://www.postgresql.org/docs/16/sql-delete.html
+			commandTag := pgconn.NewCommandTag(fmt.Sprintf("DELETE %d", tt.rowsAffected))
+			tx.On("Exec", ctx, sqlStatement, args).Return(commandTag, tt.execError)
+
+			err := deleteEvents(ctx, tx, sqlStatement, args)
+
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Equal(t, tt.expectedErrKind, errors.Kind(err))
+				for _, contains := range tt.errorContains {
+					assert.Contains(t, err.Error(), contains)
+				}
+			} else {
+				assert.NoError(t, err)
+			}
+			tx.AssertExpectations(t)
+		})
+	}
+}
+
+func TestDeleteReadingsBySubQuery(t *testing.T) {
+	ctx := context.Background()
+	subQuerySql := "SELECT id FROM event WHERE devicename = @devicename"
+	args := pgx.NamedArgs{"devicename": "test-device"}
+
+	tests := []struct {
+		name            string
+		rowsAffected    int64
+		execError       error
+		expectError     bool
+		expectedErrKind errors.ErrKind
+		errorContains   []string
+	}{
+		{
+			name:            "No rows affected - should return KindEntityDoesNotExist",
+			rowsAffected:    0,
+			execError:       nil,
+			expectError:     true,
+			expectedErrKind: errors.KindEntityDoesNotExist,
+			errorContains:   []string{"no reading found", "SQL statement:"},
+		},
+		{
+			name:            "Rows affected - should succeed",
+			rowsAffected:    1,
+			execError:       nil,
+			expectError:     false,
+			expectedErrKind: "",
+			errorContains:   nil,
+		},
+		{
+			name:            "Exec returns error - should return KindDatabaseError",
+			rowsAffected:    1, // Use non-zero to avoid RowsAffected check before error check
+			execError:       fmt.Errorf("database connection error"),
+			expectError:     true,
+			expectedErrKind: errors.KindDatabaseError,
+			errorContains:   []string{"reading(s) delete failed"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tx := new(mocks.Tx)
+			// On successful completion, a DELETE command returns a command tag of the form "DELETE count"
+			// https://www.postgresql.org/docs/16/sql-delete.html
+			commandTag := pgconn.NewCommandTag(fmt.Sprintf("DELETE %d", tt.rowsAffected))
+			tx.On("Exec", ctx, mock.MatchedBy(func(sql string) bool {
+				return strings.Contains(sql, subQuerySql)
+			}), args).Return(commandTag, tt.execError)
+
+			err := deleteReadingsBySubQuery(ctx, tx, subQuerySql, args)
+
+			if tt.expectError {
+				assert.Error(t, err)
+				assert.Equal(t, tt.expectedErrKind, errors.Kind(err))
+				for _, contains := range tt.errorContains {
+					assert.Contains(t, err.Error(), contains)
+				}
+			} else {
+				assert.NoError(t, err)
+			}
+			tx.AssertExpectations(t)
+		})
+	}
+}

internal/pkg/infrastructure/postgres/mocks/tx.go

@@ -0,0 +1,77 @@
+//
+// Copyright (C) 2025 IOTech Ltd
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package mocks
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/stretchr/testify/mock"
+)
+
+// Tx is a mock implementation of pgx.Tx for testing
+type Tx struct {
+	mock.Mock
+}
+
+func (t *Tx) Begin(ctx context.Context) (pgx.Tx, error) {
+	args := t.Called(ctx)
+	return args.Get(0).(pgx.Tx), args.Error(1)
+}
+
+func (t *Tx) Commit(ctx context.Context) error {
+	args := t.Called(ctx)
+	return args.Error(0)
+}
+
+func (t *Tx) Rollback(ctx context.Context) error {
+	args := t.Called(ctx)
+	return args.Error(0)
+}
+
+func (t *Tx) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error) {
+	args := t.Called(ctx, tableName, columnNames, rowSrc)
+	return args.Get(0).(int64), args.Error(1)
+}
+
+func (t *Tx) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults {
+	args := t.Called(ctx, b)
+	return args.Get(0).(pgx.BatchResults)
+}
+
+func (t *Tx) LargeObjects() pgx.LargeObjects {
+	args := t.Called()
+	return args.Get(0).(pgx.LargeObjects)
+}
+
+func (t *Tx) Prepare(ctx context.Context, name, sql string) (*pgconn.StatementDescription, error) {
+	args := t.Called(ctx, name, sql)
+	return args.Get(0).(*pgconn.StatementDescription), args.Error(1)
+}
+
+func (t *Tx) Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error) {
+	var callArgs []interface{}
+	callArgs = []interface{}{ctx, sql}
+	callArgs = append(callArgs, arguments...)
+	mockArgs := t.Called(callArgs...)
+	return mockArgs.Get(0).(pgconn.CommandTag), mockArgs.Error(1)
+}
+
+func (t *Tx) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) {
+	mockArgs := t.Called(ctx, sql, args)
+	return mockArgs.Get(0).(pgx.Rows), mockArgs.Error(1)
+}
+
+func (t *Tx) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
+	mockArgs := t.Called(ctx, sql, args)
+	return mockArgs.Get(0).(pgx.Row)
+}
+
+func (t *Tx) Conn() *pgx.Conn {
+	args := t.Called()
+	return args.Get(0).(*pgx.Conn)
+}

internal/pkg/infrastructure/postgres/reading.go

@@ -364,7 +364,7 @@ func deleteReadingsBySubQuery(ctx context.Context, tx pgx.Tx, subQuerySql string
 		args,
 	)
 	if commandTag.RowsAffected() == 0 {
-		return errors.NewCommonEdgeX(errors.KindContractInvalid, "no reading found", nil)
+		return errors.NewCommonEdgeX(errors.KindEntityDoesNotExist, fmt.Sprintf("no reading found; SQL statement: %s", sqlStatement), nil)
 	}
 	if err != nil {
 		return pgClient.WrapDBError("reading(s) delete failed", err)