// Copyright 2024 Dolthub, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package types

import (
	"bytes"
	"cmp"
	"context"
	"fmt"
	"math"
	"reflect"
	"sync"
	"time"

	"github.com/cockroachdb/apd/v3"
	"github.com/cockroachdb/errors"
	"github.com/dolthub/dolt/go/libraries/doltcore/schema/typeinfo"
	"github.com/dolthub/dolt/go/store/val"
	"github.com/dolthub/go-mysql-server/sql"
	"github.com/dolthub/go-mysql-server/sql/types"
	"github.com/dolthub/vitess/go/sqltypes"
	"github.com/dolthub/vitess/go/vt/proto/query"

	"github.com/dolthub/doltgresql/core/id"
	"github.com/dolthub/doltgresql/postgres/parser/duration"
	"github.com/dolthub/doltgresql/postgres/parser/timeofday"
	"github.com/dolthub/doltgresql/postgres/parser/timetz"
	"github.com/dolthub/doltgresql/postgres/parser/uuid"
	"github.com/dolthub/doltgresql/utils"
)

// DoltgresType represents a single type. Many of these fields map directly to the type definitions in the pg_types
// system table. See https://www.postgresql.org/docs/current/catalog-pg-type.html for more information.
//
// TODO: the serialization logic always serializes every field for built-in types, which is kind of silly. They are
// effectively hard-coded. We could serialize much more cheaply by only serializing values which can't be derived
// (for custom types) and hard-coding everything else.
type DoltgresType struct {
	ID          id.Type
	TypType     TypeType
	TypCategory TypeCategory
	TypLength   int16
	PassedByVal bool
	IsPreferred bool
	IsDefined   bool
	Delimiter   string

	RelID         id.Id // references the parent table relation for Composite types
	SubscriptFunc uint32
	Elem          *DoltgresType
	Array         *DoltgresType
	InputFunc     uint32 // for deserializing a text representation
	OutputFunc    uint32 // for serializing a text representation
	ReceiveFunc   uint32 // for deserializing a binary representation
	SendFunc      uint32 // for serializing a binary representation
	ModInFunc     uint32
	ModOutFunc    uint32
	AnalyzeFunc   uint32
	Align         TypeAlignment
	Storage       TypeStorage

	NotNull      bool          // for Domain types
	BaseTypeType *DoltgresType // for Domain types
	TypMod       int32         // for Domain types
	NDims        int32         // for Domain types
	TypCollation id.Collation
	DefaulBin    string // for Domain types
	Default      string
	Acl          []string // TODO: list of privileges

	// Below are not part of pg_type fields
	Checks         []*sql.CheckDefinition // TODO: should be in `pg_constraint` for Domain types
	attTypMod      int32                  // TODO: should be in `pg_attribute.atttypmod`
	CompareFunc    uint32                 // TODO: should be in `pg_amproc`
	InternalName   string                 // Name() and InternalName differ for some types. e.g.: "int2" vs "smallint"
	EnumLabels     map[string]EnumLabel   // TODO: should be in `pg_enum`
	CompositeAttrs []CompositeAttribute   // TODO: should be in `pg_attribute`

	// Below are not stored
	IsSerial            bool    // used for serial types only (e.g.: smallserial)
	IsUnresolved        bool    // used internally to know if a type has been resolved
	BaseTypeForInternal id.Type // used for INTERNAL type only
	SerializationFunc   internalSerializationFunc
	DeserializationFunc internalDeserializationFunc

	// TODO: refresh
	castCache map[*DoltgresType]Cast
	mutex     sync.Mutex
}

// internalNullType represents a type with a null ID, effectively stating that the field in the parent DoltgresType is
// empty. This type is considered resolved.
var internalNullType = &DoltgresType{ID: id.NullType, IsUnresolved: false}

// internalSerializationFunc is the function definition for internal type serialization
type internalSerializationFunc func(*sql.Context, *DoltgresType, any) ([]byte, error)

// internalDeserializationFunc is the function definition for internal type deserialization
type internalDeserializationFunc func(*sql.Context, *DoltgresType, []byte) (any, error)

var _ sql.ExtendedType = &DoltgresType{}
var _ sql.NullType = &DoltgresType{}
var _ sql.StringType = &DoltgresType{}
var _ sql.NumberType = &DoltgresType{}
var _ val.TupleTypeHandler = &DoltgresType{}
var _ typeinfo.ExtendedType = &DoltgresType{}

// NewUnresolvedDoltgresType returns a DoltgresType that is not resolved.
// The type will have the schema and name defined with given values, with IsUnresolved == true.
func NewUnresolvedDoltgresType(sch, name string) *DoltgresType {
	return NewUnresolvedDoltgresTypeFromID(id.NewType(sch, name))
}

// NewUnresolvedDoltgresTypeFromID returns a DoltgresType that is not resolved.
func NewUnresolvedDoltgresTypeFromID(idType id.Type) *DoltgresType {
	return &DoltgresType{
		ID:           idType,
		Elem:         internalNullType,
		Array:        internalNullType,
		BaseTypeType: internalNullType,
		IsUnresolved: true,
	}
}

// NewUnresolvedArrayDoltgresType returns an unresolved DoltgresType for an array of a user-defined element type.
// TypCategory and Elem are pre-filled so that IsArrayType() returns true before full resolution from the type
// collection. The array type ID follows the Postgres convention of "_" + element type name.
func NewUnresolvedArrayDoltgresType(sch, elemName string) *DoltgresType {
	return &DoltgresType{
		ID:           id.NewType(sch, "_"+elemName),
		IsUnresolved: true,
		TypCategory:  TypeCategory_ArrayTypes,
		Elem:         &DoltgresType{ID: id.NewType(sch, elemName), IsUnresolved: true},
		Array:        internalNullType,
		BaseTypeType: internalNullType,
	}
}

