zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-09 04:44:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(opensearch): repair document-metadata path broken by #14577 (#15393)

Browse Source 
### What problem does this PR solve?

Document metadata is completely broken on the OpenSearch backend
(`DOC_ENGINE=opensearch`). Both failures were introduced by #14577,
which added
a doc-metadata dispatch surface but only validated it against
Elasticsearch.

**1. Index creation rejected (`mapper_parsing_exception`).**
`OSConnection.create_doc_meta_idx` feeds `conf/doc_meta_es_mapping.json`
verbatim to OpenSearch. That file declares a top-level `"dynamic":
"runtime"`.
Runtime fields are Elasticsearch-only; OpenSearch cannot parse the
value:

mapper_parsing_exception: Could not convert [dynamic.dynamic] to boolean
(400)

**2. `search()` signature mismatch (`TypeError`).**
`DocMetadataService` (added by #14577) calls `docStoreConn.search(...)`
with
snake_case kwargs (`select_fields=`, `index_names=`,
`knowledgebase_ids=`, …),
matching `ESConnection.search`. But `OSConnection.search` still uses
camelCase
parameters (`selectFields`, `indexNames`, `knowledgebaseIds`, …):

TypeError: OSConnection.search() got an unexpected keyword argument
'select_fields'

The UI then shows "0 fields" for every document on OpenSearch.

### Fix

1. In `OSConnection.create_doc_meta_idx`, normalize a top-level
`"dynamic": "runtime"` to `True` **for the OpenSearch request only**.
The
shared mapping file is left untouched, so the Elasticsearch backend
keeps its
runtime-field behavior. Dynamic field discovery is preserved on
OpenSearch.
2. Rename the `OSConnection.search()` parameters (and their in-method
local
uses) from camelCase to snake_case so they match `ESConnection.search()`
and
the `DocMetadataService` call sites. The change is confined to
`search()`;
`get/insert/update/delete` keep their existing positional signatures
(they
   are called positionally from `rag/nlp/search.py`).

### Type of change
- [x] Bug Fix (non-breaking change which fixes an issue)

### Affected backends
OpenSearch only. Elasticsearch, Infinity and OceanBase are untouched.

### How to reproduce
1. `DOC_ENGINE=opensearch`, restart the stack.
2. Upload/parse a document, then open the dataset's document list / set
metadata.
- Before: index creation 400s (`Could not convert [dynamic.dynamic]`),
and/or
     `TypeError ... 'select_fields'`; document metadata shows 0 fields.

### Risk & backward compatibility
- ES default deployment: no change. `doc_meta_es_mapping.json` is not
modified,
  so ES still receives `"dynamic": "runtime"`.
- `search()` rename is internal; the only kwarg caller
(`DocMetadataService`)
  already uses the snake_case names this PR aligns to.

### Test plan
- [ ] `DOC_ENGINE=opensearch`: per-tenant `ragflow_doc_meta_*` index is
created
(no `mapper_parsing_exception`); document metadata reads/writes work.
- [ ] `DOC_ENGINE=elasticsearch` regression: doc-meta index still
created with
      runtime mapping; metadata unchanged.
This commit is contained in:
Rintaro
2026-05-29 22:49:36 +09:00
committed by GitHub
parent 3dfc16973c
commit 11af34a895
1 changed files with 37 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
rag/utils/opensearch_conn.py
View File

@@ -145,10 +145,23 @@ class OSConnection(DocStoreConnection):
with open(fp_mapping, "r") as f:
doc_meta_mapping = json.load(f)
mappings = doc_meta_mapping["mappings"]
# `conf/doc_meta_es_mapping.json` declares a top-level
# `"dynamic": "runtime"`. Runtime fields are an Elasticsearch-only
# feature; OpenSearch cannot parse the value and rejects index
# creation with `mapper_parsing_exception: Could not convert
# [dynamic.dynamic] to boolean`. Fall back to standard dynamic
# mapping (`true`) on OpenSearch so dynamic field discovery is kept
# without the ES-specific runtime semantics. The shared mapping file
# is left untouched so the Elasticsearch backend still gets runtime
# fields.
if mappings.get("dynamic") == "runtime":
mappings = {**mappings, "dynamic": True}
from opensearchpy.client import IndicesClient
body = {
"settings": doc_meta_mapping["settings"],
"mappings": doc_meta_mapping["mappings"],
"mappings": mappings,
}
return IndicesClient(self.os).create(index=index_name, body=body)
except Exception as e:
@@ -247,29 +260,29 @@ class OSConnection(DocStoreConnection):
"""
def search(
self, selectFields: list[str],
highlightFields: list[str],
self, select_fields: list[str],
highlight_fields: list[str],
condition: dict,
matchExprs: list[MatchExpr],
orderBy: OrderByExpr,
match_expressions: list[MatchExpr],
order_by: OrderByExpr,
offset: int,
limit: int,
indexNames: str | list[str],
knowledgebaseIds: list[str],
aggFields: list[str] = [],
index_names: str | list[str],
knowledgebase_ids: list[str],
agg_fields: list[str] = [],
rank_feature: dict | None = None
):
"""
Refers to https://github.com/opensearch-project/opensearch-py/blob/main/guides/dsl.md
"""
use_knn = False
if isinstance(indexNames, str):
indexNames = indexNames.split(",")
assert isinstance(indexNames, list) and len(indexNames) > 0
if isinstance(index_names, str):
index_names = index_names.split(",")
assert isinstance(index_names, list) and len(index_names) > 0
assert "_id" not in condition
bqry = Q("bool", must=[])
condition["kb_id"] = knowledgebaseIds
condition["kb_id"] = knowledgebase_ids
for k, v in condition.items():
if k == "available_int":
if v == 0:
@@ -290,15 +303,15 @@ class OSConnection(DocStoreConnection):
s = Search()
vector_similarity_weight = 0.5
for m in matchExprs:
for m in match_expressions:
if isinstance(m, FusionExpr) and m.method == "weighted_sum" and "weights" in m.fusion_params:
assert len(matchExprs) == 3 and isinstance(matchExprs[0], MatchTextExpr) and isinstance(matchExprs[1],
assert len(match_expressions) == 3 and isinstance(match_expressions[0], MatchTextExpr) and isinstance(match_expressions[1],
MatchDenseExpr) and isinstance(
matchExprs[2], FusionExpr)
match_expressions[2], FusionExpr)
weights = m.fusion_params["weights"]
vector_similarity_weight = float(weights.split(",")[1])
knn_query = {}
for m in matchExprs:
for m in match_expressions:
if isinstance(m, MatchTextExpr):
minimum_should_match = m.extra_options.get("minimum_should_match", 0.0)
if isinstance(minimum_should_match, float):
@@ -334,12 +347,12 @@ class OSConnection(DocStoreConnection):
if bqry:
s = s.query(bqry)
for field in highlightFields:
for field in highlight_fields:
s = s.highlight(field, force_source=True, no_match_size=30, require_field_match=False)
if orderBy:
if order_by:
orders = list()
for field, order in orderBy.fields:
for field, order in order_by.fields:
order = "asc" if order == 0 else "desc"
if field in ["page_num_int", "top_int"]:
order_info = {"order": order, "unmapped_type": "float",
@@ -351,13 +364,13 @@ class OSConnection(DocStoreConnection):
orders.append({field: order_info})
s = s.sort(*orders)
for fld in aggFields:
for fld in agg_fields:
s.aggs.bucket(f'aggs_{fld}', 'terms', field=fld, size=1000000)
if limit > 0:
s = s[offset:offset + limit]
q = s.to_dict()
logger.debug(f"OSConnection.search {str(indexNames)} query: " + json.dumps(q))
logger.debug(f"OSConnection.search {str(index_names)} query: " + json.dumps(q))
if use_knn:
del q["query"]
@@ -365,7 +378,7 @@ class OSConnection(DocStoreConnection):
for i in range(ATTEMPT_TIME):
try:
res = self.os.search(index=indexNames,
res = self.os.search(index=index_names,
body=q,
timeout=600,
# search_type="dfs_query_then_fetch",
@@ -373,10 +386,10 @@ class OSConnection(DocStoreConnection):
_source=True)
if str(res.get("timed_out", "")).lower() == "true":
raise Exception("OpenSearch Timeout.")
logger.debug(f"OSConnection.search {str(indexNames)} res: " + str(res))
logger.debug(f"OSConnection.search {str(index_names)} res: " + str(res))
return res
except Exception as e:
logger.exception(f"OSConnection.search {str(indexNames)} query: " + str(q))
logger.exception(f"OSConnection.search {str(index_names)} query: " + str(q))
if str(e).find("Timeout") > 0:
continue
raise e