package executor

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/binary"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"

	"go.uber.org/zap"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

// PagingOptions describes request paging settings.
type PagingOptions struct {
	PageSize    int
	PagingState []byte
	Query       string
}

type orderColumn struct {
	Name      string
	Direction string
}

type pagingState struct {
	QueryHash      [32]byte
	SchemaVersion  string
	Table          string
	OrderBy        string
	LastPKValues   []byte
	RemainingLimit *int64
}

const (
	pagingNullValue  = -1
	pagingUnsetValue = -2
)

type pagingUnsupportedError struct {
	reason string
}

func (e *pagingUnsupportedError) Error() string {
	return e.reason
}

func newPagingUnsupportedError(reason string) error {
	return &pagingUnsupportedError{reason: reason}
}

func isPagingUnsupported(err error) bool {
	var perr *pagingUnsupportedError
	return errors.As(err, &perr)
}

func (e *Executor) executePagedSelect(ctx context.Context, exec sqlExecutor, user, password string, stmt *parser.SelectStatement, params []interface{}, paging *PagingOptions) (*QueryResult, error) {
	if paging == nil {
		return nil, fmt.Errorf("paging options are required")
	}
	if paging.PageSize <= 0 {
		return nil, fmt.Errorf("page size must be positive")
	}
	if stmt == nil || stmt.From == nil {
		return nil, fmt.Errorf("select statement is required")
	}
	if stmt.Distinct || len(stmt.GroupBy) > 0 || selectHasAggregates(stmt) {
		return nil, newPagingUnsupportedError("paging not supported for aggregate, distinct, or group-by queries")
	}
	if stmt.From.Schema == "" {
		return nil, fmt.Errorf("keyspace is required for paging")
	}

	orderCols, orderKey, err := e.resolvePagingOrder(ctx, user, password, stmt)
	if err != nil {
		return nil, err
	}

	columnInfo, err := e.columnTypesForTable(ctx, user, password, stmt.From)
	if err != nil {
		return nil, err
	}
	orderTypes := make([]string, len(orderCols))
	for i, col := range orderCols {
		info, ok := columnInfo[strings.ToLower(col.Name)]
		if !ok {
			return nil, fmt.Errorf("paging column %q not found", col.Name)
		}
		orderTypes[i] = info.cqlType
	}

	working := cloneSelectStatement(stmt)
	applyPagingOrder(working, orderCols)
	originalCols, addedCols, err := ensurePagingColumnsSelected(working, orderCols)
	if err != nil {
		return nil, err
	}

	limitValue, err := resolvePagingLimit(working, params)
	if err != nil {
		return nil, err
	}
	offsetValue, err := resolvePagingOffset(working, params)
	if err != nil {
		return nil, err
	}
	working.Options.Limit = nil
	working.Options.PerPartitionLimit = nil
	working.Options.Offset = nil

	markerValues, err := bindMarkerValues(stmt, params)
	if err != nil {
		return nil, err
	}

	queryText := strings.TrimSpace(paging.Query)
	if queryText == "" {
		queryText = stmt.String()
	}
	expectedHash := pagingQueryHash(queryText, working.From.Schema)
	currentSchema, err := e.currentSchemaVersion(ctx, user, password)
	if err != nil {
		return nil, err
	}
	expectedTable := fmt.Sprintf("%s.%s", working.From.Schema, working.From.Name)
	remainingLimit := limitValue

	if len(paging.PagingState) > 0 {
		state, err := decodePagingState(paging.PagingState)
		if err != nil {
			return nil, newPagingUnsupportedError("paging state decode failed")
		}
		if state.QueryHash != expectedHash {
			return nil, fmt.Errorf("paging state query hash mismatch")
		}
		if state.SchemaVersion != currentSchema {
			return nil, fmt.Errorf("paging state schema version mismatch")
		}
		if state.Table != expectedTable {
			return nil, fmt.Errorf("paging state table mismatch")
		}
		if state.OrderBy != orderKey {
			return nil, fmt.Errorf("paging state order by mismatch")
		}
		lastValues, err := decodePagingValues(state.LastPKValues, orderTypes)
		if err != nil {
			return nil, newPagingUnsupportedError("paging state values decode failed")
		}
		predicate, predicateValues, err := buildPagingPredicate(orderCols, lastValues)
		if err != nil {
			return nil, err
		}
		if predicate != nil {
			if working.Where == nil {
				working.Where = predicate
			} else {
				working.Where = &parser.BinaryExpr{Op: "AND", Left: working.Where, Right: predicate}
			}
		}
		for expr, val := range predicateValues {
			markerValues[expr] = val
		}
		if state.RemainingLimit != nil {
			remainingLimit = state.RemainingLimit
		}
		offsetValue = nil
	}

	pageLimit := paging.PageSize
	if remainingLimit != nil && int64(pageLimit) > *remainingLimit {
		pageLimit = int(*remainingLimit)
	}
	if pageLimit < 0 {
		pageLimit = 0
	}
	fetchLimit := pageLimit + 1

	markerIndex := buildBindMarkerIndex(working)
	args, err := reorderParams(working, markerValues)
	if err != nil {
		return nil, err
	}
	if working.Where != nil {
		lookup := func(name string) (string, bool) {
			info, ok := columnInfo[strings.ToLower(name)]
			if !ok {
				return "", false
			}
			return normalizeCastType(info.cqlType), true
		}
		rewritten, changed, requiresProxy, err := rewriteTimeFunctionsInExpr(working.Where, args, markerIndex, lookup, true)
		if err != nil {
			return nil, err
		}
		if requiresProxy {
			return nil, newPagingUnsupportedError("paging not supported with time conversion filters")
		}
		if changed {
			working.Where = rewritten
		}
	}

	plan, err := e.buildSelectPlan(ctx, user, password, working, true, false)
	if err != nil {
		return nil, err
	}
	sql := plan.sql + " LIMIT " + strconv.Itoa(fetchLimit)
	if offsetValue != nil && *offsetValue > 0 {
		sql += " OFFSET " + strconv.FormatInt(*offsetValue, 10)
	}
	e.logger.Debug("Executing paged query",
		zap.String("user", user),
		zap.String("cql", queryText),
		zap.String("sql", sql))

	result, err := e.executeSelectWithMeta(ctx, exec, sql, args, plan.meta, mysqlOpRead)
	if err != nil {
		return nil, err
	}

	hasMore := false
	if len(result.Rows) > pageLimit {
		if remainingLimit == nil || *remainingLimit > int64(pageLimit) {
			hasMore = true
		}
	}
	if pageLimit < len(result.Rows) {
		result.Rows = result.Rows[:pageLimit]
	}

	if hasMore && len(result.Rows) > 0 {
		lastRow := result.Rows[len(result.Rows)-1]
		orderValues, err := extractOrderValues(lastRow, result.Columns, orderCols)
		if err != nil {
			return nil, err
		}
		lastValues, err := encodePagingValues(orderValues, orderTypes)
		if err != nil {
			return nil, err
		}
		state := pagingState{
			QueryHash:     expectedHash,
			SchemaVersion: currentSchema,
			Table:         expectedTable,
			OrderBy:       orderKey,
			LastPKValues:  lastValues,
		}
		if remainingLimit != nil {
			next := *remainingLimit - int64(pageLimit)
			if next > 0 {
				state.RemainingLimit = &next
			}
		}
		encoded, err := encodePagingState(state)
		if err != nil {
			return nil, err
		}
		result.HasMorePages = true
		result.PagingState = encoded
	}

	if addedCols > 0 && originalCols <= len(result.Columns) {
		trimResultColumns(result, originalCols)
	}

	if plan.transform.json || len(plan.transform.toJSON) > 0 || len(plan.transform.casts) > 0 || len(plan.transform.dropIndexes) > 0 {
		pagingState := result.PagingState
		hasMorePages := result.HasMorePages
		transformed, err := applySelectTransform(result, plan.transform)
		if err != nil {
			return nil, err
		}
		transformed.PagingState = pagingState
		transformed.HasMorePages = hasMorePages
		return transformed, nil
	}

	return result, nil
}