// AnalyzeFuncName returns the name that would be displayed in pg_type for the `typanalyze` field.
func (t *DoltgresType) AnalyzeFuncName() string {
	return globalFunctionRegistry.GetString(t.AnalyzeFunc)
}

// ArrayBaseType returns the base type of an array type.
func (t *DoltgresType) ArrayBaseType() *DoltgresType {
	if !t.IsArrayType() {
		return t
	}
	// Some array types have no declared element type for pg_catalog compatibility, but still have a logical type
	// we return for analysis.
	if t.ID == AnyArray.ID {
		return AnyElement
	}
	return t.Elem.WithAttTypMod(t.attTypMod)
}

// BaseType returns a base type of given array or vector type.
// If this type does not have base type, it returns itself.
func (t *DoltgresType) BaseType() *DoltgresType {
	if t.Elem.ID == id.NullType {
		return t
	}
	return t.Elem.WithAttTypMod(t.attTypMod)
}

// CharacterSet implements the sql.StringType interface.
func (t *DoltgresType) CharacterSet() sql.CharacterSetID {
	switch t.ID.TypeName() {
	case "varchar", "text", "name":
		return sql.CharacterSet_binary
	default:
		return sql.CharacterSet_Unspecified
	}
}

// Collation implements the sql.StringType interface.
func (t *DoltgresType) Collation() sql.CollationID {
	switch t.ID.TypeName() {
	case "varchar", "text", "name":
		return sql.Collation_Default
	default:
		return sql.Collation_Unspecified
	}
}

// CollationCoercibility implements the types.ExtendedType interface.
func (t *DoltgresType) CollationCoercibility(ctx *sql.Context) (collation sql.CollationID, coercibility byte) {
	return sql.Collation_binary, 5
}

// Compare implements the types.ExtendedType interface.
func (t *DoltgresType) Compare(ctx context.Context, v1 interface{}, v2 interface{}) (int, error) {
	// TODO: for some large types, we could do this much faster by doing it chunk-by-chunk, rather than eagerly loading
	//  the full value into memory
	var err error
	v1, err = sql.UnwrapAny(ctx, v1)
	if err != nil {
		return 0, err
	}
	v2, err = sql.UnwrapAny(ctx, v2)
	if err != nil {
		return 0, err
	}

	if v1 == nil && v2 == nil {
		return 0, nil
	} else if v1 != nil && v2 == nil {
		return 1, nil
	} else if v1 == nil && v2 != nil {
		return -1, nil
	}

	if t.TypType == TypeType_Enum {
		// TODO: temporary solution to getting the enum type (which has label info) into the 'enum_cmp' function
		// ctx is not guaranteed to be a *sql.Context when called from index comparator goroutines.
		sqlCtx, ok := ctx.(*sql.Context)
		if !ok {
			sqlCtx = sql.NewEmptyContext()
		}
		qf := globalFunctionRegistry.GetFunction(sqlCtx, t.CompareFunc)
		resTypes := qf.ResolvedTypes()
		newFunc := qf.WithResolvedTypes([]*DoltgresType{t, t, resTypes[len(resTypes)-1]})
		i, err := newFunc.(QuickFunction).CallVariadic(nil, v1, v2)
		if err != nil {
			return 0, err
		}
		return int(i.(int32)), nil
	} else if t == Oidvector {
		// ctx is not guaranteed to be a *sql.Context when called from index comparator goroutines.
		sqlCtx, ok := ctx.(*sql.Context)
		if !ok {
			sqlCtx = sql.NewEmptyContext()
		}
		i, err := globalFunctionRegistry.GetFunction(sqlCtx, t.CompareFunc).CallVariadic(nil, v1, v2)
		if err != nil {
			return 0, err
		}
		return int(i.(int32)), nil
	}

	switch ab := v1.(type) {
	case bool:
		bb := v2.(bool)
		if ab == bb {
			return 0, nil
		} else if !ab {
			return -1, nil
		} else {
			return 1, nil
		}
	case float32:
		bb := v2.(float32)
		if ab == bb {
			return 0, nil
		} else if ab < bb {
			return -1, nil
		} else {
			return 1, nil
		}
	case float64:
		bb := v2.(float64)
		if ab == bb {
			return 0, nil
		} else if ab < bb {
			return -1, nil
		} else {
			return 1, nil
		}
	case int16:
		bb := v2.(int16)
		if ab == bb {
			return 0, nil
		} else if ab < bb {
			return -1, nil
		} else {
			return 1, nil
		}
	case int32:
		bb := v2.(int32)
		if ab == bb {
			return 0, nil
		} else if ab < bb {
			return -1, nil
		} else {
			return 1, nil
		}
	case int64:
		bb := v2.(int64)
		if ab == bb {
			return 0, nil
		} else if ab < bb {
			return -1, nil
		} else {
			return 1, nil
		}
	case uint32:
		bb := v2.(uint32)
		if ab == bb {
			return 0, nil
		} else if ab < bb {
			return -1, nil
		} else {
			return 1, nil
		}
	case string:
		bb := v2.(string)
		if ab == bb {
			return 0, nil
		} else if ab < bb {
			return -1, nil
		} else {
			return 1, nil
		}
	case []byte:
		bb := v2.([]byte)
		return bytes.Compare(ab, bb), nil
	case time.Time:
		bb := v2.(time.Time)
		return ab.Compare(bb), nil
	case duration.Duration:
		bb := v2.(duration.Duration)
		return ab.Compare(bb), nil
	case sql.JSONWrapper:
		res, err := types.CompareJSON(ctx, ab, v2)
		return res, err
	case *apd.Decimal:
		bb := v2.(*apd.Decimal)
		return NumericCompare(ab, bb), nil
	case timeofday.TimeOfDay:
		bb := v2.(timeofday.TimeOfDay)
		return ab.Compare(bb), nil
	case timetz.TimeTZ:
		bb := v2.(timetz.TimeTZ)
		return ab.Compare(bb), nil
	case uuid.UUID:
		bb := v2.(uuid.UUID)
		return bytes.Compare(ab.GetBytesMut(), bb.GetBytesMut()), nil
	case id.Id:
		return cmp.Compare(id.Cache().ToOID(ab), id.Cache().ToOID(v2.(id.Id))), nil
	case id.Oid:
		return cmp.Compare(ab.OID(), v2.(id.Oid).OID()), nil
	case []any:
		if !t.IsArrayType() {
			return 0, errors.New("array value received in Compare for non array type")
		}
		bb := v2.([]any)
		minLength := utils.Min(len(ab), len(bb))
		for i := 0; i < minLength; i++ {
			res, err := t.ArrayBaseType().Compare(ctx, ab[i], bb[i])
			if err != nil {
				return 0, err
			}
			if res != 0 {
				return res, nil
			}
		}
		if len(ab) == len(bb) {
			return 0, nil
		} else if len(ab) < len(bb) {
			return -1, nil
		} else {
			return 1, nil
		}
	case []RecordValue:
		if !t.IsCompositeType() {
			return 0, errors.New("record value received in Compare for non composite type")
		}
		bb := v2.([]RecordValue)
		minLength := utils.Min(len(ab), len(bb))
		for i := 0; i < minLength; i++ {
			dgType, isDgType1 := ab[i].Type.(*DoltgresType)
			otherDgType, isDgType2 := bb[i].Type.(*DoltgresType)
			if !isDgType1 || !isDgType2 {
				return 0, errors.New("record values in Compare must use a Doltgres type")
			}
			if dgType.ID != otherDgType.ID {
				return 0, errors.New("record values in Compare must use the same type as the same index")
			}
			res, err := dgType.Compare(ctx, ab[i].Value, bb[i].Value)
			if err != nil {
				return 0, err
			}
			if res != 0 {
				return res, nil
			}
		}
		if len(ab) == len(bb) {
			return 0, nil
		} else if len(ab) < len(bb) {
			return -1, nil
		} else {
			return 1, nil
		}
	default:
		return 0, errors.Errorf("unhandled type %T in Compare", v1)
	}
}

