Merge pull request #1578 from ClickHouse/fix_http_400

Fix Native API HTTP bugs

Author: SpencerTorres

clickhouse.go

@@ -150,7 +150,7 @@ func (ch *clickhouse) Query(ctx context.Context, query string, args ...any) (row
 	if err != nil {
 		return nil, err
 	}
-	conn.debugf("[acquired] connection [%d]", conn.connID())
+	conn.debugf("[query] \"%s\"", query)
 	return conn.query(ctx, ch.release, query, args...)
 }
 
@@ -161,7 +161,8 @@ func (ch *clickhouse) QueryRow(ctx context.Context, query string, args ...any) d
 			err: err,
 		}
 	}
-	conn.debugf("[acquired] connection [%d]", conn.connID())
+
+	conn.debugf("[query row] \"%s\"", query)
 	return conn.queryRow(ctx, ch.release, query, args...)
 }
 
@@ -170,8 +171,7 @@ func (ch *clickhouse) Exec(ctx context.Context, query string, args ...any) error
 	if err != nil {
 		return err
 	}
-	conn.debugf("[acquired] connection [%d]", conn.connID())
-
+	conn.debugf("[exec] \"%s\"", query)
 	if err := conn.exec(ctx, query, args...); err != nil {
 		ch.release(conn, err)
 		return err
@@ -185,6 +185,7 @@ func (ch *clickhouse) PrepareBatch(ctx context.Context, query string, opts ...dr
 	if err != nil {
 		return nil, err
 	}
+	conn.debugf("[prepare batch] \"%s\"", query)
 	batch, err := conn.prepareBatch(ctx, ch.release, ch.acquire, query, getPrepareBatchOptions(opts...))
 	if err != nil {
 		return nil, err
@@ -207,6 +208,7 @@ func (ch *clickhouse) AsyncInsert(ctx context.Context, query string, wait bool,
 	if err != nil {
 		return err
 	}
+	conn.debugf("[async insert] \"%s\"", query)
 	if err := conn.asyncInsert(ctx, query, wait, args...); err != nil {
 		ch.release(conn, err)
 		return err
@@ -220,6 +222,7 @@ func (ch *clickhouse) Ping(ctx context.Context) (err error) {
 	if err != nil {
 		return err
 	}
+	conn.debugf("[ping]")
 	if err := conn.ping(ctx); err != nil {
 		ch.release(conn, err)
 		return err
@@ -324,6 +327,7 @@ func (ch *clickhouse) acquire(ctx context.Context) (conn nativeTransport, err er
 			}
 		}
 		conn.setReleased(false)
+		conn.debugf("[acquired from pool]")
 		return conn, nil
 	default:
 	}
@@ -334,6 +338,7 @@ func (ch *clickhouse) acquire(ctx context.Context) (conn nativeTransport, err er
 		}
 		return nil, err
 	}
+	conn.debugf("[acquired new]")
 	return conn, nil
 }
 
@@ -377,31 +382,47 @@ func (ch *clickhouse) release(conn nativeTransport, err error) {
 		return
 	}
 	conn.setReleased(true)
-	conn.debugf("[released] connection [%d]", conn.connID())
+
+	if err != nil {
+		conn.debugf("[released with error]")
+	} else {
+		conn.debugf("[released]")
+	}
 
 	select {
 	case <-ch.open:
 	default:
 	}
-	if err != nil || time.Since(conn.connectedAtTime()) >= ch.opt.ConnMaxLifetime {
+
+	if err != nil {
+		conn.debugf("[close: error] %s", err.Error())
+		conn.close()
+		return
+	} else if time.Since(conn.connectedAtTime()) >= ch.opt.ConnMaxLifetime {
+		conn.debugf("[close: lifetime expired]")
 		conn.close()
 		return
 	}
+
 	if ch.opt.FreeBufOnConnRelease {
+		conn.debugf("[free buffer]")
 		conn.freeBuffer()
 	}
+
 	select {
 	case ch.idle <- conn:
 	default:
+		conn.debugf("[close: idle pool full %d/%d]", len(ch.idle), cap(ch.idle))
 		conn.close()
 	}
 }
 
 func (ch *clickhouse) Close() error {
 	for {
 		select {
-		case c := <-ch.idle:
-			c.close()
+		case conn := <-ch.idle:
+			conn.debugf("[close: closing pool]")
+			conn.close()
 		default:
 			// In rare cases, close may be called multiple times, don't block
 			//TODO: add proper close flag to indicate this pool is unusable after Close

conn.go

@@ -64,12 +64,12 @@ func dial(ctx context.Context, addr string, num int, opt *Options) (*connect, er
 		if opt.Debugf != nil {
 			debugf = func(format string, v ...any) {
 				opt.Debugf(
-					"[clickhouse][conn=%d][%s] "+format,
-					append([]interface{}{num, conn.RemoteAddr()}, v...)...,
+					"[clickhouse][%s][id=%d] "+format,
+					append([]interface{}{conn.RemoteAddr(), num}, v...)...,
 				)
 			}
 		} else {
-			debugf = log.New(os.Stdout, fmt.Sprintf("[clickhouse][conn=%d][%s]", num, conn.RemoteAddr()), 0).Printf
+			debugf = log.New(os.Stdout, fmt.Sprintf("[clickhouse][%s][id=%d]", conn.RemoteAddr(), num), 0).Printf
 		}
 	}
 

conn_http.go

@@ -33,7 +33,6 @@ import (
 	"net/http"
 	"net/url"
 	"os"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -165,12 +164,12 @@ func dialHttp(ctx context.Context, addr string, num int, opt *Options) (*httpCon
 		if opt.Debugf != nil {
 			debugf = func(format string, v ...any) {
 				opt.Debugf(
-					"[clickhouse-http][conn=%d][%s] "+format,
-					append([]interface{}{num, addr}, v...)...,
+					"[clickhouse-http][%s][id=%d] "+format,
+					append([]interface{}{addr, num}, v...)...,
 				)
 			}
 		} else {
-			debugf = log.New(os.Stdout, fmt.Sprintf("[clickhouse-http][conn=%d][%s]", num, addr), 0).Printf
+			debugf = log.New(os.Stdout, fmt.Sprintf("[clickhouse-http][%s][id=%d]", addr, num), 0).Printf
 		}
 	}
 
@@ -216,6 +215,8 @@ func dialHttp(ctx context.Context, addr string, num int, opt *Options) (*httpCon
 	}
 
 	query.Set("default_format", "Native")
+	// TODO: we support newer revisions but for some reason this completely breaks Native format
+	//query.Set("client_protocol_version", strconv.Itoa(ClientTCPProtocolVersion))
 	u.RawQuery = query.Encode()
 
 	httpProxy := http.ProxyFromEnvironment
@@ -241,13 +242,18 @@ func dialHttp(ctx context.Context, addr string, num int, opt *Options) (*httpCon
 		}
 	}
 
-	// Temporary conn for determining timezone + version
-	conn := &httpConnect{
-		opt: opt,
+	conn := httpConnect{
+		id:          num,
+		connectedAt: time.Now(),
+		released:    false,
+		debugfFunc:  debugf,
+		opt:         opt,
 		client: &http.Client{
 			Transport: t,
 		},
-		url:             u,
+		url: u,
+		// TODO: learn more about why revision is broken
+		//revision:        ClientTCPProtocolVersion,
 		buffer:          new(chproto.Buffer),
 		compression:     opt.Compression.Method,
 		blockCompressor: compress.NewWriter(compress.Level(opt.Compression.Level), compress.Method(opt.Compression.Method)),
@@ -259,30 +265,9 @@ func dialHttp(ctx context.Context, addr string, num int, opt *Options) (*httpCon
 	if err != nil {
 		return nil, fmt.Errorf("failed to query server hello: %w", err)
 	}
-	// Close temp connection, important for freeing session_id if set
-	_ = conn.close()
-
-	// Set client revision so we can decode updated versions of Native format
-	query.Set("client_protocol_version", strconv.Itoa(int(handshake.Revision)))
-	u.RawQuery = query.Encode()
+	conn.handshake = handshake
 
-	return &httpConnect{
-		id:          num,
-		connectedAt: time.Now(),
-		released:    false,
-		debugfFunc:  debugf,
-		opt:         opt,
-		client: &http.Client{
-			Transport: t,
-		},
-		url:             u,
-		buffer:          new(chproto.Buffer),
-		compression:     opt.Compression.Method,
-		blockCompressor: compress.NewWriter(compress.Level(opt.Compression.Level), compress.Method(opt.Compression.Method)),
-		compressionPool: compressionPool,
-		blockBufferSize: opt.BlockBufferSize,
-		handshake:       handshake,
-	}, nil
+	return &conn, nil
 }
 
 type httpConnect struct {
@@ -291,6 +276,7 @@ type httpConnect struct {
 	released        bool
 	debugfFunc      func(format string, v ...any)
 	opt             *Options
+	revision        uint64
 	url             *url.URL
 	client          *http.Client
 	buffer          *chproto.Buffer
@@ -333,6 +319,7 @@ func (h *httpConnect) isBad() bool {
 }
 
 func (h *httpConnect) queryHello(ctx context.Context, release nativeTransportRelease) (proto.ServerHandshake, error) {
+	h.debugf("[query hello]")
 	ctx = Context(ctx, ignoreExternalTables())
 	query := "SELECT displayName(), version(), revision(), timezone()"
 	rows, err := h.query(ctx, release, query)
@@ -420,8 +407,8 @@ func createCompressionPool(compression *Compression) (Pool[HTTPReaderWriter], er
 func (h *httpConnect) writeData(block *proto.Block) error {
 	// Saving offset of compressible data
 	start := len(h.buffer.Buf)
-	if err := block.Encode(h.buffer, 0); err != nil {
-		return err
+	if err := block.Encode(h.buffer, h.revision); err != nil {
+		return fmt.Errorf("block encode: %w", err)
 	}
 	if h.compression == CompressionLZ4 || h.compression == CompressionZSTD {
 		// Performing compression. Supported and requires
@@ -445,8 +432,8 @@ func (h *httpConnect) readData(reader *chproto.Reader, timezone *time.Location)
 		reader.EnableCompression()
 		defer reader.DisableCompression()
 	}
-	if err := block.Decode(reader, h.handshake.Revision); err != nil {
-		return nil, err
+	if err := block.Decode(reader, h.revision); err != nil {
+		return nil, fmt.Errorf("block decode: %w", err)
 	}
 	return &block, nil
 }
@@ -564,7 +551,7 @@ func (h *httpConnect) createRequestWithExternalTables(ctx context.Context, query
 			return nil, err
 		}
 		buf.Reset()
-		err = table.Block().Encode(buf, 0)
+		err = table.Block().Encode(buf, h.revision)
 		if err != nil {
 			return nil, err
 		}
@@ -597,11 +584,12 @@ func (h *httpConnect) executeRequest(req *http.Request) (*http.Response, error)
 
 	if resp.StatusCode != http.StatusOK {
 		defer discardAndClose(resp.Body)
-		msg, err := h.readRawResponse(resp)
+		msgBytes, err := h.readRawResponse(resp)
 		if err != nil {
-			return nil, fmt.Errorf("clickhouse [execute]:: %d code: failed to read the response: %w", resp.StatusCode, err)
+			return nil, fmt.Errorf("[HTTP %d] failed to read response: %w", resp.StatusCode, err)
 		}
-		return nil, fmt.Errorf("clickhouse [execute]:: %d code: %s", resp.StatusCode, string(msg))
+
+		return nil, fmt.Errorf("[HTTP %d] response body: \"%s\"", resp.StatusCode, string(msgBytes))
 	}
 	return resp, nil
 }

conn_http_batch.go

@@ -77,11 +77,7 @@ func fetchColumnNamesAndTypesForInsert(h *httpConnect, release nativeTransportRe
 	} else {
 		// Use all columns
 		for _, colName := range allColumns {
-			colType, ok := columnsToTypes[colName]
-			if !ok {
-				return nil, fmt.Errorf("column %s is not present in the table %s", colName, tableName)
-			}
-
+			colType := columnsToTypes[colName]
 			insertColumns = append(insertColumns, ColumnNameAndType{
 				Name: colName,
 				Type: colType,
@@ -121,10 +117,12 @@ func newBlock(h *httpConnect, release nativeTransportRelease, ctx context.Contex
 }
 
 func (h *httpConnect) prepareBatch(ctx context.Context, release nativeTransportRelease, acquire nativeTransportAcquire, query string, opts driver.PrepareBatchOptions) (driver.Batch, error) {
-	// release is not used for newBlock since the connection is held for the batch.
+	// release is not used within newBlock since the connection is held for the batch.
 	query, block, err := newBlock(h, func(nativeTransport, error) {}, ctx, query)
 	if err != nil {
-		return nil, fmt.Errorf("failed to init block for HTTP batch: %w", err)
+		err = fmt.Errorf("failed to init block for HTTP batch: %w", err)
+		release(h, err)
+		return nil, err
 	}
 
 	return &httpBatch{
@@ -237,6 +235,7 @@ func (b *httpBatch) IsSent() bool {
 func (b *httpBatch) Send() (err error) {
 	defer func() {
 		b.sent = true
+		b.release(err)
 	}()
 	if b.sent {
 		return ErrBatchAlreadySent
@@ -264,33 +263,29 @@ func (b *httpBatch) Send() (err error) {
 	connWriter := compressionWriter.reset(pipeWriter)
 
 	go func() {
-		var err error = nil
+		var err error
 		defer pipeWriter.CloseWithError(err)
 		defer connWriter.Close()
 		b.conn.buffer.Reset()
-		if b.block.Rows() != 0 {
-			if err = b.conn.writeData(b.block); err != nil {
-				return
-			}
-		}
-		if err = b.conn.writeData(&proto.Block{}); err != nil {
+		if err = b.conn.writeData(b.block); err != nil {
 			return
 		}
 		if _, err = connWriter.Write(b.conn.buffer.Buf); err != nil {
 			return
 		}
-		b.release(nil)
 	}()
 
 	options.settings["query"] = b.query
 	headers["Content-Type"] = "application/octet-stream"
 
+	b.conn.debugf("[batch send start] columns=%d rows=%d", len(b.block.Columns), b.block.Rows())
 	res, err := b.conn.sendStreamQuery(b.ctx, pipeReader, &options, headers)
 	if err != nil {
-		return err
+		return fmt.Errorf("batch sendStreamQuery: %w", err)
 	}
 	discardAndClose(res.Body)
 
+	b.conn.debugf("[batch send complete]")
 	b.block.Reset()
 
 	return nil