Merge remote-tracking branch 'origin/main'

Author: jkaflik

CHANGELOG.md

@@ -1,3 +1,23 @@
+# v2.22.2, 2024-03-18 <!-- Release notes generated using configuration in .github/release.yml at main -->
+
+## What's Changed
+### Fixes 🐛
+* Fix for Map columns with Enums by @leklund in https://github.com/ClickHouse/clickhouse-go/pull/1236
+
+## New Contributors
+* @leklund made their first contribution in https://github.com/ClickHouse/clickhouse-go/pull/1236
+
+**Full Changelog**: https://github.com/ClickHouse/clickhouse-go/compare/v2.22.1...v2.22.2
+
+# v2.22.1, 2024-03-18 <!-- Release notes generated using configuration in .github/release.yml at main -->
+
+## What's Changed
+### Fixes 🐛
+* Make errors channel buffered inside query()  by @threadedstream in https://github.com/ClickHouse/clickhouse-go/pull/1237
+
+
+**Full Changelog**: https://github.com/ClickHouse/clickhouse-go/compare/v2.22.0...v2.22.1
+
 # v2.20.0, 2024-02-28 <!-- Release notes generated using configuration in .github/release.yml at main -->
 
 ## What's Changed

client_info.go

@@ -29,8 +29,8 @@ const ClientName = "clickhouse-go"
 
 const (
 	ClientVersionMajor       = 2
-	ClientVersionMinor       = 20
-	ClientVersionPatch       = 0
+	ClientVersionMinor       = 22
+	ClientVersionPatch       = 2
 	ClientTCPProtocolVersion = proto.DBMS_TCP_PROTOCOL_VERSION
 )
 

conn_batch.go

