You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by vg...@apache.org on 2024/02/04 03:17:12 UTC
(incubator-hugegraph) branch pd-store updated: chore(pd-store): minor improvement (#2443)
This is an automated email from the ASF dual-hosted git repository.
vgalaxies pushed a commit to branch pd-store
in repository https://gitbox.apache.org/repos/asf/incubator-hugegraph.git
The following commit(s) were added to refs/heads/pd-store by this push:
new bc9c1e7ac chore(pd-store): minor improvement (#2443)
bc9c1e7ac is described below
commit bc9c1e7ac962433af94121e479bf27c3a3d59ab1
Author: V_Galaxy <vg...@apache.org>
AuthorDate: Sun Feb 4 11:17:06 2024 +0800
chore(pd-store): minor improvement (#2443)
---
.../apache/hugegraph/pd/client/PDPulseTest.java | 5 +-
hugegraph-pd/pom.xml | 2 +-
.../org/apache/hugegraph/StandardHugeGraph.java | 14 +-
.../org/apache/hugegraph/config/CoreOptions.java | 7 +
.../static/conf/graphs/hugegraph.properties | 11 +
.../store/hstore/HstoreNodePartitionerImpl.java | 279 ---------------------
.../backend/store/hstore/HstoreOptions.java | 18 --
.../backend/store/hstore/HstoreSessionsImpl.java | 11 +-
.../hugegraph/store/client/grpc/KvPageScanner.java | 6 +-
hugegraph-store/pom.xml | 2 +-
10 files changed, 39 insertions(+), 316 deletions(-)
diff --git a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/client/PDPulseTest.java b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/client/PDPulseTest.java
index dfdc63cf3..220ec80f1 100644
--- a/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/client/PDPulseTest.java
+++ b/hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/client/PDPulseTest.java
@@ -24,6 +24,7 @@ import org.apache.hugegraph.pd.client.test.HgPDTestUtil;
import org.apache.hugegraph.pd.grpc.pulse.PartitionHeartbeatRequest;
import org.apache.hugegraph.pd.pulse.PulseServerNotice;
import org.junit.BeforeClass;
+import org.junit.Test;
public class PDPulseTest {
private static PDClient pdClient;
@@ -40,7 +41,7 @@ public class PDPulseTest {
pdClient.getLeader();
}
- // @Test
+ @Test
public void listen() {
PDPulse pulse = new PDPulseImpl(pdClient.getLeaderIp());
@@ -106,4 +107,4 @@ public class PDPulseTest {
HgPDTestUtil.println(this.listenerName + " is completed");
}
}
-}
\ No newline at end of file
+}
diff --git a/hugegraph-pd/pom.xml b/hugegraph-pd/pom.xml
index 743ead0d1..26181e5a4 100644
--- a/hugegraph-pd/pom.xml
+++ b/hugegraph-pd/pom.xml
@@ -135,7 +135,7 @@
<include>*.tar.gz</include>
<include>.flattened-pom.xml</include>
<!-- WARN: delete dist dir may influence dev? -->
- <!-- <include>dist/**</include>-->
+ <include>dist/**</include>
</includes>
<followSymlinks>false</followSymlinks>
</fileset>
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
index c496483ac..4551b00af 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
@@ -18,7 +18,6 @@
package org.apache.hugegraph;
import java.util.Collection;
-import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
@@ -179,8 +178,6 @@ public class StandardHugeGraph implements HugeGraph {
private final RamTable ramtable;
- private final MetaManager metaManager = MetaManager.instance();
-
private final String schedulerType;
public StandardHugeGraph(HugeConfig config) {
@@ -230,7 +227,10 @@ public class StandardHugeGraph implements HugeGraph {
}
if (isHstore()) {
- initMetaManager();
+ // TODO: parameterize the remaining configurations
+ MetaManager.instance().connect("hg", MetaManager.MetaDriverType.PD,
+ "ca", "ca", "ca",
+ config.get(CoreOptions.PD_PEERS));
}
try {
@@ -470,12 +470,6 @@ public class StandardHugeGraph implements HugeGraph {
return this.storeProvider.isHstore();
}
- private void initMetaManager() {
- this.metaManager.connect("hg", MetaManager.MetaDriverType.PD,
- "ca", "ca", "ca",
- Collections.singletonList("127.0.0.1:8686"));
- }
-
private ISchemaTransaction openSchemaTransaction() throws HugeException {
this.checkGraphNotClosed();
try {
diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
index 9f17b0cae..3697382f5 100644
--- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
+++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
@@ -674,4 +674,11 @@ public class CoreOptions extends OptionHolder {
CollectionType::valueOf,
"EC"
);
+
+ public static final ConfigOption<String> PD_PEERS = new ConfigOption<>(
+ "pd.peers",
+ "The addresses of pd nodes, separated with commas.",
+ disallowEmpty(),
+ "127.0.0.1:8686"
+ );
}
diff --git a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties
index 74e1408c7..1a3532914 100644
--- a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties
+++ b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/graphs/hugegraph.properties
@@ -24,6 +24,16 @@ serializer=binary
store=hugegraph
+# pd config
+pd.peers=127.0.0.1:8686
+
+# task config
+task.scheduler_type=local
+task.schedule_period=10
+task.retry=0
+task.wait_timeout=10
+
+# raft config
raft.mode=false
raft.path=./raft-log
raft.safe_read=true
@@ -45,6 +55,7 @@ raft.rpc_connect_timeout=5000
raft.rpc_timeout=60
raft.install_snapshot_rpc_timeout=36000
+# search config
search.text_analyzer=jieba
search.text_analyzer_mode=INDEX
diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreNodePartitionerImpl.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreNodePartitionerImpl.java
deleted file mode 100644
index 2a69fe1c0..000000000
--- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreNodePartitionerImpl.java
+++ /dev/null
@@ -1,279 +0,0 @@
-/*
- * 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.backend.store.hstore;
-
-import static org.apache.hugegraph.store.client.util.HgStoreClientConst.ALL_PARTITION_OWNER;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.hugegraph.config.HugeConfig;
-import org.apache.hugegraph.pd.client.PDClient;
-import org.apache.hugegraph.pd.common.KVPair;
-import org.apache.hugegraph.pd.common.PDException;
-import org.apache.hugegraph.pd.common.PartitionUtils;
-import org.apache.hugegraph.pd.grpc.Metapb;
-import org.apache.hugegraph.store.client.HgNodePartition;
-import org.apache.hugegraph.store.client.HgNodePartitionerBuilder;
-import org.apache.hugegraph.store.client.HgStoreNode;
-import org.apache.hugegraph.store.client.HgStoreNodeManager;
-import org.apache.hugegraph.store.client.HgStoreNodeNotifier;
-import org.apache.hugegraph.store.client.HgStoreNodePartitioner;
-import org.apache.hugegraph.store.client.HgStoreNodeProvider;
-import org.apache.hugegraph.store.client.HgStoreNotice;
-import org.apache.hugegraph.store.client.type.HgNodeStatus;
-import org.apache.hugegraph.store.client.util.HgStoreClientConst;
-import org.apache.hugegraph.util.Log;
-import org.slf4j.Logger;
-
-public class HstoreNodePartitionerImpl implements HgStoreNodePartitioner,
- HgStoreNodeProvider,
- HgStoreNodeNotifier {
-
- private static final Logger LOG = Log.logger(HstoreNodePartitionerImpl.class);
- private PDClient pdClient;
- private HgStoreNodeManager nodeManager;
-
- protected HstoreNodePartitionerImpl() {
-
- }
-
- public HstoreNodePartitionerImpl(String pdPeers) {
- pdClient = HstoreSessionsImpl.getDefaultPdClient();
- }
-
- public HstoreNodePartitionerImpl(HgStoreNodeManager nodeManager,
- String pdPeers) {
- this(pdPeers);
- this.nodeManager = nodeManager;
- }
-
- public void setPDClient(PDClient pdClient) {
- this.pdClient = pdClient;
- }
-
- /**
- * 查询分区信息，结果通过HgNodePartitionerBuilder返回
- */
- @Override
- public int partition(HgNodePartitionerBuilder builder, String graphName,
- byte[] startKey, byte[] endKey) {
- try {
- HashSet<HgNodePartition> partitions = null;
- if (HgStoreClientConst.ALL_PARTITION_OWNER == startKey) {
- List<Metapb.Store> stores = pdClient.getActiveStores(graphName);
- partitions = new HashSet<>(stores.size());
- for (Metapb.Store store : stores) {
- partitions.add(HgNodePartition.of(store.getId(), -1));
- }
-
- } else if (endKey == HgStoreClientConst.EMPTY_BYTES
- || startKey == endKey || Arrays.equals(startKey, endKey)) {
- KVPair<Metapb.Partition, Metapb.Shard> partShard =
- pdClient.getPartition(graphName, startKey);
- Metapb.Shard leader = partShard.getValue();
- partitions = new HashSet<>(1);
- partitions.add(HgNodePartition.of(leader.getStoreId(),
- pdClient.keyToCode(graphName, startKey)));
- } else {
- LOG.warn(
- "StartOwnerkey is not equal to endOwnerkey, which is meaningless!!, It is" +
- " a error!!");
- List<Metapb.Store> stores = pdClient.getActiveStores(graphName);
- for (Metapb.Store store : stores) {
- partitions.add(HgNodePartition.of(store.getId(), -1));
- }
- }
- builder.setPartitions(partitions);
- } catch (PDException e) {
- LOG.error("An error occurred while getting partition information :{}", e.getMessage());
- throw new RuntimeException(e.getMessage(), e);
- }
- return 0;
- }
-
- @Override
- public int partition(HgNodePartitionerBuilder builder, String graphName,
- int startKey, int endKey) {
- try {
- HashSet<HgNodePartition> partitions = new HashSet<>();
- Metapb.Partition partition = null;
- while ((partition == null || partition.getEndKey() < endKey)
- && startKey < PartitionUtils.MAX_VALUE) {
- KVPair<Metapb.Partition, Metapb.Shard> partShard =
- pdClient.getPartitionByCode(graphName, startKey);
- if (partShard != null) {
- partition = partShard.getKey();
- Metapb.Shard leader = partShard.getValue();
- partitions.add(HgNodePartition.of(leader.getStoreId(), startKey,
- (int) partition.getStartKey(),
- (int) partition.getEndKey()));
- startKey = (int) partition.getEndKey();
- } else {
- break;
- }
- }
- builder.setPartitions(partitions);
- } catch (PDException e) {
- LOG.error("An error occurred while getting partition information :{}", e.getMessage());
- throw new RuntimeException(e.getMessage(), e);
- }
- return 0;
-
- }
-
- /**
- * 查询hgstore信息
- *
- * @return hgstore
- */
- @Override
- public HgStoreNode apply(String graphName, Long nodeId) {
- try {
- Metapb.Store store = pdClient.getStore(nodeId);
- return nodeManager.getNodeBuilder().setNodeId(store.getId())
- .setAddress(store.getAddress()).build();
- } catch (PDException e) {
- throw new RuntimeException(e.getMessage(), e);
- }
- }
-
- /**
- * 通知更新缓存
- */
- @Override
- public int notice(String graphName, HgStoreNotice storeNotice) {
- LOG.warn(storeNotice.toString());
- if (storeNotice.getPartitionLeaders() != null) {
- storeNotice.getPartitionLeaders().forEach((partId, leader) -> {
- pdClient.updatePartitionLeader(graphName, partId, leader);
- LOG.warn("updatePartitionLeader:{}-{}-{}",
- graphName, partId, leader);
- });
- }
- if (storeNotice.getPartitionIds() != null) {
- storeNotice.getPartitionIds().forEach(partId -> {
- pdClient.invalidPartitionCache(graphName, partId);
- });
- }
- if (!storeNotice.getNodeStatus().equals(
- HgNodeStatus.PARTITION_COMMON_FAULT)
- && !storeNotice.getNodeStatus().equals(
- HgNodeStatus.NOT_PARTITION_LEADER)) {
- pdClient.invalidPartitionCache();
- LOG.warn("invalidPartitionCache:{} ", storeNotice.getNodeStatus());
- }
- return 0;
- }
-
- public Metapb.Graph delGraph(String graphName) {
- try {
- return pdClient.delGraph(graphName);
- } catch (PDException e) {
- LOG.error("delGraph {} exception, {}", graphName, e.getMessage());
- }
- return null;
- }
-
- public void setNodeManager(HgStoreNodeManager nodeManager) {
- this.nodeManager = nodeManager;
- }
-}
-
-class FakeHstoreNodePartitionerImpl extends HstoreNodePartitionerImpl {
- private static final Logger LOG = Log.logger(HstoreNodePartitionerImpl.class);
- private static final int partitionCount = 3;
- private static final Map<Integer, Long> leaderMap = new ConcurrentHashMap<>();
- private static final Map<Long, String> storeMap = new ConcurrentHashMap<>();
- HgStoreNodeManager nodeManager;
- private final String hstorePeers;
-
- public FakeHstoreNodePartitionerImpl(String pdPeers) {
- this.hstorePeers = pdPeers;
- // store列表
- for (String address : hstorePeers.split(",")) {
- storeMap.put((long) address.hashCode(), address);
- }
- // 分区列表
- for (int i = 0; i < partitionCount; i++) {
- leaderMap.put(i, storeMap.keySet().iterator().next());
- }
- }
-
- public FakeHstoreNodePartitionerImpl(HgStoreNodeManager nodeManager,
- String peers) {
- this(peers);
- this.nodeManager = nodeManager;
- }
-
- @Override
- public int partition(HgNodePartitionerBuilder builder, String graphName,
- byte[] startKey, byte[] endKey) {
- int startCode = PartitionUtils.calcHashcode(startKey);
- HashSet<HgNodePartition> partitions = new HashSet<>(storeMap.size());
- if (ALL_PARTITION_OWNER == startKey) {
- storeMap.forEach((k, v) -> {
- partitions.add(HgNodePartition.of(k, -1));
- });
- } else if (endKey == HgStoreClientConst.EMPTY_BYTES || startKey == endKey ||
- Arrays.equals(startKey, endKey)) {
- partitions.add(
- HgNodePartition.of(leaderMap.get(startCode % partitionCount), startCode));
- } else {
- LOG.error("OwnerKey转成HashCode后已经无序了， 按照OwnerKey范围查询没意义");
- storeMap.forEach((k, v) -> {
- partitions.add(HgNodePartition.of(k, -1));
- });
- }
- builder.setPartitions(partitions);
- return 0;
- }
-
- @Override
- public HgStoreNode apply(String graphName, Long nodeId) {
- return nodeManager.getNodeBuilder().setNodeId(nodeId)
- .setAddress(storeMap.get(nodeId)).build();
- }
-
- @Override
- public int notice(String graphName, HgStoreNotice storeNotice) {
- if (storeNotice.getPartitionLeaders() != null
- && storeNotice.getPartitionLeaders().size() > 0) {
- leaderMap.putAll(storeNotice.getPartitionLeaders());
- }
- return 0;
- }
-
- public static class NodePartitionerFactory {
- public static HstoreNodePartitionerImpl getNodePartitioner(
- HugeConfig config, HgStoreNodeManager nodeManager) {
- if (config.get(HstoreOptions.PD_FAKE)) {
- return new FakeHstoreNodePartitionerImpl(nodeManager,
- config.get(HstoreOptions.HSTORE_PEERS));
- } else {
- return new HstoreNodePartitionerImpl(nodeManager,
- config.get(HstoreOptions.PD_PEERS)
- );
- }
-
- }
- }
-}
diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreOptions.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreOptions.java
index bafde45f8..6de800697 100644
--- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreOptions.java
+++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreOptions.java
@@ -24,24 +24,6 @@ import org.apache.hugegraph.config.OptionHolder;
public class HstoreOptions extends OptionHolder {
- public static final ConfigOption<String> PD_PEERS = new ConfigOption<>(
- "pd.peers",
- "The addresses of pd nodes, separated with commas.",
- disallowEmpty(),
- "localhost:8686"
- );
- public static final ConfigOption<Boolean> PD_FAKE = new ConfigOption<>(
- "pd.fake",
- "Enable the fake PD service.",
- disallowEmpty(),
- false
- );
- public static final ConfigOption<String> HSTORE_PEERS = new ConfigOption<>(
- "hstore.peers",
- "The addresses of store nodes, separated with commas.",
- disallowEmpty(),
- "localhost:9080"
- );
public static final ConfigOption<Integer> PARTITION_COUNT = new ConfigOption<>(
"hstore.partition_count",
"Number of partitions, which PD controls partitions based on.",
diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreSessionsImpl.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreSessionsImpl.java
index be90902da..27de0e029 100755
--- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreSessionsImpl.java
+++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreSessionsImpl.java
@@ -37,6 +37,7 @@ import org.apache.hugegraph.backend.store.BackendEntry;
import org.apache.hugegraph.backend.store.BackendEntry.BackendColumn;
import org.apache.hugegraph.backend.store.BackendEntry.BackendColumnIterator;
import org.apache.hugegraph.backend.store.BackendEntryIterator;
+import org.apache.hugegraph.config.CoreOptions;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.pd.client.PDClient;
import org.apache.hugegraph.pd.client.PDConfig;
@@ -103,7 +104,7 @@ public class HstoreSessionsImpl extends HstoreSessions {
synchronized (this) {
if (!initializedNode) {
PDConfig pdConfig =
- PDConfig.of(config.get(HstoreOptions.PD_PEERS))
+ PDConfig.of(config.get(CoreOptions.PD_PEERS))
.setEnableCache(true);
defaultPdClient = PDClient.create(pdConfig);
hgStoreClient =
@@ -244,7 +245,9 @@ public class HstoreSessionsImpl extends HstoreSessions {
this.position = iter.position();
} else {
this.gotNext = false;
- this.position = null;
+ // QUESTION: Resetting the position may result in the caller being unable to
+ // retrieve the corresponding position.
+ // this.position = null;
}
if (!ArrayUtils.isEmpty(this.keyBegin) ||
!ArrayUtils.isEmpty(this.keyEnd)) {
@@ -315,7 +318,9 @@ public class HstoreSessionsImpl extends HstoreSessions {
if (gotNext) {
this.position = this.iter.position();
} else {
- this.position = null;
+ // QUESTION: Resetting the position may result in the caller being unable to
+ // retrieve the corresponding position.
+ // this.position = null;
}
return gotNext;
}
diff --git a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/KvPageScanner.java b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/KvPageScanner.java
index c9e99d9b7..e6ed4a729 100644
--- a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/KvPageScanner.java
+++ b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/grpc/KvPageScanner.java
@@ -194,8 +194,10 @@ class KvPageScanner implements KvCloseableIterator<Kv>, HgPageSize, HgSeekAble {
if (!this.in) {
return false;
}
- if (this.iterator != null && this.iterator.hasNext()) {
- return true;
+ // QUESTION: After `this.iterator.hasNext()` evaluates to false,
+ // no further attempts are made to reconstruct the iterator.
+ if (this.iterator != null) {
+ return this.iterator.hasNext();
}
long start = 0;
boolean debugEnabled = log.isDebugEnabled();
diff --git a/hugegraph-store/pom.xml b/hugegraph-store/pom.xml
index bc6154ad1..36f8b6982 100644
--- a/hugegraph-store/pom.xml
+++ b/hugegraph-store/pom.xml
@@ -153,7 +153,7 @@
<include>*.tar.gz</include>
<include>.flattened-pom.xml</include>
<!-- WARN: delete dist dir may influence dev? -->
- <!-- <include>dist/**</include>-->
+ <include>dist/**</include>
</includes>
<followSymlinks>false</followSymlinks>
</fileset>