[Spark]Refactor Spark project structure to combine both Dsv1 connector and kernel backed Dsv2 connector

kernel-spark/src/main/java/io/delta/kernel/spark/read/SparkBatch.java

@@ -152,7 +152,7 @@ private long calculateMaxSplitBytes(SparkSession sparkSession) {
     int minPartitionNum =
         minPartitionNumOption.isDefined()
             ? ((Number) minPartitionNumOption.get()).intValue()
-            : sparkSession.leafNodeDefaultParallelism();
+            : sparkSession.sparkContext().defaultParallelism();
     if (minPartitionNum <= 0) {
       minPartitionNum = 1;
     }

kernel-spark/src/test/java/io/delta/kernel/spark/Dsv2BasicTest.java

@@ -43,10 +43,10 @@ public void setUp(@TempDir File tempDir) {
         new SparkConf()
             .set("spark.sql.catalog.dsv2", "io.delta.kernel.spark.catalog.TestCatalog")
             .set("spark.sql.catalog.dsv2.base_path", tempDir.getAbsolutePath())
-            .set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
+            .set("spark.sql.extensions", "io.delta.sql.LegacyDeltaSparkSessionExtension")
             .set(
                 "spark.sql.catalog.spark_catalog",
-                "org.apache.spark.sql.delta.catalog.DeltaCatalog")
+                "org.apache.spark.sql.delta.catalog.LegacyDeltaCatalog")
             .setMaster("local[*]")
             .setAppName("Dsv2BasicTest");
     spark = SparkSession.builder().config(conf).getOrCreate();

kernel-spark/src/test/java/io/delta/kernel/spark/SparkDsv2TestBase.java

@@ -32,10 +32,10 @@ public static void setUpSparkAndEngine() {
         SparkSession.builder()
             .master("local[*]")
             .appName("SparkKernelDsv2Tests")
-            .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
+            .config("spark.sql.extensions", "io.delta.sql.LegacyDeltaSparkSessionExtension")
             .config(
                 "spark.sql.catalog.spark_catalog",
-                "org.apache.spark.sql.delta.catalog.DeltaCatalog")
+                "org.apache.spark.sql.delta.catalog.LegacyDeltaCatalog")
             .getOrCreate();
     defaultEngine = DefaultEngine.create(spark.sessionState().newHadoopConf());
   }

kernel-spark/src/test/java/io/delta/kernel/spark/read/SparkGoldenTableTest.java

@@ -60,10 +60,10 @@ public void setUp(@TempDir File tempDir) {
         new SparkConf()
             .set("spark.sql.catalog.dsv2", "io.delta.kernel.spark.catalog.TestCatalog")
             .set("spark.sql.catalog.dsv2.base_path", tempDir.getAbsolutePath())
-            .set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
+            .set("spark.sql.extensions", "io.delta.sql.LegacyDeltaSparkSessionExtension")
             .set(
                 "spark.sql.catalog.spark_catalog",
-                "org.apache.spark.sql.delta.catalog.DeltaCatalog")
+                "org.apache.spark.sql.delta.catalog.LegacyDeltaCatalog")
             .setMaster("local[*]")
             .setAppName("SparkGoldenTableTest");
     spark = SparkSession.builder().config(conf).getOrCreate();

kernel-spark/src/test/java/io/delta/kernel/spark/utils/StreamingHelperTest.java

@@ -140,7 +140,7 @@ public void testGetActiveCommitAtTime_pastTimestamp(@TempDir File tempDir) throw
             .history()
             .getActiveCommitAtTime(
                 timestamp,
-                Option.empty() /* catalogTable */,
+                deltaLog.initialCatalogTable() /* catalogTableOpt */,
                 false /* canReturnLastCommit */,
                 true /* mustBeRecreatable */,
                 false /* canReturnEarliestCommit */);
