Skip to content

Commit 78d5594

Browse files
albertshaualbertshau
authored
Merge pull request #48 from vladhlinskiy/feature/CDAP-15550-netezza-all-data-types
CDAP-15550 Netezza db plugin enhacements: all data types support and proper test coverage
2 parents 742c868 + f3f6d59 commit 78d5594

File tree

3 files changed

+205
-152
lines changed

3 files changed

+205
-152
lines changed

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

Lines changed: 50 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.junit.ClassRule;
3737

3838
import java.math.BigDecimal;
39+
import java.math.MathContext;
3940
import java.sql.Connection;
4041
import java.sql.Date;
4142
import java.sql.Driver;
@@ -59,7 +60,7 @@ public class NetezzaPluginTestBase extends DatabasePluginTestBase {
5960

6061
protected static String connectionUrl;
6162
protected static final int YEAR;
62-
protected static final int PRECISION = 10;
63+
protected static final int PRECISION = 16;
6364
protected static final int SCALE = 6;
6465
protected static boolean tearDown = true;
6566
private static int startCount;
@@ -122,36 +123,32 @@ protected static void createTestTables(Connection conn) throws SQLException {
122123
// create a table that the action will truncate at the end of the run
123124
stmt.execute("CREATE TABLE post_action_test (x int, day varchar(10))");
124125

125-
stmt.execute("CREATE TABLE my_table (" +
126-
"ID INT NOT NULL," +
127-
"NAME VARCHAR(40) NOT NULL," +
128-
"SCORE REAL," +
129-
"GRADUATED BOOLEAN," +
130-
"NOT_IMPORTED VARCHAR(30)," +
131-
"BYTEINT_COL BYTEINT," +
132-
"SMALLINT_COL SMALLINT," +
133-
"INTEGER_COL INTEGER," +
134-
"BIGINT_COL BIGINT," +
135-
"CHAR_COL CHARACTER(40)," +
136-
"VARCHAR_COL CHARACTER VARYING(40)," +
137-
"NCHAR_COL NATIONAL CHARACTER(40)," +
138-
"DATE_COL DATE," +
139-
"TIME_COL TIME," +
140-
"TIME_WITH_TIME_ZONE_COL TIME WITH TIME ZONE," +
141-
"TIMESTAMP_COL TIMESTAMP," +
142-
"INTERVAL_COL INTERVAL," +
143-
"DOUBLE_PRECISION_COL DOUBLE PRECISION," +
144-
"NUMERIC_COL NUMERIC(" + PRECISION + "," + SCALE + ")," +
145-
"NVARCHAR_COL NATIONAL CHARACTER VARYING(40)," +
146-
"REAL_COL REAL," +
147-
"ST_GEOMETRY_COL ST_GEOMETRY(10)," +
148-
"VARBINARY_COL BINARY VARYING(10)," +
149-
"DECIMAL_COL DECIMAL(" + PRECISION + "," + SCALE + ")," +
150-
"FLOAT_COL FLOAT(6))");
151-
stmt.execute("CREATE TABLE MY_DEST_TABLE AS " +
152-
"SELECT * FROM my_table");
153-
stmt.execute("CREATE TABLE your_table AS " +
154-
"SELECT * FROM my_table");
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)");
149+
150+
stmt.execute("CREATE TABLE MY_DEST_TABLE AS SELECT * FROM my_table");
151+
stmt.execute("CREATE TABLE YOUR_TABLE AS SELECT * FROM my_table");
155152
}
156153
}
157154

@@ -162,12 +159,12 @@ protected static void prepareTestData(Connection conn) throws SQLException {
162159
conn.prepareStatement("INSERT INTO my_table " +
163160
"VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
164161
" ?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
165-
" ?, ?, ?, ?, ?)");
162+
" ?, ?)");
166163
PreparedStatement pStmt2 =
167164
conn.prepareStatement("INSERT INTO your_table " +
168165
"VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
169166
" ?, ?, ?, ?, ?, ?, ?, ?, ?, ?," +
170-
" ?, ?, ?, ?, ?)")) {
167+
" ?, ?)")) {
171168

172169
stmt.execute("insert into db_action_test values (1, '1970-01-01')");
173170
stmt.execute("insert into post_action_test values (1, '1970-01-01')");
@@ -176,36 +173,33 @@ protected static void prepareTestData(Connection conn) throws SQLException {
176173
}
177174
}
178175

