From 2941fa666c7ac111776b8cbfb47001f7ff2d2431 Mon Sep 17 00:00:00 2001
From: Kate Ting <kathleen@apache.org>
Date: Wed, 15 May 2013 00:51:28 -0400
Subject: [PATCH] SQOOP-1046: Sqoop2: Importing single row using decimal
 splitter will not import anything

(Jarek Jarcec Cecho via Kate Ting)
---
 .../jdbc/GenericJdbcImportPartitioner.java    | 18 +++++++++++++++++-
 .../connector/jdbc/TestImportPartitioner.java | 19 +++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcImportPartitioner.java b/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcImportPartitioner.java
index 6d1a9fd7..f80f30dc 100644
--- a/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcImportPartitioner.java
+++ b/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcImportPartitioner.java
@@ -169,7 +169,7 @@ protected List<Partition> partitionFloatingPointColumn() {
   protected List<Partition> partitionNumericColumn() {
     List<Partition> partitions = new LinkedList<Partition>();
 
-    // All null valeus will result in single partition
+    // All null values will result in single partition
     if (partitionMinValue == null && partitionMaxValue == null) {
       GenericJdbcImportPartition partition = new GenericJdbcImportPartition();
       partition.setConditions(partitionColumnName + "IS NULL");
@@ -185,6 +185,13 @@ protected List<Partition> partitionNumericColumn() {
     BigDecimal minValue = new BigDecimal(partitionMinValue);
     BigDecimal maxValue = new BigDecimal(partitionMaxValue);
 
+    // Having one single value means that we can create only one single split
+    if(minValue.equals(maxValue)) {
+      GenericJdbcImportPartition partition = new GenericJdbcImportPartition();
+      partition.setConditions(constructConditions(minValue));
+      partitions.add(partition);
+    }
+
     // Get all the split points together.
     List<BigDecimal> splitPoints = new LinkedList<BigDecimal>();
 
@@ -240,4 +247,13 @@ protected String constructConditions(
     conditions.append(upperBound);
     return conditions.toString();
   }
+
+  protected String constructConditions(Object value) {
+    return new StringBuilder()
+      .append(partitionColumnName)
+      .append(" = ")
+      .append(value)
+      .toString()
+     ;
+  }
 }
diff --git a/connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestImportPartitioner.java b/connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestImportPartitioner.java
index 3150e7ce..ee314d04 100644
--- a/connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestImportPartitioner.java
+++ b/connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestImportPartitioner.java
@@ -238,6 +238,25 @@ public void testNumericUnevenPartition() throws Exception {
     });
   }
 
+  public void testNumericSinglePartition() throws Exception {
+    MutableContext context = new MutableMapContext();
+    context.setString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_COLUMNNAME, "DCOL");
+    context.setString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_COLUMNTYPE, String.valueOf(Types.NUMERIC));
+    context.setString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_MINVALUE, String.valueOf(new BigDecimal(START)));
+    context.setString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_MAXVALUE, String.valueOf(new BigDecimal(START)));
+
+    ConnectionConfiguration connConf = new ConnectionConfiguration();
+    ImportJobConfiguration jobConf = new ImportJobConfiguration();
+
+    Partitioner partitioner = new GenericJdbcImportPartitioner();
+    PartitionerContext partitionerContext = new PartitionerContext(context, 3);
+    List<Partition> partitions = partitioner.getPartitions(partitionerContext, connConf, jobConf);
+
+    verifyResult(partitions, new String[]{
+      "DCOL = -5",
+    });
+  }
+
   private void verifyResult(List<Partition> partitions,
       String[] expected) {
     assertEquals(expected.length, partitions.size());