fix(customize): skip rows with a missing raw record instead of ending the scan

backend/plugins/customize/tasks/customized_fields_extractor.go

@@ -46,20 +46,28 @@ func ExtractCustomizedFields(taskCtx plugin.SubTaskContext) errors.Error {
 		return nil
 	}
 	d := taskCtx.GetDal()
-	var err error
+	logger := taskCtx.GetLogger()
 	for _, rule := range data.Options.TransformationRules {
-		err = extractCustomizedFields(taskCtx.GetContext(), d, rule.Table, rule.RawDataTable, rule.RawDataParams, rule.Mapping)
+		orphaned, err := extractCustomizedFields(taskCtx.GetContext(), d, rule.Table, rule.RawDataTable, rule.RawDataParams, rule.Mapping)
 		if err != nil {
 			return errors.Default.Wrap(err, "error extracting customized fields")
 		}
+		if orphaned > 0 {
+			logger.Warn(nil,
+				"skipped %d row(s) of table %s: the raw record referenced by _raw_data_id no longer exists in %s",
+				orphaned, rule.Table, rule.RawDataTable)
+		}
 	}
 	return nil
 }
 
-func extractCustomizedFields(ctx context.Context, d dal.Dal, table, rawTable, rawDataParams string, extractor map[string]string) error {
+// extractCustomizedFields walks the domain layer table and copies configured JSON paths out of the
+// raw record behind each row. It returns the number of rows skipped because no raw record backed
+// them, so the caller can report that rather than leaving it invisible.
+func extractCustomizedFields(ctx context.Context, d dal.Dal, table, rawTable, rawDataParams string, extractor map[string]string) (int, error) {
 	pkFields, err := dal.GetPrimarykeyColumns(d, &models.Table{Name: table})
 	if err != nil {
-		return err
+		return 0, err
 	}
 	rawDataField := fmt.Sprintf("%s.data", rawTable)
 	// `fields` only include `_raw_data_id` and primary keys coming from the domain layer table, and `data` coming from the raw layer
@@ -76,21 +84,22 @@ func extractCustomizedFields(ctx context.Context, d dal.Dal, table, rawTable, ra
 	}
 	rows, err := d.Cursor(clauses...)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	defer rows.Close()
 
+	orphaned := 0
 	for rows.Next() {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return orphaned, ctx.Err()
 		default:
 		}
 		row := make(map[string]interface{})
 		updates := make(map[string]interface{})
 		err = d.Fetch(rows, &row)
 		if err != nil {
-			return err
+			return orphaned, err
 		}
 		switch blob := row["data"].(type) {
 		case []byte:
@@ -104,7 +113,7 @@ func extractCustomizedFields(ctx context.Context, d dal.Dal, table, rawTable, ra
 				// special case for issues custom_fields
 				rawDataId, ok := row["_raw_data_id"].(int64)
 				if !ok {
-					return errors.Default.New("_raw_data_id is not int64")
+					return orphaned, errors.Default.New("_raw_data_id is not int64")
 				}
 				if table == "issues" && result.IsArray() {
 					issueId := row["id"].(string)
@@ -115,7 +124,7 @@ func extractCustomizedFields(ctx context.Context, d dal.Dal, table, rawTable, ra
 						dal.Where("issue_id = ? AND field_id = ?", issueId, fieldId),
 					)
 					if err != nil {
-						return err
+						return orphaned, err
 					}
 
 					result.ForEach(func(_, v gjson.Result) bool {
@@ -142,7 +151,12 @@ func extractCustomizedFields(ctx context.Context, d dal.Dal, table, rawTable, ra
 				}
 			}
 		default:
-			return nil
+			// The cursor LEFT JOINs the raw table, so a domain row whose _raw_data_id points at a
+			// raw record that has since been deleted comes back with a NULL data column. There is
+			// nothing to extract from it, but the rows after it are usually fine, so skip this one
+			// instead of ending the scan.
+			orphaned++
+			continue
 		}
 
 		if len(updates) > 0 {
@@ -151,15 +165,15 @@ func extractCustomizedFields(ctx context.Context, d dal.Dal, table, rawTable, ra
 			delete(row, "data")
 			query, params, err := mkUpdate(table, updates, row)
 			if err != nil {
-				return err
+				return orphaned, err
 			}
 			err = d.Exec(query, params...)
 			if err != nil {
-				return errors.Default.Wrap(err, "Exec SQL error")
+				return orphaned, errors.Default.Wrap(err, "Exec SQL error")
 			}
 		}
 	}
-	return nil
+	return orphaned, rows.Err()
 }
 
 // fillInUpdates fills in the updates map with the result of the gjson query

backend/plugins/customize/tasks/customized_fields_extractor_test.go

@@ -0,0 +1,150 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"context"
+	"testing"
+
+	"github.com/apache/incubator-devlake/core/dal"
+	mockdal "github.com/apache/incubator-devlake/mocks/core/dal"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+// fetchedRow is one row the mocked cursor hands back, in the shape d.Fetch produces.
+type fetchedRow map[string]interface{}
+
+// newExtractorMocks wires a Dal whose cursor yields the given rows in order. It returns the Dal
+// mock plus a pointer to a slice collecting every UPDATE statement executed, so a test can tell
+// which rows actually made it through.
+func newExtractorMocks(t *testing.T, rows []fetchedRow) (*mockdal.Dal, *[]string) {
+	t.Helper()
+
+	idColumn := new(mockdal.ColumnMeta)
+	idColumn.On("Name").Return("id").Maybe()
+	idColumn.On("PrimaryKey").Return(true, true).Maybe()
+
+	cursor := new(mockdal.Rows)
+	// Next reports true once per row, then false to end the scan.
+	call := 0
+	cursor.On("Next").Return(func() bool {
+		ok := call < len(rows)
+		call++
+		return ok
+	}).Maybe()
+	cursor.On("Close").Return(nil).Maybe()
+	cursor.On("Err").Return(nil).Maybe()
+
+	d := new(mockdal.Dal)
+	d.On("GetColumns", mock.Anything, mock.Anything).
+		Return([]dal.ColumnMeta{idColumn}, nil).Maybe()
+	d.On("Cursor", mock.Anything).Return(cursor, nil).Maybe()
+
+	// Fetch copies the row for the current cursor position into the destination map.
+	fetched := 0
+	d.On("Fetch", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		dst, ok := args.Get(1).(*map[string]interface{})
+		if !ok {
+			t.Fatalf("Fetch called with %T, want *map[string]interface{}", args.Get(1))
+		}
+		for k, v := range rows[fetched] {
+			(*dst)[k] = v
+		}
+		fetched++
+	}).Return(nil).Maybe()
+
+	executed := make([]string, 0)
+	d.On("Exec", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		executed = append(executed, args.Get(0).(string))
+	}).Return(nil).Maybe()
+
+	return d, &executed
+}
+
+func Test_extractCustomizedFields_skipsOrphanedRowsAndKeepsScanning(t *testing.T) {
+	// The cursor LEFT JOINs the raw table, so a domain row whose raw record has been deleted comes
+	// back with a NULL data column. Before the fix that row ended the whole scan, leaving every
+	// later row unpopulated while the subtask still reported success.
+	rows := []fetchedRow{
+		{"_raw_data_id": int64(1), "id": "ISSUE-1", "data": nil},
+		{"_raw_data_id": int64(2), "id": "ISSUE-2", "data": `{"foo":"bar"}`},
+		{"_raw_data_id": int64(3), "id": "ISSUE-3", "data": nil},
+		{"_raw_data_id": int64(4), "id": "ISSUE-4", "data": `{"foo":"baz"}`},
+	}
+	d, executed := newExtractorMocks(t, rows)
+
+	orphaned, err := extractCustomizedFields(
+		context.Background(), d, "issues", "_raw_jira_api_issues", "params",
+		map[string]string{"x_custom": "foo"})
+
+	assert.NoError(t, err)
+	assert.Equal(t, 2, orphaned, "both rows without a raw record should be counted")
+	assert.Len(t, *executed, 2, "the two rows that do have raw data should still be updated")
+}
+
+func Test_extractCustomizedFields_allRowsOrphanedIsNotAnError(t *testing.T) {
+	// A rule whose raw records have all been cleaned up is not a failure; it just has nothing to
+	// extract. The caller reports the count so it does not pass unnoticed.
+	rows := []fetchedRow{
+		{"_raw_data_id": int64(1), "id": "ISSUE-1", "data": nil},
+		{"_raw_data_id": int64(2), "id": "ISSUE-2", "data": nil},
+	}
+	d, executed := newExtractorMocks(t, rows)
+
+	orphaned, err := extractCustomizedFields(
+		context.Background(), d, "issues", "_raw_jira_api_issues", "params",
+		map[string]string{"x_custom": "foo"})
+
+	assert.NoError(t, err)
+	assert.Equal(t, 2, orphaned)
+	assert.Empty(t, *executed)
+}
+
+func Test_extractCustomizedFields_noOrphansReportsZero(t *testing.T) {
+	rows := []fetchedRow{
+		{"_raw_data_id": int64(1), "id": "ISSUE-1", "data": `{"foo":"bar"}`},
+		{"_raw_data_id": int64(2), "id": "ISSUE-2", "data": `{"foo":"baz"}`},
+	}
+	d, executed := newExtractorMocks(t, rows)
+
+	orphaned, err := extractCustomizedFields(
+		context.Background(), d, "issues", "_raw_jira_api_issues", "params",
+		map[string]string{"x_custom": "foo"})
+
+	assert.NoError(t, err)
+	assert.Zero(t, orphaned)
+	assert.Len(t, *executed, 2)
+}
+
+func Test_extractCustomizedFields_honoursCancelledContext(t *testing.T) {
+	rows := []fetchedRow{
+		{"_raw_data_id": int64(1), "id": "ISSUE-1", "data": `{"foo":"bar"}`},
+	}
+	d, executed := newExtractorMocks(t, rows)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err := extractCustomizedFields(
+		ctx, d, "issues", "_raw_jira_api_issues", "params",
+		map[string]string{"x_custom": "foo"})
+
+	assert.ErrorIs(t, err, context.Canceled)
+	assert.Empty(t, *executed)
+}