Parquet readers support case-insensitive column names (rapidsai#21700)

Closes rapidsai#21631

This PR enables the Parquet and Hybrid scan readers to support case-insensitive column names (or paths if structs) for column selection and row filteration.

Authors:
  - Muhammad Haseeb (https://github.com/mhaseeb123)
  - Vukasin Milovanovic (https://github.com/vuule)

Approvers:
  - Vukasin Milovanovic (https://github.com/vuule)
  - Nghia Truong (https://github.com/ttnghia)
  - Matthew Murray (https://github.com/Matt711)
  - Lawrence Mitchell (https://github.com/wence-)
  - Vyas Ramasubramani (https://github.com/vyasr)

URL: rapidsai#21700

Author: mhaseeb123

cpp/include/cudf/io/parquet.hpp

@@ -103,6 +103,9 @@ class parquet_reader_options {
   type_id _decimal_width{type_id::EMPTY};
   // Whether to use JIT compilation for filtering
   bool _use_jit_filter = false;
+  // Whether column name matching is case sensitive. In case of multiple
+  // case-insensitive matches, the first matched column is selected
+  bool _case_sensitive_names = true;
 
   std::optional<std::vector<reader_column_schema>> _reader_column_schema;
 
@@ -285,6 +288,16 @@ class parquet_reader_options {
    */
   [[nodiscard]] bool is_enabled_use_jit_filter() const { return _use_jit_filter; }
 
+  /**
+   * @brief Returns whether column name matching is case sensitive.
+   *
+   * @note When disabled, if there are multiple case-insensitive matches, the first
+   * matched column is selected from the Parquet schema.
+   *
+   * @return `true` if column name matching is case sensitive (default)
+   */
+  [[nodiscard]] bool is_enabled_case_sensitive_names() const { return _case_sensitive_names; }
+
   /**
    * @brief Set a new source location
    *
@@ -510,6 +523,16 @@ class parquet_reader_options {
    * columns need to be cast. The scale of each column is preserved from the file.
    */
   void set_decimal_width(type_id width) { _decimal_width = width; }
+
+  /**
+   * @brief Sets whether column name matching is case sensitive.
+   *
+   * @note When disabled, if there are multiple case-insensitive matches, the first
+   * matched column is selected from the Parquet schema.
+   *
+   * @param val Boolean indicating whether to enable case-sensitive matching.
+   */
+  void enable_case_sensitive_names(bool val) { _case_sensitive_names = val; }
 };
 
 /**
@@ -758,6 +781,21 @@ class parquet_reader_options_builder {
     return *this;
   }
 
+  /**
+   * @brief Sets whether column name matching is case sensitive.
+   *
+   * @note When disabled, if there are multiple case-insensitive matches, the first
+   * matched column is selected from the Parquet schema.
+   *
+   * @param val Boolean indicating whether to enable case-sensitive matching
+   * @return this for chaining
+   */
+  parquet_reader_options_builder& case_sensitive_names(bool val)
+  {
+    options._case_sensitive_names = val;
+    return *this;
+  }
+
   /**
    * @brief move parquet_reader_options member once it's built.
    */

cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp

@@ -6,8 +6,8 @@
 #include "hybrid_scan_helpers.hpp"
 
 #include "io/parquet/compact_protocol_reader.hpp"
+#include "io/parquet/expression_transform_helpers.hpp"
 #include "io/parquet/reader_impl_helpers.hpp"
-#include "io/utilities/row_selection.hpp"
 
 #include <cudf/detail/nvtx/ranges.hpp>
 #include <cudf/logger.hpp>
@@ -201,7 +201,8 @@ aggregate_reader_metadata::select_payload_columns(
   bool strings_to_categorical,
   bool ignore_missing_columns,
   type_id timestamp_type_id,
-  type_id decimal_type_id)
+  type_id decimal_type_id,
+  bool case_sensitive_names)
 {
   // If neither payload nor filter columns are specified, select all columns
   if (not payload_column_names.has_value() and not filter_column_names.has_value()) {
@@ -212,27 +213,40 @@ aggregate_reader_metadata::select_payload_columns(
                           strings_to_categorical,
                           ignore_missing_columns,
                           timestamp_type_id,
-                          decimal_type_id);
+                          decimal_type_id,
+                          case_sensitive_names);
   }
 
   std::vector<std::string> valid_payload_columns;
 
+  using cudf::io::parquet::detail::normalize_column_path;
+
+  // Helper lambda to construct a set of normalized column names for O(1) lookup
+  auto construct_filter_columns_set = [](auto const& names, bool case_sensitive_names) {
+    std::unordered_set<std::string> filter_columns_set;
+    for (auto const& name : names) {
+      filter_columns_set.insert(normalize_column_path(name, case_sensitive_names));
+    }
+    return filter_columns_set;
+  };
+
   // If payload columns are specified, only select payload columns that do not appear in the filter
   // expression
   if (payload_column_names.has_value()) {
     valid_payload_columns = *payload_column_names;
     // Remove filter columns from the provided payload column names
     if (filter_column_names.has_value() and not filter_column_names->empty()) {
-      // Add filter column names to a hash set for faster lookup
-      std::unordered_set<std::string> filter_columns_set(filter_column_names->begin(),
-                                                         filter_column_names->end());
+      auto const filter_columns_set =
+        construct_filter_columns_set(*filter_column_names, case_sensitive_names);
       // Remove a payload column name if it is also present in the hash set
-      valid_payload_columns.erase(std::remove_if(valid_payload_columns.begin(),
-                                                 valid_payload_columns.end(),
-                                                 [&filter_columns_set](auto const& col) {
-                                                   return filter_columns_set.count(col) > 0;
-                                                 }),
-                                  valid_payload_columns.end());
+      valid_payload_columns.erase(
+        std::remove_if(valid_payload_columns.begin(),
+                       valid_payload_columns.end(),
+                       [&](auto const& col) {
+                         return filter_columns_set.count(
+                                  normalize_column_path(col, case_sensitive_names)) > 0;
+                       }),
+        valid_payload_columns.end());
     }
     // Call the base `select_columns()` method with valid payload columns
     return select_columns(valid_payload_columns,
@@ -241,40 +255,40 @@ aggregate_reader_metadata::select_payload_columns(
                           strings_to_categorical,
                           ignore_missing_columns,
                           timestamp_type_id,
-                          decimal_type_id);
+                          decimal_type_id,
+                          case_sensitive_names);
   }
 
   // Else if only filter columns are specified, select all columns that do not appear in the
   // filter expression
-
-  // Add filter column names to a hash set for faster lookup
-  std::unordered_set<std::string> filter_columns_set(filter_column_names->begin(),
-                                                     filter_column_names->end());
+  auto const filter_columns_set =
+    construct_filter_columns_set(*filter_column_names, case_sensitive_names);
 
   std::function<void(std::string, int)> add_column_path = [&](std::string path_till_now,
                                                               int schema_idx) {
     auto const& schema_elem     = get_schema(schema_idx);
     std::string const curr_path = path_till_now + schema_elem.name;
-    // Add the current path to the list of valid payload columns if it is not a filter column
     // TODO: Add children when AST filter expressions start supporting nested struct columns
-    if (filter_columns_set.count(curr_path) == 0) { valid_payload_columns.push_back(curr_path); }
+    if (filter_columns_set.count(normalize_column_path(curr_path, case_sensitive_names)) == 0) {
+      valid_payload_columns.push_back(curr_path);
+    }
   };
 
-  // Add all but filter columns to valid payload columns
   if (not filter_column_names->empty()) {
-    for (auto const& child_idx : get_schema(0).children_idx) {
+    auto const& root = get_schema(0);
+    for (auto const& child_idx : root.children_idx) {
       add_column_path("", child_idx);
     }
   }
 
-  // Call the base `select_columns()` method with all but filter columns
   return select_columns(valid_payload_columns,
                         {},
                         include_index,
                         strings_to_categorical,
                         ignore_missing_columns,
                         timestamp_type_id,
-                        decimal_type_id);
+                        decimal_type_id,
+                        case_sensitive_names);
 }
 
 std::vector<std::vector<cudf::size_type>>
@@ -589,19 +603,26 @@ named_to_reference_converter::named_to_reference_converter(
   std::optional<std::reference_wrapper<ast::expression const>> expr,
   table_metadata const& metadata,
   std::vector<SchemaElement> const& schema_tree,
-  cudf::io::parquet_reader_options const& options)
+  cudf::io::parquet_reader_options const& options,
+  bool case_sensitive_names)
 {
   if (!expr.has_value()) { return; }
 
-  _column_indices_to_names =
-    cudf::io::parquet::detail::map_column_indices_to_names(options, schema_tree);
+  _case_sensitive_names = case_sensitive_names;
+
+  _column_indices_to_names = cudf::io::parquet::detail::map_column_indices_to_names(
+    options, schema_tree, case_sensitive_names);
 
   // Map column names to their indices
-  std::transform(metadata.schema_info.cbegin(),
-                 metadata.schema_info.cend(),
-                 thrust::counting_iterator<size_t>(0),
-                 std::inserter(_column_name_to_index, _column_name_to_index.end()),
-                 [](auto const& sch, auto index) { return std::make_pair(sch.name, index); });
+  std::transform(
+    metadata.schema_info.cbegin(),
+    metadata.schema_info.cend(),
+    thrust::counting_iterator<size_t>(0),
+    std::inserter(_column_name_to_index, _column_name_to_index.end()),
+    [&](auto const& sch, auto index) {
+      return std::make_pair(
+        cudf::io::parquet::detail::normalize_column_path(sch.name, case_sensitive_names), index);
+    });
 
   expr.value().get().accept(*this);
 }
@@ -611,12 +632,17 @@ std::reference_wrapper<ast::expression const> named_to_reference_converter::visi
 {
   // Map the column index to its name
   auto const col_name_iter = _column_indices_to_names.find(expr.get_column_index());
-  CUDF_EXPECTS(col_name_iter != _column_indices_to_names.end(),
-               "Column index not found in column indices to names map");
+  CUDF_EXPECTS(
+    col_name_iter != _column_indices_to_names.end(),
+    "Column index in the filter expression not found in the column indices to names map. Note that "
+    "only top-level columns except structs and lists are supported in "
+    "Parquet filter expression",
+    std::invalid_argument);
   auto const col_name = col_name_iter->second;
-  // Check if the column name exists in the metadata and map it to its new column index
-  auto col_index_it = _column_name_to_index.find(col_name);
-  CUDF_EXPECTS(col_index_it != _column_name_to_index.end(), "Column name not found in metadata");
+  auto col_index_it   = _column_name_to_index.find(col_name);
+  CUDF_EXPECTS(col_index_it != _column_name_to_index.end(),
+               "Column name mapped from its index in the filter expression "
+               "not found in the metadata of selected columns");
   auto col_index = col_index_it->second;
   // Create a new column reference
   _col_ref.emplace_back(col_index);

cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp

@@ -146,6 +146,7 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
    * @param ignore_missing_columns Whether to ignore non-existent columns
    * @param timestamp_type_id Type conversion parameter
    * @param decimal_type_id Type conversion parameter
+   * @param case_sensitive_names Boolean indicating if column names are case sensitive
    *
    * @return input column information, output column buffers, list of output column schema
    * indices
@@ -158,7 +159,8 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
                            bool strings_to_categorical,
                            bool ignore_missing_columns,
                            type_id timestamp_type_id,
-                           type_id decimal_type_id);
+                           type_id decimal_type_id,
+                           bool case_sensitive_names);
 
   /**
    * @brief Filters row groups such that only the row groups that start within the byte range
@@ -369,7 +371,8 @@ class named_to_reference_converter : public parquet::detail::named_to_reference_
   named_to_reference_converter(std::optional<std::reference_wrapper<ast::expression const>> expr,
                                table_metadata const& metadata,
                                std::vector<SchemaElement> const& schema_tree,
-                               cudf::io::parquet_reader_options const& options);
+                               cudf::io::parquet_reader_options const& options,
+                               bool case_sensitive_names);
 
   using parquet::detail::named_to_reference_converter::visit;