diff --git a/cpp/src/io/parquet/decode_fixed.cu b/cpp/src/io/parquet/decode_fixed.cu
index 604c5d4c5bbb..93eb43bcddca 100644
--- a/cpp/src/io/parquet/decode_fixed.cu
+++ b/cpp/src/io/parquet/decode_fixed.cu
@@ -340,7 +340,7 @@ __device__ int update_validity_and_row_indices_nested(
   constexpr int max_batch_size = num_warps * cudf::detail::warp_size;
 
   // how many (input) values we've processed in the page so far
-  int value_count = s->input_value_count;
+  int value_count = s->progress.input_value_count;
 
   // cap by last row so that we don't process any rows past what we want to output.
   int const first_row                 = s->setup.first_row;
@@ -363,7 +363,7 @@ __device__ int update_validity_and_row_indices_nested(
     int const block_value_count  = batch_size;
 
     // compute our row index, whether we're in row bounds, and validity
-    // This ASSUMES that s->row_index_lower_bound is always -1!
+    // This ASSUMES that s->progress.row_index_lower_bound is always -1!
     // Its purpose is to handle rows than span page boundaries, which only happen for lists.
     int const row_index                     = thread_value_count + value_count;
     int const in_row_bounds                 = (row_index < last_row);
@@ -428,11 +428,11 @@ __device__ int update_validity_and_row_indices_nested(
 
   if (t == 0) {
     // update valid value count for decoding and total # of values we've processed
-    max_depth_ni.valid_count = max_depth_valid_count;
-    max_depth_ni.value_count = value_count;  // Needed AT LEAST for strings!
-    s->nz_count              = max_depth_valid_count;
-    s->input_value_count     = value_count;
-    s->input_row_count       = value_count;
+    max_depth_ni.valid_count      = max_depth_valid_count;
+    max_depth_ni.value_count      = value_count;  // Needed AT LEAST for strings!
+    s->progress.nz_count          = max_depth_valid_count;
+    s->progress.input_value_count = value_count;
+    s->progress.input_row_count   = value_count;
   }
 
   return max_depth_valid_count;
@@ -463,7 +463,7 @@ __device__ int update_validity_and_row_indices_flat(
   auto& ni = s->nesting_info[0];
 
   // how many (input) values we've processed in the page so far
-  int value_count = s->input_value_count;
+  int value_count = s->progress.input_value_count;
   int valid_count = ni.valid_count;
 
   // cap by last row so that we don't process any rows past what we want to output.
@@ -482,7 +482,7 @@ __device__ int update_validity_and_row_indices_flat(
     int const block_value_count  = batch_size;
 
     // compute our row index, whether we're in row bounds, and validity
-    // This ASSUMES that s->row_index_lower_bound is always -1!
+    // This ASSUMES that s->progress.row_index_lower_bound is always -1!
     // Its purpose is to handle rows than span page boundaries, which only happen for lists.
     int const row_index     = thread_value_count + value_count;
     int const in_row_bounds = (row_index < last_row);
@@ -538,11 +538,11 @@ __device__ int update_validity_and_row_indices_flat(
 
   if (t == 0) {
     // update valid value count for decoding and total # of values we've processed
-    ni.valid_count       = valid_count;
-    ni.value_count       = value_count;
-    s->nz_count          = valid_count;
-    s->input_value_count = value_count;
-    s->input_row_count   = value_count;
+    ni.valid_count                = valid_count;
+    ni.value_count                = value_count;
+    s->progress.nz_count          = valid_count;
+    s->progress.input_value_count = value_count;
+    s->progress.input_row_count   = value_count;
   }
 
   return valid_count;
@@ -576,16 +576,16 @@ __device__ int update_validity_and_row_indices_lists(int32_t target_value_count,
   constexpr int max_batch_size = num_warps * cudf::detail::warp_size;
 
   // how many (input) values we've processed in the page so far, prior to this loop iteration
-  int value_count = s->input_value_count;
+  int value_count = s->progress.input_value_count;
 
   // how many rows we've processed in the page so far
-  int input_row_count = s->input_row_count;
+  int input_row_count = s->progress.input_row_count;
 
   // cap by last row so that we don't process any rows past what we want to output.
   int const first_row = s->setup.first_row;
   int const last_row  = first_row + s->setup.num_rows;
 
-  int const row_index_lower_bound = s->row_index_lower_bound;
+  int const row_index_lower_bound = s->progress.row_index_lower_bound;
   int const max_depth             = s->setup.col.max_nesting_depth - 1;
   int max_depth_valid_count       = s->nesting_info[max_depth].valid_count;
 
@@ -781,11 +781,11 @@ __device__ int update_validity_and_row_indices_lists(int32_t target_value_count,
   if (t == 0) {
     // update valid value count for decoding and total # of values we've processed
     s->nesting_info[max_depth].valid_count = max_depth_valid_count;
-    s->nz_count                            = max_depth_valid_count;
-    s->input_value_count                   = value_count;
+    s->progress.nz_count                   = max_depth_valid_count;
+    s->progress.input_value_count          = value_count;
 
     // If we have lists # rows != # values
-    s->input_row_count = input_row_count;
+    s->progress.input_row_count = input_row_count;
   }
 
   return max_depth_valid_count;
@@ -797,7 +797,7 @@ inline __device__ void bool_plain_decode(page_state_s* s,
                                          int target_pos,
                                          thread_group const& group)
 {
-  int const pos = s->dict_pos;
+  int const pos = s->progress.dict_pos;
   int const t   = group.thread_rank();
   // Ensure all threads have the dict_pos
   group.sync();
@@ -856,7 +856,7 @@ __device__ void skip_ahead_in_decoding(page_state_s* s,
     if (bools_are_rle_stream) {
       skip_decode<rolling_buf_size>(bool_stream, num_to_skip, t);
     } else {
-      if (t == 0) { s->dict_pos = num_to_skip; }
+      if (t == 0) { s->progress.dict_pos = num_to_skip; }
       // For non-lists don't sync: we'll sync at the end of the lambda instead
       if constexpr (has_lists_t) { block.sync(); }
     }
@@ -897,11 +897,11 @@ __device__ void skip_ahead_in_decoding(page_state_s* s,
     auto& ni            = s->nesting_info[max_depth];
 
     // update valid value count for decoding and total # of values we've processed
-    ni.valid_count       = valid_count;
-    ni.value_count       = processed_count;
-    s->nz_count          = valid_count;
-    s->input_value_count = processed_count;
-    s->input_row_count   = processed_count;
+    ni.valid_count                = valid_count;
+    ni.value_count                = processed_count;
+    s->progress.nz_count          = valid_count;
+    s->progress.input_value_count = processed_count;
+    s->progress.input_row_count   = processed_count;
   }
   block.sync();
 }
@@ -1114,7 +1114,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8)
   int const first_row = s->setup.first_row;
   int const last_row  = first_row + s->setup.num_rows;
   while ((s->setup.error == 0) && (processed_count < s->setup.page.num_input_values) &&
-         (s->input_row_count <= last_row)) {
+         (s->progress.input_row_count <= last_row)) {
     int next_valid_count;
     block.sync();
     processed_count += min(rolling_buf_size, s->setup.page.num_input_values - processed_count);
@@ -1143,10 +1143,10 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8)
             processed_count, s, sb, nullptr, rep, t);
       } else {
         // direct copy: no nulls, no lists, no need to update validity or row indices
-        // This ASSUMES that s->row_index_lower_bound is always -1!
+        // This ASSUMES that s->progress.row_index_lower_bound is always -1!
         // Its purpose is to handle rows than span page boundaries, which only happen for lists.
         int const capped_target_value_count = min(processed_count, last_row);
-        if (t == 0) { s->input_row_count = capped_target_value_count; }
+        if (t == 0) { s->progress.input_row_count = capped_target_value_count; }
         next_valid_count = capped_target_value_count;
       }
     }
@@ -1164,7 +1164,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8)
       } else {
         auto const target_pos = next_valid_count + s->setup.page.skipped_leaf_values;
         bool_plain_decode(s, sb, target_pos, block);
-        if (t == 0) { s->dict_pos = target_pos; }
+        if (t == 0) { s->progress.dict_pos = target_pos; }
       }
       block.sync();
     }
@@ -1222,7 +1222,8 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size_t, 8)
     if constexpr (!has_lists_t) {
       if (!process_nulls) {
         if (t == 0) {
-          s->nesting_info[s->setup.col.max_nesting_depth - 1].value_count = s->input_row_count;
+          s->nesting_info[s->setup.col.max_nesting_depth - 1].value_count =
+            s->progress.input_row_count;
         }
         block.sync();
       }
diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu
index 4c743c0fb69d..f9f886cf9b02 100644
--- a/cpp/src/io/parquet/decode_preprocess.cu
+++ b/cpp/src/io/parquet/decode_preprocess.cu
@@ -69,11 +69,11 @@ __device__ void update_page_sizes(page_state_s* s,
   } temp_storage;
 
   // how many input level values we've processed in the page so far
-  int value_count = s->input_value_count;
+  int value_count = s->progress.input_value_count;
   // how many rows we've processed in the page so far
-  int row_count = s->input_row_count;
+  int row_count = s->progress.input_row_count;
   // how many leaf values we've processed in the page so far
-  int leaf_count = s->input_leaf_count;
+  int leaf_count = s->progress.input_leaf_count;
   // whether or not we need to continue checking for the first row
   bool skipped_values_set = s->setup.page.skipped_values >= 0;
 
@@ -108,7 +108,7 @@ __device__ void update_page_sizes(page_state_s* s,
 
       // if this thread is in row bounds
       int const row_index = (thread_row_count + row_count) - 1;
-      in_row_bounds       = (row_index >= s->row_index_lower_bound) &&
+      in_row_bounds       = (row_index >= s->progress.row_index_lower_bound) &&
                       (row_index < (s->setup.first_row + s->setup.num_rows));
 
       // if we have not set skipped values yet, see if we found the first in-bounds row
@@ -150,11 +150,11 @@ __device__ void update_page_sizes(page_state_s* s,
 
   // update final outputs
   if (!t) {
-    s->input_value_count = value_count;
+    s->progress.input_value_count = value_count;
 
     // only used in the skip_rows/num_rows case
-    s->input_leaf_count = leaf_count;
-    s->input_row_count  = row_count;
+    s->progress.input_leaf_count = leaf_count;
+    s->progress.input_row_count  = row_count;
   }
 
   block.sync();
@@ -288,8 +288,7 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size)
 
   // in the trim pass, for anything with lists, we only need to fully process bounding pages (those
   // at the beginning or the end of the row bounds)
-  if (!is_base_pass &&
-      !is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition)) {
+  if (!is_base_pass && !is_bounds_page(s, min_row, num_rows, has_repetition)) {
     int depth = 0;
     while (depth < s->setup.page.num_output_nesting_levels) {
       auto const thread_depth = depth + t;
@@ -297,8 +296,7 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size)
         // if we are not a bounding page (as checked above) then we are either
         // returning all rows/values from this page, or 0 of them
         pp->nesting[thread_depth].batch_size =
-          (s->setup.num_rows == 0 &&
-           !is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows))
+          (s->setup.num_rows == 0 && !is_page_contained(s, min_row, num_rows))
             ? 0
             : pp->nesting[thread_depth].size;
       }
@@ -326,15 +324,15 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size)
   if (!t) {
     s->setup.page.skipped_values      = -1;
     s->setup.page.skipped_leaf_values = 0;
-    s->input_row_count                = 0;
-    s->input_value_count              = 0;
+    s->progress.input_row_count       = 0;
+    s->progress.input_value_count     = 0;
 
     // in the base pass, we're computing the number of rows, make sure we visit absolutely
     // everything
     if (is_base_pass) {
-      s->setup.first_row       = 0;
-      s->setup.num_rows        = cuda::std::numeric_limits<int32_t>::max();
-      s->row_index_lower_bound = -1;
+      s->setup.first_row                = 0;
+      s->setup.num_rows                 = cuda::std::numeric_limits<int32_t>::max();
+      s->progress.row_index_lower_bound = -1;
     }
   }
 
diff --git a/cpp/src/io/parquet/page_data.cu b/cpp/src/io/parquet/page_data.cu
index 45ab5c9efca3..ce963ecd4f15 100644
--- a/cpp/src/io/parquet/page_data.cu
+++ b/cpp/src/io/parquet/page_data.cu
@@ -108,16 +108,17 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
 
   // skipped_leaf_values will always be 0 for flat hierarchies.
   uint32_t skipped_leaf_values = s->setup.page.skipped_leaf_values;
-  while (s->setup.error == 0 &&
-         (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) {
+  while (s->setup.error == 0 && (s->progress.input_value_count < s->setup.num_input_values ||
+                                 s->progress.src_pos < s->progress.nz_count)) {
     int target_pos;
-    int src_pos = s->src_pos;
+    int src_pos = s->progress.src_pos;
 
     if (warp.meta_group_rank() == 0) {
       target_pos = cuda::std::min(src_pos + 2 * (decode_block_size - warp.size()),
-                                  s->nz_count + (decode_block_size - warp.size()));
+                                  s->progress.nz_count + (decode_block_size - warp.size()));
     } else {
-      target_pos = cuda::std::min<int32_t>(s->nz_count, src_pos + decode_block_size - warp.size());
+      target_pos =
+        cuda::std::min<int32_t>(s->progress.nz_count, src_pos + decode_block_size - warp.size());
     }
     // This needs to be here to prevent warp 1 modifying src_pos before all threads have read it
     block.sync();
@@ -213,7 +214,9 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
         }
       }
       // Only the first thread in the warp 1 updates src_pos
-      if (warp.meta_group_rank() == 1 and warp.thread_rank() == 0) { s->src_pos = target_pos; }
+      if (warp.meta_group_rank() == 1 and warp.thread_rank() == 0) {
+        s->progress.src_pos = target_pos;
+      }
     }
     block.sync();
   }
