mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-11-11 05:40:52 +08:00
feat(infinity): implement SearchByRegexp for the agent grep tool (#21068)
This commit is contained in:
1 parent
26e6acc217
commit
54cb269d0d
8 files changed
+513
-57
No files matched your search
@@ -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": ""},
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
|
||||
@@ -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, `<tool_error tool="grep_chunks" severity="warn">`) ||
|
||||
!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"`) {
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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_<tenant>, 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_<tenant>_<datasetID>.
|
||||
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{}) {
|
||||
|
||||
@@ -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_<tenant> 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user