class SortLocalSinkState;
class SortGlobalSinkState;
class SortLocalSourceState;
class SortGlobalSourceState;

struct GlobalSortState
struct LocalSortState

SELECT
    l_returnflag,
    l_linestatus,
    sum(l_quantity) AS sum_qty,
    sum(l_extendedprice) AS sum_base_price,
    sum(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
    sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
    avg(l_quantity) AS avg_qty,
    avg(l_extendedprice) AS avg_price,
    avg(l_discount) AS avg_disc,
    count(*) AS count_order
FROM
    lineitem
WHERE
    l_shipdate <= CAST('1998-09-02' AS date)
GROUP BY
    l_returnflag,
    l_linestatus
ORDER BY
    l_returnflag,
    l_linestatus;

case LogicalOperatorType::LOGICAL_ORDER_BY:
    return CreatePlan(op.Cast<LogicalOrder>());

enum class SortKeyType : uint8_t {
  INVALID = 0,
  //! Without payload
  NO_PAYLOAD_FIXED_8 = 1,
  NO_PAYLOAD_FIXED_16 = 2,
  NO_PAYLOAD_FIXED_24 = 3,
  NO_PAYLOAD_FIXED_32 = 4,
  NO_PAYLOAD_VARIABLE_32 = 5,
  //! With payload (requires row pointer in key)
  PAYLOAD_FIXED_16 = 6,
  PAYLOAD_FIXED_24 = 7,
  PAYLOAD_FIXED_32 = 8,
  PAYLOAD_VARIABLE_32 = 9,
};

decode_sort_key->return_type = LogicalType::STRUCT(std::move(decode_child_list));

  vector<LogicalType> payload_types;
  for (idx_t output_col_idx = 0; output_col_idx < projection_map.size(); output_col_idx++) {
    const auto &input_col_idx = projection_map[output_col_idx];
    const auto it = input_column_to_key.find(input_col_idx);
    if (it != input_column_to_key.end()) {
      // Projected column also appears as a key, just reference it
      output_projection_columns.push_back({false, it->second, output_col_idx});
    } else {
      // Projected column does not appear as a key, add to payload layout
      output_projection_columns.push_back({true, payload_types.size(), output_col_idx});
      payload_types.push_back(input_types[input_col_idx]);
      input_projection_map.push_back(input_col_idx);
    }
  }
  payload_layout->Initialize(payload_types, TupleDataValidityType::CAN_HAVE_NULL_VALUES);

  std::sort(output_projection_columns.begin(), output_projection_columns.end(),
            [](const SortProjectionColumn &lhs, const SortProjectionColumn &rhs) {
              if (lhs.is_payload == rhs.is_payload) {
                return lhs.layout_col_idx < rhs.layout_col_idx;
              }
              return lhs.is_payload < rhs.is_payload;
            });

Sort::Sort(ClientContext &context, const vector<BoundOrderByNode> &orders, const vector<LogicalType> &input_types,
           vector<idx_t> projection_map, bool is_index_sort_p)
    : key_layout(make_shared_ptr<TupleDataLayout>()), payload_layout(make_shared_ptr<TupleDataLayout>()),
      is_index_sort(is_index_sort_p) {
  // Convert orders to a single "create_sort_key" expression (and corresponding "decode_sort_key")
  FunctionBinder binder(context);
  vector<unique_ptr<Expression>> create_children;
  vector<unique_ptr<Expression>> decode_children;
  child_list_t<LogicalType> decode_child_list;
  for (idx_t col_idx = 0; col_idx < orders.size(); col_idx++) {
    const auto &order = orders[col_idx];
    // Create: for each column we have two arguments: 1. the column, 2. sort specifier
    create_children.emplace_back(order.expression->Copy());
    create_children.emplace_back(make_uniq<BoundConstantExpression>(Value(order.GetOrderModifier())));
    // Avoid having unnamed structs fields (otherwise we get a parser exception while binding)
    const auto col_name = StringUtil::Format("c%llu", col_idx);
    auto col_type = order.expression->return_type;
    decode_child_list.emplace_back(col_name, col_type);
    col_type = TypeVisitor::VisitReplace(col_type, [](const LogicalType &type) {
      if (type.id() != LogicalTypeId::STRUCT) {
        return type;
      }
      child_list_t<LogicalType> internal_child_list;
      for (const auto &child : StructType::GetChildTypes(type)) {
        internal_child_list.emplace_back(StringUtil::Format("c%llu", internal_child_list.size()), child.second);
      }
      return LogicalType::STRUCT(std::move(internal_child_list));
    });
    // Decode: for each column we have two arguments: 1. col name + type, 2. sort specifier
    decode_children.emplace_back(make_uniq<BoundConstantExpression>(Value(col_name + " " + col_type.ToString())));
    decode_children.emplace_back(make_uniq<BoundConstantExpression>(order.GetOrderModifier()));
  }
  ErrorData error;
  create_sort_key = binder.BindScalarFunction(DEFAULT_SCHEMA, "create_sort_key", std::move(create_children), error);
  if (!create_sort_key) {
    throw InternalException("Unable to bind create_sort_key in Sort::Sort");
  }
  switch (create_sort_key->return_type.id()) {
  case LogicalTypeId::BIGINT:
    decode_children.insert(decode_children.begin(),
                           make_uniq<BoundReferenceExpression>(LogicalType::BIGINT, static_cast<storage_t>(0)));
    break;
  default:
    D_ASSERT(create_sort_key->return_type.id() == LogicalTypeId::BLOB);
    decode_children.insert(decode_children.begin(),
                           make_uniq<BoundReferenceExpression>(LogicalType::BLOB, static_cast<storage_t>(0)));
  }
  decode_sort_key = binder.BindScalarFunction(DecodeSortKeyFun::GetFunction(), std::move(decode_children));
  if (!decode_sort_key) {
    throw InternalException("Unable to bind decode_sort_key in Sort::Sort");
  }
  // A bit hacky, but this way we make sure that the output does contain the unnamed structs again
  decode_sort_key->return_type = LogicalType::STRUCT(std::move(decode_child_list));
  // For convenience, we fill the projection map if it is empty
  if (projection_map.empty()) {
    projection_map.reserve(input_types.size());
    for (idx_t col_idx = 0; col_idx < input_types.size(); col_idx++) {
      projection_map.push_back(col_idx);
    }
  }
  // We need to output this many columns, reserve
  output_projection_columns.reserve(projection_map.size());
  // Create mapping from input column to key (so we won't duplicate columns in key/payload)
  unordered_map<idx_t, idx_t> input_column_to_key;
  for (idx_t key_idx = 0; key_idx < orders.size(); key_idx++) {
    const auto &key_order_expr = *orders[key_idx].expression;
    if (key_order_expr.GetExpressionClass() == ExpressionClass::BOUND_REF) {
      input_column_to_key.emplace(key_order_expr.Cast<BoundReferenceExpression>().index, key_idx);
    }
  }
  // Construct payload layout (excluding columns that also appear as key)
  vector<LogicalType> payload_types;
  for (idx_t output_col_idx = 0; output_col_idx < projection_map.size(); output_col_idx++) {
    const auto &input_col_idx = projection_map[output_col_idx];
    const auto it = input_column_to_key.find(input_col_idx);
    if (it != input_column_to_key.end()) {
      // Projected column also appears as a key, just reference it
      output_projection_columns.push_back({false, it->second, output_col_idx});
    } else {
      // Projected column does not appear as a key, add to payload layout
      output_projection_columns.push_back({true, payload_types.size(), output_col_idx});
      payload_types.push_back(input_types[input_col_idx]);
      input_projection_map.push_back(input_col_idx);
    }
  }
  payload_layout->Initialize(payload_types, TupleDataValidityType::CAN_HAVE_NULL_VALUES);
  // Sort the output projection columns so we're gathering the columns in order
  std::sort(output_projection_columns.begin(), output_projection_columns.end(),
            [](const SortProjectionColumn &lhs, const SortProjectionColumn &rhs) {
              if (lhs.is_payload == rhs.is_payload) {
                return lhs.layout_col_idx < rhs.layout_col_idx;
              }
              return lhs.is_payload < rhs.is_payload;
            });
  // Finally, initialize the key layout (now that we know whether we have a payload)
  key_layout->Initialize(orders, create_sort_key->return_type, !payload_types.empty());
}

key_layout->Initialize(orders, create_sort_key->return_type, !payload_types.empty());

if (TypeIsConstantSize(physical_type)) {
    // NULL byte + fixed-width type
    sort_width += 1 + GetTypeIdSize(physical_type);
} else if (logical_type == LogicalType::VARCHAR && order.stats &&
           StringStats::HasMaxStringLength(*order.stats)) {
    // NULL byte + maximum string length + string delimiter
    sort_width += 1 + StringStats::MaxStringLength(*order.stats) + 1;
} else {
    // We don't know how long the key will be
    sort_width = DConstants::INVALID_INDEX;
    break;
}

idx_t temp_row_width = type.id() == LogicalTypeId::BIGINT ? 8 : sort_width;
  if (sort_width != DConstants::INVALID_INDEX && has_payload) {
    temp_row_width += 8;
  }
  if (temp_row_width <= 8) {
    D_ASSERT(!has_payload);
    row_width = 8;
    sort_key_type = SortKeyType::NO_PAYLOAD_FIXED_8;
  } else if (temp_row_width <= 16) {
    row_width = 16;
    sort_key_type = has_payload ? SortKeyType::PAYLOAD_FIXED_16 : SortKeyType::NO_PAYLOAD_FIXED_16;
  } else if (temp_row_width <= 24) {
    row_width = 24;
    sort_key_type = has_payload ? SortKeyType::PAYLOAD_FIXED_24 : SortKeyType::NO_PAYLOAD_FIXED_24;
  } else if (temp_row_width <= 32) {
    row_width = 32;
    sort_key_type = has_payload ? SortKeyType::PAYLOAD_FIXED_32 : SortKeyType::NO_PAYLOAD_FIXED_32;
  } else {
    row_width = 32;
    sort_key_type = has_payload ? SortKeyType::PAYLOAD_VARIABLE_32 : SortKeyType::NO_PAYLOAD_VARIABLE_32;
    // Variable-size sort key, also set these properties
    all_constant = false;
    heap_size_offset = has_payload ? SortKey<SortKeyType::PAYLOAD_VARIABLE_32>::HEAP_SIZE_OFFSET
                                   : SortKey<SortKeyType::NO_PAYLOAD_VARIABLE_32>::HEAP_SIZE_OFFSET;
  }

void SortedRun::Sink(DataChunk &key, DataChunk &payload) {
  D_ASSERT(!finalized);
  key_data->Append(key_append_state, key);
  if (payload_data) {
    D_ASSERT(key.size() == payload.size());
    payload_data->Append(payload_append_state, payload);
    SetPayloadPointer(key_append_state.chunk_state.row_locations, payload_append_state.chunk_state.row_locations,
                      key.size(), key_data->GetLayout().GetSortKeyType());
  }
}

static void SetPayloadPointer(Vector &key_locations, Vector &payload_locations, const idx_t count,
                              const SortKeyType &sort_key_type) {
  switch (sort_key_type) {
  case SortKeyType::PAYLOAD_FIXED_16:
    return TemplatedSetPayloadPointer<SortKeyType::PAYLOAD_FIXED_16>(key_locations, payload_locations, count);
  case SortKeyType::PAYLOAD_FIXED_24:
    return TemplatedSetPayloadPointer<SortKeyType::PAYLOAD_FIXED_24>(key_locations, payload_locations, count);
  case SortKeyType::PAYLOAD_FIXED_32:
    return TemplatedSetPayloadPointer<SortKeyType::PAYLOAD_FIXED_32>(key_locations, payload_locations, count);
  case SortKeyType::PAYLOAD_VARIABLE_32:
    return TemplatedSetPayloadPointer<SortKeyType::PAYLOAD_VARIABLE_32>(key_locations, payload_locations, count);
  default:
    throw NotImplementedException("SetPayloadPointer for %s", EnumUtil::ToString(sort_key_type));
  }
}

template <SortKeyType SORT_KEY_TYPE>
static void TemplatedSetPayloadPointer(Vector &key_locations, Vector &payload_locations, const idx_t count) {
  using SORT_KEY = SortKey<SORT_KEY_TYPE>;
  const auto key_locations_ptr = FlatVector::GetData<SORT_KEY *>(key_locations);
  const auto payload_locations_ptr = FlatVector::GetData<data_ptr_t>(payload_locations);
  for (idx_t i = 0; i < count; i++) {
    key_locations_ptr[i]->SetPayload(payload_locations_ptr[i]);
  }
}

static void SortSwitch(const TupleDataCollection &key_data, bool is_index_sort) {
  const auto sort_key_type = key_data.GetLayout().GetSortKeyType();
  switch (sort_key_type) {
  case SortKeyType::NO_PAYLOAD_FIXED_8:
    return TemplatedSort<SortKeyType::NO_PAYLOAD_FIXED_8>(key_data, is_index_sort);
  case SortKeyType::NO_PAYLOAD_FIXED_16:
    return TemplatedSort<SortKeyType::NO_PAYLOAD_FIXED_16>(key_data, is_index_sort);
  case SortKeyType::NO_PAYLOAD_FIXED_24:
    return TemplatedSort<SortKeyType::NO_PAYLOAD_FIXED_24>(key_data, is_index_sort);
  case SortKeyType::NO_PAYLOAD_FIXED_32:
    return TemplatedSort<SortKeyType::NO_PAYLOAD_FIXED_32>(key_data, is_index_sort);
  case SortKeyType::NO_PAYLOAD_VARIABLE_32:
    return TemplatedSort<SortKeyType::NO_PAYLOAD_VARIABLE_32>(key_data, is_index_sort);
  case SortKeyType::PAYLOAD_FIXED_16:
    return TemplatedSort<SortKeyType::PAYLOAD_FIXED_16>(key_data, is_index_sort);
  case SortKeyType::PAYLOAD_FIXED_24:
    return TemplatedSort<SortKeyType::PAYLOAD_FIXED_24>(key_data, is_index_sort);
  case SortKeyType::PAYLOAD_FIXED_32:
    return TemplatedSort<SortKeyType::PAYLOAD_FIXED_32>(key_data, is_index_sort);
  case SortKeyType::PAYLOAD_VARIABLE_32:
    return TemplatedSort<SortKeyType::PAYLOAD_VARIABLE_32>(key_data, is_index_sort);
  default:
    throw NotImplementedException("TemplatedSort for %s", EnumUtil::ToString(sort_key_type));
  }
}

const auto ska_sort_width = MinValue<idx_t>(layout.GetSortWidth(), sizeof(uint64_t));

  auto begin = BLOCK_ITERATOR(state, 0);
  auto end = BLOCK_ITERATOR(state, key_data.Count());
  const auto requires_next_sort =
      is_index_sort ? false : !SORT_KEY::CONSTANT_SIZE || SORT_KEY::INLINE_LENGTH != sizeof(uint64_t);
  const auto ska_sort_width = MinValue<idx_t>(layout.GetSortWidth(), sizeof(uint64_t));
  const auto &sort_skippable_bytes = layout.GetSortSkippableBytes();
  auto ska_extract_key =
      SkaExtractKey<SORT_KEY>(requires_next_sort, ska_sort_width, sort_skippable_bytes, context.interrupted);
  const auto fallback = [ska_extract_key](const BLOCK_ITERATOR &fb_begin, const BLOCK_ITERATOR &fb_end) {
    duckdb_ska_sort::ska_sort(fb_begin, fb_end, ska_extract_key);
  };
  duckdb_vergesort::vergesort(begin, end, std::less<SORT_KEY>(), fallback);

if (dist < 128) {
    // Vergesort is inefficient for small collections
    fallback(first, last);
    return;
}

template<typename It, typename ExtractKey>
static void ska_sort(It begin, It end, ExtractKey && extract_key)
{
    detail::inplace_radix_sort<128, 1024>(begin, end, extract_key);
}

template<typename It, typename ExtractKey>
inline void StdSortFallback(It begin, It end, ExtractKey & extract_key)
{
  // LNK note that we use the full comparison (not just extracted key) here
    static const auto comp = [&](const typename std::remove_reference<decltype(*begin)>::type & l, const typename std::remove_reference<decltype(*begin)>::type & r){ return l < r; };
  static const auto fallback = [&](const It &fb_begin, const It &fb_end) {
    duckdb_pdqsort::pdqsort_branchless(fb_begin, fb_end, comp);
  };
  duckdb_vergesort::vergesort(begin, end, comp, fallback);
}

template<class Iter, class Compare>
inline void pdqsort_branchless(Iter begin, Iter end, Compare comp) {
    if (begin == end) return;
    pdqsort_detail::pdqsort_loop<Iter, Compare, true>(
        begin, end, comp, pdqsort_detail::log2(end - begin));
}

if (size < insertion_sort_threshold) {
    if (leftmost) insertion_sort(begin, end, comp);
    else unguarded_insertion_sort(begin, end, comp);
    return;
}

    template<class Iter, class Compare>
    inline void insertion_sort(Iter begin, Iter end, Compare comp) {
        typedef typename std::iterator_traits<Iter>::value_type T;
        if (begin == end) return;
        for (Iter cur = begin + 1; cur != end; ++cur) {
            Iter sift = cur;
            Iter sift_1 = cur - 1;
            // Compare first so we can avoid 2 moves for an element already positioned correctly.
            if (comp(*sift, *sift_1)) {
                T tmp = PDQSORT_PREFER_MOVE(*sift);
                do { *sift-- = PDQSORT_PREFER_MOVE(*sift_1); }
                while (sift != begin && comp(tmp, *--sift_1));
                *sift = PDQSORT_PREFER_MOVE(tmp);
            }
        }
    }

if (count <= SortConstants::INSERTION_SORT_THRESHOLD) {
    return InsertionSort(dataptr, nullptr, count, col_offset, sort_layout.entry_size, sorting_size, 0, false);
  }

inline void InsertionSort(const data_ptr_t orig_ptr, const data_ptr_t temp_ptr, const idx_t &count,
                          const idx_t &col_offset, const idx_t &row_width, const idx_t &total_comp_width,
                          const idx_t &offset, bool swap) {
  const data_ptr_t source_ptr = swap ? temp_ptr : orig_ptr;
  const data_ptr_t target_ptr = swap ? orig_ptr : temp_ptr;
  if (count > 1) {
    const idx_t total_offset = col_offset + offset;
    auto temp_val = make_unsafe_uniq_array_uninitialized<data_t>(row_width);
    const data_ptr_t val = temp_val.get();
    const auto comp_width = total_comp_width - offset;
    for (idx_t i = 1; i < count; i++) {
      FastMemcpy(val, source_ptr + i * row_width, row_width);
      idx_t j = i;
      while (j > 0 &&
             FastMemcmp(source_ptr + (j - 1) * row_width + total_offset, val + total_offset, comp_width) > 0) {
        FastMemcpy(source_ptr + j * row_width, source_ptr + (j - 1) * row_width, row_width);
        j--;
      }
      FastMemcpy(source_ptr + j * row_width, val, row_width);
    }
  }
  if (swap) {
    memcpy(target_ptr, source_ptr, count * row_width);
  }
}

void RadixSort(BufferManager &buffer_manager, const data_ptr_t &dataptr, const idx_t &count, const idx_t &col_offset,
               const idx_t &sorting_size, const SortLayout &sort_layout, bool contains_string) {
  if (contains_string) {
    auto begin = duckdb_pdqsort::PDQIterator(dataptr, sort_layout.entry_size);
    auto end = begin + count;
    duckdb_pdqsort::PDQConstants constants(sort_layout.entry_size, col_offset, sorting_size, *end);
    return duckdb_pdqsort::pdqsort_branchless(begin, begin + count, constants);
  }
  if (count <= SortConstants::INSERTION_SORT_THRESHOLD) {
    return InsertionSort(dataptr, nullptr, count, col_offset, sort_layout.entry_size, sorting_size, 0, false);
  }
  if (sorting_size <= SortConstants::MSD_RADIX_SORT_SIZE_THRESHOLD) {
    return RadixSortLSD(buffer_manager, dataptr, count, col_offset, sort_layout.entry_size, sorting_size);
  }
  const auto block_size = buffer_manager.GetBlockSize();
  auto temp_block =
      buffer_manager.Allocate(MemoryTag::ORDER_BY, MaxValue(count * sort_layout.entry_size, block_size));
  auto pre_allocated_array =
      make_unsafe_uniq_array_uninitialized<idx_t>(sorting_size * SortConstants::MSD_RADIX_LOCATIONS);
  RadixSortMSD(dataptr, temp_block.Ptr(), count, col_offset, sort_layout.entry_size, sorting_size, 0,
               pre_allocated_array.get(), false);
}