func (e *Executor) resolvePagingOrder(ctx context.Context, user, password string, stmt *parser.SelectStatement) ([]orderColumn, string, error) {
	if stmt == nil || stmt.From == nil {
		return nil, "", fmt.Errorf("select statement is required")
	}
	if len(stmt.Options.OrderBy) > 0 {
		orderCols := make([]orderColumn, len(stmt.Options.OrderBy))
		for i, ob := range stmt.Options.OrderBy {
			if ob == nil || ob.Column == "" {
				return nil, "", fmt.Errorf("ORDER BY column is required")
			}
			dir := strings.ToUpper(ob.Direction)
			if dir == "" {
				dir = "ASC"
			} else if dir != "ASC" && dir != "DESC" {
				return nil, "", fmt.Errorf("invalid ORDER BY direction %q", ob.Direction)
			}
			orderCols[i] = orderColumn{Name: ob.Column, Direction: dir}
		}
		ok, err := e.orderByIsUnique(ctx, user, password, stmt.From, orderCols)
		if err != nil {
			return nil, "", err
		}
		if !ok {
			return nil, "", newPagingUnsupportedError("ORDER BY must use a unique key for paging")
		}
		return orderCols, orderByKey(orderCols), nil
	}

	pkCols, err := e.primaryKeyColumns(ctx, user, password, stmt.From)
	if err != nil {
		return nil, "", err
	}
	if len(pkCols) == 0 {
		return nil, "", newPagingUnsupportedError("primary key is required for paging")
	}
	orderCols := make([]orderColumn, len(pkCols))
	for i, col := range pkCols {
		orderCols[i] = orderColumn{Name: col, Direction: "ASC"}
	}
	return orderCols, orderByKey(orderCols), nil
}

