This is an automated email from the ASF dual-hosted git repository. jermy pushed a commit to branch schema-update-bug-rebase in repository https://gitbox.apache.org/repos/asf/incubator-hugegraph.git
commit bfe93f03db36e2a62e7e281bd586ca657d39273f Author: Zhangmei Li <[email protected]> AuthorDate: Thu Feb 24 17:23:50 2022 +0800 support updateIfPresent/updateIfAbsent operation Change-Id: I63f179e0b6c2e2f86e6ae185ce3a7115a4da7657 --- .../backend/cache/CachedSchemaTransaction.java | 9 +++++++ .../hugegraph/backend/store/BackendTable.java | 29 ++++++++++++++++++++++ .../backend/store/memory/InMemoryDBStore.java | 6 +++++ .../backend/store/memory/InMemoryDBTable.java | 7 ++++++ .../hugegraph/backend/tx/AbstractTransaction.java | 10 ++++++++ .../hugegraph/backend/tx/SchemaTransaction.java | 28 ++++++++++++++++++--- .../com/baidu/hugegraph/type/define/Action.java | 10 +++++++- .../backend/store/rocksdb/RocksDBStore.java | 6 +++++ .../backend/store/rocksdb/RocksDBTable.java | 8 ++++++ 9 files changed, 109 insertions(+), 4 deletions(-) diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedSchemaTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedSchemaTransaction.java index 0e7a55328..fb3fe79c2 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedSchemaTransaction.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/cache/CachedSchemaTransaction.java @@ -209,6 +209,15 @@ public final class CachedSchemaTransaction extends SchemaTransaction { return IdGenerator.of(type.string() + "-" + name); } + @Override + protected void updateSchema(SchemaElement schema) { + super.updateSchema(schema); + + this.updateCache(schema); + + this.notifyChanges(Cache.ACTION_INVALIDED, schema.type(), schema.id()); + } + @Override protected void addSchema(SchemaElement schema) { super.addSchema(schema); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java index 7126e4a63..f15f97dc7 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/BackendTable.java @@ -66,6 +66,30 @@ public abstract class BackendTable<Session extends BackendSession, Entry> { // pass } + public void updateIfPresent(Session session, Entry entry) { + // TODO: use fine-grained row lock + synchronized (this.table) { + if (this.queryExist(session, entry)) { + this.insert(session, entry); + if (session != null) { + session.commit(); + } + } + } + } + + public void updateIfAbsent(Session session, Entry entry) { + // TODO: use fine-grained row lock + synchronized (this.table) { + if (!this.queryExist(session, entry)) { + this.insert(session, entry); + if (session != null) { + session.commit(); + } + } + } + } + /** * Mapping query-type to table-type * @param query origin query @@ -112,6 +136,11 @@ public abstract class BackendTable<Session extends BackendSession, Entry> { public abstract Number queryNumber(Session session, Query query); +// public abstract boolean queryExist(Session session, Entry entry); + public boolean queryExist(Session session, Entry entry) { + return false; + } + public abstract void insert(Session session, Entry entry); public abstract void delete(Session session, Entry entry); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBStore.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBStore.java index 5d90fc95b..ead055353 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBStore.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBStore.java @@ -155,6 +155,12 @@ public abstract class InMemoryDBStore LOG.debug("[store {}] eliminate entry: {}", this.store, entry); table.eliminate(null, entry); break; + case UPDATE_IF_PRESENT: + table.updateIfPresent(null, entry); + break; + case UPDATE_IF_ABSENT: + table.updateIfAbsent(null, entry); + break; default: throw new BackendException("Unsupported mutate type: %s", item.action()); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBTable.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBTable.java index 7d5583dff..228048858 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBTable.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/store/memory/InMemoryDBTable.java @@ -50,6 +50,7 @@ import com.baidu.hugegraph.util.E; import com.baidu.hugegraph.util.InsertionOrderUtil; import com.baidu.hugegraph.util.Log; import com.google.common.base.Strings; +import com.google.common.collect.ImmutableList; public class InMemoryDBTable extends BackendTable<BackendSession, TextBackendEntry> { @@ -133,6 +134,12 @@ public class InMemoryDBTable extends BackendTable<BackendSession, } } + @Override + public boolean queryExist(BackendSession session, TextBackendEntry entry) { + List<Id> ids = ImmutableList.of(entry.id()); + return !this.queryById(ids, this.store).isEmpty(); + } + @Override public Number queryNumber(BackendSession session, Query query) { Aggregate aggregate = query.aggregateNotNull(); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java index 454ea2af4..db7d54eac 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/AbstractTransaction.java @@ -414,6 +414,16 @@ public abstract class AbstractTransaction implements Transaction { this.doAction(Action.DELETE, entry); } + @Watched(prefix = "tx") + public void doUpdateIfPresent(BackendEntry entry) { + this.doAction(Action.UPDATE_IF_PRESENT, entry); + } + + @Watched(prefix = "tx") + public void doUpdateIfAbsent(BackendEntry entry) { + this.doAction(Action.UPDATE_IF_ABSENT, entry); + } + protected void doAction(Action action, BackendEntry entry) { LOG.debug("Transaction {} entry {}", action, entry); E.checkNotNull(entry, "entry"); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/SchemaTransaction.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/SchemaTransaction.java index d0496333e..98cf4309d 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/SchemaTransaction.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/backend/tx/SchemaTransaction.java @@ -378,13 +378,20 @@ public class SchemaTransaction extends IndexableTransaction { } protected void updateSchema(SchemaElement schema) { - this.addSchema(schema); + LOG.debug("SchemaTransaction update {} with id '{}'", + schema.type(), schema.id()); + this.saveSchema(schema, true); } protected void addSchema(SchemaElement schema) { LOG.debug("SchemaTransaction add {} with id '{}'", schema.type(), schema.id()); setCreateTimeIfNeeded(schema); + this.saveSchema(schema, false); + } + + private void saveSchema(SchemaElement schema, boolean update) { + BackendEntry entry = this.serialize(schema); // System schema just put into SystemSchemaStore in memory if (schema.longId() < 0L) { @@ -396,9 +403,19 @@ public class SchemaTransaction extends IndexableTransaction { try { locks.lockWrites(LockUtil.hugeType2Group(schema.type()), schema.id()); + this.beforeWrite(); - this.doInsert(this.serialize(schema)); - this.indexTx.updateNameIndex(schema, false); + + if (update) { + this.doUpdateIfPresent(entry); + // TODO: support updateIfPresent + this.indexTx.updateNameIndex(schema, false); + } else { + // TODO: support updateIfAbsentProperty + this.doInsert(entry); + this.indexTx.updateNameIndex(schema, false); + } + this.afterWrite(); } finally { locks.unlock(); @@ -459,6 +476,11 @@ public class SchemaTransaction extends IndexableTransaction { List<T> results = new ArrayList<>(); Query query = new Query(type); Iterator<BackendEntry> entries = this.query(query).iterator(); + /* + * Can use MapperIterator instead if don't need to debug: + * new MapperIterator<>(entries, entry -> this.deserialize(entry, type)) + * QueryResults.fillList(iter, results); + */ try { while (entries.hasNext()) { BackendEntry entry = entries.next(); diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/type/define/Action.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/type/define/Action.java index 8411e929e..255627d53 100644 --- a/hugegraph-core/src/main/java/com/baidu/hugegraph/type/define/Action.java +++ b/hugegraph-core/src/main/java/com/baidu/hugegraph/type/define/Action.java @@ -27,7 +27,11 @@ public enum Action implements SerialEnum { ELIMINATE(3, "eliminate"), - DELETE(4, "delete"); + DELETE(4, "delete"), + + UPDATE_IF_PRESENT(5, "update_if_present"), + + UPDATE_IF_ABSENT(6, "update_if_absent"); private final byte code; private final String name; @@ -61,6 +65,10 @@ public enum Action implements SerialEnum { return ELIMINATE; case 4: return DELETE; + case 5: + return UPDATE_IF_PRESENT; + case 6: + return UPDATE_IF_ABSENT; default: throw new AssertionError("Unsupported action code: " + code); } diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java index cd14d3e6d..56230ac93 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBStore.java @@ -473,6 +473,12 @@ public abstract class RocksDBStore extends AbstractBackendStore<Session> { case ELIMINATE: table.eliminate(session, entry); break; + case UPDATE_IF_PRESENT: + table.updateIfPresent(session, entry); + break; + case UPDATE_IF_ABSENT: + table.updateIfAbsent(session, entry); + break; default: throw new AssertionError(String.format( "Unsupported mutate action: %s", item.action())); diff --git a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java index 427161bff..48c9a1af1 100644 --- a/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java +++ b/hugegraph-rocksdb/src/main/java/com/baidu/hugegraph/backend/store/rocksdb/RocksDBTable.java @@ -121,6 +121,14 @@ public class RocksDBTable extends BackendTable<Session, BackendEntry> { this.delete(session, entry); } + @Override + public boolean queryExist(Session session, BackendEntry entry) { + Id id = entry.id(); + try (BackendColumnIterator iter = this.queryById(session, id)) { + return iter.hasNext(); + } + } + @Override public Number queryNumber(Session session, Query query) { Aggregate aggregate = query.aggregateNotNull();
