Merge pull request #52 from mlozbin-cybervisiontech/feature/CDAP-15658-automated-db-integration-tests

CDAP-15658 Make the DB integration tests automated

Author: albertshau

README.md

@@ -18,4 +18,61 @@ mvn clean test \
 -DauroraPostgresql.clusterEndpoint=pginstance.cxywmbgwp60k.eu-central-1.rds.amazonaws.com -DauroraPostgresql.port=5432 \
 -DauroraPostgresql.database=cdappg -DauroraPostgresql.username=cdap -DauroraPostgresql.password=cdap
 -DjdbcDriversJars="/jdbc/drivers/jars/some.jar, "
-```
+```
+
+## Setup Local Environment
+MySQL, Postgresql, MSSQL, DB2 are using prebuild images.
+
+Oracle DB image should be build separately.
+
+Netezza requires VMware Player for running Netezza emulator.
+
+* [Install Docker Compose](https://docs.docker.com/compose/install/)
+* Build local docker images
+  * [Build Oracle DB docker image version 12.1.0.2-ee](https://github.com/oracle/docker-images/tree/master/OracleDatabase/SingleInstance)
+* Start docker environment by running commands:
+```bash
+cd docker-compose/db-plugins-env/
+docker-compose up -d
+```
+* [Install and start Netezza emulator](http://dwgeek.com/install-vmware-player-netezza-emulator.html/)
+
+### Properties
+#### MySQL
+* **mysql.host** - Server host. Default: localhost.
+* **mysql.port** - Server port. Default: 3306.
+* **mysql.database** - Server namespace for test databases. Default: mydb.
+* **mysql.username** - Server username Default: root.
+* **mysql.password** - Server password. Default: 123Qwe123.
+#### Postgresql
+* **postgresql.host** - Server host. Default: localhost.
+* **postgresql.port** - Server port. Default: 5432.
+* **postgresql.database** - Server namespace for test databases. Default: mydb.
+* **postgresql.username** - Server username Default: postgres.
+* **postgresql.password** - Server password. Default: 123Qwe123.
+#### MSSQL
+* **mssql.host** - Server host. Default: localhost.
+* **mssql.port** - Server port. Default: 1433.
+* **mssql.database** - Server namespace for test databases. Default: tempdb.
+* **mssql.username** - Server username Default: sa.
+* **mssql.password** - Server password. Default: 123Qwe123.
+#### DB2
+* **db2.host** - Server host. Default: localhost.
+* **db2.port** - Server port. Default: 50000. 
+* **db2.database** - Server namespace for test databases. Default: mydb.
+* **db2.username** - Server username Default: db2inst1.
+* **db2.password** - Server password. Default: 123Qwe123.
+#### Oracle
+* **oracle.host** - Server host. Default: localhost.
+* **oracle.port** - Server port. Default: 1521.
+* **oracle.username** - Server username Default: SYSTEM.
+* **oracle.password** - Server password. Default: 123Qwe123.
+* **oracle.database** - Server sid/database. Default: cdap.
+* **oracle.connectionType** - Server connection type (service/sid) Default: sid.
+#### Netezza
+* **netezza.host** - Server host. Default: localhost.
+* **netezza.port** - Server port. Default: 5480.
+* **netezza.database** - Server namespace for test databases. Default: mydb.
+* **netezza.username** - Server username Default: admin.
+* **netezza.password** - Server password. Default: password.
+  

database-commons/src/main/java/io/cdap/plugin/db/batch/sink/AbstractDBSink.java

@@ -350,6 +350,10 @@ protected boolean isFieldCompatible(Schema.Field field, ResultSetMetaData metada
 
     int sqlType = metadata.getColumnType(index);
 
+    return isFieldCompatible(fieldType, fieldLogicalType, sqlType);
+  }
+
+  protected boolean isFieldCompatible(Schema.Type fieldType, Schema.LogicalType fieldLogicalType, int sqlType) {
     // Handle logical types first
     if (fieldLogicalType != null) {
       switch (fieldLogicalType) {
@@ -375,7 +379,7 @@ protected boolean isFieldCompatible(Schema.Field field, ResultSetMetaData metada
           || sqlType == Types.BIT;
       case INT:
         return sqlType == Types.INTEGER
-          ||  sqlType == Types.SMALLINT
+          || sqlType == Types.SMALLINT
           || sqlType == Types.TINYINT;
       case LONG:
         return sqlType == Types.BIGINT;

database-commons/src/test/java/io/cdap/plugin/db/batch/DatabasePluginTestBase.java

@@ -137,7 +137,7 @@ protected void testDBSinkValidation(String inputDatasetName, String appName, Sch
 
   protected void writeDataForInvalidDataWriteTest(String inputDatasetName, String stringColumnName) throws Exception {
     Schema validSchema = Schema.recordOf(
-      "wrongDBRecord",
+      "validDBRecord",
       Schema.Field.of("ID", Schema.of(Schema.Type.INT)),
       Schema.Field.of(stringColumnName, Schema.of(Schema.Type.STRING))
     );

db2-plugin/src/main/java/io/cdap/plugin/db2/DB2Record.java

@@ -18,10 +18,12 @@
 
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.etl.api.validation.InvalidStageException;
 import io.cdap.plugin.db.ColumnType;
 import io.cdap.plugin.db.DBRecord;
 import io.cdap.plugin.db.SchemaReader;
 
+import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
@@ -31,7 +33,7 @@
  * Writable class for DB2 Source/Sink.
  */
 public class DB2Record extends DBRecord {
-
+  private static final int ILLEGAL_CONVERSION_ERROR_CODE = -4474;
   /**
    * Used in map-reduce. Do not remove.
    */
@@ -58,6 +60,21 @@ protected void handleField(ResultSet resultSet, StructuredRecord.Builder recordB
     }
   }
 
+  @Override
+  public void write(PreparedStatement stmt) throws SQLException {
+    // DB2 driver throws SQLException if data conversation fails, but SQLException is skipped.
+    // So we need to throw another exception to fail pipeline in this case.
+    try {
+      super.write(stmt);
+    } catch (SQLException e) {
+      if (e.getErrorCode() == ILLEGAL_CONVERSION_ERROR_CODE) {
+        throw new InvalidStageException(e.getMessage(), e);
+      } else {
+        throw e;
+      }
+    }
+  }
+
   private void handleSpecificType(ResultSet resultSet,
                                   StructuredRecord.Builder recordBuilder,
                                   Schema.Field field, int columnIndex) throws SQLException {

db2-plugin/src/test/java/io/cdap/plugin/db2/Db2PluginTestBase.java

@@ -74,11 +74,11 @@ public class Db2PluginTestBase extends DatabasePluginTestBase {
   }
 
   protected static final Map<String, String> BASE_PROPS = ImmutableMap.<String, String>builder()
-    .put(ConnectionConfig.HOST, System.getProperty("db2.host"))
-    .put(ConnectionConfig.PORT, System.getProperty("db2.port"))
-    .put(ConnectionConfig.DATABASE, System.getProperty("db2.database"))
-    .put(ConnectionConfig.USER, System.getProperty("db2.username"))
-    .put(ConnectionConfig.PASSWORD, System.getProperty("db2.password"))
+    .put(ConnectionConfig.HOST, System.getProperty("db2.host", "localhost"))
+    .put(ConnectionConfig.PORT, System.getProperty("db2.port", "50000"))
+    .put(ConnectionConfig.DATABASE, System.getProperty("db2.database", "mydb"))
+    .put(ConnectionConfig.USER, System.getProperty("db2.username", "db2inst1"))
+    .put(ConnectionConfig.PASSWORD, System.getProperty("db2.password", "123Qwe123"))
     .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME)
     .build();
 

db2-plugin/src/test/java/io/cdap/plugin/db2/Db2SinkTestRun.java

@@ -98,7 +98,7 @@ public void testDBSinkWithInvalidFieldLogicalType() throws Exception {
 
   @Test
   public void testDBSinkWithDBSchemaAndInvalidData() throws Exception {
-    String stringColumnName = "CHAR_COL";
+    String stringColumnName = "BINARY_COL";
     startPipelineAndWriteInvalidData(stringColumnName, getSinkConfig(), DATAPIPELINE_ARTIFACT);
     try (Connection conn = createConnection();
          Statement stmt = conn.createStatement();
@@ -169,29 +169,29 @@ public void testDBSink(String appName, String inputDatasetName, boolean setInput
   }
 
   @Override
-  protected void writeDataForInvalidDataWriteTest(String inputDatasetName, String stringColumnName) throws Exception {
+  protected void writeDataForInvalidDataWriteTest(String inputDatasetName, String columnName) throws Exception {
     Schema validSchema = Schema.recordOf(
-      "wrongDBRecord",
-      Schema.Field.of(stringColumnName, Schema.of(Schema.Type.STRING))
+      "validDBRecord",
+      Schema.Field.of(columnName, Schema.of(Schema.Type.BYTES))
     );
 
     Schema invalidSchema = Schema.recordOf(
       "wrongDBRecord",
-      Schema.Field.of(stringColumnName, Schema.of(Schema.Type.INT))
+      Schema.Field.of(columnName, Schema.of(Schema.Type.INT))
     );
 
     // add some data to the input table
     DataSetManager<Table> inputManager = getDataset(inputDatasetName);
 
     List<StructuredRecord> inputRecords = new ArrayList<>();
     inputRecords.add(StructuredRecord.builder(validSchema)
-                       .set(stringColumnName, "user1")
+                       .set(columnName, "user1".getBytes())
                        .build());
     inputRecords.add(StructuredRecord.builder(invalidSchema)
-                       .set(stringColumnName, 1)
+                       .set(columnName, 1)
                        .build());
     inputRecords.add(StructuredRecord.builder(validSchema)
-                       .set(stringColumnName, "user3")
+                       .set(columnName, "user3".getBytes())
                        .build());
     MockSource.writeInput(inputManager, inputRecords);
   }

docker-compose/db-plugins-env/docker-compose.yml

@@ -0,0 +1,59 @@
+# Copyright © 2019 Cask Data, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+
+version: '3'
+services:
+  mysql:
+    image: mysql:8.0.16
+    ports:
+      - 3306:3306
+    environment:
+      - MYSQL_ROOT_PASSWORD=123Qwe123
+      - MYSQL_DATABASE=mydb
+
+  postgres:
+    image: postgres:11.3-alpine
+    ports:
+      - 5432:5432
+    environment:
+      - POSTGRES_PASSWORD=123Qwe123
+      - POSTGRES_DB=mydb
+
+  mssql:
+    image: mcr.microsoft.com/mssql/server:2017-CU14-ubuntu
+    ports:
+      - 1433:1433
+    environment:
+      - SA_PASSWORD=123Qwe123
+      - MSSQL_DB=mydb
+      - ACCEPT_EULA=Y
+
+  db2:
+    image: ibmcom/db2:11.5.0.0
+    ports:
+      - 50000:50000
+    environment:
+      - DB2INST1_PASSWORD=123Qwe123
+      - DBNAME=mydb
+      - LICENSE=accept
+    privileged: true
+
+  oracle:
+    image: oracle/database:12.1.0.2-ee
+    ports:
+      - 1521:1521
+    environment:
+      - ORACLE_SID=cdap
+      - ORACLE_PDB=mydb
+      - ORACLE_PWD=123Qwe123

mssql-plugin/src/test/java/io/cdap/plugin/mssql/SqlServerPluginTestBase.java

@@ -82,11 +82,11 @@ public class SqlServerPluginTestBase extends DatabasePluginTestBase {
   public static final TestConfiguration CONFIG = new TestConfiguration("explore.enabled", false);
 
   protected static final Map<String, String> BASE_PROPS = ImmutableMap.<String, String>builder()
-    .put(ConnectionConfig.HOST, System.getProperty("mssql.host"))
-    .put(ConnectionConfig.PORT, System.getProperty("mssql.port"))
-    .put(ConnectionConfig.DATABASE, System.getProperty("mssql.database"))
-    .put(ConnectionConfig.USER, System.getProperty("mssql.username"))
-    .put(ConnectionConfig.PASSWORD, System.getProperty("mssql.password"))
+    .put(ConnectionConfig.HOST, System.getProperty("mssql.host", "localhost"))
+    .put(ConnectionConfig.PORT, System.getProperty("mssql.port", "1433"))
+    .put(ConnectionConfig.DATABASE, System.getProperty("mssql.database", "tempdb"))
+    .put(ConnectionConfig.USER, System.getProperty("mssql.username", "sa"))
+    .put(ConnectionConfig.PASSWORD, System.getProperty("mssql.password", "123Qwe123"))
     .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME)
     .build();
 

mysql-plugin/src/test/java/io/cdap/plugin/mysql/MysqlPluginTestBase.java

@@ -78,11 +78,11 @@ public class MysqlPluginTestBase extends DatabasePluginTestBase {
   }
 
   protected static final Map<String, String> BASE_PROPS = ImmutableMap.<String, String>builder()
-    .put(ConnectionConfig.HOST, System.getProperty("mysql.host"))
-    .put(ConnectionConfig.PORT, System.getProperty("mysql.port"))
-    .put(ConnectionConfig.DATABASE, System.getProperty("mysql.database"))
-    .put(ConnectionConfig.USER, System.getProperty("mysql.username"))
-    .put(ConnectionConfig.PASSWORD, System.getProperty("mysql.password"))
+    .put(ConnectionConfig.HOST, System.getProperty("mysql.host", "localhost"))
+    .put(ConnectionConfig.PORT, System.getProperty("mysql.port", "3306"))
+    .put(ConnectionConfig.DATABASE, System.getProperty("mysql.database", "mydb"))
+    .put(ConnectionConfig.USER, System.getProperty("mysql.username", "root"))
+    .put(ConnectionConfig.PASSWORD, System.getProperty("mysql.password", "123Qwe123"))
     .put(ConnectionConfig.JDBC_PLUGIN_NAME, JDBC_DRIVER_NAME)
     .build();
 

netezza-plugin/src/main/java/io/cdap/plugin/netezza/NetezzaDBRecord.java

@@ -33,7 +33,7 @@
  */
 public class NetezzaDBRecord extends DBRecord {
 
-  private static final int INTERVAL = 101;
+  public static final int INTERVAL = 101;
 
   private static final Set<Integer> netezzaTypes = ImmutableSet.of(
     Types.VARBINARY,