[Kernel] [CatalogManaged] ImmutableInternalTransactionState [PAUSE/ON HOLD/DO NOT REVIEW]

kernel/kernel-api/src/main/java/io/delta/kernel/ResolvedTable.java

@@ -17,6 +17,7 @@
 package io.delta.kernel;
 
 import io.delta.kernel.annotation.Experimental;
+import io.delta.kernel.commit.Committer;
 import io.delta.kernel.expressions.Column;
 import io.delta.kernel.types.StructType;
 import java.util.List;
@@ -64,4 +65,7 @@ public interface ResolvedTable {
 
   /** @return a scan builder for constructing scans to read data from this table */
   ScanBuilder getScanBuilder();
+
+  /** @return a committer that owns and controls commits to this table */
+  Committer getCommitter();
 }

kernel/kernel-api/src/main/java/io/delta/kernel/commit/CommitContext.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.commit;
+
+import io.delta.kernel.annotation.Experimental;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.utils.CloseableIterator;
+
+/** A container class for all the information than an engine needs to commit to a table. */
+@Experimental
+public interface CommitContext {
+
+  /**
+   * The finalized actions that the engine must forward to the {@link Committer} to commit to the
+   * table.
+   *
+   * <p>This iterator can only be accessed and consumed once.
+   *
+   * <p>If the engine wishes to support commit retries, the engine must materialize this actions
+   * iterator so that it can be replayed and updated in accordance with the latest table state.
+   */
+  CloseableIterator<Row> getFinalizedActions();
+
+  /**
+   * Get the {@link CommitMetadata} associated with this commit, which contains additional metadata
+   * required to commit the finalized actions to the table, such as the commit version.
+   */
+  CommitMetadata getCommitMetadata();
+}

