From 9f229565fe328487fcfbe2dd3ac422a5b6cb320a Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Mon, 30 Sep 2024 14:36:47 -0700 Subject: [PATCH 1/4] Make ColumnNodeToExpressionConverter ExpressionColumnNode into DeveloperApis instead of internal so that external advanced developers can write columns with codegen that take in columns for input. --- .../org/apache/spark/sql/internal/columnNodeSupport.scala | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala index 8b4726114890f..c61e3dca4f23a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala @@ -19,6 +19,7 @@ package org.apache.spark.sql.internal import UserDefinedFunctionUtils.toScalaUDF import org.apache.spark.SparkException +import org.apache.spark.annotation.DeveloperApi import org.apache.spark.sql.{Column, Dataset, SparkSession} import org.apache.spark.sql.catalyst.{analysis, expressions, CatalystTypeConverters} import org.apache.spark.sql.catalyst.analysis.{MultiAlias, UnresolvedAlias} @@ -248,7 +249,8 @@ private[sql] trait ColumnNodeToExpressionConverter extends (ColumnNode => Expres } } -private[sql] object ColumnNodeToExpressionConverter extends ColumnNodeToExpressionConverter { +@DeveloperApi +object ColumnNodeToExpressionConverter extends ColumnNodeToExpressionConverter { override protected def parser: ParserInterface = { SparkSession.getActiveSession.map(_.sessionState.sqlParser).getOrElse { new SparkSqlParser() @@ -261,7 +263,8 @@ private[sql] object ColumnNodeToExpressionConverter extends ColumnNodeToExpressi /** * [[ColumnNode]] wrapper for an [[Expression]]. */ -private[sql] case class ExpressionColumnNode private( +@DeveloperApi +case class ExpressionColumnNode private( expression: Expression, override val origin: Origin = CurrentOrigin.get) extends ColumnNode { override def normalize(): ExpressionColumnNode = { From df49d0d0b204d90daa03a07f74d7bf61b9186109 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Thu, 5 Dec 2024 14:34:44 -0800 Subject: [PATCH 2/4] Move columnNodeSupport --- .../apache/spark/sql/{internal => util}/columnNodeSupport.scala | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename sql/core/src/main/scala/org/apache/spark/sql/{internal => util}/columnNodeSupport.scala (100%) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala similarity index 100% rename from sql/core/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala rename to sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala From fab35676e2ceba23cd546cc5e796333a503ac430 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Thu, 5 Dec 2024 14:40:15 -0800 Subject: [PATCH 3/4] Finish moving columnNodeSupport and friends to utils --- .../main/scala/org/apache/spark/sql/classic/conversions.scala | 3 ++- .../scala/org/apache/spark/sql/util/columnNodeSupport.scala | 2 +- .../test/scala/org/apache/spark/sql/test/SQLTestUtils.scala | 3 ++- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala index e90fd4b6a6032..da1d7306ca7ef 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala @@ -21,7 +21,8 @@ import scala.language.implicitConversions import org.apache.spark.annotation.DeveloperApi import org.apache.spark.sql._ import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression} -import org.apache.spark.sql.internal.{ColumnNodeToExpressionConverter, ExpressionUtils} +import org.apache.spark.sql.internal.ExpressionUtils +import org.apache.spark.sql.utils.ColumnNodeToExpressionConverter /** * Conversions from sql interfaces to the Classic specific implementation. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala index c61e3dca4f23a..922757969655c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.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.util import UserDefinedFunctionUtils.toScalaUDF diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala index c93f17701c620..ec5aab71c4054 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala @@ -45,7 +45,8 @@ import org.apache.spark.sql.classic.{ClassicConversions, ColumnConversions} import org.apache.spark.sql.execution.FilterExec import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecution import org.apache.spark.sql.execution.datasources.DataSourceUtils -import org.apache.spark.sql.internal.{ColumnNodeToExpressionConverter, SQLConf} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.utils.ColumnNodeToExpressionConverter import org.apache.spark.util.ArrayImplicits._ import org.apache.spark.util.UninterruptibleThread import org.apache.spark.util.Utils From 54e990dfe81f380ea4446ac7978785f5f03ab5af Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Thu, 5 Dec 2024 14:59:51 -0800 Subject: [PATCH 4/4] Finish reorg from internal to utils. --- .../main/scala/org/apache/spark/ml/stat/Summarizer.scala | 2 +- .../scala/org/apache/spark/sql/internal/columnNodes.scala | 2 +- sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala | 4 ++-- .../org/apache/spark/sql/RelationalGroupedDataset.scala | 2 +- .../src/main/scala/org/apache/spark/sql/SparkSession.scala | 2 +- .../org/apache/spark/sql/api/python/PythonSQLUtils.scala | 2 +- .../scala/org/apache/spark/sql/classic/conversions.scala | 3 +-- .../sql/execution/python/UserDefinedPythonFunction.scala | 2 +- .../scala/org/apache/spark/sql/util/columnNodeSupport.scala | 6 +++--- 9 files changed, 12 insertions(+), 13 deletions(-) diff --git a/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala b/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala index e67b72e090601..1a57e173e8701 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala @@ -31,8 +31,8 @@ import org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggreg import org.apache.spark.sql.catalyst.trees.BinaryLike import org.apache.spark.sql.classic.ClassicConversions._ import org.apache.spark.sql.functions.lit -import org.apache.spark.sql.internal.ExpressionUtils.expression import org.apache.spark.sql.types._ +import org.apache.spark.sql.util.ExpressionUtils.expression import org.apache.spark.util.Utils /** diff --git a/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala b/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala index f745c152170ee..d8a8dd0d92e04 100644 --- a/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala +++ b/sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala @@ -72,7 +72,7 @@ trait ColumnNodeLike { private[internal] def sql: String } -private[internal] object ColumnNode { +private[sql] object ColumnNode { val NO_ORIGIN: Origin = Origin() def normalize[T <: ColumnNodeLike](option: Option[T]): Option[T] = option.map(_.normalize().asInstanceOf[T]) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala index 846d97b257869..61393bb24a7ac 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala @@ -62,11 +62,11 @@ import org.apache.spark.sql.execution.datasources.LogicalRelationWithTable import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, FileTable} import org.apache.spark.sql.execution.python.EvaluatePython import org.apache.spark.sql.execution.stat.StatFunctions -import org.apache.spark.sql.internal.{DataFrameWriterImpl, DataFrameWriterV2Impl, ExpressionColumnNode, MergeIntoWriterImpl, SQLConf} +import org.apache.spark.sql.internal.{DataFrameWriterImpl, DataFrameWriterV2Impl, MergeIntoWriterImpl, SQLConf} import org.apache.spark.sql.internal.TypedAggUtils.withInputType import org.apache.spark.sql.streaming.DataStreamWriter import org.apache.spark.sql.types._ -import org.apache.spark.sql.util.SchemaUtils +import org.apache.spark.sql.util.{ExpressionColumnNode, SchemaUtils} import org.apache.spark.storage.StorageLevel import org.apache.spark.unsafe.array.ByteArrayMethods import org.apache.spark.util.ArrayImplicits._ diff --git a/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala index b8c4b03fc13d2..0d0a7f85865d5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala @@ -31,10 +31,10 @@ import org.apache.spark.sql.catalyst.util.toPrettySQL import org.apache.spark.sql.classic.ClassicConversions._ import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors} import org.apache.spark.sql.execution.QueryExecution -import org.apache.spark.sql.internal.ExpressionUtils.generateAlias import org.apache.spark.sql.internal.TypedAggUtils.withInputType import org.apache.spark.sql.streaming.OutputMode import org.apache.spark.sql.types.{NumericType, StructType} +import org.apache.spark.sql.util.ExpressionUtils.generateAlias import org.apache.spark.util.ArrayImplicits._ /** diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala index 878fdc8e267a5..ca1fdfaf907ce 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala @@ -60,7 +60,7 @@ import org.apache.spark.sql.scripting.SqlScriptingExecution import org.apache.spark.sql.sources.BaseRelation import org.apache.spark.sql.streaming._ import org.apache.spark.sql.types.{DataType, StructType} -import org.apache.spark.sql.util.ExecutionListenerManager +import org.apache.spark.sql.util._ import org.apache.spark.util.{CallSite, SparkFileUtils, ThreadUtils, Utils} import org.apache.spark.util.ArrayImplicits._ diff --git a/sql/core/src/main/scala/org/apache/spark/sql/api/python/PythonSQLUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/api/python/PythonSQLUtils.scala index da03293ce743b..99204093aeb77 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/api/python/PythonSQLUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/api/python/PythonSQLUtils.scala @@ -37,9 +37,9 @@ import org.apache.spark.sql.classic.ClassicConversions._ import org.apache.spark.sql.execution.{ExplainMode, QueryExecution} import org.apache.spark.sql.execution.arrow.ArrowConverters import org.apache.spark.sql.execution.python.EvaluatePython -import org.apache.spark.sql.internal.ExpressionUtils.expression import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{DataType, StructType} +import org.apache.spark.sql.util.ExpressionUtils.expression import org.apache.spark.util.{MutableURLClassLoader, Utils} private[sql] object PythonSQLUtils extends Logging { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala index da1d7306ca7ef..31bbab15599d2 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala @@ -21,8 +21,7 @@ import scala.language.implicitConversions import org.apache.spark.annotation.DeveloperApi import org.apache.spark.sql._ import org.apache.spark.sql.catalyst.expressions.{Expression, NamedExpression} -import org.apache.spark.sql.internal.ExpressionUtils -import org.apache.spark.sql.utils.ColumnNodeToExpressionConverter +import org.apache.spark.sql.util.{ColumnNodeToExpressionConverter, ExpressionUtils} /** * Conversions from sql interfaces to the Classic specific implementation. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala index 388ede5d062e5..1e62ec04906a6 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala @@ -30,8 +30,8 @@ import org.apache.spark.sql.catalyst.parser.ParserInterface import org.apache.spark.sql.catalyst.plans.logical.{Generate, LogicalPlan, NamedParametersSupport, OneRowRelation} import org.apache.spark.sql.classic.ClassicConversions._ import org.apache.spark.sql.errors.QueryCompilationErrors -import org.apache.spark.sql.internal.ExpressionUtils.expression import org.apache.spark.sql.types.{DataType, StructType} +import org.apache.spark.sql.util.ExpressionUtils.expression /** * A user-defined Python function. This is used by the Python API. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala index 922757969655c..0e5f1b70a2dca 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/util/columnNodeSupport.scala @@ -16,8 +16,6 @@ */ package org.apache.spark.sql.util -import UserDefinedFunctionUtils.toScalaUDF - import org.apache.spark.SparkException import org.apache.spark.annotation.DeveloperApi import org.apache.spark.sql.{Column, Dataset, SparkSession} @@ -33,6 +31,8 @@ import org.apache.spark.sql.execution.SparkSqlParser import org.apache.spark.sql.execution.aggregate.{ScalaAggregator, ScalaUDAF, TypedAggregateExpression} import org.apache.spark.sql.execution.analysis.DetectAmbiguousSelfJoin import org.apache.spark.sql.expressions.{Aggregator, SparkUserDefinedFunction, UserDefinedAggregateFunction, UserDefinedAggregator} +import org.apache.spark.sql.internal._ +import org.apache.spark.sql.internal.UserDefinedFunctionUtils.toScalaUDF import org.apache.spark.sql.types.{DataType, NullType} /** @@ -285,7 +285,7 @@ private[sql] object ExpressionColumnNode { } } -private[internal] case class ColumnNodeExpression private(node: ColumnNode) extends Unevaluable { +private[sql] case class ColumnNodeExpression private(node: ColumnNode) extends Unevaluable { override def nullable: Boolean = true override def dataType: DataType = NullType override def children: Seq[Expression] = Nil