mirror of https://github.com/apache/nifi.git
PutSQLBinary
Signed-off-by: Matt Burgess <mattyb149@apache.org> NIFI-2591 Signed-off-by: Matt Burgess <mattyb149@apache.org> NIFI-2591 Signed-off-by: Matt Burgess <mattyb149@apache.org> NIFI-2591 - Added Format option for binary data types. Updated unit tests. Signed-off-by: Matt Burgess <mattyb149@apache.org> This closes #883
This commit is contained in:
Peter Wicks
Matt Burgess
parent
afb9a0016f
commit
e210172d93
|
|
@ -39,8 +39,11 @@ import org.apache.nifi.processor.io.InputStreamCallback;
|
|||
import org.apache.nifi.processor.util.StandardValidators;
|
||||
import org.apache.nifi.stream.io.StreamUtils;
|
||||
|
||||
import javax.xml.bind.DatatypeConverter;
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.UnsupportedEncodingException;
|
||||
import java.math.BigDecimal;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.sql.BatchUpdateException;
|
||||
|
|
@ -89,7 +92,13 @@ import java.util.regex.Pattern;
|
|||
@ReadsAttribute(attribute = "sql.args.N.type", description = "Incoming FlowFiles are expected to be parameterized SQL statements. The type of each Parameter is specified as an integer "
|
||||
+ "that represents the JDBC Type of the parameter."),
|
||||
@ReadsAttribute(attribute = "sql.args.N.value", description = "Incoming FlowFiles are expected to be parameterized SQL statements. The value of the Parameters are specified as "
|
||||
+ "sql.args.1.value, sql.args.2.value, sql.args.3.value, and so on. The type of the sql.args.1.value Parameter is specified by the sql.args.1.type attribute.")
|
||||
+ "sql.args.1.value, sql.args.2.value, sql.args.3.value, and so on. The type of the sql.args.1.value Parameter is specified by the sql.args.1.type attribute."),
|
||||
@ReadsAttribute(attribute = "sql.args.N.format", description = "This attribute is always optional, but default options may not always work for your data. "
|
||||
+ "Incoming FlowFiles are expected to be parameterized SQL statements. In some cases "
|
||||
+ "a format option needs to be specified, currently this is only applicable for binary data types. For binary data types "
|
||||
+ "available options are 'ascii', 'base64' and 'hex'. In 'ascii' format each string character in your attribute value represents a single byte, this is the default format "
|
||||
+ "and the format provided by Avro Processors. In 'base64' format your string is a Base64 encoded string. In 'hex' format the string is hex encoded with all "
|
||||
+ "letters in upper case and no '0x' at the beginning.")
|
||||
})
|
||||
@WritesAttributes({
|
||||
@WritesAttribute(attribute = "sql.generated.key", description = "If the database generated a key for an INSERT statement and the Obtain Generated Keys property is set to true, "
|
||||
|
|
@ -610,13 +619,17 @@ public class PutSQL extends AbstractProcessor {
|
|||
final int jdbcType = Integer.parseInt(entry.getValue());
|
||||
final String valueAttrName = "sql.args." + parameterIndex + ".value";
|
||||
final String parameterValue = attributes.get(valueAttrName);
|
||||
final String formatAttrName = "sql.args." + parameterIndex + ".format";
|
||||
final String parameterFormat = attributes.containsKey(formatAttrName)? attributes.get(formatAttrName):"";
|
||||
|
||||
try {
|
||||
setParameter(stmt, valueAttrName, parameterIndex, parameterValue, jdbcType);
|
||||
setParameter(stmt, valueAttrName, parameterIndex, parameterValue, jdbcType, parameterFormat);
|
||||
} catch (final NumberFormatException nfe) {
|
||||
throw new ProcessException("The value of the " + valueAttrName + " is '" + parameterValue + "', which cannot be converted into the necessary data type", nfe);
|
||||
} catch (ParseException pe) {
|
||||
throw new ProcessException("The value of the " + valueAttrName + " is '" + parameterValue + "', which cannot be converted to a timestamp", pe);
|
||||
} catch (UnsupportedEncodingException uee) {
|
||||
throw new ProcessException("The value of the " + valueAttrName + " is '" + parameterValue + "', which cannot be converted to UTF-8", uee);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -735,7 +748,9 @@ public class PutSQL extends AbstractProcessor {
|
|||
* @param jdbcType the JDBC Type of the SQL parameter to set
|
||||
* @throws SQLException if the PreparedStatement throws a SQLException when calling the appropriate setter
|
||||
*/
|
||||
private void setParameter(final PreparedStatement stmt, final String attrName, final int parameterIndex, final String parameterValue, final int jdbcType) throws SQLException, ParseException {
|
||||
private void setParameter(final PreparedStatement stmt, final String attrName, final int parameterIndex, final String parameterValue, final int jdbcType,
|
||||
final String valueFormat)
|
||||
throws SQLException, ParseException, UnsupportedEncodingException {
|
||||
if (parameterValue == null) {
|
||||
stmt.setNull(parameterIndex, jdbcType);
|
||||
} else {
|
||||
|
|
@ -786,6 +801,29 @@ public class PutSQL extends AbstractProcessor {
|
|||
|
||||
stmt.setTimestamp(parameterIndex, new Timestamp(lTimestamp));
|
||||
|
||||
break;
|
||||
case Types.BINARY:
|
||||
case Types.VARBINARY:
|
||||
case Types.LONGVARBINARY:
|
||||
byte[] bValue;
|
||||
|
||||
switch(valueFormat){
|
||||
case "":
|
||||
case "ascii":
|
||||
bValue = parameterValue.getBytes("ASCII");
|
||||
break;
|
||||
case "hex":
|
||||
bValue = DatatypeConverter.parseHexBinary(parameterValue);
|
||||
break;
|
||||
case "base64":
|
||||
bValue = DatatypeConverter.parseBase64Binary(parameterValue);
|
||||
break;
|
||||
default:
|
||||
throw new ParseException("Unable to parse binary data using the formatter `" + valueFormat + "`.",0);
|
||||
}
|
||||
|
||||
stmt.setBinaryStream(parameterIndex, new ByteArrayInputStream(bValue), bValue.length);
|
||||
|
||||
break;
|
||||
case Types.CHAR:
|
||||
case Types.VARCHAR:
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue;
|
|||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DriverManager;
|
||||
import java.sql.ResultSet;
|
||||
|
|
@ -30,9 +31,11 @@ import java.sql.Statement;
|
|||
import java.sql.Types;
|
||||
import java.text.ParseException;
|
||||
import java.text.SimpleDateFormat;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.commons.lang3.RandomUtils;
|
||||
import org.apache.nifi.controller.AbstractControllerService;
|
||||
import org.apache.nifi.dbcp.DBCPService;
|
||||
import org.apache.nifi.processor.exception.ProcessException;
|
||||
|
|
@ -46,6 +49,8 @@ import org.junit.Test;
|
|||
import org.junit.rules.TemporaryFolder;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import javax.xml.bind.DatatypeConverter;
|
||||
|
||||
public class TestPutSQL {
|
||||
private static final String createPersons = "CREATE TABLE PERSONS (id integer primary key, name varchar(100), code integer)";
|
||||
private static final String createPersonsAutoId = "CREATE TABLE PERSONS_AI (id INTEGER NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1), name VARCHAR(100), code INTEGER check(code <= 100))";
|
||||
|
|
@ -304,6 +309,136 @@ public class TestPutSQL {
|
|||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBinaryColumnTypes() throws InitializationException, ProcessException, SQLException, IOException, ParseException {
|
||||
final TestRunner runner = TestRunners.newTestRunner(PutSQL.class);
|
||||
try (final Connection conn = service.getConnection()) {
|
||||
try (final Statement stmt = conn.createStatement()) {
|
||||
stmt.executeUpdate("CREATE TABLE BINARYTESTS (id integer primary key, bn1 CHAR(8) FOR BIT DATA, bn2 VARCHAR(100) FOR BIT DATA, " +
|
||||
"bn3 LONG VARCHAR FOR BIT DATA)");
|
||||
}
|
||||
}
|
||||
|
||||
runner.addControllerService("dbcp", service);
|
||||
runner.enableControllerService(service);
|
||||
runner.setProperty(PutSQL.CONNECTION_POOL, "dbcp");
|
||||
|
||||
final byte[] insertStatement = "INSERT INTO BINARYTESTS (ID, bn1, bn2, bn3) VALUES (?, ?, ?, ?)".getBytes();
|
||||
|
||||
final String arg2BIN = fixedSizeByteArrayAsASCIIString(8);
|
||||
final String art3VARBIN = fixedSizeByteArrayAsASCIIString(50);
|
||||
final String art4LongBin = fixedSizeByteArrayAsASCIIString(32700); //max size supported by Derby
|
||||
|
||||
//ASCII (default) binary formatn
|
||||
Map<String, String> attributes = new HashMap<>();
|
||||
attributes.put("sql.args.1.type", String.valueOf(Types.INTEGER));
|
||||
attributes.put("sql.args.1.value", "1");
|
||||
attributes.put("sql.args.2.type", String.valueOf(Types.BINARY));
|
||||
attributes.put("sql.args.2.value", arg2BIN);
|
||||
attributes.put("sql.args.3.type", String.valueOf(Types.VARBINARY));
|
||||
attributes.put("sql.args.3.value", art3VARBIN);
|
||||
attributes.put("sql.args.4.type", String.valueOf(Types.LONGVARBINARY));
|
||||
attributes.put("sql.args.4.value", art4LongBin);
|
||||
|
||||
runner.enqueue(insertStatement, attributes);
|
||||
|
||||
//ASCII with specified format
|
||||
attributes = new HashMap<>();
|
||||
attributes.put("sql.args.1.type", String.valueOf(Types.INTEGER));
|
||||
attributes.put("sql.args.1.value", "2");
|
||||
attributes.put("sql.args.2.type", String.valueOf(Types.BINARY));
|
||||
attributes.put("sql.args.2.value", arg2BIN);
|
||||
attributes.put("sql.args.2.format", "ascii");
|
||||
attributes.put("sql.args.3.type", String.valueOf(Types.VARBINARY));
|
||||
attributes.put("sql.args.3.value", art3VARBIN);
|
||||
attributes.put("sql.args.3.format", "ascii");
|
||||
attributes.put("sql.args.4.type", String.valueOf(Types.LONGVARBINARY));
|
||||
attributes.put("sql.args.4.value", art4LongBin);
|
||||
attributes.put("sql.args.4.format", "ascii");
|
||||
|
||||
runner.enqueue(insertStatement, attributes);
|
||||
|
||||
//Hex
|
||||
final String arg2HexBIN = fixedSizeByteArrayAsHexString(8);
|
||||
final String art3HexVARBIN = fixedSizeByteArrayAsHexString(50);
|
||||
final String art4HexLongBin = fixedSizeByteArrayAsHexString(32700);
|
||||
|
||||
attributes = new HashMap<>();
|
||||
attributes.put("sql.args.1.type", String.valueOf(Types.INTEGER));
|
||||
attributes.put("sql.args.1.value", "3");
|
||||
attributes.put("sql.args.2.type", String.valueOf(Types.BINARY));
|
||||
attributes.put("sql.args.2.value", arg2HexBIN);
|
||||
attributes.put("sql.args.2.format", "hex");
|
||||
attributes.put("sql.args.3.type", String.valueOf(Types.VARBINARY));
|
||||
attributes.put("sql.args.3.value", art3HexVARBIN);
|
||||
attributes.put("sql.args.3.format", "hex");
|
||||
attributes.put("sql.args.4.type", String.valueOf(Types.LONGVARBINARY));
|
||||
attributes.put("sql.args.4.value", art4HexLongBin);
|
||||
attributes.put("sql.args.4.format", "hex");
|
||||
|
||||
runner.enqueue(insertStatement, attributes);
|
||||
|
||||
//Base64
|
||||
final String arg2Base64BIN = fixedSizeByteArrayAsBase64String(8);
|
||||
final String art3Base64VARBIN = fixedSizeByteArrayAsBase64String(50);
|
||||
final String art4Base64LongBin = fixedSizeByteArrayAsBase64String(32700);
|
||||
|
||||
attributes = new HashMap<>();
|
||||
attributes.put("sql.args.1.type", String.valueOf(Types.INTEGER));
|
||||
attributes.put("sql.args.1.value", "4");
|
||||
attributes.put("sql.args.2.type", String.valueOf(Types.BINARY));
|
||||
attributes.put("sql.args.2.value", arg2Base64BIN);
|
||||
attributes.put("sql.args.2.format", "base64");
|
||||
attributes.put("sql.args.3.type", String.valueOf(Types.VARBINARY));
|
||||
attributes.put("sql.args.3.value", art3Base64VARBIN);
|
||||
attributes.put("sql.args.3.format", "base64");
|
||||
attributes.put("sql.args.4.type", String.valueOf(Types.LONGVARBINARY));
|
||||
attributes.put("sql.args.4.value", art4Base64LongBin);
|
||||
attributes.put("sql.args.4.format", "base64");
|
||||
|
||||
runner.enqueue(insertStatement, attributes);
|
||||
|
||||
runner.run();
|
||||
|
||||
runner.assertAllFlowFilesTransferred(PutSQL.REL_SUCCESS, 4);
|
||||
|
||||
try (final Connection conn = service.getConnection()) {
|
||||
try (final Statement stmt = conn.createStatement()) {
|
||||
final ResultSet rs = stmt.executeQuery("SELECT * FROM BINARYTESTS");
|
||||
|
||||
//First Batch
|
||||
assertTrue(rs.next());
|
||||
assertEquals(1, rs.getInt(1));
|
||||
assertTrue(Arrays.equals(arg2BIN.getBytes("ASCII"), rs.getBytes(2)));
|
||||
assertTrue(Arrays.equals(art3VARBIN.getBytes("ASCII"), rs.getBytes(3)));
|
||||
assertTrue(Arrays.equals(art4LongBin.getBytes("ASCII"), rs.getBytes(4)));
|
||||
|
||||
//Second batch
|
||||
assertTrue(rs.next());
|
||||
assertEquals(2, rs.getInt(1));
|
||||
assertTrue(Arrays.equals(arg2BIN.getBytes("ASCII"), rs.getBytes(2)));
|
||||
assertTrue(Arrays.equals(art3VARBIN.getBytes("ASCII"), rs.getBytes(3)));
|
||||
assertTrue(Arrays.equals(art4LongBin.getBytes("ASCII"), rs.getBytes(4)));
|
||||
|
||||
//Third Batch (Hex)
|
||||
assertTrue(rs.next());
|
||||
assertEquals(3, rs.getInt(1));
|
||||
assertTrue(Arrays.equals(DatatypeConverter.parseHexBinary(arg2HexBIN), rs.getBytes(2)));
|
||||
assertTrue(Arrays.equals(DatatypeConverter.parseHexBinary(art3HexVARBIN), rs.getBytes(3)));
|
||||
assertTrue(Arrays.equals(DatatypeConverter.parseHexBinary(art4HexLongBin), rs.getBytes(4)));
|
||||
|
||||
//Fourth Batch (Base64)
|
||||
assertTrue(rs.next());
|
||||
assertEquals(4, rs.getInt(1));
|
||||
assertTrue(Arrays.equals(DatatypeConverter.parseBase64Binary(arg2Base64BIN), rs.getBytes(2)));
|
||||
assertTrue(Arrays.equals(DatatypeConverter.parseBase64Binary(art3Base64VARBIN), rs.getBytes(3)));
|
||||
assertTrue(Arrays.equals(DatatypeConverter.parseBase64Binary(art4Base64LongBin), rs.getBytes(4)));
|
||||
|
||||
assertFalse(rs.next());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStatementsWithPreparedParameters() throws InitializationException, ProcessException, SQLException, IOException {
|
||||
final TestRunner runner = TestRunners.newTestRunner(PutSQL.class);
|
||||
|
|
@ -671,4 +806,24 @@ public class TestPutSQL {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
private String fixedSizeByteArrayAsASCIIString(int length){
|
||||
byte[] bBinary = RandomUtils.nextBytes(length);
|
||||
ByteBuffer bytes = ByteBuffer.wrap(bBinary);
|
||||
StringBuffer sbBytes = new StringBuffer();
|
||||
for (int i = bytes.position(); i < bytes.limit(); i++)
|
||||
sbBytes.append((char)bytes.get(i));
|
||||
|
||||
return sbBytes.toString();
|
||||
}
|
||||
|
||||
private String fixedSizeByteArrayAsHexString(int length){
|
||||
byte[] bBinary = RandomUtils.nextBytes(length);
|
||||
return DatatypeConverter.printHexBinary(bBinary);
|
||||
}
|
||||
|
||||
private String fixedSizeByteArrayAsBase64String(int length){
|
||||
byte[] bBinary = RandomUtils.nextBytes(length);
|
||||
return DatatypeConverter.printBase64Binary(bBinary);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue