ARROW-9790: [Rust][Parquet] Fix PrimitiveArrayReader boundary conditions

rust/parquet/src/arrow/array_reader.rs

@@ -136,10 +136,8 @@ impl<T: DataType> ArrayReader for PrimitiveArrayReader<T> {
         while records_read < batch_size {
             let records_to_read = batch_size - records_read;
 
+            // NB can be 0 if at end of page
             let records_read_once = self.record_reader.read_records(records_to_read)?;
-            if records_read_once == 0 {
-                break; // record reader has no record
-            }
             records_read = records_read + records_read_once;
 
             // Record reader exhausted

rust/parquet/src/arrow/arrow_reader.rs

@@ -304,6 +304,45 @@ mod tests {
         >(2, 100, 2, message_type, 15, 50, converter);
     }
 
+    #[test]
+    fn test_bool_single_column_reader_test_batch_size_divides_into_row_group_size() {
+        let message_type = "
+        message test_schema {
+          REQUIRED BOOLEAN leaf;
+        }
+        ";
+
+        // Use a batch size (5) so batches to fall on
+        // row group boundaries (25 rows in 3 row groups --> row
+        // groups of 10, 10, and 5) to test edge refilling edge cases.
+        let converter = FromConverter::new();
+        single_column_reader_test::<
+            BoolType,
+            BooleanArray,
+            FromConverter<Vec<Option<bool>>, BooleanArray>,
+            BoolType,
+        >(3, 25, 2, message_type, 5, 50, converter);
+    }
+
+    #[test]
+    fn test_bool_single_column_reader_test_batch_size_divides_into_row_group_size2() {
+        let message_type = "
+        message test_schema {
+          REQUIRED BOOLEAN leaf;
+        }
+        ";
+
+        // Ensure that every batch size (25) falls exactly a row group
+        // boundary (25 in this case) to test edge case.
+        let converter = FromConverter::new();
+        single_column_reader_test::<
+            BoolType,
+            BooleanArray,
+            FromConverter<Vec<Option<bool>>, BooleanArray>,
+            BoolType,
+        >(4, 100, 2, message_type, 25, 50, converter);
+    }
+
     struct RandFixedLenGen {}
 
     impl RandGen<FixedLenByteArrayType> for RandFixedLenGen {