Add the test abstraction so that we can design delta integration tests run agaist both locally OSS UC server and external OSS UC server

.gitignore

@@ -136,3 +136,7 @@ override.tf.json
 # Ignore kernel benchmark report
 kernel/kernel-benchmarks/benchmark_report.json
 
+# Unity Catalog test artifacts
+spark/unitycatalog/etc/
+.scala-build/
+

build.sbt

@@ -705,6 +705,76 @@ lazy val contribs = (project in file("contribs"))
     Compile / compile := ((Compile / compile) dependsOn createTargetClassesDir).value
   ).configureUnidoc()
 
+
+val unityCatalogVersion = "0.3.0"
+val sparkUnityCatalogJacksonVersion = "2.15.4" // We are using Spark 4.0's Jackson version 2.15.x, to override Unity Catalog 0.3.0's version 2.18.x
+
+lazy val sparkUnityCatalog = (project in file("spark/unitycatalog"))
+  .dependsOn(spark % "compile->compile;test->test;provided->provided")
+  .disablePlugins(ScalafmtPlugin)
+  .settings(
+    name := "delta-spark-unitycatalog",
+    commonSettings,
+    skipReleaseSettings,
+    CrossSparkVersions.sparkDependentSettings(sparkVersion),
+
+    // This is a test-only module - no production sources
+    Compile / sources := Seq.empty,
+
+    // Ensure Java sources are picked up
+    Test / unmanagedSourceDirectories += baseDirectory.value / "src" / "test" / "java",
+
+    Test / javaOptions ++= Seq("-ea"),
+
+    // Don't execute in parallel since we can't have multiple Sparks in the same JVM
+    Test / parallelExecution := false,
+
+    // Force ALL Jackson dependencies to match Spark's Jackson version
+    // This overrides Jackson from Unity Catalog's transitive dependencies (e.g., Armeria)
+    dependencyOverrides ++= Seq(
+      "com.fasterxml.jackson.core" % "jackson-core" % sparkUnityCatalogJacksonVersion,
+      "com.fasterxml.jackson.core" % "jackson-annotations" % sparkUnityCatalogJacksonVersion,
+      "com.fasterxml.jackson.core" % "jackson-databind" % sparkUnityCatalogJacksonVersion,
+      "com.fasterxml.jackson.module" %% "jackson-module-scala" % sparkUnityCatalogJacksonVersion,
+      "com.fasterxml.jackson.dataformat" % "jackson-dataformat-yaml" % sparkUnityCatalogJacksonVersion,
+      "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % sparkUnityCatalogJacksonVersion,
+      "com.fasterxml.jackson.datatype" % "jackson-datatype-jdk8" % sparkUnityCatalogJacksonVersion
+    ),
+
+    libraryDependencies ++= Seq(
+      // Standard test dependencies
+      "org.scalatest" %% "scalatest" % scalaTestVersion % "test",
+
+      // Unity Catalog dependencies - exclude Jackson to use Spark's Jackson 2.15.x
+      "io.unitycatalog" %% "unitycatalog-spark" % unityCatalogVersion % "test" excludeAll(
+        ExclusionRule(organization = "com.fasterxml.jackson.core"),
+        ExclusionRule(organization = "com.fasterxml.jackson.module"),
+        ExclusionRule(organization = "com.fasterxml.jackson.datatype"),
+        ExclusionRule(organization = "com.fasterxml.jackson.dataformat")
+      ),
+      "io.unitycatalog" % "unitycatalog-server" % unityCatalogVersion % "test" excludeAll(
+        ExclusionRule(organization = "com.fasterxml.jackson.core"),
+        ExclusionRule(organization = "com.fasterxml.jackson.module"),
+        ExclusionRule(organization = "com.fasterxml.jackson.datatype"),
+        ExclusionRule(organization = "com.fasterxml.jackson.dataformat")
+      ),
+
+      // Spark test dependencies
+      "org.apache.spark" %% "spark-sql" % sparkVersion.value % "test",
+      "org.apache.spark" %% "spark-catalyst" % sparkVersion.value % "test",
+      "org.apache.spark" %% "spark-core" % sparkVersion.value % "test",
+
+      // TODO: Let's define a common junit version.
+      "org.junit.jupiter" % "junit-jupiter" % "5.10.3" % Test,
+      "org.apache.hadoop" % "hadoop-aws" % hadoopVersion % Test,
+      "net.aichler" % "jupiter-interface" % JupiterKeys.jupiterVersion.value % Test,
+      "org.assertj" % "assertj-core" % "3.26.3" % Test,
+    ),
+
+    Test / testOptions += Tests.Argument("-oDF"),
+    Test / testOptions += Tests.Argument(TestFrameworks.JUnit, "-v", "-a")
+  )
+
 lazy val sharing = (project in file("sharing"))
   .dependsOn(spark % "compile->compile;test->test;provided->provided")
   .disablePlugins(JavaFormatterPlugin, ScalafmtPlugin)