@@ -172,7 +172,7 @@ public void testGetActiveCommitAtTime_futureTimestamp_canReturnLast(@TempDir Fil
             .history()
             .getActiveCommitAtTime(
                 futureTimestamp,
-                Option.empty() /* catalogTable */,
+                deltaLog.initialCatalogTable() /* catalogTableOpt */,
                 true /* canReturnLastCommit */,
                 true /* mustBeRecreatable */,
                 false /* canReturnEarliestCommit */);
@@ -204,7 +204,7 @@ public void testGetActiveCommitAtTime_futureTimestamp_notMustBeRecreatable(@Temp
             .history()
             .getActiveCommitAtTime(
                 futureTimestamp,
-                Option.empty() /* catalogTable */,
+                deltaLog.initialCatalogTable() /* catalogTableOpt */,
                 true /* canReturnLastCommit */,
                 false /* mustBeRecreatable */,
                 false /* canReturnEarliestCommit */);
@@ -236,7 +236,7 @@ public void testGetActiveCommitAtTime_earlyTimestamp_canReturnEarliest(@TempDir
             .history()
             .getActiveCommitAtTime(
                 earlyTimestamp,
-                Option.empty() /* catalogTable */,
+                deltaLog.initialCatalogTable() /* catalogTableOpt */,
                 false /* canReturnLastCommit */,
                 true /* mustBeRecreatable */,
                 true /* canReturnEarliestCommit */);
@@ -268,7 +268,7 @@ public void testGetActiveCommitAtTime_earlyTimestamp_notMustBeRecreatable_canRet
             .history()
             .getActiveCommitAtTime(
                 earlyTimestamp,
-                Option.empty() /* catalogTable */,
+                deltaLog.initialCatalogTable() /* catalogTableOpt */,
                 false /* canReturnLastCommit */,
                 false /* mustBeRecreatable */,
                 true /* canReturnEarliestCommit */);
@@ -353,12 +353,19 @@ public void testCheckVersionExists(
               deltaLog
                   .history()
                   .checkVersionExists(
-                      versionToCheck, Option.empty(), mustBeRecreatable, allowOutOfRange));
+                      versionToCheck,
+                      deltaLog.initialCatalogTable() /* catalogTableOpt */,
+                      mustBeRecreatable,
+                      allowOutOfRange));
     } else {
       streamingHelper.checkVersionExists(versionToCheck, mustBeRecreatable, allowOutOfRange);
       deltaLog
           .history()
-          .checkVersionExists(versionToCheck, Option.empty(), mustBeRecreatable, allowOutOfRange);
+          .checkVersionExists(
+              versionToCheck,
+              deltaLog.initialCatalogTable() /* catalogTableOpt */,
+              mustBeRecreatable,
+              allowOutOfRange);
     }
   }
 }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestUtils.scala

@@ -99,7 +99,9 @@ trait AbstractTestUtils
     .appName("Spark Test Writer for Delta Kernel")
     .config("spark.master", "local")
     .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
-    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
+    .config(
+      "spark.sql.catalog.spark_catalog",
+      "org.apache.spark.sql.delta.catalog.DeltaCatalog")
     // Set this conf to empty string so that the golden tables generated
     // using with the test-prefix (i.e. there is no DELTA_TESTING set) can still work
     .config(DeltaSQLConf.TEST_DV_NAME_PREFIX.key, "")

project/TestParallelization.scala

