Skip to content
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

[SPARK-37867][SQL] Compile aggregate functions of build-in JDBC dialect #35166

Closed
wants to merge 13 commits into from
Closed

[SPARK-37867][SQL] Compile aggregate functions of build-in JDBC dialect #35166

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
9bb754e
[SPARK-37867][SQL] Compile aggregate functions of build-in JDBC dialect
beliefer Jan 11, 2022
fcf6e88
Update code
beliefer Jan 12, 2022
d342f86
Update code
beliefer Jan 14, 2022
d71b293
Update code
beliefer Jan 14, 2022
b9e648d
Update code
beliefer Jan 15, 2022
e626cd8
Update code
beliefer Jan 15, 2022
fc5f712
Update code
beliefer Jan 15, 2022
8ea6bbd
Update code
beliefer Jan 15, 2022
7aa569a
Update code
beliefer Jan 15, 2022
9c4ff55
Update code
beliefer Jan 20, 2022
7dc4597
Update code
beliefer Jan 20, 2022
c23e8e4
Update code
beliefer Jan 21, 2022
4dd1379
Update code
beliefer Jan 25, 2022
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
19 changes: 18 additions & 1 deletion ...r-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/DB2IntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.sql.jdbc.v2

import java.sql.Connection
import java.util.Locale

import org.scalatest.time.SpanSugar._

Expand All @@ -38,6 +39,7 @@ import org.apache.spark.tags.DockerTest
@DockerTest
class DB2IntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
override val catalogName: String = "db2"
override val namespaceOpt: Option[String] = Some("DB2INST1")
override val db = new DatabaseOnDocker {
override val imageName = sys.env.getOrElse("DB2_DOCKER_IMAGE_NAME", "ibmcom/db2:11.5.6.0a")
override val env = Map(
Expand All @@ -59,8 +61,19 @@ class DB2IntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
override def sparkConf: SparkConf = super.sparkConf
.set("spark.sql.catalog.db2", classOf[JDBCTableCatalog].getName)
.set("spark.sql.catalog.db2.url", db.getJdbcUrl(dockerIp, externalPort))
.set("spark.sql.catalog.db2.pushDownAggregate", "true")

override def dataPreparation(conn: Connection): Unit = {}
override def dataPreparation(conn: Connection): Unit = {
conn.prepareStatement(
"CREATE TABLE employee (dept INTEGER, name CLOB, salary DECIMAL(20, 2), bonus DOUBLE)")
Copy link
Contributor

Choose a reason for hiding this comment

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

use VARCHAR(n) instead of CLOB?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Updated

.executeUpdate()
conn.prepareStatement(
"""
|INSERT INTO employee VALUES
|(1, 'amy', 10000, 1000), (2, 'alex', 12000, 1200), (1, 'cathy', 9000, 1200),
|(2, 'david', 10000, 1300), (6, 'jen', 12000, 1200)
|""".stripMargin).executeUpdate()
}

override def testUpdateColumnType(tbl: String): Unit = {
sql(s"CREATE TABLE $tbl (ID INTEGER)")
Expand All @@ -86,4 +99,8 @@ class DB2IntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
val expectedSchema = new StructType().add("ID", IntegerType, true, defaultMetadata)
assert(t.schema === expectedSchema)
}

override def caseConvert(tableName: String): String = tableName.toUpperCase(Locale.ROOT)
Copy link
Contributor

Choose a reason for hiding this comment

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

I think DB2 objects names are not case sensitive.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I tests many time. If uses the table name with lower case, DB2 throws Exception about not find table.


testVarPop()
}
18 changes: 17 additions & 1 deletion ...ation-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,10 +58,21 @@ class MsSqlServerIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBC
override def sparkConf: SparkConf = super.sparkConf
.set("spark.sql.catalog.mssql", classOf[JDBCTableCatalog].getName)
.set("spark.sql.catalog.mssql.url", db.getJdbcUrl(dockerIp, externalPort))
.set("spark.sql.catalog.mssql.pushDownAggregate", "true")

override val connectionTimeout = timeout(7.minutes)

override def dataPreparation(conn: Connection): Unit = {}
override def dataPreparation(conn: Connection): Unit = {
conn.prepareStatement(
"CREATE TABLE employee (dept INT, name VARCHAR(32), salary NUMERIC(20, 2), bonus FLOAT)")
.executeUpdate()
conn.prepareStatement(
"""
|INSERT INTO employee VALUES
|(1, 'amy', 10000, 1000), (2, 'alex', 12000, 1200), (1, 'cathy', 9000, 1200),
|(2, 'david', 10000, 1300), (6, 'jen', 12000, 1200)
|""".stripMargin).executeUpdate()
}

override def notSupportsTableComment: Boolean = true

Expand Down Expand Up @@ -91,4 +102,9 @@ class MsSqlServerIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBC

assert(msg.contains("UpdateColumnNullability is not supported"))
}

testVarPop()
testVarSamp()
testStddevPop()
testStddevSamp()
}
15 changes: 15 additions & 0 deletions ...integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,13 +57,23 @@ class MySQLIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
override def sparkConf: SparkConf = super.sparkConf
.set("spark.sql.catalog.mysql", classOf[JDBCTableCatalog].getName)
.set("spark.sql.catalog.mysql.url", db.getJdbcUrl(dockerIp, externalPort))
.set("spark.sql.catalog.mysql.pushDownAggregate", "true")

override val connectionTimeout = timeout(7.minutes)

private var mySQLVersion = -1

