From 1882de0dcbd08e05e54974df8286726a2730b6cc Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 18 Jun 2026 23:40:41 -0700 Subject: [PATCH] [GH-3058] Add GeographyTypeSerializer to SedonaFlink --- .../sedona/flink/GeographyTypeSerializer.java | 167 ++++++++++++++++++ .../flink/GeographyTypeSerializerTest.java | 68 +++++++ 2 files changed, 235 insertions(+) create mode 100644 flink/src/main/java/org/apache/sedona/flink/GeographyTypeSerializer.java create mode 100644 flink/src/test/java/org/apache/sedona/flink/GeographyTypeSerializerTest.java diff --git a/flink/src/main/java/org/apache/sedona/flink/GeographyTypeSerializer.java b/flink/src/main/java/org/apache/sedona/flink/GeographyTypeSerializer.java new file mode 100644 index 00000000000..67e2345b654 --- /dev/null +++ b/flink/src/main/java/org/apache/sedona/flink/GeographyTypeSerializer.java @@ -0,0 +1,167 @@ +/* + * 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.sedona.flink; + +import java.io.IOException; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; +import org.apache.sedona.common.S2Geography.Geography; +import org.apache.sedona.common.S2Geography.GeographyWKBSerializer; +import org.apache.sedona.common.S2Geography.SinglePointGeography; + +public class GeographyTypeSerializer extends TypeSerializer { + + private static final long serialVersionUID = 1L; + + public static final GeographyTypeSerializer INSTANCE = new GeographyTypeSerializer(); + + public GeographyTypeSerializer() {} + + @Override + public boolean isImmutableType() { + return false; + } + + @Override + public TypeSerializer duplicate() { + return this; + } + + @Override + public Geography createInstance() { + return new SinglePointGeography(); + } + + @Override + public Geography copy(Geography from) { + if (from == null) { + return null; + } + // Geography has no copy()/clone(); a serialize/deserialize round-trip is a safe deep copy. + try { + return GeographyWKBSerializer.deserialize(GeographyWKBSerializer.serialize(from)); + } catch (IOException e) { + throw new RuntimeException("Failed to copy Geography", e); + } + } + + @Override + public Geography copy(Geography from, Geography reuse) { + return copy(from); + } + + @Override + public int getLength() { + return -1; + } + + @Override + public void serialize(Geography record, DataOutputView target) throws IOException { + if (record == null) { + target.writeInt(-1); + } else { + byte[] data = GeographyWKBSerializer.serialize(record); + target.writeInt(data.length); + target.write(data); + } + } + + @Override + public Geography deserialize(DataInputView source) throws IOException { + int length = source.readInt(); + if (length == -1) { + return null; + } + byte[] data = new byte[length]; + source.readFully(data); + return GeographyWKBSerializer.deserialize(data); + } + + @Override + public Geography deserialize(Geography reuse, DataInputView source) throws IOException { + return deserialize(source); + } + + @Override + public void copy(DataInputView source, DataOutputView target) throws IOException { + int length = source.readInt(); + target.writeInt(length); + if (length > 0) { + target.write(source, length); + } + } + + @Override + public boolean equals(Object obj) { + return obj instanceof GeographyTypeSerializer; + } + + @Override + public int hashCode() { + return getClass().hashCode(); + } + + @Override + public TypeSerializerSnapshot snapshotConfiguration() { + return new GeographySerializerSnapshot(); + } + + public static final class GeographySerializerSnapshot + implements TypeSerializerSnapshot { + private static final int CURRENT_VERSION = 1; + + @Override + public int getCurrentVersion() { + return CURRENT_VERSION; + } + + @Override + public void writeSnapshot(DataOutputView out) throws IOException {} + + @Override + public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) + throws IOException { + if (readVersion != CURRENT_VERSION) { + throw new IOException( + "Cannot read snapshot: Incompatible version " + + readVersion + + ". Expected version " + + CURRENT_VERSION); + } + } + + @Override + public TypeSerializer restoreSerializer() { + return GeographyTypeSerializer.INSTANCE; + } + + @Override + public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( + TypeSerializer newSerializer) { + if (newSerializer instanceof GeographyTypeSerializer) { + return TypeSerializerSchemaCompatibility.compatibleAsIs(); + } else { + return TypeSerializerSchemaCompatibility.incompatible(); + } + } + } +} diff --git a/flink/src/test/java/org/apache/sedona/flink/GeographyTypeSerializerTest.java b/flink/src/test/java/org/apache/sedona/flink/GeographyTypeSerializerTest.java new file mode 100644 index 00000000000..3e6a92b147d --- /dev/null +++ b/flink/src/test/java/org/apache/sedona/flink/GeographyTypeSerializerTest.java @@ -0,0 +1,68 @@ +/* + * 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.sedona.flink; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.apache.sedona.common.S2Geography.Geography; +import org.apache.sedona.common.geography.Constructors; +import org.junit.Test; + +public class GeographyTypeSerializerTest { + + private final GeographyTypeSerializer serializer = GeographyTypeSerializer.INSTANCE; + + private Geography roundTrip(Geography input) throws Exception { + DataOutputSerializer out = new DataOutputSerializer(64); + serializer.serialize(input, out); + DataInputDeserializer in = new DataInputDeserializer(out.getCopyOfBuffer()); + return serializer.deserialize(in); + } + + @Test + public void testPointRoundTrip() throws Exception { + Geography point = Constructors.geogFromWKT("POINT (1 2)", 4326); + Geography result = roundTrip(point); + assertEquals(point.toEWKT(), result.toEWKT()); + assertEquals(4326, result.getSRID()); + } + + @Test + public void testPolygonRoundTrip() throws Exception { + Geography polygon = Constructors.geogFromWKT("POLYGON ((0 0, 1 0, 1 1, 0 1, 0 0))", 4326); + Geography result = roundTrip(polygon); + assertEquals(polygon.toEWKT(), result.toEWKT()); + } + + @Test + public void testNullRoundTrip() throws Exception { + assertNull(roundTrip(null)); + } + + @Test + public void testCopy() throws Exception { + Geography point = Constructors.geogFromWKT("POINT (3 4)", 4326); + Geography copy = serializer.copy(point); + assertEquals(point.toEWKT(), copy.toEWKT()); + assertNull(serializer.copy(null)); + } +}