// 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 framework

import (
	"fmt"
	"strings"

	cerrors "github.com/cockroachdb/errors"
	"github.com/dolthub/go-mysql-server/sql"
	"github.com/dolthub/go-mysql-server/sql/expression"
	"github.com/dolthub/go-mysql-server/sql/procedures"
	"gopkg.in/src-d/go-errors.v1"

	"github.com/dolthub/doltgresql/core"
	"github.com/dolthub/doltgresql/core/casts"
	"github.com/dolthub/doltgresql/core/extensions"
	"github.com/dolthub/doltgresql/core/extensions/pg_extension"
	"github.com/dolthub/doltgresql/core/id"
	"github.com/dolthub/doltgresql/server/plpgsql"
	pgtypes "github.com/dolthub/doltgresql/server/types"
)

// ErrFunctionDoesNotExist is returned when the function in use cannot be found.
var ErrFunctionDoesNotExist = errors.NewKind(`function %s does not exist`)

// Function is an expression that represents either a CompiledFunction or a QuickFunction.
type Function interface {
	sql.FunctionExpression
	sql.NonDeterministicExpression
	specificFuncImpl()
}

// CompiledFunction is an expression that represents a fully-analyzed PostgreSQL function.
type CompiledFunction struct {
	Name          string
	Arguments     []sql.Expression
	IsOperator    bool
	overloads     *Overloads
	fnOverloads   []Overload
	overload      overloadMatch
	originalTypes []*pgtypes.DoltgresType
	callResolved  []*pgtypes.DoltgresType
	runner        sql.StatementRunner
	stashedErr    error
}

var _ sql.FunctionExpression = (*CompiledFunction)(nil)
var _ sql.NonDeterministicExpression = (*CompiledFunction)(nil)
var _ procedures.InterpreterExpr = (*CompiledFunction)(nil)
var _ sql.RowIterExpression = (*CompiledFunction)(nil)

// NewCompiledFunction returns a newly compiled function.
func NewCompiledFunction(ctx *sql.Context, name string, args []sql.Expression, functions *Overloads, isOperator bool) *CompiledFunction {
	return newCompiledFunctionInternal(ctx, name, args, functions, functions.overloadsForParams(len(args)), isOperator, nil)
}

// newCompiledFunctionInternal is called internally, which skips steps that may have already been processed.
func newCompiledFunctionInternal(
	ctx *sql.Context,
	name string,
	args []sql.Expression,
	overloads *Overloads,
	fnOverloads []Overload,
	isOperator bool,
	runner sql.StatementRunner,
) *CompiledFunction {
	c := &CompiledFunction{
		Name:        name,
		Arguments:   args,
		IsOperator:  isOperator,
		overloads:   overloads,
		fnOverloads: fnOverloads,
		runner:      runner,
	}
	// First we'll analyze all the parameters.
	originalTypes, err := c.analyzeParameters(ctx)
	if err != nil {
		// Errors should be returned from the call to Eval, so we'll stash it for now
		c.stashedErr = err
		return c
	}
	// Next we'll resolve the overload based on the parameters given.
	overload, err := c.resolve(ctx, overloads, fnOverloads, originalTypes)
	if err != nil {
		c.stashedErr = err
		return c
	}
	// If we do not receive an overload, then the parameters given did not result in a valid match
	if !overload.Valid() {
		if isOperator {
			if strings.HasPrefix(name, "internal_binary_operator_func_") {
				opStr := strings.TrimPrefix(name, "internal_binary_operator_func_")
				var leftType, rightType string
				if len(originalTypes) > 0 {
					leftType = originalTypes[0].String()
				}
				if len(originalTypes) > 1 {
					rightType = originalTypes[1].String()
				}
				c.stashedErr = cerrors.Errorf("operator does not exist: %s %s %s", leftType, opStr, rightType)
				return c
			} else if strings.HasPrefix(name, "internal_unary_operator_func_") {
				opStr := strings.TrimPrefix(name, "internal_unary_operator_func_")
				var childType string
				if len(originalTypes) > 0 {
					childType = originalTypes[0].String()
				}
				c.stashedErr = cerrors.Errorf("operator does not exist: %s%s", opStr, childType)
				return c
			}
		}
		c.stashedErr = ErrFunctionDoesNotExist.New(c.OverloadString(originalTypes))
		return c
	}

	fn := overload.Function()

	// Then we'll handle the polymorphic types
	// https://www.postgresql.org/docs/15/extend-type-system.html#EXTEND-TYPES-POLYMORPHIC
	c.callResolved = make([]*pgtypes.DoltgresType, len(overload.params.paramTypes)+1)
	hasPolymorphicParam := false
	for i, param := range overload.params.paramTypes {
		if param.IsPolymorphicType() {
			// resolve will ensure that the parameter types are valid, so we can just assign them here
			hasPolymorphicParam = true
			c.callResolved[i] = originalTypes[i]
		} else if param.ID == pgtypes.Any.ID {
			c.callResolved[i] = originalTypes[i]
		} else if i < len(args) {
			if d, ok := args[i].Type(ctx).(*pgtypes.DoltgresType); ok {
				if param.IsRecordType() && d.IsCompositeType() {
					// Preserve the composite type's field info (CompositeAttrs) so that
					// functions like row_to_json can access column names at call time.
					param = d
				} else {
					// `param` is a default type which does not have type modifier set
					param = param.WithAttTypMod(d.GetAttTypMod())
				}
			}
			c.callResolved[i] = param
		}
	}
	returnType := fn.GetReturn()
	c.callResolved[len(c.callResolved)-1] = returnType
	if returnType.IsPolymorphicType() {
		if hasPolymorphicParam {
			c.callResolved[len(c.callResolved)-1] = c.resolvePolymorphicReturnType(overload.params.paramTypes, originalTypes, returnType)
		} else if c.Name == "array_in" || c.Name == "array_recv" || c.Name == "enum_in" || c.Name == "enum_recv" || c.Name == "anyenum_in" || c.Name == "anyenum_recv" {
			// The return type should resolve to the type of OID value passed in as second argument.
			// TODO: Possible that the oid type has a special property with polymorphic return types,
			//  in that perhaps their value will set the return type in the absence of another polymorphic type in the parameter list
		} else {
			c.stashedErr = cerrors.Errorf("A result of type %s requires at least one input of type anyelement, anyarray, anynonarray, anyenum, anyrange, or anymultirange.", returnType.String())
			return c
		}
	}

	// Lastly, we assign everything to the function struct
	c.overload = overload
	c.originalTypes = originalTypes
	return c
}

