fix Map #467

clickhouse_rows.go

@@ -48,7 +48,7 @@ next:
 		}
 	}
 	r.row++
-	return true
+	return r.row <= r.block.Rows()
 }
 
 func (r *rows) Scan(dest ...interface{}) error {

clickhouse_std.go

@@ -211,6 +211,12 @@ func (r *stdRows) ColumnTypePrecisionScale(idx int) (precision, scale int64, ok
 }
 
 func (r *stdRows) Next(dest []driver.Value) error {
+	if len(r.rows.block.Columns) != len(dest) {
+		return &OpError{
+			Op:  "Next",
+			Err: fmt.Errorf("expected %d destination arguments in Next, not %d", len(r.rows.block.Columns), len(dest)),
+		}
+	}
 	if r.rows.Next() {
 		for i := range dest {
 			nullable, ok := r.ColumnTypeNullable(i)

lib/column/map.go

@@ -175,11 +175,14 @@ func (col *Map) row(n int) reflect.Value {
 	if n != 0 {
 		prev = col.offsets[n-1]
 	}
-	size := int(col.offsets[n] - prev)
+	var (
+		size = int(col.offsets[n] - prev)
+		from = int(prev)
+	)
 	for next := 0; next < size; next++ {
 		value.SetMapIndex(
-			reflect.ValueOf(col.keys.Row(n*size+next, false)),
-			reflect.ValueOf(col.values.Row(n*size+next, false)),
+			reflect.ValueOf(col.keys.Row(from+next, false)),
+			reflect.ValueOf(col.values.Row(from+next, false)),
 		)
 	}
 	return value

tests/issues/470/main.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"log"
+	"reflect"
+
+	_ "github.com/ClickHouse/clickhouse-go/v2"
+)
+
+type DatabaseFrame struct {
+	name        string
+	ColumnNames []string
+	rows        *sql.Rows
+	columnTypes []*sql.ColumnType
+	vars        []interface{}
+}
+
+func NewDatabaseFrame(name string, rows *sql.Rows) (DatabaseFrame, error) {
+	databaseFrame := DatabaseFrame{}
+	columnTypes, err := rows.ColumnTypes()
+	if err != nil {
+		return DatabaseFrame{}, err
+	}
+	databaseFrame.columnTypes = columnTypes
+	databaseFrame.name = name
+	vars := make([]interface{}, len(columnTypes), len(columnTypes))
+	columnNames := make([]string, len(columnTypes), len(columnTypes))
+	for i := range columnTypes {
+		value := reflect.Zero(columnTypes[i].ScanType()).Interface()
+		vars[i] = &value
+		columnNames[i] = columnTypes[i].Name()
+	}
+	databaseFrame.ColumnNames = columnNames
+	databaseFrame.vars = vars
+	databaseFrame.rows = rows
+	return databaseFrame, nil
+}
+
+func (f DatabaseFrame) Next() ([]interface{}, bool, error) {
+	values := make([]interface{}, len(f.columnTypes), len(f.columnTypes))
+	for f.rows.Next() {
+		if err := f.rows.Scan(f.vars...); err != nil {
+			return nil, false, err
+		}
+		for i := range f.columnTypes {
+			ptr := reflect.ValueOf(f.vars[i])
+			values[i] = ptr.Elem().Interface()
+		}
+		return values, true, nil
+	}
+	f.rows.Close()
+	return nil, false, f.rows.Err()
+}
+
+func NewNativeClient(host string, port uint16, username string, password string) (*sql.DB, error) {
+	// debug output ?debug=true
+	connection, err := sql.Open("clickhouse", fmt.Sprintf("clickhouse://%s:%s@%s:%d/", username, password, host, port))
+	if err != nil {
+		return nil, err
+	}
+	if err := connection.Ping(); err != nil {
+		return nil, err
+	}
+	return connection, nil
+}
+
+func main() {
+	c, err := NewNativeClient("localhost", 9000, "", "")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	i := 0
+	log.Printf("Reading system.%s", "system.query_thread_log")
+	rows, err := c.Query("SELECT * FROM system.query_thread_log")
+	if err != nil {
+		log.Printf("Query failed")
+		log.Fatal(err)
+	}
+	frame, err := NewDatabaseFrame("db_frame", rows)
+	if err != nil {
+		log.Println("Cant' construct frame")
+		log.Fatal(err)
+	}
+	//iterate to exhaustion
+
+	for {
+		_, ok, err := frame.Next()
+		if !ok {
+			if err != nil {
+				log.Println("Failed on termination")
+				log.Println(err)
+				break
+			}
+			break
+		}
+		if err != nil {
+			log.Fatal(err)
+		}
+		i++
+	}
+	log.Printf("Success with %d rows!!", i)
+}