// Convert implements the types.ExtendedType interface.
func (t *DoltgresType) Convert(ctx context.Context, v interface{}) (interface{}, sql.ConvertInRange, error) {
	if v == nil {
		return nil, sql.InRange, nil
	}
	switch t.ID.TypeName() {
	case "bool":
		if _, ok := v.(bool); ok {
			return v, sql.InRange, nil
		}
	case "bytea":
		if _, ok := v.([]byte); ok {
			return v, sql.InRange, nil
		}
	case "bpchar", "char", "name", "text", "varchar":
		_, ok, err := sql.Unwrap[string](ctx, v)
		if err != nil {
			return nil, sql.InRange, err
		}
		if ok {
			return v, sql.InRange, nil
		}
	case "date", "timestamp", "timestamptz":
		if _, ok := v.(time.Time); ok {
			return v, sql.InRange, nil
		}
	case "float4":
		if _, ok := v.(float32); ok {
			return v, sql.InRange, nil
		}
	case "float8":
		if _, ok := v.(float64); ok {
			return v, sql.InRange, nil
		}
	case "int2":
		if _, ok := v.(int16); ok {
			return v, sql.InRange, nil
		}
	case "int4":
		if _, ok := v.(int32); ok {
			return v, sql.InRange, nil
		}
	case "int8":
		if _, ok := v.(int64); ok {
			return v, sql.InRange, nil
		}
	case "interval":
		if _, ok := v.(duration.Duration); ok {
			return v, sql.InRange, nil
		}
	case "jsonb", "json":
		if _, ok := v.(sql.JSONWrapper); ok {
			return v, sql.InRange, nil
		}
		if _, ok := v.(string); ok {
			return v, sql.InRange, nil
		}
	case "oid", "regclass", "regproc", "regtype":
		if _, ok := v.(id.Id); ok {
			return v, sql.InRange, nil
		}
	case "time":
		if _, ok := v.(timeofday.TimeOfDay); ok {
			return v, sql.InRange, nil
		}
	case "timetz":
		if _, ok := v.(timetz.TimeTZ); ok {
			return v, sql.InRange, nil
		}
	case "xid":
		if _, ok := v.(uint32); ok {
			return v, sql.InRange, nil
		}
	case "uuid":
		if _, ok := v.(uuid.UUID); ok {
			return v, sql.InRange, nil
		}
	case "unknown":
		// TODO: implement other types, assume everything is either bool or unwrappable string for now
		switch v := v.(type) {
		case bool:
			if v {
				return "t", sql.InRange, nil
			} else {
				return "f", sql.InRange, nil
			}
		default:
			_, ok, err := sql.Unwrap[string](ctx, v)
			if err != nil {
				return nil, sql.InRange, err
			}
			if ok {
				return v, sql.InRange, nil
			}
		}
	default:
		return v, sql.InRange, nil
	}
	return nil, sql.InRange, ErrUnhandledType.New(t.String(), v)
}

// GetAssignmentCast is a reference to the assignment cast logic in the core package, which we can't use here due to
// import cycles
var GetAssignmentCast func(ctx *sql.Context, fromType *DoltgresType, toType *DoltgresType) (Cast, error)

