Add property query (#80)

Author: hanahmily

README.md

@@ -289,6 +289,25 @@ Measure m = Measure.newBuilder()
 client.define(m);
 ```
 
+### Define a Property
+
+```java
+// Define property schema
+BanyandbDatabase.Property propertyDef = 
+   BanyandbDatabase.Property.newBuilder()
+        .setMetadata(Metadata.newBuilder()
+            .setGroup("default")
+            .setName("ui_template"))
+        .addTags(
+           TagSpec.newBuilder()
+               .setName("name")
+               .setType(
+                   TagType.TAG_TYPE_STRING))
+        .build();
+
+client.define(propertyDef);
+```
+
 For more APIs usage, refer to test cases and API docs.
 
 ## Query
@@ -414,6 +433,31 @@ After response is returned, `trace` can be extracted,
 MeasureQueryResponse resp = client.query(query);
 Trace trace = resp.getTrace();
 ```
+
+### Property
+
+Query properties:
+
+```java
+BanyandbProperty.QueryRequest queryRequest = new PropertyQuery(Lists.newArrayList("default"), "ui_template", ImmutableSet.of("name")).build();
+BanyandbProperty.QueryResponse queryResponse = client.query(queryRequest);
+```
+
+Query properties based on ID:
+
+```java
+BanyandbProperty.QueryRequest queryRequest = new PropertyQuery(Lists.newArrayList("default"), "ui_template", ImmutableSet.of("name")).id("dashboard-1").build();
+BanyandbProperty.QueryResponse queryResponse = client.query(queryRequest);
+```
+
+Query properties based on tags:
+
+```java
+PropertyQuery pQuery = new PropertyQuery(Lists.newArrayList("default"), "ui_template", ImmutableSet.of("name"));
+ pQuery.criteria(PairQueryCondition.StringQueryCondition.eq("name", "foo"));
+BanyandbProperty.QueryResponse resp = client.query(pQuery.build());
+```
+
 ### Criteria
 
 Both `StreamQuery` and `MeausreQuery` support the `criteria` flag to filter data.
@@ -525,28 +569,10 @@ MeasureWrite measureWrite = client.createMeasureWrite("sw_metric", "service_cpm_
 CompletableFuture<Void> f = measureBulkWriteProcessor.add(measureWrite);
 f.get(10, TimeUnit.SECONDS);
 ```
-# Property APIs
-
-Before using properties, you need to define a property schema:
-
-```java
-// Define property schema
-BanyandbDatabase.Property propertyDef = 
-   BanyandbDatabase.Property.newBuilder()
-        .setMetadata(Metadata.newBuilder()
-            .setGroup("default")
-            .setName("sw"))
-        .addTags(
-           TagSpec.newBuilder()
-               .setName("name")
-               .setType(
-                   TagType.TAG_TYPE_STRING))
-        .build();
 
-client.define(propertyDef);
-```
+### Property
 
-After defining the schema, you can apply (create/update) properties:
+Unlike `Stream` and `Measure`, `Property` is a single write operation. The `Property` object is created and sent to the server.
 
 ```java
 // Apply a property (create or update)
@@ -570,18 +596,13 @@ You can also apply with a specific strategy:
 ApplyResponse response = client.apply(property, Strategy.STRATEGY_MERGE);
 ```
 
-Query properties:
+## Delete
 
-```java
-// Query properties
-BanyandbProperty.QueryRequest queryRequest = BanyandbProperty.QueryRequest.newBuilder()
-    .setMetadata(Metadata.newBuilder()
-        .setGroup("default")
-        .setName("ui_template"))
-    .build();
+### Stream and Measure
 
-BanyandbProperty.QueryResponse queryResponse = client.query(queryRequest);
-```
+The `Stream` and `Measure` are deleted by the TTL mechanism. You can set the TTL when defining the group schema.
+
+### Property
 
 Delete a property:
 

src/main/java/org/apache/skywalking/banyandb/v1/client/MeasureQuery.java

@@ -50,6 +50,8 @@ public class MeasureQuery extends AbstractQuery<BanyandbMeasure.QueryRequest> {
 
     private OrderBy orderBy;
 
+    private String nodeSelector;
+
     public MeasureQuery(final List<String> groups, final String name, final Set<String> tagProjections, final Set<String> fieldProjections) {
         this(groups, name, null, tagProjections, fieldProjections);
     }
@@ -136,6 +138,11 @@ public MeasureQuery offset(int offset) {
         return this;
     }
 
+    public MeasureQuery nodeSelector(String nodeSelector) {
+        this.nodeSelector = nodeSelector;
+        return this;
+    }
+
     /**
      * @return QueryRequest for gRPC level query.
      */
@@ -161,6 +168,9 @@ BanyandbMeasure.QueryRequest build(MetadataCache.EntityMetadata entityMetadata)
                     .addAllNames(fieldProjections)
                     .build());
         }
+        if (nodeSelector != null && !nodeSelector.isEmpty()) {
+            builder.setNodeSelector(nodeSelector);
+        }
         if (this.aggregation != null) {
             BanyandbMeasure.QueryRequest.GroupBy.Builder groupByBuilder = BanyandbMeasure.QueryRequest.GroupBy.newBuilder()
                     .setTagProjection(buildTagProjections(entityMetadata, this.aggregation.groupByTagsProjection));

src/main/java/org/apache/skywalking/banyandb/v1/client/PropertyQuery.java

@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.banyandb.v1.client;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import lombok.Setter;
+import org.apache.skywalking.banyandb.property.v1.BanyandbProperty;
+import org.apache.skywalking.banyandb.v1.client.grpc.exception.BanyanDBException;
+import org.apache.skywalking.banyandb.v1.client.metadata.MetadataCache;
+
+/**
+ * PropertyQuery is the high-level query API for the property model.
+ */
+@Setter
+public class PropertyQuery extends AbstractQuery<BanyandbProperty.QueryRequest> {
+    /**
+     * The limit size of the query. Default value is 20.
+     */
+    private int limit;
+    
+    /**
+     * Specific property IDs to query
+     */
+    private List<String> ids;
+    
+    /**
+     * Node selector for distributed query routing
+     */
+    private String nodeSelector;
+
+    /**
+     * Construct a property query with required fields
+     */
+    public PropertyQuery(final List<String> groups, final String name, final Set<String> projections) {
+        super(groups, name, null, projections);
+        this.limit = 20;
+        this.ids = new ArrayList<>();
+    }
+
+    /**
+     * Add a property ID to filter query results
+     * @param id property ID
+     * @return this query instance for chaining
+     */
+    public PropertyQuery id(String id) {
+        if (id != null && !id.isEmpty()) {
+            this.ids.add(id);
+        }
+        return this;
+    }
+    
+    /**
+     * Add multiple property IDs to filter query results
+     * @param ids list of property IDs
+     * @return this query instance for chaining
+     */
+    public PropertyQuery ids(List<String> ids) {
+        if (ids != null) {
+            this.ids.addAll(ids);
+        }
+        return this;
+    }
+    
+    /**
+     * Set a node selector for query routing
+     * @param nodeSelector the node selector expression
+     * @return this query instance for chaining
+     */
+    public PropertyQuery nodeSelector(String nodeSelector) {
+        this.nodeSelector = nodeSelector;
+        return this;
+    }
+
+    @Override
+    public PropertyQuery and(PairQueryCondition<?> condition) {
+        return (PropertyQuery) super.and(condition);
+    }
+
+    @Override
+    public PropertyQuery or(PairQueryCondition<?> condition) {
+        return (PropertyQuery) super.or(condition);
+    }
+
+    @Override
+    BanyandbProperty.QueryRequest build(MetadataCache.EntityMetadata ignored) throws BanyanDBException {
+        return build();
+    }
+
+    public BanyandbProperty.QueryRequest build() throws BanyanDBException {
+        final BanyandbProperty.QueryRequest.Builder builder = BanyandbProperty.QueryRequest.newBuilder();
+        builder.setName(this.name);
+        builder.addAllGroups(this.groups);
+        builder.addAllTagProjection(this.tagProjections);
+        buildCriteria().ifPresent(builder::setCriteria);
+        builder.setLimit(this.limit);
+        builder.setTrace(this.trace);
+        
+        if (!this.ids.isEmpty()) {
+            builder.addAllIds(this.ids);
+        }
+        
+        if (this.nodeSelector != null && !this.nodeSelector.isEmpty()) {
+            builder.setNodeSelector(this.nodeSelector);
+        }
+        
+        return builder.build();
+    }
+}

src/main/java/org/apache/skywalking/banyandb/v1/client/StreamQuery.java

@@ -43,6 +43,11 @@ public class StreamQuery extends AbstractQuery<BanyandbStream.QueryRequest> {
      * One order condition is supported and optional.
      */
     private OrderBy orderBy;
+    
+    /**
+     * Node selector for the query.
+     */
+    private String nodeSelector;
 
     public StreamQuery(final List<String> groups, final String name, final TimestampRange timestampRange, final Set<String> projections) {
         super(groups, name, timestampRange, projections);
@@ -63,6 +68,17 @@ public StreamQuery and(PairQueryCondition<?> condition) {
     public StreamQuery or(PairQueryCondition<?> condition) {
         return (StreamQuery) super.or(condition);
     }
+    
+    /**
+     * Set the node selector for this query.
+     *
+     * @param nodeSelector the node selector
+     * @return this query instance for chaining
+     */
+    public StreamQuery nodeSelector(String nodeSelector) {
+        this.nodeSelector = nodeSelector;
+        return this;
+    }
 
     @Override
     BanyandbStream.QueryRequest build(MetadataCache.EntityMetadata entityMetadata) throws BanyanDBException {
@@ -82,6 +98,9 @@ BanyandbStream.QueryRequest build(MetadataCache.EntityMetadata entityMetadata) t
         if (orderBy != null) {
             builder.setOrderBy(orderBy.build());
         }
+        if (nodeSelector != null && !nodeSelector.isEmpty()) {
+            builder.setNodeSelector(nodeSelector);
+        }
         builder.setTrace(this.trace);
         return builder.build();
     }

src/main/proto/banyandb/v1/banyandb-measure.proto

@@ -104,6 +104,8 @@ message QueryRequest {
   model.v1.QueryOrder order_by = 12;
   // trace is used to enable trace for the query
   bool trace = 13;
+  // node_selector is used to specify the target node for the query
+  string node_selector = 14;
 }
 
 // TopNList contains a series of topN items

src/main/proto/banyandb/v1/banyandb-stream.proto

@@ -76,6 +76,8 @@ message QueryRequest {
   model.v1.TagProjection projection = 8 [(validate.rules).message.required = true];
   // trace is used to enable trace for the query
   bool trace = 9;
+  // node_selector is used to select the node to query
+  string node_selector = 10;
 }
 
 message ElementValue {

src/test/java/org/apache/skywalking/banyandb/v1/client/ITBanyanDBPropertyTests.java

@@ -27,8 +27,13 @@
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
 import org.apache.skywalking.banyandb.common.v1.BanyandbCommon.Group;
 import org.apache.skywalking.banyandb.property.v1.BanyandbProperty.Property;
@@ -158,6 +163,7 @@ public void test_PropertyList() throws BanyanDBException {
         Assert.assertTrue(this.client.apply(property).getCreated());
 
         await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            client.query(new PropertyQuery(Lists.newArrayList("default"), "sw", ImmutableSet.of("name")).build(null));
             BanyandbProperty.QueryResponse resp = client.query(BanyandbProperty.QueryRequest.newBuilder()
                     .addGroups("default")
                     .setName("sw")
@@ -183,6 +189,36 @@ public void test_PropertyList() throws BanyanDBException {
         });
     }
 
+    @Test
+    public void test_PropertyQuery() throws BanyanDBException {
+        Property property = buildProperty("default", "sw", "id1").toBuilder().addTags(
+            Tag.newBuilder().setKey("name").setValue(
+                TagValue.newBuilder().setStr(Str.newBuilder().setValue("bar")))).build();
+        Assert.assertTrue(this.client.apply(property).getCreated());
+        property = buildProperty("default", "sw", "id2").toBuilder().addTags(
+            Tag.newBuilder().setKey("name").setValue(
+                TagValue.newBuilder().setStr(Str.newBuilder().setValue("foo")))).build();
+        Assert.assertTrue(this.client.apply(property).getCreated());
+
+        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            BanyandbProperty.QueryResponse resp = client.query(new PropertyQuery(Lists.newArrayList("default"), "sw", ImmutableSet.of("name")).build());
+            Assert.assertEquals(2, resp.getPropertiesCount());
+        });
+        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            PropertyQuery pQuery = new PropertyQuery(Lists.newArrayList("default"), "sw", ImmutableSet.of("name"));
+            pQuery.criteria(PairQueryCondition.StringQueryCondition.eq("name", "foo"));
+            BanyandbProperty.QueryResponse resp = client.query(pQuery.build());
+            Assert.assertEquals(1, resp.getPropertiesCount());
+        });
+        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            PropertyQuery pQuery = new PropertyQuery(Lists.newArrayList("default"), "sw", ImmutableSet.of("name"));
+            pQuery.criteria(Or.create(PairQueryCondition.StringQueryCondition.eq("name", "foo"), 
+            PairQueryCondition.StringQueryCondition.eq("name", "bar")));
+            BanyandbProperty.QueryResponse resp = client.query(pQuery.build());
+            Assert.assertEquals(2, resp.getPropertiesCount());
+        }); 
+    }
+
     private BanyandbProperty.Property buildProperty(String group, String name, String id) {
         BanyandbProperty.Property.Builder builder = BanyandbProperty.Property.newBuilder()
                                                                              .setMetadata(