From 06606e0dfb5409ba0c57359caa84f9016eb84e90 Mon Sep 17 00:00:00 2001
From: Thang Long VU <long.vu@databricks.com>
Date: Wed, 20 Mar 2024 01:05:16 +0100
Subject: [PATCH] Update

---
 .../sql/delta/rowid/RowIdCloneSuite.scala     | 43 +++++++++++++++++++
 1 file changed, 43 insertions(+)

diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/rowid/RowIdCloneSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/rowid/RowIdCloneSuite.scala
index 9fbc116a1e0..73164e2fec2 100644
--- a/spark/src/test/scala/org/apache/spark/sql/delta/rowid/RowIdCloneSuite.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/delta/rowid/RowIdCloneSuite.scala
@@ -86,6 +86,49 @@ class RowIdCloneSuite
     }
   }
 
+  for {
+    rowIdsEnabledOnSource <- BOOLEAN_DOMAIN
+    targetTableState <- Seq(TableState.EMPTY, TableState.NON_EXISTING)
+  } {
+    test("clone from empty source into an empty or non-existing target " +
+      s"does not assign row IDs, rowIdsEnabledOnSource=$rowIdsEnabledOnSource, " +
+      s"targetTableState=$targetTableState") {
+      withTables(
+        TableSetupInfo(tableName = "source",
+          rowIdsEnabled = rowIdsEnabledOnSource, tableState = TableState.EMPTY),
+        TableSetupInfo(tableName = "target",
+          rowIdsEnabled = false, tableState = targetTableState)) {
+        cloneTable(targetTableName = "target", sourceTableName = "source")
+
+        val (targetLog, snapshot) = DeltaLog.forTableWithSnapshot(spark, TableIdentifier("target"))
+        assertRowIdsAreNotSet(targetLog)
+        assert(RowId.isSupported(snapshot.protocol) === rowIdsEnabledOnSource)
+        assert(RowId.isEnabled(snapshot.protocol, snapshot.metadata) === rowIdsEnabledOnSource)
+      }
+    }
+  }
+
+  for (targetTableState <- Seq(TableState.EMPTY, TableState.NON_EXISTING))
+  test("clone from empty source into an empty or non-existing target " +
+    s"using property override does not assign row IDs, targetTableState=$targetTableState") {
+    withTables(
+      TableSetupInfo(tableName = "source",
+        rowIdsEnabled = false, tableState = TableState.EMPTY),
+      TableSetupInfo(tableName = "target",
+        rowIdsEnabled = false, tableState = targetTableState)) {
+
+      cloneTable(
+        targetTableName = "target",
+        sourceTableName = "source",
+        tblProperties = s"'${DeltaConfigs.ROW_TRACKING_ENABLED.key}' = true" :: Nil)
+
+      val (targetLog, snapshot) = DeltaLog.forTableWithSnapshot(spark, TableIdentifier("target"))
+      assertRowIdsAreNotSet(targetLog)
+      assert(RowId.isSupported(snapshot.protocol))
+      assert(RowId.isEnabled(snapshot.protocol, snapshot.metadata))
+    }
+  }
+
   test("clone that add row ID feature using table property override " +
     "doesn't enable row IDs on target") {
     withTables(