// ConvertToType implements the types.ExtendedType interface.
func (t *DoltgresType) ConvertToType(ctx *sql.Context, typ sql.ExtendedType, val any) (any, sql.ConvertInRange, error) {
	dt, ok := typ.(*DoltgresType)
	if !ok {
		return nil, sql.InRange, errors.Errorf("expected DoltgresType, got %T", typ)
	}

	t.mutex.Lock()
	if t.castCache == nil {
		t.castCache = make(map[*DoltgresType]Cast)
	}
	var cast Cast
	if cast, ok = t.castCache[dt]; !ok {
		var err error
		cast, err = GetAssignmentCast(ctx, dt, t)
		if err != nil {
			t.mutex.Unlock()
			return nil, sql.InRange, err
		}
		t.castCache[dt] = cast
	}
	t.mutex.Unlock()

	if cast == nil {
		// In the case that we have an unknown type string literal, we attempt to parse it with the target type's
		// input function
		// TODO: this is probably not the best place to perform this conversion, it would probably be better as an
		//  analyzer step. This currently only takes place for foreign key checks and some index lookups, and could be
		//  generalized to many more places.
		if dt.ID.TypeName() == "unknown" {
			strVal, ok, err := sql.Unwrap[string](ctx, val)
			if err != nil {
				return nil, sql.InRange, err
			}
			if ok {
				converted, err := t.IoInput(ctx, strVal)
				if err != nil {
					return nil, sql.InRange, err
				}
				return converted, sql.InRange, nil
			}
		}
		return nil, sql.InRange, errors.Errorf("no assignment cast from %s to %s", dt.Name(), t.Name())
	}

	castResult, err := cast.Eval(ctx, val, dt, t)
	if err != nil && errors.Is(err, ErrCastOutOfRange) {
		// TODO: this could be either an overflow or an underflow, we should distinguish
		return castResult, sql.Overflow, nil
	} else if err != nil {
		return nil, sql.InRange, err
	}

	return castResult, sql.InRange, nil
}

// DomainUnderlyingBaseType returns an underlying base type of this domain type.
// It can be a nested domain type, so it recursively searches for a valid base type.
// It is not valid to call this on a non-domain type.
func (t *DoltgresType) DomainUnderlyingBaseType() *DoltgresType {
	if t.BaseTypeType.TypType == TypeType_Domain {
		return t.BaseTypeType.DomainUnderlyingBaseType()
	} else {
		return t.BaseTypeType
	}
}

// Equals implements the types.ExtendedType interface.
func (t *DoltgresType) Equals(otherType sql.Type) bool {
	if otherExtendedType, ok := otherType.(*DoltgresType); ok {
		return bytes.Equal(t.Serialize(), otherExtendedType.Serialize())
	}
	return false
}

// FormatValue implements the types.ExtendedType interface. Callers with
// a session context should use FormatValueWithContext instead, since
// types whose output reads session state cannot resolve through a nil
// context.
func (t *DoltgresType) FormatValue(val any) (string, error) {
	return t.FormatValueWithContext(nil, val)
}

// FormatValueWithContext returns the postgres output representation of
// |val|, using |ctx| for any session-scoped state the type's output
// function needs.
func (t *DoltgresType) FormatValueWithContext(ctx *sql.Context, val any) (string, error) {
	if val == nil {
		return "", nil
	}
	return t.IoOutput(ctx, val)
}

// GetAttTypMod returns the attTypMod field of the type.
func (t *DoltgresType) GetAttTypMod() int32 {
	return t.attTypMod
}

// InputFuncName returns the name that would be displayed in pg_type for the `typinput` field.
func (t *DoltgresType) InputFuncName() string {
	return globalFunctionRegistry.GetString(t.InputFunc)
}

// IoInput converts input string value to given type value.
func (t *DoltgresType) IoInput(ctx *sql.Context, input string) (any, error) {
	if t.TypType == TypeType_Domain {
		return globalFunctionRegistry.GetFunction(ctx, t.InputFunc).CallVariadic(ctx, input, t.BaseTypeType.ID.AsId(), t.attTypMod)
	} else if t.ModInFunc != 0 || t.IsArrayType() {
		if t.Elem.ID != id.NullType {
			return globalFunctionRegistry.GetFunction(ctx, t.InputFunc).CallVariadic(ctx, input, t.Elem.ID.AsId(), t.attTypMod)
		} else {
			return globalFunctionRegistry.GetFunction(ctx, t.InputFunc).CallVariadic(ctx, input, t.ID.AsId(), t.attTypMod)
		}
	} else if t.TypType == TypeType_Enum {
		return globalFunctionRegistry.GetFunction(ctx, t.InputFunc).CallVariadic(ctx, input, t.ID.AsId())
	} else if t.IsCompositeType() {
		return ParseCompositeLiteral(ctx, input, t)
	} else {
		return globalFunctionRegistry.GetFunction(ctx, t.InputFunc).CallVariadic(ctx, input)
	}
}

// IoOutput converts given type value to output string.
func (t *DoltgresType) IoOutput(ctx *sql.Context, val any) (string, error) {
	var o any
	var err error
	if t.ModInFunc != 0 || t.IsArrayType() || t.IsCompositeType() {
		send := globalFunctionRegistry.GetFunction(ctx, t.OutputFunc)
		resolvedTypes := send.ResolvedTypes()
		resolvedTypes[0] = t
		o, err = send.WithResolvedTypes(resolvedTypes).(QuickFunction).CallVariadic(ctx, val)
	} else {
		o, err = globalFunctionRegistry.GetFunction(ctx, t.OutputFunc).CallVariadic(ctx, val)
	}
	if err != nil {
		return "", err
	}
	var ok bool
	os, ok, err := sql.Unwrap[string](ctx, o)
	if !ok {
		return "", errors.Errorf("unexpected type for io output, expected string, got %T", val)
	}
	return os, err
}

// IsArrayType returns true if the type is of 'array' type.
// It can be array category with empty its array attribute NULL and element attribute NOT NULL.
// Or it can be pseudo category with name 'anyarray'.
func (t *DoltgresType) IsArrayType() bool {
	return (t.TypCategory == TypeCategory_ArrayTypes && t.Elem.ID != id.NullType && t.Array.ID == id.NullType) ||
		(t.TypCategory == TypeCategory_PseudoTypes && t.ID.TypeName() == "anyarray")
}

