fix(plugin-iceberg): Incorrect Iceberg columnHandle type in Prestissimo (#26411)

## Description
<!---Describe your changes in detail-->

**Problem**
When writing to Iceberg tables with partition transforms (e.g.,
bucket(), year(), month(), day(), hour(), truncate()), the
IcebergColumnHandle type was incorrectly set to `REGULAR` for partition
source columns instead of `PARTITION_KEY`. This caused issues in
Prestissimo (native execution) where the partition column information is
needed to correctly construct partition data when writing to Iceberg
tables.
For example, for a table partitioned by bucket(10, user_id) or
year(event_time):
The source columns should be marked as `PARTITION_KEY` during INSERT
operations.
Previously, only identity partition columns were marked as
`PARTITION_KEY`
Non-identity partition transforms were ignored, causing the source
columns to be incorrectly marked as `REGULAR`.

The existing getColumns() method only identified identity partition
columns using getPartitionFields(partitionSpec, IDENTITY), which filters
out all non-identity transforms.
This PR introduces a new getColumnsForWrite() method specifically for
write operations (INSERT queries) that:
Identifies all partition source columns by examining
partitionSpec.fields() and extracting source field IDs.
Marks them as `PARTITION_KEY` so Prestissimo can access the correct
column type information during writes.

The existing getColumns() method is preserved for read operations
(SELECT queries) where only identity partition columns should be marked
as `PARTITION_KEY` to ensure partition source columns are read from data
files. This is binding to Hive implementation and because it doesn't
read partition column from data file. But for iceberg there is no way to
reverse a partitioned value to original column value.


## Motivation and Context
<!---Why is this change required? What problem does it solve?-->
<!---If it fixes an open issue, please link to the issue here.-->

This is a preliminary PR for supporting partition transform of Iceberg
native insertion.


## Impact
<!---Describe any public API or user-facing feature change or any
performance impact-->

## Test Plan
<!---Please fill in how you tested your change-->

## Contributor checklist

- [ ] Please make sure your submission complies with our [contributing
guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md),
in particular [code
style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style)
and [commit
standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards).
- [ ] PR description addresses the issue accurately and concisely. If
the change is non-trivial, a GitHub Issue is referenced.
- [ ] Documented new properties (with its default value), SQL syntax,
functions, or other functionality.
- [ ] If release notes are required, they follow the [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines).
- [ ] Adequate tests were added if applicable.
- [ ] CI passed.
- [ ] If adding new dependencies, verified they have an [OpenSSF
Scorecard](https://securityscorecards.dev/#the-checks) score of 5.0 or
higher (or obtained explicit TSC approval for lower scores).

## Release Notes
Please follow [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines)
and fill in the release notes below.

```
== NO RELEASE NOTE ==
```

Author: PingLiuPing

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

@@ -157,6 +157,7 @@
 import static com.facebook.presto.iceberg.IcebergTableType.EQUALITY_DELETES;
 import static com.facebook.presto.iceberg.IcebergUtil.MIN_FORMAT_VERSION_FOR_DELETE;
 import static com.facebook.presto.iceberg.IcebergUtil.getColumns;
+import static com.facebook.presto.iceberg.IcebergUtil.getColumnsForWrite;
 import static com.facebook.presto.iceberg.IcebergUtil.getDeleteMode;
 import static com.facebook.presto.iceberg.IcebergUtil.getFileFormat;
 import static com.facebook.presto.iceberg.IcebergUtil.getPartitionFields;
@@ -518,7 +519,7 @@ protected ConnectorInsertTableHandle beginIcebergTableInsert(ConnectorSession se
                 table.getIcebergTableName(),
                 toPrestoSchema(icebergTable.schema(), typeManager),
                 toPrestoPartitionSpec(icebergTable.spec(), typeManager),
-                getColumns(icebergTable.schema(), icebergTable.spec(), typeManager),
+                getColumnsForWrite(icebergTable.schema(), icebergTable.spec(), typeManager),
                 icebergTable.location(),
                 getFileFormat(icebergTable),
                 getCompressionCodec(session),

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

@@ -124,7 +124,7 @@
 import static com.facebook.presto.iceberg.IcebergTableProperties.getTableLocation;
 import static com.facebook.presto.iceberg.IcebergTableType.DATA;
 import static com.facebook.presto.iceberg.IcebergUtil.createIcebergViewProperties;
-import static com.facebook.presto.iceberg.IcebergUtil.getColumns;
+import static com.facebook.presto.iceberg.IcebergUtil.getColumnsForWrite;
 import static com.facebook.presto.iceberg.IcebergUtil.getHiveIcebergTable;
 import static com.facebook.presto.iceberg.IcebergUtil.isIcebergTable;
 import static com.facebook.presto.iceberg.IcebergUtil.populateTableProperties;
@@ -390,7 +390,7 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
                 new IcebergTableName(tableName, DATA, Optional.empty(), Optional.empty()),
                 toPrestoSchema(metadata.schema(), typeManager),
                 toPrestoPartitionSpec(metadata.spec(), typeManager),
-                getColumns(metadata.schema(), metadata.spec(), typeManager),
+                getColumnsForWrite(metadata.schema(), metadata.spec(), typeManager),
                 targetPath,
                 fileFormat,
                 getCompressionCodec(session),

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

@@ -67,7 +67,7 @@
 import static com.facebook.presto.iceberg.IcebergTableType.DATA;
 import static com.facebook.presto.iceberg.IcebergUtil.VIEW_OWNER;
 import static com.facebook.presto.iceberg.IcebergUtil.createIcebergViewProperties;
-import static com.facebook.presto.iceberg.IcebergUtil.getColumns;
+import static com.facebook.presto.iceberg.IcebergUtil.getColumnsForWrite;
 import static com.facebook.presto.iceberg.IcebergUtil.getNativeIcebergTable;
 import static com.facebook.presto.iceberg.IcebergUtil.getNativeIcebergView;
 import static com.facebook.presto.iceberg.IcebergUtil.populateTableProperties;
@@ -386,7 +386,7 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
                 new IcebergTableName(tableName, DATA, Optional.empty(), Optional.empty()),
                 toPrestoSchema(icebergTable.schema(), typeManager),
                 toPrestoPartitionSpec(icebergTable.spec(), typeManager),
-                getColumns(icebergTable.schema(), icebergTable.spec(), typeManager),
+                getColumnsForWrite(icebergTable.schema(), icebergTable.spec(), typeManager),
                 icebergTable.location(),
                 fileFormat,
                 getCompressionCodec(session),

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

@@ -74,7 +74,7 @@
 import static com.facebook.presto.iceberg.IcebergErrorCode.ICEBERG_INVALID_METADATA;
 import static com.facebook.presto.iceberg.IcebergErrorCode.ICEBERG_TOO_MANY_OPEN_PARTITIONS;
 import static com.facebook.presto.iceberg.IcebergErrorCode.ICEBERG_WRITER_OPEN_ERROR;
-import static com.facebook.presto.iceberg.IcebergUtil.getColumns;
+import static com.facebook.presto.iceberg.IcebergUtil.getColumnsForWrite;
 import static com.facebook.presto.iceberg.PartitionTransforms.getColumnTransform;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Verify.verify;
@@ -157,7 +157,7 @@ public IcebergPageSink(
         this.sortOrder = requireNonNull(sortOrder, "sortOrder is null");
         String tempDirectoryPath = locationProvider.newDataLocation("sort-tmp-files");
         this.tempDirectory = new Path(tempDirectoryPath);
-        this.columnTypes = getColumns(outputSchema, partitionSpec, requireNonNull(sortParameters.getTypeManager(), "typeManager is null")).stream()
+        this.columnTypes = getColumnsForWrite(outputSchema, partitionSpec, requireNonNull(sortParameters.getTypeManager(), "typeManager is null")).stream()
                 .map(IcebergColumnHandle::getType)
                 .collect(toImmutableList());
         this.sortParameters = sortParameters;

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

@@ -365,6 +365,25 @@ public static List<IcebergColumnHandle> getColumns(Stream<Integer> fields, Schem
                 .collect(toImmutableList());
     }
 
+    public static List<IcebergColumnHandle> getColumnsForWrite(Schema schema, PartitionSpec partitionSpec, TypeManager typeManager)
+    {
+        return getColumnsForWrite(schema.columns().stream().map(NestedField::fieldId), schema, partitionSpec, typeManager);
+    }
+
+    private static List<IcebergColumnHandle> getColumnsForWrite(Stream<Integer> fields, Schema schema, PartitionSpec partitionSpec, TypeManager typeManager)
+    {
+        Set<Integer> partitionSourceIds = partitionSpec.fields().stream()
+                .map(PartitionField::sourceId)
+                .collect(toImmutableSet());
+
+        return fields
+                .map(schema::findField)
+                .map(column -> partitionSourceIds.contains(column.fieldId()) ?
+                        IcebergColumnHandle.create(column, typeManager, PARTITION_KEY) :
+                        IcebergColumnHandle.create(column, typeManager, REGULAR))
+                .collect(toImmutableList());
+    }
+
     public static Map<PartitionField, Integer> getIdentityPartitions(PartitionSpec partitionSpec)
     {
         // TODO: expose transform information in Iceberg library

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

@@ -20,10 +20,10 @@
 public enum PartitionTransformType
 {
     IDENTITY("identity", 0),
-    YEAR("year", 1),
-    MONTH("month", 2),
-    DAY("day", 3),
-    HOUR("hour", 4),
+    HOUR("hour", 1),
+    DAY("day", 2),
+    MONTH("month", 3),
+    YEAR("year", 4),
     BUCKET("bucket", 5),
     TRUNCATE("truncate", 6);
 

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

@@ -131,10 +131,10 @@ void from_json(const json& j, IcebergTableName& p);
 namespace facebook::presto::protocol::iceberg {
 enum class PartitionTransformType {
   IDENTITY,
-  YEAR,
-  MONTH,
-  DAY,
   HOUR,
+  DAY,
+  MONTH,
+  YEAR,
   BUCKET,
   TRUNCATE
 };