From b5dcde8f13aad22a8d13a55c5bd1c1d4ce477b36 Mon Sep 17 00:00:00 2001
From: Qianru Lao <qianru.lao@databricks.com>
Date: Tue, 28 May 2024 11:27:17 -0700
Subject: [PATCH] [Kernel] Add Backward Compatibility for
 KernelSnapshotDelegator

---
 .../flink/internal/KernelSnapshotDelegator.scala   | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/connectors/flink/src/main/scala/io/delta/flink/internal/KernelSnapshotDelegator.scala b/connectors/flink/src/main/scala/io/delta/flink/internal/KernelSnapshotDelegator.scala
index cd41349277..761ee57e59 100644
--- a/connectors/flink/src/main/scala/io/delta/flink/internal/KernelSnapshotDelegator.scala
+++ b/connectors/flink/src/main/scala/io/delta/flink/internal/KernelSnapshotDelegator.scala
@@ -18,9 +18,11 @@
 
 package io.delta.standalone.internal
 
+import io.delta.kernel.defaults.engine.DefaultEngine
 import io.delta.kernel.engine.Engine
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
+
 import io.delta.kernel.internal.{SnapshotImpl => SnapshotImplKernel}
 import io.delta.standalone.DeltaScan
 import io.delta.standalone.actions.{AddFile => AddFileJ, Metadata => MetadataJ}
@@ -63,6 +65,18 @@ class KernelSnapshotDelegator(
     engine: Engine)
   extends SnapshotImpl(hadoopConf, path, -1, LogSegment.empty(path), -1, standaloneDeltaLog, -1) {
 
+  // For backward compatibility
+  def this(
+    kernelSnapshot: SnapshotImplKernel,
+    kernelSnapshotWrapper: KernelSnapshotWrapper,
+    hadoopConf: Configuration,
+    path: Path,
+    version: Long,
+    kernelDeltaLog: KernelDeltaLogDelegator,
+    standaloneDeltaLog: DeltaLogImpl) =
+    this(kernelSnapshot, kernelSnapshotWrapper, hadoopConf, path, version, kernelDeltaLog,
+      standaloneDeltaLog, DefaultEngine.create(hadoopConf))
+
   lazy val standaloneSnapshot: SnapshotImpl = standaloneDeltaLog.getSnapshotForVersionAsOf(getVersion())
 
   /**