mirror of
https://github.com/hanzo-ds/go.git
synced 2026-10-30 19:45:01 +00:00
Source: the unexported clickhouse struct becomes client, the ch receiver and chproto import alias follow, error message prefixes read datastore, and the JSON serializer interface drops the vendor infix. resources' metadata key and its accessor are renamed together, with a test proving the embedded YAML still unmarshals — a silent drift there would have made every connection take the unsupported-server-version branch. Codegen: all three templates still emitted the upstream module paths, so regenerating would have quietly undone the rename. Note the templates have drifted from the files they produce (regenerating today drops QBit and scanTypeDuration), so the generated files are left as tracked and only the templates are corrected. Dead links: an earlier mechanical pass invented URLs that do not resolve — github.com/ClickHouse/datastore-go, hanzo-ds/driver issue numbers that belong to upstream, and mangled names of independent projects (mailru, uptrace, nikepan, EpicStep). Third-party names are not ours to rewrite, so those directories are removed rather than repaired, and provenance comments now point at real files. Dev infra: compose runs our own server image at the tag production runs, since the vendor's alpine tag does not exist in our registry. Cluster hostnames, nginx upstreams and server configs are renamed as one set. CHANGELOG.md is upstream's release history for versions this module never released; rewriting it would fabricate a lineage, so it is dropped and the upstream base recorded in README instead. The two long-standing vet findings are fixed by keying the struct literals rather than suppressed. LICENSE and contributors/list keep the vendor name on purpose: one is the copyright notice Apache-2.0 requires, the other is real people's names. CI now enforces build, vet, test, gofmt and a grep gate.
355 lines
7.8 KiB
Go
355 lines
7.8 KiB
Go
package column
|
|
|
|
import (
|
|
"database/sql"
|
|
"database/sql/driver"
|
|
"fmt"
|
|
"github.com/hanzo-ds/native/proto"
|
|
"reflect"
|
|
"strings"
|
|
)
|
|
|
|
// https://github.com/Datastore/Datastore/blob/master/src/Columns/ColumnMap.cpp
|
|
type Map struct {
|
|
keys Interface
|
|
values Interface
|
|
chType Type
|
|
offsets Int64
|
|
scanType reflect.Type
|
|
name string
|
|
}
|
|
|
|
type OrderedMap interface {
|
|
Get(key any) (any, bool)
|
|
Put(key any, value any)
|
|
Keys() <-chan any
|
|
}
|
|
|
|
type MapIterator interface {
|
|
Next() bool
|
|
Key() any
|
|
Value() any
|
|
}
|
|
|
|
type IterableOrderedMap interface {
|
|
Put(key any, value any)
|
|
Iterator() MapIterator
|
|
}
|
|
|
|
func (col *Map) Reset() {
|
|
col.keys.Reset()
|
|
col.values.Reset()
|
|
col.offsets.Reset()
|
|
}
|
|
|
|
func (col *Map) Name() string {
|
|
return col.name
|
|
}
|
|
|
|
func (col *Map) parse(t Type, sc *ServerContext) (_ Interface, err error) {
|
|
col.chType = t
|
|
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, sc); err != nil {
|
|
return nil, err
|
|
}
|
|
if col.values, err = Type(strings.TrimSpace(types[1])).Column(col.name, sc); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if col.keys.ScanType().Comparable() {
|
|
col.scanType = reflect.MapOf(
|
|
col.keys.ScanType(),
|
|
col.values.ScanType(),
|
|
)
|
|
return col, nil
|
|
}
|
|
}
|
|
return nil, &UnsupportedColumnTypeError{
|
|
t: t,
|
|
}
|
|
}
|
|
|
|
func (col *Map) Type() Type {
|
|
return col.chType
|
|
}
|
|
|
|
func (col *Map) ScanType() reflect.Type {
|
|
return col.scanType
|
|
}
|
|
|
|
func (col *Map) Rows() int {
|
|
return col.offsets.col.Rows()
|
|
}
|
|
|
|
func (col *Map) Row(i int, ptr bool) any {
|
|
return col.row(i).Interface()
|
|
}
|
|
|
|
func (col *Map) ScanRow(dest any, i int) error {
|
|
if scanner, ok := dest.(sql.Scanner); ok {
|
|
return scanner.Scan(col.row(i).Interface())
|
|
}
|
|
value := reflect.Indirect(reflect.ValueOf(dest))
|
|
if value.Type() == col.scanType {
|
|
value.Set(col.row(i))
|
|
return nil
|
|
}
|
|
if om, ok := dest.(IterableOrderedMap); ok {
|
|
keys, values := col.orderedRow(i)
|
|
for i := range keys {
|
|
om.Put(keys[i], values[i])
|
|
}
|
|
return nil
|
|
}
|
|
if om, ok := dest.(OrderedMap); ok {
|
|
keys, values := col.orderedRow(i)
|
|
for i := range keys {
|
|
om.Put(keys[i], values[i])
|
|
}
|
|
return nil
|
|
}
|
|
return &ColumnConverterError{
|
|
Op: "ScanRow",
|
|
To: fmt.Sprintf("%T", dest),
|
|
From: string(col.chType),
|
|
Hint: fmt.Sprintf("try using %s", col.scanType),
|
|
}
|
|
}
|
|
|
|
func (col *Map) Append(v any) (nulls []uint8, err error) {
|
|
value := reflect.Indirect(reflect.ValueOf(v))
|
|
if value.Kind() != reflect.Slice {
|
|
if valuer, ok := v.(driver.Valuer); ok {
|
|
val, err := valuer.Value()
|
|
if err != nil {
|
|
return nil, &ColumnConverterError{
|
|
Op: "Append",
|
|
To: string(col.chType),
|
|
From: fmt.Sprintf("%T", v),
|
|
Hint: fmt.Sprintf("could not get driver.Valuer value, try using %s", col.scanType),
|
|
}
|
|
}
|
|
return col.Append(val)
|
|
}
|
|
return nil, &ColumnConverterError{
|
|
Op: "Append",
|
|
To: string(col.chType),
|
|
From: fmt.Sprintf("%T", v),
|
|
Hint: fmt.Sprintf("try using %s", col.scanType),
|
|
}
|
|
}
|
|
for i := 0; i < value.Len(); i++ {
|
|
if err := col.AppendRow(value.Index(i).Interface()); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func (col *Map) AppendRow(v any) error {
|
|
if v == nil {
|
|
// NOTE: successful Map.parse() make sure we have
|
|
// valid col.scanType
|
|
v = reflect.Zero(col.scanType).Interface()
|
|
}
|
|
|
|
value := reflect.Indirect(reflect.ValueOf(v))
|
|
if value.Type() == col.scanType {
|
|
var (
|
|
size int64
|
|
iter = value.MapRange()
|
|
)
|
|
for iter.Next() {
|
|
size++
|
|
if err := col.keys.AppendRow(iter.Key().Interface()); err != nil {
|
|
return err
|
|
}
|
|
if err := col.values.AppendRow(iter.Value().Interface()); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
var prev int64
|
|
if n := col.offsets.Rows(); n != 0 {
|
|
prev = col.offsets.col.Row(n - 1)
|
|
}
|
|
col.offsets.col.Append(prev + size)
|
|
return nil
|
|
}
|
|
|
|
if orderedMap, ok := v.(IterableOrderedMap); ok {
|
|
var size int64
|
|
iter := orderedMap.Iterator()
|
|
for iter.Next() {
|
|
key, value := iter.Key(), iter.Value()
|
|
size++
|
|
if err := col.keys.AppendRow(key); err != nil {
|
|
return err
|
|
}
|
|
if err := col.values.AppendRow(value); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
var prev int64
|
|
if n := col.offsets.Rows(); n != 0 {
|
|
prev = col.offsets.col.Row(n - 1)
|
|
}
|
|
col.offsets.col.Append(prev + size)
|
|
return nil
|
|
}
|
|
|
|
if orderedMap, ok := v.(OrderedMap); ok {
|
|
var size int64
|
|
for key := range orderedMap.Keys() {
|
|
value, ok := orderedMap.Get(key)
|
|
if !ok {
|
|
return fmt.Errorf("ordered map has key %v but no corresponding value", key)
|
|
}
|
|
size++
|
|
if err := col.keys.AppendRow(key); err != nil {
|
|
return err
|
|
}
|
|
if err := col.values.AppendRow(value); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
var prev int64
|
|
if n := col.offsets.Rows(); n != 0 {
|
|
prev = col.offsets.col.Row(n - 1)
|
|
}
|
|
col.offsets.col.Append(prev + size)
|
|
return nil
|
|
}
|
|
|
|
if valuer, ok := v.(driver.Valuer); ok {
|
|
val, err := valuer.Value()
|
|
if err != nil {
|
|
return &ColumnConverterError{
|
|
Op: "AppendRow",
|
|
To: string(col.chType),
|
|
From: fmt.Sprintf("%T", v),
|
|
Hint: fmt.Sprintf("could not get driver.Valuer value, try using %s", col.scanType),
|
|
}
|
|
}
|
|
return col.AppendRow(val)
|
|
}
|
|
|
|
return &ColumnConverterError{
|
|
Op: "AppendRow",
|
|
To: string(col.chType),
|
|
From: fmt.Sprintf("%T", v),
|
|
Hint: fmt.Sprintf("try using %s", col.scanType),
|
|
}
|
|
|
|
}
|
|
|
|
func (col *Map) Decode(reader *proto.Reader, rows int) error {
|
|
if err := col.offsets.col.DecodeColumn(reader, rows); err != nil {
|
|
return err
|
|
}
|
|
if i := col.offsets.Rows(); i != 0 {
|
|
size := int(col.offsets.col.Row(i - 1))
|
|
if err := col.keys.Decode(reader, size); err != nil {
|
|
return err
|
|
}
|
|
return col.values.Decode(reader, size)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (col *Map) Encode(buffer *proto.Buffer) {
|
|
col.offsets.col.EncodeColumn(buffer)
|
|
col.keys.Encode(buffer)
|
|
col.values.Encode(buffer)
|
|
}
|
|
|
|
func (col *Map) ReadStatePrefix(reader *proto.Reader) error {
|
|
if serialize, ok := col.keys.(CustomSerialization); ok {
|
|
if err := serialize.ReadStatePrefix(reader); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if serialize, ok := col.values.(CustomSerialization); ok {
|
|
if err := serialize.ReadStatePrefix(reader); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (col *Map) WriteStatePrefix(encoder *proto.Buffer) error {
|
|
if serialize, ok := col.keys.(CustomSerialization); ok {
|
|
if err := serialize.WriteStatePrefix(encoder); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if serialize, ok := col.values.(CustomSerialization); ok {
|
|
if err := serialize.WriteStatePrefix(encoder); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (col *Map) row(n int) reflect.Value {
|
|
var (
|
|
prev int64
|
|
value = reflect.MakeMap(col.scanType)
|
|
)
|
|
if n != 0 {
|
|
prev = col.offsets.col.Row(n - 1)
|
|
}
|
|
var (
|
|
size = int(col.offsets.col.Row(n) - prev)
|
|
from = int(prev)
|
|
)
|
|
for next := 0; next < size; next++ {
|
|
mapValue := col.values.Row(from+next, false)
|
|
var mapReflectValue reflect.Value
|
|
if mapValue == nil {
|
|
// Convert interface{} nil to typed nil (such as nil *string) to preserve map element
|
|
mapReflectValue = reflect.New(value.Type().Elem()).Elem()
|
|
} else {
|
|
mapReflectValue = reflect.ValueOf(mapValue)
|
|
}
|
|
|
|
value.SetMapIndex(
|
|
reflect.ValueOf(col.keys.Row(from+next, false)),
|
|
mapReflectValue,
|
|
)
|
|
}
|
|
return value
|
|
}
|
|
|
|
func (col *Map) orderedRow(n int) ([]any, []any) {
|
|
var prev int64
|
|
if n != 0 {
|
|
prev = col.offsets.col.Row(n - 1)
|
|
}
|
|
var (
|
|
size = int(col.offsets.col.Row(n) - prev)
|
|
from = int(prev)
|
|
)
|
|
keys := make([]any, size)
|
|
values := make([]any, size)
|
|
for next := 0; next < size; next++ {
|
|
keys[next] = col.keys.Row(from+next, false)
|
|
values[next] = col.values.Row(from+next, false)
|
|
}
|
|
return keys, values
|
|
}
|
|
|
|
var (
|
|
_ Interface = (*Map)(nil)
|
|
_ CustomSerialization = (*Map)(nil)
|
|
)
|