179-
private static void populateData(PreparedStatement ...stmts) throws SQLException {
176+
private static void populateData(PreparedStatement... stmts) throws SQLException {
180177
// insert the same data into both tables: my_table and your_table
181178
for (PreparedStatement pStmt : stmts) {
182179
for (int i = 1; i <= 5; i++) {
183180
String name = "user" + i;
184181
pStmt.setInt(1, i);
185-
pStmt.setString(2, name);
186-
pStmt.setDouble(3, 123.45 + i);
187-
pStmt.setBoolean(4, (i % 2 == 0));
188-
pStmt.setString(5, "random" + i);
189-
pStmt.setShort(6, (short) i);
190-
pStmt.setShort(7, (short) i);
191-
pStmt.setInt(8, i);
192-
pStmt.setLong(9, (long) i);
193-
pStmt.setString(10, name);
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)));
194191
pStmt.setString(11, name);
195192
pStmt.setString(12, name);
196-
pStmt.setDate(13, new Date(CURRENT_TS));
197-
pStmt.setTime(14, new Time(CURRENT_TS));
198-
pStmt.setTime(15, new Time(CURRENT_TS));
199-
pStmt.setTimestamp(16, new Timestamp(CURRENT_TS));
200-
pStmt.setString(17, "2 year 3 month " + i + " day");
201-
pStmt.setDouble(18, 123.45 + i);
202-
pStmt.setBigDecimal(19, new BigDecimal(123.45).add(new BigDecimal(i)));
203-
pStmt.setString(20, name);
204-
pStmt.setDouble(21, 123.45 + i);
205-
pStmt.setBytes(22, name.getBytes(Charsets.UTF_8));
206-
pStmt.setBytes(23, name.getBytes(Charsets.UTF_8));
207-
pStmt.setBigDecimal(24, new BigDecimal(123.45).add(new BigDecimal(i)));
208-
pStmt.setFloat(25, (float) 123.45 + i);
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));
209203

210204
pStmt.executeUpdate();
211205
}

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

Lines changed: 102 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -19,16 +19,19 @@
1919
import com.google.common.base.Charsets;
2020
import com.google.common.collect.ImmutableMap;
2121
import com.google.common.collect.ImmutableSet;
22+
import io.cdap.cdap.api.common.Bytes;
2223
import io.cdap.cdap.api.data.format.StructuredRecord;
2324
import io.cdap.cdap.api.data.schema.Schema;
2425
import io.cdap.cdap.api.dataset.table.Table;
2526
import io.cdap.cdap.etl.api.batch.BatchSink;
2627
import io.cdap.cdap.etl.mock.batch.MockSource;
2728
import io.cdap.cdap.etl.proto.v2.ETLPlugin;
29+
import io.cdap.cdap.test.ApplicationManager;
2830
import io.cdap.cdap.test.DataSetManager;
2931
import io.cdap.plugin.common.Constants;
3032
import io.cdap.plugin.db.batch.sink.AbstractDBSink;
3133
import org.junit.Assert;
34+
import org.junit.Before;
3235
import org.junit.Test;
3336

