[SPARK-10101] [SQL] Add maxlength to JDBC field metadata and override JDBCDialects for strings as VARCHAR.

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -161,13 +161,15 @@ object JdbcUtils extends Logging {
   /**
    * Compute the schema string for this RDD.
    */
-  def schemaString(df: DataFrame, url: String): String = {
+  def schemaString(
+                    df: DataFrame,
+                    url: String): String = {
     val sb = new StringBuilder()
     val dialect = JdbcDialects.get(url)
     df.schema.fields foreach { field => {
       val name = field.name
       val typ: String =
-        dialect.getJDBCType(field.dataType).map(_.databaseTypeDefinition).getOrElse(
+        dialect.getJDBCType(field.dataType, field.metadata).map(_.databaseTypeDefinition).getOrElse(
           field.dataType match {
             case IntegerType => "INTEGER"
             case LongType => "BIGINT"
@@ -199,7 +201,7 @@ object JdbcUtils extends Logging {
       properties: Properties = new Properties()) {
     val dialect = JdbcDialects.get(url)
     val nullTypes: Array[Int] = df.schema.fields.map { field =>
-      dialect.getJDBCType(field.dataType).map(_.jdbcNullType).getOrElse(
+      dialect.getJDBCType(field.dataType, field.metadata).map(_.jdbcNullType).getOrElse(
         field.dataType match {
           case IntegerType => java.sql.Types.INTEGER
           case LongType => java.sql.Types.BIGINT

sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala

@@ -77,9 +77,10 @@ abstract class JdbcDialect {
   /**
    * Retrieve the jdbc / sql type for a given datatype.
    * @param dt The datatype (e.g. [[org.apache.spark.sql.types.StringType]])
+   * @param md The metadata
    * @return The new JdbcType if there is an override for this DataType
    */
-  def getJDBCType(dt: DataType): Option[JdbcType] = None
+  def getJDBCType(dt: DataType, md: Metadata): Option[JdbcType] = None
 
   /**
    * Quotes the identifier. This is used to put quotes around the identifier in case the column
@@ -125,6 +126,9 @@ object JdbcDialects {
 
   registerDialect(MySQLDialect)
   registerDialect(PostgresDialect)
+  registerDialect(OracleDialect)
+  registerDialect(DB2Dialect)
+  registerDialect(NetezzaDialect)
 
   /**
    * Fetch the JdbcDialect class corresponding to a given database url.
@@ -159,8 +163,8 @@ class AggregatedDialect(dialects: List[JdbcDialect]) extends JdbcDialect {
     dialects.flatMap(_.getCatalystType(sqlType, typeName, size, md)).headOption
   }
 
-  override def getJDBCType(dt: DataType): Option[JdbcType] = {
-    dialects.flatMap(_.getJDBCType(dt)).headOption
+  override def getJDBCType(dt: DataType, md: Metadata): Option[JdbcType] = {
+    dialects.flatMap(_.getJDBCType(dt, md)).headOption
   }
 }
 
@@ -191,7 +195,7 @@ case object PostgresDialect extends JdbcDialect {
     } else None
   }
 
-  override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
+  override def getJDBCType(dt: DataType, md: Metadata): Option[JdbcType] = dt match {
     case StringType => Some(JdbcType("TEXT", java.sql.Types.CHAR))
     case BinaryType => Some(JdbcType("BYTEA", java.sql.Types.BINARY))
     case BooleanType => Some(JdbcType("BOOLEAN", java.sql.Types.BOOLEAN))
@@ -222,3 +226,66 @@ case object MySQLDialect extends JdbcDialect {
     s"`$colName`"
   }
 }
+
+/**
+ * :: DeveloperApi ::
+ * Default DB2 dialect, mapping string/boolean on write to valid DB2 types.
+ * By default string, and boolean gets mapped to db2 invalid types TEXT, and BIT(1).
+ */
+@DeveloperApi
+case object DB2Dialect extends JdbcDialect {
+
+  override def canHandle(url: String): Boolean = url.startsWith("jdbc:db2")
+
+  override def getJDBCType(dt: DataType, md: Metadata): Option[JdbcType] = {
+    if (dt == StringType && md.contains("maxlength")) {
+      Some(JdbcType(s"VARCHAR(${md.getLong("maxlength")})", java.sql.Types.CHAR))
+    } else if (dt == StringType ) {
+      Some(JdbcType("CLOB", java.sql.Types.CLOB))
+    } else if (dt == BooleanType ) {
+      Some(JdbcType("CHAR(1)", java.sql.Types.CHAR))
+    } else None
+  }
+}
+
+/**
+ * :: DeveloperApi ::
+ * Default Oracle dialect, mapping string/boolean on write to valid Oracle types.
+ */
+@DeveloperApi
+case object OracleDialect extends JdbcDialect {
+
+  override def canHandle(url: String): Boolean = url.startsWith("jdbc:oracle")
+
+  override def getJDBCType(dt: DataType, md: Metadata): Option[JdbcType] = {
+    if (dt == StringType && md.contains("maxlength")) {
+      Some(JdbcType(s"VARCHAR(${md.getLong("maxlength")})", java.sql.Types.CHAR))
+    } else if (dt == StringType ) {
+      Some(JdbcType("CLOB", java.sql.Types.CLOB))
+    } else if (dt == BooleanType ) {
+      Some(JdbcType("CHAR(1)", java.sql.Types.CHAR))
+    } else None
+  }
+}
+
+/**
+ * :: DeveloperApi ::
+ * Default Netezza dialect, mapping string/boolean on write to valid Netezza types.
+ */
+@DeveloperApi
+case object NetezzaDialect extends JdbcDialect {
+
+  override def canHandle(url: String): Boolean = url.startsWith("jdbc:netezza")
+
+  override def getJDBCType(dt: DataType, md: Metadata): Option[JdbcType] = {
+    if (dt == StringType && md.contains("maxlength")) {
+      Some(JdbcType(s"VARCHAR(${md.getLong("maxlength")})", java.sql.Types.CHAR))
+    } else if (dt == StringType ) {
+      Some(JdbcType("VARCHAR(255)", java.sql.Types.CHAR))
+    } else if (dt == BinaryType ) {
+      Some(JdbcType("BYTEINT", java.sql.Types.BINARY))
+    } else if (dt == BooleanType ) {
+      Some(JdbcType("BOOLEAN", java.sql.Types.BOOLEAN))
+    } else None
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -407,6 +407,9 @@ class JDBCSuite extends SparkFunSuite with BeforeAndAfter with SharedSQLContext
   test("Default jdbc dialect registration") {
     assert(JdbcDialects.get("jdbc:mysql://127.0.0.1/db") == MySQLDialect)
     assert(JdbcDialects.get("jdbc:postgresql://127.0.0.1/db") == PostgresDialect)
+    assert(JdbcDialects.get("jdbc:db2://127.0.0.1/db") == DB2Dialect)
+    assert(JdbcDialects.get("jdbc:oracle://127.0.0.1/db") == OracleDialect)
+    assert(JdbcDialects.get("jdbc:netezza://127.0.0.1/db") == NetezzaDialect)
     assert(JdbcDialects.get("test.invalid") == NoopDialect)
   }