Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
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
3 changes: 2 additions & 1 deletion cpp/src/io/avro/reader_impl.cu
Original file line number Diff line number Diff line change
Expand Up @@ -568,7 +568,8 @@ table_with_metadata read_avro(std::unique_ptr<cudf::io::datasource>&& source,
mr);

for (size_t i = 0; i < column_types.size(); ++i) {
out_columns.emplace_back(make_column(out_buffers[i], nullptr, std::nullopt, stream));
out_columns.emplace_back(
std::move(out_buffers[i]).make_column(nullptr, std::nullopt, stream));
}
} else {
// Create empty columns
Expand Down
12 changes: 6 additions & 6 deletions cpp/src/io/csv/reader_impl.cu
Original file line number Diff line number Diff line change
Expand Up @@ -1004,7 +1004,7 @@ table_with_metadata read_csv(cudf::io::datasource* source,
true);
if (num_quoted == 0) {
// Fast path: no rows were quoted, skip replacement entirely
out_columns[col_idx] = make_column(*buffer, nullptr, std::nullopt, col_stream);
out_columns[col_idx] = std::move(*buffer).make_column(nullptr, std::nullopt, col_stream);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
} else {
auto replaced_all_col = cudf::strings::detail::replace(
cudf::make_strings_column(
Expand Down Expand Up @@ -1074,15 +1074,15 @@ table_with_metadata read_csv(cudf::io::datasource* source,
}
}

// Create output columns for the columns that were not processed in the parallel loop
for (size_t i = 0; i < column_types.size(); ++i) {
if (!out_columns[i]) {
out_columns[i] = make_column(out_buffers[i], nullptr, std::nullopt, stream);
}
metadata.schema_info.emplace_back(out_buffers[i].name);
}

// Create output columns for the columns that were not processed in the parallel loop
for (size_t i = 0; i < column_types.size(); ++i) {
metadata.schema_info.emplace_back(out_buffers[i].name);
if (!out_columns[i]) {
out_columns[i] = std::move(out_buffers[i]).make_column(nullptr, std::nullopt, stream);
}
}
} else {
// Create empty columns
Expand Down
3 changes: 2 additions & 1 deletion cpp/src/io/orc/reader_impl_decode.cu
Original file line number Diff line number Diff line change
Expand Up @@ -1008,7 +1008,8 @@ void reader_impl::decompress_and_decode_stripes(read_mode mode)
_out_metadata.schema_info.emplace_back("");
auto col_buffer = assemble_buffer(
orc_col_meta.id, 0, *_col_meta, _metadata, _selected_columns, _out_buffers, _stream, _mr);
return make_column(col_buffer, &_out_metadata.schema_info.back(), std::nullopt, _stream);
return std::move(col_buffer)
.make_column(&_out_metadata.schema_info.back(), std::nullopt, _stream);
});
_chunk_read_data.decoded_table = std::make_unique<table>(std::move(out_columns));

Expand Down
6 changes: 4 additions & 2 deletions cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1312,9 +1312,11 @@ table_with_metadata hybrid_scan_reader_impl::read_chunk_internal(
// Only construct `out_metadata` if `_output_metadata` has not been cached.
if (!_output_metadata) {
column_name_info& col_name = out_metadata.schema_info[i];
out_columns.emplace_back(make_column(_output_buffers[i], &col_name, metadata, _stream));
out_columns.emplace_back(
std::move(_output_buffers[i]).make_column(&col_name, metadata, _stream));
} else {
out_columns.emplace_back(make_column(_output_buffers[i], nullptr, metadata, _stream));
out_columns.emplace_back(
std::move(_output_buffers[i]).make_column(nullptr, metadata, _stream));
}
}

Expand Down
6 changes: 4 additions & 2 deletions cpp/src/io/parquet/reader_impl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -761,9 +761,11 @@ table_with_metadata reader_impl::read_chunk_internal(read_mode mode)
// Only construct `out_metadata` if `_output_metadata` has not been cached.
if (!_output_metadata) {
column_name_info& col_name = out_metadata.schema_info[i];
out_columns.emplace_back(make_column(_output_buffers[i], &col_name, metadata, _stream));
out_columns.emplace_back(
std::move(_output_buffers[i]).make_column(&col_name, metadata, _stream));
} else {
out_columns.emplace_back(make_column(_output_buffers[i], nullptr, metadata, _stream));
out_columns.emplace_back(
std::move(_output_buffers[i]).make_column(nullptr, metadata, _stream));
}
}

