From 54cb269d0d0be85ca1d4bdbfb9e2be3fcceb67eb Mon Sep 17 00:00:00 2001 From: qinling0210 <88864212+qinling0210@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:15:53 +0800 Subject: [PATCH] feat(infinity): implement SearchByRegexp for the agent grep tool (#21068) --- conf/infinity_mapping.json | 2 +- internal/agent/runtime/retrieval.go | 2 +- internal/agentic_rag/grep_degrade_test.go | 10 +- internal/agentic_rag/grep_service.go | 47 +-- internal/engine/infinity/chunk.go | 352 ++++++++++++++++-- internal/engine/infinity/chunk_filter_test.go | 64 ++++ .../engine/infinity/chunk_transform_test.go | 89 +++++ internal/service/agentic_reference.go | 4 +- 8 files changed, 513 insertions(+), 57 deletions(-) diff --git a/conf/infinity_mapping.json b/conf/infinity_mapping.json index 56eb58c966..9264248dce 100644 --- a/conf/infinity_mapping.json +++ b/conf/infinity_mapping.json @@ -14,7 +14,7 @@ "important_kwd_empty_count": {"type": "integer", "default": 0}, "important_keywords": {"type": "varchar", "default": "", "analyzer": ["rag-coarse", "rag-fine"], "comment": "important_kwd, important_tks"}, "questions": {"type": "varchar", "default": "", "analyzer": ["rag-coarse", "rag-fine"], "comment": "question_kwd, question_tks"}, - "content": {"type": "varchar", "default": "", "analyzer": ["rag-coarse", "rag-fine"], "comment": "content_with_weight, content_ltks, content_sm_ltks"}, + "content": {"type": "varchar", "default": "", "analyzer": ["rag-coarse", "rag-fine", "sparsegram-3-12-fold"], "comment": "content_with_weight, content_ltks, content_sm_ltks"}, "authors": {"type": "varchar", "default": "", "analyzer": ["rag-coarse", "rag-fine"], "comment": "authors_tks, authors_sm_tks"}, "page_num_int": {"type": "varchar", "default": ""}, "top_int": {"type": "varchar", "default": ""}, diff --git a/internal/agent/runtime/retrieval.go b/internal/agent/runtime/retrieval.go index b7cf23c23d..ebf87355cf 100644 --- a/internal/agent/runtime/retrieval.go +++ b/internal/agent/runtime/retrieval.go @@ -201,7 +201,7 @@ var ErrBm25ServiceMissing = errors.New( ) // ErrRegexpNotSupported is returned when the underlying doc engine does not -// implement regex matching on chunk content (e.g. Infinity). +// implement regex matching on chunk content (no SearchByRegexp method). var ErrRegexpNotSupported = errors.New( "grep_chunks: regex matching is not supported by this document engine", ) diff --git a/internal/agentic_rag/grep_degrade_test.go b/internal/agentic_rag/grep_degrade_test.go index 5030401660..55001fe58f 100644 --- a/internal/agentic_rag/grep_degrade_test.go +++ b/internal/agentic_rag/grep_degrade_test.go @@ -156,8 +156,8 @@ func TestGrepChunksSplitsExplodingAlternation(t *testing.T) { if err != nil { t.Fatalf("split must recover, got error: %v", err) } - if !strings.Contains(out, toolErrorMarker) || !strings.Contains(out, "ES regexp query was rejected") { - t.Errorf("an exact split must explain the ES rejection, got %.200q", out) + if !strings.Contains(out, toolErrorMarker) || !strings.Contains(out, "regexp query was rejected by the engine") { + t.Errorf("an exact split must explain the engine rejection, got %.200q", out) } if !strings.Contains(out, `chunk_id="c1"`) { t.Errorf("branch hits must ship, got %.200q", out) @@ -190,8 +190,8 @@ func TestGrepChunksSanitizesLuceneUnsafePattern(t *testing.T) { if err != nil { t.Fatalf("sanitized retry must recover, got error: %v", err) } - if !strings.Contains(out, toolErrorMarker) || !strings.Contains(out, "ES regexp query was rejected") { - t.Errorf("a sanitized fallback must explain the ES rejection, got %.200q", out) + if !strings.Contains(out, toolErrorMarker) || !strings.Contains(out, "regexp query was rejected by the engine") { + t.Errorf("a sanitized fallback must explain the engine rejection, got %.200q", out) } if got := svc.calls[len(svc.calls)-1]; strings.Contains(got, `\b`) { t.Errorf("retry pattern must be sanitized, got %q", got) @@ -225,7 +225,7 @@ func TestGrepChunksDegradesToLexicalPrefilter(t *testing.T) { } if !strings.Contains(out, ``) || !strings.Contains(out, "APPROXIMATE") || - !strings.Contains(out, "ES regexp query was rejected") { + !strings.Contains(out, "regexp query was rejected by the engine") { t.Errorf("expected a warn notice about approximate coverage, got %.300q", out) } if !strings.Contains(out, `chunk_id="c1"`) { diff --git a/internal/agentic_rag/grep_service.go b/internal/agentic_rag/grep_service.go index c97327f901..6f80336e88 100644 --- a/internal/agentic_rag/grep_service.go +++ b/internal/agentic_rag/grep_service.go @@ -35,9 +35,10 @@ import ( "go.uber.org/zap" ) -// Elasticsearch it pushes the regex down to a native `regexp` query; on engines -// without native regex support it falls back to a broad recall + in-memory RE2 -// filter. It is stateless and safe to share across goroutines. +// Elasticsearch it pushes the regex down to a native `regexp` query, and on +// Infinity to a native `regex()` filter; on engines without native regex +// support it falls back to a broad recall + in-memory RE2 filter. It is +// stateless and safe to share across goroutines. type GrepAdapter struct { docEngine engine.DocEngine } @@ -48,7 +49,7 @@ func NewGrepAdapter(docEngine engine.DocEngine) *GrepAdapter { } // regexpSearchable is the narrow engine capability for native regex search. -// *elasticsearch.Engine implements it; other engines do not. +// *elasticsearch.Engine and *infinity.Engine implement it; other engines do not. type regexpSearchable interface { SearchByRegexp(ctx context.Context, req *enginetypes.RegexpSearchRequest) (*enginetypes.SearchResult, error) } @@ -56,9 +57,10 @@ type regexpSearchable interface { // ListByDocIDs reads the full original chunks belonging to the given document // ids, skipping graph (relation/entity/location) chunks so callers get the // actual document text. It backs the list_chunks deep-read runtime. -// Because content_with_weight is now a searchable keyword field, it reads the -// docs' chunks directly via the regexp pushdown (match-all "." scoped by -// doc_id), no longer needing a content_ltks-based general recall. +// Because content_with_weight is now a searchable keyword field (content with a +// sparsegram index on Infinity), it reads the docs' chunks directly via the +// regexp pushdown (match-all "." scoped by doc_id), no longer needing a +// content_ltks-based general recall. func (g *GrepAdapter) ListByDocIDs(ctx context.Context, req runtime.GrepRequest) ([]runtime.RetrievalChunk, error) { if g == nil || g.docEngine == nil { return nil, runtime.ErrGrepServiceMissing @@ -78,8 +80,8 @@ func (g *GrepAdapter) ListByDocIDs(ctx context.Context, req runtime.GrepRequest) } // Match-all on the (now keyword) content_with_weight, scoped to the docs, // restricted to ordinary text chunks (available_int=1, no compile_kwd). When - // the caller requests a sort (e.g. reading order), push it down so ES returns - // the deterministic first page. + // the caller requests a sort (e.g. reading order), push it down so the engine + // returns the deterministic first page. res, err := se.SearchByRegexp(ectx, &enginetypes.RegexpSearchRequest{ TenantID: req.TenantID, KbIDs: req.DatasetIDs, @@ -281,10 +283,11 @@ func (g *GrepAdapter) Grep(ctx context.Context, req runtime.GrepRequest) ([]runt ectx, ecancel := engineCallContext(ctx) defer ecancel() - // Path 1: native regex pushdown (Elasticsearch). With content_with_weight - // mapped as a searchable keyword field, ES regexp matching is the sole path; - // there is intentionally no in-memory RE2 fallback (a pushdown failure is a - // real error the caller should surface, not silently "no matches"). + // Path 1: native regex pushdown (Elasticsearch, Infinity). With + // content_with_weight mapped as a searchable keyword field, native regexp + // matching is the sole path; there is intentionally no in-memory RE2 + // fallback (a pushdown failure is a real error the caller should surface, + // not silently "no matches"). if se, ok := g.docEngine.(regexpSearchable); ok { res, err := se.SearchByRegexp(ectx, &enginetypes.RegexpSearchRequest{ TenantID: req.TenantID, @@ -300,7 +303,7 @@ func (g *GrepAdapter) Grep(ctx context.Context, req runtime.GrepRequest) ([]runt return translateGrepChunks(res.Chunks), nil } - // Non-ES engines do not implement regex matching on chunk content; surface + // Engines without native regex support cannot match on chunk content; surface // an explicit error rather than returning (empty) regex "matches". return nil, runtime.ErrRegexpNotSupported } @@ -399,9 +402,9 @@ func degradeGrep( return nil, "", pushErr } -// pushdownWarning keeps the original ES rejection visible whenever a fallback -// produces results. The model must distinguish a real regexp result from a -// result obtained after the engine rejected the requested pattern. +// pushdownWarning keeps the original pushdown rejection visible whenever a +// fallback produces results. The model must distinguish a real regexp result +// from a result obtained after the engine rejected the requested pattern. func pushdownWarning(pushErr error, detail string) string { if detail == "" { detail = "the pattern was split into smaller regexp branches and the returned matches were verified with the original Go regexp" @@ -413,7 +416,7 @@ func pushdownWarning(pushErr error, detail string) string { if r := []rune(summary); len(r) > 400 { summary = string(r[:400]) + "…" } - return fmt.Sprintf("ES regexp query was rejected: %s. %s", summary, detail) + return fmt.Sprintf("regexp query was rejected by the engine: %s. %s", summary, detail) } // grepBySplitAlternation pushes each top-level alternation branch as its own @@ -637,7 +640,7 @@ func engineCallContext(ctx context.Context) (context.Context, context.CancelFunc return context.WithCancel(context.WithoutCancel(ctx)) } -// translateGrepChunks converts ES regexp-search result maps into RetrievalChunk. +// translateGrepChunks converts regexp-search result maps into RetrievalChunk. // regexpPushdownHint turns an engine-side regexp rejection into something the // caller can act on. // @@ -656,8 +659,8 @@ func engineCallContext(ctx context.Context) (context.Context, context.CancelFunc func regexpPushdownHint(pattern string) string { if m := boundedRepetitionRe.FindString(pattern); m != "" { return " Retry WITHOUT the bounded repetition `" + m + "`: the engine compiles a pattern " + - "into an automaton with a fixed state budget (it already wraps every pattern in " + - "`.*(...).*`), and repetition over CJK text exhausts it. Join the terms with `.*` " + + "into a bounded automaton, and repetition over CJK text can exhaust it (Elasticsearch " + + "wraps every pattern in `.*(...).*`, which costs more states). Join the terms with `.*` " + "(e.g. `关羽.*杀`) or list them as alternatives (`关羽.*杀|关公.*斩`)." } if backreferenceRe.MatchString(pattern) { @@ -669,7 +672,7 @@ func regexpPushdownHint(pattern string) string { "express the constraint as plain terms." } return " Retry with a CHEAPER pattern: literal terms joined by `.*`, at most 2-3 alternation " + - "branches, no bounded repetition — this pattern exceeded the engine's automaton state budget." + "branches, no bounded repetition — this pattern was too expensive for the engine to compile." } var ( diff --git a/internal/engine/infinity/chunk.go b/internal/engine/infinity/chunk.go index e08c460a0a..a9b82530b2 100644 --- a/internal/engine/infinity/chunk.go +++ b/internal/engine/infinity/chunk.go @@ -1417,32 +1417,7 @@ func (e *Engine) Search(ctx context.Context, req *types.SearchRequest) (*types.S } // Convert DataFrame to chunks format (column-oriented to row-oriented) - searchChunks := make([]map[string]interface{}, 0) - for colName, colData := range df.ColumnData { - for i, val := range colData { - // A null cell must not materialize a row of its own. Infinity - // can return a single all-null cell for a projection it - // matched nothing for (observed with a full-text filter plus a - // json-list projection: source_chunk_ids came back as a nil - // slice while the other columns were empty), and fabricating - // one empty row out of it leaks phantom documents to callers — - // GetWikiGraph then read an empty slug_kwd off it and issued - // filter_fulltext('from_kwd', ''), which Infinity rejects. - if isEmptyEngineCell(val) { - continue - } - for len(searchChunks) <= i { - searchChunks = append(searchChunks, make(map[string]interface{})) - } - searchChunks[i][colName] = val - } - } - - // Infinity returns a Json/Array column as ONE cell holding one - // length-prefixed segment per row, so the loop above parks the whole - // column on row 0. Redistribute it before applyFieldMappings renames - // anything. - realignJSONColumnCells(searchChunks, df.ColumnData) + searchChunks := chunksFromColumnData(df.ColumnData) // Apply field name mapping and row_id handling // Skill index uses different schema @@ -1536,6 +1511,309 @@ func shouldDefaultAvailableFilter(req *types.SearchRequest, isSkillIndex bool) b return !isSkillIndex && !req.IncludeUnavailable } +const ( + // regexpSearchDefaultLimit caps a regexp search, matching the Elasticsearch + // engine so grep_chunks returns the same page on either backend. + regexpSearchDefaultLimit = 30 + // regexpSearchReturnAllLimit is the ceiling for list_chunks' deep read. + regexpSearchReturnAllLimit = 10000 +) + +// SearchByRegexp is a regex-match-only search over chunk content for the agent's +// grep_chunks/list_chunks tools, mirroring the Elasticsearch engine. The pattern goes +// down as regex(content, ...), which the content column's sparse gram index narrows by +// the pattern's mandatory literals; Infinity evaluates it with RE2, so the pattern is +// used verbatim (no ".*(...).*" wrapper) and "(?i)" carries case folding. +func (e *Engine) SearchByRegexp(ctx context.Context, req *types.RegexpSearchRequest) (*types.SearchResult, error) { + if req == nil { + return nil, fmt.Errorf("regexp search request cannot be nil") + } + if strings.TrimSpace(req.Pattern) == "" { + return nil, fmt.Errorf("regexp pattern cannot be empty") + } + tenantID := strings.TrimSpace(req.TenantID) + if tenantID == "" { + return nil, fmt.Errorf("tenant id cannot be empty") + } + // One index per tenant; a list would fan out to unrelated tables. + if strings.Contains(tenantID, ",") { + return nil, fmt.Errorf("regexp search supports exactly one tenant") + } + // Chunk tables are per dataset: an unscoped request would read the bare + // ragflow_, which holds no chunks, and report an empty result. + scoped := false + for _, kbID := range req.KbIDs { + if strings.TrimSpace(kbID) != "" { + scoped = true + break + } + } + if !scoped { + return nil, fmt.Errorf("regexp search requires at least one dataset id") + } + + maxResults := regexpSearchDefaultLimit + if req.ReturnAll { + maxResults = regexpSearchReturnAllLimit + } + + db, release, err := e.client.checkoutDatabase(ctx, "chunk.go") + if err != nil { + return nil, fmt.Errorf("failed to get database: %w", err) + } + defer release() + + // Chunk tables are per dataset: ragflow__. + baseName := fmt.Sprintf("ragflow_%s", tenantID) + tableNames := make([]string, 0, len(req.KbIDs)) + for _, kbID := range req.KbIDs { + if kbID == "" { + tableNames = append(tableNames, baseName) + } else { + tableNames = append(tableNames, fmt.Sprintf("%s_%s", baseName, kbID)) + } + } + + outputColumns := []string{ + "id", "doc_id", "kb_id", "content_ltks", "content_with_weight", + "title_tks", "docnm_kwd", "img_id", "available_int", "important_kwd", + "position_int", "page_num_int", "top_int", "chunk_order_int", + "create_timestamp_flt", "knowledge_graph_kwd", "question_kwd", "question_tks", + "doc_type_kwd", "mom_id", "tag_kwd", "pagerank_fea", "tag_feas", + } + if len(req.SelectFields) > 0 { + outputColumns = append([]string(nil), req.SelectFields...) + } + // The merge at the end restores the order Go-side, so the sort keys must be + // projected even when the caller did not ask for them. + outputColumns = unionSortColumns(outputColumns, req.Sort) + // No MATCH is applied, so Infinity accepts no score column to project. + outputColumns = stripScorePseudoColumns(outputColumns) + outputColumns = convertSelectFields(outputColumns, false, false) + + // Same filter for every dataset table; kb_id is implied by the table name. + filterStr := buildRegexpFilter(req.Filter, req.Pattern) + + var allResults []map[string]interface{} + totalHits := int64(0) + // A missing table is normal (created on first write); other failures surface + // only if no table answered. + var queryErr error + + for _, tableName := range tableNames { + tbl, err := db.GetTable(tableName) + if err != nil { + // A dataset table that does not exist yet is normal (tables are created on + // first write); any other open failure is real and must reach the caller. + errMsg := strings.ToLower(err.Error()) + tableMissing := strings.Contains(errMsg, "not found") || strings.Contains(errMsg, "doesn't exist") + if isConnectionLevelError(err) { + e.client.dropConnectionFor(db, "SearchByRegexp") + } + if !tableMissing && queryErr == nil { + queryErr = fmt.Errorf("open table %s: %w", tableName, err) + } + continue + } + + table := tbl.Output(outputColumns).Filter(filterStr) + + if req.Sort != nil && len(req.Sort.Fields) > 0 { + var sortFields [][2]interface{} + for _, orderField := range req.Sort.Fields { + sortType := infinity.SortTypeAsc + if orderField.Type == types.SortDesc { + sortType = infinity.SortTypeDesc + } + sortFields = append(sortFields, [2]interface{}{orderField.Field, sortType}) + } + table = table.Sort(sortFields) + } + + table = table.Limit(maxResults) + table = table.Option(map[string]interface{}{"total_hits_count": true}) + + df, err := table.ToDataFrame() + if err != nil { + common.Warn("Infinity regexp query failed", + zap.String("tableName", tableName), + zap.String("filter", filterStr), + zap.Int("limit", maxResults), + zap.Error(err)) + if isConnectionLevelError(err) { + e.client.dropConnectionFor(db, "SearchByRegexp") + } + if queryErr == nil { + queryErr = fmt.Errorf("query table %s: %w", tableName, err) + } + continue + } + + chunks := chunksFromColumnData(df.ColumnData) + applyFieldMappings(chunks) + + var tableTotal int64 + if df.ExtraInfo != "" { + var extraResult map[string]interface{} + if err := json.Unmarshal([]byte(df.ExtraInfo), &extraResult); err == nil { + if count, ok := extraResult["total_hits_count"].(float64); ok { + tableTotal = int64(count) + } + } + } + + common.Debug("Infinity regexp search", + zap.String("table", tableName), + zap.String("filter", filterStr), + zap.Int("rows", len(chunks)), + zap.Int64("total", tableTotal)) + + allResults = append(allResults, chunks...) + totalHits += tableTotal + } + + // limit() applies per table, so merge the tables' pages into the requested order + // and cap once, as the Elasticsearch engine does after its single-index query. + // The cap must follow the sort, and Total stays the pre-cap count. + sortChunksByFields(allResults, req.Sort) + if len(allResults) > maxResults { + allResults = allResults[:maxResults] + } + + // All tables failed: surface the error rather than an empty result. + if len(allResults) == 0 && totalHits == 0 && queryErr != nil { + return nil, fmt.Errorf("infinity regexp search failed on all tables: %w", queryErr) + } + + return &types.SearchResult{ + Chunks: allResults, + Total: totalHits, + }, nil +} + +// buildRegexpFilter renders the regexp search filter: the caller's scope (minus +// kb_id, already implied by the per-dataset table name) AND regex(content, ...) +// with the case-insensitive "(?i)" the grep contract requires. +func buildRegexpFilter(filter map[string]interface{}, pattern string) string { + filterCopy := map[string]interface{}{} + for k, v := range filter { + if k == "kb_id" { + continue + } + filterCopy[k] = v + } + parts := make([]string, 0, 2) + if len(filterCopy) > 0 { + if condition := equivalentConditionToStr(filterCopy, nil); condition != "" { + parts = append(parts, condition) + } + } + parts = append(parts, infinity.RegexFilter("content", "(?i)"+pattern)) + return strings.Join(parts, " AND ") +} + +// unionSortColumns appends the requested sort keys to a projection, preserving the +// caller's column order: the engine orders the merged rows itself. +func unionSortColumns(columns []string, sort *types.OrderByExpr) []string { + if sort == nil || len(sort.Fields) == 0 { + return columns + } + seen := make(map[string]struct{}, len(columns)+len(sort.Fields)) + for _, column := range columns { + seen[column] = struct{}{} + } + for _, field := range sort.Fields { + if field.Field == "" { + continue + } + if _, ok := seen[field.Field]; ok { + continue + } + seen[field.Field] = struct{}{} + columns = append(columns, field.Field) + } + return columns +} + +// sortChunksByFields orders merged rows by the requested fields, ascending unless a +// field asks for descending, mirroring the Elasticsearch engine's sortByFields: a +// missing field sorts first, numbers numerically, strings lexicographically. +// page_num_int arrives hex-decoded as an int slice, compared on its leading element. +func sortChunksByFields(chunks []map[string]interface{}, expr *types.OrderByExpr) { + if expr == nil || len(expr.Fields) == 0 { + return + } + sort.SliceStable(chunks, func(i, j int) bool { + for _, field := range expr.Fields { + vi, oki := chunks[i][field.Field] + vj, okj := chunks[j][field.Field] + if !oki || !okj { + if !oki && !okj { + continue + } + // Missing field sorts before a present one. + return !oki + } + less, equal := compareOrderValues(vi, vj) + if equal { + continue + } + if field.Type == types.SortDesc { + return !less + } + return less + } + return false + }) +} + +// compareOrderValues compares two raw field values: numbers numerically, strings +// lexicographically, anything else by text. An unset key (the empty column default, as +// a page number that was never set) ranks before a set one in either argument order, so +// the pair cannot order differently by comparison direction; equality is reported +// separately, so only a real tie falls through to the next sort field. +func compareOrderValues(a, b interface{}) (less, equal bool) { + fa, okA := numericOrderValue(a) + fb, okB := numericOrderValue(b) + if okA != okB { + return !okA, false + } + if okA { + return fa < fb, fa == fb + } + sa, okSA := a.(string) + sb, okSB := b.(string) + if okSA && okSB { + return sa < sb, sa == sb + } + ta, tb := fmt.Sprintf("%v", a), fmt.Sprintf("%v", b) + return ta < tb, ta == tb +} + +// numericOrderValue reads a scalar order key, unwrapping the int slice +// applyFieldMappings leaves behind for hex-encoded columns. +func numericOrderValue(v interface{}) (float64, bool) { + switch n := v.(type) { + case int: + return float64(n), true + case int64: + return float64(n), true + case float64: + return n, true + case float32: + return float64(n), true + case []int: + if len(n) > 0 { + return float64(n[0]), true + } + case []interface{}: + if len(n) > 0 { + return numericOrderValue(n[0]) + } + } + return 0, false +} + // GetChunk gets a chunk by ID func (e *Engine) GetChunk(ctx context.Context, tableName, chunkID string, datasetIDs []string) (interface{}, error) { if e.client == nil || e.client.pool == nil { @@ -1702,6 +1980,28 @@ func splitKeywordList(field, value string) ([]interface{}, bool) { return list, true } +// chunksFromColumnData converts an Infinity column-oriented DataFrame into the +// row-oriented chunk maps the engine returns. A null cell must not become a row +// of its own (Infinity returns one all-null cell for a projection that matched +// nothing, and a phantom row leaks to callers); a Json/Array column arrives +// parked on row 0, so realignJSONColumnCells spreads it before applyFieldMappings. +func chunksFromColumnData(columnData map[string][]interface{}) []map[string]interface{} { + rows := make([]map[string]interface{}, 0) + for colName, colData := range columnData { + for i, val := range colData { + if isEmptyEngineCell(val) { + continue + } + for len(rows) <= i { + rows = append(rows, make(map[string]interface{})) + } + rows[i][colName] = val + } + } + realignJSONColumnCells(rows, columnData) + return rows +} + // applyFieldMappings applies field mappings to chunks (side-effect only). // Used by Search() to mutate chunks with derived fields before returning. func applyFieldMappings(chunks []map[string]interface{}) { diff --git a/internal/engine/infinity/chunk_filter_test.go b/internal/engine/infinity/chunk_filter_test.go index b65e3fd52b..bf3c98742d 100644 --- a/internal/engine/infinity/chunk_filter_test.go +++ b/internal/engine/infinity/chunk_filter_test.go @@ -1,6 +1,7 @@ package infinity import ( + "context" "fmt" "strings" "testing" @@ -111,3 +112,66 @@ func TestListFilterConditionKeepsDocumentScope(t *testing.T) { }) } } + +// buildRegexpFilter must keep the caller's document scope, drop kb_id (the +// per-dataset table name already scopes the query), and append the regex() +// predicate with the case-insensitive inline flag on the content column. +func TestBuildRegexpFilterScopesAndMatchesContent(t *testing.T) { + got := buildRegexpFilter(map[string]interface{}{ + "available_int": 1, + "doc_id": []string{"d1", "d2"}, + "must_not": map[string]interface{}{"exists": "compile_kwd"}, + "kb_id": "kb-must-not-appear", + }, `foo.*bar`) + + for _, want := range []string{ + "regex(content, '(?i)foo.*bar')", + "available_int=1", + "doc_id IN ('d1', 'd2')", + "NOT (compile_kwd!='')", + } { + if !strings.Contains(got, want) { + t.Errorf("buildRegexpFilter() = %q, want it to contain %q", got, want) + } + } + if strings.Contains(got, "kb_id") { + t.Errorf("buildRegexpFilter() = %q, must not carry kb_id into the per-dataset table filter", got) + } +} + +// A pattern quoted in SQL-ish filter syntax must double its single quotes, or +// the SDK's expression parser would read a truncated literal. +func TestBuildRegexpFilterEscapesSingleQuotes(t *testing.T) { + got := buildRegexpFilter(nil, `it's a match`) + if want := `regex(content, '(?i)it''s a match')`; got != want { + t.Errorf("buildRegexpFilter() = %q, want %q", got, want) + } +} + +// A regexp search from the agent's grep tools carries only a pattern and no +// scope filter; the predicate itself must still be well-formed. +func TestBuildRegexpFilterWithoutScope(t *testing.T) { + if got, want := buildRegexpFilter(nil, `.*`), `regex(content, '(?i).*')`; got != want { + t.Errorf("buildRegexpFilter() = %q, want %q", got, want) + } +} + +// An unscoped regexp search must fail rather than scan the bare ragflow_ table: +// chunk tables are per dataset, so an empty KbIDs would come back as an error-free empty +// result. The check runs before the client is touched, so a zero Engine pins it. +func TestSearchByRegexpRejectsEmptyScope(t *testing.T) { + eng := &Engine{} + for _, kbIDs := range [][]string{nil, {}, {""}, {"", " "}} { + _, err := eng.SearchByRegexp(context.Background(), &types.RegexpSearchRequest{ + TenantID: "tenant1", + KbIDs: kbIDs, + Pattern: "foo.*bar", + }) + if err == nil { + t.Fatalf("SearchByRegexp(KbIDs=%v) must reject an empty dataset scope", kbIDs) + } + if !strings.Contains(err.Error(), "at least one dataset id") { + t.Errorf("SearchByRegexp(KbIDs=%v) error = %v, want it to name the missing dataset scope", kbIDs, err) + } + } +} diff --git a/internal/engine/infinity/chunk_transform_test.go b/internal/engine/infinity/chunk_transform_test.go index 7495fb1cd1..7a689d62ee 100644 --- a/internal/engine/infinity/chunk_transform_test.go +++ b/internal/engine/infinity/chunk_transform_test.go @@ -23,6 +23,7 @@ import ( "strings" "testing" + "ragflow/internal/engine/types" "ragflow/internal/utility" infinitysdk "github.com/infiniflow/infinity-go-sdk" @@ -953,3 +954,91 @@ func TestDedupeByChunkIDKeepsRowsWithoutID(t *testing.T) { t.Fatalf("rows = %d, want 3 (only the duplicate id collapses): %#v", len(got), got) } } + +// chunksFromColumnData must turn the column-oriented DataFrame into rows and +// must not fabricate a row out of an all-null column. +func TestChunksFromColumnDataConvertsAndSkipsNullCells(t *testing.T) { + rows := chunksFromColumnData(map[string][]interface{}{ + "id": {"a", "b"}, + "content": {"alpha", "beta"}, + "junk": {nil, nil}, + }) + if len(rows) != 2 { + t.Fatalf("chunksFromColumnData() returned %d rows, want 2", len(rows)) + } + if rows[0]["id"] != "a" || rows[0]["content"] != "alpha" { + t.Errorf("row 0 = %v, want id=a content=alpha", rows[0]) + } + if rows[1]["id"] != "b" || rows[1]["content"] != "beta" { + t.Errorf("row 1 = %v, want id=b content=beta", rows[1]) + } + if _, ok := rows[0]["junk"]; ok { + t.Errorf("row 0 = %v, a null column must not contribute a cell", rows[0]) + } +} + +// An entirely null projection (an empty match) must yield no rows rather than +// one phantom empty row. +func TestChunksFromColumnDataAllNullIsEmpty(t *testing.T) { + rows := chunksFromColumnData(map[string][]interface{}{"content": {nil}}) + if len(rows) != 0 { + t.Fatalf("chunksFromColumnData() returned %d rows, want 0", len(rows)) + } +} + +// TestSortChunksByFieldsRestoresReadingOrder covers the cross-table merge: a table's +// rows arrive in that table's order, so the requested order has to be re-applied across +// the merged set, with page_num_int compared numerically (it arrives as a hex-decoded +// int slice). +func TestSortChunksByFieldsRestoresReadingOrder(t *testing.T) { + rows := []map[string]interface{}{ + {"id": "b1", "doc_id": "doc-b", "page_num_int": []int{10}, "chunk_order_int": 0}, + {"id": "b2", "doc_id": "doc-b", "page_num_int": []int{3}, "chunk_order_int": 5}, + {"id": "a1", "doc_id": "doc-a", "page_num_int": []int{2}, "chunk_order_int": 9}, + {"id": "a2", "doc_id": "doc-a", "page_num_int": []int{2}, "chunk_order_int": 1}, + } + sortChunksByFields(rows, &types.OrderByExpr{Fields: []types.OrderByField{ + {Field: "doc_id"}, + {Field: "page_num_int"}, + {Field: "chunk_order_int"}, + }}) + + got := make([]string, 0, len(rows)) + for _, row := range rows { + got = append(got, row["id"].(string)) + } + // Document first, then the page within it (numerically), then the chunk order + // within that page. + want := []string{"a2", "a1", "b2", "b1"} + if !reflect.DeepEqual(got, want) { + t.Errorf("merged reading order = %v, want %v", got, want) + } +} + +// TestSortChunksByFieldsRanksUnsetKeyFirst pins the mixed-key case: a chunk with no page +// number arrives as the empty column default, a chunk that has one as a hex-decoded int +// slice. An unset key ranks first, and both argument orders must agree on that. +func TestSortChunksByFieldsRanksUnsetKeyFirst(t *testing.T) { + order := &types.OrderByExpr{Fields: []types.OrderByField{ + {Field: "doc_id"}, + {Field: "page_num_int"}, + }} + sorted := func(rows []map[string]interface{}) []string { + sortChunksByFields(rows, order) + ids := make([]string, 0, len(rows)) + for _, row := range rows { + ids = append(ids, row["id"].(string)) + } + return ids + } + withPage := map[string]interface{}{"id": "page-5", "doc_id": "doc-1", "page_num_int": []int{5}} + unset := map[string]interface{}{"id": "page-unset", "doc_id": "doc-1", "page_num_int": ""} + + want := []string{"page-unset", "page-5"} + if got := sorted([]map[string]interface{}{withPage, unset}); !reflect.DeepEqual(got, want) { + t.Errorf("sort(page-first input) = %v, want %v", got, want) + } + if got := sorted([]map[string]interface{}{unset, withPage}); !reflect.DeepEqual(got, want) { + t.Errorf("sort(unset-first input) = %v, want %v", got, want) + } +} diff --git a/internal/service/agentic_reference.go b/internal/service/agentic_reference.go index e34f88d055..8494f2e569 100644 --- a/internal/service/agentic_reference.go +++ b/internal/service/agentic_reference.go @@ -149,7 +149,7 @@ func (s *ChatPipelineService) buildAgenticReference(ctx context.Context, tenantI // fetchChunksByIDs loads the cited chunks from the tenant's index, keeping the // engine row shape (chunk_id / content_with_weight / docnm_kwd / doc_id / -// kb_id / img_id / positions) so chunksFormat can normalize it. The index is +// kb_id / img_id / position_int) so chunksFormat can normalize it. The index is // the chat's OWNING tenant's - the same scoping the agent's retrieval tools // use. func fetchChunksByIDs(ctx context.Context, tenantID string, datasetIDs, ids []string) []map[string]interface{} { @@ -161,7 +161,7 @@ func fetchChunksByIDs(ctx context.Context, tenantID string, datasetIDs, ids []st IndexNames: []string{fmt.Sprintf("ragflow_%s", tenantID)}, KbIDs: datasetIDs, Filter: map[string]interface{}{"id": ids}, - SelectFields: []string{"content_with_weight", "docnm_kwd", "doc_id", "kb_id", "img_id", "positions"}, + SelectFields: []string{"content_with_weight", "docnm_kwd", "doc_id", "kb_id", "img_id", "position_int"}, Limit: len(ids), } res, err := de.Search(ctx, req)