diff --git a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java index c6e0823288..77262b91ab 100755 --- a/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java +++ b/hugegraph-server/hugegraph-hstore/src/main/java/org/apache/hugegraph/backend/store/hstore/HstoreTable.java @@ -568,8 +568,13 @@ private ConditionQuery prepareConditionQuery(ConditionQuery conditionQuery) { } } if (newConditions.size() > 0) { - conditionQuery.resetConditions(newConditions); - return conditionQuery; + // NOTE: copy before reset, the origin query is still used by core + // for result filtering after the backend scan returns; drop the + // back reference so the serialized payload stays flat + ConditionQuery pushdown = conditionQuery.copy(); + pushdown.resetConditions(newConditions); + pushdown.setOriginQuery(null); + return pushdown; } else { return null; } @@ -594,8 +599,11 @@ private ConditionQuery prepareConditionQueryList(ConditionQuery conditionQuery) } } if (newConditions.size() > 0) { - conditionQuery.resetConditions(newConditions); - return conditionQuery; + // NOTE: copy before reset, see prepareConditionQuery() + ConditionQuery pushdown = conditionQuery.copy(); + pushdown.resetConditions(newConditions); + pushdown.setOriginQuery(null); + return pushdown; } else { return null; } @@ -623,7 +631,6 @@ protected BackendColumnIterator queryByRange(Session session, type |= query.inclusiveEnd() ? Session.SCAN_LTE_END : Session.SCAN_LT_END; } - ConditionQuery cq; Query origin = query.originQuery(); byte[] position = null; byte[] ownerStart = this.ownerByQueryDelegate.apply(query.resultType(), @@ -640,21 +647,20 @@ protected BackendColumnIterator queryByRange(Session session, if (query.paging() && !query.page().isEmpty()) { position = PageState.fromString(query.page()).position(); } + byte[] queryBytes = null; if (origin instanceof ConditionQuery && (query.resultType().isEdge() || query.resultType().isVertex())) { - cq = (ConditionQuery) query.originQuery(); - - // LOG.debug("query {} with ownerKeyFrom: {}, ownerKeyTo: {}, " + - // "keyFrom: {}, keyTo: {}, " + - // "scanType: {}, conditionQuery: {}", - // this.table(), bytes2String(ownerStart), - // bytes2String(ownerEnd), bytes2String(start), - // bytes2String(end), type, cq.bytes()); - return session.scan(this.table(), ownerStart, - ownerEnd, start, end, type, cq.bytes(), position); - } - return session.scan(this.table(), ownerStart, - ownerEnd, start, end, type, null, position); + // Same guard as queryByPrefix(): only push the query down to the + // store when user-prop conditions remain. A sort-key prefix/range + // query keeps sysprop conditions only (owner vertex, direction, + // label, sort values), which are already enforced by the key + // range, and the store-side row decoder cannot parse the raw + // property layout written by the server (see issue #3090). + ConditionQuery cq = prepareConditionQuery((ConditionQuery) origin); + queryBytes = cq == null ? null : cq.bytes(); + } + return session.scan(this.table(), ownerStart, ownerEnd, start, end, + type, queryBytes, position); } static boolean shouldUseOrderedRangeScan(IdRangeQuery query) { diff --git a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreTableTest.java b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreTableTest.java index b61b65a768..2f374fccf1 100644 --- a/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreTableTest.java +++ b/hugegraph-server/hugegraph-hstore/src/test/java/org/apache/hugegraph/backend/store/hstore/HstoreTableTest.java @@ -17,22 +17,32 @@ package org.apache.hugegraph.backend.store.hstore; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Iterator; import java.util.List; import java.util.NoSuchElementException; +import org.apache.commons.lang3.tuple.Pair; import org.apache.hugegraph.backend.id.Id.IdType; import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.backend.page.PageInfo; import org.apache.hugegraph.backend.page.PageState; +import org.apache.hugegraph.backend.query.Condition; +import org.apache.hugegraph.backend.query.ConditionQuery; +import org.apache.hugegraph.backend.query.IdPrefixQuery; import org.apache.hugegraph.backend.query.IdRangeQuery; import org.apache.hugegraph.backend.query.Query; import org.apache.hugegraph.backend.store.BackendEntry; import org.apache.hugegraph.backend.store.BackendEntry.BackendColumn; import org.apache.hugegraph.backend.store.BackendEntry.BackendColumnIterator; import org.apache.hugegraph.backend.store.BackendEntryIterator; +import org.apache.hugegraph.store.HgOwnerKey; import org.apache.hugegraph.store.client.util.HgStoreClientConst; import org.apache.hugegraph.type.HugeType; +import org.apache.hugegraph.type.define.Directions; +import org.apache.hugegraph.type.define.GraphMode; +import org.apache.hugegraph.type.define.HugeKeys; import org.junit.Assert; import org.junit.Test; @@ -124,6 +134,101 @@ public void testRangeScanBudgetIncludesOneLookaheadRecord() { Assert.assertEquals(14L, HstoreTable.rangeScanBudget(query)); } + @Test + public void testRangeQueryWithoutUserpropsDoesNotPushConditions() { + // Sort-key prefix/range queries keep sysprop conditions only (owner + // vertex, direction, label, sort values); those are enforced by the + // key range already and must not be pushed to the store, whose row + // decoder cannot parse the server's raw property layout (issue #3090) + ConditionQuery origin = new ConditionQuery(HugeType.EDGE); + origin.eq(HugeKeys.OWNER_VERTEX, IdGenerator.of("v1")); + origin.eq(HugeKeys.DIRECTION, Directions.OUT); + origin.eq(HugeKeys.LABEL, IdGenerator.of(1L)); + origin.gte(HugeKeys.SORT_VALUES, "ETC!"); + origin.lt(HugeKeys.SORT_VALUES, "ETC~"); + int before = origin.conditions().size(); + + ScanRecordingSession session = new ScanRecordingSession(); + this.newTestTable().queryByRange(session, edgeRangeQuery(origin)); + + Assert.assertTrue(session.scanCalled); + Assert.assertNull(session.lastQueryBytes); + Assert.assertEquals(before, origin.conditions().size()); + } + + @Test + public void testRangeQueryWithUserpropsPushesCopyAndKeepsOrigin() { + ConditionQuery origin = new ConditionQuery(HugeType.EDGE); + origin.eq(HugeKeys.OWNER_VERTEX, IdGenerator.of("v1")); + origin.query(Condition.eq(IdGenerator.of(7L), 100)); + int before = origin.conditions().size(); + + ScanRecordingSession session = new ScanRecordingSession(); + this.newTestTable().queryByRange(session, edgeRangeQuery(origin)); + + Assert.assertTrue(session.scanCalled); + Assert.assertNotNull(session.lastQueryBytes); + // the pushed-down query is a copy: the origin query keeps all its + // conditions for core-side filtering after the scan returns + Assert.assertEquals(before, origin.conditions().size()); + // pushed payload: user-prop condition survives, owner-vertex is + // dropped, and the back reference to the origin query is cleared + ConditionQuery pushed = ConditionQuery.fromBytes(session.lastQueryBytes); + Assert.assertNull(pushed.condition(HugeKeys.OWNER_VERTEX)); + Assert.assertFalse(pushed.userpropConditions().isEmpty()); + Assert.assertNull(pushed.originQuery()); + } + + @Test + public void testPrefixListQueryPushesCopyAndKeepsOrigin() { + // prepareConditionQueryList() is reached from queryByPrefixList() and + // from the streaming query(Session, Iterator, String); one origin + // query is shared by every prefix query of the batch + ConditionQuery origin = new ConditionQuery(HugeType.EDGE); + origin.eq(HugeKeys.OWNER_VERTEX, IdGenerator.of("v1")); + origin.eq(HugeKeys.DIRECTION, Directions.OUT); + origin.eq(HugeKeys.LABEL, IdGenerator.of(1L)); + origin.query(Condition.eq(IdGenerator.of(7L), 100)); + int before = origin.conditions().size(); + List queries = Arrays.asList( + new IdPrefixQuery(origin, IdGenerator.of(keyBytes(1), + IdType.STRING)), + new IdPrefixQuery(origin, IdGenerator.of(keyBytes(2), + IdType.STRING))); + + ScanRecordingSession session = new ScanRecordingSession(); + List iterators = this.newTestTable() + .queryByPrefixList(session, queries, "g+oe"); + + Assert.assertTrue(session.scanCalled); + Assert.assertEquals(2, session.lastOwnerKeys.size()); + Assert.assertEquals(2, iterators.size()); + Assert.assertNotNull(session.lastQueryBytes); + // the shared origin query keeps every condition, including the + // owner vertex that the pushed copy drops + Assert.assertEquals(before, origin.conditions().size()); + Assert.assertNotNull(origin.condition(HugeKeys.OWNER_VERTEX)); + ConditionQuery pushed = ConditionQuery.fromBytes(session.lastQueryBytes); + Assert.assertNull(pushed.condition(HugeKeys.OWNER_VERTEX)); + Assert.assertNotNull(pushed.condition(HugeKeys.LABEL)); + Assert.assertFalse(pushed.userpropConditions().isEmpty()); + Assert.assertNull(pushed.originQuery()); + } + + private HstoreTable newTestTable() { + HstoreTable table = new HstoreTable("hugegraph", "g+oe"); + table.ownerByQueryDelegate = (type, id) -> new byte[]{0}; + return table; + } + + private static IdRangeQuery edgeRangeQuery(ConditionQuery origin) { + return new IdRangeQuery(HugeType.EDGE_OUT, origin, + IdGenerator.of(keyBytes(1), IdType.STRING), + true, + IdGenerator.of(keyBytes(9), IdType.STRING), + false); + } + private static IdRangeQuery rangeIndexQuery() { return new IdRangeQuery(HugeType.RANGE_INT_INDEX, null, IdGenerator.of(keyBytes(1), IdType.STRING), @@ -139,6 +244,201 @@ private static byte[] keyBytes(int key) { return bytes; } + private static final class ScanRecordingSession extends HstoreSessions.Session { + + private boolean scanCalled = false; + private byte[] lastQueryBytes = null; + private List lastOwnerKeys = null; + + @Override + public BackendColumnIterator scan(String table, byte[] ownerKeyFrom, + byte[] ownerKeyTo, byte[] keyFrom, + byte[] keyTo, int scanType, + byte[] query, byte[] position) { + this.scanCalled = true; + this.lastQueryBytes = query; + return new TestColumnIterator(); + } + + @Override + public List scan(String table, + List keys, + int scanType, long limit, + byte[] query) { + this.scanCalled = true; + this.lastQueryBytes = query; + this.lastOwnerKeys = keys; + List iterators = new ArrayList<>(); + for (int i = 0; i < keys.size(); i++) { + iterators.add(new TestColumnIterator()); + } + return iterators; + } + + @Override + public void open() { + } + + @Override + public void close() { + } + + @Override + public Object commit() { + return null; + } + + @Override + public void rollback() { + } + + @Override + public boolean hasChanges() { + return false; + } + + @Override + public void createTable(String tableName) { + } + + @Override + public void dropTable(String tableName) { + } + + @Override + public boolean existsTable(String tableName) { + return true; + } + + @Override + public void truncateTable(String tableName) { + } + + @Override + public void deleteGraph() { + } + + @Override + public Pair keyRange(String table) { + return null; + } + + @Override + public void put(String table, byte[] ownerKey, byte[] key, + byte[] value) { + } + + @Override + public void increase(String table, byte[] ownerKey, byte[] key, + byte[] value) { + } + + @Override + public void delete(String table, byte[] ownerKey, byte[] key) { + } + + @Override + public void deletePrefix(String table, byte[] ownerKey, byte[] key) { + } + + @Override + public void deleteRange(String table, byte[] ownerKeyFrom, + byte[] ownerKeyTo, byte[] keyFrom, + byte[] keyTo) { + } + + @Override + public byte[] get(String table, byte[] key) { + return new byte[0]; + } + + @Override + public byte[] get(String table, byte[] ownerKey, byte[] key) { + return new byte[0]; + } + + @Override + public BackendColumnIterator scan(String table) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendColumnIterator scan(String table, byte[] ownerKey, + byte[] prefix) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendEntry.BackendIterator scan( + String table, Iterator keys, int scanType, + Query queryParam, byte[] query) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendColumnIterator scan(String table, byte[] ownerKeyFrom, + byte[] ownerKeyTo, byte[] keyFrom, + byte[] keyTo, int scanType) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendColumnIterator scan(String table, byte[] ownerKeyFrom, + byte[] ownerKeyTo, byte[] keyFrom, + byte[] keyTo, int scanType, + byte[] query) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendColumnIterator scan(String table, int codeFrom, + int codeTo, int scanType, + byte[] query) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendColumnIterator scan(String table, int codeFrom, + int codeTo, int scanType, + byte[] query, byte[] position) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendColumnIterator scan(String table, + byte[] conditionQueryToByte) { + throw new UnsupportedOperationException(); + } + + @Override + public BackendColumnIterator getWithBatch(String table, + List keys) { + throw new UnsupportedOperationException(); + } + + @Override + public void merge(String table, byte[] ownerKey, byte[] key, + byte[] value) { + } + + @Override + public void setMode(GraphMode mode) { + } + + @Override + public void truncate() throws Exception { + } + + @Override + public void beginTx() { + } + + @Override + public int getActiveStoreSize() { + return 0; + } + } + private static final class TestColumnIterator implements BackendColumnIterator {