Repository: apache/skywalking (language: python)

Issue:
Add zipkin storage BanyanDB  implementation 
### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/skywalking/issues?q=is%3Aissue) and found no similar feature requirement.


### Description

https://github.com/apache/skywalking/pull/9177 doesn't support BanyanDB. 
Since BanyanDB has been released, we need to add the implementation.

### Use case

_No response_

### Related issues

https://github.com/apache/skywalking/pull/9177

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)



Discussion / hints:
This feature would rely on the released BanyanDB 0.2 server and its Java client 0.2.

Fix patch:
diff --git a/.github/workflows/skywalking.yaml b/.github/workflows/skywalking.yaml
index 621883f20926..001aa30baeb2 100644
--- a/.github/workflows/skywalking.yaml
+++ b/.github/workflows/skywalking.yaml
@@ -606,6 +606,8 @@ jobs:
             config: test/e2e-v2/cases/zipkin/postgres/e2e.yaml
           - name: Zipkin Kafka
             config: test/e2e-v2/cases/zipkin/kafka/e2e.yaml
+          - name: Zipkin BanyanDB
+            config: test/e2e-v2/cases/zipkin/banyandb/e2e.yaml
 
           - name: MySQL-Sharding storage
             config: test/e2e-v2/cases/storage/mysql/sharding/e2e.yaml
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 3eeba6d6ac5d..4bda098a4099 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -116,6 +116,8 @@
 * Changed system variable `SW_SUPERDATASET_STORAGE_DAY_STEP` to `SW_STORAGE_ES_SUPER_DATASET_DAY_STEP` to be consistent with other ES storage related variables.
 * Fix ESEventQueryDAO missing metric_table boolQuery criteria.
 * Add default entity name(`_blank`) if absent to avoid NPE in the decoding. This caused `Can't split xxx id into 2 parts`.
+* Zipkin module support BanyanDB storage.
+* Zipkin traces query API, sort the result set by start time by default.
 
 #### UI
 
diff --git a/docs/en/setup/backend/zipkin-trace.md b/docs/en/setup/backend/zipkin-trace.md
index 132a688c3ff4..a857fd95154c 100644
--- a/docs/en/setup/backend/zipkin-trace.md
+++ b/docs/en/setup/backend/zipkin-trace.md
@@ -39,7 +39,6 @@ receiver-zipkin:
 ## Zipkin query
 The Zipkin receiver makes the OAP server work as an alternative Zipkin server implementation for query traces. 
 It implemented `ZipkinQueryApiV2` through the HTTP service, supporting Zipkin-lens UI.
-**Notice: Zipkin query API implementation does not support BanyanDB yet.**
 
 Use the following config to activate it.
 
diff --git a/oap-server-bom/pom.xml b/oap-server-bom/pom.xml
index 1d30803b040f..04cda371ef2c 100644
--- a/oap-server-bom/pom.xml
+++ b/oap-server-bom/pom.xml
@@ -73,7 +73,7 @@
         <awaitility.version>3.0.0</awaitility.version>
         <httpcore.version>4.4.13</httpcore.version>
         <commons-compress.version>1.21</commons-compress.version>
-        <banyandb-java-client.version>0.2.0</banyandb-java-client.version>
+        <banyandb-java-client.version>0.3.0-SNAPSHOT</banyandb-java-client.version>
         <kafka-clients.version>2.8.1</kafka-clients.version>
         <spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
     </properties>
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/zipkin/ZipkinSpanRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/zipkin/ZipkinSpanRecord.java
index d0f3a6e65906..3e44aa1f65ab 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/zipkin/ZipkinSpanRecord.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/zipkin/ZipkinSpanRecord.java
@@ -19,8 +19,10 @@
 package org.apache.skywalking.oap.server.core.zipkin;
 
 import com.google.gson.Gson;
+import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import java.util.List;
+import java.util.Map;
 import lombok.Getter;
 import lombok.Setter;
 import org.apache.skywalking.oap.server.core.Const;