// FunctionName implements the interface sql.Expression.
func (c *CompiledFunction) FunctionName() string {
	return c.Name
}

// Description implements the interface sql.Expression.
func (c *CompiledFunction) Description() string {
	return fmt.Sprintf("The PostgreSQL function `%s`", c.Name)
}

// Resolved implements the interface sql.Expression.
func (c *CompiledFunction) Resolved() bool {
	for _, param := range c.Arguments {
		if !param.Resolved() {
			return false
		}
	}
	// We don't error until evaluation time, so we need to tell the engine we're resolved if there was a stashed error
	return c.stashedErr != nil || c.overload.Valid()
}

// StashedError returns the stashed error if one exists. Otherwise, returns nil.
func (c *CompiledFunction) StashedError() error {
	if c == nil {
		return nil
	}
	return c.stashedErr
}

// String implements the interface sql.Expression.
func (c *CompiledFunction) String() string {
	sb := strings.Builder{}
	sb.WriteString(c.Name + "(")
	for i, param := range c.Arguments {
		// Aliases will output the string "x as x", which is an artifact of how we build the AST, so we'll bypass it
		if alias, ok := param.(*expression.Alias); ok {
			param = alias.Child
		}
		if i > 0 {
			sb.WriteString(", ")
		}
		sb.WriteString(param.String())
	}
	sb.WriteString(")")
	return sb.String()
}

// OverloadString returns the name of the function represented by the given overload.
func (c *CompiledFunction) OverloadString(types []*pgtypes.DoltgresType) string {
	sb := strings.Builder{}
	sb.WriteString(c.Name + "(")
	for i, t := range types {
		if i > 0 {
			sb.WriteString(", ")
		}
		sb.WriteString(t.String())
	}
	sb.WriteString(")")
	return sb.String()
}

// Type implements the interface sql.Expression.
func (c *CompiledFunction) Type(ctx *sql.Context) sql.Type {
	if len(c.callResolved) > 0 {
		rt := c.callResolved[len(c.callResolved)-1]
		rt = getTypeIfRowType(c.IsSRF(), rt)
		// If the return type is polymorphic type, we need the underlying type to be able to
		// convert the result value using the base type.
		// TODO: need to add underlying to these types for IO input and output uses
		if rt.IsPolymorphicType() && len(c.originalTypes) > 0 {
			rt = c.originalTypes[0]
			if rt.IsArrayType() {
				return rt.ArrayBaseType()
			}
		}
		return rt
	}
	// Compilation must have errored, so we'll return the unknown type
	return pgtypes.Unknown
}

// IsNullable implements the interface sql.Expression.
func (c *CompiledFunction) IsNullable(ctx *sql.Context) bool {
	// All functions seem to return NULL when given a NULL value
	return true
}

