refactor(streaming): flatten namespaces and rename to snake_case

cpp/libcudf_streaming/CMakeLists.txt

@@ -71,14 +71,8 @@ include(../cmake/thirdparty/get_cucollections.cmake)
 # * library target --------------------------------------------------------------------------------
 add_library(
   cudf_streaming SHARED
-  src/integrations/bloom_filter.cu
-  src/integrations/partition.cpp
-  src/integrations/utils.cpp
-  src/streaming/bloom_filter.cpp
-  src/streaming/channel_metadata.cpp
-  src/streaming/parquet.cpp
-  src/streaming/partition.cpp
-  src/streaming/table_chunk.cpp
+  src/bloom_filter.cpp src/channel_metadata.cpp src/detail/device_bloom_filter.cu src/parquet.cpp
+  src/partition.cpp src/partition_utils.cpp src/table_chunk.cpp src/utils.cpp
 )
 
 # ##################################################################################################
@@ -149,7 +143,13 @@ install(
   DESTINATION ${lib_dir}
   EXPORT cudf_streaming-exports
 )
-install(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/include/ DESTINATION ${CMAKE_INSTALL_INCLUDEDIR})
+install(
+  DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/include/
+  DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}
+  FILES_MATCHING
+  PATTERN "*.hpp"
+  PATTERN "detail" EXCLUDE
+)
 
 rapids_export(
   INSTALL cudf_streaming

cpp/libcudf_streaming/benchmarks/bench_partition.cpp

@@ -9,7 +9,7 @@
 #include <cudf/types.hpp>
 #include <cudf/utilities/error.hpp>
 
-#include <cudf_streaming/integrations/partition.hpp>
+#include <cudf_streaming/partition_utils.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
 #include <rmm/device_buffer.hpp>
@@ -66,14 +66,13 @@ static void BM_PartitionAndPack(benchmark::State& state)
   std::vector<cudf::size_type> columns_to_hash{0};
 
   for (auto _ : state) {
-    auto pack_partitions =
-      cudf_streaming::integrations::partition_and_pack(*table,
-                                                       columns_to_hash,
-                                                       num_partitions,
-                                                       cudf::hash_id::HASH_MURMUR3,
-                                                       cudf::DEFAULT_HASH_SEED,
-                                                       stream,
-                                                       br.get());
+    auto pack_partitions = cudf_streaming::partition_and_pack(*table,
+                                                              columns_to_hash,
+                                                              num_partitions,
+                                                              cudf::hash_id::HASH_MURMUR3,
+                                                              cudf::DEFAULT_HASH_SEED,
+                                                              stream,
+                                                              br.get());
     benchmark::DoNotOptimize(pack_partitions);
     CUDF_CUDA_TRY(cudaStreamSynchronize(stream));
   }
@@ -117,14 +116,13 @@ static void BM_PartitionAndPackCurrentImpl(benchmark::State& state)
 
   for (auto _ : state) {
     for (int i = 0; i < num_partitions; i++) {
-      auto pack_partitions =
-        cudf_streaming::integrations::partition_and_pack(*table,
-                                                         columns_to_hash,
-                                                         total_npartitions,
-                                                         cudf::hash_id::HASH_MURMUR3,
-                                                         cudf::DEFAULT_HASH_SEED,
-                                                         stream,
-                                                         br.get());
+      auto pack_partitions = cudf_streaming::partition_and_pack(*table,
+                                                                columns_to_hash,
+                                                                total_npartitions,
+                                                                cudf::hash_id::HASH_MURMUR3,
+                                                                cudf::DEFAULT_HASH_SEED,
+                                                                stream,
+                                                                br.get());
       benchmark::DoNotOptimize(pack_partitions);
     }
     CUDF_CUDA_TRY(cudaStreamSynchronize(stream));

cpp/libcudf_streaming/benchmarks/bench_shuffle.cpp

@@ -3,7 +3,7 @@
  * reserved. SPDX-License-Identifier: Apache-2.0
  */
 
-#include <cudf_streaming/integrations/partition.hpp>
+#include <cudf_streaming/partition_utils.hpp>
 
 #include <rapidsmpf/bootstrap/bootstrap.hpp>
 #include <rapidsmpf/bootstrap/utils.hpp>
@@ -286,7 +286,7 @@ rapidsmpf::Duration do_run(rapidsmpf::shuffler::PartID const total_num_partition
     shuffler.wait();
     for (auto finished_partition : shuffler.local_partitions()) {
       auto packed_chunks    = shuffler.extract(finished_partition);
-      auto output_partition = cudf_streaming::integrations::unpack_and_concat(
+      auto output_partition = cudf_streaming::unpack_and_concat(
         rapidsmpf::unspill_partitions(
           std::move(packed_chunks), br, rapidsmpf::AllowOverbooking::YES),
         stream,
@@ -304,14 +304,14 @@ rapidsmpf::Duration do_run(rapidsmpf::shuffler::PartID const total_num_partition
   // thus we only check large shuffles).
   if (args.num_local_rows >= 1000000) {
     for (const auto& output_partition : output_partitions) {
-      auto [parts, owner] = cudf_streaming::integrations::partition_and_split(
-        output_partition->view(),
-        {0},
-        static_cast<std::int32_t>(total_num_partitions),
-        cudf::hash_id::HASH_MURMUR3,
-        cudf::DEFAULT_HASH_SEED,
-        stream,
-        br);
+      auto [parts, owner] =
+        cudf_streaming::partition_and_split(output_partition->view(),
+                                            {0},
+                                            static_cast<std::int32_t>(total_num_partitions),
+                                            cudf::hash_id::HASH_MURMUR3,
+                                            cudf::DEFAULT_HASH_SEED,
+                                            stream,
+                                            br);
       RAPIDSMPF_EXPECTS(
         std::count_if(
           parts.begin(), parts.end(), [](auto const& table) { return table.num_rows() > 0; }) == 1,
@@ -406,14 +406,13 @@ rapidsmpf::Duration run_hash_partition_inline(std::shared_ptr<rapidsmpf::Communi
     generate_input_partitions(args, stream, br, std::identity{});
 
   auto make_chunk_fn = [&](cudf::table const& partition) {
-    return cudf_streaming::integrations::partition_and_pack(
-      partition,
-      {0},
-      static_cast<std::int32_t>(total_num_partitions),
-      cudf::hash_id::HASH_MURMUR3,
-      cudf::DEFAULT_HASH_SEED,
-      stream,
-      br);
+    return cudf_streaming::partition_and_pack(partition,
+                                              {0},
+                                              static_cast<std::int32_t>(total_num_partitions),
+                                              cudf::hash_id::HASH_MURMUR3,
+                                              cudf::DEFAULT_HASH_SEED,
+                                              stream,
+                                              br);
   };
 
   return do_run(total_num_partitions, comm, args, stream, br, statistics, [&](auto& shuffler) {
@@ -446,14 +445,13 @@ rapidsmpf::Duration run_hash_partition_with_datagen(
 
   std::vector<std::unordered_map<rapidsmpf::shuffler::PartID, rapidsmpf::PackedData>>
     input_partitions = generate_input_partitions(args, stream, br, [&](cudf::table&& table) {
-      return cudf_streaming::integrations::partition_and_pack(
-        table,
-        {0},
-        static_cast<std::int32_t>(total_num_partitions),
-        cudf::hash_id::HASH_MURMUR3,
-        cudf::DEFAULT_HASH_SEED,
-        stream,
-        br);
+      return cudf_streaming::partition_and_pack(table,
+                                                {0},
+                                                static_cast<std::int32_t>(total_num_partitions),
+                                                cudf::hash_id::HASH_MURMUR3,
+                                                cudf::DEFAULT_HASH_SEED,
+                                                stream,
+                                                br);
     });
 
   return do_run(total_num_partitions, comm, args, stream, br, statistics, [&](auto& shuffler) {

cpp/libcudf_streaming/benchmarks/streaming/bench_streaming_shuffle.cpp

@@ -8,9 +8,9 @@
 #include "../utils/rmm_utils.hpp"
 #include "data_generator.hpp"
 
-#include <cudf_streaming/integrations/partition.hpp>
-#include <cudf_streaming/streaming/partition.hpp>
-#include <cudf_streaming/streaming/table_chunk.hpp>
+#include <cudf_streaming/partition.hpp>
+#include <cudf_streaming/partition_utils.hpp>
+#include <cudf_streaming/table_chunk.hpp>
 
 #include <rapidsmpf/bootstrap/bootstrap.hpp>
 #include <rapidsmpf/bootstrap/utils.hpp>
@@ -233,13 +233,13 @@ rapidsmpf::Duration run(std::shared_ptr<rapidsmpf::streaming::Context> ctx,
     actors.push_back(rapidsmpf::streaming::actor::random_table_generator(
       ctx, stream, ch1, args.num_local_partitions, num_columns, num_local_rows, min_val, max_val));
     auto ch2 = ctx->create_channel();
-    actors.push_back(cudf_streaming::streaming::actor::partition_and_pack(
+    actors.push_back(cudf_streaming::actor::partition_and_pack(
       ctx, ch1, ch2, {0}, static_cast<int>(total_num_partitions), hash_function, seed));
     auto ch3 = ctx->create_channel();
     actors.push_back(
       rapidsmpf::streaming::actor::shuffler(ctx, comm, ch2, ch3, op_id, total_num_partitions));
     auto ch4 = ctx->create_channel();
-    actors.push_back(cudf_streaming::streaming::actor::unpack_and_concat(ctx, ch3, ch4));
+    actors.push_back(cudf_streaming::actor::unpack_and_concat(ctx, ch3, ch4));
     actors.push_back(consumer(ctx, ch4));
   }
   auto const t0_elapsed = rapidsmpf::Clock::now();

cpp/libcudf_streaming/benchmarks/streaming/data_generator.hpp

@@ -10,7 +10,7 @@
 #include <cudf/table/table.hpp>
 #include <cudf/types.hpp>
 
-#include <cudf_streaming/streaming/table_chunk.hpp>
+#include <cudf_streaming/table_chunk.hpp>
 
 #include <rapidsmpf/streaming/core/actor.hpp>
 #include <rapidsmpf/streaming/core/channel.hpp>
@@ -19,7 +19,7 @@
 
 namespace rapidsmpf::streaming::actor {
 
-using cudf_streaming::streaming::TableChunk;
+using cudf_streaming::table_chunk;
 
 /**
  * @brief Asynchronously generates and sends a sequence of random numeric tables.
@@ -30,12 +30,12 @@ using cudf_streaming::streaming::TableChunk;
  * It creates a specified number of cuDF tables with random `std::int32_t` values, each
  * consisting of `ncolumns` columns and `nrows` rows. The values are uniformly
  * distributed in the range [`min_val`, `max_val`]. Each generated table is wrapped
- * in a `TableChunk` and sent to the provided output channel in streaming fashion.
+ * in a `table_chunk` and sent to the provided output channel in streaming fashion.
  *
  * @param ctx The actor context to use.
  * @param stream The CUDA stream on which to create the random tables. TODO: use a pool
  * of CUDA streams.
- * @param ch_out Output channel to which generated `TableChunk` objects are sent.
+ * @param ch_out Output channel to which generated `table_chunk` objects are sent.
  * @param num_blocks Number of tables (chunks) to generate and send.
  * @param ncolumns Number of columns per generated table.
  * @param nrows Number of rows per column in each table.
@@ -61,7 +61,7 @@ inline Actor random_table_generator(std::shared_ptr<Context> ctx,
     auto res = ctx->br()->reserve_device_memory_and_spill(nbytes, AllowOverbooking::NO);
     co_await ch_out->send(
       to_message(seq,
-                 std::make_unique<TableChunk>(
+                 std::make_unique<table_chunk>(
                    std::make_unique<cudf::table>(random_table(
                      ncolumns, nrows, min_val, max_val, stream, ctx->br()->device_mr())),
                    stream)));

cpp/libcudf_streaming/benchmarks/streaming/ndsh/bench_read.cpp

@@ -9,7 +9,7 @@
 #include <cudf/io/types.hpp>
 #include <cudf/types.hpp>
 
-#include <cudf_streaming/streaming/parquet.hpp>
+#include <cudf_streaming/parquet.hpp>
 
 #include <rmm/detail/format.hpp>
 #include <rmm/mr/cuda_async_memory_resource.hpp>
@@ -52,7 +52,7 @@ rapidsmpf::streaming::Actor read_parquet(std::shared_ptr<rapidsmpf::streaming::C
     rapidsmpf::ndsh::detail::get_table_path(input_directory, input_file));
   auto options = cudf::io::parquet_reader_options::builder(cudf::io::source_info(files)).build();
   if (columns.has_value()) { options.set_column_names(*columns); }
-  return cudf_streaming::streaming::actor::read_parquet(
+  return cudf_streaming::actor::read_parquet(
     ctx, comm, ch_out, num_producers, options, num_rows_per_chunk);
 }
 
@@ -68,9 +68,8 @@ rapidsmpf::streaming::Actor consume_channel_parallel(
     while (true) {
       auto msg = co_await ch_in->receive();
       if (msg.empty()) { break; }
-      if (msg.holds<cudf_streaming::streaming::TableChunk>()) {
-        auto chunk =
-          co_await msg.release<cudf_streaming::streaming::TableChunk>().make_available(ctx);
+      if (msg.holds<cudf_streaming::table_chunk>()) {
+        auto chunk = co_await msg.release<cudf_streaming::table_chunk>().make_available(ctx);
         ctx->logger()->print("Consumed chunk with ",
                              chunk.table_view().num_rows(),
                              " rows and ",

cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp

@@ -9,7 +9,7 @@
 #include <cudf/table/table.hpp>
 #include <cudf/table/table_view.hpp>
 
-#include <cudf_streaming/streaming/table_chunk.hpp>
+#include <cudf_streaming/table_chunk.hpp>
 
 #include <rapidsmpf/cuda_event.hpp>
 #include <rapidsmpf/cuda_stream.hpp>
@@ -40,33 +40,32 @@ streaming::Actor concatenate(std::shared_ptr<streaming::Context> ctx,
   }
   if (messages.size() == 0) {
     co_await ch_out->send(
-      cudf_streaming::streaming::to_message(0,
-                                            std::make_unique<cudf_streaming::streaming::TableChunk>(
-                                              std::make_unique<cudf::table>(), concat_stream)));
+      cudf_streaming::to_message(0,
+                                 std::make_unique<cudf_streaming::table_chunk>(
+                                   std::make_unique<cudf::table>(), concat_stream)));
   } else if (messages.size() == 1) {
     co_await ch_out->send(std::move(messages[0]));
   } else {
-    std::vector<cudf_streaming::streaming::TableChunk> chunks;
+    std::vector<cudf_streaming::table_chunk> chunks;
     std::vector<cudf::table_view> views;
     if (order == ConcatOrder::LINEARIZE) {
       std::ranges::sort(messages, std::less{}, [](auto&& msg) { return msg.sequence_number(); });
     }
     chunks.reserve(messages.size());
     views.reserve(messages.size());
     for (auto&& msg : messages) {
-      auto chunk =
-        co_await msg.release<cudf_streaming::streaming::TableChunk>().make_available(ctx);
+      auto chunk = co_await msg.release<cudf_streaming::table_chunk>().make_available(ctx);
       cuda_stream_join(concat_stream, chunk.stream(), &event);
       views.push_back(chunk.table_view());
       chunks.push_back(std::move(chunk));
     }
-    auto result = std::make_unique<cudf_streaming::streaming::TableChunk>(
+    auto result = std::make_unique<cudf_streaming::table_chunk>(
       cudf::concatenate(views, concat_stream, ctx->br()->device_mr()), concat_stream);
     cuda_stream_join(chunks | std::views::transform([](auto&& chunk) { return chunk.stream(); }),
                      std::ranges::single_view(concat_stream),
                      &event);
     chunks.clear();
-    co_await ch_out->send(cudf_streaming::streaming::to_message(0, std::move(result)));
+    co_await ch_out->send(cudf_streaming::to_message(0, std::move(result)));
   }
   co_await ch_out->drain(ctx->executor());
 }

cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.hpp

@@ -22,7 +22,7 @@ enum class ConcatOrder : bool {
  * @brief Concatenate all table chunks from an input channel.
  *
  * @param ctx Streaming context.
- * @param ch_in Input channel of `TableChunk`s.
+ * @param ch_in Input channel of `table_chunk`s.
  * @param ch_out Output channel of concatenated chunks, contains at most one message.
  * @param order Do we care about maintaining the input ordering?
  *

cpp/libcudf_streaming/benchmarks/streaming/ndsh/groupby.cpp

@@ -9,7 +9,7 @@
 #include <cudf/groupby.hpp>
 #include <cudf/types.hpp>
 
-#include <cudf_streaming/streaming/table_chunk.hpp>
+#include <cudf_streaming/table_chunk.hpp>
 
 #include <rapidsmpf/communicator/communicator.hpp>
 #include <rapidsmpf/streaming/core/channel.hpp>
@@ -33,9 +33,9 @@ streaming::Actor chunkwise_group_by(std::shared_ptr<streaming::Context> ctx,
   while (!ch_out->is_shutdown()) {
     auto msg = co_await ch_in->receive();
     if (msg.empty()) { break; }
-    auto chunk  = co_await msg.release<cudf_streaming::streaming::TableChunk>().make_available(ctx);
-    auto stream = chunk.stream();
-    auto table  = chunk.table_view();
+    auto chunk        = co_await msg.release<cudf_streaming::table_chunk>().make_available(ctx);
+    auto stream       = chunk.stream();
+    auto table        = chunk.table_view();
     auto agg_requests = std::vector<cudf::groupby::aggregation_request>();
     agg_requests.reserve(requests.size());
     std::ranges::transform(requests, std::back_inserter(agg_requests), [&table](auto&& req) {
@@ -53,10 +53,10 @@ streaming::Actor chunkwise_group_by(std::shared_ptr<streaming::Context> ctx,
     for (auto&& a : aggregated) {
       std::ranges::move(a.results, std::back_inserter(result));
     }
-    co_await ch_out->send(cudf_streaming::streaming::to_message(
-      msg.sequence_number(),
-      std::make_unique<cudf_streaming::streaming::TableChunk>(
-        std::make_unique<cudf::table>(std::move(result)), stream)));
+    co_await ch_out->send(
+      cudf_streaming::to_message(msg.sequence_number(),
+                                 std::make_unique<cudf_streaming::table_chunk>(
+                                   std::make_unique<cudf::table>(std::move(result)), stream)));
   }
   co_await ch_out->drain(ctx->executor());
 }

cpp/libcudf_streaming/benchmarks/streaming/ndsh/groupby.hpp

@@ -30,8 +30,8 @@ struct groupby_request {
  * @note Grouped chunks are not further grouped together.
  *
  * @param ctx Streaming context.
- * @param ch_in `TableChunk`s to aggregate
- * @param ch_out Output channel of grouped `TableChunk`s
+ * @param ch_in `table_chunk`s to aggregate
+ * @param ch_out Output channel of grouped `table_chunk`s
  * @param keys Column indices of the key columns in the input channel.
  * @param requests Vector of aggregation requests referencing columns in the input
  * channel.