@@ -38,6 +40,8 @@
 import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
 import org.apache.skywalking.oap.server.library.util.BooleanUtils;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
+import zipkin2.Endpoint;
+import zipkin2.Span;
 
 import static org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord.TRACE_ID;
 import static org.apache.skywalking.oap.server.core.analysis.record.Record.TIME_BUCKET;
@@ -249,4 +253,51 @@ public void entity2Storage(final ZipkinSpanRecord storageData, final Convert2Sto
             }
         }
     }
+
+    public static Span buildSpanFromRecord(ZipkinSpanRecord record) {
+        Span.Builder span = Span.newBuilder();
+        span.traceId(record.getTraceId());
+        span.id(record.getSpanId());
+        span.parentId(record.getParentId());
+        span.kind(Span.Kind.valueOf(record.getKind()));
+        span.timestamp(record.getTimestamp());
+        span.duration(record.getDuration());
+        span.name(record.getName());
+        //Build localEndpoint
+        Endpoint.Builder localEndpoint = Endpoint.newBuilder();
+        localEndpoint.serviceName(record.getLocalEndpointServiceName());
+        if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
+            localEndpoint.parseIp(record.getLocalEndpointIPV4());
+        } else {
+            localEndpoint.parseIp(record.getLocalEndpointIPV6());
+        }
+        localEndpoint.port(record.getLocalEndpointPort());
+        span.localEndpoint(localEndpoint.build());
+        //Build remoteEndpoint
+        Endpoint.Builder remoteEndpoint = Endpoint.newBuilder();
+        remoteEndpoint.serviceName(record.getRemoteEndpointServiceName());
+        if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
+            remoteEndpoint.parseIp(record.getRemoteEndpointIPV4());
+        } else {
+            remoteEndpoint.parseIp(record.getRemoteEndpointIPV6());
+        }
+        remoteEndpoint.port(record.getRemoteEndpointPort());
+        span.remoteEndpoint(remoteEndpoint.build());
+
+        //Build tags
+        JsonObject tagsJson = record.getTags();
+        if (tagsJson != null) {
+            for (Map.Entry<String, JsonElement> tag : tagsJson.entrySet()) {
+                span.putTag(tag.getKey(), tag.getValue().getAsString());
+            }
+        }
+        //Build annotation
+        JsonObject annotationJson = record.getAnnotations();
+        if (annotationJson != null) {
+            for (Map.Entry<String, JsonElement> annotation : annotationJson.entrySet()) {
+                span.addAnnotation(Long.parseLong(annotation.getKey()), annotation.getValue().getAsString());
+            }
+        }
+        return span.build();
+    }
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
index b0555952cbb3..53571c6dc20f 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageProvider.java
@@ -72,7 +72,6 @@
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBSpanAttachedEventQueryDAO;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBStorageDAO;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBTraceQueryDAO;
-import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.BanyanDBZipkinQueryDAO;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
 import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBZipkinQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBZipkinQueryDAO.java
