🐛 fix(datasource): 修复 SQL Server 分页与 ClickHouse 22.8 连接兼容

- SQL Server 表数据分页改用旧版本兼容语法，避免 FETCH NEXT 报错 - ClickHouse HTTP 连接支持移除 client_protocol_version 后兼容重试 - 补充 SQL 分页与 ClickHouse 连接兼容回归测试 Refs #479
2026-07-09 14:31:35 +08:00 · 2026-05-23 19:14:40 +08:00
parent 8615265ee1
commit cf0a216329
6 changed files with 347 additions and 42 deletions
--- a/internal/db/clickhouse_impl.go
+++ b/internal/db/clickhouse_impl.go
@@ -7,6 +7,7 @@ import (
 	"database/sql"
 	"fmt"
 	"net"
+	"net/http"
 	"net/url"
 	"sort"
 	"strconv"
@@ -168,6 +169,10 @@ func defaultClickHousePortForScheme(scheme string) int {
 }

 func (c *ClickHouseDB) buildClickHouseOptions(config connection.ConnectionConfig) (*clickhouse.Options, error) {
+	return c.buildClickHouseOptionsWithHTTPCompatibility(config, false)
+}
+
+func (c *ClickHouseDB) buildClickHouseOptionsWithHTTPCompatibility(config connection.ConnectionConfig, stripHTTPClientProtocolVersion bool) (*clickhouse.Options, error) {
 	connectTimeout := getConnectTimeout(config)
 	readTimeout := connectTimeout
 	if readTimeout < minClickHouseReadTimeout {
@@ -195,9 +200,57 @@ func (c *ClickHouseDB) buildClickHouseOptions(config connection.ConnectionConfig
 		opts.TLS = tlsConfig
 	}
 	applyClickHouseConnectionParams(opts, config)
+	if stripHTTPClientProtocolVersion && protocol == clickhouse.HTTP {
+		installClickHouseHTTPClientProtocolVersionStripper(opts)
+	}
 	return opts, nil
 }

+type clickHouseHTTPClientProtocolVersionStripper struct {
+	next http.RoundTripper
+}
+
+func (rt clickHouseHTTPClientProtocolVersionStripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	next := rt.next
+	if next == nil {
+		next = http.DefaultTransport
+	}
+	if req == nil || req.URL == nil {
+		return next.RoundTrip(req)
+	}
+	query := req.URL.Query()
+	if _, ok := query["client_protocol_version"]; !ok {
+		return next.RoundTrip(req)
+	}
+
+	cloned := req.Clone(req.Context())
+	clonedURL := *req.URL
+	query.Del("client_protocol_version")
+	clonedURL.RawQuery = query.Encode()
+	cloned.URL = &clonedURL
+	return next.RoundTrip(cloned)
+}
+
+func installClickHouseHTTPClientProtocolVersionStripper(opts *clickhouse.Options) {
+	if opts == nil {
+		return
+	}
+	previous := opts.TransportFunc
+	opts.TransportFunc = func(base *http.Transport) (http.RoundTripper, error) {
+		next := http.RoundTripper(base)
+		if previous != nil {
+			wrapped, err := previous(base)
+			if err != nil {
+				return nil, err
+			}
+			if wrapped != nil {
+				next = wrapped
+			}
+		}
+		return clickHouseHTTPClientProtocolVersionStripper{next: next}, nil
+	}
+}
+
 func parseClickHouseDurationParam(raw string) (time.Duration, bool) {
 	text := strings.TrimSpace(raw)
 	if text == "" {
@@ -396,6 +449,24 @@ func isClickHouseProtocolMismatch(err error) bool {
 		strings.Contains(text, "malformed http response")
 }

+func isClickHouseHTTPClientProtocolVersionUnsupported(err error) bool {
+	if err == nil {
+		return false
+	}
+	text := strings.ToLower(strings.TrimSpace(err.Error()))
+	if text == "" || !strings.Contains(text, "client_protocol_version") {
+		return false
+	}
+	return strings.Contains(text, "unknown setting") ||
+		strings.Contains(text, "unknown_setting") ||
+		strings.Contains(text, "code: 115")
+}
+
+func shouldTryNextClickHouseProtocol(protocol clickhouse.Protocol, err error) bool {
+	return isClickHouseProtocolMismatch(err) ||
+		(protocol == clickhouse.HTTP && isClickHouseHTTPClientProtocolVersionUnsupported(err))
+}
+
 func clickHouseProtocolName(protocol clickhouse.Protocol) string {
 	if protocol == clickhouse.HTTP {
 		return "HTTP"
@@ -436,6 +507,9 @@ func sanitizeClickHouseErrorMessage(err error) string {
 }

 func clickHouseAttemptFailureMessage(protocol clickhouse.Protocol, err error) string {
+	if protocol == clickhouse.HTTP && isClickHouseHTTPClientProtocolVersionUnsupported(err) {
+		return "当前 ClickHouse HTTP 端口不支持 client_protocol_version（常见于 ClickHouse 22.8），将使用 HTTP 兼容模式重试；如仍失败请确认连接协议和端口"
+	}
 	if isClickHouseProtocolMismatch(err) {
 		if protocol == clickhouse.Native {
 			return "服务端响应不像 Native 握手，当前端口更像 HTTP/HTTPS 端口；请选择 HTTP 协议，或确认 ClickHouse Native 端口"
@@ -551,27 +625,52 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error {
 		protocols := clickHouseProtocolsForAttempt(attempt)
 		for pIdx, protocol := range protocols {
 			protocolConfig := withClickHouseProtocol(attempt, protocol)
-			logger.Infof("ClickHouse 连接尝试：第%d组/%d 协议=%s 地址=%s:%d SSL=%t",
-				idx+1, len(attempts), clickHouseProtocolName(protocol), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL)
-			opts, err := c.buildClickHouseOptions(protocolConfig)
-			if err != nil {
-				failures = append(failures, fmt.Sprintf("第%d次 TLS 配置失败(protocol=%s): %v", idx+1, protocol.String(), err))
-				logger.Warnf("ClickHouse TLS 配置失败：第%d组/%d 协议=%s 地址=%s:%d SSL=%t 原因=%v",
-					idx+1, len(attempts), clickHouseProtocolName(protocol), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL, err)
-				continue
+			compatibilityModes := []bool{false}
+			if protocol == clickhouse.HTTP {
+				compatibilityModes = append(compatibilityModes, true)
 			}
-			c.conn = clickhouse.OpenDB(opts)
-			if err := c.Ping(); err != nil {
-				failureMessage := clickHouseAttemptFailureMessage(protocol, err)
-				failures = append(failures, fmt.Sprintf("第%d次连接验证失败(protocol=%s): %s", idx+1, protocol.String(), failureMessage))
-				logger.Warnf("ClickHouse 连接尝试失败：第%d组/%d 协议=%s 地址=%s:%d SSL=%t 原因=%s",
-					idx+1, len(attempts), clickHouseProtocolName(protocol), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL, failureMessage)
-				if c.conn != nil {
-					_ = c.conn.Close()
-					c.conn = nil
+			protocolSuccess := false
+			var lastProtocolErr error
+			for compatIdx, stripHTTPClientProtocolVersion := range compatibilityModes {
+				logger.Infof("ClickHouse 连接尝试：第%d组/%d 协议=%s 地址=%s:%d SSL=%t HTTP兼容=%t",
+					idx+1, len(attempts), clickHouseProtocolName(protocol), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL, stripHTTPClientProtocolVersion)
+				opts, err := c.buildClickHouseOptionsWithHTTPCompatibility(protocolConfig, stripHTTPClientProtocolVersion)
+				if err != nil {
+					failures = append(failures, fmt.Sprintf("第%d次 TLS 配置失败(protocol=%s): %v", idx+1, protocol.String(), err))
+					logger.Warnf("ClickHouse TLS 配置失败：第%d组/%d 协议=%s 地址=%s:%d SSL=%t 原因=%v",
+						idx+1, len(attempts), clickHouseProtocolName(protocol), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL, err)
+					lastProtocolErr = err
+					break
 				}
-				if pIdx == 0 && !isClickHouseProtocolMismatch(err) {
-					// 首次连接不是协议误配特征，避免无谓重试次协议。
+				c.conn = clickhouse.OpenDB(opts)
+				if err := c.Ping(); err != nil {
+					lastProtocolErr = err
+					failureMessage := clickHouseAttemptFailureMessage(protocol, err)
+					failures = append(failures, fmt.Sprintf("第%d次连接验证失败(protocol=%s): %s", idx+1, protocol.String(), failureMessage))
+					logger.Warnf("ClickHouse 连接尝试失败：第%d组/%d 协议=%s 地址=%s:%d SSL=%t HTTP兼容=%t 原因=%s",
+						idx+1, len(attempts), clickHouseProtocolName(protocol), protocolConfig.Host, protocolConfig.Port, protocolConfig.UseSSL, stripHTTPClientProtocolVersion, failureMessage)
+					if c.conn != nil {
+						_ = c.conn.Close()
+						c.conn = nil
+					}
+					if protocol == clickhouse.HTTP &&
+						!stripHTTPClientProtocolVersion &&
+						isClickHouseHTTPClientProtocolVersionUnsupported(err) &&
+						compatIdx+1 < len(compatibilityModes) {
+						logger.Warnf("ClickHouse HTTP 端口不支持 client_protocol_version，改用 HTTP 兼容模式重试")
+						continue
+					}
+					break
+				}
+				protocolSuccess = true
+				if stripHTTPClientProtocolVersion {
+					logger.Warnf("ClickHouse HTTP 兼容模式连接成功：已移除 client_protocol_version 参数")
+				}
+				break
+			}
+			if !protocolSuccess {
+				if pIdx == 0 && !shouldTryNextClickHouseProtocol(protocol, lastProtocolErr) {
+					// 首次连接不是协议误配或已知兼容性特征，避免无谓重试次协议。
 					break
 				}
 				continue
--- a/internal/db/clickhouse_impl_test.go
+++ b/internal/db/clickhouse_impl_test.go
@@ -8,6 +8,7 @@ import (
 	"database/sql/driver"
 	"errors"
 	"io"
+	"net/http"
 	"strings"
 	"sync"
 	"testing"
@@ -304,6 +305,56 @@ func TestClickHouseProtocolMismatchIncludesHTTPParseBinaryResponse(t *testing.T)
 	}
 }

+func TestClickHouseHTTPClientProtocolVersionUnsupportedEnablesCompatibilityRetry(t *testing.T) {
+	err := errors.New(`failed to query server hello: failed to query server hello info: sendQuery: [HTTP 404] response body: "Code: 115. DB::Exception: Unknown setting client_protocol_version. (UNKNOWN_SETTING)"`)
+	if !isClickHouseHTTPClientProtocolVersionUnsupported(err) {
+		t.Fatalf("expected client_protocol_version unknown setting to be treated as HTTP compatibility issue")
+	}
+	if !shouldTryNextClickHouseProtocol(clickhouse.HTTP, err) {
+		t.Fatalf("expected HTTP client_protocol_version issue to permit protocol fallback")
+	}
+	if shouldTryNextClickHouseProtocol(clickhouse.Native, err) {
+		t.Fatalf("native protocol should not treat HTTP client_protocol_version issue as retryable")
+	}
+
+	message := clickHouseAttemptFailureMessage(clickhouse.HTTP, err)
+	if !strings.Contains(message, "client_protocol_version") || !strings.Contains(message, "兼容模式") {
+		t.Fatalf("expected compatibility retry hint, got %q", message)
+	}
+}
+
+func TestClickHouseHTTPClientProtocolVersionStripperRemovesDriverQueryParam(t *testing.T) {
+	var seenQuery string
+	stripper := clickHouseHTTPClientProtocolVersionStripper{
+		next: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+			seenQuery = req.URL.RawQuery
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Body:       io.NopCloser(strings.NewReader("")),
+				Header:     make(http.Header),
+			}, nil
+		}),
+	}
+	req, err := http.NewRequest(http.MethodPost, "http://clickhouse.local:8123/?database=default&client_protocol_version=54485", nil)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+
+	res, err := stripper.RoundTrip(req)
+	if err != nil {
+		t.Fatalf("round trip: %v", err)
+	}
+	if res != nil && res.Body != nil {
+		res.Body.Close()
+	}
+	if strings.Contains(seenQuery, "client_protocol_version") {
+		t.Fatalf("expected client_protocol_version stripped from query, got %q", seenQuery)
+	}
+	if !strings.Contains(seenQuery, "database=default") {
+		t.Fatalf("expected other query parameters to remain, got %q", seenQuery)
+	}
+}
+
 func TestWithClickHouseProtocolForcesProtocolSelection(t *testing.T) {
 	httpConfig := withClickHouseProtocol(connection.ConnectionConfig{
 		Type: "clickhouse",
@@ -381,6 +432,12 @@ func TestClickHouseProtocolsForAttemptOnlyFallsBackInAutoMode(t *testing.T) {
 	}
 }

+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (fn roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return fn(req)
+}
+
 func protocolNames(protocols []clickhouse.Protocol) []string {
 	names := make([]string, 0, len(protocols))
 	for _, protocol := range protocols {