[Kernel]Generalize the actions after commit(like checkpoint) by introducing post commit action to kernel

kernel/examples/kernel-examples/src/main/java/io/delta/kernel/examples/CreateTableAndInsertData.java

@@ -25,7 +25,8 @@
 import io.delta.kernel.data.FilteredColumnarBatch;
 import io.delta.kernel.data.Row;
 import io.delta.kernel.expressions.Literal;
-import io.delta.kernel.hook.*;
+import io.delta.kernel.hook.PostCommitHook;
+import io.delta.kernel.hook.PostCommitHook.PostCommitHookType;
 import io.delta.kernel.utils.*;
 import static io.delta.kernel.examples.utils.Utils.parseArgs;
 

kernel/kernel-api/src/main/java/io/delta/kernel/hook/PostCommitHook.java

@@ -19,8 +19,17 @@
 import io.delta.kernel.engine.Engine;
 import java.io.IOException;
 
+/**
+ * A hook for executing operation after a transaction commit. Hooks are added in the Transaction and
+ * engine need to invoke the hook explicitly for executing the operation.
+ */
 public interface PostCommitHook {
 
+  enum PostCommitHookType {
+    // Write a new checkpoint at the version committed by the txn if required.
+    CHECKPOINT,
+  }
+
   /** Invokes the post commit operation whose implementation must be thread safe. */
   void threadSafeInvoke(Engine engine) throws IOException;
 

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TransactionImpl.java

@@ -27,11 +27,11 @@
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.exceptions.ConcurrentWriteException;
 import io.delta.kernel.expressions.Column;
-import io.delta.kernel.hook.CheckpointHook;
 import io.delta.kernel.hook.PostCommitHook;
 import io.delta.kernel.internal.actions.*;
 import io.delta.kernel.internal.data.TransactionStateRow;
 import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.hook.CheckpointHook;
 import io.delta.kernel.internal.metrics.TransactionMetrics;
 import io.delta.kernel.internal.metrics.TransactionReportImpl;
 import io.delta.kernel.internal.replay.ConflictChecker;

kernel/kernel-api/src/main/java/io/delta/kernel/hook/CheckpointHook.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/hook/CheckpointHook.java

@@ -13,13 +13,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.delta.kernel.hook;
+package io.delta.kernel.internal.hook;
 
 import io.delta.kernel.Table;
 import io.delta.kernel.engine.Engine;
+import io.delta.kernel.hook.PostCommitHook;
 import io.delta.kernel.internal.fs.Path;
 import java.io.IOException;
 
+/** Write a new checkpoint at the version committed by the txn if required. */
 public class CheckpointHook implements PostCommitHook {
 
   private final Path tablePath;

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWriteSuiteBase.scala

@@ -46,7 +46,7 @@ import io.delta.kernel.types.StructType
 import io.delta.kernel.utils.CloseableIterable.{emptyIterable, inMemoryIterable}
 import io.delta.kernel.utils.CloseableIterator
 import io.delta.kernel.Operation.CREATE_TABLE
-import io.delta.kernel.hook.PostCommitHookType
+import io.delta.kernel.hook.PostCommitHook.PostCommitHookType
 import org.apache.commons.io.FileUtils
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
@@ -411,13 +411,7 @@ trait DeltaTableWriteSuiteBase extends AnyFunSuite with TestUtils {
     expVersion: Long,
     expIsReadyForCheckpoint: Boolean): Unit = {
     assert(result.getVersion === expVersion)
-    assert(
-      result.getPostCommitHooks
-        .stream()
-        .anyMatch(
-          hook => hook.getType == PostCommitHookType.CHECKPOINT
-        ) === expIsReadyForCheckpoint
-    )
+    assertCheckpointReadiness(result, expIsReadyForCheckpoint)
   }
 
   def verifyTableProperties(
@@ -439,4 +433,16 @@ trait DeltaTableWriteSuiteBase extends AnyFunSuite with TestUtils {
     builder.append("]")
     checkAnswer(resultProperties, Seq(builder.toString()).map(TestRow(_)))
   }
+
+  def assertCheckpointReadiness(
+      txnResult: TransactionCommitResult,
+      isReadyForCheckpoint: Boolean): Unit = {
+    assert(
+      txnResult.getPostCommitHooks
+        .stream()
+        .anyMatch(
+          hook => hook.getType == PostCommitHookType.CHECKPOINT
+        ) === isReadyForCheckpoint
+    )
+  }
 }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWritesSuite.scala

@@ -25,7 +25,7 @@ import io.delta.kernel.engine.Engine
 import io.delta.kernel.exceptions._
 import io.delta.kernel.expressions.Literal
 import io.delta.kernel.expressions.Literal._
-import io.delta.kernel.hook.PostCommitHookType
+import io.delta.kernel.hook.PostCommitHook.PostCommitHookType
 import io.delta.kernel.internal.checkpoints.CheckpointerSuite.selectSingleElement
 import io.delta.kernel.internal.util.SchemaUtils.casePreservingPartitionColNames
 import io.delta.kernel.internal.{SnapshotImpl, TableConfig}
@@ -132,12 +132,7 @@ class DeltaTableWritesSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBa
       val txnResult = txn.commit(engine, emptyIterable())
 
       assert(txnResult.getVersion === 0)
-      assert(
-        !txnResult.getPostCommitHooks.stream()
-          .anyMatch(
-            hook => hook.getType == PostCommitHookType.CHECKPOINT
-          )
-        )
+      assertCheckpointReadiness(txnResult, isReadyForCheckpoint = false)
 
       verifyCommitInfo(tablePath = tablePath, version = 0)
       verifyWrittenContent(tablePath, testSchema, Seq.empty)
@@ -356,11 +351,7 @@ class DeltaTableWritesSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBa
       val txnResult = txn.commit(engine, emptyIterable())
 
       assert(txnResult.getVersion === 0)
-      assert(
-        !txnResult.getPostCommitHooks
-          .stream()
-          .anyMatch(hook => hook.getType == PostCommitHookType.CHECKPOINT)
-        )
+      assertCheckpointReadiness(txnResult, isReadyForCheckpoint = false)
 
       verifyCommitInfo(tablePath, version = 0, Seq("Part1", "part2"))
       verifyWrittenContent(tablePath, schema, Seq.empty)
@@ -378,11 +369,7 @@ class DeltaTableWritesSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBa
       val txnResult = txn.commit(engine, emptyIterable())
 
       assert(txnResult.getVersion === 0)
-      assert(
-        !txnResult.getPostCommitHooks
-            .stream()
-          .anyMatch(hook => hook.getType == PostCommitHookType.CHECKPOINT)
-      )
+      assertCheckpointReadiness(txnResult, isReadyForCheckpoint = false)
 
       verifyCommitInfo(tablePath, version = 0)
       verifyWrittenContent(tablePath, schema, Seq.empty)