This is an automated email from the ASF dual-hosted git repository.
klesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/devlake.git
The following commit(s) were added to refs/heads/main by this push:
new 5eee167f3 fix(customize): skip rows with a missing raw record instead
of ending the scan (#9027)
5eee167f3 is described below
commit 5eee167f3556145ac9199367098e3aafa31698af
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Mon Aug 3 02:54:58 2026 -0400
fix(customize): skip rows with a missing raw record instead of ending the
scan (#9027)
The extractor 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. That value fell through to the default branch of the
type
switch, which returned nil and ended the whole scan.
Nothing surfaced: the subtask reported success, and every row ordered after
the
first orphan silently kept whatever custom field values it already had. The
reporter saw 1,382 orphaned rows out of 30,287 leave roughly 28,000 rows
unpopulated.
Skip the row and carry on. extractCustomizedFields now returns how many
rows it
skipped for that reason, and the caller logs a warning naming the table and
the
count, so the condition is visible instead of silent.
Also return rows.Err() at the end of the scan. An error raised while
iterating
the cursor was previously discarded, which is the same class of silent
partial
processing.
Signed-off-by: Bhanu Chander Vallabaneni <[email protected]>
---
.../customize/tasks/customized_fields_extractor.go | 40 ++++--
.../tasks/customized_fields_extractor_test.go | 150 +++++++++++++++++++++
2 files changed, 177 insertions(+), 13 deletions(-)
diff --git a/backend/plugins/customize/tasks/customized_fields_extractor.go
b/backend/plugins/customize/tasks/customized_fields_extractor.go
index 7bd1d932d..9a8454c66 100644
--- a/backend/plugins/customize/tasks/customized_fields_extractor.go
+++ b/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
diff --git
a/backend/plugins/customize/tasks/customized_fields_extractor_test.go
b/backend/plugins/customize/tasks/customized_fields_extractor_test.go
new file mode 100644
index 000000000..9fc6ff894
--- /dev/null
+++ b/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)
+}