// IsArrayCategory returns true if the type is of 'array' category.
// It can be either array types or vector types.
func (t *DoltgresType) IsArrayCategory() bool {
	return t.TypCategory == TypeCategory_ArrayTypes
}

// IsCompositeType returns true if the type is a composite type, such as an anonymous record, or a
// user-created composite type.
func (t *DoltgresType) IsCompositeType() bool {
	return t.TypType == TypeType_Composite || t.IsRecordType()
}

// IsRecordType returns true if the type is an anonymous record type.
func (t *DoltgresType) IsRecordType() bool {
	return t.TypType == TypeType_Pseudo && t.ID.TypeName() == "record"
}

// IsEmptyType returns true if the type is not valid.
func (t *DoltgresType) IsEmptyType() bool {
	return t == nil
}

// IsPolymorphicType types are special built-in pseudo-types
// that are used during function resolution to allow a function
// to handle multiple types from a single definition.
// All polymorphic types have "any" as a prefix.
// The exception is the "any" type, which is not a polymorphic type.
func (t *DoltgresType) IsPolymorphicType() bool {
	switch t.ID.TypeName() {
	case "anyelement", "anyarray", "anynonarray", "anyenum", "anyrange":
		// TODO: add other polymorphic types
		// https://www.postgresql.org/docs/15/extend-type-system.html#EXTEND-TYPES-POLYMORPHIC-TABLE
		return true
	default:
		return false
	}
}

// IsResolvedType whether the type is resolved and has complete information.
// This is used to resolve types during analyzing when non-built-in type is used.
func (t *DoltgresType) IsResolvedType() bool {
	return t != nil && !t.IsUnresolved
}

// IsValidForPolymorphicType returns whether the given type is valid for the calling polymorphic type.
func (t *DoltgresType) IsValidForPolymorphicType(target *DoltgresType) bool {
	switch t.ID.TypeName() {
	case "any":
		// "any" is not a polymorphic type like the others are, but it's useful to treat it as such for this check
		return true
	case "anyelement":
		return true
	case "anyarray":
		return target.TypCategory == TypeCategory_ArrayTypes
	case "anynonarray":
		return target.TypCategory != TypeCategory_ArrayTypes
	case "anyenum":
		return target.TypCategory == TypeCategory_EnumTypes
	case "anyrange":
		return target.TypCategory == TypeCategory_RangeTypes
	default:
		// TODO: add other polymorphic types
		// https://www.postgresql.org/docs/15/extend-type-system.html#EXTEND-TYPES-POLYMORPHIC-TABLE
		return false
	}
}

// Length implements the sql.StringType interface.
func (t *DoltgresType) Length() int64 {
	switch t.ID.TypeName() {
	case "varchar", "bpchar":
		if t.attTypMod == -1 {
			return StringUnbounded
		} else {
			return int64(GetCharLengthFromTypmod(t.attTypMod))
		}
	case "text":
		return StringUnbounded
	case "name":
		return int64(t.TypLength)
	default:
		return int64(0)
	}
}

// MaxByteLength implements the sql.StringType interface.
func (t *DoltgresType) MaxByteLength() int64 {
	if t.ID == VarChar.ID {
		return t.Length() * 4
	} else if t.TypLength == -1 {
		return StringUnbounded
	} else {
		return int64(t.TypLength) * 4
	}
}

// MaxCharacterLength implements the sql.StringType interface.
func (t *DoltgresType) MaxCharacterLength() int64 {
	if t.ID == VarChar.ID {
		return t.Length()
	} else if t.TypLength == -1 {
		return StringUnbounded
	} else {
		return int64(t.TypLength)
	}
}

// IsNumericType implements the sql.NumberType interface.
func (t *DoltgresType) IsNumericType() bool {
	return t.TypCategory == TypeCategory_NumericTypes
}

// IsFloat implements the sql.NumberType interface.
func (t *DoltgresType) IsFloat() bool {
	switch t.ID.TypeName() {
	case "float4", "float8", "numeric", "decimal":
		return true
	default:
		return false
	}
}

// DisplayWidth implements the sql.NumberType interface.
func (t *DoltgresType) DisplayWidth() int {
	switch t.ID.TypeName() {
	case "int2":
		return 6
	case "int4":
		return 11
	case "int8":
		return 20
	case "float4":
		return 14
	case "float8":
		return 25
	case "numeric", "decimal":
		return 131089 // maximum display width for numeric/decimal in Postgres
	default:
		return 0
	}
}

// IsStringType implements the sql.StringType interface.
func (t *DoltgresType) IsStringType() bool {
	return t.TypCategory == TypeCategory_StringTypes
}

// MaxSerializedWidth implements the types.ExtendedType interface.
func (t *DoltgresType) MaxSerializedWidth() sql.ExtendedTypeSerializedWidth {
	if t.TypLength < 0 {
		// Length will be 0 for any non-string type, as well as unbounded string types
		if t.Length() > 0 {
			return sql.ExtendedTypeSerializedWidth_64K
		}
		return sql.ExtendedTypeSerializedWidth_Unbounded
	}
	return sql.ExtendedTypeSerializedWidth_64K
}

// MaxTextResponseByteLength implements the types.ExtendedType interface.
func (t *DoltgresType) MaxTextResponseByteLength(ctx *sql.Context) uint32 {
	if t.ID == VarChar.ID {
		return math.MaxUint32
	} else if t.TypLength == -1 {
		return math.MaxUint32
	} else {
		return uint32(t.TypLength)
	}
}

// ModInFuncName returns the name that would be displayed in pg_type for the `typmodin` field.
func (t *DoltgresType) ModInFuncName() string {
	return globalFunctionRegistry.GetString(t.ModInFunc)
}

// ModOutFuncName returns the name that would be displayed in pg_type for the `typmodout` field.
func (t *DoltgresType) ModOutFuncName() string {
	return globalFunctionRegistry.GetString(t.ModOutFunc)
}

