import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
@Test
public void csvBatchImporterTest() throws java.lang.Exception
{

    CSVBatchImporter importer = new CSVBatchImporter();

    importer.setDbName("__TEST__1");
    try {
        importer.importCSVFiles(
                "test/src/resources/nodes.csv",
                "test/src/resources/edges.csv");
    }
    catch (OSchemaException exception) {
        exception.printStackTrace();
        throw exception;
    }

    OrientGraph graph = new OrientGraph(
            "plocal:" + System.getProperty("ORIENTDB_HOME") + "/databases/" + "__TEST__1");

    Pipe pipe = Gremlin.compile("_().map");
    pipe.setStarts(graph.getVertices());
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
@Override
public Pipe asPipe() {
    Map<BooleanClause.Occur, Collection<BooleanClause>> groupedClauses = groupClauses();

    Pipe andPipe = null;
    Collection<Pipe> andPipes = processAndClauses(groupedClauses);
    andPipes.addAll(processNotClauses(groupedClauses));
    if (! andPipes.isEmpty()) {
        andPipe = new AndFilterPipe(andPipes.toArray(new Pipe[andPipes.size()]));
    }

    Collection<Pipe> orPipes = processOrClauses(groupedClauses);
    if (! orPipes.isEmpty()) {
        if (andPipe != null) {
            orPipes.add(andPipe);
        }
        return new OrFilterPipe(orPipes.toArray(new Pipe[orPipes.size()]));
    } else {
        return andPipe;
    }
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
private Collection<Pipe> processOrClauses(Map<BooleanClause.Occur, Collection<BooleanClause>> groupedClauses) {
    Collection<BooleanClause> shouldClauses = groupedClauses.get(BooleanClause.Occur.SHOULD);
    Collection<Pipe> orPipes = new ArrayList<>();
    if (shouldClauses != null) {
        for (BooleanClause shouldClause : shouldClauses) {
            QueryExpression queryExpression = queryFactory.create(shouldClause.getQuery(), resourceDefinition);
            // don't negate expression if we negated MUST_NOT -> SHOULD
            if (negate && shouldClause.getOccur() != BooleanClause.Occur.MUST_NOT) {
                queryExpression.setNegate();
            }
            properties.addAll(queryExpression.getProperties());
            orPipes.add(queryExpression.asPipe());
        }
    }
    return orPipes;
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
@Override
protected Pipe getQueryPipe() {
    GremlinPipeline p;
    if (guid.equals("*")) {
        p = new GremlinPipeline().has(Constants.ENTITY_TEXT_PROPERTY_KEY).
                hasNot(Constants.ENTITY_TYPE_PROPERTY_KEY, "Taxonomy").outE();
    } else {
        p = new GremlinPipeline().has(Constants.GUID_PROPERTY_KEY, guid).outE();
    }
    //todo: this is basically the same pipeline used in TagRelation.asPipe()
    p.add(new FilterFunctionPipe<>(new PipeFunction<Edge, Boolean>() {
        @Override
        public Boolean compute(Edge edge) {
            String type = edge.getVertex(Direction.OUT).getProperty(Constants.ENTITY_TYPE_PROPERTY_KEY);
            VertexWrapper v = new TermVertexWrapper(edge.getVertex(Direction.IN));
            return edge.getLabel().startsWith(type) && v.getPropertyKeys().contains("available_as_tag");
        }
    }));

    return p.inV();
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public TestAtlasEntityQuery(QueryExpression queryExpression,
                            ResourceDefinition resourceDefinition,
                            Request request,
                            GremlinPipeline initialPipeline,
                            Pipe queryPipe,
                            Pipe notDeletedPipe,
                            AtlasGraph graph,
                            VertexWrapper vWrapper) {

    super(queryExpression, resourceDefinition, request);
    this.initialPipeline = initialPipeline;
    this.queryPipe = queryPipe;
    this.notDeletedPipe = notDeletedPipe;
    this.graph = graph;
    this.vWrapper = vWrapper;
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public List processNextStart() {
    if (this.starts instanceof Pipe) {
        this.starts.next();
        final List path = ((Pipe) this.starts).getCurrentPath();
        if (null == this.pathFunctions) {
            return path;
        } else {
            final List closedPath = new Path();
            int nextFunction = 0;
            for (final Object object : path) {
                closedPath.add(this.pathFunctions[nextFunction].compute(object));
                nextFunction = (nextFunction + 1) % this.pathFunctions.length;
            }
            return closedPath;
        }
    } else {
        throw new NoSuchElementException("The start of this pipe was not a pipe");
    }
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public List processNextStart() {
    if (this.starts instanceof Pipe) {
        this.starts.next();
        final List path = ((Pipe) this.starts).getCurrentPath();
        if (null == this.pathFunctions) {
            return path;
        } else {
            final List closedPath = new ArrayList();
            int nextFunction = 0;
            for (final Object object : path) {
                closedPath.add(this.pathFunctions[nextFunction].compute(object));
                nextFunction = (nextFunction + 1) % this.pathFunctions.length;
            }
            return closedPath;
        }
    } else {
        throw new NoSuchElementException("The start of this pipe was not a pipe");
    }
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public S processNextStart() {
    final S s = this.starts.next();
    final List path = ((Pipe) this.starts).getCurrentPath();
    Tree<Object> depth = this.tree;
    for (int i = 0; i < path.size(); i++) {
        Object object = path.get(i);
        if (null != this.branchFunctions) {
            object = this.branchFunctions.get(this.currentFunction).compute(object);
            this.currentFunction = (this.currentFunction + 1) % this.branchFunctions.size();
        }

        if (!depth.containsKey(object))
            depth.put(object, new Tree());

        depth = depth.get(object);
    }
    return s;
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
/**
 * A utility method to remove all the pipes back some partition step and return them as a list.
 *
 * @param namedStep the name of the step previous to remove from the pipeline
 * @return the removed pipes
 */
public static List<Pipe> removePreviousPipes(final Pipeline pipeline, final String namedStep) {
    final List<Pipe> previousPipes = new ArrayList<Pipe>();
    for (int i = pipeline.size() - 1; i >= 0; i--) {
        final Pipe pipe = pipeline.get(i);
        if (pipe instanceof AsPipe && ((AsPipe) pipe).getName().equals(namedStep)) {
            break;
        } else {
            previousPipes.add(0, pipe);
        }
    }
    for (int i = 0; i < previousPipes.size(); i++) {
        pipeline.remove(pipeline.size() - 1);
    }

    if (pipeline.size() == 1)
        pipeline.setStarts(pipeline.getStarts());

    return previousPipes;
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public S processNextStart() {
    while (true) {
        final S s = this.starts.next();
        if (this.starts instanceof Pipe) {
            final List path = ((Pipe) this.starts).getCurrentPath();
            this.set.clear();
            this.set.addAll(path);
            if (path.size() == this.set.size()) {
                return s;
            }
        } else {
            return s;
        }
    }

}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public S processNextStart() {
    int counter = 0;
    while (true) {
        counter++;
        final Pipe currentPipe = this.pipes.get(this.current);
        if (currentPipe.hasNext()) {
            final S s = (S) currentPipe.next();
            this.current = (this.current + 1) % this.total;
            return s;
        } else if (counter == this.total) {
            throw FastNoSuchElementException.instance();
        } else {
            this.current = (this.current + 1) % this.total;
        }
    }
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public void writeResult(Object result) throws IOException
{
	if (result == null)
	{
		writeMessage("");
	} else if (result instanceof Pipe)
	{
		StringBuilder sBuilder = new StringBuilder();
		Iterable<?> iterable = (Iterable<?>) result;
		for (Object obj : iterable)
		{
			if (obj != null)
			{
				sBuilder.append(obj.toString());
				sBuilder.append("\n");
			}
		}
		if (sBuilder.length() > 0)
		{
			sBuilder.deleteCharAt(sBuilder.length() - 1);
		}
		writeMessage(sBuilder.toString());
	} else
	{
		writeMessage(result.toString());
	}
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
@Override
public Pipe asPipe() {
    return new FilterFunctionPipe<>(new PipeFunction<Edge, Boolean>() {
        @Override
        public Boolean compute(Edge edge) {
            String name = edge.getVertex(Direction.OUT).getProperty(Constants.ENTITY_TYPE_PROPERTY_KEY);
            if (edge.getLabel().startsWith(name)) {
                VertexWrapper v = new TermVertexWrapper(edge.getVertex(Direction.IN));
                return ! v.getPropertyKeys().contains("available_as_tag") && ! isDeleted(v.getVertex());
            } else {
                return false;
            }
        }
    });
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
@Override
public Pipe asPipe() {
    return new FilterFunctionPipe<>(new PipeFunction<Edge, Boolean>() {
        @Override
        public Boolean compute(Edge edge) {
            String name = edge.getVertex(Direction.OUT).getProperty(Constants.ENTITY_TYPE_PROPERTY_KEY);
            if (edge.getLabel().startsWith(name)) {
                VertexWrapper v = new TermVertexWrapper(edge.getVertex(Direction.IN));
                return v.getPropertyKeys().contains("available_as_tag") && ! isDeleted(v.getVertex());
            } else {
                return false;
            }
        }
    });
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
private Collection<Pipe> processAndClauses(Map<BooleanClause.Occur, Collection<BooleanClause>> groupedClauses) {
    Collection<BooleanClause> andClauses = groupedClauses.get(BooleanClause.Occur.MUST);
    Collection<Pipe> andPipes = new ArrayList<>();
    if (andClauses != null) {
        for (BooleanClause andClause : andClauses) {
            QueryExpression queryExpression = queryFactory.create(andClause.getQuery(), resourceDefinition);
            properties.addAll(queryExpression.getProperties());
            andPipes.add(queryExpression.asPipe());
        }
    }
    return andPipes;
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
private Collection<Pipe> processNotClauses(Map<BooleanClause.Occur, Collection<BooleanClause>> groupedClauses) {
    Collection<BooleanClause> notClauses = groupedClauses.get(BooleanClause.Occur.MUST_NOT);
    Collection<Pipe> notPipes = new ArrayList<>();
    if (notClauses != null) {
        for (BooleanClause notClause : notClauses) {
            QueryExpression queryExpression = queryFactory.create(notClause.getQuery(), resourceDefinition);
            queryExpression.setNegate();
            properties.addAll(queryExpression.getProperties());
            notPipes.add(queryExpression.asPipe());
        }
    }
    return notPipes;
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
public Pipe asPipe() {
    return new FilterFunctionPipe(new PipeFunction<Vertex, Boolean>() {
        @Override
        public Boolean compute(Vertex vertex) {
            return evaluate(new VertexWrapper(vertex, resourceDefinition));
        }
    });
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
@Override
public Pipe asPipe() {
    //todo: encapsulate all of this path logic including path sep escaping and normalizing
    final int sepIdx = getField().indexOf(QueryFactory.PATH_SEP_TOKEN);
    final String edgeToken = getField().substring(0, sepIdx);
    GremlinPipeline pipeline = new GremlinPipeline();

    Relation relation = resourceDefinition.getRelations().get(fieldSegments[0]);
    if (relation != null) {
        pipeline = pipeline.outE();
        pipeline.add(relation.asPipe()).inV();
    } else {
        if (resourceDefinition.getProjections().get(fieldSegments[0]) != null) {
            return super.asPipe();
        } else {
            //todo: default Relation implementation
            pipeline = pipeline.outE().has("label", Text.REGEX, String.format(".*\\.%s", edgeToken)).inV();
        }
    }
    //todo: set resource definition from relation on underlying expression where appropriate
    String childFieldName = getField().substring(sepIdx + QueryFactory.PATH_SEP_TOKEN.length());
    underlyingExpression.setField(childFieldName);

    Pipe childPipe;
    if (childFieldName.contains(QueryFactory.PATH_SEP_TOKEN)) {
        childPipe = new ProjectionQueryExpression(underlyingExpression, resourceDefinition).asPipe();
    } else {
        childPipe = underlyingExpression.asPipe();
    }
    pipeline.add(childPipe);

    return negate ? new FilterFunctionPipe(new ExcludePipeFunction(pipeline)) : pipeline;
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
private List<Vertex> executeQuery() {
    GremlinPipeline pipeline = buildPipeline().as("root");
    Pipe expressionPipe = queryExpression.asPipe();

    // AlwaysQuery returns null for pipe
    return expressionPipe == null ? pipeline.toList() :
            pipeline.add(expressionPipe).back("root").toList();
}
 

import com.tinkerpop.pipes.Pipe; //导入依赖的package包/类
protected GremlinPipeline buildPipeline() {
    GremlinPipeline pipeline = getRootVertexPipeline();
    Pipe queryPipe = getQueryPipe();
    if (queryPipe != null) {
        pipeline.add(queryPipe);
    }
    pipeline.add(getNotDeletedPipe());
    return pipeline;
}