This is an automated email from the ASF dual-hosted git repository. kenhuuu pushed a commit to branch 3.7-dev in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 2a30be2ba3462048e659932d8c07b3d1935dfbbd Author: Yang Xia <[email protected]> AuthorDate: Mon Aug 31 17:16:21 2026 -0700 Isolate OLAP io() configuration per request OLAP io() now configures a per-request copy of the graph configuration rather than the shared, long-lived HadoopGraph configuration, so a request's reader/writer, file location, and with() options no longer persist on or race across the shared graph. Adds a protected VertexProgramStep.resolveComputeGraph seam that HadoopIoStep overrides to supply the request-local copy. Assisted-by: Claude Code:claude-opus-4-8 --- CHANGELOG.asciidoc | 1 + .../traversal/step/map/VertexProgramStep.java | 14 +- .../traversal/step/map/VertexProgramStepTest.java | 121 ++++++++++++++ .../traversal/step/sideEffect/HadoopIoStep.java | 26 ++++ .../step/sideEffect/HadoopIoStepTest.java | 173 +++++++++++++++++++++ 5 files changed, 333 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc index 7c639880af..49f4187c12 100644 --- a/CHANGELOG.asciidoc +++ b/CHANGELOG.asciidoc @@ -33,6 +33,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima * Changed GraphSON 1.0 typed deserialization to resolve a `@class` type id only when the name is in a set of exact class names. * Restricted Hadoop/Spark OLAP `io()` from a remotely submitted traversal to operator-approved reader/writer classes and `with()` keys, adding the `gremlin.io.trusted`, `gremlin.io.approvedClasses` and `gremlin.io.approvedGraphConfigKeys` options. * Restricted OLAP `GraphComputer.configure()` from a remotely submitted traversal to built-in and operator-approved keys, adding the `gremlin.io.approvedComputerConfigKeys` option. +* Fixed OLAP `io()` to configure each request against its own copy of the graph configuration, so a request's reader/writer, file location, and `with()` options no longer persist on or leak across the shared graph. * Fixed `subgraph()` to throw a descriptive error identifying the required `Edge` input instead of an internal `ClassCastException` when the traversal produces a non-edge value. * Fixed `where(P)` to throw a descriptive error identifying the required String scope key (and suggesting `is(P)` for value comparisons) instead of an internal `ClassCastException` when given a non-String predicate value. * Fixed `PeerPressure.property_name` in `gremlin-python` incorrectly mapping to the `pageRank` property name token. diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java index 77fbbe01a6..94446c5f30 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java @@ -63,14 +63,14 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com try { if (this.first && this.getPreviousStep() instanceof EmptyStep) { this.first = false; - final Graph graph = this.getTraversal().getGraph().get(); + final Graph graph = this.resolveComputeGraph(this.getTraversal().getGraph().get()); future = this.getComputer().apply(graph).program(this.generateProgram(graph, EmptyMemory.instance())).submit(); final ComputerResult result = future.get(); this.processMemorySideEffects(result.memory()); return this.getTraversal().getTraverserGenerator().generate(result, this, 1l); } else { final Traverser.Admin<ComputerResult> traverser = this.starts.next(); - final Graph graph = traverser.get().graph(); + final Graph graph = this.resolveComputeGraph(traverser.get().graph()); final Memory memory = traverser.get().memory(); future = this.getComputer().apply(graph).program(this.generateProgram(graph, memory)).submit(); final ComputerResult result = future.get(); @@ -106,6 +106,16 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com this.computer = computer; } + /** + * Resolves the {@link Graph} that this step's computer is applied to and its vertex program generated against for a + * single execution. The default returns the graph unchanged; a subclass may override it to run against a + * request-local graph (for example, one wrapping a copy of the configuration). The returned instance is used for + * both {@link #getComputer()}'s {@code apply(graph)} and {@link VertexComputing#generateProgram(Graph, Memory)}. + */ + protected Graph resolveComputeGraph(final Graph graph) { + return graph; + } + protected boolean previousTraversalVertexProgram() { Step<?, ?> currentStep = this; while (!(currentStep instanceof EmptyStep)) { diff --git a/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStepTest.java b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStepTest.java new file mode 100644 index 0000000000..efa198c300 --- /dev/null +++ b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStepTest.java @@ -0,0 +1,121 @@ +/* + * 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.tinkerpop.gremlin.process.computer.traversal.step.map; + +import org.apache.tinkerpop.gremlin.process.computer.Computer; +import org.apache.tinkerpop.gremlin.process.computer.ComputerResult; +import org.apache.tinkerpop.gremlin.process.computer.GraphComputer; +import org.apache.tinkerpop.gremlin.process.computer.Memory; +import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; +import org.apache.tinkerpop.gremlin.process.traversal.Traversal; +import org.apache.tinkerpop.gremlin.process.traversal.Traverser; +import org.apache.tinkerpop.gremlin.process.traversal.TraverserGenerator; +import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__; +import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph; +import org.junit.Test; + +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +import static org.junit.Assert.assertSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class VertexProgramStepTest { + + // The default seam is identity (returns the graph unchanged); the non-overriding subclasses rely on this to keep + // binding and generating against the graph they are handed. + @Test + public void shouldResolveComputeGraphToSameGraphByDefault() { + final Graph graph = EmptyGraph.instance(); + final VertexProgramStep step = new VertexProgramStep(__.start().asAdmin()) { + @Override + public VertexProgram generateProgram(final Graph graph, final Memory memory) { + return null; + } + }; + assertSame(graph, step.resolveComputeGraph(graph)); + } + + // The graph resolveComputeGraph returns must flow to BOTH getComputer().apply(graph) and generateProgram(graph, ...). + // Driving processNextStart with the seam returning a distinct instance, both the computer binding and the program + // generation must observe that same instance, so a reorder that binds the computer to a different graph than the + // one configured is caught here. + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void shouldBindComputerAndGenerateProgramToTheSameResolvedGraph() { + final Graph handed = mock(Graph.class); // what the traversal exposes + final Graph resolved = mock(Graph.class); // what the seam returns -- distinct, so we can tell them apart + + // Computer is final and cannot be mocked; use a real Computer.compute() whose apply(graph) calls graph.compute(). + // The GraphComputer is produced by the graph the computer is applied to, so stubbing compute() ONLY on `resolved` + // means the chain completes iff apply() bound to `resolved` -- and we verify that call directly. + final Memory memory = mock(Memory.class); + when(memory.keys()).thenReturn(Collections.emptySet()); + final ComputerResult result = mock(ComputerResult.class); + when(result.memory()).thenReturn(memory); + final GraphComputer graphComputer = mock(GraphComputer.class); + when(graphComputer.program(any())).thenReturn(graphComputer); + when(graphComputer.submit()).thenReturn(CompletableFuture.completedFuture(result)); + when(resolved.compute()).thenReturn(graphComputer); + final Computer computer = Computer.compute(); // graphComputerClass == GraphComputer.class -> apply calls graph.compute() + + // mock the traversal so no strategies/real graph are involved; getGraph() must be present (non-Empty) + final TraverserGenerator generator = mock(TraverserGenerator.class); + when(generator.generate(any(), any(), anyLong())).thenReturn(mock(Traverser.Admin.class)); + final Traversal.Admin traversal = mock(Traversal.Admin.class); + when(traversal.getTraverserSetSupplier()).thenReturn((java.util.function.Supplier) TraverserSet::new); + when(traversal.getGraph()).thenReturn(Optional.of(handed)); + when(traversal.getTraverserGenerator()).thenReturn(generator); + + final Graph[] generatedAgainst = new Graph[1]; + final VertexProgramStep step = new VertexProgramStep(traversal) { + @Override + public Computer getComputer() { + return computer; + } + + @Override + protected Graph resolveComputeGraph(final Graph graph) { + return resolved; + } + + @Override + public VertexProgram generateProgram(final Graph graph, final Memory memory) { + generatedAgainst[0] = graph; + return mock(VertexProgram.class); + } + }; + + step.processNextStart(); + + // the computer was applied to the resolved graph (apply() -> graph.compute() on `resolved`, never on `handed`) + verify(resolved).compute(); + verify(handed, never()).compute(); + // and the program was generated against that same resolved graph + assertSame("the program must be generated against the graph the seam resolved", resolved, generatedAgainst[0]); + } +} diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java index 78f01c69e3..a8f18ad191 100644 --- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java +++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStep.java @@ -19,6 +19,7 @@ package org.apache.tinkerpop.gremlin.hadoop.process.computer.traversal.step.sideEffect; import org.apache.tinkerpop.gremlin.hadoop.Constants; +import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph; import org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONInputFormat; import org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONOutputFormat; import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat; @@ -106,6 +107,31 @@ public class HadoopIoStep extends VertexProgramStep implements ReadWriting { return CloneVertexProgram.build().create(graph); } + /** + * Runs this OLAP {@code io()} operation against a request-local copy of the graph, so the reader/writer, + * input/output location, and {@code with(k,v)} options written while configuring the request are applied to that + * copy rather than to the shared {@link HadoopGraph} configuration. The base {@code processNextStart} resolves the + * graph through this seam once and uses it for both the computer binding and {@link #generateProgram(Graph, Memory)}. + */ + @Override + protected Graph resolveComputeGraph(final Graph graph) { + return isolateGraphConfiguration(graph); + } + + /** + * Returns a {@link HadoopGraph} whose configuration is an independent copy of {@code graph}'s, so configuring a + * request against it cannot mutate the shared graph. Fails closed: a non-{@link HadoopGraph} is rejected rather than + * returned unchanged (which would hand back the shared graph and defeat the isolation). Package-private for direct + * unit testing. + */ + static Graph isolateGraphConfiguration(final Graph graph) { + if (!(graph instanceof HadoopGraph)) + throw new IllegalStateException( + "OLAP io() requires a HadoopGraph to isolate its request configuration but received: " + + (null == graph ? "null" : graph.getClass().getName())); + return HadoopGraph.open(graph.configuration()); + } + @Override public HadoopIoStep clone() { return (HadoopIoStep) super.clone(); diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStepTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStepTest.java index 09c6043d27..8d4b5f9510 100644 --- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStepTest.java +++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/traversal/step/sideEffect/HadoopIoStepTest.java @@ -29,9 +29,21 @@ import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__; import org.apache.tinkerpop.gremlin.process.traversal.step.ReadWriting; import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapClassLoadingPolicy; import org.apache.tinkerpop.gremlin.hadoop.structure.io.util.OlapConfigKeyPolicy; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph; import org.junit.Test; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CyclicBarrier; +import java.util.function.BiConsumer; + import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -231,4 +243,165 @@ public class HadoopIoStepTest { assertEquals("org.provider.CustomOutputFormat", graph.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_WRITER)); } + + // Request isolation: io() must configure a per-request copy of the graph, never the shared, long-lived HadoopGraph + // configuration reused across requests. The following cover the isolation seam directly. + + @Test + public void shouldIsolateGraphConfigurationIntoADistinctInstance() { + final HadoopGraph shared = HadoopGraph.open(new BaseConfiguration()); + final Graph local = HadoopIoStep.isolateGraphConfiguration(shared); + assertNotSame("io() must run against a request-local graph, not the shared one", shared, local); + assertNotSame("the request-local graph must have its own configuration instance", + shared.configuration(), local.configuration()); + // mutating either configuration must not affect the other (isolation in both directions) + local.configuration().setProperty("only.on.local", "v"); + assertFalse(shared.configuration().containsKey("only.on.local")); + shared.configuration().setProperty("only.on.shared", "v"); + assertFalse(local.configuration().containsKey("only.on.shared")); + } + + @Test + public void shouldCarryPristineKeysIntoIsolatedConfiguration() { + final Configuration config = new BaseConfiguration(); + config.setProperty(OlapClassLoadingPolicy.TRUSTED, true); + config.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, "org.provider.CustomInputFormat"); + final HadoopGraph shared = HadoopGraph.open(config); + final Graph local = HadoopIoStep.isolateGraphConfiguration(shared); + // the copy carries every operator key forward, so trust and approved-format seeding stay intact + assertEquals(true, local.configuration().getBoolean(OlapClassLoadingPolicy.TRUSTED)); + assertEquals("org.provider.CustomInputFormat", + local.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER)); + } + + @Test + public void shouldReturnAFreshInstanceForEachIsolationCall() { + // each call yields a brand-new request-local graph/config (no caching or reuse); a structural guard, not + // concurrency coverage -- see shouldIsolateConcurrentRequestsFromEachOther. + final HadoopGraph shared = HadoopGraph.open(new BaseConfiguration()); + final Graph a = HadoopIoStep.isolateGraphConfiguration(shared); + final Graph b = HadoopIoStep.isolateGraphConfiguration(shared); + assertNotSame(a, b); + assertNotSame(a.configuration(), b.configuration()); + } + + @Test + public void shouldIsolateConcurrentRequestsFromEachOther() throws Exception { + // Two io() requests configuring against the SAME shared graph on two threads at once must each see only their + // own reader/input location, and must leave the shared graph unmutated. Isolation is structural (each request + // configures its own copy), so this is deterministic under any interleaving; the barrier forces the two + // requests to configure concurrently, exercising concurrent reads of the shared configuration. + final HadoopGraph shared = HadoopGraph.open(new BaseConfiguration()); + final CyclicBarrier barrier = new CyclicBarrier(2); + final Map<String, String> readerByThread = new ConcurrentHashMap<>(); + final Map<String, String> locationByThread = new ConcurrentHashMap<>(); + final List<Throwable> errors = Collections.synchronizedList(new ArrayList<>()); + + final BiConsumer<String, String> configureRequest = (name, file) -> { + try { + final HadoopIoStep step = new HadoopIoStep(__.start().asAdmin(), file); + step.setMode(ReadWriting.Mode.READING); + barrier.await(); // release both threads together so they configure concurrently + final Graph local = step.resolveComputeGraph(shared); + step.generateProgram(local, null); + readerByThread.put(name, local.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER)); + locationByThread.put(name, local.configuration().getString(Constants.GREMLIN_HADOOP_INPUT_LOCATION)); + } catch (final Throwable t) { + errors.add(t); + } + }; + + final Thread a = new Thread(() -> configureRequest.accept("A", "a.kryo")); + final Thread b = new Thread(() -> configureRequest.accept("B", "b.json")); + a.start(); + b.start(); + a.join(); + b.join(); + + assertTrue("no request should error: " + errors, errors.isEmpty()); + // each request observed only its own reader + input location (Gryo/a.kryo for A, GraphSON/b.json for B) + assertEquals(GryoInputFormatName(), readerByThread.get("A")); + assertEquals("a.kryo", locationByThread.get("A")); + assertEquals(GraphSONInputFormat.class.getName(), readerByThread.get("B")); + assertEquals("b.json", locationByThread.get("B")); + // and neither concurrent request mutated the shared, long-lived graph configuration + assertFalse(shared.configuration().containsKey(Constants.GREMLIN_HADOOP_GRAPH_READER)); + assertFalse(shared.configuration().containsKey(Constants.GREMLIN_HADOOP_INPUT_LOCATION)); + } + + @Test + public void shouldNotMutateSharedConfigurationWhenConfiguringAnIsolatedCopy() { + // the fix: generateProgram writes onto the request-local copy resolveComputeGraph hands it, leaving the shared + // graph configuration untouched, so a later request inherits none of this request's reader/input location. + final HadoopGraph shared = HadoopGraph.open(new BaseConfiguration()); + final HadoopIoStep step = new HadoopIoStep(__.start().asAdmin(), "graph.kryo"); + step.setMode(ReadWriting.Mode.READING); + + final Graph local = step.resolveComputeGraph(shared); + step.generateProgram(local, null); + + // the request-local copy carries the request's settings ... + assertEquals(GryoInputFormatName(), local.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER)); + assertEquals("graph.kryo", local.configuration().getString(Constants.GREMLIN_HADOOP_INPUT_LOCATION)); + // ... while the shared, long-lived configuration is left unmutated + assertFalse(shared.configuration().containsKey(Constants.GREMLIN_HADOOP_GRAPH_READER)); + assertFalse(shared.configuration().containsKey(Constants.GREMLIN_HADOOP_INPUT_LOCATION)); + } + + @Test + public void shouldApplyApprovedWithKeyToTheRequestLocalCopyOnly() { + // an operator-approved with() key is applied to the request-local copy and must not leak onto the shared graph + final Configuration config = new BaseConfiguration(); + config.setProperty(OlapConfigKeyPolicy.APPROVED_GRAPH_CONFIG_KEYS, "my.graph.option"); + final HadoopGraph shared = HadoopGraph.open(config); + final HadoopIoStep step = new HadoopIoStep(__.start().asAdmin(), "graph.kryo"); + step.setMode(ReadWriting.Mode.READING); + step.configure("my.graph.option", "v"); + + final Graph local = step.resolveComputeGraph(shared); + step.generateProgram(local, null); + + // the value landed on the request-local copy ... + assertEquals("v", local.configuration().getString("my.graph.option")); + // ... but the shared graph never received it + assertFalse(shared.configuration().containsKey("my.graph.option")); + } + + @Test + public void shouldLeaveSharedConfigurationCleanWhenARequestFailsPartway() { + // configureForRead writes the reader/input location to the graph before addParametersToConfiguration rejects an + // unapproved with() key, so the request fails partway with those already written -- onto the request-local copy. + final HadoopGraph shared = HadoopGraph.open(new BaseConfiguration()); + final HadoopIoStep step = new HadoopIoStep(__.start().asAdmin(), "graph.kryo"); + step.setMode(ReadWriting.Mode.READING); + step.configure("unapproved.key", "v"); + final Graph local = step.resolveComputeGraph(shared); + try { + step.generateProgram(local, null); + fail("an unapproved with() key must fail the request"); + } catch (final IllegalArgumentException expected) { + // expected: the request failed after the reader/input location were already written to the local copy + } + assertEquals("the partial write must have landed on the request-local copy", GryoInputFormatName(), + local.configuration().getString(Constants.GREMLIN_HADOOP_GRAPH_READER)); + // ... and none of the failed request's partial mutations reached the shared graph + assertFalse(shared.configuration().containsKey(Constants.GREMLIN_HADOOP_GRAPH_READER)); + assertFalse(shared.configuration().containsKey(Constants.GREMLIN_HADOOP_INPUT_LOCATION)); + } + + @Test + public void shouldFailClosedWhenIsolatingANonHadoopGraph() { + // a security-isolation primitive must fail closed: a non-HadoopGraph must be rejected, never returned unchanged + // (which would silently hand back the shared, long-lived graph and reinstate the cross-request leak). + try { + HadoopIoStep.isolateGraphConfiguration(EmptyGraph.instance()); + fail("request isolation must reject a non-HadoopGraph rather than silently returning the shared graph"); + } catch (final IllegalStateException ise) { + assertTrue(ise.getMessage(), ise.getMessage().contains("HadoopGraph")); + } + } + + private static String GryoInputFormatName() { + return org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat.class.getName(); + } }