// Name returns the name of the type.
func (t *DoltgresType) Name() string {
	return t.ID.TypeName()
}

// OutputFuncName returns the name that would be displayed in pg_type for the `typoutput` field.
func (t *DoltgresType) OutputFuncName() string {
	return globalFunctionRegistry.GetString(t.OutputFunc)
}

// Promote implements the types.ExtendedType interface.
func (t *DoltgresType) Promote() sql.Type {
	return t
}

// ReceiveFuncName returns the name that would be displayed in pg_type for the `typreceive` field.
func (t *DoltgresType) ReceiveFuncName() string {
	return globalFunctionRegistry.GetString(t.ReceiveFunc)
}

// Schema returns the schema that the type is contained in.
func (t *DoltgresType) Schema() string {
	return t.ID.SchemaName()
}

// SendFuncName returns the name that would be displayed in pg_type for the `typsend` field.
func (t *DoltgresType) SendFuncName() string {
	return globalFunctionRegistry.GetString(t.SendFunc)
}

// SerializedCompare implements the types.ExtendedType interface.
func (t *DoltgresType) SerializedCompare(ctx context.Context, v1 []byte, v2 []byte) (int, error) {
	if len(v1) == 0 && len(v2) == 0 {
		return 0, nil
	} else if len(v1) > 0 && len(v2) == 0 {
		return 1, nil
	} else if len(v1) == 0 && len(v2) > 0 {
		return -1, nil
	}

	switch t.TypCategory {
	case TypeCategory_StringTypes:
		return serializedStringCompare(v1, v2), nil
	default:
		// TODO: there are certainly other types that could be compared in serialized form
		return deserializeAndCompare(ctx, t, v1, v2)
	}
}

// deserializeAndCompare deserializes the given serialized values and compares them
func deserializeAndCompare(ctx context.Context, t *DoltgresType, v1 []byte, v2 []byte) (int, error) {
	val1, err := t.DeserializeValue(ctx, v1)
	if err != nil {
		return 0, err
	}
	val2, err := t.DeserializeValue(ctx, v2)
	if err != nil {
		return 0, err
	}

	return t.Compare(ctx, val1, val2)
}

// IsNullType implements the sql.NullType interface.
func (t *DoltgresType) IsNullType() bool {
	return t.ID.TypeName() == "unknown"
}

// SQL implements the types.ExtendedType interface.
func (t *DoltgresType) SQL(ctx *sql.Context, dest []byte, v interface{}) (sqltypes.Value, error) {
	if v == nil {
		return sqltypes.NULL, nil
	}
	value, err := sqlString(ctx, t, v)
	if err != nil {
		return sqltypes.Value{}, err
	}

	// TODO: check type
	return sqltypes.MakeTrusted(sqltypes.Text, types.AppendAndSliceString(dest, value)), nil
}

// String implements the types.ExtendedType interface.
func (t *DoltgresType) String() string {
	str := t.InternalName
	if t.InternalName == "" {
		str = t.Name()
	}
	if t.attTypMod != -1 {
		// TODO: need valid sql.Context
		if l, err := t.TypModOut(nil, t.attTypMod); err == nil {
			str = fmt.Sprintf("%s%s", str, l)
		}
	}
	return str
}

// SubscriptFuncName returns the name that would be displayed in pg_type for the `typsubscript` field.
func (t *DoltgresType) SubscriptFuncName() string {
	return globalFunctionRegistry.GetString(t.SubscriptFunc)
}

// ToArrayType returns an array type of given base type.
// For array types, ToArrayType causes them to return themselves.
func (t *DoltgresType) ToArrayType() *DoltgresType {
	if t.IsArrayType() {
		return t
	}
	if t.Array.IsResolvedType() {
		arr := t.Array.WithAttTypMod(t.attTypMod)
		arr.InternalName = fmt.Sprintf("%s[]", t.String())
		return arr
	}
	if t.Array.ID == id.NullType {
		// Unresolved or stub type: derive an unresolved array type using the Postgres naming convention.
		// The caller (e.g. during plan-building before the analyzer resolves types) will re-invoke
		// ToArrayType on the fully-resolved base type once the analyzer has run.
		return NewUnresolvedArrayDoltgresType(t.ID.SchemaName(), t.ID.TypeName())
	}
	// User-defined type: the array type is not in the built-in map, so build it from this base type.
	return CreateArrayTypeFromBaseType(t)
	// TODO: delete the commented out code below, only exists for referencing
	/*arr, ok := IDToBuiltInDoltgresType[t.Array]
	if !ok {
		if t.Array.ID == id.NullType {
			// Unresolved or stub type: derive an unresolved array type using the Postgres naming convention.
			// The caller (e.g. during plan-building before the analyzer resolves types) will re-invoke
			// ToArrayType on the fully-resolved base type once the analyzer has run.
			return NewUnresolvedArrayDoltgresType(t.ID.SchemaName(), t.ID.TypeName())
		}
		// User-defined type: the array type is not in the built-in map, so build it from this base type.
		return CreateArrayTypeFromBaseType(t)
	}
	newArr := *arr.WithAttTypMod(t.attTypMod)
	newArr.InternalName = fmt.Sprintf("%s[]", t.String())
	return &newArr*/
}

