Add an example to demonstrate multithreaded read_parquet pipelines

cpp/examples/parquet_io/CMakeLists.txt

@@ -16,10 +16,17 @@ project(
 
 include(../fetch_dependencies.cmake)
 
-# Configure your project here
+# Build and install parquet_io
 add_executable(parquet_io parquet_io.cpp)
 target_link_libraries(parquet_io PRIVATE cudf::cudf)
 target_compile_features(parquet_io PRIVATE cxx_std_17)
-
 install(TARGETS parquet_io DESTINATION bin/examples/libcudf)
+
+# Build and install parquet_io_multithreaded
+add_executable(parquet_io_multithreaded parquet_io_multithreaded.cpp)
+target_link_libraries(parquet_io_multithreaded PRIVATE cudf::cudf)
+target_compile_features(parquet_io_multithreaded PRIVATE cxx_std_17)
+install(TARGETS parquet_io_multithreaded DESTINATION bin/examples/libcudf)
+
+# Install the example.parquet file
 install(FILES ${CMAKE_CURRENT_LIST_DIR}/example.parquet DESTINATION bin/examples/libcudf)

cpp/examples/parquet_io/parquet_io.hpp → cpp/examples/parquet_io/common.hpp

@@ -16,17 +16,23 @@
 
 #pragma once
 
+#include "../utilities/timer.hpp"
+
+#include <cudf/concatenate.hpp>
 #include <cudf/io/parquet.hpp>
 #include <cudf/io/types.hpp>
 #include <cudf/join.hpp>
 #include <cudf/table/table_view.hpp>
 
 #include <rmm/cuda_device.hpp>
+#include <rmm/cuda_stream_pool.hpp>
 #include <rmm/mr/device/cuda_memory_resource.hpp>
 #include <rmm/mr/device/device_memory_resource.hpp>
 #include <rmm/mr/device/owning_wrapper.hpp>
 #include <rmm/mr/device/pool_memory_resource.hpp>
 
+#include <fmt/color.h>
+
 #include <chrono>
 #include <iostream>
 #include <optional>
@@ -124,3 +130,29 @@ std::shared_ptr<rmm::mr::device_memory_resource> create_memory_resource(bool is_
 
   return std::nullopt;
 }
+
+/**
+ * @brief Check if two tables are identical, throw an error otherwise
+ *
+ * @param lhs_table View to lhs table
+ * @param rhs_table View to rhs table
+ */
+inline void check_identical_tables(cudf::table_view const& lhs_table,
+                                   cudf::table_view const& rhs_table)
+{
+  try {
+    // Left anti-join the original and transcoded tables
+    // identical tables should not throw an exception and
+    // return an empty indices vector
+    auto const indices = cudf::left_anti_join(lhs_table, rhs_table, cudf::null_equality::EQUAL);
+
+    // No exception thrown, check indices
+    auto const valid = indices->size() == 0;
+    fmt::print(
+      fmt::emphasis::bold | fg(fmt::color::green_yellow), "Transcoding valid: {}\n", valid);
+  } catch (std::exception& e) {
+    std::cerr << e.what() << std::endl << std::endl;
+    throw std::runtime_error(
+      fmt::format(fmt::emphasis::bold | fg(fmt::color::red), "Transcoding valid: false\n"));
+  }
+}

cpp/examples/parquet_io/parquet_io.cpp

@@ -14,9 +14,7 @@
  * limitations under the License.
  */
 
-#include "parquet_io.hpp"
-
-#include "../utilities/timer.hpp"
+#include "common.hpp"
 
 #include <cudf/utilities/default_stream.hpp>
 
@@ -130,50 +128,32 @@ int main(int argc, char const** argv)
   // Read input parquet file
   // We do not want to time the initial read time as it may include
   // time for nvcomp, cufile loading and RMM growth
-  std::cout << std::endl << "Reading " << input_filepath << "..." << std::endl;
-  std::cout << "Note: Not timing the initial parquet read as it may include\n"
-               "times for nvcomp, cufile loading and RMM growth."
-            << std::endl
-            << std::endl;
+  fmt::print("\nReading {}...", input_filepath);
+  fmt::print(
+    "Note: Not timing the initial parquet read as it may include\n"
+    "times for nvcomp, cufile loading and RMM growth.\n\n");
   auto [input, metadata] = read_parquet(input_filepath);
 
   // Status string to indicate if page stats are set to be written or not
   auto page_stat_string = (page_stats.has_value()) ? "page stats" : "no page stats";
   // Write parquet file with the specified encoding and compression
-  std::cout << "Writing " << output_filepath << " with encoding, compression and "
-            << page_stat_string << ".." << std::endl;
+  fmt::print("Writing {} with encoding, compression and {}..\n", output_filepath, page_stat_string);
 
   // `timer` is automatically started here
   cudf::examples::timer timer;
   write_parquet(input->view(), metadata, output_filepath, encoding, compression, page_stats);
   timer.print_elapsed_millis();
 
   // Read the parquet file written with encoding and compression
-  std::cout << "Reading " << output_filepath << "..." << std::endl;
+  fmt::print("Reading {}...\n", output_filepath);
 
   // Reset the timer
   timer.reset();
   auto [transcoded_input, transcoded_metadata] = read_parquet(output_filepath);
   timer.print_elapsed_millis();
 
   // Check for validity
-  try {
-    // Left anti-join the original and transcoded tables
-    // identical tables should not throw an exception and
-    // return an empty indices vector
-    auto const indices = cudf::left_anti_join(input->view(),
-                                              transcoded_input->view(),
-                                              cudf::null_equality::EQUAL,
-                                              cudf::get_default_stream(),
-                                              resource.get());
-
-    // No exception thrown, check indices
-    auto const valid = indices->size() == 0;
-    std::cout << "Transcoding valid: " << std::boolalpha << valid << std::endl;
-  } catch (std::exception& e) {
-    std::cerr << e.what() << std::endl << std::endl;
-    std::cout << "Transcoding valid: false" << std::endl;
-  }
+  check_identical_tables(input->view(), transcoded_input->view());
 
   return 0;
 }