Fix dictionary pruning for column chunks with no dictionary page (#23746)

A chunk with no dictionary page contributes no values to the hash set built for its row group, but cuco rounds every capacity up to at least one bucket, so the set still has slots. Testing emptiness by slot count therefore missed it, and probing a set that was never built reported the literal as absent and pruned the row group. Test the value count instead.

Also add the missing group.sync() between zeroing the per-row-group results and decoding the dictionary page, without which a thread could overwrite a result another thread had already written.

Authors:
  - Paul Mattione (https://github.com/pmattione-nvidia)
  - Muhammad Haseeb (https://github.com/mhaseeb123)

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

URL: #23746

Author: pmattione-nvidia

cpp/src/io/parquet/experimental/dictionary_page_filter.cu

@@ -293,8 +293,14 @@ CUDF_KERNEL void query_dictionaries(cudf::device_span<T> decoded_data,
 
   // Evaluate the scalar against all cuco hash sets of this column
   for (auto set_idx = group.thread_rank(); set_idx < total_row_groups; set_idx += group.size()) {
-    // If the set is empty (no dictionary page data), then skip the dictionary page filter
-    if (set_offsets[set_idx + 1] - set_offsets[set_idx] == 0) {
+    // Number of values in this hash set
+    auto const num_set_values = value_offsets[set_idx + 1] - value_offsets[set_idx];
+
+    // Skip the dictionary page filter for a column chunk with no dictionary page. Emptiness must be
+    // read from the value count and not from the number of slots, because cuco rounds every
+    // capacity up to at least one bucket, so an empty dictionary still has slots. Its set was never
+    // built, so probing it would report the literal as absent and prune the row group.
+    if (num_set_values == 0) {
       result[set_idx] = operators[scalar_idx] == ast::ast_operator::EQUAL;
       continue;
     }
@@ -311,8 +317,6 @@ CUDF_KERNEL void query_dictionaries(cudf::device_span<T> decoded_data,
                                              storage_ref};
     auto set_find_ref = hash_set_ref.rebind_operators(cuco::contains);
 
-    // Number of values in this hash set
-    auto const num_set_values = value_offsets[set_idx + 1] - value_offsets[set_idx];
     // Literal value to find in this hash set
     auto const literal_value = scalar.value<T>();
 
@@ -901,6 +905,8 @@ CUDF_KERNEL void __launch_bounds__(DECODE_BLOCK_SIZE)
     results[i][row_group_idx] = false;
   }
 
+  group.sync();
+
   // Decode values from the current dictionary page with the current thread block
   for (auto value_idx = group.thread_rank(); value_idx < page.num_input_values;
        value_idx += group.num_threads()) {

cpp/tests/io/experimental/hybrid_scan_filters_test.cpp

@@ -1968,6 +1968,56 @@ TEST_P(DictionaryFilterGapTest, FilterRowGroupsWithMissingDictPages)
     EXPECT_EQ(filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr),
               expected);
   }
+
+  // The cases below give a column more than `MAX_INLINE_LITERALS` literals, which builds a hash set
+  // per dictionary instead of evaluating the literals inline. A row group with no dictionary page
+  // has no hash set built for it, so that path has to recognize it and keep the row group.
+
+  // Filtering - col0 equals any of three plain values: row group 0 is pruned as its dictionary
+  // holds none of them, row group 1 cannot be pruned
+  {
+    auto literal_value0 = cudf::string_scalar("plain_value_5", true, stream);
+    auto literal_value1 = cudf::string_scalar("plain_value_6", true, stream);
+    auto literal_value2 = cudf::string_scalar("plain_value_7", true, stream);
+    auto literal0       = cudf::ast::literal(literal_value0);
+    auto literal1       = cudf::ast::literal(literal_value1);
+    auto literal2       = cudf::ast::literal(literal_value2);
+    auto const equal0   = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col0_ref, literal0);
+    auto const equal1   = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col0_ref, literal1);
+    auto const equal2   = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col0_ref, literal2);
+    auto const either   = cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_OR, equal0, equal1);
+    auto const filter_expression =
+      cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_OR, either, equal2);
+    auto const options =
+      cudf::io::parquet_reader_options::builder().filter(filter_expression).build();
+
+    auto const expected = std::vector<cudf::size_type>{1};
+    EXPECT_EQ(filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr),
+              expected);
+  }
+
+  // Filtering - col0 equals any of three values, one of which is in row group 0's dictionary: both
+  // row groups survive
+  {
+    auto literal_value0 = cudf::string_scalar("dict_value", true, stream);
+    auto literal_value1 = cudf::string_scalar("plain_value_5", true, stream);
+    auto literal_value2 = cudf::string_scalar("plain_value_6", true, stream);
+    auto literal0       = cudf::ast::literal(literal_value0);
+    auto literal1       = cudf::ast::literal(literal_value1);
+    auto literal2       = cudf::ast::literal(literal_value2);
+    auto const equal0   = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col0_ref, literal0);
+    auto const equal1   = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col0_ref, literal1);
+    auto const equal2   = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col0_ref, literal2);
+    auto const either   = cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_OR, equal0, equal1);
+    auto const filter_expression =
+      cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_OR, either, equal2);
+    auto const options =
+      cudf::io::parquet_reader_options::builder().filter(filter_expression).build();
+
+    auto const expected = std::vector<cudf::size_type>{0, 1};
+    EXPECT_EQ(filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr),
+              expected);
+  }
 }
 
 INSTANTIATE_TEST_SUITE_P(Compression,