@@ -133,6 +133,13 @@ func (b *batch) Append(v ...any) error {
 	if b.err != nil {
 		return b.err
 	}
+
+	if len(v) > 0 {
+		if r, ok := v[0].(*rows); ok {
+			return b.appendRowsBlocks(r)
+		}
+	}
+
 	if err := b.block.Append(v...); err != nil {
 		b.err = errors.Wrap(ErrBatchInvalid, err.Error())
 		b.release(err)
@@ -141,6 +148,36 @@ func (b *batch) Append(v ...any) error {
 	return nil
 }
 
+// appendRowsBlocks is an experimental feature that allows rows blocks be appended directly to the batch.
+// This API is not stable and may be changed in the future.
+// See: tests/batch_block_test.go
+func (b *batch) appendRowsBlocks(r *rows) error {
+	var lastReadLock *proto.Block
+	var blockNum int
+
+	for r.Next() {
+		if lastReadLock == nil { // make sure the first block is logged
+			b.conn.debugf("[batch.appendRowsBlocks] blockNum = %d", blockNum)
+		}
+
+		// rows.Next() will read the next block from the server only if the current block is empty
+		// only if new block is available we should flush the current block
+		// the last block will be handled by the batch.Send() method
+		if lastReadLock != nil && lastReadLock != r.block {
+			if err := b.Flush(); err != nil {
+				return err
+			}
+			blockNum++
+			b.conn.debugf("[batch.appendRowsBlocks] blockNum = %d", blockNum)
+		}
+
+		b.block = r.block
+		lastReadLock = r.block
+	}
+
+	return nil
+}
+
 func (b *batch) AppendStruct(v any) error {
 	if b.err != nil {
 		return b.err

conn_query.go

@@ -65,7 +65,7 @@ func (c *connect) query(ctx context.Context, release func(*connect, error), quer
 		bufferSize = options.blockBufferSize
 	}
 	var (
-		errors = make(chan error)
+		errors = make(chan error, 1)
 		stream = make(chan *proto.Block, bufferSize)
 	)
 

contributors/list

@@ -1 +1 @@
-rogeryk <rogeryk@outlook.com>
+Lukas Eklund <leklund@gmail.com>

lib/column/map.go

@@ -66,7 +66,17 @@ func (col *Map) Name() string {
 
 func (col *Map) parse(t Type, tz *time.Location) (_ Interface, err error) {
 	col.chType = t
-	if types := strings.SplitN(t.params(), ",", 2); len(types) == 2 {
+	types := make([]string, 2, 2)
+	typeParams := t.params()
+	idx := strings.Index(typeParams, ",")
+	if strings.HasPrefix(typeParams, "Enum") {
+		idx = strings.Index(typeParams, "),") + 1
+	}
+	if idx > 0 {
+		types[0] = typeParams[:idx]
+		types[1] = typeParams[idx+1:]
+	}
+	if types[0] != "" && types[1] != "" {
 		if col.keys, err = Type(strings.TrimSpace(types[0])).Column(col.name, tz); err != nil {
 			return nil, err
 		}

tests/batch_block_test.go

@@ -0,0 +1,80 @@
+// Licensed to ClickHouse, Inc. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. ClickHouse, Inc. licenses this file to you 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 tests
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/context"
+)
+
+// TestBatchAppendRows tests experimental batch rows blocks append feature.
+// This API is not stable and may be changed in the future.
+func TestBatchAppendRows(t *testing.T) {
+	te, err := GetTestEnvironment(testSet)
+	require.NoError(t, err)
+	blocksRead := atomic.Uint64{}
+	opts := ClientOptionsFromEnv(te, clickhouse.Settings{})
+	opts.Debug = true
+	opts.Debugf = func(format string, args ...interface{}) {
+		if format == "[batch.appendRowsBlocks] blockNum = %d" {
+			blocksRead.Store(uint64(args[0].(int))) // store the last block number read from rows
+		}
+	}
+
+	conn, err := GetConnectionWithOptions(&opts)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+
+	// given we have two tables and a million rows in the source table
+	var tables = []string{"source", "target"}
+	for _, table := range tables {
+		require.NoError(t, conn.Exec(context.Background(), "create table if not exists "+table+" (number1 Int, number2 String, number3 Tuple(String, Int), number4 DateTime) engine = MergeTree() order by tuple()"))
+		defer conn.Exec(context.Background(), "drop table if exists "+table)
+	}
+
+	require.NoError(t, conn.Exec(ctx, "INSERT INTO source SELECT number, 'string', tuple('foo', number), now() FROM system.numbers LIMIT 1000000"))
+
+	// when we create a batch with direct data block access 10 times
+
+	selectCtx := clickhouse.Context(ctx, clickhouse.WithSettings(clickhouse.Settings{
+		"max_block_size": 1000,
+	}))
+
+	sourceRows, err := conn.Query(selectCtx, "SELECT * FROM source")
+	require.NoError(t, err)
+	defer sourceRows.Close()
+
+	b, err := conn.PrepareBatch(ctx, "INSERT INTO target")
+	require.NoError(t, err)
+	require.NoError(t, b.Append(sourceRows))
+	require.NoError(t, b.Send())
+
+	// then we should be able to see the data in the target table
+	row := conn.QueryRow(ctx, "SELECT count() FROM source")
+	require.NoError(t, row.Err())
+	var count uint64
+	require.NoError(t, row.Scan(&count))
+	assert.Equal(t, uint64(1000000), count)
+	assert.Equal(t, uint64(999), blocksRead.Load())
+}

tests/map_test.go

@@ -122,6 +122,9 @@ func TestColumnarMap(t *testing.T) {
 			  Col1 Map(String, UInt64)
 			, Col2 Map(String, UInt64)
 			, Col3 Map(String, UInt64)
+			, Col4 Map(Enum16('one' = 1, 'two' = 2), UInt64)
+			, Col5 Map(String, Enum16('one' = 1, 'two' = 2))
+			, Col6 Map(Enum8('one' = 1, 'two' = 2), Enum8('red' = 1, 'blue' = 2))
 		) Engine MergeTree() ORDER BY tuple()
 		`
 	defer func() {
@@ -134,6 +137,9 @@ func TestColumnarMap(t *testing.T) {
 		col1Data = []map[string]uint64{}
 		col2Data = []map[string]uint64{}
 		col3Data = []map[string]uint64{}
+		col4Data = []map[string]uint64{}
+		col5Data = []map[string]string{}
+		col6Data = []map[string]string{}
 	)
 	for i := 0; i < 100; i++ {
 		col1Data = append(col1Data, map[string]uint64{
@@ -145,17 +151,35 @@ func TestColumnarMap(t *testing.T) {
 			fmt.Sprintf("key_col_2_%d_2", i): uint64(i),
 		})
 		col3Data = append(col3Data, map[string]uint64{})
+		col4Data = append(col4Data, map[string]uint64{
+			"one": uint64(i),
+			"two": uint64(i),
+		})
+		col5Data = append(col5Data, map[string]string{
+			fmt.Sprintf("key_col_2_%d_1", i): "one",
+			fmt.Sprintf("key_col_2_%d_2", i): "two",
+		})
+		col6Data = append(col6Data, map[string]string{
+			"one": "red",
+			"two": "blue",
+		})
 	}
 	require.NoError(t, batch.Column(0).Append(col1Data))
 	require.NoError(t, batch.Column(1).Append(col2Data))
 	require.NoError(t, batch.Column(2).Append(col3Data))
+	require.NoError(t, batch.Column(3).Append(col4Data))
+	require.NoError(t, batch.Column(4).Append(col5Data))
+	require.NoError(t, batch.Column(5).Append(col6Data))
 	require.Equal(t, 100, batch.Rows())
 	require.NoError(t, batch.Send())
 	{
 		var (
 			col1     map[string]uint64
 			col2     map[string]uint64
 			col3     map[string]uint64
+			col4     map[string]uint64
+			col5     map[string]string
+			col6     map[string]string
 			col1Data = map[string]uint64{
 				"key_col_1_10_1": 10,
 				"key_col_1_10_2": 10,
@@ -165,11 +189,26 @@ func TestColumnarMap(t *testing.T) {
 				"key_col_2_10_2": 10,
 			}
 			col3Data = map[string]uint64{}
+			col4Data = map[string]uint64{
+				"one": 10,
+				"two": 10,
+			}
+			col5Data = map[string]string{
+				"key_col_2_10_1": "one",
+				"key_col_2_10_2": "two",
+			}
+			col6Data = map[string]string{
+				"one": "red",
+				"two": "blue",
+			}
 		)
-		require.NoError(t, conn.QueryRow(ctx, "SELECT * FROM test_map WHERE Col1['key_col_1_10_1'] = $1", 10).Scan(&col1, &col2, &col3))
+		require.NoError(t, conn.QueryRow(ctx, "SELECT * FROM test_map WHERE Col1['key_col_1_10_1'] = $1", 10).Scan(&col1, &col2, &col3, &col4, &col5, &col6))
 		assert.Equal(t, col1Data, col1)
 		assert.Equal(t, col2Data, col2)
 		assert.Equal(t, col3Data, col3)
+		assert.Equal(t, col4Data, col4)
+		assert.Equal(t, col5Data, col5)
+		assert.Equal(t, col6Data, col6)
 	}
 }