[DE-210] Feature/ignore null fields (#28)

* ignoreNullFields

* mapping options

* Spark 2.4

* test keep null in arrays

Author: rashtao

arangodb-spark-commons/src/main/scala/org/apache/spark/sql/arangodb/commons/ArangoDBConf.scala

@@ -232,6 +232,12 @@ object ArangoDBConf {
     .intConf
     .createWithDefault(DEFAULT_MAX_RETRY_DELAY)
 
+  val IGNORE_NULL_FIELDS = "ignoreNullFields"
+  val ignoreNullFieldsConf: ConfigEntry[Boolean] = ConfigBuilder(IGNORE_NULL_FIELDS)
+    .doc("whether to ignore null fields during serialization (only supported in Spark 3.x)")
+    .booleanConf
+    .createWithDefault(false)
+
   private[sql] val confEntries: Map[String, ConfigEntry[_]] = CaseInsensitiveMap(Map(
     // driver config
     USER -> userConf,
@@ -272,7 +278,8 @@ object ArangoDBConf {
     KEEP_NULL -> keepNullConf,
     MAX_ATTEMPTS -> maxAttemptsConf,
     MIN_RETRY_DELAY -> minRetryDelayConf,
-    MAX_RETRY_DELAY -> maxRetryDelayConf
+    MAX_RETRY_DELAY -> maxRetryDelayConf,
+    IGNORE_NULL_FIELDS -> ignoreNullFieldsConf
   ))
 
   /**
@@ -317,6 +324,8 @@ object ArangoDBConf {
     CaseInsensitiveMap(Map(configs.map { cfg => cfg.key -> cfg }: _*))
   }
 
+  def apply(): ArangoDBConf = new ArangoDBConf(Map.empty)
+
   def apply(options: Map[String, String]): ArangoDBConf = new ArangoDBConf(options)
 
   def apply(options: util.Map[String, String]): ArangoDBConf = ArangoDBConf(options.asScala.toMap)
@@ -335,6 +344,7 @@ class ArangoDBConf(opts: Map[String, String]) extends Serializable with Logging
   lazy val driverOptions: ArangoDBDriverConf = new ArangoDBDriverConf(settings)
   lazy val readOptions: ArangoDBReadConf = new ArangoDBReadConf(settings)
   lazy val writeOptions: ArangoDBWriteConf = new ArangoDBWriteConf(settings)
+  lazy val mappingOptions: ArangoDBMappingConf = new ArangoDBMappingConf(settings)
 
   def updated(kv: (String, String)): ArangoDBConf = new ArangoDBConf(settings + kv)
 
@@ -559,7 +569,7 @@ class ArangoDBWriteConf(opts: Map[String, String]) extends ArangoDBConf(opts) {
 
   val maxRetryDelay: Int = getConf(maxRetryDelayConf)
 
-  override def toString =
+  override def toString: String =
     s"""ArangoDBWriteConf(
        |\t db=$db
        |\t collection=$collection
@@ -576,3 +586,16 @@ class ArangoDBWriteConf(opts: Map[String, String]) extends ArangoDBConf(opts) {
        |\t maxRetryDelay=$maxRetryDelay
        |)""".stripMargin
 }
+
+
+class ArangoDBMappingConf(opts: Map[String, String]) extends ArangoDBConf(opts) {
+
+  import ArangoDBConf._
+
+  val ignoreNullFields: Boolean = getConf(ignoreNullFieldsConf)
+
+  override def toString: String =
+    s"""ArangoDBMappingConf(
+       |\t ignoreNullFields=$ignoreNullFields
+       |)""".stripMargin
+}

arangodb-spark-commons/src/main/scala/org/apache/spark/sql/arangodb/commons/mapping/ArangoGeneratorProvider.scala

@@ -1,13 +1,13 @@
 package org.apache.spark.sql.arangodb.commons.mapping
 
-import org.apache.spark.sql.arangodb.commons.ContentType
+import org.apache.spark.sql.arangodb.commons.{ArangoDBConf, ContentType}
 import org.apache.spark.sql.types.StructType
 
 import java.io.OutputStream
 import java.util.ServiceLoader
 
 trait ArangoGeneratorProvider {
-  def of(contentType: ContentType, schema: StructType, outputStream: OutputStream): ArangoGenerator
+  def of(contentType: ContentType, schema: StructType, outputStream: OutputStream, conf: ArangoDBConf): ArangoGenerator
 }
 
 object ArangoGeneratorProvider {

arangodb-spark-commons/src/main/scala/org/apache/spark/sql/arangodb/commons/mapping/ArangoParserProvider.scala

@@ -1,12 +1,12 @@
 package org.apache.spark.sql.arangodb.commons.mapping
 
-import org.apache.spark.sql.arangodb.commons.ContentType
+import org.apache.spark.sql.arangodb.commons.{ArangoDBConf, ContentType}
 import org.apache.spark.sql.types.DataType
 
 import java.util.ServiceLoader
 
 trait ArangoParserProvider {
-  def of(contentType: ContentType, schema: DataType): ArangoParser
+  def of(contentType: ContentType, schema: DataType, conf: ArangoDBConf): ArangoParser
 }
 
 object ArangoParserProvider {

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/mapping/ArangoGeneratorImpl.scala

@@ -2,7 +2,7 @@ package org.apache.spark.sql.arangodb.datasource.mapping
 
 import com.arangodb.jackson.dataformat.velocypack.VPackFactory
 import com.fasterxml.jackson.core.JsonFactory
-import org.apache.spark.sql.arangodb.commons.ContentType
+import org.apache.spark.sql.arangodb.commons.{ArangoDBConf, ContentType}
 import org.apache.spark.sql.arangodb.datasource.mapping.json.{JSONOptions, JacksonGenerator}
 import org.apache.spark.sql.types.{DataType, StructType}
 import org.apache.spark.sql.arangodb.commons.mapping.{ArangoGenerator, ArangoGeneratorProvider}
@@ -19,23 +19,23 @@ abstract sealed class ArangoGeneratorImpl(
     options) with ArangoGenerator
 
 class ArangoGeneratorProviderImpl extends ArangoGeneratorProvider {
-  override def of(contentType: ContentType, schema: StructType, outputStream: OutputStream): ArangoGeneratorImpl = contentType match {
-    case ContentType.JSON => new JsonArangoGenerator(schema, outputStream)
-    case ContentType.VPACK => new VPackArangoGenerator(schema, outputStream)
+  override def of(contentType: ContentType, schema: StructType, outputStream: OutputStream, conf: ArangoDBConf): ArangoGeneratorImpl = contentType match {
+    case ContentType.JSON => new JsonArangoGenerator(schema, outputStream, conf)
+    case ContentType.VPACK => new VPackArangoGenerator(schema, outputStream, conf)
     case _ => throw new IllegalArgumentException
   }
 }
 
-class JsonArangoGenerator(schema: StructType, outputStream: OutputStream)
+class JsonArangoGenerator(schema: StructType, outputStream: OutputStream, conf: ArangoDBConf)
   extends ArangoGeneratorImpl(
     schema,
     outputStream,
-    createOptions(new JsonFactory())
+    createOptions(new JsonFactory(), conf)
   )
 
-class VPackArangoGenerator(schema: StructType, outputStream: OutputStream)
+class VPackArangoGenerator(schema: StructType, outputStream: OutputStream, conf: ArangoDBConf)
   extends ArangoGeneratorImpl(
     schema,
     outputStream,
-    createOptions(new VPackFactory())
+    createOptions(new VPackFactory(), conf)
   )

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/mapping/ArangoParserImpl.scala

@@ -3,7 +3,7 @@ package org.apache.spark.sql.arangodb.datasource.mapping
 import com.arangodb.jackson.dataformat.velocypack.VPackFactory
 import com.arangodb.velocypack.{VPackParser, VPackSlice}
 import com.fasterxml.jackson.core.{JsonFactory, JsonParser}
-import org.apache.spark.sql.arangodb.commons.ContentType
+import org.apache.spark.sql.arangodb.commons.{ArangoDBConf, ContentType}
 import org.apache.spark.sql.arangodb.commons.mapping.{ArangoParser, ArangoParserProvider}
 import org.apache.spark.sql.arangodb.datasource.mapping.json.{JSONOptions, JacksonParser}
 import org.apache.spark.sql.catalyst.InternalRow
@@ -23,24 +23,23 @@ abstract sealed class ArangoParserImpl(
 }
 
 class ArangoParserProviderImpl extends ArangoParserProvider {
-  override def of(contentType: ContentType, schema: DataType): ArangoParserImpl = contentType match {
-    case ContentType.JSON => new JsonArangoParser(schema)
-    case ContentType.VPACK => new VPackArangoParser(schema)
+  override def of(contentType: ContentType, schema: DataType, conf: ArangoDBConf): ArangoParserImpl = contentType match {
+    case ContentType.JSON => new JsonArangoParser(schema, conf)
+    case ContentType.VPACK => new VPackArangoParser(schema, conf)
     case _ => throw new IllegalArgumentException
   }
 }
 
-class JsonArangoParser(schema: DataType)
+class JsonArangoParser(schema: DataType, conf: ArangoDBConf)
   extends ArangoParserImpl(
     schema,
-    createOptions(new JsonFactory()
-      .configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true)),
+    createOptions(new JsonFactory().configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true), conf),
     (bytes: Array[Byte]) => UTF8String.fromBytes(bytes)
   )
 
-class VPackArangoParser(schema: DataType)
+class VPackArangoParser(schema: DataType, conf: ArangoDBConf)
   extends ArangoParserImpl(
     schema,
-    createOptions(new VPackFactory()),
+    createOptions(new VPackFactory(), conf),
     (bytes: Array[Byte]) => UTF8String.fromString(new VPackParser.Builder().build().toJson(new VPackSlice(bytes), true))
   )

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/mapping/package.scala

@@ -1,10 +1,11 @@
 package org.apache.spark.sql.arangodb.datasource
 
 import com.fasterxml.jackson.core.JsonFactory
+import org.apache.spark.sql.arangodb.commons.ArangoDBConf
 import org.apache.spark.sql.arangodb.datasource.mapping.json.JSONOptions
 
 package object mapping {
-  private[mapping] def createOptions(jsonFactory: JsonFactory) = new JSONOptions(
+  private[mapping] def createOptions(jsonFactory: JsonFactory, conf: ArangoDBConf) = new JSONOptions(
     Map.empty[String, String],
     "UTC"
   ) {

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/reader/ArangoCollectionPartitionReader.scala

@@ -24,7 +24,7 @@ class ArangoCollectionPartitionReader(
   // override endpoints with partition endpoint
   private val options = opts.updated(ArangoDBConf.ENDPOINTS, inputPartition.endpoint)
   private val actualSchema = StructType(ctx.requiredSchema.filterNot(_.name == options.readOptions.columnNameOfCorruptRecord))
-  private val parser = ArangoParserProvider().of(options.driverOptions.contentType, actualSchema)
+  private val parser = ArangoParserProvider().of(options.driverOptions.contentType, actualSchema, options)
   private val safeParser = new FailureSafeParser[Array[Byte]](
     parser.parse(_).toSeq,
     options.readOptions.parseMode,

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/reader/ArangoQueryReader.scala

@@ -18,7 +18,7 @@ class ArangoQueryReader(schema: StructType, options: ArangoDBConf) extends Input
   with Logging {
 
   private val actualSchema = StructType(schema.filterNot(_.name == options.readOptions.columnNameOfCorruptRecord))
-  private val parser = ArangoParserProvider().of(options.driverOptions.contentType, actualSchema)
+  private val parser = ArangoParserProvider().of(options.driverOptions.contentType, actualSchema, options)
   private val safeParser = new FailureSafeParser[Array[Byte]](
     parser.parse(_).toSeq,
     options.readOptions.parseMode,

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/writer/ArangoDataSourceWriter.scala

@@ -92,6 +92,7 @@ class ArangoDataSourceWriter(writeUUID: String, schema: StructType, mode: SaveMo
 
     logInfo(s"Using save mode: $mode")
     logInfo(s"Using write configuration: \n${updatedOptions.writeOptions}")
+    logInfo(s"Using mapping configuration: \n${updatedOptions.mappingOptions}")
     logInfo(s"Writing schema: \n${schema.treeString}")
     logInfo(s"Can retry: $canRetry")
 

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/writer/ArangoDataWriter.scala

@@ -66,7 +66,7 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
   private def initBatch(): Unit = {
     batchCount = 0
     outVPack = new ByteArrayOutputStream()
-    vpackGenerator = ArangoGeneratorProvider().of(options.driverOptions.contentType, schema, outVPack)
+    vpackGenerator = ArangoGeneratorProvider().of(options.driverOptions.contentType, schema, outVPack, options)
     vpackGenerator.writeStartArray()
   }