Implement OpenAI chat completions in GO (#16177)

### What problem does this PR solve? Implement OpenAI chat completions in GO POST /api/v1/openai/<chat_id>/chat/completions OpenAI chat cli: internal/development.md ### Type of change - [x] Refactoring
2026-07-23 08:56:42 +08:00 · 2026-06-18 18:07:27 +08:00
parent b53b5bf12c
commit 563d855780
61 changed files with 15327 additions and 2105 deletions
--- a/internal/engine/elasticsearch/sql.go
+++ b/internal/engine/elasticsearch/sql.go
@@ -0,0 +1,206 @@
+//
+//  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 (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"regexp"
+	"strings"
+	"time"
+
+	"ragflow/internal/common"
+	"ragflow/internal/tokenizer"
+
+	"github.com/elastic/go-elasticsearch/v8/esapi"
+	"go.uber.org/zap"
+)
+
+const (
+	esSQLRequestTimeout = 2 * time.Second
+	esSQLFetchSize      = 128
+)
+
+const esSQLRetryAttempts = 2
+const esSQLRetryDelay = 3 * time.Second
+
+var whitespaceRe = regexp.MustCompile("[ `]+")
+var lktksMatchRe = regexp.MustCompile(` ([a-z_]+_l?tks)( like | ?= ?)'([^']+)'`)
+
+// Preprocess normalizes SQL for ES: collapses whitespace/backticks,
+// strips '%', and rewrites `<field>_l?tks like/= 'value'` into a
+// tokenized MATCH() call.
+func Preprocess(sql string) string {
+	sql = whitespaceRe.ReplaceAllString(sql, " ")
+	sql = strings.ReplaceAll(sql, "%", "")
+
+	// Collect replacements so we don't re-scan tokens we've already rewritten
+	type replacement struct {
+		old, new string
+	}
+	var replaces []replacement
+	for _, m := range lktksMatchRe.FindAllStringSubmatchIndex(sql, -1) {
+		match := sql[m[0]:m[1]]
+		fld := sql[m[2]:m[3]]
+		val := sql[m[6]:m[7]]
+		tokenized, err := tokenizer.Tokenize(val)
+		if err != nil {
+			continue
+		}
+		fine, err := tokenizer.FineGrainedTokenize(tokenized)
+		if err != nil {
+			continue
+		}
+		replaces = append(replaces, replacement{
+			old: match,
+			new: fmt.Sprintf(" MATCH(%s, '%s', 'operator=OR;minimum_should_match=30%%') ", fld, fine),
+		})
+	}
+	for _, r := range replaces {
+		sql = strings.Replace(sql, r.old, r.new, 1)
+	}
+	return sql
+}
+
+// RunSQL posts SQL to `/_sql`, translates the response into chunk-shaped maps.
+// Returns (nil, nil) on empty rows; (nil, error) when retries exhausted.
+func (e *elasticsearchEngine) RunSQL(ctx context.Context, tableName string, sqlText string, kbIDs []string, format string) ([]map[string]interface{}, error) {
+	if e == nil || e.client == nil {
+		return nil, fmt.Errorf("Elasticsearch RunSQL: client not initialized")
+	}
+	if sqlText == "" {
+		return nil, fmt.Errorf("Elasticsearch RunSQL: empty SQL")
+	}
+
+	common.Debug("ESConnection.sql get sql", zap.String("sql", sqlText))
+	sqlText = Preprocess(sqlText)
+	common.Debug("ESConnection.sql to es", zap.String("sql", sqlText))
+
+	var lastErr error
+	for attempt := 0; attempt < esSQLRetryAttempts; attempt++ {
+		rows, err := e.runSQLOnce(ctx, sqlText, format)
+		if err == nil {
+			return rows, nil
+		}
+		lastErr = err
+		if !isTimeoutError(err) {
+			common.Warn("ESConnection.sql got exception",
+				zap.String("sql", sqlText),
+				zap.Error(err))
+			return nil, fmt.Errorf("SQL error: %w\n\nSQL: %s", err, sqlText)
+		}
+		common.Warn("ES request timeout",
+			zap.String("sql", sqlText),
+			zap.Int("attempt", attempt+1),
+			zap.Error(err))
+		if attempt < esSQLRetryAttempts-1 {
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-time.After(esSQLRetryDelay):
+			}
+		}
+	}
+	common.Error(fmt.Sprintf("ESConnection.sql timeout after %d attempts. SQL: %s", esSQLRetryAttempts, sqlText), lastErr)
+	return nil, fmt.Errorf("Elasticsearch RunSQL: timeout after %d attempts: %w", esSQLRetryAttempts, lastErr)
+}
+
+func (e *elasticsearchEngine) runSQLOnce(ctx context.Context, sqlText string, format string) ([]map[string]interface{}, error) {
+	ctx, cancel := context.WithTimeout(ctx, esSQLRequestTimeout)
+	defer cancel()
+
+	body := map[string]interface{}{
+		"query":      sqlText,
+		"fetch_size": esSQLFetchSize,
+	}
+	buf, err := json.Marshal(body)
+	if err != nil {
+		return nil, fmt.Errorf("marshal body: %w", err)
+	}
+
+	req := esapi.SQLQueryRequest{
+		Body:   bytes.NewReader(buf),
+		Format: format,
+	}
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		return nil, fmt.Errorf("request failed: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		errBody, _ := io.ReadAll(res.Body)
+		return nil, fmt.Errorf("status=%d body=%s", res.StatusCode, string(errBody))
+	}
+
+	// Parse the SQL response.
+	var resp struct {
+		Columns []struct {
+			Name string `json:"name"`
+			Type string `json:"type"`
+		} `json:"columns"`
+		Rows [][]interface{} `json:"rows"`
+	}
+	if err := json.NewDecoder(res.Body).Decode(&resp); err != nil {
+		return nil, fmt.Errorf("decode response: %w", err)
+	}
+
+	if len(resp.Rows) == 0 {
+		return nil, nil
+	}
+
+	// Convert to chunk-shaped maps. Column names map 1:1 to JSON keys.
+	out := make([]map[string]interface{}, 0, len(resp.Rows))
+	for _, row := range resp.Rows {
+		cm := make(map[string]interface{}, len(resp.Columns))
+		for i, col := range resp.Columns {
+			if i < len(row) {
+				cm[col.Name] = row[i]
+			}
+		}
+		out = append(out, cm)
+	}
+	return out, nil
+}
+
+// isTimeoutError detects connection-level and per-attempt timeouts
+// via context.DeadlineExceeded, net.Error.Timeout(), and substring
+// matches (for SDKs that wrap without typed errors).
+func isTimeoutError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) && netErr.Timeout() {
+		return true
+	}
+	msg := err.Error()
+	for _, sub := range []string{"i/o timeout", "deadline exceeded", "connection timeout", "context deadline"} {
+		if strings.Contains(msg, sub) {
+			return true
+		}
+	}
+	return false
+}
--- a/internal/engine/elasticsearch/sql_test.go
+++ b/internal/engine/elasticsearch/sql_test.go
@@ -0,0 +1,493 @@
+//
+//  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"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"regexp"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"ragflow/internal/tokenizer"
+
+	"github.com/elastic/go-elasticsearch/v8"
+)
+
+// capturedRequest holds the request body the test server saw, for
+// assertions.
+type capturedRequest struct {
+	mu     sync.Mutex
+	path   string
+	body   string
+	method string
+}
+
+// newCapturingServer returns an httptest.Server that captures each
+// incoming request and replies with the given body / status.
+func newCapturingServer(t *testing.T, replyStatus int, replyBody string) (*httptest.Server, *capturedRequest) {
+	t.Helper()
+	cap := &capturedRequest{}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		cap.mu.Lock()
+		cap.method = r.Method
+		cap.path = r.URL.Path
+		cap.body = string(body)
+		cap.mu.Unlock()
+		w.Header().Set("X-Elastic-Product", "Elasticsearch")
+		w.WriteHeader(replyStatus)
+		_, _ = w.Write([]byte(replyBody))
+	}))
+	t.Cleanup(srv.Close)
+	return srv, cap
+}
+
+// newTestEngine constructs an elasticsearchEngine pointing at the given
+// test server. Bypasses NewEngine (which calls ES Info to verify
+// connectivity) — the test server is a stub, not a real ES cluster.
+func newTestEngine(t *testing.T, srvURL string) *elasticsearchEngine {
+	t.Helper()
+	client, err := elasticsearch.NewClient(elasticsearch.Config{
+		Addresses: []string{srvURL},
+	})
+	if err != nil {
+		t.Fatalf("elasticsearch.NewClient: %v", err)
+	}
+	return &elasticsearchEngine{client: client}
+}
+
+const sampleESResponse = `{
+  "columns": [
+    {"name": "doc_id", "type": "text"},
+    {"name": "docnm", "type": "text"},
+    {"name": "count", "type": "long"}
+  ],
+  "rows": [
+    ["d1", "report.pdf", 5],
+    ["d2", "spec.pdf", 3]
+  ]
+}`
+
+// TestRunSQL_NoFilterAdded verifies the request body is exactly
+// {"query": <sql>} — the redundant `filter` field that the previous
+// implementation added is gone. (service.addKBFilter is the source of
+// truth for kb_id scoping upstream of RunSQL.)
+func TestRunSQL_NoFilterAdded(t *testing.T) {
+	srv, cap := newCapturingServer(t, http.StatusOK, sampleESResponse)
+	e := newTestEngine(t, srv.URL)
+
+	rows, err := e.RunSQL(context.Background(), "ragflow_t1", "SELECT doc_id FROM ragflow_t1", nil, "json")
+	if err != nil {
+		t.Fatalf("RunSQL: %v", err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("rows: got %d, want 2", len(rows))
+	}
+	cap.mu.Lock()
+	got := cap.body
+	cap.mu.Unlock()
+
+	var body map[string]interface{}
+	if err := json.Unmarshal([]byte(got), &body); err != nil {
+		t.Fatalf("body is not JSON: %v\nbody=%q", err, got)
+	}
+	if _, has := body["filter"]; has {
+		t.Errorf("RunSQL request must NOT include top-level filter (addKBFilter is the source of truth upstream). body=%v", body)
+	}
+	if _, has := body["query"]; !has {
+		t.Errorf("RunSQL request must include query. body=%v", body)
+	}
+}
+
+// TestRunSQL_WhitespaceNormalizedAndPercentStripped verifies the Python
+// preprocessing step `re.sub(r"[ `]+", " ", sql)` + `sql.replace("%", "")`
+// is applied. Without these, the LLM-generated SQL with stray backticks
+// or `%` characters (e.g. from JSON decoding glitches) would fail to
+// parse in ES.
+func TestRunSQL_WhitespaceNormalizedAndPercentStripped(t *testing.T) {
+	srv, cap := newCapturingServer(t, http.StatusOK, sampleESResponse)
+	e := newTestEngine(t, srv.URL)
+
+	// Input SQL has multiple backticks/spaces and trailing % characters.
+	in := "SELECT   doc_id  FROM  `ragflow_t1`  WHERE  count  >  0  %"
+	_, err := e.RunSQL(context.Background(), "ragflow_t1", in, nil, "json")
+	if err != nil {
+		t.Fatalf("RunSQL: %v", err)
+	}
+	cap.mu.Lock()
+	got := cap.body
+	cap.mu.Unlock()
+
+	var body map[string]interface{}
+	if err := json.Unmarshal([]byte(got), &body); err != nil {
+		t.Fatalf("body is not JSON: %v\nbody=%q", err, got)
+	}
+	q, _ := body["query"].(string)
+	if strings.Contains(q, "  ") {
+		t.Errorf("query still has multiple spaces (whitespace not normalized): %q", q)
+	}
+	if strings.Contains(q, "`") {
+		t.Errorf("query still has backticks (whitespace+backtick regex not applied): %q", q)
+	}
+	if strings.Contains(q, "%") {
+		t.Errorf("query still has %% (percent strip not applied): %q", q)
+	}
+}
+
+// TestRunSQL_PerAttemptTimeout verifies the derived context has a 2s
+// deadline. We send a hanging response from the test server and assert
+// the call returns well before 30s (the Go ES client's default
+// transport-level timeout). With the retry loop in place, the total
+// time is 2s (first attempt) + 3s (sleep) + 2s (second attempt) = ~7s.
+func TestRunSQL_PerAttemptTimeout(t *testing.T) {
+	hang := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-hang
+	}))
+	t.Cleanup(func() {
+		close(hang)
+		srv.Close()
+	})
+	e := newTestEngine(t, srv.URL)
+
+	start := time.Now()
+	_, err := e.RunSQL(context.Background(), "ragflow_t1", "SELECT 1", nil, "json")
+	elapsed := time.Since(start)
+	if err == nil {
+		t.Fatalf("RunSQL: got nil error, want timeout error")
+	}
+	// 2s + 3s + 2s = 7s; allow generous upper bound for the test runner.
+	if elapsed < 6*time.Second {
+		t.Errorf("RunSQL returned in %s; expected ~7s (2 attempts + 3s sleep)", elapsed)
+	}
+	if elapsed > 10*time.Second {
+		t.Errorf("RunSQL took %s; expected ~7s, looks like the retry didn't fire", elapsed)
+	}
+	// The final error should mention timeout + 2 attempts.
+	if !strings.Contains(err.Error(), "timeout after 2 attempts") {
+		t.Errorf("err: got %q, want substring %q", err.Error(), "timeout after 2 attempts")
+	}
+}
+
+// TestRunSQL_RetryOnTimeoutThenSucceed simulates Python's
+// ConnectionTimeout-retry pattern: the first attempt times out, the
+// second attempt returns valid rows. The loop should silently retry
+// and return the rows.
+func TestRunSQL_RetryOnTimeoutThenSucceed(t *testing.T) {
+	var (
+		mu    sync.Mutex
+		calls int
+	)
+	release := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		calls++
+		attempt := calls
+		mu.Unlock()
+		if attempt == 1 {
+			// First attempt: hang so the 2s context fires.
+			select {
+			case <-release:
+			case <-r.Context().Done():
+			}
+			return
+		}
+		w.Header().Set("X-Elastic-Product", "Elasticsearch")
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(sampleESResponse))
+	}))
+	t.Cleanup(func() {
+		close(release)
+		srv.Close()
+	})
+
+	e := newTestEngine(t, srv.URL)
+	rows, err := e.RunSQL(context.Background(), "ragflow_t1", "SELECT 1", nil, "json")
+	if err != nil {
+		t.Fatalf("RunSQL: %v", err)
+	}
+	if len(rows) != 2 {
+		t.Errorf("rows: got %d, want 2 (second attempt should succeed)", len(rows))
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if calls != 2 {
+		t.Errorf("server calls: got %d, want 2 (initial + one retry)", calls)
+	}
+}
+
+// TestRunSQL_NonTimeoutErrorSurfacesImmediately verifies the non-retry
+// path: a 4xx ES response should NOT trigger a retry. The error must
+// be wrapped as `SQL error: <e>\n\nSQL: <sql>`, matching Python's
+// es_conn_base.py:400.
+func TestRunSQL_NonTimeoutErrorSurfacesImmediately(t *testing.T) {
+	var (
+		mu    sync.Mutex
+		calls int
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		calls++
+		mu.Unlock()
+		w.Header().Set("X-Elastic-Product", "Elasticsearch")
+		w.WriteHeader(http.StatusBadRequest)
+		_, _ = w.Write([]byte(`{"error": "syntax error"}`))
+	}))
+	t.Cleanup(srv.Close)
+
+	e := newTestEngine(t, srv.URL)
+	_, err := e.RunSQL(context.Background(), "ragflow_t1", "SELECT bad", nil, "json")
+	if err == nil {
+		t.Fatalf("RunSQL: got nil error, want error")
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if calls != 1 {
+		t.Errorf("server calls: got %d, want 1 (non-timeout error must NOT retry)", calls)
+	}
+	// Python wraps as `f"SQL error: {e}\n\nSQL: {sql}"`.
+	if !strings.Contains(err.Error(), "SQL error:") {
+		t.Errorf("err: got %q, want substring 'SQL error:'", err.Error())
+	}
+	if !strings.Contains(err.Error(), "SQL: SELECT bad") {
+		t.Errorf("err: got %q, want substring 'SQL: SELECT bad'", err.Error())
+	}
+}
+
+// TestRunSQL_RequestBodyHasFetchSizeAndFormat verifies the request body
+// includes fetch_size=128 and the SQLQueryRequest is built with
+// format="json", matching the Python defaults at rag/nlp/search.py:773.
+func TestRunSQL_RequestBodyHasFetchSizeAndFormat(t *testing.T) {
+	srv, cap := newCapturingServer(t, http.StatusOK, sampleESResponse)
+	e := newTestEngine(t, srv.URL)
+
+	if _, err := e.RunSQL(context.Background(), "ragflow_t1", "SELECT 1", nil, "json"); err != nil {
+		t.Fatalf("RunSQL: %v", err)
+	}
+	cap.mu.Lock()
+	got := cap.body
+	cap.mu.Unlock()
+
+	var body map[string]interface{}
+	if err := json.Unmarshal([]byte(got), &body); err != nil {
+		t.Fatalf("body is not JSON: %v\nbody=%q", err, got)
+	}
+	fs, ok := body["fetch_size"]
+	if !ok {
+		t.Errorf("body has no fetch_size; got %v", body)
+	}
+	if fmt.Sprint(fs) != "128" {
+		t.Errorf("fetch_size: got %v, want 128", fs)
+	}
+}
+
+// TestRunSQL_EmptyRowsReturnsNilNil verifies the (nil, nil) sentinel
+// for empty results — callers treat this as "fall through to vector
+// retrieval".
+func TestRunSQL_EmptyRowsReturnsNilNil(t *testing.T) {
+	empty := `{"columns": [{"name": "doc_id", "type": "text"}], "rows": []}`
+	srv, _ := newCapturingServer(t, http.StatusOK, empty)
+	e := newTestEngine(t, srv.URL)
+
+	rows, err := e.RunSQL(context.Background(), "ragflow_t1", "SELECT doc_id FROM ragflow_t1", nil, "json")
+	if err != nil {
+		t.Fatalf("RunSQL: %v", err)
+	}
+	if rows != nil {
+		t.Errorf("rows: got %v, want nil (empty-rows sentinel)", rows)
+	}
+}
+
+// TestRunSQL_PostsToSQLPath verifies the request goes to the /_sql
+// endpoint (the modern ES SQL API; the older /_xpack/sql path is
+// deprecated as of ES 7.x). The Go SDK's esapi.SQLQueryRequest hits
+// /_sql; the Python ES client is also pinned to the modern endpoint
+// at runtime even though the legacy /_xpack/sql name appears in the
+// SDK's method (`es.sql.query(...)`).
+func TestRunSQL_PostsToSQLPath(t *testing.T) {
+	srv, cap := newCapturingServer(t, http.StatusOK, sampleESResponse)
+	e := newTestEngine(t, srv.URL)
+
+	if _, err := e.RunSQL(context.Background(), "ragflow_t1", "SELECT 1", nil, "json"); err != nil {
+		t.Fatalf("RunSQL: %v", err)
+	}
+	cap.mu.Lock()
+	got := cap.path
+	cap.mu.Unlock()
+	if got != "/_sql" {
+		t.Errorf("path: got %q, want /_sql", got)
+	}
+}
+
+// TestMain registers the engine as "infinity" so tokenizer.Tokenize and
+// tokenizer.FineGrainedTokenize short-circuit and return the input
+// as-is. This lets the rewrite tests assert on the SHAPE of the MATCH()
+// substitution without depending on a real tokenizer pool.
+func TestMain(m *testing.M) {
+	tokenizer.RegisterEngineType(func() string { return "infinity" })
+	m.Run()
+}
+
+func TestPreprocess_WhitespaceAndBackticks(t *testing.T) {
+	cases := []struct {
+		in, want string
+	}{
+		{"a  b", "a b"},
+		{"a   b   c", "a b c"},
+		{"a`b`c", "a b c"},
+		{"a `` b", "a b"},
+		{"  leading and trailing  ", " leading and trailing "},
+	}
+	for _, c := range cases {
+		if got := Preprocess(c.in); got != c.want {
+			t.Errorf("Preprocess(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
+
+func TestPreprocess_StripsPercent(t *testing.T) {
+	cases := []struct {
+		in, want string
+	}{
+		{"count > 0 %", "count > 0 "},
+		{"100% match", "100 match"},
+		{"%%%", ""},
+	}
+	for _, c := range cases {
+		if got := Preprocess(c.in); got != c.want {
+			t.Errorf("Preprocess(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
+
+func TestPreprocess_LktksRewrite(t *testing.T) {
+	cases := []struct {
+		name   string
+		in     string
+		field  string
+		expect string
+	}{
+		{
+			"like with single-token value (ltks suffix)",
+			"select content_ltks like 'weather'",
+			"content_ltks",
+			"MATCH(content_ltks,",
+		},
+		{
+			"= with multi-word value (ltks suffix)",
+			"select content_ltks = 'final report'",
+			"content_ltks",
+			"MATCH(content_ltks,",
+		},
+		{
+			"tks (no l) suffix",
+			"select title_tks = 'hello'",
+			"title_tks",
+			"MATCH(title_tks,",
+		},
+		{
+			"leading-space anchor: no leading space means no match (mirrors Python regex)",
+			"content_ltks like 'weather'",
+			"content_ltks",
+			"content_ltks like 'weather'",
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			got := Preprocess(c.in)
+			isAnchorTest := c.expect == c.in
+			if isAnchorTest {
+				if got != c.in {
+					t.Errorf("Preprocess(%q) = %q, want unchanged (leading-space anchor should prevent match)", c.in, got)
+				}
+				return
+			}
+			if strings.Contains(got, c.field+" ") {
+				pattern := regexp.MustCompile(c.field + `( like | ?= ?)`)
+				if pattern.MatchString(got) {
+					t.Errorf("Preprocess(%q) = %q, still contains the original `<field> like/=` pattern", c.in, got)
+				}
+			}
+			if !strings.Contains(got, c.expect) {
+				t.Errorf("Preprocess(%q) = %q, want substring %q", c.in, got, c.expect)
+			}
+			if !strings.Contains(got, "minimum_should_match=30") {
+				t.Errorf("Preprocess(%q) = %q, want substring minimum_should_match=30", c.in, got)
+			}
+		})
+	}
+}
+
+func TestPreprocess_NoMatchLeavesSQLAlone(t *testing.T) {
+	in := "SELECT doc_id FROM ragflow_t1"
+	got := Preprocess(in)
+	if got != in {
+		t.Errorf("Preprocess(%q) = %q, want unchanged", in, got)
+	}
+}
+
+// fakeNetTimeoutErr implements net.Error with Timeout()==true.
+type fakeNetTimeoutErr struct{}
+
+func (fakeNetTimeoutErr) Error() string   { return "i/o timeout" }
+func (fakeNetTimeoutErr) Timeout() bool   { return true }
+func (fakeNetTimeoutErr) Temporary() bool { return true }
+
+func TestIsTimeoutError(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil error", nil, false},
+		{"context.DeadlineExceeded", context.DeadlineExceeded, true},
+		{"wrapped context.DeadlineExceeded", fmt.Errorf("wrap: %w", context.DeadlineExceeded), true},
+		{"net.Error.Timeout()==true", fakeNetTimeoutErr{}, true},
+		{"wrapped net.Error.Timeout", fmt.Errorf("wrap: %w", fakeNetTimeoutErr{}), true},
+		{"plain string 'i/o timeout'", errors.New("read tcp: i/o timeout"), true},
+		{"plain string 'deadline exceeded'", errors.New("context deadline exceeded"), true},
+		{"plain string 'connection timeout'", errors.New("connection timeout while reading"), true},
+		{"unrelated error", errors.New("parse: invalid character"), false},
+		{"EOF is not a timeout", errors.New("EOF"), false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if got := isTimeoutError(c.err); got != c.want {
+				t.Errorf("isTimeoutError(%v) = %v, want %v", c.err, got, c.want)
+			}
+		})
+	}
+}
+
+func TestIsTimeoutError_NonTimeoutNetError(t *testing.T) {
+	e := &net.OpError{
+		Op:  "dial",
+		Err: errors.New("connection refused"),
+	}
+	if isTimeoutError(e) {
+		t.Errorf("isTimeoutError(connection-refused) = true, want false")
+	}
+}
--- a/internal/engine/engine.go
+++ b/internal/engine/engine.go
@@ -61,6 +61,10 @@ type DocEngine interface {
 	GetFields(chunks []map[string]interface{}, fields []string) map[string]map[string]interface{}
 	GetAggregation(chunks []map[string]interface{}, fieldName string) []map[string]interface{}
 	GetHighlight(chunks []map[string]interface{}, keywords []string, fieldName string) map[string]string
+
+	// Run SQL
+	RunSQL(ctx context.Context, tableName string, sqlText string, kbIDs []string, format string) ([]map[string]interface{}, error)
+
 	GetChunkIDs(chunks []map[string]interface{}) []string
 	KNNScores(ctx context.Context, chunks []map[string]interface{}, queryVector []float64, topK int) (map[string]interface{}, error)
 	GetScores(searchResult map[string]interface{}) map[string]float64
--- a/internal/engine/global.go
+++ b/internal/engine/global.go
@@ -27,6 +27,7 @@ import (
 	"ragflow/internal/engine/infinity"

 	"go.uber.org/zap"
+	"ragflow/internal/tokenizer"
 )

 var (
@@ -40,6 +41,10 @@ var (
 func Init(cfg *server.DocEngineConfig) error {
 	var initErr error
 	once.Do(func() {
+		tokenizer.RegisterEngineType(func() string {
+			return string(GetEngineType())
+		})
+
 		engineType = EngineType(cfg.Type)
 		var err error
 		switch engineType {
--- a/internal/engine/infinity/chunk.go
+++ b/internal/engine/infinity/chunk.go
@@ -1916,6 +1916,7 @@ func convertMatchingField(fieldWeightStr string) string {
 		"authors_tks":         "authors@ft_authors_rag_coarse",
 		"authors_sm_tks":      "authors@ft_authors_rag_fine",
 		"tag_kwd":             "tag_kwd@ft_tag_kwd_whitespace__",
+		"toc_kwd":             "toc_kwd@ft_toc_kwd_whitespace__",
 		// Skill index fields
 		"name":        "name@ft_name_rag_coarse",
 		"tags":        "tags@ft_tags_rag_coarse",
--- a/internal/engine/infinity/client.go
+++ b/internal/engine/infinity/client.go
@@ -30,10 +30,18 @@ import (
 	infinity "github.com/infiniflow/infinity-go-sdk"
 )

-// infinityClient Infinity SDK client wrapper
 type infinityClient struct {
 	conn   *infinity.InfinityConnection
 	dbName string
+
+	// Original URI from config, used by RunSQL to extract the host.
+	hostURI string
+
+	// Port for psql wire-protocol listener (default 5432).
+	postgresPort int
+
+	// JSON file (under conf/) with the field-name alias map.
+	mappingFileName string
 }

 // NewInfinityClient creates a new Infinity client using the SDK
@@ -70,8 +78,11 @@ func NewInfinityClient(cfg *server.InfinityConfig) (*infinityClient, error) {
 	}

 	client := &infinityClient{
-		conn:   conn,
-		dbName: cfg.DBName,
+		conn:            conn,
+		dbName:          cfg.DBName,
+		hostURI:         cfg.URI,
+		postgresPort:    cfg.PostgresPort,
+		mappingFileName: cfg.MappingFileName,
 	}

 	return client, nil
--- a/internal/engine/infinity/metadata.go
+++ b/internal/engine/infinity/metadata.go
@@ -24,12 +24,13 @@ import (
 	"path/filepath"
 	"strings"

-	infinity "github.com/infiniflow/infinity-go-sdk"
 	"ragflow/internal/common"
 	"ragflow/internal/dao"
 	"ragflow/internal/engine/types"
 	"ragflow/internal/utility"

+	infinity "github.com/infiniflow/infinity-go-sdk"
+
 	"go.uber.org/zap"
 )

@@ -568,25 +569,113 @@ func (e *infinityEngine) SearchMetadata(ctx context.Context, req *types.SearchMe
 		}, nil
 	}

-	// Build search request for metadata - simpler than chunk search, no match expressions
-	searchReq := &types.SearchRequest{
-		IndexNames:   []string{tableName},
-		Offset:       req.Offset,
-		Limit:        req.Limit,
-		SelectFields: req.SelectFields,
-		Filter:       req.Filter,
-		MatchExprs:   nil, // No match expressions for metadata
-		OrderBy:      req.OrderBy,
-		RankFeature:  nil,
+	// Build output columns: use caller-specified fields, or "*" for all columns
+	var outputColumns []string
+	if len(req.SelectFields) > 0 {
+		outputColumns = req.SelectFields
+	} else {
+		outputColumns = []string{"*"}
 	}

-	result, err := e.Search(ctx, searchReq)
-	if err != nil {
-		return nil, err
+	// Pagination defaults
+	pageSize := req.Limit
+	if pageSize <= 0 {
+		pageSize = 30
 	}
+	offset := req.Offset
+	if offset < 0 {
+		offset = 0
+	}
+
+	// Build filter from req.Filter
+	var filterStr string
+	if req.Filter != nil {
+		filterStr = equivalentConditionToStr(req.Filter)
+	}
+
+	// Get database and table
+	db, err := e.client.conn.GetDatabase(e.client.dbName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get database: %w", err)
+	}
+
+	tbl, err := db.GetTable(tableName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get metadata table %s: %w", tableName, err)
+	}
+
+	// Build Infinity query (chainable API)
+	table := tbl.Output(outputColumns)
+	if filterStr != "" {
+		table = table.Filter(filterStr)
+	}
+
+	// Add order_by if provided
+	if req.OrderBy != nil && len(req.OrderBy.Fields) > 0 {
+		var sortFields [][2]interface{}
+		for _, orderField := range req.OrderBy.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(pageSize)
+	if offset > 0 {
+		table = table.Offset(offset)
+	}
+	table = table.Option(map[string]interface{}{"total_hits_count": true})
+
+	// Execute query
+	df, err := table.ToDataFrame()
+	if err != nil {
+		common.Warn("Infinity SearchMetadata query failed",
+			zap.String("tableName", tableName),
+			zap.Error(err))
+		return nil, fmt.Errorf("metadata query failed: %w", err)
+	}
+
+	// Convert column-oriented DataFrame to row-oriented records
+	records := make([]map[string]interface{}, 0)
+	for colName, colData := range df.ColumnData {
+		for i, val := range colData {
+			for len(records) <= i {
+				records = append(records, make(map[string]interface{}))
+			}
+			records[i][colName] = val
+		}
+	}
+
+	// Handle ROW_ID -> row_id() mapping (Infinity internal column)
+	for _, rec := range records {
+		if val, ok := rec["ROW_ID"]; ok {
+			rec["row_id()"] = val
+			delete(rec, "ROW_ID")
+		}
+	}
+
+	// Realign meta_fields column for multi-row queries (Infinity may
+	// concatenate values into one blob with 4-byte length prefix)
+	realignMetaFieldsColumn(records)
+
+	// Parse total_hits_count from ExtraInfo
+	var totalHits int64
+	if df.ExtraInfo != "" {
+		if t, ok := totalHitsFromInfinityExtraInfo(df.ExtraInfo); ok {
+			totalHits = t
+		}
+	}
+
+	common.Debug("SearchMetadata in Infinity completed",
+		zap.Int("rows", len(records)),
+		zap.Int64("total", totalHits))
+
 	return &types.SearchMetadataResult{
-		MetadataRecords: result.Chunks,
-		Total:           result.Total,
+		MetadataRecords: records,
+		Total:           totalHits,
 	}, nil
 }

--- a/internal/engine/infinity/sql.go
+++ b/internal/engine/infinity/sql.go
@@ -0,0 +1,330 @@
+//
+//  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 (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"ragflow/internal/common"
+
+	"go.uber.org/zap"
+)
+
+const (
+	psqlTimeout     = 10 * time.Second
+	defaultPsqlPath = "/usr/bin/psql"
+	defaultPsqlHost = "infinity"
+	defaultPsqlPort = "5432"
+)
+
+var whitespaceRe = regexp.MustCompile("[ `]+")
+
+var rowCountFooterRe = regexp.MustCompile(`^\(\d+ rows?`)
+
+// fieldMappingEntry is one entry in infinity_mapping.json.
+type fieldMappingEntry struct {
+	Type    string `json:"type"`
+	Comment string `json:"comment"`
+}
+
+// loadFieldMapping reads infinity_mapping.json and returns alias→actual
+// and actual→firstAlias maps. Silently returns empty maps on missing file.
+func loadFieldMapping(mappingFileName string) (aliasToActual map[string]string, actualToFirstAlias map[string]string, err error) {
+	if mappingFileName == "" {
+		mappingFileName = "infinity_mapping.json"
+	}
+	confPath := filepath.Join(projectBaseDir(), "conf", mappingFileName)
+	data, err := os.ReadFile(confPath)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return map[string]string{}, map[string]string{}, nil
+		}
+		return nil, nil, fmt.Errorf("load field mapping %q: %w", confPath, err)
+	}
+
+	fields := map[string]fieldMappingEntry{}
+	if err := json.Unmarshal(data, &fields); err != nil {
+		return nil, nil, fmt.Errorf("parse field mapping %q: %w", confPath, err)
+	}
+
+	aliasToActual = make(map[string]string, len(fields)*2)
+	actualToFirstAlias = make(map[string]string, len(fields))
+	for actual, info := range fields {
+		if info.Comment == "" {
+			continue
+		}
+		var firstAlias string
+		for _, raw := range strings.Split(info.Comment, ",") {
+			alias := strings.TrimSpace(raw)
+			if alias == "" {
+				continue
+			}
+			aliasToActual[alias] = actual
+			if firstAlias == "" {
+				firstAlias = alias
+			}
+		}
+		if firstAlias != "" {
+			actualToFirstAlias[actual] = firstAlias
+		}
+	}
+	return aliasToActual, actualToFirstAlias, nil
+}
+
+// projectBaseDir returns the project root. Honors RAG_PROJECT_BASE and
+// RAG_DEPLOY_BASE env vars; falls back to working directory.
+func projectBaseDir() string {
+	if v := os.Getenv("RAG_PROJECT_BASE"); v != "" {
+		return v
+	}
+	if v := os.Getenv("RAG_DEPLOY_BASE"); v != "" {
+		return v
+	}
+	// Fall back to the repository root. The Go engine package lives at
+	// internal/engine/infinity/; the repo root is three levels up.
+	wd, err := os.Getwd()
+	if err != nil {
+		return "."
+	}
+	return wd
+}
+
+// preprocessSQL collapses spaces/backticks and strips '%'.
+func preprocessSQL(sql string) string {
+	sql = whitespaceRe.ReplaceAllString(sql, " ")
+	sql = strings.ReplaceAll(sql, "%", "")
+	return sql
+}
+
+// rewriteFieldAliases rewrites alias field names to actual stored names
+// in SELECT, WHERE, ORDER BY, GROUP BY, and HAVING clauses.
+func rewriteFieldAliases(sql string, aliasToActual map[string]string) string {
+	if len(aliasToActual) == 0 {
+		return sql
+	}
+	selectRe := regexp.MustCompile(`(?si)(select\s+)(.+?)(\s+from\b)`)
+	sql = selectRe.ReplaceAllStringFunc(sql, func(m string) string {
+		parts := selectRe.FindStringSubmatch(m)
+		prefix, cols, suffix := parts[1], parts[2], parts[3]
+		for alias, actual := range aliasToActual {
+			pat := regexp.MustCompile(`(^|[,\s])` + regexp.QuoteMeta(alias) + `($|[,\s])`)
+			cols = pat.ReplaceAllString(cols, "${1}"+actual+"${2}")
+		}
+		return prefix + cols + suffix
+	})
+
+	clauseAliases := func(sql, keyword string) string {
+		return rewriteFirstAliasAfterKeyword(sql, keyword, aliasToActual)
+	}
+	sql = clauseAliases(sql, "where")
+	sql = clauseAliases(sql, "order by")
+	sql = clauseAliases(sql, "group by")
+	sql = clauseAliases(sql, "having")
+	return sql
+}
+
+func rewriteFirstAliasAfterKeyword(sql, keyword string, aliasToActual map[string]string) string {
+	for alias, actual := range aliasToActual {
+		aliasPat := regexp.MustCompile(`\b` + regexp.QuoteMeta(alias) + `\b`)
+		kwIdx := regexp.MustCompile(`(?i)\b` + regexp.QuoteMeta(keyword) + `\b`).FindStringIndex(sql)
+		if kwIdx == nil {
+			continue
+		}
+		tail := sql[kwIdx[1]:]
+		aliasIdx := aliasPat.FindStringIndex(tail)
+		if aliasIdx == nil {
+			continue
+		}
+		absStart := kwIdx[1] + aliasIdx[0]
+		absEnd := kwIdx[1] + aliasIdx[1]
+		sql = sql[:absStart] + actual + sql[absEnd:]
+	}
+	return sql
+}
+
+// psqlResult is the structured parse of a psql table-format output.
+type psqlResult struct {
+	Columns []string
+	Rows    [][]string
+}
+
+// runPsql shells out to psql and parses the table-format output.
+func runPsql(ctx context.Context, host, port, sql string) (*psqlResult, error) {
+	psqlPath, err := findPsqlBinary()
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, psqlTimeout)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, psqlPath, "-h", host, "-p", port, "-c", sql)
+	common.Debug("executing psql",
+		zap.String("path", psqlPath),
+		zap.String("host", host),
+		zap.String("port", port),
+	)
+	var stdout, stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+
+	if err := cmd.Run(); err != nil {
+		if ctx.Err() != nil {
+			return nil, fmt.Errorf("SQL timeout\n\nSQL: %s", sql)
+		}
+		return nil, fmt.Errorf("psql command failed: %s\nSQL: %s", strings.TrimSpace(stderr.String()), sql)
+	}
+	return parsePsqlTable(stdout.String()), nil
+}
+
+// findPsqlBinary checks PATH first, then falls back to defaultPsqlPath.
+func findPsqlBinary() (string, error) {
+	if path, err := exec.LookPath("psql"); err == nil {
+		return path, nil
+	}
+	if _, err := os.Stat(defaultPsqlPath); err == nil {
+		return defaultPsqlPath, nil
+	}
+	return "", fmt.Errorf("psql not found on PATH and not at %q", defaultPsqlPath)
+}
+
+// parsePsqlTable parses psql's pipe-delimited output:
+//
+//	col1 | col2
+//	-----+-----
+//	val1 | val2
+func parsePsqlTable(output string) *psqlResult {
+	res := &psqlResult{}
+	out := strings.TrimSpace(output)
+	if out == "" {
+		return res
+	}
+	lines := strings.Split(out, "\n")
+	if len(lines) == 0 {
+		return res
+	}
+
+	for _, raw := range strings.Split(lines[0], "|") {
+		if col := strings.TrimSpace(raw); col != "" {
+			res.Columns = append(res.Columns, col)
+		}
+	}
+
+	dataStart := 1
+	if len(lines) >= 2 && strings.Contains(lines[1], "-") {
+		dataStart = 2
+	}
+	for i := dataStart; i < len(lines); i++ {
+		line := strings.TrimSpace(lines[i])
+		if line == "" || rowCountFooterRe.MatchString(line) {
+			continue
+		}
+		cells := strings.Split(line, "|")
+		for j := range cells {
+			cells[j] = strings.TrimSpace(cells[j])
+		}
+		switch {
+		case len(cells) == len(res.Columns):
+			res.Rows = append(res.Rows, cells)
+		case len(cells) > len(res.Columns):
+			res.Rows = append(res.Rows, cells[:len(res.Columns)])
+		default:
+			padded := make([]string, len(res.Columns))
+			copy(padded, cells)
+			for k := len(cells); k < len(res.Columns); k++ {
+				padded[k] = ""
+			}
+			res.Rows = append(res.Rows, padded)
+		}
+	}
+	return res
+}
+
+// toRowMaps converts psqlResult to a slice of column-keyed maps.
+func toRowMaps(res *psqlResult) []map[string]interface{} {
+	if res == nil || len(res.Rows) == 0 {
+		return nil
+	}
+	out := make([]map[string]interface{}, 0, len(res.Rows))
+	for _, row := range res.Rows {
+		m := make(map[string]interface{}, len(res.Columns))
+		for j, col := range res.Columns {
+			if j < len(row) {
+				m[col] = row[j]
+			}
+		}
+		out = append(out, m)
+	}
+	return out
+}
+
+func resolvePsqlHostPort(hostURI string, postgresPort int) (host, port string) {
+	host = defaultPsqlHost
+	port = defaultPsqlPort
+	if postgresPort > 0 {
+		port = strconv.Itoa(postgresPort)
+	}
+	if hostURI != "" {
+		if h, _, ok := strings.Cut(hostURI, ":"); ok && h != "" {
+			host = h
+		}
+	}
+	return host, port
+}
+
+// RunSQL implements the SQL retrieval path: preprocess, rewrite aliases,
+// run psql subprocess, parse output.
+func (e *infinityEngine) RunSQL(ctx context.Context, tableName string, sqlText string, kbIDs []string, _ string) ([]map[string]interface{}, error) {
+	if e == nil || e.client == nil {
+		return nil, fmt.Errorf("infinity RunSQL: client not initialized")
+	}
+	sqlText = strings.TrimSpace(sqlText)
+	if sqlText == "" {
+		return nil, fmt.Errorf("infinity RunSQL: empty SQL")
+	}
+
+	common.Debug("InfinityConnection.sql get sql", zap.String("sql", sqlText))
+
+	sqlText = preprocessSQL(sqlText)
+
+	aliasMap, _, err := loadFieldMapping(e.client.mappingFileName)
+	if err != nil {
+		return nil, fmt.Errorf("infinity RunSQL: %w", err)
+	}
+	sqlText = rewriteFieldAliases(sqlText, aliasMap)
+
+	common.Debug("InfinityConnection.sql to execute", zap.String("sql", sqlText))
+
+	host, port := resolvePsqlHostPort(e.client.hostURI, e.client.postgresPort)
+	res, err := runPsql(ctx, host, port, sqlText)
+	if err != nil {
+		return nil, err
+	}
+
+	return toRowMaps(res), nil
+}
--- a/internal/engine/infinity/sql_test.go
+++ b/internal/engine/infinity/sql_test.go
@@ -0,0 +1,394 @@
+//
+//  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 (
+	"os"
+	"path/filepath"
+	"reflect"
+	"testing"
+)
+
+// -----------------------------------------------------------------------------
+// preprocessSQL — mirrors infinity_conn_base.py:788-789.
+// -----------------------------------------------------------------------------
+
+func TestPreprocessSQL_WhitespaceAndBackticks(t *testing.T) {
+	cases := []struct {
+		in, want string
+	}{
+		{"a  b", "a b"},
+		{"a   b   c", "a b c"},
+		{"a`b`c", "a b c"},
+		{"a `` b", "a b"},
+		// The regex collapses ALL runs of spaces/backticks — including
+		// leading and trailing whitespace. Trimming is a separate step
+		// in RunSQL (strings.TrimSpace before the preprocessing pass).
+		{"  leading and trailing  ", " leading and trailing "},
+	}
+	for _, c := range cases {
+		if got := preprocessSQL(c.in); got != c.want {
+			t.Errorf("preprocessSQL(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
+
+func TestPreprocessSQL_StripsPercent(t *testing.T) {
+	cases := []struct {
+		in, want string
+	}{
+		{"count > 0 %", "count > 0 "},
+		{"100% match", "100 match"},
+		{"%%%", ""},
+	}
+	for _, c := range cases {
+		if got := preprocessSQL(c.in); got != c.want {
+			t.Errorf("preprocessSQL(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
+
+func TestPreprocessSQL_Combined(t *testing.T) {
+	in := "SELECT   docnm_kwd  FROM  `ragflow_t1`  WHERE  count  >  0  %"
+	got := preprocessSQL(in)
+	want := "SELECT docnm_kwd FROM ragflow_t1 WHERE count > 0 "
+	if got != want {
+		t.Errorf("preprocessSQL(%q) = %q, want %q", in, got, want)
+	}
+}
+
+// -----------------------------------------------------------------------------
+// rewriteFieldAliases — mirrors infinity_conn_base.py:809-830.
+// -----------------------------------------------------------------------------
+
+func TestRewriteFieldAliases_SelectClause(t *testing.T) {
+	aliases := map[string]string{
+		"docnm_kwd":    "docnm",
+		"title_tks":    "docnm",
+		"title_sm_tks": "docnm",
+		"content_ltks": "content",
+	}
+	in := "select docnm_kwd, title_tks, content_ltks from ragflow_t1"
+	got := rewriteFieldAliases(in, aliases)
+	want := "select docnm, docnm, content from ragflow_t1"
+	if got != want {
+		t.Errorf("rewriteFieldAliases(%q) = %q, want %q", in, got, want)
+	}
+}
+
+func TestRewriteFieldAliases_WhereClause(t *testing.T) {
+	aliases := map[string]string{
+		"docnm_kwd": "docnm",
+	}
+	in := "select doc_id from ragflow_t1 where docnm_kwd = 'foo'"
+	got := rewriteFieldAliases(in, aliases)
+	want := "select doc_id from ragflow_t1 where docnm = 'foo'"
+	if got != want {
+		t.Errorf("rewriteFieldAliases(%q) = %q, want %q", in, got, want)
+	}
+}
+
+func TestRewriteFieldAliases_OrderGroupHaving(t *testing.T) {
+	aliases := map[string]string{
+		"docnm_kwd":     "docnm",
+		"important_kwd": "important_keywords",
+	}
+	in := "select doc_id from ragflow_t1 order by docnm_kwd group by important_kwd having important_kwd > 0"
+	got := rewriteFieldAliases(in, aliases)
+	want := "select doc_id from ragflow_t1 order by docnm group by important_keywords having important_keywords > 0"
+	if got != want {
+		t.Errorf("rewriteFieldAliases(%q) = %q, want %q", in, got, want)
+	}
+}
+
+func TestRewriteFieldAliases_EmptyMapIsNoop(t *testing.T) {
+	in := "select docnm_kwd from ragflow_t1"
+	if got := rewriteFieldAliases(in, map[string]string{}); got != in {
+		t.Errorf("empty alias map should not modify SQL; got %q", got)
+	}
+}
+
+func TestRewriteFieldAliases_WordBoundaryProtected(t *testing.T) {
+	// "title" is an alias; "title_sm_tks" should NOT match because
+	// word boundary is enforced.
+	aliases := map[string]string{
+		"title": "docnm",
+	}
+	in := "select title_sm_tks from ragflow_t1"
+	got := rewriteFieldAliases(in, aliases)
+	// "title" inside "title_sm_tks" should NOT be rewritten.
+	want := "select title_sm_tks from ragflow_t1"
+	if got != want {
+		t.Errorf("rewriteFieldAliases(%q) = %q, want %q (title_sm_tks must NOT be touched)", in, got, want)
+	}
+}
+
+func TestRewriteFieldAliases_NoAliasMatchLeavesSQLAlone(t *testing.T) {
+	aliases := map[string]string{
+		"docnm_kwd": "docnm",
+	}
+	in := "select content_with_weight from ragflow_t1"
+	got := rewriteFieldAliases(in, aliases)
+	if got != in {
+		t.Errorf("unrelated SQL should be unchanged; got %q", got)
+	}
+}
+
+// -----------------------------------------------------------------------------
+// parsePsqlTable — mirrors infinity_conn_base.py:894-934.
+// -----------------------------------------------------------------------------
+
+func TestParsePsqlTable_StandardOutput(t *testing.T) {
+	// Sample psql table output for `select 1 as a, 2 as b;`
+	out := ` a | b
+---+---
+ 1 | 2
+(1 row)`
+
+	res := parsePsqlTable(out)
+	wantCols := []string{"a", "b"}
+	if !reflect.DeepEqual(res.Columns, wantCols) {
+		t.Errorf("columns: got %v, want %v", res.Columns, wantCols)
+	}
+	wantRows := [][]string{{"1", "2"}}
+	if !reflect.DeepEqual(res.Rows, wantRows) {
+		t.Errorf("rows: got %v, want %v", res.Rows, wantRows)
+	}
+}
+
+func TestParsePsqlTable_EmptyOutput(t *testing.T) {
+	res := parsePsqlTable("")
+	if len(res.Columns) != 0 || len(res.Rows) != 0 {
+		t.Errorf("empty output should yield (0 cols, 0 rows); got %+v", res)
+	}
+}
+
+func TestParsePsqlTable_NoSeparatorLine(t *testing.T) {
+	// Some psql configurations skip the separator line; the parser
+	// should still recover (data starts at line 1 in that case).
+	out := "a | b\n1 | 2"
+	res := parsePsqlTable(out)
+	if len(res.Rows) != 1 {
+		t.Errorf("rows: got %d, want 1", len(res.Rows))
+	}
+}
+
+func TestParsePsqlTable_MultipleRowsAndRowCountFooter(t *testing.T) {
+	out := ` id | name
+----+------
+  1 | foo
+  2 | bar
+(2 rows)`
+	res := parsePsqlTable(out)
+	wantCols := []string{"id", "name"}
+	if !reflect.DeepEqual(res.Columns, wantCols) {
+		t.Errorf("columns: got %v, want %v", res.Columns, wantCols)
+	}
+	if len(res.Rows) != 2 {
+		t.Errorf("rows: got %d, want 2", len(res.Rows))
+	}
+	if res.Rows[0][0] != "1" || res.Rows[0][1] != "foo" {
+		t.Errorf("row[0]: got %v, want [1 foo]", res.Rows[0])
+	}
+	if res.Rows[1][0] != "2" || res.Rows[1][1] != "bar" {
+		t.Errorf("row[1]: got %v, want [2 bar]", res.Rows[1])
+	}
+}
+
+func TestParsePsqlTable_PadsAndTruncatesRows(t *testing.T) {
+	// Row with fewer cells → pad with empty strings.
+	// Row with more cells → truncate.
+	out := ` a | b | c
+---+---+---
+ 1 | 2
+ 1 | 2 | 3 | 4
+(2 rows)`
+	res := parsePsqlTable(out)
+	if len(res.Rows) != 2 {
+		t.Fatalf("rows: got %d, want 2", len(res.Rows))
+	}
+	// First row: ["1", "2", ""] (padded)
+	if !reflect.DeepEqual(res.Rows[0], []string{"1", "2", ""}) {
+		t.Errorf("padded row: got %v, want [1 2 ]", res.Rows[0])
+	}
+	// Second row: ["1", "2", "3"] (truncated)
+	if !reflect.DeepEqual(res.Rows[1], []string{"1", "2", "3"}) {
+		t.Errorf("truncated row: got %v, want [1 2 3]", res.Rows[1])
+	}
+}
+
+func TestParsePsqlTable_SkipsRowCountFooter(t *testing.T) {
+	out := " a \n---\n 1 \n(1 row)"
+	res := parsePsqlTable(out)
+	if len(res.Rows) != 1 {
+		t.Errorf("row count footer should be skipped; got %d rows", len(res.Rows))
+	}
+}
+
+// -----------------------------------------------------------------------------
+// toRowMaps — chunk-shape conversion.
+// -----------------------------------------------------------------------------
+
+func TestToRowMaps_EmptyResultsReturnsNil(t *testing.T) {
+	if rows := toRowMaps(nil); rows != nil {
+		t.Errorf("nil result: got %v, want nil", rows)
+	}
+	if rows := toRowMaps(&psqlResult{}); rows != nil {
+		t.Errorf("empty result: got %v, want nil", rows)
+	}
+}
+
+func TestToRowMaps_ConvertsToRowMaps(t *testing.T) {
+	res := &psqlResult{
+		Columns: []string{"id", "name"},
+		Rows: [][]string{
+			{"1", "foo"},
+			{"2", "bar"},
+		},
+	}
+	got := toRowMaps(res)
+	want := []map[string]interface{}{
+		{"id": "1", "name": "foo"},
+		{"id": "2", "name": "bar"},
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("toRowMaps: got %v, want %v", got, want)
+	}
+}
+
+// -----------------------------------------------------------------------------
+// resolvePsqlHostPort — mirrors infinity_conn_base.py:838-858.
+// -----------------------------------------------------------------------------
+
+func TestResolvePsqlHostPort_DefaultsWhenConfigEmpty(t *testing.T) {
+	host, port := resolvePsqlHostPort("", 0)
+	if host != defaultPsqlHost {
+		t.Errorf("host: got %q, want %q", host, defaultPsqlHost)
+	}
+	if port != defaultPsqlPort {
+		t.Errorf("port: got %q, want %q", port, defaultPsqlPort)
+	}
+}
+
+func TestResolvePsqlHostPort_OverridesFromConfig(t *testing.T) {
+	host, port := resolvePsqlHostPort("10.0.0.1:23817", 5433)
+	if host != "10.0.0.1" {
+		t.Errorf("host: got %q, want 10.0.0.1", host)
+	}
+	if port != "5433" {
+		t.Errorf("port: got %q, want 5433", port)
+	}
+}
+
+func TestResolvePsqlHostPort_EmptyHostInURIFallsBackToDefault(t *testing.T) {
+	// ":23817" parses via strings.Cut to ("", "23817") — the empty
+	// host doesn't override the default, matching Python's
+	// `re.search(r"host=(\S+)", ...)` which only matches a non-empty
+	// value.
+	host, port := resolvePsqlHostPort(":23817", 5432)
+	if host != defaultPsqlHost {
+		t.Errorf("host: got %q, want default %q (empty host in URI should not override)", host, defaultPsqlHost)
+	}
+	if port != "5432" {
+		t.Errorf("port: got %q, want 5432", port)
+	}
+}
+
+// -----------------------------------------------------------------------------
+// loadFieldMapping — mirrors infinity_conn_base.py:793-807.
+// -----------------------------------------------------------------------------
+
+func TestLoadFieldMapping_MissingFileReturnsEmpty(t *testing.T) {
+	// Use a name that doesn't exist; the function should silently
+	// return empty maps (matching Python's `os.path.exists` guard).
+	a2a, r2a, err := loadFieldMapping("nonexistent_mapping_xyz.json")
+	if err != nil {
+		t.Fatalf("missing file should be a no-op, got error: %v", err)
+	}
+	if len(a2a) != 0 || len(r2a) != 0 {
+		t.Errorf("missing file should yield empty maps; got a2a=%v r2a=%v", a2a, r2a)
+	}
+}
+
+func TestLoadFieldMapping_ParsesAliases(t *testing.T) {
+	// Write a temporary mapping file.
+	dir := t.TempDir()
+	mappingPath := filepath.Join(dir, "test_mapping.json")
+	contents := `{
+		"docnm": {"type": "varchar", "comment": "docnm_kwd, title_tks, title_sm_tks"},
+		"content": {"type": "varchar", "comment": "content_with_weight, content_ltks"},
+		"plain": {"type": "varchar"}
+	}`
+	if err := os.WriteFile(mappingPath, []byte(contents), 0o644); err != nil {
+		t.Fatalf("write mapping: %v", err)
+	}
+
+	// Set RAG_PROJECT_BASE to the temp dir's parent so loadFieldMapping
+	// finds the file at <base>/conf/<filename>.
+	os.Setenv("RAG_PROJECT_BASE", dir)
+	defer os.Unsetenv("RAG_PROJECT_BASE")
+
+	// Need to create conf/ subdir.
+	if err := os.MkdirAll(filepath.Join(dir, "conf"), 0o755); err != nil {
+		t.Fatalf("mkdir conf: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, "conf", "test_mapping.json"), []byte(contents), 0o644); err != nil {
+		t.Fatalf("write conf/mapping: %v", err)
+	}
+
+	a2a, r2a, err := loadFieldMapping("test_mapping.json")
+	if err != nil {
+		t.Fatalf("loadFieldMapping: %v", err)
+	}
+
+	// alias → actual
+	expectedAliases := map[string]string{
+		"docnm_kwd":           "docnm",
+		"title_tks":           "docnm",
+		"title_sm_tks":        "docnm",
+		"content_with_weight": "content",
+		"content_ltks":        "content",
+	}
+	if !reflect.DeepEqual(a2a, expectedAliases) {
+		t.Errorf("aliasToActual: got %v, want %v", a2a, expectedAliases)
+	}
+
+	// actual → first alias (mirrors Python at line 807)
+	if r2a["docnm"] != "docnm_kwd" {
+		t.Errorf("actualToFirstAlias[docnm]: got %q, want docnm_kwd", r2a["docnm"])
+	}
+	if r2a["content"] != "content_with_weight" {
+		t.Errorf("actualToFirstAlias[content]: got %q, want content_with_weight", r2a["content"])
+	}
+	// "plain" has no comment, so it shouldn't appear in the reverse map.
+	if _, ok := r2a["plain"]; ok {
+		t.Errorf("actualToFirstAlias should not include fields without comments")
+	}
+}
+
+func TestLoadFieldMapping_EmptyNameDefaultsToInfinityMappingJSON(t *testing.T) {
+	// Empty name → defaults to "infinity_mapping.json" (line 145).
+	// We just verify the function doesn't panic and the file-not-found
+	// path is taken silently.
+	a2a, r2a, err := loadFieldMapping("")
+	if err != nil {
+		t.Fatalf("empty name: %v", err)
+	}
+	if len(a2a) != 0 || len(r2a) != 0 {
+		t.Errorf("empty name + no file should yield empty maps; got a2a=%v r2a=%v", a2a, r2a)
+	}
+}