// Type implements the types.ExtendedType interface.
func (t *DoltgresType) Type() query.Type {
	// TODO: need better way to get accurate result
	switch t.TypCategory {
	case TypeCategory_ArrayTypes:
		return sqltypes.Text
	case TypeCategory_BooleanTypes:
		return sqltypes.Text
	case TypeCategory_CompositeTypes, TypeCategory_EnumTypes, TypeCategory_GeometricTypes, TypeCategory_NetworkAddressTypes,
		TypeCategory_RangeTypes, TypeCategory_PseudoTypes, TypeCategory_UserDefinedTypes, TypeCategory_BitStringTypes,
		TypeCategory_InternalUseTypes:
		return sqltypes.Text
	case TypeCategory_DateTimeTypes:
		switch t.ID.TypeName() {
		case "date":
			return sqltypes.Date
		case "time":
			return sqltypes.Time
		default:
			return sqltypes.Timestamp
		}
	case TypeCategory_NumericTypes:
		switch t.ID.TypeName() {
		case "float4":
			return sqltypes.Float32
		case "float8":
			return sqltypes.Float64
		case "int2":
			return sqltypes.Int16
		case "int4":
			return sqltypes.Int32
		case "int8":
			return sqltypes.Int64
		case "numeric":
			return sqltypes.Decimal
		case "oid":
			return sqltypes.VarChar
		case "regclass", "regproc", "regtype":
			return sqltypes.Text
		default:
			// TODO
			return sqltypes.Int64
		}
	case TypeCategory_StringTypes, TypeCategory_UnknownTypes:
		if t.ID.TypeName() == "varchar" {
			return sqltypes.VarChar
		}
		return sqltypes.Text
	case TypeCategory_TimespanTypes:
		return sqltypes.Text
	default:
		// shouldn't happen
		return sqltypes.Text
	}
}

// TypModIn encodes given text array value to type modifier in int32 format.
func (t *DoltgresType) TypModIn(ctx *sql.Context, val []any) (int32, error) {
	if t.ModInFunc == 0 {
		return 0, errors.Errorf("typmodin function for type '%s' doesn't exist", t.Name())
	}
	o, err := globalFunctionRegistry.GetFunction(ctx, t.ModInFunc).CallVariadic(ctx, val)
	if err != nil {
		return 0, err
	}
	output, ok := o.(int32)
	if !ok {
		return 0, errors.Errorf(`expected int32, got %T`, output)
	}
	return output, nil
}

// TypModOut decodes type modifier in int32 format to string representation of it.
func (t *DoltgresType) TypModOut(ctx *sql.Context, val int32) (string, error) {
	if t.ModOutFunc == 0 {
		return "", errors.Errorf("typmodout function for type '%s' doesn't exist", t.Name())
	}
	o, err := globalFunctionRegistry.GetFunction(ctx, t.ModOutFunc).CallVariadic(ctx, val)
	if err != nil {
		return "", err
	}
	output, ok := o.(string)
	if !ok {
		return "", errors.Errorf(`expected string, got %T`, output)
	}
	return output, nil
}

// ValueType implements the types.ExtendedType interface.
func (t *DoltgresType) ValueType() reflect.Type {
	return reflect.TypeOf(t.Zero())
}

// WithAttTypMod returns a copy of the type with attTypMod
// defined with given value. This function should be used
// to set attTypMod only, as it creates a copy of the type
// to avoid updating the original type.
func (t *DoltgresType) WithAttTypMod(tm int32) *DoltgresType {
	newDt := t.Copy()
	newDt.attTypMod = tm
	return newDt
}

// Zero implements the types.ExtendedType interface.
func (t *DoltgresType) Zero() interface{} {
	// TODO: need better way to get accurate result
	switch t.TypCategory {
	case TypeCategory_ArrayTypes:
		return []any{}
	case TypeCategory_BooleanTypes:
		return false
	case TypeCategory_CompositeTypes, TypeCategory_EnumTypes, TypeCategory_GeometricTypes, TypeCategory_NetworkAddressTypes,
		TypeCategory_RangeTypes, TypeCategory_PseudoTypes, TypeCategory_UserDefinedTypes, TypeCategory_BitStringTypes,
		TypeCategory_InternalUseTypes:
		return any(nil)
	case TypeCategory_DateTimeTypes:
		return time.Time{}
	case TypeCategory_NumericTypes:
		switch t.ID.TypeName() {
		case "float4":
			return float32(0)
		case "float8":
			return float64(0)
		case "int2":
			return int16(0)
		case "int4":
			return int32(0)
		case "int8":
			return int64(0)
		case "numeric":
			return apd.New(0, 0)
		case "oid", "regclass", "regproc", "regtype":
			return id.Null
		default:
			// TODO
			return int64(0)
		}
	case TypeCategory_StringTypes, TypeCategory_UnknownTypes:
		return ""
	case TypeCategory_TimespanTypes:
		return duration.MakeDuration(0, 0, 0)
	default:
		// shouldn't happen
		return any(nil)
	}
}

// SerializeValue implements the types.ExtendedType interface.
func (t *DoltgresType) SerializeValue(ctx context.Context, val any) ([]byte, error) {
	if val == nil {
		return nil, nil
	}
	sqlCtx, _ := ctx.(*sql.Context) // There are cases where it's okay to serialize with a nil SQL context
	if t.SerializationFunc != nil {
		return t.SerializationFunc(sqlCtx, t, val)
	}
	// If there's not a built-in serialization function, then we'll use the `send` function instead
	return t.CallSend(sqlCtx, val)
}

// DeserializeValue implements the types.ExtendedType interface.
func (t *DoltgresType) DeserializeValue(ctx context.Context, val []byte) (any, error) {
	if len(val) == 0 {
		return nil, nil
	}
	sqlCtx, _ := ctx.(*sql.Context) // There are cases where it's okay to deserialize with a nil SQL context
	if t.DeserializationFunc != nil {
		return t.DeserializationFunc(sqlCtx, t, val)
	}
	// If there's not a built-in deserialization function, then we'll use the `receive` function instead
	return t.CallReceive(sqlCtx, val)
}

// SerializationCompatible implements the val.TupleTypeHandler interface.
func (t *DoltgresType) SerializationCompatible(other val.TupleTypeHandler) bool {
	ot, ok := other.(*DoltgresType)
	return ok && t.Equals(ot)
}