Expand Down
20 changes: 9 additions & 11 deletions cpp/src/io/utilities/column_buffer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -172,10 +172,10 @@ std::string type_to_name(column_buffer_base<string_policy> const& buffer)
}

template <class string_policy>
std::unique_ptr<column> make_column(column_buffer_base<string_policy>& buffer,
column_name_info* schema_info,
std::optional<reader_column_schema> const& schema,
cuda::stream_ref stream)
std::unique_ptr<column> column_buffer_base<string_policy>::make_column(
column_name_info* schema_info,
std::optional<reader_column_schema> const& schema,
cuda::stream_ref stream) &&
{
std::function<std::unique_ptr<column>(column_buffer_base<string_policy> & buffer,
column_name_info * schema_info,
Expand Down Expand Up @@ -315,7 +315,7 @@ std::unique_ptr<column> make_column(column_buffer_base<string_policy>& buffer,
}
};

return construct_column(buffer, schema_info, schema);
return construct_column(*this, schema_info, schema);
}

/**
Expand Down Expand Up @@ -398,17 +398,15 @@ using string_type = cudf::io::detail::inline_column_buffer;
using pointer_column_buffer = column_buffer_base<pointer_type>;
using string_column_buffer = column_buffer_base<string_type>;

template std::unique_ptr<column> make_column<string_type>(
string_column_buffer& buffer,
template std::unique_ptr<column> column_buffer_base<string_type>::make_column(
column_name_info* schema_info,
std::optional<reader_column_schema> const& schema,
cuda::stream_ref stream);
cuda::stream_ref stream) &&;

template std::unique_ptr<column> make_column<pointer_type>(
pointer_column_buffer& buffer,
template std::unique_ptr<column> column_buffer_base<pointer_type>::make_column(
column_name_info* schema_info,
std::optional<reader_column_schema> const& schema,
cuda::stream_ref stream);
cuda::stream_ref stream) &&;

template std::unique_ptr<column> empty_like<string_type>(string_column_buffer& buffer,
column_name_info* schema_info,
Expand Down
48 changes: 20 additions & 28 deletions cpp/src/io/utilities/column_buffer.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -60,28 +60,6 @@ inline rmm::device_buffer create_data(data_type type,

using string_index_pair = cuda::std::pair<char const*, size_type>;

// forward declare friend functions
template <typename string_policy>
class column_buffer_base;

/**
* @brief Creates a column from an existing set of device memory buffers.
*
* @throws std::bad_alloc if device memory allocation fails
*
* @param buffer Column buffer descriptors
* @param schema_info Schema information for the column to write optionally.
* @param schema Optional schema used to control string to binary conversions.
* @param stream CUDA stream used for device memory operations and kernel launches.
*
* @return `std::unique_ptr<cudf::column>` Column from the existing device data
*/
template <class string_policy>
std::unique_ptr<column> make_column(column_buffer_base<string_policy>& buffer,
column_name_info* schema_info,
std::optional<reader_column_schema> const& schema,
cuda::stream_ref stream);

