package org.embulk.input.dynamodb.operation

import com.amazonaws.services.dynamodbv2.model.QueryRequest
import org.embulk.config.{Config, ConfigDefault}

import scala.util.chaining._

object DynamodbQueryOperation {

  trait Task extends AbstractDynamodbOperation.Task {

    // ref. https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Query.html#Query.KeyConditionExpressions
    @Config("key_condition_expression")
    def getKeyConditionExpression: String

    // TODO: Is it needed in the embulk context?
    // ref. https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Query.html#Query.KeyConditionExpressions
    @Config("scan_index_forward")
    @ConfigDefault("true")
    def getScanIndexForward: Boolean

  }
}

case class DynamodbQueryOperation(task: DynamodbQueryOperation.Task)
    extends AbstractDynamodbOperation[QueryRequest](task) {

  def newRequest: QueryRequest = {
    new QueryRequest()
      .tap(configureRequest)
      .tap(r => r.setKeyConditionExpression(task.getKeyConditionExpression))
      .tap(r => r.setScanIndexForward(task.getScanIndexForward))
  }

}

package org.embulk.input.dynamodb.operation

import com.amazonaws.services.dynamodbv2.model.QueryRequest
import org.embulk.config.{Config, ConfigDefault}

import scala.util.chaining._

object DynamodbQueryOperation {

  trait Task extends DynamodbOperationCommonOptions.Task {

    // ref. https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Query.html#Query.KeyConditionExpressions
    @Config("key_condition_expression")
    def getKeyConditionExpression: String

    // TODO: Is it needed in the embulk context?
    // ref. https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Query.html#Query.KeyConditionExpressions
    @Config("scan_index_forward")
    @ConfigDefault("true")
    def getScanIndexForward: Boolean

  }
}

case class DynamodbQueryOperation(task: DynamodbQueryOperation.Task) {

  def newRequest: QueryRequest = {
    new QueryRequest()
      .tap(r => DynamodbOperationCommonOptions.configureRequest(r, task))
      .tap(r => r.setKeyConditionExpression(task.getKeyConditionExpression))
      .tap(r => r.setScanIndexForward(task.getScanIndexForward))
  }

  def getEmbulkTaskCount: Int = 1
}

--- 
+++ 
@@ -7,7 +7,7 @@
 
 object DynamodbQueryOperation {
 
-  trait Task extends DynamodbOperationCommonOptions.Task {
+  trait Task extends AbstractDynamodbOperation.Task {
 
     // ref. https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Query.html#Query.KeyConditionExpressions
     @Config("key_condition_expression")
@@ -22,14 +22,14 @@
   }
 }
 
-case class DynamodbQueryOperation(task: DynamodbQueryOperation.Task) {
+case class DynamodbQueryOperation(task: DynamodbQueryOperation.Task)
+    extends AbstractDynamodbOperation[QueryRequest](task) {
 
   def newRequest: QueryRequest = {
     new QueryRequest()
-      .tap(r => DynamodbOperationCommonOptions.configureRequest(r, task))
+      .tap(configureRequest)
       .tap(r => r.setKeyConditionExpression(task.getKeyConditionExpression))
       .tap(r => r.setScanIndexForward(task.getScanIndexForward))
   }
 
-  def getEmbulkTaskCount: Int = 1
 }
Revert "Delete AbstractDynamodbOperation, use DynamodbOperationCommonOptions instead."

This reverts commit 3c6821bf