@@ -1515,7 +1585,7 @@ val createTargetClassesDir = taskKey[Unit]("create target classes dir")
 
 // Don't use these groups for any other projects
 lazy val sparkGroup = project
-  .aggregate(spark, sparkV1, sparkV1Filtered, sparkV2, contribs, storage, storageS3DynamoDB, sharing, hudi)
+  .aggregate(spark, sparkV1, sparkV1Filtered, sparkV2, contribs, sparkUnityCatalog, storage, storageS3DynamoDB, sharing, hudi)
   .settings(
     // crossScalaVersions must be set to Nil on the aggregating project
     crossScalaVersions := Nil,

spark/unitycatalog/src/test/java/io/sparkuctest/UCManagedTableDMLTest.java

@@ -0,0 +1,36 @@
+/*
+ * 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.sparkuctest;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Test;
+
+public class UCManagedTableDMLTest extends UnityCatalogTestBase {
+
+  @Test
+  public void testBasicOperation() {
+    String tableName = fullTableName("testBasicOperation");
+    sql("CREATE TABLE %s (id INT, val STRING) USING DELTA LOCATION '%s'",
+        tableName, path("testBasicOperation"));
+
+    sql("INSERT INTO %s VALUES (1, 'AAA'), (2, 'BBB')", tableName);
+
+    assertEquals("Should have the expected results",
+        ImmutableList.of(row(1, "AAA"), row(2, "BBB")),
+        sql("SELECT * FROM %s ORDER BY id ASC", tableName));
+  }
+}

spark/unitycatalog/src/test/java/io/sparkuctest/UnityCatalogTestBase.java

@@ -0,0 +1,110 @@
+/*
+ * 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.sparkuctest;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.common.base.Preconditions;
+import io.sparkuctest.extension.UnityCatalogExtension;
+import io.sparkuctest.extension.UnityCatalogExtensionUtil;
+import java.util.List;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class UnityCatalogTestBase {
+
+  @RegisterExtension
+  public static UnityCatalogExtension UC_EXTENSION = UnityCatalogExtensionUtil.initialize();
+
+  private static SparkSession spark;
+
+  @BeforeAll
+  public static void beforeAll() {
+    SparkConf conf = new SparkConf()
+        .setAppName("UnityCatalog Support Tests")
+        .setMaster("local[2]")
+        .set("spark.ui.enabled", "false")
+        .set("spark.sql.shuffle.partitions", "5")
+        // Delta Lake required configurations
+        .set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
+        .set("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog");
+
+    // Configure with Unity Catalog
+    UC_EXTENSION.catalogSparkConf().forEach(conf::set);
+
+    // Build the spark session.
+    spark = SparkSession.builder().config(conf).getOrCreate();
+  }
+
+  @AfterAll
+  public static void afterAll() {
+    if (spark != null) {
+      spark.stop();
+      spark = null;
+    }
+  }
+
+  protected String fullTableName(String tableName) {
+    return String.format("`%s`.`%s`.`%s`",
+        UC_EXTENSION.catalogName(), UC_EXTENSION.schemaName(), tableName);
+  }
+
+  protected String path(String basename) {
+    Preconditions.checkNotNull(basename, "basename cannot be null");
+    if (basename.startsWith("/")) {
+      return String.format("%s%s", UC_EXTENSION.rootTestingDir(), basename);
+    } else {
+      return String.format("%s/%s", UC_EXTENSION.rootTestingDir(), basename);
+    }
+  }
+
+  public static List<Row> sql(String statement, Object... args) {
+    return spark.sql(String.format(statement, args)).collectAsList();
+  }
+
+  public static Object[] row(Object... args) {
+    return args;
+  }
+
+  public static void assertEquals(String context, List<Object[]> expected, List<Row> actual) {
+    assertThat(expected)
+        .as("%s: number of results should match", context)
+        .hasSameSizeAs(actual);
+
+    for (int row = 0; row < expected.size(); row += 1) {
+      Object[] expectedRow = expected.get(row);
+      Row actualRow = actual.get(row);
+
+      assertEquals(context + ": row " + (row + 1), expectedRow, actualRow);
+    }
+  }
+
+  public static void assertEquals(String context, Object[] expected, Row actual) {
+    assertThat(expected.length)
+        .as("%s: Number of columns should match", context)
+        .isEqualTo(actual.size());
+    for (int i = 0; i < expected.length; i += 1) {
+      assertThat(expected[i])
+          .as("%s: Element does not match", context)
+          .isEqualTo(actual.get(i));
+    }
+  }
+}