func (e *Executor) orderByIsUnique(ctx context.Context, user, password string, table *parser.TableName, orderCols []orderColumn) (bool, error) {
	if table == nil {
		return false, fmt.Errorf("table is required")
	}
	if e.schemaManager == nil {
		return false, fmt.Errorf("schema manager is not configured")
	}
	indexes, err := e.schemaManager.IndexesForTable(ctx, user, password, table.Schema, table.Name)
	if err != nil {
		return false, err
	}
	if len(indexes) == 0 {
		return false, nil
	}
	byName := make(map[string][]mysqlbackend.IndexInfo)
	for _, idx := range indexes {
		if idx.NonUnique {
			continue
		}
		byName[idx.Name] = append(byName[idx.Name], idx)
	}
	orderNames := make([]string, len(orderCols))
	for i, col := range orderCols {
		orderNames[i] = strings.ToLower(col.Name)
	}
	for _, entries := range byName {
		sorted := append([]mysqlbackend.IndexInfo(nil), entries...)
		sort.Slice(sorted, func(i, j int) bool { return sorted[i].Seq < sorted[j].Seq })
		if len(sorted) != len(orderNames) {
			continue
		}
		match := true
		for i, idx := range sorted {
			if strings.ToLower(idx.Column) != orderNames[i] {
				match = false
				break
			}
		}
		if match {
			return true, nil
		}
	}
	return false, nil
}

func orderByKey(orderCols []orderColumn) string {
	parts := make([]string, len(orderCols))
	for i, col := range orderCols {
		dir := strings.ToUpper(col.Direction)
		if dir == "" {
			dir = "ASC"
		}
		parts[i] = strings.ToLower(col.Name) + " " + dir
	}
	return strings.Join(parts, ",")
}

