feat(native): Add iceberg partition transforms

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java

@@ -411,12 +411,22 @@ public static List<Column> toHiveColumns(List<NestedField> columns)
         return columns.stream()
                 .map(column -> new Column(
                         column.name(),
-                        HiveType.toHiveType(HiveSchemaUtil.convert(column.type())),
+                        icebergTypeToHiveType(column.type()),
                         Optional.empty(),
                         Optional.empty()))
                 .collect(toImmutableList());
     }
 
+    private static HiveType icebergTypeToHiveType(org.apache.iceberg.types.Type icebergType)
+    {
+        // Special handling for TIME type: use bigint instead of 'string'
+        if (icebergType.typeId() == org.apache.iceberg.types.Type.TypeID.TIME) {
+            return HiveType.HIVE_LONG;
+        }
+
+        return HiveType.toHiveType(HiveSchemaUtil.convert(icebergType));
+    }
+
     public static FileFormat getFileFormat(Table table)
     {
         return FileFormat.valueOf(table.properties()

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

@@ -16,12 +16,15 @@
 #include "presto_cpp/main/connectors/PrestoToVeloxConnectorUtils.h"
 
 #include "presto_cpp/presto_protocol/connector/iceberg/IcebergConnectorProtocol.h"
+#include "velox/connectors/hive/TableHandle.h"
 #include "velox/connectors/hive/iceberg/IcebergDataSink.h"
 #include "velox/connectors/hive/iceberg/IcebergSplit.h"
 #include "velox/type/fbhive/HiveTypeParser.h"
 
 namespace facebook::presto {
 
+using namespace velox;
+
 namespace {
 
 velox::connector::hive::iceberg::FileContent toVeloxFileContent(
@@ -299,6 +302,8 @@ IcebergPrestoToVeloxConnector::toVeloxInsertTableHandle(
           fmt::format("{}/data", icebergOutputTableHandle->outputPath),
           velox::connector::hive::LocationHandle::TableType::kNew),
       toVeloxFileFormat(icebergOutputTableHandle->fileFormat),
+      toVeloxIcebergPartitionSpec(
+          icebergOutputTableHandle->partitionSpec, typeParser),
       std::optional(
           toFileCompressionKind(icebergOutputTableHandle->compressionCodec)));
 }
@@ -327,6 +332,8 @@ IcebergPrestoToVeloxConnector::toVeloxInsertTableHandle(
           fmt::format("{}/data", icebergInsertTableHandle->outputPath),
           velox::connector::hive::LocationHandle::TableType::kExisting),
       toVeloxFileFormat(icebergInsertTableHandle->fileFormat),
+      toVeloxIcebergPartitionSpec(
+          icebergInsertTableHandle->partitionSpec, typeParser),
       std::optional(
           toFileCompressionKind(icebergInsertTableHandle->compressionCodec)));
 }
@@ -345,4 +352,36 @@ IcebergPrestoToVeloxConnector::toHiveColumns(
   return hiveColumns;
 }
 
+connector::hive::iceberg::IcebergPartitionSpec::Field
+IcebergPrestoToVeloxConnector::toVeloxIcebergPartitionField(
+    const protocol::iceberg::IcebergPartitionField& field,
+    const facebook::presto::TypeParser& typeParser,
+    const protocol::iceberg::PrestoIcebergSchema& schema) const {
+  std::string type;
+  for (const auto& column : schema.columns) {
+    if (column.name == field.name) {
+      type = column.prestoType;
+    }
+  }
+  return connector::hive::iceberg::IcebergPartitionSpec::Field(
+      field.name,
+      stringToType(type, typeParser),
+      static_cast<connector::hive::iceberg::TransformType>(field.transform),
+      field.parameter ? *field.parameter : std::optional<int32_t>());
+}
+
+std::unique_ptr<velox::connector::hive::iceberg::IcebergPartitionSpec>
+IcebergPrestoToVeloxConnector::toVeloxIcebergPartitionSpec(
+    const protocol::iceberg::PrestoIcebergPartitionSpec& spec,
+    const facebook::presto::TypeParser& typeParser) const {
+  std::vector<connector::hive::iceberg::IcebergPartitionSpec::Field> fields;
+  fields.reserve(spec.fields.size());
+  for (auto field : spec.fields) {
+    fields.emplace_back(
+        toVeloxIcebergPartitionField(field, typeParser, spec.schema));
+  }
+  return std::make_unique<connector::hive::iceberg::IcebergPartitionSpec>(
+      spec.specId, fields);
+}
+
 } // namespace facebook::presto

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

@@ -16,6 +16,8 @@
 
 #include "presto_cpp/main/connectors/PrestoToVeloxConnector.h"
 #include "presto_cpp/presto_protocol/connector/iceberg/presto_protocol_iceberg.h"
+#include "velox/connectors/hive/iceberg/IcebergDataSink.h"
+#include "velox/connectors/hive/iceberg/PartitionSpec.h"
 
 namespace facebook::presto {
 
@@ -57,6 +59,17 @@ class IcebergPrestoToVeloxConnector final : public PrestoToVeloxConnector {
       const protocol::List<protocol::iceberg::IcebergColumnHandle>&
           inputColumns,
       const TypeParser& typeParser) const;
+
+  velox::connector::hive::iceberg::IcebergPartitionSpec::Field
+  toVeloxIcebergPartitionField(
+      const protocol::iceberg::IcebergPartitionField& filed,
+      const facebook::presto::TypeParser& typeParser,
+      const protocol::iceberg::PrestoIcebergSchema& schema) const;
+
+  std::unique_ptr<velox::connector::hive::iceberg::IcebergPartitionSpec>
+  toVeloxIcebergPartitionSpec(
+      const protocol::iceberg::PrestoIcebergPartitionSpec& spec,
+      const TypeParser& typeParser) const;
 };
 
 } // namespace facebook::presto