Skip to content

Commit f3f6d59

Browse files
vhlinskyivhlinskyi
committed
CDAP-15550 Netezza db plugin enhacements: all data types support + proper test coverage - fix style
1 parent 57bb68d commit f3f6d59

File tree

3 files changed

+114
-103
lines changed

3 files changed

+114
-103
lines changed

‎netezza-plugin/src/test/java/io/cdap/plugin/netezza/NetezzaPluginTestBase.java‎

Lines changed: 77 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
import com.google.common.base.Throwables;
2121
import com.google.common.collect.ImmutableMap;
2222
import com.google.common.collect.Sets;
23-
2423
import io.cdap.cdap.api.artifact.ArtifactSummary;
2524
import io.cdap.cdap.api.plugin.PluginClass;
2625
import io.cdap.cdap.datapipeline.DataPipelineApp;
@@ -32,7 +31,6 @@
3231
import io.cdap.plugin.db.batch.DatabasePluginTestBase;
3332
import io.cdap.plugin.db.batch.sink.ETLDBOutputFormat;
3433
import io.cdap.plugin.db.batch.source.DataDrivenETLDBInputFormat;
35-
3634
import org.junit.AfterClass;
3735
import org.junit.BeforeClass;
3836
import org.junit.ClassRule;
@@ -44,13 +42,10 @@
4442
import java.sql.Driver;
4543
import java.sql.DriverManager;
4644
import java.sql.PreparedStatement;
47-
import java.sql.ResultSet;
48-
import java.sql.ResultSetMetaData;
4945
import java.sql.SQLException;
5046
import java.sql.Statement;
5147
import java.sql.Time;
5248
import java.sql.Timestamp;
53-
import java.util.Arrays;
5449
import java.util.Calendar;
5550
import java.util.Collections;
5651
import java.util.Map;
@@ -97,9 +92,9 @@ public static void setupTest() throws Exception {
9792
setupBatchArtifacts(DATAPIPELINE_ARTIFACT_ID, DataPipelineApp.class);
9893

9994
addPluginArtifact(NamespaceId.DEFAULT.artifact(JDBC_DRIVER_NAME, "1.0.0"),
100-
DATAPIPELINE_ARTIFACT_ID,
101-
NetezzaSource.class, NetezzaSink.class, DBRecord.class, ETLDBOutputFormat.class,
102-
DataDrivenETLDBInputFormat.class, DBRecord.class, NetezzaPostAction.class, NetezzaAction.class);
95+
DATAPIPELINE_ARTIFACT_ID,
96+
NetezzaSource.class, NetezzaSink.class, DBRecord.class, ETLDBOutputFormat.class,
97+
DataDrivenETLDBInputFormat.class, DBRecord.class, NetezzaPostAction.class, NetezzaAction.class);
10398

10499
connectionUrl = "jdbc:netezza://" + BASE_PROPS.get(ConnectionConfig.HOST) + ":" +
105100
BASE_PROPS.get(ConnectionConfig.PORT) + "/" + BASE_PROPS.get(ConnectionConfig.DATABASE);
@@ -108,11 +103,11 @@ public static void setupTest() throws Exception {
108103

109104
// add netezza 3rd party plugin
110105
PluginClass netezzaDriver = new PluginClass(ConnectionConfig.JDBC_PLUGIN_TYPE, JDBC_DRIVER_NAME,
111-
"netezza driver class", aClass.getCanonicalName(),
112-
null, Collections.emptyMap());
106+
"netezza driver class", aClass.getCanonicalName(),
107+
null, Collections.emptyMap());
113108
addPluginArtifact(NamespaceId.DEFAULT.artifact("netezza-jdbc-connector", "1.0.0"),
114-
DATAPIPELINE_ARTIFACT_ID,
115-
Sets.newHashSet(netezzaDriver), aClass);
109+
DATAPIPELINE_ARTIFACT_ID,
110+
Sets.newHashSet(netezzaDriver), aClass);
116111

