[SPARK-49700][CONNECT][SQL] Unified Scala Interface for Connect and C…

…lassic ### What changes were proposed in this pull request? This PR makes the shared SQL (JVM) interface the primary interface for Scala/JVM based Dataframe programming. The implementations are moved to the `classic` and `connect` sub packages. The connect client had to be moved to the sql/connect/common package because serialization requires the captured client classes to be on the classpath when deserialized on the server. ### Why are the changes needed? This is the final step in creating a unified Scala interface for both Classic and Connect. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#48818 from hvanhovell/SPARK-49700. Authored-by: Herman van Hovell <[email protected]> Signed-off-by: Herman van Hovell <[email protected]>
bersprockets · Jan 24, 2025 · 5db31ae · 5db31ae
1 parent a03c4cb
commit 5db31ae
Show file tree

Hide file tree

Showing 399 changed files with 1,851 additions and 1,730 deletions.
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
@@ -5277,6 +5277,11 @@
           "Resilient Distributed Datasets (RDDs)."
         ]
       },
+      "REGISTER_UDAF" : {
+        "message" : [
+          "Registering User Defined Aggregate Functions (UDAFs)."
+        ]
+      },
       "SESSION_BASE_RELATION_TO_DATAFRAME" : {
         "message" : [
           "Invoking SparkSession 'baseRelationToDataFrame'. This is server side developer API"

diff --git a/...ctor/connect/client/jvm/src/main/scala/org/apache/spark/sql/application/ConnectRepl.scala b/...ctor/connect/client/jvm/src/main/scala/org/apache/spark/sql/application/ConnectRepl.scala
@@ -31,8 +31,8 @@ import ammonite.util.Util.newLine
 
 import org.apache.spark.SparkBuildInfo.spark_version
 import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.SparkSession.withLocalConnectServer
+import org.apache.spark.sql.connect.SparkSession
+import org.apache.spark.sql.connect.SparkSession.withLocalConnectServer
 import org.apache.spark.sql.connect.client.{SparkConnectClient, SparkConnectClientParser}
 
 /**

diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala
diff --git a/connector/connect/client/jvm/src/test/java/org/apache/spark/sql/JavaEncoderSuite.java b/connector/connect/client/jvm/src/test/java/org/apache/spark/sql/JavaEncoderSuite.java
@@ -28,7 +28,7 @@
 import static org.apache.spark.sql.functions.*;
 import static org.apache.spark.sql.RowFactory.create;
 import org.apache.spark.api.java.function.MapFunction;
-import org.apache.spark.sql.test.SparkConnectServerUtils;
+import org.apache.spark.sql.connect.test.SparkConnectServerUtils;
 import org.apache.spark.sql.types.StructType;
 
 /**

diff --git a/...ector/connect/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala b/...ector/connect/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameSubquerySuite.scala
@@ -18,8 +18,8 @@
 package org.apache.spark.sql
 
 import org.apache.spark.SparkRuntimeException
+import org.apache.spark.sql.connect.test.{QueryTest, RemoteSparkSession}
 import org.apache.spark.sql.functions._
-import org.apache.spark.sql.test.{QueryTest, RemoteSparkSession}
 
 class DataFrameSubquerySuite extends QueryTest with RemoteSparkSession {
   import testImplicits._

diff --git a/...t/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameTableValuedFunctionsSuite.scala b/...t/client/jvm/src/test/scala/org/apache/spark/sql/DataFrameTableValuedFunctionsSuite.scala
@@ -17,8 +17,8 @@
 
 package org.apache.spark.sql
 
+import org.apache.spark.sql.connect.test.{QueryTest, RemoteSparkSession}
 import org.apache.spark.sql.functions._
-import org.apache.spark.sql.test.{QueryTest, RemoteSparkSession}
 
 class DataFrameTableValuedFunctionsSuite extends QueryTest with RemoteSparkSession {
   import testImplicits._

diff --git a/...ctor/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala b/...ctor/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
@@ -38,12 +38,13 @@ import org.apache.spark.sql.avro.{functions => avroFn}
 import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.StringEncoder
 import org.apache.spark.sql.catalyst.util.CollationFactory
+import org.apache.spark.sql.connect.{DataFrame, Dataset, SparkSession}
 import org.apache.spark.sql.connect.ConnectConversions._
 import org.apache.spark.sql.connect.client.SparkConnectClient
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, IntegrationTestUtils}
 import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions.lit
 import org.apache.spark.sql.protobuf.{functions => pbFn}
-import org.apache.spark.sql.test.{ConnectFunSuite, IntegrationTestUtils}
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.CalendarInterval
 import org.apache.spark.util.SparkFileUtils

diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/SQLExpressionsSuite.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/SQLExpressionsSuite.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql
 
-import org.apache.spark.sql.test.{QueryTest, RemoteSparkSession}
+import org.apache.spark.sql.connect.test.{QueryTest, RemoteSparkSession}
 import org.apache.spark.unsafe.types.VariantVal
 
 class SQLExpressionsSuite extends QueryTest with RemoteSparkSession {

diff --git a/...tor/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala b/...tor/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala
@@ -25,7 +25,7 @@ import scala.util.Properties
 import org.apache.commons.io.output.ByteArrayOutputStream
 import org.scalatest.BeforeAndAfterEach
 
-import org.apache.spark.sql.test.{ConnectFunSuite, IntegrationTestUtils, RemoteSparkSession}
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, IntegrationTestUtils, RemoteSparkSession}
 import org.apache.spark.tags.AmmoniteTest
 import org.apache.spark.util.IvyTestUtils
 import org.apache.spark.util.MavenUtils.MavenCoordinate

diff --git a/...a/org/apache/spark/sql/CatalogSuite.scala → ...ache/spark/sql/connect/CatalogSuite.scala b/...a/org/apache/spark/sql/CatalogSuite.scala → ...ache/spark/sql/connect/CatalogSuite.scala
@@ -15,14 +15,15 @@
  * limitations under the License.
  */
 
-package org.apache.spark.sql
+package org.apache.spark.sql.connect
 
 import java.io.{File, FilenameFilter}
 
 import org.apache.commons.io.FileUtils
 
 import org.apache.spark.SparkException
-import org.apache.spark.sql.test.{ConnectFunSuite, RemoteSparkSession, SQLHelper}
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, RemoteSparkSession, SQLHelper}
 import org.apache.spark.sql.types.{DoubleType, LongType, StructType}
 import org.apache.spark.storage.StorageLevel
 

diff --git a/...rg/apache/spark/sql/CheckpointSuite.scala → ...e/spark/sql/connect/CheckpointSuite.scala b/...rg/apache/spark/sql/CheckpointSuite.scala → ...e/spark/sql/connect/CheckpointSuite.scala
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.spark.sql
+package org.apache.spark.sql.connect
 
 import java.io.{ByteArrayOutputStream, PrintStream}
 
@@ -26,7 +26,7 @@ import org.scalatest.exceptions.TestFailedDueToTimeoutException
 
 import org.apache.spark.SparkException
 import org.apache.spark.connect.proto
-import org.apache.spark.sql.test.{ConnectFunSuite, RemoteSparkSession, SQLHelper}
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, RemoteSparkSession, SQLHelper}
 import org.apache.spark.storage.StorageLevel
 
 class CheckpointSuite extends ConnectFunSuite with RemoteSparkSession with SQLHelper {

diff --git a/.../spark/sql/ClientDataFrameStatSuite.scala → ...ql/connect/ClientDataFrameStatSuite.scala b/.../spark/sql/ClientDataFrameStatSuite.scala → ...ql/connect/ClientDataFrameStatSuite.scala
@@ -15,14 +15,15 @@
  * limitations under the License.
  */
 
-package org.apache.spark.sql
+package org.apache.spark.sql.connect
 
 import java.util.Random
 
 import org.scalatest.matchers.must.Matchers._
 
 import org.apache.spark.SparkIllegalArgumentException
-import org.apache.spark.sql.test.{ConnectFunSuite, RemoteSparkSession}
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, RemoteSparkSession}
 
 class ClientDataFrameStatSuite extends ConnectFunSuite with RemoteSparkSession {
   private def toLetter(i: Int): String = (i + 97).toChar.toString

diff --git a/...apache/spark/sql/ClientDatasetSuite.scala → ...park/sql/connect/ClientDatasetSuite.scala b/...apache/spark/sql/ClientDatasetSuite.scala → ...park/sql/connect/ClientDatasetSuite.scala
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.spark.sql
+package org.apache.spark.sql.connect
 
 import java.util.Properties
 import java.util.concurrent.TimeUnit
@@ -25,9 +25,10 @@ import io.grpc.inprocess.{InProcessChannelBuilder, InProcessServerBuilder}
 import org.scalatest.BeforeAndAfterEach
 
 import org.apache.spark.connect.proto
+import org.apache.spark.sql.Column
 import org.apache.spark.sql.connect.client.{DummySparkConnectService, SparkConnectClient}
+import org.apache.spark.sql.connect.test.ConnectFunSuite
 import org.apache.spark.sql.functions._
-import org.apache.spark.sql.test.ConnectFunSuite
 import org.apache.spark.util.SparkSerDeUtils
 
 // Add sample tests.

diff --git a/...apache/spark/sql/ClientE2ETestSuite.scala → ...park/sql/connect/ClientE2ETestSuite.scala b/...apache/spark/sql/ClientE2ETestSuite.scala → ...park/sql/connect/ClientE2ETestSuite.scala
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.spark.sql
+package org.apache.spark.sql.connect
 
 import java.io.{ByteArrayOutputStream, PrintStream}
 import java.nio.file.Files
@@ -34,15 +34,17 @@ import org.scalatest.PrivateMethodTester
 import org.apache.spark.{SparkArithmeticException, SparkException, SparkUpgradeException}
 import org.apache.spark.SparkBuildInfo.{spark_version => SPARK_VERSION}
 import org.apache.spark.internal.config.ConfigBuilder
+import org.apache.spark.sql.{functions, AnalysisException, Observation, Row, SaveMode}
 import org.apache.spark.sql.catalyst.analysis.{NamespaceAlreadyExistsException, NoSuchNamespaceException, TableAlreadyExistsException, TempTableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.StringEncoder
 import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
 import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.connect.ConnectConversions._
 import org.apache.spark.sql.connect.client.{RetryPolicy, SparkConnectClient, SparkResult}
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, IntegrationTestUtils, RemoteSparkSession, SQLHelper}
+import org.apache.spark.sql.connect.test.SparkConnectServerUtils.port
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SqlApiConf
-import org.apache.spark.sql.test.{ConnectFunSuite, IntegrationTestUtils, RemoteSparkSession, SQLHelper}
-import org.apache.spark.sql.test.SparkConnectServerUtils.port
 import org.apache.spark.sql.types._
 import org.apache.spark.util.SparkThreadUtils
 

diff --git a/...nal/ColumnNodeToProtoConverterSuite.scala → ...ect/ColumnNodeToProtoConverterSuite.scala b/...nal/ColumnNodeToProtoConverterSuite.scala → ...ect/ColumnNodeToProtoConverterSuite.scala
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.spark.sql.internal
+package org.apache.spark.sql.connect
 
 import org.apache.spark.SparkException
 import org.apache.spark.connect.proto
@@ -23,9 +23,10 @@ import org.apache.spark.sql.{Column, Encoder}
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{PrimitiveIntEncoder, PrimitiveLongEncoder}
 import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
 import org.apache.spark.sql.connect.common.{DataTypeProtoConverter, ProtoDataTypes}
+import org.apache.spark.sql.connect.test.ConnectFunSuite
 import org.apache.spark.sql.expressions.{Aggregator, SparkUserDefinedFunction, UserDefinedAggregator}
-import org.apache.spark.sql.test.ConnectFunSuite
-import org.apache.spark.sql.types.{BinaryType, DataType, DoubleType, LongType, MetadataBuilder, ShortType, StringType, StructType}
+import org.apache.spark.sql.internal._
+import org.apache.spark.sql.types._
 
 /**
  * Test suite for [[ColumnNode]] to [[proto.Expression]] conversions.
@@ -471,8 +472,8 @@ class ColumnNodeToProtoConverterSuite extends ConnectFunSuite {
   }
 }
 
-private[internal] case class Nope(override val origin: Origin = CurrentOrigin.get)
+private[connect] case class Nope(override val origin: Origin = CurrentOrigin.get)
     extends ColumnNode {
   override def sql: String = "nope"
-  override private[internal] def children: Seq[ColumnNodeLike] = Seq.empty
+  override def children: Seq[ColumnNodeLike] = Seq.empty
 }