package executor

import (
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"math"
	"math/big"
	"net"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/typemap"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/types"
)

// QueryResultTranslator converts QueryResult to CQL protocol frames
type QueryResultTranslator struct{}

// NewQueryResultTranslator creates a new result translator
func NewQueryResultTranslator() *QueryResultTranslator {
	return &QueryResultTranslator{}
}

const (
	rowBufferDefaultCap = 1024
	rowBufferMaxCap     = 64 * 1024
)

var rowBufferPool = sync.Pool{
	New: func() interface{} {
		return make([]byte, 0, rowBufferDefaultCap)
	},
}

// BuildPreparedResult builds the PREPARED result body for a prepared statement.
func (t *QueryResultTranslator) BuildPreparedResult(prepared *PreparedStatement, protoVer byte) []byte {
	if prepared == nil {
		return nil
	}

	buf := make([]byte, 0, 64)
	buf = appendInt32(buf, int32(ResultKindPrepared))
	buf = appendShortBytes(buf, []byte(prepared.ID))

	// Result metadata ID for protocol v5+.
	if protoVer >= 5 {
		metadataID := prepared.MetadataID
		if metadataID == nil {
			metadataID = []byte{}
		}
		buf = appendShortBytes(buf, metadataID)
	}

	// Prepared metadata: [flags][column_count][pk_count][column_specs...]
	paramCount := len(prepared.Parameters)
	buf = appendInt32(buf, 0)
	buf = appendInt32(buf, int32(paramCount))
	if protoVer >= 4 {
		buf = appendInt32(buf, 0)
	}
	for _, col := range prepared.Parameters {
		buf = t.appendColumnSpec(buf, col)
	}

	// Result metadata: [flags][column_count][column_specs...]
	resultCount := len(prepared.Columns)
	buf = appendInt32(buf, 0)
	buf = appendInt32(buf, int32(resultCount))
	for _, col := range prepared.Columns {
		buf = t.appendColumnSpec(buf, col)
	}

	return buf
}

// ResultKind represents the type of result in a CQL response
type ResultKind int32

// ResultKind values identify RESULT frame kinds.
const (
	ResultKindVoid         ResultKind = 0x0001
	ResultKindRows         ResultKind = 0x0002
	ResultKindSetKeyspace  ResultKind = 0x0003
	ResultKindPrepared     ResultKind = 0x0004
	ResultKindSchemaChange ResultKind = 0x0005
)

const (
	resultMetadataGlobalSpec uint32 = 0x0001
	resultMetadataHasMore    uint32 = 0x0002
	millisecondsInADay       int64  = 24 * 60 * 60 * 1000
)

// CQLType represents CQL data type IDs
type CQLType int

// CQLType values represent CQL type IDs.
const (
	CQLTypeCustom    CQLType = 0x0000
	CQLTypeAscii     CQLType = 0x0001
	CQLTypeBigInt    CQLType = 0x0002
	CQLTypeBlob      CQLType = 0x0003
	CQLTypeBoolean   CQLType = 0x0004
	CQLTypeCounter   CQLType = 0x0005
	CQLTypeDecimal   CQLType = 0x0006
	CQLTypeDouble    CQLType = 0x0007
	CQLTypeFloat     CQLType = 0x0008
	CQLTypeInt       CQLType = 0x0009
	CQLTypeTimestamp CQLType = 0x000B
	CQLTypeUUID      CQLType = 0x000C
	CQLTypeVarchar   CQLType = 0x000D
	CQLTypeVarint    CQLType = 0x000E
	CQLTypeTimeUUID  CQLType = 0x000F
	CQLTypeInet      CQLType = 0x0010
	CQLTypeDate      CQLType = 0x0011
	CQLTypeTime      CQLType = 0x0012
	CQLTypeSmallInt  CQLType = 0x0013
	CQLTypeTinyInt   CQLType = 0x0014
	CQLTypeDuration  CQLType = 0x0015
	CQLTypeList      CQLType = 0x0020
	CQLTypeMap       CQLType = 0x0021
	CQLTypeSet       CQLType = 0x0022
	CQLTypeUDT       CQLType = 0x0030
	CQLTypeTuple     CQLType = 0x0031
	CQLTypeVector    CQLType = 0x0032
)

// TranslateToResultFrame translates a QueryResult to a CQL RESULT frame (returns body and opcode)
func (t *QueryResultTranslator) TranslateToResultFrame(result *QueryResult, stream int16) (*FrameData, error) {
	if result.SetKeyspace != "" {
		return t.createSetKeyspaceResult(result.SetKeyspace, stream)
	}
	if result.SchemaChange != nil {
		return t.createSchemaChangeResult(result.SchemaChange, stream)
	}

	// For SELECT queries, return ROWS result
	if len(result.Columns) > 0 {
		return t.createRowsResult(result, stream)
	}

	// For INSERT/UPDATE/DELETE, return VOID result
	return t.createVoidResult(stream)
}

// createSetKeyspaceResult creates a SET_KEYSPACE result frame.
func (t *QueryResultTranslator) createSetKeyspaceResult(keyspace string, stream int16) (*FrameData, error) {
	buf := make([]byte, 0, 4+2+len(keyspace))
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(ResultKindSetKeyspace))
	buf = appendShortString(buf, keyspace)

	return &FrameData{
		Body:    buf,
		Opcode:  0x08, // OpcodeResult
		Version: 0x84, // ProtoVersion4
		Stream:  stream,
	}, nil
}

func (t *QueryResultTranslator) createSchemaChangeResult(change *SchemaChange, stream int16) (*FrameData, error) {
	if change == nil {
		return nil, fmt.Errorf("schema change is required")
	}

	buf := make([]byte, 0, 64)
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(ResultKindSchemaChange))
	buf = appendShortString(buf, change.ChangeType)
	buf = appendShortString(buf, change.Target)

	switch change.Target {
	case "KEYSPACE":
		buf = appendShortString(buf, change.Keyspace)
	case "TABLE", "TYPE":
		buf = appendShortString(buf, change.Keyspace)
		buf = appendShortString(buf, change.Name)
	default:
		return nil, fmt.Errorf("unsupported schema change target %q", change.Target)
	}

	return &FrameData{
		Body:    buf,
		Opcode:  0x08, // OpcodeResult
		Version: 0x84, // ProtoVersion4
		Stream:  stream,
	}, nil
}

// FrameData represents CQL frame data (body + metadata)
type FrameData struct {
	Body    []byte
	Opcode  byte
	Version byte
	Stream  int16
}