kernel/kernel-api/src/main/java/io/delta/kernel/commit/CommitFailedException.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.commit;
+
+import io.delta.kernel.annotation.Experimental;
+
+/**
+ * Exception raised by {@link Committer#commit}.
+ *
+ * <pre>
+ *  | retryable | conflict  | meaning                                                         |
+ *  |   no      |   no      | something bad happened (e.g. auth failure)                      |
+ *  |   no      |   yes     | permanent transaction conflict (e.g. multi-table commit failed) |
+ *  |   yes     |   no      | transient error (e.g. network hiccup)                           |
+ *  |   yes     |   yes     | physical conflict (allowed to rebase and retry)                 |
+ * </pre>
+ */
+@Experimental
+public class CommitFailedException extends Exception {
+
+  private final boolean retryable;
+  private final boolean conflict;
+
+  // TODO: [delta-io/delta#4908] Include the winning, conflicting catalog ratified commits here
+
+  public CommitFailedException(boolean retryable, boolean conflict, String message) {
+    super(message);
+    this.retryable = retryable;
+    this.conflict = conflict;
+  }
+
+  public CommitFailedException(
+      boolean retryable, boolean conflict, String message, Throwable cause) {
+    super(message, cause);
+    this.retryable = retryable;
+    this.conflict = conflict;
+  }
+
+  /** Returns whether the commit can be retried. */
+  public boolean isRetryable() {
+    return retryable;
+  }
+
+  /** Returns whether the commit failed due to a conflict. */
+  public boolean isConflict() {
+    return conflict;
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/commit/CommitMetadata.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.commit;
+
+import static io.delta.kernel.internal.util.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.annotation.Experimental;
+import io.delta.kernel.internal.actions.CommitInfo;
+import io.delta.kernel.internal.actions.Metadata;
+import io.delta.kernel.internal.actions.Protocol;
+import java.util.Optional;
+
+/**
+ * Contains all information (excluding the iterator of finalized actions) required to commit changes
+ * to a Delta table.
+ */
+@Experimental
+public class CommitMetadata {
+
+  private final long version;
+  private final String logPath;
+  private final CommitInfo commitInfo;
+  private final Optional<Protocol> readProtocolOpt;
+  private final Optional<Metadata> readMetadataOpt;
+  private final Optional<Protocol> newProtocolOpt;
+  private final Optional<Metadata> newMetadataOpt;
+
+  public CommitMetadata(
+      long version,
+      String logPath,
+      CommitInfo commitInfo,
+      Optional<Protocol> readProtocolOpt,
+      Optional<Metadata> readMetadataOpt,
+      Optional<Protocol> newProtocolOpt,
+      Optional<Metadata> newMetadataOpt) {
+    checkArgument(version >= 0, "version must be non-negative: %d", version);
+    this.version = version;
+    this.logPath = requireNonNull(logPath, "logPath is null");
+    this.commitInfo = requireNonNull(commitInfo, "commitInfo is null");
+    this.readProtocolOpt = requireNonNull(readProtocolOpt, "readProtocolOpt is null");
+    this.readMetadataOpt = requireNonNull(readMetadataOpt, "readMetadataOpt is null");
+    this.newProtocolOpt = requireNonNull(newProtocolOpt, "newProtocolOpt is null");
+    this.newMetadataOpt = requireNonNull(newMetadataOpt, "newMetadataOpt is null");
+  }
+
+  /** The version of the Delta table this commit is targeting. */
+  public long getVersion() {
+    return version;
+  }
+
+  /** The path to the Delta log directory. */
+  public String getLogPath() {
+    return logPath;
+  }
+
+  /** The {@link CommitInfo} that is being written as part of this commit. */
+  public CommitInfo getCommitInfo() {
+    return commitInfo;
+  }
+
+  /**
+   * The {@link Protocol} that was read at the beginning of the commit. Empty if a new table is
+   * being created.
+   */
+  public Optional<Protocol> getReadProtocolOpt() {
+    return readProtocolOpt;
+  }
+
+  /**
+   * The {@link Metadata} that was read at the beginning of the commit. Empty if a new table is
+   * being created.
+   */
+  public Optional<Metadata> getReadMetadataOpt() {
+    return readMetadataOpt;
+  }
+
+  /**
+   * The {@link Protocol} that is being written as part of this commit. Empty if the protocol is not
+   * being changed.
+   */
+  public Optional<Protocol> getNewProtocolOpt() {
+    return newProtocolOpt;
+  }
+
+  /**
+   * The {@link Metadata} that is being written as part of this commit. Empty if the metadata is not
+   * being changed.
+   */
+  public Optional<Metadata> getNewMetadataOpt() {
+    return newMetadataOpt;
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/commit/CommitResponse.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.commit;
+
+import io.delta.kernel.annotation.Experimental;
+import io.delta.kernel.internal.files.ParsedLogData;
+
+/** Response container for the result of a commit operation. */
+@Experimental
+public class CommitResponse {
+
+  // TODO: Create a DeltaLogData extends ParsedLogData that includes commit timestamp information.
+  private final ParsedLogData commitLogData;
+
+  public CommitResponse(ParsedLogData commitLogData) {
+    this.commitLogData = commitLogData;
+  }
+
+  /**
+   * The parsed log data resulting from the commit operation. Note that for catalog-managed tables,
+   * this may be the ratified staged commit, the ratified inline commit, or even a published Delta
+   * file that the {@link Committer} implementation decided to publish after committing to the
+   * managing catalog.
+   */
+  public ParsedLogData getCommitLogData() {
+    return commitLogData;
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/commit/Committer.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.commit;
+
+import io.delta.kernel.annotation.Experimental;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.utils.CloseableIterator;
+
+/**
+ * Interface for committing changes to Delta tables, supporting both filesystem-managed and
+ * catalog-managed tables.
+ */
+@Experimental
+public interface Committer {
+
+  /**
+   * Commits the given {@code finalizedActions} and {@code commitMetadata} to the table.
+   *
+   * <p>Filesystem-managed tables: Implementations must write the {@code finalizedActions} into a
+   * new Delta JSON file at version {@link CommitMetadata#getVersion()} using atomic file operations
+   * (PUT-if-absent semantics).
+   *
+   * <p>Catalog-managed tables: Implementations must follow the commit rules and requirements as
+   * dictated by the managing catalog to ensure commit atomicity and consistency. This may involve:
+   *
+   * <ol>
+   *   <li>Writing the finalized actions into a staged commit file
+   *   <li>Calling catalog commit APIs with the staged commit location (or inline content) and
+   *       additional metadata (such as the commit Protocol and Metadata)
+   *   <li>Publishing ratified catalog commits into the Delta log
+   * </ol>
+   *
+   * @return CommitResponse containing the resultant commit
+   * @throws CommitFailedException if the commit operation fails.
+   */
+  CommitResponse commit(
+      Engine engine, CloseableIterator<Row> finalizedActions, CommitMetadata commitMetadata)
+      throws CommitFailedException;
+
+  // TODO: API to get the required table properties
+}