🐛 fix(connection): 修复多数据源连接打开异常

- 修复 Milvus 等数据源类型被错误回退为 MySQL
- 为 ClickHouse 22.8 增加旧版 JSON HTTP 兼容回退
- 使用轻量别名接口验证 Elasticsearch 索引枚举
- 更新可选驱动 revision 并补充回归测试

internal/db/clickhouse_impl.go

@@ -6,6 +6,7 @@ import (
 	"bytes"
 	"context"
 	"database/sql"
+	"database/sql/driver"
 	"fmt"
 	"io"
 	"net"
@@ -41,6 +42,7 @@ const (
 
 type ClickHouseDB struct {
 	conn        *sql.DB
+	legacyHTTP  *clickHouseLegacyHTTPClient
 	pingTimeout time.Duration
 	forwarder   *ssh.LocalForwarder
 	database    string
@@ -701,6 +703,10 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error {
 		_ = c.conn.Close()
 		c.conn = nil
 	}
+	if c.legacyHTTP != nil {
+		_ = c.legacyHTTP.Close()
+		c.legacyHTTP = nil
+	}
 
 	runConfig := normalizeClickHouseConfig(config)
 	c.pingTimeout = getConnectTimeout(runConfig)
@@ -785,6 +791,20 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error {
 						}
 						continue
 					}
+					if protocol == clickhouse.HTTP && stripHTTPClientProtocolVersion {
+						legacyClient, legacyErr := c.connectClickHouseLegacyHTTP(opts)
+						if legacyErr == nil {
+							c.legacyHTTP = legacyClient
+							protocolSuccess = true
+							logger.Warnf("ClickHouse HTTP 兼容握手无法解码旧版 Native block，已切换 legacy JSON HTTP 模式")
+							break
+						}
+						lastProtocolErr = legacyErr
+						legacyFailure := sanitizeClickHouseErrorMessage(legacyErr)
+						failures = append(failures, clickHouseAttemptValidationFailedMessage(idx+1, "legacy-http", legacyFailure))
+						logger.Warnf("ClickHouse legacy JSON HTTP 连接尝试失败：第%d组/%d 地址=%s:%d SSL=%t 原因=%s",
+							idx+1, len(attempts), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL, legacyFailure)
+					}
 					break
 				}
 				protocolSuccess = true
@@ -815,6 +835,24 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error {
 	return fmt.Errorf("%s", clickHouseConnectFailureSummary(runConfig, failures))
 }
 
+func (c *ClickHouseDB) connectClickHouseLegacyHTTP(opts *clickhouse.Options) (*clickHouseLegacyHTTPClient, error) {
+	legacyClient, err := newClickHouseLegacyHTTPClient(opts)
+	if err != nil {
+		return nil, err
+	}
+	timeout := c.pingTimeout
+	if timeout <= 0 {
+		timeout = 5 * time.Second
+	}
+	ctx, cancel := utils.ContextWithTimeout(timeout)
+	defer cancel()
+	if err := legacyClient.Ping(ctx); err != nil {
+		_ = legacyClient.Close()
+		return nil, err
+	}
+	return legacyClient, nil
+}
+
 func (c *ClickHouseDB) Close() error {
 	if c.forwarder != nil {
 		if err := c.forwarder.Close(); err != nil {
@@ -823,12 +861,32 @@ func (c *ClickHouseDB) Close() error {
 		c.forwarder = nil
 	}
 	if c.conn != nil {
-		return c.conn.Close()
+		err := c.conn.Close()
+		c.conn = nil
+		if err != nil {
+			return err
+		}
+	}
+	if c.legacyHTTP != nil {
+		err := c.legacyHTTP.Close()
+		c.legacyHTTP = nil
+		if err != nil {
+			return err
+		}
 	}
 	return nil
 }
 
 func (c *ClickHouseDB) Ping() error {
+	if c.legacyHTTP != nil {
+		timeout := c.pingTimeout
+		if timeout <= 0 {
+			timeout = 5 * time.Second
+		}
+		ctx, cancel := utils.ContextWithTimeout(timeout)
+		defer cancel()
+		return c.legacyHTTP.Ping(ctx)
+	}
 	if c.conn == nil {
 		return fmt.Errorf("连接未打开")
 	}
@@ -879,6 +937,9 @@ func (c *ClickHouseDB) validateQueryPath() error {
 }
 
 func (c *ClickHouseDB) QueryContext(ctx context.Context, query string) ([]map[string]interface{}, []string, error) {
+	if c.legacyHTTP != nil {
+		return c.legacyHTTP.Query(ctx, query)
+	}
 	if c.conn == nil {
 		return nil, nil, fmt.Errorf("连接未打开")
 	}
@@ -891,6 +952,9 @@ func (c *ClickHouseDB) QueryContext(ctx context.Context, query string) ([]map[st
 }
 
 func (c *ClickHouseDB) Query(query string) ([]map[string]interface{}, []string, error) {
+	if c.legacyHTTP != nil {
+		return c.legacyHTTP.Query(context.Background(), query)
+	}
 	if c.conn == nil {
 		return nil, nil, fmt.Errorf("连接未打开")
 	}
@@ -903,6 +967,9 @@ func (c *ClickHouseDB) Query(query string) ([]map[string]interface{}, []string,
 }
 
 func (c *ClickHouseDB) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if c.legacyHTTP != nil {
+		return c.legacyHTTP.StreamQuery(ctx, query, consumer)
+	}
 	if c.conn == nil {
 		return fmt.Errorf("连接未打开")
 	}
@@ -919,6 +986,9 @@ func (c *ClickHouseDB) StreamQuery(query string, consumer QueryStreamConsumer) e
 }
 
 func (c *ClickHouseDB) ExecContext(ctx context.Context, query string) (int64, error) {
+	if c.legacyHTTP != nil {
+		return c.legacyHTTP.Exec(ctx, query)
+	}
 	if c.conn == nil {
 		return 0, fmt.Errorf("连接未打开")
 	}
@@ -930,6 +1000,9 @@ func (c *ClickHouseDB) ExecContext(ctx context.Context, query string) (int64, er
 }
 
 func (c *ClickHouseDB) Exec(query string) (int64, error) {
+	if c.legacyHTTP != nil {
+		return c.legacyHTTP.Exec(context.Background(), query)
+	}
 	if c.conn == nil {
 		return 0, fmt.Errorf("连接未打开")
 	}
@@ -1333,7 +1406,7 @@ func isClickHouseTruthy(value interface{}) bool {
 }
 
 func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
-	if c.conn == nil {
+	if c.conn == nil && c.legacyHTTP == nil {
 		return fmt.Errorf("连接未打开")
 	}
 
@@ -1349,7 +1422,7 @@ func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeS
 			continue
 		}
 		query := fmt.Sprintf("ALTER TABLE %s DELETE WHERE %s", qualifiedTable, whereExpr)
-		if _, err := c.conn.Exec(query); err != nil {
+		if _, err := c.Exec(query); err != nil {
 			return localizedDatabaseRuntimeError("db.backend.error.clickhouse_delete_failed_with_sql", map[string]any{
 				"detail": err.Error(),
 				"sql":    query,
@@ -1364,7 +1437,7 @@ func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeS
 			continue
 		}
 		query := fmt.Sprintf("ALTER TABLE %s UPDATE %s WHERE %s", qualifiedTable, setExpr, whereExpr)
-		if _, err := c.conn.Exec(query); err != nil {
+		if _, err := c.Exec(query); err != nil {
 			return localizedDatabaseRuntimeError("db.backend.error.clickhouse_update_failed_with_sql", map[string]any{
 				"detail": err.Error(),
 				"sql":    query,
@@ -1372,14 +1445,14 @@ func (c *ClickHouseDB) ApplyChanges(tableName string, changes connection.ChangeS
 		}
 	}
 
-	if err := execClickHouseInsertBatches(c.conn, qualifiedTable, changes.Inserts); err != nil {
+	if err := execClickHouseInsertBatches(c.Exec, qualifiedTable, changes.Inserts); err != nil {
 		return err
 	}
 	return nil
 }
 
-func execClickHouseInsertBatches(conn *sql.DB, qualifiedTable string, rows []map[string]interface{}) error {
-	if conn == nil {
+func execClickHouseInsertBatches(exec func(string) (int64, error), qualifiedTable string, rows []map[string]interface{}) error {
+	if exec == nil {
 		return fmt.Errorf("连接未打开")
 	}
 	return execLiteralInsertBatches(literalInsertConfig{
@@ -1388,7 +1461,8 @@ func execClickHouseInsertBatches(conn *sql.DB, qualifiedTable string, rows []map
 		QuoteColumn: quoteClickHouseIdentifier,
 		Literal:     clickHouseLiteral,
 		Exec: func(query string) (sql.Result, error) {
-			return conn.Exec(query)
+			affected, err := exec(query)
+			return driver.RowsAffected(affected), err
 		},
 	})
 }

internal/db/clickhouse_impl_test.go

@@ -8,8 +8,13 @@ import (
 	"database/sql/driver"
 	"errors"
 	"io"
+	"net"
 	"net/http"
+	"net/http/httptest"
+	"net/url"
 	"os"
+	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -19,7 +24,10 @@ import (
 	"GoNavi-Wails/internal/connection"
 	"GoNavi-Wails/shared/i18n"
 
+	chproto "github.com/ClickHouse/ch-go/proto"
 	clickhouse "github.com/ClickHouse/clickhouse-go/v2"
+	clickhousecolumn "github.com/ClickHouse/clickhouse-go/v2/lib/column"
+	clickhouseproto "github.com/ClickHouse/clickhouse-go/v2/lib/proto"
 )
 
 const fakeClickHouseDriverName = "gonavi-fake-clickhouse"
@@ -889,6 +897,143 @@ func TestClickHouseHTTPCompatibilityStripperLeavesOtherBodiesUnchanged(t *testin
 	}
 }
 
+func TestClickHouseConnectFallsBackToLegacyHTTPForRevisionZeroServer(t *testing.T) {
+	installClickHouseRuntimeMarkerForTest(t)
+
+	var (
+		mu                       sync.Mutex
+		sawProtocolVersion       bool
+		sawRevisionZeroHandshake bool
+		sawLegacyJSONQuery       bool
+	)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Errorf("read request body: %v", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		mu.Lock()
+		defer mu.Unlock()
+		if r.URL.Query().Get("client_protocol_version") != "" {
+			sawProtocolVersion = true
+			w.WriteHeader(http.StatusNotFound)
+			_, _ = io.WriteString(w, "Code: 115. DB::Exception: Unknown setting client_protocol_version. (UNKNOWN_SETTING)")
+			return
+		}
+
+		switch strings.TrimSpace(string(body)) {
+		case clickHouseServerHelloCompatQuery:
+			sawRevisionZeroHandshake = true
+			writeClickHouseRevisionZeroHelloBlock(t, w)
+		case "SELECT currentDatabase()":
+			if r.URL.Query().Get("default_format") != "JSONCompactEachRowWithNamesAndTypes" {
+				t.Errorf("legacy query format = %q", r.URL.Query().Get("default_format"))
+				w.WriteHeader(http.StatusBadRequest)
+				return
+			}
+			sawLegacyJSONQuery = true
+			_, _ = io.WriteString(w, "[\"currentDatabase()\"]\n[\"String\"]\n[\"default\"]\n")
+		default:
+			t.Errorf("unexpected ClickHouse query: %q", string(body))
+			w.WriteHeader(http.StatusBadRequest)
+		}
+	}))
+	defer server.Close()
+
+	serverURL, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatalf("parse ClickHouse test server URL: %v", err)
+	}
+	host, portText, err := net.SplitHostPort(serverURL.Host)
+	if err != nil {
+		t.Fatalf("split ClickHouse test server address: %v", err)
+	}
+	port, err := strconv.Atoi(portText)
+	if err != nil {
+		t.Fatalf("parse ClickHouse test server port: %v", err)
+	}
+	client := &ClickHouseDB{}
+	err = client.Connect(connection.ConnectionConfig{
+		Type:               "clickhouse",
+		Host:               host,
+		Port:               port,
+		Database:           "default",
+		User:               "default",
+		ClickHouseProtocol: clickHouseProtocolHTTP,
+		Timeout:            2,
+	})
+	if err != nil {
+		t.Fatalf("Connect should fall back to legacy HTTP for ClickHouse 22.8: %v", err)
+	}
+	defer client.Close()
+
+	mu.Lock()
+	defer mu.Unlock()
+	if !sawProtocolVersion || !sawRevisionZeroHandshake || !sawLegacyJSONQuery {
+		t.Fatalf(
+			"expected modern, revision-zero, and legacy requests; modern=%t revisionZero=%t legacy=%t",
+			sawProtocolVersion,
+			sawRevisionZeroHandshake,
+			sawLegacyJSONQuery,
+		)
+	}
+}
+
+func installClickHouseRuntimeMarkerForTest(t *testing.T) {
+	t.Helper()
+	previousRoot := currentExternalDriverDownloadDirectory()
+	t.Cleanup(func() { SetExternalDriverDownloadDirectory(previousRoot) })
+
+	root := t.TempDir()
+	SetExternalDriverDownloadDirectory(root)
+	markerPath, err := ResolveOptionalGoDriverMarkerPath(root, "clickhouse")
+	if err != nil {
+		t.Fatalf("resolve ClickHouse marker path: %v", err)
+	}
+	if err := os.MkdirAll(filepath.Dir(markerPath), 0o755); err != nil {
+		t.Fatalf("create ClickHouse marker directory: %v", err)
+	}
+	if err := os.WriteFile(markerPath, []byte("{}"), 0o644); err != nil {
+		t.Fatalf("write ClickHouse marker: %v", err)
+	}
+	executablePath, err := ResolveOptionalDriverAgentExecutablePath(root, "clickhouse")
+	if err != nil {
+		t.Fatalf("resolve ClickHouse agent path: %v", err)
+	}
+	if err := os.WriteFile(executablePath, []byte("test agent placeholder"), 0o755); err != nil {
+		t.Fatalf("write ClickHouse agent placeholder: %v", err)
+	}
+}
+
+func writeClickHouseRevisionZeroHelloBlock(t *testing.T, w http.ResponseWriter) {
+	t.Helper()
+	block := clickhouseproto.NewBlock()
+	for _, column := range []struct {
+		name     string
+		typeName clickhousecolumn.Type
+	}{
+		{name: "hostName()", typeName: "String"},
+		{name: "version()", typeName: "String"},
+		{name: "revision()", typeName: "UInt64"},
+		{name: "timezone()", typeName: "String"},
+	} {
+		if err := block.AddColumn(column.name, column.typeName); err != nil {
+			t.Fatalf("add ClickHouse block column: %v", err)
+		}
+	}
+	if err := block.Append("legacy-clickhouse", "22.8.20.11", uint64(54460), "UTC"); err != nil {
+		t.Fatalf("append ClickHouse block row: %v", err)
+	}
+	buffer := &chproto.Buffer{}
+	if err := block.Encode(buffer, 0); err != nil {
+		t.Fatalf("encode revision-zero ClickHouse block: %v", err)
+	}
+	w.Header().Set("Content-Type", "application/octet-stream")
+	_, _ = w.Write(buffer.Buf)
+}
+
 func TestWithClickHouseProtocolForcesProtocolSelection(t *testing.T) {
 	httpConfig := withClickHouseProtocol(connection.ConnectionConfig{
 		Type: "clickhouse",

internal/db/clickhouse_legacy_http.go

@@ -0,0 +1,322 @@
+//go:build gonavi_full_drivers || gonavi_clickhouse_driver
+
+package db
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+
+	clickhouse "github.com/ClickHouse/clickhouse-go/v2"
+)
+
+const (
+	clickHouseLegacyHTTPFormat       = "JSONCompactEachRowWithNamesAndTypes"
+	clickHouseLegacyHTTPErrorLimit   = 64 << 10
+	clickHouseLegacyHTTPUserAgent    = "GoNavi ClickHouse legacy HTTP client"
+	clickHouseLegacyHTTPDatabaseName = "database"
+)
+
+// clickHouseLegacyHTTPClient is used only after the current clickhouse-go HTTP
+// handshake proves that the server predates client_protocol_version support.
+// JSON avoids decoding Native blocks with a wire revision the driver cannot
+// negotiate with those servers.
+type clickHouseLegacyHTTPClient struct {
+	endpoint  *url.URL
+	http      *http.Client
+	transport *http.Transport
+	username  string
+	password  string
+	headers   http.Header
+	params    url.Values
+}
+
+func newClickHouseLegacyHTTPClient(opts *clickhouse.Options) (*clickHouseLegacyHTTPClient, error) {
+	if opts == nil {
+		return nil, fmt.Errorf("ClickHouse legacy HTTP options are required")
+	}
+	if len(opts.Addr) == 0 || strings.TrimSpace(opts.Addr[0]) == "" {
+		return nil, fmt.Errorf("ClickHouse legacy HTTP address is required")
+	}
+
+	scheme := "http"
+	if opts.TLS != nil {
+		scheme = "https"
+	}
+	path := strings.TrimSpace(opts.HttpUrlPath)
+	if path != "" && !strings.HasPrefix(path, "/") {
+		path = "/" + path
+	}
+	endpoint := &url.URL{
+		Scheme: scheme,
+		Host:   strings.TrimSpace(opts.Addr[0]),
+		Path:   path,
+	}
+
+	proxy := http.ProxyFromEnvironment
+	if opts.HTTPProxyURL != nil {
+		proxy = http.ProxyURL(opts.HTTPProxyURL)
+	}
+	transport := &http.Transport{
+		Proxy:                 proxy,
+		DialContext:           (&net.Dialer{Timeout: opts.DialTimeout}).DialContext,
+		MaxIdleConns:          1,
+		MaxConnsPerHost:       opts.HttpMaxConnsPerHost,
+		IdleConnTimeout:       opts.ConnMaxLifetime,
+		ResponseHeaderTimeout: opts.ReadTimeout,
+		TLSClientConfig:       opts.TLS,
+		DisableCompression:    true,
+	}
+	if opts.DialContext != nil {
+		transport.DialContext = func(ctx context.Context, _, address string) (net.Conn, error) {
+			return opts.DialContext(ctx, address)
+		}
+	}
+
+	params := make(url.Values, len(opts.Settings)+1)
+	if database := strings.TrimSpace(opts.Auth.Database); database != "" {
+		params.Set(clickHouseLegacyHTTPDatabaseName, database)
+	}
+	for key, value := range opts.Settings {
+		key = strings.TrimSpace(key)
+		if key == "" || strings.EqualFold(key, "default_format") || strings.EqualFold(key, "client_protocol_version") {
+			continue
+		}
+		if custom, ok := value.(clickhouse.CustomSetting); ok {
+			value = custom.Value
+		}
+		params.Set(key, fmt.Sprint(value))
+	}
+
+	headers := make(http.Header, len(opts.HttpHeaders)+2)
+	for key, value := range opts.HttpHeaders {
+		headers.Set(key, value)
+	}
+	headers.Set("User-Agent", clickHouseLegacyHTTPUserAgent)
+	headers.Set("Content-Type", "text/plain; charset=utf-8")
+
+	return &clickHouseLegacyHTTPClient{
+		endpoint: endpoint,
+		http: &http.Client{
+			Transport: transport,
+		},
+		transport: transport,
+		username:  opts.Auth.Username,
+		password:  opts.Auth.Password,
+		headers:   headers,
+		params:    params,
+	}, nil
+}
+
+func (c *clickHouseLegacyHTTPClient) Close() error {
+	if c != nil && c.transport != nil {
+		c.transport.CloseIdleConnections()
+	}
+	return nil
+}
+
+func (c *clickHouseLegacyHTTPClient) Ping(ctx context.Context) error {
+	rows, _, err := c.Query(ctx, "SELECT currentDatabase()")
+	if err != nil {
+		return err
+	}
+	if len(rows) == 0 {
+		return fmt.Errorf("ClickHouse legacy HTTP validation returned no rows")
+	}
+	return nil
+}
+
+func (c *clickHouseLegacyHTTPClient) Query(ctx context.Context, query string) ([]map[string]interface{}, []string, error) {
+	collector := &clickHouseLegacyHTTPCollector{}
+	if err := c.StreamQuery(ctx, query, collector); err != nil {
+		return collector.rows, collector.columns, err
+	}
+	return collector.rows, collector.columns, nil
+}
+
+func (c *clickHouseLegacyHTTPClient) StreamQuery(ctx context.Context, query string, consumer QueryStreamConsumer) error {
+	if consumer == nil {
+		return fmt.Errorf("query stream consumer required")
+	}
+	response, err := c.do(ctx, query, false)
+	if err != nil {
+		return err
+	}
+	defer response.Body.Close()
+
+	decoder := json.NewDecoder(response.Body)
+	decoder.UseNumber()
+
+	var columns []string
+	if err := decoder.Decode(&columns); err != nil {
+		return c.decodeError(decoder, response.Body, "column names", err)
+	}
+	if len(columns) == 0 {
+		return fmt.Errorf("ClickHouse legacy HTTP response has no columns")
+	}
+	columns = ensureUniqueQueryColumnNames(columns)
+
+	var typeNames []string
+	if err := decoder.Decode(&typeNames); err != nil {
+		return c.decodeError(decoder, response.Body, "column types", err)
+	}
+	if len(typeNames) != len(columns) {
+		return fmt.Errorf("ClickHouse legacy HTTP column metadata mismatch: names=%d types=%d", len(columns), len(typeNames))
+	}
+	if err := consumer.SetColumns(columns); err != nil {
+		return err
+	}
+	valueConsumer, useValueConsumer := consumer.(QueryStreamValueConsumer)
+
+	for {
+		var values []interface{}
+		err := decoder.Decode(&values)
+		if errors.Is(err, io.EOF) {
+			return nil
+		}
+		if err != nil {
+			return c.decodeError(decoder, response.Body, "row", err)
+		}
+		if len(values) != len(columns) {
+			return fmt.Errorf("ClickHouse legacy HTTP row width mismatch: columns=%d values=%d", len(columns), len(values))
+		}
+		for index := range values {
+			values[index] = normalizeQueryValueWithDBType(values[index], typeNames[index])
+		}
+		if useValueConsumer {
+			if err := valueConsumer.ConsumeRowValues(values); err != nil {
+				return err
+			}
+			continue
+		}
+		row := make(map[string]interface{}, len(columns))
+		for index, column := range columns {
+			row[column] = values[index]
+		}
+		if err := consumer.ConsumeRow(row); err != nil {
+			return err
+		}
+	}
+}
+
+func (c *clickHouseLegacyHTTPClient) Exec(ctx context.Context, query string) (int64, error) {
+	response, err := c.do(ctx, query, true)
+	if err != nil {
+		return 0, err
+	}
+	defer response.Body.Close()
+
+	body, readErr := io.ReadAll(io.LimitReader(response.Body, clickHouseLegacyHTTPErrorLimit+1))
+	if readErr != nil {
+		return 0, readErr
+	}
+	if exception := clickHouseLegacyHTTPException(body); exception != "" {
+		return 0, fmt.Errorf("%s", exception)
+	}
+	// clickhouse-go also reports zero because ClickHouse does not provide a
+	// database/sql affected-row count for ordinary HTTP executions.
+	return 0, nil
+}
+
+func (c *clickHouseLegacyHTTPClient) do(ctx context.Context, query string, waitForEnd bool) (*http.Response, error) {
+	if c == nil || c.endpoint == nil || c.http == nil {
+		return nil, fmt.Errorf("ClickHouse legacy HTTP connection is not open")
+	}
+	requestURL := *c.endpoint
+	params := cloneURLValues(c.params)
+	params.Set("default_format", clickHouseLegacyHTTPFormat)
+	params.Del("client_protocol_version")
+	if waitForEnd {
+		params.Set("wait_end_of_query", "1")
+	}
+	requestURL.RawQuery = params.Encode()
+
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, requestURL.String(), strings.NewReader(query))
+	if err != nil {
+		return nil, err
+	}
+	request.Header = c.headers.Clone()
+	if c.username != "" || c.password != "" {
+		request.SetBasicAuth(c.username, c.password)
+	}
+
+	response, err := c.http.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
+		defer response.Body.Close()
+		body, _ := io.ReadAll(io.LimitReader(response.Body, clickHouseLegacyHTTPErrorLimit))
+		detail := clickHouseLegacyHTTPException(body)
+		if detail == "" {
+			detail = strings.TrimSpace(string(body))
+		}
+		if detail == "" {
+			detail = response.Status
+		}
+		return nil, fmt.Errorf("ClickHouse legacy HTTP request failed: status=%d detail=%s", response.StatusCode, detail)
+	}
+	if code := strings.TrimSpace(response.Header.Get("X-ClickHouse-Exception-Code")); code != "" && code != "0" {
+		defer response.Body.Close()
+		body, _ := io.ReadAll(io.LimitReader(response.Body, clickHouseLegacyHTTPErrorLimit))
+		detail := clickHouseLegacyHTTPException(body)
+		if detail == "" {
+			detail = strings.TrimSpace(string(body))
+		}
+		return nil, fmt.Errorf("ClickHouse legacy HTTP exception code=%s detail=%s", code, detail)
+	}
+	return response, nil
+}
+
+func (c *clickHouseLegacyHTTPClient) decodeError(decoder *json.Decoder, body io.Reader, section string, decodeErr error) error {
+	var tail []byte
+	if decoder != nil {
+		tail, _ = io.ReadAll(io.LimitReader(io.MultiReader(decoder.Buffered(), body), clickHouseLegacyHTTPErrorLimit))
+	}
+	if exception := clickHouseLegacyHTTPException(tail); exception != "" {
+		return fmt.Errorf("%s", exception)
+	}
+	return fmt.Errorf("decode ClickHouse legacy HTTP %s: %w", section, decodeErr)
+}
+
+func clickHouseLegacyHTTPException(raw []byte) string {
+	text := sanitizeClickHouseErrorMessage(errors.New(strings.TrimSpace(string(raw))))
+	if text == "" {
+		return ""
+	}
+	lower := strings.ToLower(text)
+	if strings.Contains(lower, "db::exception") ||
+		(strings.Contains(lower, "code:") && strings.Contains(lower, "exception")) {
+		return text
+	}
+	return ""
+}
+
+func cloneURLValues(source url.Values) url.Values {
+	result := make(url.Values, len(source))
+	for key, values := range source {
+		result[key] = append([]string(nil), values...)
+	}
+	return result
+}
+
+type clickHouseLegacyHTTPCollector struct {
+	columns []string
+	rows    []map[string]interface{}
+}
+
+func (c *clickHouseLegacyHTTPCollector) SetColumns(columns []string) error {
+	c.columns = append([]string(nil), columns...)
+	return nil
+}
+
+func (c *clickHouseLegacyHTTPCollector) ConsumeRow(row map[string]interface{}) error {
+	c.rows = append(c.rows, row)
+	return nil
+}

internal/db/clickhouse_legacy_http_test.go

@@ -0,0 +1,289 @@
+//go:build gonavi_full_drivers || gonavi_clickhouse_driver
+
+package db
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"reflect"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+
+	clickhouse "github.com/ClickHouse/clickhouse-go/v2"
+)
+
+func TestClickHouseLegacyHTTPQueryPreservesValuesAndRequestOptions(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/clickhouse" {
+			t.Errorf("HTTP path = %q", r.URL.Path)
+		}
+		if got := r.URL.Query().Get("default_format"); got != clickHouseLegacyHTTPFormat {
+			t.Errorf("default_format = %q", got)
+		}
+		if got := r.URL.Query().Get("database"); got != "analytics" {
+			t.Errorf("database = %q", got)
+		}
+		if got := r.URL.Query().Get("client_protocol_version"); got != "" {
+			t.Errorf("client_protocol_version should be absent, got %q", got)
+		}
+		if got := r.URL.Query().Get("max_execution_time"); got != "60" {
+			t.Errorf("max_execution_time = %q", got)
+		}
+		username, password, ok := r.BasicAuth()
+		if !ok || username != "reporter" || password != "secret" {
+			t.Errorf("basic auth = (%q, %q, %t)", username, password, ok)
+		}
+		_, _ = io.WriteString(w, strings.Join([]string{
+			`["id","id","nullable","created_at","items","attrs"]`,
+			`["UInt64","UInt8","Nullable(String)","DateTime","Array(UInt64)","Map(String, UInt64)"]`,
+			`[9007199254740993,2,null,"2022-08-03 12:13:14",[1,9007199254740994],{"small":3,"large":9007199254740995}]`,
+			"",
+		}, "\n"))
+	}))
+	defer server.Close()
+
+	client := newClickHouseLegacyHTTPTestClientWithOptions(t, server, func(opts *clickhouse.Options) {
+		opts.Auth = clickhouse.Auth{Database: "analytics", Username: "reporter", Password: "secret"}
+		opts.HttpUrlPath = "clickhouse"
+		opts.Settings = clickhouse.Settings{
+			"max_execution_time":      60,
+			"client_protocol_version": 54485,
+		}
+	})
+
+	rows, columns, err := client.Query(context.Background(), "SELECT values")
+	if err != nil {
+		t.Fatalf("legacy query failed: %v", err)
+	}
+	if !reflect.DeepEqual(columns, []string{"id", "id_2", "nullable", "created_at", "items", "attrs"}) {
+		t.Fatalf("columns = %#v", columns)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("rows = %#v", rows)
+	}
+	row := rows[0]
+	if row["id"] != "9007199254740993" || row["id_2"] != int64(2) || row["nullable"] != nil {
+		t.Fatalf("scalar values = %#v", row)
+	}
+	if !reflect.DeepEqual(row["items"], []interface{}{int64(1), "9007199254740994"}) {
+		t.Fatalf("items = %#v", row["items"])
+	}
+	if !reflect.DeepEqual(row["attrs"], map[string]interface{}{"small": int64(3), "large": "9007199254740995"}) {
+		t.Fatalf("attrs = %#v", row["attrs"])
+	}
+}
+
+func TestClickHouseLegacyHTTPStreamDeliversRowsBeforeEOF(t *testing.T) {
+	firstChunkWritten := make(chan struct{})
+	releaseResponse := make(chan struct{})
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(releaseResponse) }) }
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			t.Error("test server does not support flushing")
+			return
+		}
+		_, _ = io.WriteString(w, "[\"n\"]\n[\"UInt64\"]\n[1]\n")
+		flusher.Flush()
+		close(firstChunkWritten)
+		<-releaseResponse
+		_, _ = io.WriteString(w, "[2]\n")
+	}))
+	defer func() {
+		release()
+		server.Close()
+	}()
+
+	client := newClickHouseLegacyHTTPTestClient(t, server, nil)
+	consumer := &legacyHTTPTestStreamConsumer{firstRow: make(chan struct{})}
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- client.StreamQuery(context.Background(), "SELECT stream", consumer)
+	}()
+
+	select {
+	case <-firstChunkWritten:
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not write the first response chunk")
+	}
+	select {
+	case <-consumer.firstRow:
+	case <-time.After(2 * time.Second):
+		t.Fatal("first row was not delivered before response EOF")
+	}
+	release()
+	if err := <-errCh; err != nil {
+		t.Fatalf("stream query failed: %v", err)
+	}
+	if !reflect.DeepEqual(consumer.columns, []string{"n"}) {
+		t.Fatalf("columns = %#v", consumer.columns)
+	}
+	if !reflect.DeepEqual(consumer.values, [][]interface{}{{int64(1)}, {int64(2)}}) {
+		t.Fatalf("values = %#v", consumer.values)
+	}
+}
+
+func TestClickHouseLegacyHTTPQueryReportsExceptionAppendedAfterRows(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "[\"n\"]\n[\"UInt64\"]\n[1]\nCode: 241. DB::Exception: memory limit exceeded. (MEMORY_LIMIT_EXCEEDED)\n")
+	}))
+	defer server.Close()
+
+	client := newClickHouseLegacyHTTPTestClient(t, server, nil)
+	rows, columns, err := client.Query(context.Background(), "SELECT fails_late")
+	if err == nil || !strings.Contains(err.Error(), "memory limit exceeded") {
+		t.Fatalf("expected trailing ClickHouse exception, got rows=%#v columns=%#v err=%v", rows, columns, err)
+	}
+	if len(rows) != 1 || len(columns) != 1 {
+		t.Fatalf("decoded prefix should remain available, rows=%#v columns=%#v", rows, columns)
+	}
+}
+
+func TestClickHouseLegacyHTTPExecHandlesSuccessAndErrors(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		if got := r.URL.Query().Get("wait_end_of_query"); got != "1" {
+			t.Errorf("wait_end_of_query = %q for %q", got, body)
+		}
+		switch string(body) {
+		case "ALTER OK":
+			w.WriteHeader(http.StatusOK)
+		case "ALTER STATUS ERROR":
+			w.WriteHeader(http.StatusInternalServerError)
+			_, _ = io.WriteString(w, "Code: 60. DB::Exception: table does not exist. (UNKNOWN_TABLE)")
+		case "ALTER TAIL ERROR":
+			w.WriteHeader(http.StatusOK)
+			_, _ = io.WriteString(w, "Code: 241. DB::Exception: memory limit exceeded. (MEMORY_LIMIT_EXCEEDED)")
+		default:
+			w.WriteHeader(http.StatusBadRequest)
+		}
+	}))
+	defer server.Close()
+
+	client := newClickHouseLegacyHTTPTestClient(t, server, nil)
+	if affected, err := client.Exec(context.Background(), "ALTER OK"); err != nil || affected != 0 {
+		t.Fatalf("successful exec = (%d, %v)", affected, err)
+	}
+	if _, err := client.Exec(context.Background(), "ALTER STATUS ERROR"); err == nil || !strings.Contains(err.Error(), "table does not exist") {
+		t.Fatalf("expected HTTP status error, got %v", err)
+	}
+	if _, err := client.Exec(context.Background(), "ALTER TAIL ERROR"); err == nil || !strings.Contains(err.Error(), "memory limit exceeded") {
+		t.Fatalf("expected HTTP 200 body exception, got %v", err)
+	}
+}
+
+func TestClickHouseApplyChangesUsesLegacyHTTPBackend(t *testing.T) {
+	var (
+		mu      sync.Mutex
+		queries []string
+	)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		mu.Lock()
+		queries = append(queries, string(body))
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	legacyClient := newClickHouseLegacyHTTPTestClient(t, server, nil)
+	client := &ClickHouseDB{legacyHTTP: legacyClient, database: "analytics"}
+	err := client.ApplyChanges("events", connection.ChangeSet{
+		Deletes: []map[string]interface{}{{"id": int64(1)}},
+		Updates: []connection.UpdateRow{{
+			Keys:   map[string]interface{}{"id": int64(2)},
+			Values: map[string]interface{}{"name": "updated"},
+		}},
+		Inserts: []map[string]interface{}{{"id": int64(3), "name": "inserted"}},
+	})
+	if err != nil {
+		t.Fatalf("ApplyChanges with legacy HTTP failed: %v", err)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if len(queries) != 3 {
+		t.Fatalf("queries = %#v", queries)
+	}
+	for _, fragment := range []string{"ALTER TABLE `analytics`.`events` DELETE", "ALTER TABLE `analytics`.`events` UPDATE", "INSERT INTO `analytics`.`events`"} {
+		if !containsStringFragment(queries, fragment) {
+			t.Fatalf("missing query fragment %q in %#v", fragment, queries)
+		}
+	}
+}
+
+func newClickHouseLegacyHTTPTestClient(t *testing.T, server *httptest.Server, settings clickhouse.Settings) *clickHouseLegacyHTTPClient {
+	t.Helper()
+	return newClickHouseLegacyHTTPTestClientWithOptions(t, server, func(opts *clickhouse.Options) {
+		opts.Settings = settings
+	})
+}
+
+func newClickHouseLegacyHTTPTestClientWithOptions(t *testing.T, server *httptest.Server, configure func(*clickhouse.Options)) *clickHouseLegacyHTTPClient {
+	t.Helper()
+	parsed, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatalf("parse test server URL: %v", err)
+	}
+	opts := &clickhouse.Options{
+		Protocol:    clickhouse.HTTP,
+		Addr:        []string{parsed.Host},
+		Auth:        clickhouse.Auth{Database: "default"},
+		DialTimeout: 2 * time.Second,
+		ReadTimeout: 2 * time.Second,
+	}
+	if configure != nil {
+		configure(opts)
+	}
+	client, err := newClickHouseLegacyHTTPClient(opts)
+	if err != nil {
+		t.Fatalf("new legacy HTTP client: %v", err)
+	}
+	t.Cleanup(func() { _ = client.Close() })
+	return client
+}
+
+type legacyHTTPTestStreamConsumer struct {
+	mu       sync.Mutex
+	columns  []string
+	values   [][]interface{}
+	firstRow chan struct{}
+	once     sync.Once
+}
+
+func (c *legacyHTTPTestStreamConsumer) SetColumns(columns []string) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.columns = append([]string(nil), columns...)
+	return nil
+}
+
+func (c *legacyHTTPTestStreamConsumer) ConsumeRow(map[string]interface{}) error {
+	return fmt.Errorf("value consumer fast path was not used")
+}
+
+func (c *legacyHTTPTestStreamConsumer) ConsumeRowValues(values []interface{}) error {
+	c.mu.Lock()
+	c.values = append(c.values, append([]interface{}(nil), values...))
+	c.mu.Unlock()
+	c.once.Do(func() { close(c.firstRow) })
+	return nil
+}
+
+func containsStringFragment(values []string, fragment string) bool {
+	for _, value := range values {
+		if strings.Contains(value, fragment) {
+			return true
+		}
+	}
+	return false
+}

