diff --git a/cpp/src/io/parquet/decode_fixed.cu b/cpp/src/io/parquet/decode_fixed.cu index 75113dfbf840..34e7e0ffaaa9 100644 --- a/cpp/src/io/parquet/decode_fixed.cu +++ b/cpp/src/io/parquet/decode_fixed.cu @@ -159,7 +159,7 @@ __device__ void decode_fixed_width_values( auto const data_out = s->nesting_info[leaf_level_index].data_out; Type const dtype = s->setup.col.physical_type; - uint32_t const dtype_len = s->dtype_len; + uint32_t const dtype_len = s->output_cvt.dtype_len; int const skipped_leaf_values = s->setup.page.skipped_leaf_values; @@ -200,9 +200,9 @@ __device__ void decode_fixed_width_values( read_fixed_width_value_fast(s, sb, src_pos, static_cast(dst)); break; default: - if (s->dtype_len_in <= sizeof(int32_t)) { + if (s->output_cvt.dtype_len_in <= sizeof(int32_t)) { read_fixed_width_byte_array_as_int(s, sb, src_pos, static_cast(dst)); - } else if (s->dtype_len_in <= sizeof(int64_t)) { + } else if (s->output_cvt.dtype_len_in <= sizeof(int64_t)) { read_fixed_width_byte_array_as_int(s, sb, src_pos, static_cast(dst)); } else { read_fixed_width_byte_array_as_int(s, sb, src_pos, static_cast<__int128_t*>(dst)); @@ -214,7 +214,7 @@ __device__ void decode_fixed_width_values( } else if (dtype == Type::INT96) { read_int96_timestamp(s, sb, src_pos, static_cast(dst)); } else if (dtype_len == 8) { - if (s->dtype_len_in == 4) { + if (s->output_cvt.dtype_len_in == 4) { // Reading INT32 TIME_MILLIS into 64-bit DURATION_MILLISECONDS // TIME_MILLIS is the only duration type stored as int32: // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#deprecated-time-convertedtype @@ -222,7 +222,7 @@ __device__ void decode_fixed_width_values( read_fixed_width_value_fast(s, sb, src_pos, dst_ptr); // zero out most significant bytes cuda::std::memset(dst_ptr + 1, 0, sizeof(int32_t)); - } else if (s->ts_scale) { + } else if (s->output_cvt.ts_scale) { read_int64_timestamp(s, sb, src_pos, static_cast(dst)); } else { read_fixed_width_value_fast(s, sb, src_pos, static_cast(dst)); @@ -253,12 +253,12 @@ __device__ inline void decode_fixed_width_split_values( auto const data_len = cuda::std::distance(s->stream.data_start, s->stream.data_end); // Check malformed BYTE_STREAM_SPLIT pages - if (s->dtype_len_in <= 0 or data_len <= 0) { + if (s->output_cvt.dtype_len_in <= 0 or data_len <= 0) { if (t == 0) { s->set_error_code(decode_error::INVALID_BYTE_STREAM_SPLIT_SIZE); } return; } - auto const num_values = data_len / s->dtype_len_in; + auto const num_values = data_len / s->output_cvt.dtype_len_in; int const skipped_leaf_values = s->setup.page.skipped_leaf_values; @@ -288,7 +288,7 @@ __device__ inline void decode_fixed_width_split_values( } }(); - uint32_t const dtype_len = s->dtype_len; + uint32_t const dtype_len = s->output_cvt.dtype_len; uint8_t const* const src = s->stream.data_start + src_pos; uint8_t* const dst = data_out + static_cast(dst_pos) * dtype_len; auto const is_decimal = s->setup.col.logical_type.has_value() and @@ -300,17 +300,17 @@ __device__ inline void decode_fixed_width_split_values( case Type::INT32: gpuOutputByteStreamSplit(dst, src, num_values); break; case Type::INT64: gpuOutputByteStreamSplit(dst, src, num_values); break; case Type::FIXED_LEN_BYTE_ARRAY: - if (s->dtype_len_in <= sizeof(int32_t)) { + if (s->output_cvt.dtype_len_in <= sizeof(int32_t)) { gpuOutputSplitFixedLenByteArrayAsInt( - reinterpret_cast(dst), src, num_values, s->dtype_len_in); + reinterpret_cast(dst), src, num_values, s->output_cvt.dtype_len_in); break; - } else if (s->dtype_len_in <= sizeof(int64_t)) { + } else if (s->output_cvt.dtype_len_in <= sizeof(int64_t)) { gpuOutputSplitFixedLenByteArrayAsInt( - reinterpret_cast(dst), src, num_values, s->dtype_len_in); + reinterpret_cast(dst), src, num_values, s->output_cvt.dtype_len_in); break; - } else if (s->dtype_len_in <= sizeof(__int128_t)) { + } else if (s->output_cvt.dtype_len_in <= sizeof(__int128_t)) { gpuOutputSplitFixedLenByteArrayAsInt( - reinterpret_cast<__int128_t*>(dst), src, num_values, s->dtype_len_in); + reinterpret_cast<__int128_t*>(dst), src, num_values, s->output_cvt.dtype_len_in); break; } // unsupported decimal precision @@ -319,15 +319,16 @@ __device__ inline void decode_fixed_width_split_values( default: s->set_error_code(decode_error::UNSUPPORTED_ENCODING); } } else if (dtype_len == 8) { - if (s->dtype_len_in == 4) { + if (s->output_cvt.dtype_len_in == 4) { // Reading INT32 TIME_MILLIS into 64-bit DURATION_MILLISECONDS // TIME_MILLIS is the only duration type stored as int32: // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#deprecated-time-convertedtype gpuOutputByteStreamSplit(dst, src, num_values); // zero out most significant bytes cuda::std::memset(dst + sizeof(int32_t), 0, sizeof(int32_t)); - } else if (s->ts_scale) { - gpuOutputSplitInt64Timestamp(reinterpret_cast(dst), src, num_values, s->ts_scale); + } else if (s->output_cvt.ts_scale) { + gpuOutputSplitInt64Timestamp( + reinterpret_cast(dst), src, num_values, s->output_cvt.ts_scale); } else { gpuOutputByteStreamSplit(dst, src, num_values); } @@ -1306,7 +1307,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8) uint32_t const dtype_len = [&]() -> uint32_t { if constexpr (is_dict_int32_t) { return sizeof(int32_t); } if constexpr (has_strings_t) { return sizeof(cudf::size_type); } - return s->dtype_len; + return s->output_cvt.dtype_len; }(); int const num_values = [&]() { if constexpr (has_lists_t) { diff --git a/cpp/src/io/parquet/page_data.cu b/cpp/src/io/parquet/page_data.cu index ce963ecd4f15..ea2d8bea8301 100644 --- a/cpp/src/io/parquet/page_data.cu +++ b/cpp/src/io/parquet/page_data.cu @@ -82,10 +82,10 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) bool const process_nulls = should_process_nulls(s); auto const data_len = cuda::std::distance(s->stream.data_start, s->stream.data_end); - auto const num_values = data_len / s->dtype_len_in; + auto const num_values = data_len / s->output_cvt.dtype_len_in; // Check malformed BYTE_STREAM_SPLIT pages - if (s->dtype_len_in <= 0 or data_len <= 0) { + if (s->output_cvt.dtype_len_in <= 0 or data_len <= 0) { cg::invoke_one(block, [&]() { set_error(static_cast(decode_error::INVALID_BYTE_STREAM_SPLIT_SIZE), error_code); @@ -162,7 +162,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // nesting level that is storing actual leaf values int leaf_level_index = s->setup.col.max_nesting_depth - 1; - uint32_t dtype_len = s->dtype_len; + uint32_t dtype_len = s->output_cvt.dtype_len; uint8_t const* src = s->stream.data_start + val_src_pos; uint8_t* dst = nesting_info_base[leaf_level_index].data_out + static_cast(dst_pos) * dtype_len; @@ -175,17 +175,17 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) case Type::INT32: gpuOutputByteStreamSplit(dst, src, num_values); break; case Type::INT64: gpuOutputByteStreamSplit(dst, src, num_values); break; case Type::FIXED_LEN_BYTE_ARRAY: - if (s->dtype_len_in <= sizeof(int32_t)) { + if (s->output_cvt.dtype_len_in <= sizeof(int32_t)) { gpuOutputSplitFixedLenByteArrayAsInt( - reinterpret_cast(dst), src, num_values, s->dtype_len_in); + reinterpret_cast(dst), src, num_values, s->output_cvt.dtype_len_in); break; - } else if (s->dtype_len_in <= sizeof(int64_t)) { + } else if (s->output_cvt.dtype_len_in <= sizeof(int64_t)) { gpuOutputSplitFixedLenByteArrayAsInt( - reinterpret_cast(dst), src, num_values, s->dtype_len_in); + reinterpret_cast(dst), src, num_values, s->output_cvt.dtype_len_in); break; - } else if (s->dtype_len_in <= sizeof(__int128_t)) { + } else if (s->output_cvt.dtype_len_in <= sizeof(__int128_t)) { gpuOutputSplitFixedLenByteArrayAsInt( - reinterpret_cast<__int128_t*>(dst), src, num_values, s->dtype_len_in); + reinterpret_cast<__int128_t*>(dst), src, num_values, s->output_cvt.dtype_len_in); break; } // unsupported decimal precision @@ -194,16 +194,16 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) default: s->set_error_code(decode_error::UNSUPPORTED_ENCODING); } } else if (dtype_len == 8) { - if (s->dtype_len_in == 4) { + if (s->output_cvt.dtype_len_in == 4) { // Reading INT32 TIME_MILLIS into 64-bit DURATION_MILLISECONDS // TIME_MILLIS is the only duration type stored as int32: // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#deprecated-time-convertedtype gpuOutputByteStreamSplit(dst, src, num_values); // zero out most significant bytes memset(dst + 4, 0, 4); - } else if (s->ts_scale) { + } else if (s->output_cvt.ts_scale) { gpuOutputSplitInt64Timestamp( - reinterpret_cast(dst), src, num_values, s->ts_scale); + reinterpret_cast(dst), src, num_values, s->output_cvt.ts_scale); } else { gpuOutputByteStreamSplit(dst, src, num_values); } @@ -227,8 +227,11 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) 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; - zero_fill_null_positions_shared( - s, s->dtype_len, init_valid_map_offset, num_values, static_cast(block.thread_rank())); + zero_fill_null_positions_shared(s, + s->output_cvt.dtype_len, + init_valid_map_offset, + num_values, + static_cast(block.thread_rank())); } } @@ -399,7 +402,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // nesting level that is storing actual leaf values int const leaf_level_index = s->setup.col.max_nesting_depth - 1; - uint32_t const dtype_len = s->dtype_len; + uint32_t const dtype_len = s->output_cvt.dtype_len; void* dst = nesting_info_base[leaf_level_index].data_out + static_cast(dst_pos) * dtype_len; @@ -428,9 +431,9 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) read_fixed_width_value_fast(s, sb, val_src_pos, static_cast(dst)); break; default: - if (s->dtype_len_in <= sizeof(int32_t)) { + if (s->output_cvt.dtype_len_in <= sizeof(int32_t)) { read_fixed_width_byte_array_as_int(s, sb, val_src_pos, static_cast(dst)); - } else if (s->dtype_len_in <= sizeof(int64_t)) { + } else if (s->output_cvt.dtype_len_in <= sizeof(int64_t)) { read_fixed_width_byte_array_as_int(s, sb, val_src_pos, static_cast(dst)); } else { read_fixed_width_byte_array_as_int( @@ -443,7 +446,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } else if (dtype == Type::INT96) { read_int96_timestamp(s, sb, val_src_pos, static_cast(dst)); } else if (dtype_len == 8) { - if (s->dtype_len_in == 4) { + if (s->output_cvt.dtype_len_in == 4) { // Reading INT32 TIME_MILLIS into 64-bit DURATION_MILLISECONDS // TIME_MILLIS is the only duration type stored as int32: // https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#deprecated-time-convertedtype @@ -451,7 +454,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) read_fixed_width_value_fast(s, sb, val_src_pos, dst_ptr); // zero out most significant bytes cuda::std::memset(dst_ptr + 1, 0, sizeof(int32_t)); - } else if (s->ts_scale) { + } else if (s->output_cvt.ts_scale) { read_int64_timestamp(s, sb, val_src_pos, static_cast(dst)); } else { read_fixed_width_value_fast(s, sb, val_src_pos, static_cast(dst)); @@ -475,8 +478,11 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) 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( - s, s->dtype_len, init_valid_map_offset, num_values, static_cast(block.thread_rank())); + zero_fill_null_positions_shared(s, + s->output_cvt.dtype_len, + init_valid_map_offset, + num_values, + static_cast(block.thread_rank())); } } diff --git a/cpp/src/io/parquet/page_data.cuh b/cpp/src/io/parquet/page_data.cuh index ebc09cee296d..d3adebca1153 100644 --- a/cpp/src/io/parquet/page_data.cuh +++ b/cpp/src/io/parquet/page_data.cuh @@ -133,7 +133,7 @@ inline __device__ void read_int96_timestamp(page_state_s* s, dict_pos = src_pos; src8 = s->stream.data_start; } - dict_pos *= (uint32_t)s->dtype_len_in; + dict_pos *= (uint32_t)s->output_cvt.dtype_len_in; ofs = 3 & reinterpret_cast(src8); src8 -= ofs; // align to 32-bit boundary ofs <<= 3; // bytes -> bits @@ -208,7 +208,7 @@ inline __device__ void read_int64_timestamp(page_state_s* s, dict_pos = src_pos; src8 = s->stream.data_start; } - dict_pos *= (uint32_t)s->dtype_len_in; + dict_pos *= (uint32_t)s->output_cvt.dtype_len_in; ofs = 3 & reinterpret_cast(src8); src8 -= ofs; // align to 32-bit boundary ofs <<= 3; // bytes -> bits @@ -226,7 +226,7 @@ inline __device__ void read_int64_timestamp(page_state_s* s, val <<= 32; val |= v.x; // Output to desired clock rate - ts = apply_ts_scale(val, s->ts_scale); + ts = apply_ts_scale(val, s->output_cvt.ts_scale); } else { ts = 0; } @@ -269,7 +269,7 @@ __device__ void read_fixed_width_byte_array_as_int(page_state_s* s, int src_pos, T* dst) { - uint32_t const dtype_len_in = s->dtype_len_in; + uint32_t const dtype_len_in = s->output_cvt.dtype_len_in; uint8_t const* data = s->stream.dict_base ? s->stream.dict_base : s->stream.data_start; uint32_t const pos = (s->stream.dict_base @@ -321,7 +321,7 @@ inline __device__ void read_fixed_width_value_fast(page_state_s* s, dict_pos = src_pos; dict = s->stream.data_start; } - dict_pos *= (uint32_t)s->dtype_len_in; + dict_pos *= (uint32_t)s->output_cvt.dtype_len_in; gpuStoreOutput(dst, dict, dict_pos, dict_size); } @@ -352,7 +352,7 @@ inline __device__ void read_nbyte_fixed_width_value( dict_pos = src_pos; dict = s->stream.data_start; } - dict_pos *= (uint32_t)s->dtype_len_in; + dict_pos *= (uint32_t)s->output_cvt.dtype_len_in; if (len & 3) { // Generic slow path for (unsigned int i = 0; i < len; i++) { diff --git a/cpp/src/io/parquet/page_decode.cuh b/cpp/src/io/parquet/page_decode.cuh index a0cd9df7d537..d4c9dbca8230 100644 --- a/cpp/src/io/parquet/page_decode.cuh +++ b/cpp/src/io/parquet/page_decode.cuh @@ -57,15 +57,25 @@ struct page_decode_progress_state { int32_t row_index_lower_bound{}; }; +// Output conversion scratch: values written by setup_local_page_info and read +// by decode kernels to shape the output data type / timestamp scale. Grouped +// so passes that need only the string byte size scan can pull in this subset +// (see page_state_composed.cuh). +struct page_decode_output_state { + 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 ts_scale{}; // timestamp scale: <0: divide by -ts_scale, >0: multiply by ts_scale +}; + struct page_state_s { CUDF_HOST_DEVICE constexpr page_state_s() noexcept {} page_decode_setup_state setup{}; page_decode_stream_state stream{}; - 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 + page_decode_output_state output_cvt{}; + uint8_t const* lvl_end{}; + int32_t first_output_value{}; // First value in page to output // (leaf) value decoding - int32_t ts_scale{}; // timestamp scale: <0: divide by -ts_scale, >0: multiply by ts_scale page_decode_progress_state progress{}; // repetition/definition level decoding @@ -588,7 +598,7 @@ __device__ size_type initialize_string_descriptors(page_state_s* s, // All group threads can participate for fixed len byte arrays. if (s->setup.col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { - int const dtype_len_in = s->dtype_len_in; + int const dtype_len_in = s->output_cvt.dtype_len_in; total_len = min((target_pos - pos) * dtype_len_in, dict_size - s->stream.dict_val); if constexpr (sizes_only == is_calc_sizes_only::NO) { for (pos += t, k += t * dtype_len_in; pos < target_pos; pos += group.size()) { @@ -1225,30 +1235,30 @@ inline __device__ bool setup_local_page_info(auto* const s, 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; - if constexpr (requires { s->dtype_len; }) { - s->ts_scale = 0; + if constexpr (requires { s->output_cvt.dtype_len; }) { + s->output_cvt.ts_scale = 0; // Validate data type 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 + s->output_cvt.dtype_len = 1; // Boolean are stored as 1 byte on the output break; case Type::INT32: [[fallthrough]]; - case Type::FLOAT: s->dtype_len = 4; break; + case Type::FLOAT: s->output_cvt.dtype_len = 4; break; case Type::INT64: if (s->setup.col.ts_clock_rate) { - s->ts_scale = + s->output_cvt.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::DOUBLE: s->output_cvt.dtype_len = 8; break; + case Type::INT96: s->output_cvt.dtype_len = 12; break; case Type::BYTE_ARRAY: if (is_decimal) { auto const decimal_precision = s->setup.col.logical_type->precision(); - s->dtype_len = [decimal_precision]() { + s->output_cvt.dtype_len = [decimal_precision]() { if (decimal_precision <= MAX_DECIMAL32_PRECISION) { return sizeof(int32_t); } else if (decimal_precision <= MAX_DECIMAL64_PRECISION) { @@ -1258,19 +1268,21 @@ inline __device__ bool setup_local_page_info(auto* const s, } }(); } else { - s->dtype_len = sizeof(string_index_pair); + s->output_cvt.dtype_len = sizeof(string_index_pair); } break; default: // FIXED_LEN_BYTE_ARRAY: - s->dtype_len = s->setup.col.type_length; - if (s->dtype_len <= 0) { s->set_error_code(decode_error::INVALID_DATA_TYPE); } + s->output_cvt.dtype_len = s->setup.col.type_length; + if (s->output_cvt.dtype_len <= 0) { + s->set_error_code(decode_error::INVALID_DATA_TYPE); + } break; } // Special check for downconversions - s->dtype_len_in = s->dtype_len; + s->output_cvt.dtype_len_in = s->output_cvt.dtype_len; if (data_type == Type::FIXED_LEN_BYTE_ARRAY) { if (is_decimal) { - s->dtype_len = [dtype_len = s->dtype_len]() { + s->output_cvt.dtype_len = [dtype_len = s->output_cvt.dtype_len]() { if (dtype_len <= sizeof(int32_t)) { return sizeof(int32_t); } else if (dtype_len <= sizeof(int64_t)) { @@ -1280,23 +1292,23 @@ inline __device__ bool setup_local_page_info(auto* const s, } }(); } else { - s->dtype_len = sizeof(string_index_pair); + s->output_cvt.dtype_len = sizeof(string_index_pair); } } else if (data_type == Type::INT32) { // check for smaller bitwidths 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; + s->output_cvt.dtype_len = lt.bit_width() / 8; } else if (lt.is_time_millis()) { // cudf outputs as INT64 - s->dtype_len = 8; + s->output_cvt.dtype_len = 8; } } } else if (data_type == Type::BYTE_ARRAY && s->setup.col.is_strings_to_cat) { - s->dtype_len = 4; // HASH32 output + s->output_cvt.dtype_len = 4; // HASH32 output } else if (data_type == Type::INT96) { - s->dtype_len = 8; // Convert to 64-bit timestamp + s->output_cvt.dtype_len = 8; // Convert to 64-bit timestamp } // during the decoding step we need to offset the global output buffers @@ -1309,51 +1321,55 @@ inline __device__ bool setup_local_page_info(auto* const s, // 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->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->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 - else { - output_offset = nesting_info->page_start_value; - } - - 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]); + if constexpr (requires { s->nesting_info; }) { + if (stage == page_processing_stage::DECODE) { + 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->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 + else { + output_offset = nesting_info->page_start_value; } - nesting_info->data_out = static_cast(s->setup.col.column_data_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]); + } - if (nesting_info->data_out != nullptr) { - // anything below max depth with a valid data pointer must be a list, so the - // element size is the size of the offset type. - 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->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->setup.page.str_offset; - } - 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); + 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 + // element size is the size of the offset type. + uint32_t len = + idx < max_depth - 1 ? sizeof(cudf::size_type) : s->output_cvt.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->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->setup.page.str_offset; + } + 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); + } } } } - } - } + } // if constexpr (requires { s->nesting_info; }) + if constexpr (requires { s->first_output_value; }) { s->first_output_value = 0; } + } // if constexpr (requires { s->output_cvt.dtype_len; }) // Find the compressed size of repetition levels cur += InitLevelSection(s, cur, end, level_type::REPETITION); diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index 98dc64ab154f..c51bae8f97a8 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -414,9 +414,10 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // place value for this thread if (dst_pos >= 0 && sp < target_pos) { - void* const dst = nesting_info_base[leaf_level_index].data_out + dst_pos * s->dtype_len; - auto const val = db->value_at(sp + skipped_leaf_values); - switch (s->dtype_len) { + void* const dst = + nesting_info_base[leaf_level_index].data_out + dst_pos * s->output_cvt.dtype_len; + auto const val = db->value_at(sp + skipped_leaf_values); + switch (s->output_cvt.dtype_len) { case 1: *static_cast(dst) = val; break; case 2: *static_cast(dst) = val; break; case 4: *static_cast(dst) = val; break; @@ -435,8 +436,11 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) 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( - s, s->dtype_len, init_valid_map_offset, num_values, static_cast(block.thread_rank())); + zero_fill_null_positions_shared(s, + s->output_cvt.dtype_len, + init_valid_map_offset, + num_values, + static_cast(block.thread_rank())); } } diff --git a/cpp/src/io/parquet/page_state_composed.cuh b/cpp/src/io/parquet/page_state_composed.cuh index 33c3cf0efe3b..cb2243e3fb88 100644 --- a/cpp/src/io/parquet/page_state_composed.cuh +++ b/cpp/src/io/parquet/page_state_composed.cuh @@ -32,6 +32,18 @@ struct level_scan_state { CUDF_PARQUET_PAGE_STATE_ERROR_METHODS }; +// Shared memory state struct used by the compute_page_string_sizes, +// compute_delta_page_string_sizes, and compute_delta_length_page_string_sizes kernels. Includes +// setup (page metadata + error), stream (page bytes + dictionary), and output_cvt (dtype_len_in) +// because these scans read the conversion scratch written by setup_local_page_info for +// FIXED_LEN_BYTE_ARRAY size math but never walk nesting info or track progress counters. +struct string_size_scan_state { + page_decode_setup_state setup; + page_decode_stream_state stream; + page_decode_output_state output_cvt; + CUDF_PARQUET_PAGE_STATE_ERROR_METHODS +}; + // Shared memory state struct used by the preprocess_string_offsets kernel. // Includes setup (page metadata + error), stream (page bytes + dictionary), and // progress (input counters) because this pass scans flat string payloads while tracking counts. @@ -41,8 +53,6 @@ struct string_offset_scan_state { page_decode_progress_state progress; CUDF_PARQUET_PAGE_STATE_ERROR_METHODS }; -static_assert(sizeof(string_offset_scan_state) < sizeof(page_state_s), - "string_offset_scan_state did not shrink after removing output conversion state"); #undef CUDF_PARQUET_PAGE_STATE_ERROR_METHODS } // namespace cudf::io::parquet::detail diff --git a/cpp/src/io/parquet/page_string_decode.cu b/cpp/src/io/parquet/page_string_decode.cu index 4d2428b44a5c..1f47b374bf1b 100644 --- a/cpp/src/io/parquet/page_string_decode.cu +++ b/cpp/src/io/parquet/page_string_decode.cu @@ -599,12 +599,12 @@ CUDF_KERNEL void __launch_bounds__(delta_preproc_block_size) size_t min_row, size_t num_rows) { - __shared__ __align__(16) page_state_s state_g; + __shared__ __align__(16) string_size_scan_state state_g; - page_state_s* const s = &state_g; - int const page_idx = blockIdx.x; - int const t = threadIdx.x; - PageInfo* const pp = &pages[page_idx]; + auto* const s = &state_g; + int const page_idx = blockIdx.x; + int const t = threadIdx.x; + PageInfo* const pp = &pages[page_idx]; // whether or not we have repetition levels (lists) bool const has_repetition = chunks[pp->chunk_idx].max_level[level_type::REPETITION] > 0; @@ -630,7 +630,7 @@ CUDF_KERNEL void __launch_bounds__(delta_preproc_block_size) // if data size is known, can short circuit here if (chunks[pp->chunk_idx].physical_type == Type::FIXED_LEN_BYTE_ARRAY) { if (t == 0) { - pp->str_bytes = pp->num_valids * s->dtype_len_in; + pp->str_bytes = pp->num_valids * s->output_cvt.dtype_len_in; // only need temp space if we're skipping values if (start_value > 0) { @@ -638,7 +638,7 @@ CUDF_KERNEL void __launch_bounds__(delta_preproc_block_size) delta_binary_decoder db; db.init_binary_block(s->stream.data_start, s->stream.data_end); // save enough for one mini-block plus some extra to save the last_string - pp->temp_string_size = s->dtype_len_in * (db.values_per_mb + 1); + pp->temp_string_size = s->output_cvt.dtype_len_in * (db.values_per_mb + 1); } } } else { @@ -695,13 +695,13 @@ CUDF_KERNEL void __launch_bounds__(delta_length_block_size) using cudf::detail::warp_size; using WarpReduce = cub::WarpReduce; __shared__ typename WarpReduce::TempStorage temp_storage; - __shared__ __align__(16) page_state_s state_g; + __shared__ __align__(16) string_size_scan_state state_g; __shared__ __align__(16) delta_binary_decoder string_lengths; - page_state_s* const s = &state_g; - int const page_idx = blockIdx.x; - int const t = threadIdx.x; - PageInfo* const pp = &pages[page_idx]; + auto* const s = &state_g; + int const page_idx = blockIdx.x; + int const t = threadIdx.x; + PageInfo* const pp = &pages[page_idx]; // whether or not we have repetition levels (lists) bool const has_repetition = chunks[pp->chunk_idx].max_level[level_type::REPETITION] > 0; @@ -810,12 +810,12 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) size_t min_row, size_t num_rows) { - __shared__ __align__(16) page_state_s state_g; + __shared__ __align__(16) string_size_scan_state state_g; - page_state_s* const s = &state_g; - int const page_idx = blockIdx.x; - int const t = threadIdx.x; - PageInfo* const pp = &pages[page_idx]; + auto* const s = &state_g; + int const page_idx = blockIdx.x; + int const t = threadIdx.x; + PageInfo* const pp = &pages[page_idx]; // whether or not we have repetition levels (lists) bool const has_repetition = chunks[pp->chunk_idx].max_level[level_type::REPETITION] > 0; @@ -852,7 +852,7 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size) size_t str_bytes = 0; // short circuit for FIXED_LEN_BYTE_ARRAY if (col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) { - str_bytes = pp->num_valids * s->dtype_len_in; + str_bytes = pp->num_valids * s->output_cvt.dtype_len_in; } else { // now process string info in the range [start_value, end_value) // set up for decoding strings...can be either plain or dictionary diff --git a/cpp/src/io/parquet/page_string_utils.cuh b/cpp/src/io/parquet/page_string_utils.cuh index 42979dad1355..8233121457ff 100644 --- a/cpp/src/io/parquet/page_string_utils.cuh +++ b/cpp/src/io/parquet/page_string_utils.cuh @@ -244,8 +244,8 @@ __device__ size_t decode_strings(page_state_s* s, int input_thread_string_offset; int string_length; 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; + input_thread_string_offset = src_pos * s->output_cvt.dtype_len_in; + string_length = s->output_cvt.dtype_len_in; } else { input_thread_string_offset = str_offsets[src_pos]; int const next_offset = str_offsets[src_pos + 1]; @@ -288,7 +288,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 split_string_length = s->output_cvt.dtype_len_in; auto const stream_length = s->setup.page.str_bytes / split_string_length; for (int ii = 0; ii < split_string_length; ii++) {