[Iceberg compaction] file metadata carriers and writer factory

sdks/java/io/iceberg/build.gradle

@@ -64,6 +64,7 @@ dependencies {
     provided "org.immutables:value:2.8.8"
     permitUnusedDeclared "org.immutables:value:2.8.8"
     implementation library.java.vendored_calcite_1_40_0
+    implementation library.java.jackson_databind
     runtimeOnly "org.apache.iceberg:iceberg-gcp:$iceberg_version"
     runtimeOnly "org.apache.iceberg:iceberg-aws:$iceberg_version"
     runtimeOnly "org.apache.iceberg:iceberg-aws-bundle:$iceberg_version"

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExecutedGroup.java

@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.maintenance;
+
+import com.google.auto.value.AutoValue;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.beam.sdk.io.iceberg.SerializableDataFile;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
+
+/** Represents the result of one rewritten subgroup as <b>compact commit descriptors</b>. */
+@AutoValue
+@DefaultSchema(AutoValueSchema.class)
+public abstract class ExecutedGroup {
+
+  @SchemaFieldNumber("0")
+  public abstract long getStartingSnapshotId();
+
+  /** The rewrite operation's id used to stamp the commit. */
+  @SchemaFieldNumber("1")
+  public abstract String getOperationId();
+
+  /** Index of the planned parent group this subgroup belongs to. */
+  @SchemaFieldNumber("2")
+  public abstract int getParentGroupIndex();
+
+  /** Total number of subgroups the parent was split into. */
+  @SchemaFieldNumber("3")
+  public abstract int getParentSubgroupCount();
+
+  /** Total input byte size of this group, for partial-progress failure accounting. */
+  @SchemaFieldNumber("4")
+  public abstract long getTotalInputByteSize();
+
+  /** Newly written compacted data files to ADD (full metrics). */
+  @SchemaFieldNumber("5")
+  public abstract List<SerializableDataFile> getNewFiles();
+
+  /** Rewritten input data files to DELETE (no metrics). */
+  @SchemaFieldNumber("6")
+  public abstract List<SerializableDataFile> getRewrittenDataFiles();
+
+  /** Dangling deletion vector JSONs to DELETE. */
+  @SchemaFieldNumber("7")
+  public abstract List<String> getDanglingDeleteFileJsons();
+
+  /** The starting snapshot's sequence number; the floor for the commit's idempotency stamp scan. */
+  @SchemaFieldNumber("8")
+  public abstract long getStartingSequenceNumber();
+
+  public static Builder builder() {
+    return new AutoValue_ExecutedGroup.Builder();
+  }
+
+  /**
+   * Locations of every newly written output file across {@code groups}. After a failed commit these
+   * are orphans; they carry the operation id so a later remove-orphan-files run can find them.
+   */
+  static List<String> newFilePaths(Iterable<ExecutedGroup> groups) {
+    List<String> paths = new ArrayList<>();
+    for (ExecutedGroup g : groups) {
+      for (SerializableDataFile sdf : g.getNewFiles()) {
+        paths.add(sdf.getPath());
+      }
+    }
+    return paths;
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder setStartingSnapshotId(long v);
+
+    public abstract Builder setStartingSequenceNumber(long v);
+
+    public abstract Builder setOperationId(String v);
+
+    public abstract Builder setParentGroupIndex(int v);
+
+    public abstract Builder setParentSubgroupCount(int v);
+
+    public abstract Builder setTotalInputByteSize(long v);
+
+    public abstract Builder setNewFiles(List<SerializableDataFile> v);
+
+    public abstract Builder setRewrittenDataFiles(List<SerializableDataFile> v);
+
+    public abstract Builder setDanglingDeleteFileJsons(List<String> v);
+
+    public abstract ExecutedGroup build();
+  }
+}

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/RangeFileScanTask.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.maintenance;
+
+import java.util.List;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * A worker-side {@link FileScanTask} reconstructed from a {@link TaskDescriptor}. Represents one
+ * row-group range of a data file, plus the delete files that apply to it.
+ */
+class RangeFileScanTask implements FileScanTask {
+  private final DataFile file;
+  private final List<DeleteFile> deletes;
+  private final long start;
+  private final long length;
+  private final PartitionSpec spec;
+
+  RangeFileScanTask(
+      DataFile file, List<DeleteFile> deletes, long start, long length, PartitionSpec spec) {
+    this.file = file;
+    this.deletes = deletes;
+    this.start = start;
+    this.length = length;
+    this.spec = spec;
+  }
+
+  @Override
+  public DataFile file() {
+    return file;
+  }
+
+  @Override
+  public List<DeleteFile> deletes() {
+    return deletes;
+  }
+
+  @Override
+  public long start() {
+    return start;
+  }
+
+  @Override
+  public long length() {
+    return length;
+  }
+
+  @Override
+  public PartitionSpec spec() {
+    return spec;
+  }
+
+  @Override
+  public Expression residual() {
+    return Expressions.alwaysTrue();
+  }
+
+  @Override
+  public Iterable<FileScanTask> split(long targetSplitSize) {
+    throw new UnsupportedOperationException(
+        "RangeFileScanTask is already a fixed row-group range and cannot be re-split.");
+  }
+}

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/RewriteSubGroup.java

@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.maintenance;
+
+import com.google.auto.value.AutoValue;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
+
+@AutoValue
+@DefaultSchema(AutoValueSchema.class)
+public abstract class RewriteSubGroup {
+  static Builder builder() {
+    return new AutoValue_RewriteSubGroup.Builder();
+  }
+
+  @SchemaFieldNumber("0")
+  abstract int getGlobalIndex();
+
+  /** Index of the planned parent group this subgroup belongs to; shared by all its subgroups. */
+  @SchemaFieldNumber("1")
+  abstract int getParentGroupIndex();
+
+  /** Total number of subgroups belonging to the parent. */
+  @SchemaFieldNumber("2")
+  abstract int getParentSubgroupCount();
+
+  /** The compact per-range descriptors this subgroup rewrites (one per row-group range). */
+  @SchemaFieldNumber("3")
+  abstract List<TaskDescriptor> getTaskDescriptors();
+
+  @SchemaFieldNumber("4")
+  abstract int getOutputSpecId();
+
+  @SchemaFieldNumber("5")
+  abstract long getWriteMaxFileSize();
+
+  @SchemaFieldNumber("6")
+  abstract long getTotalInputFileByteSize();
+
+  @SchemaFieldNumber("7")
+  abstract long getStartingSnapshotId();
+
+  /**
+   * The rewrite operation's id, unique to this pipeline execution. Used to name/tag output files
+   * and to stamp commits for idempotency.
+   */
+  @SchemaFieldNumber("8")
+  abstract String getOperationId();
+
+  /**
+   * The starting snapshot's own sequence number, captured at planning. It floors the commit's
+   * idempotency stamp scan, and still bounds the walk if that snapshot has since been expired.
+   */
+  @SchemaFieldNumber("9")
+  abstract long getStartingSequenceNumber();
+
+  @AutoValue.Builder
+  abstract static class Builder {
+    abstract Builder setGlobalIndex(int globalIndex);
+
+    abstract Builder setParentGroupIndex(int parentGroupIndex);
+
+    abstract Builder setParentSubgroupCount(int parentSubgroupCount);
+
+    abstract Builder setTaskDescriptors(List<TaskDescriptor> taskDescriptors);
+
+    /**
+     * Builds compact per-range descriptors from planned range tasks and records the group's total
+     * input byte size (the summed range lengths).
+     */
+    Builder setFileScanTasks(List<FileScanTask> tasks, Map<Integer, PartitionSpec> specs) {
+      long byteSize = 0;
+      List<TaskDescriptor> taskDescriptors = new ArrayList<>(tasks.size());
+      for (FileScanTask task : tasks) {
+        byteSize += task.length();
+        taskDescriptors.add(TaskDescriptor.from(task, specs));
+      }
+      return setTotalInputFileByteSize(byteSize).setTaskDescriptors(taskDescriptors);
+    }
+
+    abstract Builder setOutputSpecId(int outputSpecId);
+
+    abstract Builder setWriteMaxFileSize(long writeMaxFileSize);
+
+    abstract Builder setTotalInputFileByteSize(long byteSize);
+
+    abstract Builder setStartingSnapshotId(long startingSnapshotId);
+
+    abstract Builder setStartingSequenceNumber(long startingSequenceNumber);
+
+    abstract Builder setOperationId(String operationId);
+
+    abstract RewriteSubGroup build();
+  }
+}