fix: Fix panic reading empty parquet with multiple boolean columns (#23159)

Author: nameexhaustion

crates/polars-io/src/cloud/polars_object_store.rs

@@ -2,10 +2,12 @@ use std::ops::Range;
 
 use bytes::Bytes;
 use futures::{StreamExt, TryStreamExt};
+use hashbrown::hash_map::RawEntryMut;
 use object_store::path::Path;
 use object_store::{ObjectMeta, ObjectStore};
 use polars_core::prelude::{InitHashMaps, PlHashMap};
 use polars_error::{PolarsError, PolarsResult};
+use polars_utils::mmap::MemSlice;
 use tokio::io::{AsyncSeekExt, AsyncWriteExt};
 
 use crate::pl_async::{
@@ -214,14 +216,11 @@ impl PolarsObjectStore {
     ///
     /// # Panics
     /// Panics if the same range start is used by more than 1 range.
-    pub async fn get_ranges_sort<
-        K: TryFrom<usize, Error = impl std::fmt::Debug> + std::hash::Hash + Eq,
-        T: From<Bytes>,
-    >(
+    pub async fn get_ranges_sort(
         &self,
         path: &Path,
         ranges: &mut [Range<usize>],
-    ) -> PolarsResult<PlHashMap<K, T>> {
+    ) -> PolarsResult<PlHashMap<usize, MemSlice>> {
         if ranges.is_empty() {
             return Ok(Default::default());
         }
@@ -280,16 +279,23 @@ impl PolarsObjectStore {
 
                             assert_eq!(bytes.len(), full_range.len());
 
+                            let bytes = MemSlice::from_bytes(bytes);
+
                             for range in &ranges[current_offset..end] {
-                                let v = out.insert(
-                                    K::try_from(range.start).unwrap(),
-                                    T::from(bytes.slice(
-                                        range.start - full_range.start
-                                            ..range.end - full_range.start,
-                                    )),
+                                let mem_slice = bytes.slice(
+                                    range.start - full_range.start..range.end - full_range.start,
                                 );
 
-                                assert!(v.is_none()); // duplicate range start
+                                match out.raw_entry_mut().from_key(&range.start) {
+                                    RawEntryMut::Vacant(slot) => {
+                                        slot.insert(range.start, mem_slice);
+                                    },
+                                    RawEntryMut::Occupied(mut slot) => {
+                                        if slot.get_mut().len() < mem_slice.len() {
+                                            *slot.get_mut() = mem_slice;
+                                        }
+                                    },
+                                }
                             }
 
                             current_offset = end;

py-polars/tests/unit/io/test_parquet.py

@@ -3281,7 +3281,7 @@ def test_parquet_read_timezone_22506() -> None:
 
 @pytest.mark.parametrize("static", [True, False])
 @pytest.mark.parametrize("lazy", [True, False])
-def test_read_write_metadata(tmp_path: Path, static: bool, lazy: bool) -> None:
+def test_read_write_metadata(static: bool, lazy: bool) -> None:
     metadata = {"hello": "world", "something": "else"}
     md: ParquetMetadata = metadata
     if not static:
@@ -3378,3 +3378,20 @@ def multiple_test_sorting_columns() -> None:
     assert roundtrip.get_column("a").is_sorted()
     assert not roundtrip.get_column("b").is_sorted()
     assert_frame_equal(roundtrip.sort("b"), df.sort("b"))
+
+
+@pytest.mark.write_disk
+def test_read_parquet_duplicate_range_start_fetch_23139(tmp_path: Path) -> None:
+    tmp_path.mkdir(exist_ok=True)
+    path = tmp_path / "data.parquet"
+
+    df = pl.DataFrame(
+        schema={
+            "a": pl.Boolean,
+            "b": pl.Boolean,
+        }
+    )
+
+    df.write_parquet(path, use_pyarrow=True)
+
+    assert_frame_equal(pl.read_parquet(path), df)