Skip to content

refactor(parquet): extract page_decode_setup_state substruct #23471

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rapids-bot merged 2 commits into from
Jul 30, 2026
Merged

refactor(parquet): extract page_decode_setup_state substruct #23471

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3a6b48e
refactor(parquet): extract page_decode_setup_state substruct
vyasr Jul 29, 2026
9e45f1a
fix(parquet): guard delta byte array heuristic for all-null pages
vyasr Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 49 additions & 43 deletions cpp/src/io/parquet/decode_fixed.cu
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,24 +89,24 @@ __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;
while (thread_pos < end) {
// 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<state_buf::nz_buf_size>(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;
}
}();
Expand All @@ -124,7 +124,8 @@ __device__ void decode_fixed_width_values(

void* const dst = data_out + (static_cast<size_t>(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<uint32_t*>(dst));
Expand Down Expand Up @@ -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
Expand All @@ -193,18 +194,18 @@ __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;
while (thread_pos < end) {
// 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<state_buf::nz_buf_size>(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;
}
}();
Expand All @@ -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<size_t>(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) {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand All @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -1055,8 +1056,12 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8)

rle_stream<uint32_t, decode_block_size_t, rolling_buf_size> 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
Expand All @@ -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();
Expand All @@ -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_decoding<decode_block_size_t,
Expand All @@ -1101,13 +1107,13 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8)
// the core loop. decode batches of level stream data using rle_stream objects
// and pass the results to decode_values
// For chunked reads we may not process all of the rows on the page; if not stop early
int const first_row = s->first_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) {
Expand Down Expand Up @@ -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; }
}
Expand All @@ -1162,7 +1168,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8)
auto decode_values = [&]<copy_mode copy_mode_t>() {
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<decode_block_size_t, has_dict_t, has_lists_t, split_decode_t, copy_mode_t>(
s, sb, valid_count, next_valid_count, t, str_offsets, string_output_offset);
Expand Down Expand Up @@ -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<decode_block_size_t>(
Expand All @@ -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;
Expand All @@ -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
Expand Down
Loading
Loading