You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hugegraph.apache.org by "VGalaxies (via GitHub)" <gi...@apache.org> on 2023/08/29 06:07:17 UTC
[GitHub] [incubator-hugegraph] VGalaxies opened a new pull request, #2301: feat: preliminary support for hstore backend
VGalaxies opened a new pull request, #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301
<!--
Thank you very much for contributing to Apache HugeGraph, we are happy that you want to help us improve it!
Here are some tips for you:
1. If this is your first time, please read the [contributing guidelines](https://github.com/apache/hugegraph/blob/master/CONTRIBUTING.md)
2. If a PR fix/close an issue, type the message "close xxx" (xxx is the link of related
issue) in the content, GitHub will auto link it (Required)
3. Name the PR title in "Google Commit Format", start with "feat | fix | perf | refactor | doc | chore",
such like: "feat(core): support the PageRank algorithm" or "fix: wrong break in the compute loop" (module is optional)
skip it if you are unsure about which is the best component.
4. One PR address one issue, better not to mix up multiple issues.
5. Put an `x` in the `[ ]` to mark the item as CHECKED. `[x]` (or click it directly after
published)
-->
## Purpose of the PR
subtask of https://github.com/apache/incubator-hugegraph/issues/2265
Introduce `hugegraph-hstore` module to connect with `hugegraph-pd` and `hugegraph-store`, and make adjustments to other modules as well.
This PR is quite big, and some commits related to maven configuration fixes have already been cherry-picked into https://github.com/apache/incubator-hugegraph/pull/2289, awaiting merging. It might be advisable to further split this PR to facilitate the review. (How?)
<!--
Please explain more context in this section, clarify why the changes are needed.
e.g:
- If you propose a new API, clarify the use case for a new API.
- If you fix a bug, you can clarify why it is a bug, and should be associated with an issue.
-->
## Main Changes
- maven configuration fixes (see https://github.com/apache/incubator-hugegraph/pull/2289)
- migrate `hugegraph-hstore` module to `hugegraph-server` (also migrate `org.apache.hugegraph.meta` and `migrate org.apache.hugegraph.space` for compilation purpose)
- introduce `ISchemaTransaction` for adapting special version of schema tx for hstore backend
- query task and server info by graph tx for hstore backend
For detailed adaptation documentation, refer to https://hugegraph.feishu.cn/wiki/WdqXwyHkvi0UCvkBUDWcZgQHnhe
<!-- Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue. These change logs are helpful for better ant faster reviews.)
For example:
- If you introduce a new feature, please show detailed design here or add the link of design documentation.
- If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
- If there is a discussion in the mailing list, please add the link. -->
## Verifying these changes
<!-- Please pick the proper options below -->
- [ ] Trivial rework / code cleanup without any test coverage. (No Need)
- [ ] Already covered by existing tests, such as *(please modify tests here)*.
- [x] Need tests and can be verified as follows:
- validate that the functionality of existing backends remains unaffected
- verify the basic functionality of the hstore backend (basic graph import has been already verified)
## Does this PR potentially affect the following parts?
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] Nope
- [ ] Dependencies (add/update license info) <!-- Don't forget to add/update the info in "LICENSE" & "NOTICE" files (both in root & dist module) -->
- [ ] Modify configurations
- [ ] The public API
- [x] Other affects (maybe breaking changes)
## Documentation Status
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [x] `Doc - TODO` <!-- Your PR changes impact docs and you will update later -->
- [ ] `Doc - Done` <!-- Related docs have been already added or updated -->
- [ ] `Doc - No Need` <!-- Your PR changes don't impact/need docs -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1325433577
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java:
##########
@@ -0,0 +1,469 @@
+package org.apache.hugegraph.backend.cache;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.store.ram.IntObjectMap;
+import org.apache.hugegraph.backend.tx.SchemaTransactionV2;
+import org.apache.hugegraph.config.CoreOptions;
+import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventListener;
+import org.apache.hugegraph.meta.MetaDriver;
+import org.apache.hugegraph.meta.MetaManager;
+import org.apache.hugegraph.perf.PerfUtil;
+import org.apache.hugegraph.schema.SchemaElement;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Events;
+
+import com.google.common.collect.ImmutableSet;
+
+public class CachedSchemaTransactionV2 extends SchemaTransactionV2 {
+ private final Cache<Id, Object> idCache;
+ private final Cache<Id, Object> nameCache;
+
+ private final SchemaCaches<SchemaElement> arrayCaches;
+
+ private EventListener storeEventListener;
+ private EventListener cacheEventListener;
+
+ public CachedSchemaTransactionV2(MetaDriver metaDriver,
+ String cluster,
+ HugeGraphParams graphParams) {
+ super(metaDriver, cluster, graphParams);
+
+ final long capacity = graphParams.configuration()
+ .get(CoreOptions.SCHEMA_CACHE_CAPACITY);
+ this.idCache = this.cache("schema-id", capacity);
+ this.nameCache = this.cache("schema-name", capacity);
+
+ SchemaCaches<SchemaElement> attachment = this.idCache.attachment();
+ if (attachment == null) {
+ int acSize = (int) (capacity >> 3);
+ attachment = this.idCache.attachment(new SchemaCaches<>(acSize));
+ }
+ this.arrayCaches = attachment;
+ this.listenChanges();
+ }
+
+ private static Id generateId(HugeType type, Id id) {
+ // NOTE: it's slower performance to use:
+ // String.format("%x-%s", type.code(), name)
+ return IdGenerator.of(type.string() + "-" + id.asString());
+ }
+
+ private static Id generateId(HugeType type, String name) {
+ return IdGenerator.of(type.string() + "-" + name);
+ }
+
+ public void close() {
+ this.clearCache(false);
+ this.unlistenChanges();
+ }
+
+ private Cache<Id, Object> cache(String prefix, long capacity) {
+ // TODO: uncomment later - graph space
+ //final String name = prefix + "-" + this.graph().spaceGraphName();
+ final String name = prefix + "-" + "";
+ // NOTE: must disable schema cache-expire due to getAllSchema()
+ return CacheManager.instance().cache(name, capacity);
+ }
+
+ private void listenChanges() {
+ // Listen store event: "store.init", "store.clear", ...
+ Set<String> storeEvents = ImmutableSet.of(Events.STORE_INIT,
+ Events.STORE_CLEAR,
+ Events.STORE_TRUNCATE);
+ this.storeEventListener = event -> {
+ if (storeEvents.contains(event.name())) {
+ LOG.debug("Graph {} clear schema cache on event '{}'",
+ this.graph(), event.name());
+ this.clearCache(true);
+ return true;
+ }
+ return false;
+ };
+ this.graphParams().loadGraphStore().provider().listen(this.storeEventListener);
+
+ // Listen cache event: "cache"(invalid cache item)
+ this.cacheEventListener = event -> {
+ LOG.debug("Graph {} received schema cache event: {}",
+ this.graph(), event);
+ Object[] args = event.args();
+ E.checkArgument(args.length > 0 && args[0] instanceof String,
+ "Expect event action argument");
+ if (Cache.ACTION_INVALID.equals(args[0])) {
+ event.checkArgs(String.class, HugeType.class, Id.class);
+ HugeType type = (HugeType) args[1];
+ Id id = (Id) args[2];
+ this.arrayCaches.remove(type, id);
+
+ id = generateId(type, id);
+ Object value = this.idCache.get(id);
+ if (value != null) {
+ // Invalidate id cache
+ this.idCache.invalidate(id);
+
+ // Invalidate name cache
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(),
+ schema.name());
+ this.nameCache.invalidate(prefixedName);
+ }
+ this.resetCachedAll(type);
+ return true;
+ } else if (Cache.ACTION_CLEAR.equals(args[0])) {
+ event.checkArgs(String.class, HugeType.class);
+ this.clearCache(false);
+ return true;
+ }
+ return false;
+ };
+ EventHub schemaEventHub = this.graphParams().schemaEventHub();
+ if (!schemaEventHub.containsListener(Events.CACHE)) {
+ schemaEventHub.listen(Events.CACHE, this.cacheEventListener);
+ }
+ }
+
+ public void clearCache(boolean notify) {
+ this.idCache.clear();
+ this.nameCache.clear();
+ this.arrayCaches.clear();
+ }
+
+ private void resetCachedAllIfReachedCapacity() {
+ if (this.idCache.size() >= this.idCache.capacity()) {
+ LOG.warn("Schema cache reached capacity({}): {}",
+ this.idCache.capacity(), this.idCache.size());
+ this.cachedTypes().clear();
+ }
+ }
+
+ private void unlistenChanges() {
+ // Unlisten store event
+ this.graphParams().loadGraphStore().provider()
+ .unlisten(this.storeEventListener);
+
+ // Unlisten cache event
+ EventHub schemaEventHub = this.graphParams().schemaEventHub();
+ schemaEventHub.unlisten(Events.CACHE, this.cacheEventListener);
+ }
+
+ private CachedTypes cachedTypes() {
+ return this.arrayCaches.cachedTypes();
+ }
+
+ private void resetCachedAll(HugeType type) {
+ // Set the cache all flag of the schema type to false
+ this.cachedTypes().put(type, false);
+ }
+
+ private void invalidateCache(HugeType type, Id id) {
+ // remove from id cache and name cache
+ Id prefixedId = generateId(type, id);
+ Object value = this.idCache.get(prefixedId);
+ if (value != null) {
+ this.idCache.invalidate(prefixedId);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.invalidate(prefixedName);
+ }
+
+ // remove from optimized array cache
+ this.arrayCaches.remove(type, id);
+ }
+
+ @Override
+ protected void updateSchema(SchemaElement schema,
+ Consumer<SchemaElement> updateCallback) {
+ super.updateSchema(schema, updateCallback);
+
+ this.updateCache(schema);
+ }
+
+ @Override
+ protected void addSchema(SchemaElement schema) {
+ super.addSchema(schema);
+
+ this.updateCache(schema);
+
+ if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
+ MetaManager.instance()
+ // TODO: uncomment later - graph space
+ //.notifySchemaCacheClear(this.graph().graphSpace(),
+ // this.graph().name());
+ .notifySchemaCacheClear("",
+ this.graph().name());
+ }
+ }
+
+ private void updateCache(SchemaElement schema) {
+ this.resetCachedAllIfReachedCapacity();
+
+ // update id cache
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+
+ // update name cache
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+
+ // update optimized array cache
+ this.arrayCaches.updateIfNeeded(schema);
+ }
+
+ @Override
+ protected void removeSchema(SchemaElement schema) {
+ super.removeSchema(schema);
+
+ this.invalidateCache(schema.type(), schema.id());
+
+ if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
+ MetaManager.instance()
+ // TODO: uncomment later - graph space
+ //.notifySchemaCacheClear(this.graph().graphSpace(),
+ // this.graph().name());
+ .notifySchemaCacheClear("",
+ this.graph().name());
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T extends SchemaElement> T getSchema(HugeType type, Id id) {
+ // try get from optimized array cache
+ if (id.number() && id.asLong() > 0L) {
+ SchemaElement value = this.arrayCaches.get(type, id);
+ if (value != null) {
+ return (T) value;
+ }
+ }
+
+ Id prefixedId = generateId(type, id);
+ Object value = this.idCache.get(prefixedId);
+ if (value == null) {
+ value = super.getSchema(type, id);
+ if (value != null) {
+ this.resetCachedAllIfReachedCapacity();
+
+ this.idCache.update(prefixedId, value);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+ }
+ }
+
+ // update optimized array cache
+ this.arrayCaches.updateIfNeeded((SchemaElement) value);
+
+ return (T) value;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T extends SchemaElement> T getSchema(HugeType type,
+ String name) {
+ Id prefixedName = generateId(type, name);
+ Object value = this.nameCache.get(prefixedName);
+ if (value == null) {
+ value = super.getSchema(type, name);
+ if (value != null) {
+ this.resetCachedAllIfReachedCapacity();
+
+ this.nameCache.update(prefixedName, value);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+ }
+ }
+ return (T) value;
+ }
+
+ @Override
+ protected <T extends SchemaElement> List<T> getAllSchema(HugeType type) {
+ Boolean cachedAll = this.cachedTypes().getOrDefault(type, false);
+ List<T> results;
+ if (cachedAll) {
+ results = new ArrayList<>();
+ // Get from cache
+ this.idCache.traverse(value -> {
+ @SuppressWarnings("unchecked")
+ T schema = (T) value;
+ if (schema.type() == type) {
+ results.add(schema);
+ }
+ });
+ return results;
+ } else {
+ results = super.getAllSchema(type);
+ long free = this.idCache.capacity() - this.idCache.size();
+ if (results.size() <= free) {
+ // Update cache
+ for (T schema : results) {
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+ }
+ this.cachedTypes().putIfAbsent(type, true);
+ }
+ return results;
+ }
+ }
+
+ @Override
+ public void clear() {
+ // Clear schema info firstly
+ super.clear();
+ this.clearCache(false);
+ }
+
+ private static final class SchemaCaches<V extends SchemaElement> {
+
+ private final int size;
+
+ private final IntObjectMap<V> pks;
+ private final IntObjectMap<V> vls;
+ private final IntObjectMap<V> els;
+ private final IntObjectMap<V> ils;
+
+ private final CachedTypes cachedTypes;
+
+ public SchemaCaches(int size) {
+ // TODO: improve size of each type for optimized array cache
+ this.size = size;
+
+ this.pks = new IntObjectMap<>(size);
+ this.vls = new IntObjectMap<>(size);
+ this.els = new IntObjectMap<>(size);
+ this.ils = new IntObjectMap<>(size);
+
+ this.cachedTypes = new CachedTypes();
+ }
+
+ public void updateIfNeeded(V schema) {
+ if (schema == null) {
+ return;
+ }
+ Id id = schema.id();
+ if (id.number() && id.asLong() > 0L) {
+ this.set(schema.type(), id, schema);
+ }
+ }
+
+ @PerfUtil.Watched
+ public V get(HugeType type, Id id) {
+ assert id.number();
+ long longId = id.asLong();
+ if (longId <= 0L) {
+ assert false : id;
+ return null;
+ }
+ int key = (int) longId;
Review Comment:
@imbajin ignored? The existing `CachedSchemaTransaction` also has the same issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320779932
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
Review Comment:
临时代码，已删除
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
Review Comment:
临时代码，已删除
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320779888
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/ConditionQuery.java:
##########
@@ -870,4 +876,14 @@ public interface ResultsFilter {
boolean test(HugeElement element);
}
+
+ public byte[] bytes() {
+ Gson gson = new GsonBuilder()
Review Comment:
done
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/QueryAdapter.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.query.Condition;
+import org.apache.hugegraph.type.define.Directions;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.reflect.TypeToken;
+
+public class QueryAdapter extends AbstractSerializerAdapter<Condition> {
+
+ static HashMap<String, Type> cls;
+
+ static {
+ cls = new HashMap() {{
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] imbajin commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "imbajin (via GitHub)" <gi...@apache.org>.
imbajin commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1328197371
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java:
##########
@@ -0,0 +1,469 @@
+package org.apache.hugegraph.backend.cache;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.store.ram.IntObjectMap;
+import org.apache.hugegraph.backend.tx.SchemaTransactionV2;
+import org.apache.hugegraph.config.CoreOptions;
+import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventListener;
+import org.apache.hugegraph.meta.MetaDriver;
+import org.apache.hugegraph.meta.MetaManager;
+import org.apache.hugegraph.perf.PerfUtil;
+import org.apache.hugegraph.schema.SchemaElement;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Events;
+
+import com.google.common.collect.ImmutableSet;
+
+public class CachedSchemaTransactionV2 extends SchemaTransactionV2 {
+ private final Cache<Id, Object> idCache;
+ private final Cache<Id, Object> nameCache;
+
+ private final SchemaCaches<SchemaElement> arrayCaches;
+
+ private EventListener storeEventListener;
+ private EventListener cacheEventListener;
+
+ public CachedSchemaTransactionV2(MetaDriver metaDriver,
+ String cluster,
+ HugeGraphParams graphParams) {
+ super(metaDriver, cluster, graphParams);
+
+ final long capacity = graphParams.configuration()
+ .get(CoreOptions.SCHEMA_CACHE_CAPACITY);
+ this.idCache = this.cache("schema-id", capacity);
+ this.nameCache = this.cache("schema-name", capacity);
+
+ SchemaCaches<SchemaElement> attachment = this.idCache.attachment();
+ if (attachment == null) {
+ int acSize = (int) (capacity >> 3);
+ attachment = this.idCache.attachment(new SchemaCaches<>(acSize));
+ }
+ this.arrayCaches = attachment;
+ this.listenChanges();
+ }
+
+ private static Id generateId(HugeType type, Id id) {
+ // NOTE: it's slower performance to use:
+ // String.format("%x-%s", type.code(), name)
+ return IdGenerator.of(type.string() + "-" + id.asString());
+ }
+
+ private static Id generateId(HugeType type, String name) {
+ return IdGenerator.of(type.string() + "-" + name);
+ }
+
+ public void close() {
+ this.clearCache(false);
+ this.unlistenChanges();
+ }
+
+ private Cache<Id, Object> cache(String prefix, long capacity) {
+ // TODO: uncomment later - graph space
+ //final String name = prefix + "-" + this.graph().spaceGraphName();
+ final String name = prefix + "-" + "";
+ // NOTE: must disable schema cache-expire due to getAllSchema()
+ return CacheManager.instance().cache(name, capacity);
+ }
+
+ private void listenChanges() {
+ // Listen store event: "store.init", "store.clear", ...
+ Set<String> storeEvents = ImmutableSet.of(Events.STORE_INIT,
+ Events.STORE_CLEAR,
+ Events.STORE_TRUNCATE);
+ this.storeEventListener = event -> {
+ if (storeEvents.contains(event.name())) {
+ LOG.debug("Graph {} clear schema cache on event '{}'",
+ this.graph(), event.name());
+ this.clearCache(true);
+ return true;
+ }
+ return false;
+ };
+ this.graphParams().loadGraphStore().provider().listen(this.storeEventListener);
+
+ // Listen cache event: "cache"(invalid cache item)
+ this.cacheEventListener = event -> {
+ LOG.debug("Graph {} received schema cache event: {}",
+ this.graph(), event);
+ Object[] args = event.args();
+ E.checkArgument(args.length > 0 && args[0] instanceof String,
+ "Expect event action argument");
+ if (Cache.ACTION_INVALID.equals(args[0])) {
+ event.checkArgs(String.class, HugeType.class, Id.class);
+ HugeType type = (HugeType) args[1];
+ Id id = (Id) args[2];
+ this.arrayCaches.remove(type, id);
+
+ id = generateId(type, id);
+ Object value = this.idCache.get(id);
+ if (value != null) {
+ // Invalidate id cache
+ this.idCache.invalidate(id);
+
+ // Invalidate name cache
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(),
+ schema.name());
+ this.nameCache.invalidate(prefixedName);
+ }
+ this.resetCachedAll(type);
+ return true;
+ } else if (Cache.ACTION_CLEAR.equals(args[0])) {
+ event.checkArgs(String.class, HugeType.class);
+ this.clearCache(false);
+ return true;
+ }
+ return false;
+ };
+ EventHub schemaEventHub = this.graphParams().schemaEventHub();
+ if (!schemaEventHub.containsListener(Events.CACHE)) {
+ schemaEventHub.listen(Events.CACHE, this.cacheEventListener);
+ }
+ }
+
+ public void clearCache(boolean notify) {
+ this.idCache.clear();
+ this.nameCache.clear();
+ this.arrayCaches.clear();
+ }
+
+ private void resetCachedAllIfReachedCapacity() {
+ if (this.idCache.size() >= this.idCache.capacity()) {
+ LOG.warn("Schema cache reached capacity({}): {}",
+ this.idCache.capacity(), this.idCache.size());
+ this.cachedTypes().clear();
+ }
+ }
+
+ private void unlistenChanges() {
+ // Unlisten store event
+ this.graphParams().loadGraphStore().provider()
+ .unlisten(this.storeEventListener);
+
+ // Unlisten cache event
+ EventHub schemaEventHub = this.graphParams().schemaEventHub();
+ schemaEventHub.unlisten(Events.CACHE, this.cacheEventListener);
+ }
+
+ private CachedTypes cachedTypes() {
+ return this.arrayCaches.cachedTypes();
+ }
+
+ private void resetCachedAll(HugeType type) {
+ // Set the cache all flag of the schema type to false
+ this.cachedTypes().put(type, false);
+ }
+
+ private void invalidateCache(HugeType type, Id id) {
+ // remove from id cache and name cache
+ Id prefixedId = generateId(type, id);
+ Object value = this.idCache.get(prefixedId);
+ if (value != null) {
+ this.idCache.invalidate(prefixedId);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.invalidate(prefixedName);
+ }
+
+ // remove from optimized array cache
+ this.arrayCaches.remove(type, id);
+ }
+
+ @Override
+ protected void updateSchema(SchemaElement schema,
+ Consumer<SchemaElement> updateCallback) {
+ super.updateSchema(schema, updateCallback);
+
+ this.updateCache(schema);
+ }
+
+ @Override
+ protected void addSchema(SchemaElement schema) {
+ super.addSchema(schema);
+
+ this.updateCache(schema);
+
+ if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
+ MetaManager.instance()
+ // TODO: uncomment later - graph space
+ //.notifySchemaCacheClear(this.graph().graphSpace(),
+ // this.graph().name());
+ .notifySchemaCacheClear("",
+ this.graph().name());
+ }
+ }
+
+ private void updateCache(SchemaElement schema) {
+ this.resetCachedAllIfReachedCapacity();
+
+ // update id cache
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+
+ // update name cache
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+
+ // update optimized array cache
+ this.arrayCaches.updateIfNeeded(schema);
+ }
+
+ @Override
+ protected void removeSchema(SchemaElement schema) {
+ super.removeSchema(schema);
+
+ this.invalidateCache(schema.type(), schema.id());
+
+ if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
+ MetaManager.instance()
+ // TODO: uncomment later - graph space
+ //.notifySchemaCacheClear(this.graph().graphSpace(),
+ // this.graph().name());
+ .notifySchemaCacheClear("",
+ this.graph().name());
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T extends SchemaElement> T getSchema(HugeType type, Id id) {
+ // try get from optimized array cache
+ if (id.number() && id.asLong() > 0L) {
+ SchemaElement value = this.arrayCaches.get(type, id);
+ if (value != null) {
+ return (T) value;
+ }
+ }
+
+ Id prefixedId = generateId(type, id);
+ Object value = this.idCache.get(prefixedId);
+ if (value == null) {
+ value = super.getSchema(type, id);
+ if (value != null) {
+ this.resetCachedAllIfReachedCapacity();
+
+ this.idCache.update(prefixedId, value);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+ }
+ }
+
+ // update optimized array cache
+ this.arrayCaches.updateIfNeeded((SchemaElement) value);
+
+ return (T) value;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T extends SchemaElement> T getSchema(HugeType type,
+ String name) {
+ Id prefixedName = generateId(type, name);
+ Object value = this.nameCache.get(prefixedName);
+ if (value == null) {
+ value = super.getSchema(type, name);
+ if (value != null) {
+ this.resetCachedAllIfReachedCapacity();
+
+ this.nameCache.update(prefixedName, value);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+ }
+ }
+ return (T) value;
+ }
+
+ @Override
+ protected <T extends SchemaElement> List<T> getAllSchema(HugeType type) {
+ Boolean cachedAll = this.cachedTypes().getOrDefault(type, false);
+ List<T> results;
+ if (cachedAll) {
+ results = new ArrayList<>();
+ // Get from cache
+ this.idCache.traverse(value -> {
+ @SuppressWarnings("unchecked")
+ T schema = (T) value;
+ if (schema.type() == type) {
+ results.add(schema);
+ }
+ });
+ return results;
+ } else {
+ results = super.getAllSchema(type);
+ long free = this.idCache.capacity() - this.idCache.size();
+ if (results.size() <= free) {
+ // Update cache
+ for (T schema : results) {
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+ }
+ this.cachedTypes().putIfAbsent(type, true);
+ }
+ return results;
+ }
+ }
+
+ @Override
+ public void clear() {
+ // Clear schema info firstly
+ super.clear();
+ this.clearCache(false);
+ }
+
+ private static final class SchemaCaches<V extends SchemaElement> {
+
+ private final int size;
+
+ private final IntObjectMap<V> pks;
+ private final IntObjectMap<V> vls;
+ private final IntObjectMap<V> els;
+ private final IntObjectMap<V> ils;
+
+ private final CachedTypes cachedTypes;
+
+ public SchemaCaches(int size) {
+ // TODO: improve size of each type for optimized array cache
+ this.size = size;
+
+ this.pks = new IntObjectMap<>(size);
+ this.vls = new IntObjectMap<>(size);
+ this.els = new IntObjectMap<>(size);
+ this.ils = new IntObjectMap<>(size);
+
+ this.cachedTypes = new CachedTypes();
+ }
+
+ public void updateIfNeeded(V schema) {
+ if (schema == null) {
+ return;
+ }
+ Id id = schema.id();
+ if (id.number() && id.asLong() > 0L) {
+ this.set(schema.type(), id, schema);
+ }
+ }
+
+ @PerfUtil.Watched
+ public V get(HugeType type, Id id) {
+ assert id.number();
+ long longId = id.asLong();
+ if (longId <= 0L) {
+ assert false : id;
+ return null;
+ }
+ int key = (int) longId;
Review Comment:
> @imbajin ignored? The existing `CachedSchemaTransaction` also has the same issue.
Fine, both OK to ignore or fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320780010
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriver.java:
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.meta;
+
+import java.io.File;
+import java.net.URI;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.DistributedLock;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.type.define.CollectionType;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.collection.CollectionFactory;
+
+import com.google.common.base.Strings;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.ClientBuilder;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
+import io.etcd.jetcd.options.DeleteOption;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchEvent;
+import io.etcd.jetcd.watch.WatchResponse;
+import io.netty.handler.ssl.ApplicationProtocolConfig;
+import io.netty.handler.ssl.ApplicationProtocolNames;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslProvider;
+
+public class EtcdMetaDriver implements MetaDriver {
+
+ private Client client;
+ private DistributedLock lock;
+
+ public EtcdMetaDriver(String trustFile, String clientCertFile,
+ String clientKeyFile, Object... endpoints) {
+ ClientBuilder builder = this.etcdMetaDriverBuilder(endpoints);
+
+ SslContext sslContext = openSslContext(trustFile, clientCertFile,
+ clientKeyFile);
+ this.client = builder.sslContext(sslContext).build();
+ this.lock = DistributedLock.getInstance(this.client);
+ }
+
+ public EtcdMetaDriver(Object... endpoints) {
+ ClientBuilder builder = this.etcdMetaDriverBuilder(endpoints);
+ this.client = builder.build();
+ this.lock = DistributedLock.getInstance(this.client);
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content.getBytes());
+ }
+
+ private static boolean isEtcdPut(WatchEvent event) {
+ return event.getEventType() == WatchEvent.EventType.PUT;
+ }
+
+ public static SslContext openSslContext(String trustFile,
+ String clientCertFile,
+ String clientKeyFile) {
+ SslContext ssl;
+ try {
+ File trustManagerFile = FileUtils.getFile(trustFile);
+ File keyCertChainFile = FileUtils.getFile(clientCertFile);
+ File KeyFile = FileUtils.getFile(clientKeyFile);
+ ApplicationProtocolConfig alpn = new ApplicationProtocolConfig(
+ ApplicationProtocolConfig.Protocol.ALPN,
+ ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
+
+ ApplicationProtocolConfig.SelectedListenerFailureBehavior
+ .ACCEPT,
+ ApplicationProtocolNames.HTTP_2);
+
+ ssl = SslContextBuilder.forClient()
+ .applicationProtocolConfig(alpn)
+ .sslProvider(SslProvider.OPENSSL)
+ .trustManager(trustManagerFile)
+ .keyManager(keyCertChainFile, KeyFile)
+ .build();
+ } catch (Exception e) {
+ throw new HugeException("Failed to open ssl context", e);
+ }
+ return ssl;
+ }
+
+ public ClientBuilder etcdMetaDriverBuilder(Object... endpoints) {
+ int length = endpoints.length;
+ ClientBuilder builder = null;
+ if (endpoints[0] instanceof List && endpoints.length == 1) {
+ builder = Client.builder()
+ .endpoints(((List<String>) endpoints[0])
+ .toArray(new String[0]));
+ } else if (endpoints[0] instanceof String) {
+ for (int i = 1; i < length; i++) {
+ E.checkArgument(endpoints[i] instanceof String,
+ "Inconsistent endpoint %s(%s) with %s(%s)",
+ endpoints[i], endpoints[i].getClass(),
+ endpoints[0], endpoints[0].getClass());
+ }
+ builder = Client.builder().endpoints((String[]) endpoints);
+ } else if (endpoints[0] instanceof URI) {
+ for (int i = 1; i < length; i++) {
+ E.checkArgument(endpoints[i] instanceof String,
+ "Invalid endpoint %s(%s)",
+ endpoints[i], endpoints[i].getClass(),
+ endpoints[0], endpoints[0].getClass());
+ }
+ builder = Client.builder().endpoints((URI[]) endpoints);
+ } else {
+ E.checkArgument(false, "Invalid endpoint %s(%s)",
+ endpoints[0], endpoints[0].getClass());
+ }
+ return builder;
+ }
+
+ @Override
+ public long keepAlive(String key, long leaseId) {
+ try {
+ LeaseKeepAliveResponse response =
+ this.client.getLeaseClient().keepAliveOnce(leaseId).get();
+ return response.getID();
+ } catch (InterruptedException | ExecutionException e) {
+ // keepAlive once Failed
+ return 0;
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ List<KeyValue> keyValues;
+ KV kvClient = this.client.getKVClient();
+ try {
+ keyValues = kvClient.get(toByteSequence(key))
+ .get().getKvs();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new HugeException("Failed to get key '%s' from etcd", e, key);
+ }
+
+ if (keyValues.size() > 0) {
+ return keyValues.get(0).getValue().toString(Charset.defaultCharset());
+ }
+
+ return null;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ KV kvClient = this.client.getKVClient();
+ try {
+ kvClient.put(toByteSequence(key), toByteSequence(value)).get();
+ } catch (InterruptedException | ExecutionException e) {
+ try {
+ kvClient.delete(toByteSequence(key)).get();
+ } catch (Throwable t) {
+ throw new HugeException("Failed to put '%s:%s' to etcd",
+ e, key, value);
+ }
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ KV kvClient = this.client.getKVClient();
+ try {
+ kvClient.delete(toByteSequence(key)).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new HugeException(
+ "Failed to delete key '%s' from etcd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ KV kvClient = this.client.getKVClient();
+ try {
+ DeleteOption option = DeleteOption.newBuilder()
+ .isPrefix(true)
+ .build();
+ kvClient.delete(toByteSequence(prefix), option);
+ } catch (Throwable e) {
+ throw new HugeException(
+ "Failed to delete prefix '%s' from etcd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ GetOption getOption = GetOption.newBuilder()
+ .isPrefix(true)
+ .build();
+ GetResponse response;
+ try {
+ response = this.client.getKVClient().get(toByteSequence(prefix),
+ getOption).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new HugeException("Failed to scan etcd with prefix '%s'",
+ e, prefix);
+ }
+ int size = (int) response.getCount();
+ Map<String, String> keyValues = CollectionFactory.newMap(
+ CollectionType.JCF, size);
+ for (KeyValue kv : response.getKvs()) {
+ String key = kv.getKey().toString(Charset.defaultCharset());
+ String value = kv.getValue().size() == 0 ? "" :
+ kv.getValue().toString(Charset.defaultCharset());
+ keyValues.put(key, value);
+ }
+ return keyValues;
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ E.checkArgument(response instanceof WatchResponse,
+ "Invalid response type %s", response.getClass());
+ for (WatchEvent event : ((WatchResponse) response).getEvents()) {
+ // Skip if not etcd PUT event
+ if (!isEtcdPut(event)) {
+ return null;
+ }
+
+ String value = event.getKeyValue().getValue()
+ .toString(Charset.defaultCharset());
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ E.checkArgument(response instanceof WatchResponse,
+ "Invalid response type %s", response.getClass());
+
+ Map<String, String> resultMap = new HashMap<>();
+ for (WatchEvent event : ((WatchResponse) response).getEvents()) {
+ // Skip if not etcd PUT event
+ if (!isEtcdPut(event)) {
+ continue;
+ }
+
+ String key = event.getKeyValue().getKey().toString(Charset.defaultCharset());
+ String value = event.getKeyValue().getValue()
+ .toString(Charset.defaultCharset());
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult tryLock(String key, long ttl, long timeout) {
+ return this.lock.tryLock(key, ttl, timeout);
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ return this.lock.lock(key, ttl);
+ }
+
+ @Override
+ public boolean isLocked(String key) {
+ try {
+ long size = this.client.getKVClient().get(toByteSequence(key))
+ .get().getCount();
+
+ if (size > 0) {
+ return true;
+ } else {
+ return false;
+ }
Review Comment:
done
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
Review Comment:
done
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] zyxxoo merged pull request #2301: feat: preliminary support for hstore backend
Posted by "zyxxoo (via GitHub)" <gi...@apache.org>.
zyxxoo merged PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] zyxxoo commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "zyxxoo (via GitHub)" <gi...@apache.org>.
zyxxoo commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1313840490
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/IdCounter.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.tx;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hugegraph.backend.BackendException;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.Pdpb;
+import org.apache.hugegraph.store.term.HgPair;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.util.E;
+
+public class IdCounter {
+
+ private static final int TIMES = 10000;
+ private static final int DELTA = 10000;
+ private static final String DELIMITER = "/";
+ private static Map<String, HgPair> ids = new ConcurrentHashMap<>();
Review Comment:
volatile？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/PdDistributedLock.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.meta.lock;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public class PdDistributedLock extends AbstractDistributedLock {
+
+ private static int poolSize = 8;
+ private final KvClient client;
+ private ScheduledExecutorService service = new ScheduledThreadPoolExecutor(poolSize, r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ return t;
+ });
+
+ public PdDistributedLock(KvClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public LockResult lock(String key, long second) {
+ long ttl = second * 1000L;
+ try {
+ LockResponse response = this.client.lock(key, ttl);
+ boolean succeed = response.getSucceed();
+ LockResult result = new LockResult();
+ if (succeed) {
+ result.setLeaseId(response.getClientId());
+ result.lockSuccess(true);
+ long period = ttl - ttl / 4;
+ ScheduledFuture<?> future = service.scheduleAtFixedRate(() -> {
+ synchronized (result) {
+ keepAlive(key);
+ }
+ }, 10, period, TimeUnit.MILLISECONDS);
+ result.setFuture(future);
+ }
+ return result;
+ } catch (PDException e) {
+ throw new HugeException("Failed to lock '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public void unLock(String key, LockResult lockResult) {
+ try {
+ LockResponse response = this.client.unlock(key);
+ boolean succeed = response.getSucceed();
+ if (succeed == false) {
+ throw new HugeException("Failed to unlock '%s' to pd", key);
+ }
+ if (lockResult.getFuture() != null) {
+ synchronized (lockResult) {
Review Comment:
同
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/AbstractDistributedLock.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.meta.lock;
+
+import org.apache.hugegraph.pd.client.KvClient;
+
+import io.etcd.jetcd.Client;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public abstract class AbstractDistributedLock {
+
+ private static volatile AbstractDistributedLock defaultLock;
+
+ public static AbstractDistributedLock getInstance(AutoCloseable client) {
Review Comment:
这单例不合适吧？每次都要传client
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/PdDistributedLock.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.meta.lock;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public class PdDistributedLock extends AbstractDistributedLock {
+
+ private static int poolSize = 8;
+ private final KvClient client;
+ private ScheduledExecutorService service = new ScheduledThreadPoolExecutor(poolSize, r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ return t;
+ });
+
+ public PdDistributedLock(KvClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public LockResult lock(String key, long second) {
+ long ttl = second * 1000L;
+ try {
+ LockResponse response = this.client.lock(key, ttl);
+ boolean succeed = response.getSucceed();
+ LockResult result = new LockResult();
+ if (succeed) {
+ result.setLeaseId(response.getClientId());
+ result.lockSuccess(true);
+ long period = ttl - ttl / 4;
+ ScheduledFuture<?> future = service.scheduleAtFixedRate(() -> {
+ synchronized (result) {
+ keepAlive(key);
+ }
+ }, 10, period, TimeUnit.MILLISECONDS);
+ result.setFuture(future);
+ }
+ return result;
+ } catch (PDException e) {
+ throw new HugeException("Failed to lock '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public void unLock(String key, LockResult lockResult) {
+ try {
+ LockResponse response = this.client.unlock(key);
+ boolean succeed = response.getSucceed();
+ if (succeed == false) {
+ throw new HugeException("Failed to unlock '%s' to pd", key);
+ }
+ if (lockResult.getFuture() != null) {
+ synchronized (lockResult) {
Review Comment:
同
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/PdDistributedLock.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.meta.lock;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public class PdDistributedLock extends AbstractDistributedLock {
+
+ private static int poolSize = 8;
+ private final KvClient client;
+ private ScheduledExecutorService service = new ScheduledThreadPoolExecutor(poolSize, r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ return t;
+ });
+
+ public PdDistributedLock(KvClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public LockResult lock(String key, long second) {
+ long ttl = second * 1000L;
+ try {
+ LockResponse response = this.client.lock(key, ttl);
+ boolean succeed = response.getSucceed();
+ LockResult result = new LockResult();
+ if (succeed) {
+ result.setLeaseId(response.getClientId());
+ result.lockSuccess(true);
+ long period = ttl - ttl / 4;
+ ScheduledFuture<?> future = service.scheduleAtFixedRate(() -> {
+ synchronized (result) {
Review Comment:
这里为啥加锁
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
Review Comment:
Private final
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/space/SchemaTemplate.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.space;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map;
+
+import org.apache.hugegraph.util.E;
+
+import com.google.common.collect.ImmutableMap;
+
+public class SchemaTemplate {
+
+ public static SimpleDateFormat FORMATTER = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ protected Date createTime;
+ protected Date updateTime;
+ protected String creator;
+ private String name;
Review Comment:
final
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
Review Comment:
This
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/AbstractDistributedLock.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.meta.lock;
+
+import org.apache.hugegraph.pd.client.KvClient;
+
+import io.etcd.jetcd.Client;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public abstract class AbstractDistributedLock {
+
+ private static volatile AbstractDistributedLock defaultLock;
+
+ public static AbstractDistributedLock getInstance(AutoCloseable client) {
Review Comment:
这单例不合适吧？每次都要传client
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/IdCounter.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.tx;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hugegraph.backend.BackendException;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.Pdpb;
+import org.apache.hugegraph.store.term.HgPair;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.util.E;
+
+public class IdCounter {
+
+ private static final int TIMES = 10000;
+ private static final int DELTA = 10000;
+ private static final String DELIMITER = "/";
+ private static Map<String, HgPair> ids = new ConcurrentHashMap<>();
Review Comment:
volatile？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/EtcdMetaDriver.java:
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.meta;
+
+import java.io.File;
+import java.net.URI;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.DistributedLock;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.type.define.CollectionType;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.collection.CollectionFactory;
+
+import com.google.common.base.Strings;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.ClientBuilder;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
+import io.etcd.jetcd.options.DeleteOption;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchEvent;
+import io.etcd.jetcd.watch.WatchResponse;
+import io.netty.handler.ssl.ApplicationProtocolConfig;
+import io.netty.handler.ssl.ApplicationProtocolNames;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslProvider;
+
+public class EtcdMetaDriver implements MetaDriver {
+
+ private Client client;
+ private DistributedLock lock;
+
+ public EtcdMetaDriver(String trustFile, String clientCertFile,
+ String clientKeyFile, Object... endpoints) {
+ ClientBuilder builder = this.etcdMetaDriverBuilder(endpoints);
+
+ SslContext sslContext = openSslContext(trustFile, clientCertFile,
+ clientKeyFile);
+ this.client = builder.sslContext(sslContext).build();
+ this.lock = DistributedLock.getInstance(this.client);
+ }
+
+ public EtcdMetaDriver(Object... endpoints) {
+ ClientBuilder builder = this.etcdMetaDriverBuilder(endpoints);
+ this.client = builder.build();
+ this.lock = DistributedLock.getInstance(this.client);
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content.getBytes());
+ }
+
+ private static boolean isEtcdPut(WatchEvent event) {
+ return event.getEventType() == WatchEvent.EventType.PUT;
+ }
+
+ public static SslContext openSslContext(String trustFile,
+ String clientCertFile,
+ String clientKeyFile) {
+ SslContext ssl;
+ try {
+ File trustManagerFile = FileUtils.getFile(trustFile);
+ File keyCertChainFile = FileUtils.getFile(clientCertFile);
+ File KeyFile = FileUtils.getFile(clientKeyFile);
+ ApplicationProtocolConfig alpn = new ApplicationProtocolConfig(
+ ApplicationProtocolConfig.Protocol.ALPN,
+ ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
+
+ ApplicationProtocolConfig.SelectedListenerFailureBehavior
+ .ACCEPT,
+ ApplicationProtocolNames.HTTP_2);
+
+ ssl = SslContextBuilder.forClient()
+ .applicationProtocolConfig(alpn)
+ .sslProvider(SslProvider.OPENSSL)
+ .trustManager(trustManagerFile)
+ .keyManager(keyCertChainFile, KeyFile)
+ .build();
+ } catch (Exception e) {
+ throw new HugeException("Failed to open ssl context", e);
+ }
+ return ssl;
+ }
+
+ public ClientBuilder etcdMetaDriverBuilder(Object... endpoints) {
+ int length = endpoints.length;
+ ClientBuilder builder = null;
+ if (endpoints[0] instanceof List && endpoints.length == 1) {
+ builder = Client.builder()
+ .endpoints(((List<String>) endpoints[0])
+ .toArray(new String[0]));
+ } else if (endpoints[0] instanceof String) {
+ for (int i = 1; i < length; i++) {
+ E.checkArgument(endpoints[i] instanceof String,
+ "Inconsistent endpoint %s(%s) with %s(%s)",
+ endpoints[i], endpoints[i].getClass(),
+ endpoints[0], endpoints[0].getClass());
+ }
+ builder = Client.builder().endpoints((String[]) endpoints);
+ } else if (endpoints[0] instanceof URI) {
+ for (int i = 1; i < length; i++) {
+ E.checkArgument(endpoints[i] instanceof String,
+ "Invalid endpoint %s(%s)",
+ endpoints[i], endpoints[i].getClass(),
+ endpoints[0], endpoints[0].getClass());
+ }
+ builder = Client.builder().endpoints((URI[]) endpoints);
+ } else {
+ E.checkArgument(false, "Invalid endpoint %s(%s)",
+ endpoints[0], endpoints[0].getClass());
+ }
+ return builder;
+ }
+
+ @Override
+ public long keepAlive(String key, long leaseId) {
+ try {
+ LeaseKeepAliveResponse response =
+ this.client.getLeaseClient().keepAliveOnce(leaseId).get();
+ return response.getID();
+ } catch (InterruptedException | ExecutionException e) {
+ // keepAlive once Failed
+ return 0;
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ List<KeyValue> keyValues;
+ KV kvClient = this.client.getKVClient();
+ try {
+ keyValues = kvClient.get(toByteSequence(key))
+ .get().getKvs();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new HugeException("Failed to get key '%s' from etcd", e, key);
+ }
+
+ if (keyValues.size() > 0) {
+ return keyValues.get(0).getValue().toString(Charset.defaultCharset());
+ }
+
+ return null;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ KV kvClient = this.client.getKVClient();
+ try {
+ kvClient.put(toByteSequence(key), toByteSequence(value)).get();
+ } catch (InterruptedException | ExecutionException e) {
+ try {
+ kvClient.delete(toByteSequence(key)).get();
+ } catch (Throwable t) {
+ throw new HugeException("Failed to put '%s:%s' to etcd",
+ e, key, value);
+ }
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ KV kvClient = this.client.getKVClient();
+ try {
+ kvClient.delete(toByteSequence(key)).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new HugeException(
+ "Failed to delete key '%s' from etcd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ KV kvClient = this.client.getKVClient();
+ try {
+ DeleteOption option = DeleteOption.newBuilder()
+ .isPrefix(true)
+ .build();
+ kvClient.delete(toByteSequence(prefix), option);
+ } catch (Throwable e) {
+ throw new HugeException(
+ "Failed to delete prefix '%s' from etcd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ GetOption getOption = GetOption.newBuilder()
+ .isPrefix(true)
+ .build();
+ GetResponse response;
+ try {
+ response = this.client.getKVClient().get(toByteSequence(prefix),
+ getOption).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new HugeException("Failed to scan etcd with prefix '%s'",
+ e, prefix);
+ }
+ int size = (int) response.getCount();
+ Map<String, String> keyValues = CollectionFactory.newMap(
+ CollectionType.JCF, size);
+ for (KeyValue kv : response.getKvs()) {
+ String key = kv.getKey().toString(Charset.defaultCharset());
+ String value = kv.getValue().size() == 0 ? "" :
+ kv.getValue().toString(Charset.defaultCharset());
+ keyValues.put(key, value);
+ }
+ return keyValues;
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ E.checkArgument(response instanceof WatchResponse,
+ "Invalid response type %s", response.getClass());
+ for (WatchEvent event : ((WatchResponse) response).getEvents()) {
+ // Skip if not etcd PUT event
+ if (!isEtcdPut(event)) {
+ return null;
+ }
+
+ String value = event.getKeyValue().getValue()
+ .toString(Charset.defaultCharset());
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ E.checkArgument(response instanceof WatchResponse,
+ "Invalid response type %s", response.getClass());
+
+ Map<String, String> resultMap = new HashMap<>();
+ for (WatchEvent event : ((WatchResponse) response).getEvents()) {
+ // Skip if not etcd PUT event
+ if (!isEtcdPut(event)) {
+ continue;
+ }
+
+ String key = event.getKeyValue().getKey().toString(Charset.defaultCharset());
+ String value = event.getKeyValue().getValue()
+ .toString(Charset.defaultCharset());
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult tryLock(String key, long ttl, long timeout) {
+ return this.lock.tryLock(key, ttl, timeout);
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ return this.lock.lock(key, ttl);
+ }
+
+ @Override
+ public boolean isLocked(String key) {
+ try {
+ long size = this.client.getKVClient().get(toByteSequence(key))
+ .get().getCount();
+
+ if (size > 0) {
+ return true;
+ } else {
+ return false;
+ }
Review Comment:
return size>0
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/PdDistributedLock.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.meta.lock;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public class PdDistributedLock extends AbstractDistributedLock {
+
+ private static int poolSize = 8;
+ private final KvClient client;
+ private ScheduledExecutorService service = new ScheduledThreadPoolExecutor(poolSize, r -> {
+ Thread t = new Thread(r);
+ t.setDaemon(true);
+ return t;
+ });
+
+ public PdDistributedLock(KvClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public LockResult lock(String key, long second) {
+ long ttl = second * 1000L;
+ try {
+ LockResponse response = this.client.lock(key, ttl);
+ boolean succeed = response.getSucceed();
+ LockResult result = new LockResult();
+ if (succeed) {
+ result.setLeaseId(response.getClientId());
+ result.lockSuccess(true);
+ long period = ttl - ttl / 4;
+ ScheduledFuture<?> future = service.scheduleAtFixedRate(() -> {
+ synchronized (result) {
Review Comment:
这里为啥加锁
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] zyxxoo commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "zyxxoo (via GitHub)" <gi...@apache.org>.
zyxxoo commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1313828473
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/auth/SchemaDefine.java:
##########
@@ -281,6 +290,19 @@ public abstract static class Relationship extends AuthElement {
public abstract Id target();
+ public void setId() {
+ this.id(IdGenerator.of(this.source().asString() + "->" +
+ this.target().asString()));
+ }
+
+ public static <T extends Relationship> T fromMap(Map<String, Object> map, T entity) {
Review Comment:
fromMap名字有点奇怪，要不要改一下？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
+ SimpleDateFormat ft = new SimpleDateFormat("HHmmss");
+ System.out.println("当前时间为: " + ft.format(dNow));
+ }
+
+ public PDClient pdClient() {
+ return this.pdClient;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ try {
+ this.client.put(key, value);
+ } catch (PDException e) {
+ throw new HugeException("Failed to put '%s:%s' to pd", e, key, value);
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ try {
+ KResponse response = this.client.get(key);
+ return response.getValue();
+ } catch (PDException e) {
+ throw new HugeException("Failed to get '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ try {
+ this.client.delete(key);
+ } catch (PDException e) {
+ throw new HugeException("Failed to delete '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ try {
+ this.client.deletePrefix(prefix);
+ } catch (PDException e) {
+ throw new HugeException("Failed to deleteWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ try {
+ ScanPrefixResponse response = this.client.scanPrefix(prefix);
+ return response.getKvsMap();
+ } catch (PDException e) {
+ throw new HugeException("Failed to scanWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> void listen(String key, Consumer<T> consumer) {
+ try {
+ this.client.listen(key, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listen '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
+ try {
+ this.client.listenPrefix(prefix, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listenPrefix '%s' to pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ return null;
+ }
+ String value = event.getCurrent().getValue();
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ Map<String, String> resultMap = new HashMap<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not etcd PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ continue;
+ }
+
+ String key = event.getCurrent().getKey();
+ String value = event.getCurrent().getValue();
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ while (true) {
Review Comment:
没有次数限制嘛？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java:
##########
@@ -453,9 +458,25 @@ private void clearVertexCache() {
}
}
- private SchemaTransaction openSchemaTransaction() throws HugeException {
+ private boolean isHstore() {
+ return this.storeProvider.isHstore();
+ }
+
+ private void initMetaManager() {
+ this.metaManager.connect("hg", MetaManager.MetaDriverType.PD,
Review Comment:
这个链接是不是每个新的链接都要调用一次？是否只执行一次就可以呢？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/auth/HugeRole.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.auth;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.auth.SchemaDefine.Entity;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.util.E;
+import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
+import org.apache.tinkerpop.gremlin.structure.T;
+
+public class HugeRole extends Entity {
+
+ private static final long serialVersionUID = 2330399818352242686L;
+
+ private String name;
+ private String nickname;
+ private String graphSpace;
+ private String description;
+
+ public HugeRole(Id id, String name, String graphSpace) {
+ this.id = id;
+ this.name = name;
+ this.graphSpace = graphSpace;
+ this.description = null;
+ }
+
+ public HugeRole(String name, String graphSpace) {
+ this(StringUtils.isNotEmpty(name) ? IdGenerator.of(name) : null,
+ name, graphSpace);
+ }
+
+ public HugeRole(Id id, String graphSpace) {
+ this(id, id.asString(), graphSpace);
+ }
+
+ public static HugeRole fromMap(Map<String, Object> map) {
+ HugeRole role = new HugeRole("", "");
+ return fromMap(map, role);
+ }
+
+ public static Schema schema(HugeGraphParams graph) {
+ return new Schema(graph);
+ }
+
+ @Override
+ public ResourceType type() {
+ return ResourceType.GRANT;
+ }
+
+ @Override
+ public String label() {
+ return P.ROLE;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ public void name(String name) {
+ this.name = name;
+ }
+
+ public String nickname() {
+ return this.nickname;
+ }
+
+ public void nickname(String nickname) {
+ this.nickname = nickname;
+ }
+
+ public String graphSpace() {
+ return this.graphSpace;
+ }
+
+ public String description() {
+ return this.description;
+ }
+
+ public void description(String description) {
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("HugeGroup(%s)", this.id);
+ }
+
+ @Override
+ protected boolean property(String key, Object value) {
+ if (super.property(key, value)) {
+ return true;
+ }
+ switch (key) {
+ case P.GRAPHSPACE:
+ this.graphSpace = (String) value;
+ break;
+ case P.NAME:
+ this.name = (String) value;
+ break;
+ case P.NICKNAME:
+ this.nickname = (String) value;
+ break;
+ case P.DESCRIPTION:
+ this.description = (String) value;
+ break;
+ default:
+ throw new AssertionError("Unsupported key: " + key);
+ }
+ return true;
+ }
+
+ @Override
+ protected Object[] asArray() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ List<Object> list = new ArrayList<>(12);
+
+ list.add(T.label);
+ list.add(P.ROLE);
+
+ list.add(P.GRAPHSPACE);
+ list.add(this.graphSpace);
+
+ list.add(P.NAME);
+ list.add(this.name);
+
+ if (this.nickname != null) {
+ list.add(P.NICKNAME);
+ list.add(this.nickname);
+ }
+
+ if (this.description != null) {
+ list.add(P.DESCRIPTION);
+ list.add(this.description);
+ }
+
+ return super.asArray(list);
+ }
+
+ @Override
+ public Map<String, Object> asMap() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ Map<String, Object> map = new HashMap<>();
+
+ map.put(Hidden.unHide(P.NAME), this.name);
+ map.put(Hidden.unHide(P.GRAPHSPACE), this.graphSpace);
+ if (this.description != null) {
+ map.put(Hidden.unHide(P.DESCRIPTION), this.description);
+ }
+
+ if (this.nickname != null) {
+ map.put(Hidden.unHide(P.NICKNAME), this.nickname);
+ }
+
+ return super.asMap(map);
+ }
+
+ public static final class P {
+
+ public static final String ROLE = Hidden.hide("role");
+
+ public static final String ID = T.id.getAccessor();
+ public static final String LABEL = T.label.getAccessor();
+
+ public static final String NAME = "~role_name";
+ public static final String NICKNAME = "~role_nickname";
+ public static final String GRAPHSPACE = "~graphspace";
+ public static final String DESCRIPTION = "~role_description";
Review Comment:
这里咋不用hide, hide就是加～前缀吧
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/ConditionQuery.java:
##########
@@ -870,4 +876,14 @@ public interface ResultsFilter {
boolean test(HugeElement element);
}
+
+ public byte[] bytes() {
+ Gson gson = new GsonBuilder()
Review Comment:
Gson可以做成静态全局变量，不用每次创建了
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendStoreInfo.java:
##########
@@ -41,6 +41,9 @@ public boolean exists() {
}
public boolean checkVersion() {
+ if (this.storeProvider.isHstore()) { // skip for hstore now
Review Comment:
hstore暂时不用考虑兼容问题嘛？如果用户滚动升级，节点版本不一致，是否会有风险？
如果有的话，是否检查下版本比较合适呢？至少保证不兼容的版本不会产生脏数据
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/AbstractSerializerAdapter.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.BackendException;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.JsonSerializer;
+
+public abstract class AbstractSerializerAdapter<T> implements JsonSerializer<T>,
Review Comment:
以后可以考虑优化成二进制协议
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/AbstractSerializerAdapter.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.BackendException;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.JsonSerializer;
+
+public abstract class AbstractSerializerAdapter<T> implements JsonSerializer<T>,
Review Comment:
以后可以考虑优化成二进制协议
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/ConditionQuery.java:
##########
@@ -870,4 +876,14 @@ public interface ResultsFilter {
boolean test(HugeElement element);
}
+
+ public byte[] bytes() {
+ Gson gson = new GsonBuilder()
Review Comment:
Gson可以做成静态全局变量，不用每次创建了
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendStoreInfo.java:
##########
@@ -41,6 +41,9 @@ public boolean exists() {
}
public boolean checkVersion() {
+ if (this.storeProvider.isHstore()) { // skip for hstore now
Review Comment:
hstore暂时不用考虑兼容问题嘛？如果用户滚动升级，节点版本不一致，是否会有风险？
如果有的话，是否检查下版本比较合适呢？至少保证不兼容的版本不会产生脏数据
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
Review Comment:
log
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/DistributedLock.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.meta.lock;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+
+public class DistributedLock extends AbstractDistributedLock {
+
+ protected static final Logger LOG = Log.logger(DistributedLock.class);
+ private static final long UNLIMIT_TIMEOUT = -1L;
+ private final static Object mutex = new Object();
+ private static DistributedLock lockProvider = null;
+ private final KV kvClient;
+ private final Lock lockClient;
+ private final Lease leaseClient;
+
+ private DistributedLock(Client client) {
+ this.kvClient = client.getKVClient();
+ this.lockClient = client.getLockClient();
+ this.leaseClient = client.getLeaseClient();
+ }
+
+ public static DistributedLock getInstance(Client client) {
+ synchronized (mutex) {
+ if (null == lockProvider) {
+ lockProvider = new DistributedLock(client);
+ }
+ }
+ return lockProvider;
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content, Charset.defaultCharset());
+ }
+
+ public LockResult tryLock(String lockName, long ttl, long timeout) {
+ LockResult lockResult = new LockResult();
+ ScheduledExecutorService service =
+ Executors.newSingleThreadScheduledExecutor();
Review Comment:
这个是不是可以放全局队列来使用，没看错的话，这里的意思是没次lock就开一个线程来keep alive
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/auth/HugeRole.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.auth;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.auth.SchemaDefine.Entity;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.util.E;
+import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
+import org.apache.tinkerpop.gremlin.structure.T;
+
+public class HugeRole extends Entity {
+
+ private static final long serialVersionUID = 2330399818352242686L;
+
+ private String name;
+ private String nickname;
+ private String graphSpace;
+ private String description;
+
+ public HugeRole(Id id, String name, String graphSpace) {
+ this.id = id;
+ this.name = name;
+ this.graphSpace = graphSpace;
+ this.description = null;
+ }
+
+ public HugeRole(String name, String graphSpace) {
+ this(StringUtils.isNotEmpty(name) ? IdGenerator.of(name) : null,
+ name, graphSpace);
+ }
+
+ public HugeRole(Id id, String graphSpace) {
+ this(id, id.asString(), graphSpace);
+ }
+
+ public static HugeRole fromMap(Map<String, Object> map) {
+ HugeRole role = new HugeRole("", "");
+ return fromMap(map, role);
+ }
+
+ public static Schema schema(HugeGraphParams graph) {
+ return new Schema(graph);
+ }
+
+ @Override
+ public ResourceType type() {
+ return ResourceType.GRANT;
+ }
+
+ @Override
+ public String label() {
+ return P.ROLE;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ public void name(String name) {
+ this.name = name;
+ }
+
+ public String nickname() {
+ return this.nickname;
+ }
+
+ public void nickname(String nickname) {
+ this.nickname = nickname;
+ }
+
+ public String graphSpace() {
+ return this.graphSpace;
+ }
+
+ public String description() {
+ return this.description;
+ }
+
+ public void description(String description) {
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("HugeGroup(%s)", this.id);
+ }
+
+ @Override
+ protected boolean property(String key, Object value) {
+ if (super.property(key, value)) {
+ return true;
+ }
+ switch (key) {
+ case P.GRAPHSPACE:
+ this.graphSpace = (String) value;
+ break;
+ case P.NAME:
+ this.name = (String) value;
+ break;
+ case P.NICKNAME:
+ this.nickname = (String) value;
+ break;
+ case P.DESCRIPTION:
+ this.description = (String) value;
+ break;
+ default:
+ throw new AssertionError("Unsupported key: " + key);
+ }
+ return true;
+ }
+
+ @Override
+ protected Object[] asArray() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ List<Object> list = new ArrayList<>(12);
+
+ list.add(T.label);
+ list.add(P.ROLE);
+
+ list.add(P.GRAPHSPACE);
+ list.add(this.graphSpace);
+
+ list.add(P.NAME);
+ list.add(this.name);
+
+ if (this.nickname != null) {
+ list.add(P.NICKNAME);
+ list.add(this.nickname);
+ }
+
+ if (this.description != null) {
+ list.add(P.DESCRIPTION);
+ list.add(this.description);
+ }
+
+ return super.asArray(list);
+ }
+
+ @Override
+ public Map<String, Object> asMap() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ Map<String, Object> map = new HashMap<>();
+
+ map.put(Hidden.unHide(P.NAME), this.name);
+ map.put(Hidden.unHide(P.GRAPHSPACE), this.graphSpace);
+ if (this.description != null) {
+ map.put(Hidden.unHide(P.DESCRIPTION), this.description);
+ }
+
+ if (this.nickname != null) {
+ map.put(Hidden.unHide(P.NICKNAME), this.nickname);
+ }
+
+ return super.asMap(map);
+ }
+
+ public static final class P {
+
+ public static final String ROLE = Hidden.hide("role");
+
+ public static final String ID = T.id.getAccessor();
+ public static final String LABEL = T.label.getAccessor();
+
+ public static final String NAME = "~role_name";
+ public static final String NICKNAME = "~role_nickname";
+ public static final String GRAPHSPACE = "~graphspace";
+ public static final String DESCRIPTION = "~role_description";
Review Comment:
这里咋不用hide, hide就是加～前缀吧
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/QueryIdAdapter.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.id.EdgeId;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.serializer.BinaryBackendEntry;
+
+public class QueryIdAdapter extends AbstractSerializerAdapter<Id> {
+
+ static HashMap<String, Type> cls;
+
+ static {
+ cls = new HashMap() {{
Review Comment:
同
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/PdDistributedLock.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.meta.lock;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public class PdDistributedLock extends AbstractDistributedLock {
+
+ private static int poolSize = 8;
+ private final KvClient client;
+ private ScheduledExecutorService service = new ScheduledThreadPoolExecutor(poolSize, r -> {
+ Thread t = new Thread(r);
Review Comment:
线程最好有个名字，比如这里keepalive-*
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/QueryAdapter.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.query.Condition;
+import org.apache.hugegraph.type.define.Directions;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.reflect.TypeToken;
+
+public class QueryAdapter extends AbstractSerializerAdapter<Condition> {
+
+ static HashMap<String, Type> cls;
+
+ static {
+ cls = new HashMap() {{
Review Comment:
可以用immutableMap嘛？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/PdDistributedLock.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.meta.lock;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public class PdDistributedLock extends AbstractDistributedLock {
+
+ private static int poolSize = 8;
+ private final KvClient client;
+ private ScheduledExecutorService service = new ScheduledThreadPoolExecutor(poolSize, r -> {
+ Thread t = new Thread(r);
Review Comment:
线程最好有个名字，比如这里keepalive-*
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/DistributedLock.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.meta.lock;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+
+public class DistributedLock extends AbstractDistributedLock {
+
+ protected static final Logger LOG = Log.logger(DistributedLock.class);
+ private static final long UNLIMIT_TIMEOUT = -1L;
+ private final static Object mutex = new Object();
+ private static DistributedLock lockProvider = null;
+ private final KV kvClient;
+ private final Lock lockClient;
+ private final Lease leaseClient;
+
+ private DistributedLock(Client client) {
+ this.kvClient = client.getKVClient();
+ this.lockClient = client.getLockClient();
+ this.leaseClient = client.getLeaseClient();
+ }
+
+ public static DistributedLock getInstance(Client client) {
+ synchronized (mutex) {
+ if (null == lockProvider) {
+ lockProvider = new DistributedLock(client);
+ }
+ }
+ return lockProvider;
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content, Charset.defaultCharset());
+ }
+
+ public LockResult tryLock(String lockName, long ttl, long timeout) {
+ LockResult lockResult = new LockResult();
+ ScheduledExecutorService service =
+ Executors.newSingleThreadScheduledExecutor();
+
+ lockResult.lockSuccess(false);
+ lockResult.setService(service);
+
+ Long leaseId;
+
+ try {
+ leaseId = this.leaseClient.grant(ttl).get().getID();
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn(String.format("Thread {} failed to create lease for {} " +
+ "with ttl {}", Thread.currentThread().getName(),
+ lockName, ttl),
+ e);
+ return lockResult;
+ }
+
+ lockResult.setLeaseId(leaseId);
+
+ long period = ttl - ttl / 5;
+ service.scheduleAtFixedRate(new KeepAliveTask(this.leaseClient, leaseId),
+ period, period, TimeUnit.SECONDS);
+
+ try {
+ if (timeout == UNLIMIT_TIMEOUT) {
+ this.lockClient.lock(toByteSequence(lockName), leaseId).get();
+
+ } else {
+ this.lockClient.lock(toByteSequence(lockName), leaseId)
+ .get(1, TimeUnit.SECONDS);
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn(String.format("Thread {} failed to lock {}",
+ Thread.currentThread().getName(), lockName),
+ e);
+ service.shutdown();
+ this.revokeLease(leaseId);
+ return lockResult;
+ } catch (TimeoutException e) {
+ // 获取锁超时
+ LOG.warn("Thread {} timeout to lock {}",
+ Thread.currentThread().getName(), lockName);
+ service.shutdown();
+ this.revokeLease(leaseId);
+ return lockResult;
+ }
+
+ lockResult.lockSuccess(true);
+
+ return lockResult;
+ }
+
+ @Override
+ public LockResult lock(String lockName, long ttl) {
+ return tryLock(lockName, ttl, UNLIMIT_TIMEOUT);
+ }
+
+ @Override
+ public void unLock(String lockName, LockResult lockResult) {
+ LOG.debug("Thread {} start to unlock {}",
+ Thread.currentThread().getName(), lockName);
+
+ lockResult.getService().shutdown();
+
+ if (lockResult.getLeaseId() != 0L) {
+ this.revokeLease(lockResult.getLeaseId());
+ }
+
+ LOG.debug("Thread {} unlock {} successfully",
+ Thread.currentThread().getName(), lockName);
+ }
+
+ private void revokeLease(long leaseId) {
+ try {
+ this.leaseClient.revoke(leaseId).get();
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn(String.format("Thread %s failed to revoke release %s",
+ Thread.currentThread().getName(), leaseId), e);
+ }
+ }
+
+ public static class KeepAliveTask implements Runnable {
+
+ private final Lease leaseClient;
+ private final long leaseId;
+
+ KeepAliveTask(Lease leaseClient, long leaseId) {
+ this.leaseClient = leaseClient;
+ this.leaseId = leaseId;
+ }
+
+ @Override
+ public void run() {
+ this.leaseClient.keepAliveOnce(this.leaseId);
+ }
Review Comment:
这里可以统计下调用的时间间隔
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java:
##########
@@ -453,9 +458,25 @@ private void clearVertexCache() {
}
}
- private SchemaTransaction openSchemaTransaction() throws HugeException {
+ private boolean isHstore() {
+ return this.storeProvider.isHstore();
+ }
+
+ private void initMetaManager() {
+ this.metaManager.connect("hg", MetaManager.MetaDriverType.PD,
Review Comment:
这个链接是不是每个新的链接都要调用一次？是否只执行一次就可以呢？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/DistributedLock.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.meta.lock;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+
+public class DistributedLock extends AbstractDistributedLock {
+
+ protected static final Logger LOG = Log.logger(DistributedLock.class);
+ private static final long UNLIMIT_TIMEOUT = -1L;
+ private final static Object mutex = new Object();
+ private static DistributedLock lockProvider = null;
+ private final KV kvClient;
+ private final Lock lockClient;
+ private final Lease leaseClient;
+
+ private DistributedLock(Client client) {
+ this.kvClient = client.getKVClient();
+ this.lockClient = client.getLockClient();
+ this.leaseClient = client.getLeaseClient();
+ }
+
+ public static DistributedLock getInstance(Client client) {
+ synchronized (mutex) {
+ if (null == lockProvider) {
+ lockProvider = new DistributedLock(client);
+ }
+ }
+ return lockProvider;
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content, Charset.defaultCharset());
+ }
+
+ public LockResult tryLock(String lockName, long ttl, long timeout) {
+ LockResult lockResult = new LockResult();
+ ScheduledExecutorService service =
+ Executors.newSingleThreadScheduledExecutor();
+
+ lockResult.lockSuccess(false);
+ lockResult.setService(service);
+
+ Long leaseId;
+
+ try {
+ leaseId = this.leaseClient.grant(ttl).get().getID();
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn(String.format("Thread {} failed to create lease for {} " +
+ "with ttl {}", Thread.currentThread().getName(),
+ lockName, ttl),
+ e);
+ return lockResult;
+ }
+
+ lockResult.setLeaseId(leaseId);
+
+ long period = ttl - ttl / 5;
+ service.scheduleAtFixedRate(new KeepAliveTask(this.leaseClient, leaseId),
+ period, period, TimeUnit.SECONDS);
+
+ try {
+ if (timeout == UNLIMIT_TIMEOUT) {
+ this.lockClient.lock(toByteSequence(lockName), leaseId).get();
+
+ } else {
+ this.lockClient.lock(toByteSequence(lockName), leaseId)
+ .get(1, TimeUnit.SECONDS);
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn(String.format("Thread {} failed to lock {}",
+ Thread.currentThread().getName(), lockName),
+ e);
+ service.shutdown();
+ this.revokeLease(leaseId);
+ return lockResult;
+ } catch (TimeoutException e) {
+ // 获取锁超时
+ LOG.warn("Thread {} timeout to lock {}",
+ Thread.currentThread().getName(), lockName);
+ service.shutdown();
+ this.revokeLease(leaseId);
+ return lockResult;
+ }
+
+ lockResult.lockSuccess(true);
+
+ return lockResult;
+ }
+
+ @Override
+ public LockResult lock(String lockName, long ttl) {
+ return tryLock(lockName, ttl, UNLIMIT_TIMEOUT);
+ }
+
+ @Override
+ public void unLock(String lockName, LockResult lockResult) {
+ LOG.debug("Thread {} start to unlock {}",
+ Thread.currentThread().getName(), lockName);
+
+ lockResult.getService().shutdown();
+
+ if (lockResult.getLeaseId() != 0L) {
+ this.revokeLease(lockResult.getLeaseId());
+ }
+
+ LOG.debug("Thread {} unlock {} successfully",
+ Thread.currentThread().getName(), lockName);
+ }
+
+ private void revokeLease(long leaseId) {
+ try {
+ this.leaseClient.revoke(leaseId).get();
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.warn(String.format("Thread %s failed to revoke release %s",
+ Thread.currentThread().getName(), leaseId), e);
+ }
+ }
+
+ public static class KeepAliveTask implements Runnable {
+
+ private final Lease leaseClient;
+ private final long leaseId;
+
+ KeepAliveTask(Lease leaseClient, long leaseId) {
+ this.leaseClient = leaseClient;
+ this.leaseId = leaseId;
+ }
+
+ @Override
+ public void run() {
+ this.leaseClient.keepAliveOnce(this.leaseId);
+ }
Review Comment:
这里可以统计下调用的时间间隔
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/auth/SchemaDefine.java:
##########
@@ -281,6 +290,19 @@ public abstract static class Relationship extends AuthElement {
public abstract Id target();
+ public void setId() {
+ this.id(IdGenerator.of(this.source().asString() + "->" +
+ this.target().asString()));
+ }
+
+ public static <T extends Relationship> T fromMap(Map<String, Object> map, T entity) {
Review Comment:
fromMap名字有点奇怪，要不要改一下？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/QueryAdapter.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.query.Condition;
+import org.apache.hugegraph.type.define.Directions;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.reflect.TypeToken;
+
+public class QueryAdapter extends AbstractSerializerAdapter<Condition> {
+
+ static HashMap<String, Type> cls;
+
+ static {
+ cls = new HashMap() {{
Review Comment:
可以用immutableMap嘛？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/QueryIdAdapter.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.id.EdgeId;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.serializer.BinaryBackendEntry;
+
+public class QueryIdAdapter extends AbstractSerializerAdapter<Id> {
+
+ static HashMap<String, Type> cls;
+
+ static {
+ cls = new HashMap() {{
Review Comment:
同
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
+ SimpleDateFormat ft = new SimpleDateFormat("HHmmss");
+ System.out.println("当前时间为: " + ft.format(dNow));
+ }
+
+ public PDClient pdClient() {
+ return this.pdClient;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ try {
+ this.client.put(key, value);
+ } catch (PDException e) {
+ throw new HugeException("Failed to put '%s:%s' to pd", e, key, value);
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ try {
+ KResponse response = this.client.get(key);
+ return response.getValue();
+ } catch (PDException e) {
+ throw new HugeException("Failed to get '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ try {
+ this.client.delete(key);
+ } catch (PDException e) {
+ throw new HugeException("Failed to delete '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ try {
+ this.client.deletePrefix(prefix);
+ } catch (PDException e) {
+ throw new HugeException("Failed to deleteWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ try {
+ ScanPrefixResponse response = this.client.scanPrefix(prefix);
+ return response.getKvsMap();
+ } catch (PDException e) {
+ throw new HugeException("Failed to scanWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> void listen(String key, Consumer<T> consumer) {
+ try {
+ this.client.listen(key, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listen '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
+ try {
+ this.client.listenPrefix(prefix, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listenPrefix '%s' to pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ return null;
+ }
+ String value = event.getCurrent().getValue();
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ Map<String, String> resultMap = new HashMap<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not etcd PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ continue;
+ }
+
+ String key = event.getCurrent().getKey();
+ String value = event.getCurrent().getValue();
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ while (true) {
Review Comment:
没有次数限制嘛？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/DistributedLock.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.meta.lock;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+
+public class DistributedLock extends AbstractDistributedLock {
+
+ protected static final Logger LOG = Log.logger(DistributedLock.class);
+ private static final long UNLIMIT_TIMEOUT = -1L;
+ private final static Object mutex = new Object();
+ private static DistributedLock lockProvider = null;
+ private final KV kvClient;
+ private final Lock lockClient;
+ private final Lease leaseClient;
+
+ private DistributedLock(Client client) {
+ this.kvClient = client.getKVClient();
+ this.lockClient = client.getLockClient();
+ this.leaseClient = client.getLeaseClient();
+ }
+
+ public static DistributedLock getInstance(Client client) {
+ synchronized (mutex) {
+ if (null == lockProvider) {
+ lockProvider = new DistributedLock(client);
+ }
+ }
+ return lockProvider;
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content, Charset.defaultCharset());
+ }
+
+ public LockResult tryLock(String lockName, long ttl, long timeout) {
+ LockResult lockResult = new LockResult();
+ ScheduledExecutorService service =
+ Executors.newSingleThreadScheduledExecutor();
Review Comment:
这个是不是可以放全局队列来使用，没看错的话，这里的意思是没次lock就开一个线程来keep alive
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320595217
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/auth/HugeRole.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.auth;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.auth.SchemaDefine.Entity;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.util.E;
+import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
+import org.apache.tinkerpop.gremlin.structure.T;
+
+public class HugeRole extends Entity {
+
+ private static final long serialVersionUID = 2330399818352242686L;
+
+ private String name;
+ private String nickname;
+ private String graphSpace;
+ private String description;
+
+ public HugeRole(Id id, String name, String graphSpace) {
+ this.id = id;
+ this.name = name;
+ this.graphSpace = graphSpace;
+ this.description = null;
+ }
+
+ public HugeRole(String name, String graphSpace) {
+ this(StringUtils.isNotEmpty(name) ? IdGenerator.of(name) : null,
+ name, graphSpace);
+ }
+
+ public HugeRole(Id id, String graphSpace) {
+ this(id, id.asString(), graphSpace);
+ }
+
+ public static HugeRole fromMap(Map<String, Object> map) {
+ HugeRole role = new HugeRole("", "");
+ return fromMap(map, role);
+ }
+
+ public static Schema schema(HugeGraphParams graph) {
+ return new Schema(graph);
+ }
+
+ @Override
+ public ResourceType type() {
+ return ResourceType.GRANT;
+ }
+
+ @Override
+ public String label() {
+ return P.ROLE;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ public void name(String name) {
+ this.name = name;
+ }
+
+ public String nickname() {
+ return this.nickname;
+ }
+
+ public void nickname(String nickname) {
+ this.nickname = nickname;
+ }
+
+ public String graphSpace() {
+ return this.graphSpace;
+ }
+
+ public String description() {
+ return this.description;
+ }
+
+ public void description(String description) {
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("HugeGroup(%s)", this.id);
+ }
+
+ @Override
+ protected boolean property(String key, Object value) {
+ if (super.property(key, value)) {
+ return true;
+ }
+ switch (key) {
+ case P.GRAPHSPACE:
+ this.graphSpace = (String) value;
+ break;
+ case P.NAME:
+ this.name = (String) value;
+ break;
+ case P.NICKNAME:
+ this.nickname = (String) value;
+ break;
+ case P.DESCRIPTION:
+ this.description = (String) value;
+ break;
+ default:
+ throw new AssertionError("Unsupported key: " + key);
+ }
+ return true;
+ }
+
+ @Override
+ protected Object[] asArray() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ List<Object> list = new ArrayList<>(12);
+
+ list.add(T.label);
+ list.add(P.ROLE);
+
+ list.add(P.GRAPHSPACE);
+ list.add(this.graphSpace);
+
+ list.add(P.NAME);
+ list.add(this.name);
+
+ if (this.nickname != null) {
+ list.add(P.NICKNAME);
+ list.add(this.nickname);
+ }
+
+ if (this.description != null) {
+ list.add(P.DESCRIPTION);
+ list.add(this.description);
+ }
+
+ return super.asArray(list);
+ }
+
+ @Override
+ public Map<String, Object> asMap() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ Map<String, Object> map = new HashMap<>();
+
+ map.put(Hidden.unHide(P.NAME), this.name);
+ map.put(Hidden.unHide(P.GRAPHSPACE), this.graphSpace);
+ if (this.description != null) {
+ map.put(Hidden.unHide(P.DESCRIPTION), this.description);
+ }
+
+ if (this.nickname != null) {
+ map.put(Hidden.unHide(P.NICKNAME), this.nickname);
+ }
+
+ return super.asMap(map);
+ }
+
+ public static final class P {
+
+ public static final String ROLE = Hidden.hide("role");
+
+ public static final String ID = T.id.getAccessor();
+ public static final String LABEL = T.label.getAccessor();
+
+ public static final String NAME = "~role_name";
+ public static final String NICKNAME = "~role_nickname";
+ public static final String GRAPHSPACE = "~graphspace";
+ public static final String DESCRIPTION = "~role_description";
Review Comment:
已有的代码如 `HugeProject`，对于 property key 都是这种写法 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1325431135
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/AbstractSerializerAdapter.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.BackendException;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.JsonSerializer;
+
+public abstract class AbstractSerializerAdapter<T> implements JsonSerializer<T>,
Review Comment:
已添加注释
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/AbstractSerializerAdapter.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.BackendException;
+
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.JsonSerializer;
+
+public abstract class AbstractSerializerAdapter<T> implements JsonSerializer<T>,
Review Comment:
已添加注释
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320699183
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
+ SimpleDateFormat ft = new SimpleDateFormat("HHmmss");
+ System.out.println("当前时间为: " + ft.format(dNow));
+ }
+
+ public PDClient pdClient() {
+ return this.pdClient;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ try {
+ this.client.put(key, value);
+ } catch (PDException e) {
+ throw new HugeException("Failed to put '%s:%s' to pd", e, key, value);
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ try {
+ KResponse response = this.client.get(key);
+ return response.getValue();
+ } catch (PDException e) {
+ throw new HugeException("Failed to get '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ try {
+ this.client.delete(key);
+ } catch (PDException e) {
+ throw new HugeException("Failed to delete '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ try {
+ this.client.deletePrefix(prefix);
+ } catch (PDException e) {
+ throw new HugeException("Failed to deleteWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ try {
+ ScanPrefixResponse response = this.client.scanPrefix(prefix);
+ return response.getKvsMap();
+ } catch (PDException e) {
+ throw new HugeException("Failed to scanWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> void listen(String key, Consumer<T> consumer) {
+ try {
+ this.client.listen(key, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listen '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
+ try {
+ this.client.listenPrefix(prefix, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listenPrefix '%s' to pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ return null;
+ }
+ String value = event.getCurrent().getValue();
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ Map<String, String> resultMap = new HashMap<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not etcd PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ continue;
+ }
+
+ String key = event.getCurrent().getKey();
+ String value = event.getCurrent().getValue();
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ while (true) {
Review Comment:
这个方法注解为 Deprecated 且未被引用，直接删除？
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
+ SimpleDateFormat ft = new SimpleDateFormat("HHmmss");
+ System.out.println("当前时间为: " + ft.format(dNow));
+ }
+
+ public PDClient pdClient() {
+ return this.pdClient;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ try {
+ this.client.put(key, value);
+ } catch (PDException e) {
+ throw new HugeException("Failed to put '%s:%s' to pd", e, key, value);
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ try {
+ KResponse response = this.client.get(key);
+ return response.getValue();
+ } catch (PDException e) {
+ throw new HugeException("Failed to get '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ try {
+ this.client.delete(key);
+ } catch (PDException e) {
+ throw new HugeException("Failed to delete '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ try {
+ this.client.deletePrefix(prefix);
+ } catch (PDException e) {
+ throw new HugeException("Failed to deleteWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ try {
+ ScanPrefixResponse response = this.client.scanPrefix(prefix);
+ return response.getKvsMap();
+ } catch (PDException e) {
+ throw new HugeException("Failed to scanWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> void listen(String key, Consumer<T> consumer) {
+ try {
+ this.client.listen(key, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listen '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
+ try {
+ this.client.listenPrefix(prefix, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listenPrefix '%s' to pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ return null;
+ }
+ String value = event.getCurrent().getValue();
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ Map<String, String> resultMap = new HashMap<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not etcd PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ continue;
+ }
+
+ String key = event.getCurrent().getKey();
+ String value = event.getCurrent().getValue();
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ while (true) {
Review Comment:
这个方法注解为 Deprecated 且未被引用，直接删除？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1327891256
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/tx/IdCounter.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.tx;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hugegraph.backend.BackendException;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.Pdpb;
+import org.apache.hugegraph.store.term.HgPair;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.util.E;
+
+public class IdCounter {
+
+ private static final int TIMES = 10000;
+ private static final int DELTA = 10000;
+ private static final String DELIMITER = "/";
+ private static Map<String, HgPair> ids = new ConcurrentHashMap<>();
Review Comment:
should be `final`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320595217
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/auth/HugeRole.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.auth;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.auth.SchemaDefine.Entity;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.util.E;
+import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
+import org.apache.tinkerpop.gremlin.structure.T;
+
+public class HugeRole extends Entity {
+
+ private static final long serialVersionUID = 2330399818352242686L;
+
+ private String name;
+ private String nickname;
+ private String graphSpace;
+ private String description;
+
+ public HugeRole(Id id, String name, String graphSpace) {
+ this.id = id;
+ this.name = name;
+ this.graphSpace = graphSpace;
+ this.description = null;
+ }
+
+ public HugeRole(String name, String graphSpace) {
+ this(StringUtils.isNotEmpty(name) ? IdGenerator.of(name) : null,
+ name, graphSpace);
+ }
+
+ public HugeRole(Id id, String graphSpace) {
+ this(id, id.asString(), graphSpace);
+ }
+
+ public static HugeRole fromMap(Map<String, Object> map) {
+ HugeRole role = new HugeRole("", "");
+ return fromMap(map, role);
+ }
+
+ public static Schema schema(HugeGraphParams graph) {
+ return new Schema(graph);
+ }
+
+ @Override
+ public ResourceType type() {
+ return ResourceType.GRANT;
+ }
+
+ @Override
+ public String label() {
+ return P.ROLE;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ public void name(String name) {
+ this.name = name;
+ }
+
+ public String nickname() {
+ return this.nickname;
+ }
+
+ public void nickname(String nickname) {
+ this.nickname = nickname;
+ }
+
+ public String graphSpace() {
+ return this.graphSpace;
+ }
+
+ public String description() {
+ return this.description;
+ }
+
+ public void description(String description) {
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("HugeGroup(%s)", this.id);
+ }
+
+ @Override
+ protected boolean property(String key, Object value) {
+ if (super.property(key, value)) {
+ return true;
+ }
+ switch (key) {
+ case P.GRAPHSPACE:
+ this.graphSpace = (String) value;
+ break;
+ case P.NAME:
+ this.name = (String) value;
+ break;
+ case P.NICKNAME:
+ this.nickname = (String) value;
+ break;
+ case P.DESCRIPTION:
+ this.description = (String) value;
+ break;
+ default:
+ throw new AssertionError("Unsupported key: " + key);
+ }
+ return true;
+ }
+
+ @Override
+ protected Object[] asArray() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ List<Object> list = new ArrayList<>(12);
+
+ list.add(T.label);
+ list.add(P.ROLE);
+
+ list.add(P.GRAPHSPACE);
+ list.add(this.graphSpace);
+
+ list.add(P.NAME);
+ list.add(this.name);
+
+ if (this.nickname != null) {
+ list.add(P.NICKNAME);
+ list.add(this.nickname);
+ }
+
+ if (this.description != null) {
+ list.add(P.DESCRIPTION);
+ list.add(this.description);
+ }
+
+ return super.asArray(list);
+ }
+
+ @Override
+ public Map<String, Object> asMap() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ Map<String, Object> map = new HashMap<>();
+
+ map.put(Hidden.unHide(P.NAME), this.name);
+ map.put(Hidden.unHide(P.GRAPHSPACE), this.graphSpace);
+ if (this.description != null) {
+ map.put(Hidden.unHide(P.DESCRIPTION), this.description);
+ }
+
+ if (this.nickname != null) {
+ map.put(Hidden.unHide(P.NICKNAME), this.nickname);
+ }
+
+ return super.asMap(map);
+ }
+
+ public static final class P {
+
+ public static final String ROLE = Hidden.hide("role");
+
+ public static final String ID = T.id.getAccessor();
+ public static final String LABEL = T.label.getAccessor();
+
+ public static final String NAME = "~role_name";
+ public static final String NICKNAME = "~role_nickname";
+ public static final String GRAPHSPACE = "~graphspace";
+ public static final String DESCRIPTION = "~role_description";
Review Comment:
已有的代码如 `HugeProject`，对于 property key 都是这样写的 🤔
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/auth/HugeRole.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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.auth;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.auth.SchemaDefine.Entity;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.util.E;
+import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
+import org.apache.tinkerpop.gremlin.structure.T;
+
+public class HugeRole extends Entity {
+
+ private static final long serialVersionUID = 2330399818352242686L;
+
+ private String name;
+ private String nickname;
+ private String graphSpace;
+ private String description;
+
+ public HugeRole(Id id, String name, String graphSpace) {
+ this.id = id;
+ this.name = name;
+ this.graphSpace = graphSpace;
+ this.description = null;
+ }
+
+ public HugeRole(String name, String graphSpace) {
+ this(StringUtils.isNotEmpty(name) ? IdGenerator.of(name) : null,
+ name, graphSpace);
+ }
+
+ public HugeRole(Id id, String graphSpace) {
+ this(id, id.asString(), graphSpace);
+ }
+
+ public static HugeRole fromMap(Map<String, Object> map) {
+ HugeRole role = new HugeRole("", "");
+ return fromMap(map, role);
+ }
+
+ public static Schema schema(HugeGraphParams graph) {
+ return new Schema(graph);
+ }
+
+ @Override
+ public ResourceType type() {
+ return ResourceType.GRANT;
+ }
+
+ @Override
+ public String label() {
+ return P.ROLE;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ public void name(String name) {
+ this.name = name;
+ }
+
+ public String nickname() {
+ return this.nickname;
+ }
+
+ public void nickname(String nickname) {
+ this.nickname = nickname;
+ }
+
+ public String graphSpace() {
+ return this.graphSpace;
+ }
+
+ public String description() {
+ return this.description;
+ }
+
+ public void description(String description) {
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("HugeGroup(%s)", this.id);
+ }
+
+ @Override
+ protected boolean property(String key, Object value) {
+ if (super.property(key, value)) {
+ return true;
+ }
+ switch (key) {
+ case P.GRAPHSPACE:
+ this.graphSpace = (String) value;
+ break;
+ case P.NAME:
+ this.name = (String) value;
+ break;
+ case P.NICKNAME:
+ this.nickname = (String) value;
+ break;
+ case P.DESCRIPTION:
+ this.description = (String) value;
+ break;
+ default:
+ throw new AssertionError("Unsupported key: " + key);
+ }
+ return true;
+ }
+
+ @Override
+ protected Object[] asArray() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ List<Object> list = new ArrayList<>(12);
+
+ list.add(T.label);
+ list.add(P.ROLE);
+
+ list.add(P.GRAPHSPACE);
+ list.add(this.graphSpace);
+
+ list.add(P.NAME);
+ list.add(this.name);
+
+ if (this.nickname != null) {
+ list.add(P.NICKNAME);
+ list.add(this.nickname);
+ }
+
+ if (this.description != null) {
+ list.add(P.DESCRIPTION);
+ list.add(this.description);
+ }
+
+ return super.asArray(list);
+ }
+
+ @Override
+ public Map<String, Object> asMap() {
+ E.checkState(this.name != null, "Group name can't be null");
+
+ Map<String, Object> map = new HashMap<>();
+
+ map.put(Hidden.unHide(P.NAME), this.name);
+ map.put(Hidden.unHide(P.GRAPHSPACE), this.graphSpace);
+ if (this.description != null) {
+ map.put(Hidden.unHide(P.DESCRIPTION), this.description);
+ }
+
+ if (this.nickname != null) {
+ map.put(Hidden.unHide(P.NICKNAME), this.nickname);
+ }
+
+ return super.asMap(map);
+ }
+
+ public static final class P {
+
+ public static final String ROLE = Hidden.hide("role");
+
+ public static final String ID = T.id.getAccessor();
+ public static final String LABEL = T.label.getAccessor();
+
+ public static final String NAME = "~role_name";
+ public static final String NICKNAME = "~role_nickname";
+ public static final String GRAPHSPACE = "~graphspace";
+ public static final String DESCRIPTION = "~role_description";
Review Comment:
已有的代码如 `HugeProject`，对于 property key 都是这样写的 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] github-code-scanning[bot] commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "github-code-scanning[bot] (via GitHub)" <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320780559
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/cache/CachedSchemaTransactionV2.java:
##########
@@ -0,0 +1,469 @@
+package org.apache.hugegraph.backend.cache;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.store.ram.IntObjectMap;
+import org.apache.hugegraph.backend.tx.SchemaTransactionV2;
+import org.apache.hugegraph.config.CoreOptions;
+import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventListener;
+import org.apache.hugegraph.meta.MetaDriver;
+import org.apache.hugegraph.meta.MetaManager;
+import org.apache.hugegraph.perf.PerfUtil;
+import org.apache.hugegraph.schema.SchemaElement;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Events;
+
+import com.google.common.collect.ImmutableSet;
+
+public class CachedSchemaTransactionV2 extends SchemaTransactionV2 {
+ private final Cache<Id, Object> idCache;
+ private final Cache<Id, Object> nameCache;
+
+ private final SchemaCaches<SchemaElement> arrayCaches;
+
+ private EventListener storeEventListener;
+ private EventListener cacheEventListener;
+
+ public CachedSchemaTransactionV2(MetaDriver metaDriver,
+ String cluster,
+ HugeGraphParams graphParams) {
+ super(metaDriver, cluster, graphParams);
+
+ final long capacity = graphParams.configuration()
+ .get(CoreOptions.SCHEMA_CACHE_CAPACITY);
+ this.idCache = this.cache("schema-id", capacity);
+ this.nameCache = this.cache("schema-name", capacity);
+
+ SchemaCaches<SchemaElement> attachment = this.idCache.attachment();
+ if (attachment == null) {
+ int acSize = (int) (capacity >> 3);
+ attachment = this.idCache.attachment(new SchemaCaches<>(acSize));
+ }
+ this.arrayCaches = attachment;
+ this.listenChanges();
+ }
+
+ private static Id generateId(HugeType type, Id id) {
+ // NOTE: it's slower performance to use:
+ // String.format("%x-%s", type.code(), name)
+ return IdGenerator.of(type.string() + "-" + id.asString());
+ }
+
+ private static Id generateId(HugeType type, String name) {
+ return IdGenerator.of(type.string() + "-" + name);
+ }
+
+ public void close() {
+ this.clearCache(false);
+ this.unlistenChanges();
+ }
+
+ private Cache<Id, Object> cache(String prefix, long capacity) {
+ // TODO: uncomment later - graph space
+ //final String name = prefix + "-" + this.graph().spaceGraphName();
+ final String name = prefix + "-" + "";
+ // NOTE: must disable schema cache-expire due to getAllSchema()
+ return CacheManager.instance().cache(name, capacity);
+ }
+
+ private void listenChanges() {
+ // Listen store event: "store.init", "store.clear", ...
+ Set<String> storeEvents = ImmutableSet.of(Events.STORE_INIT,
+ Events.STORE_CLEAR,
+ Events.STORE_TRUNCATE);
+ this.storeEventListener = event -> {
+ if (storeEvents.contains(event.name())) {
+ LOG.debug("Graph {} clear schema cache on event '{}'",
+ this.graph(), event.name());
+ this.clearCache(true);
+ return true;
+ }
+ return false;
+ };
+ this.graphParams().loadGraphStore().provider().listen(this.storeEventListener);
+
+ // Listen cache event: "cache"(invalid cache item)
+ this.cacheEventListener = event -> {
+ LOG.debug("Graph {} received schema cache event: {}",
+ this.graph(), event);
+ Object[] args = event.args();
+ E.checkArgument(args.length > 0 && args[0] instanceof String,
+ "Expect event action argument");
+ if (Cache.ACTION_INVALID.equals(args[0])) {
+ event.checkArgs(String.class, HugeType.class, Id.class);
+ HugeType type = (HugeType) args[1];
+ Id id = (Id) args[2];
+ this.arrayCaches.remove(type, id);
+
+ id = generateId(type, id);
+ Object value = this.idCache.get(id);
+ if (value != null) {
+ // Invalidate id cache
+ this.idCache.invalidate(id);
+
+ // Invalidate name cache
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(),
+ schema.name());
+ this.nameCache.invalidate(prefixedName);
+ }
+ this.resetCachedAll(type);
+ return true;
+ } else if (Cache.ACTION_CLEAR.equals(args[0])) {
+ event.checkArgs(String.class, HugeType.class);
+ this.clearCache(false);
+ return true;
+ }
+ return false;
+ };
+ EventHub schemaEventHub = this.graphParams().schemaEventHub();
+ if (!schemaEventHub.containsListener(Events.CACHE)) {
+ schemaEventHub.listen(Events.CACHE, this.cacheEventListener);
+ }
+ }
+
+ public void clearCache(boolean notify) {
+ this.idCache.clear();
+ this.nameCache.clear();
+ this.arrayCaches.clear();
+ }
+
+ private void resetCachedAllIfReachedCapacity() {
+ if (this.idCache.size() >= this.idCache.capacity()) {
+ LOG.warn("Schema cache reached capacity({}): {}",
+ this.idCache.capacity(), this.idCache.size());
+ this.cachedTypes().clear();
+ }
+ }
+
+ private void unlistenChanges() {
+ // Unlisten store event
+ this.graphParams().loadGraphStore().provider()
+ .unlisten(this.storeEventListener);
+
+ // Unlisten cache event
+ EventHub schemaEventHub = this.graphParams().schemaEventHub();
+ schemaEventHub.unlisten(Events.CACHE, this.cacheEventListener);
+ }
+
+ private CachedTypes cachedTypes() {
+ return this.arrayCaches.cachedTypes();
+ }
+
+ private void resetCachedAll(HugeType type) {
+ // Set the cache all flag of the schema type to false
+ this.cachedTypes().put(type, false);
+ }
+
+ private void invalidateCache(HugeType type, Id id) {
+ // remove from id cache and name cache
+ Id prefixedId = generateId(type, id);
+ Object value = this.idCache.get(prefixedId);
+ if (value != null) {
+ this.idCache.invalidate(prefixedId);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.invalidate(prefixedName);
+ }
+
+ // remove from optimized array cache
+ this.arrayCaches.remove(type, id);
+ }
+
+ @Override
+ protected void updateSchema(SchemaElement schema,
+ Consumer<SchemaElement> updateCallback) {
+ super.updateSchema(schema, updateCallback);
+
+ this.updateCache(schema);
+ }
+
+ @Override
+ protected void addSchema(SchemaElement schema) {
+ super.addSchema(schema);
+
+ this.updateCache(schema);
+
+ if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
+ MetaManager.instance()
+ // TODO: uncomment later - graph space
+ //.notifySchemaCacheClear(this.graph().graphSpace(),
+ // this.graph().name());
+ .notifySchemaCacheClear("",
+ this.graph().name());
+ }
+ }
+
+ private void updateCache(SchemaElement schema) {
+ this.resetCachedAllIfReachedCapacity();
+
+ // update id cache
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+
+ // update name cache
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+
+ // update optimized array cache
+ this.arrayCaches.updateIfNeeded(schema);
+ }
+
+ @Override
+ protected void removeSchema(SchemaElement schema) {
+ super.removeSchema(schema);
+
+ this.invalidateCache(schema.type(), schema.id());
+
+ if (!this.graph().option(CoreOptions.TASK_SYNC_DELETION)) {
+ MetaManager.instance()
+ // TODO: uncomment later - graph space
+ //.notifySchemaCacheClear(this.graph().graphSpace(),
+ // this.graph().name());
+ .notifySchemaCacheClear("",
+ this.graph().name());
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T extends SchemaElement> T getSchema(HugeType type, Id id) {
+ // try get from optimized array cache
+ if (id.number() && id.asLong() > 0L) {
+ SchemaElement value = this.arrayCaches.get(type, id);
+ if (value != null) {
+ return (T) value;
+ }
+ }
+
+ Id prefixedId = generateId(type, id);
+ Object value = this.idCache.get(prefixedId);
+ if (value == null) {
+ value = super.getSchema(type, id);
+ if (value != null) {
+ this.resetCachedAllIfReachedCapacity();
+
+ this.idCache.update(prefixedId, value);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+ }
+ }
+
+ // update optimized array cache
+ this.arrayCaches.updateIfNeeded((SchemaElement) value);
+
+ return (T) value;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T extends SchemaElement> T getSchema(HugeType type,
+ String name) {
+ Id prefixedName = generateId(type, name);
+ Object value = this.nameCache.get(prefixedName);
+ if (value == null) {
+ value = super.getSchema(type, name);
+ if (value != null) {
+ this.resetCachedAllIfReachedCapacity();
+
+ this.nameCache.update(prefixedName, value);
+
+ SchemaElement schema = (SchemaElement) value;
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+ }
+ }
+ return (T) value;
+ }
+
+ @Override
+ protected <T extends SchemaElement> List<T> getAllSchema(HugeType type) {
+ Boolean cachedAll = this.cachedTypes().getOrDefault(type, false);
+ List<T> results;
+ if (cachedAll) {
+ results = new ArrayList<>();
+ // Get from cache
+ this.idCache.traverse(value -> {
+ @SuppressWarnings("unchecked")
+ T schema = (T) value;
+ if (schema.type() == type) {
+ results.add(schema);
+ }
+ });
+ return results;
+ } else {
+ results = super.getAllSchema(type);
+ long free = this.idCache.capacity() - this.idCache.size();
+ if (results.size() <= free) {
+ // Update cache
+ for (T schema : results) {
+ Id prefixedId = generateId(schema.type(), schema.id());
+ this.idCache.update(prefixedId, schema);
+
+ Id prefixedName = generateId(schema.type(), schema.name());
+ this.nameCache.update(prefixedName, schema);
+ }
+ this.cachedTypes().putIfAbsent(type, true);
+ }
+ return results;
+ }
+ }
+
+ @Override
+ public void clear() {
+ // Clear schema info firstly
+ super.clear();
+ this.clearCache(false);
+ }
+
+ private static final class SchemaCaches<V extends SchemaElement> {
+
+ private final int size;
+
+ private final IntObjectMap<V> pks;
+ private final IntObjectMap<V> vls;
+ private final IntObjectMap<V> els;
+ private final IntObjectMap<V> ils;
+
+ private final CachedTypes cachedTypes;
+
+ public SchemaCaches(int size) {
+ // TODO: improve size of each type for optimized array cache
+ this.size = size;
+
+ this.pks = new IntObjectMap<>(size);
+ this.vls = new IntObjectMap<>(size);
+ this.els = new IntObjectMap<>(size);
+ this.ils = new IntObjectMap<>(size);
+
+ this.cachedTypes = new CachedTypes();
+ }
+
+ public void updateIfNeeded(V schema) {
+ if (schema == null) {
+ return;
+ }
+ Id id = schema.id();
+ if (id.number() && id.asLong() > 0L) {
+ this.set(schema.type(), id, schema);
+ }
+ }
+
+ @PerfUtil.Watched
+ public V get(HugeType type, Id id) {
+ assert id.number();
+ long longId = id.asLong();
+ if (longId <= 0L) {
+ assert false : id;
+ return null;
+ }
+ int key = (int) longId;
Review Comment:
## User-controlled data in numeric cast
This cast to a narrower type depends on a [user-provided value](1), potentially causing truncation.
This cast to a narrower type depends on a [user-provided value](2), potentially causing truncation.
This cast to a narrower type depends on a [user-provided value](3), potentially causing truncation.
This cast to a narrower type depends on a [user-provided value](4), potentially causing truncation.
This cast to a narrower type depends on a [user-provided value](5), potentially causing truncation.
This cast to a narrower type depends on a [user-provided value](6), potentially causing truncation.
[Show more details](https://github.com/apache/incubator-hugegraph/security/code-scanning/66)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320780092
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/AbstractDistributedLock.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.meta.lock;
+
+import org.apache.hugegraph.pd.client.KvClient;
+
+import io.etcd.jetcd.Client;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public abstract class AbstractDistributedLock {
+
+ private static volatile AbstractDistributedLock defaultLock;
+
+ public static AbstractDistributedLock getInstance(AutoCloseable client) {
Review Comment:
未被引用的抽象类，已删除
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/AbstractDistributedLock.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.meta.lock;
+
+import org.apache.hugegraph.pd.client.KvClient;
+
+import io.etcd.jetcd.Client;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public abstract class AbstractDistributedLock {
+
+ private static volatile AbstractDistributedLock defaultLock;
+
+ public static AbstractDistributedLock getInstance(AutoCloseable client) {
Review Comment:
未被引用的抽象类，已删除
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/space/SchemaTemplate.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.space;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map;
+
+import org.apache.hugegraph.util.E;
+
+import com.google.common.collect.ImmutableMap;
+
+public class SchemaTemplate {
+
+ public static SimpleDateFormat FORMATTER = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ protected Date createTime;
+ protected Date updateTime;
+ protected String creator;
+ private String name;
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320784541
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/serializer/QueryIdAdapter.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.query.serializer;
+
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hugegraph.backend.id.EdgeId;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.serializer.BinaryBackendEntry;
+
+public class QueryIdAdapter extends AbstractSerializerAdapter<Id> {
+
+ static HashMap<String, Type> cls;
+
+ static {
+ cls = new HashMap() {{
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320785819
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendStoreInfo.java:
##########
@@ -41,6 +41,9 @@ public boolean exists() {
}
public boolean checkVersion() {
+ if (this.storeProvider.isHstore()) { // skip for hstore now
Review Comment:
此处跳过是因为 hstore 不提供 system store，考虑从 hstore 的 graph store 中获取 driver version
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/store/BackendStoreInfo.java:
##########
@@ -41,6 +41,9 @@ public boolean exists() {
}
public boolean checkVersion() {
+ if (this.storeProvider.isHstore()) { // skip for hstore now
Review Comment:
此处跳过是因为 hstore 不提供 system store，考虑从 hstore 的 graph store 中获取 driver version
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1325430891
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java:
##########
@@ -453,9 +458,25 @@ private void clearVertexCache() {
}
}
- private SchemaTransaction openSchemaTransaction() throws HugeException {
+ private boolean isHstore() {
+ return this.storeProvider.isHstore();
+ }
+
+ private void initMetaManager() {
+ this.metaManager.connect("hg", MetaManager.MetaDriverType.PD,
Review Comment:
考虑在 StandardHugeGraph 构造函数中初始化 meta manager
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java:
##########
@@ -453,9 +458,25 @@ private void clearVertexCache() {
}
}
- private SchemaTransaction openSchemaTransaction() throws HugeException {
+ private boolean isHstore() {
+ return this.storeProvider.isHstore();
+ }
+
+ private void initMetaManager() {
+ this.metaManager.connect("hg", MetaManager.MetaDriverType.PD,
Review Comment:
考虑在 StandardHugeGraph 构造函数中初始化 meta manager
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] zyxxoo commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "zyxxoo (via GitHub)" <gi...@apache.org>.
zyxxoo commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1325528585
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
+ SimpleDateFormat ft = new SimpleDateFormat("HHmmss");
+ System.out.println("当前时间为: " + ft.format(dNow));
+ }
+
+ public PDClient pdClient() {
+ return this.pdClient;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ try {
+ this.client.put(key, value);
+ } catch (PDException e) {
+ throw new HugeException("Failed to put '%s:%s' to pd", e, key, value);
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ try {
+ KResponse response = this.client.get(key);
+ return response.getValue();
+ } catch (PDException e) {
+ throw new HugeException("Failed to get '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ try {
+ this.client.delete(key);
+ } catch (PDException e) {
+ throw new HugeException("Failed to delete '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ try {
+ this.client.deletePrefix(prefix);
+ } catch (PDException e) {
+ throw new HugeException("Failed to deleteWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ try {
+ ScanPrefixResponse response = this.client.scanPrefix(prefix);
+ return response.getKvsMap();
+ } catch (PDException e) {
+ throw new HugeException("Failed to scanWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> void listen(String key, Consumer<T> consumer) {
+ try {
+ this.client.listen(key, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listen '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
+ try {
+ this.client.listenPrefix(prefix, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listenPrefix '%s' to pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ return null;
+ }
+ String value = event.getCurrent().getValue();
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ Map<String, String> resultMap = new HashMap<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not etcd PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ continue;
+ }
+
+ String key = event.getCurrent().getKey();
+ String value = event.getCurrent().getValue();
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ while (true) {
Review Comment:
如果是 deprecated，就直接删除吧
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
+ SimpleDateFormat ft = new SimpleDateFormat("HHmmss");
+ System.out.println("当前时间为: " + ft.format(dNow));
+ }
+
+ public PDClient pdClient() {
+ return this.pdClient;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ try {
+ this.client.put(key, value);
+ } catch (PDException e) {
+ throw new HugeException("Failed to put '%s:%s' to pd", e, key, value);
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ try {
+ KResponse response = this.client.get(key);
+ return response.getValue();
+ } catch (PDException e) {
+ throw new HugeException("Failed to get '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ try {
+ this.client.delete(key);
+ } catch (PDException e) {
+ throw new HugeException("Failed to delete '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ try {
+ this.client.deletePrefix(prefix);
+ } catch (PDException e) {
+ throw new HugeException("Failed to deleteWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ try {
+ ScanPrefixResponse response = this.client.scanPrefix(prefix);
+ return response.getKvsMap();
+ } catch (PDException e) {
+ throw new HugeException("Failed to scanWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> void listen(String key, Consumer<T> consumer) {
+ try {
+ this.client.listen(key, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listen '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
+ try {
+ this.client.listenPrefix(prefix, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listenPrefix '%s' to pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ return null;
+ }
+ String value = event.getCurrent().getValue();
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ Map<String, String> resultMap = new HashMap<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not etcd PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ continue;
+ }
+
+ String key = event.getCurrent().getKey();
+ String value = event.getCurrent().getValue();
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ while (true) {
Review Comment:
如果是 deprecated，就直接删除吧
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320779976
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/DistributedLock.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.meta.lock;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+
+public class DistributedLock extends AbstractDistributedLock {
+
+ protected static final Logger LOG = Log.logger(DistributedLock.class);
+ private static final long UNLIMIT_TIMEOUT = -1L;
+ private final static Object mutex = new Object();
+ private static DistributedLock lockProvider = null;
+ private final KV kvClient;
+ private final Lock lockClient;
+ private final Lease leaseClient;
+
+ private DistributedLock(Client client) {
+ this.kvClient = client.getKVClient();
+ this.lockClient = client.getLockClient();
+ this.leaseClient = client.getLeaseClient();
+ }
+
+ public static DistributedLock getInstance(Client client) {
+ synchronized (mutex) {
+ if (null == lockProvider) {
+ lockProvider = new DistributedLock(client);
+ }
+ }
+ return lockProvider;
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content, Charset.defaultCharset());
+ }
+
+ public LockResult tryLock(String lockName, long ttl, long timeout) {
+ LockResult lockResult = new LockResult();
+ ScheduledExecutorService service =
+ Executors.newSingleThreadScheduledExecutor();
Review Comment:
参考 PdDistributedLock 进行了修改
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/DistributedLock.java:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.meta.lock;
+
+import java.nio.charset.Charset;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+
+public class DistributedLock extends AbstractDistributedLock {
+
+ protected static final Logger LOG = Log.logger(DistributedLock.class);
+ private static final long UNLIMIT_TIMEOUT = -1L;
+ private final static Object mutex = new Object();
+ private static DistributedLock lockProvider = null;
+ private final KV kvClient;
+ private final Lock lockClient;
+ private final Lease leaseClient;
+
+ private DistributedLock(Client client) {
+ this.kvClient = client.getKVClient();
+ this.lockClient = client.getLockClient();
+ this.leaseClient = client.getLeaseClient();
+ }
+
+ public static DistributedLock getInstance(Client client) {
+ synchronized (mutex) {
+ if (null == lockProvider) {
+ lockProvider = new DistributedLock(client);
+ }
+ }
+ return lockProvider;
+ }
+
+ private static ByteSequence toByteSequence(String content) {
+ return ByteSequence.from(content, Charset.defaultCharset());
+ }
+
+ public LockResult tryLock(String lockName, long ttl, long timeout) {
+ LockResult lockResult = new LockResult();
+ ScheduledExecutorService service =
+ Executors.newSingleThreadScheduledExecutor();
Review Comment:
参考 PdDistributedLock 进行了修改
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/lock/PdDistributedLock.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.meta.lock;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+
+/**
+ * @author zhangyingjie
+ * @date 2022/6/18
+ **/
+public class PdDistributedLock extends AbstractDistributedLock {
+
+ private static int poolSize = 8;
+ private final KvClient client;
+ private ScheduledExecutorService service = new ScheduledThreadPoolExecutor(poolSize, r -> {
+ Thread t = new Thread(r);
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org
[GitHub] [incubator-hugegraph] VGalaxies commented on a diff in pull request #2301: feat: preliminary support for hstore backend
Posted by "VGalaxies (via GitHub)" <gi...@apache.org>.
VGalaxies commented on code in PR #2301:
URL: https://github.com/apache/incubator-hugegraph/pull/2301#discussion_r1320699183
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/PdMetaDriver.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.meta;
+
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.meta.lock.LockResult;
+import org.apache.hugegraph.meta.lock.PdDistributedLock;
+import org.apache.hugegraph.pd.client.KvClient;
+import org.apache.hugegraph.pd.client.PDClient;
+import org.apache.hugegraph.pd.client.PDConfig;
+import org.apache.hugegraph.pd.common.PDException;
+import org.apache.hugegraph.pd.grpc.kv.KResponse;
+import org.apache.hugegraph.pd.grpc.kv.LockResponse;
+import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse;
+import org.apache.hugegraph.pd.grpc.kv.TTLResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchEvent;
+import org.apache.hugegraph.pd.grpc.kv.WatchResponse;
+import org.apache.hugegraph.pd.grpc.kv.WatchType;
+
+import com.google.common.base.Strings;
+
+public class PdMetaDriver implements MetaDriver {
+
+ KvClient<WatchResponse> client = null;
+ PDClient pdClient = null;
+ private PdDistributedLock lock;
+
+ public PdMetaDriver(String pdPeer) {
+ PDConfig pdConfig = PDConfig.of(pdPeer);
+ this.client = new KvClient<>(pdConfig);
+ this.pdClient = PDClient.create(pdConfig);
+ lock = new PdDistributedLock(this.client);
+ }
+
+ public static void main(String[] args) {
+ PDConfig pdConfig = PDConfig.of("127.0.0.1:8686");
+ KvClient<WatchResponse> client = new KvClient<>(pdConfig);
+ ScanPrefixResponse contents;
+ try {
+ contents = client.scanPrefix("HUGEGRAPH/METRICS");
+ Map<String, String> map = contents.getKvsMap();
+ for (Map.Entry<String, String> entry : map.entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ System.out.println(map.size());
+ } catch (PDException e) {
+ e.printStackTrace();
+ }
+ Date dNow = new Date();
+ System.out.println(dNow);
+ SimpleDateFormat ft = new SimpleDateFormat("HHmmss");
+ System.out.println("当前时间为: " + ft.format(dNow));
+ }
+
+ public PDClient pdClient() {
+ return this.pdClient;
+ }
+
+ @Override
+ public void put(String key, String value) {
+ try {
+ this.client.put(key, value);
+ } catch (PDException e) {
+ throw new HugeException("Failed to put '%s:%s' to pd", e, key, value);
+ }
+ }
+
+ @Override
+ public String get(String key) {
+ try {
+ KResponse response = this.client.get(key);
+ return response.getValue();
+ } catch (PDException e) {
+ throw new HugeException("Failed to get '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void delete(String key) {
+ try {
+ this.client.delete(key);
+ } catch (PDException e) {
+ throw new HugeException("Failed to delete '%s' from pd", e, key);
+ }
+ }
+
+ @Override
+ public void deleteWithPrefix(String prefix) {
+ try {
+ this.client.deletePrefix(prefix);
+ } catch (PDException e) {
+ throw new HugeException("Failed to deleteWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public Map<String, String> scanWithPrefix(String prefix) {
+ try {
+ ScanPrefixResponse response = this.client.scanPrefix(prefix);
+ return response.getKvsMap();
+ } catch (PDException e) {
+ throw new HugeException("Failed to scanWithPrefix '%s' from pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> void listen(String key, Consumer<T> consumer) {
+ try {
+ this.client.listen(key, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listen '%s' to pd", e, key);
+ }
+ }
+
+ @Override
+ public <T> void listenPrefix(String prefix, Consumer<T> consumer) {
+ try {
+ this.client.listenPrefix(prefix, (Consumer<WatchResponse>) consumer);
+ } catch (PDException e) {
+ throw new HugeException("Failed to listenPrefix '%s' to pd", e, prefix);
+ }
+ }
+
+ @Override
+ public <T> List<String> extractValuesFromResponse(T response) {
+ List<String> values = new ArrayList<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ return null;
+ }
+ String value = event.getCurrent().getValue();
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public <T> Map<String, String> extractKVFromResponse(T response) {
+ Map<String, String> resultMap = new HashMap<>();
+ WatchResponse res = (WatchResponse) response;
+ for (WatchEvent event : res.getEventsList()) {
+ // Skip if not etcd PUT event
+ if (!event.getType().equals(WatchType.Put)) {
+ continue;
+ }
+
+ String key = event.getCurrent().getKey();
+ String value = event.getCurrent().getValue();
+ if (Strings.isNullOrEmpty(key)) {
+ continue;
+ }
+ resultMap.put(key, value);
+ }
+ return resultMap;
+ }
+
+ @Override
+ public LockResult lock(String key, long ttl) {
+ while (true) {
Review Comment:
这个方法注解为 Deprecated 且目前未被引用，直接删除？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@hugegraph.apache.org
For additional commands, e-mail: issues-help@hugegraph.apache.org