Repository: apache/skywalking (language: python)

Issue:
[Feature] Upgrade BanyanDB storage module
### 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

Since #9761 merged, we could kick off upgrading the OAP BanyanDB storage module.

The client lib introduces several improvements:

* `exist` endpoints of metadata registry. stream and measure `exist` methods' [return value](https://github.com/apache/skywalking-banyandb-java-client/blob/main/src/main/java/org/apache/skywalking/banyandb/v1/client/metadata/ResourceExist.java#L25
) has `hasGroup`, which will simplify the interaction of creating groups.
* `AbstractQuery`'s [`and` method](https://github.com/apache/skywalking-banyandb-java-client/blob/main/README.md#example-of-and) has the same semantic as before. We could leave it alone.
* `AbstractQuery`'s [`or` method](https://github.com/apache/skywalking-banyandb-java-client/blob/main/README.md#example-of-or) could fit the profile query scenario. 
* `Property`'s `apply`, `get` and `delete` operations support selecting tags. We could disable a UI template by updating a tag instead of a whole property.
* Tag specification supports the `index_only` flag. We could set trace segment's `searchable` tags to `index_only=true`.
* `WriteProcessor`'s `add` method could return a `CompletableFuture<Void>`. We could return it to the OAP's core module.

### Use case

_No response_

### Related issues

_No response_

### Are you willing to submit a PR?

- [ ] 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:
As @hanahmily requests, I talked with @wankai123 to take cover this after https://github.com/apache/skywalking/issues/9745 is done.
@wankai123 We could do this with https://github.com/apache/skywalking/issues/9183 together, as it is a TODO from 0.1 integration as well.

Fix patch:
diff --git a/.licenserc.yaml b/.licenserc.yaml
index e7a33ff930b8..ec0af0005296 100644
--- a/.licenserc.yaml
+++ b/.licenserc.yaml
@@ -119,3 +119,9 @@ dependency:
     - name: com.google.flatbuffers:flatbuffers-java
       version: 1.12.0
       license: Apache-2.0
+    - name: build.buf.protoc-gen-validate:pgv-java-stub
+      version: 0.6.13
+      license: Apache-2.0
+    - name: build.buf.protoc-gen-validate:protoc-gen-validate
+      version: 0.6.13
+      license: Apache-2.0
diff --git a/dist-material/release-docs/LICENSE b/dist-material/release-docs/LICENSE
index 2a4cf10095e1..6bb498a288f9 100755
--- a/dist-material/release-docs/LICENSE
+++ b/dist-material/release-docs/LICENSE
@@ -208,6 +208,8 @@ Apache-2.0 licenses
 ========================================================================
 The following components are provided under the Apache-2.0 License. See project link for details.
 The text of each license is the standard Apache 2.0 license.
+    https://mvnrepository.com/artifact/build.buf.protoc-gen-validate/pgv-java-stub/0.6.13 Apache-2.0
+    https://mvnrepository.com/artifact/build.buf.protoc-gen-validate/protoc-gen-validate/0.6.13 Apache-2.0
     https://mvnrepository.com/artifact/com.aayushatharva.brotli4j/brotli4j/1.7.1 Apache-2.0
     https://mvnrepository.com/artifact/com.alibaba.nacos/nacos-api/1.4.2 Apache-2.0
     https://mvnrepository.com/artifact/com.alibaba.nacos/nacos-client/1.4.2 Apache-2.0
@@ -263,6 +265,7 @@ The text of each license is the standard Apache 2.0 license.
     https://mvnrepository.com/artifact/commons-codec/commons-codec/1.15 Apache-2.0
     https://mvnrepository.com/artifact/commons-io/commons-io/2.7 Apache-2.0
     https://mvnrepository.com/artifact/commons-logging/commons-logging/1.2 Apache-2.0
+    https://mvnrepository.com/artifact/commons-validator/commons-validator/1.7 Apache-2.0
     https://npmjs.com/package/d3-flame-graph/v/4.1.3 4.1.3 Apache-2.0
     https://npmjs.com/package/echarts/v/5.2.2 5.2.2 Apache-2.0
     https://mvnrepository.com/artifact/io.etcd/jetcd-common/0.5.3 Apache-2.0
@@ -618,6 +621,14 @@ The text of each license is also included in licenses/LICENSE-[project].txt.
 
     https://npmjs.com/package/robust-predicates/v/3.0.1 3.0.1 Unlicense
 
+========================================================================
+https://golang.org/LICENSE licenses
+========================================================================
+The following components are provided under the https://golang.org/LICENSE License. See project link for details.
+The text of each license is also included in licenses/LICENSE-[project].txt.
+
+    https://mvnrepository.com/artifact/com.google.re2j/re2j/1.5 https://golang.org/LICENSE
+
 ========================================================================
 https://spdx.org/licenses/MIT-0.html licenses
 ========================================================================
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 6c024b1da24f..7705fcf3f6d6 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -83,6 +83,19 @@
 * Support `sampledTrace` in LAL.
 * Support multiple rules with different names under the same layer of LAL script.
 * (Optimization) Reduce the buffer size(queue) of MAL(only) metric streams. Set L1 queue size as 1/20, L2 queue size as 1/2.
+* [**Breaking Change**] Migrate to BanyanDB v0.2.0.
+  * Adopt new OR logical operator for,
+    1. `MeasureIDs` query
+    2. `BanyanDBProfileThreadSnapshotQueryDAO` query
+    3. Multiple `Event` conditions query
+    4. Metrics query
+  * Simplify Group check and creation
+  * Partially apply `UITemplate` changes
+  * Support `index_only`
+  * Return `CompletableFuture<Void>` directly from BanyanDB client
+  * Optimize data binary parse methods in *LogQueryDAO
+  * Support different indexType
+  * Support configuration for TTL and (block|segment) intervals
 
 #### UI
 
diff --git a/docs/en/setup/backend/backend-storage.md b/docs/en/setup/backend/backend-storage.md
index 8bcbcc63a546..bac970ca185f 100644
--- a/docs/en/setup/backend/backend-storage.md
+++ b/docs/en/setup/backend/backend-storage.md
@@ -308,6 +308,10 @@ storage:
     superDatasetShardsFactor: ${SW_STORAGE_BANYANDB_SUPERDATASET_SHARDS_FACTOR:2}
     concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
     profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_PROFILE_TASK_QUERY_MAX_SIZE:200} # the max number of fetch task in a request