// CallSend is a way to call the `send` function for this type.
func (t *DoltgresType) CallSend(ctx *sql.Context, val any) ([]byte, error) {
	var o any
	var err error
	if t.ModInFunc != 0 || t.IsArrayType() {
		send := globalFunctionRegistry.GetFunction(ctx, t.SendFunc)
		resolvedTypes := send.ResolvedTypes()
		resolvedTypes[0] = t
		o, err = send.WithResolvedTypes(resolvedTypes).(QuickFunction).CallVariadic(ctx, val)
	} else {
		o, err = globalFunctionRegistry.GetFunction(ctx, t.SendFunc).CallVariadic(ctx, val)
	}
	if err != nil || o == nil {
		return nil, err
	}
	return o.([]byte), nil
}

// CallReceive is a way to call the `receive` function for this type.
func (t *DoltgresType) CallReceive(ctx *sql.Context, val []byte) (any, error) {
	if t.TypType == TypeType_Domain {
		return globalFunctionRegistry.GetFunction(ctx, t.ReceiveFunc).CallVariadic(ctx, val, t.BaseTypeType.ID.AsId(), t.attTypMod)
	} else if t.ModInFunc != 0 || t.IsArrayType() {
		if t.Elem.ID != id.NullType {
			return globalFunctionRegistry.GetFunction(ctx, t.ReceiveFunc).CallVariadic(ctx, val, t.Elem.ID.AsId(), t.attTypMod)
		} else {
			return globalFunctionRegistry.GetFunction(ctx, t.ReceiveFunc).CallVariadic(ctx, val, t.ID.AsId(), t.attTypMod)
		}
	} else if t.TypType == TypeType_Enum {
		return globalFunctionRegistry.GetFunction(ctx, t.ReceiveFunc).CallVariadic(ctx, val, t.ID.AsId())
	} else if t.IsCompositeType() {
		return globalFunctionRegistry.GetFunction(ctx, t.ReceiveFunc).CallVariadic(ctx, val, t.ID.AsId(), t.attTypMod)
	} else {
		return globalFunctionRegistry.GetFunction(ctx, t.ReceiveFunc).CallVariadic(ctx, val)
	}
}

// ConvertSerialized implements the val.TupleTypeHandler interface.
func (t *DoltgresType) ConvertSerialized(ctx context.Context, other val.TupleTypeHandler, val []byte) ([]byte, error) {
	ot, ok := other.(*DoltgresType)
	if !ok {
		if other == nil {
			// TODO: replace this with something much better than this hack
			//  `other` will be nil in cases where the parent and child are mismatched (AdaptiveEncoding vs StringEnc for example)
			//  Since we do restrict foreign keys to similar types, if the calling type is a string type, we can fairly
			//  safely assume that the other type is also a string type. This is implemented specifically for a customer
			//  issue, as should be replaced as soon as a real fix is found.
			switch t.ID.TypeName() {
			case "bpchar", "text", "varchar":
				var str string
				if len(val) > 0 {
					if val[len(val)-1] == 0 {
						str = string(val[:len(val)-1])
					} else {
						str = string(val)
					}
				}
				return t.SerializeValue(ctx, str)
			}
		}
		return nil, errors.Errorf("expected DoltgresType, got %T", other)
	}

	value, err := ot.DeserializeValue(ctx, val)
	if err != nil {
		return nil, err
	}

	sqlCtx, _ := ctx.(*sql.Context)
	toValue, _, err := t.ConvertToType(sqlCtx, ot, value)
	if err != nil {
		return nil, err
	}

	return t.SerializeValue(ctx, toValue)
}

// TypeInfo implements the typeinfo.ExtendedType interface.
func (t *DoltgresType) TypeInfo() typeinfo.TypeInfo {
	return typeInfo{
		Type: t,
	}
}

// Copy returns a copy of the type without the cache and mutex
func (t *DoltgresType) Copy() *DoltgresType {
	return &DoltgresType{
		ID:          t.ID,
		TypType:     t.TypType,
		TypCategory: t.TypCategory,
		TypLength:   t.TypLength,
		PassedByVal: t.PassedByVal,
		IsPreferred: t.IsPreferred,
		IsDefined:   t.IsDefined,
		Delimiter:   t.Delimiter,

		RelID:         t.RelID,
		SubscriptFunc: t.SubscriptFunc,
		Elem:          t.Elem,
		Array:         t.Array,
		InputFunc:     t.InputFunc,
		OutputFunc:    t.OutputFunc,
		ReceiveFunc:   t.ReceiveFunc,
		SendFunc:      t.SendFunc,
		ModInFunc:     t.ModInFunc,
		ModOutFunc:    t.ModOutFunc,
		AnalyzeFunc:   t.AnalyzeFunc,
		Align:         t.Align,
		Storage:       t.Storage,

		NotNull:      t.NotNull,
		BaseTypeType: t.BaseTypeType,
		TypMod:       t.TypMod,
		NDims:        t.NDims,
		TypCollation: t.TypCollation,
		DefaulBin:    t.DefaulBin,
		Default:      t.Default,
		Acl:          t.Acl,

		Checks:         t.Checks,
		attTypMod:      t.attTypMod,
		CompareFunc:    t.CompareFunc,
		InternalName:   t.InternalName,
		EnumLabels:     t.EnumLabels,
		CompositeAttrs: t.CompositeAttrs,

		IsSerial:            t.IsSerial,
		IsUnresolved:        t.IsUnresolved,
		BaseTypeForInternal: t.BaseTypeForInternal,
		SerializationFunc:   t.SerializationFunc,
		DeserializationFunc: t.DeserializationFunc,
	}
}

// TypeCastFunction is a function that takes a value of a particular kind of type, and returns it as another kind of type.
// The targetType given should match the "To" type used to obtain the cast.
type TypeCastFunction func(ctx *sql.Context, val any, sourceType *DoltgresType, targetType *DoltgresType) (any, error)