Code style

Author: jacobfi

lib/sqlite4java/libsqlite4java-osx.dylib

@@ -178,15 +178,14 @@ private[dynamodb] class TableConnector(tableName: String, parallelism: Int, para
 
         val tableWriteItems = new TableWriteItems(tableName)
         val tableWriteItemsWithItems: TableWriteItems =
-            // check if hash key only or also range key
+        // Check if hash key only or also range key.
             columnSchema.keys() match {
                 case Left((hashKey, hashKeyIndex, hashKeyType)) =>
                     val hashKeys = items.map(row =>
                         JavaConverter.convertRowValue(row, hashKeyIndex, hashKeyType).asInstanceOf[AnyRef])
                     tableWriteItems.withHashOnlyKeysToDelete(hashKey, hashKeys: _*)
-
                 case Right(((hashKey, hashKeyIndex, hashKeyType), (rangeKey, rangeKeyIndex, rangeKeyType))) =>
-                    val alternatingHashAndRangeKeys = items.flatMap { case row =>
+                    val alternatingHashAndRangeKeys = items.flatMap { row =>
                         val hashKeyValue = JavaConverter.convertRowValue(row, hashKeyIndex, hashKeyType)
                         val rangeKeyValue = JavaConverter.convertRowValue(row, rangeKeyIndex, rangeKeyType)
                         Seq(hashKeyValue.asInstanceOf[AnyRef], rangeKeyValue.asInstanceOf[AnyRef])

src/main/scala/com/audienceproject/spark/dynamodb/connector/TableConnector.scala

@@ -0,0 +1,40 @@
+/**
+  * Licensed to the Apache Software Foundation (ASF) under one
+  * or more contributor license agreements.  See the NOTICE file
+  * distributed with this work for additional information
+  * regarding copyright ownership.  The ASF licenses this file
+  * to you under the Apache License, Version 2.0 (the
+  * "License"); you may not use this file except in compliance
+  * with the License.  You may obtain a copy of the License at
+  *
+  * http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing,
+  * software distributed under the License is distributed on an
+  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  * KIND, either express or implied.  See the License for the
+  * specific language governing permissions and limitations
+  * under the License.
+  *
+  * Copyright © 2020 AudienceProject. All rights reserved.
+  */
+
+package com.audienceproject.spark.dynamodb.datasource
+
+import com.amazonaws.services.dynamodbv2.document.DynamoDB
+import com.audienceproject.spark.dynamodb.connector.{ColumnSchema, TableConnector}
+
+class DynamoBatchDeleteWriter(batchSize: Int,
+                              columnSchema: ColumnSchema,
+                              connector: TableConnector,
+                              client: DynamoDB)
+    extends DynamoBatchWriter(batchSize, columnSchema, connector, client) {
+
+    protected override def flush(): Unit = {
+        if (buffer.nonEmpty) {
+            connector.deleteItems(columnSchema, buffer)(client, rateLimiter)
+            buffer.clear()
+        }
+    }
+
+}

src/main/scala/com/audienceproject/spark/dynamodb/datasource/DynamoBatchDeleteWriter.scala

@@ -31,12 +31,11 @@ import scala.collection.mutable.ArrayBuffer
 class DynamoBatchWriter(batchSize: Int,
                         columnSchema: ColumnSchema,
                         connector: TableConnector,
-                        client: DynamoDB
-                       )
+                        client: DynamoDB)
     extends DataWriter[InternalRow] {
 
-    protected val buffer = new ArrayBuffer[InternalRow](batchSize)
-    protected val rateLimiter = RateLimiter.create(connector.writeLimit)
+    protected val buffer: ArrayBuffer[InternalRow] = new ArrayBuffer[InternalRow](batchSize)
+    protected val rateLimiter: RateLimiter = RateLimiter.create(connector.writeLimit)
 
     override def write(record: InternalRow): Unit = {
         buffer += record.copy()

src/main/scala/com/audienceproject/spark/dynamodb/datasource/DynamoBatchDeleter.scala

@@ -43,12 +43,11 @@ class DynamoWriterFactory(connector: TableConnector,
         if (update) {
             assert(!delete, "Please provide exactly one of 'update' or 'delete' options.")
             new DynamoUpdateWriter(columnSchema, connector, client)
-        }
-        else if (delete) {
-            new DynamoBatchDeleter(batchSize, columnSchema, connector, client)
-        }
-        else
+        } else if (delete) {
+            new DynamoBatchDeleteWriter(batchSize, columnSchema, connector, client)
+        } else {
             new DynamoBatchWriter(batchSize, columnSchema, connector, client)
+        }
     }
 
 }