Skip to content

Consolidate protobuf scan and decode interfaces #4836

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
thirtiseven merged 12 commits into from
Aug 6, 2026
Merged

Consolidate protobuf scan and decode interfaces #4836

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
f0d1897
Consolidate protobuf scan and decode interfaces
thirtiseven Jul 17, 2026
48d80c8
add a regression tests
thirtiseven Jul 20, 2026
964ba8a
AssertUtils.assertStructColumnsAreEqual refactor
thirtiseven Jul 20, 2026
6fe0c3e
address comment
thirtiseven Jul 20, 2026
a33b516
address comments
thirtiseven Jul 23, 2026
dd80cb5
address comments
thirtiseven Jul 24, 2026
661c357
address comments
thirtiseven Jul 27, 2026
882e82d
style
thirtiseven Jul 27, 2026
65cbdf7
address comments
thirtiseven Jul 29, 2026
eaa2a8e
address comments
thirtiseven Jul 30, 2026
93f8e48
address comments
thirtiseven Aug 5, 2026
67907da
Merge branch 'main' into protobuf-refactor-consolidate-interfaces
igorpeshansky Aug 5, 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
415 changes: 159 additions & 256 deletions src/main/cpp/src/protobuf/protobuf.cu
View file
Open in desktop

Large diffs are not rendered by default.

278 changes: 111 additions & 167 deletions src/main/cpp/src/protobuf/protobuf_builders.cu
View file
Open in desktop

Large diffs are not rendered by default.

109 changes: 47 additions & 62 deletions src/main/cpp/src/protobuf/protobuf_host_helpers.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@
#include <cstdint>
#include <limits>
#include <memory>
#include <source_location>
#include <span>
#include <string>
#include <utility>
#include <vector>
Expand Down Expand Up @@ -87,26 +89,13 @@ struct field_descriptor_bundle {
field_descriptor_bundle make_field_descriptors(std::vector<int> const& field_indices,
protobuf_schema const& schema,
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr);
rmm::device_async_resource_ref mr,
std::span<int const> output_indices = {});

// ============================================================================
// Nested decode view bundles
// ============================================================================

struct protobuf_input_view {
uint8_t const* message_data;
cudf::size_type message_data_size;
cudf::size_type const* row_offsets;
cudf::size_type base_offset;
int num_rows;
};

struct nested_parent_view {
field_location const* locations;
std::size_t location_count;
int32_t const* top_row_indices;
};