// createVoidResult creates a VOID result frame (no data returned)
func (t *QueryResultTranslator) createVoidResult(stream int16) (*FrameData, error) {
	body := make([]byte, 4)

	// Write result kind (4 bytes, big endian)
	binary.BigEndian.PutUint32(body[0:4], uint32(ResultKindVoid))

	return &FrameData{
		Body:    body,
		Opcode:  0x08, // OpcodeResult
		Version: 0x84, // ProtoVersion4
		Stream:  stream,
	}, nil
}

// createRowsResult creates a ROWS result frame (SELECT query results)
func (t *QueryResultTranslator) createRowsResult(result *QueryResult, stream int16) (*FrameData, error) {
	buf := make([]byte, 0, 1024)

	// Result kind (4 bytes)
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(ResultKindRows))

	ksName, tableName, ok := globalSpecInfo(result.Columns)
	useGlobalSpec := ok

	flags := uint32(0)
	if useGlobalSpec {
		flags |= resultMetadataGlobalSpec
	}
	if result.HasMorePages && len(result.PagingState) > 0 {
		flags |= resultMetadataHasMore
	}

	// Flags (4 bytes)
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], flags)

	// Column count (4 bytes)
	colCount := len(result.Columns)
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(colCount))

	if flags&resultMetadataHasMore != 0 {
		buf = appendBytes(buf, result.PagingState)
	}

	// Column specifications
	if useGlobalSpec {
		buf = appendShortString(buf, ksName)
		buf = appendShortString(buf, tableName)
		for _, col := range result.Columns {
			buf = appendShortString(buf, col.Name)
			buf = append(buf, t.writeTypeInfo(col.Type, col.TypeInfo)...)
		}
	} else {
		for _, col := range result.Columns {
			buf = t.appendColumnSpec(buf, col)
		}
	}

	// Row count (4 bytes)
	rowCount := len(result.Rows)
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(rowCount))

	// Rows
	for _, row := range result.Rows {
		rowBuf := getRowBuffer()
		rowBuf = t.appendRow(rowBuf, row, result.Columns)
		buf = append(buf, rowBuf...)
		putRowBuffer(rowBuf)
	}

	return &FrameData{
		Body:    buf,
		Opcode:  0x08, // OpcodeResult
		Version: 0x84, // ProtoVersion4
		Stream:  stream,
	}, nil
}

func globalSpecInfo(columns []ColumnInfo) (string, string, bool) {
	if len(columns) == 0 {
		return "", "", false
	}
	keyspace := columns[0].Keyspace
	table := columns[0].Table
	for _, col := range columns[1:] {
		if col.Keyspace != keyspace || col.Table != table {
			return "", "", false
		}
	}
	return keyspace, table, true
}

// translateColumnSpec translates a ColumnInfo to CQL column spec format
func (t *QueryResultTranslator) translateColumnSpec(col ColumnInfo) []byte {
	return t.appendColumnSpec(make([]byte, 0, 64), col)
}

func (t *QueryResultTranslator) appendColumnSpec(buf []byte, col ColumnInfo) []byte {
	buf = appendShortString(buf, col.Keyspace)
	buf = appendShortString(buf, col.Table)
	buf = appendShortString(buf, col.Name)
	buf = append(buf, t.writeTypeInfo(col.Type, col.TypeInfo)...)
	return buf
}

// writeTypeInfo writes a complete TypeInfo structure for the given CQL type string.
// This handles both simple types and complex types like set<>, list<>, map<>, tuple<>, udt, and vector.
func (t *QueryResultTranslator) writeTypeInfo(cqlType string, typeInfo types.TypeInfo) []byte {
	buf := make([]byte, 0)

	typeStr := strings.TrimSpace(cqlType)
	if strings.HasPrefix(typeStr, "frozen<") && strings.HasSuffix(typeStr, ">") {
		typeStr = typeStr[7 : len(typeStr)-1]
	}

	switch info := typeInfo.(type) {
	case *types.UDTTypeInfo:
		buf = append(buf, make([]byte, 2)...)
		binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(CQLTypeUDT))
		buf = appendShortString(buf, info.Keyspace)
		buf = appendShortString(buf, info.Name)
		buf = append(buf, make([]byte, 2)...)
		binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(len(info.Fields)))
		for _, field := range info.Fields {
			buf = appendShortString(buf, field.Name)
			buf = append(buf, t.writeTypeInfo(field.Type, field.TypeInfo)...)
		}
		return buf
	case *types.VectorTypeInfo:
		return t.writeVectorTypeInfo(typeStr, info)
	}

	if strings.HasPrefix(typeStr, "'") && strings.HasSuffix(typeStr, "'") {
		buf = append(buf, make([]byte, 2)...)
		binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(CQLTypeCustom))
		customName := strings.Trim(typeStr, "'")
		buf = appendShortString(buf, customName)
		return buf
	}

	if strings.HasPrefix(typeStr, "vector<") {
		return t.writeVectorTypeInfo(typeStr, nil)
	}

	if strings.HasPrefix(typeStr, "set<") || strings.HasPrefix(typeStr, "list<") {
		elemType := innerType(typeStr)
		var elemInfo types.TypeInfo
		switch info := typeInfo.(type) {
		case *types.SetTypeInfo:
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		case *types.ListTypeInfo:
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		}
		code := CQLTypeList
		if strings.HasPrefix(typeStr, "set<") {
			code = CQLTypeSet
		}
		buf = append(buf, make([]byte, 2)...)
		binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(code))
		buf = append(buf, t.writeTypeInfo(elemType, elemInfo)...)
		return buf
	}

	if strings.HasPrefix(typeStr, "map<") {
		parts := splitTopLevelTypes(innerType(typeStr))
		if len(parts) >= 2 {
			keyType := parts[0]
			valType := parts[1]
			var keyInfo types.TypeInfo
			var valInfo types.TypeInfo
			if info, ok := typeInfo.(*types.MapTypeInfo); ok {
				if info.KeyType != "" {
					keyType = info.KeyType
				}
				if info.ValueType != "" {
					valType = info.ValueType
				}
				keyInfo = info.KeyInfo
				valInfo = info.ValueInfo
			}
			buf = append(buf, make([]byte, 2)...)
			binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(CQLTypeMap))
			buf = append(buf, t.writeTypeInfo(keyType, keyInfo)...)
			buf = append(buf, t.writeTypeInfo(valType, valInfo)...)
		}
		return buf
	}

	if strings.HasPrefix(typeStr, "tuple<") {
		parts := splitTopLevelTypes(innerType(typeStr))
		var elemInfos []types.TypeInfo
		if info, ok := typeInfo.(*types.TupleTypeInfo); ok {
			if len(info.ElemTypes) > 0 {
				parts = info.ElemTypes
			}
			elemInfos = info.ElemInfos
		}
		buf = append(buf, make([]byte, 2)...)
		binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(CQLTypeTuple))
		buf = append(buf, make([]byte, 2)...)
		binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(len(parts)))
		for i, part := range parts {
			var info types.TypeInfo
			if i < len(elemInfos) {
				info = elemInfos[i]
			}
			buf = append(buf, t.writeTypeInfo(part, info)...)
		}
		return buf
	}

	typeCode := t.mysqlTypeToCQLType(typeStr)
	buf = append(buf, make([]byte, 2)...)
	binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(typeCode))
	return buf
}