new file mode 100644
index 000000000000..96560abb2863
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBZipkinQueryDAO.java
@@ -0,0 +1,297 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.storage.plugin.banyandb;
+
+import com.google.common.collect.ImmutableSet;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
+import org.apache.skywalking.banyandb.v1.client.AbstractQuery;
+import org.apache.skywalking.banyandb.v1.client.DataPoint;
+import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
+import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.banyandb.v1.client.RowEntity;
+import org.apache.skywalking.banyandb.v1.client.StreamQuery;
+import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
+import org.apache.skywalking.banyandb.v1.client.TimestampRange;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.core.storage.query.IZipkinQueryDAO;
+import org.apache.skywalking.oap.server.core.zipkin.ZipkinServiceRelationTraffic;
+import org.apache.skywalking.oap.server.core.zipkin.ZipkinServiceSpanTraffic;
+import org.apache.skywalking.oap.server.core.zipkin.ZipkinServiceTraffic;
+import org.apache.skywalking.oap.server.core.zipkin.ZipkinSpanRecord;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
+import zipkin2.Span;
+import zipkin2.storage.QueryRequest;
+
+public class BanyanDBZipkinQueryDAO extends AbstractBanyanDBDAO implements IZipkinQueryDAO {
+    private final static int QUERY_MAX_SIZE = Integer.MAX_VALUE;
+    private static final Set<String> SERVICE_TRAFFIC_TAGS = ImmutableSet.of(ZipkinServiceTraffic.SERVICE_NAME);
+    private static final Set<String> REMOTE_SERVICE_TRAFFIC_TAGS = ImmutableSet.of(
+        ZipkinServiceRelationTraffic.REMOTE_SERVICE_NAME);
+    private static final Set<String> SPAN_TRAFFIC_TAGS = ImmutableSet.of(ZipkinServiceSpanTraffic.SPAN_NAME);
+    private static final Set<String> TRACE_ID = ImmutableSet.of(ZipkinSpanRecord.TRACE_ID);
+    private static final Set<String> TRACE_TAGS = ImmutableSet.of(
+        ZipkinSpanRecord.TRACE_ID,
+        ZipkinSpanRecord.SPAN_ID,
+        ZipkinSpanRecord.PARENT_ID,
+        ZipkinSpanRecord.KIND,
+        ZipkinSpanRecord.TIMESTAMP,
+        ZipkinSpanRecord.TIMESTAMP_MILLIS,
+        ZipkinSpanRecord.DURATION,
+        ZipkinSpanRecord.NAME,
+        ZipkinSpanRecord.DEBUG,
+        ZipkinSpanRecord.SHARED,
+        ZipkinSpanRecord.LOCAL_ENDPOINT_SERVICE_NAME,
+        ZipkinSpanRecord.LOCAL_ENDPOINT_IPV4,
+        ZipkinSpanRecord.LOCAL_ENDPOINT_IPV6,
+        ZipkinSpanRecord.LOCAL_ENDPOINT_PORT,
+        ZipkinSpanRecord.REMOTE_ENDPOINT_SERVICE_NAME,
+        ZipkinSpanRecord.REMOTE_ENDPOINT_IPV4,
+        ZipkinSpanRecord.REMOTE_ENDPOINT_IPV6,
+        ZipkinSpanRecord.REMOTE_ENDPOINT_PORT,
+        ZipkinSpanRecord.TAGS,
+        ZipkinSpanRecord.ANNOTATIONS
+    );
+
+    public BanyanDBZipkinQueryDAO(BanyanDBStorageClient client) {
+        super(client);
+    }
+
+    @Override
+    public List<String> getServiceNames() throws IOException {
+        MeasureQueryResponse resp =
+            query(ZipkinServiceTraffic.INDEX_NAME,
+                  SERVICE_TRAFFIC_TAGS,
+                  Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
+
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        query.setLimit(QUERY_MAX_SIZE);
+                    }
+                }
+            );
+        final List<String> services = new ArrayList<>();
+        for (final DataPoint dataPoint : resp.getDataPoints()) {
+            services.add(dataPoint.getTagValue(ZipkinServiceTraffic.SERVICE_NAME));
+        }
+        return services;
+    }
+
+    @Override
+    public List<String> getRemoteServiceNames(final String serviceName) throws IOException {
+        MeasureQueryResponse resp =
+            query(ZipkinServiceRelationTraffic.INDEX_NAME,
+                  REMOTE_SERVICE_TRAFFIC_TAGS,
+                  Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
+
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        if (StringUtil.isNotEmpty(serviceName)) {
+                            query.and(eq(ZipkinServiceRelationTraffic.SERVICE_NAME, serviceName));
+                        }
+                        query.setLimit(QUERY_MAX_SIZE);
+                    }
+                }
+            );
+        final List<String> remoteServices = new ArrayList<>();
+        for (final DataPoint dataPoint : resp.getDataPoints()) {
+            remoteServices.add(dataPoint.getTagValue(ZipkinServiceRelationTraffic.REMOTE_SERVICE_NAME));
+        }
+        return remoteServices;
+    }
+
+    @Override
+    public List<String> getSpanNames(final String serviceName) throws IOException {
+        MeasureQueryResponse resp =
+            query(ZipkinServiceSpanTraffic.INDEX_NAME,
+                  SPAN_TRAFFIC_TAGS,
+                  Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
+
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        if (StringUtil.isNotEmpty(serviceName)) {
+                            query.and(eq(ZipkinServiceSpanTraffic.SERVICE_NAME, serviceName));
+                        }
+                        query.setLimit(QUERY_MAX_SIZE);
+                    }
+                }
+            );
+        final List<String> spanNames = new ArrayList<>();
+        for (final DataPoint dataPoint : resp.getDataPoints()) {
+            spanNames.add(dataPoint.getTagValue(ZipkinServiceSpanTraffic.SPAN_NAME));
+        }
+        return spanNames;
+    }
+
+    @Override
+    public List<Span> getTrace(final String traceId) throws IOException {
+        StreamQueryResponse resp =
+            query(ZipkinSpanRecord.INDEX_NAME, TRACE_TAGS,
+                  new QueryBuilder<StreamQuery>() {
+
+                      @Override
+                      protected void apply(StreamQuery query) {
+                          query.and(eq(ZipkinSpanRecord.TRACE_ID, traceId));
+                          query.setLimit(QUERY_MAX_SIZE);
+                      }
+                  }
+            );
+
+        List<Span> trace = new ArrayList<>(resp.getElements().size());
+
+        for (final RowEntity rowEntity : resp.getElements()) {
+            ZipkinSpanRecord spanRecord = new ZipkinSpanRecord.Builder().storage2Entity(
+                new BanyanDBConverter.StorageToStream(ZipkinSpanRecord.INDEX_NAME, rowEntity));
+            trace.add(ZipkinSpanRecord.buildSpanFromRecord(spanRecord));
+        }
+
+        return trace;
+    }
+
+    @Override
+    public List<List<Span>> getTraces(final QueryRequest request, Duration duration) throws IOException {
+        final int tracesLimit = request.limit();
+        int scrollLimit = 1000;
+        int scrollFrom = 0;
+        Set<String> traceIds = new HashSet<>();
+        while (traceIds.size() < tracesLimit) {
+            Set<String> resp = getTraceIds(request, duration, scrollFrom, scrollLimit);
+            if (resp.size() == 0) {
+                break;
+            }
+            for (String traceId : resp) {
+                traceIds.add(traceId);
+                if (traceIds.size() >= tracesLimit) {
+                    break;
+                }
+            }
+            scrollFrom = scrollFrom + scrollLimit;
+        }
+
+        return getTraces(traceIds);
+    }
+
+    private Set<String> getTraceIds(final QueryRequest request,
+                                    Duration duration,
+                                    int from,
+                                    int limit) throws IOException {
+        final long startTimeMillis = duration.getStartTimestamp();
+        final long endTimeMillis = duration.getEndTimestamp();
+
+        TimestampRange tsRange = null;
+        if (startTimeMillis > 0 && endTimeMillis > 0) {
+            tsRange = new TimestampRange(startTimeMillis, endTimeMillis);
+        }
+        final QueryBuilder<StreamQuery> queryBuilder = new QueryBuilder<StreamQuery>() {
+
+            @Override
+            public void apply(final StreamQuery query) {
+                if (!StringUtil.isEmpty(request.serviceName())) {
+                    query.and(eq(ZipkinSpanRecord.LOCAL_ENDPOINT_SERVICE_NAME, request.serviceName()));
+                }
+
+                if (!StringUtil.isEmpty(request.remoteServiceName())) {
+                    query.and(eq(ZipkinSpanRecord.REMOTE_ENDPOINT_SERVICE_NAME, request.remoteServiceName()));
+                }
+
+                if (!StringUtil.isEmpty(request.spanName())) {
+                    query.and(eq(ZipkinSpanRecord.NAME, request.spanName()));
+                }
+
+                if (!CollectionUtils.isEmpty(request.annotationQuery())) {
+                    for (Map.Entry<String, String> annotation : request.annotationQuery().entrySet()) {
+                        if (annotation.getValue().isEmpty()) {
+                            query.and(eq(ZipkinSpanRecord.QUERY, annotation.getKey()));
+                        } else {
+                            query.and(eq(ZipkinSpanRecord.QUERY, annotation.getKey() + "=" + annotation.getValue()));
+                        }
+                    }
+                }
+
+                if (request.minDuration() != null) {
+                    query.and(gte(ZipkinSpanRecord.DURATION, request.minDuration()));
+                }
+                if (request.maxDuration() != null) {
+                    query.and(lte(ZipkinSpanRecord.DURATION, request.maxDuration()));
+                }
+                query.setOrderBy(new StreamQuery.OrderBy(ZipkinSpanRecord.TIMESTAMP_MILLIS, AbstractQuery.Sort.DESC));
+                query.setLimit(limit);
+                query.setOffset(from);
+            }
+        };
+        StreamQueryResponse resp = query(ZipkinSpanRecord.INDEX_NAME, TRACE_TAGS, tsRange, queryBuilder);
+        Set<String> traceIds = new LinkedHashSet<>(); //needs to keep order here
+        for (final RowEntity rowEntity : resp.getElements()) {
+            traceIds.add(rowEntity.getTagValue(ZipkinSpanRecord.TRACE_ID));
+        }
+        return traceIds;
+    }
+
+    @Override
+    public List<List<Span>> getTraces(final Set<String> traceIds) throws IOException {
+        if (CollectionUtils.isEmpty(traceIds)) {
+            return Collections.EMPTY_LIST;
+        }
+        List<AbstractCriteria> conditions = new ArrayList<>(traceIds.size());
+        StreamQueryResponse resp =
+            query(ZipkinSpanRecord.INDEX_NAME, TRACE_TAGS,
+                  new QueryBuilder<StreamQuery>() {
+
+                      @Override
+                      protected void apply(StreamQuery query) {
+                          for (String traceId : traceIds) {
+                              conditions.add(eq(ZipkinSpanRecord.TRACE_ID, traceId));
+                          }
+                          if (conditions.size() == 1) {
+                              query.criteria(conditions.get(0));
+                          } else if (conditions.size() > 1) {
+                              query.criteria(or(conditions));
+                          }
+                          query.setOrderBy(
+                              new StreamQuery.OrderBy(ZipkinSpanRecord.TIMESTAMP_MILLIS, AbstractQuery.Sort.DESC));
+                          query.setLimit(QUERY_MAX_SIZE);
+                      }
+                  }
+            );
+        return buildTraces(resp);
+    }
+
+    private List<List<Span>> buildTraces(StreamQueryResponse resp) {
+        Map<String, List<Span>> groupedByTraceId = new LinkedHashMap<String, List<Span>>();
+        for (final RowEntity rowEntity : resp.getElements()) {
+            ZipkinSpanRecord spanRecord = new ZipkinSpanRecord.Builder().storage2Entity(
+                new BanyanDBConverter.StorageToStream(ZipkinSpanRecord.INDEX_NAME, rowEntity));
+            Span span = ZipkinSpanRecord.buildSpanFromRecord(spanRecord);
+            String traceId = span.traceId();
+            groupedByTraceId.putIfAbsent(traceId, new ArrayList<>());
+            groupedByTraceId.get(traceId).add(span);
+        }
+        return new ArrayList<>(groupedByTraceId.values());
+    }
+}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBZipkinQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBZipkinQueryDAO.java
deleted file mode 100644
index 9da423e02d8b..000000000000
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBZipkinQueryDAO.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
-import org.apache.skywalking.oap.server.core.query.input.Duration;
-import org.apache.skywalking.oap.server.core.storage.query.IZipkinQueryDAO;
-import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
-import zipkin2.Span;
-import zipkin2.storage.QueryRequest;
-
-//TODO: Not support BanyanDB for query yet.
-public class BanyanDBZipkinQueryDAO extends AbstractBanyanDBDAO implements IZipkinQueryDAO {
-
-    public BanyanDBZipkinQueryDAO(BanyanDBStorageClient client) {
-        super(client);
-    }
-
-    @Override
-    public List<String> getServiceNames() throws IOException {
-        return new ArrayList<>();
-    }
-
-    @Override
-    public List<String> getRemoteServiceNames(final String serviceName) throws IOException {
-        return new ArrayList<>();
-    }
-
-    @Override
-    public List<String> getSpanNames(final String serviceName) throws IOException {
-        return new ArrayList<>();
-    }
-
-    @Override
-    public List<Span> getTrace(final String traceId) throws IOException {
-        return new ArrayList<>();
-    }
-
-    @Override
-    public List<List<Span>> getTraces(final QueryRequest request, Duration duration) throws IOException {
-        return new ArrayList<>();
-    }
-
-    @Override
-    public List<List<Span>> getTraces(final Set<String> traceIds) throws IOException {
-        return new ArrayList<>();
-    }
-}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
index 74aa1faa15be..3db659d515ad 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/zipkin/ZipkinQueryEsDAO.java
@@ -18,8 +18,6 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.zipkin;
 
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
@@ -31,6 +29,7 @@
 import org.apache.skywalking.library.elasticsearch.requests.search.Search;
 import org.apache.skywalking.library.elasticsearch.requests.search.SearchBuilder;
 import org.apache.skywalking.library.elasticsearch.requests.search.SearchParams;