struct protobuf_decode_runtime_context {
rmm::device_uvector<bool>* row_force_null;
rmm::device_uvector<protobuf_error>* error;
Expand All @@ -118,6 +107,13 @@ struct recursive_decode_context {
protobuf_decode_runtime_context runtime;
};

struct protobuf_field_decode_request {
Comment thread
igorpeshansky marked this conversation as resolved.
recursive_decode_context context;
uint8_t const* message_data;
int schema_idx;
protobuf_value_domain_view values;
};

struct list_offsets_from_counts_result {
int32_t total_count;
rmm::device_uvector<int32_t> offsets;
Expand All @@ -138,6 +134,21 @@ struct repeated_field_work {
}
};

inline void validate_nonempty_repeated_field_work(
repeated_field_work const& work,
int num_rows,
std::source_location const& location = std::source_location::current())
{
auto const caller = location.function_name();
auto const message = [caller](char const* detail) { return std::string{caller} + ": " + detail; };
CUDF_EXPECTS(work.total_count > 0, message("total count must be positive"));
CUDF_EXPECTS(work.offsets.size() == static_cast<size_t>(num_rows) + 1,
message("offsets size must match row count"));
CUDF_EXPECTS(work.occurrences != nullptr, message("repeated occurrences must be present"));
CUDF_EXPECTS(work.occurrences->size() == static_cast<size_t>(work.total_count),
message("occurrence count mismatch"));
}

template <typename CountIterator>
inline list_offsets_from_counts_result make_list_offsets_from_counts(
CountIterator counts_begin,
Expand Down Expand Up @@ -203,16 +214,6 @@ inline cudf::detail::host_vector<int> build_lookup_table(FieldNumberFn get_field
return table;
}

inline cudf::detail::host_vector<int> build_index_lookup_table(
nested_field_descriptor const* schema,
int const* field_indices,
int num_indices,
rmm::cuda_stream_view stream)
{
return build_lookup_table(
[&](int i) { return schema[field_indices[i]].field_number; }, num_indices, stream);
}

template <typename FieldDesc>
inline cudf::detail::host_vector<int> build_field_lookup_table(FieldDesc const* descs,
int num_fields,
Expand Down Expand Up @@ -346,32 +347,29 @@ std::unique_ptr<cudf::column> make_empty_struct_column_with_schema(
return make_empty_struct_column_from_children(schema, child_indices, stream, mr);
}

void maybe_check_required_fields(field_location const* locations,
void maybe_check_required_fields(required_field_input_view input,
std::vector<int> const& field_indices,
std::vector<nested_field_descriptor> const& schema,
int num_rows,
cudf::bitmask_type const* input_null_mask,
cudf::size_type input_offset,
field_location const* parent_locs,
bool* row_force_null,
int32_t const* top_row_indices,
protobuf_error* error_flag,
protobuf_decode_runtime_context decode_ctx,
rmm::cuda_stream_view stream);

void propagate_invalid_enum_flags_to_rows(rmm::device_uvector<bool> const& item_invalid,
rmm::device_uvector<bool>& row_invalid,
int num_items,
int32_t const* top_row_indices,
bool propagate_to_rows,
protobuf_decode_runtime_context decode_ctx,
protobuf_value_domain_view value_domain,
rmm::cuda_stream_view stream);

void validate_enum_and_propagate_rows(rmm::device_uvector<int32_t> const& values,
rmm::device_uvector<bool>& valid,
enum_domain_device_view enum_domain,
protobuf_decode_runtime_context decode_ctx,
protobuf_value_domain_view value_domain,
rmm::cuda_stream_view stream);

void validate_enum_and_propagate_rows(rmm::device_uvector<int32_t> const& values,
rmm::device_uvector<bool>& valid,
cudf::detail::host_vector<int32_t> const& valid_enums,
rmm::device_uvector<bool>& row_invalid,
int num_items,
int32_t const* top_row_indices,
bool propagate_to_rows,
protobuf_decode_runtime_context decode_ctx,
protobuf_value_domain_view value_domain,
rmm::cuda_stream_view stream);

// ============================================================================
Expand All @@ -398,16 +396,11 @@ std::unique_ptr<cudf::column> make_null_list_column_with_child(
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr);

std::unique_ptr<cudf::column> build_enum_string_column(
rmm::device_uvector<int32_t>& enum_values,
rmm::device_uvector<bool>& valid,
cudf::detail::host_vector<int32_t> const& valid_enums,
std::vector<cudf::detail::host_vector<uint8_t>> const& enum_name_bytes,
protobuf_decode_runtime_context decode_ctx,
int num_rows,
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr,
int32_t const* top_row_indices = nullptr);
std::unique_ptr<cudf::column> build_enum_string_column(rmm::device_uvector<int32_t>& enum_values,
rmm::device_uvector<bool>& valid,
protobuf_field_decode_request request,
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr);

// Wrap offsets + child into a LIST column, propagating the input's null mask. Note: when
// `binary_input` has no nulls, `mr` is effectively unused — only the with-nulls path
Expand All @@ -420,26 +413,18 @@ std::unique_ptr<cudf::column> make_list_column_with_input_nulls(
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr);

// `d_field_offsets` is the pre-built LIST row-offsets buffer (size num_rows + 1) from the
// orchestrator (allocated against `mr`); each builder moves it into its output column.
std::unique_ptr<cudf::column> build_repeated_enum_string_column(
cudf::column_view const& binary_input,
protobuf_input_view input,
rmm::device_uvector<int32_t> d_field_offsets,
rmm::device_uvector<field_occurrence>& d_occurrences,
int total_count,
cudf::detail::host_vector<int32_t> const& valid_enums,
std::vector<cudf::detail::host_vector<uint8_t>> const& enum_name_bytes,
protobuf_decode_runtime_context decode_ctx,
recursive_decode_context context,
repeated_field_work work,
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr);

std::unique_ptr<cudf::column> build_repeated_string_column(
cudf::column_view const& binary_input,
protobuf_input_view input,
rmm::device_uvector<int32_t> d_field_offsets,
rmm::device_uvector<field_occurrence>& d_occurrences,
int total_count,
repeated_field_work work,
bool is_bytes,
rmm::device_uvector<protobuf_error>& d_error,
rmm::cuda_stream_view stream,
Expand Down
Loading