117112
TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
118113

@@ -128,64 +123,85 @@ protected static void createTestTables(Connection conn) throws SQLException {
128123
// create a table that the action will truncate at the end of the run
129124
stmt.execute("CREATE TABLE post_action_test (x int, day varchar(10))");
130125

131-
stmt.execute("create table MY_TABLE (INTEGER_COL INTEGER, BYTEINT_COL BYTEINT, SMALLINT_COL SMALLINT, " +
132-
"BIGINT_COL BIGINT, REAL_COL REAL, REAL_FLOAT_COL FLOAT(1), DOUBLE_FLOAT_COL FLOAT(7), " +
133-
"DOUBLE_PRECISION_COL DOUBLE PRECISION, NUMERIC_COL NUMERIC(" + PRECISION + "," + SCALE + "), " +
134-
"DECIMAL_COL DECIMAL(" + PRECISION + "," + SCALE + "), CHAR_COL CHAR(40), VARCHAR_COL VARCHAR(40), " +
135-
"NCHAR_COL NCHAR(40), NVARCHAR_COL NVARCHAR(40), VARBINARY_COL BINARY VARYING(10), " +
136-
"ST_GEOMETRY_COL ST_GEOMETRY(10), DATE_COL DATE, TIME_COL TIME, TIMETZ_COL TIMETZ, TIMESTAMP_COL TIMESTAMP, " +
137-
"INTERVAL_COL INTERVAL, BOOLEAN_COL BOOLEAN)");
126+
stmt.execute("create table MY_TABLE (" +
127+
"INTEGER_COL INTEGER, " +
128+
"BYTEINT_COL BYTEINT, " +
129+
"SMALLINT_COL SMALLINT, " +
130+
"BIGINT_COL BIGINT, " +
131+
"REAL_COL REAL, " +
132+
"REAL_FLOAT_COL FLOAT(1), " +
133+
"DOUBLE_FLOAT_COL FLOAT(7), " +
134+
"DOUBLE_PRECISION_COL DOUBLE PRECISION, " +
135+
"NUMERIC_COL NUMERIC(" + PRECISION + "," + SCALE + "), " +
136+
"DECIMAL_COL DECIMAL(" + PRECISION + "," + SCALE + "), " +
137+
"CHAR_COL CHAR(40), " +
138+
"VARCHAR_COL VARCHAR(40), " +
139+
"NCHAR_COL NCHAR(40), " +
140+
"NVARCHAR_COL NVARCHAR(40), " +
141+
"VARBINARY_COL BINARY VARYING(10), " +
142+
"ST_GEOMETRY_COL ST_GEOMETRY(10), " +
143+
"DATE_COL DATE, " +
144+
"TIME_COL TIME, " +
145+
"TIMETZ_COL TIMETZ, " +
146+
"TIMESTAMP_COL TIMESTAMP, " +
147+
"INTERVAL_COL INTERVAL, " +
148+
"BOOLEAN_COL BOOLEAN)");
138149

139150
stmt.execute("CREATE TABLE MY_DEST_TABLE AS SELECT * FROM my_table");
140151
stmt.execute("CREATE TABLE YOUR_TABLE AS SELECT * FROM my_table");
141152
}
142153
}
143154

144155
protected static void prepareTestData(Connection conn) throws SQLException {
145-
try (Statement stmt = conn.createStatement()) {
156+
try (
157+
Statement stmt = conn.createStatement();
158+
PreparedStatement pStmt1 =
159+
conn.prepareStatement("INSERT INTO my_table " +
160+
"VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
161+
" ?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
162+
" ?, ?)");
163+
PreparedStatement pStmt2 =
164+
conn.prepareStatement("INSERT INTO your_table " +
165+
"VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
166+
" ?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
167+
" ?, ?)")) {
168+
146169
stmt.execute("insert into db_action_test values (1, '1970-01-01')");
147170
stmt.execute("insert into post_action_test values (1, '1970-01-01')");
148171

149-
for (String tableName : Arrays.asList("MY_TABLE", "YOUR_TABLE")) {
150-
for (int i = 1; i <= 5; i++) {
151-
String name = "user" + i;
152-
try (PreparedStatement pStmt = conn.prepareStatement("insert into " + tableName + " values(" +
153-
"?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")) {
154-
pStmt.setString(11, name);
155-
pStmt.setString(12, name);
156-
pStmt.setString(13, name);
157-
pStmt.setString(14, name);
158-
pStmt.setBytes(15, name.getBytes(Charsets.UTF_8));
159-
pStmt.setBytes(16, name.getBytes(Charsets.UTF_8));
160-
161-
pStmt.setInt(1, i);
162-
pStmt.setInt(2, i);
163-
pStmt.setInt(3, i);
164-
pStmt.setLong(4, i);
165-
166-
pStmt.setFloat(5, 123.45f + i);
167-
pStmt.setFloat(6, 123.45f + i);
168-
pStmt.setDouble(7, 123.45 + i);
169-
pStmt.setDouble(8, 123.45 + i);
170-
pStmt.setBigDecimal(
171-
9,
172-
new BigDecimal(123.45, MathContext.DECIMAL64).add(new BigDecimal(i, MathContext.DECIMAL64))
173-
);
174-
pStmt.setBigDecimal(
175-
10,
176-
new BigDecimal(123.45, MathContext.DECIMAL64).add(new BigDecimal(i, MathContext.DECIMAL64))
177-
);
178-
pStmt.setDate(17, new Date(CURRENT_TS));
179-
pStmt.setTime(18, new Time(CURRENT_TS));
180-
pStmt.setString(19, "13:24:16+03");
181-
pStmt.setTimestamp(20, new Timestamp(CURRENT_TS));
182-
pStmt.setString(21, "2 year 3 month " + i + " day");
183-
// other
184-
pStmt.setBoolean(22, (i % 2 == 0));
185-
186-
pStmt.execute();
187-
}
188-
}
172+
populateData(pStmt1, pStmt2);
173+
}
174+
}
175+
176+
private static void populateData(PreparedStatement... stmts) throws SQLException {
177+
// insert the same data into both tables: my_table and your_table
178+
for (PreparedStatement pStmt : stmts) {
179+
for (int i = 1; i <= 5; i++) {
180+
String name = "user" + i;
181+
pStmt.setInt(1, i);
182+
pStmt.setInt(2, i);
183+
pStmt.setInt(3, i);
184+
pStmt.setLong(4, i);
185+
pStmt.setFloat(5, 123.45f + i);
186+
pStmt.setFloat(6, 123.45f + i);
187+
pStmt.setDouble(7, 123.45 + i);
188+
pStmt.setDouble(8, 123.45 + i);
189+
pStmt.setBigDecimal(9, new BigDecimal(123.45, MathContext.DECIMAL64).add(new BigDecimal(i, MathContext.DECIMAL64)));
190+
pStmt.setBigDecimal(10, new BigDecimal(123.45, MathContext.DECIMAL64).add(new BigDecimal(i, MathContext.DECIMAL64)));
191+
pStmt.setString(11, name);
192+
pStmt.setString(12, name);
193+
pStmt.setString(13, name);
194+
pStmt.setString(14, name);
195+
pStmt.setBytes(15, name.getBytes(Charsets.UTF_8));
196+
pStmt.setBytes(16, name.getBytes(Charsets.UTF_8));
197+
pStmt.setDate(17, new Date(CURRENT_TS));
198+
pStmt.setTime(18, new Time(CURRENT_TS));
199+
pStmt.setString(19, "13:24:16+03");
200+
pStmt.setTimestamp(20, new Timestamp(CURRENT_TS));
201+
pStmt.setString(21, "2 year 3 month " + i + " day");
202+
pStmt.setBoolean(22, (i % 2 == 0));
203+
204+
pStmt.executeUpdate();
189205
}
190206
}
191207
}
@@ -194,7 +210,7 @@ public static Connection createConnection() {
194210
try {
195211
Class.forName(Driver.class.getCanonicalName());
196212
return DriverManager.getConnection(connectionUrl, BASE_PROPS.get(ConnectionConfig.USER),
197-
BASE_PROPS.get(ConnectionConfig.PASSWORD));
213+
BASE_PROPS.get(ConnectionConfig.PASSWORD));
198214
} catch (Exception e) {
199215
throw Throwables.propagate(e);
200216
}

‎netezza-plugin/src/test/java/io/cdap/plugin/netezza/NetezzaSinkTestRun.java‎

Lines changed: 23 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
import com.google.common.base.Charsets;
2020
import com.google.common.collect.ImmutableMap;
2121
import com.google.common.collect.ImmutableSet;
22-
2322
import io.cdap.cdap.api.common.Bytes;
2423
import io.cdap.cdap.api.data.format.StructuredRecord;
2524
import io.cdap.cdap.api.data.schema.Schema;
@@ -31,7 +30,6 @@
3130
import io.cdap.cdap.test.DataSetManager;
3231
import io.cdap.plugin.common.Constants;
3332
import io.cdap.plugin.db.batch.sink.AbstractDBSink;
34-
3533
import org.junit.Assert;
3634
import org.junit.Before;
3735
import org.junit.Test;
@@ -165,35 +163,34 @@ public void testDBSink(String appName, String inputDatasetName, boolean setInput
165163
private void createInputData(String inputDatasetName) throws Exception {
166164
// add some data to the input table
167165
DataSetManager<Table> inputManager = getDataset(inputDatasetName);
168-
169166
List<StructuredRecord> inputRecords = new ArrayList<>();
170167
LocalDateTime localDateTime = new Timestamp(CURRENT_TS).toLocalDateTime();
171168
for (int i = 1; i <= 2; i++) {
172169
String name = "user" + i;
173170
inputRecords.add(StructuredRecord.builder(SCHEMA)
174-
.set("INTEGER_COL", i)
175-
.set("BYTEINT_COL", i)
176-
.set("SMALLINT_COL", i)
177-
.set("BIGINT_COL", (long) i)
178-
.set("REAL_COL", 3.451f + i)
179-
.set("REAL_FLOAT_COL", 3.451f + i)
180-
.set("DOUBLE_FLOAT_COL", 3.451 + i)
181-
.set("DOUBLE_PRECISION_COL", 3.451 + i)
182-
.setDecimal("NUMERIC_COL", NUMERIC_VALUE)
183-
.setDecimal("DECIMAL_COL", NUMERIC_VALUE)
184-
.set("CHAR_COL", name)
185-
.set("VARCHAR_COL", name)
186-
.set("NCHAR_COL", name)
187-
.set("NVARCHAR_COL", name)
188-
.set("VARBINARY_COL", name.getBytes(Charsets.UTF_8))
189-
.set("ST_GEOMETRY_COL", name.getBytes(Charsets.UTF_8))
190-
.setDate("DATE_COL", localDateTime.toLocalDate())
191-
.setTime("TIME_COL", localDateTime.toLocalTime())
192-
.set("TIMETZ_COL", "13:24:16+03")
193-
.setTimestamp("TIMESTAMP_COL", localDateTime.atZone(ZoneId.ofOffset("UTC", ZoneOffset.UTC)))
194-
.set("INTERVAL_COL", "2 years 3 mons 2 days")
195-
.set("BOOLEAN_COL", true)
196-
.build());
171+
.set("INTEGER_COL", i)
172+
.set("BYTEINT_COL", i)
173+
.set("SMALLINT_COL", i)
174+
.set("BIGINT_COL", (long) i)
175+
.set("REAL_COL", 3.451f + i)
176+
.set("REAL_FLOAT_COL", 3.451f + i)
177+
.set("DOUBLE_FLOAT_COL", 3.451 + i)
178+
.set("DOUBLE_PRECISION_COL", 3.451 + i)
179+
.setDecimal("NUMERIC_COL", NUMERIC_VALUE)
180+
.setDecimal("DECIMAL_COL", NUMERIC_VALUE)
181+
.set("CHAR_COL", name)
182+
.set("VARCHAR_COL", name)
183+
.set("NCHAR_COL", name)
184+
.set("NVARCHAR_COL", name)
185+
.set("VARBINARY_COL", name.getBytes(Charsets.UTF_8))
186+
.set("ST_GEOMETRY_COL", name.getBytes(Charsets.UTF_8))
187+
.setDate("DATE_COL", localDateTime.toLocalDate())
188+
.setTime("TIME_COL", localDateTime.toLocalTime())
189+
.set("TIMETZ_COL", "13:24:16+03")
190+
.setTimestamp("TIMESTAMP_COL", localDateTime.atZone(ZoneId.ofOffset("UTC", ZoneOffset.UTC)))
191+
.set("INTERVAL_COL", "2 years 3 mons 2 days")
192+
.set("BOOLEAN_COL", true)
193+
.build());
197194
}
198195
MockSource.writeInput(inputManager, inputRecords);
199196
}

‎netezza-plugin/src/test/java/io/cdap/plugin/netezza/NetezzaSourceTestRun.java‎

Lines changed: 14 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package io.cdap.plugin.netezza;
1818

1919
import com.google.common.collect.ImmutableMap;
20-
2120
import io.cdap.cdap.api.common.Bytes;
2221
import io.cdap.cdap.api.data.format.StructuredRecord;
2322
import io.cdap.cdap.api.dataset.table.Table;
@@ -34,7 +33,6 @@
3433
import io.cdap.plugin.common.Constants;
3534
import io.cdap.plugin.db.ConnectionConfig;
3635
import io.cdap.plugin.db.batch.source.AbstractDBSource;
37-
3836
import org.junit.Assert;
3937
import org.junit.Test;
4038

@@ -79,7 +77,7 @@ public void testDBMacroSupport() throws Exception {
7977
ETLPlugin sinkConfig = MockSink.getPlugin("macroOutputTable");
8078

8179
ApplicationManager appManager = deployETL(sourceConfig, sinkConfig,
82-
DATAPIPELINE_ARTIFACT, "testDBMacro");
80+
DATAPIPELINE_ARTIFACT, "testDBMacro");
8381
runETLOnce(appManager, ImmutableMap.of("logical.start.time", String.valueOf(CURRENT_TS)));
8482

8583
DataSetManager<Table> outputManager = getDataset("macroOutputTable");
@@ -112,7 +110,7 @@ public void testDBSource() throws Exception {
112110
ETLPlugin sinkConfig = MockSink.getPlugin(outputDatasetName);
113111

114112
ApplicationManager appManager = deployETL(sourceConfig, sinkConfig,
115-
DATAPIPELINE_ARTIFACT, "testDBSource");
113+
DATAPIPELINE_ARTIFACT, "testDBSource");
116114
runETLOnce(appManager);
117115

118116
DataSetManager<Table> outputManager = getDataset(outputDatasetName);
@@ -142,13 +140,13 @@ public void testDBSource() throws Exception {
142140
Assert.assertEquals(124.45, row1.get("DOUBLE_PRECISION_COL"), 0.000001);
143141
Assert.assertEquals(125.45, row2.get("DOUBLE_PRECISION_COL"), 0.000001);
144142
Assert.assertEquals(new BigDecimal(124.45, new MathContext(PRECISION)).setScale(SCALE),
145-
row1.getDecimal("NUMERIC_COL"));
143+
row1.getDecimal("NUMERIC_COL"));
146144
Assert.assertEquals(new BigDecimal(125.45, new MathContext(PRECISION)).setScale(SCALE),
147-
row2.getDecimal("NUMERIC_COL"));
145+
row2.getDecimal("NUMERIC_COL"));
148146
Assert.assertEquals(new BigDecimal(124.45, new MathContext(PRECISION)).setScale(SCALE),
149-
row1.getDecimal("DECIMAL_COL"));
147+
row1.getDecimal("DECIMAL_COL"));
150148
Assert.assertEquals(new BigDecimal(125.45, new MathContext(PRECISION)).setScale(SCALE),
151-
row2.getDecimal("DECIMAL_COL"));
149+
row2.getDecimal("DECIMAL_COL"));
152150

