You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sqoop.apache.org by ja...@apache.org on 2013/07/21 19:28:34 UTC
git commit: SQOOP-1049: Sqoop2: Record not imported if partition
column value is NULL
Updated Branches:
refs/heads/sqoop2 abb8cf875 -> 91b121cca
SQOOP-1049: Sqoop2: Record not imported if partition column value is NULL
(Mengwei Ding via Jarek Jarcec Cecho)
Project: http://git-wip-us.apache.org/repos/asf/sqoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/sqoop/commit/91b121cc
Tree: http://git-wip-us.apache.org/repos/asf/sqoop/tree/91b121cc
Diff: http://git-wip-us.apache.org/repos/asf/sqoop/diff/91b121cc
Branch: refs/heads/sqoop2
Commit: 91b121ccaa9e40dc8a86b3a00d68300e7ba4c890
Parents: abb8cf8
Author: Jarek Jarcec Cecho <ja...@apache.org>
Authored: Sun Jul 21 10:27:55 2013 -0700
Committer: Jarek Jarcec Cecho <ja...@apache.org>
Committed: Sun Jul 21 10:27:55 2013 -0700
----------------------------------------------------------------------
.../java/org/apache/sqoop/model/FormUtils.java | 8 +++--
.../jdbc/GenericJdbcImportPartitioner.java | 34 ++++++++++++++++----
.../jdbc/configuration/ImportTableForm.java | 1 +
.../generic-jdbc-connector-resources.properties | 4 +++
.../connector/jdbc/TestImportPartitioner.java | 30 +++++++++++++++++
5 files changed, 69 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/sqoop/blob/91b121cc/common/src/main/java/org/apache/sqoop/model/FormUtils.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/sqoop/model/FormUtils.java b/common/src/main/java/org/apache/sqoop/model/FormUtils.java
index f3dee8e..27db8af 100644
--- a/common/src/main/java/org/apache/sqoop/model/FormUtils.java
+++ b/common/src/main/java/org/apache/sqoop/model/FormUtils.java
@@ -363,7 +363,9 @@ public class FormUtils {
jsonForm.put(inputName, value);
} else if(type.isEnum()) {
jsonForm.put(inputName, value.toString());
- } else {
+ } else if(type == Boolean.class) {
+ jsonForm.put(inputName, value);
+ }else {
throw new SqoopException(ModelError.MODEL_004,
"Unsupported type " + type.getName() + " for input " + formName + "." + inputName);
}
@@ -450,7 +452,9 @@ public class FormUtils {
inputField.set(formValue, ((Long)jsonInputs.get(inputName)).intValue());
} else if(type.isEnum()) {
inputField.set(formValue, Enum.valueOf((Class<? extends Enum>) inputField.getType(), (String) jsonInputs.get(inputName)));
- } else {
+ } else if(type == Boolean.class) {
+ inputField.set(formValue, (Boolean) jsonInputs.get(inputName));
+ }else {
throw new SqoopException(ModelError.MODEL_004,
"Unsupported type " + type.getName() + " for input " + formName + "." + inputName);
}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/91b121cc/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/GenericJdbcImportPartitioner.java
----------------------------------------------------------------------
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 7792c57..8d0c4ab 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
@@ -45,41 +45,58 @@ public class GenericJdbcImportPartitioner extends Partitioner<ConnectionConfigur
private int partitionColumnType;
private String partitionMinValue;
private String partitionMaxValue;
+ private Boolean partitionColumnNull;
@Override
public List<Partition> getPartitions(PartitionerContext context,ConnectionConfiguration connection, ImportJobConfiguration job) {
+ List<Partition> partitions = new LinkedList<Partition>();
+
numberPartitions = context.getMaxPartitions();
partitionColumnName = context.getString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_COLUMNNAME);
partitionColumnType = context.getInt(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_COLUMNTYPE, -1);
partitionMinValue = context.getString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_MINVALUE);
partitionMaxValue = context.getString(GenericJdbcConnectorConstants.CONNECTOR_JDBC_PARTITION_MAXVALUE);
+ partitionColumnNull = job.table.partitionColumnNull;
+ if (partitionColumnNull == null) {
+ partitionColumnNull = false;
+ }
+
if (partitionMinValue == null && partitionMaxValue == null) {
- List<Partition> partitions = new LinkedList<Partition>();
GenericJdbcImportPartition partition = new GenericJdbcImportPartition();
partition.setConditions(partitionColumnName + " IS NULL");
partitions.add(partition);
return partitions;
}
+ if (partitionColumnNull) {
+ GenericJdbcImportPartition partition = new GenericJdbcImportPartition();
+ partition.setConditions(partitionColumnName + " IS NULL");
+ partitions.add(partition);
+ numberPartitions -= 1;
+ }
+
switch (partitionColumnType) {
case Types.TINYINT:
case Types.SMALLINT:
case Types.INTEGER:
case Types.BIGINT:
// Integer column
- return partitionIntegerColumn();
+ partitions.addAll(partitionIntegerColumn());
+ break;
case Types.REAL:
case Types.FLOAT:
case Types.DOUBLE:
// Floating point column
- return partitionFloatingPointColumn();
+ partitions.addAll(partitionFloatingPointColumn());
+ break;
case Types.NUMERIC:
case Types.DECIMAL:
// Decimal column
- return partitionNumericColumn();
+ partitions.addAll(partitionNumericColumn());
+ break;
case Types.BIT:
case Types.BOOLEAN:
@@ -90,20 +107,25 @@ public class GenericJdbcImportPartitioner extends Partitioner<ConnectionConfigur
case Types.TIME:
case Types.TIMESTAMP:
// Date time column
- return partitionDateTimeColumn();
+ partitions.addAll(partitionDateTimeColumn());
+ break;
case Types.CHAR:
case Types.VARCHAR:
case Types.LONGVARCHAR:
// Text column
- return partitionTextColumn();
+ partitions.addAll(partitionTextColumn());
+ break;
default:
throw new SqoopException(
GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0011,
String.valueOf(partitionColumnType));
}
+
+ return partitions;
}
+
protected List<Partition> partitionDateTimeColumn() {
List<Partition> partitions = new LinkedList<Partition>();
http://git-wip-us.apache.org/repos/asf/sqoop/blob/91b121cc/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/configuration/ImportTableForm.java
----------------------------------------------------------------------
diff --git a/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/configuration/ImportTableForm.java b/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/configuration/ImportTableForm.java
index ef27236..0991b28 100644
--- a/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/configuration/ImportTableForm.java
+++ b/connector/connector-generic-jdbc/src/main/java/org/apache/sqoop/connector/jdbc/configuration/ImportTableForm.java
@@ -30,5 +30,6 @@ public class ImportTableForm {
@Input(size = 2000) public String sql;
@Input(size = 50) public String columns;
@Input(size = 50) public String partitionColumn;
+ @Input public Boolean partitionColumnNull;
@Input(size = 50) public String boundaryQuery;
}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/91b121cc/connector/connector-generic-jdbc/src/main/resources/generic-jdbc-connector-resources.properties
----------------------------------------------------------------------
diff --git a/connector/connector-generic-jdbc/src/main/resources/generic-jdbc-connector-resources.properties b/connector/connector-generic-jdbc/src/main/resources/generic-jdbc-connector-resources.properties
index 44fc984..0950e32 100644
--- a/connector/connector-generic-jdbc/src/main/resources/generic-jdbc-connector-resources.properties
+++ b/connector/connector-generic-jdbc/src/main/resources/generic-jdbc-connector-resources.properties
@@ -81,6 +81,10 @@ table.dataDirectory.help = The sub-directory under warehouse for data
table.partitionColumn.label = Partition column name
table.partitionColumn.help = A specific column for data partition
+# Table pcol is null
+table.partitionColumnNull.label = Nulls in partition column
+table.partitionColumnNull.help = Whether there are null values in partition column
+
# Table boundary
table.boundaryQuery.label = Boundary query
table.boundaryQuery.help = The boundary query for data partition
http://git-wip-us.apache.org/repos/asf/sqoop/blob/91b121cc/connector/connector-generic-jdbc/src/test/java/org/apache/sqoop/connector/jdbc/TestImportPartitioner.java
----------------------------------------------------------------------
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 7ecc900..a33dd6c 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
@@ -457,6 +457,36 @@ public class TestImportPartitioner extends TestCase {
}
+ public void testPatitionWithNullValues() throws Exception {
+ MutableContext context = new MutableMapContext();
+ context.setString(GenericJdbcConnectorConstants
+ .CONNECTOR_JDBC_PARTITION_COLUMNNAME, "VCCOL");
+ context.setString(GenericJdbcConnectorConstants
+ .CONNECTOR_JDBC_PARTITION_COLUMNTYPE, String.valueOf(Types.VARCHAR));
+ context.setString(GenericJdbcConnectorConstants
+ .CONNECTOR_JDBC_PARTITION_MINVALUE, "AAA");
+ context.setString(GenericJdbcConnectorConstants
+ .CONNECTOR_JDBC_PARTITION_MAXVALUE, "AAE");
+
+ ConnectionConfiguration connConf = new ConnectionConfiguration();
+ ImportJobConfiguration jobConf = new ImportJobConfiguration();
+ jobConf.table.partitionColumnNull = true;
+
+ Partitioner partitioner = new GenericJdbcImportPartitioner();
+ PartitionerContext partitionerContext = new PartitionerContext(context, 5, null);
+
+ List<Partition> partitions = partitioner.getPartitions(partitionerContext, connConf, jobConf);
+
+ verifyResult(partitions, new String[] {
+ "VCCOL IS NULL",
+ "'AAA' <= VCCOL AND VCCOL < 'AAB'",
+ "'AAB' <= VCCOL AND VCCOL < 'AAC'",
+ "'AAC' <= VCCOL AND VCCOL < 'AAD'",
+ "'AAD' <= VCCOL AND VCCOL <= 'AAE'",
+ });
+
+ }
+
private void verifyResult(List<Partition> partitions,
String[] expected) {
assertEquals(expected.length, partitions.size());