[iotdb] branch master updated: [IOTDB-3196] Add wal read interface for consensus group (#5931)

[ta...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new 71e9389e95 [IOTDB-3196] Add wal read interface for consensus group (#5931)
71e9389e95 is described below

commit 71e9389e95718a3b7b78e2f19e9f58b3e1a39fc2
Author: Alan Choo <43...@users.noreply.github.com>
AuthorDate: Wed May 18 10:09:53 2022 +0800

    [IOTDB-3196] Add wal read interface for consensus group (#5931)
    
    * add search interface for consensus group
    
    * spotless
    
    * remove
---
 .../iotdb/consensus/wal/ConsensusReqReader.java    | 86 ++++++++++++++++++++++
 .../plan/planner/plan/node/write/InsertNode.java   | 22 ++++--
 .../org/apache/iotdb/db/wal/node/IWALNode.java     |  3 +-
 .../org/apache/iotdb/db/wal/node/WALFakeNode.java  | 18 +++++
 .../java/org/apache/iotdb/db/wal/node/WALNode.java | 46 ++++++++++++
 5 files changed, 168 insertions(+), 7 deletions(-)

diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/wal/ConsensusReqReader.java b/consensus/src/main/java/org/apache/iotdb/consensus/wal/ConsensusReqReader.java
new file mode 100644
index 0000000000..1b1929c8c8
--- /dev/null
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/wal/ConsensusReqReader.java
@@ -0,0 +1,86 @@
+/*
+ * 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.iotdb.consensus.wal;
+
+import org.apache.iotdb.consensus.common.request.IConsensusRequest;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
+/** This interface provides search interface for consensus requests via index. */
+public interface ConsensusReqReader {
+  /**
+   * Gets the consensus request at the specified position.
+   *
+   * @param index index of the consensus request to return
+   * @return the consensus request at the specified position
+   */
+  IConsensusRequest getReq(long index);
+
+  /**
+   * Gets the consensus requests from the specified start position.
+   *
+   * @param startIndex index of the start consensus request
+   * @param num number of consensus requests to return, the number of actual returned consensus
+   *     requests may less than this value
+   * @return the consensus requests from the specified start position
+   */
+  List<IConsensusRequest> getReqs(long startIndex, int num);
+
+  /**
+   * Gets the consensus requests iterator from the specified start position.
+   *
+   * @param startIndex index of the start consensus request
+   * @return the consensus requests iterator from the specified start position.
+   */
+  ReqIterator getReqIterator(long startIndex);
+
+  /** This iterator provides blocking and non-blocking interfaces to read consensus request. */
+  interface ReqIterator {
+    /** Like {@link Iterator#hasNext()} */
+    boolean hasNext();
+
+    /** Like {@link Iterator#next()} */
+    IConsensusRequest next();
+
+    /**
+     * Returns the next element in the iteration, blocked until next element is available.
+     *
+     * @return the next element in the iteration
+     */
+    IConsensusRequest waitForNext() throws InterruptedException;
+
+    /**
+     * Returns the next element in the iteration, blocked until next element is available or a
+     * specified amount of time has elapsed.
+     *
+     * @return the next element in the iteration
+     */
+    IConsensusRequest waitForNext(long timeout) throws InterruptedException, TimeoutException;
+
+    /**
+     * Skips to target position of next element in the iteration <br>
+     * Notice: The correctness of forward skipping should be guaranteed by the caller
+     *
+     * @param targetIndex target position of next element in the iteration
+     */
+    void skipTo(long targetIndex);
+  }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java
index 48252bbab7..2b79cb36a3 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertNode.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.db.mpp.plan.planner.plan.node.write;
 
 import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
 import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.consensus.common.request.IConsensusRequest;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.exception.metadata.DataTypeMismatchException;
 import org.apache.iotdb.db.metadata.idtable.entry.IDeviceID;
@@ -42,7 +43,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-public abstract class InsertNode extends WritePlanNode {
+public abstract class InsertNode extends WritePlanNode implements IConsensusRequest {
 
   /**
    * if use id table, this filed is id form of device path <br>
@@ -138,6 +139,20 @@ public abstract class InsertNode extends WritePlanNode {
     this.deviceID = deviceID;
   }
 
+  /**
+   * Deserialize via {@link
+   * org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNodeType#deserialize(ByteBuffer)}
+   */
+  @Override
+  public void serializeRequest(ByteBuffer buffer) {
+    serializeAttributes(buffer);
+  }
+
+  @Override
+  protected void serializeAttributes(ByteBuffer byteBuffer) {
+    throw new NotImplementedException("serializeAttributes of InsertNode is not implemented");
+  }
+
   /** Serialized size of measurement schemas, ignoring failed time series */
   protected int serializeMeasurementSchemasSize() {
     int byteLen = 0;
@@ -268,11 +283,6 @@ public abstract class InsertNode extends WritePlanNode {
   }
   // endregion
 
-  @Override
-  protected void serializeAttributes(ByteBuffer byteBuffer) {
-    throw new NotImplementedException("serializeAttributes of InsertNode is not implemented");
-  }
-
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java b/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java
index ffaffb9561..9925015dda 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/node/IWALNode.java
@@ -18,6 +18,7 @@
  */
 package org.apache.iotdb.db.wal.node;
 
+import org.apache.iotdb.consensus.wal.ConsensusReqReader;
 import org.apache.iotdb.db.engine.flush.FlushListener;
 import org.apache.iotdb.db.engine.memtable.IMemTable;
 import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
@@ -28,7 +29,7 @@ import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
 import org.apache.iotdb.db.wal.utils.listener.WALFlushListener;
 
 /** This interface provides uniform interface for writing wal and making checkpoints. */
-public interface IWALNode extends FlushListener, AutoCloseable {
+public interface IWALNode extends FlushListener, AutoCloseable, ConsensusReqReader {
   /** Log InsertRowPlan */
   WALFlushListener log(int memTableId, InsertRowPlan insertRowPlan);
 
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java b/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java
index df3dab778e..0ed3ab7d3d 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/node/WALFakeNode.java
@@ -18,6 +18,7 @@
  */
 package org.apache.iotdb.db.wal.node;
 
+import org.apache.iotdb.consensus.common.request.IConsensusRequest;
 import org.apache.iotdb.db.engine.memtable.IMemTable;
 import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertRowNode;
 import org.apache.iotdb.db.mpp.plan.planner.plan.node.write.InsertTabletNode;
@@ -27,6 +28,8 @@ import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
 import org.apache.iotdb.db.wal.exception.WALException;
 import org.apache.iotdb.db.wal.utils.listener.WALFlushListener;
 
+import java.util.List;
+
 /** This class provides fake wal node when wal is disabled or exception happens. */
 public class WALFakeNode implements IWALNode {
   private final WALFlushListener.Status status;
@@ -96,6 +99,21 @@ public class WALFakeNode implements IWALNode {
     // do nothing
   }
 
+  @Override
+  public IConsensusRequest getReq(long index) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public List<IConsensusRequest> getReqs(long startIndex, int num) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public ReqIterator getReqIterator(long startIndex) {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public void close() {
     // do nothing
diff --git a/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java b/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java
index 3cf5706c54..6412658895 100644
--- a/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/wal/node/WALNode.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.file.SystemFileFactory;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.consensus.common.request.IConsensusRequest;
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.engine.StorageEngine;
@@ -50,8 +51,10 @@ import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.FileNotFoundException;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -373,6 +376,49 @@ public class WALNode implements IWALNode {
   }
   // endregion
 
+  // region Search interfaces for consensus group
+  @Override
+  public IConsensusRequest getReq(long index) {
+    return null;
+  }
+
+  @Override
+  public List<IConsensusRequest> getReqs(long startIndex, int num) {
+    return null;
+  }
+
+  @Override
+  public ReqIterator getReqIterator(long startIndex) {
+    return new PlanNodeIterator();
+  }
+
+  private class PlanNodeIterator implements ReqIterator {
+    @Override
+    public boolean hasNext() {
+      return false;
+    }
+
+    @Override
+    public IConsensusRequest next() {
+      return null;
+    }
+
+    @Override
+    public IConsensusRequest waitForNext() throws InterruptedException {
+      return null;
+    }
+
+    @Override
+    public IConsensusRequest waitForNext(long timeout)
+        throws InterruptedException, TimeoutException {
+      return null;
+    }
+
+    @Override
+    public void skipTo(long targetIndex) {}
+  }
+  // endregion
+
   @Override
   public void close() {
     buffer.close();