Skip to content

[SPARK-49828][SQL] Expose ColumnNode to/from Expression utils as developer APIs #48306

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
holdenk wants to merge 4 commits into from
+20 −16
Closed

[SPARK-49828][SQL] Expose ColumnNode to/from Expression utils as developer APIs #48306

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9f22956
Make ColumnNodeToExpressionConverter ExpressionColumnNode into Develo…
holdenk Sep 30, 2024
df49d0d
Move columnNodeSupport
holdenk Dec 5, 2024
fab3567
Finish moving columnNodeSupport and friends to utils
holdenk Dec 5, 2024
54e990d
Finish reorg from internal to utils.
holdenk Dec 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

/**
Expand Down
2 changes: 1 addition & 1 deletion sql/api/src/main/scala/org/apache/spark/sql/internal/columnNodes.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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])
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._

/**
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._

Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/api/python/PythonSQLUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/classic/conversions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +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.{ColumnNodeToExpressionConverter, ExpressionUtils}
import org.apache.spark.sql.util.{ColumnNodeToExpressionConverter, ExpressionUtils}

/**
* Conversions from sql interfaces to the Classic specific implementation.
Expand Down
2 changes: 1 addition & 1 deletion ...core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
15 changes: 9 additions & 6 deletions ...park/sql/internal/columnNodeSupport.scala → ...he/spark/sql/util/columnNodeSupport.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.spark.sql.internal

import UserDefinedFunctionUtils.toScalaUDF
package org.apache.spark.sql.util

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}
Expand All @@ -32,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}

/**
Expand Down Expand Up @@ -248,7 +249,8 @@ private[sql] trait ColumnNodeToExpressionConverter extends (ColumnNode => Expres
}
}

private[sql] object ColumnNodeToExpressionConverter extends ColumnNodeToExpressionConverter {
@DeveloperApi
object ColumnNodeToExpressionConverter extends ColumnNodeToExpressionConverter {

@hvanhovell hvanhovell Sep 30, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Quick question, we have an implicit class in SparkSession called RichColumn. That will 'restore' the expr functionality. Is that something that would work for you? Alternatively, if the use of SparkSession is cumbersome, we could also add another implicit class to org.apache.spark.sql.classic.ClassicConversions.

@holdenk holdenk Sep 30, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So for Java users?

@hvanhovell hvanhovell Sep 30, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, I see your point. I would love to meet the Java developer that works with Spark internals :)...

They can still use these objects.

override protected def parser: ParserInterface = {
SparkSession.getActiveSession.map(_.sessionState.sqlParser).getOrElse {
new SparkSqlParser()
Expand All @@ -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(

@hvanhovell hvanhovell Sep 30, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You could use org.apache.spark.sql.classic.ClassicConversions for this. I am not saying that you should, some people don't like implicits.

@holdenk holdenk Sep 30, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So it doesn't do both the directions of the conversion and, as you point out, it's also an implicit conversion which (in my opinion) is some pretty unnecessary magic for this.

@hvanhovell hvanhovell Sep 30, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In this case implicits provide a way for a developer to make a minimal amount of changes, that is why they are there. I am generally not a big fan, but in this case they do make migration a lot easier.

You could call org.apache.spark.sql.classic.ClassicConversions.column directly if you want to avoid implicits.

expression: Expression,
override val origin: Origin = CurrentOrigin.get) extends ColumnNode {
override def normalize(): ExpressionColumnNode = {
Expand All @@ -282,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
Expand Down
3 changes: 2 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down