// translateRow translates a row to CQL row format
func (t *QueryResultTranslator) translateRow(row Row, columns []ColumnInfo) []byte {
	return t.appendRow(make([]byte, 0, 64), row, columns)
}

func (t *QueryResultTranslator) appendRow(buf []byte, row Row, columns []ColumnInfo) []byte {
	for i, value := range row.Values {
		valueBytes := t.translateValue(value, columns[i].Type, columns[i].TypeInfo)

		// Length prefix (4 bytes)
		buf = appendInt32(buf, 0)
		if valueBytes == nil {
			// NULL value - set length to -1 (0xFFFFFFFF)
			binary.BigEndian.PutUint32(buf[len(buf)-4:], 0xFFFFFFFF)
		} else {
			binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(valueBytes)))
			buf = append(buf, valueBytes...)
		}
	}

	return buf
}

// translateValue translates a value to CQL wire format.
func (t *QueryResultTranslator) translateValue(value interface{}, cqlType string, typeInfo types.TypeInfo) []byte {
	if value == nil {
		return nil
	}

	// Handle complex types that need special serialization based on CQL type
	// First, normalize the type string (handle frozen<>, set<>, list<>, map<>, etc.)
	normalizedType := normalizeTypeString(cqlType)
	if strings.HasPrefix(normalizedType, "frozen<") {
		normalizedType = normalizedType[7 : len(normalizedType)-1]
	}

	if info, ok := typeInfo.(*types.UDTTypeInfo); ok {
		return t.writeUDTValue(value, info)
	}
	if info, ok := typeInfo.(*types.VectorTypeInfo); ok {
		return t.writeVectorValue(value, normalizedType, info)
	}

	// Handle set, list, map types
	if strings.HasPrefix(normalizedType, "set<") || strings.HasPrefix(normalizedType, "list<") {
		return t.writeCollection(value, normalizedType, typeInfo)
	}
	if strings.HasPrefix(normalizedType, "map<") {
		return t.writeMap(value, normalizedType, typeInfo)
	}
	if strings.HasPrefix(normalizedType, "tuple<") {
		return t.writeTuple(value, normalizedType, typeInfo)
	}
	if strings.HasPrefix(normalizedType, "vector<") {
		return t.writeVectorValue(value, normalizedType, nil)
	}
	if strings.HasPrefix(normalizedType, "'") && strings.HasSuffix(normalizedType, "'") {
		if str, ok := value.(string); ok {
			return t.writeString(str)
		}
		if bytes, ok := value.([]byte); ok {
			return t.writeBytes(bytes)
		}
		return t.writeString(fmt.Sprintf("%v", value))
	}

	// Handle simple types based on CQL type
	switch normalizedType {
	case "inet":
		// Value is a string like "127.0.0.1", need to convert to IP bytes
		if str, ok := value.(string); ok {
			return t.writeInet(str)
		}
		// If it's already bytes, return as-is
		if bytes, ok := value.([]byte); ok {
			return bytes
		}
		return t.writeString(fmt.Sprintf("%v", value))

	case "uuid", "timeuuid":
		// UUIDs must be 16 bytes in CQL wire format
		if str, ok := value.(string); ok {
			return t.writeUUID(str)
		}
		// If it's already bytes, return as-is (assuming 16 bytes)
		if bytes, ok := value.([]byte); ok {
			return bytes
		}
		// Fallback: try to format as string then convert
		return t.writeUUID(fmt.Sprintf("%v", value))

	case "timestamp":
		if ts, ok := parseTimestampValue(value); ok {
			return t.writeTimestamp(ts)
		}
		if str, ok := value.(string); ok {
			return t.writeString(str)
		}
		return t.writeString(fmt.Sprintf("%v", value))

	case "date":
		switch v := value.(type) {
		case time.Time:
			return t.writeDate(v)
		case string:
			if parsed, ok := parseDateString(v); ok {
				return t.writeDate(parsed)
			}
			return t.writeString(v)
		case []byte:
			s := string(v)
			if parsed, ok := parseDateString(s); ok {
				return t.writeDate(parsed)
			}
			return t.writeString(s)
		default:
			return t.writeString(fmt.Sprintf("%v", value))
		}

	case "time":
		switch v := value.(type) {
		case time.Duration:
			return t.writeTime(v)
		case time.Time:
			return t.writeTimeOfDay(v)
		case string:
			if parsed, ok := parseTimeString(v); ok {
				return t.writeTime(parsed)
			}
			return t.writeString(v)
		case []byte:
			s := string(v)
			if parsed, ok := parseTimeString(s); ok {
				return t.writeTime(parsed)
			}
			return t.writeString(s)
		default:
			return t.writeString(fmt.Sprintf("%v", value))
		}

	case "blob":
		switch v := value.(type) {
		case []byte:
			return t.writeBytes(v)
		case string:
			if decoded, ok := parseHexBytes(v); ok {
				return t.writeBytes(decoded)
			}
			return t.writeString(v)
		default:
			return t.writeString(fmt.Sprintf("%v", value))
		}

	case "boolean":
		if b, ok := boolFromValue(value); ok {
			return t.writeBoolean(b)
		}
		return t.writeBoolean(false)

	case "int":
		return t.writeInt32(toInt32(value))
	case "smallint":
		return t.writeInt16(toInt16(value))
	case "tinyint":
		return t.writeInt8(toInt8(value))

	case "bigint", "counter":
		switch v := value.(type) {
		case int:
			return t.writeInt64(int64(v))
		case int8:
			return t.writeInt64(int64(v))
		case int16:
			return t.writeInt64(int64(v))
		case int32:
			return t.writeInt64(int64(v))
		case int64:
			return t.writeInt64(v)
		case uint:
			return t.writeInt64(int64(v))
		case uint8:
			return t.writeInt64(int64(v))
		case uint16:
			return t.writeInt64(int64(v))
		case uint32:
			return t.writeInt64(int64(v))
		case uint64:
			return t.writeInt64(int64(v))
		case string:
			var i int64
			fmt.Sscanf(v, "%d", &i)
			return t.writeInt64(i)
		case json.Number:
			if i, err := v.Int64(); err == nil {
				return t.writeInt64(i)
			}
		default:
			return t.writeInt64(0)
		}

	case "float":
		switch v := value.(type) {
		case float32:
			return t.writeFloat32(v)
		case float64:
			return t.writeFloat32(float32(v))
		case string:
			var f float64
			fmt.Sscanf(v, "%f", &f)
			return t.writeFloat32(float32(f))
		case json.Number:
			if f, err := v.Float64(); err == nil {
				return t.writeFloat32(float32(f))
			}
		default:
			return t.writeFloat32(0)
		}

	case "double":
		switch v := value.(type) {
		case float64:
			return t.writeFloat64(v)
		case float32:
			return t.writeFloat64(float64(v))
		case string:
			var f float64
			fmt.Sscanf(v, "%f", &f)
			return t.writeFloat64(f)
		case json.Number:
			if f, err := v.Float64(); err == nil {
				return t.writeFloat64(f)
			}
		default:
			return t.writeFloat64(0)
		}

	case "decimal":
		return encodeDecimalValue(value)
	case "varint":
		return encodeVarintValue(value)
	case "duration":
		if encoded, ok := encodeDurationValue(value); ok {
			return encoded
		}
		return nil

	case "text", "varchar", "ascii":
		if str, ok := value.(string); ok {
			return t.writeString(str)
		}
		return t.writeString(fmt.Sprintf("%v", value))

	default:
		// Fallback based on Go type
		switch v := value.(type) {
		case int:
			return t.writeInt32(int32(v))
		case int8:
			return t.writeInt32(int32(v))
		case int16:
			return t.writeInt32(int32(v))
		case int32:
			return t.writeInt32(v)
		case int64:
			return t.writeInt64(v)
		case uint:
			return t.writeInt64(int64(v))
		case uint8:
			return t.writeInt32(int32(v))
		case uint16:
			return t.writeInt32(int32(v))
		case uint32:
			return t.writeInt64(int64(v))
		case uint64:
			return t.writeInt64(int64(v))
		case float32:
			return t.writeFloat32(v)
		case float64:
			return t.writeFloat64(v)
		case bool:
			return t.writeBoolean(v)
		case string:
			return t.writeString(v)
		case []byte:
			return t.writeBytes(v)
		case time.Time:
			return t.writeTimestamp(v)
		default:
			return t.writeString(fmt.Sprintf("%v", v))
		}
	}

	return nil
}

