diff --git a/wrangler-api/src/main/java/io/cdap/wrangler/api/NullHandlingException.java b/wrangler-api/src/main/java/io/cdap/wrangler/api/NullHandlingException.java deleted file mode 100644 index 0d13c6711..000000000 --- a/wrangler-api/src/main/java/io/cdap/wrangler/api/NullHandlingException.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright © 2016-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; - -/** - * A Null Handling specific exception used for communicating issues with Null Handling in a column. - */ -public class NullHandlingException extends Exception { - public NullHandlingException(Exception e) { - super(e); - } - - public NullHandlingException(String message) { - super(message); - } - -} diff --git a/wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java b/wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java index a09301f5a..03673fa38 100644 --- a/wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java +++ b/wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java @@ -30,19 +30,18 @@ import io.cdap.wrangler.api.ReportErrorAndProceed; import io.cdap.wrangler.api.Row; import io.cdap.wrangler.api.TransientVariableScope; -import io.cdap.wrangler.proto.workspace.v2.Workspace.UserDefinedAction; +import io.cdap.wrangler.proto.workspace.v2.UserDefinedAction; import io.cdap.wrangler.schema.DirectiveOutputSchemaGenerator; import io.cdap.wrangler.schema.DirectiveSchemaResolutionContext; import io.cdap.wrangler.schema.TransientStoreKeys; import io.cdap.wrangler.utils.RecordConvertor; import io.cdap.wrangler.utils.RecordConvertorException; import io.cdap.wrangler.utils.SchemaConverter; +import java.util.HashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.util.ArrayList; -import java.util.HashMap; -import java.util.Iterator; +import java.util.Map; import java.util.List; import javax.annotation.Nullable; @@ -59,13 +58,13 @@ public final class RecipePipelineExecutor implements RecipePipeline directives; - private HashMap nullabilityMap; + private final Map nullabilityMap; public RecipePipelineExecutor(RecipeParser recipeParser, @Nullable ExecutorContext context, - HashMap nullabilityMap) { + Map nullabilityMap) { this.context = context; this.recipeParser = recipeParser; - this.nullabilityMap = nullabilityMap; + this.nullabilityMap = new HashMap<>(nullabilityMap); } /** diff --git a/wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java b/wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java index 5507f0770..3e30d47db 100644 --- a/wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java +++ b/wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java @@ -40,6 +40,7 @@ import io.cdap.wrangler.registry.CompositeDirectiveRegistry; import io.cdap.wrangler.registry.SystemDirectiveRegistry; import io.cdap.wrangler.schema.TransientStoreKeys; +import java.util.Collections; import org.junit.Assert; import java.util.Iterator; @@ -89,7 +90,7 @@ public static List execute(String[] recipe, List rows, ExecutorContext String migrate = new MigrateToV2(recipe).migrate(); RecipeParser parser = new GrammarBasedParser(Contexts.SYSTEM, migrate, registry); - return new RecipePipelineExecutor(parser, context, null).execute(rows); + return new RecipePipelineExecutor(parser, context, Collections.emptyMap()).execute(rows); } /** diff --git a/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/DirectiveExecutionRequest.java b/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/DirectiveExecutionRequest.java index ffd96e946..b3105f057 100644 --- a/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/DirectiveExecutionRequest.java +++ b/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/DirectiveExecutionRequest.java @@ -17,10 +17,11 @@ package io.cdap.wrangler.proto.workspace.v2; -import io.cdap.wrangler.proto.workspace.v2.Workspace.UserDefinedAction; +import io.cdap.wrangler.proto.workspace.v2.UserDefinedAction; import java.util.Collections; import java.util.HashMap; import java.util.List; +import java.util.Map; /** * Directive execution request for v2 endpoint @@ -28,11 +29,11 @@ public class DirectiveExecutionRequest { private final List directives; private final int limit; - private final HashMap nullabilityMap; + private final Map nullabilityMap; public DirectiveExecutionRequest(List directives, int limit, - HashMap nullabilityMap) { + Map nullabilityMap) { this.directives = directives; this.limit = limit; this.nullabilityMap = nullabilityMap; @@ -46,7 +47,8 @@ public List getDirectives() { return directives == null ? Collections.emptyList() : directives; } - public HashMap getNullabilityMap() { - return nullabilityMap; + public Map getNullabilityMap() { + + return nullabilityMap == null ? Collections.emptyMap() : nullabilityMap; } } diff --git a/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/UserDefinedAction.java b/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/UserDefinedAction.java new file mode 100644 index 000000000..99bddc175 --- /dev/null +++ b/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/UserDefinedAction.java @@ -0,0 +1,10 @@ +package io.cdap.wrangler.proto.workspace.v2; + +/** + * UserDefinedAction enum. + */ +public enum UserDefinedAction { + FILTER, + SEND_TO_ERROR_COLLECTOR, + ERROR, +} diff --git a/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/Workspace.java b/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/Workspace.java index 33b422e53..abcb63fde 100644 --- a/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/Workspace.java +++ b/wrangler-proto/src/main/java/io/cdap/wrangler/proto/workspace/v2/Workspace.java @@ -20,8 +20,10 @@ import com.google.gson.JsonObject; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import javax.annotation.Nullable; @@ -39,11 +41,11 @@ public class Workspace { // this is for insights page in UI private final JsonObject insights; - private HashMap nullabilityMap; + private final Map nullabilityMap; private Workspace(String workspaceName, String workspaceId, List directives, long createdTimeMillis, long updatedTimeMillis, @Nullable SampleSpec sampleSpec, - JsonObject insights, HashMap nullabilityMap) { + JsonObject insights, Map nullabilityMap) { this.workspaceName = workspaceName; this.workspaceId = workspaceId; this.directives = directives; @@ -51,8 +53,7 @@ private Workspace(String workspaceName, String workspaceId, List directi this.updatedTimeMillis = updatedTimeMillis; this.sampleSpec = sampleSpec; this.insights = insights; - this.nullabilityMap = nullabilityMap == null || nullabilityMap.isEmpty() ? - new HashMap<>() : nullabilityMap; + this.nullabilityMap = Collections.unmodifiableMap(new HashMap(nullabilityMap)); } public String getWorkspaceName() { @@ -84,15 +85,10 @@ public JsonObject getInsights() { return insights; } - public HashMap getNullabilityMap() { + public Map getNullabilityMap() { return nullabilityMap; } - public void setNullabilityMap( - HashMap nullabilityMap) { - this.nullabilityMap = nullabilityMap; - } - @Override public boolean equals(Object o) { if (this == o) { @@ -140,7 +136,7 @@ public static class Builder { private long updatedTimeMillis; private SampleSpec sampleSpec; private JsonObject insights; - private HashMap nullabilityMap; + private Map nullabilityMap; Builder(String name, String workspaceId) { this.workspaceName = name; @@ -175,7 +171,7 @@ public Builder setInsights(JsonObject insights) { return this; } - public Builder setNullabilityMap (HashMap nullabilityMap) { + public Builder setNullabilityMap(Map nullabilityMap) { this.nullabilityMap = nullabilityMap; return this; } @@ -185,13 +181,4 @@ public Workspace build() { insights, nullabilityMap); } } - - /** - * UserDefinedAction enum. - */ - public enum UserDefinedAction { - SKIP_ROW, - SEND_TO_ERROR_COLLECTOR, - ERROR_PIPELINE, - } } diff --git a/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/AbstractDirectiveHandler.java b/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/AbstractDirectiveHandler.java index 056c9da0c..99163ac39 100644 --- a/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/AbstractDirectiveHandler.java +++ b/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/AbstractDirectiveHandler.java @@ -48,7 +48,7 @@ import io.cdap.wrangler.proto.workspace.v2.DirectiveExecutionResponse; import io.cdap.wrangler.proto.workspace.v2.SampleSpec; import io.cdap.wrangler.proto.workspace.v2.Workspace; -import io.cdap.wrangler.proto.workspace.v2.Workspace.UserDefinedAction; +import io.cdap.wrangler.proto.workspace.v2.UserDefinedAction; import io.cdap.wrangler.registry.CompositeDirectiveRegistry; import io.cdap.wrangler.registry.DirectiveRegistry; import io.cdap.wrangler.registry.SystemDirectiveRegistry; @@ -122,7 +122,8 @@ protected List executeDirectives( List directives, List sample, GrammarWalker.Visitor grammarVisitor, - HashMap nullabilityMap) throws DirectiveParseException, E, RecipeException { + Map nullabilityMap) + throws DirectiveParseException, E, RecipeException { if (directives.isEmpty()) { return sample; diff --git a/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/WorkspaceHandler.java b/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/WorkspaceHandler.java index 4f4a677cd..5da2e4967 100644 --- a/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/WorkspaceHandler.java +++ b/wrangler-service/src/main/java/io/cdap/wrangler/service/directive/WorkspaceHandler.java @@ -67,7 +67,7 @@ import io.cdap.wrangler.proto.workspace.v2.ServiceResponse; import io.cdap.wrangler.proto.workspace.v2.StageSpec; import io.cdap.wrangler.proto.workspace.v2.Workspace; -import io.cdap.wrangler.proto.workspace.v2.Workspace.UserDefinedAction; +import io.cdap.wrangler.proto.workspace.v2.UserDefinedAction; import io.cdap.wrangler.proto.workspace.v2.WorkspaceCreationRequest; import io.cdap.wrangler.proto.workspace.v2.WorkspaceDetail; import io.cdap.wrangler.proto.workspace.v2.WorkspaceId; @@ -474,10 +474,9 @@ private DirectiveExecutionResponse execute(NamespaceSummary ns, HttpServiceReque WorkspaceDetail detail = wsStore.getWorkspaceDetail(workspaceId); UserDirectivesCollector userDirectivesCollector = new UserDirectivesCollector(); - HashMap nullabilityMap = executionRequest.getNullabilityMap() == null ? - new HashMap<>() : executionRequest.getNullabilityMap(); + Map nullabilityMap = executionRequest.getNullabilityMap(); if (!nullabilityMap.isEmpty()) { - //change nullabilityMap in Workspace Object + //create new workspace object with the new nullabilityMap changeNullability(nullabilityMap, workspaceId); } List result = executeDirectives(ns.getName(), directives, detail, @@ -492,16 +491,18 @@ private DirectiveExecutionResponse execute(NamespaceSummary ns, HttpServiceReque return response; } - private void changeNullability(HashMap columnMappings, - WorkspaceId workspaceId) throws Exception { + private void changeNullability(Map nullabilityMap, + WorkspaceId workspaceId) throws Exception { try { Workspace workspace = wsStore.getWorkspace(workspaceId); - workspace.setNullabilityMap(columnMappings); - wsStore.updateWorkspace(workspaceId, workspace); + Workspace newWorkspace = Workspace.builder(workspace) + .setUpdatedTimeMillis(System.currentTimeMillis()) + .setNullabilityMap(nullabilityMap).build(); + wsStore.updateWorkspace(workspaceId, newWorkspace); } catch (Exception e) { throw new RuntimeException("Error in setting nullabilityMap of columns ", e); } - } + } /**