buildProperties(
Builder variant,
int size,
Iterator eps) {
@@ -215,7 +207,7 @@ private
> List buildProperties(
pSize : size);
Graphpb.Property.Builder pb = Graphpb.Property.newBuilder();
while (eps.hasNext()) {
- HugeProperty property = (HugeProperty) eps.next();
+ BaseProperty property = eps.next();
PropertyKey key = property.propertyKey();
long pkId = key.id().asLong();
if (pSize > 0 && !properties.contains(pkId)) {
@@ -309,8 +301,8 @@ private void buildId(Builder variant, Id id) {
}
}
- private Edge parseEdge(HugeElement element) {
- HugeEdge e = (HugeEdge) element;
+ private Edge parseEdge(BaseElement element) {
+ BaseEdge e = (BaseEdge) element;
edge.clear();
EdgeLabel label = e.schemaLabel();
edge.setLabel(label.longId());
@@ -323,14 +315,14 @@ private Edge parseEdge(HugeElement element) {
buildId(variant, e.targetVertex().id());
edge.setTargetId(variant.build());
int size = e.sizeOfProperties();
- Iterator> eps = e.properties();
+ Iterator> eps = e.properties().iterator();
List props = buildProperties(variant, size, eps);
edge.setField(propertiesDesEdge, props);
return edge.build();
}
- private Vertex parseVertex(HugeElement element) {
- HugeVertex v = (HugeVertex) element;
+ private Vertex parseVertex(BaseElement element) {
+ BaseVertex v = (BaseVertex) element;
vertex.clear();
VertexLabel label = v.schemaLabel();
vertex.setLabel(label.longId());
@@ -338,7 +330,7 @@ private Vertex parseVertex(HugeElement element) {
buildId(variant, v.id());
vertex.setId(variant.build());
int size = v.sizeOfProperties();
- Iterator> vps = v.properties();
+ Iterator> vps = v.properties().iterator();
List props = buildProperties(variant, size, vps);
vertex.setField(propertiesDesVertex, props);
return vertex.build();
@@ -348,4 +340,8 @@ private Vertex parseVertex(HugeElement element) {
public void close() {
iter.close();
}
+
+ public Exception getStopCause() {
+ return stopCause;
+ }
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyCreator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyCreator.java
index 072d09cc4a..43db542f6b 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyCreator.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyCreator.java
@@ -30,7 +30,7 @@
public class InnerKeyCreator {
final BusinessHandler businessHandler;
- private final Map graphIdCache = new ConcurrentHashMap<>();
+ private volatile Map graphIdCache = new ConcurrentHashMap<>();
public InnerKeyCreator(BusinessHandler businessHandler) {
this.businessHandler = businessHandler;
@@ -49,6 +49,26 @@ public int getGraphId(Integer partId, String graphName) throws HgStoreException
}
}
+ /**
+ *
+ * @param partId partition id
+ * @param graphName graph name
+ * @return 65535 如果不存在
+ * @throws HgStoreException
+ */
+ public int getGraphIdOrCreate(Integer partId, String graphName) throws HgStoreException {
+ try {
+ GraphIdManager manager;
+ if ((manager = graphIdCache.get(partId)) == null) {
+ manager = new GraphIdManager(businessHandler, partId);
+ graphIdCache.put(partId, manager);
+ }
+ return (int) manager.getGraphIdOrCreate(graphName);
+ } catch (Exception e) {
+ throw new HgStoreException(HgStoreException.EC_RKDB_PD_FAIL, e.getMessage());
+ }
+ }
+
public void delGraphId(Integer partId, String graphName) {
if (graphIdCache.containsKey(partId)) {
graphIdCache.get(partId).releaseGraphId(graphName);
@@ -68,6 +88,15 @@ public int parseKeyCode(byte[] innerKey) {
return Bits.getShort(innerKey, innerKey.length - Short.BYTES);
}
+ public byte[] getKeyOrCreate(Integer partId, String graph, int code, byte[] key) {
+ int graphId = getGraphIdOrCreate(partId, graph);
+ byte[] buf = new byte[Short.BYTES + key.length + Short.BYTES];
+ Bits.putShort(buf, 0, graphId);
+ Bits.put(buf, Short.BYTES, key);
+ Bits.putShort(buf, key.length + Short.BYTES, code);
+ return buf;
+ }
+
public byte[] getKey(Integer partId, String graph, int code, byte[] key) {
int graphId = getGraphId(partId, graph);
byte[] buf = new byte[Short.BYTES + key.length + Short.BYTES];
@@ -77,6 +106,21 @@ public byte[] getKey(Integer partId, String graph, int code, byte[] key) {
return buf;
}
+ /**
+ *
+ * @param partId
+ * @param graph
+ * @param key
+ * @return
+ */
+ public byte[] getKey(Integer partId, String graph, byte[] key) {
+ int graphId = getGraphId(partId, graph);
+ byte[] buf = new byte[Short.BYTES + key.length];
+ Bits.putShort(buf, 0, graphId);
+ Bits.put(buf, Short.BYTES, key);
+ return buf;
+ }
+
public byte[] getStartKey(Integer partId, String graph) {
int graphId = getGraphId(partId, graph);
byte[] buf = new byte[Short.BYTES];
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyFilter.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyFilter.java
index 34dc46063b..368032f2ce 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyFilter.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/InnerKeyFilter.java
@@ -40,6 +40,14 @@ public InnerKeyFilter(ScanIterator iterator) {
moveNext();
}
+ public InnerKeyFilter(ScanIterator iterator, boolean codeFilter) {
+ this.iterator = iterator;
+ this.codeFrom = Integer.MIN_VALUE;
+ this.codeTo = Integer.MAX_VALUE;
+ this.codeFilter = codeFilter;
+ moveNext();
+ }
+
public InnerKeyFilter(ScanIterator iterator, int codeFrom, int codeTo) {
this.iterator = iterator;
this.codeFrom = codeFrom;
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/MultiPartitionIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/MultiPartitionIterator.java
index 44d77935d5..72cc472b21 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/MultiPartitionIterator.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/MultiPartitionIterator.java
@@ -24,6 +24,7 @@
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.function.BiFunction;
+import java.util.stream.Collectors;
import org.apache.hugegraph.rocksdb.access.ScanIterator;
@@ -198,4 +199,16 @@ private byte[] getPositionKey(int partitionId) {
}
+ /**
+ * obtain iteration list of all partitions
+ *
+ * @return iteration list
+ */
+ public List getIterators() {
+ return this.partitions.stream()
+ .map(id -> supplier.apply(id, getPositionKey(id)))
+ .filter(ScanIterator::hasNext)
+ .collect(Collectors.toList());
+ }
+
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/SelectIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/SelectIterator.java
index 41a47efccf..2b51e98778 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/SelectIterator.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/SelectIterator.java
@@ -21,10 +21,10 @@
import java.util.List;
import java.util.Set;
-import org.apache.hugegraph.backend.id.Id;
-import org.apache.hugegraph.backend.serializer.BytesBuffer;
+import org.apache.hugegraph.id.Id;
import org.apache.hugegraph.rocksdb.access.RocksDBSession.BackendColumn;
import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.apache.hugegraph.serializer.BytesBuffer;
import org.apache.hugegraph.type.define.DataType;
import org.apache.hugegraph.type.define.SerialEnum;
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/BatchGetIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/BatchGetIterator.java
new file mode 100644
index 0000000000..948a46aa89
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/BatchGetIterator.java
@@ -0,0 +1,88 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.Iterator;
+import java.util.function.Function;
+
+import org.apache.hugegraph.rocksdb.access.RocksDBSession;
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.apache.hugegraph.store.query.QueryTypeParam;
+
+/**
+ * 传入多个id，根据id去查询数据，返回iterator
+ * ID 查询
+ */
+public class BatchGetIterator implements ScanIterator {
+
+ private final Iterator iterator;
+
+ private final Function retriveFunction;
+
+ private byte[] pos;
+
+ public BatchGetIterator(Iterator iterator,
+ Function retriveFunction) {
+ this.iterator = iterator;
+ this.retriveFunction = retriveFunction;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.iterator.hasNext();
+ }
+
+ @Override
+ public boolean isValid() {
+ return this.iterator.hasNext();
+ }
+
+ @Override
+ public RocksDBSession.BackendColumn next() {
+ var param = iterator.next();
+ byte[] key = param.getStart();
+ this.pos = key;
+ var value = retriveFunction.apply(param);
+ return value == null ? null : RocksDBSession.BackendColumn.of(key, value);
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public byte[] position() {
+ return this.pos;
+ }
+
+ @Override
+ public long count() {
+ long count = 0L;
+ while (this.iterator.hasNext()) {
+ this.iterator.next();
+ count += 1;
+ }
+ return count;
+ }
+
+ @Override
+ public void seek(byte[] position) {
+ // not supported
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/FileObjectIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/FileObjectIterator.java
new file mode 100644
index 0000000000..23fd7fbb55
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/FileObjectIterator.java
@@ -0,0 +1,71 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Iterator;
+
+import org.apache.hugegraph.store.business.itrv2.io.SortShuffleSerializer;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class FileObjectIterator implements Iterator {
+
+ private FileInputStream fis = null;
+ private T current;
+ private String fn;
+ private SortShuffleSerializer serializer;
+
+ public FileObjectIterator(String filePath, SortShuffleSerializer serializer) {
+ this.fn = filePath;
+ this.serializer = serializer;
+ }
+
+ @Override
+ public boolean hasNext() {
+ try {
+ if (fis == null) {
+ fis = new FileInputStream(this.fn);
+ }
+ current = readObject(fis);
+
+ if (current != null) {
+ return true;
+ } else {
+ String parent = new File(this.fn).getParent();
+ new File(parent).delete();
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return false;
+ }
+
+ @Override
+ public T next() {
+ return current;
+ }
+
+ public T readObject(InputStream input) throws IOException {
+ return serializer.read(input);
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/InAccurateIntersectionIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/InAccurateIntersectionIterator.java
new file mode 100644
index 0000000000..69d1699df6
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/InAccurateIntersectionIterator.java
@@ -0,0 +1,103 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.NoSuchElementException;
+import java.util.function.ToLongFunction;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+
+/**
+ * 不适用于单个的iterator, 单个的要使用union的版本(只是做去重)
+ *
+ * @param
+ */
+public class InAccurateIntersectionIterator implements ScanIterator {
+
+ private final Roaring64Bitmap workBitmap;
+
+ private final ToLongFunction toLongFunction;
+
+ private final ScanIterator iterator;
+
+ private T current;
+
+ public InAccurateIntersectionIterator(ScanIterator iterator, ToLongFunction toLongFunction) {
+ assert (iterator instanceof MultiListIterator &&
+ ((MultiListIterator) iterator).getIterators().size() > 0);
+ this.iterator = iterator;
+ this.workBitmap = new Roaring64Bitmap();
+ this.toLongFunction = toLongFunction;
+ }
+
+ @Override
+ public boolean hasNext() {
+ current = null;
+ while (iterator.hasNext()) {
+ var element = (T) iterator.next();
+ if (element == null) {
+ continue;
+ }
+
+ var key = toLongFunction.applyAsLong(element);
+ if (workBitmap.contains(key)) {
+ current = element;
+ return true;
+ } else {
+ workBitmap.add(key);
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean isValid() {
+ return iterator.isValid();
+ }
+
+ @Override
+ public E next() {
+ if (current == null) {
+ throw new NoSuchElementException();
+ }
+ return (E) current;
+ }
+
+ @Override
+ public long count() {
+ return iterator.count();
+ }
+
+ @Override
+ public byte[] position() {
+ return iterator.position();
+ }
+
+ @Override
+ public void seek(byte[] position) {
+ iterator.seek(position);
+ }
+
+ @Override
+ public void close() {
+ iterator.close();
+ this.workBitmap.clear();
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/InAccurateUnionFilterIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/InAccurateUnionFilterIterator.java
new file mode 100644
index 0000000000..8e818c54f6
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/InAccurateUnionFilterIterator.java
@@ -0,0 +1,100 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.NoSuchElementException;
+import java.util.function.ToLongFunction;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+
+/**
+ * 不精确的去重，直接使用位图
+ *
+ * @param
+ */
+public class InAccurateUnionFilterIterator implements ScanIterator {
+
+ private final Roaring64Bitmap workBitmap;
+
+ private final ToLongFunction toLongFunction;
+
+ private final ScanIterator iterator;
+
+ private T current;
+
+ public InAccurateUnionFilterIterator(ScanIterator iterator, ToLongFunction toLongFunction) {
+ this.iterator = iterator;
+ this.workBitmap = new Roaring64Bitmap();
+ this.toLongFunction = toLongFunction;
+ }
+
+ @Override
+ public boolean hasNext() {
+ current = null;
+ while (iterator.hasNext()) {
+ var element = (T) iterator.next();
+ if (element == null) {
+ continue;
+ }
+
+ var key = toLongFunction.applyAsLong(element);
+ if (!workBitmap.contains(key)) {
+ current = element;
+ workBitmap.add(key);
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean isValid() {
+ return iterator.isValid();
+ }
+
+ @Override
+ public E next() {
+ if (current == null) {
+ throw new NoSuchElementException();
+ }
+ return (E) current;
+ }
+
+ @Override
+ public long count() {
+ return iterator.count();
+ }
+
+ @Override
+ public byte[] position() {
+ return iterator.position();
+ }
+
+ @Override
+ public void seek(byte[] position) {
+ iterator.seek(position);
+ }
+
+ @Override
+ public void close() {
+ iterator.close();
+ this.workBitmap.clear();
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/IntersectionFilterIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/IntersectionFilterIterator.java
new file mode 100644
index 0000000000..8651b96f7d
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/IntersectionFilterIterator.java
@@ -0,0 +1,243 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.hugegraph.rocksdb.access.RocksDBSession;
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.apache.hugegraph.store.business.itrv2.io.SortShuffleSerializer;
+import org.apache.hugegraph.store.util.SortShuffle;
+
+/**
+ * 目前应用：2个及以上的iterator去重 (大数据量）
+ * 痛点： iterator 内部可能有重复的， 怎么处理 ？？
+ */
+public class IntersectionFilterIterator implements ScanIterator {
+
+ private static final Integer MAX_SIZE = 100000;
+ protected Map map;
+ private ScanIterator iterator;
+ private IntersectionWrapper wrapper;
+ private boolean processed = false;
+ private Iterator innerIterator;
+ private SortShuffle sortShuffle;
+
+ private int size = -1;
+
+ @Deprecated
+ public IntersectionFilterIterator(ScanIterator iterator, IntersectionWrapper wrapper) {
+ this.iterator = iterator;
+ this.wrapper = wrapper;
+ this.map = new HashMap<>();
+ }
+
+ /**
+ * iterator中取交集（可以是multi list iterator
+ * 问题：对于multi list iterator，无法做到每个都存在，需要外部去重。但是保证总数
+ *
+ * @param iterator 待遍历的iterator
+ * @param wrapper bitmap, 不在bitmap中的，丢弃
+ * @param size the element count in the iterator by filtering
+ */
+ public IntersectionFilterIterator(ScanIterator iterator, IntersectionWrapper wrapper,
+ int size) {
+ this(iterator, wrapper);
+ this.size = size;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!processed) {
+ try {
+ dedup();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ processed = true;
+ }
+
+ return innerIterator.hasNext();
+ }
+
+ // TODO: 优化序列化器
+ private void saveElements() throws IOException, ClassNotFoundException {
+ for (var entry : this.map.entrySet()) {
+ for (int i = 0; i < entry.getValue(); i++) {
+ sortShuffle.append((RocksDBSession.BackendColumn) entry.getKey());
+ }
+ }
+
+ this.map.clear();
+ }
+
+ /**
+ * todo: 如果一个iterator中存在重复的，目前是无解。 去重的代价太高了。
+ *
+ * @throws IOException
+ * @throws ClassNotFoundException
+ */
+ protected void dedup() throws IOException, ClassNotFoundException {
+ while (this.iterator.hasNext()) {
+ var object = this.iterator.next();
+ if (wrapper.contains(object)) {
+ this.map.put(object, map.getOrDefault(object, 0) + 1);
+ if (this.map.size() >= MAX_SIZE) {
+ if (this.sortShuffle == null) {
+ this.sortShuffle =
+ new SortShuffle<>((o1, o2) -> Arrays.compare(o1.name, o2.name),
+ SortShuffleSerializer.ofBackendColumnSerializer());
+ }
+ saveElements();
+ }
+ }
+ }
+
+ // last batch
+ if (this.sortShuffle != null) {
+ saveElements();
+ this.sortShuffle.finish();
+ }
+
+ if (this.sortShuffle == null) {
+ // map 没填满
+ this.innerIterator =
+ new MapValueFilterIterator<>(this.map, x -> x == size || size == -1 && x > 1);
+ } else {
+ // 需要读取文件
+ var fileIterator =
+ (Iterator) this.sortShuffle.getIterator();
+ this.innerIterator = new ReduceIterator<>(fileIterator,
+ (o1, o2) -> Arrays.compare(o1.name, o2.name),
+ this.size);
+ }
+ }
+
+ @Override
+ public boolean isValid() {
+ if (this.processed) {
+ return false;
+ }
+ return iterator.isValid();
+ }
+
+ @Override
+ public T next() {
+ return (T) this.innerIterator.next();
+ }
+
+ @Override
+ public void close() {
+ this.iterator.close();
+ this.map.clear();
+ }
+
+ @Override
+ public long count() {
+ return this.iterator.count();
+ }
+
+ @Override
+ public byte[] position() {
+ return this.iterator.position();
+ }
+
+ @Override
+ public void seek(byte[] position) {
+ this.iterator.seek(position);
+ }
+
+ /**
+ * 只保留有重复元素的
+ *
+ * @param
+ */
+ public static class ReduceIterator implements Iterator {
+
+ private E prev = null;
+
+ private E current = null;
+
+ private E data = null;
+
+ private int count = 0;
+
+ private Iterator iterator;
+
+ private Comparator comparator;
+
+ private int adjacent;
+
+ public ReduceIterator(Iterator iterator, Comparator comparator, int adjacent) {
+ this.count = 0;
+ this.iterator = iterator;
+ this.comparator = comparator;
+ this.adjacent = adjacent;
+ }
+
+ /**
+ * 连续重复结果消除. 当prev == current的时候，记录data
+ * 当不等的时候，放回之前的data.
+ * 注意最后的结果，可能重复
+ */
+ @Override
+ public boolean hasNext() {
+ while (iterator.hasNext()) {
+ if (prev == null) {
+ prev = iterator.next();
+ continue;
+ }
+
+ current = iterator.next();
+ if (comparator.compare(prev, current) == 0) {
+ data = current;
+ count += 1;
+ } else {
+ // count starts from 0, so the size is count + 1
+ if (count > 0 && this.adjacent == -1 || count + 1 == this.adjacent) {
+ count = 0;
+ prev = current;
+ return true;
+ } else {
+ count = 0;
+ prev = current;
+ }
+ }
+ }
+
+ // 最后一个结果
+ if (count > 0) {
+ count = 0;
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public E next() {
+ return data;
+ }
+ }
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/IntersectionWrapper.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/IntersectionWrapper.java
new file mode 100644
index 0000000000..e6dd9f398b
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/IntersectionWrapper.java
@@ -0,0 +1,115 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.List;
+import java.util.function.ToLongFunction;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+
+public class IntersectionWrapper {
+
+ private Roaring64Bitmap workBitmap;
+ private Roaring64Bitmap resultBitmap;
+ private ScanIterator iterator;
+ private ToLongFunction hashFunction;
+ private boolean matchAll;
+
+ public IntersectionWrapper(ScanIterator iterator, ToLongFunction hashFunction) {
+ this.iterator = iterator;
+ this.hashFunction = hashFunction;
+ this.workBitmap = new Roaring64Bitmap();
+ this.resultBitmap = new Roaring64Bitmap();
+ this.matchAll = false;
+ }
+
+ /**
+ * 记录iterator中hash值相同的部分.
+ *
+ * @param iterator iterator
+ * @param hashFunction mapping the element to a long value
+ * @param matchAllIterator a value that all exists in the iterator( MultiListIterator)
+ */
+ public IntersectionWrapper(ScanIterator iterator, ToLongFunction hashFunction,
+ boolean matchAllIterator) {
+ this(iterator, hashFunction);
+ this.matchAll = matchAllIterator;
+ }
+
+ public void proc() {
+ if (matchAll && iterator instanceof MultiListIterator) {
+ var mIterators = ((MultiListIterator) iterator).getIterators();
+ if (mIterators.size() > 1) {
+ procMulti(mIterators);
+ }
+ return;
+ }
+
+ procSingle(this.iterator, false);
+ }
+
+ /**
+ * 对于multi list iterator 取所有 iterator的交集
+ *
+ * @param iterators iterators
+ */
+ private void procMulti(List iterators) {
+ var itr = iterators.get(0);
+ procSingle(itr, true);
+
+ for (int i = 1; i < iterators.size(); i++) {
+ // change last round result to the work map
+ workBitmap = resultBitmap.clone();
+ resultBitmap.clear();
+ procSingle(iterators.get(i), false);
+ }
+ }
+
+ private void procSingle(ScanIterator itr, boolean firstRound) {
+ while (itr.hasNext()) {
+ var n = itr.next();
+ if (n == null) {
+ continue;
+ }
+ var key = hashFunction.applyAsLong((T) n);
+
+ if (firstRound) {
+ resultBitmap.add(key);
+ } else {
+ if (workBitmap.contains(key)) {
+ resultBitmap.add(key);
+ } else {
+ workBitmap.add(key);
+ }
+ }
+ }
+ workBitmap.clear();
+ }
+
+ /**
+ * return contains
+ *
+ * @param o input element
+ * @return true: 如果是hash的，可能存在，false则肯定不存在
+ */
+ public boolean contains(T o) {
+ return resultBitmap.contains(hashFunction.applyAsLong(o));
+ }
+}
+
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapJoinIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapJoinIterator.java
new file mode 100644
index 0000000000..1da9b90d4a
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapJoinIterator.java
@@ -0,0 +1,122 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+
+public class MapJoinIterator implements ScanIterator {
+
+ private final List iteratorList;
+
+ private final Function keyFunction;
+
+ private final Map map = new HashMap<>();
+
+ private Iterator iterator;
+
+ private int loc = -1;
+
+ private boolean flag;
+
+ /**
+ * 多个iterator取交集。
+ *
+ * @param iteratorList iterator list
+ * @param loc the location of the iterator having smallest size
+ * @param keyFunction key mapping mapping
+ */
+ public MapJoinIterator(List iteratorList, int loc, Function keyFunction) {
+ assert (iteratorList != null);
+ assert (loc >= 0 && loc < iteratorList.size());
+ this.iteratorList = iteratorList;
+ this.keyFunction = keyFunction;
+ this.loc = loc;
+ this.flag = false;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!flag) {
+ proc();
+ }
+ return this.iterator.hasNext();
+ }
+
+ @Override
+ public boolean isValid() {
+ return true;
+ }
+
+ @Override
+ public T next() {
+ return (T) this.iterator.next();
+ }
+
+ @Override
+ public void close() {
+ iteratorList.forEach(ScanIterator::close);
+ this.map.clear();
+ }
+
+ public void reset() {
+ this.iterator = this.map.values().iterator();
+ }
+
+ private void proc() {
+ var itr = iteratorList.get(loc);
+ while (itr.hasNext()) {
+ var tmp = (T) itr.next();
+ if (tmp != null) {
+ map.put(keyFunction.apply(tmp), tmp);
+ }
+ }
+
+ for (int i = 0; i < iteratorList.size(); i++) {
+
+ if (i == loc) {
+ continue;
+ }
+
+ var workMap = new HashMap();
+
+ itr = iteratorList.get(i);
+ while (itr.hasNext()) {
+ var tmp = (T) itr.next();
+ if (tmp != null) {
+ var key = keyFunction.apply(tmp);
+ if (map.containsKey(key)) {
+ workMap.put(key, tmp);
+ }
+ }
+ }
+
+ map.clear();
+ map.putAll(workMap);
+ }
+
+ this.iterator = this.map.values().iterator();
+
+ this.flag = true;
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapLimitIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapLimitIterator.java
new file mode 100644
index 0000000000..354fc9bf3e
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapLimitIterator.java
@@ -0,0 +1,133 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.Set;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+
+import com.alipay.sofa.jraft.util.concurrent.ConcurrentHashSet;
+
+/**
+ * 针对一个iterator做去重，前SET_MAX_SIZE做精确去重，后面的直接返回
+ *
+ * @param
+ */
+public class MapLimitIterator implements ScanIterator {
+
+ private static final Integer SET_MAX_SIZE = 100000;
+ private ScanIterator iterator;
+ private Set set;
+ private T current = null;
+
+ public MapLimitIterator(ScanIterator iterator) {
+ this.iterator = iterator;
+ set = new ConcurrentHashSet<>();
+ }
+
+ /**
+ * {@inheritDoc}
+ * 返回下一个元素是否存在。
+ * 检查集合中是否还有下一个可用的元素，如果有则返回true，否则返回false。
+ * 如果当前元素为空或者已在集合中，将跳过该元素继续检查下一个元素。
+ * 在检查完所有符合条件的元素后，再次调用hasNext方法会重新检查一遍元素，
+ * 如果满足条件（即不为null且未包含在集合中），则将当前元素添加到集合中并返回true。
+ * 当集合中已经包含SET_MAX_SIZE个元素时，将不会再添加任何新元素，并且返回false。
+ *
+ * @return 下一个元素是否存在
+ */
+ @Override
+ public boolean hasNext() {
+ current = null;
+ while (iterator.hasNext()) {
+ var tmp = (T) iterator.next();
+ if (tmp != null && !set.contains(tmp)) {
+ current = tmp;
+ break;
+ }
+ }
+
+ // 控制set的大小
+ if (current != null && set.size() <= SET_MAX_SIZE) {
+ set.add(current);
+ }
+
+ return current != null;
+ }
+
+ /**
+ * {@inheritDoc}
+ * 返回当前对象。
+ *
+ * @return 当前对象的类型为T1的引用。
+ */
+ @Override
+ public T1 next() {
+ return (T1) current;
+ }
+
+ /**
+ * 返回当前迭代器是否有效。
+ *
+ * @return 当前迭代器是否有效，即是否还有下一个元素。
+ */
+ @Override
+ public boolean isValid() {
+ return iterator.isValid();
+ }
+
+ /**
+ * 返回集合的数量。
+ *
+ * @return 集合的数量。
+ */
+ @Override
+ public long count() {
+ return iterator.count();
+ }
+
+ /**
+ * 返回当前迭代器的位置。
+ *
+ * @return 当前迭代器的位置。
+ */
+ @Override
+ public byte[] position() {
+ return iterator.position();
+ }
+
+ /**
+ * {@inheritDoc}
+ * 将文件指针移动到指定的位置。
+ *
+ * @param position 指定的字节数组，表示要移动到的位置。
+ */
+ @Override
+ public void seek(byte[] position) {
+ iterator.seek(position);
+ }
+
+ /**
+ * 关闭迭代器。
+ */
+ @Override
+ public void close() {
+ iterator.close();
+ this.set.clear();
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapUnionIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapUnionIterator.java
new file mode 100644
index 0000000000..8fc7ecee8f
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapUnionIterator.java
@@ -0,0 +1,82 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+
+public class MapUnionIterator implements ScanIterator {
+
+ private final List iteratorList;
+
+ private final Function keyFunction;
+
+ private final Map map = new HashMap<>();
+
+ private Iterator iterator;
+
+ private boolean flag = false;
+
+ public MapUnionIterator(List iteratorList, Function keyFunction) {
+ this.iteratorList = iteratorList;
+ this.keyFunction = keyFunction;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!this.flag) {
+ this.proc();
+ }
+ return this.iterator.hasNext();
+ }
+
+ @Override
+ public boolean isValid() {
+ return true;
+ }
+
+ @Override
+ public T next() {
+ return (T) this.iterator.next();
+ }
+
+ @Override
+ public void close() {
+ iteratorList.forEach(ScanIterator::close);
+ this.map.clear();
+ }
+
+ private void proc() {
+ for (ScanIterator itr : this.iteratorList) {
+ while (itr.hasNext()) {
+ var item = (T) itr.next();
+ if (item != null) {
+ map.put(keyFunction.apply(item), item);
+ }
+ }
+ }
+
+ this.iterator = map.values().iterator();
+ this.flag = true;
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapValueFilterIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapValueFilterIterator.java
new file mode 100644
index 0000000000..94d9402bbc
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MapValueFilterIterator.java
@@ -0,0 +1,57 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.function.IntPredicate;
+
+public class MapValueFilterIterator implements Iterator {
+
+ Iterator> mapIterator;
+ private IntPredicate filter;
+ private K value;
+
+ public MapValueFilterIterator(Map map, IntPredicate filter) {
+ this.mapIterator = map.entrySet().iterator();
+ this.filter = filter;
+ }
+
+ @Override
+ public boolean hasNext() {
+ while (mapIterator.hasNext()) {
+ Map.Entry entry = mapIterator.next();
+ if (filter.test(entry.getValue())) {
+ value = entry.getKey();
+ return true;
+ }
+ }
+ this.value = null;
+ return false;
+ }
+
+ @Override
+ public K next() {
+ if (value == null) {
+ throw new NoSuchElementException();
+ }
+
+ return value;
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java
new file mode 100644
index 0000000000..909f1fa873
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/MultiListIterator.java
@@ -0,0 +1,158 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.apache.hugegraph.store.business.MultiPartitionIterator;
+
+/**
+ * 一组的同类型iterator，依次按照iterator输出
+ */
+public class MultiListIterator implements ScanIterator {
+
+ /**
+ * iterator 容器
+ */
+ private List iterators;
+
+ /**
+ * iterator 容器的迭代器
+ */
+ private Iterator innerListIterator;
+
+ /**
+ * 当前的element
+ */
+ private ScanIterator innerIterator;
+
+ public MultiListIterator() {
+ this.iterators = new CopyOnWriteArrayList<>();
+ }
+
+ public MultiListIterator(List iterators) {
+ this.iterators = new CopyOnWriteArrayList<>(iterators);
+ }
+
+ /**
+ * 将迭代器添加到扫描迭代器列表中。
+ *
+ * @param iterator 要添加的扫描迭代器。
+ */
+ public void addIterator(ScanIterator iterator) {
+ this.iterators.add(iterator);
+ }
+
+ public List getIterators() {
+ return iterators;
+ }
+
+ /**
+ * 获取内部迭代器
+ */
+ private void getInnerIterator() {
+ if (this.innerIterator != null && this.innerIterator.hasNext()) {
+ return;
+ }
+
+ // close prev one
+ if (this.innerIterator != null) {
+ this.innerIterator.close();
+ }
+
+ if (this.innerListIterator == null) {
+ this.innerListIterator = this.iterators.iterator();
+ }
+
+ while (this.innerListIterator.hasNext()) {
+ this.innerIterator = this.innerListIterator.next();
+ if (this.innerIterator.hasNext()) {
+ return;
+ } else {
+ // whole empty
+ this.innerIterator.close();
+ }
+ }
+
+ this.innerIterator = null;
+ }
+
+ @Override
+ public boolean hasNext() {
+ getInnerIterator();
+ return this.innerIterator != null;
+ }
+
+ @Override
+ public boolean isValid() {
+ getInnerIterator();
+ if (this.innerIterator != null) {
+ return this.innerIterator.isValid();
+ }
+ return true;
+ }
+
+ /**
+ * 关闭迭代器。
+ */
+ @Override
+ public void close() {
+ if (this.innerIterator != null) {
+ this.innerIterator.close();
+ }
+ if (this.innerListIterator != null) {
+ while (this.innerListIterator.hasNext()) {
+ this.innerListIterator.next().close();
+ }
+ }
+ this.iterators.clear();
+ }
+
+ @Override
+ public T next() {
+ return (T) this.innerIterator.next();
+ }
+
+ @Override
+ public long count() {
+ long count = 0;
+ while (hasNext()) {
+ next();
+ count += 1;
+ }
+ return count;
+ }
+
+ @Override
+ public byte[] position() {
+ return this.innerIterator.position();
+ }
+
+ @Override
+ public void seek(byte[] position) {
+ if (this.iterators.size() == 1) {
+ // range scan or prefix scan
+ if (this.innerIterator instanceof MultiPartitionIterator) {
+ this.innerIterator.seek(position);
+ }
+ }
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/TypeTransIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/TypeTransIterator.java
new file mode 100644
index 0000000000..610d495bcc
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/TypeTransIterator.java
@@ -0,0 +1,176 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+
+/**
+ * 封装 iterator， 通过 function 做type的转化,
+ * 最后发送一个supplier.get的指令
+ *
+ * @param 原来的类型
+ * @param 转换后的类型
+ */
+public class TypeTransIterator implements ScanIterator {
+
+ private final Iterator iterator;
+ private final Function function;
+ private String name = "";
+ private ScanIterator originalIterator;
+ private Supplier additionSupplier;
+
+ /**
+ * is used once. return supper. apply and set to true.
+ */
+ private boolean flag = false;
+
+ private E data;
+
+ public TypeTransIterator(ScanIterator scanIterator, Function function) {
+ this.originalIterator = scanIterator;
+ this.iterator = new Iterator() {
+ @Override
+ public boolean hasNext() {
+ return scanIterator.hasNext();
+ }
+
+ @Override
+ public F next() {
+ return scanIterator.next();
+ }
+ };
+ this.function = function;
+ }
+
+ public TypeTransIterator(ScanIterator scanIterator, Function function, String name) {
+ this(scanIterator, function);
+ this.name = name;
+ }
+
+ public TypeTransIterator(Iterator iterator, Function function) {
+ this.iterator = iterator;
+ this.function = function;
+ }
+
+ public TypeTransIterator(Iterator iterator, Function function, Supplier supplier) {
+ this.iterator = iterator;
+ this.function = function;
+ this.additionSupplier = supplier;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (this.data != null) {
+ return true;
+ }
+
+ while (this.iterator.hasNext()) {
+ var n = this.iterator.next();
+ if (n != null && (data = this.function.apply(n)) != null) {
+ return true;
+ }
+ }
+
+ // look up for the default supplier
+ if (this.additionSupplier != null && !this.flag) {
+ data = this.additionSupplier.get();
+ this.flag = true;
+ }
+
+ return data != null;
+ }
+
+ @Override
+ public boolean isValid() {
+ return true;
+ }
+
+ @Override
+ public T next() {
+ if (this.data == null) {
+ throw new NoSuchElementException();
+ }
+ try {
+ return (T) this.data;
+ } finally {
+ // 取出去之后，将data置空
+ this.data = null;
+ }
+ }
+
+ @Override
+ public void close() {
+ if (this.originalIterator != null) {
+ this.originalIterator.close();
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "TypeTransIterator{" +
+ "name='" + name + '\'' +
+ ", function=" + function +
+ ", additionSupplier=" + additionSupplier +
+ ", flag=" + flag +
+ ", iterator=" + (originalIterator == null ? iterator : originalIterator) +
+ '}';
+ }
+
+ /**
+ * to java.util.Iterator
+ *
+ * @return iterator
+ */
+ public Iterator toIterator() {
+ return new InnerIterator(this);
+ }
+
+ private class InnerIterator implements Iterator, ScanIterator {
+
+ private final TypeTransIterator iterator;
+
+ public InnerIterator(TypeTransIterator iterator) {
+ this.iterator = iterator;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.iterator.hasNext();
+ }
+
+ @Override
+ public boolean isValid() {
+ return true;
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ @Override
+ public E next() {
+ return this.iterator.next();
+ }
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/UnionFilterIterator.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/UnionFilterIterator.java
new file mode 100644
index 0000000000..d08bc7dd73
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/UnionFilterIterator.java
@@ -0,0 +1,228 @@
+/*
+ * 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.hugegraph.store.business.itrv2;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import org.apache.hugegraph.pd.common.HgAssert;
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+import org.apache.hugegraph.store.business.itrv2.io.SortShuffleSerializer;
+import org.apache.hugegraph.store.util.SortShuffle;
+
+public class UnionFilterIterator implements ScanIterator {
+
+ private static final Integer MAP_SIZE = 10000;
+
+ private final ScanIterator iterator;
+
+ private final IntersectionWrapper wrapper;
+ private final Comparator comparator;
+ protected Map map;
+ private Iterator innerIterator;
+ private SortShuffle sortShuffle;
+ private SortShuffleSerializer serializer;
+ private Object current;
+ private boolean isProcessed = false;
+
+ public UnionFilterIterator(ScanIterator iterator, IntersectionWrapper wrapper,
+ Comparator comparator, SortShuffleSerializer serializer) {
+ HgAssert.isNotNull(wrapper, "wrapper is null");
+ this.iterator = iterator;
+ this.wrapper = wrapper;
+ this.map = new HashMap<>();
+ this.comparator = comparator;
+ this.serializer = serializer;
+ }
+
+ /**
+ * 将当前元素保存到sortShuffle中。
+ */
+ private void saveElement() {
+ for (var entry : this.map.entrySet()) {
+ try {
+ sortShuffle.append(entry.getKey());
+ if (entry.getValue() > 1) {
+ sortShuffle.append(entry.getKey());
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ this.map.clear();
+ }
+
+ @Override
+ public boolean hasNext() {
+ while (this.iterator.hasNext()) {
+ var obj = (T) this.iterator.next();
+ // batch get or index lookup may generate null
+ if (obj == null) {
+ continue;
+ }
+
+ // 肯定是唯一的
+ if (!wrapper.contains(obj)) {
+ this.current = obj;
+ return true;
+ } else {
+ // System.out.println("----->" + Arrays.toString(((RocksDBSession.BackendColumn)
+ // obj).name));
+ this.map.put(obj, map.getOrDefault(obj, 0) + 1);
+ if (this.map.size() > MAP_SIZE) {
+ if (this.sortShuffle == null) {
+ sortShuffle = new SortShuffle<>(this.comparator, this.serializer);
+ }
+ saveElement();
+ }
+ }
+ }
+
+ if (!isProcessed) {
+ if (sortShuffle != null) {
+ try {
+ saveElement();
+ sortShuffle.finish();
+
+ var fileIterator = sortShuffle.getIterator();
+ this.innerIterator = new NoRepeatValueIterator<>(fileIterator, this.comparator);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ this.innerIterator = new MapValueFilterIterator<>(this.map, x -> x > 0);
+ }
+
+ isProcessed = true;
+ }
+
+ var ret = this.innerIterator.hasNext();
+ if (ret) {
+ this.current = this.innerIterator.next();
+ return true;
+ }
+
+ if (sortShuffle != null) {
+ sortShuffle.close();
+ sortShuffle = null;
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean isValid() {
+ // todo: check logic
+ return this.iterator.isValid() || hasNext();
+ }
+
+ @Override
+ public X next() {
+ if (current == null) {
+ throw new NoSuchElementException();
+ }
+
+ return (X) current;
+ }
+
+ @Override
+ public void close() {
+ this.iterator.close();
+ if (this.sortShuffle != null) {
+ this.sortShuffle.close();
+ }
+ }
+
+ @Override
+ public long count() {
+ return this.iterator.count();
+ }
+
+ @Override
+ public byte[] position() {
+ return this.iterator.position();
+ }
+
+ @Override
+ public void seek(byte[] position) {
+ this.iterator.seek(position);
+ }
+
+ private static class NoRepeatValueIterator implements Iterator {
+
+ private final Iterator iterator;
+ private final Comparator comparator;
+ private E prev = null;
+ private E data = null;
+ private int count = 0;
+
+ public NoRepeatValueIterator(Iterator iterator, Comparator comparator) {
+ this.count = 0;
+ this.iterator = iterator;
+ this.comparator = comparator;
+ }
+
+ @Override
+ public boolean hasNext() {
+ while (iterator.hasNext()) {
+ var n = iterator.next();
+ if (prev == null) {
+ // prev = iterator.next();
+ prev = n;
+ continue;
+ }
+
+ // E current = iterator.next();
+ E current = n;
+
+ if (comparator.compare(prev, current) == 0) {
+ count += 1;
+ } else {
+ if (count > 0) {
+ // --- pre is dup
+ prev = current;
+ } else {
+ data = prev;
+ prev = current;
+ return true;
+ }
+ count = 0;
+ }
+ }
+
+ // 最后一个结果
+ if (count == 0) {
+ data = prev;
+ count = 1;
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public E next() {
+ return data;
+ }
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/io/SortShuffleSerializer.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/io/SortShuffleSerializer.java
new file mode 100644
index 0000000000..c321c33a14
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/itrv2/io/SortShuffleSerializer.java
@@ -0,0 +1,275 @@
+/*
+ * 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.hugegraph.store.business.itrv2.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+import org.apache.hugegraph.backend.BackendColumn;
+import org.apache.hugegraph.rocksdb.access.RocksDBSession;
+import org.apache.hugegraph.serializer.BinaryElementSerializer;
+import org.apache.hugegraph.store.query.KvSerializer;
+import org.apache.hugegraph.store.query.Tuple2;
+import org.apache.hugegraph.store.util.MultiKv;
+import org.apache.hugegraph.structure.BaseEdge;
+import org.apache.hugegraph.structure.BaseElement;
+import org.apache.hugegraph.structure.BaseVertex;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * support backend column, Multi kv, BaseElement
+ * format : object | object | object
+ * todo: need write object type header ?
+ *
+ * @param object type
+ */
+@Slf4j
+public class SortShuffleSerializer {
+
+ private static final byte TYPE_HEADER_MULTI_KV = 1;
+ private static final byte TYPE_HEADER_BACKEND_COLUMN = 2;
+ private static final byte TYPE_HEADER_BASE_ELEMENT = 3;
+
+ private static SortShuffleSerializer backendSerializer =
+ new SortShuffleSerializer<>(new BackendColumnSerializer());
+
+ private static SortShuffleSerializer mkv =
+ new SortShuffleSerializer<>(new MultiKvSerializer());
+
+ private static SortShuffleSerializer element =
+ new SortShuffleSerializer<>(new BaseElementSerializer());
+
+ private final ObjectSerializer serializer;
+
+ private SortShuffleSerializer(ObjectSerializer serializer) {
+ this.serializer = serializer;
+ }
+
+ public static SortShuffleSerializer ofBackendColumnSerializer() {
+ return backendSerializer;
+ }
+
+ public static SortShuffleSerializer ofMultiKvSerializer() {
+ return mkv;
+ }
+
+ public static SortShuffleSerializer ofBaseElementSerializer() {
+ return element;
+ }
+
+ public static byte[] toByte(int i) {
+ byte[] result = new byte[4];
+ result[0] = (byte) ((i >> 24) & 0xff);
+ result[1] = (byte) ((i >> 16) & 0xff);
+ result[2] = (byte) ((i >> 8) & 0xff);
+ result[3] = (byte) (i & 0xff);
+ return result;
+ }
+
+ public static int toInt(byte[] b) {
+ assert b.length == 4;
+ int value = 0;
+ for (int i = 0; i < 4; i++) {
+ int shift = (3 - i) * 8;
+ value += (b[i] & 0xff) << shift;
+ }
+ return value;
+ }
+
+ private static byte[] kvBytesToByte(byte[] key, byte[] value) {
+
+ int len = (key == null ? 0 : key.length) + (value == null ? 0 : value.length) + 8;
+ ByteBuffer buffer = ByteBuffer.allocate(len);
+ buffer.putInt(key == null ? 0 : key.length);
+ if (key != null) {
+ buffer.put(key);
+ }
+ buffer.putInt(value == null ? 0 : value.length);
+ if (value != null) {
+ buffer.put(value);
+ }
+ return buffer.array();
+ }
+
+ private static Tuple2 fromKvBytes(byte[] bytes) {
+ assert bytes != null;
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+
+ int nameLen = buffer.getInt();
+ byte[] name = null;
+ if (nameLen != 0) {
+ name = new byte[nameLen];
+ buffer.get(name);
+ }
+
+ int valueLen = buffer.getInt();
+ byte[] value = null;
+ if (valueLen != 0) {
+ value = new byte[valueLen];
+ buffer.get(value);
+ }
+
+ return Tuple2.of(name, value);
+ }
+
+ public void write(OutputStream output, T data) throws IOException {
+ // input.write(serializer.getTypeHeader());
+ var b = serializer.getBytes(data);
+ output.write(toByte(b.length));
+ output.write(b);
+ }
+
+ public T read(InputStream input) {
+ try {
+ var bytes = input.readNBytes(4);
+
+ if (bytes.length == 0) {
+ return null;
+ }
+
+ int sz = toInt(bytes);
+ return serializer.fromBytes(input.readNBytes(sz));
+ } catch (IOException e) {
+ log.debug("error: {}", e.getMessage());
+ return null;
+ }
+ }
+
+ private abstract static class ObjectSerializer {
+
+ public abstract T fromBytes(byte[] bytes);
+
+ public abstract byte[] getBytes(T t);
+
+ public abstract byte getTypeHeader();
+ }
+
+ /**
+ * format :
+ * key bytes len| key | value bytes len | value bytes
+ */
+
+ private static class MultiKvSerializer extends ObjectSerializer {
+
+ @Override
+ public MultiKv fromBytes(byte[] bytes) {
+ var tuple = fromKvBytes(bytes);
+ return MultiKv.of(KvSerializer.fromObjectBytes(tuple.getV1()),
+ KvSerializer.fromObjectBytes(tuple.getV2()));
+ }
+
+ @Override
+ public byte[] getBytes(MultiKv multiKv) {
+ return kvBytesToByte(KvSerializer.toBytes(multiKv.getKeys()),
+ KvSerializer.toBytes(multiKv.getValues()));
+ }
+
+ @Override
+ public byte getTypeHeader() {
+ return TYPE_HEADER_MULTI_KV;
+ }
+ }
+
+ /**
+ * format:
+ * name.len | name | value.len | value
+ */
+ private static class BackendColumnSerializer extends
+ ObjectSerializer {
+
+ @Override
+ public RocksDBSession.BackendColumn fromBytes(byte[] bytes) {
+ var tuple = fromKvBytes(bytes);
+ return RocksDBSession.BackendColumn.of(tuple.getV1(), tuple.getV2());
+ }
+
+ @Override
+ public byte[] getBytes(RocksDBSession.BackendColumn column) {
+ return kvBytesToByte(column.name, column.value);
+ }
+
+ @Override
+ public byte getTypeHeader() {
+ return TYPE_HEADER_BACKEND_COLUMN;
+ }
+ }
+
+ /**
+ * format:
+ * vertex/edge | name.len | name | value.len | value
+ */
+ private static class BaseElementSerializer extends ObjectSerializer {
+
+ private final BinaryElementSerializer serializer = new BinaryElementSerializer();
+
+ @Override
+ public BaseElement fromBytes(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+
+ boolean isVertex = buffer.get() == 0;
+
+ int nameLen = buffer.getInt();
+ byte[] name = new byte[nameLen];
+ buffer.get(name);
+ int valueLen = buffer.getInt();
+ byte[] value = new byte[valueLen];
+ buffer.get(value);
+
+ if (isVertex) {
+ return serializer.parseVertex(null, BackendColumn.of(name, value), null);
+ }
+ return serializer.parseEdge(null, BackendColumn.of(name, value), null, true);
+ }
+
+ @Override
+ public byte[] getBytes(BaseElement element) {
+ assert element != null;
+
+ BackendColumn column;
+ boolean isVertex = false;
+ if (element instanceof BaseVertex) {
+ column = serializer.writeVertex((BaseVertex) element);
+ isVertex = true;
+ } else {
+ column = serializer.writeEdge((BaseEdge) element);
+ }
+
+ ByteBuffer buffer = ByteBuffer.allocate(column.name.length + column.value.length + 9);
+ if (isVertex) {
+ buffer.put((byte) 0);
+ } else {
+ buffer.put((byte) 1);
+ }
+
+ buffer.putInt(column.name.length);
+ buffer.put(column.name);
+ buffer.putInt(column.value.length);
+ buffer.put(column.value);
+ return buffer.array();
+ }
+
+ @Override
+ public byte getTypeHeader() {
+ return TYPE_HEADER_BASE_ELEMENT;
+ }
+ }
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutRequest.java
index a776e6d4e1..ba51f82759 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutRequest.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutRequest.java
@@ -23,6 +23,7 @@
import lombok.Data;
@Data
+@Deprecated
public class BatchPutRequest extends HgCmdBase.BaseRequest {
private List entries;
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutResponse.java
index 98a72f5655..ad2bd4b638 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutResponse.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/BatchPutResponse.java
@@ -17,6 +17,7 @@
package org.apache.hugegraph.store.cmd;
+@Deprecated
public class BatchPutResponse extends HgCmdBase.BaseResponse {
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CleanDataResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CleanDataResponse.java
index f7773075de..ce4f5fa98a 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CleanDataResponse.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CleanDataResponse.java
@@ -16,7 +16,7 @@
*/
package org.apache.hugegraph.store.cmd;
-
+@Deprecated
public class CleanDataResponse extends HgCmdBase.BaseResponse {
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftRequest.java
index be5c384205..ad3cb063b4 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftRequest.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftRequest.java
@@ -28,6 +28,7 @@
import lombok.extern.slf4j.Slf4j;
@Slf4j
+@Deprecated
public class CreateRaftRequest extends HgCmdBase.BaseRequest {
List values = new ArrayList<>();
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftResponse.java
index 9e14ffc97d..8bfdf9c551 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftResponse.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CreateRaftResponse.java
@@ -17,6 +17,7 @@
package org.apache.hugegraph.store.cmd;
+@Deprecated
public class CreateRaftResponse extends HgCmdBase.BaseResponse {
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DbCompactionResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DbCompactionResponse.java
index 228aae1078..ae589f212f 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DbCompactionResponse.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DbCompactionResponse.java
@@ -17,6 +17,7 @@
package org.apache.hugegraph.store.cmd;
+@Deprecated
public class DbCompactionResponse extends HgCmdBase.BaseResponse {
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DestroyRaftResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DestroyRaftResponse.java
index cb24b2fc49..8d015f2f94 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DestroyRaftResponse.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DestroyRaftResponse.java
@@ -17,6 +17,7 @@
package org.apache.hugegraph.store.cmd;
+@Deprecated
public class DestroyRaftResponse extends HgCmdBase.BaseResponse {
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/FutureClosureAdapter.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/FutureClosureAdapter.java
index 8579a7d4f1..b7633160d4 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/FutureClosureAdapter.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/FutureClosureAdapter.java
@@ -22,6 +22,7 @@
import com.alipay.sofa.jraft.Closure;
import com.alipay.sofa.jraft.Status;
+@Deprecated
public class FutureClosureAdapter implements Closure {
public final CompletableFuture future = new CompletableFuture<>();
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/GetStoreInfoResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/GetStoreInfoResponse.java
index df32cd99fe..7ed2e3d054 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/GetStoreInfoResponse.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/GetStoreInfoResponse.java
@@ -25,6 +25,7 @@
import lombok.extern.slf4j.Slf4j;
@Slf4j
+@Deprecated
public class GetStoreInfoResponse extends HgCmdBase.BaseResponse {
private byte[] store;
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdBase.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdBase.java
index b612f3fc44..0f7923e210 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdBase.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdBase.java
@@ -32,6 +32,10 @@ public class HgCmdBase {
public static final byte ROCKSDB_COMPACTION = 0x05;
public static final byte CREATE_RAFT = 0x06;
public static final byte DESTROY_RAFT = 0x07;
+ public static final byte TTL_CLEAN = 0x08;
+ public static final byte BLANK_TASK = 0x09;
+
+ public static final byte REDIRECT_RAFT_TASK = 0x10;
@Data
public abstract static class BaseRequest implements Serializable {
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdClient.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdClient.java
index 6a73639e67..890dfe095f 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdClient.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdClient.java
@@ -23,6 +23,18 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.hugegraph.store.cmd.request.BatchPutRequest;
+import org.apache.hugegraph.store.cmd.request.CleanDataRequest;
+import org.apache.hugegraph.store.cmd.request.CreateRaftRequest;
+import org.apache.hugegraph.store.cmd.request.DestroyRaftRequest;
+import org.apache.hugegraph.store.cmd.request.GetStoreInfoRequest;
+import org.apache.hugegraph.store.cmd.request.RedirectRaftTaskRequest;
+import org.apache.hugegraph.store.cmd.request.UpdatePartitionRequest;
+import org.apache.hugegraph.store.cmd.response.BatchPutResponse;
+import org.apache.hugegraph.store.cmd.response.CleanDataResponse;
+import org.apache.hugegraph.store.cmd.response.GetStoreInfoResponse;
+import org.apache.hugegraph.store.cmd.response.RedirectRaftTaskResponse;
+import org.apache.hugegraph.store.cmd.response.UpdatePartitionResponse;
import org.apache.hugegraph.store.meta.Partition;
import org.apache.hugegraph.store.meta.Store;
@@ -137,6 +149,10 @@ public UpdatePartitionResponse raftUpdatePartition(UpdatePartitionRequest reques
return (UpdatePartitionResponse) tryInternalCallSyncWithRpc(request);
}
+ public RedirectRaftTaskResponse redirectRaftTask(RedirectRaftTaskRequest request) {
+ return (RedirectRaftTaskResponse) tryInternalCallSyncWithRpc(request);
+ }
+
/**
* Find Leader, retry on error, handle Leader redirection
*
@@ -165,6 +181,7 @@ public HgCmdBase.BaseResponse tryInternalCallSyncWithRpc(HgCmdBase.BaseRequest r
&& response.partitionLeaders != null
) {
// When returning leader drift, and partitionLeaders is not empty, need to reset the leader.
+ Thread.sleep(i * 1000L);
} else {
log.error(
"HgCmdClient tryInternalCallSyncWithRpc error msg {} leaders is {}",
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdProcessor.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdProcessor.java
index e0710ef97e..56bc7918bc 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdProcessor.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/HgCmdProcessor.java
@@ -22,6 +22,23 @@
import java.util.concurrent.TimeUnit;
import org.apache.hugegraph.store.HgStoreEngine;
+import org.apache.hugegraph.store.PartitionEngine;
+import org.apache.hugegraph.store.cmd.request.BatchPutRequest;
+import org.apache.hugegraph.store.cmd.request.BlankTaskRequest;
+import org.apache.hugegraph.store.cmd.request.CleanDataRequest;
+import org.apache.hugegraph.store.cmd.request.CreateRaftRequest;
+import org.apache.hugegraph.store.cmd.request.DestroyRaftRequest;
+import org.apache.hugegraph.store.cmd.request.GetStoreInfoRequest;
+import org.apache.hugegraph.store.cmd.request.RedirectRaftTaskRequest;
+import org.apache.hugegraph.store.cmd.request.UpdatePartitionRequest;
+import org.apache.hugegraph.store.cmd.response.BatchPutResponse;
+import org.apache.hugegraph.store.cmd.response.CleanDataResponse;
+import org.apache.hugegraph.store.cmd.response.CreateRaftResponse;
+import org.apache.hugegraph.store.cmd.response.DefaultResponse;
+import org.apache.hugegraph.store.cmd.response.DestroyRaftResponse;
+import org.apache.hugegraph.store.cmd.response.GetStoreInfoResponse;
+import org.apache.hugegraph.store.cmd.response.RedirectRaftTaskResponse;
+import org.apache.hugegraph.store.cmd.response.UpdatePartitionResponse;
import org.apache.hugegraph.store.meta.Partition;
import org.apache.hugegraph.store.raft.RaftClosure;
import org.apache.hugegraph.store.raft.RaftOperation;
@@ -56,6 +73,8 @@ public static void registerProcessor(final RpcServer rpcServer, final HgStoreEng
rpcServer.registerProcessor(new HgCmdProcessor<>(UpdatePartitionRequest.class, engine));
rpcServer.registerProcessor(new HgCmdProcessor<>(CreateRaftRequest.class, engine));
rpcServer.registerProcessor(new HgCmdProcessor<>(DestroyRaftRequest.class, engine));
+ rpcServer.registerProcessor(new HgCmdProcessor<>(BlankTaskRequest.class, engine));
+ rpcServer.registerProcessor(new HgCmdProcessor<>(ProcessBuilder.Redirect.class, engine));
}
@Override
@@ -93,6 +112,17 @@ public void handleRequest(RpcContext rpcCtx, T request) {
handleDestroyRaft((DestroyRaftRequest) request, (DestroyRaftResponse) response);
break;
}
+ case HgCmdBase.BLANK_TASK: {
+ response = new DefaultResponse();
+ addBlankTask((BlankTaskRequest) request, (DefaultResponse) response);
+ break;
+ }
+ case HgCmdBase.REDIRECT_RAFT_TASK: {
+ response = new RedirectRaftTaskResponse();
+ handleRedirectRaftTask((RedirectRaftTaskRequest) request,
+ (RedirectRaftTaskResponse) response);
+ break;
+ }
default: {
log.warn("HgCmdProcessor magic {} is not recognized ", request.magic());
}
@@ -138,6 +168,39 @@ public void handleDestroyRaft(DestroyRaftRequest request, DestroyRaftResponse re
response.setStatus(Status.OK);
}
+ public void handleRedirectRaftTask(RedirectRaftTaskRequest request,
+ RedirectRaftTaskResponse response) {
+ log.info("RedirectRaftTaskNode rpc call received, {}", request.getPartitionId());
+ raftSyncTask(request.getGraphName(), request.getPartitionId(), request.getRaftOp(),
+ request.getData(), response);
+ response.setStatus(Status.OK);
+ }
+
+ public void addBlankTask(BlankTaskRequest request, DefaultResponse response) {
+ try {
+ int partitionId = request.getPartitionId();
+ PartitionEngine pe = engine.getPartitionEngine(partitionId);
+ if (pe.isLeader()) {
+ CountDownLatch latch = new CountDownLatch(1);
+ RaftClosure closure = s -> {
+ if (s.isOk()) {
+ response.setStatus(Status.OK);
+ } else {
+ log.error("doBlankTask in cmd with error: {}", s.getErrorMsg());
+ response.setStatus(Status.EXCEPTION);
+ }
+ latch.countDown();
+ };
+ pe.addRaftTask(RaftOperation.create(RaftOperation.SYNC_BLANK_TASK), closure);
+ latch.await();
+ } else {
+ response.setStatus(Status.LEADER_REDIRECT);
+ }
+ } catch (Exception e) {
+ response.setStatus(Status.EXCEPTION);
+ }
+ }
+
/**
* raft notify replica synchronization execution
*
@@ -147,9 +210,14 @@ public void handleDestroyRaft(DestroyRaftRequest request, DestroyRaftResponse re
*/
private void raftSyncTask(HgCmdBase.BaseRequest request, HgCmdBase.BaseResponse response,
final byte op) {
+ raftSyncTask(request.getGraphName(), request.getPartitionId(), op, request, response);
+ }
+
+ private void raftSyncTask(String graph, int partId, byte op, Object raftReq,
+ HgCmdBase.BaseResponse response) {
CountDownLatch latch = new CountDownLatch(1);
- engine.addRaftTask(request.getGraphName(), request.getPartitionId(),
- RaftOperation.create(op, request), new RaftClosure() {
+ engine.addRaftTask(graph, partId,
+ RaftOperation.create(op, raftReq), new RaftClosure() {
@Override
public void run(com.alipay.sofa.jraft.Status status) {
Status responseStatus = Status.UNKNOWN;
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionRequest.java
index 016b162870..12dbc372ce 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionRequest.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionRequest.java
@@ -22,6 +22,7 @@
import lombok.Data;
@Data
+@Deprecated
public class UpdatePartitionRequest extends HgCmdBase.BaseRequest {
private int startKey;
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionResponse.java
index 49bb1c7cb5..5cec121442 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionResponse.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/UpdatePartitionResponse.java
@@ -17,6 +17,7 @@
package org.apache.hugegraph.store.cmd;
+@Deprecated
public class UpdatePartitionResponse extends HgCmdBase.BaseResponse {
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/BatchPutRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/BatchPutRequest.java
new file mode 100644
index 0000000000..1e09424da1
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/BatchPutRequest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.hugegraph.store.cmd.request;
+
+import java.io.Serializable;
+import java.util.List;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+import lombok.Data;
+
+@Data
+public class BatchPutRequest extends HgCmdBase.BaseRequest {
+
+ private List entries;
+
+ @Override
+ public byte magic() {
+ return HgCmdBase.BATCH_PUT;
+ }
+
+ @Data
+ public static class KV implements Serializable {
+
+ private String table;
+ private int code;
+ private byte[] key;
+ private byte[] value;
+
+ public static KV of(String table, int code, byte[] key, byte[] value) {
+ KV kv = new KV();
+ kv.table = table;
+ kv.code = code;
+ kv.key = key;
+ kv.value = value;
+ return kv;
+ }
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/BlankTaskRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/BlankTaskRequest.java
new file mode 100644
index 0000000000..08abc59a21
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/BlankTaskRequest.java
@@ -0,0 +1,32 @@
+/*
+ * 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.hugegraph.store.cmd.request;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+/**
+ * @author zhangyingjie
+ * @date 2023/8/21
+ **/
+public class BlankTaskRequest extends HgCmdBase.BaseRequest {
+
+ @Override
+ public byte magic() {
+ return HgCmdBase.BLANK_TASK;
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CleanDataRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/CleanDataRequest.java
similarity index 96%
rename from hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CleanDataRequest.java
rename to hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/CleanDataRequest.java
index 35540687bf..1fbfb5656d 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/CleanDataRequest.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/CleanDataRequest.java
@@ -15,10 +15,11 @@
* limitations under the License.
*/
-package org.apache.hugegraph.store.cmd;
+package org.apache.hugegraph.store.cmd.request;
import org.apache.hugegraph.pd.grpc.pulse.CleanPartition;
import org.apache.hugegraph.pd.grpc.pulse.CleanType;
+import org.apache.hugegraph.store.cmd.HgCmdBase;
import org.apache.hugegraph.store.meta.Partition;
import lombok.Data;
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/CreateRaftRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/CreateRaftRequest.java
new file mode 100644
index 0000000000..1897c850c8
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/CreateRaftRequest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.hugegraph.store.cmd.request;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hugegraph.pd.grpc.Metapb;
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+import com.alipay.sofa.jraft.conf.Configuration;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CreateRaftRequest extends HgCmdBase.BaseRequest {
+
+ List values = new ArrayList<>();
+ String peers;
+
+ public List getPartitions() {
+ try {
+ List partitions = new ArrayList<>();
+ for (byte[] partition : values) {
+ partitions.add(Metapb.Partition.parseFrom(partition));
+ }
+ return partitions;
+ } catch (InvalidProtocolBufferException e) {
+ log.error("CreateRaftNodeProcessor parse partition exception }", e);
+ }
+ return new ArrayList<>();
+ }
+
+ public void addPartition(Metapb.Partition partition) {
+ values.add(partition.toByteArray());
+ }
+
+ public Configuration getConf() {
+ Configuration conf = null;
+ if (peers != null) {
+ conf = new Configuration();
+ conf.parse(this.peers);
+ }
+ return conf;
+ }
+
+ public void setConf(Configuration conf) {
+ if (conf != null) {
+ this.peers = conf.toString();
+ }
+ }
+
+ @Override
+ public byte magic() {
+ return HgCmdBase.CREATE_RAFT;
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DbCompactionRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/DbCompactionRequest.java
similarity index 91%
rename from hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DbCompactionRequest.java
rename to hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/DbCompactionRequest.java
index 7952f170d1..5da60f0bf0 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DbCompactionRequest.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/DbCompactionRequest.java
@@ -15,7 +15,9 @@
* limitations under the License.
*/
-package org.apache.hugegraph.store.cmd;
+package org.apache.hugegraph.store.cmd.request;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
import lombok.Data;
@@ -29,3 +31,4 @@ public byte magic() {
return HgCmdBase.ROCKSDB_COMPACTION;
}
}
+
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DestroyRaftRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/DestroyRaftRequest.java
similarity index 87%
rename from hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DestroyRaftRequest.java
rename to hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/DestroyRaftRequest.java
index 10bf1c30b7..ecd7e7cf0e 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/DestroyRaftRequest.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/DestroyRaftRequest.java
@@ -15,17 +15,19 @@
* limitations under the License.
*/
-package org.apache.hugegraph.store.cmd;
+package org.apache.hugegraph.store.cmd.request;
import java.util.ArrayList;
import java.util.List;
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
import lombok.Data;
@Data
public class DestroyRaftRequest extends HgCmdBase.BaseRequest {
- private final List graphNames = new ArrayList<>();
+ private List graphNames = new ArrayList<>();
public void addGraphName(String graphName) {
graphNames.add(graphName);
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/GetStoreInfoRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/GetStoreInfoRequest.java
similarity index 90%
rename from hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/GetStoreInfoRequest.java
rename to hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/GetStoreInfoRequest.java
index 68f0d7f329..0b194a5051 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/GetStoreInfoRequest.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/GetStoreInfoRequest.java
@@ -15,7 +15,9 @@
* limitations under the License.
*/
-package org.apache.hugegraph.store.cmd;
+package org.apache.hugegraph.store.cmd.request;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
public class GetStoreInfoRequest extends HgCmdBase.BaseRequest {
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/RedirectRaftTaskRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/RedirectRaftTaskRequest.java
new file mode 100644
index 0000000000..efb430a696
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/RedirectRaftTaskRequest.java
@@ -0,0 +1,42 @@
+/*
+ * 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.hugegraph.store.cmd.request;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+import lombok.Data;
+
+@Data
+public class RedirectRaftTaskRequest extends HgCmdBase.BaseRequest {
+
+ final byte raftOp;
+
+ private Object data;
+
+ public RedirectRaftTaskRequest(String graph, Integer partitionId, byte raftOp, Object data) {
+ this.raftOp = raftOp;
+ this.data = data;
+ setGraphName(graph);
+ setPartitionId(partitionId);
+ }
+
+ @Override
+ public byte magic() {
+ return HgCmdBase.REDIRECT_RAFT_TASK;
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/UpdatePartitionRequest.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/UpdatePartitionRequest.java
new file mode 100644
index 0000000000..430756178a
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/request/UpdatePartitionRequest.java
@@ -0,0 +1,37 @@
+/*
+ * 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.hugegraph.store.cmd.request;
+
+import org.apache.hugegraph.pd.grpc.Metapb;
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+import lombok.Data;
+
+@Data
+public class UpdatePartitionRequest extends HgCmdBase.BaseRequest {
+
+ private int startKey;
+ private int endKey;
+
+ private Metapb.PartitionState workState;
+
+ @Override
+ public byte magic() {
+ return HgCmdBase.RAFT_UPDATE_PARTITION;
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/BatchPutResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/BatchPutResponse.java
new file mode 100644
index 0000000000..c687a1c8ea
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/BatchPutResponse.java
@@ -0,0 +1,24 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+public class BatchPutResponse extends HgCmdBase.BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/CleanDataResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/CleanDataResponse.java
new file mode 100644
index 0000000000..cfa9454166
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/CleanDataResponse.java
@@ -0,0 +1,24 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+public class CleanDataResponse extends HgCmdBase.BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/CreateRaftResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/CreateRaftResponse.java
new file mode 100644
index 0000000000..c58dddfe1c
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/CreateRaftResponse.java
@@ -0,0 +1,24 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+public class CreateRaftResponse extends HgCmdBase.BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DbCompactionResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DbCompactionResponse.java
new file mode 100644
index 0000000000..5c81833aa2
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DbCompactionResponse.java
@@ -0,0 +1,24 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+public class DbCompactionResponse extends HgCmdBase.BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DefaultResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DefaultResponse.java
new file mode 100644
index 0000000000..bfa27cbe49
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DefaultResponse.java
@@ -0,0 +1,28 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase.BaseResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2023/8/21
+ **/
+public class DefaultResponse extends BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DestroyRaftResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DestroyRaftResponse.java
new file mode 100644
index 0000000000..0e037e0435
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/DestroyRaftResponse.java
@@ -0,0 +1,24 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+public class DestroyRaftResponse extends HgCmdBase.BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/GetStoreInfoResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/GetStoreInfoResponse.java
new file mode 100644
index 0000000000..779c2785de
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/GetStoreInfoResponse.java
@@ -0,0 +1,45 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.pd.grpc.Metapb;
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+import org.apache.hugegraph.store.meta.Store;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class GetStoreInfoResponse extends HgCmdBase.BaseResponse {
+
+ private byte[] store;
+
+ public Store getStore() {
+ try {
+ return new Store(Metapb.Store.parseFrom(this.store));
+ } catch (InvalidProtocolBufferException e) {
+ log.error("GetStoreResponse parse exception {}", e);
+ }
+ return null;
+ }
+
+ public void setStore(Store store) {
+ this.store = store.getProtoObj().toByteArray();
+ }
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/RedirectRaftTaskResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/RedirectRaftTaskResponse.java
new file mode 100644
index 0000000000..9ee7ca45c5
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/RedirectRaftTaskResponse.java
@@ -0,0 +1,24 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+public class RedirectRaftTaskResponse extends HgCmdBase.BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/UpdatePartitionResponse.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/UpdatePartitionResponse.java
new file mode 100644
index 0000000000..9901ab5428
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/cmd/response/UpdatePartitionResponse.java
@@ -0,0 +1,24 @@
+/*
+ * 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.hugegraph.store.cmd.response;
+
+import org.apache.hugegraph.store.cmd.HgCmdBase;
+
+public class UpdatePartitionResponse extends HgCmdBase.BaseResponse {
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/consts/PoolNames.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/consts/PoolNames.java
new file mode 100644
index 0000000000..b3d2ed586a
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/consts/PoolNames.java
@@ -0,0 +1,35 @@
+/*
+ * 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.hugegraph.store.consts;
+
+/**
+ * @author zhangyingjie
+ * @date 2023/10/30
+ **/
+public class PoolNames {
+
+ public static final String GRPC = "hg-grpc";
+ public static final String SCAN = "hg-scan";
+ public static final String SCAN_V2 = "hg-scan-v2";
+ public static final String I_JOB = "hg-i-job";
+ public static final String U_JOB = "hg-u-job";
+ public static final String COMPACT = "hg-compact";
+ public static final String HEARTBEAT = "hg-heartbeat";
+ public static final String P_HEARTBEAT = "hg-p-heartbeat";
+
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/PartitionChangedListener.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/PartitionChangedListener.java
new file mode 100644
index 0000000000..5b4e5f80f4
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/PartitionChangedListener.java
@@ -0,0 +1,34 @@
+/*
+ * 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.hugegraph.store.listener;
+
+import org.apache.hugegraph.store.cmd.request.UpdatePartitionRequest;
+import org.apache.hugegraph.store.cmd.response.UpdatePartitionResponse;
+import org.apache.hugegraph.store.meta.Partition;
+
+/**
+ * @author zhangyingjie
+ * @date 2023/9/11
+ * Partition对象被修改消息
+ **/
+public interface PartitionChangedListener {
+
+ void onChanged(Partition partition);
+
+ UpdatePartitionResponse rangeOrStateChanged(UpdatePartitionRequest request);
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/PartitionStateListener.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/PartitionStateListener.java
new file mode 100644
index 0000000000..06b9918005
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/PartitionStateListener.java
@@ -0,0 +1,34 @@
+/*
+ * 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.hugegraph.store.listener;
+
+import java.util.List;
+
+import org.apache.hugegraph.pd.grpc.Metapb;
+import org.apache.hugegraph.store.meta.Partition;
+import org.apache.hugegraph.store.meta.PartitionRole;
+
+public interface PartitionStateListener {
+
+ // 分区角色发生改变
+ void partitionRoleChanged(Partition partition, PartitionRole newRole);
+
+ // 分区发生改变
+ void partitionShardChanged(Partition partition, List oldShards,
+ List newShards);
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/StoreStateListener.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/StoreStateListener.java
new file mode 100644
index 0000000000..11c607338b
--- /dev/null
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/listener/StoreStateListener.java
@@ -0,0 +1,26 @@
+/*
+ * 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.hugegraph.store.listener;
+
+import org.apache.hugegraph.pd.grpc.Metapb;
+import org.apache.hugegraph.store.meta.Store;
+
+public interface StoreStateListener {
+
+ void stateChanged(Store store, Metapb.StoreState oldState, Metapb.StoreState newState);
+}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java
index c98b03935d..92a8440ac3 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java
@@ -18,19 +18,25 @@
package org.apache.hugegraph.store.meta;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
import org.apache.hugegraph.store.meta.base.DBSessionBuilder;
import org.apache.hugegraph.store.meta.base.PartitionMetaStore;
+import org.apache.hugegraph.store.term.Bits;
import org.apache.hugegraph.store.util.HgStoreException;
import com.google.protobuf.Int64Value;
+import lombok.extern.slf4j.Slf4j;
+
/**
* GraphId Manager, maintains a self-incrementing circular ID, responsible for managing the mapping between GraphName and GraphId.
*/
+@Slf4j
public class GraphIdManager extends PartitionMetaStore {
protected static final String GRAPH_ID_PREFIX = "@GRAPH_ID@";
@@ -39,27 +45,6 @@ public class GraphIdManager extends PartitionMetaStore {
static Object cidLock = new Object();
final DBSessionBuilder sessionBuilder;
final int partitionId;
- // public long getGraphId(String graphName) {
- // if (!graphIdCache.containsKey(graphName)) {
- // synchronized (graphIdLock) {
- // if (!graphIdCache.containsKey(graphName)) {
- // byte[] key = MetadataKeyHelper.getGraphIDKey(graphName);
- // Int64Value id = get(Int64Value.parser(), key);
- // if (id == null) {
- // id = Int64Value.of(getCId(GRAPH_ID_PREFIX, maxGraphID));
- // if (id.getValue() == -1) {
- // throw new HgStoreException(HgStoreException.EC_FAIL,
- // "The number of graphs exceeds the maximum 65535");
- // }
- // put(key, id);
- // flush();
- // }
- // graphIdCache.put(graphName, id.getValue());
- // }
- // }
- // }
- // return graphIdCache.get(graphName);
- // }
private final Map graphIdCache = new ConcurrentHashMap<>();
public GraphIdManager(DBSessionBuilder sessionBuilder, int partitionId) {
@@ -79,12 +64,34 @@ public long getGraphId(String graphName) {
byte[] key = MetadataKeyHelper.getGraphIDKey(graphName);
Int64Value id = get(Int64Value.parser(), key);
if (id == null) {
- id = Int64Value.of(getCId(GRAPH_ID_PREFIX, maxGraphID));
+ id = Int64Value.of(maxGraphID);
+ }
+ l = id.getValue();
+ graphIdCache.put(graphName, l);
+ }
+ }
+ }
+ return l;
+ }
+
+ public long getGraphIdOrCreate(String graphName) {
+
+ Long l = graphIdCache.get(graphName);
+ if (l == null || l == maxGraphID) {
+ synchronized (graphIdLock) {
+ if ((l = graphIdCache.get(graphName)) == null || l == maxGraphID) {
+ byte[] key = MetadataKeyHelper.getGraphIDKey(graphName);
+ Int64Value id = get(Int64Value.parser(), key);
+ if (id == null) {
+ id = Int64Value.of(getCId(GRAPH_ID_PREFIX, maxGraphID - 1));
if (id.getValue() == -1) {
throw new HgStoreException(HgStoreException.EC_FAIL,
"The number of graphs exceeds the maximum " +
"65535");
}
+ log.info("partition: {}, Graph ID {} is allocated for graph {}, stack: {}",
+ this.partitionId, id.getValue(), graphName,
+ Arrays.toString(Thread.currentThread().getStackTrace()));
put(key, id);
flush();
}
@@ -112,7 +119,20 @@ public long releaseGraphId(String graphName) {
}
/**
- * Get auto-increment non-repetitive id, start from 0 after reaching the limit.
+ * 为了兼容受影响的 graph，保证 g+v 表中没有数据
+ *
+ * @return 有数据返回 false，没有返回 true
+ */
+ private boolean checkCount(long l) {
+ var start = new byte[2];
+ Bits.putShort(start, 0, (short) l);
+ try (var itr = sessionBuilder.getSession(partitionId).sessionOp().scan("g+v", start)) {
+ return itr == null || !itr.hasNext();
+ }
+ }
+
+ /**
+ * 获取自增循环不重复 id, 达到上限后从 0 开始自增
*
* @param key key
* @param max max id limit, after reaching this value, it will reset to 0 and start incrementing again.
@@ -127,24 +147,19 @@ protected long getCId(String key, long max) {
// Find an unused cid
List ids =
scan(Int64Value.parser(), genCIDSlotKey(key, current), genCIDSlotKey(key, max));
- for (Int64Value id : ids) {
- if (current == id.getValue()) {
+ var idSet = ids.stream().map(Int64Value::getValue).collect(Collectors.toSet());
+
+ while (idSet.contains(current) || !checkCount(current)) {
current++;
- } else {
- break;
- }
}
- if (current == max) {
+ if (current == max - 1) {
current = 0;
ids = scan(Int64Value.parser(), genCIDSlotKey(key, current),
genCIDSlotKey(key, last));
- for (Int64Value id : ids) {
- if (current == id.getValue()) {
+ idSet = ids.stream().map(Int64Value::getValue).collect(Collectors.toSet());
+ while (idSet.contains(current) || !checkCount(current)) {
current++;
- } else {
- break;
- }
}
}
@@ -162,7 +177,7 @@ protected long getCId(String key, long max) {
/**
* Return key with used Cid
*/
- private byte[] genCIDSlotKey(String key, long value) {
+ public byte[] genCIDSlotKey(String key, long value) {
byte[] keySlot = MetadataKeyHelper.getCidSlotKeyPrefix(key);
ByteBuffer buf = ByteBuffer.allocate(keySlot.length + Long.SIZE);
buf.put(keySlot);
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java
index ffd1349a91..ac98d39282 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/PartitionManager.java
@@ -26,6 +26,7 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
@@ -37,8 +38,10 @@
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.store.HgStoreEngine;
import org.apache.hugegraph.store.business.BusinessHandlerImpl;
-import org.apache.hugegraph.store.cmd.UpdatePartitionRequest;
-import org.apache.hugegraph.store.cmd.UpdatePartitionResponse;
+import org.apache.hugegraph.store.cmd.HgCmdClient;
+import org.apache.hugegraph.store.cmd.request.UpdatePartitionRequest;
+import org.apache.hugegraph.store.cmd.response.UpdatePartitionResponse;
+import org.apache.hugegraph.store.listener.PartitionChangedListener;
import org.apache.hugegraph.store.meta.base.GlobalMetaStore;
import org.apache.hugegraph.store.options.HgStoreEngineOptions;
import org.apache.hugegraph.store.options.MetadataOptions;
@@ -72,6 +75,7 @@ public class PartitionManager extends GlobalMetaStore {
// Record all partition information of this machine, consistent with rocksdb storage.
private Map> partitions;
+ private HgCmdClient cmdClient;
public PartitionManager(PdProvider pdProvider, HgStoreEngineOptions options) {
super(new MetadataOptions() {{
@@ -225,7 +229,7 @@ private void loadPartitions() {
var partIds = new HashSet();
for (String path : this.options.getDataPath().split(",")) {
File[] dirs = new File(path + "/" + HgStoreEngineOptions.DB_Path_Prefix).listFiles();
- if (dirs == null) {
+ if (dirs == null || dirs.length == 0) {
continue;
}
@@ -241,6 +245,8 @@ private void loadPartitions() {
}
}
+ Set normalPartitions = new HashSet<>();
+
// Once according to the partition read
for (int partId : partIds) {
if (!resetPartitionPath(partId)) {
@@ -249,18 +255,23 @@ private void loadPartitions() {
continue;
}
- for (var metaPart : wrapper.scan(partId, Metapb.Partition.parser(), key)) {
+ var metaParts = wrapper.scan(partId, Metapb.Partition.parser(), key);
+ int countOfPartition = 0;
+
+ var shards = pdProvider.getShardGroup(partId).getShardsList();
+
+ for (var metaPart : metaParts) {
var graph = metaPart.getGraphName();
var pdPartition = pdProvider.getPartitionByID(graph, metaPart.getId());
boolean isLegeal = false;
- var shards = pdProvider.getShardGroup(metaPart.getId()).getShardsList();
-
if (pdPartition != null) {
// Check if it contains this store id
if (shards.stream().anyMatch(s -> s.getStoreId() == storeId)) {
isLegeal = true;
}
+ } else {
+ continue;
}
if (isLegeal) {
@@ -268,6 +279,8 @@ private void loadPartitions() {
partitions.put(graph, new ConcurrentHashMap<>());
}
+ countOfPartition += 1;
+
Partition partition = new Partition(metaPart);
partition.setWorkState(Metapb.PartitionState.PState_Normal); // Start recovery work state
partitions.get(graph).put(partition.getId(), partition);
@@ -284,6 +297,19 @@ private void loadPartitions() {
System.exit(0);
}
}
+
+ if (countOfPartition > 0) {
+ // 分区数据正常
+ normalPartitions.add(partId);
+ }
+ wrapper.close(partId);
+ }
+
+ // 删掉多余的分区存储路径，被迁移走的分区，有可能还会迁回来
+ for (var location : storeMetadata.getPartitionStores()) {
+ if (!normalPartitions.contains(location.getPartitionId())) {
+ storeMetadata.removePartitionStore(location.getPartitionId());
+ }
}
}
@@ -611,7 +637,7 @@ public ShardGroup getShardGroup(int partitionId) {
Metapb.ShardGroup.parser());
if (shardGroup == null) {
- shardGroup = pdProvider.getShardGroup(partitionId);
+ shardGroup = pdProvider.getShardGroupDirect(partitionId);
if (shardGroup != null) {
// local not found, write back to db from pd
@@ -726,6 +752,18 @@ public List getLeaderPartitionIds(String graph) {
return ids;
}
+ public Set getLeaderPartitionIdSet() {
+ Set ids = new HashSet<>();
+ partitions.forEach((key, value) -> {
+ value.forEach((k, v) -> {
+ if (!useRaft || v.isLeader()) {
+ ids.add(k);
+ }
+ });
+ });
+ return ids;
+ }
+
/**
* Generate partition peer string, containing priority information *
*
@@ -833,15 +871,15 @@ public Store getStoreByRaftEndpoint(ShardGroup group, String endpoint) {
return result[0];
}
- public Shard getShardByRaftEndpoint(ShardGroup group, String endpoint) {
- final Shard[] result = {new Shard()};
- group.getShards().forEach((shard) -> {
+ public Shard getShardByEndpoint(ShardGroup group, String endpoint) {
+ List shards = group.getShards();
+ for (Shard shard : shards) {
Store store = getStore(shard.getStoreId());
if (store != null && store.getRaftAddress().equalsIgnoreCase(endpoint)) {
- result[0] = shard;
+ return shard;
}
- });
- return result[0];
+ }
+ return new Shard();
}
/**
@@ -885,6 +923,16 @@ public String getDbDataPath(int partitionId, String dbName) {
return location;
}
+ /**
+ * db 存储路径
+ *
+ * @return location/db
+ */
+ public String getDbDataPath(int partitionId) {
+ String dbName = BusinessHandlerImpl.getDbName(partitionId);
+ return getDbDataPath(partitionId, dbName);
+ }
+
public void reportTask(MetaTask.Task task) {
try {
pdProvider.reportTask(task);
@@ -908,14 +956,39 @@ public PartitionMetaStoreWrapper getWrapper() {
return wrapper;
}
- /**
- * Partition object is modified message
- */
- public interface PartitionChangedListener {
+ public void setCmdClient(HgCmdClient client) {
+ this.cmdClient = client;
+ }
- void onChanged(Partition partition);
+ public UpdatePartitionResponse updateState(Metapb.Partition partition,
+ Metapb.PartitionState state) {
+ // 分区分裂时，主动需要查找 leader 进行同步信息
+ UpdatePartitionRequest request = new UpdatePartitionRequest();
+ request.setWorkState(state);
+ request.setPartitionId(partition.getId());
+ request.setGraphName(partition.getGraphName());
+ return cmdClient.raftUpdatePartition(request);
+ }
- UpdatePartitionResponse rangeOrStateChanged(UpdatePartitionRequest request);
+ public UpdatePartitionResponse updateRange(Metapb.Partition partition, int startKey,
+ int endKey) {
+ // 分区分裂时，主动需要查找 leader 进行同步信息
+ UpdatePartitionRequest request = new UpdatePartitionRequest();
+ request.setStartKey(startKey);
+ request.setEndKey(endKey);
+ request.setPartitionId(partition.getId());
+ request.setGraphName(partition.getGraphName());
+ return cmdClient.raftUpdatePartition(request);
+ }
+
+ public List getPartitionIds(String graph) {
+ List ids = new ArrayList<>();
+ if (partitions.containsKey(graph)) {
+ partitions.get(graph).forEach((k, v) -> {
+ ids.add(k);
+ });
+ }
+ return ids;
}
}
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/ShardGroup.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/ShardGroup.java
index 892af940b3..4b3a5a618f 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/ShardGroup.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/ShardGroup.java
@@ -52,11 +52,17 @@ public static ShardGroup from(Metapb.ShardGroup meta) {
shardGroup.setId(meta.getId());
shardGroup.setVersion(meta.getVersion());
shardGroup.setConfVersion(meta.getConfVer());
- shardGroup.setShards(meta.getShardsList().stream().map(Shard::fromMetaPbShard)
- .collect(Collectors.toList()));
+ shardGroup.setShards(new CopyOnWriteArrayList<>(
+ meta.getShardsList().stream().map(Shard::fromMetaPbShard)
+ .collect(Collectors.toList())));
return shardGroup;
}
+ public ShardGroup addShard(Shard shard) {
+ this.shards.add(shard);
+ return this;
+ }
+
public synchronized ShardGroup changeLeader(long storeId) {
shards.forEach(shard -> {
shard.setRole(shard.getStoreId() == storeId ? Metapb.ShardRole.Leader :
diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/StoreMetadata.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/StoreMetadata.java
index 662b6521f1..6fb54710cc 100644
--- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/StoreMetadata.java
+++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/StoreMetadata.java
@@ -24,6 +24,7 @@
import java.util.List;
import java.util.Map;
+import org.apache.commons.io.FileUtils;
import org.apache.hugegraph.pd.grpc.Metapb;
import org.apache.hugegraph.store.meta.base.GlobalMetaStore;
import org.apache.hugegraph.store.options.MetadataOptions;
@@ -115,6 +116,16 @@ public Metapb.PartitionStore getPartitionStore(int partitionId) {
return get(Metapb.PartitionStore.parser(), key);
}
+ /**
+ * 删除指定分区对应的存储元数据。
+ *
+ * @param partitionId 分区ID。
+ */
+ public void removePartitionStore(int partitionId) {
+ byte[] key = MetadataKeyHelper.getPartitionStoreKey(partitionId);
+ delete(key);
+ }
+
public List getPartitionStores() {
byte[] key = MetadataKeyHelper.getPartitionStorePrefix();
return scan(Metapb.PartitionStore.parser(), key);
@@ -141,16 +152,14 @@ public void savePartitionRaft(Metapb.PartitionRaft partitionRaft) {
}
private String getMinDataLocation() {
- Map counter = new HashMap<>();
- dataLocations.forEach(l -> {
- counter.put(l, Integer.valueOf(0));
- });
- getPartitionStores().forEach(ptStore -> {
- if (counter.containsKey(ptStore.getStoreLocation())) {
- counter.put(ptStore.getStoreLocation(),
- counter.get(ptStore.getStoreLocation()) + 1);
+ var counter = stateLocByFreeSpace();
+ if (counter.isEmpty()) {
+ counter = stateLocByPartitionCount();
+ log.info("allocate db path using partition count: db count stats: {}", counter);
+ } else {
+ log.info("allocate db path using free space: db size stats: {}", counter);
}
- });
+
int min = Integer.MAX_VALUE;
String location = "";
for (String k : counter.keySet()) {
@@ -162,6 +171,91 @@ private String getMinDataLocation() {
return location;
}
+ /**
+ * get location count by allocated db count
+ *
+ * @return loc -> db count
+ */
+ private Map stateLocByPartitionCount() {
+ Map