func applyPagingOrder(stmt *parser.SelectStatement, orderCols []orderColumn) {
	orderBy := make([]*parser.OrderByClause, len(orderCols))
	for i, col := range orderCols {
		orderBy[i] = &parser.OrderByClause{Column: col.Name, Direction: col.Direction}
	}
	stmt.Options.OrderBy = orderBy
}

func ensurePagingColumnsSelected(stmt *parser.SelectStatement, orderCols []orderColumn) (int, int, error) {
	original := len(stmt.Columns)
	if selectHasStar(stmt.Columns) {
		return original, 0, nil
	}
	canAdd := !stmt.Distinct && len(stmt.GroupBy) == 0
	selected := make(map[string]struct{}, len(stmt.Columns))
	for _, expr := range stmt.Columns {
		if name, ok := columnExprName(expr); ok {
			selected[strings.ToLower(name)] = struct{}{}
		}
	}
	added := 0
	for _, col := range orderCols {
		if _, ok := selected[strings.ToLower(col.Name)]; ok {
			continue
		}
		if !canAdd {
			return original, added, newPagingUnsupportedError("paging requires ORDER BY columns to be selected")
		}
		stmt.Columns = append(stmt.Columns, &parser.ColumnExpr{Name: col.Name})
		selected[strings.ToLower(col.Name)] = struct{}{}
		added++
	}
	return original, added, nil
}

func selectHasStar(columns []parser.Expression) bool {
	for _, expr := range columns {
		if _, ok := expr.(*parser.StarExpr); ok {
			return true
		}
	}
	return false
}

func columnExprName(expr parser.Expression) (string, bool) {
	if expr == nil {
		return "", false
	}
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		expr = aliased.Expr
	}
	col, ok := expr.(*parser.ColumnExpr)
	if !ok {
		return "", false
	}
	return col.Name, true
}

func buildPagingPredicate(orderCols []orderColumn, lastValues []interface{}) (parser.Expression, map[parser.Expression]interface{}, error) {
	if len(orderCols) == 0 {
		return nil, nil, fmt.Errorf("paging columns are required")
	}
	if len(orderCols) != len(lastValues) {
		return nil, nil, fmt.Errorf("paging column/value count mismatch")
	}
	values := make(map[parser.Expression]interface{}, len(orderCols)*2)
	var predicate parser.Expression
	for i := range orderCols {
		branch, branchValues := buildPagingBranch(orderCols, lastValues, i)
		for expr, val := range branchValues {
			values[expr] = val
		}
		if predicate == nil {
			predicate = branch
		} else {
			predicate = &parser.BinaryExpr{Op: "OR", Left: predicate, Right: branch}
		}
	}
	return predicate, values, nil
}

func buildPagingBranch(orderCols []orderColumn, lastValues []interface{}, index int) (parser.Expression, map[parser.Expression]interface{}) {
	values := make(map[parser.Expression]interface{}, index+1)
	var branch parser.Expression
	for i := 0; i <= index; i++ {
		col := &parser.ColumnExpr{Name: orderCols[i].Name}
		marker := &parser.BindMarker{}
		values[marker] = lastValues[i]
		op := "="
		if i == index {
			op = orderComparisonOp(orderCols[i].Direction)
		}
		cond := &parser.BinaryExpr{Op: op, Left: col, Right: marker}
		if branch == nil {
			branch = cond
		} else {
			branch = &parser.BinaryExpr{Op: "AND", Left: branch, Right: cond}
		}
	}
	return branch, values
}

func orderComparisonOp(direction string) string {
	switch strings.ToUpper(direction) {
	case "DESC":
		return "<"
	default:
		return ">"
	}
}

