diff --git a/internal/dao/kb.go b/internal/dao/kb.go index 6f40e9d2d7..5d70fda66d 100644 --- a/internal/dao/kb.go +++ b/internal/dao/kb.go @@ -478,37 +478,6 @@ func (dao *KnowledgebaseDAO) GetAllIDs(ctx context.Context, db *gorm.DB) ([]stri return kbIDs, err } -// UpdateParserConfig updates the parser configuration with deep merge -// This matches the Python update_parser_config method -func (dao *KnowledgebaseDAO) UpdateParserConfig(ctx context.Context, db *gorm.DB, id string, config map[string]interface{}) error { - var kb entity.Knowledgebase - if err := db.WithContext(ctx).Where("id = ? AND status = ?", id, string(entity.StatusValid)).First(&kb).Error; err != nil { - return err - } - - mergedConfig := mergeConfig(kb.ParserConfig, config) - return db.WithContext(ctx).Model(&entity.Knowledgebase{}). - Where("id = ?", id). - Update("parser_config", mergedConfig).Error -} - -// DeleteFieldMap removes the field_map from parser_config -// This matches the Python delete_field_map method -func (dao *KnowledgebaseDAO) DeleteFieldMap(ctx context.Context, db *gorm.DB, id string) error { - var kb entity.Knowledgebase - if err := db.WithContext(ctx).Where("id = ? AND status = ?", id, string(entity.StatusValid)).First(&kb).Error; err != nil { - return err - } - - if kb.ParserConfig != nil { - delete(kb.ParserConfig, "field_map") - return db.WithContext(ctx).Model(&entity.Knowledgebase{}). - Where("id = ?", id). - Update("parser_config", kb.ParserConfig).Error - } - return nil -} - // GetFieldMap retrieves field mappings from multiple knowledge bases // This matches the Python get_field_map method func (dao *KnowledgebaseDAO) GetFieldMap(ctx context.Context, db *gorm.DB, ids []string) (map[string]interface{}, error) { @@ -598,43 +567,6 @@ func (dao *KnowledgebaseDAO) GetList(ctx context.Context, db *gorm.DB, tenantIDs return kbs, total, nil } -// mergeConfig performs a deep merge of configuration maps -func mergeConfig(old, new map[string]interface{}) map[string]interface{} { - result := make(map[string]interface{}) - for k, v := range old { - result[k] = v - } - - for k, v := range new { - if existing, ok := result[k]; ok { - if existingMap, ok := existing.(map[string]interface{}); ok { - if newMap, ok := v.(map[string]interface{}); ok { - result[k] = mergeConfig(existingMap, newMap) - continue - } - } - if existingSlice, ok := existing.([]interface{}); ok { - if newSlice, ok := v.([]interface{}); ok { - merged := append(existingSlice, newSlice...) - seen := make(map[interface{}]bool) - unique := make([]interface{}, 0) - for _, item := range merged { - if !seen[item] { - seen[item] = true - unique = append(unique, item) - } - } - result[k] = unique - continue - } - } - } - result[k] = v - } - - return result -} - // DeleteByTenantID deletes all knowledge bases by tenant ID (hard delete) func (dao *KnowledgebaseDAO) DeleteByTenantID(ctx context.Context, db *gorm.DB, tenantID string) (int64, error) { result := db.WithContext(ctx).Unscoped().Where("tenant_id = ?", tenantID).Delete(&entity.Knowledgebase{}) diff --git a/internal/ingestion/component/chunker/table.go b/internal/ingestion/component/chunker/table.go index dc405675bb..986733f3c6 100644 --- a/internal/ingestion/component/chunker/table.go +++ b/internal/ingestion/component/chunker/table.go @@ -15,8 +15,8 @@ // // TableChunker emits one chunk per upstream table row. It is the faithful -// Go port of the Python `table` chunk method (rag/app/table.py), whose -// docstring states: "Every row in table will be treated as a chunk." +// Go port of the Python `table` chunk method, whose docstring states: +// "Every row in table will be treated as a chunk." // // Unlike TokenChunker (which token-shreds a row's text into multiple // pieces) or OneChunker (which merges many rows into a single chunk), @@ -189,7 +189,7 @@ func expandHTMLTableRows(item schema.ChunkDoc) []schema.ChunkDoc { return out } -// tableRowRecordText renders one row in rag/app/table.py's line format: +// tableRowRecordText renders one row in the Python table chunker's line format: // "- {field}: {value}" per non-empty cell, newline joined. A cell whose // column has no header name keeps its value alone rather than being dropped. func tableRowRecordText(names, cells []string) string { diff --git a/internal/ingestion/component/chunker/table_test.go b/internal/ingestion/component/chunker/table_test.go index 99e653c4da..36979b2a24 100644 --- a/internal/ingestion/component/chunker/table_test.go +++ b/internal/ingestion/component/chunker/table_test.go @@ -40,7 +40,7 @@ func tableChunksOf(t *testing.T, inputs map[string]any) []map[string]any { return chunks } -// TestTableChunker_OneChunkPerRow ports rag/app/table.py's contract: +// TestTableChunker_OneChunkPerRow ports the Python table chunker's contract: // "Every row in table will be treated as a chunk." Two upstream rows must // yield exactly two chunks, each preserving its own content + metadata. func TestTableChunker_OneChunkPerRow(t *testing.T) { @@ -150,7 +150,7 @@ func TestTableChunker_ExpandsHTMLRowsWithAlignedPositions(t *testing.T) { if chunks[0]["text"] != "- ID: A-1\n- Status: paid" { t.Errorf("chunk0 text = %v", chunks[0]["text"]) } - // The empty cell is skipped, exactly like rag/app/table.py:656. + // The empty cell is skipped, matching the Python table chunker's per-cell skip. if chunks[1]["text"] != "- ID: A-2" { t.Errorf("chunk1 text = %v", chunks[1]["text"]) } diff --git a/internal/service/document/document_test.go b/internal/service/document/document_test.go index 94ffd75255..452886b4aa 100644 --- a/internal/service/document/document_test.go +++ b/internal/service/document/document_test.go @@ -974,6 +974,55 @@ func TestSyncDocumentUpsertRemovesStagedBlobWhenInsertFails(t *testing.T) { } } +func TestSyncDocumentUpsertWithNilDatasetParserConfig(t *testing.T) { + ctx := t.Context() + db := setupServiceTestDB(t) + pushServiceDB(t, db) + + kb := entity.Knowledgebase{ + ID: "kb-sync-nil-config", TenantID: "tenant-1", Name: "Sync KB", ParserID: "naive", + } + if err := db.Create(&kb).Error; err != nil { + t.Fatalf("create dataset: %v", err) + } + if err := db.First(&kb, "id = ?", kb.ID).Error; err != nil { + t.Fatalf("load dataset: %v", err) + } + if kb.ParserConfig != nil { + t.Fatalf("dataset parser_config = %#v, want nil", kb.ParserConfig) + } + + factory := storage.GetStorageFactory() + originalStorage := factory.GetStorage() + factory.SetStorage(newFakeUploadStorage()) + t.Cleanup(func() { factory.SetStorage(originalStorage) }) + + result, err := testDocumentService(t).Upsert(ctx, service.DocumentUpsertInput{ + TaskContext: service.SyncTaskContext{ + Connector: entity.Connector{TenantID: "tenant-1"}, + Knowledgebase: kb, + }, + SourceType: "github", + DocumentID: "doc-sync-nil-config", + SourceDocument: syncerconnector.SourceDocument{ + SourceID: "source-1", SemanticIdentifier: "source", Extension: ".txt", Blob: []byte("content"), + }, + }) + if err != nil { + t.Fatalf("sync document: %v", err) + } + if result.Action != service.DocumentActionAdded { + t.Fatalf("sync action = %q, want added", result.Action) + } + var doc entity.Document + if err := db.First(&doc, "id = ?", result.DocID).Error; err != nil { + t.Fatalf("load synced document: %v", err) + } + if doc.ParserConfig == nil || len(doc.ParserConfig) != 0 { + t.Fatalf("document parser_config = %#v, want empty object", doc.ParserConfig) + } +} + func TestSyncDocumentUpsertRemovesStagedBlobWhenUpdateFails(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) db := setupServiceTestDB(t) diff --git a/internal/service/document/document_upload.go b/internal/service/document/document_upload.go index 829774f93c..0d56b0be29 100644 --- a/internal/service/document/document_upload.go +++ b/internal/service/document/document_upload.go @@ -327,6 +327,7 @@ func normalizeWebDocumentName(name, contentType string, blob []byte) string { // resolveDocumentParser); blob may be nil for the empty/virtual document. func (s *DocumentService) newDatasetDocument(kb *entity.Knowledgebase, tenantID, filename, location, filetype, parserID string, parserConfig entity.JSONMap, src string, size int64, blob []byte) *entity.Document { docID := utility.GenerateToken() + parserConfig = cloneParserConfigForDocument(parserConfig) status := "1" suffix := "" if i := strings.LastIndex(filename, "."); i >= 0 { @@ -358,6 +359,14 @@ func (s *DocumentService) newDatasetDocument(kb *entity.Knowledgebase, tenantID, return doc } +// cloneParserConfigForDocument copies a dataset's parser_config for a new +// document row. A configuration loaded from the database hands out live nested +// maps, so a component entry written for one document would otherwise be +// written for the dataset and every sibling document too. +func cloneParserConfigForDocument(config entity.JSONMap) entity.JSONMap { + return entity.JSONMap(common.DeepMergeMaps(config, nil)) +} + // docToRawMap serialises a freshly created Document into the raw key shape the // handler remaps (chunk_num→chunk_count, kb_id→dataset_id). func docToRawMap(doc *entity.Document) map[string]interface{} { diff --git a/internal/service/document/document_upload_test.go b/internal/service/document/document_upload_test.go index 66639f497a..0b7588db4c 100644 --- a/internal/service/document/document_upload_test.go +++ b/internal/service/document/document_upload_test.go @@ -1,6 +1,47 @@ package document -import "testing" +import ( + "testing" + + "ragflow/internal/entity" +) + +// cloneParserConfigForDocument must deep-copy the nested maps a database load +// hands out, so writing a component entry onto one document's config cannot +// leak into the dataset's config or a sibling document's config. +func TestCloneParserConfigForDocument(t *testing.T) { + if clone := cloneParserConfigForDocument(nil); clone == nil || len(clone) != 0 { + t.Fatalf("nil config must clone to an empty object, got %#v", clone) + } + + original := entity.JSONMap{ + "Parser:HipSignsRhyme": map[string]interface{}{ + "spreadsheet": map[string]interface{}{"output_format": "html"}, + }, + "topn": 10, + } + + clone := cloneParserConfigForDocument(original) + + nested, ok := clone["Parser:HipSignsRhyme"].(map[string]interface{}) + if !ok { + t.Fatalf("clone lost the component entry: %#v", clone["Parser:HipSignsRhyme"]) + } + ss, ok := nested["spreadsheet"].(map[string]interface{}) + if !ok { + t.Fatalf("clone lost the spreadsheet setup: %#v", nested["spreadsheet"]) + } + ss["output_format"] = "json" + clone["topn"] = 42 + + origNested := original["Parser:HipSignsRhyme"].(map[string]interface{})["spreadsheet"].(map[string]interface{}) + if origNested["output_format"] != "html" { + t.Fatalf("original nested map was mutated: %#v", origNested) + } + if original["topn"] != 10 { + t.Fatalf("original top-level map was aliased: topn=%#v", original["topn"]) + } +} func TestNormalizeWebDocumentName(t *testing.T) { pdfBlob := []byte("%PDF-1.4 fake") diff --git a/internal/service/document/file2document.go b/internal/service/document/file2document.go index 664e1ae942..0b2e97f640 100644 --- a/internal/service/document/file2document.go +++ b/internal/service/document/file2document.go @@ -224,7 +224,7 @@ func (s *File2DocumentService) convertFiles(ctx context.Context, fileIDs, kbIDs continue } - parserID, parserConfig := resolveDocumentParser(ctx, kb, docName, utility.FileType(file.Type), kb.ParserConfig) + parserID, parserConfig := resolveDocumentParser(ctx, kb, docName, utility.FileType(file.Type), cloneParserConfigForDocument(kb.ParserConfig)) suffix := strings.TrimPrefix(filepath.Ext(docName), ".") doc := &entity.Document{ ID: utility.GenerateUUID(), diff --git a/internal/service/document/sync_document.go b/internal/service/document/sync_document.go index 678cc4e186..08b3e2c4bc 100644 --- a/internal/service/document/sync_document.go +++ b/internal/service/document/sync_document.go @@ -114,7 +114,7 @@ func (s *DocumentService) insertSyncDocument(ctx context.Context, input service. } // create the 'doc' - parserID, parserConfig := resolveDocumentParser(ctx, kb, filename, utility.FileType(filetype), copyJSONMap(kb.ParserConfig)) + parserID, parserConfig := resolveDocumentParser(ctx, kb, filename, utility.FileType(filetype), kb.ParserConfig) doc := s.newDatasetDocument(kb, tenantID, filename, "", filetype, parserID, parserConfig, input.SourceType, int64(len(input.SourceDocument.Blob)), input.SourceDocument.Blob) doc.ID = input.DocumentID @@ -301,15 +301,6 @@ func syncDocumentStagedLocation(sourceType, docID, filename string) string { return fmt.Sprintf("sync/%s/.staged/%s/%s%s", sourceType, utility.GenerateToken(), docID, ext) } -// copyJSONMap returns a shallow copy of a JSON map. -func copyJSONMap(value entity.JSONMap) entity.JSONMap { - out := entity.JSONMap{} - for k, v := range value { - out[k] = v - } - return out -} - // errorsIsRecordNotFound reports whether an error is GORM's not-found error. func errorsIsRecordNotFound(err error) bool { return errors.Is(err, gorm.ErrRecordNotFound)