diff --git a/cpp/src/io/parquet/decode_fixed.cu b/cpp/src/io/parquet/decode_fixed.cu index 5e3122d9fa37..f1f5ecd0a3f3 100644 --- a/cpp/src/io/parquet/decode_fixed.cu +++ b/cpp/src/io/parquet/decode_fixed.cu @@ -89,13 +89,13 @@ __device__ void decode_fixed_width_values( constexpr int max_batch_size = num_warps * cudf::detail::warp_size; // nesting level that is storing actual leaf values - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; auto const data_out = s->nesting_info[leaf_level_index].data_out; - Type const dtype = s->col.physical_type; + Type const dtype = s->setup.col.physical_type; uint32_t const dtype_len = s->dtype_len; - int const skipped_leaf_values = s->page.skipped_leaf_values; + int const skipped_leaf_values = s->setup.page.skipped_leaf_values; // decode values int thread_pos = start + t; @@ -103,10 +103,10 @@ __device__ void decode_fixed_width_values( // Index from value buffer (doesn't include nulls) to final array (has gaps for nulls) int const dst_pos = [&]() { if constexpr (copy_mode_t == copy_mode::DIRECT) { - return thread_pos - s->first_row; + return thread_pos - s->setup.first_row; } else { int dst_pos = sb->nz_idx[rolling_index(thread_pos)]; - if constexpr (!has_lists_t) { dst_pos -= s->first_row; } + if constexpr (!has_lists_t) { dst_pos -= s->setup.first_row; } return dst_pos; } }(); @@ -124,7 +124,8 @@ __device__ void decode_fixed_width_values( void* const dst = data_out + (static_cast(dst_pos) * dtype_len); - if (s->col.logical_type.has_value() && s->col.logical_type->type == LogicalType::DECIMAL) { + if (s->setup.col.logical_type.has_value() && + s->setup.col.logical_type->type == LogicalType::DECIMAL) { switch (dtype) { case Type::INT32: read_fixed_width_value_fast(s, sb, src_pos, static_cast(dst)); @@ -179,10 +180,10 @@ __device__ inline void decode_fixed_width_split_values( constexpr int max_batch_size = num_warps * warp_size; // nesting level that is storing actual leaf values - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; auto const data_out = s->nesting_info[leaf_level_index].data_out; - Type const dtype = s->col.physical_type; + Type const dtype = s->setup.col.physical_type; auto const data_len = cuda::std::distance(s->data_start, s->data_end); // Check malformed BYTE_STREAM_SPLIT pages @@ -193,7 +194,7 @@ __device__ inline void decode_fixed_width_split_values( auto const num_values = data_len / s->dtype_len_in; - int const skipped_leaf_values = s->page.skipped_leaf_values; + int const skipped_leaf_values = s->setup.page.skipped_leaf_values; // decode values int thread_pos = start + t; @@ -201,10 +202,10 @@ __device__ inline void decode_fixed_width_split_values( // Index from value buffer (doesn't include nulls) to final array (has gaps for nulls) int const dst_pos = [&]() { if constexpr (copy_mode_t == copy_mode::DIRECT) { - return thread_pos - s->first_row; + return thread_pos - s->setup.first_row; } else { int dst_pos = sb->nz_idx[rolling_index(thread_pos)]; - if constexpr (!has_lists_t) { dst_pos -= s->first_row; } + if constexpr (!has_lists_t) { dst_pos -= s->setup.first_row; } return dst_pos; } }(); @@ -224,8 +225,8 @@ __device__ inline void decode_fixed_width_split_values( uint32_t const dtype_len = s->dtype_len; uint8_t const* const src = s->data_start + src_pos; uint8_t* const dst = data_out + static_cast(dst_pos) * dtype_len; - auto const is_decimal = - s->col.logical_type.has_value() and s->col.logical_type->type == LogicalType::DECIMAL; + auto const is_decimal = s->setup.col.logical_type.has_value() and + s->setup.col.logical_type->type == LogicalType::DECIMAL; // Note: non-decimal FIXED_LEN_BYTE_ARRAY will be handled in the string reader if (is_decimal) { @@ -293,7 +294,7 @@ __device__ int skip_validity_and_row_indices_nonlist( int32_t target_value_count, page_state_s* s, level_t const* const def, bool is_nested, int t) { int const max_def_level = - is_nested ? s->nesting_info[s->col.max_nesting_depth - 1].max_def_level : 1; + is_nested ? s->nesting_info[s->setup.col.max_nesting_depth - 1].max_def_level : 1; int max_depth_valid_count = 0; int value_count = 0; @@ -342,11 +343,11 @@ __device__ int update_validity_and_row_indices_nested( int value_count = s->input_value_count; // cap by last row so that we don't process any rows past what we want to output. - int const first_row = s->first_row; - int const last_row = first_row + s->num_rows; + int const first_row = s->setup.first_row; + int const last_row = first_row + s->setup.num_rows; int const capped_target_value_count = min(target_value_count, last_row); - int const max_depth = s->col.max_nesting_depth - 1; + int const max_depth = s->setup.col.max_nesting_depth - 1; auto& max_depth_ni = s->nesting_info[max_depth]; int max_depth_valid_count = max_depth_ni.valid_count; @@ -466,8 +467,8 @@ __device__ int update_validity_and_row_indices_flat( int valid_count = ni.valid_count; // cap by last row so that we don't process any rows past what we want to output. - int const first_row = s->first_row; - int const last_row = first_row + s->num_rows; + int const first_row = s->setup.first_row; + int const last_row = first_row + s->setup.num_rows; int const capped_target_value_count = min(target_value_count, last_row); int const valid_map_offset = ni.valid_map_offset; @@ -581,11 +582,11 @@ __device__ int update_validity_and_row_indices_lists(int32_t target_value_count, int input_row_count = s->input_row_count; // cap by last row so that we don't process any rows past what we want to output. - int const first_row = s->first_row; - int const last_row = first_row + s->num_rows; + int const first_row = s->setup.first_row; + int const last_row = first_row + s->setup.num_rows; int const row_index_lower_bound = s->row_index_lower_bound; - int const max_depth = s->col.max_nesting_depth - 1; + int const max_depth = s->setup.col.max_nesting_depth - 1; int max_depth_valid_count = s->nesting_info[max_depth].valid_count; int const warp_index = t / cudf::detail::warp_size; @@ -862,7 +863,7 @@ __device__ void skip_ahead_in_decoding(page_state_s* s, }; if constexpr (has_lists_t) { - auto const skipped_leaf_values = s->page.skipped_leaf_values; + auto const skipped_leaf_values = s->setup.page.skipped_leaf_values; if (skipped_leaf_values > 0) { processed_count = skipped_leaf_values; if constexpr (has_dict_t) { @@ -875,7 +876,7 @@ __device__ void skip_ahead_in_decoding(page_state_s* s, } // Non-lists - int const first_row = s->first_row; + int const first_row = s->setup.first_row; if (first_row <= 0) { return; } // Nothing to skip // Count the number of valids we're skipping. @@ -892,7 +893,7 @@ __device__ void skip_ahead_in_decoding(page_state_s* s, } if (t == 0) { - int const max_depth = s->col.max_nesting_depth - 1; + int const max_depth = s->setup.col.max_nesting_depth - 1; auto& ni = s->nesting_info[max_depth]; // update valid value count for decoding and total # of values we've processed @@ -1055,8 +1056,12 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) rle_stream dict_stream{dict_runs}; if constexpr (has_dict_t) { - dict_stream.init( - block, s->dict_bits, s->data_start, s->data_end, sb->dict_idx, s->page.num_input_values); + dict_stream.init(block, + s->dict_bits, + s->data_start, + s->data_end, + sb->dict_idx, + s->setup.page.num_input_values); } // Use dictionary stream memory for bools @@ -1065,7 +1070,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) if constexpr (has_bools_t) { if (bools_are_rle_stream) { bool_stream.init( - block, 1, s->data_start, s->data_end, sb->dict_idx, s->page.num_input_values); + block, 1, s->data_start, s->data_end, sb->dict_idx, s->setup.page.num_input_values); } } block.sync(); @@ -1077,10 +1082,11 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) // - valid_count: number of non-null values we have decoded so far. In each iteration of the // loop below, we look at the number of valid items (which could be all for non-nullable), // and valid_count is that running count. - int processed_count = 0; - int valid_count = 0; - size_t string_output_offset = 0; - int const init_valid_map_offset = s->nesting_info[s->col.max_nesting_depth - 1].valid_map_offset; + int processed_count = 0; + int valid_count = 0; + size_t string_output_offset = 0; + int const init_valid_map_offset = + s->nesting_info[s->setup.col.max_nesting_depth - 1].valid_map_offset; // Skip ahead in the decoding so that we don't repeat work skip_ahead_in_decodingfirst_row; - int const last_row = first_row + s->num_rows; - while ((s->error == 0) && (processed_count < s->page.num_input_values) && + int const first_row = s->setup.first_row; + int const last_row = first_row + s->setup.num_rows; + while ((s->setup.error == 0) && (processed_count < s->setup.page.num_input_values) && (s->input_row_count <= last_row)) { int next_valid_count; block.sync(); - processed_count += min(rolling_buf_size, s->page.num_input_values - processed_count); + processed_count += min(rolling_buf_size, s->setup.page.num_input_values - processed_count); // only need to process definition levels if this is a nullable column if (process_nulls) { @@ -1152,7 +1158,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) if (bools_are_rle_stream) { bool_stream.decode_next(t, next_valid_count - valid_count); } else { - auto const target_pos = next_valid_count + s->page.skipped_leaf_values; + auto const target_pos = next_valid_count + s->setup.page.skipped_leaf_values; bool_plain_decode(s, sb, target_pos, block); if (t == 0) { s->dict_pos = target_pos; } } @@ -1162,7 +1168,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) auto decode_values = [&]() { if constexpr (has_strings_t) { uint32_t* const str_offsets = - s->col.column_string_offset_base + page_string_offset_indices[page_idx]; + s->setup.col.column_string_offset_base + page_string_offset_indices[page_idx]; string_output_offset = decode_strings( s, sb, valid_count, next_valid_count, t, str_offsets, string_output_offset); @@ -1194,10 +1200,10 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) uint32_t const dtype_len = has_strings_t ? sizeof(cudf::size_type) : s->dtype_len; int const num_values = [&]() { if constexpr (has_lists_t) { - auto const& ni = s->nesting_info[s->col.max_nesting_depth - 1]; + auto const& ni = s->nesting_info[s->setup.col.max_nesting_depth - 1]; return ni.valid_map_offset - init_valid_map_offset; } else { - return s->num_rows; + return s->setup.num_rows; } }(); zero_fill_null_positions_shared( @@ -1212,13 +1218,13 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) if constexpr (!has_lists_t) { if (!process_nulls) { if (t == 0) { - s->nesting_info[s->col.max_nesting_depth - 1].value_count = s->input_row_count; + s->nesting_info[s->setup.col.max_nesting_depth - 1].value_count = s->input_row_count; } block.sync(); } } - if (s->col.is_large_string_col) { + if (s->setup.col.is_large_string_col) { // page.chunk_idx are ordered by input_col_idx and row_group_idx respectively. auto const chunks_per_rowgroup = initial_str_offsets.size(); auto const input_col_idx = pages[page_idx].chunk_idx % chunks_per_rowgroup; @@ -1228,7 +1234,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) } } - if (t == 0 and s->error != 0) { set_error(s->error, error_code); } + if (t == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } } } // anonymous namespace diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 946861671a07..46270acd1a52 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -55,7 +55,7 @@ __device__ void update_page_sizes(page_state_s* s, cg::thread_block const& block) { // max nesting depth of the column - int const max_depth = s->col.max_nesting_depth; + int const max_depth = s->setup.col.max_nesting_depth; int const t = block.thread_rank(); constexpr int num_warps = preprocess_block_size / cudf::detail::warp_size; constexpr int max_batch_size = num_warps * cudf::detail::warp_size; @@ -74,7 +74,7 @@ __device__ void update_page_sizes(page_state_s* s, // how many leaf values we've processed in the page so far int leaf_count = s->input_leaf_count; // whether or not we need to continue checking for the first row - bool skipped_values_set = s->page.skipped_values >= 0; + bool skipped_values_set = s->setup.page.skipped_values >= 0; while (value_count < target_value_count) { int const batch_size = @@ -107,8 +107,8 @@ __device__ void update_page_sizes(page_state_s* s, // if this thread is in row bounds int const row_index = (thread_row_count + row_count) - 1; - in_row_bounds = - (row_index >= s->row_index_lower_bound) && (row_index < (s->first_row + s->num_rows)); + in_row_bounds = (row_index >= s->row_index_lower_bound) && + (row_index < (s->setup.first_row + s->setup.num_rows)); // if we have not set skipped values yet, see if we found the first in-bounds row if (!skipped_values_set) { @@ -121,8 +121,8 @@ __device__ void update_page_sizes(page_state_s* s, if (global_count > 0) { // this is the thread that represents the first row. if (local_count == 1 && in_row_bounds) { - s->page.skipped_values = value_count + t; - s->page.skipped_leaf_values = + s->setup.page.skipped_values = value_count + t; + s->setup.page.skipped_leaf_values = leaf_count + (is_new_leaf ? thread_leaf_count - 1 : thread_leaf_count); } skipped_values_set = true; @@ -139,7 +139,7 @@ __device__ void update_page_sizes(page_state_s* s, int const count = block_reduce(temp_storage.reduce_storage).Sum(in_nesting_bounds); block.sync(); if (!t) { - PageNestingInfo* pni = &s->page.nesting[s_idx]; + PageNestingInfo* pni = &s->setup.page.nesting[s_idx]; pni->batch_size += count; } } @@ -179,7 +179,7 @@ __device__ void compute_page_sizes_for_pruned_pages(PageInfo* page, if (not has_repetition and max_depth == 1) { if (!block.thread_rank()) { if (is_base_pass) { page->nesting[0].size = page->num_rows; } - page->nesting[0].batch_size = state->num_rows; + page->nesting[0].batch_size = state->setup.num_rows; } return; } @@ -213,12 +213,12 @@ __device__ void compute_page_sizes_for_pruned_pages(PageInfo* page, // Write size information for all depths up to the list depth for (auto depth = warp.thread_rank(); depth < list_depth; depth += warp.size()) { if (is_base_pass) { page->nesting[depth].size = page->num_rows; } - page->nesting[depth].batch_size = state->num_rows; + page->nesting[depth].batch_size = state->setup.num_rows; } // Write size information at the list depth (zero if no list) if (warp.thread_rank() == 0) { if (is_base_pass) { page->nesting[list_depth].size = page->num_rows; } - page->nesting[list_depth].batch_size = state->num_rows; + page->nesting[list_depth].batch_size = state->setup.num_rows; } } } @@ -274,9 +274,9 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) // it directly. if (!has_repetition) { int depth = 0; - while (depth < s->page.num_output_nesting_levels) { + while (depth < s->setup.page.num_output_nesting_levels) { auto const thread_depth = depth + t; - if (thread_depth < s->page.num_output_nesting_levels) { + if (thread_depth < s->setup.page.num_output_nesting_levels) { if (is_base_pass) { pp->nesting[thread_depth].size = pp->num_input_values; } pp->nesting[thread_depth].batch_size = pp->num_input_values; } @@ -288,15 +288,16 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) // in the trim pass, for anything with lists, we only need to fully process bounding pages (those // at the beginning or the end of the row bounds) if (!is_base_pass && - !is_bounds_page(s->page, s->col.start_row, min_row, num_rows, has_repetition)) { + !is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition)) { int depth = 0; - while (depth < s->page.num_output_nesting_levels) { + while (depth < s->setup.page.num_output_nesting_levels) { auto const thread_depth = depth + t; - if (thread_depth < s->page.num_output_nesting_levels) { + if (thread_depth < s->setup.page.num_output_nesting_levels) { // if we are not a bounding page (as checked above) then we are either // returning all rows/values from this page, or 0 of them pp->nesting[thread_depth].batch_size = - (s->num_rows == 0 && !is_page_contained(s->page, s->col.start_row, min_row, num_rows)) + (s->setup.num_rows == 0 && + !is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows)) ? 0 : pp->nesting[thread_depth].size; } @@ -307,10 +308,10 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) // zero sizes int depth = 0; - while (depth < s->page.num_output_nesting_levels) { + while (depth < s->setup.page.num_output_nesting_levels) { auto const thread_depth = depth + t; - if (thread_depth < s->page.num_output_nesting_levels) { - s->page.nesting[thread_depth].batch_size = 0; + if (thread_depth < s->setup.page.num_output_nesting_levels) { + s->setup.page.nesting[thread_depth].batch_size = 0; } depth += blockDim.x; } @@ -322,16 +323,16 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) : reinterpret_cast(pp->lvl_decode_buf[level_type::DEFINITION]); if (!t) { - s->page.skipped_values = -1; - s->page.skipped_leaf_values = 0; - s->input_row_count = 0; - s->input_value_count = 0; + s->setup.page.skipped_values = -1; + s->setup.page.skipped_leaf_values = 0; + s->input_row_count = 0; + s->input_value_count = 0; // in the base pass, we're computing the number of rows, make sure we visit absolutely // everything if (is_base_pass) { - s->first_row = 0; - s->num_rows = cuda::std::numeric_limits::max(); + s->setup.first_row = 0; + s->setup.num_rows = cuda::std::numeric_limits::max(); s->row_index_lower_bound = -1; } } @@ -339,7 +340,7 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) block.sync(); // update_page_sizes - update_page_sizes(s, s->page.num_input_values, rep, def, !is_base_pass, block); + update_page_sizes(s, s->setup.page.num_input_values, rep, def, !is_base_pass, block); // update output results: // - real number of rows for the whole page @@ -348,13 +349,13 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) // - string bytes if (is_base_pass) { // nesting level 0 is the root column, so the size is also the # of rows - if (!t) { pp->num_rows = s->page.nesting[0].batch_size; } + if (!t) { pp->num_rows = s->setup.page.nesting[0].batch_size; } // store off this batch size as the "full" size int depth = 0; - while (depth < s->page.num_output_nesting_levels) { + while (depth < s->setup.page.num_output_nesting_levels) { auto const thread_depth = depth + t; - if (thread_depth < s->page.num_output_nesting_levels) { + if (thread_depth < s->setup.page.num_output_nesting_levels) { pp->nesting[thread_depth].size = pp->nesting[thread_depth].batch_size; } depth += block.size(); @@ -362,8 +363,8 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) } if (!t) { - pp->skipped_values = s->page.skipped_values; - pp->skipped_leaf_values = s->page.skipped_leaf_values; + pp->skipped_values = s->setup.page.skipped_values; + pp->skipped_leaf_values = s->setup.page.skipped_leaf_values; } } @@ -446,7 +447,7 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) cg::invoke_one(block, [&]() { init(©_barrier, block.size()); }); block.sync(); decoders[level_type::REPETITION].init(block, - s->col.level_bits[level_type::REPETITION], + s->setup.col.level_bits[level_type::REPETITION], s->abs_lvl_start[level_type::REPETITION], s->abs_lvl_end[level_type::REPETITION], rep, @@ -469,7 +470,7 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) cg::invoke_one(block, [&]() { init(©_barrier, block.size()); }); block.sync(); decoders[level_type::DEFINITION].init(block, - s->col.level_bits[level_type::DEFINITION], + s->setup.col.level_bits[level_type::DEFINITION], s->abs_lvl_start[level_type::DEFINITION], s->abs_lvl_end[level_type::DEFINITION], def, diff --git a/cpp/src/io/parquet/page_data.cu b/cpp/src/io/parquet/page_data.cu index e5737b790b3c..2e31b0f7c6ba 100644 --- a/cpp/src/io/parquet/page_data.cu +++ b/cpp/src/io/parquet/page_data.cu @@ -78,7 +78,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } // Must be evaluated after setup_local_page_info - bool const has_repetition = s->col.max_level[level_type::REPETITION] > 0; + bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0; bool const process_nulls = should_process_nulls(s); auto const data_len = cuda::std::distance(s->data_start, s->data_end); @@ -103,12 +103,13 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) auto* const rep = reinterpret_cast(pp->lvl_decode_buf[level_type::REPETITION]); // Capture initial valid_map_offset before any processing that might modify it - int const init_valid_map_offset = s->nesting_info[s->col.max_nesting_depth - 1].valid_map_offset; + int const init_valid_map_offset = + s->nesting_info[s->setup.col.max_nesting_depth - 1].valid_map_offset; // skipped_leaf_values will always be 0 for flat hierarchies. - uint32_t skipped_leaf_values = s->page.skipped_leaf_values; - while (s->error == 0 && - (s->input_value_count < s->num_input_values || s->src_pos < s->nz_count)) { + uint32_t skipped_leaf_values = s->setup.page.skipped_leaf_values; + while (s->setup.error == 0 && + (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) { int target_pos; int src_pos = s->src_pos; @@ -129,7 +130,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) gpuDecodeLevels(s, sb, target_pos, rep, def, warp); } else { // WARP1..WARP3: Decode values - Type const dtype = s->col.physical_type; + Type const dtype = s->setup.col.physical_type; src_pos += block.thread_rank() - warp.size(); // The position in the output column/buffer @@ -146,7 +147,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // - so we will end up ignoring the first two input rows, and input rows 2..n will // get written to the output starting at position 0. // - if (!has_repetition) { dst_pos -= s->first_row; } + if (!has_repetition) { dst_pos -= s->setup.first_row; } // target_pos will always be properly bounded by num_rows, but dst_pos may be negative (values // before first_row) in the flat hierarchy case. @@ -158,14 +159,14 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) uint32_t val_src_pos = src_pos + skipped_leaf_values; // nesting level that is storing actual leaf values - int leaf_level_index = s->col.max_nesting_depth - 1; + int leaf_level_index = s->setup.col.max_nesting_depth - 1; uint32_t dtype_len = s->dtype_len; uint8_t const* src = s->data_start + val_src_pos; uint8_t* dst = nesting_info_base[leaf_level_index].data_out + static_cast(dst_pos) * dtype_len; - auto const is_decimal = - s->col.logical_type.has_value() and s->col.logical_type->type == LogicalType::DECIMAL; + auto const is_decimal = s->setup.col.logical_type.has_value() and + s->setup.col.logical_type->type == LogicalType::DECIMAL; // Note: non-decimal FIXED_LEN_BYTE_ARRAY will be handled in the string reader if (is_decimal) { @@ -219,7 +220,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // Zero-fill null positions after decoding valid values if (has_repetition) { - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; auto const& ni = s->nesting_info[leaf_level_index]; if (ni.valid_map != nullptr) { int const num_values = ni.valid_map_offset - init_valid_map_offset; @@ -228,8 +229,8 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } } - if (s->error != 0) { - cg::invoke_one(block, [&]() { set_error(s->error, error_code); }); + if (s->setup.error != 0) { + cg::invoke_one(block, [&]() { set_error(s->setup.error, error_code); }); } } @@ -286,18 +287,19 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } // Must be evaluated after setup_local_page_info - bool const has_repetition = s->col.max_level[level_type::REPETITION] > 0; + bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0; bool const process_nulls = should_process_nulls(s); PageNestingDecodeInfo* nesting_info_base = s->nesting_info; // Capture initial valid_map_offset before any processing that might modify it - int const init_valid_map_offset = s->nesting_info[s->col.max_nesting_depth - 1].valid_map_offset; + int const init_valid_map_offset = + s->nesting_info[s->setup.col.max_nesting_depth - 1].valid_map_offset; if (s->dict_base) { out_warp_id = (s->dict_bits > 0) ? 2 : 1; } else { - switch (s->col.physical_type) { + switch (s->setup.col.physical_type) { case Type::BOOLEAN: [[fallthrough]]; case Type::BYTE_ARRAY: [[fallthrough]]; case Type::FIXED_LEN_BYTE_ARRAY: out_warp_id = 2; break; @@ -312,15 +314,15 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) : reinterpret_cast(pp->lvl_decode_buf[level_type::DEFINITION]); auto* const rep = reinterpret_cast(pp->lvl_decode_buf[level_type::REPETITION]); - auto const is_decimal = - s->col.logical_type.has_value() and s->col.logical_type->type == LogicalType::DECIMAL; - Type const dtype = s->col.physical_type; + auto const is_decimal = s->setup.col.logical_type.has_value() and + s->setup.col.logical_type->type == LogicalType::DECIMAL; + Type const dtype = s->setup.col.physical_type; auto const first_out_thread_id = out_warp_id * warp.size(); // skipped_leaf_values will always be 0 for flat hierarchies. - uint32_t skipped_leaf_values = s->page.skipped_leaf_values; - while (s->error == 0 && - (s->input_value_count < s->num_input_values || s->src_pos < s->nz_count)) { + uint32_t skipped_leaf_values = s->setup.page.skipped_leaf_values; + while (s->setup.error == 0 && + (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) { int target_pos; int src_pos = s->src_pos; @@ -352,10 +354,10 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) if (s->dict_base) { src_target_pos = decode_dictionary_indices(s, sb, src_target_pos, warp).first; - } else if (s->col.physical_type == Type::BOOLEAN) { + } else if (s->setup.col.physical_type == Type::BOOLEAN) { src_target_pos = decode_rle_booleans(s, sb, src_target_pos, warp); - } else if (s->col.physical_type == Type::BYTE_ARRAY or - s->col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { + } else if (s->setup.col.physical_type == Type::BYTE_ARRAY or + s->setup.col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { initialize_string_descriptors(s, sb, src_target_pos, warp); } if (warp.thread_rank() == 0) { s->dict_pos = src_target_pos; } @@ -377,7 +379,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // - so we will end up ignoring the first two input rows, and input rows 2..n will // get written to the output starting at position 0. // - if (!has_repetition) { dst_pos -= s->first_row; } + if (!has_repetition) { dst_pos -= s->setup.first_row; } // target_pos will always be properly bounded by num_rows, but dst_pos may be negative (values // before first_row) in the flat hierarchy case. @@ -389,7 +391,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) uint32_t val_src_pos = src_pos + skipped_leaf_values; // nesting level that is storing actual leaf values - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; uint32_t const dtype_len = s->dtype_len; void* dst = @@ -398,7 +400,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) if (dtype == Type::BYTE_ARRAY) { if (is_decimal) { auto const [ptr, len] = gpuGetStringData(s, sb, val_src_pos); - auto const decimal_precision = s->col.logical_type->precision(); + auto const decimal_precision = s->setup.col.logical_type->precision(); if (decimal_precision <= MAX_DECIMAL32_PRECISION) { gpuOutputByteArrayAsInt(ptr, len, static_cast(dst)); } else if (decimal_precision <= MAX_DECIMAL64_PRECISION) { @@ -464,7 +466,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) auto const is_string = ((dtype == Type::BYTE_ARRAY) && !is_decimal) || (dtype == Type::FIXED_LEN_BYTE_ARRAY); if (is_string || has_repetition) { - auto const& ni = s->nesting_info[s->col.max_nesting_depth - 1]; + auto const& ni = s->nesting_info[s->setup.col.max_nesting_depth - 1]; if (ni.valid_map != nullptr) { int const num_values = ni.valid_map_offset - init_valid_map_offset; zero_fill_null_positions_shared( @@ -472,8 +474,8 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } } - if (s->error != 0) { - cg::invoke_one(block, [&]() { set_error(s->error, error_code); }); + if (s->setup.error != 0) { + cg::invoke_one(block, [&]() { set_error(s->setup.error, error_code); }); } } diff --git a/cpp/src/io/parquet/page_data.cuh b/cpp/src/io/parquet/page_data.cuh index ccc8935c22ab..98a66b842892 100644 --- a/cpp/src/io/parquet/page_data.cuh +++ b/cpp/src/io/parquet/page_data.cuh @@ -1,6 +1,6 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #pragma once @@ -24,7 +24,7 @@ template inline __device__ void gpuOutputString(page_state_s* s, state_buf* sb, int src_pos, void* dstv) { auto [ptr, len] = gpuGetStringData(s, sb, src_pos); - if (s->col.is_strings_to_cat and s->col.physical_type == Type::BYTE_ARRAY) { + if (s->setup.col.is_strings_to_cat and s->setup.col.physical_type == Type::BYTE_ARRAY) { // Output hash. This hash value is used if the option to convert strings to // categoricals is enabled. The seed value is chosen arbitrarily. uint32_t constexpr hash_seed = 33; @@ -162,7 +162,7 @@ inline __device__ void read_int96_timestamp(page_state_s* s, days - 2440588}; // TBD: Should be noon instead of midnight, but this matches pyarrow *dst = [&]() { - switch (s->col.ts_clock_rate) { + switch (s->setup.col.ts_clock_rate) { case 1: // seconds return duration_cast(d_d).count() + duration_cast(duration_ns{nanos}).count(); diff --git a/cpp/src/io/parquet/page_decode.cuh b/cpp/src/io/parquet/page_decode.cuh index ae3eff032838..3ad736a278a2 100644 --- a/cpp/src/io/parquet/page_decode.cuh +++ b/cpp/src/io/parquet/page_decode.cuh @@ -21,15 +21,22 @@ namespace cg = cooperative_groups; enum class copy_mode : bool { INDIRECT, DIRECT }; +struct page_decode_setup_state { + PageInfo page{}; + ColumnChunkDesc col{}; + int32_t first_row{}; // First row in page to output + int32_t num_rows{}; // Rows in page to decode (including rows to be skipped) + int32_t num_input_values{}; // total # of input/level values in the page + kernel_error::value_type error{}; +}; + struct page_state_s { CUDF_HOST_DEVICE constexpr page_state_s() noexcept {} + page_decode_setup_state setup{}; uint8_t const* data_start{}; uint8_t const* data_end{}; uint8_t const* dict_base{}; // ptr to dictionary page data int32_t dict_size{}; // size of dictionary data - int32_t first_row{}; // First row in page to output - int32_t num_rows{}; // Rows in page to decode (including rows to be skipped) - int32_t num_input_values{}; // total # of input/level values in the page int32_t dtype_len{}; // Output data type length int32_t dtype_len_in{}; // Can be larger than dtype_len if truncating 32-bit into 8-bit int32_t dict_bits{}; // # of bits to store dictionary indices @@ -37,9 +44,6 @@ struct page_state_s { int32_t dict_val{}; uint32_t initial_rle_run[NUM_LEVEL_TYPES]{}; // [def,rep] int32_t initial_rle_value[NUM_LEVEL_TYPES]{}; // [def,rep] - kernel_error::value_type error{}; - PageInfo page{}; - ColumnChunkDesc col{}; // (leaf) value decoding int32_t nz_count{}; // number of valid entries in nz_idx (write position in circular buffer) @@ -64,13 +68,13 @@ struct page_state_s { inline __device__ void set_error_code(decode_error err) { - cuda::atomic_ref ref{error}; + cuda::atomic_ref ref{setup.error}; ref.fetch_or(static_cast(err), cuda::std::memory_order_relaxed); } inline __device__ void reset_error_code() { - cuda::atomic_ref ref{error}; + cuda::atomic_ref ref{setup.error}; ref.store(0, cuda::std::memory_order_release); } }; @@ -96,10 +100,10 @@ struct null_count_back_copier { { if (s->nesting_info != nullptr and s->nesting_info == s->nesting_decode_cache) { int depth = 0; - while (depth < s->page.num_output_nesting_levels) { + while (depth < s->setup.page.num_output_nesting_levels) { int const thread_depth = depth + t; - if (thread_depth < s->page.num_output_nesting_levels) { - s->page.nesting_decode[thread_depth].null_count = + if (thread_depth < s->setup.page.num_output_nesting_levels) { + s->setup.page.nesting_decode[thread_depth].null_count = s->nesting_decode_cache[thread_depth].null_count; } depth += blockDim.x; @@ -117,7 +121,7 @@ struct null_count_back_copier { __device__ inline bool is_nullable(page_state_s* s) { auto const lvl = level_type::DEFINITION; - auto const max_def_level = s->col.max_level[lvl]; + auto const max_def_level = s->setup.col.max_level[lvl]; return max_def_level > 0; } @@ -142,13 +146,13 @@ __device__ inline bool maybe_has_nulls(page_state_s* s) // repeated run with number of items in the run not equal // to the rows in the page, assume that means we could have nulls - if (s->page.num_input_values != (init_run >> 1)) { return true; } + if (s->setup.page.num_input_values != (init_run >> 1)) { return true; } - auto const lvl_bits = s->col.level_bits[lvl]; + auto const lvl_bits = s->setup.col.level_bits[lvl]; auto const run_val = lvl_bits == 0 ? 0 : s->initial_rle_value[lvl]; // the encoded repeated value isn't valid, we have (all) nulls - return run_val != s->col.max_level[lvl]; + return run_val != s->setup.col.max_level[lvl]; } /** @@ -552,7 +556,7 @@ __device__ size_type initialize_string_descriptors(page_state_s* s, int total_len = 0; // All group threads can participate for fixed len byte arrays. - if (s->col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { + if (s->setup.col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { int const dtype_len_in = s->dtype_len_in; total_len = min((target_pos - pos) * dtype_len_in, dict_size - s->dict_val); if constexpr (sizes_only == is_calc_sizes_only::NO) { @@ -690,17 +694,17 @@ inline __device__ void get_nesting_bounds(int& start_depth, // Clamp to decoded level count; for chunked reads the level buffers may be // smaller than num_input_values when skip_rows/num_rows reduces the range. auto const actual_num_values = - (s->page.num_decoded_level_values > 0) - ? cuda::std::min(s->page.num_input_values, s->page.num_decoded_level_values) - : s->page.num_input_values; + (s->setup.page.num_decoded_level_values > 0) + ? cuda::std::min(s->setup.page.num_input_values, s->setup.page.num_decoded_level_values) + : s->setup.page.num_input_values; auto const max_idx = cuda::std::min(target_input_value_count, actual_num_values); if (input_value_count + t < max_idx) { int const index = input_value_count + t; - d = (def != nullptr) ? def[index] : s->col.max_level[level_type::DEFINITION]; + d = (def != nullptr) ? def[index] : s->setup.col.max_level[level_type::DEFINITION]; // if we have repetition (there are list columns involved) we have to // bound what nesting levels we apply values to - if (s->col.max_level[level_type::REPETITION] > 0) { + if (s->setup.col.max_level[level_type::REPETITION] > 0) { int const r = rep[index]; start_depth = s->nesting_info[r].start_depth; end_depth = s->nesting_info[d].end_depth; @@ -709,7 +713,7 @@ inline __device__ void get_nesting_bounds(int& start_depth, // traverse the entire hierarchy. else { start_depth = 0; - end_depth = s->col.max_nesting_depth - 1; + end_depth = s->setup.col.max_nesting_depth - 1; } } } @@ -740,8 +744,8 @@ __device__ void gpuUpdateValidityOffsetsAndRowIndices(int32_t target_input_value if (s->input_value_count >= target_input_value_count) { return; } // max nesting depth of the column - int const max_depth = s->col.max_nesting_depth; - bool const has_repetition = s->col.max_level[level_type::REPETITION] > 0; + int const max_depth = s->setup.col.max_nesting_depth; + bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0; // how many (input) values we've processed in the page so far int input_value_count = s->input_value_count; // how many rows we've processed in the page so far @@ -775,7 +779,7 @@ __device__ void gpuUpdateValidityOffsetsAndRowIndices(int32_t target_input_value input_row_count += __popc(warp_row_count_mask); // is this thread within read row bounds? int const in_row_bounds = thread_row_index >= s->row_index_lower_bound && - thread_row_index < (s->first_row + s->num_rows) + thread_row_index < (s->setup.first_row + s->setup.num_rows) ? 1 : 0; @@ -852,9 +856,9 @@ __device__ void gpuUpdateValidityOffsetsAndRowIndices(int32_t target_input_value // the correct position to start reading. since we are about to write the validity vector here // we need to adjust our computed mask to take into account the write row bounds. int const in_write_row_bounds = - !has_repetition - ? thread_row_index >= s->first_row && thread_row_index < (s->first_row + s->num_rows) - : in_row_bounds; + !has_repetition ? thread_row_index >= s->setup.first_row && + thread_row_index < (s->setup.first_row + s->setup.num_rows) + : in_row_bounds; int const first_thread_in_write_range = !has_repetition ? __ffs(ballot(in_write_row_bounds)) - 1 : 0; @@ -926,12 +930,12 @@ __device__ void gpuDecodeLevels( cg::thread_block_tile const& warp) { auto cur_leaf_count = target_leaf_count; - while (s->error == 0 && s->nz_count < target_leaf_count && - s->input_value_count < s->num_input_values) { + while (s->setup.error == 0 && s->nz_count < target_leaf_count && + s->input_value_count < s->setup.num_input_values) { // because the rep and def streams are encoded separately, we cannot request an exact // # of values to be decoded at once. we can only process the lowest # of decoded rep/def // levels we get. - auto const actual_leaf_count = cuda::std::min(cur_leaf_count, s->num_input_values); + auto const actual_leaf_count = cuda::std::min(cur_leaf_count, s->setup.num_input_values); // process what we got back gpuUpdateValidityOffsetsAndRowIndices( @@ -958,9 +962,9 @@ inline __device__ uint32_t InitLevelSection(page_state_s* s, level_type lvl) { int32_t len; - int const level_bits = s->col.level_bits[lvl]; - auto const encoding = lvl == level_type::DEFINITION ? s->page.definition_level_encoding - : s->page.repetition_level_encoding; + int const level_bits = s->setup.col.level_bits[lvl]; + auto const encoding = lvl == level_type::DEFINITION ? s->setup.page.definition_level_encoding + : s->setup.page.repetition_level_encoding; auto start = cur; @@ -987,13 +991,13 @@ inline __device__ uint32_t InitLevelSection(page_state_s* s, // this is a little redundant. if level_bits == 0, then nothing should be encoded // for the level, but some V2 files in the wild violate this and encode the data anyway. // thus we will handle V2 headers separately. - if ((s->page.flags & PAGEINFO_FLAGS_V2) != 0 && (len = s->page.lvl_bytes[lvl]) != 0) { + if ((s->setup.page.flags & PAGEINFO_FLAGS_V2) != 0 && (len = s->setup.page.lvl_bytes[lvl]) != 0) { // V2 only uses RLE encoding so no need to check encoding s->abs_lvl_start[lvl] = cur; init_rle(cur, cur + len); } else if (level_bits == 0) { len = 0; - s->initial_rle_run[lvl] = s->page.num_input_values * 2; // repeated value + s->initial_rle_run[lvl] = s->setup.page.num_input_values * 2; // repeated value s->initial_rle_value[lvl] = 0; s->abs_lvl_start[lvl] = cur; } else if (encoding == Encoding::RLE) { // V1 header with RLE encoding @@ -1009,8 +1013,8 @@ inline __device__ uint32_t InitLevelSection(page_state_s* s, s->set_error_code(decode_error::LEVEL_STREAM_OVERRUN); } } else if (encoding == Encoding::BIT_PACKED) { - len = (s->page.num_input_values * level_bits + 7) >> 3; - s->initial_rle_run[lvl] = ((s->page.num_input_values + 7) >> 3) * 2 + 1; // literal run + len = (s->setup.page.num_input_values * level_bits + 7) >> 3; + s->initial_rle_run[lvl] = ((s->setup.page.num_input_values + 7) >> 3) * 2 + 1; // literal run s->initial_rle_value[lvl] = 0; s->abs_lvl_start[lvl] = cur; } else { @@ -1078,71 +1082,74 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, // Fetch page info if (!t) { - s->page = *p; + s->setup.page = *p; s->nesting_info = nullptr; - s->col = chunks[s->page.chunk_idx]; + s->setup.col = chunks[s->setup.page.chunk_idx]; } __syncthreads(); // return false if this is a dictionary page or it does not pass the filter condition - if ((s->page.flags & PAGEINFO_FLAGS_DICTIONARY) != 0 || !filter(s->page)) { return false; } + if ((s->setup.page.flags & PAGEINFO_FLAGS_DICTIONARY) != 0 || !filter(s->setup.page)) { + return false; + } // our starting row (absolute index) is // col.start_row == absolute row index // page.chunk-row == relative row index within the chunk - size_t const page_start_row = s->col.start_row + s->page.chunk_row; + size_t const page_start_row = s->setup.col.start_row + s->setup.page.chunk_row; // if we can use the nesting decode cache, set it up now - auto const can_use_decode_cache = s->page.nesting_info_size <= max_cacheable_nesting_decode_info; + auto const can_use_decode_cache = + s->setup.page.nesting_info_size <= max_cacheable_nesting_decode_info; if (can_use_decode_cache) { int depth = 0; - while (depth < s->page.nesting_info_size) { + while (depth < s->setup.page.nesting_info_size) { int const thread_depth = depth + t; - if (thread_depth < s->page.nesting_info_size) { + if (thread_depth < s->setup.page.nesting_info_size) { // these values need to be copied over from global s->nesting_decode_cache[thread_depth].max_def_level = - s->page.nesting_decode[thread_depth].max_def_level; + s->setup.page.nesting_decode[thread_depth].max_def_level; s->nesting_decode_cache[thread_depth].page_start_value = - s->page.nesting_decode[thread_depth].page_start_value; + s->setup.page.nesting_decode[thread_depth].page_start_value; s->nesting_decode_cache[thread_depth].start_depth = - s->page.nesting_decode[thread_depth].start_depth; + s->setup.page.nesting_decode[thread_depth].start_depth; s->nesting_decode_cache[thread_depth].end_depth = - s->page.nesting_decode[thread_depth].end_depth; + s->setup.page.nesting_decode[thread_depth].end_depth; } depth += blockDim.x; } } if (!t) { - s->nesting_info = can_use_decode_cache ? s->nesting_decode_cache : s->page.nesting_decode; + s->nesting_info = can_use_decode_cache ? s->nesting_decode_cache : s->setup.page.nesting_decode; - // NOTE: s->page.num_rows, s->col.chunk_row, s->first_row and s->num_rows will be - // invalid/bogus during first pass of the preprocess step for nested types. this is ok - // because we ignore these values in that stage. + // NOTE: s->setup.page.num_rows, s->setup.col.chunk_row, s->setup.first_row and + // s->setup.num_rows will be invalid/bogus during first pass of the preprocess step for nested + // types. this is ok because we ignore these values in that stage. auto const end_row = min_row + num_rows; // if we are totally outside the range of the input, do nothing - auto const page_end_row = page_start_row + s->page.num_rows; + auto const page_end_row = page_start_row + s->setup.page.num_rows; if ((page_start_row >= end_row) || (page_end_row <= min_row)) { - s->first_row = 0; - s->num_rows = 0; + s->setup.first_row = 0; + s->setup.num_rows = 0; } // otherwise else { - s->first_row = page_start_row >= min_row ? 0 : min_row - page_start_row; - auto const max_page_rows = s->page.num_rows - s->first_row; - s->num_rows = (page_start_row + s->first_row) + max_page_rows <= end_row + s->setup.first_row = page_start_row >= min_row ? 0 : min_row - page_start_row; + auto const max_page_rows = s->setup.page.num_rows - s->setup.first_row; + s->setup.num_rows = (page_start_row + s->setup.first_row) + max_page_rows <= end_row ? max_page_rows - : end_row - (page_start_row + s->first_row); + : end_row - (page_start_row + s->setup.first_row); } } __syncthreads(); // zero counts int depth = 0; - while (depth < s->page.num_output_nesting_levels) { + while (depth < s->setup.page.num_output_nesting_levels) { int const thread_depth = depth + t; - if (thread_depth < s->page.num_output_nesting_levels) { + if (thread_depth < s->setup.page.num_output_nesting_levels) { s->nesting_info[thread_depth].valid_count = 0; s->nesting_info[thread_depth].value_count = 0; s->nesting_info[thread_depth].null_count = 0; @@ -1162,9 +1169,10 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, // P1 will contain 0 rows // // NOTE: this check needs to be done after the null counts have been zeroed out - bool const has_repetition = s->col.max_level[level_type::REPETITION] > 0; + bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0; if ((stage == page_processing_stage::STRING_BOUNDS || stage == page_processing_stage::DECODE) && - !page_has_rows_to_process(s->page, s->col.start_row, min_row, num_rows, has_repetition)) { + !page_has_rows_to_process( + s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition)) { return false; } @@ -1175,15 +1183,15 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, // values. The case is: // - On page N-1, the last row starts, with 2/6 values encoded // - On page N, the remaining 4/6 values are encoded, but there are no new rows. - // if (s->page.num_input_values > 0 && s->page.num_rows > 0) { - if (s->page.num_input_values > 0) { - uint8_t* cur = s->page.page_data; - uint8_t* end = cur + s->page.uncompressed_page_size; + // if (s->setup.page.num_input_values > 0 && s->setup.page.num_rows > 0) { + if (s->setup.page.num_input_values > 0) { + uint8_t* cur = s->setup.page.page_data; + uint8_t* end = cur + s->setup.page.uncompressed_page_size; s->ts_scale = 0; // Validate data type - auto const data_type = s->col.physical_type; - auto const is_decimal = - s->col.logical_type.has_value() and s->col.logical_type->type == LogicalType::DECIMAL; + auto const data_type = s->setup.col.physical_type; + auto const is_decimal = s->setup.col.logical_type.has_value() and + s->setup.col.logical_type->type == LogicalType::DECIMAL; switch (data_type) { case Type::BOOLEAN: s->dtype_len = 1; // Boolean are stored as 1 byte on the output @@ -1191,15 +1199,16 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, case Type::INT32: [[fallthrough]]; case Type::FLOAT: s->dtype_len = 4; break; case Type::INT64: - if (s->col.ts_clock_rate) { - s->ts_scale = calc_timestamp_scale(s->col.logical_type, s->col.ts_clock_rate); + if (s->setup.col.ts_clock_rate) { + s->ts_scale = + calc_timestamp_scale(s->setup.col.logical_type, s->setup.col.ts_clock_rate); } [[fallthrough]]; case Type::DOUBLE: s->dtype_len = 8; break; case Type::INT96: s->dtype_len = 12; break; case Type::BYTE_ARRAY: if (is_decimal) { - auto const decimal_precision = s->col.logical_type->precision(); + auto const decimal_precision = s->setup.col.logical_type->precision(); s->dtype_len = [decimal_precision]() { if (decimal_precision <= MAX_DECIMAL32_PRECISION) { return sizeof(int32_t); @@ -1214,7 +1223,7 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, } break; default: // FIXED_LEN_BYTE_ARRAY: - s->dtype_len = s->col.type_length; + s->dtype_len = s->setup.col.type_length; if (s->dtype_len <= 0) { s->set_error_code(decode_error::INVALID_DATA_TYPE); } break; } @@ -1236,8 +1245,8 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, } } else if (data_type == Type::INT32) { // check for smaller bitwidths - if (s->col.logical_type.has_value()) { - auto const& lt = *s->col.logical_type; + if (s->setup.col.logical_type.has_value()) { + auto const& lt = *s->setup.col.logical_type; if (lt.type == LogicalType::INTEGER) { s->dtype_len = lt.bit_width() / 8; } else if (lt.is_time_millis()) { @@ -1245,7 +1254,7 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, s->dtype_len = 8; } } - } else if (data_type == Type::BYTE_ARRAY && s->col.is_strings_to_cat) { + } else if (data_type == Type::BYTE_ARRAY && s->setup.col.is_strings_to_cat) { s->dtype_len = 4; // HASH32 output } else if (data_type == Type::INT96) { s->dtype_len = 8; // Convert to 64-bit timestamp @@ -1257,17 +1266,18 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, // - for flat schemas, we can do this directly by using row counts // - for nested schemas, these offsets are computed during the preprocess step // - // NOTE: in a chunked read situation, s->col.column_data_base and s->col.valid_map_base - // will be aliased to memory that has been freed when we get here in the non-decode step, so - // we cannot check against nullptr. we'll just check a flag directly. + // NOTE: in a chunked read situation, s->setup.col.column_data_base and + // s->setup.col.valid_map_base will be aliased to memory that has been freed when we get here + // in the non-decode step, so we cannot check against nullptr. we'll just check a flag + // directly. if (stage == page_processing_stage::DECODE) { - int max_depth = s->col.max_nesting_depth; + int max_depth = s->setup.col.max_nesting_depth; for (int idx = 0; idx < max_depth; idx++) { PageNestingDecodeInfo* nesting_info = &s->nesting_info[idx]; size_t output_offset; // schemas without lists - if (s->col.max_level[level_type::REPETITION] == 0) { + if (s->setup.col.max_level[level_type::REPETITION] == 0) { output_offset = page_start_row >= min_row ? page_start_row - min_row : 0; } // for schemas with lists, we've already got the exact value precomputed @@ -1275,13 +1285,14 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, output_offset = nesting_info->page_start_value; } - if (s->col.column_data_base != nullptr) { - nesting_info->data_out = static_cast(s->col.column_data_base[idx]); - if (s->col.column_string_base != nullptr) { - nesting_info->string_out = static_cast(s->col.column_string_base[idx]); + if (s->setup.col.column_data_base != nullptr) { + nesting_info->data_out = static_cast(s->setup.col.column_data_base[idx]); + if (s->setup.col.column_string_base != nullptr) { + nesting_info->string_out = + static_cast(s->setup.col.column_string_base[idx]); } - nesting_info->data_out = static_cast(s->col.column_data_base[idx]); + nesting_info->data_out = static_cast(s->setup.col.column_data_base[idx]); if (nesting_info->data_out != nullptr) { // anything below max depth with a valid data pointer must be a list, so the @@ -1289,13 +1300,13 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, uint32_t len = idx < max_depth - 1 ? sizeof(cudf::size_type) : s->dtype_len; // if this is a string column, then dtype_len is a lie. data will be offsets rather // than (ptr,len) tuples. - if (is_string_col(s->col)) { len = sizeof(cudf::size_type); } + if (is_string_col(s->setup.col)) { len = sizeof(cudf::size_type); } nesting_info->data_out += (output_offset * len); } if (nesting_info->string_out != nullptr) { - nesting_info->string_out += s->page.str_offset; + nesting_info->string_out += s->setup.page.str_offset; } - nesting_info->valid_map = s->col.valid_map_base[idx]; + nesting_info->valid_map = s->setup.col.valid_map_base[idx]; if (nesting_info->valid_map != nullptr) { nesting_info->valid_map += output_offset >> 5; nesting_info->valid_map_offset = (int32_t)(output_offset & 0x1f); @@ -1315,26 +1326,27 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, s->dict_val = 0; // NOTE: if additional encodings are supported in the future, modifications must // be made to is_supported_encoding() in reader_impl_preprocess.cu - switch (s->page.encoding) { + switch (s->setup.page.encoding) { case Encoding::PLAIN_DICTIONARY: case Encoding::RLE_DICTIONARY: { // RLE-packed dictionary indices, first byte indicates index length in bits - auto const is_decimal = - s->col.logical_type.has_value() and s->col.logical_type->type == LogicalType::DECIMAL; - if ((s->col.physical_type == Type::BYTE_ARRAY or - s->col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) and - not is_decimal and s->col.str_dict_index != nullptr) { + auto const is_decimal = s->setup.col.logical_type.has_value() and + s->setup.col.logical_type->type == LogicalType::DECIMAL; + if ((s->setup.col.physical_type == Type::BYTE_ARRAY or + s->setup.col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) and + not is_decimal and s->setup.col.str_dict_index != nullptr) { // String dictionary: use index - s->dict_base = reinterpret_cast(s->col.str_dict_index); - s->dict_size = s->col.dict_page->num_input_values * sizeof(string_index_pair); + s->dict_base = reinterpret_cast(s->setup.col.str_dict_index); + s->dict_size = s->setup.col.dict_page->num_input_values * sizeof(string_index_pair); } else { - s->dict_base = s->col.dict_page->page_data; - s->dict_size = s->col.dict_page->uncompressed_page_size; + s->dict_base = s->setup.col.dict_page->page_data; + s->dict_size = s->setup.col.dict_page->uncompressed_page_size; } s->dict_run = 0; s->dict_val = 0; s->dict_bits = (cur < end) ? *cur++ : 0; - if (s->dict_bits > 32 || (!s->dict_base && s->col.dict_page->num_input_values > 0)) { + if (s->dict_bits > 32 || + (!s->dict_base && s->setup.col.dict_page->num_input_values > 0)) { s->set_error_code(decode_error::INVALID_DICT_WIDTH); } } break; @@ -1342,7 +1354,7 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, case Encoding::BYTE_STREAM_SPLIT: s->dict_size = static_cast(end - cur); s->dict_val = 0; - if (s->col.physical_type == Type::BOOLEAN) { s->dict_run = s->dict_size * 2 + 1; } + if (s->setup.col.physical_type == Type::BOOLEAN) { s->dict_run = s->dict_size * 2 + 1; } break; case Encoding::RLE: { // first 4 bytes are length of RLE data @@ -1368,22 +1380,22 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, s->set_error_code(decode_error::EMPTY_PAGE); } - s->nz_count = 0; - s->num_input_values = s->page.num_input_values; - s->dict_pos = 0; - s->src_pos = 0; + s->nz_count = 0; + s->setup.num_input_values = s->setup.page.num_input_values; + s->dict_pos = 0; + s->src_pos = 0; // for flat hierarchies, we can't know how many leaf values to skip unless we do a full // preprocess of the definition levels (since nulls will have no actual decodable value, there // is no direct correlation between # of rows and # of decodable values). so we will start // processing at the beginning of the value stream and disregard any indices that start // before the first row. - if (s->col.max_level[level_type::REPETITION] == 0) { - s->page.skipped_values = 0; - s->page.skipped_leaf_values = 0; - s->input_value_count = 0; - s->input_row_count = 0; - s->input_leaf_count = 0; + if (s->setup.col.max_level[level_type::REPETITION] == 0) { + s->setup.page.skipped_values = 0; + s->setup.page.skipped_leaf_values = 0; + s->input_value_count = 0; + s->input_row_count = 0; + s->input_leaf_count = 0; // The fixed-width decode kernel ASSUMES this is always -1 for non-lists! s->row_index_lower_bound = -1; @@ -1393,7 +1405,7 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, else { // input_row_count translates to "how many rows we have processed so far", so since we are // skipping directly to where we want to start decoding, set it to first_row - s->input_row_count = s->first_row; + s->input_row_count = s->setup.first_row; // return the lower bound to compare (page-relative) thread row index against. Explanation: // In the case of nested schemas, rows can span page boundaries. That is to say, @@ -1408,19 +1420,19 @@ inline __device__ bool setup_local_page_info(page_state_s* const s, if (min_row < page_start_row && max_row >= page_start_row - 1) { s->row_index_lower_bound = -1; } else { - s->row_index_lower_bound = s->first_row; + s->row_index_lower_bound = s->setup.first_row; } // if we're in the decoding step, jump directly to the first // value we care about if (stage == page_processing_stage::DECODE) { - s->input_value_count = s->page.skipped_values > -1 ? s->page.skipped_values : 0; + s->input_value_count = s->setup.page.skipped_values > -1 ? s->setup.page.skipped_values : 0; } else if (stage == page_processing_stage::PREPROCESS) { s->input_value_count = 0; s->input_leaf_count = 0; // magic number to indicate it hasn't been set for use inside UpdatePageSizes - s->page.skipped_values = -1; - s->page.skipped_leaf_values = 0; + s->setup.page.skipped_values = -1; + s->setup.page.skipped_leaf_values = 0; } } @@ -1454,7 +1466,7 @@ __device__ void zero_fill_null_positions_shared( auto const warp = cg::tiled_partition(block); // nesting level that is storing actual leaf values - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; auto const& ni = s->nesting_info[leaf_level_index]; // Check if we have nulls to fill diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index 7cb5fbd59d26..5769fe0f406c 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -335,11 +335,12 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) } // Must be evaluated after setup_local_page_info - bool const has_repetition = s->col.max_level[level_type::REPETITION] > 0; + bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0; bool const process_nulls = should_process_nulls(s); // Capture initial valid_map_offset before any processing that might modify it - int const init_valid_map_offset = s->nesting_info[s->col.max_nesting_depth - 1].valid_map_offset; + int const init_valid_map_offset = + s->nesting_info[s->setup.col.max_nesting_depth - 1].valid_map_offset; // copying logic from gpuDecodePageData. PageNestingDecodeInfo const* nesting_info_base = s->nesting_info; @@ -352,7 +353,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) auto* const rep = reinterpret_cast(pp->lvl_decode_buf[level_type::REPETITION]); // skipped_leaf_values will always be 0 for flat hierarchies. - uint32_t const skipped_leaf_values = s->page.skipped_leaf_values; + uint32_t const skipped_leaf_values = s->setup.page.skipped_leaf_values; // initialize delta state if (block.thread_rank() == 0) { db->init_binary_block(s->data_start, s->data_end); } @@ -371,8 +372,8 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // that has a value we need. if (skipped_leaf_values > 0) { db->skip_values(skipped_leaf_values); } - while (s->error == 0 && - (s->input_value_count < s->num_input_values || s->src_pos < s->nz_count)) { + while (s->setup.error == 0 && + (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) { uint32_t target_pos; uint32_t const src_pos = s->src_pos; @@ -400,7 +401,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) } else if (src_pos < target_pos) { // warp 2 // nesting level that is storing actual leaf values - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; // process the mini-block using warps for (uint32_t sp = src_pos + warp.thread_rank(); sp < src_pos + batch_size; @@ -409,7 +410,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) int32_t dst_pos = sb->nz_idx[rolling_index(sp)]; // handle skip_rows here. flat hierarchies can just skip up to first_row. - if (!has_repetition) { dst_pos -= s->first_row; } + if (!has_repetition) { dst_pos -= s->setup.first_row; } // place value for this thread if (dst_pos >= 0 && sp < target_pos) { @@ -431,7 +432,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) if (has_repetition) { // Zero-fill null positions after decoding valid values - auto const& ni = s->nesting_info[s->col.max_nesting_depth - 1]; + auto const& ni = s->nesting_info[s->setup.col.max_nesting_depth - 1]; if (ni.valid_map != nullptr) { int const num_values = ni.valid_map_offset - init_valid_map_offset; zero_fill_null_positions_shared( @@ -439,7 +440,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) } } - if (block.thread_rank() == 0 and s->error != 0) { set_error(s->error, error_code); } + if (block.thread_rank() == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } } // Decode page data that is DELTA_BYTE_ARRAY packed. This encoding consists of a DELTA_BINARY_PACKED @@ -484,7 +485,8 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) return; } - if (s->col.logical_type.has_value() && s->col.logical_type->type == LogicalType::DECIMAL) { + if (s->setup.col.logical_type.has_value() && + s->setup.col.logical_type->type == LogicalType::DECIMAL) { // we cannot read decimal encoded with DELTA_BYTE_ARRAY yet if (block.thread_rank() == 0) { set_error(static_cast(decode_error::INVALID_DATA_TYPE), error_code); @@ -492,14 +494,17 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) return; } - bool const has_repetition = s->col.max_level[level_type::REPETITION] > 0; + bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0; bool const process_nulls = should_process_nulls(s); // Capture initial valid_map_offset before any processing that might modify it - int const init_valid_map_offset = s->nesting_info[s->col.max_nesting_depth - 1].valid_map_offset; + int const init_valid_map_offset = + s->nesting_info[s->setup.col.max_nesting_depth - 1].valid_map_offset; // choose a character parallel string copy when the average string is longer than a warp - auto const use_char_ll = (s->page.str_bytes / s->page.num_valids) > cudf::detail::warp_size; + auto const use_char_ll = + s->setup.page.num_valids > 0 && + (s->setup.page.str_bytes / s->setup.page.num_valids) > cudf::detail::warp_size; // copying logic from decode_page_data. PageNestingDecodeInfo const* nesting_info_base = s->nesting_info; @@ -512,11 +517,11 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) auto* const rep = reinterpret_cast(pp->lvl_decode_buf[level_type::REPETITION]); // skipped_leaf_values will always be 0 for flat hierarchies. - uint32_t const skipped_leaf_values = s->page.skipped_leaf_values; + uint32_t const skipped_leaf_values = s->setup.page.skipped_leaf_values; if (block.thread_rank() == 0) { // initialize the prefixes and suffixes blocks - dba->init(s->data_start, s->data_end, s->page.start_val, s->page.temp_string_buf); + dba->init(s->data_start, s->data_end, s->setup.page.start_val, s->setup.page.temp_string_buf); } block.sync(); @@ -539,7 +544,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } // pointer to location to output final strings - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; auto strings_data = nesting_info_base[leaf_level_index].string_out; // sanity check to make sure we can process this page @@ -554,17 +559,18 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // if this is a bounds page and nested, then we need to skip up front. non-nested will work // its way through the page. - int string_pos = has_repetition ? s->page.start_val : 0; + int string_pos = has_repetition ? s->setup.page.start_val : 0; auto const is_bounds_pg = - is_bounds_page(s->page, s->col.start_row, min_row, num_rows, has_repetition); + is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); if (is_bounds_pg && string_pos > 0) { dba->skip(use_char_ll); } - while (!s->error && (s->input_value_count < s->num_input_values || s->src_pos < s->nz_count)) { + while (!s->setup.error && + (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) { uint32_t target_pos; uint32_t const src_pos = s->src_pos; if (warp.meta_group_rank() < 3) { // warp 0..2 - target_pos = min(src_pos + 2 * batch_size, s->nz_count + s->first_row + batch_size); + target_pos = min(src_pos + 2 * batch_size, s->nz_count + s->setup.first_row + batch_size); } else { // warp 3 target_pos = min(s->nz_count, src_pos + batch_size); } @@ -588,7 +594,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) suffix_db->decode_batch(); } else if (warp.meta_group_rank() == 3 and src_pos < target_pos) { // warp 3 - int const nproc = min(batch_size, s->page.end_val - string_pos); + int const nproc = min(batch_size, s->setup.page.end_val - string_pos); strings_data += use_char_ll ? dba->calculate_string_values_cp(strings_data, string_pos, nproc, warp.thread_rank()) @@ -602,7 +608,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) int dst_pos = sb->nz_idx[rolling_index(sp)]; // handle skip_rows here. flat hierarchies can just skip up to first_row. - if (!has_repetition) { dst_pos -= s->first_row; } + if (!has_repetition) { dst_pos -= s->setup.first_row; } if (dst_pos >= 0 && sp < target_pos) { auto const offptr = @@ -632,7 +638,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // For large strings, update the initial string buffer offset to be used during large string // column construction. Otherwise, convert string sizes to final offsets. - if (s->col.is_large_string_col) { + if (s->setup.col.is_large_string_col) { // page.chunk_idx are ordered by input_col_idx and row_group_idx respectively. auto const chunks_per_rowgroup = initial_str_offsets.size(); auto const input_col_idx = pages[page_idx].chunk_idx % chunks_per_rowgroup; @@ -649,7 +655,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } } - if (block.thread_rank() == 0 and s->error != 0) { set_error(s->error, error_code); } + if (block.thread_rank() == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } } // Decode page data that is DELTA_LENGTH_BYTE_ARRAY packed. This encoding consists of a @@ -690,7 +696,8 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) return; } - if (s->col.logical_type.has_value() && s->col.logical_type->type == LogicalType::DECIMAL) { + if (s->setup.col.logical_type.has_value() && + s->setup.col.logical_type->type == LogicalType::DECIMAL) { // we cannot read decimal encoded with DELTA_LENGTH_BYTE_ARRAY yet if (block.thread_rank() == 0) { set_error(static_cast(decode_error::INVALID_DATA_TYPE), error_code); @@ -698,11 +705,12 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) return; } - bool const has_repetition = s->col.max_level[level_type::REPETITION] > 0; + bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0; bool const process_nulls = should_process_nulls(s); // Capture initial valid_map_offset before any processing that might modify it - int const init_valid_map_offset = s->nesting_info[s->col.max_nesting_depth - 1].valid_map_offset; + int const init_valid_map_offset = + s->nesting_info[s->setup.col.max_nesting_depth - 1].valid_map_offset; // copying logic from gpuDecodePageData. PageNestingDecodeInfo const* nesting_info_base = s->nesting_info; @@ -715,7 +723,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) auto* const rep = reinterpret_cast(pp->lvl_decode_buf[level_type::REPETITION]); // skipped_leaf_values will always be 0 for flat hierarchies. - uint32_t const skipped_leaf_values = s->page.skipped_leaf_values; + uint32_t const skipped_leaf_values = s->setup.page.skipped_leaf_values; // initialize delta state if (block.thread_rank() == 0) { @@ -734,7 +742,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) return; } - int const leaf_level_index = s->col.max_nesting_depth - 1; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; // db->init_binary_block below resets db->values_per_mb block.sync(); @@ -742,11 +750,11 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // that has a value we need, and set string_offset to the position of the first value in the // string data block. auto const is_bounds_pg = - is_bounds_page(s->page, s->col.start_row, min_row, num_rows, has_repetition); - if (is_bounds_pg && s->page.start_val > 0) { + is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); + if (is_bounds_pg && s->setup.page.start_val > 0) { if (warp.meta_group_rank() == 0) { // string_off is only valid on thread 0 - auto const string_off = db->skip_values_and_sum(s->page.start_val); + auto const string_off = db->skip_values_and_sum(s->setup.page.start_val); // Threads in the warp might diverge and read in skip_values_and_sum // after lane 0 reinits below. warp.sync(); @@ -761,9 +769,10 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) block.sync(); } - int string_pos = has_repetition ? s->page.start_val : 0; + int string_pos = has_repetition ? s->setup.page.start_val : 0; - while (!s->error && (s->input_value_count < s->num_input_values || s->src_pos < s->nz_count)) { + while (!s->setup.error && + (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) { uint32_t target_pos; uint32_t const src_pos = s->src_pos; @@ -791,7 +800,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) db->decode_batch(); } else if (warp.meta_group_rank() == 2 && src_pos < target_pos) { // warp 2 - int const nproc = min(batch_size, s->page.end_val - string_pos); + int const nproc = min(batch_size, s->setup.page.end_val - string_pos); string_pos += nproc; // process the mini-block in batches of 32 @@ -801,7 +810,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) int dst_pos = sb->nz_idx[rolling_index(sp)]; // handle skip_rows here. flat hierarchies can just skip up to first_row. - if (!has_repetition) { dst_pos -= s->first_row; } + if (!has_repetition) { dst_pos -= s->setup.first_row; } // fill in offsets array if (dst_pos >= 0 && sp < target_pos) { @@ -830,7 +839,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // For large strings, update the initial string buffer offset to be used during large string // column construction. Otherwise, convert string sizes to final offsets. - if (s->col.is_large_string_col) { + if (s->setup.col.is_large_string_col) { // page.chunk_idx are ordered by input_col_idx and row_group_idx respectively. auto const chunks_per_rowgroup = initial_str_offsets.size(); auto const input_col_idx = pages[page_idx].chunk_idx % chunks_per_rowgroup; @@ -851,9 +860,9 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // finally, copy the string data into place auto const dst = nesting_info_base[leaf_level_index].string_out; auto const src = page_string_data + string_offset; - memcpy_block(dst, src, s->page.str_bytes, block); + memcpy_block(dst, src, s->setup.page.str_bytes, block); - if (block.thread_rank() == 0 and s->error != 0) { set_error(s->error, error_code); } + if (block.thread_rank() == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } } } // anonymous namespace diff --git a/cpp/src/io/parquet/page_string_decode.cu b/cpp/src/io/parquet/page_string_decode.cu index adc5c375b8f7..749028490c95 100644 --- a/cpp/src/io/parquet/page_string_decode.cu +++ b/cpp/src/io/parquet/page_string_decode.cu @@ -57,10 +57,10 @@ __device__ cuda::std::pair page_bounds( auto const block = cg::this_thread_block(); auto const t = block.thread_rank(); - int const max_depth = s->col.max_nesting_depth; + int const max_depth = s->setup.col.max_nesting_depth; int const max_def = s->nesting_info[max_depth - 1].max_def_level; - auto const pp = &s->page; + auto const pp = &s->setup.page; // Clamp to decoded level count; for chunked reads we may have decoded fewer values than // num_input_values (skip_rows/num_rows), and the level buffers are only that large. int const actual_num_values = @@ -71,8 +71,8 @@ __device__ cuda::std::pair page_bounds( // can skip all this if we know there are no nulls if (max_def == 0 && !is_bounds_pg) { if (t == 0) { - s->page.num_valids = actual_num_values; - s->page.num_nulls = 0; + s->setup.page.num_valids = actual_num_values; + s->setup.page.num_nulls = 0; } return {0, actual_num_values}; } @@ -93,7 +93,7 @@ __device__ cuda::std::pair page_bounds( __shared__ int end_val_idx; // need these for skip_rows case - auto const page_start_row = s->col.start_row + pp->chunk_row; + auto const page_start_row = s->setup.col.start_row + pp->chunk_row; auto const max_row = min_row + num_rows; auto const begin_row = page_start_row >= min_row ? 0 : min_row - page_start_row; auto const max_page_rows = pp->num_rows - begin_row; @@ -550,7 +550,7 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) } bool const is_bounds_pg = - is_bounds_page(s->page, s->col.start_row, min_row, num_rows, has_repetition); + is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); // if we have size info, then we only need to do this for bounds pages if (pp->has_value_info && !is_bounds_pg) { return; } @@ -572,8 +572,8 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) // need to save num_nulls and num_valids calculated in page_bounds in this page if (t == 0) { - pp->num_nulls = s->page.num_nulls; - pp->num_valids = s->page.num_valids; + pp->num_nulls = s->setup.page.num_nulls; + pp->num_valids = s->setup.page.num_valids; pp->start_val = start_value; pp->end_val = end_value; } @@ -643,12 +643,12 @@ CUDF_KERNEL void __launch_bounds__(delta_preproc_block_size) } } else { bool const is_bounds_pg = - is_bounds_page(s->page, s->col.start_row, min_row, num_rows, has_repetition); + is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); // if we have size info, then we only need to do this for bounds pages if (pp->has_value_info && !is_bounds_pg) { // check if we need to store values from the index - if (t == 0 && is_page_contained(s->page, s->col.start_row, min_row, num_rows)) { + if (t == 0 && is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows)) { pp->str_bytes = pp->str_bytes_from_index; } return; @@ -725,12 +725,12 @@ CUDF_KERNEL void __launch_bounds__(delta_length_block_size) } bool const is_bounds_pg = - is_bounds_page(s->page, s->col.start_row, min_row, num_rows, has_repetition); + is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); // if we have size info, then we only need to do this for bounds pages if (pp->has_value_info && !is_bounds_pg) { // check if we need to store values from the index - if (t == 0 && is_page_contained(s->page, s->col.start_row, min_row, num_rows)) { + if (t == 0 && is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows)) { pp->str_bytes = pp->str_bytes_from_index; } return; @@ -839,18 +839,18 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) } bool const is_bounds_pg = - is_bounds_page(s->page, s->col.start_row, min_row, num_rows, has_repetition); + is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); // if we have size info, then we only need to do this for bounds pages if (pp->has_value_info && !is_bounds_pg) { // check if we need to store values from the index - if (t == 0 && is_page_contained(s->page, s->col.start_row, min_row, num_rows)) { + if (t == 0 && is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows)) { pp->str_bytes = pp->str_bytes_from_index; } return; } - auto const& col = s->col; + auto const& col = s->setup.col; size_t str_bytes = 0; // short circuit for FIXED_LEN_BYTE_ARRAY if (col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { @@ -1319,9 +1319,9 @@ CUDF_KERNEL void preprocess_string_offsets_kernel( // We don't know how many values we'll need to read, because we don't know // how many nulls we'll skip. So we have to read through the skipped rows. // This runs before we know skipped_leaf_values so can't skip for lists either. - bool const is_list = (chunk.max_level[level_type::REPETITION] != 0); - size_t const num_values_to_process = - is_list ? pp->nesting[chunk.max_nesting_depth - 1].batch_size : s->num_rows + s->first_row; + bool const is_list = (chunk.max_level[level_type::REPETITION] != 0); + size_t const num_values_to_process = is_list ? pp->nesting[chunk.max_nesting_depth - 1].batch_size + : s->setup.num_rows + s->setup.first_row; if (num_values_to_process == 0) { return; } diff --git a/cpp/src/io/parquet/page_string_utils.cuh b/cpp/src/io/parquet/page_string_utils.cuh index 85f8bf06200e..c9a4e44d9c3c 100644 --- a/cpp/src/io/parquet/page_string_utils.cuh +++ b/cpp/src/io/parquet/page_string_utils.cuh @@ -98,17 +98,17 @@ __device__ void convert_small_string_lengths_to_offsets(page_state_s const* cons { // If this is a large string column. In the // latter case, offsets will be computed during string column creation. - auto& ni = state->nesting_info[state->col.max_nesting_depth - 1]; + auto& ni = state->nesting_info[state->setup.col.max_nesting_depth - 1]; int value_count = ni.value_count; // if no repetition we haven't calculated start/end bounds and instead just skipped // values until we reach first_row. account for that here. - if constexpr (not has_lists) { value_count -= state->first_row; } + if constexpr (not has_lists) { value_count -= state->setup.first_row; } // Convert the array of lengths into offsets if (value_count > 0) { auto const offptr = reinterpret_cast(ni.data_out); - auto const initial_value = state->page.str_offset; + auto const initial_value = state->setup.page.str_offset; block_excl_sum(offptr, value_count, initial_value); } } @@ -122,16 +122,16 @@ inline __device__ void compute_initial_large_strings_offset(page_state_s const* size_t& initial_str_offset) { // Values decoded by this page. - int value_count = state->nesting_info[state->col.max_nesting_depth - 1].value_count; + int value_count = state->nesting_info[state->setup.col.max_nesting_depth - 1].value_count; // if no repetition we haven't calculated start/end bounds and instead just skipped // values until we reach first_row. account for that here. - if constexpr (not has_lists) { value_count -= state->first_row; } + if constexpr (not has_lists) { value_count -= state->setup.first_row; } // Atomically update the initial string offset if this is a large string column. This initial // offset will be used to compute (64-bit) offsets during large string column construction. if (value_count > 0 and threadIdx.x == 0) { - auto const initial_value = state->page.str_offset; + auto const initial_value = state->setup.page.str_offset; cuda::atomic_ref initial_str_offsets_ref{ initial_str_offset}; initial_str_offsets_ref.fetch_min(initial_value, cuda::std::memory_order_relaxed); @@ -199,8 +199,8 @@ __device__ size_t decode_strings(page_state_s* s, size_t string_output_offset) { // nesting level that is storing actual leaf values - int const leaf_level_index = s->col.max_nesting_depth - 1; - int const skipped_leaf_values = s->page.skipped_leaf_values; + int const leaf_level_index = s->setup.col.max_nesting_depth - 1; + int const skipped_leaf_values = s->setup.page.skipped_leaf_values; auto const& ni = s->nesting_info[leaf_level_index]; @@ -215,10 +215,10 @@ __device__ size_t decode_strings(page_state_s* s, // Index from value buffer (doesn't include nulls) to final array (has gaps for nulls) int const dst_pos = [&]() { if constexpr (copy_mode_t == copy_mode::DIRECT) { - return thread_pos - s->first_row; + return thread_pos - s->setup.first_row; } else { int dst_pos = sb->nz_idx[rolling_index(thread_pos)]; - if constexpr (!has_lists_t) { dst_pos -= s->first_row; } + if constexpr (!has_lists_t) { dst_pos -= s->setup.first_row; } return dst_pos; } }(); @@ -243,7 +243,7 @@ __device__ size_t decode_strings(page_state_s* s, } else { int input_thread_string_offset; int string_length; - if (s->col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { + if (s->setup.col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { input_thread_string_offset = src_pos * s->dtype_len_in; string_length = s->dtype_len_in; } else { @@ -289,7 +289,7 @@ __device__ size_t decode_strings(page_state_s* s, if constexpr (split_decode_t) { if (in_range) { auto const split_string_length = s->dtype_len_in; - auto const stream_length = s->page.str_bytes / split_string_length; + auto const stream_length = s->setup.page.str_bytes / split_string_length; for (int ii = 0; ii < split_string_length; ii++) { thread_output_string[ii] = s->data_start[src_pos + ii * stream_length];