class HugeVertexStepByBatch<E extends Element>
        extends HugeVertexStep<E> {

    private static final long serialVersionUID = -3609787815053052222L;

    private BatchMapperIterator<Traverser.Admin<Vertex>, E> batchIterator;
    private Traverser.Admin<Vertex> head;
    private Iterator<E> iterator;

    public HugeVertexStepByBatch(final VertexStep<E> originalVertexStep) {
        super(originalVertexStep);
        this.batchIterator = null;
        this.head = null;
        this.iterator = null;
    }

    @Override
    protected Traverser.Admin<E> processNextStart() {
        /* Override super.processNextStart() */
        if (this.batchIterator == null) {
            int batchSize = (int) Query.QUERY_BATCH;
            this.batchIterator = new BatchMapperIterator<>(
                    batchSize, this.starts, this::flatMap);
        }

        if (this.batchIterator.hasNext()) {
            assert this.head != null;
            E item = this.batchIterator.next();
            // TODO: find the parent node accurately instead the head
            return this.head.split(item, this);
        }

        throw FastNoSuchElementException.instance();
    }

    @Override
    public void reset() {
        super.reset();
        this.closeIterator();
        this.batchIterator = null;
        this.head = null;
    }

    @Override
    public Iterator<?> lastTimeResults() {<FILL_FUNCTION_BODY>}

    @Override
    protected void closeIterator() {
        CloseableIterator.closeIterator(this.batchIterator);
    }

    @SuppressWarnings("unchecked")
    private Iterator<E> flatMap(List<Traverser.Admin<Vertex>> traversers) {
        if (this.head == null && !traversers.isEmpty()) {
            this.head = traversers.get(0);
        }
        boolean queryVertex = this.returnsVertex();
        boolean queryEdge = this.returnsEdge();
        assert queryVertex || queryEdge;
        if (queryVertex) {
            this.iterator = (Iterator<E>) this.vertices(traversers);
        } else {
            assert queryEdge;
            this.iterator = (Iterator<E>) this.edges(traversers);
        }
        return this.iterator;
    }

    private Iterator<Vertex> vertices(
            List<Traverser.Admin<Vertex>> traversers) {
        assert !traversers.isEmpty();

        Iterator<Edge> edges = this.edges(traversers);
        return this.queryAdjacentVertices(edges);
    }

    private Iterator<Edge> edges(List<Traverser.Admin<Vertex>> traversers) {
        assert !traversers.isEmpty();

        BatchConditionQuery batchQuery = new BatchConditionQuery(
                HugeType.EDGE, traversers.size());

        for (Traverser.Admin<Vertex> traverser : traversers) {
            ConditionQuery query = this.constructEdgesQuery(traverser);
            /*
             * Merge each query into batch query through IN condition
             * NOTE: duplicated results may be removed by backend store
             */
            batchQuery.mergeToIN(query, HugeKeys.OWNER_VERTEX);
        }

        this.injectQueryInfo(batchQuery);
        return this.queryEdges(batchQuery);
    }
}

        /*
         * NOTE: fetch page from this iterator， can only get page info of
         * the lowest level, may lost info of upper levels.
         */
        return this.iterator;
    
<methods>public void <init>(VertexStep<E>) ,public void addHasContainer(HasContainer) ,public boolean equals(java.lang.Object) ,public List<HasContainer> getHasContainers() ,public int hashCode() ,public Iterator<?> lastTimeResults() ,public org.apache.hugegraph.backend.query.Query queryInfo() ,public java.lang.String toString() <variables>private static final Logger LOG,private final List<HasContainer> hasContainers,private Iterator<E> iterator,private final org.apache.hugegraph.backend.query.Query queryInfo,private static final long serialVersionUID