feat: [velox+prestissimo][iceberg] Iceberg V3 full C++ support: deletion vectors, equality deletes, sequence number conflict resolution, DV writer, DWRF data sink, Manifold filesystem, PUFFIN protocol

Summary:
X-link: facebookincubator/velox#16959

Combined velox/prestissimo diffs for Iceberg V3 C++ support:
- Improve IcebergSplitReader error handling and fix test file handle leaks
- Add Iceberg V3 deletion vector support (DeletionVectorReader)
- Add Iceberg equality delete file reader (EqualityDeleteFileReader)
- Add sequence number conflict resolution for equality deletes
- Add sequence number conflict resolution for positional deletes and deletion vectors
- Add Iceberg V3 deletion vector writer (DeletionVectorWriter)
- Add DWRF file format support for Iceberg data sink
- Add Manifold filesystem support with CAT token authentication
- Reformat FileContent enum to multi-line for extensibility
- Wire PUFFIN file format through C++ protocol and connector layer

Original diffs preserved: D97530140, D97530142, D97530141, D97530136, D97530139, D97530137, D97530138, D97599411, D97531548, D97531555

Differential Revision: D98704718

Author: apurva-meta

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

@@ -40,6 +40,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));
 }
@@ -171,7 +179,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);
@@ -191,14 +199,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,
@@ -211,8 +232,7 @@ IcebergPrestoToVeloxConnector::toVeloxSplit(
   }
 
   std::unordered_map<std::string, std::string> infoColumns = {
-      {"$data_sequence_number",
-       std::to_string(icebergSplit->dataSequenceNumber)},
+      {"$data_sequence_number", std::to_string(icebergSplit->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

@@ -78,12 +78,16 @@ void to_json(json& j, const ChangelogSplitInfo& p);
 void from_json(const json& j, ChangelogSplitInfo& p);
 } // namespace facebook::presto::protocol::iceberg
 namespace facebook::presto::protocol::iceberg {
-enum class FileContent { DATA, POSITION_DELETES, EQUALITY_DELETES };
+enum class FileContent {
+  DATA,
+  POSITION_DELETES,
+  EQUALITY_DELETES,
+};
 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