153151
Assert.assertEquals("user1", ((String) row1.get("CHAR_COL")).trim());
154152
Assert.assertEquals("user2", ((String) row2.get("CHAR_COL")).trim());
@@ -205,7 +203,7 @@ public void testDbSourceMultipleTables() throws Exception {
205203
ETLPlugin sinkConfig = MockSink.getPlugin(outputDatasetName);
206204

207205
ApplicationManager appManager = deployETL(sourceConfig, sinkConfig,
208-
DATAPIPELINE_ARTIFACT, "testDBSourceWithMultipleTables");
206+
DATAPIPELINE_ARTIFACT, "testDBSourceWithMultipleTables");
209207
runETLOnce(appManager);
210208

211209
// records should be written
@@ -261,22 +259,22 @@ public void testUserNamePasswordCombinations() throws Exception {
261259
Map<String, String> noUser = new HashMap<>(baseSourceProps);
262260
noUser.put(ConnectionConfig.PASSWORD, "password");
263261
database = new ETLStage("databaseSource",
264-
new ETLPlugin(NetezzaConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, noUser, null));
262+
new ETLPlugin(NetezzaConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, noUser, null));
265263
etlConfig = ETLBatchConfig.builder()
266264
.addStage(database)
267265
.addStage(table)
268266
.addConnection(database.getName(), table.getName())
269267
.build();
270268
assertDeploymentFailure(appId, etlConfig, DATAPIPELINE_ARTIFACT,
271-
"Deploying DB Source with null username but non-null password should have failed.");
269+
"Deploying DB Source with null username but non-null password should have failed.");
272270

