feat: add query and view (#153)

Author: s12f

client/src/main/java/io/hstream/HStreamClient.java

@@ -126,4 +126,20 @@ static HStreamClientBuilder builder() {
   void deleteSubscription(String subscriptionId, boolean force);
 
   Cluster describeCluster();
+
+  Query createQuery(String sql);
+
+  List<Query> listQueries();
+
+  Query getQuery(String id);
+
+  void deleteQuery(String id);
+
+  void createView(String sql);
+
+  List<View> listViews();
+
+  View getView(String name);
+
+  void deleteView(String name);
 }

client/src/main/java/io/hstream/Query.java

@@ -0,0 +1,64 @@
+package io.hstream;
+
+public class Query {
+  String id;
+  TaskStatus status;
+  long createdTime;
+  String queryText;
+
+  public String getId() {
+    return id;
+  }
+
+  public TaskStatus getStatus() {
+    return status;
+  }
+
+  public long getCreatedTime() {
+    return createdTime;
+  }
+
+  public String getQueryText() {
+    return queryText;
+  }
+
+  public static final class Builder {
+    private String id;
+    private TaskStatus status;
+    private long createdTime;
+    private String queryText;
+
+    public Builder id(String id) {
+      this.id = id;
+      return this;
+    }
+
+    public Builder status(TaskStatus status) {
+      this.status = status;
+      return this;
+    }
+
+    public Builder createdTime(long createdTime) {
+      this.createdTime = createdTime;
+      return this;
+    }
+
+    public Builder queryText(String queryText) {
+      this.queryText = queryText;
+      return this;
+    }
+
+    public Query build() {
+      Query query = new Query();
+      query.queryText = this.queryText;
+      query.id = this.id;
+      query.createdTime = this.createdTime;
+      query.status = this.status;
+      return query;
+    }
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+}

client/src/main/java/io/hstream/TaskStatus.java

@@ -0,0 +1,10 @@
+package io.hstream;
+
+public enum TaskStatus {
+  TASK_CREATING,
+  TASK_CREATED,
+  TASK_RUNNING,
+  TASK_CREATION_ABORT,
+  TASK_CONNECTION_ABORT,
+  TASK_TERMINATED,
+}

client/src/main/java/io/hstream/View.java

@@ -0,0 +1,78 @@
+package io.hstream;
+
+import java.util.List;
+
+public class View {
+  String name;
+  TaskStatus status;
+  long createdTime;
+  String sql;
+  List<String> schema;
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public TaskStatus getStatus() {
+    return status;
+  }
+
+  public long getCreatedTime() {
+    return createdTime;
+  }
+
+  public String getSql() {
+    return sql;
+  }
+
+  public List<String> getSchema() {
+    return schema;
+  }
+
+  public static final class Builder {
+    private String name;
+    private TaskStatus status;
+    private long createdTime;
+    private String sql;
+    private List<String> schema;
+
+    public Builder name(String name) {
+      this.name = name;
+      return this;
+    }
+
+    public Builder status(TaskStatus status) {
+      this.status = status;
+      return this;
+    }
+
+    public Builder createdTime(long createdTime) {
+      this.createdTime = createdTime;
+      return this;
+    }
+
+    public Builder sql(String sql) {
+      this.sql = sql;
+      return this;
+    }
+
+    public Builder schema(List<String> schema) {
+      this.schema = schema;
+      return this;
+    }
+
+    public View build() {
+      View view = new View();
+      view.createdTime = this.createdTime;
+      view.schema = this.schema;
+      view.status = this.status;
+      view.name = this.name;
+      view.sql = this.sql;
+      return view;
+    }
+  }
+}

client/src/main/java/io/hstream/util/GrpcUtils.java

@@ -5,6 +5,7 @@
 import io.hstream.*;
 import io.hstream.internal.RecordId;
 import io.hstream.internal.SpecialOffset;
+import io.hstream.internal.TaskStatusPB;
 
 /**
  * A class of utility functions to convert between the GRPC generated classes and the custom classes
@@ -154,4 +155,42 @@ public static io.hstream.CompressionType compressionTypeFromInternal(
         throw new IllegalArgumentException("Unknown compressionType: " + compressionType);
     }
   }
+
+  public static Query queryFromInternal(io.hstream.internal.Query query) {
+    return Query.newBuilder()
+        .id(query.getId())
+        .status(taskStatusFromInternal(query.getStatus()))
+        .createdTime(query.getCreatedTime())
+        .queryText(query.getQueryText())
+        .build();
+  }
+
+  public static View viewFromInternal(io.hstream.internal.View view) {
+    return View.newBuilder()
+        .name(view.getViewId())
+        .status(taskStatusFromInternal(view.getStatus()))
+        .sql(view.getSql())
+        .createdTime(view.getCreatedTime())
+        .schema(view.getSchemaList())
+        .build();
+  }
+
+  public static TaskStatus taskStatusFromInternal(TaskStatusPB statusPB) {
+    switch (statusPB) {
+      case TASK_CREATING:
+        return TaskStatus.TASK_CREATING;
+      case TASK_CREATED:
+        return TaskStatus.TASK_CREATED;
+      case TASK_RUNNING:
+        return TaskStatus.TASK_RUNNING;
+      case TASK_CONNECTION_ABORT:
+        return TaskStatus.TASK_CONNECTION_ABORT;
+      case TASK_CREATION_ABORT:
+        return TaskStatus.TASK_CREATION_ABORT;
+      case TASK_TERMINATED:
+        return TaskStatus.TASK_TERMINATED;
+      default:
+        throw new IllegalArgumentException("Unknown task status: " + statusPB);
+    }
+  }
 }

client/src/main/kotlin/io/hstream/impl/HStreamClientKtImpl.kt

@@ -7,23 +7,34 @@ import io.hstream.Cluster
 import io.hstream.ConsumerBuilder
 import io.hstream.HStreamClient
 import io.hstream.ProducerBuilder
+import io.hstream.Query
 import io.hstream.QueryerBuilder
 import io.hstream.ReaderBuilder
 import io.hstream.Shard
 import io.hstream.Stream
 import io.hstream.Subscription
+import io.hstream.View
+import io.hstream.internal.CommandQuery
+import io.hstream.internal.CreateQueryRequest
+import io.hstream.internal.DeleteQueryRequest
 import io.hstream.internal.DeleteStreamRequest
 import io.hstream.internal.DeleteSubscriptionRequest
+import io.hstream.internal.DeleteViewRequest
+import io.hstream.internal.GetQueryRequest
+import io.hstream.internal.GetViewRequest
 import io.hstream.internal.HStreamApiGrpcKt
+import io.hstream.internal.ListQueriesRequest
 import io.hstream.internal.ListShardsRequest
 import io.hstream.internal.ListStreamsRequest
 import io.hstream.internal.ListSubscriptionsRequest
+import io.hstream.internal.ListViewsRequest
 import io.hstream.internal.LookupSubscriptionRequest
 import io.hstream.util.GrpcUtils
 import kotlinx.coroutines.runBlocking
 import org.slf4j.LoggerFactory
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.atomic.AtomicReference
+import kotlin.streams.toList
 
 class HStreamClientKtImpl(bootstrapServerUrls: List<String>, credentials: ChannelCredentials? = null) : HStreamClient {
 
@@ -193,6 +204,64 @@ class HStreamClientKtImpl(bootstrapServerUrls: List<String>, credentials: Channe
         }
     }
 
+    override fun createQuery(sql: String?): Query? {
+        checkNotNull(sql)
+        return unaryCallBlocked {
+            val query = it.createQuery(CreateQueryRequest.newBuilder().setSql(sql).build())
+            GrpcUtils.queryFromInternal(query)
+        }
+    }
+
+    override fun listQueries(): List<Query> {
+        return unaryCallBlocked {
+            val result = it.listQueries(ListQueriesRequest.getDefaultInstance())
+            result.queriesList.stream()
+                .map(GrpcUtils::queryFromInternal)
+                .toList()
+        }
+    }
+
+    override fun getQuery(id: String?): Query {
+        return unaryCallBlocked {
+            val result = it.getQuery(GetQueryRequest.newBuilder().setId(id).build())
+            GrpcUtils.queryFromInternal(result)
+        }
+    }
+
+    override fun deleteQuery(id: String?) {
+        unaryCallBlocked {
+            it.deleteQuery(DeleteQueryRequest.newBuilder().setId(id).build())
+        }
+    }
+
+    override fun createView(sql: String?) {
+        unaryCallBlocked {
+            it.executeQuery(CommandQuery.newBuilder().setStmtText(sql).build())
+        }
+    }
+
+    override fun listViews(): List<View> {
+        return unaryCallBlocked {
+            it.listViews(ListViewsRequest.getDefaultInstance())
+                .viewsList.stream()
+                .map(GrpcUtils::viewFromInternal)
+                .toList()
+        }
+    }
+
+    override fun getView(name: String?): View {
+        return unaryCallBlocked {
+            val result = it.getView(GetViewRequest.newBuilder().setViewId(name).build())
+            GrpcUtils.viewFromInternal(result)
+        }
+    }
+
+    override fun deleteView(name: String?) {
+        unaryCallBlocked {
+            it.deleteView(DeleteViewRequest.newBuilder().setViewId(name).build())
+        }
+    }
+
     private final suspend fun lookupSubscriptionServerUrl(subscriptionId: String?): String {
         return unaryCallCoroutine {
             val req: LookupSubscriptionRequest =