Fix Parquet UNKNOWN annotation IT writes on Dataproc[databricks] (NVIDIA#15428)

fixes NVIDIA#15426.

### Description

- Dataproc nightlies failed in
`test_parquet_unknown_type_annotation_pre_411_physical` because PyArrow
wrote directly to `spark_tmp_path`, which is created via Hadoop
`FileSystem.mkdirs` (GCS on Dataproc) rather than the local filesystem;
write the sample Parquet file locally and `copy_from_local` into the
Hadoop path so cluster reads succeed.
- Move the existing `copy_from_local` helper from
`fastparquet_compatibility_test.py` into `parquet_test_utils.py` and
reuse it from both call sites to avoid duplicating the Hadoop copy path.

### 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: Firestarman <firestarmanllc@gmail.com>

Author: firestarman

integration_tests/src/main/python/fastparquet_compatibility_test.py

@@ -17,6 +17,7 @@
 from asserts import assert_gpu_and_cpu_are_equal_collect
 from data_gen import *
 from fastparquet_utils import get_fastparquet_result_canonicalizer
+from parquet_test_utils import copy_from_local
 from spark_session import is_databricks_runtime, spark_version, with_cpu_session, with_gpu_session
 
 
@@ -247,17 +248,6 @@ def test_reading_file_written_with_gpu(spark_tmp_path, column_gen):
         delete_local_directory(local_base_path)
 
 
-def copy_from_local(spark, local_source, hdfs_target):
-    """
-    Copies contents of local_source to hdfs_target.
-    """
-    sc = spark.sparkContext
-    Path = sc._jvm.org.apache.hadoop.fs.Path
-    config = sc._jsc.hadoopConfiguration()
-    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(config)
-    fs.copyFromLocalFile(Path(local_source), Path(hdfs_target))
-
-
 @pytest.mark.skipif(condition=fastparquet_unavailable(),
                     reason="fastparquet is required for testing fastparquet compatibility")
 @pytest.mark.parametrize('column_gen', [

integration_tests/src/main/python/parquet_test.py

@@ -13,6 +13,8 @@
 # limitations under the License.
 import os
 import re
+import shutil
+import tempfile
 
 import pytest
 
@@ -23,7 +25,7 @@
 from marks import *
 import pyarrow as pa
 import pyarrow.parquet as pq
-from parquet_test_utils import parquet_row_group_midpoints
+from parquet_test_utils import copy_from_local, parquet_row_group_midpoints
 from pyspark.sql.types import *
 from pyspark.sql.functions import *
 from spark_init_internal import spark_version
@@ -1996,7 +1998,12 @@ def setup_table(spark):
 
 def _write_parquet_unknown_null_table(
         data_path, with_list=False, with_map=False, field_id=None):
-    """Write INT32 physical + UNKNOWN/Null logical annotation (Spark void_in_parquet shape)."""
+    """Write INT32 physical + UNKNOWN/Null logical annotation (Spark void_in_parquet shape).
+
+    PyArrow writes only to the local filesystem, while spark_tmp_path is created via
+    Hadoop FileSystem.mkdirs. On Dataproc the default FS is typically GCS, so writing
+    directly to data_path fails with FileNotFoundError. Write locally then copy.
+    """
     if with_list:
         table = pa.table({
             'list_void': pa.array([[None, None], [None], None], type=pa.list_(pa.null())),
@@ -2018,7 +2025,13 @@ def _write_parquet_unknown_null_table(
             'id': pa.array([1, 2, 3], type=pa.int32()),
             'void_col': pa.array([None, None, None], type=pa.null()),
         })
-    pq.write_table(table, data_path)
+    local_dir = tempfile.mkdtemp(prefix='parquet_unknown_')
+    try:
+        local_file = os.path.join(local_dir, 'part.parquet')
+        pq.write_table(table, local_file)
+        with_cpu_session(lambda spark: copy_from_local(spark, local_file, data_path))
+    finally:
+        shutil.rmtree(local_dir, ignore_errors=True)
 
 
 # SPARK-56045 / SPARK-54220: Parquet UNKNOWN logical type annotation. PyArrow null columns are

integration_tests/src/main/python/parquet_test_utils.py

@@ -12,6 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+def copy_from_local(spark, local_source, hdfs_target):
+    """Copy a local path onto the Hadoop FS used by spark_tmp_path (e.g. GCS on Dataproc)."""
+    sc = spark.sparkContext
+    Path = sc._jvm.org.apache.hadoop.fs.Path
+    config = sc._jsc.hadoopConfiguration()
+    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(config)
+    fs.copyFromLocalFile(Path(local_source), Path(hdfs_target))
+
+
 def parquet_row_group_midpoints(spark, path):
     """Returns an approximate byte midpoint for each Parquet row group."""
     jvm = spark.sparkContext._jvm