+    streamBlockInterval: ${SW_STORAGE_BANYANDB_STREAM_BLOCK_INTERVAL:4} # Unit is hour
+    streamSegmentInterval: ${SW_STORAGE_BANYANDB_STREAM_SEGMENT_INTERVAL:24} # Unit is hour
+    measureBlockInterval: ${SW_STORAGE_BANYANDB_MEASURE_BLOCK_INTERVAL:4} # Unit is hour
+    measureSegmentInterval: ${SW_STORAGE_BANYANDB_MEASURE_SEGMENT_INTERVAL:24} # Unit is hour
 ```
 
 For more details, please refer to the documents of [BanyanDB](https://skywalking.apache.org/docs/skywalking-banyandb/next/readme/)
diff --git a/oap-server-bom/pom.xml b/oap-server-bom/pom.xml
index bd1792d1abb3..d14d8a0a94fc 100644
--- a/oap-server-bom/pom.xml
+++ b/oap-server-bom/pom.xml
@@ -73,7 +73,8 @@
         <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.1.0</banyandb-java-client.version>
+        <!-- TODO: use release version of banyandb java client -->
+        <banyandb-java-client.version>0.2.0-SNAPSHOT</banyandb-java-client.version>
         <kafka-clients.version>2.4.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/storage/annotation/BanyanDB.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/BanyanDB.java
index 87890c5f2209..80acbdfc426d 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/BanyanDB.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/BanyanDB.java
@@ -31,12 +31,12 @@
 public @interface BanyanDB {
     /**
      * GlobalIndex declares advanced global index, which are only available in BanyanDB.
-     *
+     * <p>
      * Global index should only be considered if a column value has a huge value candidates, but we will need a direct
      * equal
      * query without timestamp.
      * The typical global index is designed for huge candidate of indexed values, such as `trace ID` or `segment ID`.
-     *
+     * <p>
      * Only work with {@link Column}
      */
     @Target({ElementType.FIELD})
@@ -51,17 +51,17 @@
      * ServiceA's traffic gauge, service call per minute, includes following timestamp values, then it should be sharded
      * by service ID
      * [ServiceA(encoded ID): 01-28 18:30 values-1, 01-28 18:31 values-2, 01-28 18:32 values-3, 01-28 18:32 values-4]
-     *
+     * <p>
      * BanyanDB is the 1st storage implementation supporting this. It would make continuous time series metrics stored
      * closely and compressed better.
-     *
+     * <p>
      * 1. One entity could have multiple sharding keys
      * 2. If no column is appointed for this, {@link org.apache.skywalking.oap.server.core.storage.StorageData#id}
      * would be used by the storage implementation accordingly.
-     *
+     * <p>
      * NOTICE, this sharding concept is NOT just for splitting data into different database instances or physical
      * files.
-     *
+     * <p>
      * Only work with {@link Column}
      *
      * @return non-negative if this column be used for sharding. -1 means not as a sharding key
@@ -91,4 +91,31 @@
     @interface NoIndexing {
 
     }
+
+    /**
+     * Additional information for constructing Index in BanyanDB.
+     *
+     * @since 9.3.0
+     */
+    @Target({ElementType.FIELD})
+    @Retention(RetentionPolicy.RUNTIME)
+    @interface IndexRule {
+        /**
+         * IndexRule supports selecting two distinct kinds of index structures.
+         */
+        IndexType indexType() default IndexType.INVERTED;
+
+        enum IndexType {
+            /**
+             * The `INVERTED` index is the primary option when users set up an index rule.
+             * It's suitable for most tag indexing due to a better memory usage ratio and query performance.
+             */
+            INVERTED,
+            /**
+             * The `TREE` index could be better when there are high cardinalities, such as the `ID` tag and numeric duration tag.
+             * In these cases, it saves much memory space.
+             */
+            TREE;
+        }
+    }
 }
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java
index fe90b85ef9d1..d1fdb008c870 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/BanyanDBExtension.java
@@ -55,6 +55,14 @@ public class BanyanDBExtension {
      */
     private final boolean shouldIndex;
 
+    /**
+     * indexType is the type of index built for a {@link ModelColumn} in BanyanDB.
+     *
+     * @since 9.3.0
+     */
+    @Getter
+    private final BanyanDB.IndexRule.IndexType indexType;
+
     /**
      * @return true if this column is a part of sharding key
      */
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
index 7b8ce242e14d..b0aee436784c 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
@@ -203,10 +203,13 @@ private void retrieval(final Class<?> clazz,
                     BanyanDB.GlobalIndex.class);
                 final BanyanDB.NoIndexing banyanDBNoIndex = field.getAnnotation(
                     BanyanDB.NoIndexing.class);
+                final BanyanDB.IndexRule banyanDBIndexRule = field.getAnnotation(
+                        BanyanDB.IndexRule.class);
                 BanyanDBExtension banyanDBExtension = new BanyanDBExtension(
                     banyanDBShardingKey == null ? -1 : banyanDBShardingKey.index(),
-                    banyanDBGlobalIndex == null ? false : true,
-                    banyanDBNoIndex != null ? false : column.storageOnly()
+                    banyanDBGlobalIndex != null,
+                    banyanDBNoIndex == null && column.storageOnly(),
+                    banyanDBIndexRule == null ? BanyanDB.IndexRule.IndexType.INVERTED : banyanDBIndexRule.indexType()
                 );
 
                 final ModelColumn modelColumn = new ModelColumn(
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java
index 0f0f51e98e3a..593057067a33 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/IBrowserLogQueryDAO.java
@@ -36,15 +36,18 @@ BrowserErrorLogs queryBrowserErrorLogs(String serviceId,
                                            int limit,
                                            int from) throws IOException;
 
+    default BrowserErrorLog parserDataBinary(String dataBinaryBase64) {
+        return parserDataBinary(Base64.getDecoder().decode(dataBinaryBase64));
+    }
+
     /**
      * Parser the raw error log.
      */
-    default BrowserErrorLog parserDataBinary(
-        String dataBinaryBase64) {
+    default BrowserErrorLog parserDataBinary(byte[] dataBinary) {
         try {
             BrowserErrorLog log = new BrowserErrorLog();
             org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog browserErrorLog = org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog
-                .parseFrom(Base64.getDecoder().decode(dataBinaryBase64));
+                .parseFrom(dataBinary);
 
             log.setService(browserErrorLog.getService());
             log.setServiceVersion(browserErrorLog.getServiceVersion());
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java
index f4365fbbacbb..3e412d5178a6 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ILogQueryDAO.java
@@ -19,9 +19,11 @@
 package org.apache.skywalking.oap.server.core.storage.query;
 
 import com.google.protobuf.InvalidProtocolBufferException;
+
 import java.io.IOException;
 import java.util.Base64;
 import java.util.List;
+
 import org.apache.skywalking.apm.network.logging.v3.LogTags;
 import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.Tag;
 import org.apache.skywalking.oap.server.core.query.enumeration.Order;
@@ -50,11 +52,15 @@ Logs queryLogs(String serviceId,
                    final List<String> excludingKeywordsOfContent) throws IOException;
 
     /**
-     * Parser the raw tags.
+     * Parse the raw tags with base64 representation of data binary
      */
     default void parserDataBinary(String dataBinaryBase64, List<KeyValue> tags) {
+        parserDataBinary(Base64.getDecoder().decode(dataBinaryBase64), tags);
+    }
+
+    default void parserDataBinary(byte[] dataBinary, List<KeyValue> tags) {
         try {
-            LogTags logTags = LogTags.parseFrom(Base64.getDecoder().decode(dataBinaryBase64));
+            LogTags logTags = LogTags.parseFrom(dataBinary);
             logTags.getDataList().forEach(pair -> tags.add(new KeyValue(pair.getKey(), pair.getValue())));
         } catch (InvalidProtocolBufferException e) {
             throw new RuntimeException(e);
diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml
index 751c0380451b..06e7eb565e62 100644
--- a/oap-server/server-starter/src/main/resources/application.yml
+++ b/oap-server/server-starter/src/main/resources/application.yml
@@ -236,6 +236,10 @@ storage:
     superDatasetShardsFactor: ${SW_STORAGE_BANYANDB_SUPERDATASET_SHARDS_FACTOR:2}
     concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
     profileTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_PROFILE_TASK_QUERY_MAX_SIZE:200} # the max number of fetch task in a request
+    streamBlockInterval: ${SW_STORAGE_BANYANDB_STREAM_BLOCK_INTERVAL:4} # Unit is hour
+    streamSegmentInterval: ${SW_STORAGE_BANYANDB_STREAM_SEGMENT_INTERVAL:24} # Unit is hour
+    measureBlockInterval: ${SW_STORAGE_BANYANDB_MEASURE_BLOCK_INTERVAL:4} # Unit is hour
+    measureSegmentInterval: ${SW_STORAGE_BANYANDB_MEASURE_SEGMENT_INTERVAL:24} # Unit is hour
 
 agent-analyzer:
   selector: ${SW_AGENT_ANALYZER:default}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java
index 0928361ef509..58d61c9831f6 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java
@@ -77,7 +77,7 @@ protected void apply(MeasureQuery query) {
             return Collections.emptyList();
         }
 
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName, duration.getStep());
         if (schema == null) {
             throw new IOException("schema is not registered");
         }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java
index 0e6d761aa991..9564de27dad1 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBBatchDAO.java
@@ -31,6 +31,7 @@
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
 
 public class BanyanDBBatchDAO extends AbstractDAO<BanyanDBStorageClient> implements IBatchDAO {
     private static final Object STREAM_SYNCHRONIZER = new Object();
@@ -64,16 +65,16 @@ public void insert(InsertRequest insertRequest) {
     @Override
     public CompletableFuture<Void> flush(List<PrepareRequest> prepareRequests) {
         if (CollectionUtils.isNotEmpty(prepareRequests)) {
-            for (final PrepareRequest r : prepareRequests) {
+            return CompletableFuture.allOf(prepareRequests.stream().map((Function<PrepareRequest, CompletableFuture<Void>>) r -> {
                 if (r instanceof BanyanDBStreamInsertRequest) {
-                    // TODO: return CompletableFuture<Void>
-                    getStreamBulkWriteProcessor().add(((BanyanDBStreamInsertRequest) r).getStreamWrite());
+                    return getStreamBulkWriteProcessor().add(((BanyanDBStreamInsertRequest) r).getStreamWrite());
                 } else if (r instanceof BanyanDBMeasureInsertRequest) {
-                    getMeasureBulkWriteProcessor().add(((BanyanDBMeasureInsertRequest) r).getMeasureWrite());
+                    return getMeasureBulkWriteProcessor().add(((BanyanDBMeasureInsertRequest) r).getMeasureWrite());
                 } else if (r instanceof BanyanDBMeasureUpdateRequest) {
-                    getMeasureBulkWriteProcessor().add(((BanyanDBMeasureUpdateRequest) r).getMeasureWrite());
+                    return getMeasureBulkWriteProcessor().add(((BanyanDBMeasureUpdateRequest) r).getMeasureWrite());
                 }
-            }
+                return CompletableFuture.completedFuture(null);
+            }).toArray(CompletableFuture[]::new));
         }
 
         return CompletableFuture.completedFuture(null);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java
index c4ddc13c8f23..e40c6b70155e 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBConverter.java
@@ -41,8 +41,8 @@ public static class StorageToStream implements Convert2Entity {
         private final MetadataRegistry.Schema schema;
         private final RowEntity rowEntity;
 
-        public StorageToStream(String modelName, RowEntity rowEntity) {
-            this.schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+        public StorageToStream(String streamModelName, RowEntity rowEntity) {
+            this.schema = MetadataRegistry.INSTANCE.findRecordMetadata(streamModelName);
             this.rowEntity = rowEntity;
         }
 
@@ -215,8 +215,8 @@ public static class StorageToMeasure implements Convert2Entity {
         private final MetadataRegistry.Schema schema;
         private final DataPoint dataPoint;
 
-        public StorageToMeasure(String modelName, DataPoint dataPoint) {
-            this.schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+        public StorageToMeasure(MetadataRegistry.Schema schema, DataPoint dataPoint) {
+            this.schema = schema;
             this.dataPoint = dataPoint;
         }
 
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java
index e2f281a0b8a0..aabe4dbd499a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java
@@ -21,9 +21,10 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.banyandb.v1.client.BanyanDBClient;
 import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
-import org.apache.skywalking.banyandb.v1.client.metadata.Group;
 import org.apache.skywalking.banyandb.v1.client.metadata.Measure;
 import org.apache.skywalking.banyandb.v1.client.metadata.Stream;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.config.ConfigService;
 import org.apache.skywalking.oap.server.core.storage.StorageException;
 import org.apache.skywalking.oap.server.core.storage.model.Model;
 import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller;
@@ -35,29 +36,37 @@
 @Slf4j
 public class BanyanDBIndexInstaller extends ModelInstaller {
     private final BanyanDBStorageConfig config;
+    private final ConfigService configService;
 
     public BanyanDBIndexInstaller(Client client, ModuleManager moduleManager, BanyanDBStorageConfig config) {
         super(client, moduleManager);
         this.config = config;
+        this.configService = moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class);
     }
 
     @Override
     protected boolean isExists(Model model) throws StorageException {
-        final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(model, config);
+        final MetadataRegistry.SchemaMetadata metadata = MetadataRegistry.INSTANCE.parseMetadata(model, config, configService);
         try {
             final BanyanDBClient c = ((BanyanDBStorageClient) this.client).client;
-            // first check group
-            Group g = metadata.getOrCreateGroup(c);
-            if (g == null) {
-                throw new StorageException("fail to create group " + metadata.getGroup());
+            // first check resource existence and create group if necessary
+            final boolean resourceExist = metadata.checkResourceExistence(c);
+            if (!resourceExist) {
+                return false;
             }
-            log.info("group {} created", g.name());
+
             // then check entity schema
             if (metadata.findRemoteSchema(c).isPresent()) {
-                MetadataRegistry.INSTANCE.registerModel(model, config);
+                // register models only locally but not remotely
+                if (model.isTimeSeries() && model.isRecord()) { // stream
+                    MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
+                } else if (model.isTimeSeries() && !model.isRecord()) { // measure
+                    MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
+                }
                 return true;
             }
-            return false;
+
+            throw new IllegalStateException("inconsistent state");
         } catch (BanyanDBException ex) {
             throw new StorageException("fail to check existence", ex);
         }
@@ -67,13 +76,13 @@ protected boolean isExists(Model model) throws StorageException {
     protected void createTable(Model model) throws StorageException {
         try {
             if (model.isTimeSeries() && model.isRecord()) { // stream
-                Stream stream = (Stream) MetadataRegistry.INSTANCE.registerModel(model, config);
+                Stream stream = MetadataRegistry.INSTANCE.registerStreamModel(model, config, configService);
                 if (stream != null) {
                     log.info("install stream schema {}", model.getName());
                     ((BanyanDBStorageClient) client).define(stream);
                 }
             } else if (model.isTimeSeries() && !model.isRecord()) { // measure
-                Measure measure = (Measure) MetadataRegistry.INSTANCE.registerModel(model, config);
+                Measure measure = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, configService);
                 if (measure != null) {
                     log.info("install measure schema {}", model.getName());
                     ((BanyanDBStorageClient) client).define(measure);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java
index 316e0af413e9..12b03a94eb0a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBNoneStreamDAO.java
@@ -41,12 +41,12 @@ public BanyanDBNoneStreamDAO(BanyanDBStorageClient client, StorageBuilder<NoneSt
 
     @Override
     public void insert(Model model, NoneStream noneStream) throws IOException {
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findRecordMetadata(model.getName());
         if (schema == null) {
             throw new IOException(model.getName() + " is not registered");
         }
         StreamWrite streamWrite = new StreamWrite(schema.getMetadata().getGroup(), // group name
-                model.getName(), // index-name
+                schema.getMetadata().name(), // stream-name
                 noneStream.id(), // identity
                 TimeBucket.getTimestamp(noneStream.getTimeBucket(), model.getDownsampling())); // timestamp
         Convert2Storage<StreamWrite> convert2Storage = new BanyanDBConverter.StreamToStorage(schema, streamWrite);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java
index 3955fac14226..d099a0a41994 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java
@@ -71,7 +71,7 @@ protected void apply(MeasureQuery query) {
             return Collections.emptyList();
         }
 
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findRecordMetadata(modelName);
         if (schema == null) {
             throw new IOException("schema is not registered");
         }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
index 9ba378125c8b..f2408f86952a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
@@ -118,7 +118,7 @@ public MeasureQueryResponse query(MeasureQuery q) throws IOException {
 
     public void define(Property property) throws IOException {
         try {
-            this.client.save(property);
+            this.client.apply(property);
             this.healthChecker.health();
         } catch (BanyanDBException ex) {
             healthChecker.unHealth(ex);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
index e31874e7f53b..3a6569135b30 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
@@ -56,4 +56,32 @@ public class BanyanDBStorageConfig extends ModuleConfig {
      * Shards Factor for a super dataset
      */
     private int superDatasetShardsFactor;
+    /**
+     * block interval for Stream group.
+     * Unit is hours.
+     *
+     * @since 9.3.0
+     */
+    private int streamBlockInterval;
+    /**
+     * segment interval for Stream group.
+     * Unit is hours.
+     *
+     * @since 9.3.0
+     */
+    private int streamSegmentInterval;
+    /**
+     * block interval for Measure group.
+     * Unit is hours.
+     *
+     * @since 9.3.0
+     */
+    private int measureBlockInterval;
+    /**
+     * segment interval for Measure group.
+     * Unit is hours.
+     *
+     * @since 9.3.0
+     */
+    private int measureSegmentInterval;
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java
index ab0dd31f7108..d616ed032e0f 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBUITemplateManagementDAO.java
@@ -26,7 +26,6 @@
 import org.apache.skywalking.oap.server.core.query.type.DashboardConfiguration;
 import org.apache.skywalking.oap.server.core.query.type.TemplateChangeStatus;
 import org.apache.skywalking.oap.server.core.storage.management.UITemplateManagementDAO;
-import org.apache.skywalking.oap.server.library.util.BooleanUtils;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
 
 import java.io.IOException;
@@ -60,7 +59,7 @@ public List<DashboardConfiguration> getAllTemplates(Boolean includingDisabled) t
 
     @Override
     public TemplateChangeStatus addTemplate(DashboardSetting setting) {
-        Property newTemplate = convert(setting.toEntity());
+        Property newTemplate = applyAll(setting.toEntity());
         try {
             this.getClient().define(newTemplate);
             return TemplateChangeStatus.builder()
@@ -76,7 +75,7 @@ public TemplateChangeStatus addTemplate(DashboardSetting setting) {
 
     @Override
     public TemplateChangeStatus changeTemplate(DashboardSetting setting) {
-        Property newTemplate = convert(setting.toEntity());
+        Property newTemplate = applyConfiguration(setting.toEntity());
         try {
             this.getClient().define(newTemplate);
             return TemplateChangeStatus.builder()
@@ -98,9 +97,8 @@ public TemplateChangeStatus disableTemplate(String id) throws IOException {
                     .build();
         }
         UITemplate uiTemplate = parse(oldProperty);
-        uiTemplate.setDisabled(BooleanUtils.FALSE);
         try {
-            this.getClient().define(convert(uiTemplate));
+            this.getClient().define(applyStatus(uiTemplate));
             return TemplateChangeStatus.builder()
                     .status(true)
                     .id(uiTemplate.id())
@@ -134,11 +132,37 @@ public UITemplate parse(Property property) {
         return uiTemplate;
     }
 
-    public Property convert(UITemplate uiTemplate) {
+    public Property applyAll(UITemplate uiTemplate) {
         return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id())
                 .addTag(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration()))
                 .addTag(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled()))
                 .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()))
                 .build();
     }
+
+    /**
+     * Partial apply status, i.e. disable tag.
+     *
+     * @param uiTemplate previous UITemplate
+     * @return new property (patch) to be applied
+     */
+    public Property applyStatus(UITemplate uiTemplate) {
+        return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id())
+                .addTag(TagAndValue.newLongTag(UITemplate.DISABLED, uiTemplate.getDisabled()))
+                .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()))
+                .build();
+    }
+
+    /**
+     * Partial apply configuration, i.e. configuration tag.
+     *
+     * @param uiTemplate previous UITemplate
+     * @return new property (patch) to be applied
+     */
+    public Property applyConfiguration(UITemplate uiTemplate) {
+        return Property.create(GROUP, UITemplate.INDEX_NAME, uiTemplate.id())
+                .addTag(TagAndValue.newStringTag(UITemplate.CONFIGURATION, uiTemplate.getConfiguration()))
+                .addTag(TagAndValue.newLongTag(UITemplate.UPDATE_TIME, uiTemplate.getUpdateTime()))
+                .build();
+    }
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java
index 9287d5c710fe..9f6f2e4cf8b3 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/MetadataRegistry.java
@@ -33,18 +33,23 @@
 import org.apache.skywalking.banyandb.v1.client.metadata.Duration;
 import org.apache.skywalking.banyandb.v1.client.metadata.Group;
 import org.apache.skywalking.banyandb.v1.client.metadata.IndexRule;
+import org.apache.skywalking.banyandb.v1.client.metadata.IntervalRule;
 import org.apache.skywalking.banyandb.v1.client.metadata.Measure;
 import org.apache.skywalking.banyandb.v1.client.metadata.NamedSchema;
+import org.apache.skywalking.banyandb.v1.client.metadata.ResourceExist;
 import org.apache.skywalking.banyandb.v1.client.metadata.Stream;
 import org.apache.skywalking.banyandb.v1.client.metadata.TagFamilySpec;
 import org.apache.skywalking.oap.server.core.analysis.DownSampling;
 import org.apache.skywalking.oap.server.core.analysis.metrics.IntList;
+import org.apache.skywalking.oap.server.core.config.ConfigService;
+import org.apache.skywalking.oap.server.core.query.enumeration.Step;
+import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
 import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetadata;
 import org.apache.skywalking.oap.server.core.storage.model.Model;
 import org.apache.skywalking.oap.server.core.storage.model.ModelColumn;
 import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject;
 
-import javax.annotation.Nullable;
+import javax.annotation.Nonnull;
 import java.lang.reflect.ParameterizedType;
 import java.util.ArrayList;
 import java.util.Comparator;
@@ -63,8 +68,8 @@ public enum MetadataRegistry {
 
     private final Map<String, Schema> registry = new HashMap<>();
 
-    public NamedSchema<?> registerModel(Model model, BanyanDBStorageConfig config) {
-        final SchemaMetadata schemaMetadata = parseMetadata(model, config);
+    public Stream registerStreamModel(Model model, BanyanDBStorageConfig config, ConfigService configService) {
+        final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
         Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
         Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
                 .collect(Collectors.toMap(modelColumn -> modelColumn.getColumnName().getStorageName(), Function.identity()));
@@ -87,38 +92,100 @@ public NamedSchema<?> registerModel(Model model, BanyanDBStorageConfig config) {
                 .filter(Objects::nonNull)
                 .collect(Collectors.toList());
 
-        if (schemaMetadata.getKind() == Kind.STREAM) {
-            final Stream.Builder builder = Stream.create(schemaMetadata.getGroup(), schemaMetadata.getName());
-            if (entities.isEmpty()) {
-                throw new IllegalStateException("sharding keys of model[stream." + model.getName() + "] must not be empty");
+        final Stream.Builder builder = Stream.create(schemaMetadata.getGroup(), schemaMetadata.name());
+        if (entities.isEmpty()) {
+            throw new IllegalStateException("sharding keys of model[stream." + model.getName() + "] must not be empty");
+        }
+        builder.setEntityRelativeTags(entities);
+        builder.addTagFamilies(tagFamilySpecs);
+        builder.addIndexes(indexRules);
+        registry.put(schemaMetadata.name(), schemaBuilder.build());
+        return builder.build();
+    }
+
+    public Measure registerMeasureModel(Model model, BanyanDBStorageConfig config, ConfigService configService) {
+        final SchemaMetadata schemaMetadata = parseMetadata(model, config, configService);
+        Schema.SchemaBuilder schemaBuilder = Schema.builder().metadata(schemaMetadata);
+        Map<String, ModelColumn> modelColumnMap = model.getColumns().stream()
+                .collect(Collectors.toMap(modelColumn -> modelColumn.getColumnName().getStorageName(), Function.identity()));
+        // parse and set sharding keys
+        List<String> entities = parseEntityNames(modelColumnMap);
+        // parse tag metadata
+        // this can be used to build both
+        // 1) a list of TagFamilySpec,
+        // 2) a list of IndexRule,
+        List<TagMetadata> tags = parseTagMetadata(model, schemaBuilder);
+        List<TagFamilySpec> tagFamilySpecs = schemaMetadata.extractTagFamilySpec(tags);
+        // iterate over tagFamilySpecs to save tag names
+        for (final TagFamilySpec tagFamilySpec : tagFamilySpecs) {
+            for (final TagFamilySpec.TagSpec tagSpec : tagFamilySpec.tagSpecs()) {
+                schemaBuilder.tag(tagSpec.getTagName());
             }
-            builder.setEntityRelativeTags(entities);
-            builder.addTagFamilies(tagFamilySpecs);
-            builder.addIndexes(indexRules);
-            registry.put(model.getName(), schemaBuilder.build());
-            return builder.build();
+        }
+        List<IndexRule> indexRules = tags.stream()
+                .map(TagMetadata::getIndexRule)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toList());
+
+        final Measure.Builder builder = Measure.create(schemaMetadata.getGroup(), schemaMetadata.name(),
+                downSamplingDuration(model.getDownsampling()));
+        if (entities.isEmpty()) { // if shardingKeys is empty, for measure, we can use ID as a single sharding key.
+            builder.setEntityRelativeTags(Measure.ID);
         } else {
-            final Measure.Builder builder = Measure.create(schemaMetadata.getGroup(), schemaMetadata.getName(),
-                    downSamplingDuration(model.getDownsampling()));
-            if (entities.isEmpty()) { // if shardingKeys is empty, for measure, we can use ID as a single sharding key.
-                builder.setEntityRelativeTags(Measure.ID);
-            } else {
-                builder.setEntityRelativeTags(entities);
-            }
-            builder.addTagFamilies(tagFamilySpecs);
-            builder.addIndexes(indexRules);
-            // parse and set field
-            Optional<ValueColumnMetadata.ValueColumn> valueColumnOpt = ValueColumnMetadata.INSTANCE
-                    .readValueColumnDefinition(model.getName());
-            valueColumnOpt.ifPresent(valueColumn -> builder.addField(parseFieldSpec(modelColumnMap.get(valueColumn.getValueCName()), valueColumn)));
-            valueColumnOpt.ifPresent(valueColumn -> schemaBuilder.field(valueColumn.getValueCName()));
-            registry.put(model.getName(), schemaBuilder.build());
-            return builder.build();
+            builder.setEntityRelativeTags(entities);
         }
+        builder.addTagFamilies(tagFamilySpecs);
+        builder.addIndexes(indexRules);
+        // parse and set field
+        Optional<ValueColumnMetadata.ValueColumn> valueColumnOpt = ValueColumnMetadata.INSTANCE
+                .readValueColumnDefinition(model.getName());
+        valueColumnOpt.ifPresent(valueColumn -> builder.addField(parseFieldSpec(modelColumnMap.get(valueColumn.getValueCName()), valueColumn)));
+        valueColumnOpt.ifPresent(valueColumn -> schemaBuilder.field(valueColumn.getValueCName()));
+        registry.put(schemaMetadata.name(), schemaBuilder.build());
+        return builder.build();
     }
 
-    public Schema findMetadata(final String name) {
-        return this.registry.get(name);
+    public Schema findMetadata(final Model model) {
+        if (model.isRecord()) {
+            return findRecordMetadata(model.getName());
+        }
+        return findMetadata(model.getName(), model.getDownsampling());
+    }
+
+    public Schema findRecordMetadata(final String recordModelName) {
+        final Schema s = this.registry.get(recordModelName);
+        if (s == null) {
+            return null;
+        }
+        // impose sanity check
+        if (s.getMetadata().getKind() != Kind.STREAM) {
+            throw new IllegalArgumentException(recordModelName + "is not a record");
+        }
+        return s;
+    }
+
+    static DownSampling deriveFromStep(Step step) {
+        switch (step) {
+            case DAY:
+                return DownSampling.Day;
+            case HOUR:
+                return DownSampling.Hour;
+            case SECOND:
+                return DownSampling.Second;
+            default:
+                return DownSampling.Minute;
+        }
+    }
+
+    public Schema findMetadata(final String modelName, Step step) {
+        return findMetadata(modelName, deriveFromStep(step));
+    }
+
+    /**
+     * Find metadata with down-sampling
+     */
+    public Schema findMetadata(final String modelName, DownSampling downSampling) {
+        return this.registry.get(SchemaMetadata.formatName(modelName, downSampling));
     }
 
     private Measure.FieldSpec parseFieldSpec(ModelColumn modelColumn, ValueColumnMetadata.ValueColumn valueColumn) {
@@ -167,10 +234,17 @@ IndexRule parseIndexRule(String tagName, ModelColumn modelColumn) {
             return IndexRule.create(tagName, IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.SERIES);
         }
         if (modelColumn.getBanyanDBExtension().isGlobalIndexing()) {
-            return IndexRule.create(tagName, IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.GLOBAL);
+            return IndexRule.create(tagName, parseIndexType(modelColumn.getBanyanDBExtension().getIndexType()), IndexRule.IndexLocation.GLOBAL);
         } else {
-            return IndexRule.create(tagName, IndexRule.IndexType.INVERTED, IndexRule.IndexLocation.SERIES);
+            return IndexRule.create(tagName, parseIndexType(modelColumn.getBanyanDBExtension().getIndexType()), IndexRule.IndexLocation.SERIES);
+        }
+    }
+
+    private static IndexRule.IndexType parseIndexType(BanyanDB.IndexRule.IndexType indexType) {
+        if (indexType == BanyanDB.IndexRule.IndexType.INVERTED) {
+            return IndexRule.IndexType.INVERTED;
         }
+        return IndexRule.IndexType.TREE;
     }
 
     /**
@@ -203,9 +277,6 @@ List<TagMetadata> parseTagMetadata(Model model, Schema.SchemaBuilder builder) {
                 continue;
             }
             final TagFamilySpec.TagSpec tagSpec = parseTagSpec(col);
-            if (tagSpec == null) {
-                continue;
-            }
             builder.spec(col.getColumnName().getStorageName(), new ColumnSpec(ColumnType.TAG, col.getType()));
             if (col.shouldIndex()) {
                 // build indexRule
@@ -225,33 +296,40 @@ List<TagMetadata> parseTagMetadata(Model model, Schema.SchemaBuilder builder) {
      * @param modelColumn the column in the model to be parsed
      * @return a typed tag spec
      */
-    @Nullable
+    @Nonnull
     private TagFamilySpec.TagSpec parseTagSpec(ModelColumn modelColumn) {
         final Class<?> clazz = modelColumn.getType();
         final String colName = modelColumn.getColumnName().getStorageName();
+        TagFamilySpec.TagSpec tagSpec = null;
         if (String.class.equals(clazz) || StorageDataComplexObject.class.isAssignableFrom(clazz) || JsonObject.class.equals(clazz)) {
-            return TagFamilySpec.TagSpec.newStringTag(colName);
+            tagSpec = TagFamilySpec.TagSpec.newStringTag(colName);
         } else if (int.class.equals(clazz) || long.class.equals(clazz)) {
-            return TagFamilySpec.TagSpec.newIntTag(colName);
+            tagSpec = TagFamilySpec.TagSpec.newIntTag(colName);
         } else if (byte[].class.equals(clazz)) {
-            return TagFamilySpec.TagSpec.newBinaryTag(colName);
+            tagSpec = TagFamilySpec.TagSpec.newBinaryTag(colName);
         } else if (clazz.isEnum()) {
-            return TagFamilySpec.TagSpec.newIntTag(colName);
+            tagSpec = TagFamilySpec.TagSpec.newIntTag(colName);
         } else if (double.class.equals(clazz) || Double.class.equals(clazz)) {
             // serialize double as binary
-            return TagFamilySpec.TagSpec.newBinaryTag(colName);
+            tagSpec = TagFamilySpec.TagSpec.newBinaryTag(colName);
         } else if (IntList.class.isAssignableFrom(clazz)) {
-            return TagFamilySpec.TagSpec.newIntArrayTag(colName);
+            tagSpec = TagFamilySpec.TagSpec.newIntArrayTag(colName);
         } else if (List.class.isAssignableFrom(clazz)) { // handle exceptions
             ParameterizedType t = (ParameterizedType) modelColumn.getGenericType();
             if (String.class.equals(t.getActualTypeArguments()[0])) {
-                return TagFamilySpec.TagSpec.newStringArrayTag(colName);
+                tagSpec = TagFamilySpec.TagSpec.newStringArrayTag(colName);
             }
         }
-        throw new IllegalStateException("type " + modelColumn.getType().toString() + " is not supported");
+        if (tagSpec == null) {
+            throw new IllegalStateException("type " + modelColumn.getType().toString() + " is not supported");
+        }
+        if (modelColumn.isIndexOnly()) {
+            tagSpec.indexedOnly();
+        }
+        return tagSpec;
     }
 
-    public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config) {
+    public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config, ConfigService configService) {
         if (model.isRecord()) {
             String group = "stream-default";
             if (model.isSuperDataset()) {
@@ -261,29 +339,61 @@ public SchemaMetadata parseMetadata(Model model, BanyanDBStorageConfig config) {
             return new SchemaMetadata(group,
                     model.getName(),
                     Kind.STREAM,
+                    model.getDownsampling(),
                     config.getRecordShardsNumber() *
-                            (model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1)
+                            (model.isSuperDataset() ? config.getSuperDatasetShardsFactor() : 1),
+                    config.getStreamBlockInterval(),
+                    config.getStreamSegmentInterval(),
+                    configService.getRecordDataTTL()
             );
         }
-        return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE, config.getMetricsShardsNumber());
+        return new SchemaMetadata("measure-default", model.getName(), Kind.MEASURE,
+                model.getDownsampling(),
+                config.getMetricsShardsNumber(),
+                config.getMeasureBlockInterval(),
+                config.getMeasureSegmentInterval(),
+                configService.getMetricsDataTTL());
     }
 
     @RequiredArgsConstructor
     @Data
     public static class SchemaMetadata {
         private final String group;
-        private final String name;
+        /**
+         * name of the {@link Model}
+         */
+        private final String modelName;
         private final Kind kind;
-
+        /**
+         * down-sampling of the {@link Model}
+         */
+        private final DownSampling downSampling;
         private final int shard;
+        private final int blockIntervalHrs;
+        private final int segmentIntervalHrs;
+        private final int ttlDays;
+
+        /**
+         * Format the entity name for BanyanDB
+         *
+         * @param modelName    name of the model
+         * @param downSampling not used if it is {@link DownSampling#None}
+         * @return entity (e.g. measure, stream) name
+         */
+        static String formatName(String modelName, DownSampling downSampling) {
+            if (downSampling == null || downSampling == DownSampling.None) {
+                return modelName;
+            }
+            return modelName + "_" + downSampling.getName();
+        }
 
         public Optional<NamedSchema<?>> findRemoteSchema(BanyanDBClient client) throws BanyanDBException {
             try {
                 switch (kind) {
                     case STREAM:
-                        return Optional.ofNullable(client.findStream(this.group, this.name));
+                        return Optional.ofNullable(client.findStream(this.group, this.name()));
                     case MEASURE:
-                        return Optional.ofNullable(client.findMeasure(this.group, this.name));
+                        return Optional.ofNullable(client.findMeasure(this.group, this.name()));
                     default:
                         throw new IllegalStateException("should not reach here");
                 }
@@ -315,21 +425,45 @@ private List<TagFamilySpec> extractTagFamilySpec(List<TagMetadata> tagMetadataLi
             return tagFamilySpecs;
         }
 
-        public Group getOrCreateGroup(BanyanDBClient client) throws BanyanDBException {
-            Group g = client.findGroup(this.group);
-            if (g != null) {
-                return g;
-            }
+        public boolean checkResourceExistence(BanyanDBClient client) throws BanyanDBException {
+            ResourceExist resourceExist;
             switch (kind) {
                 case STREAM:
-                    return client.define(Group.create(this.group, Catalog.STREAM, this.shard, 0, Duration.ofDays(7)));
+                    resourceExist = client.existStream(this.group, this.name());
+                    if (!resourceExist.hasGroup()) {
+                        Group g = client.define(Group.create(this.group, Catalog.STREAM, this.shard,
+                                IntervalRule.create(IntervalRule.Unit.HOUR, this.blockIntervalHrs),
+                                IntervalRule.create(IntervalRule.Unit.HOUR, this.segmentIntervalHrs),
+                                IntervalRule.create(IntervalRule.Unit.DAY, this.ttlDays)));
+                        if (g != null) {
+                            log.info("group {} created", g.name());
+                        }
+                    }
+                    return resourceExist.hasResource();
                 case MEASURE:
-                    return client.define(Group.create(this.group, Catalog.MEASURE, this.shard, 12, Duration.ofDays(7)));
+                    resourceExist = client.existMeasure(this.group, this.name());
+                    if (!resourceExist.hasGroup()) {
+                        Group g = client.define(Group.create(this.group, Catalog.MEASURE, this.shard,
+                                IntervalRule.create(IntervalRule.Unit.HOUR, this.blockIntervalHrs),
+                                IntervalRule.create(IntervalRule.Unit.HOUR, this.segmentIntervalHrs),
+                                IntervalRule.create(IntervalRule.Unit.DAY, this.ttlDays)));
+                        if (g != null) {
+                            log.info("group {} created", g.name());
+                        }
+                    }
+                    return resourceExist.hasResource();
                 default:
                     throw new IllegalStateException("should not reach here");
             }
         }
 
+        public String name() {
+            if (this.kind == Kind.MEASURE) {
+                return formatName(this.modelName, this.downSampling);
+            }
+            return this.modelName;
+        }
+
         public String indexFamily() {
             switch (kind) {
                 case MEASURE:
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java
index 6a522f76306d..17ebe6d1eccc 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java
@@ -18,16 +18,17 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.banyandb.measure;
 
+import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import com.google.common.collect.ImmutableSet;
+import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
 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.PairQueryCondition;
 import org.apache.skywalking.oap.server.core.analysis.Layer;
 import org.apache.skywalking.oap.server.core.query.PaginationUtils;
 import org.apache.skywalking.oap.server.core.query.enumeration.Order;
@@ -56,48 +57,7 @@ public BanyanDBEventQueryDAO(final BanyanDBStorageClient client) {
     @Override
     public Events queryEvents(EventQueryCondition condition) throws Exception {
         MeasureQueryResponse resp = query(Event.INDEX_NAME, TAGS,
-                Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
-                    @Override
-                    protected void apply(MeasureQuery query) {
-                        if (!isNullOrEmpty(condition.getUuid())) {
-                            query.and(eq(Event.UUID, condition.getUuid()));
-                        }
-                        final Source source = condition.getSource();
-                        if (source != null) {
-                            if (!isNullOrEmpty(source.getService())) {
-                                query.and(eq(Event.SERVICE, source.getService()));
-                            }
-                            if (!isNullOrEmpty(source.getServiceInstance())) {
-                                query.and(eq(Event.SERVICE_INSTANCE, source.getServiceInstance()));
-                            }
-                            if (!isNullOrEmpty(source.getEndpoint())) {
-                                query.and(eq(Event.ENDPOINT, source.getEndpoint()));
-                            }
-                        }
-
-                        if (!isNullOrEmpty(condition.getName())) {
-                            query.and(eq(Event.NAME, condition.getName()));
-                        }
-
-                        if (condition.getType() != null) {
-                            query.and(eq(Event.TYPE, condition.getType().name()));
-                        }
-
-                        final Duration startTime = condition.getTime();
-                        if (startTime != null) {
-                            if (startTime.getStartTimestamp() > 0) {
-                                query.and(gte(Event.START_TIME, startTime.getStartTimestamp()));
-                            }
-                            if (startTime.getEndTimestamp() > 0) {
-                                query.and(lte(Event.END_TIME, startTime.getEndTimestamp()));
-                            }
-                        }
-
-                        if (!isNullOrEmpty(condition.getLayer())) {
-                            query.and(eq(Event.LAYER, Layer.valueOf(condition.getLayer()).value()));
-                        }
-                    }
-                });
+                Collections.emptySet(), buildQuery(Collections.singletonList(condition)));
         Events events = new Events();
         if (resp.size() == 0) {
             return events;
@@ -105,18 +65,87 @@ protected void apply(MeasureQuery query) {
         for (final DataPoint dataPoint : resp.getDataPoints()) {
             events.getEvents().add(buildEventView(dataPoint));
         }
-        sortEvents(events, condition);
         return events;
     }
 
     @Override
     public Events queryEvents(List<EventQueryCondition> conditionList) throws Exception {
-        Events totalEvents = new Events();
-        for (final EventQueryCondition cond : conditionList) {
-            final Events singleEvents = this.queryEvents(cond);
-            totalEvents.getEvents().addAll(singleEvents.getEvents());
+        MeasureQueryResponse resp = query(Event.INDEX_NAME, TAGS,
+                Collections.emptySet(), buildQuery(conditionList));
+        Events events = new Events();
+        if (resp.size() == 0) {
+            return events;
+        }
+        for (final DataPoint dataPoint : resp.getDataPoints()) {
+            events.getEvents().add(buildEventView(dataPoint));
         }
-        return totalEvents;
+        return events;
+    }
+
+    public QueryBuilder<MeasureQuery> buildQuery(final List<EventQueryCondition> conditionList) {
+        EventQueryCondition condition = conditionList.get(0);
+        final Order queryOrder = isNull(condition.getOrder()) ? Order.DES : condition.getOrder();
+        final PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
+
+        return new QueryBuilder<MeasureQuery>() {
+            @Override
+            protected void apply(MeasureQuery query) {
+                List<AbstractCriteria> eventsQueryConditions = new ArrayList<>(conditionList.size());
+                query.limit(page.getLimit());
+                query.offset(page.getFrom());
+                if (queryOrder == Order.ASC) {
+                    query.setOrderBy(asc(Event.START_TIME));
+                } else {
+                    query.setOrderBy(desc(Event.START_TIME));
+                }
+                for (final EventQueryCondition condition : conditionList) {
+                    List<PairQueryCondition<?>> queryConditions = new ArrayList<>();
+                    if (!isNullOrEmpty(condition.getUuid())) {
+                        queryConditions.add(eq(Event.UUID, condition.getUuid()));
+                    }
+                    final Source source = condition.getSource();
+                    if (source != null) {
+                        if (!isNullOrEmpty(source.getService())) {
+                            queryConditions.add(eq(Event.SERVICE, source.getService()));
+                        }
+                        if (!isNullOrEmpty(source.getServiceInstance())) {
+                            queryConditions.add(eq(Event.SERVICE_INSTANCE, source.getServiceInstance()));
+                        }
+                        if (!isNullOrEmpty(source.getEndpoint())) {
+                            queryConditions.add(eq(Event.ENDPOINT, source.getEndpoint()));
+                        }
+                    }
+
+                    if (!isNullOrEmpty(condition.getName())) {
+                        queryConditions.add(eq(Event.NAME, condition.getName()));
+                    }
+
+                    if (condition.getType() != null) {
+                        queryConditions.add(eq(Event.TYPE, condition.getType().name()));
+                    }
+
+                    final Duration startTime = condition.getTime();
+                    if (startTime != null) {
+                        if (startTime.getStartTimestamp() > 0) {
+                            queryConditions.add(gte(Event.START_TIME, startTime.getStartTimestamp()));
+                        }
+                        if (startTime.getEndTimestamp() > 0) {
+                            queryConditions.add(lte(Event.END_TIME, startTime.getEndTimestamp()));
+                        }
+                    }
+
+                    if (!isNullOrEmpty(condition.getLayer())) {
+                        queryConditions.add(eq(Event.LAYER, Layer.valueOf(condition.getLayer()).value()));
+                    }
+                    eventsQueryConditions.add(and(queryConditions));
+                }
+                if (eventsQueryConditions.size() == 1) {
+                    query.criteria(eventsQueryConditions.get(0));
+                } else if (eventsQueryConditions.size() > 1) {
+                    query.criteria(or(eventsQueryConditions));
+                }
+            }
+        };
     }
 
     protected org.apache.skywalking.oap.server.core.query.type.event.Event buildEventView(
@@ -142,30 +171,4 @@ protected org.apache.skywalking.oap.server.core.query.type.event.Event buildEven
 
         return event;
     }
-
-    private void sortEvents(Events events, EventQueryCondition condition) {
-        if (events.getEvents().isEmpty()) {
-            return;
-        }
-
-        final Comparator<org.apache.skywalking.oap.server.core.query.type.event.Event> c =
-                buildComparator(isNull(condition.getOrder()) ? Order.DES : condition.getOrder());
-        final PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging());
-        events.setEvents(
-                events.getEvents()
-                        .stream()
-                        .sorted(c)
-                        .skip(page.getFrom())
-                        .limit(page.getLimit())
-                        .collect(Collectors.toList())
-        );
-    }
-
-    private Comparator<org.apache.skywalking.oap.server.core.query.type.event.Event> buildComparator(Order queryOrder) {
-        Comparator<org.apache.skywalking.oap.server.core.query.type.event.Event> c = Comparator.comparingLong(org.apache.skywalking.oap.server.core.query.type.event.Event::getStartTime);
-        if (queryOrder == Order.DES) {
-            c = c.reversed();
-        }
-        return c;
-    }
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
index e666929a762e..d10b92a22ff2 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetadataQueryDAO.java
@@ -167,7 +167,7 @@ public ServiceInstance getInstance(String instanceId) throws IOException {
                     @Override
                     protected void apply(MeasureQuery query) {
                         if (StringUtil.isNotEmpty(instanceId)) {
-                            query.andWithID(instanceId);
+                            query.and(id(instanceId));
                         }
                     }
                 });
@@ -204,17 +204,17 @@ protected void apply(MeasureQuery query) {
     @Override
     public List<Process> listProcesses(String serviceId, ProfilingSupportStatus supportStatus, long lastPingStartTimeBucket, long lastPingEndTimeBucket) throws IOException {
         MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
-            PROCESS_TRAFFIC_TAGS,
-            Collections.emptySet(),
-            new QueryBuilder<MeasureQuery>() {
-                @Override
-                protected void apply(MeasureQuery query) {
-                    query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
-                    query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
-                    query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, supportStatus.value()));
-                    query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
-                }
-            });
+                PROCESS_TRAFFIC_TAGS,
+                Collections.emptySet(),
+                new QueryBuilder<MeasureQuery>() {
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
+                        query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
+                        query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, supportStatus.value()));
+                        query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+                    }
+                });
 
         final List<Process> processes = new ArrayList<>();
 
@@ -228,20 +228,19 @@ protected void apply(MeasureQuery query) {
     @Override
     public List<Process> listProcesses(String serviceInstanceId, Duration duration, boolean includeVirtual) throws IOException {
         long lastPingStartTimeBucket = duration.getStartTimeBucket();
-        long lastPingEndTimeBucket = duration.getEndTimeBucket();
         MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
-            PROCESS_TRAFFIC_TAGS,
-            Collections.emptySet(),
-            new QueryBuilder<MeasureQuery>() {
-                @Override
-                protected void apply(MeasureQuery query) {
-                    query.and(eq(ProcessTraffic.INSTANCE_ID, serviceInstanceId));
-                    query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
-                    if (!includeVirtual) {
-                        query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+                PROCESS_TRAFFIC_TAGS,
+                Collections.emptySet(),
+                new QueryBuilder<MeasureQuery>() {
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        query.and(eq(ProcessTraffic.INSTANCE_ID, serviceInstanceId));
+                        query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
+                        if (!includeVirtual) {
+                            query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+                        }
                     }
-                }
-            });
+                });
 
         final List<Process> processes = new ArrayList<>();
 
@@ -255,15 +254,15 @@ protected void apply(MeasureQuery query) {
     @Override
     public List<Process> listProcesses(String agentId) throws IOException {
         MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
-            PROCESS_TRAFFIC_TAGS,
-            Collections.emptySet(),
-            new QueryBuilder<MeasureQuery>() {
-                @Override
-                protected void apply(MeasureQuery query) {
-                    query.and(eq(ProcessTraffic.AGENT_ID, agentId));
-                    query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
-                }
-            });
+                PROCESS_TRAFFIC_TAGS,
+                Collections.emptySet(),
+                new QueryBuilder<MeasureQuery>() {
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        query.and(eq(ProcessTraffic.AGENT_ID, agentId));
+                        query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+                    }
+                });
 
         final List<Process> processes = new ArrayList<>();
 
@@ -277,41 +276,41 @@ protected void apply(MeasureQuery query) {
     @Override
     public long getProcessCount(String serviceId, ProfilingSupportStatus profilingSupportStatus, long lastPingStartTimeBucket, long lastPingEndTimeBucket) throws IOException {
         MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
-            PROCESS_TRAFFIC_TAGS,
-            Collections.emptySet(),
-            new QueryBuilder<MeasureQuery>() {
-                @Override
-                protected void apply(MeasureQuery query) {
-                    query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
-                    query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
-                    query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, profilingSupportStatus.value()));
-                    query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
-                }
-            });
+                PROCESS_TRAFFIC_TAGS,
+                Collections.emptySet(),
+                new QueryBuilder<MeasureQuery>() {
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        query.and(eq(ProcessTraffic.SERVICE_ID, serviceId));
+                        query.and(gte(ProcessTraffic.LAST_PING_TIME_BUCKET, lastPingStartTimeBucket));
+                        query.and(eq(ProcessTraffic.PROFILING_SUPPORT_STATUS, profilingSupportStatus.value()));
+                        query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+                    }
+                });
 
         return resp.getDataPoints()
-            .stream()
-            .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
-            .size();
+                .stream()
+                .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
+                .size();
     }
 
     @Override
     public long getProcessCount(String instanceId) throws IOException {
         MeasureQueryResponse resp = query(ProcessTraffic.INDEX_NAME,
-            PROCESS_TRAFFIC_TAGS,
-            Collections.emptySet(),
-            new QueryBuilder<MeasureQuery>() {
-                @Override
-                protected void apply(MeasureQuery query) {
-                    query.and(eq(ProcessTraffic.INSTANCE_ID, instanceId));
-                    query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
-                }
-            });
+                PROCESS_TRAFFIC_TAGS,
+                Collections.emptySet(),
+                new QueryBuilder<MeasureQuery>() {
+                    @Override
+                    protected void apply(MeasureQuery query) {
+                        query.and(eq(ProcessTraffic.INSTANCE_ID, instanceId));
+                        query.and(ne(ProcessTraffic.DETECT_TYPE, ProcessDetectType.VIRTUAL.value()));
+                    }
+                });
 
         return resp.getDataPoints()
-            .stream()
-            .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
-            .size();
+                .stream()
+                .collect(Collectors.groupingBy((Function<DataPoint, String>) dataPoint -> dataPoint.getTagValue(ProcessTraffic.PROPERTIES)))
+                .size();
     }
 
     @Override
@@ -323,7 +322,7 @@ public Process getProcess(String processId) throws IOException {
                     @Override
                     protected void apply(MeasureQuery query) {
                         if (StringUtil.isNotEmpty(processId)) {
-                            query.andWithID(processId);
+                            query.and(id(processId));
                         }
                     }
                 });
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java
index f0343e5579ab..0b24eb0e291d 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsDAO.java
@@ -37,6 +37,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 @Slf4j
@@ -51,25 +52,26 @@ public BanyanDBMetricsDAO(BanyanDBStorageClient client, StorageBuilder<Metrics>
     @Override
     public List<Metrics> multiGet(Model model, List<Metrics> metrics) throws IOException {
         log.info("multiGet {} from BanyanDB", model.getName());
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
         if (schema == null) {
             throw new IOException(model.getName() + " is not registered");
         }
-        // TODO: add time range
         List<Metrics> metricsInStorage = new ArrayList<>(metrics.size());
-        for (final Metrics missCachedMetric : metrics) {
-            MeasureQueryResponse resp = query(model.getName(), schema.getTags(), schema.getFields(), new QueryBuilder<MeasureQuery>() {
-                @Override
-                protected void apply(MeasureQuery query) {
-                    query.andWithID(missCachedMetric.id());
+        MeasureQueryResponse resp = query(model.getName(), schema.getTags(), schema.getFields(), new QueryBuilder<MeasureQuery>() {
+            @Override
+            protected void apply(MeasureQuery query) {
+                for (final Metrics missCachedMetric : metrics) {
+                    query.or(id(missCachedMetric.id()));
                 }
-            });
-            if (resp.size() == 0) {
-                continue;
-            }
-            for (final DataPoint dataPoint : resp.getDataPoints()) {
-                metricsInStorage.add(storageBuilder.storage2Entity(new BanyanDBConverter.StorageToMeasure(model.getName(), dataPoint)));
             }
+        });
+
+        if (resp.size() == 0) {
+            return Collections.emptyList();
+        }
+
+        for (final DataPoint dataPoint : resp.getDataPoints()) {
+            metricsInStorage.add(storageBuilder.storage2Entity(new BanyanDBConverter.StorageToMeasure(schema, dataPoint)));
         }
         return metricsInStorage;
     }
@@ -77,12 +79,12 @@ protected void apply(MeasureQuery query) {
     @Override
     public InsertRequest prepareBatchInsert(Model model, Metrics metrics) throws IOException {
         log.info("prepare to insert {}", model.getName());
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
         if (schema == null) {
             throw new IOException(model.getName() + " is not registered");
         }
         MeasureWrite measureWrite = new MeasureWrite(schema.getMetadata().getGroup(), // group name
-                model.getName(), // index-name
+                schema.getMetadata().name(), // measure-name
                 TimeBucket.getTimestamp(metrics.getTimeBucket(), model.getDownsampling())); // timestamp
         final BanyanDBConverter.MeasureToStorage toStorage = new BanyanDBConverter.MeasureToStorage(schema, measureWrite);
         storageBuilder.entity2Storage(metrics, toStorage);
@@ -93,12 +95,12 @@ public InsertRequest prepareBatchInsert(Model model, Metrics metrics) throws IOE
     @Override
     public UpdateRequest prepareBatchUpdate(Model model, Metrics metrics) throws IOException {
         log.info("prepare to update {}", model.getName());
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
         if (schema == null) {
             throw new IOException(model.getName() + " is not registered");
         }
         MeasureWrite measureWrite = new MeasureWrite(schema.getMetadata().getGroup(), // group name
-                model.getName(), // index-name
+                schema.getMetadata().name(), // measure-name
                 TimeBucket.getTimestamp(metrics.getTimeBucket(), model.getDownsampling())); // timestamp
         final BanyanDBConverter.MeasureToStorage toStorage = new BanyanDBConverter.MeasureToStorage(schema, measureWrite);
         storageBuilder.entity2Storage(metrics, toStorage);
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
index 74cc7876254e..ee306052c8b4 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
@@ -19,13 +19,15 @@
 package org.apache.skywalking.oap.server.storage.plugin.banyandb.measure;
 
 import com.google.common.collect.ImmutableSet;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
+
+import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.banyandb.v1.client.DataPoint;
 import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
 import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
@@ -48,6 +50,7 @@
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.util.ByteUtil;
 
+@Slf4j
 public class BanyanDBMetricsQueryDAO extends AbstractBanyanDBDAO implements IMetricsQueryDAO {
     public BanyanDBMetricsQueryDAO(BanyanDBStorageClient client) {
         super(client);
@@ -55,6 +58,12 @@ public BanyanDBMetricsQueryDAO(BanyanDBStorageClient client) {
 
     @Override
     public long readMetricsValue(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
+        String modelName = condition.getName();
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName, duration.getStep());
+        if (schema == null) {
+            throw new IOException("schema is not registered");
+        }
+
         int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName());
         Function function = ValueColumnMetadata.INSTANCE.getValueFunction(condition.getName());
         if (function == Function.Latest) {
@@ -62,9 +71,8 @@ public long readMetricsValue(MetricsCondition condition, String valueColumnName,
                     .getValues().latestValue(defaultValue);
         }
 
-        final String modelName = condition.getName();
         TimestampRange timestampRange = new TimestampRange(duration.getStartTimestamp(), duration.getEndTimestamp());
-        final MeasureQueryResponse resp = query(modelName,
+        final MeasureQueryResponse resp = query(schema,
                 ImmutableSet.of(Metrics.ENTITY_ID),
                 ImmutableSet.of(valueColumnName),
                 timestampRange,
@@ -99,18 +107,17 @@ private void buildAggregationQuery(MeasureQuery query, String valueColumnName, F
     @Override
     public MetricsValues readMetricsValues(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
         String modelName = condition.getName();
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName, duration.getStep());
         if (schema == null) {
             throw new IOException("schema is not registered");
         }
-        final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
-        final List<String> ids = pointOfTimes.stream().map(pointOfTime -> {
-            String id = pointOfTime.id(condition.getEntity().buildId());
-            return id;
-        }).collect(Collectors.toList());
+
+        final String entityID = condition.getEntity().buildId();
+        Map<String, DataPoint> idMap = queryByEntityID(schema, valueColumnName, duration, entityID);
+
+        List<String> ids = extractMeasureIDs(duration, entityID);
 
         MetricsValues metricsValues = new MetricsValues();
-        Map<String, DataPoint> idMap = queryIDs(modelName, valueColumnName, ids);
         if (!idMap.isEmpty()) {
             // Label is null, because in readMetricsValues, no label parameter.
             IntValues intValues = metricsValues.getValues();
@@ -150,15 +157,10 @@ private long extractFieldValue(MetadataRegistry.Schema schema, String fieldName,
 
     @Override
     public List<MetricsValues> readLabeledMetricsValues(MetricsCondition condition, String valueColumnName, List<String> labels, Duration duration) throws IOException {
-        final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
-        String modelName = condition.getName();
-        List<String> ids = new ArrayList<>(pointOfTimes.size());
-        pointOfTimes.forEach(pointOfTime -> {
-            String id = pointOfTime.id(condition.getEntity().buildId());
-            ids.add(id);
-        });
+        Map<String, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
+
+        List<String> ids = extractMeasureIDs(duration, condition.getEntity().buildId());
 
-        Map<String, DataPoint> idMap = queryIDs(modelName, valueColumnName, ids);
         Map<String, DataTable> dataTableMap = new HashMap<>(idMap.size());
         for (final Map.Entry<String, DataPoint> entry : idMap.entrySet()) {
             dataTableMap.put(
@@ -168,29 +170,23 @@ public List<MetricsValues> readLabeledMetricsValues(MetricsCondition condition,
         }
 
         return Util.sortValues(
-            Util.composeLabelValue(condition, labels, ids, dataTableMap),
-            ids,
-            ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName())
+                Util.composeLabelValue(condition, labels, ids, dataTableMap),
+                ids,
+                ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName())
         );
     }
 
     @Override
     public HeatMap readHeatMap(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
-        final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
-        String modelName = condition.getName();
-        List<String> ids = new ArrayList<>(pointOfTimes.size());
-        pointOfTimes.forEach(pointOfTime -> {
-            String id = pointOfTime.id(condition.getEntity().buildId());
-            ids.add(id);
-        });
+        Map<String, DataPoint> idMap = queryByEntityID(condition, valueColumnName, duration);
 
         HeatMap heatMap = new HeatMap();
-        Map<String, DataPoint> idMap = queryIDs(modelName, valueColumnName, ids);
-
         if (idMap.isEmpty()) {
             return heatMap;
         }
 
+        List<String> ids = extractMeasureIDs(duration, condition.getEntity().buildId());
+
         final int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName());
         for (String id : ids) {
             DataPoint dataPoint = idMap.get(id);
@@ -205,19 +201,40 @@ public HeatMap readHeatMap(MetricsCondition condition, String valueColumnName, D
         return heatMap;
     }
 
-    private Map<String, DataPoint> queryIDs(String modelName, String valueColumnName, List<String> measureIDs) throws IOException {
-        Map<String, DataPoint> map = new HashMap<>(measureIDs.size());
-        for (final String id : measureIDs) {
-            MeasureQueryResponse resp = query(modelName, Collections.emptySet(), ImmutableSet.of(valueColumnName), new QueryBuilder<MeasureQuery>() {
-                @Override
-                protected void apply(MeasureQuery query) {
-                    query.andWithID(id);
-                }
-            });
-            if (resp.size() > 0) {
-                map.putIfAbsent(resp.getDataPoints().get(0).getId(), resp.getDataPoints().get(0));
+    private List<String> extractMeasureIDs(Duration duration, String entityID) {
+        final List<PointOfTime> pointOfTimes = duration.assembleDurationPoints();
+        List<String> ids = new ArrayList<>(pointOfTimes.size());
+        pointOfTimes.forEach(pointOfTime -> {
+            String id = pointOfTime.id(entityID);
+            ids.add(id);
+        });
+        return ids;
+    }
+
+    private Map<String, DataPoint> queryByEntityID(final MetricsCondition condition, String valueColumnName, Duration duration) throws IOException {
+        final MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(condition.getName(), duration.getStep());
+        if (schema == null) {
+            throw new IOException("schema is not registered");
+        }
+        return queryByEntityID(schema, valueColumnName, duration, condition.getEntity().buildId());
+    }
+
+    private Map<String, DataPoint> queryByEntityID(MetadataRegistry.Schema schema, String valueColumnName, Duration duration, String entityID) throws IOException {
+        TimestampRange timestampRange = new TimestampRange(duration.getStartTimestamp(), duration.getEndTimestamp());
+
+        Map<String, DataPoint> map = new HashMap<>();
+        MeasureQueryResponse resp = query(schema, Collections.emptySet(), ImmutableSet.of(valueColumnName), timestampRange, new QueryBuilder<MeasureQuery>() {
+            @Override
+            protected void apply(MeasureQuery query) {
+                query.and(eq(Metrics.ENTITY_ID, entityID));
+            }
+        });
+        for (final DataPoint dp : resp.getDataPoints()) {
+            if (map.putIfAbsent(dp.getId(), dp) != null) {
+                log.warn("duplicated data point");
             }
         }
+
         return map;
     }
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java
index 94bf71d72698..679568e24e74 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBNetworkAddressAliasDAO.java
@@ -29,16 +29,20 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
 import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
 import org.apache.skywalking.oap.server.core.analysis.manual.networkalias.NetworkAddressAlias;
 import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBConverter.StorageToMeasure;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO;
 
 @Slf4j
 public class BanyanDBNetworkAddressAliasDAO extends AbstractBanyanDBDAO implements INetworkAddressAliasDAO {
     private final NetworkAddressAlias.Builder builder = new NetworkAddressAlias.Builder();
 
+    private MetadataRegistry.Schema schema;
+
     private static final Set<String> TAGS = ImmutableSet.of(NetworkAddressAlias.ADDRESS,
             NetworkAddressAlias.TIME_BUCKET, NetworkAddressAlias.LAST_UPDATE_TIME_BUCKET,
             NetworkAddressAlias.REPRESENT_SERVICE_ID, NetworkAddressAlias.REPRESENT_SERVICE_INSTANCE_ID);
@@ -47,6 +51,13 @@ public BanyanDBNetworkAddressAliasDAO(final BanyanDBStorageClient client) {
         super(client);
     }
 
+    private MetadataRegistry.Schema getSchema() {
+        if (schema == null) {
+            schema = MetadataRegistry.INSTANCE.findMetadata(NetworkAddressAlias.INDEX_NAME, DownSampling.Minute);
+        }
+        return schema;
+    }
+
     @Override
     public List<NetworkAddressAlias> loadLastUpdate(long timeBucket) {
         try {
@@ -61,10 +72,14 @@ protected void apply(final MeasureQuery query) {
                         }
                     }
             );
+            /**
+             * Currently, only used by {@link org.apache.skywalking.oap.server.storage.plugin.banyandb.measure.BanyanDBNetworkAddressAliasDAO}.
+             * The default DownSampling strategy, i.e. {@link DownSampling#Minute} is assumed in this case.
+             */
             return resp.getDataPoints()
                     .stream()
                     .map(
-                            point -> builder.storage2Entity(new StorageToMeasure(NetworkAddressAlias.INDEX_NAME, point))
+                            point -> builder.storage2Entity(new StorageToMeasure(getSchema(), point))
                     )
                     .collect(Collectors.toList());
         } catch (IOException e) {
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
index bee2069770be..4978b3f64a8a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
@@ -18,13 +18,18 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
 
+import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
 import org.apache.skywalking.banyandb.v1.client.AbstractQuery;
+import org.apache.skywalking.banyandb.v1.client.And;
 import org.apache.skywalking.banyandb.v1.client.MeasureQuery;
 import org.apache.skywalking.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.banyandb.v1.client.Or;
 import org.apache.skywalking.banyandb.v1.client.PairQueryCondition;
 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.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.storage.AbstractDAO;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
@@ -33,6 +38,7 @@
 import java.time.Instant;
 import java.util.List;
 import java.util.Set;
+import java.util.function.BiFunction;
 
 public abstract class AbstractBanyanDBDAO extends AbstractDAO<BanyanDBStorageClient> {
     private static final Instant UPPER_BOUND = Instant.ofEpochSecond(0, Long.MAX_VALUE);
@@ -43,21 +49,21 @@ protected AbstractBanyanDBDAO(BanyanDBStorageClient client) {
         super(client);
     }
 
-    protected StreamQueryResponse query(String modelName, Set<String> tags, QueryBuilder<StreamQuery> builder) throws IOException {
-        return this.query(modelName, tags, null, builder);
+    protected StreamQueryResponse query(String streamModelName, Set<String> tags, QueryBuilder<StreamQuery> builder) throws IOException {
+        return this.query(streamModelName, tags, null, builder);
     }
 
-    protected StreamQueryResponse query(String modelName, Set<String> tags, TimestampRange timestampRange,
+    protected StreamQueryResponse query(String streamModelName, Set<String> tags, TimestampRange timestampRange,
                                         QueryBuilder<StreamQuery> builder) throws IOException {
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findRecordMetadata(streamModelName);
         if (schema == null) {
-            throw new IllegalStateException("schema is not registered");
+            throw new IllegalArgumentException("schema is not registered");
         }
         final StreamQuery query;
         if (timestampRange == null) {
-            query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), LARGEST_TIME_RANGE, tags);
+            query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), LARGEST_TIME_RANGE, tags);
         } else {
-            query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), timestampRange, tags);
+            query = new StreamQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), timestampRange, tags);
         }
 
         builder.apply(query);
@@ -65,22 +71,27 @@ protected StreamQueryResponse query(String modelName, Set<String> tags, Timestam
         return getClient().query(query);
     }
 
-    protected MeasureQueryResponse query(String modelName, Set<String> tags, Set<String> fields,
+    protected MeasureQueryResponse query(String measureModelName, Set<String> tags, Set<String> fields,
                                          QueryBuilder<MeasureQuery> builder) throws IOException {
-        return this.query(modelName, tags, fields, null, builder);
+        return this.query(measureModelName, tags, fields, null, builder);
     }
 
-    protected MeasureQueryResponse query(String modelName, Set<String> tags, Set<String> fields,
+    protected MeasureQueryResponse query(String measureModelName, Set<String> tags, Set<String> fields,
                                          TimestampRange timestampRange, QueryBuilder<MeasureQuery> builder) throws IOException {
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(modelName);
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(measureModelName, DownSampling.Minute);
         if (schema == null) {
-            throw new IllegalStateException("schema is not registered");
+            throw new IllegalArgumentException("measure is not registered");
         }
+        return this.query(schema, tags, fields, timestampRange, builder);
+    }
+
+    protected MeasureQueryResponse query(MetadataRegistry.Schema schema, Set<String> tags, Set<String> fields,
+                                         TimestampRange timestampRange, QueryBuilder<MeasureQuery> builder) throws IOException {
         final MeasureQuery query;
         if (timestampRange == null) {
-            query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), LARGEST_TIME_RANGE, tags, fields);
+            query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), LARGEST_TIME_RANGE, tags, fields);
         } else {
-            query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().getName(), timestampRange, tags, fields);
+            query = new MeasureQuery(schema.getMetadata().getGroup(), schema.getMetadata().name(), timestampRange, tags, fields);
         }
 
         builder.apply(query);
@@ -126,5 +137,44 @@ protected PairQueryCondition<String> eq(String name, String value) {
         protected PairQueryCondition<Long> ne(String name, long value) {
             return PairQueryCondition.LongQueryCondition.ne(name, value);
         }
+
+        protected PairQueryCondition<String> id(String value) {
+            return PairQueryCondition.IDQueryCondition.eq(Metrics.ID, value);
+        }
+
+        protected AbstractQuery.OrderBy desc(String name) {
+            return new AbstractQuery.OrderBy(name, AbstractQuery.Sort.DESC);
+        }
+
+        protected AbstractQuery.OrderBy asc(String name) {
+            return new AbstractQuery.OrderBy(name, AbstractQuery.Sort.ASC);
+        }
+
+        protected AbstractCriteria and(List<? extends AbstractCriteria> conditions) {
+            if (conditions.isEmpty()) {
+                return null;
+            }
+            if (conditions.size() == 1) {
+                return conditions.get(0);
+            }
+
+            return conditions.subList(2, conditions.size()).stream().reduce(
+                    And.create(conditions.get(0), conditions.get(1)),
+                    (BiFunction<AbstractCriteria, AbstractCriteria, AbstractCriteria>) And::create,
+                    And::create);
+        }
+
+        protected AbstractCriteria or(List<? extends AbstractCriteria> conditions) {
+            if (conditions.isEmpty()) {
+                return null;
+            }
+            if (conditions.size() == 1) {
+                return conditions.get(0);
+            }
+            return conditions.subList(2, conditions.size()).stream().reduce(
+                    Or.create(conditions.get(0), conditions.get(1)),
+                    (BiFunction<AbstractCriteria, AbstractCriteria, AbstractCriteria>) Or::create,
+                    Or::create);
+        }
     }
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java
index 2ac1be38c707..1733953db81a 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBBrowserLogQueryDAO.java
@@ -19,7 +19,6 @@
 package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
 
 import com.google.common.collect.ImmutableSet;
-import com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.skywalking.banyandb.v1.client.RowEntity;
 import org.apache.skywalking.banyandb.v1.client.StreamQuery;
 import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
@@ -30,7 +29,6 @@
 import org.apache.skywalking.oap.server.core.query.input.Duration;
 import org.apache.skywalking.oap.server.core.query.type.BrowserErrorLog;
 import org.apache.skywalking.oap.server.core.query.type.BrowserErrorLogs;
-import org.apache.skywalking.oap.server.core.query.type.ErrorCategory;
 import org.apache.skywalking.oap.server.core.storage.query.IBrowserLogQueryDAO;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
@@ -98,33 +96,4 @@ public void apply(StreamQuery query) {
         }
         return logs;
     }
-
-    /**
-     * TODO: merge the default method in the interface
-     */
-    private BrowserErrorLog parserDataBinary(
-            byte[] dataBinary) {
-        try {
-            BrowserErrorLog log = new BrowserErrorLog();
-            org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog browserErrorLog = org.apache.skywalking.apm.network.language.agent.v3.BrowserErrorLog
-                    .parseFrom(dataBinary);
-
-            log.setService(browserErrorLog.getService());
-            log.setServiceVersion(browserErrorLog.getServiceVersion());
-            log.setTime(browserErrorLog.getTime());
-            log.setPagePath(browserErrorLog.getPagePath());
-            log.setCategory(ErrorCategory.valueOf(browserErrorLog.getCategory().name().toUpperCase()));
-            log.setGrade(browserErrorLog.getGrade());
-            log.setMessage(browserErrorLog.getMessage());
-            log.setLine(browserErrorLog.getLine());
-            log.setCol(browserErrorLog.getCol());
-            log.setStack(browserErrorLog.getStack());
-            log.setErrorUrl(browserErrorLog.getErrorUrl());
-            log.setFirstReportedError(browserErrorLog.getFirstReportedError());
-
-            return log;
-        } catch (InvalidProtocolBufferException e) {
-            throw new RuntimeException(e);
-        }
-    }
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java
index 6c3c9d35de6e..12c9205e54d0 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBLogQueryDAO.java
@@ -19,8 +19,6 @@
 package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
 
 import com.google.common.collect.ImmutableSet;
-import com.google.protobuf.InvalidProtocolBufferException;
-import org.apache.skywalking.apm.network.logging.v3.LogTags;
 import org.apache.skywalking.banyandb.v1.client.RowEntity;
 import org.apache.skywalking.banyandb.v1.client.StreamQuery;
 import org.apache.skywalking.banyandb.v1.client.StreamQueryResponse;
@@ -34,7 +32,6 @@
 import org.apache.skywalking.oap.server.core.query.input.Duration;
 import org.apache.skywalking.oap.server.core.query.input.TraceScopeCondition;
 import org.apache.skywalking.oap.server.core.query.type.ContentType;
-import org.apache.skywalking.oap.server.core.query.type.KeyValue;
 import org.apache.skywalking.oap.server.core.query.type.Log;
 import org.apache.skywalking.oap.server.core.query.type.Logs;
 import org.apache.skywalking.oap.server.core.storage.query.ILogQueryDAO;
@@ -149,17 +146,4 @@ public void apply(StreamQuery query) {
         }
         return logs;
     }
-
-    /**
-     * Parser the raw tags.
-     * TODO: merge default method
-     */
-    private void parserDataBinary(byte[] dataBinary, List<KeyValue> tags) {
-        try {
-            LogTags logTags = LogTags.parseFrom(dataBinary);
-            logTags.getDataList().forEach(pair -> tags.add(new KeyValue(pair.getKey(), pair.getValue())));
-        } catch (InvalidProtocolBufferException e) {
-            throw new RuntimeException(e);
-        }
-    }
 }
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java
index 2def41744819..29108e6e9cdd 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileTaskQueryDAO.java
@@ -103,9 +103,9 @@ public ProfileTask getById(String id) throws IOException {
                     @Override
                     protected void apply(StreamQuery query) {
                         if (StringUtil.isNotEmpty(id)) {
-                            // TODO: support search by ID
+                            query.and(eq(ProfileTaskRecord.TASK_ID, id));
                         }
-                        // query.setLimit(1);
+                        query.setLimit(1);
                     }
                 });
 
@@ -113,9 +113,7 @@ protected void apply(StreamQuery query) {
             return null;
         }
 
-        RowEntity first = resp.getElements().stream().filter(e -> id.equals(e.getTagValue(ProfileTaskRecord.TASK_ID)))
-            .findFirst().orElse(null);
-        return first == null ? null : buildProfileTask(first);
+        return buildProfileTask(resp.getElements().get(0));
     }
 
     private ProfileTask buildProfileTask(RowEntity data) {
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
index a0bff696b6eb..985544bbe91c 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBProfileThreadSnapshotQueryDAO.java
@@ -34,12 +34,9 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Comparator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
-import java.util.function.Function;
-import java.util.stream.Collectors;
 
 /**
  * {@link ProfileThreadSnapshotRecord} is a stream
@@ -92,6 +89,7 @@ public List<BasicTrace> queryProfiledSegments(String taskId) throws IOException
                     public void apply(StreamQuery query) {
                         query.and(eq(ProfileThreadSnapshotRecord.TASK_ID, taskId))
                                 .and(eq(ProfileThreadSnapshotRecord.SEQUENCE, 0L));
+                        query.setOrderBy(desc(ProfileThreadSnapshotRecord.DUMP_TIME));
                         query.setLimit(querySegmentMaxSize);
                     }
                 });
@@ -105,43 +103,41 @@ public void apply(StreamQuery query) {
             segmentIds.add(rowEntity.getTagValue(ProfileThreadSnapshotRecord.SEGMENT_ID));
         }
 
-        // TODO: support `IN` or `OR` logic operation in BanyanDB
-        List<BasicTrace> basicTraces = new ArrayList<>();
-        for (String segmentID : segmentIds) {
-            final StreamQueryResponse segmentRecordResp = query(SegmentRecord.INDEX_NAME,
-                    TAGS_TRACE,
-                    new QueryBuilder<StreamQuery>() {
-                        @Override
-                        public void apply(StreamQuery traceQuery) {
-                            traceQuery.and(eq(SegmentRecord.SEGMENT_ID, segmentID));
-                        }
-                    });
-
-            for (final RowEntity row : segmentRecordResp.getElements()) {
-                BasicTrace basicTrace = new BasicTrace();
-
-                basicTrace.setSegmentId(row.getId());
-                basicTrace.setStart(String.valueOf((Number) row.getTagValue(SegmentRecord.START_TIME)));
-                basicTrace.getEndpointNames().add(IDManager.EndpointID.analysisId(
-                        row.getTagValue(SegmentRecord.ENDPOINT_ID)
-                ).getEndpointName());
-                basicTrace.setDuration(((Number) row.getTagValue(SegmentRecord.LATENCY)).intValue());
-                basicTrace.setError(BooleanUtils.valueToBoolean(
-                        ((Number) row.getTagValue(SegmentRecord.IS_ERROR)).intValue()
-                ));
-                basicTrace.getTraceIds().add(row.getTagValue(SegmentRecord.TRACE_ID));
-
-                basicTraces.add(basicTrace);
-            }
+        if (segmentIds.isEmpty()) {
+            return Collections.emptyList();
         }
 
-        // TODO: Sort in DB with DESC
-        basicTraces = basicTraces.stream()
-                // comparing start_time
-                .sorted(Comparator.comparing((Function<BasicTrace, Long>) basicTrace -> Long.parseLong(basicTrace.getStart()))
-                        // and sort in reverse order
-                        .reversed())
-                .collect(Collectors.toList());
+        final StreamQueryResponse segmentRecordResp = query(SegmentRecord.INDEX_NAME,
+                TAGS_TRACE,
+                new QueryBuilder<StreamQuery>() {
+                    @Override
+                    public void apply(StreamQuery traceQuery) {
+                        // TODO: use "in" operator
+                        for (final String segmentID : segmentIds) {
+                            traceQuery.or(eq(SegmentRecord.SEGMENT_ID, segmentID));
+                        }
+                        traceQuery.setLimit(segmentIds.size());
+                        traceQuery.setOrderBy(desc(SegmentRecord.START_TIME));
+                    }
+                });
+
+        List<BasicTrace> basicTraces = new ArrayList<>();
+        for (final RowEntity row : segmentRecordResp.getElements()) {
+            BasicTrace basicTrace = new BasicTrace();
+
+            basicTrace.setSegmentId(row.getId());
+            basicTrace.setStart(String.valueOf((Number) row.getTagValue(SegmentRecord.START_TIME)));
+            basicTrace.getEndpointNames().add(IDManager.EndpointID.analysisId(
+                    row.getTagValue(SegmentRecord.ENDPOINT_ID)
+            ).getEndpointName());
+            basicTrace.setDuration(((Number) row.getTagValue(SegmentRecord.LATENCY)).intValue());
+            basicTrace.setError(BooleanUtils.valueToBoolean(
+                    ((Number) row.getTagValue(SegmentRecord.IS_ERROR)).intValue()
+            ));
+            basicTrace.getTraceIds().add(row.getTagValue(SegmentRecord.TRACE_ID));
+
+            basicTraces.add(basicTrace);
+        }
         return basicTraces;
     }
 
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java
index 1598dc73d195..11b0d9abecb6 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBRecordDAO.java
@@ -38,7 +38,7 @@ public class BanyanDBRecordDAO implements IRecordDAO {
 
     @Override
     public InsertRequest prepareBatchInsert(Model model, Record record) throws IOException {
-        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model.getName());
+        MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(model);
         if (schema == null) {
             throw new IOException(model.getName() + " is not registered");
         }


Test patch:
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
index d9fc9dd6a0fe..e2574a1dde90 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/storage/model/ModelColumnTest.java
@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.server.core.storage.model;
 
 import org.apache.skywalking.oap.server.core.analysis.metrics.DataTable;
+import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
 import org.apache.skywalking.oap.server.core.storage.annotation.ElasticSearch;
 import org.junit.Assert;
 import org.junit.Test;
@@ -31,7 +32,7 @@ public void testColumnDefine() {
                                              new SQLDatabaseExtension(),
                                              new ElasticSearchExtension(
                                                  ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
-                                             new BanyanDBExtension(-1, false, true)
+                                             new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
         );
         Assert.assertEquals(true, column.isStorageOnly());
         Assert.assertEquals("abc", column.getColumnName().getName());
@@ -40,7 +41,7 @@ public void testColumnDefine() {
                                  false, false, true, 200,
                                  new SQLDatabaseExtension(),
                                  new ElasticSearchExtension(ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
-                                 new BanyanDBExtension(-1, false, true)
+                                 new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
         );
         Assert.assertEquals(true, column.isStorageOnly());
         Assert.assertEquals("abc", column.getColumnName().getName());
@@ -50,7 +51,7 @@ public void testColumnDefine() {
                                  false, false, true, 200,
                                  new SQLDatabaseExtension(),
                                  new ElasticSearchExtension(ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
-                                 new BanyanDBExtension(-1, false, true)
+                                 new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
         );
         Assert.assertEquals(false, column.isStorageOnly());
         Assert.assertEquals("abc", column.getColumnName().getName());
@@ -63,7 +64,7 @@ public void testConflictDefinition() {
                                              new SQLDatabaseExtension(),
                                              new ElasticSearchExtension(
                                                  ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
-                                             new BanyanDBExtension(-1, false, true)
+                                             new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
         );
     }
 
@@ -74,7 +75,7 @@ public void testConflictDefinitionIndexOnly() {
                                              new SQLDatabaseExtension(),
                                              new ElasticSearchExtension(
                                                  ElasticSearch.MatchQuery.AnalyzerType.OAP_ANALYZER, "abc"),
-                                             new BanyanDBExtension(-1, false, true)
+                                             new BanyanDBExtension(-1, false, true, BanyanDB.IndexRule.IndexType.INVERTED)
         );
     }
 }
diff --git a/test/e2e-v2/script/env b/test/e2e-v2/script/env
index 7c08ebc39cea..50ce99a58e1e 100644
--- a/test/e2e-v2/script/env
+++ b/test/e2e-v2/script/env
@@ -23,6 +23,6 @@ SW_AGENT_CLIENT_JS_COMMIT=af0565a67d382b683c1dbd94c379b7080db61449
 SW_AGENT_CLIENT_JS_TEST_COMMIT=4f1eb1dcdbde3ec4a38534bf01dded4ab5d2f016
 SW_KUBERNETES_COMMIT_SHA=0f3ec68e5a7e1608cec8688716b848ed15e971e5
 SW_ROVER_COMMIT=d956eaede57b62108b78bca48045bd09ba88e653
-SW_BANYANDB_COMMIT=5a326d7e36a008c5ea10e3ae506309cb29733c53
+SW_BANYANDB_COMMIT=6b344929519dfd90203076e471fbf1136bde2262
 
 SW_CTL_COMMIT=e684fae0107045fc23799146d62f04cb68bd5a3b