// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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 compute

import (
	
	
	

	
	
	
	
	
)

var (
	isinDoc = FunctionDoc{
		Summary: "Find each element in a set of values",
		Description: `For each element in "values", return true if it is found 
in a given set, false otherwise`,
		ArgNames:        []string{"values"},
		OptionsType:     "SetOptions",
		OptionsRequired: true,
	}
)

type NullMatchingBehavior = kernels.NullMatchingBehavior

const (
	NullMatchingMatch        = kernels.NullMatchingMatch
	NullMatchingSkip         = kernels.NullMatchingSkip
	NullMatchingEmitNull     = kernels.NullMatchingEmitNull
	NullMatchingInconclusive = kernels.NullMatchingInconclusive
)

type setLookupFunc struct {
	ScalarFunction
}

func ( *setLookupFunc) ( context.Context,  FunctionOptions,  ...Datum) (Datum, error) {
	return execInternal(, , , -1, ...)
}

func ( *setLookupFunc) ( ...arrow.DataType) (exec.Kernel, error) {
	ensureDictionaryDecoded(...)
	return .DispatchExact(...)
}

type SetOptions struct {
	ValueSet     Datum
	NullBehavior NullMatchingBehavior
}

func (*SetOptions) () string { return "SetOptions" }

func initSetLookup( *exec.KernelCtx,  exec.KernelInitArgs) (exec.KernelState, error) {
	if .Options == nil {
		return nil, fmt.Errorf("%w: calling a set lookup function without SetOptions", ErrInvalid)
	}

	,  := .Options.(*SetOptions)
	if ! {
		return nil, fmt.Errorf("%w: expected SetOptions, got %T", ErrInvalid, .Options)
	}

	,  := .ValueSet.(ArrayLikeDatum)
	if ! {
		return nil, fmt.Errorf("%w: expected array-like datum, got %T", ErrInvalid, .ValueSet)
	}

	 := .Inputs[0]
	if (.ID() == arrow.STRING || .ID() == arrow.LARGE_STRING) && !arrow.IsBaseBinary(.Type().ID()) {
		// don't implicitly cast from a non-binary type to string
		// since most types support casting to string and that may lead to
		// surprises. However we do want most other implicit casts
		return nil, fmt.Errorf("%w: array type doesn't match type of values set: %s vs %s",
			ErrInvalid, , .Type())
	}

	if !arrow.TypeEqual(.Type(), ) {
		,  := CastDatum(.Ctx, , SafeCastOptions())
		if  == nil {
			defer .Release()
			 = .(ArrayLikeDatum)
		} else if CanCast(, .Type()) {
			// avoid casting from non-binary types to string like above
			// otherwise will try to cast input array to valueset during
			// execution
			if (.Type().ID() == arrow.STRING || .Type().ID() == arrow.LARGE_STRING) && !arrow.IsBaseBinary(.ID()) {
				return nil, fmt.Errorf("%w: array type doesn't match type of values set: %s vs %s",
					ErrInvalid, , .Type())
			}
		} else {
			return nil, fmt.Errorf("%w: array type doesn't match type of values set: %s vs %s",
				ErrInvalid, , .Type())
		}

	}

	 := kernels.SetLookupOptions{
		ValueSet:     make([]exec.ArraySpan, 1),
		TotalLen:     .ValueSet.Len(),
		NullBehavior: .NullBehavior,
	}

	switch .Kind() {
	case KindArray:
		.ValueSet[0].SetMembers(.(*ArrayDatum).Value)
		.ValueSetType = .(*ArrayDatum).Type()
	case KindChunked:
		 := .(*ChunkedDatum).Value
		.ValueSetType = .DataType()
		.ValueSet = make([]exec.ArraySpan, len(.Chunks()))
		for ,  := range .Chunks() {
			.ValueSet[].SetMembers(.Data())
		}
	default:
		return nil, fmt.Errorf("%w: expected array or chunked array, got %s", ErrInvalid, .ValueSet.Kind())
	}

	return kernels.CreateSetLookupState(, exec.GetAllocator(.Ctx))
}

type setLookupState interface {
	Init(kernels.SetLookupOptions) error
	ValueType() arrow.DataType
}

func execIsIn( *exec.KernelCtx,  *exec.ExecSpan,  *exec.ExecResult) error {
	 := .State.(setLookupState)
	.Kernel.(*exec.ScalarKernel).Data = 
	 := .Values[0]

	if !arrow.TypeEqual(.Type(), .ValueType()) {
		 := .Array.MakeArray()
		defer .Release()

		,  := CastArray(.Ctx, , SafeCastOptions(.ValueType()))
		if  != nil {
			if errors.Is(, arrow.ErrNotImplemented) {
				return fmt.Errorf("%w: array type doesn't match type of values  set: %s vs %s",
					ErrInvalid, .Type(), .ValueType())
			}
			return 
		}
		defer .Release()

		var  exec.ArraySpan
		.SetMembers(.Data())
		return kernels.DispatchIsIn(, &, )
	}

	return kernels.DispatchIsIn(, &.Array, )
}

func ( context.Context,  SetOptions,  Datum) (Datum, error) {
	return CallFunction(, "is_in", &, )
}

func ( context.Context, ,  Datum) (Datum, error) {
	return IsIn(, SetOptions{ValueSet: }, )
}

func ( FunctionRegistry) {
	 := NewScalarFunction("is_in", Unary(), isinDoc)

	 := []exec.InputType{
		exec.NewMatchedInput(exec.Primitive()),
		exec.NewIDInput(arrow.DECIMAL32),
		exec.NewIDInput(arrow.DECIMAL64),
	}

	 := exec.NewOutputType(arrow.FixedWidthTypes.Boolean)
	for ,  := range  {
		 := exec.NewScalarKernel([]exec.InputType{}, , execIsIn, initSetLookup)
		.MemAlloc = exec.MemPrealloc
		.NullHandling = exec.NullComputedPrealloc
		if  := .AddKernel();  != nil {
			panic()
		}
	}

	 := []exec.InputType{
		exec.NewMatchedInput(exec.BinaryLike()),
		exec.NewMatchedInput(exec.LargeBinaryLike()),
		exec.NewExactInput(extensions.NewUUIDType()),
		exec.NewIDInput(arrow.FIXED_SIZE_BINARY),
		exec.NewIDInput(arrow.DECIMAL128),
		exec.NewIDInput(arrow.DECIMAL256),
	}
	for ,  := range  {
		 := exec.NewScalarKernel([]exec.InputType{}, , execIsIn, initSetLookup)
		.MemAlloc = exec.MemPrealloc
		.NullHandling = exec.NullComputedPrealloc
		.CleanupFn = func( exec.KernelState) error {
			 := .(*kernels.SetLookupState[[]byte])
			.Lookup.(*hashing.BinaryMemoTable).Release()
			return nil
		}

		if  := .AddKernel();  != nil {
			panic()
		}
	}

	.AddFunction(&setLookupFunc{*}, false)
}