// IsNonDeterministic implements the interface sql.NonDeterministicExpression.
func (c *CompiledFunction) IsNonDeterministic() bool {
	if c.overload.Valid() {
		return c.overload.Function().NonDeterministic()
	}
	// Compilation must have errored, so we'll just return true
	return true
}

// IsStrict returns whether this function has the STRICT property regarding nulls.
func (c *CompiledFunction) IsStrict() bool {
	if c.overload.Valid() {
		return c.overload.Function().IsStrict()
	}
	return false
}

// IsSRF returns whether this function is a set returning function.
func (c *CompiledFunction) IsSRF() bool {
	if c.overload.Valid() {
		return c.overload.Function().IsSRF()
	}
	return false
}

// IsVariadic returns whether this function has any variadic parameters.
func (c *CompiledFunction) IsVariadic() bool {
	if c.overload.Valid() {
		return c.overload.params.variadic != -1
	}
	// Compilation must have errored, so we'll just return true
	return true
}

// Eval implements the interface sql.Expression.
func (c *CompiledFunction) Eval(ctx *sql.Context, row sql.Row) (interface{}, error) {
	// If we have a stashed error, then we should return that now. Errors are stashed when they're supposed to be
	// returned during the call to Eval. This helps to ensure consistency with how errors are returned in Postgres.
	if c.stashedErr != nil {
		return nil, c.stashedErr
	}

	// Evaluate all arguments, returning immediately if we encounter a null argument and the function is marked STRICT
	var err error
	isStrict := c.overload.Function().IsStrict()
	args := make([]any, len(c.Arguments))
	exprTypes := make([]*pgtypes.DoltgresType, len(args))
	for i, arg := range c.Arguments {
		args[i], err = arg.Eval(ctx, row)
		if err != nil {
			return nil, err
		}
		var ok bool
		if exprTypes[i], ok = arg.Type(ctx).(*pgtypes.DoltgresType); !ok {
			dt, err := pgtypes.FromGmsTypeToDoltgresType(arg.Type(ctx))
			if err != nil {
				return nil, err
			}
			args[i], _, _ = dt.Convert(ctx, args[i])
			exprTypes[i] = dt
		}
		if args[i] == nil && isStrict {
			return nil, nil
		}
	}

	if len(c.overload.casts) > 0 {
		targetParamTypes := c.overload.params.paramTypes
		for i, arg := range args {
			// For variadic params, we need to identify the corresponding target type
			var targetType *pgtypes.DoltgresType
			isVariadicArg := c.overload.params.variadic >= 0 && i >= len(c.overload.params.paramTypes)-1
			if isVariadicArg {
				targetType = targetParamTypes[c.overload.params.variadic]
				if !targetType.IsArrayType() {
					// should be impossible, we check this at function compile time
					return nil, cerrors.Errorf("variadic arguments must be array types, was %T", targetType)
				}
				targetType = targetType.ArrayBaseType()
			} else {
				targetType = targetParamTypes[i]
				// When the declared parameter type is anyarray, the implicit cast from an
				// unknown/text argument (via UseInOut) would target anyarray.IoInput which
				// cannot be loaded as a QuickFunction. Resolve to the concrete array type
				// (e.g. _aggtype) so the cast uses the real element-type I/O path instead.
				// TODO: If targetType.ID can be resolved to the concrete type earlier in
				//       processing, then we don't need this check here anymore.
				if targetType.ID == pgtypes.AnyArray.ID {
					targetType = c.resolvePolymorphicReturnType(targetParamTypes, exprTypes, targetType)
				}
			}

			if c.overload.casts[i].ID.IsValid() {
				args[i], err = c.overload.casts[i].Eval(ctx, arg, exprTypes[i], targetType)
				if err != nil {
					return nil, err
				}
			} else {
				return nil, cerrors.Errorf("function %s is missing the appropriate implicit cast", c.OverloadString(c.originalTypes))
			}
		}
	}

	args = c.overload.params.coalesceVariadicValues(args)

	// Call the function
	switch f := c.overload.Function().(type) {
	case Function0:
		return f.Callable(ctx)
	case Function1:
		return f.Callable(ctx, ([2]*pgtypes.DoltgresType)(c.callResolved), args[0])
	case Function1N:
		return f.Callable(ctx, c.callResolved, args[0], args[1:])
	case Function2:
		return f.Callable(ctx, ([3]*pgtypes.DoltgresType)(c.callResolved), args[0], args[1])
	case Function2N:
		return f.Callable(ctx, c.callResolved, args[0], args[1], args[2:])
	case Function3:
		return f.Callable(ctx, ([4]*pgtypes.DoltgresType)(c.callResolved), args[0], args[1], args[2])
	case Function4:
		return f.Callable(ctx, ([5]*pgtypes.DoltgresType)(c.callResolved), args[0], args[1], args[2], args[3])
	case Function5:
		return f.Callable(ctx, ([6]*pgtypes.DoltgresType)(c.callResolved), args[0], args[1], args[2], args[3], args[4])
	case Function6:
		return f.Callable(ctx, ([7]*pgtypes.DoltgresType)(c.callResolved), args[0], args[1], args[2], args[3], args[4], args[5])
	case Function7:
		return f.Callable(ctx, ([8]*pgtypes.DoltgresType)(c.callResolved), args[0], args[1], args[2], args[3], args[4], args[5], args[6])
	case InterpretedFunction:
		return plpgsql.Call(ctx, f, c.runner, c.callResolved, args)
	case CFunction:
		cfunc, err := extensions.GetExtensionFunction(f.ExtensionName, f.ExtensionSymbol)
		if err != nil {
			return nil, err
		}
		cargs := make([]pg_extension.NullableDatum, len(args))
		for i, argType := range f.ParameterTypes { // TODO: ParameterTypes does not account for variadic parameters
			cConvFunc, ok := cConversionToDatumMap[argType.ID]
			if !ok {
				return nil, cerrors.Errorf("no conversion function from Go to C for `%s`", argType.ID.TypeName())
			}
			cargs[i], err = cConvFunc(args[i])
			if err != nil {
				return nil, err
			}
		}
		result, isNotNull := pg_extension.CallFmgrFunction(cfunc.Ptr, cargs...)
		if isNotNull {
			cConvFunc, ok := cConversionFromDatumMap[f.ReturnType.ID]
			if !ok {
				return nil, cerrors.Errorf("no conversion function from C to Go for `%s`", f.ReturnType.ID.TypeName())
			}
			retVal, err := cConvFunc(result)
			if err != nil {
				return nil, err
			}
			return retVal, nil
		} else {
			return nil, nil
		}
	case SQLFunction:
		return CallSqlFunction(ctx, f, c.runner, args)
	default:
		return nil, cerrors.Errorf("unknown function type in CompiledFunction::Eval %T", f)
	}
}

// EvalRowIter implements sql.RowIterExpression
func (c *CompiledFunction) EvalRowIter(ctx *sql.Context, r sql.Row) (sql.RowIter, error) {
	eval, err := c.Eval(ctx, r)
	if err != nil {
		return nil, err
	}

	switch v := eval.(type) {
	case sql.RowIter:
		return v, nil
	case nil:
		return nil, nil
	default:
		return nil, cerrors.Errorf("function %s returned a value of type %T, which is not a RowIter", c.Name, eval)
	}
}

// ReturnsRowIter implements the interface sql.RowIterExpression
func (c *CompiledFunction) ReturnsRowIter() bool {
	return c.IsSRF()
}

// Children implements the interface sql.Expression.
func (c *CompiledFunction) Children() []sql.Expression {
	return c.Arguments
}

// WithChildren implements the interface sql.Expression.
func (c *CompiledFunction) WithChildren(ctx *sql.Context, children ...sql.Expression) (sql.Expression, error) {
	if len(children) != len(c.Arguments) {
		return nil, sql.ErrInvalidChildrenNumber.New(len(children), len(c.Arguments))
	}

	// We have to re-resolve here, since the change in children may require it (e.g. we have more type info than we did)
	return newCompiledFunctionInternal(ctx, c.Name, children, c.overloads, c.fnOverloads, c.IsOperator, c.runner), nil
}

// SetStatementRunner implements the interface analyzer.Interpreter.
func (c *CompiledFunction) SetStatementRunner(ctx *sql.Context, runner sql.StatementRunner) sql.Expression {
	nc := *c
	nc.runner = runner
	return &nc
}

