From c5ff2365390a10c87da04ffec437d8d218d98490 Mon Sep 17 00:00:00 2001
From: Thang Long Vu <107926660+longvu-db@users.noreply.github.com>
Date: Wed, 20 Mar 2024 16:20:48 +0100
Subject: [PATCH] [Spark][Test-Only] Add Row Tracking Clone tests with empty
 source table (#2742)

## Description
Add Materialized Row Tracking columns CLONE tests, making sure we have
enough Row Tracking test coverage necessary to gear us towards [enabling
Row Tracking outside of
testing](https://github.com/delta-io/delta/pull/2059).

## How was this patch tested?
Existing + added UTs.

## Does this PR introduce _any_ user-facing changes?
No.
---
 .../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(