273271
// non-null username, non-null, but empty password. Should succeed.
274272
// as source
275273
Map<String, String> emptyPassword = new HashMap<>(baseSourceProps);
276274
emptyPassword.put(ConnectionConfig.USER, "root");
277275
emptyPassword.put(ConnectionConfig.PASSWORD, "");
278276
database = new ETLStage("databaseSource",
279-
new ETLPlugin(NetezzaConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, emptyPassword, null));
277+
new ETLPlugin(NetezzaConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, emptyPassword, null));
280278
etlConfig = ETLBatchConfig.builder()
281279
.addStage(database)
282280
.addStage(table)
@@ -314,8 +312,8 @@ public void testNonExistentDBTable() throws Exception {
314312
.build();
315313
ApplicationId appId = NamespaceId.DEFAULT.app("dbSourceNonExistingTest");
316314
assertRuntimeFailure(appId, etlConfig, DATAPIPELINE_ARTIFACT,
317-
"ETL Application with DB Source should have failed because of a " +
318-
"non-existent source table.", 1);
315+
"ETL Application with DB Source should have failed because of a " +
316+
"non-existent source table.", 1);
319317

320318
// Bad connection
321319
ETLPlugin sourceBadConnConfig = new ETLPlugin(
@@ -341,7 +339,7 @@ public void testNonExistentDBTable() throws Exception {
341339
.addConnection(sourceBadConn.getName(), sink.getName())
342340
.build();
343341
assertRuntimeFailure(appId, etlConfig, DATAPIPELINE_ARTIFACT,
344-
"ETL Application with DB Source should have failed because of a " +
345-
"non-existent source database.", 2);
342+
"ETL Application with DB Source should have failed because of a " +
343+
"non-existent source database.", 2);
346344
}
347345
}

0 commit comments

Comments
 (0)