// GetQuickFunction returns the QuickFunction form of this function, if it exists. If one does not exist, then this
// return nil.
func (c *CompiledFunction) GetQuickFunction() QuickFunction {
	if c.stashedErr != nil || !c.Resolved() || !c.overload.Valid() || c.overload.params.variadic != -1 ||
		len(c.overload.casts) > 0 {
		return nil
	}
	switch f := c.overload.Function().(type) {
	case Function1:
		return &QuickFunction1{
			Name:         c.Name,
			Argument:     c.Arguments[0],
			IsStrict:     c.overload.Function().IsStrict(),
			IsSRF:        c.IsSRF(),
			callResolved: ([2]*pgtypes.DoltgresType)(c.callResolved),
			function:     f,
		}
	case Function2:
		return &QuickFunction2{
			Name:         c.Name,
			Arguments:    ([2]sql.Expression)(c.Arguments),
			IsStrict:     c.overload.Function().IsStrict(),
			IsSRF:        c.IsSRF(),
			callResolved: ([3]*pgtypes.DoltgresType)(c.callResolved),
			function:     f,
		}
	case Function3:
		return &QuickFunction3{
			Name:         c.Name,
			Arguments:    ([3]sql.Expression)(c.Arguments),
			IsStrict:     c.overload.Function().IsStrict(),
			IsSRF:        c.IsSRF(),
			callResolved: ([4]*pgtypes.DoltgresType)(c.callResolved),
			function:     f,
		}
	default:
		return nil
	}
}

// resolve returns an overloadMatch that either matches the given parameters exactly, or is a viable match after casting.
// Returns an invalid overloadMatch if a viable match is not found.
func (c *CompiledFunction) resolve(ctx *sql.Context, overloads *Overloads, fnOverloads []Overload, argTypes []*pgtypes.DoltgresType) (overloadMatch, error) {
	// First check for an exact match
	exactMatch, found := overloads.ExactMatchForTypes(argTypes...)
	if found {
		return overloadMatch{
			params: Overload{
				function:   exactMatch,
				paramTypes: argTypes,
				argTypes:   argTypes,
				variadic:   -1,
			},
		}, nil
	}
	// There are no exact matches, so now we'll look through all overloads to determine the best match. This is
	// much more work, but there's a performance penalty for runtime overload resolution in Postgres as well.
	if c.IsOperator {
		return c.resolveOperator(ctx, argTypes, overloads, fnOverloads)
	} else {
		return c.resolveFunction(ctx, argTypes, fnOverloads)
	}
}

// resolveOperator resolves an operator according to the rules defined by Postgres.
// https://www.postgresql.org/docs/15/typeconv-oper.html
func (c *CompiledFunction) resolveOperator(ctx *sql.Context, argTypes []*pgtypes.DoltgresType, overloads *Overloads, fnOverloads []Overload) (overloadMatch, error) {
	// Binary operators treat unknown literals as the other type, so we'll account for that here to see if we can find
	// an "exact" match.
	if len(argTypes) == 2 {
		leftUnknownType := argTypes[0].ID == pgtypes.Unknown.ID
		rightUnknownType := argTypes[1].ID == pgtypes.Unknown.ID
		if (leftUnknownType && !rightUnknownType) || (!leftUnknownType && rightUnknownType) {
			var typ *pgtypes.DoltgresType
			identity := casts.Cast{
				ID:       id.NewCast(argTypes[0].ID, argTypes[1].ID),
				CastType: casts.CastType_Explicit,
				Function: id.NullFunction,
				UseInOut: false,
			}
			opCasts := []casts.Cast{identity, identity}
			if leftUnknownType {
				opCasts[0].UseInOut = true
				typ = argTypes[1]
			} else {
				opCasts[1].UseInOut = true
				typ = argTypes[0]
			}
			if exactMatch, ok := overloads.ExactMatchForTypes(typ, typ); ok {
				return overloadMatch{
					params: Overload{
						function:   exactMatch,
						paramTypes: []*pgtypes.DoltgresType{typ, typ},
						argTypes:   []*pgtypes.DoltgresType{typ, typ},
						variadic:   -1,
					},
					casts: opCasts,
				}, nil
			}
		}
	}
	// From this point, the steps appear to be the same for functions and operators
	return c.resolveFunction(ctx, argTypes, fnOverloads)
}

// resolveFunction resolves a function according to the rules defined by Postgres.
// https://www.postgresql.org/docs/15/typeconv-func.html
func (c *CompiledFunction) resolveFunction(ctx *sql.Context, argTypes []*pgtypes.DoltgresType, overloads []Overload) (overloadMatch, error) {
	// First we'll discard all overloads that do not have implicitly-convertible param types
	compatibleOverloads, err := c.typeCompatibleOverloads(ctx, overloads, argTypes)
	if err != nil {
		return overloadMatch{}, err
	}

	// No compatible overloads available, return early
	if len(compatibleOverloads) == 0 {
		return overloadMatch{}, nil
	}

	// If we've found exactly one match then we'll return that one
	// TODO: we need to also prefer non-variadic functions here over variadic ones (no such conflict can exist for now)
	//  https://www.postgresql.org/docs/15/typeconv-func.html
	if len(compatibleOverloads) == 1 {
		return compatibleOverloads[0], nil
	}

	// Next rank the candidates by the number of params whose types match exactly
	closestMatches := c.closestTypeMatches(argTypes, compatibleOverloads)

	// Now check again for exactly one match
	if len(closestMatches) == 1 {
		return closestMatches[0], nil
	}

	// If there was more than a single match, try to find the one with the most preferred type conversions
	preferredOverloads := c.preferredTypeMatches(argTypes, closestMatches)

	// Check once more for exactly one match
	if len(preferredOverloads) == 1 {
		return preferredOverloads[0], nil
	}

	// Next we'll check the type categories for `unknown` types
	unknownOverloads, ok := c.unknownTypeCategoryMatches(argTypes, preferredOverloads)
	if !ok {
		return overloadMatch{}, nil
	}

	// Check again for exactly one match
	if len(unknownOverloads) == 1 {
		return unknownOverloads[0], nil
	}

	// No matching function overload found
	return overloadMatch{}, nil
}

// typeCompatibleOverloads returns all overloads that have a matching number of params whose types can be
// implicitly converted to the ones provided. This is the set of all possible overloads that could be used with the
// param types provided.
func (c *CompiledFunction) typeCompatibleOverloads(ctx *sql.Context, fnOverloads []Overload, argTypes []*pgtypes.DoltgresType) ([]overloadMatch, error) {
	castsColl, err := core.GetCastsCollectionFromContext(ctx, "")
	if err != nil {
		return nil, err
	}
	var compatible []overloadMatch
	for _, overload := range fnOverloads {
		isConvertible := true
		overloadCasts := make([]casts.Cast, len(argTypes))
		// Polymorphic parameters must be gathered so that we can later verify that they all have matching base types
		var polymorphicParameters []*pgtypes.DoltgresType
		var polymorphicTargets []*pgtypes.DoltgresType
		for i := range argTypes {
			paramType := overload.argTypes[i]
			if paramType.IsValidForPolymorphicType(argTypes[i]) {
				overloadCasts[i] = casts.Cast{
					ID:       id.NewCast(argTypes[i].ID, paramType.ID),
					CastType: casts.CastType_Explicit,
					Function: id.NullFunction,
					UseInOut: false,
				}
				polymorphicParameters = append(polymorphicParameters, paramType)
				polymorphicTargets = append(polymorphicTargets, argTypes[i])
			} else if paramType.IsRecordType() && argTypes[i].IsCompositeType() {
				// Composite types (e.g. table row types) are compatible with the generic Record parameter.
				overloadCasts[i] = casts.Cast{
					ID:       id.NewCast(argTypes[i].ID, paramType.ID),
					CastType: casts.CastType_Explicit,
					Function: id.NullFunction,
					UseInOut: false,
				}
			} else {
				var err error
				overloadCasts[i], err = castsColl.GetImplicitCast(ctx, argTypes[i], paramType)
				if err != nil {
					return nil, err
				}
				if !overloadCasts[i].ID.IsValid() {
					isConvertible = false
					break
				}
			}
		}

		if isConvertible && c.polymorphicTypesCompatible(polymorphicParameters, polymorphicTargets) {
			compatible = append(compatible, overloadMatch{params: overload, casts: overloadCasts})
		}
	}
	return compatible, nil
}

// closestTypeMatches returns the set of overload candidates that have the most exact type matches for the arg types
// provided.
func (*CompiledFunction) closestTypeMatches(argTypes []*pgtypes.DoltgresType, candidates []overloadMatch) []overloadMatch {
	matchCount := 0
	var matches []overloadMatch
	for _, cand := range candidates {
		currentMatchCount := 0
		for argIdx := range argTypes {
			argType := cand.params.argTypes[argIdx]
			if argTypes[argIdx].ID == argType.ID || (argTypes[argIdx].ID == pgtypes.Unknown.ID && argType.ID == pgtypes.Text.ID) {
				currentMatchCount++
			}
		}
		if currentMatchCount > matchCount {
			matchCount = currentMatchCount
			matches = append([]overloadMatch{}, cand)
		} else if currentMatchCount == matchCount {
			matches = append(matches, cand)
		}
	}
	return matches
}

// preferredTypeMatches returns the overload candidates that have the most preferred types for args that require casts.
func (*CompiledFunction) preferredTypeMatches(argTypes []*pgtypes.DoltgresType, candidates []overloadMatch) []overloadMatch {
	preferredCount := 0
	var preferredOverloads []overloadMatch
	for _, cand := range candidates {
		currentPreferredCount := 0
		for argIdx := range argTypes {
			argType := cand.params.argTypes[argIdx]
			if argTypes[argIdx].ID != argType.ID && argType.IsPreferred {
				currentPreferredCount++
			}
		}

		if currentPreferredCount > preferredCount {
			preferredCount = currentPreferredCount
			preferredOverloads = append([]overloadMatch{}, cand)
		} else if currentPreferredCount == preferredCount {
			preferredOverloads = append(preferredOverloads, cand)
		}
	}
	return preferredOverloads
}

// unknownTypeCategoryMatches checks the type categories of `unknown` types. These types have an inherent bias toward
// the string category since an `unknown` literal resembles a string. Returns false if the resolution should fail.
func (c *CompiledFunction) unknownTypeCategoryMatches(argTypes []*pgtypes.DoltgresType, candidates []overloadMatch) ([]overloadMatch, bool) {
	matches := make([]overloadMatch, len(candidates))
	copy(matches, candidates)
	// For our first loop, we'll filter matches based on whether they accept the string category
	for argIdx := range argTypes {
		// We're only concerned with `unknown` types
		if argTypes[argIdx].ID != pgtypes.Unknown.ID {
			continue
		}
		var newMatches []overloadMatch
		for _, match := range matches {
			if match.params.argTypes[argIdx].TypCategory == pgtypes.TypeCategory_StringTypes {
				newMatches = append(newMatches, match)
			}
		}
		// If we've found matches in this step, then we'll update our match set
		if len(newMatches) > 0 {
			matches = newMatches
		}
	}
	// Return early if we've filtered down to a single match
	if len(matches) == 1 {
		return matches, true
	}
	// TODO: implement the remainder of step 4.e. from the documentation (following code assumes it has been implemented)
	// ...

	// If we've discarded every function, then we'll actually return all original candidates
	if len(matches) == 0 {
		return candidates, true
	}
	// In this case, we've trimmed at least one candidate, so we'll return our new matches
	return matches, true
}

// polymorphicTypesCompatible returns whether any polymorphic types given are compatible with the expression types given
func (*CompiledFunction) polymorphicTypesCompatible(paramTypes []*pgtypes.DoltgresType, exprTypes []*pgtypes.DoltgresType) bool {
	if len(paramTypes) != len(exprTypes) {
		return false
	}
	// If there are less than two parameters then we don't even need to check
	if len(paramTypes) < 2 {
		return true
	}

	// If one of the types is anyarray, then anyelement behaves as anynonarray, so we can convert them to anynonarray
	for _, paramType := range paramTypes {
		if paramType.ID == pgtypes.AnyArray.ID {
			// At least one parameter is anyarray, so copy all parameters to a new slice and replace anyelement with anynonarray
			newParamTypes := make([]*pgtypes.DoltgresType, len(paramTypes))
			copy(newParamTypes, paramTypes)
			for i := range newParamTypes {
				if paramTypes[i].ID == pgtypes.AnyElement.ID {
					newParamTypes[i] = pgtypes.AnyNonArray
				}
			}
			paramTypes = newParamTypes
			break
		}
	}

	// The base type is the type that must match between all polymorphic types.
	var baseType *pgtypes.DoltgresType
	for i, paramType := range paramTypes {
		if paramType.IsPolymorphicType() && exprTypes[i].ID != pgtypes.Unknown.ID {
			// Although we do this check before we ever reach this function, we do it again as we may convert anyelement
			// to anynonarray, which changes type validity
			if !paramType.IsValidForPolymorphicType(exprTypes[i]) {
				return false
			}
			// Get the base expression type that we'll compare against
			baseExprType := exprTypes[i]
			if baseExprType.IsArrayType() {
				baseExprType = baseExprType.ArrayBaseType()
			}
			// TODO: handle range types
			// Check that the base expression type matches the previously-found base type
			if baseType.IsEmptyType() {
				baseType = baseExprType
			} else if baseType.ID != baseExprType.ID {
				return false
			}
		}
	}
	return true
}

// resolvePolymorphicReturnType returns the type that should be used for the return type. If the return type is not a
// polymorphic type, then the return type is directly returned. However, if the return type is a polymorphic type, then
// the type is determined using the expression types and parameter types. This makes the assumption that everything has
// already been validated.
func (c *CompiledFunction) resolvePolymorphicReturnType(functionInterfaceTypes []*pgtypes.DoltgresType, originalTypes []*pgtypes.DoltgresType, returnType *pgtypes.DoltgresType) *pgtypes.DoltgresType {
	if !returnType.IsPolymorphicType() {
		return returnType
	}
	// We can use the first polymorphic non-unknown type that we find, since we can morph it into any type that we need.
	// We've verified that all polymorphic types are compatible in a previous step, so this is safe to do.
	var firstPolymorphicType *pgtypes.DoltgresType
	for i, functionInterfaceType := range functionInterfaceTypes {
		if functionInterfaceType.IsPolymorphicType() && originalTypes[i].ID != pgtypes.Unknown.ID {
			firstPolymorphicType = originalTypes[i]
			break
		}
	}

	// if all types are `unknown`, use `text` type
	if firstPolymorphicType.IsEmptyType() {
		firstPolymorphicType = pgtypes.Text
	}

	switch returnType.ID {
	case pgtypes.AnyElement.ID, pgtypes.AnyNonArray.ID:
		// For return types, anyelement behaves the same as anynonarray.
		// This isn't explicitly in the documentation, however it does note that:
		// "...anynonarray and anyenum do not represent separate type variables; they are the same type as anyelement..."
		// The implication of this being that anyelement will always return the base type even for array types,
		// just like anynonarray would.
		if firstPolymorphicType.IsArrayType() {
			return firstPolymorphicType.ArrayBaseType()
		} else {
			return firstPolymorphicType
		}
	case pgtypes.AnyArray.ID:
		// Array types will return themselves, so this is safe
		if firstPolymorphicType.IsArrayType() {
			return firstPolymorphicType
		} else if firstPolymorphicType.ID == pgtypes.Internal.ID {
			return pgtypes.IDToBuiltInDoltgresType[firstPolymorphicType.BaseTypeForInternal]
		} else {
			return firstPolymorphicType.ToArrayType()
		}
	default:
		panic(cerrors.Errorf("`%s` is not yet handled during function compilation", returnType.String()))
	}
}

// analyzeParameters analyzes the parameters within an Eval call.
func (c *CompiledFunction) analyzeParameters(ctx *sql.Context) (originalTypes []*pgtypes.DoltgresType, err error) {
	originalTypes = make([]*pgtypes.DoltgresType, len(c.Arguments))
	for i, param := range c.Arguments {
		returnType := param.Type(ctx)
		if extendedType, ok := returnType.(*pgtypes.DoltgresType); ok && !extendedType.IsEmptyType() {
			if extendedType.TypType == pgtypes.TypeType_Domain {
				extendedType = extendedType.DomainUnderlyingBaseType()
			}
			originalTypes[i] = extendedType
		} else {
			// TODO: we need to remove GMS types from all of our expressions so that we can remove this
			dt, err := pgtypes.FromGmsTypeToDoltgresType(returnType)
			if err != nil {
				return nil, err
			}
			originalTypes[i] = dt
		}
	}
	return originalTypes, nil
}

// specificFuncImpl implements the interface sql.Expression.
func (*CompiledFunction) specificFuncImpl() {}

// getTypeIfRowType returns the underlying type if it's Row Type;
// otherwise, it returns the type that is passed.
func getTypeIfRowType(isSRF bool, t *pgtypes.DoltgresType) *pgtypes.DoltgresType {
	if isSRF {
		// TODO: need support for used defined types
		if typ, ok := pgtypes.IDToBuiltInDoltgresType[t.Elem.ID]; ok {
			return typ
		}
	}
	return t
}

// ResolveDefaultValues adds missing arguments if there is any using the default value set on the parameter.
// It checks if it's a valid SQL function that has fewer arguments than defined parameters.
func (c *CompiledFunction) ResolveDefaultValues(ctx *sql.Context, getDefExpr func(defExpr string) (sql.Expression, error)) error {
	if !c.overload.Valid() {
		return nil
	}
	sqlFunc, ok := c.overload.params.function.(SQLFunction)
	if !ok {
		return nil
	}

	if len(c.Arguments) < len(sqlFunc.ParameterTypes) {
		castsColl, err := core.GetCastsCollectionFromContext(ctx, "")
		if err != nil {
			return err
		}
		for i, param := range sqlFunc.ParameterTypes {
			if i < len(c.Arguments) {
				if exprTypeId := c.Arguments[i].Type(ctx).(*pgtypes.DoltgresType).ID; exprTypeId != pgtypes.Unknown.ID && param.ID != exprTypeId {
					// if non-matching type, then skip appending defaults
					break
				}
			} else if sqlFunc.ParameterDefaults[i] != "" {
				// only if there is default, then append
				cdv, err := getDefExpr(sqlFunc.ParameterDefaults[i])
				if err != nil {
					return err
				}
				c.Arguments = append(c.Arguments, cdv)
				implicitCast, err := castsColl.GetImplicitCast(ctx, cdv.Type(ctx).(*pgtypes.DoltgresType), sqlFunc.ParameterTypes[i])
				if err != nil {
					return err
				}
				c.overload.casts = append(c.overload.casts, implicitCast)
			}
		}
	}

	return nil
}