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 (#27462)

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


Thrift ODR Violation Blocking Native Parquet Writes in Velox
Problem
Velox's Parquet writer crashes with SIGSEGV when linked into any binary that also uses FBThrift (e.g., Prestissimo presto_server). The crash is a C++ One Definition Rule (ODR) violation.

Root Cause
Velox's Parquet writer depends on OSS Apache Thrift (third-party2/apache-thrift/) for serializing Parquet page headers and file metadata. FBThrift (fbcode/thrift/) is Meta's fork used by RPC services. Both libraries declare classes in the same namespace (apache::thrift::protocol::TProtocol, apache::thrift::transport::TTransport, etc.) but with incompatible class layouts:

OSS Apache Thrift (Parquet)	FBThrift (Prestissimo RPC)
Namespace	apache::thrift	apache::thrift
TTransport size	~40 bytes (has TConfiguration shared_ptr, message size fields)	~8 bytes (vtable pointer only)
fd_ offset in TFDTransport	~40+	~8
When both are linked into one binary, the linker picks one definition. Code compiled against the other layout reads wrong memory offsets → SIGSEGV.

Crash Signature
Signal 11 (SIGSEGV) (0x0)
  std::_Sp_counted_base<>::_M_release_slow_last_use()    ← null shared_ptr control block
  apache::thrift::protocol::TProtocol::TProtocol()       ← wrong TTransport layout
  ThriftSerializer::ThriftSerializer()                    ← Parquet page header serialization
  SerializedPageWriter::SerializedPageWriter()
  Writer::close() → flush() → writeTable()
  IcebergDataSink::closeInternal()                        ← triggered by any native Parquet write

Impact
All native Parquet writes crash (INSERT, CTAS) in any Velox binary that links FBThrift
DWRF/ORC writes are unaffected (they don't use thrift serialization)
Parquet reads are unaffected (reads use a different thrift code path that happens to not trigger the ODR)
Affects Prestissimo, and potentially any Velox embedder (Gluten/Spark, etc.) that links both Parquet and another thrift variant
Prior Art
SEV 635079 — same ODR caused SIGSEGV crashes in Spark F3 pipelines (March 2026, SEV-2)
Apache Arrow already solved this by vendoring OSS thrift in private_parquet::apache::thrift namespace (D47918122, 2023)
T262970501 — tracking task for addressing the Parquet thrift dependency
GitHub issue #13175 — upstream tracking

Solution:
X-link: facebookincubator/velox#16019 this will help fix it.

Differential Revision: D98704718

Author: apurva-meta

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

@@ -37,9 +37,19 @@ velox::connector::hive::iceberg::FileContent toVeloxFileContent(
 velox::dwio::common::FileFormat toVeloxFileFormat(
     const presto::protocol::iceberg::FileFormat format) {
   if (format == protocol::iceberg::FileFormat::ORC) {
-    return velox::dwio::common::FileFormat::ORC;
+    // Iceberg metadata stores DWRF as ORC since Iceberg has no native DWRF
+    // format. At Meta, all Iceberg ORC files are actually DWRF-encoded.
+    return velox::dwio::common::FileFormat::DWRF;
   } 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 +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);
@@ -191,14 +201,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,

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

@@ -320,7 +320,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

@@ -79,12 +79,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