github/sqoop
5
0
Fork 0
mirror of https://github.com/apache/sqoop.git synced 2025-05-03 22:20:52 +08:00
Code Issues Projects Releases Wiki Activity

SQOOP-2847. Sqoop --incremental + missing parent --target-dir reports success with no data

Browse Source 
(Jarcec via Hari)
This commit is contained in:
Hari Shreedharan 2016-03-01 10:42:21 -08:00
parent 5c4d06d727
commit c2351511e6
2 changed files with 33 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/java/org/apache/sqoop/tool/ImportTool.java
View File

@ -430,6 +430,7 @@ private boolean initIncrementalConstraints(SqoopOptions options,
protected void lastModifiedMerge(SqoopOptions options, ImportJobContext context) throws IOException {
FileSystem fs = FileSystem.get(options.getConf());
if (context.getDestination() != null && fs.exists(context.getDestination())) {
LOG.info("Final destination exists, will run merge job.");
Path userDestDir = getOutputPath(options, context.getTableName(), false);
if (fs.exists(userDestDir)) {
String tableClassName = null;
@ -461,7 +462,16 @@ protected void lastModifiedMerge(SqoopOptions options, ImportJobContext context)
unloadJars();
} else {
fs.rename(context.getDestination(), userDestDir);
// Create parent directory(ies), otherwise fs.rename would fail
if(!fs.exists(userDestDir.getParent())) {
fs.mkdirs(userDestDir.getParent());
}
// And finally move the data
LOG.info("Moving data from temporary directory " + context.getDestination() + " to final destination " + userDestDir);
if(!fs.rename(context.getDestination(), userDestDir)) {
throw new RuntimeException("Couldn't move data from temporary directory " + context.getDestination() + " to final destination " + userDestDir);
}
}
}
}

23
src/test/com/cloudera/sqoop/TestIncrementalImport.java
View File

@ -49,6 +49,8 @@
import com.cloudera.sqoop.tool.ImportTool;
import com.cloudera.sqoop.tool.JobTool;
import javax.management.Query;
/**
* Test the incremental import functionality.
*
@ -543,12 +545,16 @@ private List<String> getArgListForQuery(String query, String directoryName,
if (commonArgs) {
CommonArgs.addHadoopFlags(args);
}
String [] directoryNames = directoryName.split("/");
String className = directoryNames[directoryNames.length -1];
args.add("--connect");
args.add(SOURCE_DB_URL);
args.add("--query");
args.add(query);
args.add("--class-name");
args.add(directoryName);
args.add(className);
args.add("--target-dir");
args.add(BaseSqoopTestCase.LOCAL_WAREHOUSE_DIR
+ System.getProperty("file.separator") + directoryName);
@ -775,6 +781,21 @@ public void testEmptyLastModified() throws Exception {
assertDirOfNumbers(TABLE_NAME, 0);
}
public void testEmptyLastModifiedWithNonExistingParentDirectory() throws Exception {
final String TABLE_NAME = "emptyLastModifiedNoParent";
final String QUERY = "SELECT id, last_modified FROM \"" + TABLE_NAME + "\" WHERE $CONDITIONS";
final String DIRECTORY = "non-existing/parents/" + TABLE_NAME;
createTimestampTable(TABLE_NAME, 0, null);
List<String> args = getArgListForQuery(QUERY, DIRECTORY, true, false, false);
Configuration conf = newConf();
SqoopOptions options = new SqoopOptions();
options.setConf(conf);
runImport(options, args);
assertDirOfNumbers(DIRECTORY, 0);
}
public void testFullLastModifiedImport() throws Exception {
// Given a table of rows imported in the past,
// see that they are imported.