From ac6bf9c3d4f7a165ba7e07618560363f335ae7e0 Mon Sep 17 00:00:00 2001
From: Muhammet Orazov <m.orazow@gmail.com>
Date: Thu, 9 Apr 2020 04:43:34 +0200
Subject: [PATCH] Remove `NOT NULL` constraint from String types. (#63)

We create an Exasol table, if it did not exist, before saving the Spark
dataframe. The `NOT NULL` constraint was added to the create table DDL,
if the Spark schema field type is not nullable.

However, this can be problem in Exasol side. Because, Exasol puts `null`
if the string is empty for `VARCHAR` or `CLOB` column types. Therefore,
putting not null constraints fails when inserting empty strings.

This commit removes the `NOT NULL` constraints from string types even if
they are not nullable.

Fixes #60.
---
 src/main/scala/com/exasol/spark/util/Types.scala  | 15 +++++++++------
 .../scala/com/exasol/spark/util/TypesSuite.scala  | 13 ++++++++++++-
 2 files changed, 21 insertions(+), 7 deletions(-)

diff --git a/src/main/scala/com/exasol/spark/util/Types.scala b/src/main/scala/com/exasol/spark/util/Types.scala
index 4a0af37c..3880a69f 100644
--- a/src/main/scala/com/exasol/spark/util/Types.scala
+++ b/src/main/scala/com/exasol/spark/util/Types.scala
@@ -218,8 +218,11 @@ object Types extends Logging {
   }
 
   /**
-   * Returns comma separated column name and column types for Exasol table from
-   * Spark schema.
+   * Returns comma separated column name and column types for Exasol
+   * table from Spark schema.
+   *
+   * It skips the `NOT NULL` constraint if the Spark dataframe schema
+   * type is a [[org.apache.spark.sql.types.StringType$]] type.
    *
    * @param schema A Spark [[org.apache.spark.sql.types.StructType]] schema
    * @return A comma separated column names and their types
@@ -229,10 +232,10 @@ object Types extends Logging {
       .map { field =>
         val fieldType = Types.exasolTypeFromSparkDataType(field.dataType)
         val nameType = s"${field.name} $fieldType"
-        if (field.nullable) {
-          nameType
-        } else {
-          nameType + " NOT NULL"
+        field.nullable match {
+          case true                                  => nameType
+          case false if field.dataType != StringType => nameType + " NOT NULL"
+          case false                                 => nameType
         }
       }
       .mkString(", ")
diff --git a/src/test/scala/com/exasol/spark/util/TypesSuite.scala b/src/test/scala/com/exasol/spark/util/TypesSuite.scala
index 7dc79425..34bb211c 100644
--- a/src/test/scala/com/exasol/spark/util/TypesSuite.scala
+++ b/src/test/scala/com/exasol/spark/util/TypesSuite.scala
@@ -156,7 +156,7 @@ class TypesSuite extends FunSuite with Matchers {
     )
   }
 
-  test("`createTableSchema` should create a comma separated column names and types") {
+  test("createTableSchema should create a comma separated column names and types") {
     val schema: StructType = StructType(
       Seq(
         StructField("bool_col", BooleanType),
@@ -176,4 +176,15 @@ class TypesSuite extends FunSuite with Matchers {
     assert(createTableSchema(schema) === expectedStr)
   }
 
+  test("createTableSchema returns string type without not null constraint") {
+    val schema = StructType(
+      Seq(
+        StructField("str_col", StringType),
+        StructField("text_col", StringType, false)
+      )
+    )
+    val expectedConvertedSchema = "str_col CLOB, text_col CLOB"
+    assert(createTableSchema(schema) === expectedConvertedSchema)
+  }
+
 }