Array(T) performance improvement #2

Author: kshvakov

lib/column/array.go

@@ -9,10 +9,6 @@ import (
 	"github.com/kshvakov/clickhouse/lib/binary"
 )
 
-type ArrayWriter interface {
-	WriteArray(encoder *binary.Encoder, column Column) (uint64, error)
-}
-
 type Array struct {
 	base
 	depth  int

lib/data/block.go

@@ -95,7 +95,7 @@ func (block *Block) Read(serverInfo *ServerInfo, decoder *binary.Decoder) (err e
 	return nil
 }
 
-func (block *Block) prepareArray(value reflect.Value, num, level int) (values []interface{}) {
+func (block *Block) writeArray(column column.Column, value reflect.Value, num, level int) error {
 	switch {
 	case value.Kind() == reflect.Slice:
 		if len(block.offsets[num]) < level {
@@ -107,12 +107,16 @@ func (block *Block) prepareArray(value reflect.Value, num, level int) (values []
 			)
 		}
 		for i := 0; i < value.Len(); i++ {
-			values = append(values, block.prepareArray(value.Index(i), num, level+1)...)
+			if err := block.writeArray(column, value.Index(i), num, level+1); err != nil {
+				return err
+			}
 		}
 	default:
-		values = append(values, value.Interface())
+		if err := column.Write(block.buffers[num].Column, value.Interface()); err != nil {
+			return err
+		}
 	}
-	return values
+	return nil
 }
 
 func (block *Block) AppendRow(args []driver.Value) error {
@@ -130,10 +134,8 @@ func (block *Block) AppendRow(args []driver.Value) error {
 			if value.Kind() != reflect.Slice {
 				return fmt.Errorf("unsupported Array(T) type [%T]", value.Interface())
 			}
-			for _, value := range block.prepareArray(value, num, 1) {
-				if err := column.Write(block.buffers[num].Column, value); err != nil {
-					return err
-				}
+			if err := block.writeArray(c, value, num, 1); err != nil {
+				return err
 			}
 		case *column.Nullable:
 			if err := column.WriteNull(block.buffers[num].Offset, block.buffers[num].Column, args[num]); err != nil {

lib/data/block_write_column.go

@@ -94,10 +94,5 @@ func (block *Block) WriteArray(c int, v interface{}) error {
 	if value.Kind() != reflect.Slice {
 		return fmt.Errorf("unsupported Array(T) type [%T]", value.Interface())
 	}
-	for _, value := range block.prepareArray(value, c, 1) {
-		if err := block.Columns[c].Write(block.buffers[c].Column, value); err != nil {
-			return err
-		}
-	}
-	return nil
+	return block.writeArray(block.Columns[c], value, c, 1)
 }