You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2022/05/17 00:26:04 UTC
[skywalking-banyandb-java-client] branch main updated: Add topN impl (#13)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb-java-client.git
The following commit(s) were added to refs/heads/main by this push:
new 00e81de Add topN impl (#13)
00e81de is described below
commit 00e81de25495a3f1722aeabf900ff72946f3659e
Author: Jiajing LU <lu...@gmail.com>
AuthorDate: Tue May 17 08:26:00 2022 +0800
Add topN impl (#13)
---
README.md | 21 +++++
pom.xml | 21 ++---
.../banyandb/v1/client/AbstractQuery.java | 27 +++++++
.../banyandb/v1/client/MeasureQuery.java | 89 ++++++++++++++++++++--
.../skywalking/banyandb/v1/client/StreamQuery.java | 26 -------
.../banyandb/v1/client/TimestampRange.java | 5 ++
.../banyandb/v1/client/util/TimeUtils.java | 10 +++
src/main/proto/banyandb/v1/banyandb-measure.proto | 21 +++++
.../v1/client/BanyanDBClientStreamQueryTest.java | 4 +-
.../banyandb/v1/client/BanyanDBClientTestCI.java | 2 +-
10 files changed, 174 insertions(+), 52 deletions(-)
diff --git a/README.md b/README.md
index accf6e2..2ee3148 100644
--- a/README.md
+++ b/README.md
@@ -164,6 +164,27 @@ MeasureQueryResponse resp = client.query(query);
List<DataPoint> dataPointList = resp.getDataPoints();
```
+Measure API supports `TopN`/`BottomN` search. The results or (grouped-)results are
+ordered by the given `field`,
+
+```java
+MeasureQuery query = new MeasureQuery("sw_metrics", "service_instance_cpm_day",
+ new TimestampRange(begin.toEpochMilli(), end.toEpochMilli()),
+ ImmutableSet.of("id", "scope", "service_id"),
+ ImmutableSet.of("total"));
+query.topN(5, "total"); // bottomN
+```
+
+Besides, `limit` and `offset` are used to support pagination. `Tag`-based sort can also
+be done to the final results,
+
+```java
+query.limit(5);
+query.offset(1);
+query.orderBy("service_id", Sort.DESC);
+```
+
+
## Write
### Stream
diff --git a/pom.xml b/pom.xml
index 06d6109..b15ba28 100644
--- a/pom.xml
+++ b/pom.xml
@@ -83,13 +83,13 @@
<!-- core lib dependency -->
<bytebuddy.version>1.10.19</bytebuddy.version>
<!-- grpc version should align with the Skywalking main repo -->
- <grpc.version>1.43.2</grpc.version>
+ <grpc.version>1.46.0</grpc.version>
<protobuf.version>3.19.2</protobuf.version>
<protoc.version>3.19.2</protoc.version>
<gson.version>2.8.6</gson.version>
<os-maven-plugin.version>1.6.2</os-maven-plugin.version>
<protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
- <netty.tcnative.version>2.0.46.Final</netty.tcnative.version>
+ <netty.tcnative.version>2.0.51.Final</netty.tcnative.version>
<javax.annotation-api.version>1.3.2</javax.annotation-api.version>
<auto-value.version>1.9</auto-value.version>
<testcontainers.version>1.16.3</testcontainers.version>
@@ -118,33 +118,23 @@
<maven-checkstyle-plugin.version>3.1.0</maven-checkstyle-plugin.version>
<jmh.version>1.21</jmh.version>
<checkstyle.fails.on.error>true</checkstyle.fails.on.error>
-
</properties>
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>io.grpc</groupId>
- <artifactId>grpc-bom</artifactId>
- <version>${grpc.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
<dependencies>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
+ <version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
+ <version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
+ <version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
@@ -173,6 +163,7 @@
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-testing</artifactId>
+ <version>${grpc.version}</version>
<scope>test</scope>
</dependency>
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractQuery.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractQuery.java
index 0662770..829b77c 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractQuery.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/AbstractQuery.java
@@ -22,6 +22,7 @@ import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import lombok.AccessLevel;
import lombok.Getter;
+import lombok.RequiredArgsConstructor;
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
@@ -141,4 +142,30 @@ public abstract class AbstractQuery<T> {
}
return b.build();
}
+
+ @RequiredArgsConstructor
+ public static class OrderBy {
+ /**
+ * The field name for ordering.
+ */
+ private final String indexRuleName;
+ /**
+ * The type of ordering.
+ */
+ private final Sort type;
+
+ BanyandbModel.QueryOrder build() {
+ final BanyandbModel.QueryOrder.Builder builder = BanyandbModel.QueryOrder.newBuilder();
+ builder.setIndexRuleName(indexRuleName);
+ builder.setSort(
+ Sort.DESC.equals(type) ? BanyandbModel.Sort.SORT_DESC : BanyandbModel.Sort.SORT_ASC);
+ return builder.build();
+ }
+ }
+
+ @RequiredArgsConstructor
+ @Getter(AccessLevel.PROTECTED)
+ public enum Sort {
+ ASC, DESC;
+ }
}
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureQuery.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureQuery.java
index dedfc40..5c6d9aa 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureQuery.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureQuery.java
@@ -19,6 +19,7 @@
package org.apache.skywalking.banyandb.v1.client;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import com.google.common.collect.ImmutableSet;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
@@ -41,6 +42,14 @@ public class MeasureQuery extends AbstractQuery<BanyandbMeasure.QueryRequest> {
private Aggregation aggregation;
+ private TopN topN;
+
+ private Integer limit;
+
+ private int offset;
+
+ private OrderBy orderBy;
+
public MeasureQuery(final String group, final String name, final Set<String> tagProjections, final Set<String> fieldProjections) {
this(group, name, null, tagProjections, fieldProjections);
}
@@ -50,6 +59,12 @@ public class MeasureQuery extends AbstractQuery<BanyandbMeasure.QueryRequest> {
this.fieldProjections = fieldProjections;
}
+ public MeasureQuery groupBy(Set<String> groupByKeys) {
+ Preconditions.checkArgument(this.tagProjections.containsAll(groupByKeys), "groupBy tags should be selected first");
+ this.aggregation = new Aggregation(null, Aggregation.Type.UNSPECIFIED, groupByKeys);
+ return this;
+ }
+
public MeasureQuery maxBy(String field, Set<String> groupByKeys) {
Preconditions.checkArgument(fieldProjections.contains(field), "field should be selected first");
Preconditions.checkArgument(this.tagProjections.containsAll(groupByKeys), "groupBy tags should be selected first");
@@ -89,6 +104,33 @@ public class MeasureQuery extends AbstractQuery<BanyandbMeasure.QueryRequest> {
return this;
}
+ public MeasureQuery topN(int number, String field) {
+ Preconditions.checkArgument(fieldProjections.contains(field), "field should be selected first");
+ this.topN = new TopN(field, number, Sort.DESC);
+ return this;
+ }
+
+ public MeasureQuery bottomN(int number, String field) {
+ Preconditions.checkArgument(fieldProjections.contains(field), "field should be selected first");
+ this.topN = new TopN(field, number, Sort.ASC);
+ return this;
+ }
+
+ public MeasureQuery limit(int limit) {
+ this.limit = limit;
+ return this;
+ }
+
+ public MeasureQuery offset(int offset) {
+ this.offset = offset;
+ return this;
+ }
+
+ public MeasureQuery orderBy(String tagName, Sort sort) {
+ this.orderBy = new OrderBy(tagName, sort);
+ return this;
+ }
+
/**
* Query ID column with given value.
*
@@ -107,27 +149,57 @@ public class MeasureQuery extends AbstractQuery<BanyandbMeasure.QueryRequest> {
builder.setMetadata(buildMetadata());
if (timestampRange != null) {
builder.setTimeRange(timestampRange.build());
+ } else {
+ builder.setTimeRange(TimestampRange.MAX_RANGE);
}
builder.setTagProjection(buildTagProjections());
builder.setFieldProjection(BanyandbMeasure.QueryRequest.FieldProjection.newBuilder()
.addAllNames(fieldProjections)
.build());
if (this.aggregation != null) {
- BanyandbMeasure.QueryRequest.GroupBy groupBy = BanyandbMeasure.QueryRequest.GroupBy.newBuilder()
- .setTagProjection(buildTagProjections(this.aggregation.groupByTagsProjection))
- .setFieldName(this.aggregation.fieldName)
- .build();
- BanyandbMeasure.QueryRequest.Aggregation aggr = BanyandbMeasure.QueryRequest.Aggregation.newBuilder()
- .setFunction(this.aggregation.aggregationType.function)
- .setFieldName(this.aggregation.fieldName)
+ BanyandbMeasure.QueryRequest.GroupBy.Builder groupByBuilder = BanyandbMeasure.QueryRequest.GroupBy.newBuilder()
+ .setTagProjection(buildTagProjections(this.aggregation.groupByTagsProjection));
+ if (Strings.isNullOrEmpty(this.aggregation.fieldName)) {
+ if (this.aggregation.aggregationType != Aggregation.Type.UNSPECIFIED) {
+ throw new IllegalArgumentException("field name cannot be null or empty if aggregation is specified");
+ }
+ } else {
+ builder.setGroupBy(groupByBuilder.build());
+ groupByBuilder.setFieldName(this.aggregation.fieldName);
+ BanyandbMeasure.QueryRequest.Aggregation aggr = BanyandbMeasure.QueryRequest.Aggregation.newBuilder()
+ .setFunction(this.aggregation.aggregationType.function)
+ .setFieldName(this.aggregation.fieldName)
+ .build();
+ builder.setGroupBy(groupByBuilder.build()).setAgg(aggr);
+ }
+ }
+ if (this.topN != null) {
+ BanyandbMeasure.QueryRequest.Top top = BanyandbMeasure.QueryRequest.Top.newBuilder()
+ .setFieldName(this.topN.fieldName)
+ .setNumber(this.topN.number)
+ .setFieldValueSort(Sort.DESC.equals(this.topN.sort) ? BanyandbModel.Sort.SORT_DESC : BanyandbModel.Sort.SORT_ASC)
.build();
- builder.setGroupBy(groupBy).setAgg(aggr);
+ builder.setTop(top);
+ }
+ builder.setOffset(this.offset);
+ if (this.limit != null) {
+ builder.setLimit(this.limit);
+ }
+ if (this.orderBy != null) {
+ builder.setOrderBy(orderBy.build());
}
// add all criteria
builder.addAllCriteria(buildCriteria());
return builder.build();
}
+ @RequiredArgsConstructor
+ public static class TopN {
+ private final String fieldName;
+ private final int number;
+ private final AbstractQuery.Sort sort;
+ }
+
@RequiredArgsConstructor
public static class Aggregation {
private final String fieldName;
@@ -136,6 +208,7 @@ public class MeasureQuery extends AbstractQuery<BanyandbMeasure.QueryRequest> {
@RequiredArgsConstructor
public enum Type {
+ UNSPECIFIED(BanyandbModel.AggregationFunction.AGGREGATION_FUNCTION_UNSPECIFIED),
MEAN(BanyandbModel.AggregationFunction.AGGREGATION_FUNCTION_MEAN),
MAX(BanyandbModel.AggregationFunction.AGGREGATION_FUNCTION_MAX),
MIN(BanyandbModel.AggregationFunction.AGGREGATION_FUNCTION_MIN),
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamQuery.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamQuery.java
index f0fefc7..7012dc4 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamQuery.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/StreamQuery.java
@@ -20,9 +20,7 @@ package org.apache.skywalking.banyandb.v1.client;
import java.util.Set;
-import lombok.RequiredArgsConstructor;
import lombok.Setter;
-import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
import org.apache.skywalking.banyandb.stream.v1.BanyandbStream;
import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
@@ -76,28 +74,4 @@ public class StreamQuery extends AbstractQuery<BanyandbStream.QueryRequest> {
}
return builder.build();
}
-
- @RequiredArgsConstructor
- public static class OrderBy {
- /**
- * The field name for ordering.
- */
- private final String indexRuleName;
- /**
- * The type of ordering.
- */
- private final Type type;
-
- private BanyandbModel.QueryOrder build() {
- final BanyandbModel.QueryOrder.Builder builder = BanyandbModel.QueryOrder.newBuilder();
- builder.setIndexRuleName(indexRuleName);
- builder.setSort(
- Type.DESC.equals(type) ? BanyandbModel.Sort.SORT_DESC : BanyandbModel.Sort.SORT_ASC);
- return builder.build();
- }
-
- public enum Type {
- ASC, DESC
- }
- }
}
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/TimestampRange.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/TimestampRange.java
index 7ea4e3f..95e7763 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/TimestampRange.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/TimestampRange.java
@@ -23,10 +23,15 @@ import lombok.AccessLevel;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
+import org.apache.skywalking.banyandb.v1.client.util.TimeUtils;
@RequiredArgsConstructor
@Getter(AccessLevel.PROTECTED)
public class TimestampRange {
+ static final BanyandbModel.TimeRange MAX_RANGE = BanyandbModel.TimeRange.newBuilder()
+ .setBegin(TimeUtils.fromEpochNanos(Long.MIN_VALUE))
+ .setEnd(TimeUtils.fromEpochNanos(Long.MAX_VALUE))
+ .build();
/**
* start timestamp in timeunit of milliseconds. inclusive.
*/
diff --git a/src/main/java/org/apache/skywalking/banyandb/v1/client/util/TimeUtils.java b/src/main/java/org/apache/skywalking/banyandb/v1/client/util/TimeUtils.java
index 50432a5..39fcaa3 100644
--- a/src/main/java/org/apache/skywalking/banyandb/v1/client/util/TimeUtils.java
+++ b/src/main/java/org/apache/skywalking/banyandb/v1/client/util/TimeUtils.java
@@ -38,4 +38,14 @@ public class TimeUtils {
.setNanos(zdt.toInstant().getNano())
.build();
}
+
+ public static Timestamp fromEpochNanos(long nanos) {
+ long seconds = nanos / 1000_000_000;
+ int nanosInt = (int) (nanos - seconds * 1000_000_000);
+ int res = nanosInt % 1000_000;
+ return Timestamp.newBuilder()
+ .setSeconds(seconds)
+ .setNanos(nanosInt - res)
+ .build();
+ }
}
diff --git a/src/main/proto/banyandb/v1/banyandb-measure.proto b/src/main/proto/banyandb/v1/banyandb-measure.proto
index 28d90d9..2a8214d 100644
--- a/src/main/proto/banyandb/v1/banyandb-measure.proto
+++ b/src/main/proto/banyandb/v1/banyandb-measure.proto
@@ -75,6 +75,27 @@ message QueryRequest {
}
// agg aggregates data points based on a field
Aggregation agg = 8;
+ message Top {
+ // number set the how many items should be returned
+ int32 number = 1;
+ // field_name must be one of files indicated by the field_projection
+ string field_name = 2;
+ // field_value_sort indicates how to sort fields
+ // ASC: bottomN
+ // DESC: topN
+ // UNSPECIFIED: topN
+ model.v1.Sort field_value_sort = 3;
+ }
+ // top limit the result based on a particular field
+ Top top = 9;
+ // offset is used to support pagination, together with the following limit.
+ // If top is specified, offset processes the dataset based on top's output
+ uint32 offset = 10;
+ // limit is used to impose a boundary on the number of records being returned.
+ // If top is specified, limit processes the dataset based on top's output
+ uint32 limit = 11;
+ // order_by is given to specify the sort for a tag.
+ model.v1.QueryOrder order_by = 12;
}
//TopNList contains a series of topN items
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamQueryTest.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamQueryTest.java
index f3f0c17..9656955 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamQueryTest.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientStreamQueryTest.java
@@ -103,7 +103,7 @@ public class BanyanDBClientStreamQueryTest extends AbstractBanyanDBClientTest {
ImmutableSet.of("state", "start_time", "duration", "trace_id"));
// search for all states
query.and(PairQueryCondition.LongQueryCondition.eq("state", 0L));
- query.setOrderBy(new StreamQuery.OrderBy("duration", StreamQuery.OrderBy.Type.DESC));
+ query.setOrderBy(new StreamQuery.OrderBy("duration", AbstractQuery.Sort.DESC));
client.query(query);
verify(streamQueryServiceImpl).query(requestCaptor.capture(), ArgumentMatchers.any());
@@ -151,7 +151,7 @@ public class BanyanDBClientStreamQueryTest extends AbstractBanyanDBClientTest {
.and(PairQueryCondition.StringQueryCondition.eq("endpoint_id", endpointId))
.and(PairQueryCondition.LongQueryCondition.ge("duration", minDuration))
.and(PairQueryCondition.LongQueryCondition.le("duration", maxDuration))
- .setOrderBy(new StreamQuery.OrderBy("start_time", StreamQuery.OrderBy.Type.ASC));
+ .setOrderBy(new StreamQuery.OrderBy("start_time", AbstractQuery.Sort.ASC));
client.query(query);
diff --git a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java
index 48e43ee..fa41f36 100644
--- a/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java
+++ b/src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java
@@ -30,7 +30,7 @@ import java.io.IOException;
public class BanyanDBClientTestCI {
private static final String REGISTRY = "ghcr.io";
private static final String IMAGE_NAME = "apache/skywalking-banyandb";
- private static final String TAG = "ad94bb9e39276a985d1647cacae4afc565b17d83";
+ private static final String TAG = "f868a5dcd881c21dbf044013228994f49f15bd83";
private static final String IMAGE = REGISTRY + "/" + IMAGE_NAME + ":" + TAG;