feat: [prestissimo][iceberg] Wire PUFFIN file format through C++ protocol and connector layer (#27394)

Summary:

This is the C++ counterpart to the Java PUFFIN support diff. It wires
the PUFFIN file format through the Prestissimo protocol and connector
conversion layer so that Iceberg V3 deletion vector files can be
deserialized and handled by native workers.

Changes:
1. Adds PUFFIN to the C++ protocol FileFormat enum and its JSON
   serialization table in presto_protocol_iceberg.{h,cpp}.
2. Handles PUFFIN in toVeloxFileFormat() in
   IcebergPrestoToVeloxConnector.cpp, mapping it to DWRF as a
   placeholder since DeletionVectorReader reads raw binary and
   does not use the DWRF/Parquet reader infrastructure.

== RELEASE NOTES ==
General Changes
* Upgrade Apache Iceberg library from 1.10.0 to 1.10.1.
Hive Connector Changes
* Add Iceberg V3 deletion vector (DV) support using Puffin-encoded roaring�bitmaps, including a DV reader, writer, page sink, and compaction procedure.
* Add Iceberg equality delete file reader with sequence number conflict�resolution per the Iceberg V2+ spec: equality deletes skip when�deleteFileSeqNum <= dataFileSeqNum; positional deletes and DVs skip when�deleteFileSeqNum < dataFileSeqNum; sequence number 0 (V1 legacy) never skips.
* Wire dataSequenceNumber through the Presto protocol layer (Java → C++)�to enable server-side sequence number conflict resolution for all delete�file types.
* Add PUFFIN file format support for deletion vector discovery, enabling�the coordinator to locate DV files during split creation.
* Add Iceberg V3 deletion vector write path with DV page sink and�rewrite_delete_files compaction procedure for DV maintenance.
* Add nanosecond timestamp (TIMESTAMP_NANO) type support for Iceberg V3�tables.
* Add Variant type support for Iceberg V3, enabling semi-structured data�columns in Iceberg tables.
* Eagerly collect delete files during split creation with improved logging�for easier debugging of Iceberg delete file resolution.
* Improve IcebergSplitReader error handling and fix test file handle leaks.
* Add end-to-end integration tests for Iceberg V3 covering snapshot�lifecycle (INSERT, DELETE with equality/positional/DV deletes, UPDATE,�MERGE, time-travel) and all 99 TPC-DS queries.

Differential Revision: D97531555

Author: apurva-meta

presto-native-execution/presto_cpp/main/connectors/IcebergPrestoToVeloxConnector.cpp

@@ -42,6 +42,14 @@ velox::dwio::common::FileFormat toVeloxFileFormat(
     return velox::dwio::common::FileFormat::ORC;
   } else if (format == protocol::iceberg::FileFormat::PARQUET) {
     return velox::dwio::common::FileFormat::PARQUET;
+  } else if (format == protocol::iceberg::FileFormat::PUFFIN) {
+    // PUFFIN is used for Iceberg V3 deletion vectors. The DeletionVectorReader
+    // reads raw binary from the file and does not use the DWRF/Parquet reader,
+    // so we map PUFFIN to DWRF as a placeholder — the format value is not
+    // actually used by the reader. This mapping is only safe for deletion
+    // vector files; if PUFFIN is encountered for other file content types,
+    // the DV routing logic in toHiveIcebergSplit() must reclassify it first.
+    return velox::dwio::common::FileFormat::DWRF;
   }
   VELOX_UNSUPPORTED("Unsupported file format: {}", fmt::underlying(format));
 }
@@ -173,7 +181,7 @@ IcebergPrestoToVeloxConnector::toVeloxSplit(
     const protocol::ConnectorId& catalogId,
     const protocol::ConnectorSplit* connectorSplit,
     const protocol::SplitContext* splitContext) const {
-  auto icebergSplit =
+  const auto* icebergSplit =
       dynamic_cast<const protocol::iceberg::IcebergSplit*>(connectorSplit);
   VELOX_CHECK_NOT_NULL(
       icebergSplit, "Unexpected split type {}", connectorSplit->_type);
@@ -196,14 +204,27 @@ IcebergPrestoToVeloxConnector::toVeloxSplit(
   std::vector<velox::connector::hive::iceberg::IcebergDeleteFile> deletes;
   deletes.reserve(icebergSplit->deletes.size());
   for (const auto& deleteFile : icebergSplit->deletes) {
-    std::unordered_map<int32_t, std::string> lowerBounds(
+    const std::unordered_map<int32_t, std::string> lowerBounds(
         deleteFile.lowerBounds.begin(), deleteFile.lowerBounds.end());
 
-    std::unordered_map<int32_t, std::string> upperBounds(
+    const std::unordered_map<int32_t, std::string> upperBounds(
         deleteFile.upperBounds.begin(), deleteFile.upperBounds.end());
 
-    velox::connector::hive::iceberg::IcebergDeleteFile icebergDeleteFile(
-        toVeloxFileContent(deleteFile.content),
+    // Iceberg V3 deletion vectors arrive from the coordinator as
+    // POSITION_DELETES with PUFFIN format. Reclassify them as
+    // kDeletionVector so that IcebergSplitReader routes them to
+    // DeletionVectorReader instead of PositionalDeleteFileReader.
+    velox::connector::hive::iceberg::FileContent veloxContent =
+        toVeloxFileContent(deleteFile.content);
+    if (veloxContent ==
+            velox::connector::hive::iceberg::FileContent::kPositionalDeletes &&
+        deleteFile.format == protocol::iceberg::FileFormat::PUFFIN) {
+      veloxContent =
+          velox::connector::hive::iceberg::FileContent::kDeletionVector;
+    }
+
+    const velox::connector::hive::iceberg::IcebergDeleteFile icebergDeleteFile(
+        veloxContent,
         deleteFile.path,
         toVeloxFileFormat(deleteFile.format),
         deleteFile.recordCount,
@@ -218,8 +239,7 @@ IcebergPrestoToVeloxConnector::toVeloxSplit(
 
 
   std::unordered_map<std::string, std::string> infoColumns = {
-      {"$data_sequence_number",
-       std::to_string(dataSequenceNumber)},
+      {"$data_sequence_number", std::to_string(dataSequenceNumber)},
       {"$path", icebergSplit->path}};
 
   return std::make_unique<velox::connector::hive::iceberg::HiveIcebergSplit>(

presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.cpp

@@ -306,7 +306,8 @@ static const std::pair<FileFormat, json> FileFormat_enum_table[] =
         {FileFormat::ORC, "ORC"},
         {FileFormat::PARQUET, "PARQUET"},
         {FileFormat::AVRO, "AVRO"},
-        {FileFormat::METADATA, "METADATA"}};
+        {FileFormat::METADATA, "METADATA"},
+        {FileFormat::PUFFIN, "PUFFIN"}};
 void to_json(json& j, const FileFormat& e) {
   static_assert(std::is_enum<FileFormat>::value, "FileFormat must be an enum!");
   const auto* it = std::find_if(

presto-native-execution/presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.h

@@ -87,7 +87,7 @@ extern void to_json(json& j, const FileContent& e);
 extern void from_json(const json& j, FileContent& e);
 } // namespace facebook::presto::protocol::iceberg
 namespace facebook::presto::protocol::iceberg {
-enum class FileFormat { ORC, PARQUET, AVRO, METADATA };
+enum class FileFormat { ORC, PARQUET, AVRO, METADATA, PUFFIN };
 extern void to_json(json& j, const FileFormat& e);
 extern void from_json(const json& j, FileFormat& e);
 } // namespace facebook::presto::protocol::iceberg