3437
import java.math.BigDecimal;
@@ -51,12 +54,56 @@
5154
* Test for ETL using databases.
5255
*/
5356
public class NetezzaSinkTestRun extends NetezzaPluginTestBase {
57+
private static final Schema SCHEMA = Schema.recordOf(
58+
"dbRecord",
59+
Schema.Field.of("INTEGER_COL", Schema.of(Schema.Type.INT)),
60+
Schema.Field.of("BYTEINT_COL", Schema.of(Schema.Type.INT)),
61+
Schema.Field.of("SMALLINT_COL", Schema.of(Schema.Type.INT)),
62+
Schema.Field.of("BIGINT_COL", Schema.of(Schema.Type.LONG)),
63+
Schema.Field.of("REAL_COL", Schema.of(Schema.Type.FLOAT)),
64+
Schema.Field.of("REAL_FLOAT_COL", Schema.of(Schema.Type.FLOAT)),
65+
Schema.Field.of("DOUBLE_FLOAT_COL", Schema.of(Schema.Type.DOUBLE)),
66+
Schema.Field.of("DOUBLE_PRECISION_COL", Schema.of(Schema.Type.DOUBLE)),
67+
Schema.Field.of("NUMERIC_COL", Schema.decimalOf(PRECISION, SCALE)),
68+
Schema.Field.of("DECIMAL_COL", Schema.decimalOf(PRECISION, SCALE)),
69+
Schema.Field.of("CHAR_COL", Schema.of(Schema.Type.STRING)),
70+
Schema.Field.of("VARCHAR_COL", Schema.of(Schema.Type.STRING)),
71+
Schema.Field.of("NCHAR_COL", Schema.of(Schema.Type.STRING)),
72+
Schema.Field.of("NVARCHAR_COL", Schema.of(Schema.Type.STRING)),
73+
Schema.Field.of("VARBINARY_COL", Schema.of(Schema.Type.BYTES)),
74+
Schema.Field.of("ST_GEOMETRY_COL", Schema.of(Schema.Type.BYTES)),
75+
Schema.Field.of("DATE_COL", Schema.of(Schema.LogicalType.DATE)),
76+
Schema.Field.of("TIME_COL", Schema.of(Schema.LogicalType.TIME_MICROS)),
77+
Schema.Field.of("TIMETZ_COL", Schema.of(Schema.Type.STRING)),
78+
Schema.Field.of("TIMESTAMP_COL", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)),
79+
Schema.Field.of("INTERVAL_COL", Schema.of(Schema.Type.STRING)),
80+
Schema.Field.of("BOOLEAN_COL", Schema.of(Schema.Type.BOOLEAN))
81+
);
82+
83+
private static final BigDecimal NUMERIC_VALUE = new BigDecimal(3.458d, new MathContext(PRECISION)).setScale(SCALE);
84+
85+
@Before
86+
public void setup() throws Exception {
87+
try (Statement stmt = createConnection().createStatement()) {
88+
stmt.execute("DELETE FROM MY_DEST_TABLE");
89+
}
90+
}
5491

