code change

Author: Sun Cao

.github/workflows/build-and-test.yml

@@ -1,5 +1,5 @@
 name: Build and Test
-on: [push, pull_request, workflow_dispatch]
+on: [pull_request, workflow_dispatch]
 jobs:
   build-and-test-server:
     runs-on: ubuntu-24.04
@@ -55,6 +55,33 @@ jobs:
           java-version: '17'
       - run: ./build/sbt client/test spark/test
 
+  build-and-test-client-scala-2.12:
+    runs-on: ubuntu-24.04
+    env:
+      SPARK_LOCAL_IP: localhost
+      AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
+      AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+      AZURE_TEST_ACCOUNT_KEY: ${{ secrets.AZURE_TEST_ACCOUNT_KEY }}
+      GOOGLE_APPLICATION_CREDENTIALS: /tmp/google_service_account_key.json
+      GOOGLE_SERVICE_ACCOUNT_KEY: ${{ secrets.GOOGLE_SERVICE_ACCOUNT_KEY }}
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+      - name: Cache Scala, SBT
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.sbt
+            ~/.ivy2
+            ~/.cache/coursier
+          key: build-and-test-client
+      - name: Install Java 8
+        uses: actions/setup-java@v3
+        with:
+          distribution: 'zulu'
+          java-version: '8'
+      - run: ./build/sbt ++2.12.10 client/test
+
   python:
     runs-on: ubuntu-24.04
     strategy:

build.sbt

@@ -18,10 +18,17 @@ import sbt.ExclusionRule
 
 ThisBuild / parallelExecution := false
 
-val sparkVersion = "4.0.0"
+val previousSparkVersion = "3.5.3"
+val latestSparkVersion = "4.0.0"
 val scala212 = "2.12.10"
 val scala213 = "2.13.13"
 
