SQOOP-2160: Sqoop2: Datatypes: Provide foundation for the exhaustive type checks

(Jarek Jarcec Cecho via Abraham Elmahrek)

common-test/src/main/java/org/apache/sqoop/common/test/asserts/ProviderAsserts.java

@@ -42,9 +42,22 @@ public class ProviderAsserts {
    * @param values Values that should be present in the table
    */
   public static void assertRow(DatabaseProvider provider, String tableName,  Object []conditions, Object ...values) {
+    assertRow(provider, tableName, true, conditions, values);
+  }
+
+  /**
+   * Assert row in the table.
+   *
+   * @param provider Provider that should be used to query the database
+   * @param tableName Table name
+   * @param escapeValues Flag whether the values should be escaped based on their type when using in the generated queries or not
+   * @param conditions Conditions for identifying the row
+   * @param values Values that should be present in the table
+   */
+  public static void assertRow(DatabaseProvider provider, String tableName, boolean escapeValues, Object []conditions, Object ...values) {
     ResultSet rs = null;
     try {
-      rs = provider.getRows(tableName, conditions);
+      rs = provider.getRows(tableName, escapeValues, conditions);
 
       if(! rs.next()) {
         fail("No rows found.");

common-test/src/main/java/org/apache/sqoop/common/test/db/DatabaseProvider.java

@@ -19,6 +19,9 @@
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.log4j.Logger;
+import org.apache.sqoop.common.test.db.types.DatabaseType;
+import org.apache.sqoop.common.test.db.types.DatabaseTypeList;
+import org.apache.sqoop.common.test.db.types.EmptyTypeList;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -142,6 +145,17 @@ public String getJdbcDriver() {
     return null;
   }
 
+  /**
+   * Return type overview for this database.
+   *
+   * This method must work even in case that the provider hasn't been started.
+   *
+   * @return
+   */
+  public DatabaseTypeList getDatabaseTypes() {
+    return new EmptyTypeList();
+  }
+
   /**
    * Get full table name with qualifications
    * @param schemaName
@@ -343,13 +357,24 @@ public void createTable(String name, String primaryKey, String ...columns) {
    * @param values List of objects that should be inserted
    */
   public void insertRow(String tableName, Object ...values) {
+    insertRow(tableName, true, values);
+  }
+
+  /**
+   * Insert new row into the table.
+   *
+   * @param tableName Table name
+   * @param escapeValues Should the values be escaped based on their type or not
+   * @param values List of objects that should be inserted
+   */
+  public void insertRow(String tableName, boolean escapeValues, Object ...values) {
     StringBuilder sb = new StringBuilder("INSERT INTO ");
     sb.append(escapeTableName(tableName));
     sb.append(" VALUES (");
 
     List<String> valueList = new LinkedList<String>();
     for(Object value : values) {
-      valueList.add(convertObjectToQueryString(value));
+      valueList.add(escapeValues ? convertObjectToQueryString(value) : value.toString());
     }
 
     sb.append(StringUtils.join(valueList, ", "));
@@ -366,6 +391,18 @@ public void insertRow(String tableName, Object ...values) {
    * @return ResultSet with given criteria
    */
   public ResultSet getRows(String tableName, Object []conditions) {
+    return getRows(tableName, true, conditions);
+  }
+
+  /**
+   * Return rows that match given conditions.
+   *
+   * @param tableName Table name
+   * @param escapeValues Should the values be escaped based on their type or not
+   * @param conditions Conditions in form of double values - column name and value, for example: "id", 1 or "last_update_date", null
+   * @return ResultSet with given criteria
+   */
+  public ResultSet getRows(String tableName, boolean escapeValues, Object []conditions) {
     // Columns are in form of two strings - name and value
     if(conditions.length % 2 != 0) {
       throw new RuntimeException("Incorrect number of parameters.");
@@ -386,7 +423,7 @@ public ResultSet getRows(String tableName, Object []conditions) {
       if(value == null) {
         conditionList.add(escapeColumnName((String) columnName) + " IS NULL");
       } else {
-        conditionList.add(escapeColumnName((String) columnName) + " = " + convertObjectToQueryString(value));
+        conditionList.add(escapeColumnName((String) columnName) + " = " + (escapeValues ? convertObjectToQueryString(value) : value));
       }
     }
 

common-test/src/main/java/org/apache/sqoop/common/test/db/DerbyProvider.java

@@ -20,6 +20,8 @@
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.derby.drda.NetworkServerControl;
+import org.apache.sqoop.common.test.db.types.DatabaseTypeList;
+import org.apache.sqoop.common.test.db.types.DerbyTypeList;
 import org.apache.sqoop.common.test.utils.LoggerWriter;
 import org.apache.sqoop.common.test.utils.NetworkUtils;
 
@@ -165,4 +167,9 @@ public String getConnectionUsername() {
   public String getConnectionPassword() {
     return null;
   }
+
+  @Override
+  public DatabaseTypeList getDatabaseTypes() {
+    return new DerbyTypeList();
+  }
 }

common-test/src/main/java/org/apache/sqoop/common/test/db/types/DatabaseType.java

@@ -0,0 +1,96 @@
+/**
+ * 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.
+ */
+package org.apache.sqoop.common.test.db.types;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Class describing type and example values that can be stored in the database.
+ */
+public class DatabaseType {
+
+  /**
+   * Builder for simpler creation of the DatabaseType objects
+   */
+  public static class Builder {
+    private String name;
+    List<ExampleValue> values;
+
+    public Builder(String name) {
+      this.name = name;
+      values = new LinkedList<ExampleValue>();
+    }
+
+    public Builder addExample(String insertStatement, Object objectValue, String escapedStringValue) {
+      values.add(new ExampleValue(insertStatement, objectValue, escapedStringValue));
+      return this;
+    }
+
+    public DatabaseType build() {
+      return new DatabaseType(name, values);
+    }
+  }
+
+  /**
+   * Return new instance of builder.
+   */
+  public static Builder builder(String name) {
+    return new Builder(name);
+  }
+
+  /**
+   * Name of type as can appear in the CREATE TABLE statement.
+   */
+  public final String name;
+
+  /**
+   * Example values for given data type.
+   *
+   * Small number of the values, not exhaustive list.
+   */
+  public final List<ExampleValue> values;
+
+  public DatabaseType(String name, List<ExampleValue> values) {
+    this.name = name;
+    this.values = values;
+  }
+
+  @Override
+  public String toString() {
+    return name;
+  }
+
+  /**
+   * Returns escaped strings from all values in it's own array.
+   *
+   * The order is defined and will always be in the same order as
+   * is present in the values array.
+   *
+   * @return Array where each item represents escapeStringValue field from the
+   *   corresponding values element
+   */
+  public String []escapedStringValues() {
+    String [] ret = new String[values.size()];
+    int i = 0;
+    for(ExampleValue value : values) {
+      ret[i++] = value.escapedStringValue;
+    }
+    return ret;
+  }
+}

common-test/src/main/java/org/apache/sqoop/common/test/db/types/DatabaseTypeList.java

@@ -0,0 +1,47 @@
+/**
+ * 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.
+ */
+package org.apache.sqoop.common.test.db.types;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * List of all types provided by given database.
+ */
+abstract public class DatabaseTypeList {
+
+  /**
+   * Internal list of all types.
+   */
+  List<DatabaseType> types;
+
+  public DatabaseTypeList() {
+    types = new LinkedList<DatabaseType>();
+  }
+
+  protected void add(DatabaseType type) {
+    types.add(type);
+  }
+
+  /**
+   * Returns all types for given database.
+   */
+  public List<DatabaseType> getAllTypes() {
+    return types;
+  }
+}

common-test/src/main/java/org/apache/sqoop/common/test/db/types/DerbyTypeList.java

@@ -0,0 +1,93 @@
+/**
+ * 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.
+ */
+package org.apache.sqoop.common.test.db.types;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+
+/**
+ * Source: https://db.apache.org/derby/docs/10.7/ref/crefsqlj31068.html
+ */
+public class DerbyTypeList extends DatabaseTypeList {
+  public DerbyTypeList() {
+    super();
+
+    // Numeric types
+    add(DatabaseType.builder("SMALLINT")
+      .addExample("-32768", new Integer(-32768), "-32768")
+      .addExample(    "-1",     new Integer(-1),     "-1")
+      .addExample(     "0",      new Integer(0),      "0")
+      .addExample(     "1",      new Integer(1),      "1")
+      .addExample( "32767",  new Integer(32767),  "32767")
+      .build());
+    add(DatabaseType.builder("INT")
+      .addExample("-2147483648", new Integer(-2147483648), "-2147483648")
+      .addExample(         "-1",          new Integer(-1),          "-1")
+      .addExample(          "0",           new Integer(0),           "0")
+      .addExample(          "1",           new Integer(1),           "1")
+      .addExample( "2147483647",  new Integer(2147483647),  "2147483647")
+      .build());
+    add(DatabaseType.builder("BIGINT")
+      .addExample("-9223372036854775808", new Long(-9223372036854775808L), "-9223372036854775808")
+      .addExample(                  "-1",                   new Long(-1L),                   "-1")
+      .addExample(                   "0",                    new Long(0L),                    "0")
+      .addExample(                   "1",                    new Long(1L),                    "1")
+      .addExample( "9223372036854775807",  new Long(9223372036854775807L),  "9223372036854775807")
+      .build());
+
+    // Floating points
+    add(DatabaseType.builder("REAL")
+      .addExample("CAST(-3.402E+38 AS REAL)", new Float(-3.402E+38),  "-3.402E38")
+      .addExample( "CAST(3.402E+38 AS REAL)",  new Float(3.402E+38),   "3.402E38")
+      .addExample(                       "0",          new Float(0),        "0.0")
+      .addExample( "CAST(1.175E-37 AS REAL)",  new Float(1.175E-37),  "1.175E-37")
+      .addExample("CAST(-1.175E-37 AS REAL)", new Float(-1.175E-37), "-1.175E-37")
+      .build());
+    add(DatabaseType.builder("DOUBLE")
+      .addExample("-1.79769E+308", new Double(-1.79769E+308), "-1.79769E308")
+      .addExample( "1.79769E+308",  new Double(1.79769E+308),  "1.79769E308")
+      .addExample(            "0",             new Double(0),          "0.0")
+      .addExample(   "2.225E-307",    new Double(2.225E-307),   "2.225E-307")
+      .addExample(  "-2.225E-307",   new Double(-2.225E-307),  "-2.225E-307")
+      .build());
+
+    // Fixed point
+    add(DatabaseType.builder("DECIMAL(5, 2)")
+      .addExample("-999.99", new BigDecimal(-999.99).setScale(2, RoundingMode.CEILING), "-999.99")
+      .addExample( "-999.9",  new BigDecimal(-999.9).setScale(2,   RoundingMode.FLOOR), "-999.90")
+      .addExample(  "-99.9",   new BigDecimal(-99.9).setScale(2, RoundingMode.CEILING),  "-99.90")
+      .addExample(   "-9.9",    new BigDecimal(-9.9).setScale(2, RoundingMode.CEILING),   "-9.90")
+      .addExample(     "-9",      new BigDecimal(-9).setScale(2, RoundingMode.CEILING),   "-9.00")
+      .addExample(      "0",       new BigDecimal(0).setScale(2, RoundingMode.CEILING),    "0.00")
+      .addExample(      "9",       new BigDecimal(9).setScale(2, RoundingMode.CEILING),    "9.00")
+      .addExample(    "9.9",     new BigDecimal(9.9).setScale(2,   RoundingMode.FLOOR),    "9.90")
+      .addExample(   "99.9",    new BigDecimal(99.9).setScale(2,   RoundingMode.FLOOR),   "99.90")
+      .addExample(  "999.9",   new BigDecimal(999.9).setScale(2, RoundingMode.CEILING),  "999.90")
+      .addExample( "999.99",  new BigDecimal(999.99).setScale(2,   RoundingMode.FLOOR),  "999.99")
+      .build());
+
+    // BLOB
+    // Boolean
+    // Char
+    // CLOB
+    // Date
+    // Time
+    // Timestamp
+    // XML?
+  }
+}

common-test/src/main/java/org/apache/sqoop/common/test/db/types/EmptyTypeList.java

@@ -0,0 +1,31 @@
+/**
+ * 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.
+ */
+package org.apache.sqoop.common.test.db.types;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Default implementation that don't have any types.
+ */
+public class EmptyTypeList extends DatabaseTypeList {
+  @Override
+  public List<DatabaseType> getAllTypes() {
+    return new LinkedList<DatabaseType>();
+  }
+}

common-test/src/main/java/org/apache/sqoop/common/test/db/types/ExampleValue.java

@@ -0,0 +1,44 @@
+/**
+ * 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.
+ */
+package org.apache.sqoop.common.test.db.types;
+
+/**
+ * Example value with different representations.
+ */
+public class ExampleValue {
+  /**
+   * Properly escaped value so that it can be used in INSERT statement.
+   */
+  public final String insertStatement;
+
+  /**
+   * Object value that should be returned from JDBC driver getObject().
+   */
+  public final Object objectValue;
+
+  /**
+   * Escaped string value that will be stored by HDFS connector.
+   */
+  public final String escapedStringValue;
+
+  public ExampleValue(String insertStatement, Object objectValue, String escapedStringValue) {
+    this.insertStatement = insertStatement;
+    this.objectValue = objectValue;
+    this.escapedStringValue = escapedStringValue;
+  }
+}

test/src/main/java/org/apache/sqoop/test/testcases/ConnectorTestCase.java

@@ -17,7 +17,6 @@
  */
 package org.apache.sqoop.test.testcases;
 
-import static org.testng.Assert.fail;
 import static org.testng.AssertJUnit.assertEquals;
 import static org.testng.Assert.assertNotSame;
 
@@ -46,9 +45,6 @@
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeSuite;
 
-import java.sql.ResultSet;
-import java.sql.SQLException;
-
 /**
  * Base test case suitable for connector testing.
  *
@@ -123,6 +119,14 @@ protected void insertRow(Object ...values) {
     provider.insertRow(getTableName(), values);
   }
 
+  protected void insertRow(boolean escapeValues, Object ...values) {
+    provider.insertRow(getTableName(), escapeValues, values);
+  }
+
+  protected long rowCount() {
+    return provider.rowCount(getTableName());
+  }
+
   protected void dumpTable() {
     provider.dumpTable(getTableName());
   }
@@ -223,6 +227,17 @@ protected void assertRow(Object[] conditions, Object ...values) {
     ProviderAsserts.assertRow(provider, getTableName(), conditions, values);
   }
 
+  /**
+   * Assert row in testing table.
+   *
+   * @param conditions Conditions in config that are expected by the database provider
+   * @param escapeValues Flag whether the values should be escaped based on their type when using in the generated queries or not
+   * @param values Values that are expected in the table (with corresponding types)
+   */
+  protected void assertRow(Object []conditions, boolean escapeValues, Object ...values) {
+    ProviderAsserts.assertRow(provider, getTableName(), escapeValues, conditions, values);
+  }
+
   /**
    * Assert row in table "cities".
    *

test/src/main/java/org/apache/sqoop/test/utils/ParametrizedUtils.java

@@ -20,6 +20,7 @@
 import org.apache.commons.lang.ArrayUtils;
 
 import java.util.LinkedList;
+import java.util.List;
 
 /**
  * Various utils to help build expected structure for JUNIT Parametrized runner.
@@ -69,6 +70,25 @@ public class ParametrizedUtils {
     return ret;
   }
 
+  /**
+   * Convert single list to array of arrays by putting each element in the source
+   * list into it's own one-item big array.
+   *
+   * This method is suitable for conversions required by Parametrized test runner.
+   *
+   * @param list List to be converted
+   * @return Converted array
+   */
+  public static Iterable<Object []>toArrayOfArrays(List<? extends Object> list) {
+     LinkedList<Object []> ret = new LinkedList<Object []>();
+
+    for(Object o : list) {
+      ret.add(toArray(o));
+    }
+
+    return ret;
+  }
+
   /**
    * Merge two objects into array.
    *

test/src/test/java/org/apache/sqoop/integration/connector/jdbc/generic/AllTypesTest.java

@@ -0,0 +1,164 @@
+/**
+ * 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.
+ */
+package org.apache.sqoop.integration.connector.jdbc.generic;
+
+import com.google.common.collect.Iterables;
+import org.apache.sqoop.common.Direction;
+import org.apache.sqoop.common.test.db.DatabaseProvider;
+import org.apache.sqoop.common.test.db.DatabaseProviderFactory;
+import org.apache.sqoop.common.test.db.types.DatabaseType;
+import org.apache.sqoop.common.test.db.types.ExampleValue;
+import org.apache.sqoop.connector.hdfs.configuration.ToFormat;
+import org.apache.sqoop.model.MConfigList;
+import org.apache.sqoop.model.MDriverConfig;
+import org.apache.sqoop.model.MJob;
+import org.apache.sqoop.model.MLink;
+import org.apache.sqoop.test.testcases.ConnectorTestCase;
+import org.apache.sqoop.test.utils.ParametrizedUtils;
+import org.testng.ITest;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Factory;
+import org.testng.annotations.Test;
+
+import java.lang.reflect.Method;
+
+import static org.testng.Assert.assertEquals;
+
+/**
+ * Test transfer of all supported data types.
+ */
+public class AllTypesTest extends ConnectorTestCase implements ITest {
+
+  @DataProvider(name="all-types-test", parallel=true)
+  public static Object[][] data() throws Exception {
+    DatabaseProvider provider = DatabaseProviderFactory.getProvider(System.getProperties());
+    return Iterables.toArray(ParametrizedUtils.toArrayOfArrays(provider.getDatabaseTypes().getAllTypes()), Object[].class);
+  }
+
+  @Factory(dataProvider="all-types-test")
+  public AllTypesTest(DatabaseType type) {
+    this.type = type;
+  }
+
+  private DatabaseType type;
+
+  @Test
+  public void testFrom() throws Exception {
+    createTable("id",
+      "id", "INT",
+      "value", type.name
+    );
+
+    int i = 1;
+    for(ExampleValue value: type.values) {
+      insertRow(false, Integer.toString(i++), value.insertStatement);
+    }
+
+    // RDBMS link
+    MLink rdbmsConnection = getClient().createLink("generic-jdbc-connector");
+    fillRdbmsLinkConfig(rdbmsConnection);
+    saveLink(rdbmsConnection);
+
+    // HDFS link
+    MLink hdfsConnection = getClient().createLink("hdfs-connector");
+    saveLink(hdfsConnection);
+
+    // Job creation
+    MJob job = getClient().createJob(rdbmsConnection.getPersistenceId(), hdfsConnection.getPersistenceId());
+
+    // Fill rdbms "FROM" config
+    MConfigList fromConfig = job.getJobConfig(Direction.FROM);
+    fromConfig.getStringInput("fromJobConfig.tableName").setValue(provider.escapeTableName(getTableName()));
+    fromConfig.getStringInput("fromJobConfig.columns").setValue(provider.escapeColumnName("value"));
+    fromConfig.getStringInput("fromJobConfig.partitionColumn").setValue(provider.escapeColumnName("id"));
+
+    // Fill the hdfs "TO" config
+    fillHdfsToConfig(job, ToFormat.TEXT_FILE);
+
+    // driver config
+    MDriverConfig driverConfig = job.getDriverConfig();
+    driverConfig.getIntegerInput("throttlingConfig.numExtractors").setValue(1);
+
+    saveJob(job);
+    executeJob(job);
+
+    // Assert correct output
+    assertTo(type.escapedStringValues());
+
+    // Clean up testing table
+    dropTable();
+  }
+
+  @Test
+  public void testTo() throws Exception {
+    createTable(null,
+      "value", type.name
+    );
+
+    createFromFile("input-0001", type.escapedStringValues());
+
+    // RDBMS link
+    MLink rdbmsLink = getClient().createLink("generic-jdbc-connector");
+    fillRdbmsLinkConfig(rdbmsLink);
+    saveLink(rdbmsLink);
+
+    // HDFS link
+    MLink hdfsLink = getClient().createLink("hdfs-connector");
+    saveLink(hdfsLink);
+
+    // Job creation
+    MJob job = getClient().createJob(hdfsLink.getPersistenceId(), rdbmsLink.getPersistenceId());
+    fillHdfsFromConfig(job);
+
+    // Set the rdms "TO" config here
+    MConfigList toConfig = job.getJobConfig(Direction.TO);
+    toConfig.getStringInput("toJobConfig.tableName").setValue(provider.escapeTableName(getTableName()));
+
+    // Driver config
+    MDriverConfig driverConfig = job.getDriverConfig();
+    driverConfig.getIntegerInput("throttlingConfig.numExtractors").setValue(1);
+
+    saveJob(job);
+    executeJob(job);
+    dumpTable();
+
+    assertEquals(type.values.size(), rowCount());
+    for(ExampleValue value : type.values) {
+      assertRow(
+        new Object[] {"value", value.insertStatement},
+        false,
+        value.objectValue);
+    }
+
+    // Clean up testing table
+    dropTable();
+  }
+
+  private String testName;
+
+  @BeforeMethod(alwaysRun = true)
+  public void beforeMethod(Method aMethod) {
+    this.testName = aMethod.getName();
+  }
+
+  @Override
+  public String getTestName() {
+    return testName + "[" + type.name + "]";
+  }
+}