You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sqoop.apache.org by an...@apache.org on 2018/01/18 14:03:48 UTC
[02/32] sqoop git commit: SQOOP-3273: Removing com.cloudera.sqoop
packages
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/BaseSqoopTestCase.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/BaseSqoopTestCase.java b/src/test/org/apache/sqoop/testutil/BaseSqoopTestCase.java
new file mode 100644
index 0000000..588f439
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/BaseSqoopTestCase.java
@@ -0,0 +1,624 @@
+/**
+ * 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.testutil;
+
+import org.apache.sqoop.ConnFactory;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.manager.ConnManager;
+import org.apache.sqoop.metastore.JobData;
+import org.apache.sqoop.tool.ImportTool;
+import com.google.common.collect.ObjectArrays;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.BasicConfigurator;
+import org.apache.sqoop.SqoopJobDataPublisher;
+import org.junit.After;
+import org.junit.Before;
+
+import java.io.File;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Arrays;
+
+import static org.junit.Assert.fail;
+
+/**
+ * Class that implements common methods required for tests.
+ */
+public abstract class BaseSqoopTestCase {
+
+ public static class DummyDataPublisher extends SqoopJobDataPublisher {
+ public static String hiveTable;
+ public static String storeTable;
+ public static String storeType;
+ public static String operation;
+
+ @Override
+ public void publish(Data data) {
+ hiveTable = data.getHiveTable();
+ storeTable = data.getStoreTable();
+ storeType = data.getStoreType();
+ operation = data.getOperation();
+ }
+ }
+
+ public static final Log LOG = LogFactory.getLog(
+ BaseSqoopTestCase.class.getName());
+
+ public static boolean isOnPhysicalCluster() {
+ return onPhysicalCluster;
+ }
+ private static void setOnPhysicalCluster(boolean value) {
+ onPhysicalCluster = value;
+ }
+
+ private static boolean onPhysicalCluster = false;
+
+ /** Base directory for all temporary data. */
+ public static final String TEMP_BASE_DIR;
+
+ /** Where to import table data to in the local filesystem for testing. */
+ public static final String LOCAL_WAREHOUSE_DIR;
+
+ // Initializer for the above
+ static {
+ String tmpDir = System.getProperty("test.build.data", "/tmp/");
+ if (!tmpDir.endsWith(File.separator)) {
+ tmpDir = tmpDir + File.separator;
+ }
+
+ TEMP_BASE_DIR = tmpDir;
+ LOCAL_WAREHOUSE_DIR = TEMP_BASE_DIR + "sqoop/warehouse";
+ }
+
+ // Used if a test manually sets the table name to be used.
+ private String curTableName;
+
+ protected void setCurTableName(String curName) {
+ this.curTableName = curName;
+ }
+
+ /**
+ * Because of how classloading works, we don't actually want to name
+ * all the tables the same thing -- they'll actually just use the same
+ * implementation of the Java class that was classloaded before. So we
+ * use this counter to uniquify table names.
+ */
+ private static int tableNum = 0;
+
+ /** When creating sequentially-identified tables, what prefix should
+ * be applied to these tables?
+ */
+ protected String getTablePrefix() {
+ return "SQOOP_TABLE_";
+ }
+
+ protected String getTableName() {
+ if (null != curTableName) {
+ return curTableName;
+ } else {
+ return getTablePrefix() + Integer.toString(tableNum);
+ }
+ }
+
+ protected String getWarehouseDir() {
+ return LOCAL_WAREHOUSE_DIR;
+ }
+
+ private String [] colNames;
+ protected String [] getColNames() {
+ return colNames;
+ }
+
+ protected void setColNames(String [] cols) {
+ if (null == cols) {
+ this.colNames = null;
+ } else {
+ this.colNames = Arrays.copyOf(cols, cols.length);
+ }
+ }
+
+ protected HsqldbTestServer getTestServer() {
+ return testServer;
+ }
+
+ protected ConnManager getManager() {
+ return manager;
+ }
+
+ protected void setManager(ConnManager manager) {
+ this.manager = manager;
+ }
+
+ /**
+ * @return a connection to the database under test.
+ */
+ protected Connection getConnection() {
+ try {
+ return getTestServer().getConnection();
+ } catch (SQLException sqlE) {
+ LOG.error("Could not get connection to test server: " + sqlE);
+ return null;
+ }
+ }
+
+ // instance variables populated during setUp, used during tests
+ private HsqldbTestServer testServer;
+ private ConnManager manager;
+
+ private static boolean isLog4jConfigured = false;
+
+ protected void incrementTableNum() {
+ tableNum++;
+ }
+
+ /**
+ * @return true if we need an in-memory database to run these tests.
+ */
+ protected boolean useHsqldbTestServer() {
+ return true;
+ }
+
+ /**
+ * @return the connect string to use for interacting with the database.
+ * If useHsqldbTestServer is false, you need to override this and provide
+ * a different connect string.
+ */
+ protected String getConnectString() {
+ return HsqldbTestServer.getUrl();
+ }
+
+ /**
+ * @return a Configuration object used to configure tests.
+ */
+ protected Configuration getConf() {
+ return new Configuration();
+ }
+
+ /**
+ * @return a new SqoopOptions customized for this particular test, but one
+ * which has not had any arguments parsed yet.
+ */
+ protected SqoopOptions getSqoopOptions(Configuration conf) {
+ return new SqoopOptions(conf);
+ }
+
+ @Before
+ public void setUp() {
+ // The assumption is that correct HADOOP configuration will have it set to
+ // hdfs://namenode
+ setOnPhysicalCluster(
+ !CommonArgs.LOCAL_FS.equals(System.getProperty(
+ CommonArgs.FS_DEFAULT_NAME)));
+ incrementTableNum();
+
+ if (!isLog4jConfigured) {
+ BasicConfigurator.configure();
+ isLog4jConfigured = true;
+ LOG.info("Configured log4j with console appender.");
+ }
+
+ if (useHsqldbTestServer()) {
+ testServer = new HsqldbTestServer();
+ try {
+ testServer.resetServer();
+ } catch (SQLException sqlE) {
+ LOG.error("Got SQLException: " + StringUtils.stringifyException(sqlE));
+ fail("Got SQLException: " + StringUtils.stringifyException(sqlE));
+ } catch (ClassNotFoundException cnfe) {
+ LOG.error("Could not find class for db driver: "
+ + StringUtils.stringifyException(cnfe));
+ fail("Could not find class for db driver: "
+ + StringUtils.stringifyException(cnfe));
+ }
+
+ manager = testServer.getManager();
+ } else {
+ Configuration conf = getConf();
+ //Need to disable OraOop for existing tests
+ conf.set("oraoop.disabled", "true");
+ SqoopOptions opts = getSqoopOptions(conf);
+ opts.setConnectString(getConnectString());
+ opts.setTableName(getTableName());
+ ConnFactory f = new ConnFactory(conf);
+ try {
+ this.manager = f.getManager(new JobData(opts, new ImportTool()));
+ } catch (IOException ioe) {
+ fail("IOException instantiating manager: "
+ + StringUtils.stringifyException(ioe));
+ }
+ }
+ }
+
+ private void guaranteeCleanWarehouse() {
+ if (isOnPhysicalCluster()) {
+ Path warehousePath = new Path(this.getWarehouseDir());
+ try {
+ FileSystem fs = FileSystem.get(getConf());
+ fs.delete(warehousePath, true);
+ } catch (IOException e) {
+ LOG.warn(e);
+ }
+ } else {
+ File s = new File(getWarehouseDir());
+ if (!s.delete()) {
+ LOG.warn("Cannot delete " + s.getPath());
+ }
+ }
+ }
+
+ @After
+ public void tearDown() {
+ setCurTableName(null); // clear user-override table name.
+
+ try {
+ if (null != manager) {
+ manager.close();
+ manager = null;
+ }
+ } catch (SQLException sqlE) {
+ LOG.error("Got SQLException: " + StringUtils.stringifyException(sqlE));
+ fail("Got SQLException: " + StringUtils.stringifyException(sqlE));
+ }
+ guaranteeCleanWarehouse();
+ }
+
+ public static final String BASE_COL_NAME = "DATA_COL";
+
+ protected String getColName(int i) {
+ return BASE_COL_NAME + i;
+ }
+
+ /**
+ * Drop a table if it already exists in the database.
+ * @param table the name of the table to drop.
+ * @throws SQLException if something goes wrong.
+ */
+ protected void dropTableIfExists(String table) throws SQLException {
+ Connection conn = getManager().getConnection();
+ PreparedStatement statement = conn.prepareStatement(dropTableIfExistsCommand(table),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ try {
+ statement.executeUpdate();
+ conn.commit();
+ } finally {
+ statement.close();
+ }
+ }
+
+ protected String dropTableIfExistsCommand(String table) {
+ return "DROP TABLE " + manager.escapeTableName(table) + " IF EXISTS";
+ }
+
+ protected void createTableWithColTypesAndNames(String[] colNames,
+ String[] colTypes,
+ String[] vals) {
+ createTableWithColTypesAndNames(getTableName(), colNames, colTypes, vals);
+ }
+
+ /**
+ * Create a table with a set of columns with their names and add a row of values.
+ * @param newTableName The name of the new table
+ * @param colNames Column names
+ * @param colTypes the types of the columns to make
+ * @param vals the SQL text for each value to insert
+ */
+ protected void createTableWithColTypesAndNames(String newTableName,
+ String[] colNames,
+ String[] colTypes,
+ String[] vals) {
+ assert colNames != null;
+ assert colTypes != null;
+ assert colNames.length == colTypes.length;
+
+ Connection conn = null;
+ PreparedStatement statement = null;
+ String createTableStr = null;
+ String columnDefStr = "";
+
+ try {
+ try {
+ dropTableIfExists(newTableName);
+
+ conn = getManager().getConnection();
+
+ for (int i = 0; i < colTypes.length; i++) {
+ columnDefStr += manager.escapeColName(colNames[i].toUpperCase()) + " " + colTypes[i];
+ if (i < colTypes.length - 1) {
+ columnDefStr += ", ";
+ }
+ }
+
+ createTableStr = "CREATE TABLE " + manager.escapeTableName(newTableName) + "(" + columnDefStr + ")";
+ LOG.info("Creating table: " + createTableStr);
+ statement = conn.prepareStatement(
+ createTableStr,
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ statement.executeUpdate();
+ } catch (SQLException sqlException) {
+ fail("Could not create table: "
+ + StringUtils.stringifyException(sqlException));
+ } finally {
+ if (null != statement) {
+ try {
+ statement.close();
+ } catch (SQLException se) {
+ // Ignore exception on close.
+ }
+
+ statement = null;
+ }
+ }
+
+ for (int count=0; vals != null && count < vals.length/colTypes.length;
+ ++count ) {
+ String columnListStr = "";
+ String valueListStr = "";
+ for (int i = 0; i < colTypes.length; i++) {
+ columnListStr += manager.escapeColName(colNames[i].toUpperCase());
+ valueListStr += vals[count * colTypes.length + i];
+ if (i < colTypes.length - 1) {
+ columnListStr += ", ";
+ valueListStr += ", ";
+ }
+ }
+ try {
+ String insertValsStr = "INSERT INTO " + manager.escapeTableName(newTableName) + "(" + columnListStr + ")"
+ + " VALUES(" + valueListStr + ")";
+ LOG.info("Inserting values: " + insertValsStr);
+ statement = conn.prepareStatement(
+ insertValsStr,
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ statement.executeUpdate();
+ } catch (SQLException sqlException) {
+ fail("Could not insert into table: "
+ + StringUtils.stringifyException(sqlException));
+ } finally {
+ if (null != statement) {
+ try {
+ statement.close();
+ } catch (SQLException se) {
+ // Ignore exception on close.
+ }
+
+ statement = null;
+ }
+ }
+ }
+
+ conn.commit();
+ this.colNames = colNames;
+ } catch (SQLException se) {
+ if (null != conn) {
+ try {
+ conn.close();
+ } catch (SQLException connSE) {
+ // Ignore exception on close.
+ }
+ }
+ fail("Could not create table: " + StringUtils.stringifyException(se));
+ }
+ }
+
+ /**
+ * insert into a table with a set of columns values for a given row.
+ * @param colTypes the types of the columns to make
+ * @param vals the SQL text for each value to insert
+ */
+ protected void insertIntoTable(String[] colTypes, String[] vals) {
+ assert colNames != null;
+ assert colNames.length == vals.length;
+
+ Connection conn = null;
+ PreparedStatement statement = null;
+
+ String[] colNames = new String[vals.length];
+ for( int i = 0; i < vals.length; i++) {
+ colNames[i] = BASE_COL_NAME + Integer.toString(i);
+ }
+ try {
+ conn = getManager().getConnection();
+ for (int count=0; vals != null && count < vals.length/colTypes.length;
+ ++count ) {
+ String columnListStr = "";
+ String valueListStr = "";
+ for (int i = 0; i < colTypes.length; i++) {
+ columnListStr += manager.escapeColName(colNames[i].toUpperCase());
+ valueListStr += vals[count * colTypes.length + i];
+ if (i < colTypes.length - 1) {
+ columnListStr += ", ";
+ valueListStr += ", ";
+ }
+ }
+ try {
+ String insertValsStr = "INSERT INTO " + manager.escapeTableName(getTableName()) + "(" + columnListStr + ")"
+ + " VALUES(" + valueListStr + ")";
+ LOG.info("Inserting values: " + insertValsStr);
+ statement = conn.prepareStatement(
+ insertValsStr,
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ statement.executeUpdate();
+ } catch (SQLException sqlException) {
+ fail("Could not insert into table: "
+ + StringUtils.stringifyException(sqlException));
+ } finally {
+ if (null != statement) {
+ try {
+ statement.close();
+ } catch (SQLException se) {
+ // Ignore exception on close.
+ }
+
+ statement = null;
+ }
+ }
+ }
+ conn.commit();
+ this.colNames = colNames;
+ } catch (SQLException se) {
+ if (null != conn) {
+ try {
+ conn.close();
+ } catch (SQLException connSE) {
+ // Ignore exception on close.
+ }
+ }
+ fail("Could not create table: " + StringUtils.stringifyException(se));
+ }
+
+ }
+
+ /**
+ * update a table with a set of columns values for a given row.
+ * @param colTypes the types of the columns to make
+ * @param vals the SQL text for each value to insert
+ */
+ protected void updateTable(String[] colTypes, String[] vals) {
+ assert colNames != null;
+ assert colNames.length == vals.length;
+
+ Connection conn = null;
+ PreparedStatement statement = null;
+
+ String[] colNames = new String[vals.length];
+ for( int i = 0; i < vals.length; i++) {
+ colNames[i] = BASE_COL_NAME + Integer.toString(i);
+ }
+
+ try {
+ conn = getManager().getConnection();
+ for (int count=0; vals != null && count < vals.length/colNames.length;
+ ++count ) {
+ String updateStr = "";
+ for (int i = 1; i < colNames.length; i++) {
+ updateStr += manager.escapeColName(colNames[i].toUpperCase()) + " = "+vals[count * colNames.length + i];
+ if (i < colNames.length - 1) {
+ updateStr += ", ";
+ }
+ }
+ updateStr += " WHERE "+colNames[0]+"="+vals[0]+"";
+ try {
+ String updateValsStr = "UPDATE " + manager.escapeTableName(getTableName()) + " SET " + updateStr;
+ LOG.info("updating values: " + updateValsStr);
+ statement = conn.prepareStatement(
+ updateValsStr,
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ statement.executeUpdate();
+ } catch (SQLException sqlException) {
+ fail("Could not update table: "
+ + StringUtils.stringifyException(sqlException));
+ } finally {
+ if (null != statement) {
+ try {
+ statement.close();
+ } catch (SQLException se) {
+ // Ignore exception on close.
+ }
+ statement = null;
+ }
+ }
+ }
+
+ conn.commit();
+ this.colNames = colNames;
+ } catch (SQLException se) {
+ if (null != conn) {
+ try {
+ conn.close();
+ } catch (SQLException connSE) {
+ // Ignore exception on close.
+ }
+ }
+ fail("Could not update table: " + StringUtils.stringifyException(se));
+ }
+ }
+
+ /**
+ * Create a table with a set of columns and add a row of values.
+ * @param colTypes the types of the columns to make
+ * @param vals the SQL text for each value to insert
+ */
+ protected void createTableWithColTypes(String [] colTypes, String [] vals) {
+ String[] colNames = new String[colTypes.length];
+ for( int i = 0; i < colTypes.length; i++) {
+ colNames[i] = BASE_COL_NAME + Integer.toString(i);
+ }
+ createTableWithColTypesAndNames(colNames, colTypes, vals);
+ }
+
+ /**
+ * Create a table with a single column and put a data element in it.
+ * @param colType the type of the column to create
+ * @param val the value to insert (reformatted as a string)
+ */
+ protected void createTableForColType(String colType, String val) {
+ String [] types = { colType };
+ String [] vals = { val };
+
+ createTableWithColTypes(types, vals);
+ }
+
+ protected Path getTablePath() {
+ Path warehousePath = new Path(getWarehouseDir());
+ Path tablePath = new Path(warehousePath, getTableName());
+ return tablePath;
+ }
+
+ protected Path getDataFilePath() {
+ return new Path(getTablePath(), "part-m-00000");
+ }
+
+ protected void removeTableDir() {
+ File tableDirFile = new File(getTablePath().toString());
+ if (tableDirFile.exists()) {
+ // Remove the directory where the table will be imported to,
+ // prior to running the MapReduce job.
+ if (!DirUtil.deleteDir(tableDirFile)) {
+ LOG.warn("Could not delete table directory: "
+ + tableDirFile.getAbsolutePath());
+ }
+ }
+ }
+
+ /**
+ * Create a new string array with 'moreEntries' appended to the 'entries'
+ * array.
+ * @param entries initial entries in the array
+ * @param moreEntries variable-length additional entries.
+ * @return an array containing entries with all of moreEntries appended.
+ */
+ protected String [] newStrArray(String [] entries, String... moreEntries) {
+ if (null == moreEntries) {
+ return entries;
+ }
+
+ if (null == entries) {
+ entries = new String[0];
+ }
+
+ return ObjectArrays.concat(entries, moreEntries, String.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/CommonArgs.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/CommonArgs.java b/src/test/org/apache/sqoop/testutil/CommonArgs.java
new file mode 100644
index 0000000..a91addf
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/CommonArgs.java
@@ -0,0 +1,58 @@
+/**
+ * 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.testutil;
+
+import java.util.List;
+
+/**
+ * Sets common arguments to Sqoop sub-instances for testing.
+ */
+public final class CommonArgs {
+
+ private CommonArgs() {
+ }
+
+ public static final String LOCAL_FS="file:///";
+ // this key is deprecated past 0.21
+ public static final String FS_DEFAULT_NAME="fs.defaultfs.name";
+ public static final String FS_DEFAULTFS="fs.defaultFS";
+
+ public static String getJobtrackerAddress() {
+ return System.getProperty("mapreduce.jobtracker.address", "local");
+ }
+ public static String getDefaultFS() {
+ return System.getProperty(FS_DEFAULT_NAME, LOCAL_FS);
+ }
+ /**
+ * Craft a list of arguments that are common to (virtually)
+ * all Sqoop programs.
+ */
+ public static void addHadoopFlags(List<String> args) {
+ args.add("-D");
+ args.add("mapreduce.jobtracker.address=local");
+ args.add("-D");
+ args.add("mapreduce.job.maps=1");
+ args.add("-D");
+ args.add("fs.defaultFS=file:///");
+ args.add("-D");
+ args.add("jobclient.completion.poll.interval=50");
+ args.add("-D");
+ args.add("jobclient.progress.monitor.poll.interval=50");
+ }
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/DirUtil.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/DirUtil.java b/src/test/org/apache/sqoop/testutil/DirUtil.java
new file mode 100644
index 0000000..7ca5d53
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/DirUtil.java
@@ -0,0 +1,59 @@
+/**
+ * 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.testutil;
+
+import java.io.File;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Misc directory operations.
+ */
+public final class DirUtil {
+
+ private DirUtil() {
+ }
+
+ public static final Log LOG = LogFactory.getLog(DirUtil.class.getName());
+
+ /**
+ * recursively delete a dir and its children.
+ * @param dir
+ * @return true on succesful removal of a dir
+ */
+ public static boolean deleteDir(File dir) {
+ if (dir.isDirectory()) {
+ String [] children = dir.list();
+ for (int i = 0; i < children.length; i++) {
+ File f = new File(dir, children[i]);
+ boolean success = deleteDir(f);
+ if (!success) {
+ LOG.warn("Could not delete " + f.getAbsolutePath());
+ return false;
+ }
+ }
+ }
+
+ // The directory is now empty so delete it too.
+ LOG.debug("Removing: " + dir);
+ return dir.delete();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/ExplicitSetMapper.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/ExplicitSetMapper.java b/src/test/org/apache/sqoop/testutil/ExplicitSetMapper.java
new file mode 100644
index 0000000..c1122c0
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/ExplicitSetMapper.java
@@ -0,0 +1,109 @@
+/**
+ * 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.testutil;
+
+import java.io.IOException;
+
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.sqoop.lib.SqoopRecord;
+import org.apache.hadoop.util.ReflectionUtils;
+
+/**
+ * Test harness mapper. Instantiate the user's specific type, explicitly
+ * set the value of a field with setField(), and read the field value
+ * back via the field map. Throw an IOException if it doesn't get set
+ * correctly.
+ */
+public class ExplicitSetMapper extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, NullWritable> {
+
+ public static final Log LOG = LogFactory.getLog(
+ ExplicitSetMapper.class.getName());
+
+ public static final String USER_TYPE_NAME_KEY = "sqoop.user.class";
+ public static final String SET_COL_KEY = "sqoop.explicit.set.col";
+ public static final String SET_VAL_KEY = "sqoop.explicit.set.val";
+
+ private SqoopRecord userRecord;
+ private String setCol;
+ private String setVal;
+
+ public void configure(JobConf job) {
+ String userTypeName = job.get(USER_TYPE_NAME_KEY);
+ if (null == userTypeName) {
+ throw new RuntimeException("Unconfigured parameter: "
+ + USER_TYPE_NAME_KEY);
+ }
+
+ setCol = job.get(SET_COL_KEY);
+ setVal = job.get(SET_VAL_KEY);
+
+ LOG.info("User type name set to " + userTypeName);
+ LOG.info("Will try to set col " + setCol + " to " + setVal);
+
+ this.userRecord = null;
+
+ try {
+ Configuration conf = new Configuration();
+ Class userClass = Class.forName(userTypeName, true,
+ Thread.currentThread().getContextClassLoader());
+ this.userRecord =
+ (SqoopRecord) ReflectionUtils.newInstance(userClass, conf);
+ } catch (ClassNotFoundException cnfe) {
+ // handled by the next block.
+ LOG.error("ClassNotFound exception: " + cnfe.toString());
+ } catch (Exception e) {
+ LOG.error("Got an exception reflecting user class: " + e.toString());
+ }
+
+ if (null == this.userRecord) {
+ LOG.error("Could not instantiate user record of type " + userTypeName);
+ throw new RuntimeException("Could not instantiate user record of type "
+ + userTypeName);
+ }
+ }
+
+ public void map(LongWritable key, Text val,
+ OutputCollector<Text, NullWritable> out, Reporter r) throws IOException {
+
+ // Try to set the field.
+ userRecord.setField(setCol, setVal);
+ Map<String, Object> fieldVals = userRecord.getFieldMap();
+ if (!fieldVals.get(setCol).equals(setVal)) {
+ throw new IOException("Could not set column value! Got back "
+ + fieldVals.get(setCol));
+ } else {
+ LOG.info("Correctly changed value for col " + setCol + " to " + setVal);
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/ExportJobTestCase.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/ExportJobTestCase.java b/src/test/org/apache/sqoop/testutil/ExportJobTestCase.java
new file mode 100644
index 0000000..aa5960e
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/ExportJobTestCase.java
@@ -0,0 +1,366 @@
+/**
+ * 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.testutil;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.StringUtils;
+import org.junit.Before;
+
+import org.apache.sqoop.Sqoop;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.mapreduce.ExportOutputFormat;
+import org.apache.sqoop.tool.ExportTool;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * Class that implements common methods required for tests which export data
+ * from HDFS to databases, to verify correct export.
+ */
+public abstract class ExportJobTestCase extends BaseSqoopTestCase {
+
+ public static final Log LOG = LogFactory.getLog(
+ ExportJobTestCase.class.getName());
+
+ @Before
+ public void setUp() {
+ // start the server
+ super.setUp();
+
+ if (useHsqldbTestServer()) {
+ // throw away any existing data that might be in the database.
+ try {
+ this.getTestServer().dropExistingSchema();
+ } catch (SQLException sqlE) {
+ fail(sqlE.toString());
+ }
+ }
+ }
+
+ protected String getTablePrefix() {
+ return "EXPORT_TABLE_";
+ }
+
+ /**
+ * @return the maximum rows to fold into an INSERT statement.
+ * HSQLDB can only support the single-row INSERT syntax. Other databases
+ * can support greater numbers of rows per statement.
+ */
+ protected int getMaxRowsPerStatement() {
+ return 1;
+ }
+
+ /**
+ * Create the argv to pass to Sqoop.
+ * @param includeHadoopFlags if true, then include -D various.settings=values
+ * @param rowsPerStmt number of rows to export in a single INSERT statement.
+ * @param statementsPerTx ## of statements to use in a transaction.
+ * @return the argv as an array of strings.
+ */
+ protected String [] getArgv(boolean includeHadoopFlags,
+ int rowsPerStmt, int statementsPerTx, String... additionalArgv) {
+ ArrayList<String> args = new ArrayList<String>();
+
+ if (includeHadoopFlags) {
+ CommonArgs.addHadoopFlags(args);
+ args.add("-D");
+ int realRowsPerStmt = Math.min(rowsPerStmt, getMaxRowsPerStatement());
+ if (realRowsPerStmt != rowsPerStmt) {
+ LOG.warn("Rows per statement set to " + realRowsPerStmt
+ + " by getMaxRowsPerStatement() limit.");
+ }
+ args.add(ExportOutputFormat.RECORDS_PER_STATEMENT_KEY + "="
+ + realRowsPerStmt);
+ args.add("-D");
+ args.add(ExportOutputFormat.STATEMENTS_PER_TRANSACTION_KEY + "="
+ + statementsPerTx);
+ }
+
+ // Any additional Hadoop flags (-D foo=bar) are prepended.
+ if (null != additionalArgv) {
+ boolean prevIsFlag = false;
+ for (String arg : additionalArgv) {
+ if (arg.equals("-D")) {
+ args.add(arg);
+ prevIsFlag = true;
+ } else if (prevIsFlag) {
+ args.add(arg);
+ prevIsFlag = false;
+ }
+ }
+ }
+ boolean isHCatJob = false;
+ // The sqoop-specific additional args are then added.
+ if (null != additionalArgv) {
+ boolean prevIsFlag = false;
+ for (String arg : additionalArgv) {
+ if (arg.equals("-D")) {
+ prevIsFlag = true;
+ continue;
+ } else if (prevIsFlag) {
+ prevIsFlag = false;
+ continue;
+ } else {
+ // normal argument.
+ if (!isHCatJob && arg.equals("--hcatalog-table")) {
+ isHCatJob = true;
+ }
+ args.add(arg);
+ }
+ }
+ }
+
+ if (usesSQLtable()) {
+ args.add("--table");
+ args.add(getTableName());
+ }
+ // Only add export-dir if hcatalog-table is not there in additional argv
+ if (!isHCatJob) {
+ args.add("--export-dir");
+ args.add(getTablePath().toString());
+ }
+ args.add("--connect");
+ args.add(getConnectString());
+ args.add("--fields-terminated-by");
+ args.add("\\t");
+ args.add("--lines-terminated-by");
+ args.add("\\n");
+ args.add("-m");
+ args.add("1");
+
+ LOG.debug("args:");
+ for (String a : args) {
+ LOG.debug(" " + a);
+ }
+
+ return args.toArray(new String[0]);
+ }
+
+ protected boolean usesSQLtable() {
+ return true;
+ }
+
+ /** When exporting text columns, what should the text contain? */
+ protected String getMsgPrefix() {
+ return "textfield";
+ }
+
+
+ /** @return the minimum 'id' value in the table */
+ protected int getMinRowId(Connection conn) throws SQLException {
+ PreparedStatement statement = conn.prepareStatement(
+ "SELECT MIN(\"ID\") FROM " + getTableName(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ int minVal = 0;
+ try {
+ ResultSet rs = statement.executeQuery();
+ try {
+ rs.next();
+ minVal = rs.getInt(1);
+ } finally {
+ rs.close();
+ }
+ } finally {
+ statement.close();
+ }
+
+ return minVal;
+ }
+
+ /** @return the maximum 'id' value in the table */
+ protected int getMaxRowId(Connection conn) throws SQLException {
+ PreparedStatement statement = conn.prepareStatement(
+ "SELECT MAX(\"ID\") FROM " + getTableName(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ int maxVal = 0;
+ try {
+ ResultSet rs = statement.executeQuery();
+ try {
+ rs.next();
+ maxVal = rs.getInt(1);
+ } finally {
+ rs.close();
+ }
+ } finally {
+ statement.close();
+ }
+
+ return maxVal;
+ }
+
+ /**
+ * Check that we got back the expected row set.
+ * @param expectedNumRecords The number of records we expected to load
+ * into the database.
+ */
+ protected void verifyExport(int expectedNumRecords)
+ throws IOException, SQLException {
+ Connection conn = getConnection();
+ verifyExport(expectedNumRecords, conn);
+ }
+
+ /**
+ * Check that we got back the expected row set.
+ * @param expectedNumRecords The number of records we expected to load
+ * into the database.
+ * @param conn the db connection to use.
+ */
+ protected void verifyExport(int expectedNumRecords, Connection conn)
+ throws IOException, SQLException {
+ LOG.info("Verifying export: " + getTableName());
+ // Check that we got back the correct number of records.
+ PreparedStatement statement = conn.prepareStatement(
+ "SELECT COUNT(*) FROM " + getTableName(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ int actualNumRecords = 0;
+ ResultSet rs = null;
+ try {
+ rs = statement.executeQuery();
+ try {
+ rs.next();
+ actualNumRecords = rs.getInt(1);
+ } finally {
+ rs.close();
+ }
+ } finally {
+ statement.close();
+ }
+
+ assertEquals("Got back unexpected row count", expectedNumRecords,
+ actualNumRecords);
+
+ if (expectedNumRecords == 0) {
+ return; // Nothing more to verify.
+ }
+
+ // Check that we start with row 0.
+ int minVal = getMinRowId(conn);
+ assertEquals("Minimum row was not zero", 0, minVal);
+
+ // Check that the last row we loaded is numRows - 1
+ int maxVal = getMaxRowId(conn);
+ assertEquals("Maximum row had invalid id", expectedNumRecords - 1, maxVal);
+
+ // Check that the string values associated with these points match up.
+ statement = conn.prepareStatement("SELECT \"MSG\" FROM " + getTableName()
+ + " WHERE \"ID\" = " + minVal,
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ String minMsg = "";
+ try {
+ rs = statement.executeQuery();
+ try {
+ rs.next();
+ minMsg = rs.getString(1);
+ } finally {
+ rs.close();
+ }
+ } finally {
+ statement.close();
+ }
+
+ assertEquals("Invalid msg field for min value", getMsgPrefix() + minVal,
+ minMsg);
+
+ statement = conn.prepareStatement("SELECT \"MSG\" FROM " + getTableName()
+ + " WHERE \"ID\" = " + maxVal,
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ String maxMsg = "";
+ try {
+ rs = statement.executeQuery();
+ try {
+ rs.next();
+ maxMsg = rs.getString(1);
+ } finally {
+ rs.close();
+ }
+ } finally {
+ statement.close();
+ }
+
+ assertEquals("Invalid msg field for min value", getMsgPrefix() + maxVal, maxMsg);
+ }
+
+ // Verify Export Method For checking update's : Issue [SQOOP-2846]
+ protected void verifyExport(String expectedValue) throws IOException, SQLException {
+ Connection conn = getConnection();
+ LOG.info("Verifying export: " + getTableName());
+ // Check that we got back the correct number of records.
+ PreparedStatement statement = conn.prepareStatement("SELECT MSG FROM " + getTableName(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ String actualValue = null;
+ ResultSet rs = null;
+ try {
+ rs = statement.executeQuery();
+ try {
+ rs.next();
+ actualValue = rs.getString(1);
+ } finally {
+ rs.close();
+ }
+ } finally {
+ statement.close();
+ }
+
+ assertEquals("Got back unexpected row count", expectedValue, actualValue);
+ }
+
+ /**
+ * Run a MapReduce-based export (using the argv provided to control
+ * execution).
+ * @return the generated jar filename
+ */
+ protected List<String> runExport(String [] argv) throws IOException {
+ // run the tool through the normal entry-point.
+ int ret;
+ List<String> generatedJars = null;
+ try {
+ ExportTool exporter = new ExportTool();
+ Configuration conf = getConf();
+ //Need to disable OraOop for existing tests
+ conf.set("oraoop.disabled", "true");
+ SqoopOptions opts = getSqoopOptions(conf);
+ Sqoop sqoop = new Sqoop(exporter, conf, opts);
+ ret = Sqoop.runSqoop(sqoop, argv);
+ generatedJars = exporter.getGeneratedJarFiles();
+ } catch (Exception e) {
+ LOG.error("Got exception running Sqoop: "
+ + StringUtils.stringifyException(e));
+ ret = 1;
+ }
+
+ // expect a successful return.
+ if (0 != ret) {
+ throw new IOException("Failure during job; return status " + ret);
+ }
+
+ return generatedJars;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/HsqldbTestServer.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/HsqldbTestServer.java b/src/test/org/apache/sqoop/testutil/HsqldbTestServer.java
new file mode 100644
index 0000000..c63a8f2
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/HsqldbTestServer.java
@@ -0,0 +1,294 @@
+/**
+ * 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.testutil;
+
+import java.util.Arrays;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.hsqldb.Server;
+
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.manager.ConnManager;
+import org.apache.sqoop.manager.HsqldbManager;
+
+/**
+ * Create a simple hsqldb server and schema to use for testing.
+ */
+public class HsqldbTestServer {
+ public static final Log LOG =
+ LogFactory.getLog(HsqldbTestServer.class.getName());
+
+ // singleton server instance.
+ private static Server server;
+
+ // if -Dhsql.server.host hasn't been set to something like
+ // hsql://localhost.localdomain/ a defaul in-mem DB will be used
+ private static final String IN_MEM = "mem:";
+ public static String getServerHost() {
+ String host = System.getProperty("hsql.server.host", IN_MEM);
+ if (!host.endsWith("/")) { host += "/"; }
+ return host;
+ }
+
+ private static boolean inMemoryDB = IN_MEM.equals(getServerHost());
+
+ // Database name can be altered too
+ private static final String DATABASE_NAME =
+ System.getProperty("hsql.database.name", "db1");
+
+ // hsqldb always capitalizes table and column names
+ private static final String DUMMY_TABLE_NAME = "TWOINTTABLE";
+ private static final String [] TWO_INT_TABLE_FIELDS = {
+ "INTFIELD1",
+ "INTFIELD2",
+ };
+
+ private static final String EMPLOYEE_TABLE_NAME = "EMPLOYEES";
+
+ private static final String DB_URL = "jdbc:hsqldb:"
+ + getServerHost() + DATABASE_NAME;
+ private static final String DRIVER_CLASS = "org.hsqldb.jdbcDriver";
+
+ // all user-created HSQLDB tables are in the "PUBLIC" schema when connected
+ // to a database.
+ private static final String HSQLDB_SCHEMA_NAME = "PUBLIC";
+
+ public static String getSchemaName() {
+ return HSQLDB_SCHEMA_NAME;
+ }
+
+ public static String [] getFieldNames() {
+ return Arrays.copyOf(TWO_INT_TABLE_FIELDS, TWO_INT_TABLE_FIELDS.length);
+ }
+
+ public static String getUrl() {
+ return DB_URL;
+ }
+
+ public static String getTableName() {
+ return DUMMY_TABLE_NAME;
+ }
+
+ public static String getDatabaseName() {
+ return DATABASE_NAME;
+ }
+
+ /**
+ * start the server.
+ */
+ public void start() {
+ if (null == server) {
+ LOG.info("Starting new hsqldb server; database=" + DATABASE_NAME);
+ String tmpDir = System.getProperty("test.build.data", "/tmp/");
+ String dbLocation = tmpDir + "/sqoop/testdb.file";
+ if (inMemoryDB) {dbLocation = IN_MEM; }
+ server = new Server();
+
+ server.setDatabaseName(0, DATABASE_NAME);
+ server.putPropertiesFromString("database.0=" + dbLocation
+ + ";no_system_exit=true");
+ server.start();
+ }
+ }
+
+ public void stop() {
+ if (null == server) {
+ return;
+ }
+ server.stop();
+ server = null;
+ }
+
+ public Connection getConnection() throws SQLException {
+ return getConnection(null, null);
+ }
+
+ public Connection getConnection(String user, String password) throws SQLException {
+ try {
+ Class.forName(DRIVER_CLASS);
+ } catch (ClassNotFoundException cnfe) {
+ LOG.error("Could not get connection; driver class not found: "
+ + DRIVER_CLASS);
+ return null;
+ }
+
+ Connection connection = DriverManager.getConnection(DB_URL, user, password);
+ connection.setAutoCommit(false);
+ return connection;
+ }
+
+ /**
+ * Returns database URL for the server instance.
+ * @return String representation of DB_URL
+ */
+ public static String getDbUrl() {
+ return DB_URL;
+ }
+
+ /**
+ * Create a table.
+ */
+ public void createSchema() throws SQLException {
+
+ Connection connection = null;
+ Statement st = null;
+
+ try {
+ connection = getConnection();
+
+ st = connection.createStatement();
+ st.executeUpdate("DROP TABLE \"" + DUMMY_TABLE_NAME + "\" IF EXISTS");
+ st.executeUpdate("CREATE TABLE \"" + DUMMY_TABLE_NAME + "\"(intField1 INT, intField2 INT)");
+
+ connection.commit();
+ } finally {
+ if (null != st) {
+ st.close();
+ }
+
+ if (null != connection) {
+ connection.close();
+ }
+ }
+ }
+
+ /**
+ * @return the sum of the integers in the first column of TWOINTTABLE.
+ */
+ public static int getFirstColSum() {
+ return 1 + 3 + 5 + 7;
+ }
+
+ /**
+ * Fill the table with some data.
+ */
+ public void populateData() throws SQLException {
+
+ Connection connection = null;
+ Statement st = null;
+
+ try {
+ connection = getConnection();
+
+ st = connection.createStatement();
+ st.executeUpdate("INSERT INTO \"" + DUMMY_TABLE_NAME + "\" VALUES(1, 8)");
+ st.executeUpdate("INSERT INTO \"" + DUMMY_TABLE_NAME + "\" VALUES(3, 6)");
+ st.executeUpdate("INSERT INTO \"" + DUMMY_TABLE_NAME + "\" VALUES(5, 4)");
+ st.executeUpdate("INSERT INTO \"" + DUMMY_TABLE_NAME + "\" VALUES(7, 2)");
+
+ connection.commit();
+ } finally {
+ if (null != st) {
+ st.close();
+ }
+
+ if (null != connection) {
+ connection.close();
+ }
+ }
+ }
+
+ public void createEmployeeDemo() throws SQLException, ClassNotFoundException {
+ Class.forName(DRIVER_CLASS);
+
+ Connection connection = null;
+ Statement st = null;
+
+ try {
+ connection = getConnection();
+
+ st = connection.createStatement();
+ st.executeUpdate("DROP TABLE \"" + EMPLOYEE_TABLE_NAME + "\" IF EXISTS");
+ st.executeUpdate("CREATE TABLE \"" + EMPLOYEE_TABLE_NAME + "\"(emp_id INT NOT NULL PRIMARY KEY, name VARCHAR(64))");
+
+ st.executeUpdate("INSERT INTO \"" + EMPLOYEE_TABLE_NAME + "\" VALUES(1, 'Aaron')");
+ st.executeUpdate("INSERT INTO \"" + EMPLOYEE_TABLE_NAME + "\" VALUES(2, 'Joe')");
+ st.executeUpdate("INSERT INTO \"" + EMPLOYEE_TABLE_NAME + "\" VALUES(3, 'Jim')");
+ st.executeUpdate("INSERT INTO \"" + EMPLOYEE_TABLE_NAME + "\" VALUES(4, 'Lisa')");
+
+ connection.commit();
+ } finally {
+ if (null != st) {
+ st.close();
+ }
+
+ if (null != connection) {
+ connection.close();
+ }
+ }
+ }
+
+ /**
+ * Delete any existing tables.
+ */
+ public void dropExistingSchema() throws SQLException {
+ ConnManager mgr = getManager();
+ String [] tables = mgr.listTables();
+ if (null != tables) {
+ Connection conn = mgr.getConnection();
+ for (String table : tables) {
+ Statement s = conn.createStatement();
+ try {
+ s.executeUpdate("DROP TABLE \"" + table + "\"");
+ conn.commit();
+ } finally {
+ s.close();
+ }
+ }
+ }
+ }
+
+ /**
+ * Creates an hsqldb server, fills it with tables and data.
+ */
+ public void resetServer() throws ClassNotFoundException, SQLException {
+ start();
+ dropExistingSchema();
+ createSchema();
+ populateData();
+ }
+
+ public SqoopOptions getSqoopOptions() {
+ return new SqoopOptions(HsqldbTestServer.getUrl(),
+ HsqldbTestServer.getTableName());
+ }
+
+ public ConnManager getManager() {
+ return new HsqldbManager(getSqoopOptions());
+ }
+
+ public void createNewUser(String username, String password) throws SQLException {
+ try (Connection connection = getConnection(); Statement statement = connection.createStatement()) {
+ statement.executeUpdate(String.format("CREATE USER %s PASSWORD %s ADMIN", username, password));
+ }
+ }
+
+ public void changePasswordForUser(String username, String newPassword) throws SQLException {
+ try (Connection connection = getConnection(); Statement statement = connection.createStatement()) {
+ statement.executeUpdate(String.format("ALTER USER %s SET PASSWORD %s", username, newPassword));
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/ImportJobTestCase.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/ImportJobTestCase.java b/src/test/org/apache/sqoop/testutil/ImportJobTestCase.java
new file mode 100644
index 0000000..dbefe20
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/ImportJobTestCase.java
@@ -0,0 +1,245 @@
+/**
+ * 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.testutil;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.Sqoop;
+import org.apache.sqoop.orm.CompilationManager;
+import org.apache.sqoop.tool.SqoopTool;
+import org.apache.sqoop.tool.ImportTool;
+import org.apache.sqoop.util.ClassLoaderStack;
+import org.junit.Before;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * Class that implements common methods required for tests which import data
+ * from SQL into HDFS and verify correct import.
+ */
+public abstract class ImportJobTestCase extends BaseSqoopTestCase {
+
+ public static final Log LOG = LogFactory.getLog(
+ ImportJobTestCase.class.getName());
+
+ @Before
+ public void setUp() {
+ super.setUp();
+ removeTableDir();
+ }
+
+ protected String getTablePrefix() {
+ return "IMPORT_TABLE_";
+ }
+
+ /**
+ * @return a list of additional args to pass to the sqoop command line.
+ */
+ protected List<String> getExtraArgs(Configuration conf) {
+ return new ArrayList<String>();
+ }
+
+ /**
+ * Create the argv to pass to Sqoop.
+ * @param includeHadoopFlags if true, then include -D various.settings=values
+ * @param colNames the columns to import. If null, all columns are used.
+ * @param conf a Configuration specifying additional properties to use when
+ * determining the arguments.
+ * @return the argv as an array of strings.
+ */
+ protected String [] getArgv(boolean includeHadoopFlags, String [] colNames,
+ Configuration conf) {
+ if (null == colNames) {
+ colNames = getColNames();
+ }
+
+ String splitByCol = colNames[0];
+ String columnsString = "";
+ for (String col : colNames) {
+ columnsString += col + ",";
+ }
+
+ ArrayList<String> args = new ArrayList<String>();
+
+ if (includeHadoopFlags) {
+ CommonArgs.addHadoopFlags(args);
+ }
+
+ args.add("--table");
+ args.add(getTableName());
+ args.add("--columns");
+ args.add(columnsString);
+ args.add("--split-by");
+ args.add(splitByCol);
+ args.add("--warehouse-dir");
+ args.add(getWarehouseDir());
+ args.add("--connect");
+ args.add(getConnectString());
+ args.add("--as-sequencefile");
+ args.add("--num-mappers");
+ args.add("1");
+
+ args.addAll(getExtraArgs(conf));
+
+ return args.toArray(new String[0]);
+ }
+
+ /**
+ * Do a MapReduce-based import of the table and verify that the results
+ * were imported as expected. (tests readFields(ResultSet) and toString())
+ * @param expectedVal the value we injected into the table.
+ * @param importCols the columns to import. If null, all columns are used.
+ */
+ protected void verifyImport(String expectedVal, String [] importCols) {
+
+ // paths to where our output file will wind up.
+ Path tableDirPath = getTablePath();
+
+ removeTableDir();
+
+ Configuration conf = getConf();
+ //Need to disable OraOop for existing tests
+ conf.set("oraoop.disabled", "true");
+ SqoopOptions opts = getSqoopOptions(conf);
+
+ // run the tool through the normal entry-point.
+ int ret;
+ try {
+ Sqoop importer = new Sqoop(new ImportTool(), conf, opts);
+ ret = Sqoop.runSqoop(importer, getArgv(true, importCols, conf));
+ } catch (Exception e) {
+ LOG.error("Got exception running Sqoop: " + e.toString());
+ throw new RuntimeException(e);
+ }
+
+ // expect a successful return.
+ assertEquals("Failure during job", 0, ret);
+
+ opts = getSqoopOptions(conf);
+ try {
+ ImportTool importTool = new ImportTool();
+ opts = importTool.parseArguments(getArgv(false, importCols, conf), conf,
+ opts, true);
+ } catch (Exception e) {
+ fail(e.toString());
+ }
+
+ CompilationManager compileMgr = new CompilationManager(opts);
+ String jarFileName = compileMgr.getJarFilename();
+ ClassLoader prevClassLoader = null;
+ try {
+ prevClassLoader = ClassLoaderStack.addJarFile(jarFileName,
+ getTableName());
+
+ // Now open and check all part-files in the table path until we find
+ // a non-empty one that we can verify contains the value.
+ if (!BaseSqoopTestCase.isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus [] stats = fs.listStatus(tableDirPath);
+
+ if (stats == null || stats.length == 0) {
+ fail("Error: no files in " + tableDirPath);
+ }
+
+ boolean foundRecord = false;
+ for (FileStatus stat : stats) {
+ if (!stat.getPath().getName().startsWith("part-")
+ && !stat.getPath().getName().startsWith("data-")) {
+ // This isn't a data file. Ignore it.
+ continue;
+ }
+
+ try {
+ Object readValue = SeqFileReader.getFirstValue(
+ stat.getPath().toString());
+ LOG.info("Read back from sequencefile: " + readValue);
+ foundRecord = true;
+ // Add trailing '\n' to expected value since SqoopRecord.toString()
+ // encodes the record delim.
+ if (null == expectedVal) {
+ assertEquals("Error validating result from SeqFile", "null\n",
+ readValue.toString());
+ } else {
+ assertEquals("Error validating result from SeqFile",
+ expectedVal + "\n", readValue.toString());
+ }
+ } catch (EOFException eoe) {
+ // EOF in a file isn't necessarily a problem. We may have some
+ // empty sequence files, which will throw this. Just continue
+ // in the loop.
+ }
+ }
+
+ if (!foundRecord) {
+ fail("Couldn't read any records from SequenceFiles");
+ }
+ } catch (IOException ioe) {
+ fail("IOException: " + ioe.toString());
+ } finally {
+ if (null != prevClassLoader) {
+ ClassLoaderStack.setCurrentClassLoader(prevClassLoader);
+ }
+ }
+ }
+
+ /**
+ * Run a MapReduce-based import (using the argv provided to control
+ * execution).
+ */
+ protected void runImport(SqoopTool tool, String [] argv) throws IOException {
+ // run the tool through the normal entry-point.
+ int ret;
+ try {
+ Configuration conf = getConf();
+ //Need to disable OraOop for existing tests
+ conf.set("oraoop.disabled", "true");
+ SqoopOptions opts = getSqoopOptions(conf);
+ Sqoop sqoop = new Sqoop(tool, conf, opts);
+ ret = Sqoop.runSqoop(sqoop, argv);
+ } catch (Exception e) {
+ LOG.error("Got exception running Sqoop: " + e.toString());
+ e.printStackTrace();
+ ret = 1;
+ }
+
+ // expect a successful return.
+ if (0 != ret) {
+ throw new IOException("Failure during job; return status " + ret);
+ }
+ }
+
+ /** run an import using the default ImportTool. */
+ protected void runImport(String [] argv) throws IOException {
+ runImport(new ImportTool(), argv);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/InjectableConnManager.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/InjectableConnManager.java b/src/test/org/apache/sqoop/testutil/InjectableConnManager.java
new file mode 100644
index 0000000..b4917fe
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/InjectableConnManager.java
@@ -0,0 +1,92 @@
+/**
+ * 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.testutil;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.manager.HsqldbManager;
+import org.apache.sqoop.manager.ImportJobContext;
+import org.apache.sqoop.mapreduce.ImportJobBase;
+import org.apache.sqoop.util.ImportException;
+import org.apache.hadoop.util.ReflectionUtils;
+
+/**
+ * A ConnManager that uses dependency injection to control "import jobs"
+ * that are mocked for testing.
+ */
+public class InjectableConnManager extends HsqldbManager {
+
+ // The Configuration is used to control the injected classes.
+ public static final String MAPPER_KEY = "sqoop.inject.mapper.class";
+ public static final String INPUT_FORMAT_KEY =
+ "sqoop.inject.input.format.class";
+ public static final String OUTPUT_FORMAT_KEY =
+ "sqoop.inject.output.format.class";
+ public static final String IMPORT_JOB_KEY =
+ "sqoop.inject.import.job.class";
+
+ public InjectableConnManager(final SqoopOptions options) {
+ super(options);
+ }
+
+ /**
+ * Allow the user to inject custom mapper, input, and output formats
+ * into the importTable() process.
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public void importTable(ImportJobContext context)
+ throws IOException, ImportException {
+
+ SqoopOptions options = context.getOptions();
+ Configuration conf = options.getConf();
+
+ Class<? extends Mapper> mapperClass = (Class<? extends Mapper>)
+ conf.getClass(MAPPER_KEY, Mapper.class);
+ Class<? extends InputFormat> ifClass = (Class<? extends InputFormat>)
+ conf.getClass(INPUT_FORMAT_KEY, TextInputFormat.class);
+ Class<? extends OutputFormat> ofClass = (Class<? extends OutputFormat>)
+ conf.getClass(OUTPUT_FORMAT_KEY, TextOutputFormat.class);
+
+ Class<? extends ImportJobBase> jobClass = (Class<? extends ImportJobBase>)
+ conf.getClass(IMPORT_JOB_KEY, ImportJobBase.class);
+
+ String tableName = context.getTableName();
+
+ // Instantiate the user's chosen ImportJobBase instance.
+ ImportJobBase importJob = ReflectionUtils.newInstance(jobClass, conf);
+
+ // And configure the dependencies to inject
+ importJob.setOptions(options);
+ importJob.setMapperClass(mapperClass);
+ importJob.setInputFormatClass(ifClass);
+ importJob.setOutputFormatClass(ofClass);
+
+ importJob.runImport(tableName, context.getJarFile(),
+ getSplitColumn(options, tableName), conf);
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/InjectableManagerFactory.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/InjectableManagerFactory.java b/src/test/org/apache/sqoop/testutil/InjectableManagerFactory.java
new file mode 100644
index 0000000..48c3461
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/InjectableManagerFactory.java
@@ -0,0 +1,36 @@
+/**
+ * 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.testutil;
+
+import org.apache.sqoop.manager.ConnManager;
+import org.apache.sqoop.manager.ManagerFactory;
+import org.apache.sqoop.metastore.JobData;
+import org.apache.sqoop.testutil.InjectableConnManager;
+
+/**
+ * ManagerFactory that is used for testing; this accepts any
+ * connection string.
+ */
+public class InjectableManagerFactory extends ManagerFactory {
+
+ public ConnManager accept(JobData data) {
+ // Always accept and use the injectable manager.
+ return new InjectableConnManager(data.getSqoopOptions());
+ }
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/testutil/LobAvroImportTestCase.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/testutil/LobAvroImportTestCase.java b/src/test/org/apache/sqoop/testutil/LobAvroImportTestCase.java
new file mode 100644
index 0000000..20d6151
--- /dev/null
+++ b/src/test/org/apache/sqoop/testutil/LobAvroImportTestCase.java
@@ -0,0 +1,397 @@
+/**
+ * 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.testutil;
+
+import org.junit.After;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runners.MethodSorters;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.ArrayList;
+
+import org.apache.avro.file.DataFileConstants;
+import org.apache.avro.file.DataFileReader;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.DatumReader;
+import org.apache.avro.mapred.FsInput;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.sqoop.io.CodecMap;
+import org.apache.sqoop.lib.BlobRef;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests BLOB/CLOB import for Avro.
+ */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public abstract class LobAvroImportTestCase extends ImportJobTestCase {
+
+ private Log log;
+
+ public LobAvroImportTestCase() {
+ this.log = LogFactory.getLog(LobAvroImportTestCase.class.getName());
+ }
+
+ /**
+ * @return the Log object to use for reporting during this test
+ */
+ protected abstract Log getLogger();
+
+ /**
+ * @return a "friendly" name for the database. e.g "mysql" or "oracle".
+ */
+ protected abstract String getDbFriendlyName();
+
+ @Override
+ protected String getTablePrefix() {
+ return "LOB_" + getDbFriendlyName().toUpperCase() + "_";
+ }
+
+ @Override
+ protected boolean useHsqldbTestServer() {
+ // Hsqldb does not support BLOB/CLOB
+ return false;
+ }
+
+ @After
+ public void tearDown() {
+ try {
+ // Clean up the database on our way out.
+ dropTableIfExists(getTableName());
+ } catch (SQLException e) {
+ log.warn("Error trying to drop table '" + getTableName()
+ + "' on tearDown: " + e);
+ }
+ super.tearDown();
+ }
+
+ protected String [] getArgv(String ... additionalArgs) {
+ // Import every column of the table
+ String [] colNames = getColNames();
+ String splitByCol = colNames[0];
+ String columnsString = "";
+ for (String col : colNames) {
+ columnsString += col + ",";
+ }
+
+ ArrayList<String> args = new ArrayList<String>();
+
+ CommonArgs.addHadoopFlags(args);
+
+ args.add("--table");
+ args.add(getTableName());
+ args.add("--columns");
+ args.add(columnsString);
+ args.add("--split-by");
+ args.add(splitByCol);
+ args.add("--warehouse-dir");
+ args.add(getWarehouseDir());
+ args.add("--connect");
+ args.add(getConnectString());
+ args.add("--as-avrodatafile");
+ args.add("--num-mappers");
+ args.add("1");
+
+ for (String arg : additionalArgs) {
+ args.add(arg);
+ }
+
+ return args.toArray(new String[0]);
+ }
+
+ protected String getBlobType() {
+ return "BLOB";
+ }
+
+ protected String getBlobInsertStr(String blobData) {
+ return "'" + blobData + "'";
+ }
+
+ /**
+ * Return the current table number as a string. In test, table number is used
+ * to name .lob files.
+ * @return current table number.
+ */
+ private String getTableNum() {
+ return getTableName().substring(getTablePrefix().length());
+ }
+
+ /**
+ * Return an instance of DataFileReader for the given filename.
+ * @param filename path that we're opening a reader for.
+ * @return instance of DataFileReader.
+ * @throws IOException
+ */
+ private DataFileReader<GenericRecord> read(Path filename)
+ throws IOException {
+ Configuration conf = getConf();
+ if (!BaseSqoopTestCase.isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FsInput fsInput = new FsInput(filename, conf);
+ DatumReader<GenericRecord> datumReader =
+ new GenericDatumReader<GenericRecord>();
+ return new DataFileReader<GenericRecord>(fsInput, datumReader);
+ }
+
+ /** Import blob data that is smaller than inline lob limit. Blob data
+ * should be saved as Avro bytes.
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testBlobAvroImportInline() throws IOException, SQLException {
+ String [] types = { getBlobType() };
+ String expectedVal = "This is short BLOB data";
+ String [] vals = { getBlobInsertStr(expectedVal) };
+
+ createTableWithColTypes(types, vals);
+
+ runImport(getArgv());
+
+ Path outputFile = new Path(getTablePath(), "part-m-00000.avro");
+ DataFileReader<GenericRecord> reader = read(outputFile);
+ GenericRecord record = reader.next();
+
+ // Verify that blob data is imported as Avro bytes.
+ ByteBuffer buf = (ByteBuffer) record.get(getColName(0));
+ String returnVal = new String(buf.array());
+
+ assertEquals(getColName(0), expectedVal, returnVal);
+ }
+
+ /**
+ * Import blob data that is larger than inline lob limit. The reference file
+ * should be saved as Avro bytes. Blob data should be saved in LOB file
+ * format.
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testBlobAvroImportExternal() throws IOException, SQLException {
+ String [] types = { getBlobType() };
+ String data = "This is short BLOB data";
+ String [] vals = { getBlobInsertStr(data) };
+
+ createTableWithColTypes(types, vals);
+
+ // Set inline lob limit to a small value so that blob data will be
+ // written to an external file.
+ runImport(getArgv("--inline-lob-limit", "1"));
+
+ Path outputFile = new Path(getTablePath(), "part-m-00000.avro");
+ DataFileReader<GenericRecord> reader = read(outputFile);
+ GenericRecord record = reader.next();
+
+ // Verify that the reference file is written in Avro bytes.
+ ByteBuffer buf = (ByteBuffer) record.get(getColName(0));
+ String returnVal = new String(buf.array());
+ String expectedStart = "externalLob(lf,_lob/large_obj";
+ String expectedEnd = getTableNum() + "_m_0000000.lob,68,"
+ + data.length() + ")";
+
+ assertNotNull(returnVal);
+ assertTrue("ExpectedStart: " + expectedStart + ", value: " + returnVal, returnVal.startsWith(expectedStart));
+ assertTrue("ExpectedEnd: " + expectedEnd + ", value: " + returnVal, returnVal.endsWith(expectedEnd));
+
+ // Verify that blob data stored in the external lob file is correct.
+ BlobRef br = BlobRef.parse(returnVal);
+ Path lobFileDir = new Path(getWarehouseDir(), getTableName());
+ InputStream in = br.getDataStream(getConf(), lobFileDir);
+
+ byte [] bufArray = new byte[data.length()];
+ int chars = in.read(bufArray);
+ in.close();
+
+ assertEquals(chars, data.length());
+
+ returnVal = new String(bufArray);
+ String expectedVal = data;
+
+ assertEquals(getColName(0), returnVal, expectedVal);
+ }
+
+ /**
+ * Import blob data that is smaller than inline lob limit and compress with
+ * deflate codec. Blob data should be encoded and saved as Avro bytes.
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testBlobCompressedAvroImportInline()
+ throws IOException, SQLException {
+ String [] types = { getBlobType() };
+ String expectedVal = "This is short BLOB data";
+ String [] vals = { getBlobInsertStr(expectedVal) };
+
+ createTableWithColTypes(types, vals);
+
+ runImport(getArgv("--compression-codec", CodecMap.DEFLATE));
+
+ Path outputFile = new Path(getTablePath(), "part-m-00000.avro");
+ DataFileReader<GenericRecord> reader = read(outputFile);
+ GenericRecord record = reader.next();
+
+ // Verify that the data block of the Avro file is compressed with deflate
+ // codec.
+ assertEquals(CodecMap.DEFLATE,
+ reader.getMetaString(DataFileConstants.CODEC));
+
+ // Verify that all columns are imported correctly.
+ ByteBuffer buf = (ByteBuffer) record.get(getColName(0));
+ String returnVal = new String(buf.array());
+
+ assertEquals(getColName(0), expectedVal, returnVal);
+ }
+
+ /**
+ * Import blob data that is larger than inline lob limit and compress with
+ * deflate codec. The reference file should be encoded and saved as Avro
+ * bytes. Blob data should be saved in LOB file format without compression.
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testBlobCompressedAvroImportExternal()
+ throws IOException, SQLException {
+ String [] types = { getBlobType() };
+ String data = "This is short BLOB data";
+ String [] vals = { getBlobInsertStr(data) };
+
+ createTableWithColTypes(types, vals);
+
+ // Set inline lob limit to a small value so that blob data will be
+ // written to an external file.
+ runImport(getArgv(
+ "--inline-lob-limit", "1", "--compression-codec", CodecMap.DEFLATE));
+
+ Path outputFile = new Path(getTablePath(), "part-m-00000.avro");
+ DataFileReader<GenericRecord> reader = read(outputFile);
+ GenericRecord record = reader.next();
+
+ // Verify that the data block of the Avro file is compressed with deflate
+ // codec.
+ assertEquals(CodecMap.DEFLATE,
+ reader.getMetaString(DataFileConstants.CODEC));
+
+ // Verify that the reference file is written in Avro bytes.
+ ByteBuffer buf = (ByteBuffer) record.get(getColName(0));
+ String returnVal = new String(buf.array());
+ String expectedStart = "externalLob(lf,_lob/large_obj";
+ String expectedEnd = getTableNum() + "_m_0000000.lob,68,"
+ + data.length() + ")";
+
+ assertNotNull(returnVal);
+ assertTrue("ExpectedStart: " + expectedStart + ", value: " + returnVal, returnVal.startsWith(expectedStart));
+ assertTrue("ExpectedEnd: " + expectedEnd + ", value: " + returnVal, returnVal.endsWith(expectedEnd));
+
+ // Verify that blob data stored in the external lob file is correct.
+ BlobRef br = BlobRef.parse(returnVal);
+ Path lobFileDir = new Path(getWarehouseDir(), getTableName());
+ InputStream in = br.getDataStream(getConf(), lobFileDir);
+
+ byte [] bufArray = new byte[data.length()];
+ int chars = in.read(bufArray);
+ in.close();
+
+ assertEquals(chars, data.length());
+
+ returnVal = new String(bufArray);
+ String expectedVal = data;
+
+ assertEquals(getColName(0), returnVal, expectedVal);
+ }
+
+ /**
+ * Import multiple columns of blob data. Blob data should be saved as Avro
+ * bytes.
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testBlobAvroImportMultiCols() throws IOException, SQLException {
+ String [] types = { getBlobType(), getBlobType(), getBlobType(), };
+ String expectedVal1 = "This is short BLOB data1";
+ String expectedVal2 = "This is short BLOB data2";
+ String expectedVal3 = "This is short BLOB data3";
+ String [] vals = { getBlobInsertStr(expectedVal1),
+ getBlobInsertStr(expectedVal2),
+ getBlobInsertStr(expectedVal3), };
+
+ createTableWithColTypes(types, vals);
+
+ runImport(getArgv());
+
+ Path outputFile = new Path(getTablePath(), "part-m-00000.avro");
+ DataFileReader<GenericRecord> reader = read(outputFile);
+ GenericRecord record = reader.next();
+
+ // Verify that all columns are imported correctly.
+ ByteBuffer buf = (ByteBuffer) record.get(getColName(0));
+ String returnVal = new String(buf.array());
+
+ assertEquals(getColName(0), expectedVal1, returnVal);
+
+ buf = (ByteBuffer) record.get(getColName(1));
+ returnVal = new String(buf.array());
+
+ assertEquals(getColName(1), expectedVal2, returnVal);
+
+ buf = (ByteBuffer) record.get(getColName(2));
+ returnVal = new String(buf.array());
+
+ assertEquals(getColName(2), expectedVal3, returnVal);
+ }
+
+ @Test
+ public void testClobAvroImportInline() throws IOException, SQLException {
+ // TODO: add tests for CLOB support for Avro import
+ }
+
+ @Test
+ public void testClobAvroImportExternal() throws IOException, SQLException {
+ // TODO: add tests for CLOB support for Avro import
+ }
+
+ @Test
+ public void testClobCompressedAvroImportInline()
+ throws IOException, SQLException {
+ // TODO: add tests for CLOB support for Avro import
+ }
+
+ @Test
+ public void testClobCompressedAvroImportExternal()
+ throws IOException, SQLException {
+ // TODO: add tests for CLOB support for Avro import
+ }
+
+ @Test
+ public void testClobAvroImportMultiCols() throws IOException, SQLException {
+ // TODO: add tests for CLOB support for Avro import
+ }
+}