func resolvePagingLimit(stmt *parser.SelectStatement, params []interface{}) (*int64, error) {
	if stmt == nil {
		return nil, nil
	}
	limitExpr := stmt.Options.Limit
	ppExpr := stmt.Options.PerPartitionLimit
	if limitExpr == nil && ppExpr == nil {
		return nil, nil
	}
	markerIndex := buildBindMarkerIndex(stmt)

	var limitVal *int64
	if limitExpr != nil {
		term, err := parseLimitTerm(limitExpr, markerIndex)
		if err != nil {
			return nil, err
		}
		value, err := limitTermValue(term, params)
		if err != nil {
			return nil, err
		}
		limitVal = &value
	}
	if ppExpr != nil {
		term, err := parseLimitTerm(ppExpr, markerIndex)
		if err != nil {
			return nil, err
		}
		value, err := limitTermValue(term, params)
		if err != nil {
			return nil, err
		}
		if limitVal == nil || value < *limitVal {
			limitVal = &value
		}
	}

	return limitVal, nil
}

func resolvePagingOffset(stmt *parser.SelectStatement, params []interface{}) (*int64, error) {
	if stmt == nil {
		return nil, nil
	}
	offsetExpr := stmt.Options.Offset
	if offsetExpr == nil {
		return nil, nil
	}
	markerIndex := buildBindMarkerIndex(stmt)
	term, err := parseOffsetTerm(offsetExpr, markerIndex)
	if err != nil {
		return nil, err
	}
	value, err := limitTermValue(term, params)
	if err != nil {
		return nil, err
	}
	return &value, nil
}

func bindMarkerValues(stmt parser.Statement, params []interface{}) (map[parser.Expression]interface{}, error) {
	markerIndex := buildBindMarkerIndex(stmt)
	if len(markerIndex) == 0 {
		return map[parser.Expression]interface{}{}, nil
	}
	if len(params) != len(markerIndex) {
		return nil, fmt.Errorf("parameter count mismatch for paging")
	}
	values := make(map[parser.Expression]interface{}, len(markerIndex))
	for expr, idx := range markerIndex {
		if idx < 0 || idx >= len(params) {
			return nil, fmt.Errorf("parameter index %d out of range", idx)
		}
		values[expr] = params[idx]
	}
	return values, nil
}

func reorderParams(stmt parser.Statement, values map[parser.Expression]interface{}) ([]interface{}, error) {
	markerIndex := buildBindMarkerIndex(stmt)
	if len(markerIndex) == 0 {
		return nil, nil
	}
	params := make([]interface{}, len(markerIndex))
	for expr, idx := range markerIndex {
		val, ok := values[expr]
		if !ok {
			return nil, fmt.Errorf("missing bind marker value")
		}
		params[idx] = val
	}
	return params, nil
}

func extractOrderValues(row Row, columns []ColumnInfo, orderCols []orderColumn) ([]interface{}, error) {
	if len(orderCols) == 0 {
		return nil, fmt.Errorf("order columns are required")
	}
	indexByName := make(map[string]int, len(columns))
	for i, col := range columns {
		indexByName[strings.ToLower(col.Name)] = i
	}
	values := make([]interface{}, len(orderCols))
	for i, col := range orderCols {
		idx, ok := indexByName[strings.ToLower(col.Name)]
		if !ok || idx >= len(row.Values) {
			return nil, fmt.Errorf("paging column %q missing from result", col.Name)
		}
		values[i] = row.Values[idx]
	}
	return values, nil
}

func trimResultColumns(result *QueryResult, count int) {
	if result == nil {
		return
	}
	if count < 0 || count > len(result.Columns) {
		return
	}
	result.Columns = result.Columns[:count]
	for i := range result.Rows {
		if count <= len(result.Rows[i].Values) {
			result.Rows[i].Values = result.Rows[i].Values[:count]
		}
	}
}

func cloneSelectStatement(stmt *parser.SelectStatement) *parser.SelectStatement {
	clone := *stmt
	clone.Columns = append([]parser.Expression(nil), stmt.Columns...)
	clone.GroupBy = append([]parser.Expression(nil), stmt.GroupBy...)
	clone.Options = stmt.Options
	clone.Options.OrderBy = append([]*parser.OrderByClause(nil), stmt.Options.OrderBy...)
	clone.IndexHint = stmt.IndexHint
	return &clone
}

