diff --git a/internal/engine/elasticsearch/chunk_readback_integration_test.go b/internal/engine/elasticsearch/chunk_readback_integration_test.go new file mode 100644 index 0000000000..abf200491a --- /dev/null +++ b/internal/engine/elasticsearch/chunk_readback_integration_test.go @@ -0,0 +1,110 @@ +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build integration + +package elasticsearch + +import ( + "context" + "reflect" + "testing" + + "ragflow/internal/common" + "ragflow/internal/server/config" +) + +// TestInsertChunks_ReadBackSuffixedFields is the T0 integration-tier read-back +// baseline (issue #17371): it writes a chunk to a real Elasticsearch instance +// and reads it back, asserting the stored document keeps the index-physical +// field names the ingestion pipeline emits. This catches regressions that the +// unit-tier test (which only inspects the request body) cannot — e.g. an ES +// mapping coercion that changes the stored value. +// +// Requires a running Elasticsearch; set ES_TEST=1 to run. +func TestInsertChunks_ReadBackSuffixedFields(t *testing.T) { + if common.GetEnv(common.EnvESTest) != "1" { + t.Skip("Skipping ES integration test; set ES_TEST=1 to run") + } + + engine, err := NewEngine(getESTestConfig()) + if err != nil { + t.Fatalf("NewEngine: %v", err) + } + + ctx := context.Background() + baseName := "ragflow_chunk_readback_test" + datasetID := "kb-1" + chunkID := "readback-chunk-1" + chunk := map[string]interface{}{ + "doc_id": "doc-1", + "id": chunkID, + "kb_id": "producer-kb-1", + "docnm_kwd": "sample.md", + "content_with_weight": "hello world", + "create_timestamp_flt": float64(123.0), + "question_kwd": []string{"q1", "q2"}, + "important_kwd": []string{"k1"}, + "page_num_int": int(1), + "position_int": int(2), + } + + if _, err := engine.InsertChunks(ctx, []map[string]interface{}{chunk}, baseName, datasetID); err != nil { + t.Fatalf("InsertChunks: %v", err) + } + + got, err := engine.GetChunk(ctx, baseName, chunkID, []string{datasetID}) + if err != nil { + t.Fatalf("GetChunk: %v", err) + } + stored, ok := got.(map[string]interface{}) + if !ok { + t.Fatalf("GetChunk returned %T, want map[string]interface{}", got) + } + + assertStoredField(t, stored, "docnm_kwd", "sample.md") + assertStoredField(t, stored, "content_with_weight", "hello world") + assertStoredField(t, stored, "create_timestamp_flt", float64(123.0)) + assertStoredField(t, stored, "page_num_int", float64(1)) + // The input kb_id ("producer-kb-1") differs from datasetID ("kb-1"); the + // stored value must equal datasetID, proving InsertChunks overrides it. + assertStoredField(t, stored, "kb_id", datasetID) + if v, ok := stored["question_kwd"]; !ok || !reflect.DeepEqual(v, []interface{}{"q1", "q2"}) { + t.Errorf("stored question_kwd = %#v, want [q1 q2]", v) + } +} + +// getESTestConfig builds an Elasticsearch config for integration tests from +// the environment, falling back to localhost defaults. It is kept local to this +// file so the read-back test does not depend on kg_test.go (whose getTestConfig +// is an unrelated main-branch compile fix tracked separately). +func getESTestConfig() config.ElasticsearchConfig { + hosts := common.GetEnv(common.EnvESHost) + if hosts == "" { + hosts = "http://localhost:1200" + } + username := common.GetEnv(common.EnvESUsername) + if username == "" { + username = "elastic" + } + password := common.GetEnv(common.EnvESPassword) + if password == "" { + password = "infini_rag_flow" + } + return config.ElasticsearchConfig{ + Hosts: hosts, + Username: username, + Password: password, + } +} diff --git a/internal/engine/elasticsearch/chunk_readback_test.go b/internal/engine/elasticsearch/chunk_readback_test.go new file mode 100644 index 0000000000..9d9f96df09 --- /dev/null +++ b/internal/engine/elasticsearch/chunk_readback_test.go @@ -0,0 +1,112 @@ +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package elasticsearch + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" +) + +// assertStoredField compares a field of the (JSON-decoded) captured bulk +// document against the expected value using reflect.DeepEqual. The bulk body +// round-trips through JSON, so numbers arrive as float64 and arrays as +// []interface{}. +func assertStoredField(t *testing.T, doc map[string]interface{}, key string, want interface{}) { + t.Helper() + got, ok := doc[key] + if !ok { + t.Errorf("stored doc missing field %q", key) + return + } + if !reflect.DeepEqual(got, want) { + t.Errorf("field %q = %#v, want %#v", key, got, want) + } +} + +// TestInsertChunks_WritesIngestionShape is the T0 unit-tier read-back baseline +// (issue #17371). It stands up an in-memory fake Elasticsearch, captures the +// bulk request that InsertChunks sends, and asserts the stored document keeps +// the index-physical field names that the ingestion pipeline currently emits +// (docnm_kwd, content_with_weight, create_timestamp_flt, question_kwd, +// important_kwd, page_num_int, position_int, kb_id overridden to datasetID). +// +// This guards the engine write boundary: when the suffixing is later moved +// from ingestion into this boundary (T2), the document we send to ES must stay +// byte-identical. No real ES is required, so it runs in the default unit tier. +func TestInsertChunks_WritesIngestionShape(t *testing.T) { + var bulkBody []byte + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/_bulk" { + bulkBody, _ = io.ReadAll(r.Body) + w.Header().Set("X-Elastic-Product", "Elasticsearch") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"errors":false,"items":[]}`)) + return + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + engine := newTestEngine(t, srv.URL) + + // Mirrors the shape produced by indexdoc.ProcessChunksForPipeline after T1 + // (kb_id is a plain string). InsertChunks overrides kb_id with datasetID, + // so the stored value is the datasetID regardless of the input form. The + // input kb_id ("producer-kb-1") is intentionally DISTINCT from datasetID + // ("kb-1") so the test actually exercises the override rather than passing + // through an already-equal value. + chunk := map[string]interface{}{ + "doc_id": "doc-1", + "id": "chunk-1", + "kb_id": "producer-kb-1", + "docnm_kwd": "sample.md", + "content_with_weight": "hello world", + "create_time": "2026-08-04 00:00:00", + "create_timestamp_flt": float64(123.0), + "question_kwd": []string{"q1", "q2"}, + "important_kwd": []string{"k1"}, + "page_num_int": int(1), + "position_int": int(2), + } + + if _, err := engine.InsertChunks(context.Background(), []map[string]interface{}{chunk}, "ragflow_chunk_readback_test", "kb-1"); err != nil { + t.Fatalf("InsertChunks: %v", err) + } + + lines := strings.Split(strings.TrimSpace(string(bulkBody)), "\n") + if len(lines) < 2 { + t.Fatalf("bulk body has %d lines, want >=2: %q", len(lines), string(bulkBody)) + } + var doc map[string]interface{} + if err := json.Unmarshal([]byte(lines[1]), &doc); err != nil { + t.Fatalf("unmarshal doc line: %v", err) + } + + assertStoredField(t, doc, "docnm_kwd", "sample.md") + assertStoredField(t, doc, "content_with_weight", "hello world") + assertStoredField(t, doc, "create_timestamp_flt", float64(123.0)) + assertStoredField(t, doc, "question_kwd", []interface{}{"q1", "q2"}) + assertStoredField(t, doc, "important_kwd", []interface{}{"k1"}) + assertStoredField(t, doc, "page_num_int", float64(1)) + assertStoredField(t, doc, "position_int", float64(2)) + // InsertChunks overrides kb_id with datasetID on write. + assertStoredField(t, doc, "kb_id", "kb-1") +} diff --git a/internal/engine/infinity/chunk_transform_test.go b/internal/engine/infinity/chunk_transform_test.go new file mode 100644 index 0000000000..60af458523 --- /dev/null +++ b/internal/engine/infinity/chunk_transform_test.go @@ -0,0 +1,80 @@ +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package infinity + +import ( + "reflect" + "testing" +) + +// TestTransformChunkFields_IngestionShape is the T0 unit-tier read-back +// baseline for the Infinity engine (issue #17371). Unlike Elasticsearch, which +// stores the suffixed field names verbatim, Infinity reverse-maps the +// ingestion-emitted names to its own base names inside transformChunkFields. +// This test calls that transform directly (the engine requires a live Infinity +// instance, so it cannot be exercised via httptest) and asserts the mapping +// stays stable. It guards T2: when the suffixing is moved out of ingestion into +// the engine write boundary, transformChunkFields must keep producing the same +// base-name document. +// +// The input mirrors the shape produced by indexdoc.ProcessChunksForPipeline +// after T1 (kb_id is a plain string). +func TestTransformChunkFields_IngestionShape(t *testing.T) { + chunk := map[string]interface{}{ + "doc_id": "doc-1", + "id": "chunk-1", + "kb_id": "kb-1", + "docnm_kwd": "sample.md", + "content_with_weight": "hello world", + "create_timestamp_flt": float64(123.0), + "question_kwd": []interface{}{"q1", "q2"}, + "important_kwd": []interface{}{"k1"}, + "page_num_int": int(1), + "position_int": int(2), + } + + got := transformChunkFields(chunk, nil) + + want := map[string]interface{}{ + "doc_id": "doc-1", + "id": "chunk-1", + "kb_id": "kb-1", + "docnm": "sample.md", + "content": "hello world", + "create_timestamp_flt": float64(123.0), + "questions": "q1\nq2", + "important_keywords": "k1", + "page_num_int": int(1), + "position_int": int(2), + } + + for k, wv := range want { + gv, ok := got[k] + if !ok { + t.Errorf("transformed doc missing field %q", k) + continue + } + if !reflect.DeepEqual(gv, wv) { + t.Errorf("field %q = %#v, want %#v", k, gv, wv) + } + } + + // The suffixed ingestion-only names must not leak into the transformed doc. + for _, leaked := range []string{"docnm_kwd", "content_with_weight", "question_kwd", "important_kwd"} { + if _, ok := got[leaked]; ok { + t.Errorf("transform leaked ingestion-only field %q into Infinity doc", leaked) + } + } +} diff --git a/internal/ingestion/task/indexdoc/process.go b/internal/ingestion/task/indexdoc/process.go index d5516e8d94..867741fbbd 100644 --- a/internal/ingestion/task/indexdoc/process.go +++ b/internal/ingestion/task/indexdoc/process.go @@ -77,7 +77,7 @@ func ProcessChunksForPipeline( for _, ck := range chunks { ck["doc_id"] = docID - ck["kb_id"] = []string{kbID} + ck["kb_id"] = kbID ck["docnm_kwd"] = docName ck["create_time"] = timeStr ck["create_timestamp_flt"] = timestamp diff --git a/internal/ingestion/task/indexdoc/process_test.go b/internal/ingestion/task/indexdoc/process_test.go index 61f1c0b2da..26e633c3d6 100644 --- a/internal/ingestion/task/indexdoc/process_test.go +++ b/internal/ingestion/task/indexdoc/process_test.go @@ -53,12 +53,8 @@ func TestProcessChunksForPipeline_SetsDocIDAndKBID(t *testing.T) { if chunks[0]["doc_id"] != "doc-1" { t.Errorf("doc_id = %q, want \"doc-1\"", chunks[0]["doc_id"]) } - if kbIDs, ok := chunks[0]["kb_id"].([]string); ok { - if len(kbIDs) != 1 || kbIDs[0] != "kb-1" { - t.Errorf("kb_id = %v, want [\"kb-1\"]", chunks[0]["kb_id"]) - } - } else { - t.Errorf("kb_id should be []string, got %T", chunks[0]["kb_id"]) + if kbID, ok := chunks[0]["kb_id"].(string); !ok || kbID != "kb-1" { + t.Errorf("kb_id = %v, want \"kb-1\" (string)", chunks[0]["kb_id"]) } } diff --git a/internal/ingestion/task/pipeline_e2e_test.go b/internal/ingestion/task/pipeline_e2e_test.go index b68970c335..0225419ab3 100644 --- a/internal/ingestion/task/pipeline_e2e_test.go +++ b/internal/ingestion/task/pipeline_e2e_test.go @@ -32,7 +32,7 @@ import ( "ragflow/internal/engine/infinity" indexdoc "ragflow/internal/ingestion/task/indexdoc" "ragflow/internal/ingestion/testutil" - "ragflow/internal/server" + "ragflow/internal/server/config" "ragflow/internal/service" ) @@ -67,7 +67,7 @@ func setupTestDocEngine(t *testing.T, engineType engine.EngineType, tenantID, da esPassword = "infini_rag_flow" } - cfg := &server.ElasticsearchConfig{ + cfg := config.ElasticsearchConfig{ Hosts: esHost, Username: esUser, Password: esPassword, @@ -91,7 +91,7 @@ func setupTestDocEngine(t *testing.T, engineType engine.EngineType, tenantID, da return nil, func() {} } - cfg := &server.InfinityConfig{ + cfg := config.InfinityConfig{ URI: infURI, DBName: "ragflow_e2e_test", PostgresPort: 5432, diff --git a/internal/ingestion/task/pipeline_real_integration_test.go b/internal/ingestion/task/pipeline_real_integration_test.go index 9c3068445d..5ade785ac6 100644 --- a/internal/ingestion/task/pipeline_real_integration_test.go +++ b/internal/ingestion/task/pipeline_real_integration_test.go @@ -720,8 +720,8 @@ func taskS3SafeBucketName(s string) string { } // taskChunkFieldEqualsStr compares a chunk field to a plain string, tolerating -// the slice form used internally (e.g. kb_id is []string{kbID} and survives a -// JSON round-trip as []any{kbID}). +// either form a producer may emit: a plain string (e.g. kb_id is "kb-1" after +// T1) or a single-element slice that survives a JSON round-trip as []any{"kb-1"}. func taskChunkFieldEqualsStr(v any, want string) bool { switch val := v.(type) { case string: