text: #include "duckdb/function/aggregate/distributive_functions.hpp"
  #include "duckdb/common/exception.hpp"
  #include "duckdb/common/vector_operations/vector_operations.hpp"
  #include "duckdb/common/operator/comparison_operators.hpp"
  #include "duckdb/common/value_operations/value_operations.hpp"
  #include "duckdb/common/vector_operations/aggregate_executor.hpp"
  #include "duckdb/common/vector_operations/vector_operations.hpp"
  #include "duckdb/common/operator/aggregate_operators.hpp"
  #include "duckdb/common/types/null_value.hpp"
  #include "duckdb/planner/expression.hpp"
  
  namespace duckdb {
  
  template <class T>
  struct MinMaxState {
  	T value;
  	bool isset;
  };
  
  template <class OP>
  static AggregateFunction GetUnaryAggregate(LogicalType type) {
  	switch (type.id()) {
  	case LogicalTypeId::BOOLEAN:
  		return AggregateFunction::UnaryAggregate<MinMaxState<int8_t>, int8_t, int8_t, OP>(type, type);
  	case LogicalTypeId::TINYINT:
  		return AggregateFunction::UnaryAggregate<MinMaxState<int8_t>, int8_t, int8_t, OP>(type, type);
  	case LogicalTypeId::SMALLINT:
  		return AggregateFunction::UnaryAggregate<MinMaxState<int16_t>, int16_t, int16_t, OP>(type, type);
  	case LogicalTypeId::DATE:
  	case LogicalTypeId::DATE_TZ:
  	case LogicalTypeId::INTEGER:
  		return AggregateFunction::UnaryAggregate<MinMaxState<int32_t>, int32_t, int32_t, OP>(type, type);
  	case LogicalTypeId::TIMESTAMP:
  	case LogicalTypeId::TIME:
  	case LogicalTypeId::TIMESTAMP_TZ:
  	case LogicalTypeId::TIME_TZ:
  	case LogicalTypeId::BIGINT:
  		return AggregateFunction::UnaryAggregate<MinMaxState<int64_t>, int64_t, int64_t, OP>(type, type);
  	case LogicalTypeId::UTINYINT:
  		return AggregateFunction::UnaryAggregate<MinMaxState<uint8_t>, uint8_t, uint8_t, OP>(type, type);
  	case LogicalTypeId::USMALLINT:
  		return AggregateFunction::UnaryAggregate<MinMaxState<uint16_t>, uint16_t, uint16_t, OP>(type, type);
  	case LogicalTypeId::UINTEGER:
  		return AggregateFunction::UnaryAggregate<MinMaxState<uint32_t>, uint32_t, uint32_t, OP>(type, type);
  	case LogicalTypeId::UBIGINT:
  		return AggregateFunction::UnaryAggregate<MinMaxState<uint64_t>, uint64_t, uint64_t, OP>(type, type);
  	case LogicalTypeId::HUGEINT:
  	case LogicalTypeId::UUID:
  		return AggregateFunction::UnaryAggregate<MinMaxState<hugeint_t>, hugeint_t, hugeint_t, OP>(type, type);
  	case LogicalTypeId::FLOAT:
  		return AggregateFunction::UnaryAggregate<MinMaxState<float>, float, float, OP>(type, type);
  	case LogicalTypeId::DOUBLE:
  		return AggregateFunction::UnaryAggregate<MinMaxState<double>, double, double, OP>(type, type);
  	case LogicalTypeId::INTERVAL:
  		return AggregateFunction::UnaryAggregate<MinMaxState<interval_t>, interval_t, interval_t, OP>(type, type);
  	default:
  		throw InternalException("Unimplemented type for min/max aggregate");
  	}
  }
  
  struct MinMaxBase {
  	template <class STATE>
  	static void Initialize(STATE *state) {
  		state->isset = false;
  	}
  
  	template <class INPUT_TYPE, class STATE, class OP>
  	static void ConstantOperation(STATE *state, FunctionData *bind_data, INPUT_TYPE *input, ValidityMask &mask,
  	                              idx_t count) {
  		D_ASSERT(mask.RowIsValid(0));
  		if (!state->isset) {
  			OP::template Assign<INPUT_TYPE, STATE>(state, input[0]);
  			state->isset = true;
  		} else {
  			OP::template Execute<INPUT_TYPE, STATE>(state, input[0]);
  		}
  	}
  
  	template <class INPUT_TYPE, class STATE, class OP>
  	static void Operation(STATE *state, FunctionData *bind_data, INPUT_TYPE *input, ValidityMask &mask, idx_t idx) {
  		if (!state->isset) {
  			OP::template Assign<INPUT_TYPE, STATE>(state, input[idx]);
  			state->isset = true;
  		} else {
  			OP::template Execute<INPUT_TYPE, STATE>(state, input[idx]);
  		}
  	}
  
  	static bool IgnoreNull() {
  		return true;
  	}
  };
  
  struct NumericMinMaxBase : public MinMaxBase {
  	template <class INPUT_TYPE, class STATE>
  	static void Assign(STATE *state, INPUT_TYPE input) {
  		state->value = input;
  	}
  
  	template <class T, class STATE>
  	static void Finalize(Vector &result, FunctionData *, STATE *state, T *target, ValidityMask &mask, idx_t idx) {
  		mask.Set(idx, state->isset);
  		target[idx] = state->value;
  	}
  };
  
  struct MinOperation : public NumericMinMaxBase {
  	template <class INPUT_TYPE, class STATE>
  	static void Execute(STATE *state, INPUT_TYPE input) {
  		if (LessThan::Operation<INPUT_TYPE>(input, state->value)) {
  			state->value = input;
  		}
  	}
  
  	template <class STATE, class OP>
  	static void Combine(const STATE &source, STATE *target) {
  		if (!source.isset) {
  			// source is NULL, nothing to do
  			return;
  		}
  		if (!target->isset) {
  			// target is NULL, use source value directly
  			*target = source;
  		} else if (GreaterThan::Operation(target->value, source.value)) {
  			target->value = source.value;
  		}
  	}
  };
  
  struct MaxOperation : public NumericMinMaxBase {
  	template <class INPUT_TYPE, class STATE>
  	static void Execute(STATE *state, INPUT_TYPE input) {
  		if (GreaterThan::Operation<INPUT_TYPE>(input, state->value)) {
  			state->value = input;
  		}
  	}
  
  	template <class STATE, class OP>
  	static void Combine(const STATE &source, STATE *target) {
  		if (!source.isset) {
  			// source is NULL, nothing to do
  			return;
  		}
  		if (!target->isset) {
  			// target is NULL, use source value directly
  			*target = source;
  		} else if (LessThan::Operation(target->value, source.value)) {
  			target->value = source.value;
  		}
  	}
  };
  
  struct StringMinMaxBase : public MinMaxBase {
  	template <class STATE>
  	static void Destroy(STATE *state) {
  		if (state->isset && !state->value.IsInlined()) {
  			delete[] state->value.GetDataUnsafe();
  		}
  	}
  
  	template <class INPUT_TYPE, class STATE>
  	static void Assign(STATE *state, INPUT_TYPE input) {
  		Destroy(state);
  		if (input.IsInlined()) {
  			state->value = input;
  		} else {
  			// non-inlined string, need to allocate space for it
  			auto len = input.GetSize();
  			auto ptr = new char[len];
  			memcpy(ptr, input.GetDataUnsafe(), len);
  
  			state->value = string_t(ptr, len);
  		}
  	}
  
  	template <class T, class STATE>
  	static void Finalize(Vector &result, FunctionData *, STATE *state, T *target, ValidityMask &mask, idx_t idx) {
  		if (!state->isset) {
  			mask.SetInvalid(idx);
  		} else {
  			target[idx] = StringVector::AddStringOrBlob(result, state->value);
  		}
  	}
  
  	template <class STATE, class OP>
  	static void Combine(const STATE &source, STATE *target) {
  		if (!source.isset) {
  			// source is NULL, nothing to do
  			return;
  		}
  		if (!target->isset) {
  			// target is NULL, use source value directly
  			Assign(target, source.value);
  			target->isset = true;
  		} else {
  			OP::template Execute<string_t, STATE>(target, source.value);
  		}
  	}
  };
  
  struct MinOperationString : public StringMinMaxBase {
  	template <class INPUT_TYPE, class STATE>
  	static void Execute(STATE *state, INPUT_TYPE input) {
  		if (LessThan::Operation<INPUT_TYPE>(input, state->value)) {
  			Assign(state, input);
  		}
  	}
  };
  
  struct MaxOperationString : public StringMinMaxBase {
  	template <class INPUT_TYPE, class STATE>
  	static void Execute(STATE *state, INPUT_TYPE input) {
  		if (GreaterThan::Operation<INPUT_TYPE>(input, state->value)) {
  			Assign(state, input);
  		}
  	}
  };
  
  template <typename T, class OP>
  static bool TemplatedOptimumType(Vector &left, idx_t lidx, idx_t lcount, Vector &right, idx_t ridx, idx_t rcount) {
  	VectorData lvdata, rvdata;
  	left.Orrify(lcount, lvdata);
  	right.Orrify(rcount, rvdata);
  
  	lidx = lvdata.sel->get_index(lidx);
  	ridx = rvdata.sel->get_index(ridx);
  
  	auto ldata = (const T *)lvdata.data;
  	auto rdata = (const T *)rvdata.data;
  
  	auto &lval = ldata[lidx];
  	auto &rval = rdata[ridx];
  
  	auto lnull = !lvdata.validity.RowIsValid(lidx);
  	auto rnull = !rvdata.validity.RowIsValid(ridx);
  
  	return OP::Operation(lval, rval, lnull, rnull);
  }
  
  template <class OP>
  static bool TemplatedOptimumList(Vector &left, idx_t lidx, idx_t lcount, Vector &right, idx_t ridx, idx_t rcount);
  
  template <class OP>
  static bool TemplatedOptimumStruct(Vector &left, idx_t lidx, idx_t lcount, Vector &right, idx_t ridx, idx_t rcount);
  
  template <class OP>
  static bool TemplatedOptimumValue(Vector &left, idx_t lidx, idx_t lcount, Vector &right, idx_t ridx, idx_t rcount) {
  	D_ASSERT(left.GetType() == right.GetType());
  	switch (left.GetType().InternalType()) {
  	case PhysicalType::BOOL:
  	case PhysicalType::INT8:
  		return TemplatedOptimumType<int8_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::INT16:
  		return TemplatedOptimumType<int16_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::INT32:
  		return TemplatedOptimumType<int32_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::INT64:
  		return TemplatedOptimumType<int64_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::UINT8:
  		return TemplatedOptimumType<uint8_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::UINT16:
  		return TemplatedOptimumType<uint16_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::UINT32:
  		return TemplatedOptimumType<uint32_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::UINT64:
  		return TemplatedOptimumType<uint64_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::INT128:
  		return TemplatedOptimumType<hugeint_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::FLOAT:
  		return TemplatedOptimumType<float, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::DOUBLE:
  		return TemplatedOptimumType<double, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::INTERVAL:
  		return TemplatedOptimumType<interval_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::VARCHAR:
  		return TemplatedOptimumType<string_t, OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::LIST:
  		return TemplatedOptimumList<OP>(left, lidx, lcount, right, ridx, rcount);
  	case PhysicalType::MAP:
  	case PhysicalType::STRUCT:
  		return TemplatedOptimumStruct<OP>(left, lidx, lcount, right, ridx, rcount);
  	default:
  		throw InternalException("Invalid type for distinct comparison");
  	}
  }
  
  template <class OP>
  static bool TemplatedOptimumStruct(Vector &left, idx_t lidx, idx_t lcount, Vector &right, idx_t ridx, idx_t rcount) {
  	// STRUCT dictionaries apply to all the children
  	// so map the indexes first
  	VectorData lvdata, rvdata;
  	left.Orrify(lcount, lvdata);
  	right.Orrify(rcount, rvdata);
  
  	lidx = lvdata.sel->get_index(lidx);
  	ridx = rvdata.sel->get_index(ridx);
  
  	// DISTINCT semantics are in effect for nested types
  	auto lnull = !lvdata.validity.RowIsValid(lidx);
  	auto rnull = !rvdata.validity.RowIsValid(ridx);
  	if (lnull || rnull) {
  		return OP::Operation(0, 0, lnull, rnull);
  	}
  
  	auto &lchildren = StructVector::GetEntries(left);
  	auto &rchildren = StructVector::GetEntries(right);
  
  	D_ASSERT(lchildren.size() == rchildren.size());
  	for (idx_t col_no = 0; col_no < lchildren.size(); ++col_no) {
  		auto &lchild = *lchildren[col_no];
  		auto &rchild = *rchildren[col_no];
  
  		// Strict comparisons use the OP for definite
  		if (TemplatedOptimumValue<OP>(lchild, lidx, lcount, rchild, ridx, rcount)) {
  			return true;
  		}
  
  		if (col_no == lchildren.size() - 1) {
  			break;
  		}
  
  		// Strict comparisons use IS NOT DISTINCT for possible
  		if (!TemplatedOptimumValue<NotDistinctFrom>(lchild, lidx, lcount, rchild, ridx, rcount)) {
  			return false;
  		}
  	}
  
  	return false;
  }
  
  template <class OP>
  static bool TemplatedOptimumList(Vector &left, idx_t lidx, idx_t lcount, Vector &right, idx_t ridx, idx_t rcount) {
  	VectorData lvdata, rvdata;
  	left.Orrify(lcount, lvdata);
  	right.Orrify(rcount, rvdata);
  
  	// Update the indexes and vector sizes for recursion.
  	lidx = lvdata.sel->get_index(lidx);
  	ridx = rvdata.sel->get_index(ridx);
  
  	lcount = ListVector::GetListSize(left);
  	rcount = ListVector::GetListSize(left);
  
  	// DISTINCT semantics are in effect for nested types
  	auto lnull = !lvdata.validity.RowIsValid(lidx);
  	auto rnull = !rvdata.validity.RowIsValid(ridx);
  	if (lnull || rnull) {
  		return OP::Operation(0, 0, lnull, rnull);
  	}
  
  	auto &lchild = ListVector::GetEntry(left);
  	auto &rchild = ListVector::GetEntry(right);
  
  	auto ldata = (const list_entry_t *)lvdata.data;
  	auto rdata = (const list_entry_t *)rvdata.data;
  
  	auto &lval = ldata[lidx];
  	auto &rval = rdata[ridx];
  
  	for (idx_t pos = 0;; ++pos) {
  		// Tie-breaking uses the OP
  		if (pos == lval.length || pos == rval.length) {
  			return OP::Operation(lval.length, rval.length, false, false);
  		}
  
  		// Strict comparisons use the OP for definite
  		lidx = lval.offset + pos;
  		ridx = rval.offset + pos;
  		if (TemplatedOptimumValue<OP>(lchild, lidx, lcount, rchild, ridx, rcount)) {
  			return true;
  		}
  
  		// Strict comparisons use IS NOT DISTINCT for possible
  		if (!TemplatedOptimumValue<NotDistinctFrom>(lchild, lidx, lcount, rchild, ridx, rcount)) {
  			return false;
  		}
  	}
  
  	return false;
  }
  
  struct VectorMinMaxState {
  	Vector *value;
  };
  
  struct VectorMinMaxBase {
  	static bool IgnoreNull() {
  		return true;
  	}
  
  	template <class STATE>
  	static void Initialize(STATE *state) {
  		state->value = nullptr;
  	}
  
  	template <class STATE>
  	static void Destroy(STATE *state) {
  		if (state->value) {
  			delete state->value;
  		}
  		state->value = nullptr;
  	}
  
  	template <class STATE>
  	static void Assign(STATE *state, Vector &input, const idx_t idx) {
  		if (!state->value) {
  			state->value = new Vector(input.GetType());
  			state->value->SetVectorType(VectorType::CONSTANT_VECTOR);
  		}
  		sel_t selv = idx;
  		SelectionVector sel(&selv);
  		VectorOperations::Copy(input, *state->value, sel, 1, 0, 0);
  	}
  
  	template <class STATE>
  	static void Execute(STATE *state, Vector &input, const idx_t idx, const idx_t count) {
  		Assign(state, input, idx);
  	}
  
  	template <class STATE, class OP>
  	static void Update(Vector inputs[], FunctionData *, idx_t input_count, Vector &state_vector, idx_t count) {
  		auto &input = inputs[0];
  		VectorData idata;
  		input.Orrify(count, idata);
  
  		VectorData sdata;
  		state_vector.Orrify(count, sdata);
  
  		auto states = (STATE **)sdata.data;
  		for (idx_t i = 0; i < count; i++) {
  			const auto idx = idata.sel->get_index(i);
  			if (!idata.validity.RowIsValid(idx)) {
  				continue;
  			}
  			const auto sidx = sdata.sel->get_index(i);
  			auto state = states[sidx];
  			if (!state->value) {
  				Assign(state, input, idx);
  			} else {
  				OP::template Execute(state, input, idx, count);
  			}
  		}
  	}
  
  	template <class STATE, class OP>
  	static void Combine(const STATE &source, STATE *target) {
  		if (!source.value) {
  			return;
  		} else if (!target->value) {
  			Assign(target, *source.value, 0);
  		} else {
  			OP::template Execute(target, *source.value, 0, 1);
  		}
  	}
  
  	template <class T, class STATE>
  	static void Finalize(Vector &result, FunctionData *, STATE *state, T *target, ValidityMask &mask, idx_t idx) {
  		if (!state->value) {
  			// we need to use FlatVector::SetNull here
  			// since for STRUCT columns only setting the validity mask of the struct is incorrect
  			// as for a struct column, we need to also set ALL child columns to NULL
  			FlatVector::SetNull(result, idx, true);
  		} else {
  			VectorOperations::Copy(*state->value, result, 1, 0, idx);
  		}
  	}
  
  	static unique_ptr<FunctionData> Bind(ClientContext &context, AggregateFunction &function,
  	                                     vector<unique_ptr<Expression>> &arguments) {
  		function.arguments[0] = arguments[0]->return_type;
  		function.return_type = arguments[0]->return_type;
  		return nullptr;
  	}
  };
  
  struct MinOperationVector : public VectorMinMaxBase {
  	template <class STATE>
  	static void Execute(STATE *state, Vector &input, const idx_t idx, const idx_t count) {
  		if (TemplatedOptimumValue<DistinctLessThan>(input, idx, count, *state->value, 0, 1)) {
  			Assign(state, input, idx);
  		}
  	}
  };
  
  struct MaxOperationVector : public VectorMinMaxBase {
  	template <class STATE>
  	static void Execute(STATE *state, Vector &input, const idx_t idx, const idx_t count) {
  		if (TemplatedOptimumValue<DistinctGreaterThan>(input, idx, count, *state->value, 0, 1)) {
  			Assign(state, input, idx);
  		}
  	}
  };
  
  template <class OP>
  unique_ptr<FunctionData> BindDecimalMinMax(ClientContext &context, AggregateFunction &function,
                                             vector<unique_ptr<Expression>> &arguments) {
  	auto decimal_type = arguments[0]->return_type;
  	switch (decimal_type.InternalType()) {
  	case PhysicalType::INT16:
  		function = GetUnaryAggregate<OP>(LogicalType::SMALLINT);
  		break;
  	case PhysicalType::INT32:
  		function = GetUnaryAggregate<OP>(LogicalType::INTEGER);
  		break;
  	case PhysicalType::INT64:
  		function = GetUnaryAggregate<OP>(LogicalType::BIGINT);
  		break;
  	default:
  		function = GetUnaryAggregate<OP>(LogicalType::HUGEINT);
  		break;
  	}
  	function.arguments[0] = decimal_type;
  	function.return_type = decimal_type;
  	return nullptr;
  }
  
  template <typename OP, typename STATE>
  static AggregateFunction GetMinMaxFunction(const LogicalType &type) {
  	return AggregateFunction({type}, type, AggregateFunction::StateSize<STATE>,
  	                         AggregateFunction::StateInitialize<STATE, OP>, OP::template Update<STATE, OP>,
  	                         AggregateFunction::StateCombine<STATE, OP>,
  	                         AggregateFunction::StateFinalize<STATE, void, OP>, nullptr, OP::Bind,
  	                         AggregateFunction::StateDestroy<STATE, OP>);
  }
  
  template <class OP, class OP_STRING, class OP_VECTOR>
  static void AddMinMaxOperator(AggregateFunctionSet &set) {
  	for (auto &type : LogicalType::AllTypes()) {
  		if (type.id() == LogicalTypeId::VARCHAR || type.id() == LogicalTypeId::BLOB) {
  			set.AddFunction(
  			    AggregateFunction::UnaryAggregateDestructor<MinMaxState<string_t>, string_t, string_t, OP_STRING>(
  			        type.id(), type.id()));
  		} else if (type.id() == LogicalTypeId::DECIMAL) {
  			set.AddFunction(AggregateFunction({type}, type, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr,
  			                                  BindDecimalMinMax<OP>));
  		} else if (type.id() == LogicalTypeId::LIST || type.id() == LogicalTypeId::MAP ||
  		           type.id() == LogicalTypeId::STRUCT) {
  			set.AddFunction(GetMinMaxFunction<OP_VECTOR, VectorMinMaxState>(type));
  
  		} else {
  			set.AddFunction(GetUnaryAggregate<OP>(type));
  		}
  	}
  }
  
  void MinFun::RegisterFunction(BuiltinFunctions &set) {
  	AggregateFunctionSet min("min");
  	AddMinMaxOperator<MinOperation, MinOperationString, MinOperationVector>(min);
  	set.AddFunction(min);
  }
  
  void MaxFun::RegisterFunction(BuiltinFunctions &set) {
  	AggregateFunctionSet max("max");
  	AddMinMaxOperator<MaxOperation, MaxOperationString, MaxOperationVector>(max);
  	set.AddFunction(max);
  }
  
  } // namespace duckdb
  