Relational directives

wrangler-api/src/main/java/io/cdap/wrangler/api/RelationalDirective.java

@@ -0,0 +1,30 @@
+/*
+ *  Copyright © 2017-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.
+ */
+
+package io.cdap.wrangler.api;
+
+/**
+ * Directive interface which supports Relational transformations
+ */
+public interface RelationalDirective extends Directive {
+
+  /**
+   * returns sql expression
+   * @return expression
+   */
+  String getSQL();
+
+}

wrangler-core/src/main/java/io/cdap/directives/column/Drop.java

@@ -24,6 +24,7 @@
 import io.cdap.wrangler.api.DirectiveExecutionException;
 import io.cdap.wrangler.api.DirectiveParseException;
 import io.cdap.wrangler.api.ExecutorContext;
+import io.cdap.wrangler.api.RelationalDirective;
 import io.cdap.wrangler.api.Row;
 import io.cdap.wrangler.api.annotations.Categories;
 import io.cdap.wrangler.api.lineage.Lineage;
@@ -43,7 +44,7 @@
 @Name(Drop.NAME)
 @Categories(categories = { "column"})
 @Description("Drop one or more columns.")
-public class Drop implements Directive, Lineage {
+public class Drop implements RelationalDirective, Lineage {
   public static final String NAME = "drop";
 
   // Columns to be dropped.
@@ -88,4 +89,14 @@ public Mutation lineage() {
       .drop(Many.of(columns))
       .build();
   }
+
+  @Override
+  public String getSQL() {
+    String sql = "DROP COLUMN ";
+    for (String col : columns) {
+      sql += col + ",";
+    }
+    sql = sql.substring(0, sql.length() - 1);
+    return sql;
+  }
 }

wrangler-core/src/main/java/io/cdap/wrangler/registry/SystemDirectiveRegistry.java

@@ -20,6 +20,7 @@
 import io.cdap.cdap.api.artifact.ArtifactSummary;
 import io.cdap.wrangler.api.Directive;
 import io.cdap.wrangler.api.DirectiveLoadException;
+import io.cdap.wrangler.api.RelationalDirective;
 import org.reflections.Reflections;
 
 import java.util.ArrayList;
@@ -84,6 +85,7 @@ public SystemDirectiveRegistry(List<String> namespaces) throws DirectiveLoadExce
       try {
         Reflections reflections = new Reflections(namespace);
         Set<Class<? extends Directive>> system = reflections.getSubTypesOf(Directive.class);
+        system.addAll(reflections.getSubTypesOf(RelationalDirective.class));
         for (Class<? extends Directive> directive : system) {
           DirectiveInfo info = DirectiveInfo.fromSystem(directive);
           registry.put(info.name(), info);

wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java

@@ -44,6 +44,7 @@
 import io.cdap.cdap.etl.api.relational.StringExpressionFactoryType;
 import io.cdap.cdap.features.Feature;
 import io.cdap.directives.aggregates.DefaultTransientStore;
+import io.cdap.directives.column.Drop;
 import io.cdap.wrangler.api.CompileException;
 import io.cdap.wrangler.api.CompileStatus;
 import io.cdap.wrangler.api.Compiler;
@@ -53,9 +54,11 @@
 import io.cdap.wrangler.api.EntityCountMetric;
 import io.cdap.wrangler.api.ErrorRecord;
 import io.cdap.wrangler.api.ExecutorContext;
+import io.cdap.wrangler.api.RecipeException;
 import io.cdap.wrangler.api.RecipeParser;
 import io.cdap.wrangler.api.RecipePipeline;
 import io.cdap.wrangler.api.RecipeSymbol;
+import io.cdap.wrangler.api.RelationalDirective;
 import io.cdap.wrangler.api.Row;
 import io.cdap.wrangler.api.TokenGroup;
 import io.cdap.wrangler.api.TransientStore;
@@ -190,7 +193,7 @@ public void configurePipeline(PipelineConfigurer configurer) {
           if (!config.containsMacro(Config.NAME_PRECONDITION_SQL)) {
             validatePrecondition(config.getPreconditionSQL(), true, collector);
           }
-          validateSQLModeDirectives(collector);
+          //validateSQLModeDirectives(collector);
         } else {
           if (!config.containsMacro(Config.NAME_PRECONDITION)) {
             validatePrecondition(config.getPreconditionJEXL(), false, collector);
@@ -569,23 +572,66 @@ private RecipeParser getRecipeParser(StageContext context)
 
   @Override
   public Relation transform(RelationalTranformContext relationalTranformContext, Relation relation) {
-    if (PRECONDITION_LANGUAGE_SQL.equalsIgnoreCase(config.getPreconditionLanguage())
-            && checkPreconditionNotEmpty(true)) {
+    if (!(PRECONDITION_LANGUAGE_SQL.equalsIgnoreCase(config.getPreconditionLanguage())
+            && checkPreconditionNotEmpty(true))) {
+      return new InvalidRelation("Plugin is not configured for relational transformation");
+    }
 
-      if (!Feature.WRANGLER_PRECONDITION_SQL.isEnabled(relationalTranformContext)) {
-        throw new RuntimeException("SQL Precondition feature is not available");
-      }
+    if (!Feature.WRANGLER_PRECONDITION_SQL.isEnabled(relationalTranformContext)) {
+      throw new RuntimeException("SQL Precondition feature is not available");
+    }
 
-      Optional<ExpressionFactory<String>> expressionFactory = getExpressionFactory(relationalTranformContext);
-      if (!expressionFactory.isPresent()) {
-        return new InvalidRelation("Cannot find an Expression Factory");
-      }
+    Optional<ExpressionFactory<String>> expressionFactory = getExpressionFactory(relationalTranformContext);
+    if (!expressionFactory.isPresent()) {
+      return new InvalidRelation("Cannot find an Expression Factory");
+    }
+
+    Expression filterExpression = expressionFactory.get().compile(config.getPreconditionSQL());
+    Relation filteredRelation = relation.filter(filterExpression);
+
+    ExpressionFactory<String> expFactory = expressionFactory.get();
+
+    String recipe = config.getDirectives();
+
+    registry = SystemDirectiveRegistry.INSTANCE;
+    try {
+      registry.reload("default");
+    } catch (DirectiveLoadException e) {
+      throw new RuntimeException(e);
+    }
 
-      Expression filterExpression = expressionFactory.get().compile(config.getPreconditionSQL());
-      return relation.filter(filterExpression);
+    List<Directive> directives = null;
+    try {
+      GrammarBasedParser parser = new GrammarBasedParser("default", new MigrateToV2(recipe).migrate(), registry);
+      directives = parser.parse();
+    } catch (DirectiveParseException e) {
+      throw new RuntimeException(e);
+    } catch (RecipeException e) {
+      throw new RuntimeException(e);
+    }
+
+    for (Directive directive : directives) {
+      // Expression exp = expFactory.compile(sql);
+      if (!(directive instanceof RelationalDirective)) {
+        throw new RuntimeException("Directive is not relational Directive");
+      }
+      // currently supporting only drop column
+      // SQL will be returned as "DROP COLUMN col1, col2"
+      String sql = ((RelationalDirective) directive).getSQL();
+      List<String> cols = getColumnsOfDropSQL(sql);
+      for (String col : cols) {
+        filteredRelation = filteredRelation.dropColumn(col);
+      }
     }
+    return filteredRelation;
+  }
 
-    return new InvalidRelation("Plugin is not configured for relational transformation");
+  private List<String> getColumnsOfDropSQL(String sql) {
+    List<String> cols = new ArrayList<>();
+    for (String col : sql.split(" ")[2].split(",")) {
+      cols.add(col.trim());
+    }
+    return cols;
   }
 
   private Optional<ExpressionFactory<String>> getExpressionFactory(RelationalTranformContext ctx) {
@@ -642,6 +688,7 @@ public static class Config extends PluginConfig {
     static final String NAME_PRECONDITION_LANGUAGE = "expressionLanguage";
     static final String NAME_FIELD = "field";
     static final String NAME_DIRECTIVES = "directives";
+    static final String NAME_RELATIONAL_DIRECTIVES = "relationalDirectives";
     static final String NAME_UDD = "udd";
     static final String NAME_SCHEMA = "schema";
     static final String NAME_ON_ERROR = "on-error";
@@ -672,6 +719,7 @@ public static class Config extends PluginConfig {
     @Nullable
     private String directives;
 
+
     @Name(NAME_UDD)
     @Description("List of User Defined Directives (UDD) that have to be loaded.")
     @Nullable
@@ -694,7 +742,7 @@ public static class Config extends PluginConfig {
     private final String onError;
 
     public Config(String preconditionLanguage, String precondition, String directives, String udds,
-                  String field, String schema, String onError) {
+                  String field, String schema, String onError, String relationalDirectives) {
       this.preconditionLanguage = preconditionLanguage;
       this.precondition = precondition;
       this.directives = directives;

wrangler-transform/widgets/Wrangler-transform.json

@@ -73,6 +73,24 @@
         }
       ]
     },
+    {
+      "label": "RelationalDirectives",
+      "properties": [
+        {
+          "widget-type": "wrangler-relational-directives",
+          "label": "RelationalRecipe",
+          "name": "RelationalDirectives",
+          "widget-attributes" : {
+            "placeholder" : "#pragma load-directives my-directive; my-directive :body;"
+          }
+        },
+        {
+          "widget-type": "csv",
+          "label": "User Defined SQL(UDS)",
+          "name": "uds"
+        }
+      ]
+    },
     {
       "label" : "Error Handling",
       "properties" : [