func encodeDecimalValue(value interface{}) []byte {
	scale, unscaled, ok := decimalPartsFromValue(value)
	if !ok {
		return nil
	}
	buf := make([]byte, 4)
	binary.BigEndian.PutUint32(buf, uint32(scale))
	return append(buf, encodeSignedBigInt(unscaled)...)
}

func encodeVarintValue(value interface{}) []byte {
	intVal, ok := bigIntFromValue(value)
	if !ok {
		return nil
	}
	return encodeSignedBigInt(intVal)
}

func decimalPartsFromValue(value interface{}) (int32, *big.Int, bool) {
	switch v := value.(type) {
	case int:
		return 0, big.NewInt(int64(v)), true
	case int8:
		return 0, big.NewInt(int64(v)), true
	case int16:
		return 0, big.NewInt(int64(v)), true
	case int32:
		return 0, big.NewInt(int64(v)), true
	case int64:
		return 0, big.NewInt(v), true
	case uint:
		return 0, new(big.Int).SetUint64(uint64(v)), true
	case uint8:
		return 0, new(big.Int).SetUint64(uint64(v)), true
	case uint16:
		return 0, new(big.Int).SetUint64(uint64(v)), true
	case uint32:
		return 0, new(big.Int).SetUint64(uint64(v)), true
	case uint64:
		return 0, new(big.Int).SetUint64(v), true
	case float32:
		return decimalPartsFromString(strconv.FormatFloat(float64(v), 'f', -1, 32))
	case float64:
		return decimalPartsFromString(strconv.FormatFloat(v, 'f', -1, 64))
	case string:
		return decimalPartsFromString(v)
	case []byte:
		return decimalPartsFromString(string(v))
	case fmt.Stringer:
		return decimalPartsFromString(v.String())
	default:
		return decimalPartsFromString(fmt.Sprint(value))
	}
}

func decimalPartsFromString(raw string) (int32, *big.Int, bool) {
	s := strings.TrimSpace(raw)
	if s == "" {
		return 0, nil, false
	}
	sign := 1
	switch s[0] {
	case '-':
		sign = -1
		s = s[1:]
	case '+':
		s = s[1:]
	}
	if s == "" {
		return 0, nil, false
	}

	exp := int64(0)
	if idx := strings.IndexAny(s, "eE"); idx != -1 {
		expPart := s[idx+1:]
		s = s[:idx]
		if expPart == "" {
			return 0, nil, false
		}
		parsedExp, err := strconv.ParseInt(expPart, 10, 32)
		if err != nil {
			return 0, nil, false
		}
		exp = parsedExp
	}

	parts := strings.SplitN(s, ".", 2)
	intPart := parts[0]
	fracPart := ""
	if len(parts) == 2 {
		fracPart = parts[1]
	}
	if intPart == "" {
		intPart = "0"
	}

	scale := int64(len(fracPart)) - exp
	digits := intPart + fracPart
	if exp > 0 && exp > int64(len(fracPart)) {
		zeros := int(exp - int64(len(fracPart)))
		digits += strings.Repeat("0", zeros)
		scale = 0
	}
	if scale < 0 || scale > math.MaxInt32 {
		return 0, nil, false
	}

	digits = strings.TrimLeft(digits, "0")
	if digits == "" {
		digits = "0"
	}

	unscaled := new(big.Int)
	if _, ok := unscaled.SetString(digits, 10); !ok {
		return 0, nil, false
	}
	if sign < 0 {
		unscaled.Neg(unscaled)
	}
	return int32(scale), unscaled, true
}

func bigIntFromValue(value interface{}) (*big.Int, bool) {
	switch v := value.(type) {
	case int:
		return big.NewInt(int64(v)), true
	case int8:
		return big.NewInt(int64(v)), true
	case int16:
		return big.NewInt(int64(v)), true
	case int32:
		return big.NewInt(int64(v)), true
	case int64:
		return big.NewInt(v), true
	case uint:
		return new(big.Int).SetUint64(uint64(v)), true
	case uint8:
		return new(big.Int).SetUint64(uint64(v)), true
	case uint16:
		return new(big.Int).SetUint64(uint64(v)), true
	case uint32:
		return new(big.Int).SetUint64(uint64(v)), true
	case uint64:
		return new(big.Int).SetUint64(v), true
	case string:
		return parseBigIntString(v)
	case []byte:
		return parseBigIntString(string(v))
	case fmt.Stringer:
		return parseBigIntString(v.String())
	default:
		return parseBigIntString(fmt.Sprint(value))
	}
}

