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:58 UTC
[12/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/TestBoundaryQuery.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestBoundaryQuery.java b/src/test/org/apache/sqoop/TestBoundaryQuery.java
new file mode 100644
index 0000000..4461006
--- /dev/null
+++ b/src/test/org/apache/sqoop/TestBoundaryQuery.java
@@ -0,0 +1,188 @@
+/**
+ * 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;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.util.ReflectionUtils;
+
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.SqoopOptions.InvalidOptionsException;
+import org.apache.sqoop.orm.CompilationManager;
+import org.apache.sqoop.testutil.CommonArgs;
+import org.apache.sqoop.testutil.HsqldbTestServer;
+import org.apache.sqoop.testutil.ImportJobTestCase;
+import org.apache.sqoop.testutil.SeqFileReader;
+import org.apache.sqoop.tool.ImportTool;
+import org.apache.sqoop.util.ClassLoaderStack;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * Test that --boundary-query works in Sqoop.
+ */
+public class TestBoundaryQuery extends ImportJobTestCase {
+
+ /**
+ * Create the argv to pass to Sqoop.
+ * @return the argv as an array of strings.
+ */
+ protected String [] getArgv(boolean includeHadoopFlags, boolean tableImport,
+ String boundaryQuery,
+ String targetDir, String... extraArgs) {
+
+ ArrayList<String> args = new ArrayList<String>();
+
+ if (includeHadoopFlags) {
+ CommonArgs.addHadoopFlags(args);
+ }
+
+ if (tableImport) {
+ args.add("--table");
+ args.add(HsqldbTestServer.getTableName());
+ } else {
+ args.add("--query");
+ args.add("SELECT INTFIELD1, INTFIELD2 FROM "
+ + HsqldbTestServer.getTableName() + " WHERE $CONDITIONS");
+ }
+ args.add("--connect");
+ args.add(HsqldbTestServer.getUrl());
+ if (boundaryQuery != null) {
+ args.add("--boundary-query");
+ args.add(boundaryQuery);
+ }
+ args.add("--as-sequencefile");
+ args.add("--target-dir");
+ args.add(targetDir);
+ args.add("--class-name");
+ args.add(getTableName());
+ args.add("--verbose");
+ for (String extraArg : extraArgs) {
+ args.add(extraArg);
+ }
+
+ return args.toArray(new String[0]);
+ }
+
+ // this test just uses the two int table.
+ protected String getTableName() {
+ return HsqldbTestServer.getTableName();
+ }
+
+
+ /**
+ * Given a comma-delimited list of integers, grab and parse the first int.
+ * @param str a comma-delimited list of values, the first of which is an int.
+ * @return the first field in the string, cast to int
+ */
+ private int getFirstInt(String str) {
+ String [] parts = str.split(",");
+ return Integer.parseInt(parts[0]);
+ }
+
+ public void runQueryTest(String query, boolean tableImport,
+ int numExpectedResults, int expectedSum, String targetDir,
+ String... extraArgs) throws IOException {
+
+ ClassLoader prevClassLoader = null;
+ SequenceFile.Reader reader = null;
+
+ String [] argv = getArgv(true, tableImport, query, targetDir, extraArgs);
+ runImport(argv);
+ try {
+ SqoopOptions opts = new ImportTool().parseArguments(
+ getArgv(false, tableImport, query, targetDir, extraArgs),
+ null, null, true);
+
+ CompilationManager compileMgr = new CompilationManager(opts);
+ String jarFileName = compileMgr.getJarFilename();
+
+ prevClassLoader = ClassLoaderStack.addJarFile(jarFileName,
+ getTableName());
+
+ reader = SeqFileReader.getSeqFileReader(getDataFilePath().toString());
+
+ // here we can actually instantiate (k, v) pairs.
+ Configuration conf = new Configuration();
+ Object key = ReflectionUtils.newInstance(reader.getKeyClass(), conf);
+ Object val = ReflectionUtils.newInstance(reader.getValueClass(), conf);
+
+ if (reader.next(key) == null) {
+ fail("Empty SequenceFile during import");
+ }
+
+ // make sure that the value we think should be at the top, is.
+ reader.getCurrentValue(val);
+
+ // We know that these values are two ints separated by a ',' character.
+ // Since this is all dynamic, though, we don't want to actually link
+ // against the class and use its methods. So we just parse this back
+ // into int fields manually. Sum them up and ensure that we get the
+ // expected total for the first column, to verify that we got all the
+ // results from the db into the file.
+ int curSum = getFirstInt(val.toString());
+ int totalResults = 1;
+
+ // now sum up everything else in the file.
+ while (reader.next(key) != null) {
+ reader.getCurrentValue(val);
+ curSum += getFirstInt(val.toString());
+ totalResults++;
+ }
+
+ assertEquals("Total sum of first db column mismatch", expectedSum,
+ curSum);
+ assertEquals("Incorrect number of results for query", numExpectedResults,
+ totalResults);
+ } catch (InvalidOptionsException ioe) {
+ fail(ioe.toString());
+ } catch (ParseException pe) {
+ fail(pe.toString());
+ } finally {
+ IOUtils.closeStream(reader);
+
+ if (null != prevClassLoader) {
+ ClassLoaderStack.setCurrentClassLoader(prevClassLoader);
+ }
+ }
+ }
+
+ @Test
+ public void testBoundaryQuery() throws IOException {
+ System.out.println("PCYO");
+ String query = "select min(intfield1), max(intfield1) from "
+ + getTableName() +" where intfield1 in (3, 5)";
+
+ runQueryTest(query, true, 2, 8, getTablePath().toString(),
+ "--m", "1", "--split-by", "INTFIELD1");
+ }
+
+ @Test
+ public void testNoBoundaryQuerySingleMapper() throws IOException {
+
+ runQueryTest(null, false, 4, 16, getTablePath().toString(),
+ "--m", "1");
+ }
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/TestColumnTypes.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestColumnTypes.java b/src/test/org/apache/sqoop/TestColumnTypes.java
new file mode 100644
index 0000000..9674030
--- /dev/null
+++ b/src/test/org/apache/sqoop/TestColumnTypes.java
@@ -0,0 +1,80 @@
+/**
+ * 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;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.sqoop.testutil.ManagerCompatTestCase;
+
+/**
+ * Test that each of the different SQL Column types that we support
+ * can, in fact, be imported into HDFS. Test that the writable
+ * that we expect to work, does.
+ *
+ * This uses hsqldb as its test database.
+ *
+ * This requires testing:
+ * - That we can pull from the database into HDFS:
+ * readFields(ResultSet), toString()
+ * - That we can pull from mapper to reducer:
+ * write(DataOutput), readFields(DataInput)
+ * - And optionally, that we can push to the database:
+ * write(PreparedStatement)
+ */
+public class TestColumnTypes extends ManagerCompatTestCase {
+
+ public static final Log LOG = LogFactory.getLog(
+ TestColumnTypes.class.getName());
+
+ @Override
+ protected Log getLogger() {
+ return LOG;
+ }
+
+ @Override
+ protected String getDbFriendlyName() {
+ return "HSQLDB";
+ }
+
+ @Override
+ protected boolean useHsqldbTestServer() {
+ return true;
+ }
+
+ // Don't need to override getConnectString() because the default uses hsqldb.
+
+ // HSQLdb does not support these types over JDBC.
+
+ @Override
+ protected boolean supportsClob() {
+ return false;
+ }
+
+ @Override
+ protected boolean supportsBlob() {
+ return false;
+ }
+
+ @Override
+ protected String getVarBinarySeqOutput(String asInserted) {
+ return toLowerHexString(asInserted);
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/TestCompression.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestCompression.java b/src/test/org/apache/sqoop/TestCompression.java
new file mode 100644
index 0000000..6480c44
--- /dev/null
+++ b/src/test/org/apache/sqoop/TestCompression.java
@@ -0,0 +1,203 @@
+/**
+ * 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;
+
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.orm.CompilationManager;
+import org.apache.sqoop.testutil.BaseSqoopTestCase;
+import org.apache.sqoop.testutil.CommonArgs;
+import org.apache.sqoop.testutil.HsqldbTestServer;
+import org.apache.sqoop.testutil.ImportJobTestCase;
+import org.apache.sqoop.testutil.SeqFileReader;
+import org.apache.sqoop.tool.ImportTool;
+import org.apache.sqoop.util.ClassLoaderStack;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.compress.BZip2Codec;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.GzipCodec;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test that compression options (--compress, --compression-codec) work.
+ */
+public class TestCompression extends ImportJobTestCase {
+
+ /**
+ * Create the argv to pass to Sqoop.
+ * @return the argv as an array of strings.
+ */
+ protected String [] getArgv(boolean includeHadoopFlags, String [] colNames,
+ CompressionCodec codec, String fileFormat) {
+ String columnsString = "";
+ for (String col : colNames) {
+ columnsString += col + ",";
+ }
+
+ ArrayList<String> args = new ArrayList<String>();
+
+ if (includeHadoopFlags) {
+ CommonArgs.addHadoopFlags(args);
+ }
+
+ args.add("--table");
+ args.add(HsqldbTestServer.getTableName());
+ args.add("--columns");
+ args.add(columnsString);
+ args.add("--compress");
+ if (codec != null) {
+ args.add("--compression-codec");
+ args.add(codec.getClass().getName());
+ }
+ args.add("--warehouse-dir");
+ args.add(getWarehouseDir());
+ args.add("--connect");
+ args.add(HsqldbTestServer.getUrl());
+ args.add(fileFormat);
+ args.add("--num-mappers");
+ args.add("1");
+
+ return args.toArray(new String[0]);
+ }
+
+ // this test just uses the two int table.
+ protected String getTableName() {
+ return HsqldbTestServer.getTableName();
+ }
+
+ public void runSequenceFileCompressionTest(CompressionCodec codec,
+ int expectedNum) throws Exception {
+
+ String [] columns = HsqldbTestServer.getFieldNames();
+ ClassLoader prevClassLoader = null;
+ SequenceFile.Reader reader = null;
+
+ String [] argv = getArgv(true, columns, codec, "--as-sequencefile");
+ runImport(argv);
+ try {
+ SqoopOptions opts = new ImportTool().parseArguments(
+ getArgv(false, columns, codec, "--as-sequencefile"),
+ null, null, true);
+
+ CompilationManager compileMgr = new CompilationManager(opts);
+ String jarFileName = compileMgr.getJarFilename();
+ LOG.debug("Got jar from import job: " + jarFileName);
+
+ prevClassLoader = ClassLoaderStack.addJarFile(jarFileName,
+ getTableName());
+
+ reader = SeqFileReader.getSeqFileReader(getDataFilePath().toString());
+
+ if (codec == null) {
+ codec = new GzipCodec();
+ }
+ assertTrue("Block compressed", reader.isBlockCompressed());
+ assertEquals(codec.getClass(), reader.getCompressionCodec().getClass());
+
+ // here we can actually instantiate (k, v) pairs.
+ Configuration conf = new Configuration();
+ Object key = ReflectionUtils.newInstance(reader.getKeyClass(), conf);
+ Object val = ReflectionUtils.newInstance(reader.getValueClass(), conf);
+
+ // We know that these values are two ints separated by a ',' character.
+ // Since this is all dynamic, though, we don't want to actually link
+ // against the class and use its methods. So we just parse this back
+ // into int fields manually. Sum them up and ensure that we get the
+ // expected total for the first column, to verify that we got all the
+ // results from the db into the file.
+
+ // Sum up everything in the file.
+ int numLines = 0;
+ while (reader.next(key) != null) {
+ reader.getCurrentValue(val);
+ numLines++;
+ }
+
+ assertEquals(expectedNum, numLines);
+ } finally {
+ IOUtils.closeStream(reader);
+
+ if (null != prevClassLoader) {
+ ClassLoaderStack.setCurrentClassLoader(prevClassLoader);
+ }
+ }
+ }
+
+ public void runTextCompressionTest(CompressionCodec codec, int expectedNum)
+ throws IOException {
+
+ String [] columns = HsqldbTestServer.getFieldNames();
+ String [] argv = getArgv(true, columns, codec, "--as-textfile");
+ runImport(argv);
+
+ Configuration conf = new Configuration();
+ if (!BaseSqoopTestCase.isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+
+ if (codec == null) {
+ codec = new GzipCodec();
+ }
+ ReflectionUtils.setConf(codec, getConf());
+ Path p = new Path(getDataFilePath().toString()
+ + codec.getDefaultExtension());
+ InputStream is = codec.createInputStream(fs.open(p));
+ BufferedReader r = new BufferedReader(new InputStreamReader(is));
+ int numLines = 0;
+ while (true) {
+ String ln = r.readLine();
+ if (ln == null) {
+ break;
+ }
+ numLines++;
+ }
+ r.close();
+ assertEquals(expectedNum, numLines);
+ }
+
+ @Test
+ public void testDefaultTextCompression() throws IOException {
+ runTextCompressionTest(null, 4);
+ }
+
+ @Test
+ public void testBzip2TextCompression() throws IOException {
+ runTextCompressionTest(new BZip2Codec(), 4);
+ }
+
+ @Test
+ public void testBzip2SequenceFileCompression() throws Exception {
+ runSequenceFileCompressionTest(new BZip2Codec(), 4);
+ }
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/TestConnFactory.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestConnFactory.java b/src/test/org/apache/sqoop/TestConnFactory.java
new file mode 100644
index 0000000..fb6c940
--- /dev/null
+++ b/src/test/org/apache/sqoop/TestConnFactory.java
@@ -0,0 +1,200 @@
+/**
+ * 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;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.sqoop.ConnFactory;
+import org.apache.sqoop.SqoopOptions;
+import org.apache.sqoop.manager.ConnManager;
+import org.apache.sqoop.manager.ImportJobContext;
+import org.apache.sqoop.manager.ManagerFactory;
+import org.apache.sqoop.metastore.JobData;
+import org.apache.sqoop.tool.ImportTool;
+import org.junit.Test;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import org.junit.Rule;
+import org.junit.rules.ExpectedException;
+
+/**
+ * Test the ConnFactory implementation and its ability to delegate to multiple
+ * different ManagerFactory implementations using reflection.
+ */
+
+public class TestConnFactory {
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+ @Test
+ public void testCustomFactory() throws IOException {
+ Configuration conf = new Configuration();
+ conf.set(ConnFactory.FACTORY_CLASS_NAMES_KEY,
+ AlwaysDummyFactory.class.getName());
+
+ ConnFactory factory = new ConnFactory(conf);
+ ConnManager manager = factory.getManager(
+ new JobData(new SqoopOptions(), new ImportTool()));
+ assertNotNull("No manager returned", manager);
+ assertTrue("Expected a DummyManager", manager instanceof DummyManager);
+ }
+
+ @Test
+
+ public void testExceptionForNoManager() throws IOException {
+ Configuration conf = new Configuration();
+ conf.set(ConnFactory.FACTORY_CLASS_NAMES_KEY, EmptyFactory.class.getName());
+
+ ConnFactory factory = new ConnFactory(conf);
+
+ thrown.expect(IOException.class);
+ thrown.reportMissingExceptionWithMessage("Expected IOException because of missing ConnManager ");
+ factory.getManager(
+ new JobData(new SqoopOptions(), new ImportTool()));
+ }
+
+ @Test
+ public void testMultipleManagers() throws IOException {
+ Configuration conf = new Configuration();
+ // The AlwaysDummyFactory is second in this list. Nevertheless, since
+ // we know the first factory in the list will return null, we should still
+ // get a DummyManager out.
+ String classNames = EmptyFactory.class.getName()
+ + "," + AlwaysDummyFactory.class.getName();
+ conf.set(ConnFactory.FACTORY_CLASS_NAMES_KEY, classNames);
+
+ ConnFactory factory = new ConnFactory(conf);
+ ConnManager manager = factory.getManager(
+ new JobData(new SqoopOptions(), new ImportTool()));
+ assertNotNull("No manager returned", manager);
+ assertTrue("Expected a DummyManager", manager instanceof DummyManager);
+ }
+
+ ////// mock classes used for test cases above //////
+
+ /**
+ * Factory that always returns a DummyManager, regardless of the
+ * configuration.
+ */
+ public static class AlwaysDummyFactory extends ManagerFactory {
+ public ConnManager accept(JobData data) {
+ // Always return a new DummyManager
+ return new DummyManager();
+ }
+ }
+
+ /**
+ * ManagerFactory that accepts no configurations.
+ */
+ public static class EmptyFactory extends ManagerFactory {
+ public ConnManager accept(JobData data) {
+ // Never instantiate a proper ConnManager;
+ return null;
+ }
+ }
+
+ /**
+ * This implementation doesn't do anything special.
+ */
+ public static class DummyManager extends ConnManager {
+ public void close() {
+ }
+
+ public String [] listDatabases() {
+ return null;
+ }
+
+ public String [] listTables() {
+ return null;
+ }
+
+ public String [] getColumnNames(String tableName) {
+ return null;
+ }
+
+ public String[] getColumnNamesForProcedure(String procedureName) {
+ return null;
+ }
+
+ public String getPrimaryKey(String tableName) {
+ return null;
+ }
+
+ public Map<String,List<Integer>> getColumnInfo(String tableName) {
+ return null;
+ }
+
+ /**
+ * Default implementation.
+ * @param sqlType sql data type
+ * @return java data type
+ */
+ public String toJavaType(int sqlType) {
+ return null;
+ }
+
+ /**
+ * Default implementation.
+ * @param sqlType sql data type
+ * @return hive data type
+ */
+ public String toHiveType(int sqlType) {
+ return null;
+ }
+
+ public Map<String, Integer> getColumnTypes(String tableName) {
+ return null;
+ }
+
+ @Override
+ public Map<String, Integer> getColumnTypesForProcedure(
+ String procedureName) {
+ return null;
+ }
+
+ public ResultSet readTable(String tableName, String [] columns) {
+ return null;
+ }
+
+ public Connection getConnection() {
+ return null;
+ }
+
+ public String getDriverClass() {
+ return null;
+ }
+
+ public void execAndPrint(String s) {
+ }
+
+ public void importTable(ImportJobContext context) {
+ }
+
+ public void release() {
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/TestDirectImport.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestDirectImport.java b/src/test/org/apache/sqoop/TestDirectImport.java
new file mode 100644
index 0000000..edad143
--- /dev/null
+++ b/src/test/org/apache/sqoop/TestDirectImport.java
@@ -0,0 +1,81 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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;
+
+import org.apache.sqoop.testutil.CommonArgs;
+import org.apache.sqoop.testutil.HsqldbTestServer;
+import org.apache.sqoop.testutil.ImportJobTestCase;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+public class TestDirectImport extends ImportJobTestCase {
+
+ @Rule
+ public ExpectedException exception = ExpectedException.none();
+
+ protected String[] getArgv(boolean includeHadoopFlags, String[] colNames, boolean isDirect) {
+ String columnsString = "";
+ for (String col : colNames) {
+ columnsString += col + ",";
+ }
+
+ ArrayList<String> args = new ArrayList<String>();
+
+ if (includeHadoopFlags) {
+ CommonArgs.addHadoopFlags(args);
+ }
+
+ args.add("--table");
+ args.add(HsqldbTestServer.getTableName());
+ args.add("--columns");
+ args.add(columnsString);
+ if (isDirect) args.add("--direct");
+ args.add("--split-by");
+ args.add("INTFIELD1");
+ args.add("--connect");
+ args.add(HsqldbTestServer.getUrl());
+
+ args.add("--delete-target-dir");
+
+ return args.toArray(new String[0]);
+ }
+
+ @Test
+ public void testDirectFlagWithHSQL() throws IOException {
+ String[] columns = HsqldbTestServer.getFieldNames();
+
+ String[] argv = getArgv(true, columns, true);
+ exception.expect(IOException.class);
+ runImport(argv);
+ }
+
+ @Test
+ public void testNonDirectFlagWithHSQL() throws IOException {
+ String[] columns = HsqldbTestServer.getFieldNames();
+
+ String[] argv = getArgv(true, columns, false);
+ runImport(argv);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/TestExport.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestExport.java b/src/test/org/apache/sqoop/TestExport.java
new file mode 100644
index 0000000..f32f620
--- /dev/null
+++ b/src/test/org/apache/sqoop/TestExport.java
@@ -0,0 +1,1087 @@
+/**
+ * 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;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+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.sqoop.SqoopOptions;
+import org.apache.sqoop.testutil.BaseSqoopTestCase;
+import org.apache.sqoop.testutil.CommonArgs;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.compress.CodecPool;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.compress.Decompressor;
+import org.apache.hadoop.util.ReflectionUtils;
+
+import org.apache.sqoop.lib.RecordParser;
+import org.apache.sqoop.lib.SqoopRecord;
+import org.apache.sqoop.testutil.ExportJobTestCase;
+import org.apache.sqoop.tool.CodeGenTool;
+import org.apache.sqoop.util.ClassLoaderStack;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test that we can export data from HDFS into databases.
+ */
+public class TestExport extends ExportJobTestCase {
+
+ /**
+ * @return an argv for the CodeGenTool to use when creating tables to export.
+ */
+ protected String [] getCodeGenArgv(String... extraArgs) {
+ List<String> codeGenArgv = new ArrayList<String>();
+
+ if (null != extraArgs) {
+ for (String arg : extraArgs) {
+ codeGenArgv.add(arg);
+ }
+ }
+
+ codeGenArgv.add("--table");
+ codeGenArgv.add(getTableName());
+ codeGenArgv.add("--connect");
+ codeGenArgv.add(getConnectString());
+ codeGenArgv.add("--fields-terminated-by");
+ codeGenArgv.add("\\t");
+ codeGenArgv.add("--lines-terminated-by");
+ codeGenArgv.add("\\n");
+
+ return codeGenArgv.toArray(new String[0]);
+ }
+
+
+ protected String getRecordLine(int recordNum, ColumnGenerator... extraCols) {
+ String idStr = Integer.toString(recordNum);
+ StringBuilder sb = new StringBuilder();
+
+ sb.append(idStr);
+ sb.append("\t");
+ sb.append(getMsgPrefix());
+ sb.append(idStr);
+ for (ColumnGenerator gen : extraCols) {
+ sb.append("\t");
+ sb.append(gen.getExportText(recordNum));
+ }
+ sb.append("\n");
+
+ return sb.toString();
+ }
+
+ /** When generating data for export tests, each column is generated
+ according to a ColumnGenerator. Methods exist for determining
+ what to put into text strings in the files to export, as well
+ as what the string representation of the column as returned by
+ the database should look like.
+ */
+ public interface ColumnGenerator {
+ /** For a row with id rowNum, what should we write into that
+ line of the text file to export?
+ */
+ String getExportText(int rowNum);
+
+ /** For a row with id rowNum, what should the database return
+ for the given column's value?
+ */
+ String getVerifyText(int rowNum);
+
+ /** Return the column type to put in the CREATE TABLE statement. */
+ String getType();
+ }
+
+ protected static class IntColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ int day = rowNum + 1;
+ return String.valueOf(day);
+ }
+ public String getVerifyText(int rowNum) {
+ int day = rowNum + 1;
+ return String.valueOf(day);
+ }
+ public String getType() {
+ return "INT";
+ }
+ }
+
+ /**
+ * Create a data file that gets exported to the db.
+ * @param fileNum the number of the file (for multi-file export)
+ * @param numRecords how many records to write to the file.
+ * @param gzip is true if the file should be gzipped.
+ */
+ protected void createTextFile(int fileNum, int numRecords, boolean gzip,
+ ColumnGenerator... extraCols) throws IOException {
+ int startId = fileNum * numRecords;
+
+ String ext = ".txt";
+ if (gzip) {
+ ext = ext + ".gz";
+ }
+ Path tablePath = getTablePath();
+ Path filePath = new Path(tablePath, "part" + fileNum + ext);
+
+ Configuration conf = new Configuration();
+ if (!BaseSqoopTestCase.isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+ fs.mkdirs(tablePath);
+ OutputStream os = fs.create(filePath);
+ if (gzip) {
+ CompressionCodecFactory ccf = new CompressionCodecFactory(conf);
+ CompressionCodec codec = ccf.getCodec(filePath);
+ os = codec.createOutputStream(os);
+ }
+ BufferedWriter w = new BufferedWriter(new OutputStreamWriter(os));
+ for (int i = 0; i < numRecords; i++) {
+ w.write(getRecordLine(startId + i, extraCols));
+ }
+ w.close();
+ os.close();
+
+ if (gzip) {
+ verifyCompressedFile(filePath, numRecords);
+ }
+ }
+
+ private void verifyCompressedFile(Path f, int expectedNumLines)
+ throws IOException {
+ Configuration conf = new Configuration();
+ if (!BaseSqoopTestCase.isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+ InputStream is = fs.open(f);
+ CompressionCodecFactory ccf = new CompressionCodecFactory(conf);
+ CompressionCodec codec = ccf.getCodec(f);
+ LOG.info("gzip check codec is " + codec);
+ Decompressor decompressor = CodecPool.getDecompressor(codec);
+ if (null == decompressor) {
+ LOG.info("Verifying gzip sanity with null decompressor");
+ } else {
+ LOG.info("Verifying gzip sanity with decompressor: "
+ + decompressor.toString());
+ }
+ is = codec.createInputStream(is, decompressor);
+ BufferedReader r = new BufferedReader(new InputStreamReader(is));
+ int numLines = 0;
+ while (true) {
+ String ln = r.readLine();
+ if (ln == null) {
+ break;
+ }
+ numLines++;
+ }
+
+ r.close();
+ assertEquals("Did not read back correct number of lines",
+ expectedNumLines, numLines);
+ LOG.info("gzip sanity check returned " + numLines + " lines; ok.");
+ }
+
+ /**
+ * Create a data file in SequenceFile format that gets exported to the db.
+ * @param fileNum the number of the file (for multi-file export).
+ * @param numRecords how many records to write to the file.
+ * @param className the table class name to instantiate and populate
+ * for each record.
+ */
+ private void createSequenceFile(int fileNum, int numRecords, String className)
+ throws IOException {
+
+ try {
+ // Instantiate the value record object via reflection.
+ Class cls = Class.forName(className, true,
+ Thread.currentThread().getContextClassLoader());
+ SqoopRecord record = (SqoopRecord) ReflectionUtils.newInstance(
+ cls, new Configuration());
+
+ // Create the SequenceFile.
+ Configuration conf = new Configuration();
+ if (!BaseSqoopTestCase.isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+ Path tablePath = getTablePath();
+ Path filePath = new Path(tablePath, "part" + fileNum);
+ fs.mkdirs(tablePath);
+ SequenceFile.Writer w = SequenceFile.createWriter(
+ fs, conf, filePath, LongWritable.class, cls);
+
+ // Now write the data.
+ int startId = fileNum * numRecords;
+ for (int i = 0; i < numRecords; i++) {
+ record.parse(getRecordLine(startId + i));
+ w.append(new LongWritable(startId + i), record);
+ }
+
+ w.close();
+ } catch (ClassNotFoundException cnfe) {
+ throw new IOException(cnfe);
+ } catch (RecordParser.ParseError pe) {
+ throw new IOException(pe);
+ }
+ }
+
+ /** Return the column name for a column index.
+ * Each table contains two columns named 'id' and 'msg', and then an
+ * arbitrary number of additional columns defined by ColumnGenerators.
+ * These columns are referenced by idx 0, 1, 2...
+ * @param idx the index of the ColumnGenerator in the array passed to
+ * createTable().
+ * @return the name of the column
+ */
+ protected String forIdx(int idx) {
+ return "COL" + idx;
+ }
+
+ /**
+ * Return a SQL statement that drops a table, if it exists.
+ * @param tableName the table to drop.
+ * @return the SQL statement to drop that table.
+ */
+ protected String getDropTableStatement(String tableName) {
+ return "DROP TABLE " + tableName + " IF EXISTS";
+ }
+
+ /** Create the table definition to export to, removing any prior table.
+ By specifying ColumnGenerator arguments, you can add extra columns
+ to the table of arbitrary type.
+ */
+ public void createTable(ColumnGenerator... extraColumns) throws SQLException {
+ Connection conn = getConnection();
+ PreparedStatement statement = conn.prepareStatement(
+ getDropTableStatement(getTableName()),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ try {
+ statement.executeUpdate();
+ conn.commit();
+ } finally {
+ statement.close();
+ }
+
+ StringBuilder sb = new StringBuilder();
+ sb.append("CREATE TABLE ");
+ sb.append(getTableName());
+ sb.append(" (\"ID\" INT NOT NULL PRIMARY KEY, \"MSG\" VARCHAR(64)");
+ int colNum = 0;
+ for (ColumnGenerator gen : extraColumns) {
+ sb.append(", \"" + forIdx(colNum++) + "\" " + gen.getType());
+ }
+ sb.append(")");
+
+ statement = conn.prepareStatement(sb.toString(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ try {
+ statement.executeUpdate();
+ conn.commit();
+ } finally {
+ statement.close();
+ }
+ }
+
+ /**
+ * @return the name of the staging table to be used for testing.
+ */
+ public String getStagingTableName() {
+ return getTableName() + "_STAGE";
+ }
+
+ /**
+ * Creates the staging table.
+ * @param extraColumns extra columns that go in the staging table
+ * @throws SQLException if an error occurs during export
+ */
+ public void createStagingTable(ColumnGenerator... extraColumns)
+ throws SQLException {
+ String stageTableName = getStagingTableName();
+ Connection conn = getConnection();
+ PreparedStatement statement = conn.prepareStatement(
+ getDropTableStatement(stageTableName),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ try {
+ statement.executeUpdate();
+ conn.commit();
+ } finally {
+ statement.close();
+ }
+
+ StringBuilder sb = new StringBuilder();
+ sb.append("CREATE TABLE ");
+ sb.append(stageTableName);
+ sb.append(" (\"ID\" INT NOT NULL PRIMARY KEY, \"MSG\" VARCHAR(64)");
+ int colNum = 0;
+ for (ColumnGenerator gen : extraColumns) {
+ sb.append(", \"" + forIdx(colNum++) + "\" " + gen.getType());
+ }
+ sb.append(")");
+
+ statement = conn.prepareStatement(sb.toString(),
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ try {
+ statement.executeUpdate();
+ conn.commit();
+ } finally {
+ statement.close();
+ }
+
+ }
+
+ /** Removing an existing table directory from the filesystem. */
+ private void removeTablePath() throws IOException {
+ Configuration conf = new Configuration();
+ if (!BaseSqoopTestCase.isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+ fs.delete(getTablePath(), true);
+ }
+
+ /** Verify that on a given row, a column has a given value.
+ * @param id the id column specifying the row to test.
+ */
+ private void assertColValForRowId(int id, String colName, String expectedVal)
+ throws SQLException {
+ Connection conn = getConnection();
+ LOG.info("Verifying column " + colName + " has value " + expectedVal);
+
+ PreparedStatement statement = conn.prepareStatement(
+ "SELECT \"" + colName + "\" FROM " + getTableName() + " WHERE \"ID\" = " + id,
+ ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+ String actualVal = null;
+ try {
+ ResultSet rs = statement.executeQuery();
+ try {
+ rs.next();
+ actualVal = rs.getString(1);
+ } finally {
+ rs.close();
+ }
+ } finally {
+ statement.close();
+ }
+
+ assertEquals("Got unexpected column value", expectedVal, actualVal);
+ }
+
+ /** Verify that for the max and min values of the 'id' column, the values
+ for a given column meet the expected values.
+ */
+ protected void assertColMinAndMax(String colName, ColumnGenerator generator)
+ throws SQLException {
+ Connection conn = getConnection();
+ int minId = getMinRowId(conn);
+ int maxId = getMaxRowId(conn);
+
+ LOG.info("Checking min/max for column " + colName + " with type "
+ + generator.getType());
+
+ String expectedMin = generator.getVerifyText(minId);
+ String expectedMax = generator.getVerifyText(maxId);
+
+ assertColValForRowId(minId, colName, expectedMin);
+ assertColValForRowId(maxId, colName, expectedMax);
+ }
+
+ /**
+ * 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];
+ }
+
+ int newSize = entries.length + moreEntries.length;
+ String [] out = new String [newSize];
+
+ int i = 0;
+ for (String e : entries) {
+ out[i++] = e;
+ }
+
+ for (String e : moreEntries) {
+ out[i++] = e;
+ }
+
+ return out;
+ }
+
+ protected void multiFileTest(int numFiles, int recordsPerMap, int numMaps,
+ String... argv) throws IOException, SQLException {
+
+ final int TOTAL_RECORDS = numFiles * recordsPerMap;
+
+ try {
+ LOG.info("Beginning test: numFiles=" + numFiles + "; recordsPerMap="
+ + recordsPerMap + "; numMaps=" + numMaps);
+
+ for (int i = 0; i < numFiles; i++) {
+ createTextFile(i, recordsPerMap, false);
+ }
+
+ createTable();
+ runExport(getArgv(true, 10, 10, newStrArray(argv, "-m", "" + numMaps)));
+ verifyExport(TOTAL_RECORDS);
+ } finally {
+ LOG.info("multi-file test complete");
+ }
+ }
+
+ /**
+ * Run an "export" on an empty file.
+ */
+ @Test
+ public void testEmptyExport() throws IOException, SQLException {
+ multiFileTest(1, 0, 1);
+ }
+
+ /** Export 10 rows, make sure they load in correctly. */
+ @Test
+ public void testTextExport() throws IOException, SQLException {
+ multiFileTest(1, 10, 1);
+ }
+
+ /** Make sure we can use CombineFileInputFormat to handle multiple
+ * files in a single mapper.
+ */
+ @Test
+ public void testMultiFilesOneMapper() throws IOException, SQLException {
+ multiFileTest(2, 10, 1);
+ }
+
+ /** Make sure we can use CombineFileInputFormat to handle multiple
+ * files and multiple maps.
+ */
+ @Test
+ public void testMultiFilesMultiMaps() throws IOException, SQLException {
+ multiFileTest(2, 10, 2);
+ }
+
+ /** Export 10 rows from gzipped text files. */
+ @Test
+ public void testGzipExport() throws IOException, SQLException {
+
+ LOG.info("Beginning gzip export test");
+
+ final int TOTAL_RECORDS = 10;
+
+ createTextFile(0, TOTAL_RECORDS, true);
+ createTable();
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ LOG.info("Complete gzip export test");
+ }
+
+ /**
+ * Ensure that we use multiple statements in a transaction.
+ */
+ @Test
+ public void testMultiStatement() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 20;
+ createTextFile(0, TOTAL_RECORDS, true);
+ createTable();
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ }
+
+ /**
+ * Ensure that we use multiple transactions in a single mapper.
+ */
+ @Test
+ public void testMultiTransaction() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 20;
+ createTextFile(0, TOTAL_RECORDS, true);
+ createTable();
+ runExport(getArgv(true, 5, 2));
+ verifyExport(TOTAL_RECORDS);
+ }
+
+ /**
+ * Exercises the testMultiTransaction test with staging table specified.
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testMultiTransactionWithStaging()
+ throws IOException, SQLException {
+ final int TOTAL_RECORDS = 20;
+ createTextFile(0, TOTAL_RECORDS, true);
+ createTable();
+ createStagingTable();
+ runExport(getArgv(true, 5, 2, "--staging-table", getStagingTableName()));
+ verifyExport(TOTAL_RECORDS);
+ }
+
+ /**
+ * Ensure that when we don't force a commit with a statement cap,
+ * it happens anyway.
+ */
+ @Test
+ public void testUnlimitedTransactionSize() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 20;
+ createTextFile(0, TOTAL_RECORDS, true);
+ createTable();
+ runExport(getArgv(true, 5, -1));
+ verifyExport(TOTAL_RECORDS);
+ }
+
+ /** Run 2 mappers, make sure all records load in correctly. */
+ @Test
+ public void testMultiMapTextExport() throws IOException, SQLException {
+
+ final int RECORDS_PER_MAP = 10;
+ final int NUM_FILES = 2;
+
+ for (int f = 0; f < NUM_FILES; f++) {
+ createTextFile(f, RECORDS_PER_MAP, false);
+ }
+
+ createTable();
+ runExport(getArgv(true, 10, 10));
+ verifyExport(RECORDS_PER_MAP * NUM_FILES);
+ }
+
+ /**
+ * Run 2 mappers with staging enabled,
+ * make sure all records load in correctly.
+ */
+ @Test
+ public void testMultiMapTextExportWithStaging()
+ throws IOException, SQLException {
+
+ final int RECORDS_PER_MAP = 10;
+ final int NUM_FILES = 2;
+
+ for (int f = 0; f < NUM_FILES; f++) {
+ createTextFile(f, RECORDS_PER_MAP, false);
+ }
+
+ createTable();
+ createStagingTable();
+ runExport(getArgv(true, 10, 10, "--staging-table", getStagingTableName()));
+ verifyExport(RECORDS_PER_MAP * NUM_FILES);
+ }
+
+ /** Export some rows from a SequenceFile, make sure they import correctly. */
+ @Test
+ public void testSequenceFileExport() throws Exception {
+
+ final int TOTAL_RECORDS = 10;
+
+ // First, generate class and jar files that represent the table
+ // we're exporting to.
+ LOG.info("Creating initial schema for SeqFile test");
+ createTable();
+ LOG.info("Generating code...");
+ CodeGenTool codeGen = new CodeGenTool();
+ String [] codeGenArgs = getCodeGenArgv();
+ SqoopOptions options = codeGen.parseArguments(
+ codeGenArgs, null, null, true);
+ codeGen.validateOptions(options);
+ int ret = codeGen.run(options);
+ assertEquals(0, ret);
+ List<String> generatedJars = codeGen.getGeneratedJarFiles();
+
+ // Now, wipe the created table so we can export on top of it again.
+ LOG.info("Resetting schema and data...");
+ createTable();
+
+ // Wipe the directory we use when creating files to export to ensure
+ // it's ready for new SequenceFiles.
+ removeTablePath();
+
+ assertNotNull(generatedJars);
+ assertEquals("Expected 1 generated jar file", 1, generatedJars.size());
+ String jarFileName = generatedJars.get(0);
+ // Sqoop generates jars named "foo.jar"; by default, this should contain a
+ // class named 'foo'. Extract the class name.
+ Path jarPath = new Path(jarFileName);
+ String jarBaseName = jarPath.getName();
+ assertTrue(jarBaseName.endsWith(".jar"));
+ assertTrue(jarBaseName.length() > ".jar".length());
+ String className = jarBaseName.substring(0, jarBaseName.length()
+ - ".jar".length());
+
+ LOG.info("Using jar filename: " + jarFileName);
+ LOG.info("Using class name: " + className);
+
+ ClassLoader prevClassLoader = null;
+
+ try {
+ if (null != jarFileName) {
+ prevClassLoader = ClassLoaderStack.addJarFile(jarFileName, className);
+ }
+
+ // Now use this class and jar name to create a sequence file.
+ LOG.info("Writing data to SequenceFiles");
+ createSequenceFile(0, TOTAL_RECORDS, className);
+
+ // Now run and verify the export.
+ LOG.info("Exporting SequenceFile-based data");
+ runExport(getArgv(true, 10, 10, "--class-name", className,
+ "--jar-file", jarFileName));
+ verifyExport(TOTAL_RECORDS);
+ } finally {
+ if (null != prevClassLoader) {
+ ClassLoaderStack.setCurrentClassLoader(prevClassLoader);
+ }
+ }
+ }
+
+ @Test
+ public void testIntCol() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 10;
+
+ // generate a column equivalent to rownum.
+ ColumnGenerator gen = new ColumnGenerator() {
+ public String getExportText(int rowNum) {
+ return "" + rowNum;
+ }
+ public String getVerifyText(int rowNum) {
+ return "" + rowNum;
+ }
+ public String getType() {
+ return "INTEGER";
+ }
+ };
+
+ createTextFile(0, TOTAL_RECORDS, false, gen);
+ createTable(gen);
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ assertColMinAndMax(forIdx(0), gen);
+ }
+
+ /** @return the column type for a large integer */
+ protected String getBigIntType() {
+ return "BIGINT";
+ }
+
+ @Test
+ public void testBigIntCol() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 10;
+
+ // generate a column that won't fit in a normal int.
+ ColumnGenerator gen = new ColumnGenerator() {
+ public String getExportText(int rowNum) {
+ long val = (long) rowNum * 1000000000;
+ return "" + val;
+ }
+ public String getVerifyText(int rowNum) {
+ long val = (long) rowNum * 1000000000;
+ return "" + val;
+ }
+ public String getType() {
+ return getBigIntType();
+ }
+ };
+
+ createTextFile(0, TOTAL_RECORDS, false, gen);
+ createTable(gen);
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ assertColMinAndMax(forIdx(0), gen);
+ }
+
+ protected String pad(int n) {
+ if (n <= 9) {
+ return "0" + n;
+ } else {
+ return String.valueOf(n);
+ }
+ }
+
+ /**
+ * Get a column generator for DATE columns.
+ */
+ protected ColumnGenerator getDateColumnGenerator() {
+ return new ColumnGenerator() {
+ public String getExportText(int rowNum) {
+ int day = rowNum + 1;
+ return "2009-10-" + pad(day);
+ }
+ public String getVerifyText(int rowNum) {
+ int day = rowNum + 1;
+ return "2009-10-" + pad(day);
+ }
+ public String getType() {
+ return "DATE";
+ }
+ };
+ }
+
+ /**
+ * Get a column generator for TIME columns.
+ */
+ protected ColumnGenerator getTimeColumnGenerator() {
+ return new ColumnGenerator() {
+ public String getExportText(int rowNum) {
+ return "10:01:" + pad(rowNum);
+ }
+ public String getVerifyText(int rowNum) {
+ return "10:01:" + pad(rowNum);
+ }
+ public String getType() {
+ return "TIME";
+ }
+ };
+ }
+
+ @Test
+ public void testDatesAndTimes() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 10;
+
+ ColumnGenerator genDate = getDateColumnGenerator();
+ ColumnGenerator genTime = getTimeColumnGenerator();
+
+ createTextFile(0, TOTAL_RECORDS, false, genDate, genTime);
+ createTable(genDate, genTime);
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ assertColMinAndMax(forIdx(0), genDate);
+ assertColMinAndMax(forIdx(1), genTime);
+ }
+
+ @Test
+ public void testNumericTypes() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 9;
+
+ // Check floating point values
+ ColumnGenerator genFloat = new ColumnGenerator() {
+ public String getExportText(int rowNum) {
+ double v = 3.141 * (double) (rowNum + 1);
+ return "" + v;
+ }
+ public String getVerifyText(int rowNum) {
+ double v = 3.141 * (double) (rowNum + 1);
+ return "" + v;
+ }
+ public String getType() {
+ return "FLOAT";
+ }
+ };
+
+ // Check precise decimal placement. The first of ten
+ // rows will be 2.7181; the last of ten rows will be
+ // 2.71810.
+ ColumnGenerator genNumeric = new ColumnGenerator() {
+ public String getExportText(int rowNum) {
+ int digit = rowNum + 1;
+ return "2.718" + digit;
+ }
+ public String getVerifyText(int rowNum) {
+ int digit = rowNum + 1;
+ return "2.718" + digit;
+ }
+ public String getType() {
+ return "NUMERIC(6,4)";
+ }
+ };
+
+ createTextFile(0, TOTAL_RECORDS, false, genFloat, genNumeric);
+ createTable(genFloat, genNumeric);
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ assertColMinAndMax(forIdx(0), genFloat);
+ assertColMinAndMax(forIdx(1), genNumeric);
+ }
+
+ @Test
+ public void testColumnsExport() throws IOException, SQLException {
+ testColumnsExport("ID,MSG," + forIdx(0) + "," + forIdx(2));
+ }
+
+ /**
+ * It's possible to change the column string that
+ * {@link #testColumnsExport()} uses - you might want to do
+ * this if your database randomly generates column names, instead
+ * of using the given ones (e.g. stored procedure parameter
+ * names in H2)
+ */
+ protected void testColumnsExport(
+ String columnsStr) throws IOException, SQLException {
+ final int TOTAL_COLUMNS = 3;
+ final int TOTAL_RECORDS = 10;
+
+ // This class is used to generate a column whose entries have unique
+ // sequential values that are computed as follows:
+ // (row number * total number of columns) + column number
+ class MultiColumnGenerator implements ColumnGenerator {
+ private int col;
+ MultiColumnGenerator(int col) {
+ this.col = col;
+ }
+ public String getExportText(int rowNum) {
+ return new Integer(rowNum * TOTAL_COLUMNS + col).toString();
+ }
+ public String getVerifyText(int rowNum) {
+ return new Integer(rowNum * TOTAL_COLUMNS + col).toString();
+ }
+ public String getType() {
+ return "INTEGER";
+ }
+ }
+
+ // This class is used to generate a null column.
+ class NullColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ return null;
+ }
+ public String getVerifyText(int rowNum) {
+ return null;
+ }
+ public String getType() {
+ return "INTEGER";
+ }
+ }
+
+ // Generate an input file that only contains partial columns: col0
+ // and col2. Export these columns to the DB w/ the --columns option.
+ // Finally, verify that these two columns are exported properly. In
+ // addition, verify that the values of col1 in the DB are all null
+ // as they are not exported.
+ ColumnGenerator gen0 = new MultiColumnGenerator(0);
+ ColumnGenerator gen1 = new MultiColumnGenerator(1);
+ ColumnGenerator gen2 = new MultiColumnGenerator(2);
+
+ createTextFile(0, TOTAL_RECORDS, false, gen0, gen2);
+ createTable(gen0, gen1, gen2);
+
+ runExport(getArgv(true, 10, 10, "--columns", columnsStr));
+
+ ColumnGenerator genNull = new NullColumnGenerator();
+
+ verifyExport(TOTAL_RECORDS);
+ assertColMinAndMax(forIdx(0), gen0);
+ assertColMinAndMax(forIdx(2), gen2);
+ assertColMinAndMax(forIdx(1), genNull);
+ }
+
+ protected void testExportToTableWithName(String tableName) throws IOException, SQLException {
+ final int TOTAL_RECORDS = 10;
+ setCurTableName(tableName);
+ createTextFile(0, TOTAL_RECORDS, false);
+ createTable();
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ }
+
+ /**
+ * When we have less columns in the export file than in the export table, without changes in SQOOP-3158
+ * it will fail with errors. After SQOOP-3158, it should succeed without errors
+ *
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testLessColumnsInFileThanInTable() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 10;
+
+ ColumnGenerator genInteger = new IntColumnGenerator();
+ ColumnGenerator genTime = getTimeColumnGenerator();
+
+ createTextFile(0, TOTAL_RECORDS, false, genInteger);
+ createTable(genInteger, genTime);
+ runExport(getArgv(true, 10, 10));
+ verifyExport(TOTAL_RECORDS);
+ assertColMinAndMax(forIdx(0), genInteger);
+
+ // test that the Time column is with NULL values
+ class NullColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ return null;
+ }
+ public String getVerifyText(int rowNum) {
+ return null;
+ }
+ public String getType() {
+ return "Timestamp";
+ }
+ }
+
+ ColumnGenerator genNull = new NullColumnGenerator();
+ assertColMinAndMax(forIdx(1), genNull);
+ }
+
+ /**
+ * This test case covers the following:
+ *
+ * - two files under export directory
+ * - first file has one missing column
+ * - second file has correct number of columns
+ * - second file has last column with the value defined by parameter --input-null-string
+ * - last column is of STRING data type
+ *
+ * This case will happen when destination table has been modified to add a new column and
+ * user has data for both before and after the table modification, some files has less columns
+ * and some files has new columns
+ *
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testLessColumnsInFileThanInTableInputNullStringPassed() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 10;
+
+ // a column that contains string value of "STRING_NULL",
+ // this should be treated as NULL value during test import
+ // due to --input-null-string is defined using "STRING_NULL"
+ class StringNULLValueColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ return "STRING_NULL";
+ }
+ public String getVerifyText(int rowNum) {
+ return "STRING_NULL";
+ }
+ public String getType() {
+ return "VARCHAR(255)";
+ }
+ }
+
+ // a normal string column
+ class StringColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ int day = rowNum + 1;
+ return "ROW_" + pad(day);
+ }
+ public String getVerifyText(int rowNum) {
+ int day = rowNum + 1;
+ return "ROW_" + pad(day);
+ }
+ public String getType() {
+ return "VARCHAR(255)";
+ }
+ }
+
+ // test that the second column is with NULL values after import
+ class NullColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ return null;
+ }
+ public String getVerifyText(int rowNum) {
+ return null;
+ }
+ public String getType() {
+ return "VARCHAR(255)";
+ }
+ }
+
+ ColumnGenerator genString = new StringColumnGenerator();
+ ColumnGenerator genNullString = new StringNULLValueColumnGenerator();
+
+ createTextFile(0, TOTAL_RECORDS, false, genString);
+ createTextFile(1, TOTAL_RECORDS, false, genString, genNullString);
+ createTable(genString, genNullString);
+ runExport(getArgv(true, 10, 10, "--input-null-string", "STRING_NULL"));
+
+ verifyExport(TOTAL_RECORDS * 2);
+ assertColMinAndMax(forIdx(0), genString);
+
+ ColumnGenerator genNull = new NullColumnGenerator();
+ assertColMinAndMax(forIdx(1), genNull);
+ }
+
+ /**
+ * This test case covers the following:
+ *
+ * - two files under export directory
+ * - first file has one missing column
+ * - second file has correct number of columns
+ * - second file has last column with the value defined by parameter --input-null-non-string
+ * - last column is of INTEGER data type
+ *
+ * This case will happen when destination table has been modified to add a new column and
+ * user has data for both before and after the table modification, some files has less columns
+ * and some files has new columns
+ *
+ * @throws IOException
+ * @throws SQLException
+ */
+ @Test
+ public void testLessColumnsInFileThanInTableInputNullIntPassed() throws IOException, SQLException {
+ final int TOTAL_RECORDS = 10;
+
+ // a column that contains string value of "INT_NULL",
+ // this should be treated as NULL value during test import
+ // due to --input-null-non-string is defined using "INT_NULL"
+ class IntNULLValueColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ return "INT_NULL";
+ }
+ public String getVerifyText(int rowNum) {
+ return "INT_NULL";
+ }
+ public String getType() {
+ return "INT";
+ }
+ }
+
+ // test that the second column is with NULL values after import
+ class NullColumnGenerator implements ColumnGenerator {
+ public String getExportText(int rowNum) {
+ return null;
+ }
+ public String getVerifyText(int rowNum) {
+ return null;
+ }
+ public String getType() {
+ return "INT";
+ }
+ }
+
+ ColumnGenerator genInt = new IntColumnGenerator();
+ ColumnGenerator genNullInt = new IntNULLValueColumnGenerator();
+
+ createTextFile(0, TOTAL_RECORDS, false, genInt);
+ createTextFile(1, TOTAL_RECORDS, false, genInt, genNullInt);
+ createTable(genInt, genNullInt);
+ runExport(getArgv(true, 10, 10, "--input-null-non-string", "INT_NULL"));
+ verifyExport(TOTAL_RECORDS * 2);
+ assertColMinAndMax(forIdx(0), genInt);
+ assertColMinAndMax(forIdx(1), new NullColumnGenerator());
+ }
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/TestExportUpdate.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestExportUpdate.java b/src/test/org/apache/sqoop/TestExportUpdate.java
new file mode 100644
index 0000000..d28edd7
--- /dev/null
+++ b/src/test/org/apache/sqoop/TestExportUpdate.java
@@ -0,0 +1,710 @@
+/**
+ * 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;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+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.sqoop.testutil.CommonArgs;
+import org.apache.sqoop.testutil.ExportJobTestCase;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+/**
+ * Test that we can update a copy of data in the database,
+ * based on newer data in HDFS.
+ */
+public class TestExportUpdate extends ExportJobTestCase {
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ @Override
+ protected String getTablePrefix() {
+ return "UPDATE_TABLE_";
+ }
+
+ /**
+ * Create the underlying table to update.
+ */
+ private void populateDatabase(int numRows) throws SQLException {
+ Connection conn = getConnection();
+
+ PreparedStatement statement = conn.prepareStatement(
+ "CREATE TABLE " + getTableName()
+ + " (A INT NOT NULL, B VARCHAR(32), C INT)");
+ try {
+ statement.executeUpdate();
+ conn.commit();
+ } finally {
+ statement.close();
+ statement = null;
+ }
+
+ try {
+ for (int i = 0; i < numRows; i++) {
+ statement = conn.prepareStatement("INSERT INTO " + getTableName()
+ + " VALUES (" + i + ", 'foo" + i + "', " + i + ")");
+ statement.executeUpdate();
+ statement.close();
+ statement = null;
+ }
+ } finally {
+ if (null != statement) {
+ statement.close();
+ }
+ }
+
+ conn.commit();
+ }
+
+ /**
+ * <p>Creates a table with three columns - A INT, B INT and C VARCHAR(32).
+ * This table is populated with records in a set of three with total records
+ * with the total number of unique values of A equal to the specified aMax
+ * value. For each value of A, there will be three records with value of
+ * B ranging from 0-2, and a corresponding value of C.</p>
+ * <p>For example if <tt>aMax = 2</tt>, the table will contain the
+ * following records:
+ * <pre>
+ * A | B | C
+ * ----------------------
+ * 0 | 0 | 0foo0
+ * 0 | 1 | 0foo1
+ * 0 | 2 | 0foo2
+ * 1 | 0 | 1foo0
+ * 1 | 1 | 1foo1
+ * 1 | 2 | 1foo2
+ * </pre></p>
+ * @param aMax the number of
+ * @throws SQLException
+ */
+ private void createMultiKeyTable(int aMax) throws SQLException {
+ Connection conn = getConnection();
+
+ PreparedStatement statement = conn.prepareStatement(
+ "CREATE TABLE " + getTableName()
+ + " (A INT NOT NULL, B INT NOT NULL, C VARCHAR(32))");
+ try {
+ statement.executeUpdate();
+ conn.commit();
+ } finally {
+ statement.close();
+ statement = null;
+ }
+
+ try {
+ for (int i = 0; i< aMax; i++) {
+ for (int j = 0; j < 3; j++) {
+ statement = conn.prepareStatement("INSERT INTO " + getTableName()
+ + " VALUES (" + i + ", " + j + ", '"
+ + i + "foo" + j + "')");
+ statement.executeUpdate();
+ statement.close();
+ statement = null;
+ }
+ }
+ } finally {
+ if (null != statement) {
+ statement.close();
+ }
+ }
+
+ conn.commit();
+ }
+
+ /**
+ * <p>Creates update files for multi-key update test. The total number of
+ * update records will be number of files times the number of aKeysPerFile
+ * times 3. Column A value will start with the specified <tt>startAtValue</tt>
+ * and for each value there will be three records corresponding to Column
+ * B values [0-2].</p>
+ * @param numFiles number of files to create
+ * @param aKeysPerFile number of records sets with different column A values
+ * @param startAtValue the starting value of column A
+ * @param bKeyValues the list of values for the column B
+ * @throws IOException
+ */
+ private void createMultiKeyUpdateFiles(int numFiles, int aKeysPerFile,
+ int startAtValue, int[] bKeyValues)
+ throws IOException {
+ Configuration conf = getConf();
+ if (!isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+
+ int aValue = startAtValue;
+ for (int i = 0; i < numFiles; i++) {
+ OutputStream os = fs.create(new Path(getTablePath(), "" + i + ".txt"));
+ BufferedWriter w = new BufferedWriter(new OutputStreamWriter(os));
+
+ for (int j = 0; j < aKeysPerFile; j++) {
+ for (int k = 0; k < bKeyValues.length; k++) {
+ w.write(getUpdateStringForMultiKeyRow(aValue, bKeyValues[k]));
+ }
+ aValue++;
+ }
+
+ w.close();
+ os.close();
+ }
+ }
+
+ /**
+ * Generate a string of text representing an update for one row
+ * of the multi-key table. The values of columns A and B are given
+ * and the value of column C is generated as <em>a</em>bar<em>b</em>.
+ * @param a the value of column a
+ * @param b the value of column b
+ */
+ private String getUpdateStringForMultiKeyRow(int a, int b) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(a).append("\t").append(b).append("\t").append(a);
+ sb.append("bar").append(b).append("\n");
+
+ return sb.toString();
+ }
+
+
+ /**
+ * Create a set of files that will be used as the input to the update
+ * process.
+ * @param numFiles the number of files to generate
+ * @param updatesPerFile the number of rows to create in each file
+ * @param keyCol a value between 0 and 2 specifying whether 'a',
+ * 'b', or 'c' ({@see populateDatabase()}) is the key column to keep
+ * the same.
+ * @param startOffsets is an optional list of row ids/values for a/c
+ * which are the record ids at which the update files begin.
+ * For instance, if numFiles=3, updatesPerFile=2, and keyCol=0 then
+ * if startOffsets is {5, 10, 12}, files will be generated to update
+ * rows with A=5,6; A=10,11; A=12,13.
+ *
+ * If startOffsets is empty or underspecified (given numFiles), then
+ * subsequent files will start immediately after the previous file.
+ */
+ private void createUpdateFiles(int numFiles, int updatesPerFile,
+ int keyCol, int... startOffsets) throws IOException {
+ Configuration conf = getConf();
+ if (!isOnPhysicalCluster()) {
+ conf.set(CommonArgs.FS_DEFAULT_NAME, CommonArgs.LOCAL_FS);
+ }
+ FileSystem fs = FileSystem.get(conf);
+
+ int rowId = 0;
+ for (int i = 0; i < numFiles; i++) {
+ OutputStream os = fs.create(new Path(getTablePath(), "" + i + ".txt"));
+ BufferedWriter w = new BufferedWriter(new OutputStreamWriter(os));
+
+ if (null != startOffsets && startOffsets.length > i) {
+ // If a start offset has been specified for this file, go there.
+ // Otherwise, just carry over from the previous file iteration.
+ rowId = startOffsets[i];
+ }
+
+ for (int j = 0; j < updatesPerFile; j++) {
+ w.write(getUpdateStringForRow(keyCol, rowId++));
+ }
+
+ w.close();
+ os.close();
+ }
+ }
+
+ /**
+ * Generate a string of text representing an update for one row
+ * of the database. keyCol is a value in [0, 2] representing which
+ * column is kept fixed. rowId specifies the row being updated.
+ */
+ private String getUpdateStringForRow(int keyCol, int rowId) {
+ StringBuilder sb = new StringBuilder();
+
+ int [] rowInts = new int[3]; // There are 3 columns in the table.
+ for (int i = 0; i < 3; i++) {
+ if (keyCol == i) {
+ // Keep this column fixed.
+ rowInts[i] = rowId;
+ } else {
+ // Update the int in this column.
+ rowInts[i] = rowId * 2;
+ }
+ }
+
+ sb.append(rowInts[0]);
+ sb.append("\tfoo");
+ sb.append(rowInts[1]);
+ sb.append("\t");
+ sb.append(rowInts[2]);
+ sb.append("\n");
+
+ return sb.toString();
+ }
+
+ /**
+ * Verifies the number of rows in the table.
+ */
+ private void verifyRowCount(int expectedCount) throws SQLException {
+ String query = "SELECT COUNT(*) FROM " + getTableName();
+ PreparedStatement statement = null;
+ ResultSet rs = null;
+
+ try {
+ Connection conn = getConnection();
+ statement = conn.prepareStatement(query);
+ rs = statement.executeQuery();
+
+ boolean success = rs.next();
+ assertTrue("Expected at least one result", success);
+
+ int trueCount = rs.getInt(1);
+ assertEquals(expectedCount, trueCount);
+
+ // This query should have returned exactly one row.
+ success = rs.next();
+ assertFalse("Expected no more than one output record", success);
+ } finally {
+ if (null != rs) {
+ try {
+ rs.close();
+ } catch (SQLException sqle) {
+ LOG.error("Error closing result set: "
+ + StringUtils.stringifyException(sqle));
+ }
+ }
+
+ if (null != statement) {
+ statement.close();
+ }
+ }
+ }
+
+ private void verifyMultiKeyRow(String[] keyColumnNames, int[] keyValues,
+ Object ...expectedVals) throws SQLException {
+ StringBuilder querySb = new StringBuilder("SELECT A, B, C FROM ");
+ querySb.append(getTableName()).append(" WHERE ");
+ boolean first = true;
+ for (int i = 0; i< keyColumnNames.length; i++) {
+ if (first) {
+ first = false;
+ } else {
+ querySb.append(" AND ");
+ }
+ querySb.append(keyColumnNames[i]).append(" = ");
+ querySb.append(keyValues[i]);
+ }
+
+ String query = querySb.toString();
+ PreparedStatement statement = null;
+ ResultSet rs = null;
+
+ try {
+ Connection conn = getConnection();
+ statement = conn.prepareStatement(query);
+ rs = statement.executeQuery();
+
+ boolean success = rs.next();
+ assertTrue("Expected at least one output record", success);
+
+ // Assert that all three columns have the correct values.
+ for (int i = 0; i < expectedVals.length; i++) {
+ String expected = expectedVals[i].toString();
+ String result = rs.getString(i + 1);
+ assertEquals("Invalid response for column " + i + "; got " + result
+ + " when expected " + expected, expected, result);
+ }
+
+ // This query should have returned exactly one row.
+ success = rs.next();
+ assertFalse("Expected no more than one output record", success);
+ } finally {
+ if (null != rs) {
+ try {
+ rs.close();
+ } catch (SQLException sqle) {
+ LOG.error("Error closing result set: "
+ + StringUtils.stringifyException(sqle));
+ }
+ }
+
+ if (null != statement) {
+ statement.close();
+ }
+ }
+ }
+
+ /**
+ * Verify that a particular row has the expected values.
+ */
+ private void verifyRow(String keyColName, String keyVal,
+ String... expectedVals) throws SQLException {
+ String query = "SELECT A, B, C FROM " + getTableName() + " WHERE "
+ + keyColName + " = " + keyVal;
+ PreparedStatement statement = null;
+ ResultSet rs = null;
+
+ try {
+ Connection conn = getConnection();
+ statement = conn.prepareStatement(query);
+ rs = statement.executeQuery();
+
+ boolean success = rs.next();
+ assertTrue("Expected at least one output record", success);
+
+ // Assert that all three columns have the correct values.
+ for (int i = 0; i < expectedVals.length; i++) {
+ String expected = expectedVals[i];
+ String result = rs.getString(i + 1);
+ assertEquals("Invalid response for column " + i + "; got " + result
+ + " when expected " + expected, expected, result);
+ }
+
+ // This query should have returned exactly one row.
+ success = rs.next();
+ assertFalse("Expected no more than one output record", success);
+ } finally {
+ if (null != rs) {
+ try {
+ rs.close();
+ } catch (SQLException sqle) {
+ LOG.error("Error closing result set: "
+ + StringUtils.stringifyException(sqle));
+ }
+ }
+
+ if (null != statement) {
+ statement.close();
+ }
+ }
+ }
+
+ private void runUpdate(int numMappers, String updateCol) throws IOException {
+ runExport(getArgv(true, 2, 2, "-m", "" + numMappers,
+ "--update-key", updateCol));
+ }
+
+ @Test
+ public void testBasicUpdate() throws Exception {
+ // Test that we can do a single-task single-file update.
+ // This updates the entire database.
+
+ populateDatabase(10);
+ createUpdateFiles(1, 10, 0, 0);
+ runUpdate(1, "A");
+ verifyRowCount(10);
+ // Check a few rows...
+ verifyRow("A", "0", "0", "foo0", "0");
+ verifyRow("A", "1", "1", "foo2", "2");
+ verifyRow("A", "9", "9", "foo18", "18");
+ }
+
+ /**
+ * Creates a table with two columns that together act as unique keys
+ * and then modifies a subset of the rows via update.
+ * @throws Exception
+ */
+ @Test
+ public void testMultiKeyUpdate() throws Exception {
+ createMultiKeyTable(3);
+
+ createMultiKeyUpdateFiles(1, 1, 1, new int[] {0, 1, 3});
+
+ runExport(getArgv(true, 2, 2, "-m", "1",
+ "--update-key", "A,B"));
+ verifyRowCount(9);
+ // Check a few rows...
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 0 }, 0, 0, "0foo0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 1 }, 0, 1, "0foo1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 2 }, 0, 2, "0foo2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 0 }, 1, 0, "1bar0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 1 }, 1, 1, "1bar1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 2 }, 1, 2, "1foo2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 0 }, 2, 0, "2foo0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 1 }, 2, 1, "2foo1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 2 }, 2, 2, "2foo2");
+
+ }
+
+ /**
+ * Creates a table with two columns that together act as unique keys
+ * and then modifies a subset of the rows via update.
+ * @throws Exception
+ */
+ @Test
+ public void testMultiKeyUpdateMultipleFilesNoUpdate() throws Exception {
+ createMultiKeyTable(4);
+
+ createMultiKeyUpdateFiles(2, 1, 1, new int[] {3, 4, 5});
+
+ runExport(getArgv(true, 2, 2, "-m", "1",
+ "--update-key", "A,B"));
+ verifyRowCount(12);
+ // Check a few rows...
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 0 }, 0, 0, "0foo0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 1 }, 0, 1, "0foo1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 2 }, 0, 2, "0foo2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 0 }, 1, 0, "1foo0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 1 }, 1, 1, "1foo1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 2 }, 1, 2, "1foo2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 0 }, 2, 0, "2foo0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 1 }, 2, 1, "2foo1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 2 }, 2, 2, "2foo2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 3, 0 }, 3, 0, "3foo0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 3, 1 }, 3, 1, "3foo1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 3, 2 }, 3, 2, "3foo2");
+ }
+
+ /**
+ * Creates a table with two columns that together act as unique keys
+ * and then modifies a subset of the rows via update.
+ * @throws Exception
+ */
+ @Test
+ public void testMultiKeyUpdateMultipleFilesFullUpdate() throws Exception {
+ createMultiKeyTable(4);
+
+ createMultiKeyUpdateFiles(2, 2, 0, new int[] {0, 1, 2});
+
+ runExport(getArgv(true, 2, 2, "-m", "1",
+ "--update-key", "A,B"));
+ verifyRowCount(12);
+ // Check a few rows...
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 0 }, 0, 0, "0bar0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 1 }, 0, 1, "0bar1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 0, 2 }, 0, 2, "0bar2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 0 }, 1, 0, "1bar0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 1 }, 1, 1, "1bar1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 1, 2 }, 1, 2, "1bar2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 0 }, 2, 0, "2bar0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 1 }, 2, 1, "2bar1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 2, 2 }, 2, 2, "2bar2");
+
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 3, 0 }, 3, 0, "3bar0");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 3, 1 }, 3, 1, "3bar1");
+ verifyMultiKeyRow(new String[] { "A", "B"},
+ new int[] { 3, 2 }, 3, 2, "3bar2");
+ }
+
+ @Test
+ public void testEmptyTable() throws Exception {
+ // Test that an empty table will "accept" updates that modify
+ // no rows; no new data is injected into the database.
+ populateDatabase(0);
+ createUpdateFiles(1, 10, 0, 0);
+ runUpdate(1, "A");
+ verifyRowCount(0);
+ }
+
+ @Test
+ public void testEmptyFiles() throws Exception {
+ // An empty input file results in no changes to a db table.
+ populateDatabase(10);
+ createUpdateFiles(1, 0, 0);
+ runUpdate(1, "A");
+ verifyRowCount(10);
+ // Check that a few rows have not changed at all.
+ verifyRow("A", "0", "0", "foo0", "0");
+ verifyRow("A", "1", "1", "foo1", "1");
+ verifyRow("A", "9", "9", "foo9", "9");
+ }
+
+ @Test
+ public void testStringCol() throws Exception {
+ // Test that we can do modifications based on the string "B" column.
+ populateDatabase(10);
+ createUpdateFiles(1, 10, 1);
+ runUpdate(1, "B");
+ verifyRowCount(10);
+ verifyRow("B", "'foo0'", "0", "foo0", "0");
+ verifyRow("B", "'foo1'", "2", "foo1", "2");
+ verifyRow("B", "'foo9'", "18", "foo9", "18");
+ }
+
+ @Test
+ public void testLastCol() throws Exception {
+ // Test that we can do modifications based on the third int column.
+ populateDatabase(10);
+ createUpdateFiles(1, 10, 2);
+ runUpdate(1, "C");
+ verifyRowCount(10);
+ verifyRow("C", "0", "0", "foo0", "0");
+ verifyRow("C", "1", "2", "foo2", "1");
+ verifyRow("C", "9", "18", "foo18", "9");
+ }
+
+ @Test
+ public void testMultiMaps() throws Exception {
+ // Test that we can handle multiple map tasks.
+ populateDatabase(20);
+ createUpdateFiles(2, 10, 0);
+ runUpdate(1, "A");
+ verifyRowCount(20);
+ verifyRow("A", "0", "0", "foo0", "0");
+ verifyRow("A", "1", "1", "foo2", "2");
+ verifyRow("A", "9", "9", "foo18", "18");
+ verifyRow("A", "10", "10", "foo20", "20");
+ verifyRow("A", "15", "15", "foo30", "30");
+ verifyRow("A", "19", "19", "foo38", "38");
+ }
+
+ @Test
+ public void testSubsetUpdate() throws Exception {
+ // Update only a few rows in the middle of the table.
+ populateDatabase(10);
+ createUpdateFiles(1, 5, 0, 3); // only rows A=3..7 change.
+ runUpdate(1, "A");
+ verifyRowCount(10);
+
+ // Verify these rows are unchanged.
+ verifyRow("A", "0", "0", "foo0", "0");
+ verifyRow("A", "2", "2", "foo2", "2");
+ verifyRow("A", "8", "8", "foo8", "8");
+ verifyRow("A", "9", "9", "foo9", "9");
+
+ // Verify these rows have been updated.
+ verifyRow("A", "3", "3", "foo6", "6");
+ verifyRow("A", "5", "5", "foo10", "10");
+ verifyRow("A", "7", "7", "foo14", "14");
+ }
+
+ @Test
+ public void testSubsetUpdate2() throws Exception {
+ // Update only some of the rows in the db. Also include some
+ // updates that do not affect actual rows in the table.
+ // These should just be ignored.
+
+ populateDatabase(10);
+ // Create two files that update four rows each.
+ // File0 updates A=-2..1 (-2 and -1 don't exist).
+ // File1 updates A=8..11 (10 and 11 don't exist).
+ createUpdateFiles(2, 4, 0, -2, 8);
+ runUpdate(2, "A");
+ verifyRowCount(10);
+
+ // Verify these rows are unchanged.
+ verifyRow("A", "4", "4", "foo4", "4");
+ verifyRow("A", "7", "7", "foo7", "7");
+
+ // Verify these updates succeeded.
+ verifyRow("A", "1", "1", "foo2", "2");
+ verifyRow("A", "8", "8", "foo16", "16");
+ verifyRow("A", "9", "9", "foo18", "18");
+ }
+
+ /**
+ * Test updating only subset of the columns.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testUpdateColumnSubset() throws Exception {
+ populateDatabase(4);
+ createUpdateFiles(1, 3, 0);
+
+ runExport(getArgv(true, 2, 2, "-m", "1",
+ "--update-key", "A", "--columns", "A,B"));
+
+ verifyRowCount(4);
+
+ // First column should not have any changes (even though it was updated)
+ verifyRow("A", "0", "0", "foo0", "0");
+
+ // Second column have updated column B, but C should be left untouched
+ verifyRow("A", "1", "1", "foo2", "1");
+
+ // Third column have updated column B, but C should be left untouched
+ verifyRow("A", "2", "2", "foo4", "2");
+
+ // Last columns should be completely untouched
+ verifyRow("A", "3", "3", "foo3", "3");
+ }
+
+ /**
+ * Parameter --columns must be superset of --update-key in order for
+ * CompilationManager and other parts of the framework work correctly.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testUpdateColumnNotInColumns() throws Exception {
+ populateDatabase(1);
+
+ thrown.expect(IOException.class);
+ thrown.reportMissingExceptionWithMessage("Expected IOException as --columns is not a superset of --update-key");
+ runExport(getArgv(true, 2, 2, "-m", "1",
+ "--update-key", "A", "--columns", "B"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sqoop/blob/6984a36c/src/test/org/apache/sqoop/TestExportUsingProcedure.java
----------------------------------------------------------------------
diff --git a/src/test/org/apache/sqoop/TestExportUsingProcedure.java b/src/test/org/apache/sqoop/TestExportUsingProcedure.java
index 1783676..64e7af3 100644
--- a/src/test/org/apache/sqoop/TestExportUsingProcedure.java
+++ b/src/test/org/apache/sqoop/TestExportUsingProcedure.java
@@ -37,8 +37,6 @@ import org.h2.Driver;
import org.junit.After;
import org.junit.Before;
-import com.cloudera.sqoop.SqoopOptions;
-import com.cloudera.sqoop.TestExport;
import org.junit.Rule;
import org.junit.rules.TestName;