You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geaflow.apache.org by lt...@apache.org on 2026/08/22 02:32:55 UTC
(geaflow) branch master updated: [ISSUE-796] Clean up PartitionType TODOs in rocksdb store module (#823)
This is an automated email from the ASF dual-hosted git repository.
Leomrlin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/geaflow.git
The following commit(s) were added to refs/heads/master by this push:
new 47a1b2e10 [ISSUE-796] Clean up PartitionType TODOs in rocksdb store module (#823)
47a1b2e10 is described below
commit 47a1b2e10ccb9bcad325b218dc883db062815ad8
Author: Heracles <26...@qq.com>
AuthorDate: Sat Aug 22 10:32:50 2026 +0800
[ISSUE-796] Clean up PartitionType TODOs in rocksdb store module (#823)
* [ISSUE-796] Clean up PartitionType TODOs in rocksdb store module
The `PartitionType` enum in the rocksdb store carried two stale `TODO`s:
- `// TODO: Support dt partition` on `DT` — DT partition is already
implemented (`SyncGraphDtPartitionProxy`, dispatched by `ProxyBuilder`),
so the TODO is removed.
- `// TODO: Support label dt partition` on `DT_LABEL` — this combination
has no proxy implementation. Configuring it previously fell through to a
generic "unexpected partition type" error. `ProxyBuilder.build` now
rejects `DT_LABEL` explicitly with a message that names the unsupported
type, and the TODO is replaced with a NOTE documenting the gap.
Adds `PartitionTypeTest` covering the enum lookup/flags, the explicit
`DT_LABEL` rejection, and that `DT` is still dispatched to a real proxy.
The test targets `ProxyBuilder` directly so it does not depend on opening
a native RocksDB instance.
Co-Authored-By: Claude Opus 4.8 <no...@anthropic.com>
* [ISSUE-796] Address review: fail fast on DT_LABEL before opening RocksDB
- Extract ProxyBuilder.checkPartitionTypeSupported and call it in
StaticGraphRocksdbStoreBase.init before super.init() opens the RocksDB
instance, so an unsupported partition.type fails fast without leaving a
half-initialized store on disk.
- Rewrite the DT-dispatch test to mock RocksdbClient/encoder and assert the
returned proxy is a SyncGraphDtPartitionProxy, instead of inferring dispatch
from an incidental NullPointerException.
Co-Authored-By: Claude Opus 4.8 <no...@anthropic.com>
---------
Co-authored-by: Claude Opus 4.8 <no...@anthropic.com>
---
.../geaflow-store/geaflow-store-rocksdb/pom.xml | 6 ++
.../geaflow/store/rocksdb/PartitionType.java | 4 +-
.../store/rocksdb/StaticGraphRocksdbStoreBase.java | 3 +
.../geaflow/store/rocksdb/proxy/ProxyBuilder.java | 18 ++++
.../geaflow/store/rocksdb/PartitionTypeTest.java | 114 +++++++++++++++++++++
5 files changed, 143 insertions(+), 2 deletions(-)
diff --git a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/pom.xml b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/pom.xml
index b652e1d53..7b7835684 100644
--- a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/pom.xml
+++ b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/pom.xml
@@ -68,5 +68,11 @@
<artifactId>testng</artifactId>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/PartitionType.java b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/PartitionType.java
index 5c71cee93..4f77042bc 100644
--- a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/PartitionType.java
+++ b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/PartitionType.java
@@ -24,9 +24,9 @@ import org.apache.geaflow.common.exception.GeaflowRuntimeException;
// Partition type for rocksdb graph store
public enum PartitionType {
LABEL(false, true),
- // TODO: Support dt partition
DT(true, false),
- // TODO: Support label dt partition
+ // NOTE: DT_LABEL (partition by both timestamp and label) is not supported yet.
+ // ProxyBuilder rejects it with an explicit error until a proxy implementation is added.
DT_LABEL(true, true),
NONE(false, false);
diff --git a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/StaticGraphRocksdbStoreBase.java b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/StaticGraphRocksdbStoreBase.java
index 81bffca80..6359c62d0 100644
--- a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/StaticGraphRocksdbStoreBase.java
+++ b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/StaticGraphRocksdbStoreBase.java
@@ -54,6 +54,9 @@ public class StaticGraphRocksdbStoreBase<K, VV, EV> extends BaseRocksdbGraphStor
// Init partition type for rocksdb graph store
partitionType = PartitionType.getEnum(storeContext.getConfig()
.getString(RocksdbConfigKeys.ROCKSDB_GRAPH_STORE_PARTITION_TYPE));
+ // Fail fast on unsupported partition types before super.init opens the RocksDB instance,
+ // so an unsupported partition.type does not leave a half-initialized store on disk.
+ ProxyBuilder.checkPartitionTypeSupported(partitionType);
super.init(storeContext);
IGraphKVEncoder<K, VV, EV> encoder = GraphKVEncoderFactory.build(config,
diff --git a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/proxy/ProxyBuilder.java b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/proxy/ProxyBuilder.java
index 0f7e786ab..216c70d7c 100644
--- a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/proxy/ProxyBuilder.java
+++ b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/main/java/org/apache/geaflow/store/rocksdb/proxy/ProxyBuilder.java
@@ -29,11 +29,29 @@ import org.apache.geaflow.store.rocksdb.RocksdbConfigKeys;
public class ProxyBuilder {
+ /**
+ * Rejects partition types that have no proxy implementation yet. Callers should invoke this
+ * before opening the RocksDB instance so an unsupported {@code partition.type} fails fast
+ * without leaving a half-initialized store on disk.
+ *
+ * <p>{@code DT_LABEL} (partition by both timestamp and label) has no proxy, so it is rejected
+ * with an explicit message that names the type rather than falling through to a generic
+ * "unexpected partition type" error later in {@link #build}.
+ */
+ public static void checkPartitionTypeSupported(PartitionType partitionType) {
+ if (partitionType == PartitionType.DT_LABEL) {
+ throw new GeaflowRuntimeException(
+ "partition type DT_LABEL (partition by both timestamp and label) is not "
+ + "supported yet");
+ }
+ }
+
public static <K, VV, EV> IGraphRocksdbProxy<K, VV, EV> build(
Configuration config, RocksdbClient rocksdbClient,
IGraphKVEncoder<K, VV, EV> encoder) {
PartitionType partitionType = PartitionType.getEnum(
config.getString(RocksdbConfigKeys.ROCKSDB_GRAPH_STORE_PARTITION_TYPE));
+ checkPartitionTypeSupported(partitionType);
if (partitionType.isPartition()) {
if (partitionType == PartitionType.LABEL) {
// TODO: Support async graph proxy partitioned by label
diff --git a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/test/java/org/apache/geaflow/store/rocksdb/PartitionTypeTest.java b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/test/java/org/apache/geaflow/store/rocksdb/PartitionTypeTest.java
new file mode 100644
index 000000000..a2ab06b0c
--- /dev/null
+++ b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/test/java/org/apache/geaflow/store/rocksdb/PartitionTypeTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.geaflow.store.rocksdb;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.exception.GeaflowRuntimeException;
+import org.apache.geaflow.state.graph.encoder.IEdgeKVEncoder;
+import org.apache.geaflow.state.graph.encoder.IGraphKVEncoder;
+import org.apache.geaflow.state.graph.encoder.IVertexKVEncoder;
+import org.apache.geaflow.store.rocksdb.proxy.IGraphRocksdbProxy;
+import org.apache.geaflow.store.rocksdb.proxy.ProxyBuilder;
+import org.apache.geaflow.store.rocksdb.proxy.SyncGraphDtPartitionProxy;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class PartitionTypeTest {
+
+ @Test
+ public void testGetEnumIsCaseInsensitive() {
+ Assert.assertEquals(PartitionType.getEnum("dt"), PartitionType.DT);
+ Assert.assertEquals(PartitionType.getEnum("DT_LABEL"), PartitionType.DT_LABEL);
+ Assert.assertEquals(PartitionType.getEnum("none"), PartitionType.NONE);
+ }
+
+ @Test(expectedExceptions = GeaflowRuntimeException.class)
+ public void testGetEnumRejectsUnknownType() {
+ PartitionType.getEnum("unknown");
+ }
+
+ @Test
+ public void testPartitionFlags() {
+ Assert.assertTrue(PartitionType.DT.isDtPartition());
+ Assert.assertFalse(PartitionType.DT.isLabelPartition());
+ Assert.assertTrue(PartitionType.DT.isPartition());
+
+ Assert.assertTrue(PartitionType.DT_LABEL.isDtPartition());
+ Assert.assertTrue(PartitionType.DT_LABEL.isLabelPartition());
+
+ Assert.assertFalse(PartitionType.NONE.isPartition());
+ }
+
+ /**
+ * DT_LABEL partition has no proxy implementation yet, so {@link ProxyBuilder} must reject it
+ * fast with an explicit error that names the partition type, rather than falling through to a
+ * generic "unexpected partition type" message. The rejection happens before the RocksDB
+ * client or encoder are touched, so {@code null} arguments are fine here.
+ */
+ @Test
+ public void testDtLabelPartitionIsRejectedWithClearMessage() {
+ try {
+ ProxyBuilder.build(newConfig("dt_label"), null, (IGraphKVEncoder<Object, Object, Object>) null);
+ Assert.fail("expected DT_LABEL partition to be rejected");
+ } catch (GeaflowRuntimeException e) {
+ Assert.assertTrue(e.getMessage() != null && e.getMessage().contains("DT_LABEL"),
+ "error message should name the unsupported partition type, but was: "
+ + e.getMessage());
+ }
+ }
+
+ /**
+ * DT partition is implemented (see {@link SyncGraphDtPartitionProxy}), so {@link ProxyBuilder}
+ * must dispatch it to the DT proxy instead of rejecting it. We assert the concrete proxy type
+ * returned by the builder rather than inferring dispatch from an incidental
+ * {@link NullPointerException}, which is more explicit about what "dispatched to proxy" means.
+ */
+ @Test
+ public void testDtPartitionIsDispatchedToProxy() {
+ IGraphRocksdbProxy<Object, Object, Object> proxy =
+ ProxyBuilder.build(newConfig("dt"), mockClient(), mockEncoder());
+ Assert.assertTrue(proxy instanceof SyncGraphDtPartitionProxy,
+ "DT partition should be dispatched to SyncGraphDtPartitionProxy, but was: "
+ + proxy.getClass().getName());
+ }
+
+ private Configuration newConfig(String partitionType) {
+ Map<String, String> map = new HashMap<>();
+ map.put(RocksdbConfigKeys.ROCKSDB_GRAPH_STORE_PARTITION_TYPE.getKey(), partitionType);
+ return new Configuration(map);
+ }
+
+ @SuppressWarnings("unchecked")
+ private IGraphKVEncoder<Object, Object, Object> mockEncoder() {
+ IGraphKVEncoder<Object, Object, Object> encoder = mock(IGraphKVEncoder.class);
+ when(encoder.getVertexEncoder()).thenReturn(mock(IVertexKVEncoder.class));
+ when(encoder.getEdgeEncoder()).thenReturn(mock(IEdgeKVEncoder.class));
+ return encoder;
+ }
+
+ private RocksdbClient mockClient() {
+ return mock(RocksdbClient.class);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@geaflow.apache.org
For additional commands, e-mail: commits-help@geaflow.apache.org