func parseBigIntString(raw string) (*big.Int, bool) {
	s := strings.TrimSpace(raw)
	if s == "" || strings.ContainsAny(s, ".eE") {
		return nil, false
	}
	val := new(big.Int)
	if _, ok := val.SetString(s, 10); !ok {
		return nil, false
	}
	return val, true
}

func encodeSignedBigInt(value *big.Int) []byte {
	if value == nil || value.Sign() == 0 {
		return []byte{0}
	}

	if value.Sign() > 0 {
		bytes := value.Bytes()
		if bytes[0]&0x80 != 0 {
			return append([]byte{0x00}, bytes...)
		}
		return bytes
	}

	abs := new(big.Int).Abs(value)
	bits := abs.BitLen()
	length := (bits + 7) / 8
	if length == 0 {
		length = 1
	}

	limit := new(big.Int).Lsh(big.NewInt(1), uint(length*8-1))
	if abs.Cmp(limit) == 1 {
		length++
	}

	buf := make([]byte, length)
	abs.FillBytes(buf)
	for i := range buf {
		buf[i] = ^buf[i]
	}
	carry := byte(1)
	for i := len(buf) - 1; i >= 0; i-- {
		sum := buf[i] + carry
		buf[i] = sum
		if sum != 0 {
			carry = 0
			break
		}
	}

	return buf
}

// writeCollection writes a set or list value
func (t *QueryResultTranslator) writeCollection(value interface{}, cqlType string, typeInfo types.TypeInfo) []byte {
	buf := make([]byte, 0)

	// Get the element type from set<type> or list<type>
	var elemType string
	var elemInfo types.TypeInfo
	if strings.HasPrefix(cqlType, "set<") {
		elemType = innerType(cqlType)
		if info, ok := typeInfo.(*types.SetTypeInfo); ok {
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		}
	} else if strings.HasPrefix(cqlType, "list<") {
		elemType = innerType(cqlType)
		if info, ok := typeInfo.(*types.ListTypeInfo); ok {
			if info.ElemType != "" {
				elemType = info.ElemType
			}
			elemInfo = info.ElemInfo
		}
	}

	// Handle slice values
	var elements []interface{}
	if value == nil {
		elements = []interface{}{}
	} else {
		switch v := value.(type) {
		case []interface{}:
			elements = v
		default:
			rv := reflect.ValueOf(value)
			if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) {
				for i := 0; i < rv.Len(); i++ {
					elements = append(elements, rv.Index(i).Interface())
				}
			} else {
				elements = []interface{}{v}
			}
		}
	}

	// Write element count (4 bytes)
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(elements)))

	// Write each element
	for _, elem := range elements {
		elemBytes := t.translateValue(elem, elemType, elemInfo)
		// Each element is prefixed with its length (4 bytes)
		buf = append(buf, make([]byte, 4)...)
		binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(elemBytes)))
		buf = append(buf, elemBytes...)
	}

	return buf
}

func normalizeTypeString(typeStr string) string {
	trimmed := strings.TrimSpace(typeStr)
	if trimmed == "" {
		return trimmed
	}
	if strings.Contains(trimmed, "'") {
		return trimmed
	}

	lower := strings.ToLower(trimmed)
	if strings.Contains(lower, "<") {
		return lower
	}

	base := lower
	if idx := strings.IndexAny(base, " ("); idx != -1 {
		base = base[:idx]
	}

	switch base {
	case "text", "varchar", "ascii",
		"int", "bigint", "uuid", "timeuuid", "inet", "boolean",
		"timestamp", "date", "time",
		"double", "float",
		"blob",
		"smallint", "tinyint",
		"decimal", "varint", "duration",
		"counter":
		return base
	default:
		return typemap.MySQLTypeToCQLTypeString(base)
	}
}

// writeMap writes a map value
func (t *QueryResultTranslator) writeMap(value interface{}, cqlType string, typeInfo types.TypeInfo) []byte {
	buf := make([]byte, 0)

	// Parse map<keyType, valueType>
	parts := splitTopLevelTypes(innerType(cqlType))
	if len(parts) < 2 {
		return buf
	}
	keyType := parts[0]
	valType := parts[1]
	var keyInfo types.TypeInfo
	var valInfo types.TypeInfo
	if info, ok := typeInfo.(*types.MapTypeInfo); ok {
		if info.KeyType != "" {
			keyType = info.KeyType
		}
		if info.ValueType != "" {
			valType = info.ValueType
		}
		keyInfo = info.KeyInfo
		valInfo = info.ValueInfo
	}

	if value == nil {
		return buf
	}

	rv := reflect.ValueOf(value)
	if rv.Kind() != reflect.Map {
		return buf
	}

	// Write pair count (4 bytes)
	keys := rv.MapKeys()
	buf = append(buf, make([]byte, 4)...)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(keys)))

	// Write each key-value pair
	for _, key := range keys {
		val := rv.MapIndex(key)
		keyBytes := t.translateValue(key.Interface(), keyType, keyInfo)
		valBytes := t.translateValue(val.Interface(), valType, valInfo)

		// Key with length prefix
		buf = append(buf, make([]byte, 4)...)
		binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(keyBytes)))
		buf = append(buf, keyBytes...)

		// Value with length prefix
		buf = append(buf, make([]byte, 4)...)
		binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(valBytes)))
		buf = append(buf, valBytes...)
	}

	return buf
}

func (t *QueryResultTranslator) writeUDTValue(value interface{}, info *types.UDTTypeInfo) []byte {
	if info == nil {
		return nil
	}

	var values []interface{}
	var valueMap map[string]interface{}

	switch v := value.(type) {
	case map[string]interface{}:
		valueMap = v
	case []interface{}:
		values = v
	default:
		rv := reflect.ValueOf(value)
		if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) {
			for i := 0; i < rv.Len(); i++ {
				values = append(values, rv.Index(i).Interface())
			}
		}
	}

	buf := make([]byte, 0)
	for i, field := range info.Fields {
		var fieldVal interface{}
		if valueMap != nil {
			if val, ok := valueMap[field.Name]; ok {
				fieldVal = val
			} else if val, ok := valueMap[strings.ToLower(field.Name)]; ok {
				fieldVal = val
			}
		} else if i < len(values) {
			fieldVal = values[i]
		}

		fieldBytes := t.translateValue(fieldVal, field.Type, field.TypeInfo)
		buf = append(buf, make([]byte, 4)...)
		if fieldBytes == nil {
			binary.BigEndian.PutUint32(buf[len(buf)-4:], 0xFFFFFFFF)
			continue
		}
		binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(fieldBytes)))
		buf = append(buf, fieldBytes...)
	}

	return buf
}