@@ -321,18 +324,21 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
   auto const first_out_thread_id = out_warp_id * warp.size();
   // skipped_leaf_values will always be 0 for flat hierarchies.
   uint32_t skipped_leaf_values = s->setup.page.skipped_leaf_values;
-  while (s->setup.error == 0 &&
-         (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) {
+  while (s->setup.error == 0 && (s->progress.input_value_count < s->setup.num_input_values ||
+                                 s->progress.src_pos < s->progress.nz_count)) {
     int target_pos;
-    int src_pos = s->src_pos;
+    int src_pos = s->progress.src_pos;
 
     if (warp.meta_group_rank() < out_warp_id) {
-      target_pos = cuda::std::min<int32_t>(src_pos + 2 * (decode_block_size - first_out_thread_id),
-                                           s->nz_count + (decode_block_size - first_out_thread_id));
-    } else {
       target_pos =
-        cuda::std::min<int32_t>(s->nz_count, src_pos + decode_block_size - first_out_thread_id);
-      if (out_warp_id > 1) { target_pos = cuda::std::min<int32_t>(target_pos, s->dict_pos); }
+        cuda::std::min<int32_t>(src_pos + 2 * (decode_block_size - first_out_thread_id),
+                                s->progress.nz_count + (decode_block_size - first_out_thread_id));
+    } else {
+      target_pos = cuda::std::min<int32_t>(s->progress.nz_count,
+                                           src_pos + decode_block_size - first_out_thread_id);
+      if (out_warp_id > 1) {
+        target_pos = cuda::std::min<int32_t>(target_pos, s->progress.dict_pos);
+      }
     }
     // this needs to be here to prevent warp 3 modifying src_pos before all threads have read it
     block.sync();
@@ -347,10 +353,10 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
       uint32_t src_target_pos = target_pos + skipped_leaf_values;
 
       // WARP1: Decode dictionary indices, booleans or string positions
-      // NOTE: racecheck complains of a RAW error involving the s->dict_pos assignment below.
-      // This is likely a false positive in practice, but could be solved by wrapping the next
-      // 9 lines in `if (s->dict_pos < src_target_pos) {}`. If that change is made here, it will
-      // be needed in the other DecodeXXX kernels.
+      // NOTE: racecheck complains of a RAW error involving the s->progress.dict_pos assignment
+      // below. This is likely a false positive in practice, but could be solved by wrapping the
+      // next 9 lines in `if (s->progress.dict_pos < src_target_pos) {}`. If that change is made
+      // here, it will be needed in the other DecodeXXX kernels.
       if (s->stream.dict_base) {
         src_target_pos =
           decode_dictionary_indices<is_calc_sizes_only::NO>(s, sb, src_target_pos, warp).first;
@@ -360,7 +366,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
                  s->setup.col.physical_type == Type::FIXED_LEN_BYTE_ARRAY) {
         initialize_string_descriptors<is_calc_sizes_only::NO>(s, sb, src_target_pos, warp);
       }
-      if (warp.thread_rank() == 0) { s->dict_pos = src_target_pos; }
+      if (warp.thread_rank() == 0) { s->progress.dict_pos = src_target_pos; }
     } else {
       // WARP1..WARP3: Decode values
       src_pos += block.thread_rank() - first_out_thread_id;
@@ -457,7 +463,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
         }
       }
 
-      if (block.thread_rank() == first_out_thread_id) { s->src_pos = target_pos; }
+      if (block.thread_rank() == first_out_thread_id) { s->progress.src_pos = target_pos; }
     }
     __syncthreads();
   }
