Skip to content

[SPARK-12010][SQL] Spark JDBC requires support for column-name-free INSERT syntax #10380

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
CK50 wants to merge 7 commits into from
Closed

[SPARK-12010][SQL] Spark JDBC requires support for column-name-free INSERT syntax #10380

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
3211963
Initial version of suggested simple approach
CK50 Dec 18, 2015
0772db3
comment removed
CK50 Dec 18, 2015
a59c1aa
dissolved insertStatement method, as only used once
CK50 Dec 18, 2015
c130e31
Reintroduced insertStatement method
CK50 Dec 22, 2015
79d3e0d
Removed dialect parameter from insertStatement method and
CK50 Dec 22, 2015
cae0f58
Removing obsolete method from JdbcDialects.scala
CK50 Dec 22, 2015
5a7f262
Final trimming
CK50 Dec 22, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
Initial version of suggested simple approach
  • Loading branch information
@CK50
CK50 committed Dec 18, 2015
commit 3211963805edff521a08f69193aaf2b8a09b85f8
18 changes: 8 additions & 10 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,15 +62,12 @@ object JdbcUtils extends Logging {
/**
* Returns a PreparedStatement that inserts a row into table via conn.
Copy link
Member

@srowen srowen Dec 20, 2015

Choose a reason for hiding this comment

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

Hm, the only problem here is that this is a public method, and while it feels like it was intended to be a Spark-only utility method, I'm not sure it's marked as such.

It's not a big deal to retain it and implement in terms of the new method. However it's now a function of a dialect, which is not an argument here. I suppose any dialect will do since they all behave the same now. This method could then be deprecated.

However: yeah, the behavior is actually the same for all dialects now. Really, this has come full circle and can just be a modification to this method, which was already the same for all dialects. Is there reason to believe the insert statement might vary later? Then I could see keeping the current structure here and just deprecating this method.

Copy link
Contributor

@hvanhovell hvanhovell Dec 20, 2015

Choose a reason for hiding this comment

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

Yeah, you are right about that. We'll break a public API. Is that a problem, since this is probably going into 2.0?

Copy link
Member

@srowen srowen Dec 21, 2015

Choose a reason for hiding this comment

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

That's a fair point, though I think the intent was to back-port this to 1.6.x at least, as it's a moderately important fix. Conceptually, I don't think the API has changed; the insert statement is still not dialect-specific. Hence it seems like the current API is even desirable to maintain for now.

Copy link
Contributor

@hvanhovell hvanhovell Dec 21, 2015

Choose a reason for hiding this comment

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

Ok, we have keep the method if we are back-porting. I have yet to encouter a database that doesn't support this insert syntax (did a bit more googling); so it seems safe to put in the generic method.

*/
def insertStatement(conn: Connection, table: String, rddSchema: StructType): PreparedStatement = {
val sql = new StringBuilder(s"INSERT INTO $table VALUES (")
var fieldsLeft = rddSchema.fields.length
while (fieldsLeft > 0) {
sql.append("?")
if (fieldsLeft > 1) sql.append(", ") else sql.append(")")
fieldsLeft = fieldsLeft - 1
}
conn.prepareStatement(sql.toString())
def insertStatement(conn: Connection,
Copy link
Contributor

@hvanhovell hvanhovell Dec 18, 2015

Choose a reason for hiding this comment

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

Nit: This method is only used by the savePartition we could integrate it, since most of the code was moved to JdbcDialects.getInsertStatement.

Copy link
Contributor Author

@CK50 CK50 Dec 18, 2015

Choose a reason for hiding this comment

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

@hvanhovell
It works fine on Oracle and on Cassandra (Progress JDBC driver for Cassandra).
Commenting on other RDBMS: I was surprised that the column-free syntax is supported on so many databases. From all my work on different RDBMS, the syntax with column-names is much more the standard. - But I have not tested on other than Oracle and Cassandra.

Copy link
Contributor Author

@CK50 CK50 Dec 18, 2015

Choose a reason for hiding this comment

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

@hvanhovell
re integrating JdbcDialects.getInsertStatement: I can certainly do so, if desired.

Copy link
Contributor

@hvanhovell hvanhovell Dec 18, 2015

Choose a reason for hiding this comment

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

@CK50 I don't have a strong opinion on this. I'd personally remove it, that's all.

dialect: JdbcDialect,
table: String,
rddSchema: StructType): PreparedStatement = {
val sql = dialect.getInsertStatement(table, rddSchema)
conn.prepareStatement(sql)
}

/**
Expand Down Expand Up @@ -126,6 +123,7 @@ object JdbcUtils extends Logging {
dialect: JdbcDialect): Iterator[Byte] = {
val conn = getConnection()
var committed = false
// Temp work-around. Not to be commited!!!!
val supportsTransactions = try {
conn.getMetaData().supportsDataManipulationTransactionsOnly() ||
conn.getMetaData().supportsDataDefinitionAndDataManipulationTransactions()
Expand All @@ -139,7 +137,7 @@ object JdbcUtils extends Logging {
if (supportsTransactions) {
conn.setAutoCommit(false) // Everything in the same db transaction.
}
val stmt = insertStatement(conn, table, rddSchema)
val stmt = insertStatement(conn, dialect, table, rddSchema)
try {
var rowCount = 0
while (iterator.hasNext) {
Expand Down
18 changes: 18 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,24 @@ abstract class JdbcDialect extends Serializable {
def beforeFetch(connection: Connection, properties: Map[String, String]): Unit = {
}

/**
* Get the SQL statement that should be used to insert new records into the table.
* Dialects can override this method to return a statement that works best in a particular
* database.
* @param table The name of the table.
* @param rddSchema The schema of DataFrame to be inserted
* @param columnMapping An optional mapping from DataFrame field names to database column
* names
* @return The SQL statement to use for inserting into the table.
*/
def getInsertStatement(table: String,
rddSchema: StructType): String = {
rddSchema.fields.map(field => field.name)
.mkString(s"INSERT INTO $table ( ", ", ", " ) " ) +
rddSchema.fields.map(field => "?")
.mkString("VALUES ( ", ", ", " )" )
}

}

/**
Expand Down