@Override
  public Span next() {
    if (!hasNext()) throw new NoSuchElementException();
    Record row = delegate.peek();

    long spanId = row.getValue(ZipkinSpans.ZIPKIN_SPANS.ID);
    boolean error = false;
    String lcService = null, srService = null, csService = null, caService = null, saService = null,
      maService = null, mrService = null, msService = null;
    while (hasNext()) { // there are more values for this trace
      if (spanId != delegate.peek().getValue(ZipkinSpans.ZIPKIN_SPANS.ID)) {
        break; // if we are in a new span
      }
      Record next = delegate.next(); // row for the same span

      String key = emptyToNull(next, ZIPKIN_ANNOTATIONS.A_KEY);
      String value = emptyToNull(next, ZIPKIN_ANNOTATIONS.ENDPOINT_SERVICE_NAME);
      if (key == null || value == null) continue; // neither client nor server
      switch (key) {
        case "lc":
          lcService = value;
          break;
        case "ca":
          caService = value;
          break;
        case "cs":
          csService = value;
          break;
        case "ma":
          maService = value;
          break;
        case "mr":
          mrService = value;
          break;
        case "ms":
          msService = value;
          break;
        case "sa":
          saService = value;
          break;
        case "sr":
          srService = value;
          break;
        case "error":
          // a span is in error if it has a tag, not an annotation, of name "error"
          error = V1BinaryAnnotation.TYPE_STRING == next.get(ZIPKIN_ANNOTATIONS.A_TYPE);
      }
    }

    // The client address is more authoritative than the client send owner.
    if (caService == null) caService = csService;

    // Finagle labels two sides of the same socket ("ca", "sa") with the same name.
    // Skip the client side, so it isn't mistaken for a loopback request
    if (saService != null && saService.equals(caService)) caService = null;

    long parentId = maybeGet(row, ZipkinSpans.ZIPKIN_SPANS.PARENT_ID, 0L);
    Span.Builder result =
        Span.newBuilder().traceId(traceIdHi, traceIdLo).parentId(parentId).id(spanId);

    if (error) {
      result.putTag("error", "" /* actual value doesn't matter */);
    }

    if (srService != null) {
      return result
          .kind(Span.Kind.SERVER)
          .localEndpoint(ep(srService))
          .remoteEndpoint(ep(caService))
          .build();
    } else if (saService != null) {
      Endpoint localEndpoint = ep(caService);
      // When span.kind is missing, the local endpoint is "lc" and the remote endpoint is "sa"
      if (localEndpoint == null) localEndpoint = ep(lcService);
      return result
          .kind(csService != null ? Span.Kind.CLIENT : null)
          .localEndpoint(localEndpoint)
          .remoteEndpoint(ep(saService))
          .build();
    } else if (csService != null) {
      return result.kind(Span.Kind.SERVER).localEndpoint(ep(caService)).build();
    } else if (mrService != null) {
      return result
        .kind(Span.Kind.CONSUMER)
        .localEndpoint(ep(mrService))
        .remoteEndpoint(ep(maService))
        .build();
    } else if (msService != null) {
      return result
        .kind(Span.Kind.PRODUCER)
        .localEndpoint(ep(msService))
        .remoteEndpoint(ep(maService))
        .build();
    }
    return result.build();
  }
Assumes the input records are sorted by trace id, span id
zipkin-storage/mysql-v1/src/main/java/zipkin2/storage/mysql/v1/DependencyLinkV2SpanIterator.java