add the rpc layer

Author: polyzos

fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/TabletServerGateway.java

@@ -42,6 +42,10 @@
 import org.apache.fluss.rpc.messages.ProduceLogResponse;
 import org.apache.fluss.rpc.messages.PutKvRequest;
 import org.apache.fluss.rpc.messages.PutKvResponse;
+import org.apache.fluss.rpc.messages.ScanKvRequest;
+import org.apache.fluss.rpc.messages.ScanKvResponse;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveRequest;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveResponse;
 import org.apache.fluss.rpc.messages.StopReplicaRequest;
 import org.apache.fluss.rpc.messages.StopReplicaResponse;
 import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
@@ -172,4 +176,10 @@ CompletableFuture<NotifyKvSnapshotOffsetResponse> notifyKvSnapshotOffset(
     @RPC(api = ApiKeys.NOTIFY_LAKE_TABLE_OFFSET)
     CompletableFuture<NotifyLakeTableOffsetResponse> notifyLakeTableOffset(
             NotifyLakeTableOffsetRequest request);
+
+    @RPC(api = ApiKeys.SCAN_KV)
+    CompletableFuture<ScanKvResponse> scanKv(ScanKvRequest request);
+
+    @RPC(api = ApiKeys.SCANNER_KEEP_ALIVE)
+    CompletableFuture<ScannerKeepAliveResponse> scannerKeepAlive(ScannerKeepAliveRequest request);
 }

fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java

@@ -80,7 +80,9 @@ public enum ApiKeys {
     REBALANCE(1049, 0, 0, PUBLIC),
     LIST_REBALANCE_PROGRESS(1050, 0, 0, PUBLIC),
     CANCEL_REBALANCE(1051, 0, 0, PUBLIC),
-    PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 0, PRIVATE);
+    PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 0, PRIVATE),
+    SCAN_KV(1053, 0, 0, PUBLIC),
+    SCANNER_KEEP_ALIVE(1054, 0, 0, PUBLIC);
 
     private static final Map<Integer, ApiKeys> ID_TO_TYPE =
             Arrays.stream(ApiKeys.values())

fluss-rpc/src/main/proto/FlussApi.proto

@@ -272,6 +272,93 @@ message LimitScanResponse{
   optional bytes records = 4;
 }
 
