public static boolean isLegal(final Traversal.Admin<?, ?> traversal) {
        final Step<?, ?> startStep = traversal.getStartStep();
        final Step<?, ?> endStep = traversal.getEndStep();
        // right now this is not supported because of how the SparkStarBarrierInterceptor mutates the traversal prior to local evaluation
        if (traversal.getStrategies().getStrategy(SubgraphStrategy.class).isPresent() ||
                traversal.getStrategies().getStrategy(ProductiveByStrategy.class).isPresent())
            return false;
        if (!startStep.getClass().equals(GraphStep.class) || ((GraphStep) startStep).returnsEdge())
            return false;
        if (!endStep.getClass().equals(CountGlobalStep.class) &&
                !endStep.getClass().equals(SumGlobalStep.class) &&
                !endStep.getClass().equals(MeanGlobalStep.class) &&
                !endStep.getClass().equals(MaxGlobalStep.class) &&
                !endStep.getClass().equals(MinGlobalStep.class) &&
                !endStep.getClass().equals(FoldStep.class) &&
                !endStep.getClass().equals(GroupStep.class) &&
                !endStep.getClass().equals(GroupCountStep.class))
            // TODO: tree()
            return false;
        if (TraversalHelper.getStepsOfAssignableClassRecursively(Scope.global, Barrier.class, traversal).size() != 1)
            return false;
        if (traversal.getTraverserRequirements().contains(TraverserRequirement.SACK))
            return false;
        return TraversalHelper.isLocalStarGraph(traversal);
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java