imbajin commented on code in PR #3140:
URL: https://github.com/apache/hugegraph/pull/3140#discussion_r3732280189
##########
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/backend/query/QueryResults.java:
##########
@@ -102,12 +102,12 @@ public <T extends Idfiable> Iterator<T>
keepInputOrderIfNeeded(
return origin;
}
Collection<Id> ids;
- if (!this.mustSortByInputIds() || this.paging() ||
+ if (!this.mustSortByInputIds() ||
Review Comment:
‼️ critical — Removing the paging short-circuit makes
`keepInputOrderIfNeeded()` call `fillMap(origin, map)` on the lazy page
iterator. The new HStore paging path sets `mustSortByInputIds`, so an unbounded
multi-page query is now fully materialized before returning, which can exhaust
all pages and cause large memory use/OOM. Preserve lazy page boundaries while
ordering each page, and add an unbounded multi-page regression test.
##########
hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxSessionProxyTest.java:
##########
@@ -0,0 +1,398 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.store.client;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hugegraph.store.HgKvEntry;
+import org.apache.hugegraph.store.HgKvIterator;
+import org.apache.hugegraph.store.HgOwnerKey;
+import org.apache.hugegraph.store.HgStoreSession;
+import org.apache.hugegraph.store.grpc.common.ScanMethod;
+import org.apache.hugegraph.store.grpc.common.ScanOrderType;
+import org.apache.hugegraph.store.grpc.stream.ScanStreamReq.Builder;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class NodeTxSessionProxyTest {
+
+ @Test
+ public void testNodeTkvDoesNotMutateSharedOwnerKeys() {
+ HgOwnerKey start = HgOwnerKey.of(keyBytes(9), keyBytes(1));
+ HgOwnerKey end = HgOwnerKey.of(keyBytes(9), keyBytes(5));
+ start.setSerialNo(7);
+ end.setSerialNo(8);
+
+ NodeTkv first = new NodeTkv(HgNodePartition.of(1L, 0, 10, 20),
+ "g+index", start, end);
+ NodeTkv second = new NodeTkv(HgNodePartition.of(2L, 0, 30, 40),
+ "g+index", start, end);
+
+ Assert.assertEquals(0, start.getKeyCode());
+ Assert.assertEquals(0, end.getKeyCode());
+ Assert.assertEquals(10, first.getKey().getKeyCode());
+ Assert.assertEquals(20, first.getEndKey().getKeyCode());
+ Assert.assertEquals(30, second.getKey().getKeyCode());
+ Assert.assertEquals(40, second.getEndKey().getKeyCode());
+ Assert.assertEquals(7, first.getKey().getSerialNo());
+ Assert.assertEquals(8, first.getEndKey().getSerialNo());
+ }
+
+ @Test
+ public void testScanIteratorOrderedUsesOneStreamPerStoreLazily()
Review Comment:
⚠️ important — The new tests use fake iterators and do not exercise the
cross-module contract: two or more Stores/partitions, more than the 64-entry
page, empty and partially failing sources, continuation cursors, and early
close. Add an integration-level regression covering interleaved keys across
multiple pages and the final cursor/limit behavior.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.store.client;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.PriorityQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.hugegraph.store.HgKvEntry;
+import org.apache.hugegraph.store.HgKvIterator;
+import org.apache.hugegraph.store.client.util.ExecutorPool;
+import org.apache.hugegraph.store.client.util.HgStoreClientConst;
+
+final class OrderedKvIterator implements HgKvIterator<HgKvEntry> {
+
+ private static final int INITIALIZE_THREADS = 8;
+ private static final ExecutorService INITIALIZER =
Review Comment:
⚠️ important — All ordered scans share one fixed eight-thread initializer.
Because each scan uses `invokeAll()` for every Store, slow partitions from
eight concurrent queries can occupy the global pool and block unrelated queries
behind them. Add bounded cancellation/timeouts and isolation or use an existing
query-scoped executor; cover a slow-query concurrency case.
##########
hugegraph-store/hg-store-node/src/main/java/org/apache/hugegraph/store/node/grpc/ScanUtil.java:
##########
@@ -71,7 +74,14 @@ static ScanIterator getIterator(ScanStreamReq request,
HgStoreWrapperEx wrapper)
iter = wrapper.scanPrefix(graph, partition, table, prefix,
scanType, query);
break;
case RANGE:
- iter = wrapper.scan(graph, partition, table, start, end,
scanType, query);
+ if (partition == SCAN_ALL_PARTITIONS_ID &&
Review Comment:
⚠️ important — The `ORDER_BY_KEY` routing is implemented only for
`ScanStreamReq`. One-shot scans call `getIterator(toSq(request), wrapper)`, but
`ScanQuery` drops `orderType` and its RANGE branch always invokes the legacy
partition scan. An ordered one-shot request can therefore return
partition-concatenated results and apply its limit incorrectly. Preserve the
order flag through `ScanQuery` or route one-shot requests through the request
overload, with an interleaved-partition test.
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/OrderedKvIterator.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.store.client;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.PriorityQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.hugegraph.store.HgKvEntry;
+import org.apache.hugegraph.store.HgKvIterator;
+import org.apache.hugegraph.store.client.util.ExecutorPool;
+import org.apache.hugegraph.store.client.util.HgStoreClientConst;
+
+final class OrderedKvIterator implements HgKvIterator<HgKvEntry> {
+
+ private static final int INITIALIZE_THREADS = 8;
+ private static final ExecutorService INITIALIZER =
+ Executors.newFixedThreadPool(
+ INITIALIZE_THREADS,
+ ExecutorPool.newThreadFactory("ordered-scan-init"));
+
+ private final List<? extends HgKvIterator<? extends HgKvEntry>> iterators;
+ private final PriorityQueue<SourceEntry> queue;
+ private final boolean[] sourceClosed;
+ private final long limit;
+ private final ExecutorService initializer;
+
+ private boolean initialized;
+ private boolean closed;
+ private long count;
+ private HgKvEntry current;
+ private byte[] position;
+
+ OrderedKvIterator(List<? extends HgKvIterator<? extends HgKvEntry>>
iterators,
+ long limit) {
+ this(iterators, limit, INITIALIZER);
+ }
+
+ OrderedKvIterator(List<? extends HgKvIterator<? extends HgKvEntry>>
iterators,
+ long limit, ExecutorService initializer) {
+ this.iterators = iterators;
+ this.queue = new PriorityQueue<>((left, right) -> {
+ int result = Arrays.compareUnsigned(left.entry.key(),
+ right.entry.key());
+ if (result != 0) {
+ return result;
+ }
+ return Integer.compare(left.source, right.source);
+ });
+ this.sourceClosed = new boolean[iterators.size()];
+ this.limit = limit <= HgStoreClientConst.NO_LIMIT ? Long.MAX_VALUE :
+ limit;
+ this.initializer = Objects.requireNonNull(initializer);
+ this.initialized = false;
+ this.closed = false;
+ this.count = 0L;
+ this.current = null;
+ this.position = HgStoreClientConst.EMPTY_BYTES;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (this.closed) {
+ return false;
+ }
+ this.initialize();
+ boolean hasNext = this.count < this.limit && !this.queue.isEmpty();
+ if (!hasNext) {
+ this.close();
+ }
+ return hasNext;
+ }
+
+ @Override
+ public HgKvEntry next() {
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ SourceEntry sourceEntry = this.queue.poll();
+ this.current = sourceEntry.entry;
+ this.position = this.current.key();
+ this.count++;
+
+ try {
+ if (this.count < this.limit) {
+ this.addNext(sourceEntry.source);
+ } else {
+ this.close();
+ }
+ } catch (RuntimeException | Error e) {
+ this.closeAfterFailure(e);
+ throw e;
+ }
+ return this.current;
+ }
+
+ @Override
+ public byte[] key() {
+ return this.current == null ? null : this.current.key();
+ }
+
+ @Override
+ public byte[] value() {
+ return this.current == null ? null : this.current.value();
+ }
+
+ @Override
+ public byte[] position() {
+ return this.position;
+ }
+
+ @Override
+ public void close() {
+ if (this.closed) {
+ return;
+ }
+ this.closed = true;
+ Throwable failure = null;
+ for (int i = 0; i < this.iterators.size(); i++) {
+ try {
+ this.closeSource(i);
+ } catch (RuntimeException | Error e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ this.queue.clear();
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ if (failure != null) {
+ throw (Error) failure;
+ }
+ }
+
+ private void initialize() {
+ if (this.initialized) {
+ return;
+ }
+ this.initialized = true;
+ List<Callable<SourceEntry>> tasks =
+ new ArrayList<>(this.iterators.size());
+ for (int i = 0; i < this.iterators.size(); i++) {
+ int source = i;
+ tasks.add(() -> this.firstEntry(source));
+ }
+ try {
+ List<Future<SourceEntry>> futures =
+ this.initializer.invokeAll(tasks);
Review Comment:
⚠️ important — `invokeAll(tasks)` waits for every first-entry task before
observing any `ExecutionException`. If one Store fails quickly while another is
slow or blocked, the failure cannot close the other streams until the slowest
task returns. Use completion/future handling that cancels unfinished tasks on
the first failure and closes all iterators in a `finally` path.
##########
hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/OrderedMultiPartitionIterator.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.store.business;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.PriorityQueue;
+import java.util.function.Function;
+
+import org.apache.hugegraph.rocksdb.access.RocksDBSession.BackendColumn;
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+
+public final class OrderedMultiPartitionIterator implements ScanIterator {
+
+ private static final byte[] EMPTY_BYTES = new byte[0];
+
+ private final List<Integer> partitionIds;
+ private final Function<Integer, ScanIterator> supplier;
+ private final List<SourceEntry> sources;
+ private final PriorityQueue<SourceEntry> queue;
+
+ private boolean initialized;
+ private boolean closed;
+ private Integer currentPartitionId;
+
+ private OrderedMultiPartitionIterator(List<Integer> partitionIds,
+ Function<Integer, ScanIterator>
supplier) {
+ this.partitionIds = new
ArrayList<>(Objects.requireNonNull(partitionIds));
+ Collections.sort(this.partitionIds);
+ this.supplier = Objects.requireNonNull(supplier);
+ this.sources = new ArrayList<>(this.partitionIds.size());
+ this.queue = new PriorityQueue<>((left, right) -> {
+ int result = Arrays.compareUnsigned(left.entry.name,
+ right.entry.name);
+ if (result != 0) {
+ return result;
+ }
+ return Integer.compare(left.partitionId, right.partitionId);
+ });
+ this.initialized = false;
+ this.closed = false;
+ this.currentPartitionId = null;
+ }
+
+ public static OrderedMultiPartitionIterator of(
+ List<Integer> partitionIds,
+ Function<Integer, ScanIterator> supplier) {
+ return new OrderedMultiPartitionIterator(partitionIds, supplier);
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (this.closed) {
+ return false;
+ }
+ this.initialize();
+ if (this.queue.isEmpty()) {
+ this.close();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public boolean isValid() {
+ return this.hasNext();
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T next() {
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ SourceEntry source = this.queue.poll();
+ BackendColumn current = source.entry;
+ this.currentPartitionId = source.partitionId;
+ try {
+ if (source.iterator.hasNext()) {
+ source.entry = source.iterator.next();
+ this.queue.add(source);
+ } else {
+ this.closeSource(source);
+ }
+ } catch (RuntimeException | Error e) {
+ this.closeAfterFailure(e);
+ throw e;
+ }
+ return (T) current;
+ }
+
+ @Override
+ public byte[] position() {
+ if (this.currentPartitionId == null) {
+ return EMPTY_BYTES;
+ }
+ return ByteBuffer.allocate(Integer.BYTES)
+ .putInt(this.currentPartitionId)
+ .array();
+ }
+
+ @Override
+ public void seek(byte[] position) {
Review Comment:
⚠️ important — `ScanUtil.getIterator(ScanStreamReq)` unconditionally calls
`iter.seek(request.getPosition())`, but this new iterator throws for every
non-empty position. A continuation request carrying the previous ordered-scan
cursor therefore fails instead of resuming the next page. Implement a
per-partition resume strategy or reject this mode before opening the scan and
document the contract; add a non-empty-cursor test.
##########
hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/OrderedMultiPartitionIterator.java:
##########
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hugegraph.store.business;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.PriorityQueue;
+import java.util.function.Function;
+
+import org.apache.hugegraph.rocksdb.access.RocksDBSession.BackendColumn;
+import org.apache.hugegraph.rocksdb.access.ScanIterator;
+
+public final class OrderedMultiPartitionIterator implements ScanIterator {
+
+ private static final byte[] EMPTY_BYTES = new byte[0];
+
+ private final List<Integer> partitionIds;
+ private final Function<Integer, ScanIterator> supplier;
+ private final List<SourceEntry> sources;
+ private final PriorityQueue<SourceEntry> queue;
+
+ private boolean initialized;
+ private boolean closed;
+ private Integer currentPartitionId;
+
+ private OrderedMultiPartitionIterator(List<Integer> partitionIds,
+ Function<Integer, ScanIterator>
supplier) {
+ this.partitionIds = new
ArrayList<>(Objects.requireNonNull(partitionIds));
+ Collections.sort(this.partitionIds);
+ this.supplier = Objects.requireNonNull(supplier);
+ this.sources = new ArrayList<>(this.partitionIds.size());
+ this.queue = new PriorityQueue<>((left, right) -> {
+ int result = Arrays.compareUnsigned(left.entry.name,
+ right.entry.name);
+ if (result != 0) {
+ return result;
+ }
+ return Integer.compare(left.partitionId, right.partitionId);
+ });
+ this.initialized = false;
+ this.closed = false;
+ this.currentPartitionId = null;
+ }
+
+ public static OrderedMultiPartitionIterator of(
+ List<Integer> partitionIds,
+ Function<Integer, ScanIterator> supplier) {
+ return new OrderedMultiPartitionIterator(partitionIds, supplier);
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (this.closed) {
+ return false;
+ }
+ this.initialize();
+ if (this.queue.isEmpty()) {
+ this.close();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public boolean isValid() {
+ return this.hasNext();
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T next() {
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ SourceEntry source = this.queue.poll();
+ BackendColumn current = source.entry;
+ this.currentPartitionId = source.partitionId;
+ try {
+ if (source.iterator.hasNext()) {
+ source.entry = source.iterator.next();
+ this.queue.add(source);
+ } else {
+ this.closeSource(source);
+ }
+ } catch (RuntimeException | Error e) {
+ this.closeAfterFailure(e);
+ throw e;
+ }
+ return (T) current;
+ }
+
+ @Override
Review Comment:
⚠️ important — This iterator does not override `ScanIterator.count()`, whose
default is `0`; `FilterIterator` and `SelectIterator` delegate to it, while the
existing `MultiPartitionIterator` sums source counts. Ordered scans therefore
report zero to count callers. Implement consistent count semantics and add a
count regression test.
##########
hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java:
##########
@@ -69,6 +69,9 @@ void doPut(String graph, int code, String table, byte[] key,
byte[] value) throw
ScanIterator scan(String graph, int code, String table, byte[] start,
byte[] end, int scanType) throws HgStoreException;
+ ScanIterator scanOrdered(String graph, String table, byte[] start,
Review Comment:
⚠️ important — Adding a new abstract method to the public `BusinessHandler`
interface (and the matching abstract `scanOrdered` method on public
`HstoreSessions.Session`) is a source and binary compatibility break for
external implementations, plugins, and test doubles. Prefer a default
capability method or a separate optional interface with a guarded fallback, and
document the rollout requirement.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
