You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by chutium <gi...@git.apache.org> on 2014/07/28 01:12:55 UTC
[GitHub] spark pull request: SPARK-2710 Build SchemaRDD from a JdbcRDD with...
GitHub user chutium opened a pull request:
https://github.com/apache/spark/pull/1612
SPARK-2710 Build SchemaRDD from a JdbcRDD with MetaData
SPARK-2710 Build SchemaRDD from a JdbcRDD with MetaData
and a small bug fix on JdbcRDD, line 109
it seems conn will never be closed
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/chutium/spark jdbc-schema-rdd
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/1612.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1612
----
commit 46a30628e8a7c8109ac647caa850cd8229431ddf
Author: chutium <te...@gmail.com>
Date: 2014-07-27T23:03:25Z
SPARK-2710 Build SchemaRDD from a JdbcRDD with MetaData
and a small bug fix on JdbcRDD, line 106, it seems conn will never be closed
commit 1f8c76ec15dc8cf0bb0bac8db01f76f0bfe58e52
Author: chutium <te...@gmail.com>
Date: 2014-07-27T23:11:24Z
SPARK-2710 Build SchemaRDD from a JdbcRDD with MetaData
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15506243
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +105,10 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) stmt.setLong(1, part.lower)
+ if (parameterCount > 1) stmt.setLong(2, part.upper)
+
--- End diff --
I just checked the unit test, seems the JdbcRDD is quite limit in making partitions, probably we need to think more about it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17101449
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,72 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): StructType = {
+ StructType(createSchema(jdbcResultSet.getSchema))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private[sql] def jdbcResultSetToRow(
+ jdbcResultSet: JdbcRDD[ResultSet],
+ schema: StructType) : RDD[Row] = {
+ val row = new GenericMutableRow(schema.fields.length)
+ jdbcResultSet.map(asRow(_, row, schema.fields))
+ }
+
+ private def asRow(rs: ResultSet, row: GenericMutableRow, schema: Seq[StructField]): Row = {
+ schema.zipWithIndex.foreach {
--- End diff --
changed to while, learned to write from your ```fillObject``` in hive ```TableReader``` :D https://github.com/apache/spark/blob/2b8d89e30ebfe2272229a1eddd7542d7437c9924/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala#L279-L288
thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r16704875
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -57,6 +61,8 @@ class JdbcRDD[T: ClassTag](
mapRow: (ResultSet) => T = JdbcRDD.resultSetToObjectArray _)
extends RDD[T](sc, Nil) with Logging {
+ private var schema: Seq[(String, Int, Boolean)] = null
--- End diff --
yep, i tried to do like you said before, but there is no public method or attribute to get ResultSet or Statement from this ```JdbcRDD``` in spark core, so in ```JdbcResultSetRDD``` i have no idea how can we get the metadata from ```JdbcRDD```... otherwise we do something like ```jdbcRDD.head``` then we can get the metadata from first row, but it may execute the whole query at plan phase.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15506096
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) schema
--- End diff --
Seems a dead code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15528919
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
+ schema.zipWithIndex.foreach {
+ case (StructField(name, dataType, nullable), i) => {
+ dataType match {
+ case StringType => row.update(i, rs.getString(i+1))
--- End diff --
i find a easier way, we can check rs.wasNull after the dataType match {}, then just once
if (rs.wasNull) row.update(i, null)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15505789
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
+ schema.zipWithIndex.foreach {
--- End diff --
Use `while` loop instead of the `zipWithIndex` and `foreach`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15564232
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
+ schema.zipWithIndex.foreach {
+ case (StructField(name, dataType, nullable), i) => {
+ dataType match {
+ case StringType => row.update(i, rs.getString(i+1))
--- End diff --
If we reused the row object, this is a bug here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-53942630
ok to test
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-54433054
[QA tests have started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/19775/consoleFull) for PR 1612 at commit [`566d154`](https://github.com/apache/spark/commit/566d154b23d0d7163b48889cfcd523e706c9a480).
* This patch merges cleanly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15506043
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +105,10 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) stmt.setLong(1, part.lower)
+ if (parameterCount > 1) stmt.setLong(2, part.upper)
+
--- End diff --
This is not what your mean here. It is something like `SELECT * FROM tbl limit ?, ?`. But there is probably a bug in original implementation, as I remember the second parameter of `LIMIT` clause in MySQL is the `row_count`, not the `upper`. see: http://dev.mysql.com/doc/refman/5.0/en/select.html
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15573636
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +105,10 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) stmt.setLong(1, part.lower)
+ if (parameterCount > 1) stmt.setLong(2, part.upper)
+
--- End diff --
done, https://github.com/apache/spark/pull/1612/files#diff-0
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r16642111
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,72 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): StructType = {
+ StructType(createSchema(jdbcResultSet.getSchema))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private[sql] def jdbcResultSetToRow(
+ jdbcResultSet: JdbcRDD[ResultSet],
+ schema: StructType) : RDD[Row] = {
+ val row = new GenericMutableRow(schema.fields.length)
+ jdbcResultSet.map(asRow(_, row, schema.fields))
+ }
+
+ private def asRow(rs: ResultSet, row: GenericMutableRow, schema: Seq[StructField]): Row = {
+ schema.zipWithIndex.foreach {
--- End diff --
`zipWithIndex` will cause some overhead, let's use for loop instead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r16642168
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDDSuite.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.jdbc
+
+import java.sql._
+
+import org.scalatest.BeforeAndAfter
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.test.TestSQLContext._
+
+class JdbcResultSetRDDSuite extends QueryTest with BeforeAndAfter {
+
+ before {
+ Class.forName("org.apache.derby.jdbc.EmbeddedDriver")
+ val conn = DriverManager.getConnection("jdbc:derby:target/JdbcSchemaRDDSuiteDb;create=true")
+ try {
+ val create = conn.createStatement
+ create.execute("""
+ CREATE TABLE FOO(
+ ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
+ DATA INTEGER
+ )""")
+ create.close()
+ val insert = conn.prepareStatement("INSERT INTO FOO(DATA) VALUES(?)")
+ (1 to 100).foreach { i =>
+ insert.setInt(1, i * 2)
+ insert.executeUpdate
+ }
+ insert.close()
+ } catch {
+ case e: SQLException if e.getSQLState == "X0Y32" =>
+ // table exists
+ } finally {
+ conn.close()
+ }
+ }
+
+ test("basic functionality") {
+ val jdbcResultSetRDD = jdbcResultSet("jdbc:derby:target/JdbcSchemaRDDSuiteDb", "SELECT DATA FROM FOO")
+ jdbcResultSetRDD.registerAsTable("foo")
+
+ checkAnswer(
+ sql("select count(*) from foo"),
+ 100
+ )
+ checkAnswer(
+ sql("select sum(DATA) from foo"),
+ 10100
+ )
+ }
+
+ after {
+ try {
+ DriverManager.getConnection("jdbc:derby:;shutdown=true")
+ } catch {
+ case se: SQLException if se.getSQLState == "XJ015" =>
--- End diff --
Either logging the exception or remove the `try ... catch`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279050
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala ---
@@ -205,6 +208,54 @@ class SQLContext(@transient val sparkContext: SparkContext)
}
/**
+ * Loads from JDBC, returning the ResultSet as a [[SchemaRDD]].
+ * It gets MetaData from ResultSet of PreparedStatement to determine the schema.
+ *
+ * @group userf
+ */
+ def jdbcResultSet(
--- End diff --
For the 1.2 release we are going to be focusing on adding more external datasources. As part of this we are trying to change the way we add them to avoid SQLContext getting to large. What do you think about adding an object, `org.apache.spark.sql.jdbc.JDBC` that has these methods instead of adding them here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by koeninger <gi...@git.apache.org>.
Github user koeninger commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17280404
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +73,32 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) {
+ return schema
+ }
+
+ val conn = getConnection()
--- End diff --
Is this connection guaranteed to get closed? It won't benefit from the addOnCompleteCallback below, for instance.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15451376
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
--- End diff --
here i tried to return a java.sql.ResultSetMetaData object, then build the Seq[(String, Int, Boolean)] for schemaRDD in Spark SQL scope, but when i run this SchemaRDD, i got "java.io.NotSerializableException: org.postgresql.jdbc4.Jdbc4ResultSetMetaData"
so i let this method return a Seq[(String, Int, Boolean)], and in Spark SQL scope, create Seq[StructField] from this Seq[(String, Int, Boolean)]
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15522485
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) schema
+
+ val conn = getConnection()
+ val stmt = conn.prepareStatement(sql)
+ val metadata = stmt.getMetaData
+ try {
+ if (null != stmt && ! stmt.isClosed()) stmt.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing statement", e)
+ }
+ schema = Seq[(String, Int, Boolean)]()
--- End diff --
i will try, am a scala newbie
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15509342
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcTypes.scala ---
@@ -0,0 +1,56 @@
+/*
+* 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.jdbc
+
+import org.apache.spark.Logging
+import org.apache.spark.sql.catalyst.types._
+
+private[sql] object JdbcTypes extends Logging {
+
+ def toPrimitiveDataType(jdbcType: Int): DataType =
+ jdbcType match {
+ case java.sql.Types.LONGVARCHAR
+ | java.sql.Types.VARCHAR
+ | java.sql.Types.CHAR => StringType
+ case java.sql.Types.NUMERIC
+ | java.sql.Types.DECIMAL => DecimalType
+ case java.sql.Types.BIT => BooleanType
+ case java.sql.Types.TINYINT => ByteType
+ case java.sql.Types.SMALLINT => ShortType
+ case java.sql.Types.INTEGER => IntegerType
+ case java.sql.Types.BIGINT => LongType
+ case java.sql.Types.REAL => FloatType
+ case java.sql.Types.FLOAT
+ | java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.LONGVARBINARY
+ | java.sql.Types.VARBINARY
+ | java.sql.Types.BINARY => BinaryType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.DATE
--- End diff --
Why not cast into TimestampType?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15522818
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
+ schema.zipWithIndex.foreach {
+ case (StructField(name, dataType, nullable), i) => {
+ dataType match {
+ case StringType => row.update(i, rs.getString(i+1))
--- End diff --
oh, this is serious, thanks a lot
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279616
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDDSuite.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.jdbc
+
+import java.sql._
+
+import org.scalatest.BeforeAndAfter
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.test.TestSQLContext._
+
+class JdbcResultSetRDDSuite extends QueryTest with BeforeAndAfter {
+
+ before {
+ Class.forName("org.apache.derby.jdbc.EmbeddedDriver")
+ val conn = DriverManager.getConnection("jdbc:derby:target/JdbcSchemaRDDSuiteDb;create=true")
+ try {
+ val create = conn.createStatement
+ create.execute("""
+ CREATE TABLE FOO(
+ ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
+ DATA INTEGER
+ )""")
+ create.close()
+ val insert = conn.prepareStatement("INSERT INTO FOO(DATA) VALUES(?)")
+ (1 to 100).foreach { i =>
+ insert.setInt(1, i * 2)
+ insert.executeUpdate
+ }
+ insert.close()
+ } catch {
+ case e: SQLException if e.getSQLState == "X0Y32" =>
+ // table exists
+ } finally {
+ conn.close()
+ }
+ }
+
+ test("basic functionality") {
+ val jdbcResultSetRDD = jdbcResultSet("jdbc:derby:target/JdbcSchemaRDDSuiteDb", "SELECT DATA FROM FOO")
+ jdbcResultSetRDD.registerAsTable("foo")
+
+ checkAnswer(
+ sql("select count(*) from foo"),
+ 100
+ )
+ checkAnswer(
+ sql("select sum(DATA) from foo"),
+ 10100
+ )
+ }
+
+ after {
+ try {
+ DriverManager.getConnection("jdbc:derby:;shutdown=true")
+ } catch {
+ case se: SQLException if se.getSQLState == "XJ015" =>
--- End diff --
Thanks! Maybe we can add this as a comment in both places?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15509316
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
+ schema.zipWithIndex.foreach {
+ case (StructField(name, dataType, nullable), i) => {
+ dataType match {
+ case StringType => row.update(i, rs.getString(i+1))
--- End diff --
This probably contains 2 bugs:
* Not consistent with `JdbcTypes.toPrimitiveDataType`
As `java.sql.Types.DATE`, `java.sql.Types.TIME` and `java.sql.Types.CLOB` all of them will cast into `StringType` in `JdbcTypes.toPrimitiveDataType`, `rs.getString` for `java.sql.Types.TIME` here may cause exception.
* Null value checking
According to the document of ResultSet, the primitive type getter methods will return a default value if null value found. For example the API `int getInt(int columnIndex)` will returns `0` if the actual value is `null`.
We can write the code like:
```
val iVal = rs.getInt("ID_PARENT")
if (rs.wasNull()) {
row.update(i, null)
} else {
row.update(i, iVal)
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-55317048
thanks for the review, i will try to improve it soon, adding more external datasources is always helpful, then we can use Spark SQL as a data integration platform, and of course SQL92 is also important, now Spark SQL ist more like a tool for quering hadoop files.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15505774
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
--- End diff --
This will cause object created for every input row, it's better to move out of function `asRow`, like in the `inferSchema`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15564039
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) schema
--- End diff --
I mean `if (null != schema) schema` is the dead code. Probably what you want is
```
if (null != schema) return schema
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-50617738
Test Suite added
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r16642061
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -57,6 +61,8 @@ class JdbcRDD[T: ClassTag](
mapRow: (ResultSet) => T = JdbcRDD.resultSetToObjectArray _)
extends RDD[T](sc, Nil) with Logging {
+ private var schema: Seq[(String, Int, Boolean)] = null
--- End diff --
Move the schema stuff to `JdbcResultSetRDD`? We'd better keep the Spark core clean and same implementation pattern with the other Core RDDs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-50435389
Actually I am not sure the motivation of this PR, seems fix the bug in line 109 of `JdbcRDD.scala` is enough.
Creating the SchemaRDD from a normal RDD is easy while the `SQLContext` instance declared, we can write the code like: `jdbcRDD.where(xxx)` directly, see the implicit function `SQLContext.createSchemaRDD`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15506119
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) schema
+
+ val conn = getConnection()
+ val stmt = conn.prepareStatement(sql)
+ val metadata = stmt.getMetaData
+ try {
+ if (null != stmt && ! stmt.isClosed()) stmt.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing statement", e)
+ }
+ schema = Seq[(String, Int, Boolean)]()
--- End diff --
Use `scala.collection.mutable.ArrayBuffer` instead?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279173
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
--- End diff --
We should probably also make this `private[spark]`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-53942797
[QA tests have finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/19508/consoleFull) for PR 1612 at commit [`917a753`](https://github.com/apache/spark/commit/917a753ac3de77ca30f33433c42968c38b9b6865).
* This patch **fails** unit tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279021
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,72 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): StructType = {
+ StructType(createSchema(jdbcResultSet.getSchema))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private[sql] def jdbcResultSetToRow(
+ jdbcResultSet: JdbcRDD[ResultSet],
+ schema: StructType) : RDD[Row] = {
+ val row = new GenericMutableRow(schema.fields.length)
+ jdbcResultSet.map(asRow(_, row, schema.fields))
+ }
+
+ private def asRow(rs: ResultSet, row: GenericMutableRow, schemaFields: Seq[StructField]): Row = {
+ var i = 0
+ while (i < schemaFields.length) {
+ schemaFields(i).dataType match {
+ case StringType => row.update(i, rs.getString(i + 1))
+ case DecimalType => row.update(i, rs.getBigDecimal(i + 1))
+ case BooleanType => row.update(i, rs.getBoolean(i + 1))
+ case ByteType => row.update(i, rs.getByte(i + 1))
+ case ShortType => row.update(i, rs.getShort(i + 1))
+ case IntegerType => row.update(i, rs.getInt(i + 1))
+ case LongType => row.update(i, rs.getLong(i + 1))
+ case FloatType => row.update(i, rs.getFloat(i + 1))
+ case DoubleType => row.update(i, rs.getDouble(i + 1))
+ case BinaryType => row.update(i, rs.getBytes(i + 1))
+ case TimestampType => row.update(i, rs.getTimestamp(i + 1))
+ case _ => sys.error(
+ s"Unsupported jdbc datatype")
--- End diff --
Would be good to print what the unsupported type is. Also, try to wrap at the highest syntatic level, for example:
```scala
case unsupportedType =>
sys.error(s"Unsupported jdbc datatype: $unsupportedType")
```
(Though actually in this case I think it'll all fit on one line).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-65164393
Thanks for working on this! I think this will be a really useful addition. However, with the new external data sources api that is part of 1.2, I think it might be better to do this as an external library (for example: https://github.com/databricks/spark-avro). This would make it easier to make releases, and also help us keep spark core's size manageable. If you agree, maybe we can close this issue? Let me know if you have any questions.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r16705825
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +113,14 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) {
--- End diff --
i am afraid they do not think it is a problem, in the original comment of JdbcRDD:
```
* @param sql the text of the query.
* The query must contain two ? placeholders for parameters used to partition the results.
* E.g. "select title, author from books where ? <= id and id <= ?"
```
but i believe many users just want to get the whole table out of RDBMS simply, and then do some calculation in Spark's magic world... how many partitions will be created is no matter, so in the normal use case, the tables stored in RDBMS are small, therefore these two ? placeholders for partitioning is not always necessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17278868
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
--- End diff --
Can you add this as a comment here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-53942746
[QA tests have started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/19508/consoleFull) for PR 1612 at commit [`917a753`](https://github.com/apache/spark/commit/917a753ac3de77ca30f33433c42968c38b9b6865).
* This patch merges cleanly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-54433211
[QA tests have finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/19775/consoleFull) for PR 1612 at commit [`566d154`](https://github.com/apache/spark/commit/566d154b23d0d7163b48889cfcd523e706c9a480).
* This patch **fails** unit tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by koeninger <gi...@git.apache.org>.
Github user koeninger commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279539
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDDSuite.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.jdbc
+
+import java.sql._
+
+import org.scalatest.BeforeAndAfter
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.test.TestSQLContext._
+
+class JdbcResultSetRDDSuite extends QueryTest with BeforeAndAfter {
+
+ before {
+ Class.forName("org.apache.derby.jdbc.EmbeddedDriver")
+ val conn = DriverManager.getConnection("jdbc:derby:target/JdbcSchemaRDDSuiteDb;create=true")
+ try {
+ val create = conn.createStatement
+ create.execute("""
+ CREATE TABLE FOO(
+ ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
+ DATA INTEGER
+ )""")
+ create.close()
+ val insert = conn.prepareStatement("INSERT INTO FOO(DATA) VALUES(?)")
+ (1 to 100).foreach { i =>
+ insert.setInt(1, i * 2)
+ insert.executeUpdate
+ }
+ insert.close()
+ } catch {
+ case e: SQLException if e.getSQLState == "X0Y32" =>
+ // table exists
+ } finally {
+ conn.close()
+ }
+ }
+
+ test("basic functionality") {
+ val jdbcResultSetRDD = jdbcResultSet("jdbc:derby:target/JdbcSchemaRDDSuiteDb", "SELECT DATA FROM FOO")
+ jdbcResultSetRDD.registerAsTable("foo")
+
+ checkAnswer(
+ sql("select count(*) from foo"),
+ 100
+ )
+ checkAnswer(
+ sql("select sum(DATA) from foo"),
+ 10100
+ )
+ }
+
+ after {
+ try {
+ DriverManager.getConnection("jdbc:derby:;shutdown=true")
+ } catch {
+ case se: SQLException if se.getSQLState == "XJ015" =>
--- End diff --
http://db.apache.org/derby/papers/DerbyTut/embedded_intro.html
" A clean shutdown always throws SQL exception XJ015, which can be ignored. "
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15564165
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
--- End diff --
OK, I see, @yhuai actually added a TODO in https://github.com/apache/spark/pull/1346.
https://github.com/yhuai/spark/blob/dataTypeAndSchema/sql/core/src/main/scala/org/apache/spark/sql/json/JsonRDD.scala#L360
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15573659
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) schema
--- End diff --
oops, yep, thanks a lot
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r16705207
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDDSuite.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.jdbc
+
+import java.sql._
+
+import org.scalatest.BeforeAndAfter
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.test.TestSQLContext._
+
+class JdbcResultSetRDDSuite extends QueryTest with BeforeAndAfter {
+
+ before {
+ Class.forName("org.apache.derby.jdbc.EmbeddedDriver")
+ val conn = DriverManager.getConnection("jdbc:derby:target/JdbcSchemaRDDSuiteDb;create=true")
+ try {
+ val create = conn.createStatement
+ create.execute("""
+ CREATE TABLE FOO(
+ ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
+ DATA INTEGER
+ )""")
+ create.close()
+ val insert = conn.prepareStatement("INSERT INTO FOO(DATA) VALUES(?)")
+ (1 to 100).foreach { i =>
+ insert.setInt(1, i * 2)
+ insert.executeUpdate
+ }
+ insert.close()
+ } catch {
+ case e: SQLException if e.getSQLState == "X0Y32" =>
+ // table exists
+ } finally {
+ conn.close()
+ }
+ }
+
+ test("basic functionality") {
+ val jdbcResultSetRDD = jdbcResultSet("jdbc:derby:target/JdbcSchemaRDDSuiteDb", "SELECT DATA FROM FOO")
+ jdbcResultSetRDD.registerAsTable("foo")
+
+ checkAnswer(
+ sql("select count(*) from foo"),
+ 100
+ )
+ checkAnswer(
+ sql("select sum(DATA) from foo"),
+ 10100
+ )
+ }
+
+ after {
+ try {
+ DriverManager.getConnection("jdbc:derby:;shutdown=true")
+ } catch {
+ case se: SQLException if se.getSQLState == "XJ015" =>
--- End diff --
i have also no idea what is this "XJ015"... i taked it from https://github.com/apache/spark/blob/master/core/src/test/scala/org/apache/spark/rdd/JdbcRDDSuite.scala :P
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15564209
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,70 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.sql.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): LogicalPlan = {
+ val schema = createSchema(jdbcResultSet.getSchema)
+
+ SparkLogicalPlan(ExistingRdd(asAttributes(schema), jdbcResultSet.map(asRow(_, schema))))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private def asRow(rs: ResultSet, schema: Seq[StructField]): Row = {
+ val row = new GenericMutableRow(schema.length)
+ schema.zipWithIndex.foreach {
--- End diff --
This is actually improvement for performance, it probably be merged very quickly if you do that in a new PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-54915903
Thanks for working on this! Several people have asked for it :)
Aside from the few minor style comments, it would be great if we could add APIs for java and python as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15452368
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +105,10 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) stmt.setLong(1, part.lower)
+ if (parameterCount > 1) stmt.setLong(2, part.upper)
+
--- End diff --
user do not need to set parameters in SQL query if they only want to use 1 partition
such as i want to SELECT * FROM some_tbl
i do not need to write it like SELECT * FROM some_tbl WHERE 1=? and 2=? , then setLong(1, 1) and setLong(2, 2)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-54436556
[QA tests have started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/19782/consoleFull) for PR 1612 at commit [`2013303`](https://github.com/apache/spark/commit/2013303c37eb8408ea31d285d2d3ac23ded7b3e9).
* This patch merges cleanly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17504535
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala ---
@@ -205,6 +208,54 @@ class SQLContext(@transient val sparkContext: SparkContext)
}
/**
+ * Loads from JDBC, returning the ResultSet as a [[SchemaRDD]].
+ * It gets MetaData from ResultSet of PreparedStatement to determine the schema.
+ *
+ * @group userf
+ */
+ def jdbcResultSet(
--- End diff --
What we did for Spark Streaming is that each of the external connectors has a module called XUtils with static utility functions for creating things:
```
object KafkaUtils {
def createKafkaStream(streamingContext: StreamingContext)
}
```
For this reason it might be good to call this `JDBCUtils` in similar fashion.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chutium <gi...@git.apache.org>.
Github user chutium commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17203025
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,72 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): StructType = {
+ StructType(createSchema(jdbcResultSet.getSchema))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private[sql] def jdbcResultSetToRow(
+ jdbcResultSet: JdbcRDD[ResultSet],
+ schema: StructType) : RDD[Row] = {
+ val row = new GenericMutableRow(schema.fields.length)
+ jdbcResultSet.map(asRow(_, row, schema.fields))
+ }
+
+ private def asRow(rs: ResultSet, row: GenericMutableRow, schema: Seq[StructField]): Row = {
+ schema.zipWithIndex.foreach {
--- End diff --
Hi, @chenghao-intel how about use ```for (i <- 1 to schemaFields.length) { ... }``` here? better than ```while (i < schemaFields.length) { ... i+=1 }``` ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279150
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +113,14 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) {
--- End diff --
Yeah, I agree. I was annoyed when I had to work around this. As long as its backwards compatible I'm okay including it here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by koeninger <gi...@git.apache.org>.
Github user koeninger commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279806
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +113,14 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) {
--- End diff --
Not that there's anything wrong with backwards compatible fixes/enhancements, but a few things I noticed here:
1. If it's a sufficiently small table that a user is only using 1 partition, why not encourage them to query it from the driver and broadcast it?
2. As it stands, it looks like you allow 0, 1, 2, or more ? placeholders, but the doc comment change only describes the 0 or 2 case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-54453005
[QA tests have finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/19782/consoleFull) for PR 1612 at commit [`2013303`](https://github.com/apache/spark/commit/2013303c37eb8408ea31d285d2d3ac23ded7b3e9).
* This patch **passes** unit tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15506092
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +69,28 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) schema
+
+ val conn = getConnection()
+ val stmt = conn.prepareStatement(sql)
+ val metadata = stmt.getMetaData
+ try {
+ if (null != stmt && ! stmt.isClosed()) stmt.close()
+ } catch {
+ case e: Exception => logWarning("Exception closing statement", e)
+ }
+ schema = Seq[(String, Int, Boolean)]()
+ for(i <- 1 to metadata.getColumnCount) {
+ schema :+= (
+ metadata.getColumnName(i),
+ metadata.getColumnType(i),
+ metadata.isNullable(i) == java.sql.ResultSetMetaData.columnNullable
--- End diff --
Is that's correct with `java.sql.ResultSetMetaData.columnNullable`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1612#issuecomment-50289234
Can one of the admins verify this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279109
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDDSuite.scala ---
@@ -0,0 +1,75 @@
+/*
+ * 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.jdbc
+
+import java.sql._
+
+import org.scalatest.BeforeAndAfter
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.test.TestSQLContext._
+
+class JdbcResultSetRDDSuite extends QueryTest with BeforeAndAfter {
+
+ before {
+ Class.forName("org.apache.derby.jdbc.EmbeddedDriver")
+ val conn = DriverManager.getConnection("jdbc:derby:target/JdbcSchemaRDDSuiteDb;create=true")
+ try {
+ val create = conn.createStatement
+ create.execute("""
+ CREATE TABLE FOO(
+ ID INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1),
+ DATA INTEGER
+ )""")
+ create.close()
+ val insert = conn.prepareStatement("INSERT INTO FOO(DATA) VALUES(?)")
+ (1 to 100).foreach { i =>
+ insert.setInt(1, i * 2)
+ insert.executeUpdate
+ }
+ insert.close()
+ } catch {
+ case e: SQLException if e.getSQLState == "X0Y32" =>
+ // table exists
+ } finally {
+ conn.close()
+ }
+ }
+
+ test("basic functionality") {
+ val jdbcResultSetRDD = jdbcResultSet("jdbc:derby:target/JdbcSchemaRDDSuiteDb", "SELECT DATA FROM FOO")
+ jdbcResultSetRDD.registerAsTable("foo")
+
+ checkAnswer(
+ sql("select count(*) from foo"),
+ 100
+ )
+ checkAnswer(
+ sql("select sum(DATA) from foo"),
+ 10100
+ )
+ }
+
+ after {
+ try {
+ DriverManager.getConnection("jdbc:derby:;shutdown=true")
+ } catch {
+ case se: SQLException if se.getSQLState == "XJ015" =>
--- End diff --
Hmm, yeah good question... Perhaps @koeninger, who wrote the original test, could enlighten us?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by koeninger <gi...@git.apache.org>.
Github user koeninger commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17280425
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -67,6 +73,32 @@ class JdbcRDD[T: ClassTag](
}).toArray
}
+ def getSchema: Seq[(String, Int, Boolean)] = {
+ if (null != schema) {
+ return schema
+ }
+
+ val conn = getConnection()
+ val stmt = conn.prepareStatement(sql)
+ val metadata = stmt.getMetaData
+ try {
+ if (null != stmt && ! stmt.isClosed()) {
+ stmt.close()
+ }
+ } catch {
+ case e: Exception => logWarning("Exception closing statement", e)
+ }
+ schema = Seq[(String, Int, Boolean)]()
+ for(i <- 1 to metadata.getColumnCount) {
+ schema :+= (
--- End diff --
Are there any thread safety concerns regarding mutating schema here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279068
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcTypes.scala ---
@@ -0,0 +1,77 @@
+/*
+ * 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.jdbc
+
+import org.apache.spark.Logging
+import org.apache.spark.sql.catalyst.types._
+
+private[sql] object JdbcTypes extends Logging {
+
+ /**
+ * More about JDBC types mapped to Java types:
+ * http://docs.oracle.com/javase/6/docs/technotes/guides/jdbc/getstart/mapping.html#1051555
+ *
+ * Compatibility of ResultSet getter Methods defined in JDBC spec:
+ * http://download.oracle.com/otn-pub/jcp/jdbc-4_1-mrel-spec/jdbc4.1-fr-spec.pdf
+ * page 211
+ */
+ def toPrimitiveDataType(jdbcType: Int): DataType =
+ jdbcType match {
+ case java.sql.Types.LONGVARCHAR
+ | java.sql.Types.VARCHAR
+ | java.sql.Types.CHAR => StringType
+ case java.sql.Types.NUMERIC
+ | java.sql.Types.DECIMAL => DecimalType
+ case java.sql.Types.BIT => BooleanType
+ case java.sql.Types.TINYINT => ByteType
+ case java.sql.Types.SMALLINT => ShortType
+ case java.sql.Types.INTEGER => IntegerType
+ case java.sql.Types.BIGINT => LongType
+ case java.sql.Types.REAL => FloatType
+ case java.sql.Types.FLOAT
+ | java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.LONGVARBINARY
+ | java.sql.Types.VARBINARY
+ | java.sql.Types.BINARY => BinaryType
+ // Timestamp's getter should also be able to get DATE and TIME according to JDBC spec
+ case java.sql.Types.TIMESTAMP
+ | java.sql.Types.DATE
+ | java.sql.Types.TIME => TimestampType
+
+ // TODO: CLOB only works with getClob or getAscIIStream
+ // case java.sql.Types.CLOB
+
+ // TODO: BLOB only works with getBlob or getBinaryStream
+ // case java.sql.Types.BLOB
+
+ // TODO: nested types
+ // case java.sql.Types.ARRAY => ArrayType
+ // case java.sql.Types.STRUCT => StructType
+
+ // TODO: unsupported types
+ // case java.sql.Types.DISTINCT
+ // case java.sql.Types.REF
+
+ // TODO: more about JAVA_OBJECT:
+ // http://docs.oracle.com/javase/6/docs/technotes/guides/jdbc/getstart/mapping.html#1038181
+ // case java.sql.Types.JAVA_OBJECT => BinaryType
+
+ case _ => sys.error(
--- End diff --
Same here: include the type that isn't supported.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/1612
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r16642093
--- Diff: core/src/main/scala/org/apache/spark/rdd/JdbcRDD.scala ---
@@ -81,8 +113,14 @@ class JdbcRDD[T: ClassTag](
logInfo("statement fetch size set to: " + stmt.getFetchSize + " to force MySQL streaming ")
}
- stmt.setLong(1, part.lower)
- stmt.setLong(2, part.upper)
+ val parameterCount = stmt.getParameterMetaData.getParameterCount
+ if (parameterCount > 0) {
--- End diff --
Should we put this as a new PR? Which seems like an enhancement / bug fixing for `JdbcRDD`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by chenghao-intel <gi...@git.apache.org>.
Github user chenghao-intel commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r15509361
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcTypes.scala ---
@@ -0,0 +1,56 @@
+/*
+* 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.jdbc
+
+import org.apache.spark.Logging
+import org.apache.spark.sql.catalyst.types._
+
+private[sql] object JdbcTypes extends Logging {
+
+ def toPrimitiveDataType(jdbcType: Int): DataType =
+ jdbcType match {
+ case java.sql.Types.LONGVARCHAR
+ | java.sql.Types.VARCHAR
+ | java.sql.Types.CHAR => StringType
+ case java.sql.Types.NUMERIC
+ | java.sql.Types.DECIMAL => DecimalType
+ case java.sql.Types.BIT => BooleanType
+ case java.sql.Types.TINYINT => ByteType
+ case java.sql.Types.SMALLINT => ShortType
+ case java.sql.Types.INTEGER => IntegerType
+ case java.sql.Types.BIGINT => LongType
+ case java.sql.Types.REAL => FloatType
+ case java.sql.Types.FLOAT
+ | java.sql.Types.DOUBLE => DoubleType
+ case java.sql.Types.LONGVARBINARY
+ | java.sql.Types.VARBINARY
+ | java.sql.Types.BINARY => BinaryType
+ case java.sql.Types.TIMESTAMP => TimestampType
+ case java.sql.Types.DATE
+ | java.sql.Types.TIME => StringType
+ case java.sql.Types.CLOB => StringType //should be just to string?
--- End diff --
Add `TODO` will be more helpful for the future improvement.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2710] [SQL] Build SchemaRDD from a Jdbc...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/1612#discussion_r17279042
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcResultSetRDD.scala ---
@@ -0,0 +1,72 @@
+/*
+ * 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.jdbc
+
+import java.sql.ResultSet
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.rdd.JdbcRDD
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.types._
+import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
+import org.apache.spark.Logging
+
+private[sql] object JdbcResultSetRDD extends Logging {
+
+ private[sql] def inferSchema(
+ jdbcResultSet: JdbcRDD[ResultSet]): StructType = {
+ StructType(createSchema(jdbcResultSet.getSchema))
+ }
+
+ private def createSchema(metaSchema: Seq[(String, Int, Boolean)]): Seq[StructField] = {
+ metaSchema.map(e => StructField(e._1, JdbcTypes.toPrimitiveDataType(e._2), e._3))
+ }
+
+ private[sql] def jdbcResultSetToRow(
+ jdbcResultSet: JdbcRDD[ResultSet],
+ schema: StructType) : RDD[Row] = {
+ val row = new GenericMutableRow(schema.fields.length)
+ jdbcResultSet.map(asRow(_, row, schema.fields))
--- End diff --
If you are are going to reuse the row object (which is a good idea), I'd use `mapPartitions` instead and create the object inside of the closure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org