Fix test_parquet_interleaved_file_splits_partition_value_alignment on GCS again [databricks] (NVIDIA#15289)

Fixes NVIDIA#15256.

### Description

`test_parquet_interleaved_file_splits_partition_value_alignment` fails
on Dataproc when the Spark temporary directory uses HDFS. The test reads
Parquet row-group metadata with PyArrow, whose HDFS filesystem
implementation uses `libhdfs` and requires the `CLASSPATH` environment
variable to contain the Hadoop JARs. Without that environment variable,
the test fails with `OSError: HDFS connection failed`.

This change replaces the PyArrow metadata reader with Spark's JVM-side
`ParquetFileReader`. It uses the active Spark session's Hadoop
configuration, allowing the helper to access files through the same
filesystem implementation Spark uses.

The row-group midpoint calculation now uses
`BlockMetaData.getStartingPos()` and
`BlockMetaData.getCompressedSize()`, which provide the same offsets and
compressed-size information used by the previous PyArrow implementation.
The Parquet reader is closed in a `finally` block.

### Checklists

Documentation
- [ ] Updated for new or modified user-facing features or behaviors
- [x] No user-facing change

Testing
- [x] Added or modified tests to cover new code paths
- [ ] Covered by existing tests
(Please provide the names of the existing tests in the PR description.)
- [ ] Not required

Performance
- [ ] Tests ran and results are added in the PR description
- [ ] Issue filed with a link in the PR description
- [x] Not required

Signed-off-by: Jihoon Son <ghoonson@gmail.com>

Author: jihoonson

integration_tests/src/main/python/delta_lake_test.py

@@ -693,8 +693,11 @@ def setup_table(spark):
         f"b={b_size}, max_split={max_split}")
 
     a_tail_start = a_size - a_tail
-    a_midpoints = parquet_row_group_midpoints(a_path)
-    b_midpoints = parquet_row_group_midpoints(b_path)
+    a_midpoints, b_midpoints = with_cpu_session(
+        lambda spark: (
+            parquet_row_group_midpoints(spark, a_path),
+            parquet_row_group_midpoints(spark, b_path),
+        ))
     assert any(a_tail_start <= midpoint < a_size for midpoint in a_midpoints), (
         f"A tail split [{a_tail_start}, {a_size}) has no row-group midpoint; "
         f"midpoints={a_midpoints}")

integration_tests/src/main/python/parquet_test.py

@@ -1020,8 +1020,11 @@ def parquet_file_info_by_part(spark):
         f"b={b_size}, max_split={max_split}")
 
     a_tail_start = a_size - a_tail
-    a_midpoints = parquet_row_group_midpoints(a_path)
-    b_midpoints = parquet_row_group_midpoints(b_path)
+    a_midpoints, b_midpoints = with_cpu_session(
+        lambda spark: (
+            parquet_row_group_midpoints(spark, a_path),
+            parquet_row_group_midpoints(spark, b_path),
+        ))
     assert any(a_tail_start <= midpoint < a_size for midpoint in a_midpoints), (
         f"A tail split [{a_tail_start}, {a_size}) has no row-group midpoint; "
         f"midpoints={a_midpoints}")

integration_tests/src/main/python/parquet_test_utils.py

@@ -12,29 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from urllib.parse import urlparse
-
-import pyarrow.fs as pa_fs
-import pyarrow.parquet as pa_pq
-
-
-def parquet_row_group_midpoints(path):
+def parquet_row_group_midpoints(spark, path):
     """Returns an approximate byte midpoint for each Parquet row group."""
-    if urlparse(path).scheme:
-        filesystem, path = pa_fs.FileSystem.from_uri(path)
-        meta = pa_pq.read_metadata(path, filesystem=filesystem)
-    else:
-        meta = pa_pq.read_metadata(path)
-    midpoints = []
-    for rg_index in range(meta.num_row_groups):
-        row_group = meta.row_group(rg_index)
-        first_col = row_group.column(0)
-        start = first_col.data_page_offset
-        dict_offset = first_col.dictionary_page_offset
-        if dict_offset is not None and dict_offset > 0:
-            start = min(start, dict_offset)
-        total_size = 0
-        for col_index in range(row_group.num_columns):
-            total_size += row_group.column(col_index).total_compressed_size
-        midpoints.append(start + total_size // 2)
-    return midpoints
+    jvm = spark.sparkContext._jvm
+    hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
+    hadoop_path = jvm.org.apache.hadoop.fs.Path(path)
+    reader = jvm.org.apache.parquet.hadoop.ParquetFileReader.open(
+        hadoop_conf, hadoop_path)
+    try:
+        blocks = reader.getFooter().getBlocks()
+        return [
+            block.getStartingPos() + block.getCompressedSize() // 2
+            for block in blocks
+        ]
+    finally:
+        reader.close()