override def dataPreparation(conn: Connection): Unit = {
mySQLVersion = conn.getMetaData.getDatabaseMajorVersion
conn.prepareStatement(
"CREATE TABLE employee (dept INT, name VARCHAR(32), salary DECIMAL(20, 2)," +
" bonus DOUBLE)").executeUpdate()
conn.prepareStatement(
"""
|INSERT INTO employee VALUES
| (1, 'amy', 10000, 1000), (2, 'alex', 12000, 1200), (1, 'cathy', 9000, 1200),
| (2, 'david', 10000, 1300), (6, 'jen', 12000, 1200)
|""".stripMargin).executeUpdate()
}

override def testUpdateColumnType(tbl: String): Unit = {
Expand Down Expand Up @@ -119,4 +129,9 @@ class MySQLIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
override def supportsIndex: Boolean = true

override def indexOptions: String = "KEY_BLOCK_SIZE=10"

testVarPop()
testVarSamp()
testStddevPop()
testStddevSamp()
}
30 changes: 29 additions & 1 deletion ...ntegration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/OracleIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.sql.jdbc.v2

import java.sql.Connection
import java.util.Locale

import org.scalatest.time.SpanSugar._

Expand Down Expand Up @@ -56,6 +57,7 @@ import org.apache.spark.tags.DockerTest
@DockerTest
class OracleIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest {
override val catalogName: String = "oracle"
override val namespaceOpt: Option[String] = Some("SYSTEM")
override val db = new DatabaseOnDocker {
lazy override val imageName =
sys.env.getOrElse("ORACLE_DOCKER_IMAGE_NAME", "gvenzl/oracle-xe:18.4.0")
Expand All @@ -73,9 +75,25 @@ class OracleIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest
override def sparkConf: SparkConf = super.sparkConf
.set("spark.sql.catalog.oracle", classOf[JDBCTableCatalog].getName)
.set("spark.sql.catalog.oracle.url", db.getJdbcUrl(dockerIp, externalPort))
.set("spark.sql.catalog.oracle.pushDownAggregate", "true")

override val connectionTimeout = timeout(7.minutes)
override def dataPreparation(conn: Connection): Unit = {}

override def dataPreparation(conn: Connection): Unit = {
conn.prepareStatement(
"CREATE TABLE employee (dept NUMBER(32), name VARCHAR2(32), salary NUMBER(20, 2)," +
" bonus BINARY_DOUBLE)").executeUpdate()
conn.prepareStatement("INSERT INTO employee VALUES (1, 'amy', 10000, 1000)")
.executeUpdate()
conn.prepareStatement("INSERT INTO employee VALUES (2, 'alex', 12000, 1200)")
.executeUpdate()
conn.prepareStatement("INSERT INTO employee VALUES (1, 'cathy', 9000, 1200)")
.executeUpdate()
conn.prepareStatement("INSERT INTO employee VALUES (2, 'david', 10000, 1300)")
.executeUpdate()
conn.prepareStatement("INSERT INTO employee VALUES (6, 'jen', 12000, 1200)")
Copy link
Contributor

Choose a reason for hiding this comment

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

Seems like every database can create table then insert values with multiple INSERT commands. Can we do it in the base class?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

First, each database supports different data types? There is no common CREATE TABLE statement.
Second, some databases support insert into table Values followed by multiple lines in succession. For example, Oracle does not support this syntax and only supports inserting one by one.
In addition, the performance of inserting multiple items at one time is generally better than inserting one item multiple times.

.executeUpdate()
}

override def testUpdateColumnType(tbl: String): Unit = {
sql(s"CREATE TABLE $tbl (ID INTEGER)")
Expand All @@ -93,4 +111,14 @@ class OracleIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTest
assert(msg1.contains(
s"Cannot update $catalogName.alt_table field ID: string cannot be cast to int"))
}

override def caseConvert(tableName: String): String = tableName.toUpperCase(Locale.ROOT)

testVarPop()
testVarSamp()
testStddevPop()
testStddevSamp()
testCovarPop()
testCovarSamp()
testCorr()
}
21 changes: 20 additions & 1 deletion ...egration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,19 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTes
.set("spark.sql.catalog.postgresql.url", db.getJdbcUrl(dockerIp, externalPort))
.set("spark.sql.catalog.postgresql.pushDownTableSample", "true")
.set("spark.sql.catalog.postgresql.pushDownLimit", "true")
.set("spark.sql.catalog.postgresql.pushDownAggregate", "true")

override def dataPreparation(conn: Connection): Unit = {}
override def dataPreparation(conn: Connection): Unit = {
conn.prepareStatement(
"CREATE TABLE employee (dept INTEGER, name VARCHAR(32), salary NUMERIC(20, 2)," +
" bonus double precision)").executeUpdate()
conn.prepareStatement(
"""
|INSERT INTO employee VALUES
| (1, 'amy', 10000, 1000), (2, 'alex', 12000, 1200), (1, 'cathy', 9000, 1200),
| (2, 'david', 10000, 1300), (6, 'jen', 12000, 1200)
|""".stripMargin).executeUpdate()
}

override def testUpdateColumnType(tbl: String): Unit = {
sql(s"CREATE TABLE $tbl (ID INTEGER)")
Expand Down Expand Up @@ -84,4 +95,12 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite with V2JDBCTes
override def supportsIndex: Boolean = true

override def indexOptions: String = "FILLFACTOR=70"

testVarPop()
testVarSamp()
testStddevPop()
testStddevSamp()
testCovarPop()
testCovarSamp()
testCorr()
}
Loading