Fix potential malformed headers in parquet delta decoder

cpp/src/io/parquet/delta_binary.cuh

@@ -91,6 +91,7 @@ struct delta_binary_decoder {
   uint32_t cur_mb;               // index of the current mini-block within the block
   uint8_t const* cur_mb_start;   // pointer to the start of the current mini-block data
   uint8_t const* cur_bitwidths;  // pointer to the bitwidth array in the block
+  bool error;                    // flag to catch malformed headers
 
   zigzag128_t value[delta_rolling_buf_size];  // circular buffer of delta values
 
@@ -148,7 +149,21 @@ struct delta_binary_decoder {
     last_value       = first_value;
 
     current_value_idx = 0;
-    values_per_mb     = block_size / mini_block_count;
+    error             = false;
+
+    // Validate header against the DELTA_BINARY_PACKED spec invariants
+    if (mini_block_count == 0 or block_size == 0 or (block_size % mini_block_count) != 0) {
+      error         = true;
+      value_count   = 0;
+      values_per_mb = 1;
+      block_start   = d_end;
+      cur_mb        = 0;
+      cur_mb_start  = d_end;
+      cur_bitwidths = d_end;
+      return;
+    }
+
+    values_per_mb = block_size / mini_block_count;
 
     // init the first mini-block
     block_start = d_start;

cpp/src/io/parquet/page_delta_decode.cu

@@ -366,9 +366,11 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size)
   block.sync();
 
   auto const batch_size = db->values_per_mb;
-  if (batch_size > max_delta_mini_block_size) {
-    set_error(static_cast<kernel_error::value_type>(decode_error::DELTA_PARAMS_UNSUPPORTED),
-              error_code);
+  if (db->error or batch_size > max_delta_mini_block_size) {
+    if (block.thread_rank() == 0) {
+      set_error(static_cast<kernel_error::value_type>(decode_error::DELTA_PARAMS_UNSUPPORTED),
+                error_code);
+    }
     return;
   }
 
@@ -546,6 +548,15 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
   }
   block.sync();
 
+  // Propagate malformed-header errors from either underlying DELTA_BINARY_PACKED decoder.
+  if (prefix_db->error or suffix_db->error) {
+    if (block.thread_rank() == 0) {
+      set_error(static_cast<kernel_error::value_type>(decode_error::DELTA_PARAMS_UNSUPPORTED),
+                error_code);
+    }
+    return;
+  }
+
   // assert that prefix and suffix have same mini-block size
   if (prefix_db->values_per_mb != suffix_db->values_per_mb or
       prefix_db->block_size != suffix_db->block_size or
@@ -562,8 +573,10 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
   // sanity check to make sure we can process this page
   auto const batch_size = prefix_db->values_per_mb;
   if (batch_size > max_delta_mini_block_size) {
-    set_error(static_cast<kernel_error::value_type>(decode_error::DELTA_PARAMS_UNSUPPORTED),
-              error_code);
+    if (block.thread_rank() == 0) {
+      set_error(static_cast<kernel_error::value_type>(decode_error::DELTA_PARAMS_UNSUPPORTED),
+                error_code);
+    }
     return;
   }
 
@@ -759,14 +772,18 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
   }
   block.sync();
 
-  int const leaf_level_index = s->col.max_nesting_depth - 1;
-
   // sanity check to make sure we can process this page
   auto const batch_size = db->values_per_mb;
-  if (batch_size > max_delta_mini_block_size) {
-    set_error(static_cast<int32_t>(decode_error::DELTA_PARAMS_UNSUPPORTED), error_code);
+  if (db->error or batch_size > max_delta_mini_block_size) {
+    if (block.thread_rank() == 0) {
+      set_error(static_cast<kernel_error::value_type>(decode_error::DELTA_PARAMS_UNSUPPORTED),
+                error_code);
+    }
     return;
   }
+
+  int const leaf_level_index = s->col.max_nesting_depth - 1;
+
   // db->init_binary_block below resets db->values_per_mb
   block.sync();
   // if this is a bounds page, then we need to decode up to the first mini-block