Sync to master

Conflicts:
	sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
	sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

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

@@ -133,8 +133,12 @@ object JdbcDialects {
 
   registerDialect(MySQLDialect)
   registerDialect(PostgresDialect)
+<<<<<<< HEAD
   registerDialect(OracleDialect)
   registerDialect(NetezzaDialect)
+=======
+  registerDialect(DB2Dialect)
+>>>>>>> upstream/master
 
   /**
    * Fetch the JdbcDialect class corresponding to a given database url.
@@ -235,9 +239,26 @@ case object MySQLDialect extends JdbcDialect {
 
 /**
  * :: DeveloperApi ::
- * Default Oracle dialect, mapping string/boolean on write to valid Oracle types.
+ * 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] = dt match {
+    case StringType => Some(JdbcType("CLOB", java.sql.Types.CLOB))
+    case BooleanType => Some(JdbcType("CHAR(1)", java.sql.Types.CHAR))
+    case _ => 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")
@@ -273,4 +294,4 @@ case object NetezzaDialect extends JdbcDialect {
       Some(JdbcType("BOOLEAN", java.sql.Types.BOOLEAN))
     } else None
   }
-}
+}

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

@@ -409,6 +409,7 @@ class JDBCSuite extends SparkFunSuite with BeforeAndAfter with SharedSQLContext
     assert(JdbcDialects.get("jdbc:postgresql://127.0.0.1/db") == PostgresDialect)
     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("jdbc:db2://127.0.0.1/db") == DB2Dialect)
     assert(JdbcDialects.get("test.invalid") == NoopDialect)
   }