Add support for branching in Iceberg

Author: ebyhr

Diff for: plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -49,6 +49,7 @@ public class IcebergTableHandle
     private final int formatVersion;
     private final String tableLocation;
     private final Map<String, String> storageProperties;
+    private final Optional<String> branch;
 
     // Filter used during split generation and table scan, but not required to be strictly enforced by Iceberg Connector
     private final TupleDomain<IcebergColumnHandle> unenforcedPredicate;
@@ -92,7 +93,8 @@ public static IcebergTableHandle fromJsonForDeserializationOnly(
             @JsonProperty("projectedColumns") Set<IcebergColumnHandle> projectedColumns,
             @JsonProperty("nameMappingJson") Optional<String> nameMappingJson,
             @JsonProperty("tableLocation") String tableLocation,
-            @JsonProperty("storageProperties") Map<String, String> storageProperties)
+            @JsonProperty("storageProperties") Map<String, String> storageProperties,
+            @JsonProperty("branch") Optional<String> branch)
     {
         return new IcebergTableHandle(
                 catalog,
@@ -111,6 +113,7 @@ public static IcebergTableHandle fromJsonForDeserializationOnly(
                 tableLocation,
                 storageProperties,
                 Optional.empty(),
+                branch,
                 false,
                 Optional.empty(),
                 ImmutableSet.of(),
@@ -134,6 +137,7 @@ public IcebergTableHandle(
             String tableLocation,
             Map<String, String> storageProperties,
             Optional<IcebergTablePartitioning> tablePartitioning,
+            Optional<String> branch,
             boolean recordScannedFiles,
             Optional<DataSize> maxScannedFileSize,
             Set<IcebergColumnHandle> constraintColumns,
@@ -155,6 +159,7 @@ public IcebergTableHandle(
         this.tableLocation = requireNonNull(tableLocation, "tableLocation is null");
         this.storageProperties = ImmutableMap.copyOf(requireNonNull(storageProperties, "storageProperties is null"));
         this.tablePartitioning = requireNonNull(tablePartitioning, "tablePartitioning is null");
+        this.branch = requireNonNull(branch, "branch is null");
         this.recordScannedFiles = recordScannedFiles;
         this.maxScannedFileSize = requireNonNull(maxScannedFileSize, "maxScannedFileSize is null");
         this.constraintColumns = ImmutableSet.copyOf(requireNonNull(constraintColumns, "constraintColumns is null"));
@@ -261,6 +266,12 @@ public Optional<IcebergTablePartitioning> getTablePartitioning()
         return tablePartitioning;
     }
 
+    @JsonProperty
+    public Optional<String> getBranch()
+    {
+        return branch;
+    }
+
     @JsonIgnore
     public boolean isRecordScannedFiles()
     {
@@ -314,6 +325,7 @@ public IcebergTableHandle withProjectedColumns(Set<IcebergColumnHandle> projecte
                 tableLocation,
                 storageProperties,
                 tablePartitioning,
+                branch,
                 recordScannedFiles,
                 maxScannedFileSize,
                 constraintColumns,
@@ -339,6 +351,7 @@ public IcebergTableHandle forAnalyze()
                 tableLocation,
                 storageProperties,
                 tablePartitioning,
+                branch,
                 recordScannedFiles,
                 maxScannedFileSize,
                 constraintColumns,
@@ -364,6 +377,7 @@ public IcebergTableHandle forOptimize(boolean recordScannedFiles, DataSize maxSc
                 tableLocation,
                 storageProperties,
                 tablePartitioning,
+                branch,
                 recordScannedFiles,
                 Optional.of(maxScannedFileSize),
                 constraintColumns,
@@ -389,6 +403,7 @@ public IcebergTableHandle withTablePartitioning(Optional<IcebergTablePartitionin
                 tableLocation,
                 storageProperties,
                 requiredTablePartitioning,
+                branch,
                 recordScannedFiles,
                 maxScannedFileSize,
                 constraintColumns,
@@ -422,6 +437,7 @@ public boolean equals(Object o)
                 Objects.equals(nameMappingJson, that.nameMappingJson) &&
                 Objects.equals(tableLocation, that.tableLocation) &&
                 Objects.equals(storageProperties, that.storageProperties) &&
+                Objects.equals(branch, that.branch) &&
                 Objects.equals(maxScannedFileSize, that.maxScannedFileSize) &&
                 Objects.equals(constraintColumns, that.constraintColumns) &&
                 Objects.equals(forAnalyze, that.forAnalyze);
@@ -446,6 +462,7 @@ public int hashCode()
                 nameMappingJson,
                 tableLocation,
                 storageProperties,
+                branch,
                 recordScannedFiles,
                 maxScannedFileSize,
                 constraintColumns,

Diff for: plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTableHandle.java

@@ -22,6 +22,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
@@ -37,7 +38,8 @@ public record IcebergWritableTableHandle(
         IcebergFileFormat fileFormat,
         Map<String, String> storageProperties,
         RetryMode retryMode,
-        Map<String, String> fileIoProperties)
+        Map<String, String> fileIoProperties,
+        Optional<String> branch)
         implements ConnectorInsertTableHandle, ConnectorOutputTableHandle
 {
     public IcebergWritableTableHandle
@@ -53,6 +55,7 @@ public record IcebergWritableTableHandle(
         requireNonNull(retryMode, "retryMode is null");
         checkArgument(partitionsSpecsAsJson.containsKey(partitionSpecId), "partitionSpecId missing from partitionSpecs");
         fileIoProperties = ImmutableMap.copyOf(requireNonNull(fileIoProperties, "fileIoProperties is null"));
+        requireNonNull(branch, "branch is null");
     }
 
     @Override

Diff for: plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergWritableTableHandle.java

@@ -265,6 +265,7 @@ private static IcebergTableHandle getIcebergTableHandle(PartitionSpec partitionS
                 "dummy_table_location",
                 ImmutableMap.of(),
                 Optional.empty(),
+                Optional.empty(),
                 false,
                 Optional.empty(),
                 ImmutableSet.of(),