| 2105 | namespace |
| 2106 | { |
| 2107 | void NtileState::windowInsertResultInto( |
| 2108 | const WindowTransform * transform, |
| 2109 | size_t function_index, |
| 2110 | const DataTypes & argument_types) |
| 2111 | { |
| 2112 | if (!buckets) [[unlikely]] |
| 2113 | { |
| 2114 | const auto & current_block = transform->blockAt(transform->current_row); |
| 2115 | const auto & workspace = transform->workspaces[function_index]; |
| 2116 | const auto & arg_col = *current_block.original_input_columns[workspace.argument_column_indices[0]]; |
| 2117 | if (!isColumnConst(arg_col)) |
| 2118 | throw Exception(ErrorCodes::BAD_ARGUMENTS, "Argument of 'ntile' function must be a constant"); |
| 2119 | auto type_id = argument_types[0]->getTypeId(); |
| 2120 | if (type_id == TypeIndex::UInt8) |
| 2121 | buckets = arg_col[transform->current_row.row].safeGet<UInt8>(); |
| 2122 | else if (type_id == TypeIndex::UInt16) |
| 2123 | buckets = arg_col[transform->current_row.row].safeGet<UInt16>(); |
| 2124 | else if (type_id == TypeIndex::UInt32) |
| 2125 | buckets = arg_col[transform->current_row.row].safeGet<UInt32>(); |
| 2126 | else if (type_id == TypeIndex::UInt64) |
| 2127 | buckets = arg_col[transform->current_row.row].safeGet<UInt64>(); |
| 2128 | |
| 2129 | if (!buckets) |
| 2130 | { |
| 2131 | throw Exception(ErrorCodes::BAD_ARGUMENTS, "Argument of 'ntile' function must be greater than zero"); |
| 2132 | } |
| 2133 | } |
| 2134 | // new partition |
| 2135 | if (WindowFunctionHelpers::checkPartitionEnterFirstRow(transform)) [[unlikely]] |
| 2136 | { |
| 2137 | current_partition_rows = 0; |
| 2138 | current_partition_inserted_row = 0; |
| 2139 | start_row = transform->current_row; |
| 2140 | } |
| 2141 | current_partition_rows++; |
| 2142 | |
| 2143 | // Only do the action when we meet the last row in this partition. |
| 2144 | if (!WindowFunctionHelpers::checkPartitionEnterLastRow(transform)) |
| 2145 | return; |
| 2146 | |
| 2147 | auto bucket_capacity = current_partition_rows / buckets; |
| 2148 | auto capacity_diff = current_partition_rows - bucket_capacity * buckets; |
| 2149 | |
| 2150 | // bucket number starts from 1. |
| 2151 | UInt64 bucket_num = 1; |
| 2152 | while (current_partition_inserted_row < current_partition_rows) |
| 2153 | { |
| 2154 | auto current_bucket_capacity = bucket_capacity; |
| 2155 | if (capacity_diff > 0) |
| 2156 | { |
| 2157 | current_bucket_capacity += 1; |
| 2158 | capacity_diff--; |
| 2159 | } |
| 2160 | auto left_rows = current_bucket_capacity; |
| 2161 | while (left_rows) |
| 2162 | { |
| 2163 | auto available_block_rows = transform->blockRowsNumber(start_row) - start_row.row; |
| 2164 | IColumn & to = *transform->blockAt(start_row).output_columns[function_index]; |
no test coverage detected