Support Iceberg Identifier fields

docs/src/main/sphinx/connector/iceberg.md

@@ -1052,6 +1052,10 @@ connector using a {doc}`WITH </sql/create-table-as>` clause.
     Defaults to `false`. 
 * - `data_location`
   - Optionally specifies the file system location URI for the table's data files
+* - `identifier_fields`
+  - Optionally specifies table `identifier-field-ids`. If a table is `identifier-field-ids` by
+    columns `c1` and `c2`, the identifier_fields property is `identifier_fields =
+    ARRAY['c1', 'c2']`.
 * - `extra_properties`
   - Additional properties added to an Iceberg table. The properties are not used by Trino,
     and are available in the `$properties` metadata table.

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -308,12 +308,14 @@
 import static io.trino.plugin.iceberg.IcebergTableProperties.EXTRA_PROPERTIES_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.FILE_FORMAT_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.FORMAT_VERSION_PROPERTY;
+import static io.trino.plugin.iceberg.IcebergTableProperties.IDENTIFIER_FIELDS_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.MAX_COMMIT_RETRY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.OBJECT_STORE_LAYOUT_ENABLED_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.ORC_BLOOM_FILTER_COLUMNS_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.PARTITIONING_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.SORTED_BY_PROPERTY;
+import static io.trino.plugin.iceberg.IcebergTableProperties.getIdentifierFields;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getPartitioning;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getTableLocation;
 import static io.trino.plugin.iceberg.IcebergTableProperties.validateCompression;
@@ -445,6 +447,7 @@ public class IcebergMetadata
             .add(PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY)
             .add(PARTITIONING_PROPERTY)
             .add(SORTED_BY_PROPERTY)
+            .add(IDENTIFIER_FIELDS_PROPERTY)
             .build();
     private static final String SYSTEM_SCHEMA = "system";
 
@@ -1224,7 +1227,7 @@ public void setMaterializedViewColumnComment(ConnectorSession session, SchemaTab
     @Override
     public Optional<ConnectorTableLayout> getNewTableLayout(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
-        Schema schema = schemaFromMetadata(tableMetadata.getColumns());
+        Schema schema = schemaFromMetadata(tableMetadata.getColumns(), getIdentifierFields(tableMetadata.getProperties()));
         PartitionSpec partitionSpec = parsePartitionFields(schema, getPartitioning(tableMetadata.getProperties()));
         return getWriteLayout(schema, partitionSpec, false);
     }
@@ -2491,6 +2494,13 @@ public void setTableProperties(ConnectorSession session, ConnectorTableHandle ta
             }
         }
 
+        if (properties.containsKey(IDENTIFIER_FIELDS_PROPERTY)) {
+            @SuppressWarnings("unchecked")
+            List<String> identifierFields = (List<String>) properties.get(IDENTIFIER_FIELDS_PROPERTY)
+                    .orElse(Collections.emptyList());
+            updateIdentifierFields(icebergTable, transaction, identifierFields);
+        }
+
         commitTransaction(transaction, "set table properties");
     }
 
@@ -2538,6 +2548,30 @@ private static void updatePartitioning(Table icebergTable, Transaction transacti
         }
     }
 
+    private void updateIdentifierFields(Table icebergTable, Transaction transaction, List<String> identifierFields)
+    {
+        UpdateSchema updateSchema = transaction.updateSchema();
+        if (!identifierFields.isEmpty()) {
+            Schema schema = icebergTable.schema();
+            for (String identifierField : identifierFields) {
+                NestedField field = schema.findField(identifierField);
+                if (field == null) {
+                    throw new TrinoException(COLUMN_NOT_FOUND, "Field '" + identifierField + "' does not exist");
+                }
+                if (field.isOptional()) {
+                    throw new TrinoException(NOT_SUPPORTED, "Identifier field '" + identifierField + "' cannot be optional");
+                }
+            }
+        }
+        updateSchema.setIdentifierFields(identifierFields);
+        try {
+            updateSchema.commit();
+        }
+        catch (Exception e) {
+            throw new TrinoException(ICEBERG_COMMIT_ERROR, "Failed to set new identifierFields value", e);
+        }
+    }
+
     private static Term toIcebergTerm(Schema schema, PartitionField partitionField)
     {
         return Expressions.transform(schema.findColumnName(partitionField.sourceId()), partitionField.transform());

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTableProperties.java

@@ -70,6 +70,7 @@ public class IcebergTableProperties
     public static final String OBJECT_STORE_LAYOUT_ENABLED_PROPERTY = "object_store_layout_enabled";
     public static final String DATA_LOCATION_PROPERTY = "data_location";
     public static final String EXTRA_PROPERTIES_PROPERTY = "extra_properties";
+    public static final String IDENTIFIER_FIELDS_PROPERTY = "identifier_fields";
 
     public static final Set<String> SUPPORTED_PROPERTIES = ImmutableSet.<String>builder()
             .add(FILE_FORMAT_PROPERTY)
@@ -85,6 +86,7 @@ public class IcebergTableProperties
             .add(DATA_LOCATION_PROPERTY)
             .add(EXTRA_PROPERTIES_PROPERTY)
             .add(PARQUET_BLOOM_FILTER_COLUMNS_PROPERTY)
+            .add(IDENTIFIER_FIELDS_PROPERTY)
             .build();
 
     // These properties are used by Trino or Iceberg internally and cannot be set directly by users through extra_properties
@@ -126,6 +128,15 @@ public IcebergTableProperties(
                         false,
                         value -> (List<?>) value,
                         value -> value))