func (t *QueryResultTranslator) writeVectorTypeInfo(typeStr string, info *types.VectorTypeInfo) []byte {
	elemType := ""
	size := 0
	var elemInfo types.TypeInfo
	if info != nil {
		elemType = info.ElemType
		size = info.Size
		elemInfo = info.ElemInfo
	}
	if (elemType == "" || size <= 0) && strings.HasPrefix(typeStr, "vector<") {
		if parsedElem, parsedSize, ok := parseVectorType(typeStr); ok {
			elemType = parsedElem
			size = parsedSize
		}
	}
	if elemType == "" {
		elemType = "float"
	}

	buf := make([]byte, 0, 8)
	buf = append(buf, make([]byte, 2)...)
	binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(CQLTypeVector))
	buf = append(buf, t.writeTypeInfo(elemType, elemInfo)...)
	buf = appendInt32(buf, int32(size))
	return buf
}

func (t *QueryResultTranslator) writeVectorValue(value interface{}, cqlType string, info *types.VectorTypeInfo) []byte {
	elemType := ""
	size := 0
	if info != nil {
		elemType = info.ElemType
		size = info.Size
	}
	if (elemType == "" || size <= 0) && strings.HasPrefix(cqlType, "vector<") {
		if parsedElem, parsedSize, ok := parseVectorType(cqlType); ok {
			elemType = parsedElem
			size = parsedSize
		}
	}
	elemBytes := vectorElementBytes(elemType)
	if size <= 0 || elemBytes == 0 {
		return nil
	}
	expectedLen := size * elemBytes

	switch v := value.(type) {
	case []byte:
		if len(v) != expectedLen {
			return nil
		}
		return v
	case []float32:
		if elemBytes != 4 || len(v) != size {
			return nil
		}
		buf := make([]byte, expectedLen)
		for i, f := range v {
			binary.BigEndian.PutUint32(buf[i*4:], math.Float32bits(f))
		}
		return buf
	case []float64:
		if elemBytes != 8 || len(v) != size {
			return nil
		}
		buf := make([]byte, expectedLen)
		for i, f := range v {
			binary.BigEndian.PutUint64(buf[i*8:], math.Float64bits(f))
		}
		return buf
	default:
		rv := reflect.ValueOf(value)
		if !rv.IsValid() || rv.Kind() != reflect.Slice {
			return nil
		}
		if rv.Len() != size {
			return nil
		}
		buf := make([]byte, expectedLen)
		for i := 0; i < rv.Len(); i++ {
			item := rv.Index(i).Interface()
			switch elemBytes {
			case 4:
				f, ok := toFloat32(item)
				if !ok {
					return nil
				}
				binary.BigEndian.PutUint32(buf[i*4:], math.Float32bits(f))
			case 8:
				f, ok := toFloat64(item)
				if !ok {
					return nil
				}
				binary.BigEndian.PutUint64(buf[i*8:], math.Float64bits(f))
			}
		}
		return buf
	}
}

// writeInet converts an IP string to CQL wire format
func (t *QueryResultTranslator) writeInet(ipStr string) []byte {
	// Parse IP address
	ip := net.ParseIP(ipStr)
	if ip == nil {
		return []byte{}
	}

	// Convert to 4-byte (IPv4) or 16-byte (IPv6) format
	if ip4 := ip.To4(); ip4 != nil {
		return []byte(ip4)
	}
	return []byte(ip.To16())
}

// mysqlTypeToCQLType maps MySQL type or CQL type string to CQL type code
func (t *QueryResultTranslator) mysqlTypeToCQLType(typeStr string) CQLType {
	// First check if it's already a CQL type (from system tables)
	switch typeStr {
	case "text", "varchar", "ascii":
		return CQLTypeVarchar
	case "int":
		return CQLTypeInt
	case "bigint":
		return CQLTypeBigInt
	case "uuid":
		return CQLTypeUUID
	case "timeuuid":
		return CQLTypeTimeUUID
	case "inet":
		return CQLTypeInet
	case "boolean":
		return CQLTypeBoolean
	case "timestamp":
		return CQLTypeTimestamp
	case "date":
		return CQLTypeDate
	case "time":
		return CQLTypeTime
	case "double":
		return CQLTypeDouble
	case "float":
		return CQLTypeFloat
	case "blob":
		return CQLTypeBlob
	case "smallint":
		return CQLTypeSmallInt
	case "tinyint":
		return CQLTypeTinyInt
	case "decimal":
		return CQLTypeDecimal
	case "varint":
		return CQLTypeVarint
	case "duration":
		return CQLTypeDuration
	case "counter":
		return CQLTypeCounter
	}

	// Check for complex types (set, list, map, frozen)
	if strings.HasPrefix(typeStr, "set<") {
		return CQLTypeSet
	}
	if strings.HasPrefix(typeStr, "list<") {
		return CQLTypeList
	}
	if strings.HasPrefix(typeStr, "map<") {
		return CQLTypeMap
	}
	if strings.HasPrefix(typeStr, "vector<") {
		return CQLTypeVector
	}
	if strings.HasPrefix(typeStr, "tuple<") {
		return CQLTypeTuple
	}
	if strings.HasPrefix(typeStr, "'") && strings.HasSuffix(typeStr, "'") {
		return CQLTypeCustom
	}
	if strings.HasPrefix(typeStr, "frozen<") {
		// Extract the inner type
		inner := typeStr[7 : len(typeStr)-1]
		return t.mysqlTypeToCQLType(inner)
	}

	// Handle MySQL types (for backward compatibility)
	switch typeStr {
	case "TINYINT":
		return CQLTypeTinyInt
	case "SMALLINT":
		return CQLTypeSmallInt
	case "INT":
		return CQLTypeInt
	case "BIGINT":
		return CQLTypeBigInt
	case "FLOAT":
		return CQLTypeFloat
	case "DOUBLE":
		return CQLTypeDouble
	case "DECIMAL":
		return CQLTypeDecimal
	case "VARCHAR", "CHAR", "TEXT":
		return CQLTypeVarchar
	case "BOOLEAN":
		return CQLTypeBoolean
	case "DATETIME", "TIMESTAMP":
		return CQLTypeTimestamp
	case "DATE":
		return CQLTypeDate
	case "TIME":
		return CQLTypeTime
	case "BLOB", "LONGBLOB", "MEDIUMBLOB", "TINYBLOB", "BINARY", "VARBINARY":
		return CQLTypeBlob
	case "JSON":
		return CQLTypeVarchar
	}

	// Default to varchar
	return CQLTypeVarchar
}