+// A scan request. Initially, it should specify a scan. Later on, you
+// can use the scanner id returned to fetch result batches with a different
+// scan request.
+//
+// The scanner will remain open if there are more results, and it's not
+// asked to be closed explicitly.
+//
+// Clients may choose to retry scan requests that fail to complete (due to, for
+// example, a timeout or network error).
+//
+// You can fetch the results and ask the scanner to be closed to save
+// a trip if you are not interested in remaining results.
+message ScanKvRequest {
+  // If continuing an existing scan, then you must set scanner_id.
+  // Otherwise, you must set 'new_scan_request'.
+  optional bytes scanner_id = 1;
+  optional PbScanReqForBucket bucket_scan_req = 2;
+
+  // The sequence ID of this call. The sequence ID should start at 0
+  // with the request for a new scanner, and after each successful request,
+  // the client should increment it by 1. When retrying a request, the client
+  // should _not_ increment this value. If the server detects that the client
+  // missed a chunk of rows from the middle of a scan, it will respond with an
+  // error.
+  optional uint32 call_seq_id = 3;
+
+  // The maximum number of bytes to send in the response.
+  optional uint32 batch_size_bytes = 4;
+
+  // If set, the server will close the scanner after responding to
+  // this request, regardless of whether all rows have been delivered.
+  optional bool close_scanner = 5;
+}
+
+message PbScanReqForBucket {
+  // The tablet to scan.
+  required int64 table_id = 1;
+  optional int64 partition_id = 2;
+  required int32 bucket_id = 3;
+
+  // The maximum number of rows to scan with the new scanner.
+  //
+  // The scanner will automatically stop yielding results and close itself
+  // after reaching this number of result rows.
+  optional uint64 limit = 4;
+}
+
+message ScanKvResponse {
+  // The error, if an error occurred with this request.
+  optional int32 error_code = 1;
+  optional string error_message = 2;
+
+  // When a scanner is created, returns the scanner ID which may be used
+  // to pull new rows from the scanner.
+  optional bytes scanner_id = 3;
+
+  // Set to true to indicate that there may be further results to be fetched
+  // from this scanner. If the scanner has no more results, then the scanner
+  // ID will become invalid and cannot continue to be used.
+  //
+  // Note that if a scan returns no results, then the initial response from
+  // the first RPC may return false in this flag, in which case there will
+  // be no scanner ID assigned.
+  optional bool has_more_results = 4;
+
+  // The block of returned rows.
+  //
+  // NOTE: the schema-related fields will not be present in this row block.
+  // The schema will match the schema requested by the client when it created
+  // the scanner.
+  optional bytes records = 5;
+
+  // Returns the corresponding log offset at the time the scanner is created
+  optional int64 log_offset = 6;
+}
+
+// A scanner keep-alive request.
+// Updates the scanner access time, increasing its time-to-live.
+message ScannerKeepAliveRequest {
+  required bytes scanner_id = 1;
+}
+
+message ScannerKeepAliveResponse {
+  // The error, if an error occurred with this request.
+  optional int32 error_code = 1;
+  optional string error_message = 2;
+}
 
 // notify bucket leader and isr request
 message NotifyLeaderAndIsrRequest {

fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java

@@ -71,6 +71,10 @@
 import org.apache.fluss.rpc.messages.ProduceLogResponse;
 import org.apache.fluss.rpc.messages.PutKvRequest;
 import org.apache.fluss.rpc.messages.PutKvResponse;
+import org.apache.fluss.rpc.messages.ScanKvRequest;
+import org.apache.fluss.rpc.messages.ScanKvResponse;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveRequest;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveResponse;
 import org.apache.fluss.rpc.messages.StopReplicaRequest;
 import org.apache.fluss.rpc.messages.StopReplicaResponse;
 import org.apache.fluss.rpc.messages.TableExistsRequest;
@@ -163,6 +167,17 @@ public CompletableFuture<NotifyLakeTableOffsetResponse> notifyLakeTableOffset(
         return null;
     }
 
+    @Override
+    public CompletableFuture<ScanKvResponse> scanKv(ScanKvRequest request) {
+        return null;
+    }
+
+    @Override
+    public CompletableFuture<ScannerKeepAliveResponse> scannerKeepAlive(
+            ScannerKeepAliveRequest request) {
+        return null;
+    }
+
     @Override
     public CompletableFuture<ListDatabasesResponse> listDatabases(ListDatabasesRequest request) {
         return null;

fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java

@@ -56,6 +56,10 @@
 import org.apache.fluss.rpc.messages.ProduceLogResponse;
 import org.apache.fluss.rpc.messages.PutKvRequest;
 import org.apache.fluss.rpc.messages.PutKvResponse;
+import org.apache.fluss.rpc.messages.ScanKvRequest;
+import org.apache.fluss.rpc.messages.ScanKvResponse;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveRequest;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveResponse;
 import org.apache.fluss.rpc.messages.StopReplicaRequest;
 import org.apache.fluss.rpc.messages.StopReplicaResponse;
 import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
@@ -386,6 +390,17 @@ public CompletableFuture<NotifyLakeTableOffsetResponse> notifyLakeTableOffset(
         return response;
     }
 
+    @Override
+    public CompletableFuture<ScanKvResponse> scanKv(ScanKvRequest request) {
+        return null;
+    }
+
+    @Override
+    public CompletableFuture<ScannerKeepAliveResponse> scannerKeepAlive(
+            ScannerKeepAliveRequest request) {
+        return null;
+    }
+
     @Override
     public void authorizeTable(OperationType operationType, long tableId) {
         if (authorizer != null) {

fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java

@@ -81,6 +81,10 @@
 import org.apache.fluss.rpc.messages.ProduceLogResponse;
 import org.apache.fluss.rpc.messages.PutKvRequest;
 import org.apache.fluss.rpc.messages.PutKvResponse;
+import org.apache.fluss.rpc.messages.ScanKvRequest;
+import org.apache.fluss.rpc.messages.ScanKvResponse;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveRequest;
+import org.apache.fluss.rpc.messages.ScannerKeepAliveResponse;
 import org.apache.fluss.rpc.messages.StopReplicaRequest;
 import org.apache.fluss.rpc.messages.StopReplicaResponse;
 import org.apache.fluss.rpc.messages.TableExistsRequest;
@@ -324,6 +328,17 @@ public CompletableFuture<NotifyLakeTableOffsetResponse> notifyLakeTableOffset(
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public CompletableFuture<ScanKvResponse> scanKv(ScanKvRequest request) {
+        return null;
+    }
+
+    @Override
+    public CompletableFuture<ScannerKeepAliveResponse> scannerKeepAlive(
+            ScannerKeepAliveRequest request) {
+        return null;
+    }
+
     @Override
     public CompletableFuture<ListAclsResponse> listAcls(ListAclsRequest request) {
         throw new UnsupportedOperationException();