internal/db/driver_agent_revisions_gen.go

@@ -22,8 +22,8 @@ func init() {
 		"mongodb":       "src-2610395b35c2e708",
 		"tdengine":      "src-779b9b537f08856f",
 		"iotdb":         "src-7edea4aba8d4869e",
-		"clickhouse":    "src-0197342ca5afa8b5",
-		"elasticsearch": "src-08e8e80cb17a409a",
+		"clickhouse":    "src-5872eb0d28e8ab82",
+		"elasticsearch": "src-85079509f9c31623",
 		"trino":         "src-ba947f211ce7b19f",
 	}
 }

internal/db/elasticsearch_impl.go

@@ -102,6 +102,12 @@ func (e *ElasticsearchDB) Connect(config connection.ConnectionConfig) error {
 			lastErr = err
 			continue
 		}
+		if _, err := e.GetDatabases(); err != nil {
+			e.client = nil
+			logger.Warnf("Elasticsearch 索引枚举验证失败：%d/%d 模式=%s 错误=%v", idx+1, len(attempts), sslLabel, err)
+			lastErr = err
+			continue
+		}
 
 		logger.Infof("Elasticsearch 连接成功：%d/%d 模式=%s", idx+1, len(attempts), sslLabel)
 		if idx > 0 {
@@ -331,9 +337,13 @@ func (e *ElasticsearchDB) GetDatabases() ([]string, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
-	res, err := e.client.Indices.Get(
-		[]string{"*"},
-		e.client.Indices.Get.WithContext(ctx),
+	// Alias API 只返回索引名和别名，避免列表加载下载全部 settings/mappings。
+	res, err := e.client.Indices.GetAlias(
+		e.client.Indices.GetAlias.WithContext(ctx),
+		e.client.Indices.GetAlias.WithIndex("*"),
+		e.client.Indices.GetAlias.WithExpandWildcards("all"),
+		e.client.Indices.GetAlias.WithAllowNoIndices(true),
+		e.client.Indices.GetAlias.WithIgnoreUnavailable(true),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("获取索引列表失败：%w", err)

internal/db/elasticsearch_impl_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"slices"
 	"strings"
+	"sync/atomic"
 	"testing"
 
 	"GoNavi-Wails/internal/connection"
@@ -104,20 +105,101 @@ func TestElasticsearchPing(t *testing.T) {
 	})
 }
 
+func TestElasticsearchConnectValidatesIndexListing(t *testing.T) {
+	var aliasListingRequested atomic.Bool
+	server := newMockESServer(t, func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.Method == http.MethodHead && r.URL.Path == "/":
+			w.WriteHeader(http.StatusOK)
+		case r.Method == http.MethodGet && r.URL.Path == "/*/_alias":
+			aliasListingRequested.Store(true)
+			w.WriteHeader(http.StatusForbidden)
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	})
+
+	host, port, ok := parseHostPortWithDefault(strings.TrimPrefix(server.URL, "http://"), defaultEsPort)
+	if !ok {
+		t.Fatalf("无法解析测试服务器地址：%s", server.URL)
+	}
+
+	db := &ElasticsearchDB{}
+	err := db.Connect(connection.ConnectionConfig{
+		Type:    "elasticsearch",
+		Host:    host,
+		Port:    port,
+		Timeout: 2,
+	})
+	if err == nil {
+		t.Fatal("Connect 应在索引枚举被拒绝时失败")
+	}
+	if !aliasListingRequested.Load() {
+		t.Fatal("Connect 应使用轻量 alias 端点验证索引枚举能力")
+	}
+	if !strings.Contains(err.Error(), "获取索引列表失败") || !strings.Contains(err.Error(), "403") {
+		t.Fatalf("Connect 应返回索引枚举错误，实际：%v", err)
+	}
+}
+
+func TestElasticsearchConnectAllowsEmptyCluster(t *testing.T) {
+	var aliasListingRequested atomic.Bool
+	server := newMockESServer(t, func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.Method == http.MethodHead && r.URL.Path == "/":
+			w.WriteHeader(http.StatusOK)
+		case r.Method == http.MethodGet && r.URL.Path == "/*/_alias":
+			aliasListingRequested.Store(true)
+			query := r.URL.Query()
+			if query.Get("allow_no_indices") != "true" || query.Get("ignore_unavailable") != "true" {
+				w.WriteHeader(http.StatusBadRequest)
+				return
+			}
+			writeJSON(w, map[string]interface{}{})
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	})
+
+	host, port, ok := parseHostPortWithDefault(strings.TrimPrefix(server.URL, "http://"), defaultEsPort)
+	if !ok {
+		t.Fatalf("无法解析测试服务器地址：%s", server.URL)
+	}
+
+	db := &ElasticsearchDB{}
+	if err := db.Connect(connection.ConnectionConfig{
+		Type:    "elasticsearch",
+		Host:    host,
+		Port:    port,
+		Timeout: 2,
+	}); err != nil {
+		t.Fatalf("Connect 应允许没有索引的空集群：%v", err)
+	}
+	if !aliasListingRequested.Load() {
+		t.Fatal("Connect 应验证空集群的索引枚举能力")
+	}
+}
+
 // TestElasticsearchGetDatabases 测试获取索引列表。
 func TestElasticsearchGetDatabases(t *testing.T) {
-	t.Run("正常获取全部索引", func(t *testing.T) {
+	t.Run("使用轻量别名端点获取全部索引", func(t *testing.T) {
+		var fullIndexDefinitionsRequested atomic.Bool
 		server := newMockESServer(t, func(w http.ResponseWriter, r *http.Request) {
-			if r.Method == http.MethodGet && (r.URL.Path == "/" || r.URL.Path == "/*") && !strings.Contains(r.URL.Path, "_") {
+			if r.Method == http.MethodGet && r.URL.Path == "/*/_alias" {
 				writeJSON(w, map[string]interface{}{
-					"logs-2024": map[string]interface{}{},
-					"users":     map[string]interface{}{},
-					".security": map[string]interface{}{},
-					".kibana_1": map[string]interface{}{},
-					"products":  map[string]interface{}{},
+					"logs-2024": map[string]interface{}{"aliases": map[string]interface{}{}},
+					"users":     map[string]interface{}{"aliases": map[string]interface{}{}},
+					".security": map[string]interface{}{"aliases": map[string]interface{}{}},
+					".kibana_1": map[string]interface{}{"aliases": map[string]interface{}{}},
+					"products":  map[string]interface{}{"aliases": map[string]interface{}{}},
 				})
 				return
 			}
+			if r.Method == http.MethodGet && r.URL.Path == "/*" {
+				fullIndexDefinitionsRequested.Store(true)
+				w.WriteHeader(http.StatusForbidden)
+				return
+			}
 			w.WriteHeader(http.StatusNotFound)
 		})
 
@@ -137,6 +219,9 @@ func TestElasticsearchGetDatabases(t *testing.T) {
 				t.Fatalf("索引 [%d] 期望 %q，实际 %q", i, name, databases[i])
 			}
 		}
+		if fullIndexDefinitionsRequested.Load() {
+			t.Fatal("GetDatabases 不应请求包含 settings/mappings 的完整索引定义")
+		}
 	})
 
 	t.Run("连接未打开时返回错误", func(t *testing.T) {
@@ -998,13 +1083,9 @@ func TestESMockIntegration(t *testing.T) {
 		case r.Method == http.MethodHead && path == "/":
 			w.WriteHeader(http.StatusOK)
 
-		// Indices.Get("*") — 返回所有索引
-		case r.Method == http.MethodGet && (path == "/" || path == "/*") && !strings.Contains(path, "_"):
-			writeJSON(w, map[string]interface{}{
-				"products":  map[string]interface{}{},
-				"orders":    map[string]interface{}{},
-				".internal": map[string]interface{}{},
-			})
+		// 完整索引定义响应可能过大，列表加载不应请求该端点。
+		case r.Method == http.MethodGet && path == "/*":
+			w.WriteHeader(http.StatusForbidden)
 
 		// Indices.GetAlias — 返回别名映射
 		case strings.Contains(path, "/_alias") && r.Method == http.MethodGet:
@@ -1014,6 +1095,8 @@ func TestESMockIntegration(t *testing.T) {
 						"products-alias": map[string]interface{}{},
 					},
 				},
+				"orders":    map[string]interface{}{"aliases": map[string]interface{}{}},
+				".internal": map[string]interface{}{"aliases": map[string]interface{}{}},
 			})
 
 		// Mapping