// Write functions for different CQL types

func (t *QueryResultTranslator) writeInt32(v int32) []byte {
	buf := make([]byte, 4)
	binary.BigEndian.PutUint32(buf, uint32(v))
	return buf
}

func (t *QueryResultTranslator) writeInt16(v int16) []byte {
	buf := make([]byte, 2)
	binary.BigEndian.PutUint16(buf, uint16(v))
	return buf
}

func (t *QueryResultTranslator) writeInt8(v int8) []byte {
	return []byte{byte(v)}
}

func (t *QueryResultTranslator) writeInt64(v int64) []byte {
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, uint64(v))
	return buf
}

func (t *QueryResultTranslator) writeFloat32(v float32) []byte {
	buf := make([]byte, 4)
	binary.BigEndian.PutUint32(buf, math.Float32bits(v))
	return buf
}

func (t *QueryResultTranslator) writeFloat64(v float64) []byte {
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, math.Float64bits(v))
	return buf
}

func (t *QueryResultTranslator) writeBoolean(v bool) []byte {
	if v {
		return []byte{0x01}
	}
	return []byte{0x00}
}

func (t *QueryResultTranslator) writeString(v string) []byte {
	return []byte(v)
}

func (t *QueryResultTranslator) writeBytes(v []byte) []byte {
	return v
}

func (t *QueryResultTranslator) writeTimestamp(v time.Time) []byte {
	// CQL timestamps are milliseconds since epoch
	millis := v.UnixNano() / int64(time.Millisecond)
	return t.writeInt64(millis)
}

func (t *QueryResultTranslator) writeDate(v time.Time) []byte {
	millis := v.UTC().UnixNano() / int64(time.Millisecond)
	days := millis/millisecondsInADay + (1 << 31)
	return t.writeInt32(int32(days))
}

func (t *QueryResultTranslator) writeTime(d time.Duration) []byte {
	if d < 0 {
		d = 0
	}
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, uint64(d.Nanoseconds()))
	return buf
}

func (t *QueryResultTranslator) writeTimeOfDay(v time.Time) []byte {
	utc := v.UTC()
	d := time.Duration(utc.Hour())*time.Hour +
		time.Duration(utc.Minute())*time.Minute +
		time.Duration(utc.Second())*time.Second +
		time.Duration(utc.Nanosecond())
	return t.writeTime(d)
}

func parseDateString(value string) (time.Time, bool) {
	if value == "" {
		return time.Time{}, false
	}
	parsed, err := time.Parse("2006-01-02", value)
	if err != nil {
		return time.Time{}, false
	}
	return parsed, true
}

func parseTimeString(value string) (time.Duration, bool) {
	if value == "" {
		return 0, false
	}
	layout := "15:04:05"
	if strings.Contains(value, ".") {
		layout = "15:04:05.999999999"
	}
	parsed, err := time.Parse(layout, value)
	if err != nil {
		return 0, false
	}
	d := time.Duration(parsed.Hour())*time.Hour +
		time.Duration(parsed.Minute())*time.Minute +
		time.Duration(parsed.Second())*time.Second +
		time.Duration(parsed.Nanosecond())
	return d, true
}

func parseTimestampValue(value interface{}) (time.Time, bool) {
	switch v := value.(type) {
	case time.Time:
		return v, true
	case int:
		return time.Unix(0, int64(v)*int64(time.Millisecond)), true
	case int64:
		return time.Unix(0, v*int64(time.Millisecond)), true
	case uint:
		return time.Unix(0, int64(v)*int64(time.Millisecond)), true
	case uint64:
		return time.Unix(0, int64(v)*int64(time.Millisecond)), true
	case json.Number:
		if i, err := v.Int64(); err == nil {
			return time.Unix(0, i*int64(time.Millisecond)), true
		}
		if f, err := v.Float64(); err == nil {
			return time.Unix(0, int64(f)*int64(time.Millisecond)), true
		}
	case string:
		return parseTimestampString(v)
	}
	return time.Time{}, false
}

func parseTimestampString(value string) (time.Time, bool) {
	s := strings.TrimSpace(value)
	if s == "" {
		return time.Time{}, false
	}
	if ms, ok := parseInt64String(s); ok {
		return time.Unix(0, ms*int64(time.Millisecond)), true
	}
	layouts := []string{
		"2006-01-02 15:04:05.000000-0700",
		"2006-01-02 15:04:05.000-0700",
		"2006-01-02 15:04:05-0700",
		"2006-01-02 15:04:05.000000-07:00",
		"2006-01-02 15:04:05.000-07:00",
		"2006-01-02 15:04:05-07:00",
		"2006-01-02T15:04:05.000000-0700",
		"2006-01-02T15:04:05.000-0700",
		"2006-01-02T15:04:05-0700",
		"2006-01-02T15:04:05.000000-07:00",
		"2006-01-02T15:04:05.000-07:00",
		"2006-01-02T15:04:05-07:00",
		"2006-01-02 15:04:05.000000",
		"2006-01-02 15:04:05.000",
		"2006-01-02 15:04:05",
		time.RFC3339Nano,
		time.RFC3339,
	}
	for _, layout := range layouts {
		if parsed, err := time.ParseInLocation(layout, s, time.UTC); err == nil {
			return parsed, true
		}
	}
	return time.Time{}, false
}

func parseInt64String(value string) (int64, bool) {
	if value == "" {
		return 0, false
	}
	n, err := strconv.ParseInt(value, 10, 64)
	if err != nil {
		return 0, false
	}
	return n, true
}

func parseHexBytes(value string) ([]byte, bool) {
	s := strings.TrimSpace(value)
	if strings.HasPrefix(s, "0x") || strings.HasPrefix(s, "0X") {
		s = s[2:]
	}
	if s == "" {
		return []byte{}, true
	}
	if len(s)%2 != 0 {
		return nil, false
	}
	decoded, err := hex.DecodeString(s)
	if err != nil {
		return nil, false
	}
	return decoded, true
}

func boolFromValue(value interface{}) (bool, bool) {
	switch v := value.(type) {
	case bool:
		return v, true
	case int:
		return v != 0, true
	case int8:
		return v != 0, true
	case int16:
		return v != 0, true
	case int32:
		return v != 0, true
	case int64:
		return v != 0, true
	case uint:
		return v != 0, true
	case uint8:
		return v != 0, true
	case uint16:
		return v != 0, true
	case uint32:
		return v != 0, true
	case uint64:
		return v != 0, true
	case string:
		s := strings.TrimSpace(strings.ToLower(v))
		switch s {
		case "true":
			return true, true
		case "false":
			return false, true
		case "1":
			return true, true
		case "0":
			return false, true
		}
		if i, err := strconv.ParseInt(s, 10, 64); err == nil {
			return i != 0, true
		}
	case []byte:
		return boolFromValue(string(v))
	case json.Number:
		if i, err := v.Int64(); err == nil {
			return i != 0, true
		}
		if f, err := v.Float64(); err == nil {
			return f != 0, true
		}
	}
	return false, false
}

