fix: Dropped/shifted rows in parquet scan with streaming=True (#18766)

crates/polars-io/src/parquet/read/read_impl.rs

@@ -604,18 +604,15 @@ fn rg_to_dfs_par_over_rg(
             continue;
         }
 
-        row_groups.push((i, rg_md, rg_slice, row_count_start));
+        row_groups.push((rg_md, rg_slice, row_count_start));
     }
 
     let dfs = POOL.install(|| {
         // Set partitioned fields to prevent quadratic behavior.
         // Ensure all row groups are partitioned.
         row_groups
             .into_par_iter()
-            .enumerate()
-            .map(|(iter_idx, (_rg_idx, _md, slice, row_count_start))| {
-                let md = &file_metadata.row_groups[iter_idx];
-
+            .map(|(md, slice, row_count_start)| {
                 if slice.1 == 0 || use_statistics && !read_this_row_group(predicate, md, schema)? {
                     return Ok(None);
                 }

py-polars/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "py-polars"
-version = "1.7.1"
+version = "1.7.2"
 edition = "2021"
 
 [lib]

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

@@ -529,3 +529,15 @@ def trim_to_metadata(path: str | Path) -> None:
         assert_frame_equal(
             pl.scan_parquet(path).slice(-1, (1 << 32) - 1).collect(), df.tail(1)
         )
+
+
+@pytest.mark.parametrize("streaming", [True, False])
+def test_parquet_row_groups_shift_bug_18739(tmp_path: Path, streaming: bool) -> None:
+    tmp_path.mkdir(exist_ok=True)
+    path = tmp_path / "data.bin"
+
+    df = pl.DataFrame({"id": range(100)})
+    df.write_parquet(path, row_group_size=1)
+
+    lf = pl.scan_parquet(path)
+    assert_frame_equal(df, lf.collect(streaming=streaming))