+import org.apache.skywalking.library.elasticsearch.requests.search.Sort;
 import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.Aggregation;
 import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.BucketOrder;
 import org.apache.skywalking.library.elasticsearch.requests.search.aggregation.TermsAggregationBuilder;
@@ -50,7 +49,6 @@
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.IndexController;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.TimeRangeIndexNameGenerator;
-import zipkin2.Endpoint;
 import zipkin2.Span;
 import zipkin2.storage.QueryRequest;
 
@@ -153,7 +151,7 @@ public List<Span> getTrace(final String traceId) {
                     Map<String, Object> sourceAsMap = searchHit.getSource();
                     ZipkinSpanRecord record = new ZipkinSpanRecord.Builder().storage2Entity(
                         new ElasticSearchConverter.ToEntity(ZipkinSpanRecord.INDEX_NAME, sourceAsMap));
-                    trace.add(buildSpanFromRecord(record));
+                    trace.add(ZipkinSpanRecord.buildSpanFromRecord(record));
                 }
                 if (response.getHits().getHits().size() < SCROLLING_BATCH_SIZE) {
                     break;
@@ -236,7 +234,8 @@ public List<List<Span>> getTraces(final QueryRequest request, Duration duration)
     public List<List<Span>> getTraces(final Set<String> traceIds) {
         String index = IndexController.LogicIndicesRegister.getPhysicalTableName(ZipkinSpanRecord.INDEX_NAME);
         BoolQueryBuilder query = Query.bool().must(Query.terms(ZipkinSpanRecord.TRACE_ID, new ArrayList<>(traceIds)));
-        SearchBuilder search = Search.builder().query(query).size(SCROLLING_BATCH_SIZE); //max span size for 1 scroll
+        SearchBuilder search = Search.builder().query(query).sort(ZipkinSpanRecord.TIMESTAMP_MILLIS, Sort.Order.DESC)
+                                     .size(SCROLLING_BATCH_SIZE); //max span size for 1 scroll
         final SearchParams params = new SearchParams().scroll(SCROLL_CONTEXT_RETENTION);
 
         SearchResponse response = getClient().search(index, search.build(), params);
@@ -263,57 +262,10 @@ private void buildTraces(SearchResponse response, Map<String, List<Span>> groupe
             Map<String, Object> sourceAsMap = searchHit.getSource();
             ZipkinSpanRecord record = new ZipkinSpanRecord.Builder().storage2Entity(
                 new ElasticSearchConverter.ToEntity(ZipkinSpanRecord.INDEX_NAME, sourceAsMap));
-            Span span = buildSpanFromRecord(record);
+            Span span = ZipkinSpanRecord.buildSpanFromRecord(record);
             String traceId = span.traceId();
             groupedByTraceId.putIfAbsent(traceId, new ArrayList<>());
             groupedByTraceId.get(traceId).add(span);
         }
     }
-
-    private Span buildSpanFromRecord(ZipkinSpanRecord record) {
-        Span.Builder span = Span.newBuilder();
-        span.traceId(record.getTraceId());
-        span.id(record.getSpanId());
-        span.parentId(record.getParentId());
-        span.kind(Span.Kind.valueOf(record.getKind()));
-        span.timestamp(record.getTimestamp());
-        span.duration(record.getDuration());
-        span.name(record.getName());
-        //Build localEndpoint
-        Endpoint.Builder localEndpoint = Endpoint.newBuilder();
-        localEndpoint.serviceName(record.getLocalEndpointServiceName());
-        if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
-            localEndpoint.parseIp(record.getLocalEndpointIPV4());
-        } else {
-            localEndpoint.parseIp(record.getLocalEndpointIPV6());
-        }
-        localEndpoint.port(record.getLocalEndpointPort());
-        span.localEndpoint(localEndpoint.build());
-        //Build remoteEndpoint
-        Endpoint.Builder remoteEndpoint = Endpoint.newBuilder();
-        remoteEndpoint.serviceName(record.getRemoteEndpointServiceName());
-        if (!StringUtil.isEmpty(record.getLocalEndpointIPV4())) {
-            remoteEndpoint.parseIp(record.getRemoteEndpointIPV4());
-        } else {
-            remoteEndpoint.parseIp(record.getRemoteEndpointIPV6());
-        }
-        remoteEndpoint.port(record.getRemoteEndpointPort());
-        span.remoteEndpoint(remoteEndpoint.build());
-
-        //Build tags
-        JsonObject tagsJson = record.getTags();
-        if (tagsJson != null) {
-            for (Map.Entry<String, JsonElement> tag : tagsJson.entrySet()) {
-                span.putTag(tag.getKey(), tag.getValue().getAsString());
-            }
-        }
-        //Build annotation
-        JsonObject annotationJson = record.getAnnotations();
-        if (annotationJson != null) {
-            for (Map.Entry<String, JsonElement> annotation : annotationJson.entrySet()) {
-                span.addAnnotation(Long.parseLong(annotation.getKey()), annotation.getValue().getAsString());
-            }
-        }
-        return span.build();
-    }
 }
diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCZipkinQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCZipkinQueryDAO.java
index 2ab576ae7455..476b4d0f9fde 100644
--- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCZipkinQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCZipkinQueryDAO.java
@@ -143,7 +143,7 @@ public List<List<Span>> getTraces(final QueryRequest request, Duration duration)
         List<Object> condition = new ArrayList<>(5);
         List<Map.Entry<String, String>> annotations = new ArrayList<>(request.annotationQuery().entrySet());
         sql.append("select ").append(ZipkinSpanRecord.INDEX_NAME).append(".").append(ZipkinSpanRecord.TRACE_ID).append(", ")
-            .append("max(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(")").append(" from ");
+            .append("min(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(")").append(" from ");
         sql.append(ZipkinSpanRecord.INDEX_NAME);
         /**
          * This is an AdditionalEntity feature, see:
@@ -208,7 +208,7 @@ public List<List<Span>> getTraces(final QueryRequest request, Duration duration)
             }
         }
         sql.append(" group by ").append(ZipkinSpanRecord.TRACE_ID);
-        sql.append(" order by max(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(") desc");
+        sql.append(" order by min(").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(") desc");
         sql.append(" limit ").append(request.limit());
         Set<String> traceIds = new HashSet<>();
         try (Connection connection = h2Client.getConnection()) {
@@ -244,6 +244,8 @@ public List<List<Span>> getTraces(final Set<String> traceIds) throws IOException
             i++;
         }
 
+        sql.append(" order by ").append(ZipkinSpanRecord.TIMESTAMP_MILLIS).append(" desc");
+
         try (Connection connection = h2Client.getConnection()) {
             ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]));
             return buildTraces(resultSet);


Test patch:
diff --git a/test/e2e-v2/cases/zipkin/banyandb/docker-compose.yml b/test/e2e-v2/cases/zipkin/banyandb/docker-compose.yml
new file mode 100644
index 000000000000..325f9acd2a7c
--- /dev/null
+++ b/test/e2e-v2/cases/zipkin/banyandb/docker-compose.yml
@@ -0,0 +1,66 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+version: '2.1'
+
+services:
+  banyandb:
+    extends:
+      file: ../../../script/docker-compose/base-compose.yml
+      service: banyandb
+    networks:
+      - e2e
+
+  oap:
+    extends:
+      file: ../../../script/docker-compose/base-compose.yml
+      service: oap
+    environment:
+      SW_STORAGE: banyandb
+      SW_QUERY_ZIPKIN: default
+      SW_RECEIVER_ZIPKIN: default
+    expose:
+      - 9411
+    ports:
+      - 9412
+    entrypoint: ['sh', '-c', '/download-mysql.sh /skywalking/oap-libs && /skywalking/docker-entrypoint.sh']
+    networks:
+      - e2e
+    depends_on:
+      banyandb:
+        condition: service_healthy
+
+  frontend:
+    extends:
+      file: ../docker-compose-brave.yml
+      service: frontend
+    depends_on:
+      backend:
+        condition: service_healthy
+      oap:
+        condition: service_healthy
+    ports:
+      - 8081
+
+  backend:
+    extends:
+      file: ../docker-compose-brave.yml
+      service: backend
+    depends_on:
+      oap:
+        condition: service_healthy
+
+networks:
+  e2e:
diff --git a/test/e2e-v2/cases/zipkin/banyandb/e2e.yaml b/test/e2e-v2/cases/zipkin/banyandb/e2e.yaml
new file mode 100644
index 000000000000..8749cd70d7dc
--- /dev/null
+++ b/test/e2e-v2/cases/zipkin/banyandb/e2e.yaml
@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This file is used to show how to write configuration files and can be used to test.
+
+setup:
+  env: compose
+  file: docker-compose.yml
+  timeout: 20m
+  init-system-environment: ../../../script/env
+  steps:
+    - name: set PATH
+      command: export PATH=/tmp/skywalking-infra-e2e/bin:$PATH
+    - name: install yq
+      command: bash test/e2e-v2/script/prepare/setup-e2e-shell/install.sh yq
+    - name: install swctl
+      command: bash test/e2e-v2/script/prepare/setup-e2e-shell/install.sh swctl
+
+trigger:
+  action: http
+  interval: 3s
+  times: 10
+  url: http://${frontend_host}:${frontend_8081}
+  method: POST
+
+verify:
+  # verify with retry strategy
+  retry:
+    # max retry count
+    count: 20
+    # the interval between two retries, in millisecond.
+    interval: 10s
+  cases:
+    - includes:
+      - ../zipkin-cases.yaml