5592
@Test
56-
public void testDBSink() throws Exception {
57-
String inputDatasetName = "input-dbsinktest";
93+
public void testDBSinkWithExplicitInputSchema() throws Exception {
94+
testDBSink("testDBSinkWithExplicitInputSchema", "input-dbsinktest-explicit", true);
95+
}
96+
97+
98+
@Test
99+
public void testDBSinkWithInferredInputSchema() throws Exception {
100+
testDBSink("testDBSinkWithInferredInputSchema", "input-dbsinktest-inferred", false);
101+
}
58102

59-
ETLPlugin sourceConfig = MockSource.getPlugin(inputDatasetName);
103+
public void testDBSink(String appName, String inputDatasetName, boolean setInputSchema) throws Exception {
104+
ETLPlugin sourceConfig = (setInputSchema)
105+
? MockSource.getPlugin(inputDatasetName, SCHEMA)
106+
: MockSource.getPlugin(inputDatasetName);
60107
ETLPlugin sinkConfig = new ETLPlugin(
61108
NetezzaConstants.PLUGIN_NAME,
62109
BatchSink.PLUGIN_TYPE,
@@ -67,76 +114,82 @@ public void testDBSink() throws Exception {
67114
.build(),
68115
null);
69116

70-
deployETL(sourceConfig, sinkConfig, DATAPIPELINE_ARTIFACT, "testDBSink");
117+
ApplicationManager appManager = deployETL(sourceConfig, sinkConfig, DATAPIPELINE_ARTIFACT, appName);
71118
createInputData(inputDatasetName);
72119

120+
runETLOnce(appManager, ImmutableMap.of("logical.start.time", String.valueOf(CURRENT_TS)));
73121

74122
try (Connection conn = createConnection();
75123
Statement stmt = conn.createStatement();
76-
ResultSet resultSet = stmt.executeQuery("SELECT * FROM my_table ORDER BY ID")) {
124+
ResultSet resultSet = stmt.executeQuery("SELECT * FROM MY_DEST_TABLE ORDER BY INTEGER_COL")) {
125+
77126
Set<String> users = new HashSet<>();
78127
Assert.assertTrue(resultSet.next());
79-
users.add(resultSet.getString("NAME"));
128+
users.add(resultSet.getString("VARCHAR_COL"));
129+
130+
Assert.assertEquals(1, resultSet.getInt("INTEGER_COL"));
131+
Assert.assertEquals(1, resultSet.getInt("BYTEINT_COL"));
132+
Assert.assertEquals(1, resultSet.getInt("SMALLINT_COL"));
133+
Assert.assertEquals(1, resultSet.getLong("BIGINT_COL"));
134+
135+
Assert.assertEquals(3.451f + 1, resultSet.getFloat("REAL_COL"), 0.00001f);
136+
Assert.assertEquals(3.451f + 1, resultSet.getFloat("REAL_FLOAT_COL"), 0.00001f);
137+
Assert.assertEquals(3.451 + 1, resultSet.getFloat("DOUBLE_FLOAT_COL"), 0.000001);
138+
Assert.assertEquals(3.451 + 1, resultSet.getFloat("DOUBLE_PRECISION_COL"), 0.000001);
139+
Assert.assertEquals(NUMERIC_VALUE, resultSet.getBigDecimal("NUMERIC_COL"));
140+
141+
Assert.assertEquals("user1", resultSet.getString("CHAR_COL").trim());
142+
Assert.assertEquals("user1", resultSet.getString("VARCHAR_COL"));
143+
Assert.assertEquals("user1", resultSet.getString("NCHAR_COL").trim());
144+
Assert.assertEquals("user1", resultSet.getString("NVARCHAR_COL"));
145+
Assert.assertEquals("user1", Bytes.toString(resultSet.getBytes("VARBINARY_COL")));
146+
Assert.assertEquals("user1", Bytes.toString(resultSet.getBytes("ST_GEOMETRY_COL")));
147+
80148
Assert.assertEquals(new Date(CURRENT_TS).toString(), resultSet.getDate("DATE_COL").toString());
81149
Assert.assertEquals(new Time(CURRENT_TS).toString(), resultSet.getTime("TIME_COL").toString());
82-
Assert.assertEquals(new Timestamp(CURRENT_TS),
83-
resultSet.getTimestamp("TIMESTAMP_COL"));
150+
Assert.assertEquals("13:24:16+03", resultSet.getString("TIMETZ_COL"));
151+
Assert.assertEquals(new Timestamp(CURRENT_TS), resultSet.getTimestamp("TIMESTAMP_COL"));
152+
Assert.assertEquals("2 years 3 mons 2 days", resultSet.getString("INTERVAL_COL"));
153+
154+
Assert.assertTrue(resultSet.getBoolean("BOOLEAN_COL"));
155+
84156
Assert.assertTrue(resultSet.next());
85-
Assert.assertEquals(new BigDecimal(3.458, new MathContext(PRECISION)).setScale(SCALE),
86-
resultSet.getBigDecimal("NUMERIC_COL"));
87-
Assert.assertEquals(new BigDecimal(3.459, new MathContext(PRECISION)).setScale(SCALE),
88-
resultSet.getBigDecimal("DECIMAL_COL"));
89-
users.add(resultSet.getString("NAME"));
90-
Assert.assertEquals(ImmutableSet.of("user1", "user2"), users);
157+
users.add(resultSet.getString("VARCHAR_COL"));
91158

159+
Assert.assertEquals(ImmutableSet.of("user1", "user2"), users);
92160
}
93161
}
94162

95163
private void createInputData(String inputDatasetName) throws Exception {
96164
// add some data to the input table
97165
DataSetManager<Table> inputManager = getDataset(inputDatasetName);
98-
Schema schema = Schema.recordOf(
99-
"dbRecord",
100-
Schema.Field.of("ID", Schema.of(Schema.Type.INT)),
101-
Schema.Field.of("NAME", Schema.of(Schema.Type.STRING)),
102-
Schema.Field.of("SCORE", Schema.of(Schema.Type.FLOAT)),
103-
Schema.Field.of("GRADUATED", Schema.of(Schema.Type.BOOLEAN)),
104-
Schema.Field.of("TINY", Schema.of(Schema.Type.INT)),
105-
Schema.Field.of("SMALL", Schema.of(Schema.Type.INT)),
106-
Schema.Field.of("BIG", Schema.of(Schema.Type.LONG)),
107-
Schema.Field.of("FLOAT_COL", Schema.of(Schema.Type.FLOAT)),
108-
Schema.Field.of("REAL_COL", Schema.of(Schema.Type.FLOAT)),
109-
Schema.Field.of("NUMERIC_COL", Schema.decimalOf(PRECISION, SCALE)),
110-
Schema.Field.of("DECIMAL_COL", Schema.decimalOf(PRECISION, SCALE)),
111-
Schema.Field.of("BIT_COL", Schema.of(Schema.Type.BOOLEAN)),
112-
Schema.Field.of("DATE_COL", Schema.of(Schema.LogicalType.DATE)),
113-
Schema.Field.of("TIME_COL", Schema.of(Schema.LogicalType.TIME_MICROS)),
114-
Schema.Field.of("TIMESTAMP_COL", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)),
115-
Schema.Field.of("BINARY_COL", Schema.of(Schema.Type.BYTES)),
116-
Schema.Field.of("BLOB_COL", Schema.of(Schema.Type.BYTES))
117-
);
118166
List<StructuredRecord> inputRecords = new ArrayList<>();
119167
LocalDateTime localDateTime = new Timestamp(CURRENT_TS).toLocalDateTime();
120168
for (int i = 1; i <= 2; i++) {
121169
String name = "user" + i;
122-
inputRecords.add(StructuredRecord.builder(schema)
123-
.set("ID", i)
124-
.set("NAME", name)
125-
.set("SCORE", 3.451f)
126-
.set("GRADUATED", (i % 2 == 0))
127-
.set("TINY", i + 1)
128-
.set("SMALL", i + 2)
129-
.set("BIG", 3456987L)
130-
.set("FLOAT_COL", 3.456f)
131-
.set("REAL_COL", 3.457f)
132-
.setDecimal("NUMERIC_COL", new BigDecimal(3.458d, new MathContext(PRECISION)).setScale(SCALE))
133-
.setDecimal("DECIMAL_COL", new BigDecimal(3.459d, new MathContext(PRECISION)).setScale(SCALE))
134-
.set("BIT_COL", (i % 2 == 1))
170+
inputRecords.add(StructuredRecord.builder(SCHEMA)
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))
135187
.setDate("DATE_COL", localDateTime.toLocalDate())
136188
.setTime("TIME_COL", localDateTime.toLocalTime())
189+
.set("TIMETZ_COL", "13:24:16+03")
137190
.setTimestamp("TIMESTAMP_COL", localDateTime.atZone(ZoneId.ofOffset("UTC", ZoneOffset.UTC)))
138-
.set("BINARY_COL", name.getBytes(Charsets.UTF_8))
139-
.set("BLOB_COL", name.getBytes(Charsets.UTF_8))
191+
.set("INTERVAL_COL", "2 years 3 mons 2 days")
192+
.set("BOOLEAN_COL", true)
140193
.build());
141194
}
142195
MockSource.writeInput(inputManager, inputRecords);

0 commit comments

Comments
 (0)