From c0d6bc2e7ee292406926e14937f7fa5c141d0583 Mon Sep 17 00:00:00 2001 From: yqz <2678785492@qq.com> Date: Sun, 2 Aug 2026 18:23:52 +0800 Subject: [PATCH] [ISSUE-796] Clean up PartitionType TODOs in rocksdb store module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../geaflow/store/rocksdb/PartitionType.java | 4 +- .../store/rocksdb/proxy/ProxyBuilder.java | 4 + .../store/rocksdb/PartitionTypeTest.java | 91 +++++++++++++++++++ 3 files changed, 97 insertions(+), 2 deletions(-) create mode 100644 geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/test/java/org/apache/geaflow/store/rocksdb/PartitionTypeTest.java 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 @@ // 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/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..99eb73712 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 @@ -40,6 +40,10 @@ public static IGraphRocksdbProxy build( return new SyncGraphLabelPartitionProxy<>(rocksdbClient, encoder, config); } else if (partitionType == PartitionType.DT) { return new SyncGraphDtPartitionProxy<>(rocksdbClient, encoder, config); + } else if (partitionType == PartitionType.DT_LABEL) { + throw new GeaflowRuntimeException( + "partition type DT_LABEL (partition by both timestamp and label) is not " + + "supported yet"); } throw new GeaflowRuntimeException("unexpected partition type: " + config.getString( RocksdbConfigKeys.ROCKSDB_GRAPH_STORE_PARTITION_TYPE)); 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..a9dd27385 --- /dev/null +++ b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-rocksdb/src/test/java/org/apache/geaflow/store/rocksdb/PartitionTypeTest.java @@ -0,0 +1,91 @@ +/* + * 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 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.IGraphKVEncoder; +import org.apache.geaflow.store.rocksdb.proxy.ProxyBuilder; +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) 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 {@code SyncGraphDtPartitionProxy}), so {@link ProxyBuilder} + * must dispatch it to a real proxy instead of rejecting it. With a {@code null} client the + * proxy constructor fails with a {@link NullPointerException}, which confirms the builder got + * past partition-type dispatch rather than throwing a {@link GeaflowRuntimeException}. + */ + @Test(expectedExceptions = NullPointerException.class) + public void testDtPartitionIsDispatchedToProxy() { + ProxyBuilder.build(newConfig("dt"), null, (IGraphKVEncoder) null); + } + + private Configuration newConfig(String partitionType) { + Map map = new HashMap<>(); + map.put(RocksdbConfigKeys.ROCKSDB_GRAPH_STORE_PARTITION_TYPE.getKey(), partitionType); + return new Configuration(map); + } +}