github/sqoop
5
0
Fork 0
mirror of https://github.com/apache/sqoop.git synced 2025-05-02 16:21:18 +08:00
Code Issues Projects Releases Wiki Activity

Merge 60aafe073c into f8beae32a0

Browse Source
This commit is contained in:
fescandell 2021-03-17 22:31:31 +08:00 committed by GitHub
commit 4e5b06bf3e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 37 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/java/org/apache/sqoop/manager/DirectNetezzaManager.java
View File

@ -23,6 +23,7 @@
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Arrays;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.GnuParser;
@ -86,6 +87,8 @@ public class DirectNetezzaManager extends NetezzaManager {
public static final String NETEZZA_TABLE_ENCODING_LONG_ARG =
"encoding";
public static final String NETEZZA_SCHEMA_OPT = "netezza.schema";
public static final String NETEZZA_TABLE_SCHEMA_LONG_ARG = "schema";
private static final String QUERY_CHECK_DICTIONARY_FOR_TABLE =
"SELECT 1 FROM _V_TABLE WHERE OWNER= ? "
@ -293,6 +296,9 @@ protected RelatedOptions getNetezzaExtraOpts() {
netezzaOpts.addOption(OptionBuilder.withArgName(NETEZZA_TABLE_ENCODING_OPT)
.hasArg().withDescription("Table encoding")
.withLongOpt(NETEZZA_TABLE_ENCODING_LONG_ARG).create());
netezzaOpts.addOption(OptionBuilder.withArgName(NETEZZA_SCHEMA_OPT)
.hasArg().withDescription("Allow Schema")
.withLongOpt(NETEZZA_TABLE_SCHEMA_LONG_ARG).create());
return netezzaOpts;
}
@ -302,6 +308,8 @@ private void handleNetezzaExtraArgs(SqoopOptions opts)
Configuration conf = opts.getConf();
String[] extraArgs = opts.getExtraArgs();
LOG.debug("extraArgs " + Arrays.toString(extraArgs));
RelatedOptions netezzaOpts = getNetezzaExtraOpts();
CommandLine cmdLine = new GnuParser().parse(netezzaOpts, extraArgs, true);
@ -320,6 +328,12 @@ private void handleNetezzaExtraArgs(SqoopOptions opts)
conf.set(NETEZZA_TABLE_ENCODING_OPT, encoding);
}
if (cmdLine.hasOption(NETEZZA_TABLE_SCHEMA_LONG_ARG)) {
String schemaName = cmdLine.getOptionValue(NETEZZA_TABLE_SCHEMA_LONG_ARG);
LOG.info("We will use schema " + schemaName);
conf.set(NETEZZA_SCHEMA_OPT, schemaName);
}
conf.setBoolean(NETEZZA_CTRL_CHARS_OPT,
cmdLine.hasOption(NETEZZA_CTRL_CHARS_LONG_ARG));
@ -353,6 +367,20 @@ public boolean isORMFacilitySelfManaged() {
public boolean isDirectModeHCatSupported() {
return true;
}
@Override
public String escapeTableName(String tableName) {
Configuration conf = options.getConf();
String schema = conf.get(NETEZZA_SCHEMA_OPT);
// Return table name including schema if requested
if (schema != null && !schema.isEmpty()) {
return escapeIdentifier(schema) + "." + escapeIdentifier(tableName);
}
return escapeIdentifier(tableName);
}
public static String getLocalLogDir(TaskAttemptID attemptId) {

11
src/java/org/apache/sqoop/mapreduce/db/netezza/NetezzaExternalTableImportMapper.java
View File

@ -76,8 +76,15 @@ private String getSqlStatement(int myId) throws IOException {
char ec = (char) conf.getInt(DelimiterSet.OUTPUT_ESCAPED_BY_KEY, 0);
String nullValue = conf.get(DirectNetezzaManager.NETEZZA_NULL_VALUE);
String schema = conf.get(DirectNetezzaManager.NETEZZA_SCHEMA_OPT);
String tableName = dbc.getInputTableName();
if (schema!=null)
{
tableName = schema +"."+tableName;
}
int errorThreshold = conf.getInt(
DirectNetezzaManager.NETEZZA_ERROR_THRESHOLD_OPT, 1);
String logDir = conf.get(DirectNetezzaManager.NETEZZA_LOG_DIR_OPT);
@ -130,7 +137,7 @@ private String getSqlStatement(int myId) throws IOException {
sqlStmt.append(',').append(cols[i]);
}
}
sqlStmt.append(" FROM ").append(dbc.getInputTableName()).append(' ');
sqlStmt.append(" FROM ").append(tableName).append(' ');
sqlStmt.append("WHERE (DATASLICEID % ");
sqlStmt.append(numMappers).append(") = ").append(myId);
if (inputConds != null && inputConds.length() > 0) {