From cc28035881c7b93b485fc97f6d34caabe53fa441 Mon Sep 17 00:00:00 2001 From: euvre <93761161+euvre@users.noreply.github.com> Date: Thu, 20 Aug 2026 04:04:12 -0700 Subject: [PATCH] fix(ingestion): show data pipeline column for builtin pipeline ingestion logs (#18484) --- internal/ingestion/task/pipeline_executor.go | 30 ++++-- .../ingestion/task/pipeline_executor_test.go | 101 +++++++++++++++++- 2 files changed, 120 insertions(+), 11 deletions(-) diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go index ad9a480c33..dfab1be12e 100644 --- a/internal/ingestion/task/pipeline_executor.go +++ b/internal/ingestion/task/pipeline_executor.go @@ -19,6 +19,7 @@ package task import ( "context" "encoding/json" + "errors" "fmt" "ragflow/internal/utility" "sort" @@ -646,16 +647,25 @@ func (s *PipelineExecutor) recordPipelineLog(ctx context.Context, db *gorm.DB, d } } - pipelineTitle := "" - var pipelineAvatar *string - if db != nil { - if canvas, err := dao.NewUserCanvasDAO().GetByID(ctx, db, s.canvasID); err == nil && canvas != nil { - if canvas.Title != nil { - pipelineTitle = *canvas.Title + // Pipeline identity for the log row. A document without a user pipeline + // selection runs on a builtin registry pipeline: its canvasID is the + // parser_id, not a canvas row, so the log is titled with the document's + // parser_id, reuses the document thumbnail as avatar, and leaves + // pipeline_id empty. + pipelineTitle := doc.ParserID + pipelineAvatar := doc.Thumbnail + var pipelineID *string + if s.taskCtx.PipelineID != "" { + pipelineID = &s.canvasID + if db != nil { + if canvas, err := dao.NewUserCanvasDAO().GetByID(ctx, db, s.canvasID); err == nil && canvas != nil { + if canvas.Title != nil { + pipelineTitle = *canvas.Title + } + pipelineAvatar = canvas.Avatar + } else if err != nil && !errors.Is(err, dao.ErrUserCanvasNotFound) { + common.Warn(fmt.Sprintf("failed to reload pipeline %s for operation log: %v", s.canvasID, err)) } - pipelineAvatar = canvas.Avatar - } else if err != nil { - common.Warn(fmt.Sprintf("failed to reload pipeline %s for operation log: %v", s.canvasID, err)) } } @@ -680,7 +690,7 @@ func (s *PipelineExecutor) recordPipelineLog(ctx context.Context, db *gorm.DB, d TenantID: s.Tenant().ID, KbID: s.KB().ID, DocumentID: docID, - PipelineID: &s.canvasID, + PipelineID: pipelineID, PipelineTitle: &pipelineTitle, TaskType: string(entity.PipelineTaskTypeParse), DSL: dslMap, diff --git a/internal/ingestion/task/pipeline_executor_test.go b/internal/ingestion/task/pipeline_executor_test.go index 021d409fa9..af4b8a3f74 100644 --- a/internal/ingestion/task/pipeline_executor_test.go +++ b/internal/ingestion/task/pipeline_executor_test.go @@ -315,6 +315,102 @@ func TestRecordPipelineLog_ValidJSONParsed(t *testing.T) { } } +func TestRecordPipelineLog_BuiltinUsesParserIDFallback(t *testing.T) { + cleanup := setupPipelineExecutorTestDB(t) + defer cleanup() + + taskCtx := makeTaskCtx() + taskCtx.Doc.ParserID = "general" + taskCtx.Doc.Thumbnail = strPtr("thumb.png") + + var captured *entity.PipelineOperationLog + svc := mustNewPipelineExecutor(t, taskCtx, "general", 0). + WithLogCreateFunc(func(ctx context.Context, db *gorm.DB, log *entity.PipelineOperationLog) error { + captured = log + return nil + }) + svc.recordPipelineLog(t.Context(), dao.DB, "doc-1", `{}`, "done") + + if captured == nil { + t.Fatal("logCreateFunc was not called") + } + if captured.PipelineTitle == nil || *captured.PipelineTitle != "general" { + t.Fatalf("PipelineTitle = %v, want \"general\"", captured.PipelineTitle) + } + if captured.Avatar == nil || *captured.Avatar != "thumb.png" { + t.Fatalf("Avatar = %v, want \"thumb.png\"", captured.Avatar) + } + if captured.PipelineID != nil { + t.Fatalf("PipelineID = %q, want nil for builtin pipeline", *captured.PipelineID) + } +} + +func TestRecordPipelineLog_CustomCanvasTitle(t *testing.T) { + cleanup := setupPipelineExecutorTestDB(t) + defer cleanup() + + if err := dao.DB.Create(&entity.UserCanvas{ + ID: "canvas-1", + UserID: "tenant-1", + Title: strPtr("My Pipeline"), + Avatar: strPtr("a.png"), + }).Error; err != nil { + t.Fatalf("seed canvas: %v", err) + } + + taskCtx := makeTaskCtx() + taskCtx.Doc.ParserID = "general" + taskCtx.PipelineID = "canvas-1" + + var captured *entity.PipelineOperationLog + svc := mustNewPipelineExecutor(t, taskCtx, "canvas-1", 0). + WithLogCreateFunc(func(ctx context.Context, db *gorm.DB, log *entity.PipelineOperationLog) error { + captured = log + return nil + }) + svc.recordPipelineLog(t.Context(), dao.DB, "doc-1", `{}`, "done") + + if captured == nil { + t.Fatal("logCreateFunc was not called") + } + if captured.PipelineTitle == nil || *captured.PipelineTitle != "My Pipeline" { + t.Fatalf("PipelineTitle = %v, want \"My Pipeline\"", captured.PipelineTitle) + } + if captured.Avatar == nil || *captured.Avatar != "a.png" { + t.Fatalf("Avatar = %v, want \"a.png\"", captured.Avatar) + } + if captured.PipelineID == nil || *captured.PipelineID != "canvas-1" { + t.Fatalf("PipelineID = %v, want \"canvas-1\"", captured.PipelineID) + } +} + +func TestRecordPipelineLog_CustomCanvasMissingFallsBackToParserID(t *testing.T) { + cleanup := setupPipelineExecutorTestDB(t) + defer cleanup() + + taskCtx := makeTaskCtx() + taskCtx.Doc.ParserID = "general" + taskCtx.PipelineID = "canvas-gone" + + var captured *entity.PipelineOperationLog + svc := mustNewPipelineExecutor(t, taskCtx, "canvas-gone", 0). + WithLogCreateFunc(func(ctx context.Context, db *gorm.DB, log *entity.PipelineOperationLog) error { + captured = log + return nil + }) + svc.recordPipelineLog(t.Context(), dao.DB, "doc-1", `{}`, "done") + + if captured == nil { + t.Fatal("logCreateFunc was not called") + } + if captured.PipelineTitle == nil || *captured.PipelineTitle != "general" { + t.Fatalf("PipelineTitle = %v, want \"general\" fallback", captured.PipelineTitle) + } + if captured.PipelineID == nil || *captured.PipelineID != "canvas-gone" { + t.Fatalf("PipelineID = %v, want \"canvas-gone\"", captured.PipelineID) + } +} + // ============================================================================= // updateDocumentMetadata // ============================================================================= @@ -405,7 +501,10 @@ func TestPipelineExecutor_Run_MainFlowWithStubs(t *testing.T) { logged := false inserted := false - svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0). + taskCtx := makeTaskCtx() + taskCtx.PipelineID = "flow-1" + + svc := mustNewPipelineExecutor(t, taskCtx, "flow-1", 0). WithLoadDSLFunc(func(ctx context.Context, canvasID string) (string, string, error) { return `{"nodes":[{"id":"n1"}],"edges":[]}`, "flow-corrected", nil }).