diff --git a/cpp/src/io/parquet/page_decode.cuh b/cpp/src/io/parquet/page_decode.cuh
index 58dc7b1b294f..a0cd9df7d537 100644
--- a/cpp/src/io/parquet/page_decode.cuh
+++ b/cpp/src/io/parquet/page_decode.cuh
@@ -47,6 +47,16 @@ struct page_decode_stream_state {
   int32_t dict_val{};
 };
 
+struct page_decode_progress_state {
+  int32_t nz_count{};
+  int32_t dict_pos{};
+  int32_t src_pos{};
+  int32_t input_value_count{};
+  int32_t input_row_count{};
+  int32_t input_leaf_count{};
+  int32_t row_index_lower_bound{};
+};
+
 struct page_state_s {
   CUDF_HOST_DEVICE constexpr page_state_s() noexcept {}
   page_decode_setup_state setup{};
@@ -55,16 +65,11 @@ struct page_state_s {
   int32_t dtype_len_in{};  // Can be larger than dtype_len if truncating 32-bit into 8-bit
 
   // (leaf) value decoding
-  int32_t nz_count{};  // number of valid entries in nz_idx (write position in circular buffer)
-  int32_t dict_pos{};  // write position of dictionary indices
-  int32_t src_pos{};   // input read position of final output value
   int32_t ts_scale{};  // timestamp scale: <0: divide by -ts_scale, >0: multiply by ts_scale
+  page_decode_progress_state progress{};
 
   // repetition/definition level decoding
-  int32_t input_value_count{};      // how many values of the input we've processed
-  int32_t input_row_count{};        // how many rows of the input we've processed
-  int32_t input_leaf_count{};       // how many leaf values of the input we've processed
-  int32_t row_index_lower_bound{};  // lower bound of row indices we should process
+  uint8_t const* lvl_start[NUM_LEVEL_TYPES]{};  // [def,rep]
 
   // a shared-memory cache of frequently used data when decoding. The source of this data is
   // normally stored in global memory which can yield poor performance. So, when possible
@@ -188,21 +193,20 @@ __device__ constexpr bool is_string_col(PageInfo const& page,
  * @brief Returns whether or not a page spans either the beginning or the end of the
  * specified row bounds
  *
- * @param page The page to be checked
- * @param chunk_start_row Absolute row index of the first row in the page's column chunk
+ * @param s Page decode state containing the page and column chunk metadata
  * @param start_row The starting row index
  * @param num_rows The number of rows
  * @param has_repetition True if the schema has nesting
  *
  * @return True if the page spans the beginning or the end of the row bounds
  */
-inline __device__ bool is_bounds_page(PageInfo const& page,
-                                      size_t chunk_start_row,
+inline __device__ bool is_bounds_page(auto* const s,
                                       size_t start_row,
                                       size_t num_rows,
                                       bool has_repetition)
 {
-  size_t const page_begin = chunk_start_row + page.chunk_row;
+  auto const& page        = s->setup.page;
+  size_t const page_begin = s->setup.col.start_row + page.chunk_row;
   size_t const page_end   = page_begin + page.num_rows;
   size_t const begin      = start_row;
   size_t const end        = start_row + num_rows;
@@ -227,19 +231,16 @@ inline __device__ bool is_bounds_page(PageInfo const& page,
  * @brief Returns whether or not a page is completely contained within the specified
  * row bounds
  *
- * @param page The page to be checked
- * @param chunk_start_row Absolute row index of the first row in the page's column chunk
+ * @param s Page decode state containing the page and column chunk metadata
  * @param start_row The starting row index
  * @param num_rows The number of rows
  *
  * @return True if the page is completely contained within the row bounds
  */
-inline __device__ bool is_page_contained(PageInfo const& page,
-                                         size_t chunk_start_row,
-                                         size_t start_row,
-                                         size_t num_rows)
+inline __device__ bool is_page_contained(auto* const s, size_t start_row, size_t num_rows)
 {
-  size_t const page_begin = chunk_start_row + page.chunk_row;
+  auto const& page        = s->setup.page;
+  size_t const page_begin = s->setup.col.start_row + page.chunk_row;
   size_t const page_end   = page_begin + page.num_rows;
   size_t const begin      = start_row;
   size_t const end        = start_row + num_rows;
@@ -257,21 +258,20 @@ inline __device__ bool is_page_contained(PageInfo const& page,
  * carry values while containing zero of its own rows. Such a page must still be processed when
  * it spans (is a "bounds" page for) or is fully contained within the requested range.
  *
- * @param page The page to be checked
- * @param chunk_start_row Absolute row index of the first row in the page's column chunk
+ * @param s Page decode state containing the page and column chunk metadata
  * @param min_row Absolute index of the first requested row
  * @param num_rows Number of requested rows
  * @param has_repetition True if the schema has nesting (list) columns
  *
  * @return True if the page has rows/values to process for the requested range
  */
-inline __device__ bool page_has_rows_to_process(PageInfo const& page,
-                                                size_t chunk_start_row,
+inline __device__ bool page_has_rows_to_process(auto* const s,
                                                 size_t min_row,
                                                 size_t num_rows,
                                                 bool has_repetition)
 {
-  size_t const page_start_row = chunk_start_row + page.chunk_row;
+  auto const& page            = s->setup.page;
+  size_t const page_start_row = s->setup.col.start_row + page.chunk_row;
   size_t const page_end_row   = page_start_row + page.num_rows;
   size_t const end_row        = min_row + num_rows;
 
@@ -282,8 +282,30 @@ inline __device__ bool page_has_rows_to_process(PageInfo const& page,
 
   // A single list row can span pages, so a list page can carry values (and offsets) with 0 rows;
   // such a page carries no rows of its own but must still be processed.
-  return is_bounds_page(page, chunk_start_row, min_row, num_rows, has_repetition) ||
-         is_page_contained(page, chunk_start_row, min_row, num_rows);
+  return is_bounds_page(s, min_row, num_rows, has_repetition) ||
+         is_page_contained(s, min_row, num_rows);
+}
+
+inline __device__ bool page_has_rows_to_process(PageInfo const& page,
+                                                size_t chunk_start_row,
+                                                size_t min_row,
+                                                size_t num_rows,
+                                                bool has_repetition)
+{
+  size_t const page_start_row = chunk_start_row + page.chunk_row;
+  size_t const page_end_row   = page_start_row + page.num_rows;
+  size_t const end_row        = min_row + num_rows;
+  bool const has_rows =
+    (page.num_rows > 0) && (page_start_row < end_row) && (page_end_row > min_row);
+  if (has_rows || !has_repetition) { return has_rows; }
+
+  auto const test_page_end_nonlists =
+    page.is_num_rows_adjusted ? page_end_row >= end_row : page_end_row > end_row;
+  auto const bounds = ((page_start_row <= min_row and page_end_row >= min_row) or
+                       (page_start_row <= end_row and page_end_row >= end_row)) or
+                      ((page_start_row < min_row and page_end_row > min_row) or
+                       (page_start_row < end_row and test_page_end_nonlists));
+  return bounds || (page_start_row >= min_row && page_end_row <= end_row);
 }
 
 /**
@@ -354,13 +376,13 @@ __device__ cuda::std::pair<int, int> decode_dictionary_indices(
 {
   uint8_t const* end = s->stream.data_end;
   int dict_bits      = s->stream.dict_bits;
-  int pos            = s->dict_pos;
+  int pos            = s->progress.dict_pos;
   int str_len        = 0;
   int const t        = warp.thread_rank();
 
-  // NOTE: racecheck warns about a RAW involving s->dict_pos, which is likely a false
+  // NOTE: racecheck warns about a RAW involving s->progress.dict_pos, which is likely a false
   // positive because the only path that does not include a sync will lead to
-  // s->dict_pos being overwritten with the same value
+  // s->progress.dict_pos being overwritten with the same value
 
   while (pos < target_pos) {
     int is_literal, batch_len;
@@ -480,12 +502,12 @@ inline __device__ int decode_rle_booleans(
   cg::thread_block_tile<cudf::detail::warp_size, cg::thread_block> const& warp)
 {
   uint8_t const* end = s->stream.data_end;
-  int64_t pos        = s->dict_pos;
+  int64_t pos        = s->progress.dict_pos;
   int const t        = warp.thread_rank();
 
-  // NOTE: racecheck warns about a RAW involving s->dict_pos, which is likely a false positive
-  // because the only path that does not include a sync will lead to s->dict_pos being overwritten
-  // with the same value
+  // NOTE: racecheck warns about a RAW involving s->progress.dict_pos, which is likely a false
+  // positive because the only path that does not include a sync will lead to s->progress.dict_pos
+  // being overwritten with the same value
 
   while (pos < target_pos) {
     int is_literal, batch_len;
@@ -561,7 +583,7 @@ __device__ size_type initialize_string_descriptors(page_state_s* s,
   int const t         = group.thread_rank();
   int const dict_size = s->stream.dict_size;
   int k               = s->stream.dict_val;
-  int pos             = s->dict_pos;
+  int pos             = s->progress.dict_pos;
   int total_len       = 0;
 
   // All group threads can participate for fixed len byte arrays.
@@ -750,15 +772,15 @@ __device__ void gpuUpdateValidityOffsetsAndRowIndices(int32_t target_input_value
                                                       int t)
 {
   // exit early if there's no work to do
-  if (s->input_value_count >= target_input_value_count) { return; }
+  if (s->progress.input_value_count >= target_input_value_count) { return; }
 
   // max nesting depth of the column
   int const max_depth       = s->setup.col.max_nesting_depth;
   bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0;
   // how many (input) values we've processed in the page so far
-  int input_value_count = s->input_value_count;
+  int input_value_count = s->progress.input_value_count;
   // how many rows we've processed in the page so far
-  int input_row_count = s->input_row_count;
+  int input_row_count = s->progress.input_row_count;
 
   PageNestingDecodeInfo* nesting_info_base = s->nesting_info;
 
@@ -787,7 +809,7 @@ __device__ void gpuUpdateValidityOffsetsAndRowIndices(int32_t target_input_value
       input_row_count + ((__popc(warp_row_count_mask & ((1 << t) - 1)) + is_new_row) - 1);
     input_row_count += __popc(warp_row_count_mask);
     // is this thread within read row bounds?
-    int const in_row_bounds = thread_row_index >= s->row_index_lower_bound &&
+    int const in_row_bounds = thread_row_index >= s->progress.row_index_lower_bound &&
                                   thread_row_index < (s->setup.first_row + s->setup.num_rows)
                                 ? 1
                                 : 0;
@@ -902,9 +924,9 @@ __device__ void gpuUpdateValidityOffsetsAndRowIndices(int32_t target_input_value
   // update
   if (!t) {
     // update valid value count for decoding and total # of values we've processed
-    s->nz_count          = nesting_info_base[max_depth - 1].valid_count;
-    s->input_value_count = input_value_count;
-    s->input_row_count   = input_row_count;
+    s->progress.nz_count          = nesting_info_base[max_depth - 1].valid_count;
+    s->progress.input_value_count = input_value_count;
+    s->progress.input_row_count   = input_row_count;
   }
 }
 
@@ -939,8 +961,8 @@ __device__ void gpuDecodeLevels(
   cg::thread_block_tile<cudf::detail::warp_size, cg::thread_block> const& warp)
 {
   auto cur_leaf_count = target_leaf_count;
-  while (s->setup.error == 0 && s->nz_count < target_leaf_count &&
-         s->input_value_count < s->setup.num_input_values) {
+  while (s->setup.error == 0 && s->progress.nz_count < target_leaf_count &&
+         s->progress.input_value_count < s->setup.num_input_values) {
     // because the rep and def streams are encoded separately, we cannot request an exact
     // # of values to be decoded at once. we can only process the lowest # of decoded rep/def
     // levels we get.
@@ -1188,8 +1210,7 @@ inline __device__ bool setup_local_page_info(auto* const s,
   // NOTE: this check needs to be done after the null counts have been zeroed out
   bool const has_repetition = s->setup.col.max_level[level_type::REPETITION] > 0;
   if ((stage == page_processing_stage::STRING_BOUNDS || stage == page_processing_stage::DECODE) &&
-      !page_has_rows_to_process(
-        s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition)) {
+      !page_has_rows_to_process(s, min_row, num_rows, has_repetition)) {
     return false;
   }
 
@@ -1403,10 +1424,10 @@ inline __device__ bool setup_local_page_info(auto* const s,
     }
 
     s->setup.num_input_values = s->setup.page.num_input_values;
-    if constexpr (requires { s->input_value_count; }) {
-      s->nz_count = 0;
-      s->dict_pos = 0;
-      s->src_pos  = 0;
+    if constexpr (requires { s->progress.input_value_count; }) {
+      s->progress.nz_count = 0;
+      s->progress.dict_pos = 0;
+      s->progress.src_pos  = 0;
 
       // for flat hierarchies, we can't know how many leaf values to skip unless we do a full
       // preprocess of the definition levels (since nulls will have no actual decodable value, there
@@ -1416,19 +1437,19 @@ inline __device__ bool setup_local_page_info(auto* const s,
       if (s->setup.col.max_level[level_type::REPETITION] == 0) {
         s->setup.page.skipped_values      = 0;
         s->setup.page.skipped_leaf_values = 0;
-        s->input_value_count              = 0;
-        s->input_row_count                = 0;
-        s->input_leaf_count               = 0;
+        s->progress.input_value_count     = 0;
+        s->progress.input_row_count       = 0;
+        s->progress.input_leaf_count      = 0;
 
         // The fixed-width decode kernel ASSUMES this is always -1 for non-lists!
-        s->row_index_lower_bound = -1;
+        s->progress.row_index_lower_bound = -1;
       }
       // for nested hierarchies, we have run a preprocess that lets us skip directly to the values
       // we need to start decoding at
       else {
         // input_row_count translates to "how many rows we have processed so far", so since we are
         // skipping directly to where we want to start decoding, set it to first_row
-        s->input_row_count = s->setup.first_row;
+        s->progress.input_row_count = s->setup.first_row;
 
         // return the lower bound to compare (page-relative) thread row index against. Explanation:
         // In the case of nested schemas, rows can span page boundaries.  That is to say,
@@ -1441,25 +1462,25 @@ inline __device__ bool setup_local_page_info(auto* const s,
         // relative row index -1
         int const max_row = (min_row + num_rows) - 1;
         if (min_row < page_start_row && max_row >= page_start_row - 1) {
-          s->row_index_lower_bound = -1;
+          s->progress.row_index_lower_bound = -1;
         } else {
-          s->row_index_lower_bound = s->setup.first_row;
+          s->progress.row_index_lower_bound = s->setup.first_row;
         }
 
         // if we're in the decoding step, jump directly to the first
         // value we care about
         if (stage == page_processing_stage::DECODE) {
-          s->input_value_count =
+          s->progress.input_value_count =
             s->setup.page.skipped_values > -1 ? s->setup.page.skipped_values : 0;
         } else if (stage == page_processing_stage::PREPROCESS) {
-          s->input_value_count = 0;
-          s->input_leaf_count  = 0;
+          s->progress.input_value_count = 0;
+          s->progress.input_leaf_count  = 0;
           // magic number to indicate it hasn't been set for use inside UpdatePageSizes
           s->setup.page.skipped_values      = -1;
           s->setup.page.skipped_leaf_values = 0;
         }
       }
-    }  // if constexpr (requires { s->input_value_count; })
+    }  // if constexpr (requires { s->progress.input_value_count; })
 
     __threadfence_block();
   }
diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu
index 8baead9eb596..98dc64ab154f 100644
--- a/cpp/src/io/parquet/page_delta_decode.cu
+++ b/cpp/src/io/parquet/page_delta_decode.cu
@@ -372,15 +372,15 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size)
   // that has a value we need.
   if (skipped_leaf_values > 0) { db->skip_values(skipped_leaf_values); }
 
-  while (s->setup.error == 0 &&
-         (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) {
+  while (s->setup.error == 0 && (s->progress.input_value_count < s->setup.num_input_values ||
+                                 s->progress.src_pos < s->progress.nz_count)) {
     uint32_t target_pos;
-    uint32_t const src_pos = s->src_pos;
+    uint32_t const src_pos = s->progress.src_pos;
 
     if (warp.meta_group_rank() < 2) {  // warp0..1
-      target_pos = min(src_pos + 2 * batch_size, s->nz_count + batch_size);
+      target_pos = min(src_pos + 2 * batch_size, s->progress.nz_count + batch_size);
     } else {  // warp2
-      target_pos = min(s->nz_count, src_pos + batch_size);
+      target_pos = min(s->progress.nz_count, src_pos + batch_size);
     }
     // This needs to be here to prevent warp 2 modifying src_pos before all threads have read it
     block.sync();
@@ -424,7 +424,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size)
           }
         }
       }
-      if (warp.thread_rank() == 0) { s->src_pos = src_pos + batch_size; }
+      if (warp.thread_rank() == 0) { s->progress.src_pos = src_pos + batch_size; }
     }
 
     block.sync();
@@ -562,20 +562,20 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
 
   // if this is a bounds page and nested, then we need to skip up front. non-nested will work
   // its way through the page.
-  int string_pos = has_repetition ? s->setup.page.start_val : 0;
-  auto const is_bounds_pg =
-    is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition);
+  int string_pos          = has_repetition ? s->setup.page.start_val : 0;
+  auto const is_bounds_pg = is_bounds_page(s, min_row, num_rows, has_repetition);
   if (is_bounds_pg && string_pos > 0) { dba->skip(use_char_ll); }
 
-  while (!s->setup.error &&
-         (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) {
+  while (!s->setup.error && (s->progress.input_value_count < s->setup.num_input_values ||
+                             s->progress.src_pos < s->progress.nz_count)) {
     uint32_t target_pos;
-    uint32_t const src_pos = s->src_pos;
+    uint32_t const src_pos = s->progress.src_pos;
 
     if (warp.meta_group_rank() < 3) {  // warp 0..2
-      target_pos = min(src_pos + 2 * batch_size, s->nz_count + s->setup.first_row + batch_size);
+      target_pos =
+        min(src_pos + 2 * batch_size, s->progress.nz_count + s->setup.first_row + batch_size);
     } else {  // warp 3
-      target_pos = min(s->nz_count, src_pos + batch_size);
+      target_pos = min(s->progress.nz_count, src_pos + batch_size);
     }
     // this needs to be here to prevent warp 3 modifying src_pos before all threads have read it
     block.sync();
@@ -622,7 +622,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
         warp.sync();
       }
 
-      if (warp.thread_rank() == 0) { s->src_pos = src_pos + batch_size; }
+      if (warp.thread_rank() == 0) { s->progress.src_pos = src_pos + batch_size; }
     }
 
     block.sync();
@@ -752,8 +752,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
   // if this is a bounds page, then we need to decode up to the first mini-block
   // that has a value we need, and set string_offset to the position of the first value in the
   // string data block.
-  auto const is_bounds_pg =
-    is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition);
+  auto const is_bounds_pg = is_bounds_page(s, min_row, num_rows, has_repetition);
   if (is_bounds_pg && s->setup.page.start_val > 0) {
     if (warp.meta_group_rank() == 0) {
       // string_off is only valid on thread 0
@@ -774,15 +773,15 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
 
   int string_pos = has_repetition ? s->setup.page.start_val : 0;
 
-  while (!s->setup.error &&
-         (s->input_value_count < s->setup.num_input_values || s->src_pos < s->nz_count)) {
+  while (!s->setup.error && (s->progress.input_value_count < s->setup.num_input_values ||
+                             s->progress.src_pos < s->progress.nz_count)) {
     uint32_t target_pos;
-    uint32_t const src_pos = s->src_pos;
+    uint32_t const src_pos = s->progress.src_pos;
 
     if (warp.meta_group_rank() < 2) {  // warp0..1
-      target_pos = min(src_pos + 2 * batch_size, s->nz_count + batch_size);
+      target_pos = min(src_pos + 2 * batch_size, s->progress.nz_count + batch_size);
     } else {  // warp2
-      target_pos = min(s->nz_count, src_pos + batch_size);
+      target_pos = min(s->progress.nz_count, src_pos + batch_size);
     }
     // this needs to be here to prevent warp 2 modifying src_pos before all threads have read it
     __syncthreads();
@@ -824,7 +823,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size)
         warp.sync();
       }
 
-      if (warp.thread_rank() == 0) { s->src_pos = src_pos + batch_size; }
+      if (warp.thread_rank() == 0) { s->progress.src_pos = src_pos + batch_size; }
     }
     block.sync();
   }
diff --git a/cpp/src/io/parquet/page_state_composed.cuh b/cpp/src/io/parquet/page_state_composed.cuh
index 70a6213567bc..33c3cf0efe3b 100644
--- a/cpp/src/io/parquet/page_state_composed.cuh
+++ b/cpp/src/io/parquet/page_state_composed.cuh
@@ -32,6 +32,17 @@ struct level_scan_state {
   CUDF_PARQUET_PAGE_STATE_ERROR_METHODS
 };
 
+// Shared memory state struct used by the preprocess_string_offsets kernel.
+// Includes setup (page metadata + error), stream (page bytes + dictionary), and
+// progress (input counters) because this pass scans flat string payloads while tracking counts.
+struct string_offset_scan_state {
+  page_decode_setup_state setup;
+  page_decode_stream_state stream;
+  page_decode_progress_state progress;
+  CUDF_PARQUET_PAGE_STATE_ERROR_METHODS
+};
+static_assert(sizeof(string_offset_scan_state) < sizeof(page_state_s),
+              "string_offset_scan_state did not shrink after removing output conversion state");
 #undef CUDF_PARQUET_PAGE_STATE_ERROR_METHODS
 
 }  // namespace cudf::io::parquet::detail
diff --git a/cpp/src/io/parquet/page_string_decode.cu b/cpp/src/io/parquet/page_string_decode.cu
index b3559bfae496..4d2428b44a5c 100644
--- a/cpp/src/io/parquet/page_string_decode.cu
+++ b/cpp/src/io/parquet/page_string_decode.cu
@@ -6,6 +6,7 @@
 #include "delta_binary.cuh"
 #include "error.hpp"
 #include "page_decode.cuh"
+#include "page_state_composed.cuh"
 #include "page_string_utils.cuh"
 
 #include <cudf/detail/algorithms/reduce.cuh>
@@ -549,8 +550,7 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size)
     return;
   }
 
-  bool const is_bounds_pg =
-    is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition);
+  bool const is_bounds_pg = is_bounds_page(s, min_row, num_rows, has_repetition);
 
   // if we have size info, then we only need to do this for bounds pages
   if (pp->has_value_info && !is_bounds_pg) { return; }
@@ -642,13 +642,12 @@ CUDF_KERNEL void __launch_bounds__(delta_preproc_block_size)
       }
     }
   } else {
-    bool const is_bounds_pg =
-      is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition);
+    bool const is_bounds_pg = is_bounds_page(s, min_row, num_rows, has_repetition);
 
     // if we have size info, then we only need to do this for bounds pages
     if (pp->has_value_info && !is_bounds_pg) {
       // check if we need to store values from the index
-      if (t == 0 && is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows)) {
+      if (t == 0 && is_page_contained(s, min_row, num_rows)) {
         pp->str_bytes = pp->str_bytes_from_index;
       }
       return;
@@ -724,13 +723,12 @@ CUDF_KERNEL void __launch_bounds__(delta_length_block_size)
     return;
   }
 
-  bool const is_bounds_pg =
-    is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition);
+  bool const is_bounds_pg = is_bounds_page(s, min_row, num_rows, has_repetition);
 
   // if we have size info, then we only need to do this for bounds pages
   if (pp->has_value_info && !is_bounds_pg) {
     // check if we need to store values from the index
-    if (t == 0 && is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows)) {
+    if (t == 0 && is_page_contained(s, min_row, num_rows)) {
       pp->str_bytes = pp->str_bytes_from_index;
     }
     return;
@@ -839,13 +837,12 @@ CUDF_KERNEL void __launch_bounds__(preprocess_block_size)
     return;
   }
 
-  bool const is_bounds_pg =
-    is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition);
+  bool const is_bounds_pg = is_bounds_page(s, min_row, num_rows, has_repetition);
 
   // if we have size info, then we only need to do this for bounds pages
   if (pp->has_value_info && !is_bounds_pg) {
     // check if we need to store values from the index
-    if (t == 0 && is_page_contained(s->setup.page, s->setup.col.start_row, min_row, num_rows)) {
+    if (t == 0 && is_page_contained(s, min_row, num_rows)) {
       pp->str_bytes = pp->str_bytes_from_index;
     }
     return;
@@ -1104,7 +1101,7 @@ inline __device__ bool prefetch_string_data(int t,
  * @param error_code Error code to set if a string length overruns the page
  */
 template <int32_t block_size, size_t prefetch_size>
-inline __device__ void read_string_offsets_buffered(page_state_s* s,
+inline __device__ void read_string_offsets_buffered(auto* s,
                                                     size_t num_values_to_process,
                                                     uint32_t* str_offsets,
                                                     kernel_error::pointer error_code)
@@ -1197,7 +1194,7 @@ inline __device__ void read_string_offsets_buffered(page_state_s* s,
  * @param error_code Error code to set if a string length overruns the page
  */
 template <int32_t block_size>
-inline __device__ void read_string_offsets_sequential(page_state_s* s,
+inline __device__ void read_string_offsets_sequential(auto* s,
                                                       size_t num_values_to_process,
                                                       uint32_t* str_offsets,
                                                       kernel_error::pointer error_code)
@@ -1304,8 +1301,8 @@ CUDF_KERNEL void preprocess_string_offsets_kernel(
           decode_kernel_mask::STRING_STREAM_SPLIT_NESTED,
           decode_kernel_mask::STRING_STREAM_SPLIT_LIST);
 
-  __shared__ __align__(16) page_state_s state_g;
-  page_state_s* const s = &state_g;
+  __shared__ __align__(16) string_offset_scan_state state_g;
+  auto* const s = &state_g;
   if (!setup_local_page_info(s,
                              pp,
                              chunks,