Skip to content

[SPARK-44284][CONNECT] Create simple conf system for sql/api #41838

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
hvanhovell wants to merge 5 commits into from
Closed

[SPARK-44284][CONNECT] Create simple conf system for sql/api #41838

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b1fc1f4
Create simple conf system for sql/api
hvanhovell Jul 3, 2023
75ab6d3
Executor side behavior.
hvanhovell Jul 3, 2023
2ab2aa4
Order paranoia
hvanhovell Jul 3, 2023
a63b07c
Update sql/api/src/main/scala/org/apache/spark/sql/SqlApiConf.scala
hvanhovell Jul 5, 2023
37899d3
Update sql/api/src/main/scala/org/apache/spark/sql/SqlApiConf.scala
hvanhovell Jul 5, 2023
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
62 changes: 62 additions & 0 deletions sql/api/src/main/scala/org/apache/spark/sql/SqlApiConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.util.SparkClassUtils

import java.util.concurrent.atomic.AtomicReference
import scala.util.Try

/**
* Configuration for all objects that are placed in the `sql/api` project. The normal way of
* accessing this class is through `SqlApiConf.get`. If this code is being used with sql/core
* then it values are bound to the currently set SQLConf. In connect mode it will default to
* hardcoded values.
*/
private[sql] trait SqlApiConf {
Copy link
Member

@HyukjinKwon HyukjinKwon Jul 4, 2023

Choose a reason for hiding this comment

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

no biggie and just dropping a comment. I wonder if we should name it like just SQLConfiguration or just SQLConf (since org.apache.spark.sql.internal.SQLConf at sql/core isn't API anyway)

Copy link
Contributor

@amaliujia amaliujia Jul 4, 2023

Choose a reason for hiding this comment

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

Probably not name as SQLConf, as that will triggers a huge amount of refactorings.

Copy link
Member

@pan3793 pan3793 Jul 5, 2023

Choose a reason for hiding this comment

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

Is org.apache.spark.SparkConf a public API? How about naming it SparkSQLConf?

Copy link
Member

@HyukjinKwon HyukjinKwon Jul 5, 2023

Choose a reason for hiding this comment

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

SparkSQLConf SGTM

Copy link
Contributor

@LuciferYang LuciferYang Jul 5, 2023

Choose a reason for hiding this comment

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

SparkSQLConf SGTM

+1

Copy link
Contributor Author

@hvanhovell hvanhovell Jul 5, 2023

Choose a reason for hiding this comment

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

It really should be an internal thing. It is marked private[sql] for this reason. Users should only interact with RuntimeConf, and perhaps SparkSession.Builder to get/set options.

Copy link
Member

@HyukjinKwon HyukjinKwon Jul 5, 2023

Choose a reason for hiding this comment

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

Yeah, if it's internal, I am good with whatever name :-)

def ansiEnabled: Boolean
def caseSensitiveAnalysis: Boolean
def maxToStringFields: Int
}

private[sql] object SqlApiConf {
/**
* Defines a getter that returns the [[SqlApiConf]] within scope.
*/
private val confGetter = new AtomicReference[() => SqlApiConf](() => DefaultSqlApiConf)

/**
* Sets the active config getter.
*/
private[sql] def setConfGetter(getter: () => SqlApiConf): Unit = {
confGetter.set(getter)
}

def get: SqlApiConf = confGetter.get()()

// Force load SQLConf. This will trigger the installation of a confGetter that points to SQLConf.
Try(SparkClassUtils.classForName("org.apache.spark.sql.internal.SQLConf$"))
Copy link
Member

@turboFei turboFei Dec 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Hi @hvanhovell We meet dead lock issue in a iceberg compaction job and seems related with this patch.

I have raised ticket https://issues.apache.org/jira/browse/SPARK-50620 and provide the thread dump: https://issues.apache.org/jira/secure/attachment/13073575/iceberg_compaction.txt

"Executor 92 task launch worker for task 2910, task 69.0 in stage 7.0 (TID 2910)" #152 daemon prio=5 os_prio=0 cpu=616.25ms elapsed=258408.34s tid=0x00007f77d67cc330 nid=0x22c9e in Object.wait()  [0x00007f77755fb000]
   java.lang.Thread.State: RUNNABLE
	at org.apache.spark.sql.internal.SQLConf$.<init>(SQLConf.scala:184)
	- waiting on the Class initialization monitor for org.apache.spark.sql.internal.SqlApiConf$
	at org.apache.spark.sql.internal.SQLConf$.<clinit>(SQLConf.scala)

"Executor 92 task launch worker for task 5362, task 521.0 in stage 10.0 (TID 5362)" #123 daemon prio=5 os_prio=0 cpu=2443.78ms elapsed=258409.29s tid=0x00007f77d60ecad0 nid=0x22c7c in Object.wait()  [0x00007f777e591000]
   java.lang.Thread.State: RUNNABLE
	at java.lang.Class.forName0([email protected]/Native Method)
	- waiting on the Class initialization monitor for org.apache.spark.sql.internal.SQLConf$
	at java.lang.Class.forName([email protected]/Class.java:467)
	at org.apache.spark.util.SparkClassUtils.classForName(SparkClassUtils.scala:41)
	at org.apache.spark.util.SparkClassUtils.classForName$(SparkClassUtils.scala:36)
	at org.apache.spark.util.SparkClassUtils$.classForName(SparkClassUtils.scala:83)
	at org.apache.spark.sql.internal.SqlApiConf$.$anonfun$new$1(SqlApiConf.scala:73)

For the first task: at org.apache.spark.sql.internal.SQLConf$.<clinit>(SQLConf.scala) - waiting on the Class initialization monitor for org.apache.spark.sql.internal.SqlApiConf$.

For the second one: at org.apache.spark.sql.internal.SqlApiConf$.$anonfun$new$1(SqlApiConf.scala:73) - waiting on the Class initialization monitor for org.apache.spark.sql.internal.SQLConf$

Copy link
Member

@turboFei turboFei Dec 19, 2024

Choose a reason for hiding this comment

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

resolved by #44602, please ignore

}

/**
* Defaults configurations used when no other [[SqlApiConf]] getter is set.
*/
private[sql] object DefaultSqlApiConf extends SqlApiConf {
override def ansiEnabled: Boolean = false
override def caseSensitiveAnalysis: Boolean = false
override def maxToStringFields: Int = 50
}
4 changes: 2 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/DataTypeUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,9 @@
*/
package org.apache.spark.sql.catalyst.types

import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, Literal}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy.{ANSI, STRICT}
import org.apache.spark.sql.types.{ArrayType, AtomicType, DataType, Decimal, DecimalType, MapType, NullType, StructField, StructType}
Expand All @@ -30,7 +30,7 @@ object DataTypeUtils {
* (`StructField.nullable`, `ArrayType.containsNull`, and `MapType.valueContainsNull`).
*/
def sameType(left: DataType, right: DataType): Boolean =
Copy link
Contributor Author

@hvanhovell hvanhovell Jul 3, 2023

Choose a reason for hiding this comment

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

We should move this back to DataType.

Copy link
Contributor

@amaliujia amaliujia Jul 4, 2023

Choose a reason for hiding this comment

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

sure we should do it.

if (SQLConf.get.caseSensitiveAnalysis) {
if (SqlApiConf.get.caseSensitiveAnalysis) {
equalsIgnoreNullability(left, right)
} else {
equalsIgnoreCaseAndNullability(left, right)
Expand Down
14 changes: 10 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.{IGNORE_MISSING_FILES => SPARK_IGNORE_MISSING_FILES}
import org.apache.spark.network.util.ByteUnit
import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.catalyst.analysis.{HintErrorLogger, Resolver}
import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode
Expand Down Expand Up @@ -176,9 +177,14 @@ object SQLConf {
* See [[get]] for more information.
*/
def setSQLConfGetter(getter: () => SQLConf): Unit = {
SqlApiConf.setConfGetter(getter)
Copy link
Contributor Author

@hvanhovell hvanhovell Jul 3, 2023

Choose a reason for hiding this comment

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

This is just a bit more efficient.

confGetter.set(getter)
}

// Make sure SqlApiConf is always in sync with SQLConf. SqlApiConf will always try to
// load SqlConf to make sure both classes are in sync from the get go.
SqlApiConf.setConfGetter(() => SQLConf.get)

/**
* Returns the active config object within the current scope. If there is an active SparkSession,
* the proper SQLConf associated with the thread's active session is used. If it's called from
Expand Down Expand Up @@ -4439,7 +4445,7 @@ object SQLConf {
*
* SQLConf is thread-safe (internally synchronized, so safe to be used in multiple threads).
*/
class SQLConf extends Serializable with Logging {
class SQLConf extends Serializable with Logging with SqlApiConf {
import SQLConf._

/** Only low degree of contention is expected for conf, thus NOT using ConcurrentHashMap. */
Expand Down Expand Up @@ -4685,7 +4691,7 @@ class SQLConf extends Serializable with Logging {

def subqueryReuseEnabled: Boolean = getConf(SUBQUERY_REUSE_ENABLED)

def caseSensitiveAnalysis: Boolean = getConf(SQLConf.CASE_SENSITIVE)
override def caseSensitiveAnalysis: Boolean = getConf(SQLConf.CASE_SENSITIVE)

def constraintPropagationEnabled: Boolean = getConf(CONSTRAINT_PROPAGATION_ENABLED)

Expand Down Expand Up @@ -5001,7 +5007,7 @@ class SQLConf extends Serializable with Logging {
def storeAssignmentPolicy: StoreAssignmentPolicy.Value =
StoreAssignmentPolicy.withName(getConf(STORE_ASSIGNMENT_POLICY))

def ansiEnabled: Boolean = getConf(ANSI_ENABLED)
override def ansiEnabled: Boolean = getConf(ANSI_ENABLED)

def enableDefaultColumns: Boolean = getConf(SQLConf.ENABLE_DEFAULT_COLUMNS)

Expand Down Expand Up @@ -5071,7 +5077,7 @@ class SQLConf extends Serializable with Logging {
def nameNonStructGroupingKeyAsValue: Boolean =
getConf(SQLConf.NAME_NON_STRUCT_GROUPING_KEY_AS_VALUE)

def maxToStringFields: Int = getConf(SQLConf.MAX_TO_STRING_FIELDS)
override def maxToStringFields: Int = getConf(SQLConf.MAX_TO_STRING_FIELDS)

def maxPlanStringLength: Int = getConf(SQLConf.MAX_PLAN_STRING_LENGTH).toInt

Expand Down
4 changes: 2 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/types/StructType.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import scala.util.control.NonFatal
import org.json4s.JsonDSL._

import org.apache.spark.annotation.Stable
import org.apache.spark.sql.SqlApiConf
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, LegacyTypeStringParser}
Expand All @@ -33,7 +34,6 @@ import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
import org.apache.spark.sql.catalyst.util.StringConcat
import org.apache.spark.sql.catalyst.util.truncatedString
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.util.collection.Utils

/**
Expand Down Expand Up @@ -426,7 +426,7 @@ case class StructType(fields: Array[StructField]) extends DataType with Seq[Stru
truncatedString(
fieldTypes,
"struct<", ",", ">",
SQLConf.get.maxToStringFields)
SqlApiConf.get.maxToStringFields)
}

override def catalogString: String = {
Expand Down