From 11af34a895a1de1ed8d1db15a93bdc46ab2bdea2 Mon Sep 17 00:00:00 2001 From: Rintaro Date: Fri, 29 May 2026 22:49:36 +0900 Subject: [PATCH] fix(opensearch): repair document-metadata path broken by #14577 (#15393) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### 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. --- rag/utils/opensearch_conn.py | 61 ++++++++++++++++++++++-------------- 1 file changed, 37 insertions(+), 24 deletions(-) diff --git a/rag/utils/opensearch_conn.py b/rag/utils/opensearch_conn.py index e9909d978..e68a033d8 100644 --- a/rag/utils/opensearch_conn.py +++ b/rag/utils/opensearch_conn.py @@ -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