Merge pull request #58 from mlozbin-cybervisiontech/feature/PLUGIN-8_memsql-plugin

PLUGIN-8 MemSQL database Source and Sink plugins

Author: albertshau

README.md

@@ -13,20 +13,43 @@ mvn clean test \
 ```
 Notice that you must change properties for Aurora MySQL and Aurora Postgresql to real before running tests.
 ## Setup Local Environment
-MySQL, Postgresql, MSSQL, DB2 are using prebuild images.
+MySQL, Postgresql, MSSQL, DB2, MemSQL are using prebuild images.
 
 Oracle DB image should be build separately.
 
+MemSQL image should be configure after start.
+
 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:
+* Enter the folder with docker-compose file:
 ```bash
 cd docker-compose/db-plugins-env/
+```
+* Export your license key for MemSQL to environment variable:
+```bash
+export MEMSQL_LICENSE_KEY=YOUR_LICENSE_KEY
+```
+* Initialize Memsql container:
+```bash
+docker-compose up memsql
+```
+* Start docker environment by running commands:
+```bash
 docker-compose up -d
 ```
+* Connect to MemSQL Studio at [http://localhost:8888](http://localhost:8888) 
+The default Username is root and Password should be left blank.
+* Create `mydb` database in MemSQL Studio
+```sql
+create database mydb
+```
+* Set password for `root` user in MemSQL Studio
+```sql
+grant all on *.* to 'root'@'%' identified by 'root' with grant option;
+```
 * [Install and start Netezza emulator](http://dwgeek.com/install-vmware-player-netezza-emulator.html/)
 * Create database `mydb` in Netezza emulator
 
@@ -68,6 +91,12 @@ docker-compose up -d
 * **netezza.database** - Server namespace for test databases. Default: mydb.
 * **netezza.username** - Server username. Default: admin.
 * **netezza.password** - Server password. Default: password.
+#### MemSQL
+* **memsql.host** - Server host. Default: localhost.
+* **memsql.port** - Server port. Default: 3308.
+* **memsql.database** - Server namespace for test databases. Default: mydb.
+* **memsql.username** - Server username. Default: root.
+* **memsql.password** - Server password. Default: root.
 #### Aurora MySQL
 * **auroraMysql.clusterEndpoint** - Cluster endpoint.
 * **auroraMysql.port** - Server port.

database-commons/src/main/java/io/cdap/plugin/db/batch/source/AbstractDBSource.java

@@ -32,8 +32,6 @@
 import io.cdap.cdap.etl.api.StageConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
 import io.cdap.cdap.etl.api.batch.BatchSourceContext;
-import io.cdap.cdap.etl.api.validation.InvalidConfigPropertyException;
-import io.cdap.cdap.etl.api.validation.InvalidStageException;
 import io.cdap.cdap.internal.io.SchemaTypeAdapter;
 import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.common.ReferenceBatchSource;
@@ -259,7 +257,8 @@ public void prepareRun(BatchSourceContext context) throws Exception {
 
     Schema schemaFromDB = loadSchemaFromDB(driverClass);
     if (sourceConfig.schema != null) {
-      sourceConfig.validateSchema(schemaFromDB);
+      sourceConfig.validateSchema(schemaFromDB, collector);
+      collector.getOrThrowException();
       connectionConfigAccessor.setSchema(sourceConfig.schema);
     } else {
       String schemaStr = SCHEMA_TYPE_ADAPTER.toJson(schemaFromDB);
@@ -394,37 +393,43 @@ private void validate(FailureCollector collector) {
       }
     }
 
-    private void validateSchema(Schema actualSchema) {
-      validateSchema(actualSchema, getSchema());
+    protected void validateSchema(Schema actualSchema, FailureCollector collector) {
+      validateSchema(actualSchema, getSchema(), collector);
     }
 
     @VisibleForTesting
-    static void validateSchema(Schema actualSchema, Schema configSchema) {
+    static void validateSchema(Schema actualSchema, Schema configSchema, FailureCollector collector) {
       if (configSchema == null) {
-
-        throw new InvalidConfigPropertyException("Schema should not be null or empty", SCHEMA);
+        collector.addFailure("Schema should not be null or empty.", null)
+          .withConfigProperty(SCHEMA);
+        return;
       }
+
       for (Schema.Field field : configSchema.getFields()) {
         Schema.Field actualField = actualSchema.getField(field.getName());
         if (actualField == null) {
-          throw new InvalidConfigPropertyException(String.format("Schema field '%s' is not present in actual record",
-                                                                 field.getName()), SCHEMA);
+          collector.addFailure(
+            String.format("Schema field '%s' is not present in actual record", field.getName()), null)
+            .withOutputSchemaField(field.getName());
+          continue;
         }
+
         Schema actualFieldSchema = actualField.getSchema().isNullable() ?
           actualField.getSchema().getNonNullable() : actualField.getSchema();
         Schema expectedFieldSchema = field.getSchema().isNullable() ?
           field.getSchema().getNonNullable() : field.getSchema();
 
         if (!actualFieldSchema.equals(expectedFieldSchema)) {
-          throw new IllegalArgumentException(
+          collector.addFailure(
             String.format("Schema field '%s' has type '%s' but found '%s' in input record",
-                          field.getName(), expectedFieldSchema.getType(), actualFieldSchema.getType()));
+                          field.getName(), expectedFieldSchema.getType(), actualFieldSchema.getType()), null)
+            .withOutputSchemaField(field.getName());
         }
       }
     }
 
     @Nullable
-    private Schema getSchema() {
+    protected Schema getSchema() {
       try {
         return Strings.isNullOrEmpty(schema) ? null : Schema.parseJson(schema);
       } catch (IOException e) {

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

@@ -17,14 +17,21 @@
 package io.cdap.plugin.db.batch.source;
 
 import io.cdap.cdap.api.data.schema.Schema;
-import io.cdap.cdap.etl.api.validation.InvalidConfigPropertyException;
+import io.cdap.cdap.etl.api.validation.CauseAttributes;
+import io.cdap.cdap.etl.api.validation.ValidationFailure;
+import io.cdap.cdap.etl.mock.validation.MockFailureCollector;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.List;
+import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
+
 /**
  * Test class for source schema validation.
  */
 public class AbstractDBSourceTest {
+  private static final String MOCK_STAGE = "mockStage";
   private static final Schema SCHEMA = Schema.recordOf(
     "schema",
     Schema.Field.of("id", Schema.nullableOf(Schema.of(Schema.Type.INT))),
@@ -39,7 +46,9 @@ public class AbstractDBSourceTest {
 
   @Test
   public void testValidateSourceSchemaCorrectSchema() {
-    AbstractDBSource.DBSourceConfig.validateSchema(SCHEMA, SCHEMA);
+    MockFailureCollector collector = new MockFailureCollector(MOCK_STAGE);
+    AbstractDBSource.DBSourceConfig.validateSchema(SCHEMA, SCHEMA, collector);
+    Assert.assertEquals(0, collector.getValidationFailures().size());
   }
 
   @Test
@@ -55,12 +64,9 @@ public void testValidateSourceSchemaMismatchFields() {
       Schema.Field.of("double_column", Schema.nullableOf(Schema.of(Schema.Type.DOUBLE)))
     );
 
-    try {
-      AbstractDBSource.DBSourceConfig.validateSchema(actualSchema, SCHEMA);
-      Assert.fail(String.format("Expected to throw %s", InvalidConfigPropertyException.class.getName()));
-    } catch (InvalidConfigPropertyException e) {
-      Assert.assertEquals(AbstractDBSource.DBSourceConfig.SCHEMA, e.getProperty());
-    }
+    MockFailureCollector collector = new MockFailureCollector(MOCK_STAGE);
+    AbstractDBSource.DBSourceConfig.validateSchema(actualSchema, SCHEMA, collector);
+    assertPropertyValidationFailed(collector, "boolean_column");
   }
 
   @Test
@@ -77,12 +83,26 @@ public void testValidateSourceSchemaInvalidFieldType() {
       Schema.Field.of("boolean_column", Schema.nullableOf(Schema.of(Schema.Type.INT)))
     );
 
-    try {
-      AbstractDBSource.DBSourceConfig.validateSchema(actualSchema, SCHEMA);
-      Assert.fail(String.format("Expected to throw %s", IllegalArgumentException.class.getName()));
-    } catch (IllegalArgumentException e) {
-      String errorMessage = "Schema field 'boolean_column' has type 'BOOLEAN' but found 'INT' in input record";
-      Assert.assertEquals(errorMessage, e.getMessage());
-    }
+    MockFailureCollector collector = new MockFailureCollector(MOCK_STAGE);
+    AbstractDBSource.DBSourceConfig.validateSchema(actualSchema, SCHEMA, collector);
+    assertPropertyValidationFailed(collector, "boolean_column");
+  }
+
+  private static void assertPropertyValidationFailed(MockFailureCollector failureCollector, String paramName) {
+    List<ValidationFailure> failureList = failureCollector.getValidationFailures();
+    Assert.assertEquals(1, failureList.size());
+    ValidationFailure failure = failureList.get(0);
+    List<ValidationFailure.Cause> causeList = getCauses(failure, CauseAttributes.OUTPUT_SCHEMA_FIELD);
+    Assert.assertEquals(1, causeList.size());
+    ValidationFailure.Cause cause = causeList.get(0);
+    Assert.assertEquals(paramName, cause.getAttribute(CauseAttributes.OUTPUT_SCHEMA_FIELD));
+  }
+
+  @Nonnull
+  private static List<ValidationFailure.Cause> getCauses(ValidationFailure failure, String stacktrace) {
+    return failure.getCauses()
+      .stream()
+      .filter(cause -> cause.getAttribute(stacktrace) != null)
+      .collect(Collectors.toList());
   }
 }

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

@@ -56,4 +56,11 @@ services:
     environment:
       - ORACLE_SID=cdap
       - ORACLE_PDB=mydb
-      - ORACLE_PWD=123Qwe123
+      - ORACLE_PWD=123Qwe123
+  memsql:
+    image: memsql/cluster-in-a-box:centos-6.8.10-a53e479edc-1.9.0-1.3.0
+    ports:
+      - 3308:3306
+      - 8888:8080
+    environment:
+      - LICENSE_KEY=${MEMSQL_LICENSE_KEY}

memsql-plugin/docs/Memsql-action.md

@@ -0,0 +1,67 @@
+# MemSQL Action
+
+
+Description
+-----------
+Action that runs a MemSQL command.
+
+
+Use Case
+--------
+The action can be used whenever you want to run a MemSQL command before or after a data pipeline.
+For example, you may want to run a sql update command on a database before the pipeline source pulls data from tables.
+
+
+Properties
+----------
+**Driver Name:** Name of the JDBC driver to use.
+
+**Database Command:** Database command to execute.
+
+**Host:** Host that MemSQL is running on.
+
+**Port:** Port that MemSQL is running on.
+
+**Database:** MemSQL database name.
+
+**Username:** User identity for connecting to the specified database.
+
+**Password:** Password to use to connect to the specified database.
+
+**Connection Arguments:** A list of arbitrary string key/value pairs as connection arguments. These arguments
+will be passed to the JDBC driver as connection arguments for JDBC drivers that may need additional configurations.
+
+**Auto Reconnect:** Should the driver try to re-establish stale and/or dead connections.
+
+**Use SSL:** Turns on SSL encryption. The connection will fail if SSL is not available.
+
+**Keystore URL:** URL to the client certificate KeyStore (if not specified, use defaults). Must be accessible at the
+same location on host where CDAP Master is running and all hosts on which at least one HDFS, MapReduce, or YARN daemon
+role is running.
+
+**Keystore Password:** Password for the client certificates KeyStore.
+
+**Truststore URL:** URL to the trusted root certificate KeyStore (if not specified, use defaults). Must be accessible at
+the same location on host where CDAP Master is running and all hosts on which at least one HDFS, MapReduce, or YARN
+daemon role is running.
+
+**Truststore Password:** Password for the trusted root certificates KeyStore
+
+**Use Compression:** Use zlib compression when communicating with the server. Select this option for WAN
+connections.
+
+**Use ANSI Quotes:** Treats " as an identifier quote character and not as a string quote character.
+
+
+Example
+-------
+Suppose you want to execute a query against a MemSQL database named "prod" that is running on "localhost" 
+port 3306, then configure the plugin with:
+
+```
+Driver Name: "mariadb"
+Database Command: "UPDATE table_name SET price = 20 WHERE ID = 6"
+Host: "localhost"
+Port: 3306
+Database: "prod"
+```