BanyanDB: Support update the Schema when OAP starting.

docs/en/changes/changes.md

@@ -30,6 +30,7 @@
 * Bump up netty to 4.1.115, grpc to 1.68.1, boringssl to 2.0.69.
 * BanyanDB: Support update the Group settings when OAP starting.
 * BanyanDB: Introduce index mode and refactor banyandb group settings.
+* BanyanDB: Support update the Schema when OAP starting.
 
 #### UI
 

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBAggregationQueryDAO.java

@@ -70,7 +70,7 @@ public List<SelectedRecord> sortMetrics(TopNCondition condition, String valueCol
             // 2) additional conditions are all group by tags
             if (CollectionUtils.isEmpty(additionalConditions) ||
                     additionalConditions.stream().map(KeyValue::getKey).collect(Collectors.toSet())
-                            .equals(ImmutableSet.copyOf(schema.getTopNSpec().getGroupByTagNames()))) {
+                            .equals(ImmutableSet.copyOf(schema.getTopNSpec().getGroupByTagNamesList()))) {
                 return serverSideTopN(condition, schema, spec, timestampRange, additionalConditions);
             }
         }

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBRecordsQueryDAO.java

@@ -56,9 +56,15 @@ public List<Record> readRecords(RecordCondition condition, String valueColumnNam
                     protected void apply(StreamQuery query) {
                         query.and(eq(TopN.ENTITY_ID, condition.getParentEntity().buildId()));
                         if (condition.getOrder() == Order.DES) {
-                            query.setOrderBy(new StreamQuery.OrderBy(valueColumnName, AbstractQuery.Sort.DESC));
+                            query.setOrderBy(new StreamQuery.OrderBy(
+                                MetadataRegistry.getIndexRuleName(modelName, valueColumnName),
+                                AbstractQuery.Sort.DESC
+                            ));
                         } else {
-                            query.setOrderBy(new StreamQuery.OrderBy(valueColumnName, AbstractQuery.Sort.ASC));
+                            query.setOrderBy(
+                                new StreamQuery.OrderBy(MetadataRegistry.getIndexRuleName(modelName, valueColumnName),
+                                                        AbstractQuery.Sort.ASC
+                                ));
                         }
                         query.setLimit(condition.getTopN());
                     }

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEBPFProfilingScheduleQueryDAO.java

@@ -55,13 +55,17 @@ public List<EBPFProfilingSchedule> querySchedules(String taskId) throws IOExcept
          MetadataRegistry.Schema schema = MetadataRegistry.INSTANCE.findMetadata(EBPFProfilingScheduleRecord.INDEX_NAME, DownSampling.Minute);
          MeasureQueryResponse resp = query(schema,
                  TAGS,
-                 Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
-                     @Override
-                     protected void apply(MeasureQuery query) {
-                         query.and(eq(EBPFProfilingScheduleRecord.TASK_ID, taskId));
-                         query.setOrderBy(new AbstractQuery.OrderBy(EBPFProfilingScheduleRecord.START_TIME, AbstractQuery.Sort.DESC));
-                     }
-                 });
+                                           Collections.emptySet(), new QueryBuilder<MeasureQuery>() {
+                 @Override
+                 protected void apply(MeasureQuery query) {
+                     query.and(eq(EBPFProfilingScheduleRecord.TASK_ID, taskId));
+                     query.setOrderBy(new AbstractQuery.OrderBy(
+                         MetadataRegistry.getIndexRuleName(EBPFProfilingScheduleRecord.INDEX_NAME,
+                                                           EBPFProfilingScheduleRecord.START_TIME
+                         ), AbstractQuery.Sort.DESC));
+                 }
+             }
+         );
 
          return resp.getDataPoints().stream().map(this::buildEBPFProfilingSchedule).collect(Collectors.toList());
      }

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBEventQueryDAO.java

@@ -120,9 +120,15 @@ protected void apply(MeasureQuery query) {
                 query.limit(page.getLimit());
                 query.offset(page.getFrom());
                 if (queryOrder == Order.ASC) {
-                    query.setOrderBy(new AbstractQuery.OrderBy(Event.START_TIME, AbstractQuery.Sort.ASC));
+                    query.setOrderBy(
+                        new AbstractQuery.OrderBy(MetadataRegistry.getIndexRuleName(Event.INDEX_NAME, Event.START_TIME),
+                                                  AbstractQuery.Sort.ASC
+                        ));
                 } else {
-                    query.setOrderBy(new AbstractQuery.OrderBy(Event.START_TIME, AbstractQuery.Sort.DESC));
+                    query.setOrderBy(
+                        new AbstractQuery.OrderBy(MetadataRegistry.getIndexRuleName(Event.INDEX_NAME, Event.START_TIME),
+                                                  AbstractQuery.Sort.DESC
+                        ));
                 }
                 for (final EventQueryCondition condition : conditionList) {
                     List<PairQueryCondition<?>> queryConditions = new ArrayList<>();

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
 
 import com.google.gson.Gson;
+import java.util.Objects;
 import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
 import org.apache.skywalking.banyandb.v1.client.AbstractCriteria;
 import org.apache.skywalking.banyandb.v1.client.AbstractQuery;
@@ -183,9 +184,10 @@ private TopNQueryResponse topNQuery(MetadataRegistry.Schema schema,
                                         AbstractQuery.Sort sort,
                                         List<KeyValue> additionalConditions,
                                         List<AttrCondition> attributes) throws IOException {
-        final TopNQuery q = new TopNQuery(List.of(schema.getMetadata().getGroup()), schema.getTopNSpec().getName(),
-                timestampRange,
-                number, sort);
+        final TopNQuery q = new TopNQuery(List.of(schema.getMetadata().getGroup()), Objects.requireNonNull(
+            schema.getTopNSpec()).getMetadata().getName(),
+                                          timestampRange,
+                                          number, sort);
         q.setAggregationType(MeasureQuery.Aggregation.Type.MEAN);
         List<PairQueryCondition<?>> conditions = new ArrayList<>();
         if (CollectionUtils.isNotEmpty(additionalConditions)) {

oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/BanyanDBTraceQueryDAO.java

@@ -48,6 +48,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
 
 import static java.util.Objects.nonNull;
 
@@ -128,7 +129,10 @@ public void apply(StreamQuery query) {
                         query.setOrderBy(new StreamQuery.OrderBy(AbstractQuery.Sort.DESC));
                         break;
                     case BY_DURATION:
-                        query.setOrderBy(new StreamQuery.OrderBy(SegmentRecord.LATENCY, AbstractQuery.Sort.DESC));
+                        query.setOrderBy(new StreamQuery.OrderBy(
+                            MetadataRegistry.getIndexRuleName(SegmentRecord.INDEX_NAME, SegmentRecord.LATENCY),
+                            AbstractQuery.Sort.DESC
+                        ));
                         break;
                 }