GryoRegistrar is the new name for TinkerPopKryoRegistrar. Added GryoPoolKryoShimSerivce for use by non-Hadoop projects. Updated VertexProgramHelper's serizlizer to use the KryoShimService. Few issues there, need to talk to @dalaro. Only thing failing are some traversal interruption exception tests... they were working prior to GryoPoolKryoShimService usage.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/4a185a37 Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/4a185a37 Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/4a185a37 Branch: refs/heads/TINKERPOP-1321 Commit: 4a185a3765ed5c19e1d7d14413a33c11fc2e0d29 Parents: 55eb1e6 ebd2f85 Author: Marko A. Rodriguez <okramma...@gmail.com> Authored: Mon Jun 6 10:31:01 2016 -0600 Committer: Marko A. Rodriguez <okramma...@gmail.com> Committed: Mon Jun 6 10:31:01 2016 -0600 ---------------------------------------------------------------------- .../process/computer/GiraphWorkerContext.java | 3 +- .../computer/util/VertexProgramHelper.java | 30 +- .../gremlin/structure/io/gryo/GryoMapper.java | 447 ++++++++++++------- .../gremlin/structure/io/gryo/GryoPool.java | 1 + .../structure/io/gryo/GryoSerializers.java | 40 +- .../structure/io/gryo/JavaTimeSerializers.java | 125 ++---- .../structure/io/gryo/PairSerializer.java | 11 +- .../structure/io/gryo/TypeRegistration.java | 82 ++++ .../structure/io/gryo/URISerializer.java | 23 +- .../structure/io/gryo/UUIDSerializer.java | 24 +- .../gryo/kryoshim/GryoPoolKryoShimService.java | 65 +++ .../structure/io/gryo/kryoshim/InputShim.java | 38 ++ .../structure/io/gryo/kryoshim/KryoShim.java | 41 ++ .../io/gryo/kryoshim/KryoShimService.java | 99 ++++ .../io/gryo/kryoshim/KryoShimServiceLoader.java | 214 +++++++++ .../structure/io/gryo/kryoshim/OutputShim.java | 40 ++ .../io/gryo/kryoshim/SerializerShim.java | 36 ++ .../io/gryo/kryoshim/package-info.java | 55 +++ .../kryoshim/shaded/ShadedInputAdapter.java | 66 +++ .../gryo/kryoshim/shaded/ShadedKryoAdapter.java | 67 +++ .../kryoshim/shaded/ShadedOutputAdapter.java | 72 +++ .../shaded/ShadedSerializerAdapter.java | 54 +++ .../io/gryo/kryoshim/shaded/package-info.java | 25 ++ .../util/star/StarGraphGryoSerializer.java | 130 +----- .../util/star/StarGraphSerializer.java | 150 +++++++ ...n.structure.io.gryo.kryoshim.KryoShimService | 1 + .../hadoop/process/computer/HadoopCombine.java | 3 +- .../hadoop/process/computer/HadoopMap.java | 3 +- .../hadoop/process/computer/HadoopReduce.java | 3 +- .../structure/io/HadoopPoolShimService.java | 76 ++++ .../hadoop/structure/io/HadoopPools.java | 1 + .../structure/io/HadoopPoolsConfigurable.java | 4 +- .../hadoop/structure/io/ObjectWritable.java | 25 +- .../hadoop/structure/io/VertexWritable.java | 39 +- .../structure/io/gryo/GryoRecordReader.java | 3 +- .../structure/io/gryo/GryoRecordWriter.java | 4 +- ...n.structure.io.gryo.kryoshim.KryoShimService | 1 + .../spark/process/computer/SparkExecutor.java | 3 +- .../structure/io/gryo/GryoRegistrator.java | 194 ++++++++ .../spark/structure/io/gryo/GryoSerializer.java | 2 +- .../io/gryo/IoRegistryAwareKryoSerializer.java | 116 +++++ .../io/gryo/ObjectWritableSerializer.java | 12 +- .../io/gryo/VertexWritableSerializer.java | 12 +- .../kryoshim/unshaded/UnshadedInputAdapter.java | 78 ++++ .../kryoshim/unshaded/UnshadedKryoAdapter.java | 74 +++ .../unshaded/UnshadedKryoShimService.java | 154 +++++++ .../unshaded/UnshadedOutputAdapter.java | 83 ++++ .../unshaded/UnshadedSerializerAdapter.java | 57 +++ ...n.structure.io.gryo.kryoshim.KryoShimService | 1 + .../spark/structure/io/ToyGraphInputRDD.java | 3 +- 50 files changed, 2403 insertions(+), 487 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4a185a37/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/VertexProgramHelper.java ---------------------------------------------------------------------- diff --cc gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/VertexProgramHelper.java index bc67866,bc67866..74bfd2e --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/VertexProgramHelper.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/VertexProgramHelper.java @@@ -25,8 -25,8 +25,14 @@@ import org.apache.tinkerpop.gremlin.pro import org.apache.tinkerpop.gremlin.process.traversal.Traversal; import org.apache.tinkerpop.gremlin.process.traversal.step.map.EdgeVertexStep; import org.apache.tinkerpop.gremlin.process.traversal.step.map.VertexStep; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.GryoPoolKryoShimService; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShim; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimServiceLoader; import org.apache.tinkerpop.gremlin.util.Serializer; ++import java.io.ByteArrayInputStream; ++import java.io.ByteArrayOutputStream; import java.io.IOException; import java.util.Arrays; import java.util.HashSet; @@@ -63,25 -63,25 +69,19 @@@ public final class VertexProgramHelper public static void serialize(final Object object, final Configuration configuration, final String key) { if (configuration instanceof AbstractConfiguration) ((AbstractConfiguration) configuration).setDelimiterParsingDisabled(true); -- try { -- final String byteString = Arrays.toString(Serializer.serializeObject(object)); -- configuration.setProperty(key, byteString.substring(1, byteString.length() - 1)); -- } catch (final IOException e) { -- throw new IllegalArgumentException(e.getMessage(), e); -- } ++ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); ++ new GryoPoolKryoShimService().writeClassAndObject(object, outputStream); ++ final String byteString = Arrays.toString(outputStream.toByteArray()); ++ configuration.setProperty(key, byteString.substring(1, byteString.length() - 1)); } public static <T> T deserialize(final Configuration configuration, final String key) { -- try { -- final String[] stringBytes = configuration.getString(key).split(","); -- byte[] bytes = new byte[stringBytes.length]; -- for (int i = 0; i < stringBytes.length; i++) { -- bytes[i] = Byte.valueOf(stringBytes[i].trim()); -- } -- return (T) Serializer.deserializeObject(bytes); -- } catch (final IOException | ClassNotFoundException e) { -- throw new IllegalArgumentException(e.getMessage(), e); ++ final String[] stringBytes = configuration.getString(key).split(","); ++ byte[] bytes = new byte[stringBytes.length]; ++ for (int i = 0; i < stringBytes.length; i++) { ++ bytes[i] = Byte.valueOf(stringBytes[i].trim()); } ++ return (T) new GryoPoolKryoShimService().readClassAndObject(new ByteArrayInputStream(bytes)); } public static <S, E> Traversal.Admin<S, E> reverse(final Traversal.Admin<S, E> traversal) { http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4a185a37/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java ---------------------------------------------------------------------- diff --cc gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java index 9cae845,41ca44d..8c5fdcd --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/GryoMapper.java @@@ -43,10 -45,10 +45,7 @@@ import org.apache.tinkerpop.gremlin.pro import org.apache.tinkerpop.gremlin.process.traversal.traverser.O_OB_S_SE_SL_Traverser; import org.apache.tinkerpop.gremlin.process.traversal.traverser.O_Traverser; import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet; --import org.apache.tinkerpop.gremlin.process.traversal.util.DefaultTraversalMetrics; --import org.apache.tinkerpop.gremlin.process.traversal.util.ImmutableMetrics; --import org.apache.tinkerpop.gremlin.process.traversal.util.MutableMetrics; --import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalExplanation; ++import org.apache.tinkerpop.gremlin.process.traversal.util.*; import org.apache.tinkerpop.gremlin.structure.Direction; import org.apache.tinkerpop.gremlin.structure.Edge; import org.apache.tinkerpop.gremlin.structure.Graph; @@@ -244,119 -243,119 +240,121 @@@ public final class GryoMapper implement * Note that the following are pre-registered boolean, Boolean, byte, Byte, char, Character, double, Double, * int, Integer, float, Float, long, Long, short, Short, String, void. */ - private final List<Triplet<Class, Function<Kryo, Serializer>, Integer>> serializationList = new ArrayList<Triplet<Class, Function<Kryo, Serializer>, Integer>>() {{ - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(byte[].class, null, 25)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(char[].class, null, 26)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(short[].class, null, 27)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(int[].class, null, 28)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(long[].class, null, 29)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(float[].class, null, 30)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(double[].class, null, 31)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(String[].class, null, 32)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Object[].class, null, 33)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ArrayList.class, null, 10)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(BigInteger.class, null, 34)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(BigDecimal.class, null, 35)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Calendar.class, null, 39)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Class.class, null, 41)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Collection.class, null, 37)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Collections.EMPTY_LIST.getClass(), null, 51)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Collections.EMPTY_MAP.getClass(), null, 52)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Collections.EMPTY_SET.getClass(), null, 53)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Collections.singleton(null).getClass(), null, 54)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Collections.singletonList(null).getClass(), null, 24)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Collections.singletonMap(null, null).getClass(), null, 23)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Contains.class, null, 49)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Currency.class, null, 40)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Date.class, null, 38)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Direction.class, null, 12)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(DetachedEdge.class, null, 21)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(DetachedVertexProperty.class, null, 20)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(DetachedProperty.class, null, 18)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(DetachedVertex.class, null, 19)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(DetachedPath.class, null, 60)); + private final List<TypeRegistration<?>> typeRegistrations = new ArrayList<TypeRegistration<?>>() {{ + add(GryoTypeReg.of(byte[].class, 25)); + add(GryoTypeReg.of(char[].class, 26)); + add(GryoTypeReg.of(short[].class, 27)); + add(GryoTypeReg.of(int[].class, 28)); + add(GryoTypeReg.of(long[].class, 29)); + add(GryoTypeReg.of(float[].class, 30)); + add(GryoTypeReg.of(double[].class, 31)); + add(GryoTypeReg.of(String[].class, 32)); + add(GryoTypeReg.of(Object[].class, 33)); + add(GryoTypeReg.of(ArrayList.class, 10)); + add(GryoTypeReg.of(BigInteger.class, 34)); + add(GryoTypeReg.of(BigDecimal.class, 35)); + add(GryoTypeReg.of(Calendar.class, 39)); + add(GryoTypeReg.of(Class.class, 41)); + add(GryoTypeReg.of(Collection.class, 37)); + add(GryoTypeReg.of(Collections.EMPTY_LIST.getClass(), 51)); + add(GryoTypeReg.of(Collections.EMPTY_MAP.getClass(), 52)); + add(GryoTypeReg.of(Collections.EMPTY_SET.getClass(), 53)); + add(GryoTypeReg.of(Collections.singleton(null).getClass(), 54)); + add(GryoTypeReg.of(Collections.singletonList(null).getClass(), 24)); + add(GryoTypeReg.of(Collections.singletonMap(null, null).getClass(), 23)); + add(GryoTypeReg.of(Contains.class, 49)); + add(GryoTypeReg.of(Currency.class, 40)); + add(GryoTypeReg.of(Date.class, 38)); + add(GryoTypeReg.of(Direction.class, 12)); + add(GryoTypeReg.of(DetachedEdge.class, 21)); + add(GryoTypeReg.of(DetachedVertexProperty.class, 20)); + add(GryoTypeReg.of(DetachedProperty.class, 18)); + add(GryoTypeReg.of(DetachedVertex.class, 19)); + add(GryoTypeReg.of(DetachedPath.class, 60)); // skip 14 - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(EnumSet.class, null, 46)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(HashMap.class, null, 11)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(HashMap.Entry.class, null, 16)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(HASH_MAP_NODE, null, 92)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(KryoSerializable.class, null, 36)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LinkedHashMap.class, null, 47)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LinkedHashSet.class, null, 71)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LinkedList.class, null, 116)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LINKED_HASH_MAP_ENTRY_CLASS, null, 15)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Locale.class, null, 22)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(StringBuffer.class, null, 43)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(StringBuilder.class, null, 44)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(T.class, null, 48)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(TimeZone.class, null, 42)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(TreeMap.class, null, 45)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(TreeSet.class, null, 50)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(UUID.class, kryo -> new UUIDSerializer(), 17)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(URI.class, kryo -> new URISerializer(), 72)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(VertexTerminator.class, null, 13)); - - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferenceEdge.class, null, 81)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferenceVertexProperty.class, null, 82)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferenceProperty.class, null, 83)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferenceVertex.class, null, 84)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ReferencePath.class, null, 85)); - - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(StarGraph.class, kryo -> StarGraphGryoSerializer.with(Direction.BOTH), 86)); - - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Edge.class, kryo -> new GryoSerializers.EdgeSerializer(), 65)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Vertex.class, kryo -> new GryoSerializers.VertexSerializer(), 66)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Property.class, kryo -> new GryoSerializers.PropertySerializer(), 67)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(VertexProperty.class, kryo -> new GryoSerializers.VertexPropertySerializer(), 68)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Path.class, kryo -> new GryoSerializers.PathSerializer(), 59)); + add(GryoTypeReg.of(EnumSet.class, 46)); + add(GryoTypeReg.of(HashMap.class, 11)); + add(GryoTypeReg.of(HashMap.Entry.class, 16)); + add(GryoTypeReg.of(HASH_MAP_NODE, 92)); + add(GryoTypeReg.of(KryoSerializable.class, 36)); + add(GryoTypeReg.of(LinkedHashMap.class, 47)); + add(GryoTypeReg.of(LinkedHashSet.class, 71)); + add(GryoTypeReg.of(LinkedList.class, 116)); + add(GryoTypeReg.of(LINKED_HASH_MAP_ENTRY_CLASS, 15)); + add(GryoTypeReg.of(Locale.class, 22)); + add(GryoTypeReg.of(StringBuffer.class, 43)); + add(GryoTypeReg.of(StringBuilder.class, 44)); + add(GryoTypeReg.of(T.class, 48)); + add(GryoTypeReg.of(TimeZone.class, 42)); + add(GryoTypeReg.of(TreeMap.class, 45)); + add(GryoTypeReg.of(TreeSet.class, 50)); + add(GryoTypeReg.of(UUID.class, 17, new UUIDSerializer())); + add(GryoTypeReg.of(URI.class, 72, new URISerializer())); + add(GryoTypeReg.of(VertexTerminator.class, 13)); + + add(GryoTypeReg.of(ReferenceEdge.class, 81)); + add(GryoTypeReg.of(ReferenceVertexProperty.class, 82)); + add(GryoTypeReg.of(ReferenceProperty.class, 83)); + add(GryoTypeReg.of(ReferenceVertex.class, 84)); + add(GryoTypeReg.of(ReferencePath.class, 85)); + + add(GryoTypeReg.of(StarGraph.class, 86, new StarGraphSerializer(Direction.BOTH, new GraphFilter()))); + + add(GryoTypeReg.of(Edge.class, 65, new GryoSerializers.EdgeSerializer())); + add(GryoTypeReg.of(Vertex.class, 66, new GryoSerializers.VertexSerializer())); + add(GryoTypeReg.of(Property.class, 67, new GryoSerializers.PropertySerializer())); + add(GryoTypeReg.of(VertexProperty.class, 68, new GryoSerializers.VertexPropertySerializer())); + add(GryoTypeReg.of(Path.class, 59, new GryoSerializers.PathSerializer())); // skip 55 - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(B_O_Traverser.class, null, 75)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(O_Traverser.class, null, 76)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(B_LP_O_P_S_SE_SL_Traverser.class, null, 77)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(B_O_S_SE_SL_Traverser.class, null, 78)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(B_LP_O_S_SE_SL_Traverser.class, null, 87)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(O_OB_S_SE_SL_Traverser.class, null, 89)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LP_O_OB_S_SE_SL_Traverser.class, null, 90)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LP_O_OB_P_S_SE_SL_Traverser.class, null, 91)); - - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(TraverserSet.class, null, 58)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Tree.class, null, 61)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(HashSet.class, null, 62)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(BulkSet.class, null, 64)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(MutableMetrics.class, null, 69)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ImmutableMetrics.class, null, 115)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(DefaultTraversalMetrics.class, null, 70)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(MapMemory.class, null, 73)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(MapReduce.NullObject.class, null, 74)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(AtomicLong.class, null, 79)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Pair.class, kryo -> new PairSerializer(), 88)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(TraversalExplanation.class, kryo -> new JavaSerializer(), 106)); - - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Duration.class, kryo -> new JavaTimeSerializers.DurationSerializer(), 93)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Instant.class, kryo -> new JavaTimeSerializers.InstantSerializer(), 94)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LocalDate.class, kryo -> new JavaTimeSerializers.LocalDateSerializer(), 95)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LocalDateTime.class, kryo -> new JavaTimeSerializers.LocalDateTimeSerializer(), 96)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(LocalTime.class, kryo -> new JavaTimeSerializers.LocalTimeSerializer(), 97)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(MonthDay.class, kryo -> new JavaTimeSerializers.MonthDaySerializer(), 98)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(OffsetDateTime.class, kryo -> new JavaTimeSerializers.OffsetDateTimeSerializer(), 99)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(OffsetTime.class, kryo -> new JavaTimeSerializers.OffsetTimeSerializer(), 100)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Period.class, kryo -> new JavaTimeSerializers.PeriodSerializer(), 101)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Year.class, kryo -> new JavaTimeSerializers.YearSerializer(), 102)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(YearMonth.class, kryo -> new JavaTimeSerializers.YearMonthSerializer(), 103)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ZonedDateTime.class, kryo -> new JavaTimeSerializers.ZonedDateTimeSerializer(), 104)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ZoneOffset.class, kryo -> new JavaTimeSerializers.ZoneOffsetSerializer(), 105)); - - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(Operator.class, null, 107)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(FoldStep.FoldBiOperator.class, null, 108)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(GroupCountStep.GroupCountBiOperator.class, null, 109)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(GroupStep.GroupBiOperator.class, kryo -> new JavaSerializer(), 117)); // because they contain traversals - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(MeanGlobalStep.MeanGlobalBiOperator.class, null, 110)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(MeanGlobalStep.MeanNumber.class, null, 111)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(TreeStep.TreeBiOperator.class, null, 112)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(GroupStepV3d0.GroupBiOperatorV3d0.class, null, 113)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(RangeGlobalStep.RangeBiOperator.class, null, 114)); - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(OrderGlobalStep.OrderBiOperator.class, kryo -> new JavaSerializer(), 118)); // because they contain traversals - add(Triplet.<Class, Function<Kryo, Serializer>, Integer>with(ProfileStep.ProfileBiOperator.class, null, 119)); // ***LAST ID*** + add(GryoTypeReg.of(B_O_Traverser.class, 75)); + add(GryoTypeReg.of(O_Traverser.class, 76)); + add(GryoTypeReg.of(B_LP_O_P_S_SE_SL_Traverser.class, 77)); + add(GryoTypeReg.of(B_O_S_SE_SL_Traverser.class, 78)); + add(GryoTypeReg.of(B_LP_O_S_SE_SL_Traverser.class, 87)); + add(GryoTypeReg.of(O_OB_S_SE_SL_Traverser.class, 89)); + add(GryoTypeReg.of(LP_O_OB_S_SE_SL_Traverser.class, 90)); + add(GryoTypeReg.of(LP_O_OB_P_S_SE_SL_Traverser.class, 91)); + + add(GryoTypeReg.of(TraverserSet.class, 58)); + add(GryoTypeReg.of(Tree.class, 61)); + add(GryoTypeReg.of(HashSet.class, 62)); + add(GryoTypeReg.of(BulkSet.class, 64)); + add(GryoTypeReg.of(MutableMetrics.class, 69)); + add(GryoTypeReg.of(ImmutableMetrics.class, 115)); + add(GryoTypeReg.of(DefaultTraversalMetrics.class, 70)); + add(GryoTypeReg.of(MapMemory.class, 73)); + add(GryoTypeReg.of(MapReduce.NullObject.class, 74)); + add(GryoTypeReg.of(AtomicLong.class, 79)); + add(GryoTypeReg.of(Pair.class, 88, new PairSerializer())); + add(GryoTypeReg.of(TraversalExplanation.class, 106, new JavaSerializer())); ++ add(GryoTypeReg.of(PureTraversal.class, 120, new JavaSerializer())); ++ add(GryoTypeReg.of(GraphFilter.class, 121, new JavaSerializer())); // ***LAST ID*** + + add(GryoTypeReg.of(Duration.class, 93, new JavaTimeSerializers.DurationSerializer())); + add(GryoTypeReg.of(Instant.class, 94, new JavaTimeSerializers.InstantSerializer())); + add(GryoTypeReg.of(LocalDate.class, 95, new JavaTimeSerializers.LocalDateSerializer())); + add(GryoTypeReg.of(LocalDateTime.class, 96, new JavaTimeSerializers.LocalDateTimeSerializer())); + add(GryoTypeReg.of(LocalTime.class, 97, new JavaTimeSerializers.LocalTimeSerializer())); + add(GryoTypeReg.of(MonthDay.class, 98, new JavaTimeSerializers.MonthDaySerializer())); + add(GryoTypeReg.of(OffsetDateTime.class, 99, new JavaTimeSerializers.OffsetDateTimeSerializer())); + add(GryoTypeReg.of(OffsetTime.class, 100, new JavaTimeSerializers.OffsetTimeSerializer())); + add(GryoTypeReg.of(Period.class, 101, new JavaTimeSerializers.PeriodSerializer())); + add(GryoTypeReg.of(Year.class, 102, new JavaTimeSerializers.YearSerializer())); + add(GryoTypeReg.of(YearMonth.class, 103, new JavaTimeSerializers.YearMonthSerializer())); + add(GryoTypeReg.of(ZonedDateTime.class, 104, new JavaTimeSerializers.ZonedDateTimeSerializer())); + add(GryoTypeReg.of(ZoneOffset.class, 105, new JavaTimeSerializers.ZoneOffsetSerializer())); + + add(GryoTypeReg.of(Operator.class, 107)); + add(GryoTypeReg.of(FoldStep.FoldBiOperator.class, 108)); + add(GryoTypeReg.of(GroupCountStep.GroupCountBiOperator.class, 109)); + add(GryoTypeReg.of(GroupStep.GroupBiOperator.class, 117, new JavaSerializer())); // because they contain traversals + add(GryoTypeReg.of(MeanGlobalStep.MeanGlobalBiOperator.class, 110)); + add(GryoTypeReg.of(MeanGlobalStep.MeanNumber.class, 111)); + add(GryoTypeReg.of(TreeStep.TreeBiOperator.class, 112)); + add(GryoTypeReg.of(GroupStepV3d0.GroupBiOperatorV3d0.class, 113)); + add(GryoTypeReg.of(RangeGlobalStep.RangeBiOperator.class, 114)); + add(GryoTypeReg.of(OrderGlobalStep.OrderBiOperator.class, 118, new JavaSerializer())); // because they contain traversals - add(GryoTypeReg.of(ProfileStep.ProfileBiOperator.class, 119)); // ***LAST ID*** ++ add(GryoTypeReg.of(ProfileStep.ProfileBiOperator.class, 119)); }}; private final List<IoRegistry> registries = new ArrayList<>(); http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4a185a37/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/kryoshim/GryoPoolKryoShimService.java ---------------------------------------------------------------------- diff --cc gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/kryoshim/GryoPoolKryoShimService.java index 0000000,0000000..526b702 new file mode 100644 --- /dev/null +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/gryo/kryoshim/GryoPoolKryoShimService.java @@@ -1,0 -1,0 +1,65 @@@ ++/* ++ * 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.structure.io.gryo.kryoshim; ++ ++import org.apache.commons.configuration.BaseConfiguration; ++import org.apache.commons.configuration.Configuration; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoPool; ++import org.apache.tinkerpop.shaded.kryo.io.Input; ++ ++import java.io.InputStream; ++import java.io.OutputStream; ++import java.util.Collections; ++ ++/** ++ * @author Marko A. Rodriguez (http://markorodriguez.com) ++ */ ++public class GryoPoolKryoShimService implements KryoShimService { ++ ++ private static GryoPool GRYO_POOL = null; ++ ++ public Object readClassAndObject(final InputStream inputStream) { ++ applyConfiguration(new BaseConfiguration()); ++ return GRYO_POOL.readWithKryo(reader -> reader.readClassAndObject(new Input(inputStream))); ++ } ++ ++ @Override ++ public void writeClassAndObject(final Object object, final OutputStream outputStream) { ++ applyConfiguration(new BaseConfiguration()); ++ GRYO_POOL.doWithWriter(writer -> writer.writeObject(outputStream, object)); ++ ++ } ++ ++ @Override ++ public int getPriority() { ++ return 1; ++ } ++ ++ @Override ++ public void applyConfiguration(final Configuration configuration) { ++ if (null == GRYO_POOL) { ++ GRYO_POOL = GryoPool.build(). ++ poolSize(configuration.getInt(GryoPool.CONFIG_IO_GRYO_POOL_SIZE, 256)). ++ ioRegistries(configuration.getList(GryoPool.CONFIG_IO_REGISTRY, Collections.emptyList())). ++ initializeMapper(m -> m.registrationRequired(false)). ++ create(); ++ } ++ } ++} http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4a185a37/gremlin-core/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService ---------------------------------------------------------------------- diff --cc gremlin-core/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService index 0000000,0000000..5045e05 new file mode 100644 --- /dev/null +++ b/gremlin-core/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService @@@ -1,0 -1,0 +1,1 @@@ ++org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.GryoPoolKryoShimService http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/4a185a37/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/GryoRegistrator.java ---------------------------------------------------------------------- diff --cc spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/GryoRegistrator.java index 0000000,0000000..1b98284 new file mode 100644 --- /dev/null +++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/GryoRegistrator.java @@@ -1,0 -1,0 +1,194 @@@ ++/* ++ * 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.spark.structure.io.gryo; ++ ++import com.esotericsoftware.kryo.Kryo; ++import com.esotericsoftware.kryo.Serializer; ++import com.esotericsoftware.kryo.serializers.JavaSerializer; ++import org.apache.spark.serializer.KryoRegistrator; ++import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable; ++import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable; ++import org.apache.tinkerpop.gremlin.spark.process.computer.payload.MessagePayload; ++import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewIncomingPayload; ++import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewOutgoingPayload; ++import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewPayload; ++import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.kryoshim.unshaded.UnshadedSerializerAdapter; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoMapper; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.TypeRegistration; ++import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.SerializerShim; ++import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph; ++import org.slf4j.Logger; ++import org.slf4j.LoggerFactory; ++ ++import java.util.*; ++ ++/** ++ * A spark.kryo.registrator implementation that installs TinkerPop types. ++ * This is intended for use with spark.serializer=KryoSerializer, not GryoSerializer. ++ */ ++public class GryoRegistrator implements KryoRegistrator { ++ ++ private static final Logger log = LoggerFactory.getLogger(GryoRegistrator.class); ++ ++ @Override ++ public void registerClasses(Kryo kryo) { ++ registerClasses(kryo, Collections.emptyMap(), Collections.emptySet()); ++ } ++ ++ /** ++ * Register TinkerPop's classes with the supplied {@link Kryo} instance ++ * while honoring optional overrides and optional class blacklist ("blackset"?). ++ * ++ * @param kryo the Kryo serializer instance with which to register types ++ * @param serializerOverrides serializer mappings that override this class's defaults ++ * @param blacklist classes which should not be registered at all, even if there is an override entry ++ * or if they would be registered by this class by default (does not affect Kryo's ++ * built-in registrations, e.g. String.class). ++ */ ++ public void registerClasses(Kryo kryo, Map<Class<?>, Serializer<?>> serializerOverrides, Set<Class<?>> blacklist) { ++ // Apply TinkerPop type registrations copied from GyroSerializer's constructor ++ for (Map.Entry<Class<?>, Serializer<?>> ent : getExtraRegistrations().entrySet()) { ++ Class<?> targetClass = ent.getKey(); ++ Serializer<?> ser = ent.getValue(); ++ ++ // Is this class blacklisted? Skip it. (takes precedence over serializerOverrides) ++ if (blacklist.contains(targetClass)) { ++ log.debug("Not registering serializer for {} (blacklisted)", targetClass); ++ continue; ++ } ++ ++ if (checkForAndApplySerializerOverride(serializerOverrides, kryo, targetClass)) { ++ // do nothing but skip the remaining else(-if) clauses ++ } else if (null == ser) { ++ log.debug("Registering {} with default serializer", targetClass); ++ kryo.register(targetClass); ++ } else { ++ log.debug("Registering {} with serializer {}", targetClass, ser); ++ kryo.register(targetClass, ser); ++ } ++ } ++ ++ Set<Class<?>> shimmedClassesFromGryoMapper = new HashSet<>(); ++ ++ // Apply GryoMapper's default registrations ++ for (TypeRegistration<?> tr : GryoMapper.build().create().getTypeRegistrations()) { ++ // Is this class blacklisted? Skip it. (takes precedence over serializerOverrides) ++ if (blacklist.contains(tr.getTargetClass())) { ++ log.debug("Not registering serializer for {} (blacklisted)", tr.getTargetClass()); ++ continue; ++ } ++ ++ final org.apache.tinkerpop.shaded.kryo.Serializer<?> shadedSerializer = tr.getShadedSerializer(); ++ final SerializerShim<?> serializerShim = tr.getSerializerShim(); ++ final java.util.function.Function< ++ org.apache.tinkerpop.shaded.kryo.Kryo, ++ org.apache.tinkerpop.shaded.kryo.Serializer> functionOfShadedKryo = tr.getFunctionOfShadedKryo(); ++ ++ // Apply overrides with the highest case-precedence ++ if (checkForAndApplySerializerOverride(serializerOverrides, kryo, tr.getTargetClass())) { ++ // do nothing but skip the remaining else(-if) clauses ++ } else if (null != shadedSerializer) { ++ if (shadedSerializer.getClass().equals(org.apache.tinkerpop.shaded.kryo.serializers.JavaSerializer.class)) { ++ // Convert GryoMapper's shaded JavaSerializer mappings to their unshaded equivalents ++ log.debug("Registering {} with JavaSerializer", tr.getTargetClass()); ++ kryo.register(tr.getTargetClass(), new JavaSerializer()); ++ } else { ++ // There's supposed to be a check in GryoMapper that prevents this from happening ++ log.error("GryoMapper's default serialization registration for {} is a {}. " + ++ "This is probably a bug in TinkerPop (this is not a valid default registration). " + ++ "I am configuring Spark to use Kryo's default serializer for this class, " + ++ "but this may cause serialization failures at runtime.", ++ tr.getTargetClass(), ++ org.apache.tinkerpop.shaded.kryo.Serializer.class.getCanonicalName()); ++ kryo.register(tr.getTargetClass()); ++ } ++ } else if (null != serializerShim) { ++ // Wrap shim serializers in an adapter for Spark's unshaded Kryo ++ log.debug("Registering {} to serializer shim {} (serializer shim {})", ++ tr.getTargetClass(), serializerShim, serializerShim.getClass()); ++ kryo.register(tr.getTargetClass(), new UnshadedSerializerAdapter<>(serializerShim)); ++ shimmedClassesFromGryoMapper.add(tr.getTargetClass()); ++ } else if (null != functionOfShadedKryo) { ++ // As with shaded serializers, there's supposed to be a check in GryoMapper that prevents this from happening ++ log.error("GryoMapper's default serialization registration for {} is a Function<{},{}>. " + ++ "This is probably a bug in TinkerPop (this is not a valid default registration). " + ++ "I am configuring Spark to use Kryo's default serializer instead of this function, " + ++ "but this may cause serialization failures at runtime.", ++ tr.getTargetClass(), ++ org.apache.tinkerpop.shaded.kryo.Kryo.class.getCanonicalName(), ++ org.apache.tinkerpop.shaded.kryo.Serializer.class.getCanonicalName()); ++ kryo.register(tr.getTargetClass()); ++ } else { ++ // Register all other classes with the default behavior (FieldSerializer) ++ log.debug("Registering {} with default serializer", tr.getTargetClass()); ++ kryo.register(tr.getTargetClass()); ++ } ++ } ++ ++ // StarGraph's shim serializer is especially important on Spark for efficiency reasons, ++ // so log a warning if we failed to register it somehow ++ if (!shimmedClassesFromGryoMapper.contains(StarGraph.class)) { ++ log.warn("No SerializerShim found for StarGraph"); ++ } ++ } ++ ++ private LinkedHashMap<Class<?>, Serializer<?>> getExtraRegistrations() { ++ ++ /* The map returned by this method MUST have a fixed iteration order! ++ * ++ * The order itself is irrelevant, so long as it is completely stable at runtime. ++ * ++ * LinkedHashMap satisfies this requirement (its contract specifies ++ * iteration in key-insertion-order). ++ */ ++ ++ LinkedHashMap<Class<?>, Serializer<?>> m = new LinkedHashMap<>(); ++ // The following entries were copied from GryoSerializer's constructor ++ // This could be turned into a static collection on GryoSerializer to avoid ++ // duplication, but it would be a bit cumbersome to do so without disturbing ++ // the ordering of the existing entries in that constructor, since not all ++ // of the entries are for TinkerPop (and the ordering is significant). ++ m.put(MessagePayload.class, null); ++ m.put(ViewIncomingPayload.class, null); ++ m.put(ViewOutgoingPayload.class, null); ++ m.put(ViewPayload.class, null); ++ m.put(VertexWritable.class, new UnshadedSerializerAdapter<>(new VertexWritableSerializer())); ++ m.put(ObjectWritable.class, new UnshadedSerializerAdapter<>(new ObjectWritableSerializer<>())); ++ ++ return m; ++ } ++ ++ private boolean checkForAndApplySerializerOverride(Map<Class<?>, Serializer<?>> serializerOverrides, ++ Kryo kryo, Class<?> targetClass) { ++ if (serializerOverrides.containsKey(targetClass)) { ++ Serializer<?> ser = serializerOverrides.get(targetClass); ++ if (null == ser) { ++ // null means use Kryo's default serializer ++ log.debug("Registering {} with default serializer per overrides", targetClass); ++ kryo.register(targetClass); ++ } else { ++ // nonnull means use that serializer ++ log.debug("Registering {} with serializer {} per overrides", targetClass, ser); ++ kryo.register(targetClass, ser); ++ } ++ return true; ++ } ++ return false; ++ } ++}