[Kernel] [CatalogManaged] New Transaction and Committer etc. APIs

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,60 @@
+/*
+ * 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.transaction.TransactionV2;
+import io.delta.kernel.utils.CloseableIterator;
+
+/**
+ * A container class for all the information that an engine needs to commit to a table.
+ *
+ * <p>This interface encapsulates both the actions to be committed and the associated metadata
+ * required for the commit process. Engines use this context to provide inputs to the {@link
+ * Committer#commit} method.
+ *
+ * @see TransactionV2#getInitialCommitContext
+ */
+@Experimental
+public interface CommitContext {
+
+  /**
+   * Returns the finalized actions that the engine must forward to the {@link Committer} to commit
+   * to the table.
+   *
+   * <p>These actions represent the changes this transaction will make, including data file
+   * additions, metadata updates, and protocol changes.
+   *
+   * <p><b>Important limitations:</b>
+   *
+   * <ul>
+   *   <li>This iterator can only be accessed and consumed once
+   *   <li>For retry support, engines must materialize these actions before each commit attempt,
+   *       allowing them to be replayed and then updated during conflict resolution with the latest
+   *       table state
+   * </ul>
+   */
+  CloseableIterator<Row> getFinalizedActions();
+
+  /**
+   * Returns 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,
+   * Delta log path, and more.
+   */
+  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, located at {@code <table_root>/_delta_log}. */
+  public String getDeltaLogDirPath() {
+    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,68 @@
+/*
+ * 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>
+   *
+   * @param engine the {@link Engine} instance used for committing changes
+   * @param finalizedActions the iterator of finalized actions to be committed, taken from {@link
+   *     CommitContext#getFinalizedActions()}. Callers must either
+   *     <ul>
+   *       <li>Pass the iterator directly from the {@link CommitContext} into this call site
+   *       <li>First materialize the iterator contents (e.g., into a {@link java.util.List}) and
+   *           then create a new iterator from the materialized data
+   *     </ul>
+   *
+   * @param commitMetadata 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, Delta log path, and more.
+   * @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
+}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/table/ResolvedTableInternalImpl.java

@@ -19,6 +19,7 @@
 import static java.util.Objects.requireNonNull;
 
 import io.delta.kernel.ScanBuilder;
+import io.delta.kernel.commit.Committer;
 import io.delta.kernel.expressions.Column;
 import io.delta.kernel.internal.ScanBuilderImpl;
 import io.delta.kernel.internal.actions.DomainMetadata;
@@ -115,6 +116,11 @@ public ScanBuilder getScanBuilder() {
         new Path(getPath()), getProtocol(), getMetadata(), getSchema(), logReplay, snapshotReport);
   }
 
+  @Override
+  public Committer getCommitter() {
+    throw new UnsupportedOperationException("not implemented");
+  }
+
   ///////////////////////////////////////
   // ResolvedTableInternalImpl Methods //
   ///////////////////////////////////////