GH-43807: [C++][Python] Add UUID extension type conversion support to/from Parquet

cpp/src/arrow/dataset/file_parquet.cc

@@ -133,6 +133,8 @@ parquet::ArrowReaderProperties MakeArrowReaderProperties(
   arrow_properties.set_io_context(
       parquet_scan_options.arrow_reader_properties->io_context());
   arrow_properties.set_use_threads(options.use_threads);
+  arrow_properties.set_arrow_extensions_enabled(
+      parquet_scan_options.arrow_reader_properties->get_arrow_extensions_enabled());
   return arrow_properties;
 }
 

cpp/src/parquet/arrow/arrow_schema_test.cc

@@ -33,6 +33,7 @@
 
 #include "arrow/array.h"
 #include "arrow/extension/json.h"
+#include "arrow/extension/uuid.h"
 #include "arrow/ipc/writer.h"
 #include "arrow/testing/gtest_util.h"
 #include "arrow/type.h"
@@ -866,7 +867,7 @@ Status ArrowSchemaToParquetMetadata(std::shared_ptr<::arrow::Schema>& arrow_sche
   return Status::OK();
 }
 
-TEST_F(TestConvertParquetSchema, ParquetSchemaArrowExtensions) {
+TEST_F(TestConvertParquetSchema, ParquetSchemaArrowJsonExtension) {
   std::vector<NodePtr> parquet_fields;
   parquet_fields.push_back(PrimitiveNode::Make(
       "json_1", Repetition::OPTIONAL, ParquetType::BYTE_ARRAY, ConvertedType::JSON));
@@ -948,6 +949,68 @@ TEST_F(TestConvertParquetSchema, ParquetSchemaArrowExtensions) {
   }
 }
 
+TEST_F(TestConvertParquetSchema, ParquetSchemaArrowUuidExtension) {
+  std::vector<NodePtr> parquet_fields;
+  parquet_fields.push_back(PrimitiveNode::Make("uuid", Repetition::OPTIONAL,
+                                               LogicalType::UUID(),
+                                               ParquetType::FIXED_LEN_BYTE_ARRAY, 16));
+
+  {
+    // Parquet file does not contain Arrow schema.
+    // By default, field should be treated as fixed_size_binary(16) in Arrow.
+    auto arrow_schema =
+        ::arrow::schema({::arrow::field("uuid", ::arrow::fixed_size_binary(16), true)});
+    std::shared_ptr<KeyValueMetadata> metadata{};
+    ASSERT_OK(ConvertSchema(parquet_fields, metadata));
+    CheckFlatSchema(arrow_schema);
+  }
+
+  {
+    // Parquet file does not contain Arrow schema.
+    // If Arrow extensions are enabled, field will be interpreted as uuid()
+    // extension field.
+    ArrowReaderProperties props;
+    props.set_arrow_extensions_enabled(true);
+    auto arrow_schema =
+        ::arrow::schema({::arrow::field("uuid", ::arrow::extension::uuid(), true)});
+    std::shared_ptr<KeyValueMetadata> metadata{};
+    ASSERT_OK(ConvertSchema(parquet_fields, metadata, props));
+    CheckFlatSchema(arrow_schema);
+  }
+
+  {
+    // Parquet file contains Arrow schema.
+    // uuid will be interpreted as uuid() field
+    ArrowReaderProperties props;
+    props.set_arrow_extensions_enabled(false);
+    std::shared_ptr<KeyValueMetadata> field_metadata =
+        ::arrow::key_value_metadata({"foo", "bar"}, {"biz", "baz"});
+    auto arrow_schema = ::arrow::schema(
+        {::arrow::field("uuid", ::arrow::extension::uuid(), true, field_metadata)});
+
+    std::shared_ptr<KeyValueMetadata> metadata;
+    ASSERT_OK(ArrowSchemaToParquetMetadata(arrow_schema, metadata));
+    ASSERT_OK(ConvertSchema(parquet_fields, metadata, props));
+    CheckFlatSchema(arrow_schema, true /* check_metadata */);
+  }
+
+  {
+    // Parquet file contains Arrow schema.
+    // uuid will be interpreted as uuid() field even though extensions are not enabled.
+    ArrowReaderProperties props;
+    props.set_arrow_extensions_enabled(true);
+    std::shared_ptr<KeyValueMetadata> field_metadata =
+        ::arrow::key_value_metadata({"foo", "bar"}, {"biz", "baz"});
+    auto arrow_schema = ::arrow::schema(
+        {::arrow::field("uuid", ::arrow::extension::uuid(), true, field_metadata)});
+
+    std::shared_ptr<KeyValueMetadata> metadata;
+    ASSERT_OK(ArrowSchemaToParquetMetadata(arrow_schema, metadata));
+    ASSERT_OK(ConvertSchema(parquet_fields, metadata, props));
+    CheckFlatSchema(arrow_schema, true /* check_metadata */);
+  }
+}
+
 class TestConvertArrowSchema : public ::testing::Test {
  public:
   virtual void SetUp() {}

cpp/src/parquet/arrow/schema.cc

@@ -22,6 +22,7 @@
 #include <vector>
 
 #include "arrow/extension/json.h"
+#include "arrow/extension/uuid.h"
 #include "arrow/extension_type.h"
 #include "arrow/io/memory.h"
 #include "arrow/ipc/api.h"
@@ -434,7 +435,13 @@ Status FieldToNode(const std::string& name, const std::shared_ptr<Field>& field,
         type = ParquetType::BYTE_ARRAY;
         logical_type = LogicalType::JSON();
         break;
+      } else if (ext_type->extension_name() == std::string("arrow.uuid")) {
+        type = ParquetType::FIXED_LEN_BYTE_ARRAY;
+        logical_type = LogicalType::UUID();
+        length = 16;
+        break;
       }
+
       std::shared_ptr<::arrow::Field> storage_field = ::arrow::field(
           name, ext_type->storage_type(), field->nullable(), field->metadata());
       return FieldToNode(name, storage_field, properties, arrow_properties, out);
@@ -1053,6 +1060,29 @@ Result<bool> ApplyOriginalMetadata(const Field& origin_field, SchemaField* infer
       // inferred_type is arrow::extension::json(arrow::utf8())
       auto origin_storage_field = origin_field.WithType(ex_type.storage_type());
 
+      // Apply metadata recursively to storage type
+      RETURN_NOT_OK(ApplyOriginalStorageMetadata(*origin_storage_field, inferred));
+      inferred->field = inferred->field->WithType(origin_type);
+    } else if (inferred_type->id() == ::arrow::Type::FIXED_SIZE_BINARY &&
+               ex_type.extension_name() == std::string("arrow.uuid")) {
+      // Schema mismatch.
+      //
+      // Arrow extensions are DISABLED in Parquet.
+      // origin_type is ::arrow::extension::uuid()
+      // inferred_type is ::arrow::fixed_size_binary()
+      //
+      // Origin type is restored as Arrow should be considered the source of truth.
+      inferred->field = inferred->field->WithType(origin_type);
+      RETURN_NOT_OK(ApplyOriginalStorageMetadata(origin_field, inferred));
+    } else if (inferred_type->id() == ::arrow::Type::EXTENSION &&
+               ex_type.extension_name() == std::string("arrow.uuid")) {
+      // Schema match.
+      //
+      // Arrow extensions are ENABLED in Parquet.
+      // origin_type is arrow::extension::uuid()
+      // inferred_type is arrow::extension::uuid()
+      auto origin_storage_field = origin_field.WithType(ex_type.storage_type());
+
       // Apply metadata recursively to storage type
       RETURN_NOT_OK(ApplyOriginalStorageMetadata(*origin_storage_field, inferred));
       inferred->field = inferred->field->WithType(origin_type);

cpp/src/parquet/arrow/schema_internal.cc

@@ -18,6 +18,7 @@
 #include "parquet/arrow/schema_internal.h"
 
 #include "arrow/extension/json.h"
+#include "arrow/extension/uuid.h"
 #include "arrow/type.h"
 
 #include "parquet/properties.h"
@@ -134,16 +135,22 @@ Result<std::shared_ptr<ArrowType>> FromByteArray(
   }
 }
 
-Result<std::shared_ptr<ArrowType>> FromFLBA(const LogicalType& logical_type,
-                                            int32_t physical_length) {
+Result<std::shared_ptr<ArrowType>> FromFLBA(
+    const LogicalType& logical_type, int32_t physical_length,
+    const ArrowReaderProperties& reader_properties) {
   switch (logical_type.type()) {
     case LogicalType::Type::DECIMAL:
       return MakeArrowDecimal(logical_type);
     case LogicalType::Type::FLOAT16:
       return ::arrow::float16();
     case LogicalType::Type::NONE:
     case LogicalType::Type::INTERVAL:
+      return ::arrow::fixed_size_binary(physical_length);
     case LogicalType::Type::UUID:
+      if (reader_properties.get_arrow_extensions_enabled()) {
+        return ::arrow::extension::uuid();
+      }
+
       return ::arrow::fixed_size_binary(physical_length);
     default:
       return Status::NotImplemented("Unhandled logical logical_type ",
@@ -211,7 +218,7 @@ Result<std::shared_ptr<ArrowType>> GetArrowType(
     case ParquetType::BYTE_ARRAY:
       return FromByteArray(logical_type, reader_properties);
     case ParquetType::FIXED_LEN_BYTE_ARRAY:
-      return FromFLBA(logical_type, type_length);
+      return FromFLBA(logical_type, type_length, reader_properties);
     default: {
       // PARQUET-1565: This can occur if the file is corrupt
       return Status::IOError("Invalid physical column type: ",

python/pyarrow/_dataset_parquet.pyx

@@ -703,7 +703,7 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
     cache_options : pyarrow.CacheOptions, default None
         Cache options used when pre_buffer is enabled. The default values should
         be good for most use cases. You may want to adjust these for example if
-        you have exceptionally high latency to the file system. 
+        you have exceptionally high latency to the file system.
     thrift_string_size_limit : int, default None
         If not None, override the maximum total string size allocated
         when decoding Thrift structures. The default limit should be
@@ -720,6 +720,9 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
         Parquet file.
     page_checksum_verification : bool, default False
         If True, verify the page checksum for each page read from the file.
+    arrow_extensions_enabled : bool, default False
+        If True, read Parquet logical types as Arrow Extension Types where possible,
+        (e.g., JSON as arrow.json or UUID as arrow.uuid).
     """
 
     # Avoid mistakingly creating attributes
@@ -733,7 +736,8 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
                  thrift_container_size_limit=None,
                  decryption_config=None,
                  decryption_properties=None,
-                 bint page_checksum_verification=False):
+                 bint page_checksum_verification=False,
+                 bint arrow_extensions_enabled=False):
         self.init(shared_ptr[CFragmentScanOptions](
             new CParquetFragmentScanOptions()))
         self.use_buffered_stream = use_buffered_stream
@@ -752,6 +756,7 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
         if decryption_properties is not None:
             self.decryption_properties = decryption_properties
         self.page_checksum_verification = page_checksum_verification
+        self.arrow_extensions_enabled = arrow_extensions_enabled
 
     cdef void init(self, const shared_ptr[CFragmentScanOptions]& sp):
         FragmentScanOptions.init(self, sp)
@@ -868,6 +873,14 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
     def page_checksum_verification(self, bint page_checksum_verification):
         self.reader_properties().set_page_checksum_verification(page_checksum_verification)
 
+    @property
+    def arrow_extensions_enabled(self):
+        return self.arrow_reader_properties().get_arrow_extensions_enabled()
+
+    @arrow_extensions_enabled.setter
+    def arrow_extensions_enabled(self, bint arrow_extensions_enabled):
+        self.arrow_reader_properties().set_arrow_extensions_enabled(arrow_extensions_enabled)
+
     def equals(self, ParquetFragmentScanOptions other):
         """
         Parameters
@@ -881,11 +894,12 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
         attrs = (
             self.use_buffered_stream, self.buffer_size, self.pre_buffer, self.cache_options,
             self.thrift_string_size_limit, self.thrift_container_size_limit,
-            self.page_checksum_verification)
+            self.page_checksum_verification, self.arrow_extensions_enabled)
         other_attrs = (
             other.use_buffered_stream, other.buffer_size, other.pre_buffer, other.cache_options,
             other.thrift_string_size_limit,
-            other.thrift_container_size_limit, other.page_checksum_verification)
+            other.thrift_container_size_limit, other.page_checksum_verification,
+            other.arrow_extensions_enabled)
         return attrs == other_attrs
 
     @staticmethod
@@ -902,7 +916,8 @@ cdef class ParquetFragmentScanOptions(FragmentScanOptions):
             cache_options=self.cache_options,
             thrift_string_size_limit=self.thrift_string_size_limit,
             thrift_container_size_limit=self.thrift_container_size_limit,
-            page_checksum_verification=self.page_checksum_verification
+            page_checksum_verification=self.page_checksum_verification,
+            arrow_extensions_enabled=self.arrow_extensions_enabled
         )
         return ParquetFragmentScanOptions._reconstruct, (kwargs,)
 

python/pyarrow/_parquet.pxd

@@ -405,6 +405,8 @@ cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
         CCacheOptions cache_options() const
         void set_coerce_int96_timestamp_unit(TimeUnit unit)
         TimeUnit coerce_int96_timestamp_unit() const
+        void set_arrow_extensions_enabled(c_bool extensions_enabled)
+        c_bool get_arrow_extensions_enabled() const
 
     ArrowReaderProperties default_arrow_reader_properties()
 

python/pyarrow/_parquet.pyx

@@ -1454,7 +1454,8 @@ cdef class ParquetReader(_Weakrefable):
              FileDecryptionProperties decryption_properties=None,
              thrift_string_size_limit=None,
              thrift_container_size_limit=None,
-             page_checksum_verification=False):
+             page_checksum_verification=False,
+             arrow_extensions_enabled=False):
         """
         Open a parquet file for reading.
 
@@ -1471,6 +1472,7 @@ cdef class ParquetReader(_Weakrefable):
         thrift_string_size_limit : int, optional
         thrift_container_size_limit : int, optional
         page_checksum_verification : bool, default False
+        arrow_extensions_enabled : bool, default False
         """
         cdef:
             shared_ptr[CFileMetaData] c_metadata
@@ -1520,6 +1522,8 @@ cdef class ParquetReader(_Weakrefable):
             arrow_props.set_coerce_int96_timestamp_unit(
                 string_to_timeunit(coerce_int96_timestamp_unit))
 
+        arrow_props.set_arrow_extensions_enabled(arrow_extensions_enabled)
+
         self.source = source
         get_reader(source, use_memory_map, &self.rd_handle)
 

python/pyarrow/parquet/core.py

@@ -254,6 +254,9 @@ class ParquetFile:
         it will be parsed as an URI to determine the filesystem.
     page_checksum_verification : bool, default False
         If True, verify the checksum for each page read from the file.
+    arrow_extensions_enabled : bool, default False
+        If True, read Parquet logical types as Arrow Extension Types where possible,
+        (e.g., JSON as arrow.json or UUID as arrow.uuid).
 
     Examples
     --------
@@ -302,7 +305,7 @@ def __init__(self, source, *, metadata=None, common_metadata=None,
                  pre_buffer=False, coerce_int96_timestamp_unit=None,
                  decryption_properties=None, thrift_string_size_limit=None,
                  thrift_container_size_limit=None, filesystem=None,
-                 page_checksum_verification=False):
+                 page_checksum_verification=False, arrow_extensions_enabled=False):
 
         self._close_source = getattr(source, 'closed', True)
 
@@ -322,6 +325,7 @@ def __init__(self, source, *, metadata=None, common_metadata=None,
             thrift_string_size_limit=thrift_string_size_limit,
             thrift_container_size_limit=thrift_container_size_limit,
             page_checksum_verification=page_checksum_verification,
+            arrow_extensions_enabled=arrow_extensions_enabled,
         )
         self.common_metadata = common_metadata
         self._nested_paths_by_prefix = self._build_nested_paths()
@@ -1264,6 +1268,9 @@ class ParquetDataset:
     sufficient for most Parquet files.
 page_checksum_verification : bool, default False
     If True, verify the page checksum for each page read from the file.
+arrow_extensions_enabled : bool, default False
+    If True, read Parquet logical types as Arrow Extension Types where possible,
+    (e.g., JSON as arrow.json or UUID as arrow.uuid).
 
 Examples
 --------
@@ -1276,7 +1283,8 @@ def __init__(self, path_or_paths, filesystem=None, schema=None, *, filters=None,
                  coerce_int96_timestamp_unit=None,
                  decryption_properties=None, thrift_string_size_limit=None,
                  thrift_container_size_limit=None,
-                 page_checksum_verification=False):
+                 page_checksum_verification=False,
+                 arrow_extensions_enabled=False):
 
         import pyarrow.dataset as ds
 
@@ -1287,6 +1295,7 @@ def __init__(self, path_or_paths, filesystem=None, schema=None, *, filters=None,
             "thrift_string_size_limit": thrift_string_size_limit,
             "thrift_container_size_limit": thrift_container_size_limit,
             "page_checksum_verification": page_checksum_verification,
+            "arrow_extensions_enabled": arrow_extensions_enabled,
         }
         if buffer_size:
             read_options.update(use_buffered_stream=True,
@@ -1674,6 +1683,9 @@ def partitioning(self):
     sufficient for most Parquet files.
 page_checksum_verification : bool, default False
     If True, verify the checksum for each page read from the file.
+arrow_extensions_enabled : bool, default False
+    If True, read Parquet logical types as Arrow Extension Types where possible,
+    (e.g., JSON as arrow.json or UUID as arrow.uuid).
 
 Returns
 -------
@@ -1768,7 +1780,8 @@ def read_table(source, *, columns=None, use_threads=True,
                pre_buffer=True, coerce_int96_timestamp_unit=None,
                decryption_properties=None, thrift_string_size_limit=None,
                thrift_container_size_limit=None,
-               page_checksum_verification=False):
+               page_checksum_verification=False,
+               arrow_extensions_enabled=False):
 
     try:
         dataset = ParquetDataset(
@@ -1787,6 +1800,7 @@ def read_table(source, *, columns=None, use_threads=True,
             thrift_string_size_limit=thrift_string_size_limit,
             thrift_container_size_limit=thrift_container_size_limit,
             page_checksum_verification=page_checksum_verification,
+            arrow_extensions_enabled=arrow_extensions_enabled,
         )
     except ImportError:
         # fall back on ParquetFile for simple cases when pyarrow.dataset