+                .add(new PropertyMetadata<>(
+                        IDENTIFIER_FIELDS_PROPERTY,
+                        "Identifier fields",
+                        new ArrayType(VARCHAR),
+                        List.class,
+                        ImmutableList.of(),
+                        false,
+                        value -> (List<?>) value,
+                        value -> value))
                 .add(new PropertyMetadata<>(
                         SORTED_BY_PROPERTY,
                         "Sorted columns",
@@ -247,6 +258,13 @@ public static List<String> getPartitioning(Map<String, Object> tableProperties)
         return partitioning == null ? ImmutableList.of() : ImmutableList.copyOf(partitioning);
     }
 
+    @SuppressWarnings("unchecked")
+    public static List<String> getIdentifierFields(Map<String, Object> tableProperties)
+    {
+        List<String> identifierFields = (List<String>) tableProperties.get(IDENTIFIER_FIELDS_PROPERTY);
+        return identifierFields == null ? ImmutableList.of() : ImmutableList.copyOf(identifierFields);
+    }
+
     @SuppressWarnings("unchecked")
     public static List<String> getSortOrder(Map<String, Object> tableProperties)
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergUtil.java

@@ -91,6 +91,7 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Base64;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -131,6 +132,7 @@
 import static io.trino.plugin.iceberg.IcebergTableProperties.DATA_LOCATION_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.FILE_FORMAT_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.FORMAT_VERSION_PROPERTY;
+import static io.trino.plugin.iceberg.IcebergTableProperties.IDENTIFIER_FIELDS_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.LOCATION_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.MAX_COMMIT_RETRY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.OBJECT_STORE_LAYOUT_ENABLED_PROPERTY;
@@ -141,6 +143,7 @@
 import static io.trino.plugin.iceberg.IcebergTableProperties.PROTECTED_ICEBERG_NATIVE_PROPERTIES;
 import static io.trino.plugin.iceberg.IcebergTableProperties.SORTED_BY_PROPERTY;
 import static io.trino.plugin.iceberg.IcebergTableProperties.SUPPORTED_PROPERTIES;
+import static io.trino.plugin.iceberg.IcebergTableProperties.getIdentifierFields;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getPartitioning;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getSortOrder;
 import static io.trino.plugin.iceberg.IcebergTableProperties.validateCompression;
@@ -153,6 +156,7 @@
 import static io.trino.plugin.iceberg.TypeConverter.toIcebergTypeForNewColumn;
 import static io.trino.plugin.iceberg.TypeConverter.toTrinoType;
 import static io.trino.plugin.iceberg.util.Timestamps.timestampTzFromMicros;
+import static io.trino.spi.StandardErrorCode.COLUMN_NOT_FOUND;
 import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static io.trino.spi.StandardErrorCode.INVALID_ARGUMENTS;
 import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
@@ -329,6 +333,11 @@ public static Map<String, Object> getIcebergTableProperties(BaseTable icebergTab
 
         compressionCodec.ifPresent(hiveCompressionCodec -> properties.put(COMPRESSION_CODEC, HiveCompressionOption.valueOf(hiveCompressionCodec.name())));
 
+        ImmutableList<String> identifierFieldNames = icebergTable.schema().identifierFieldNames().stream().collect(toImmutableList());
+        if (!identifierFieldNames.isEmpty()) {
+            properties.put(IDENTIFIER_FIELDS_PROPERTY, identifierFieldNames);
+        }
+
         SortOrder sortOrder = icebergTable.sortOrder();
         // TODO: Support sort column transforms (https://github.com/trinodb/trino/issues/15088)
         if (sortOrder.isSorted() && sortOrder.fields().stream().allMatch(sortField -> sortField.transform().isIdentity())) {
@@ -848,6 +857,23 @@ public static Schema schemaFromMetadata(List<ColumnMetadata> columns)
         return new Schema(icebergSchema.asStructType().fields());
     }
 
+    public static Schema schemaFromMetadata(List<ColumnMetadata> columns, List<String> identifierFields)
+    {
+        Schema schema = schemaFromMetadata(columns);
+        if (identifierFields.isEmpty()) {
+            return schema;
+        }
+        Set<Integer> ids = new HashSet<>();
+        for (String identifierField : identifierFields) {
+            NestedField field = schema.findField(identifierField);
+            if (field == null) {
+                throw new TrinoException(COLUMN_NOT_FOUND, "Unknown identifier field: " + identifierField);
+            }
+            ids.add(field.fieldId());
+        }
+        return new Schema(schema.schemaId(), schema.columns(), schema.getAliases(), ids);
+    }
+
     public static Schema schemaFromViewColumns(TypeManager typeManager, List<ViewColumn> columns)
     {
         List<NestedField> icebergColumns = new ArrayList<>();
@@ -872,7 +898,7 @@ public static List<ViewColumn> viewColumnsFromSchema(TypeManager typeManager, Sc
     public static Transaction newCreateTableTransaction(TrinoCatalog catalog, ConnectorTableMetadata tableMetadata, ConnectorSession session, boolean replace, String tableLocation, Predicate<String> allowedExtraProperties)
     {
         SchemaTableName schemaTableName = tableMetadata.getTable();
-        Schema schema = schemaFromMetadata(tableMetadata.getColumns());
+        Schema schema = schemaFromMetadata(tableMetadata.getColumns(), getIdentifierFields(tableMetadata.getProperties()));
         PartitionSpec partitionSpec = parsePartitionFields(schema, getPartitioning(tableMetadata.getProperties()));
         SortOrder sortOrder = parseSortFields(schema, getSortOrder(tableMetadata.getProperties()));
 

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/AbstractTrinoCatalog.java

@@ -79,6 +79,7 @@
 import static io.trino.plugin.iceberg.IcebergMaterializedViewProperties.getStorageSchema;
 import static io.trino.plugin.iceberg.IcebergSessionProperties.isUseFileSizeFromMetadata;
 import static io.trino.plugin.iceberg.IcebergTableName.tableNameWithType;
+import static io.trino.plugin.iceberg.IcebergTableProperties.getIdentifierFields;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getPartitioning;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getSortOrder;
 import static io.trino.plugin.iceberg.IcebergTableProperties.getTableLocation;
@@ -309,7 +310,7 @@ protected Location createMaterializedViewStorage(
                 .orElseGet(() -> defaultTableLocation(session, viewName));
         List<ColumnMetadata> columns = columnsForMaterializedView(definition, materializedViewProperties);
 
-        Schema schema = schemaFromMetadata(columns);
+        Schema schema = schemaFromMetadata(columns, getIdentifierFields(materializedViewProperties));
         PartitionSpec partitionSpec = parsePartitionFields(schema, getPartitioning(materializedViewProperties));
         SortOrder sortOrder = parseSortFields(schema, getSortOrder(materializedViewProperties));
         Map<String, String> properties = createTableProperties(new ConnectorTableMetadata(storageTableName, columns, materializedViewProperties, Optional.empty()), _ -> false);
@@ -371,7 +372,7 @@ private List<ColumnMetadata> columnsForMaterializedView(ConnectorMaterializedVie
                     }
                     return new ColumnMetadata(column.getName(), type);
                 })
-                .collect(toImmutableList()));
+                .collect(toImmutableList()), getIdentifierFields(materializedViewProperties));
         PartitionSpec partitionSpec = parsePartitionFields(schemaWithTimestampTzPreserved, getPartitioning(materializedViewProperties));
         Set<String> temporalPartitioningSources = partitionSpec.fields().stream()
                 .flatMap(partitionField -> {

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorSmokeTest.java

@@ -893,6 +893,26 @@ public void testIcebergTablesSystemTable()
         }
     }
 
+    @Test
+    public void testCreateTableWithIdentifierFields()
+    {
+        if (!hasBehavior(SUPPORTS_CREATE_TABLE)) {
+            return;
+        }
+        try (TestTable table = newTrinoTable(
+                "test_identifier_fields",
+                " (id BIGINT NOT NULL, name VARCHAR NOT NULL) WITH (identifier_fields = ARRAY['id'])"))
+        {
+
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .contains("identifier_fields = ARRAY['id']");
+
+            assertUpdate("ALTER TABLE " + table.getName() + " SET PROPERTIES identifier_fields = ARRAY['name']");
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .contains("identifier_fields = ARRAY['name']");
+        }
+    }
+
     protected void dropSchema(String schema)
             throws Exception
     {

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergV2.java

@@ -1485,6 +1485,72 @@ public void testMetadataDeleteAfterCommitEnabled()
         assertUpdate("DROP TABLE " + tableName);
     }
 
+    @Test
+    public void testCreateTableWithIdentifierFields()
+    {
+        try (TestTable table = newTrinoTable("test_identifier_fields",
+                "(id BIGINT NOT NULL, name VARCHAR NOT NULL) WITH (identifier_fields = ARRAY['id'])")) {
+
+            BaseTable icebergTable = loadTable(table.getName());
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .contains("identifier_fields = ARRAY['id']");
+            Set<String> identifierFieldNames = icebergTable.schema().identifierFieldNames();
+            assertThat(identifierFieldNames).containsExactlyInAnyOrder("id");
+
+            assertUpdate("ALTER TABLE " + table.getName() + " SET PROPERTIES identifier_fields = ARRAY['name']");
+            icebergTable = loadTable(table.getName());
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .contains("identifier_fields = ARRAY['name']");
+            identifierFieldNames = icebergTable.schema().identifierFieldNames();
+            assertThat(identifierFieldNames).containsExactlyInAnyOrder("name");
+        }
+    }
+
+    @Test
+    public void testCreateTableWithMutilIdentifierFields()
+    {
+        try (TestTable table = newTrinoTable("test_identifier_fields",
+                "(id BIGINT NOT NULL, name VARCHAR NOT NULL, area INT NOT NULL) WITH (identifier_fields = ARRAY['id','name'])")) {
+            BaseTable icebergTable = loadTable(table.getName());
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .containsAnyOf("identifier_fields = ARRAY['id','name']", "identifier_fields = ARRAY['name','id']");
+            Set<String> identifierFieldNames = icebergTable.schema().identifierFieldNames();
+            assertThat(identifierFieldNames).containsExactlyInAnyOrder("id", "name");
+
+            assertUpdate("ALTER TABLE " + table.getName() + " SET PROPERTIES identifier_fields = ARRAY['name','area']");
+            icebergTable = loadTable(table.getName());
+            assertThat((String) computeScalar("SHOW CREATE TABLE " + table.getName()))
+                    .containsAnyOf("identifier_fields = ARRAY['name','area']", "identifier_fields = ARRAY['area', 'name']");
+            identifierFieldNames = icebergTable.schema().identifierFieldNames();
+            assertThat(identifierFieldNames).containsExactlyInAnyOrder("id", "area");
+        }
+    }
+
+    @Test
+    public void testUpdateIdentifierFields()
+    {
+        try (TestTable table = newTrinoTable("test_identifier_fields",
+                "(id BIGINT NOT NULL, name VARCHAR NOT NULL, area INT)")) {
+            BaseTable icebergTable = loadTable(table.getName());
+
+            Set<String> identifierFieldNames = icebergTable.schema().identifierFieldNames();
+            assertThat(identifierFieldNames).isEmpty();
+
+            assertThat(query("ALTER TABLE " + table.getName() + " SET PROPERTIES identifier_fields = ARRAY['not_exist_col']"))
+                    .failure()
+                    .hasMessage("Field 'not_exist_col' does not exist");
+
+            assertThat(query("ALTER TABLE " + table.getName() + " SET PROPERTIES identifier_fields = ARRAY['area']"))
+                    .failure()
+                    .hasMessage("Identifier field 'area' cannot be optional");
+
+            assertUpdate("ALTER TABLE " + table.getName() + " SET PROPERTIES identifier_fields = ARRAY[]");
+            icebergTable.refresh();
+            identifierFieldNames = icebergTable.schema().identifierFieldNames();
+            assertThat(identifierFieldNames).isEmpty();
+        }
+    }
+
     @Test
     void testAnalyzeNoSnapshot()
     {