mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-11-10 21:32:43 +08:00
fix(document): clone parser_config to break dataset aliasing; drop dead parser-config helpers (#20381)
This commit is contained in:
1 parent
f03ee77821
commit
83b8e2ca24
8 files changed
+107
-85
No files matched your search
@@ -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{})
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"])
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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{} {
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user