template <typename string_policy>
class column_buffer_base {
public:
Expand Down Expand Up @@ -150,6 +128,26 @@ class column_buffer_base {
return static_cast<string_policy*>(this)->make_string_column_impl(stream);
}

/**
* @brief Creates a column from an existing set of device memory buffers.
*
* The buffer contents are moved into the returned column where possible; the buffer must be
* treated as consumed and must not be reused after the call. The method is only callable on
* rvalues so that call sites make the consumption explicit via `std::move`.
*
* @throws std::bad_alloc if device memory allocation fails
*
* @param schema_info Schema information for the column to write optionally.
* @param schema Optional schema used to control string to binary conversions.
* @param stream CUDA stream used for device memory operations and kernel launches.
*
* @return Column from the existing device data
*/
[[nodiscard]] std::unique_ptr<column> make_column(
column_name_info* schema_info,
std::optional<reader_column_schema> const& schema,
cuda::stream_ref stream) &&;

protected:
rmm::device_buffer _data{};
rmm::device_buffer _null_mask{};
Expand All @@ -166,12 +164,6 @@ class column_buffer_base {
std::string name;

std::vector<string_policy> children;

friend std::unique_ptr<column> make_column<string_policy>(
column_buffer_base& buffer,
column_name_info* schema_info,
std::optional<reader_column_schema> const& schema,
cuda::stream_ref stream);
};

// column buffer that uses a string_index_pair for strings data, requiring a gather step when
Expand Down
1 change: 1 addition & 0 deletions cpp/tests/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -316,6 +316,7 @@ ConfigureTest(
# ##################################################################################################
# * io tests --------------------------------------------------------------------------------------
ConfigureTest(COMPRESSION_TEST io/comp/comp_test.cpp)
ConfigureTest(COLUMN_BUFFER_TEST io/column_buffer_test.cpp)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
ConfigureTest(ROW_SELECTION_TEST io/row_selection_test.cpp)
ConfigureTest(FILEPATH_SOURCE_TEST io/filepath_source_test.cpp)

Expand Down
64 changes: 64 additions & 0 deletions cpp/tests/io/column_buffer_test.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
/*
* SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*/

#include "io/utilities/column_buffer.hpp"

#include <cudf_test/base_fixture.hpp>
#include <cudf_test/column_utilities.hpp>
#include <cudf_test/column_wrapper.hpp>
#include <cudf_test/cudf_gtest.hpp>

#include <cudf/types.hpp>

#include <cstring>
#include <vector>

struct ColumnBufferTest : public cudf::test::BaseFixture {};

TEST_F(ColumnBufferTest, MakeColumnFromRvalueFixedWidth)
{
auto const stream = cudf::get_default_stream();
auto const mr = cudf::get_current_device_resource_ref();

cudf::io::detail::inline_column_buffer buffer{cudf::data_type{cudf::type_id::INT32}, false};
buffer.create(4, stream, mr);

std::vector<int32_t> const host_values{1, 2, 3, 4};
CUDF_CUDA_TRY(cudaMemcpyAsync(buffer.data(),
host_values.data(),
host_values.size() * sizeof(int32_t),
cudaMemcpyDefault,
stream.value()));
stream.synchronize();

auto column = std::move(buffer).make_column(nullptr, std::nullopt, stream);

cudf::test::fixed_width_column_wrapper<int32_t> const expected{{1, 2, 3, 4}};
CUDF_TEST_EXPECT_COLUMNS_EQUAL(column->view(), expected);
}

TEST_F(ColumnBufferTest, MakeColumnFromRvalueNullable)
{
auto const stream = cudf::get_default_stream();
auto const mr = cudf::get_current_device_resource_ref();

cudf::test::fixed_width_column_wrapper<int8_t> const expected{{1, 2, 0, 4}, {1, 1, 0, 1}};

cudf::io::detail::inline_column_buffer buffer{cudf::data_type{cudf::type_id::INT8}, true};
buffer.create_with_mask(4, cudf::mask_state::ALL_VALID, false, stream, mr);

int8_t const host_values[]{1, 2, 3, 4};
CUDF_CUDA_TRY(cudaMemcpyAsync(
buffer.data(), host_values, sizeof(host_values), cudaMemcpyDefault, stream.value()));
uint32_t const valid_bits{0b00001011u};
CUDF_CUDA_TRY(cudaMemcpyAsync(
buffer.null_mask(), &valid_bits, sizeof(valid_bits), cudaMemcpyDefault, stream.value()));
buffer.null_count() = 1;
stream.synchronize();

auto column = std::move(buffer).make_column(nullptr, std::nullopt, stream);

CUDF_TEST_EXPECT_COLUMNS_EQUAL(column->view(), expected);
}
Loading