// writeUUID converts a UUID string to 16-byte binary format for CQL wire protocol
// UUID format: "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" -> 16 bytes
func (t *QueryResultTranslator) writeUUID(uuidStr string) []byte {
	// Remove dashes from UUID string
	uuidStr = strings.ReplaceAll(uuidStr, "-", "")

	// Validate and parse hex
	if len(uuidStr) != 32 {
		// Invalid UUID, return empty or zero UUID
		return make([]byte, 16)
	}

	bytes, err := hex.DecodeString(uuidStr)
	if err != nil {
		// Invalid hex, return zero UUID
		return make([]byte, 16)
	}

	return bytes
}

func appendShortString(buf []byte, value string) []byte {
	valBytes := []byte(value)
	buf = append(buf, 0, 0)
	binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(len(valBytes)))
	buf = append(buf, valBytes...)
	return buf
}

func appendShortBytes(buf []byte, value []byte) []byte {
	if value == nil {
		buf = append(buf, 0xFF, 0xFF)
		return buf
	}
	buf = append(buf, 0, 0)
	binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(len(value)))
	buf = append(buf, value...)
	return buf
}

func appendInt32(buf []byte, value int32) []byte {
	buf = append(buf, 0, 0, 0, 0)
	binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(value))
	return buf
}

func appendBytes(buf []byte, value []byte) []byte {
	if value == nil {
		return appendInt32(buf, -1)
	}
	buf = appendInt32(buf, int32(len(value)))
	buf = append(buf, value...)
	return buf
}

func getRowBuffer() []byte {
	if buf, ok := rowBufferPool.Get().([]byte); ok {
		return buf[:0]
	}
	return make([]byte, 0, rowBufferDefaultCap)
}

func putRowBuffer(buf []byte) {
	if cap(buf) > rowBufferMaxCap {
		return
	}
	rowBufferPool.Put(buf[:0])
}

func innerType(typeStr string) string {
	start := strings.Index(typeStr, "<")
	end := strings.LastIndex(typeStr, ">")
	if start == -1 || end == -1 || end <= start {
		return ""
	}
	return strings.TrimSpace(typeStr[start+1 : end])
}

func splitTopLevelTypes(typeStr string) []string {
	var parts []string
	depth := 0
	start := 0
	for i, ch := range typeStr {
		switch ch {
		case '<':
			depth++
		case '>':
			depth--
		case ',':
			if depth == 0 {
				parts = append(parts, strings.TrimSpace(typeStr[start:i]))
				start = i + 1
			}
		}
	}
	last := strings.TrimSpace(typeStr[start:])
	if last != "" {
		parts = append(parts, last)
	}
	return parts
}

func vectorElementBytes(elemType string) int {
	switch strings.ToLower(strings.TrimSpace(elemType)) {
	case "float":
		return 4
	case "double":
		return 8
	default:
		return 0
	}
}

func (t *QueryResultTranslator) writeTuple(value interface{}, cqlType string, typeInfo types.TypeInfo) []byte {
	elemTypes := splitTopLevelTypes(innerType(cqlType))
	var elemInfos []types.TypeInfo
	if info, ok := typeInfo.(*types.TupleTypeInfo); ok {
		if len(info.ElemTypes) > 0 {
			elemTypes = info.ElemTypes
		}
		elemInfos = info.ElemInfos
	}
	buf := make([]byte, 0)

	var elements []interface{}
	if value == nil {
		elements = []interface{}{}
	} else {
		rv := reflect.ValueOf(value)
		if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) {
			for i := 0; i < rv.Len(); i++ {
				elements = append(elements, rv.Index(i).Interface())
			}
		} else {
			elements = []interface{}{value}
		}
	}

	for i, elemType := range elemTypes {
		var elem interface{}
		if i < len(elements) {
			elem = elements[i]
		}
		var elemInfo types.TypeInfo
		if i < len(elemInfos) {
			elemInfo = elemInfos[i]
		}
		elemBytes := t.translateValue(elem, elemType, elemInfo)
		buf = append(buf, make([]byte, 4)...)
		if elemBytes == nil {
			binary.BigEndian.PutUint32(buf[len(buf)-4:], 0xFFFFFFFF)
		} else {
			binary.BigEndian.PutUint32(buf[len(buf)-4:], uint32(len(elemBytes)))
			buf = append(buf, elemBytes...)
		}
	}

	return buf
}

func toInt64(value interface{}) int64 {
	switch v := value.(type) {
	case int:
		return int64(v)
	case int8:
		return int64(v)
	case int16:
		return int64(v)
	case int32:
		return int64(v)
	case int64:
		return v
	case uint:
		return int64(v)
	case uint8:
		return int64(v)
	case uint16:
		return int64(v)
	case uint32:
		return int64(v)
	case uint64:
		return int64(v)
	case float32:
		return int64(v)
	case float64:
		return int64(v)
	case string:
		var i int64
		fmt.Sscanf(v, "%d", &i)
		return i
	case json.Number:
		if i, err := v.Int64(); err == nil {
			return i
		}
		if f, err := v.Float64(); err == nil {
			return int64(f)
		}
		return 0
	default:
		return 0
	}
}

func toInt32(value interface{}) int32 {
	return int32(toInt64(value))
}

func toInt16(value interface{}) int16 {
	return int16(toInt64(value))
}

func toInt8(value interface{}) int8 {
	return int8(toInt64(value))
}

func toFloat64(value interface{}) (float64, bool) {
	switch v := value.(type) {
	case float64:
		return v, true
	case float32:
		return float64(v), true
	case int:
		return float64(v), true
	case int8:
		return float64(v), true
	case int16:
		return float64(v), true
	case int32:
		return float64(v), true
	case int64:
		return float64(v), true
	case uint:
		return float64(v), true
	case uint8:
		return float64(v), true
	case uint16:
		return float64(v), true
	case uint32:
		return float64(v), true
	case uint64:
		return float64(v), true
	case json.Number:
		if f, err := v.Float64(); err == nil {
			return f, true
		}
		if i, err := v.Int64(); err == nil {
			return float64(i), true
		}
	case string:
		f, err := strconv.ParseFloat(v, 64)
		if err == nil {
			return f, true
		}
	}
	return 0, false
}

func toFloat32(value interface{}) (float32, bool) {
	if f, ok := toFloat64(value); ok {
		return float32(f), true
	}
	return 0, false
}