@Override
    public void apply(final Traversal.Admin<?, ?> traversal) {
        // nothing partitioning can do will alter the behavior of an AbstractLambdaTraversal implementation unless
        // it has a bypass in which case the strategy will operate on that
        if (traversal instanceof AbstractLambdaTraversal && null == ((AbstractLambdaTraversal<?, ?>) traversal).getBypassTraversal())
            return;

        // if there are vertexFeatures assigned then it means that includeMetaProperties is enabled and the graph
        // can support their usage. the only reason the VertexFeatures are needed is for cardinality checks for
        // writes.
        final Optional<Graph.Features.VertexFeatures> vertexFeatures;
        if (includeMetaProperties) {
            final Graph graph = traversal.getGraph().orElseThrow(
                    () -> new IllegalStateException("PartitionStrategy does not work with anonymous Traversals when includeMetaProperties is enabled"));
            final Graph.Features.VertexFeatures vf = graph.features().vertex();
            final boolean supportsMetaProperties = vf.supportsMetaProperties();
            if (!supportsMetaProperties)
                throw new IllegalStateException("PartitionStrategy is configured to include meta-properties but the Graph does not support them");
            vertexFeatures = Optional.of(vf);
        } else {
            vertexFeatures = Optional.empty();
        }

        // no need to add has after mutating steps because we want to make it so that the write partition can
        // be independent of the read partition.  in other words, i don't need to be able to read from a partition
        // in order to write to it. Seems like ElementStep isn't necessary here? a Property can't be loaded that
        // isn't within the partition so element() could only ever traverse back to something within the partition.
        final List<Step> stepsToInsertHasAfter = new ArrayList<>();
        stepsToInsertHasAfter.addAll(TraversalHelper.getStepsOfAssignableClass(GraphStep.class, traversal));
        stepsToInsertHasAfter.addAll(TraversalHelper.getStepsOfAssignableClass(VertexStep.class, traversal));
        stepsToInsertHasAfter.addAll(TraversalHelper.getStepsOfAssignableClass(EdgeOtherVertexStep.class, traversal));
        stepsToInsertHasAfter.addAll(TraversalHelper.getStepsOfAssignableClass(EdgeVertexStep.class, traversal));

        // all steps that return a vertex need to have has(partitionKey,within,partitionValues) injected after it.
        // attempt to put the partition filter at the end of other has() following that step so that the order of
        // the has() is maintained on the chance that order up to that point is somehow intentional. this seems
        // only relevant to Vertex/Edge steps and not properties where this order would seemingly have less
        // significance to read performance
        stepsToInsertHasAfter.forEach(step -> {
            // find the last has() following the insert step
            Step insertAfter = step;
            while (insertAfter.getNextStep() instanceof HasStep) {
                insertAfter = insertAfter.getNextStep();
            }
            TraversalHelper.insertAfterStep(
                    new HasStep(traversal, new HasContainer(partitionKey, P.within(new ArrayList<>(readPartitions)))), insertAfter, traversal);
        });

        if (vertexFeatures.isPresent()) {
            final List<PropertiesStep> propertiesSteps = TraversalHelper.getStepsOfAssignableClass(PropertiesStep.class, traversal);
            propertiesSteps.forEach(step -> {
                // check length first because keyExists will return true otherwise
                if (step.getPropertyKeys().length > 0 && ElementHelper.keyExists(partitionKey, step.getPropertyKeys()))
                    throw new IllegalStateException("Cannot explicitly request the partitionKey in the traversal");

                if (step.getReturnType() == PropertyType.PROPERTY) {
                    // check the following step to see if it is a has(partitionKey, *) - if so then this strategy was
                    // already applied down below via g.V().values() which injects a properties() step
                    final Step next = step.getNextStep();
                    if (!(next instanceof HasStep) || !((HasContainer) ((HasStep) next).getHasContainers().get(0)).getKey().equals(partitionKey)) {
                        // use choose() to determine if the properties() step is called on a Vertex to get a VertexProperty
                        // if not, pass it through.
                        final Traversal choose = __.choose(
                                __.filter(new TypeChecker<>(VertexProperty.class)),
                                __.has(partitionKey, P.within(new ArrayList<>(readPartitions))),
                                __.__()).filter(new PartitionKeyHider());
                        TraversalHelper.insertTraversal(step, choose.asAdmin(), traversal);
                    }
                } else if (step.getReturnType() == PropertyType.VALUE) {
                    // use choose() to determine if the values() step is called on a Vertex to get a VertexProperty
                    // if not, pass it through otherwise explode g.V().values() to g.V().properties().has().value()
                    final Traversal choose = __.choose(
                            __.filter(new TypeChecker<>(Vertex.class)),
                            __.properties(step.getPropertyKeys()).has(partitionKey, P.within(new ArrayList<>(readPartitions))).filter(new PartitionKeyHider()).value(),
                            __.__().filter(new PartitionKeyHider()));
                    TraversalHelper.insertTraversal(step, choose.asAdmin(), traversal);
                    traversal.removeStep(step);
                } else {
                    throw new IllegalStateException(String.format("%s is not accounting for a particular %s %s",
                            PartitionStrategy.class.getSimpleName(), PropertyType.class.toString(), step.getReturnType()));
                }
            });

            final List<PropertyMapStep> propertyMapSteps = TraversalHelper.getStepsOfAssignableClass(PropertyMapStep.class, traversal);
            propertyMapSteps.forEach(step -> {
                // check length first because keyExists will return true otherwise
                if (step.getPropertyKeys().length > 0 && ElementHelper.keyExists(partitionKey, step.getPropertyKeys()))
                    throw new IllegalStateException("Cannot explicitly request the partitionKey in the traversal");

                if (step.getReturnType() == PropertyType.PROPERTY) {
                    // via map() filter out properties that aren't in the partition if it is a PropertyVertex,
                    // otherwise just let them pass through
                    TraversalHelper.insertAfterStep(new LambdaMapStep<>(traversal, new MapPropertiesFilter()), step, traversal);
                } else if (step.getReturnType() == PropertyType.VALUE) {
                    // as this is a value map, replace that step with propertiesMap() that returns PropertyType.VALUE.
                    // from there, add the filter as shown above and then unwrap the properties as they would have
                    // been done under valueMap()
                    final PropertyMapStep propertyMapStep = new PropertyMapStep(traversal, PropertyType.PROPERTY, step.getPropertyKeys());
                    propertyMapStep.configure(WithOptions.tokens, step.getIncludedTokens());
                    TraversalHelper.replaceStep(step, propertyMapStep, traversal);

                    final LambdaMapStep mapPropertiesFilterStep = new LambdaMapStep<>(traversal, new MapPropertiesFilter());
                    TraversalHelper.insertAfterStep(mapPropertiesFilterStep, propertyMapStep, traversal);
                    TraversalHelper.insertAfterStep(new LambdaMapStep<>(traversal, new MapPropertiesConverter()), mapPropertiesFilterStep, traversal);
                } else {
                    throw new IllegalStateException(String.format("%s is not accounting for a particular %s %s",
                            PartitionStrategy.class.getSimpleName(), PropertyType.class.toString(), step.getReturnType()));
                }
            });
        }

        final List<Step> stepsToInsertPropertyMutations = traversal.getSteps().stream().filter(step ->
                step instanceof MergeVertexStep || step instanceof MergeEdgeStep ||
                step instanceof AddEdgeStep || step instanceof AddVertexStep ||
                        step instanceof AddEdgeStartStep || step instanceof AddVertexStartStep ||
                        (includeMetaProperties && step instanceof AddPropertyStep)
        ).collect(Collectors.toList());

        stepsToInsertPropertyMutations.forEach(step -> {
            // note that with AddPropertyStep we just add the partition key/value regardless of whether this
            // ends up being a Vertex or not.  AddPropertyStep currently chooses to simply not bother
            // to use the additional "property mutations" if the Element being mutated is a Edge or
            // VertexProperty
            ((Mutating) step).configure(partitionKey, writePartition);

            if (vertexFeatures.isPresent()) {
                // GraphTraversal folds g.addV().property('k','v') to just AddVertexStep/AddVertexStartStep so this
                // has to be exploded back to g.addV().property(cardinality, 'k','v','partition','A')
                if (step instanceof AddVertexStartStep || step instanceof AddVertexStep) {
                    final Parameters parameters = ((Parameterizing) step).getParameters();
                    final Map<Object, List<Object>> params = parameters.getRaw();
                    params.forEach((k, v) -> {

                        // need to filter out T based keys
                        if (k instanceof String) {
                            final List<Step> addPropertyStepsToAppend = new ArrayList<>(v.size());
                            final VertexProperty.Cardinality cardinality = vertexFeatures.get().getCardinality((String) k);
                            v.forEach(o -> {
                                final AddPropertyStep addPropertyStep = new AddPropertyStep(traversal, cardinality, k, o);
                                addPropertyStep.configure(partitionKey, writePartition);
                                addPropertyStepsToAppend.add(addPropertyStep);

                                // need to remove the parameter from the AddVertex/StartStep because it's now being added
                                // via the AddPropertyStep
                                parameters.remove(k);
                            });

                            Collections.reverse(addPropertyStepsToAppend);
                            addPropertyStepsToAppend.forEach(s -> TraversalHelper.insertAfterStep(s, step, traversal));
                        }
                    });
                }
            }
        });
    }
{@code PartitionStrategy} partitions the vertices, edges and vertex properties of a graph into String named
partitions (i.e. buckets, subgraphs, etc.).  It blinds a {@link Traversal} from "seeing" specified areas of
the graph given the partition names assigned to {@link Builder#readPartitions(String...)}.  The traversal will
ignore all graph elements not in those "read" partitions.

@author Stephen Mallette (http://stephen.genoprime.com)
@author Marko A. Rodriguez (http://markorodriguez.com)
gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/decoration/PartitionStrategy.java