func selectHasAggregates(stmt *parser.SelectStatement) bool {
	if stmt == nil {
		return false
	}
	for _, expr := range stmt.Columns {
		if exprHasAggregate(expr) {
			return true
		}
	}
	return false
}

func exprHasAggregate(expr parser.Expression) bool {
	if expr == nil {
		return false
	}
	switch e := expr.(type) {
	case *parser.AliasedExpr:
		return exprHasAggregate(e.Expr)
	case *parser.TypeHintExpr:
		return exprHasAggregate(e.Expr)
	case *parser.UnaryExpr:
		return exprHasAggregate(e.Expr)
	case *parser.BinaryExpr:
		return exprHasAggregate(e.Left) || exprHasAggregate(e.Right)
	case *parser.FunctionCall:
		if isAggregateFunction(e) {
			return true
		}
		for _, arg := range e.Args {
			if exprHasAggregate(arg) {
				return true
			}
		}
		return false
	case *parser.CollectionAccessExpr:
		return exprHasAggregate(e.Target) || exprHasAggregate(e.Start) || exprHasAggregate(e.End)
	case *parser.ListLiteral:
		for _, elem := range e.Elements {
			if exprHasAggregate(elem) {
				return true
			}
		}
		return false
	case *parser.SetLiteral:
		for _, elem := range e.Elements {
			if exprHasAggregate(elem) {
				return true
			}
		}
		return false
	case *parser.TupleLiteral:
		for _, elem := range e.Elements {
			if exprHasAggregate(elem) {
				return true
			}
		}
		return false
	case *parser.VectorLiteral:
		for _, elem := range e.Elements {
			if exprHasAggregate(elem) {
				return true
			}
		}
		return false
	case *parser.MapLiteral:
		for _, entry := range e.Entries {
			if exprHasAggregate(entry.Key) || exprHasAggregate(entry.Value) {
				return true
			}
		}
		return false
	default:
		return false
	}
}

func pagingQueryHash(query, keyspace string) [32]byte {
	value := strings.TrimSpace(query) + "|" + keyspace
	return sha256.Sum256([]byte(value))
}

func encodePagingState(state pagingState) ([]byte, error) {
	buf := make([]byte, 0, 64)
	buf = append(buf, state.QueryHash[:]...)
	buf = appendShortString(buf, state.SchemaVersion)
	buf = appendShortString(buf, state.Table)
	buf = appendShortString(buf, state.OrderBy)
	buf = appendBytes(buf, state.LastPKValues)
	if state.RemainingLimit != nil {
		buf = appendInt64(buf, *state.RemainingLimit)
	}

	encoded := make([]byte, base64.StdEncoding.EncodedLen(len(buf)))
	base64.StdEncoding.Encode(encoded, buf)
	return encoded, nil
}

func decodePagingState(raw []byte) (pagingState, error) {
	if len(raw) == 0 {
		return pagingState{}, fmt.Errorf("paging state is empty")
	}
	decoded := make([]byte, base64.StdEncoding.DecodedLen(len(raw)))
	n, err := base64.StdEncoding.Decode(decoded, raw)
	if err != nil {
		return pagingState{}, fmt.Errorf("failed to decode paging state: %w", err)
	}
	decoded = decoded[:n]
	if len(decoded) < 32 {
		return pagingState{}, fmt.Errorf("paging state too short")
	}
	state := pagingState{}
	copy(state.QueryHash[:], decoded[:32])
	offset := 32
	state.SchemaVersion, n, err = readShortString(decoded[offset:])
	if err != nil {
		return pagingState{}, err
	}
	offset += n
	state.Table, n, err = readShortString(decoded[offset:])
	if err != nil {
		return pagingState{}, err
	}
	offset += n
	state.OrderBy, n, err = readShortString(decoded[offset:])
	if err != nil {
		return pagingState{}, err
	}
	offset += n
	state.LastPKValues, n, err = readBytesWithLen(decoded[offset:])
	if err != nil {
		return pagingState{}, err
	}
	offset += n
	if offset < len(decoded) {
		if len(decoded[offset:]) < 8 {
			return pagingState{}, fmt.Errorf("paging state remaining limit truncated")
		}
		remaining := int64(binary.BigEndian.Uint64(decoded[offset : offset+8]))
		state.RemainingLimit = &remaining
		offset += 8
	}
	if offset != len(decoded) {
		return pagingState{}, fmt.Errorf("paging state has trailing bytes")
	}
	return state, nil
}