+def sparkVersionFor(scalaVer: String): String = scalaVer match {
+  case v if v.startsWith("2.12") => previousSparkVersion
+  case v if v.startsWith("2.13") => latestSparkVersion
+  case _ => sys.error(s"Unsupported Scala version: $scalaVer")
+}
+
 lazy val commonSettings = Seq(
   organization := "io.delta",
   fork := true,
@@ -66,20 +73,24 @@ lazy val root = (project in file(".")).aggregate(client, spark, server)
 lazy val client = (project in file("client")) settings(
   name := "delta-sharing-client",
   scalaVersion := scala213,
-  crossScalaVersions := Seq(scala213),
+  crossScalaVersions := Seq(scala212, scala213),
   commonSettings,
   java17Settings,
   scalaStyleSettings,
   releaseSettings,
-  libraryDependencies ++= Seq(
-    "org.apache.httpcomponents" % "httpclient" % "4.5.14",
-    "org.apache.spark" %% "spark-sql" % sparkVersion % "provided",
-    "org.apache.spark" %% "spark-catalyst" % sparkVersion % "test" classifier "tests",
-    "org.apache.spark" %% "spark-core" % sparkVersion % "test" classifier "tests",
-    "org.apache.spark" %% "spark-sql" % sparkVersion % "test" classifier "tests",
-    "org.scalatest" %% "scalatest" % "3.2.3" % "test",
-    "org.scalatestplus" %% "mockito-4-11" % "3.2.18.0" % "test"
-  ),
+  libraryDependencies ++= {
+    val sv = scalaVersion.value
+    val sparkVer = sparkVersionFor(sv)
+    Seq(
+      "org.apache.httpcomponents" % "httpclient" % "4.5.14",
+      "org.apache.spark" %% "spark-sql" % sparkVer % "provided",
+      "org.apache.spark" %% "spark-catalyst" % sparkVer % "test" classifier "tests",
+      "org.apache.spark" %% "spark-core" % sparkVer % "test" classifier "tests",
+      "org.apache.spark" %% "spark-sql" % sparkVer % "test" classifier "tests",
+      "org.scalatest" %% "scalatest" % "3.2.3" % "test",
+      "org.scalatestplus" %% "mockito-4-11" % "3.2.18.0" % "test"
+    )
+  },
   Compile / sourceGenerators += Def.task {
     val file = (Compile / sourceManaged).value / "io" / "delta" / "sharing" / "client" / "package.scala"
     IO.write(file,
@@ -90,6 +101,16 @@ lazy val client = (project in file("client")) settings(
          |}
          |""".stripMargin)
     Seq(file)
+  },
+  // Use scala-2.12 and scala-2.13 folders for version-specific overrides
+  Compile / unmanagedSourceDirectories ++= {
+    val sv = scalaVersion.value
+    val base = (Compile / sourceDirectory).value
+    sv match {
+      case v if v.startsWith("2.12") => Seq(base / "scala-2.12")
+      case v if v.startsWith("2.13") => Seq(base / "scala-2.13")
+      case _ => Seq.empty
+    }
   }
 )
 
@@ -102,10 +123,10 @@ lazy val spark = (project in file("spark")) dependsOn(client) settings(
   scalaStyleSettings,
   releaseSettings,
   libraryDependencies ++= Seq(
-    "org.apache.spark" %% "spark-sql" % sparkVersion % "provided",
-    "org.apache.spark" %% "spark-catalyst" % sparkVersion % "test" classifier "tests",
-    "org.apache.spark" %% "spark-core" % sparkVersion % "test" classifier "tests",
-    "org.apache.spark" %% "spark-sql" % sparkVersion % "test" classifier "tests",
+    "org.apache.spark" %% "spark-sql" % latestSparkVersion % "provided",
+    "org.apache.spark" %% "spark-catalyst" % latestSparkVersion % "test" classifier "tests",
+    "org.apache.spark" %% "spark-core" % latestSparkVersion % "test" classifier "tests",
+    "org.apache.spark" %% "spark-sql" % latestSparkVersion % "test" classifier "tests",
     "org.scalatest" %% "scalatest" % "3.2.3" % "test"
   ),
   Compile / sourceGenerators += Def.task {

client/src/main/scala-2.12/io/delta/sharing/spark/perf/DeltaSharingLimitPushDown.scala

@@ -0,0 +1,110 @@
+/*
+ * 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.sharing.spark.perf
+
+import scala.reflect.runtime.universe.{termNames, typeOf, typeTag}
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.catalyst.expressions.IntegerLiteral
+import org.apache.spark.sql.catalyst.plans.logical.{LocalLimit, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation}
+import org.apache.spark.sql.sources.BaseRelation
+
+import io.delta.sharing.client.util.ConfUtils
+import io.delta.sharing.spark.RemoteDeltaSnapshotFileIndex
+
+object DeltaSharingLimitPushDown extends Rule[LogicalPlan] {
+
+  def setup(spark: SparkSession): Unit = synchronized {
+    if (!spark.experimental.extraOptimizations.contains(DeltaSharingLimitPushDown) ) {
+      spark.experimental.extraOptimizations ++= Seq(DeltaSharingLimitPushDown)
+    }
+  }
+
+  def apply(p: LogicalPlan): LogicalPlan = {
+    if (ConfUtils.limitPushdownEnabled(p.conf)) {
+      p transform {
+        case localLimit @ LocalLimit(
+        literalExpr @ IntegerLiteral(limit),
+        l @ LogicalRelationWithTable(
+        r @ HadoopFsRelation(remoteIndex: RemoteDeltaSnapshotFileIndex, _, _, _, _, _),
+        _)
+        ) =>
+          if (remoteIndex.limitHint.isEmpty) {
+            val spark = SparkSession.active
+            LocalLimit(literalExpr,
+              LogicalRelationShim.copyWithNewRelation(
+                l,
+                r.copy(
+                  location = remoteIndex.copy(limitHint = Some(limit)))(spark))
+            )
+          } else {
+            localLimit
+          }
+      }
+    } else {
+      p
+    }
+  }
+}
+
+/**
+ * Extract the [[BaseRelation]] and [[CatalogTable]] from [[LogicalRelation]]. You can also
+ * retrieve the instance of LogicalRelation like following:
+ *
+ * case l @ LogicalRelationWithTable(relation, catalogTable) => ...
+ *
+ * NOTE: This is copied from Spark 4.0 codebase - license: Apache-2.0.
+ */
+object LogicalRelationWithTable {
+  def unapply(plan: LogicalRelation): Option[(BaseRelation, Option[CatalogTable])] = {
+    Some(plan.relation, plan.catalogTable)
+  }
+}
+
+/**
+ * This class helps the codebase to address the differences among multiple Spark versions.
+ */
+object LogicalRelationShim {
+  /**
+   * This method provides the ability of copying LogicalRelation instance across Spark versions,
+   * when the caller only wants to replace the relation in the LogicalRelation.
+   */
+  def copyWithNewRelation(src: LogicalRelation, newRelation: BaseRelation): LogicalRelation = {
+    // We assume Spark would not change the order of the existing parameter, but it's even safe
+    // as long as the first parameter is reserved to the `relation`.
+    val paramsForPrimaryConstructor = src.productIterator.toArray
+    paramsForPrimaryConstructor(0) = newRelation
+
+    val constructor = typeOf[LogicalRelation]
+      .decl(termNames.CONSTRUCTOR)
+      // Getting all the constructors
+      .alternatives
+      .map(_.asMethod)
+      // Picking the primary constructor
+      .find(_.isPrimaryConstructor)
+      // A class must always have a primary constructor, so this is safe
+      .get
+    val constructorMirror = typeTag[LogicalRelation].mirror
+      .reflectClass(typeOf[LogicalRelation].typeSymbol.asClass)
+      .reflectConstructor(constructor)
+
+    constructorMirror.apply(paramsForPrimaryConstructor: _*).asInstanceOf[LogicalRelation]
+  }
+}

client/src/main/scala-2.12/org/apache/spark/sql/DeltaSharingScanUtils.scala

@@ -0,0 +1,34 @@
+/*
+ * 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
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+
+object DeltaSharingScanUtils {
+  // A wrapper to expose Dataset.ofRows function.
+  // This is needed because Dataset object is in private[sql] scope.
+  def ofRows(spark: SparkSession, plan: LogicalRelation): DataFrame = {
+    Dataset.ofRows(spark, plan)
+  }
+
+  // A wrapper to expose Column.apply(expr: Expression) function.
+  // This is needed because the Column object is in private[sql] scope.
+  def toColumn(expr: Expression): Column = {
+    Column(expr)
+  }
+}