@@ -54,6 +54,7 @@ object TestParallelization {
     Test / testGroupingStrategy := {
       val groupsCount = (Test / forkTestJVMCount).value
       val shard = (Test / shardId).value
+      // Use regular baseDirectory for target directory (not Test/baseDirectory)
       val baseJvmDir = baseDirectory.value
       MinShardGroupDurationStrategy(groupsCount, baseJvmDir, shard, defaultForkOptions.value)
     },
@@ -81,7 +82,9 @@ object TestParallelization {
       javaHome = javaHome.value,
       outputStrategy = outputStrategy.value,
       bootJars = Vector.empty,
-      workingDirectory = Some(baseDirectory.value),
+      // Use Test/baseDirectory instead of baseDirectory to support modules where these differ
+      // (e.g. spark-combined module where Test/baseDirectory points to spark/ source directory)
+      workingDirectory = Some((Test / baseDirectory).value),
       runJVMOptions = (Test / javaOptions).value.toVector,
       connectInput = connectInput.value,
       envVars = (Test / envVars).value

spark-combined/src/main/java/org/apache/spark/sql/delta/catalog/DeltaCatalog.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright (2021) 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 org.apache.spark.sql.delta.catalog;
+
+/**
+ * Delta Catalog implementation that can delegate to both V1 and V2 implementations.
+ * This class sits in delta-spark (combined) module and can access:
+ * - V1: org.apache.spark.sql.delta.* (full version with DeltaLog)
+ * - V2: io.delta.kernel.spark.*
+ */
+public class DeltaCatalog extends AbstractDeltaCatalog {
+
+}
+
+

spark-combined/src/main/scala/io/delta/sql/DeltaSparkSessionExtension.scala

@@ -0,0 +1,37 @@
+/*
+ * Copyright (2021) 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.delta.sql
+
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+
+/**
+ * Delta Spark Session Extension that can register both V1 and V2 implementations.
+ * This class sits in delta-spark (combined) module and can access:
+ * - V1: org.apache.spark.sql.delta.* (full version with DeltaLog)
+ * - V2: io.delta.kernel.spark.*
+ */
+class DeltaSparkSessionExtension extends AbstractDeltaSparkSessionExtension {
+
+  /**
+   * NoOpRule for binary compatibility with Delta 3.3.0
+   * This class must remain here to satisfy MiMa checks
+   */
+  class NoOpRule extends Rule[LogicalPlan] {
+    override def apply(plan: LogicalPlan): LogicalPlan = plan
+  }
+}

spark/src/main/scala/io/delta/sql/DeltaSparkSessionExtension.scala

@@ -80,7 +80,8 @@ import org.apache.spark.sql.internal.SQLConf
  *
  * @since 0.4.0
  */
-class DeltaSparkSessionExtension extends (SparkSessionExtensions => Unit) {
+class LegacyDeltaSparkSessionExtension extends AbstractDeltaSparkSessionExtension
+class AbstractDeltaSparkSessionExtension extends (SparkSessionExtensions => Unit) {
   override def apply(extensions: SparkSessionExtensions): Unit = {
     extensions.injectParser { (_, parser) =>
       new DeltaSqlParser(parser)

spark/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala

@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.TimeTravel
 import org.apache.spark.sql.delta.DataFrameUtils
 import org.apache.spark.sql.delta.DeltaErrors.{TemporallyUnstableInputException, TimestampEarlierThanCommitRetentionException}
 import org.apache.spark.sql.delta.actions.TableFeatureProtocolUtils
-import org.apache.spark.sql.delta.catalog.DeltaCatalog
+import org.apache.spark.sql.delta.catalog.LegacyDeltaCatalog
 import org.apache.spark.sql.delta.catalog.DeltaTableV2
 import org.apache.spark.sql.delta.catalog.IcebergTablePlaceHolder
 import org.apache.spark.sql.delta.commands._
@@ -245,7 +245,7 @@ class DeltaAnalysis(session: SparkSession)
         case _ =>
           protocol
       }
-      val newDeltaCatalog = new DeltaCatalog()
+      val newDeltaCatalog = new LegacyDeltaCatalog()
       val existingTableOpt = newDeltaCatalog.getExistingTableIfExists(catalogTableTarget.identifier)
       val newTable = newDeltaCatalog
         .verifyTableAndSolidify(

spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala

@@ -20,12 +20,10 @@ package org.apache.spark.sql.delta
 import java.io.{FileNotFoundException, IOException}
 import java.nio.file.FileAlreadyExistsException
 import java.util.{ConcurrentModificationException, UUID}
-
 import scala.collection.JavaConverters._
-
-import org.apache.spark.sql.delta.skipping.clustering.temp.{ClusterBySpec}
+import org.apache.spark.sql.delta.skipping.clustering.temp.ClusterBySpec
 import org.apache.spark.sql.delta.actions.{CommitInfo, Metadata, Protocol, TableFeatureProtocolUtils}
-import org.apache.spark.sql.delta.catalog.DeltaCatalog
+import org.apache.spark.sql.delta.catalog.AbstractDeltaCatalog
 import org.apache.spark.sql.delta.commands.{AlterTableDropFeatureDeltaCommand, DeltaGenerateCommand}
 import org.apache.spark.sql.delta.constraints.Constraints
 import org.apache.spark.sql.delta.hooks.AutoCompactType
@@ -37,9 +35,8 @@ import org.apache.spark.sql.delta.redirect.RedirectState
 import org.apache.spark.sql.delta.schema.{DeltaInvariantViolationException, InvariantViolationException, SchemaUtils, UnsupportedDataTypeInfo}
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.util.JsonUtils
-import io.delta.sql.DeltaSparkSessionExtension
+import io.delta.sql.AbstractDeltaSparkSessionExtension
 import org.apache.hadoop.fs.{ChecksumException, Path}
-
 import org.apache.spark.{SparkConf, SparkEnv, SparkException}
 import org.apache.spark.sql.{AnalysisException, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
@@ -1881,10 +1878,10 @@ trait DeltaErrorsBase
     val catalogImplConfig = SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key
     new DeltaAnalysisException(
       errorClass = "DELTA_CONFIGURE_SPARK_SESSION_WITH_EXTENSION_AND_CATALOG",
-      messageParameters = Array(classOf[DeltaSparkSessionExtension].getName,
-        catalogImplConfig, classOf[DeltaCatalog].getName,
-        classOf[DeltaSparkSessionExtension].getName,
-        catalogImplConfig, classOf[DeltaCatalog].getName),
+      messageParameters = Array(classOf[AbstractDeltaSparkSessionExtension].getName,
+        catalogImplConfig, classOf[AbstractDeltaCatalog].getName,
+        classOf[AbstractDeltaSparkSessionExtension].getName,
+        catalogImplConfig, classOf[AbstractDeltaCatalog].getName),
       cause = originalException)
   }
 

spark/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala

@@ -67,7 +67,9 @@ import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
  * A Catalog extension which can properly handle the interaction between the HiveMetaStore and
  * Delta tables. It delegates all operations DataSources other than Delta to the SparkCatalog.
  */
-class DeltaCatalog extends DelegatingCatalogExtension
+class LegacyDeltaCatalog extends AbstractDeltaCatalog
+
+class AbstractDeltaCatalog extends DelegatingCatalogExtension
   with StagingTableCatalog
   with SupportsPathIdentifier
   with DeltaLogging {
@@ -933,7 +935,7 @@ class DeltaCatalog extends DelegatingCatalogExtension
  * A trait for handling table access through delta.`/some/path`. This is a stop-gap solution
  * until PathIdentifiers are implemented in Apache Spark.
  */
-trait SupportsPathIdentifier extends TableCatalog { self: DeltaCatalog =>
+trait SupportsPathIdentifier extends TableCatalog { self: AbstractDeltaCatalog =>
 
   private def supportSQLOnFile: Boolean = spark.sessionState.conf.runSQLonFile
 

spark/src/main/scala/org/apache/spark/sql/delta/implicits/package.scala

@@ -17,12 +17,11 @@
 package org.apache.spark.sql.delta
 
 import org.apache.spark.sql.delta.actions.AddFile
-import org.apache.spark.sql.delta.implicits.RichSparkClasses
 import org.apache.spark.sql.delta.util.DeltaEncoders
 
 import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
 
-package object implicits extends DeltaEncoders with RichSparkClasses {
+package object implicits extends DeltaEncoders with implicits.RichSparkClasses {
   // Define a few implicit classes to provide the `toDF` method. These classes are not using generic
   // types to avoid touching Scala reflection.
   implicit class RichAddFileSeq(files: Seq[AddFile]) {

spark/src/test/scala/org/apache/spark/sql/delta/DeltaErrorsSuite.scala

@@ -29,7 +29,7 @@ import scala.sys.process.Process
 import org.apache.spark.sql.delta.DeltaErrors.generateDocsLink
 import org.apache.spark.sql.delta.actions.{Action, Metadata, Protocol}
 import org.apache.spark.sql.delta.actions.TableFeatureProtocolUtils.{TABLE_FEATURES_MIN_READER_VERSION, TABLE_FEATURES_MIN_WRITER_VERSION}
-import org.apache.spark.sql.delta.catalog.DeltaCatalog
+import org.apache.spark.sql.delta.catalog.AbstractDeltaCatalog
 import org.apache.spark.sql.delta.constraints.CharVarcharConstraint
 import org.apache.spark.sql.delta.constraints.Constraints
 import org.apache.spark.sql.delta.constraints.Constraints.NotNull
@@ -38,7 +38,7 @@ import org.apache.spark.sql.delta.schema.{DeltaInvariantViolationException, Inva
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
 import org.apache.spark.sql.delta.test.DeltaSQLTestUtils
-import io.delta.sql.DeltaSparkSessionExtension
+import io.delta.sql.AbstractDeltaSparkSessionExtension
 import org.apache.hadoop.fs.Path
 import org.json4s.JString
 import org.scalatest.GivenWhenThen
@@ -1977,9 +1977,9 @@ trait DeltaErrorsSuiteBase
       }
 
       checkError(e, "DELTA_CONFIGURE_SPARK_SESSION_WITH_EXTENSION_AND_CATALOG", "56038", Map(
-        "sparkSessionExtensionName" -> classOf[DeltaSparkSessionExtension].getName,
+        "sparkSessionExtensionName" -> classOf[AbstractDeltaSparkSessionExtension].getName,
         "catalogKey" -> SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key,
-        "catalogClassName" -> classOf[DeltaCatalog].getName
+        "catalogClassName" -> classOf[AbstractDeltaCatalog].getName
       ))
     }
     {