func encodePagingValues(values []interface{}, types []string) ([]byte, error) {
	if len(values) != len(types) {
		return nil, fmt.Errorf("paging values/type count mismatch")
	}
	buf := make([]byte, 0, 2+len(values)*8)
	buf = append(buf, make([]byte, 2)...)
	binary.BigEndian.PutUint16(buf[len(buf)-2:], uint16(len(values)))
	translator := NewQueryResultTranslator()
	for i, value := range values {
		encoded := translator.translateValue(value, types[i], nil)
		if encoded == nil {
			buf = appendInt32(buf, -1)
			continue
		}
		buf = appendInt32(buf, int32(len(encoded)))
		buf = append(buf, encoded...)
	}
	return buf, nil
}

func decodePagingValues(payload []byte, types []string) ([]interface{}, error) {
	if len(payload) < 2 {
		return nil, fmt.Errorf("paging values payload too short")
	}
	count := int(binary.BigEndian.Uint16(payload[0:2]))
	if count != len(types) {
		return nil, fmt.Errorf("paging values count mismatch")
	}
	values := make([]interface{}, count)
	offset := 2
	for i := 0; i < count; i++ {
		if len(payload) < offset+4 {
			return nil, fmt.Errorf("paging value length truncated")
		}
		length := int32(binary.BigEndian.Uint32(payload[offset : offset+4]))
		offset += 4
		if length == pagingNullValue || length == pagingUnsetValue {
			values[i] = nil
			continue
		}
		if length < 0 {
			return nil, fmt.Errorf("invalid paging value length %d", length)
		}
		if len(payload) < offset+int(length) {
			return nil, fmt.Errorf("paging value truncated")
		}
		raw := payload[offset : offset+int(length)]
		offset += int(length)
		val, err := decodeCQLValue(raw, types[i], nil)
		if err != nil {
			return nil, err
		}
		values[i] = val
	}
	if offset != len(payload) {
		return nil, fmt.Errorf("paging values trailing bytes")
	}
	return values, nil
}

func readShortString(data []byte) (string, int, error) {
	if len(data) < 2 {
		return "", 0, fmt.Errorf("not enough bytes for string")
	}
	length := int(binary.BigEndian.Uint16(data[0:2]))
	if len(data) < 2+length {
		return "", 0, fmt.Errorf("not enough bytes for string value")
	}
	return string(data[2 : 2+length]), 2 + length, nil
}

func readBytesWithLen(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes for bytes value")
	}
	length := int32(binary.BigEndian.Uint32(data[0:4]))
	if length == pagingNullValue || length == pagingUnsetValue {
		return nil, 4, nil
	}
	if length < 0 {
		return nil, 0, fmt.Errorf("invalid bytes length %d", length)
	}
	if len(data) < 4+int(length) {
		return nil, 0, fmt.Errorf("not enough bytes for bytes value")
	}
	return data[4 : 4+int(length)], 4 + int(length), nil
}

func appendInt64(buf []byte, value int64) []byte {
	buf = append(buf, make([]byte, 8)...)
	binary.BigEndian.PutUint64(buf[len(buf)-8:], uint64(value))
	return buf
}