You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sedona.apache.org by ji...@apache.org on 2026/06/10 20:14:47 UTC
(sedona) branch master updated: [GH-3036] Flink Box3D foundation: Box3DTypeSerializer + ST_Box3D / ST_3DMakeBox (#3037)
This is an automated email from the ASF dual-hosted git repository.
jiayuasu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sedona.git
The following commit(s) were added to refs/heads/master by this push:
new 52e9f382bc [GH-3036] Flink Box3D foundation: Box3DTypeSerializer + ST_Box3D / ST_3DMakeBox (#3037)
52e9f382bc is described below
commit 52e9f382bce8313ab310ad1a7913081b43ce353d
Author: Jia Yu <ji...@apache.org>
AuthorDate: Wed Jun 10 13:14:41 2026 -0700
[GH-3036] Flink Box3D foundation: Box3DTypeSerializer + ST_Box3D / ST_3DMakeBox (#3037)
---
.../apache/sedona/flink/Box3DTypeSerializer.java | 181 +++++++++++++++++++++
.../main/java/org/apache/sedona/flink/Catalog.java | 2 +
.../sedona/flink/expressions/Constructors.java | 20 +++
.../apache/sedona/flink/expressions/Functions.java | 15 ++
.../sedona/flink/Box3DTypeSerializerTest.java | 120 ++++++++++++++
.../org/apache/sedona/flink/ConstructorTest.java | 16 ++
.../java/org/apache/sedona/flink/FunctionTest.java | 31 ++++
7 files changed, 385 insertions(+)
diff --git a/flink/src/main/java/org/apache/sedona/flink/Box3DTypeSerializer.java b/flink/src/main/java/org/apache/sedona/flink/Box3DTypeSerializer.java
new file mode 100644
index 0000000000..24ed196cf8
--- /dev/null
+++ b/flink/src/main/java/org/apache/sedona/flink/Box3DTypeSerializer.java
@@ -0,0 +1,181 @@
+/*
+ * 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.geometryObjects.Box3D;
+
+/**
+ * Flink {@link TypeSerializer} for {@link Box3D}. Serializes as a presence byte followed by six
+ * doubles (xmin, ymin, zmin, xmax, ymax, zmax). Total payload for a non-null value is 49 bytes.
+ */
+public class Box3DTypeSerializer extends TypeSerializer<Box3D> {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final Box3DTypeSerializer INSTANCE = new Box3DTypeSerializer();
+
+ public Box3DTypeSerializer() {}
+
+ @Override
+ public boolean isImmutableType() {
+ return true;
+ }
+
+ @Override
+ public TypeSerializer<Box3D> duplicate() {
+ return this;
+ }
+
+ @Override
+ public Box3D createInstance() {
+ // Box3D has no in-band "empty" sentinel — absence is represented by SQL NULL.
+ // Returning null here is the honest default; (0, 0, 0, 0, 0, 0) would alias to a real box at
+ // the origin and could silently corrupt any caller that treats the createInstance() result as
+ // a usable bbox.
+ return null;
+ }
+
+ @Override
+ public Box3D copy(Box3D from) {
+ if (from == null) {
+ return null;
+ }
+ return new Box3D(
+ from.getXMin(),
+ from.getYMin(),
+ from.getZMin(),
+ from.getXMax(),
+ from.getYMax(),
+ from.getZMax());
+ }
+
+ @Override
+ public Box3D copy(Box3D from, Box3D reuse) {
+ return copy(from);
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(Box3D record, DataOutputView target) throws IOException {
+ if (record == null) {
+ target.writeByte(0);
+ } else {
+ target.writeByte(1);
+ target.writeDouble(record.getXMin());
+ target.writeDouble(record.getYMin());
+ target.writeDouble(record.getZMin());
+ target.writeDouble(record.getXMax());
+ target.writeDouble(record.getYMax());
+ target.writeDouble(record.getZMax());
+ }
+ }
+
+ @Override
+ public Box3D deserialize(DataInputView source) throws IOException {
+ byte present = source.readByte();
+ if (present == 0) {
+ return null;
+ }
+ double xmin = source.readDouble();
+ double ymin = source.readDouble();
+ double zmin = source.readDouble();
+ double xmax = source.readDouble();
+ double ymax = source.readDouble();
+ double zmax = source.readDouble();
+ return new Box3D(xmin, ymin, zmin, xmax, ymax, zmax);
+ }
+
+ @Override
+ public Box3D deserialize(Box3D reuse, DataInputView source) throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ byte present = source.readByte();
+ target.writeByte(present);
+ if (present != 0) {
+ target.write(source, 48);
+ }
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return obj instanceof Box3DTypeSerializer;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+
+ @Override
+ public TypeSerializerSnapshot<Box3D> snapshotConfiguration() {
+ return new Box3DSerializerSnapshot();
+ }
+
+ public static final class Box3DSerializerSnapshot implements TypeSerializerSnapshot<Box3D> {
+ 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<Box3D> restoreSerializer() {
+ return Box3DTypeSerializer.INSTANCE;
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<Box3D> resolveSchemaCompatibility(
+ TypeSerializer<Box3D> newSerializer) {
+ if (newSerializer instanceof Box3DTypeSerializer) {
+ return TypeSerializerSchemaCompatibility.compatibleAsIs();
+ } else {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ }
+ }
+}
diff --git a/flink/src/main/java/org/apache/sedona/flink/Catalog.java b/flink/src/main/java/org/apache/sedona/flink/Catalog.java
index ba813a3d3b..65097b0e44 100644
--- a/flink/src/main/java/org/apache/sedona/flink/Catalog.java
+++ b/flink/src/main/java/org/apache/sedona/flink/Catalog.java
@@ -43,6 +43,7 @@ public class Catalog {
new Constructors.ST_LinestringFromWKB(),
new Constructors.ST_GeomFromBox2D(),
new Constructors.ST_MakeBox2D(),
+ new Constructors.ST_3DMakeBox(),
new Constructors.ST_MakeEnvelope(),
new Constructors.ST_MakePoint(),
new Constructors.ST_MakePointM(),
@@ -82,6 +83,7 @@ public class Catalog {
new Functions.ST_CrossesDateLine(),
new Functions.ST_Expand(),
new Functions.ST_Box2D(),
+ new Functions.ST_Box3D(),
new Functions.ST_Envelope(),
new Functions.ST_OrientedEnvelope(),
new Functions.ST_Difference(),
diff --git a/flink/src/main/java/org/apache/sedona/flink/expressions/Constructors.java b/flink/src/main/java/org/apache/sedona/flink/expressions/Constructors.java
index 9f7c4fc0f6..ccee88a9c2 100644
--- a/flink/src/main/java/org/apache/sedona/flink/expressions/Constructors.java
+++ b/flink/src/main/java/org/apache/sedona/flink/expressions/Constructors.java
@@ -24,9 +24,11 @@ import org.apache.sedona.common.Functions;
import org.apache.sedona.common.enums.FileDataSplitter;
import org.apache.sedona.common.enums.GeometryType;
import org.apache.sedona.common.geometryObjects.Box2D;
+import org.apache.sedona.common.geometryObjects.Box3D;
import org.apache.sedona.common.utils.FormatUtils;
import org.apache.sedona.common.utils.GeoHashDecoder;
import org.apache.sedona.flink.Box2DTypeSerializer;
+import org.apache.sedona.flink.Box3DTypeSerializer;
import org.apache.sedona.flink.GeometryTypeSerializer;
import org.locationtech.jts.geom.*;
import org.locationtech.jts.io.ParseException;
@@ -332,6 +334,24 @@ public class Constructors {
}
}
+ public static class ST_3DMakeBox extends ScalarFunction {
+ @DataTypeHint(value = "RAW", rawSerializer = Box3DTypeSerializer.class, bridgedTo = Box3D.class)
+ public Box3D eval(
+ @DataTypeHint(
+ value = "RAW",
+ rawSerializer = GeometryTypeSerializer.class,
+ bridgedTo = Geometry.class)
+ Object lowerLeft,
+ @DataTypeHint(
+ value = "RAW",
+ rawSerializer = GeometryTypeSerializer.class,
+ bridgedTo = Geometry.class)
+ Object upperRight) {
+ return org.apache.sedona.common.Constructors.make3DBox(
+ (Geometry) lowerLeft, (Geometry) upperRight);
+ }
+ }
+
public static class ST_PolygonFromEnvelope extends ScalarFunction {
@DataTypeHint(
value = "RAW",
diff --git a/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java b/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
index 8465e80ab5..6d6560910a 100644
--- a/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
+++ b/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
@@ -24,7 +24,9 @@ import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.sedona.common.geometryObjects.Box2D;
+import org.apache.sedona.common.geometryObjects.Box3D;
import org.apache.sedona.flink.Box2DTypeSerializer;
+import org.apache.sedona.flink.Box3DTypeSerializer;
import org.apache.sedona.flink.GeometryArrayTypeSerializer;
import org.apache.sedona.flink.GeometryTypeSerializer;
import org.geotools.api.referencing.FactoryException;
@@ -517,6 +519,19 @@ public class Functions {
}
}
+ public static class ST_Box3D extends ScalarFunction {
+ @DataTypeHint(value = "RAW", rawSerializer = Box3DTypeSerializer.class, bridgedTo = Box3D.class)
+ public Box3D eval(
+ @DataTypeHint(
+ value = "RAW",
+ rawSerializer = GeometryTypeSerializer.class,
+ bridgedTo = Geometry.class)
+ Object o) {
+ Geometry geom = (Geometry) o;
+ return org.apache.sedona.common.Functions.box3D(geom);
+ }
+ }
+
public static class ST_OrientedEnvelope extends ScalarFunction {
@DataTypeHint(
value = "RAW",
diff --git a/flink/src/test/java/org/apache/sedona/flink/Box3DTypeSerializerTest.java b/flink/src/test/java/org/apache/sedona/flink/Box3DTypeSerializerTest.java
new file mode 100644
index 0000000000..5916f167c1
--- /dev/null
+++ b/flink/src/test/java/org/apache/sedona/flink/Box3DTypeSerializerTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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 static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.sedona.common.geometryObjects.Box3D;
+import org.junit.Test;
+
+public class Box3DTypeSerializerTest {
+
+ private final Box3DTypeSerializer serializer = Box3DTypeSerializer.INSTANCE;
+
+ private byte[] serialize(Box3D box) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ serializer.serialize(box, new DataOutputViewStreamWrapper(out));
+ return out.toByteArray();
+ }
+
+ private Box3D deserialize(byte[] bytes) throws IOException {
+ return serializer.deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStream(bytes)));
+ }
+
+ @Test
+ public void roundTripOrderedBox() throws IOException {
+ Box3D box = new Box3D(0.0, 0.0, -3.0, 5.0, 10.0, 7.0);
+ byte[] bytes = serialize(box);
+ // 1 presence byte + 6 doubles.
+ assertEquals(49, bytes.length);
+ assertEquals(box, deserialize(bytes));
+ }
+
+ @Test
+ public void roundTripInvertedBoundsVerbatim() throws IOException {
+ // The serializer stores bounds as-is; ordering validation lives in the predicates, not here.
+ Box3D inverted = new Box3D(5.0, 5.0, 5.0, 0.0, 0.0, 0.0);
+ assertEquals(inverted, deserialize(serialize(inverted)));
+ }
+
+ @Test
+ public void roundTripNegativeAndFractionalBounds() throws IOException {
+ Box3D box = new Box3D(-1.5, -2.25, -3.125, 4.5, 6.75, 8.875);
+ assertEquals(box, deserialize(serialize(box)));
+ }
+
+ @Test
+ public void serializeNullWritesSinglePresenceByte() throws IOException {
+ byte[] bytes = serialize(null);
+ assertEquals(1, bytes.length);
+ assertEquals(0, bytes[0]);
+ assertNull(deserialize(bytes));
+ }
+
+ @Test
+ public void rawCopyTransfersBytesForNonNull() throws IOException {
+ Box3D box = new Box3D(1.0, 2.0, 3.0, 4.0, 5.0, 6.0);
+ byte[] original = serialize(box);
+
+ // Exercise copy(DataInputView, DataOutputView): the raw byte-shuffling path that skips
+ // deserialization. The copied bytes must round-trip back to the same Box3D.
+ ByteArrayOutputStream copied = new ByteArrayOutputStream();
+ serializer.copy(
+ new DataInputViewStreamWrapper(new ByteArrayInputStream(original)),
+ new DataOutputViewStreamWrapper(copied));
+
+ byte[] copiedBytes = copied.toByteArray();
+ assertEquals(original.length, copiedBytes.length);
+ assertEquals(box, deserialize(copiedBytes));
+ }
+
+ @Test
+ public void rawCopyTransfersBytesForNull() throws IOException {
+ byte[] original = serialize(null);
+ ByteArrayOutputStream copied = new ByteArrayOutputStream();
+ serializer.copy(
+ new DataInputViewStreamWrapper(new ByteArrayInputStream(original)),
+ new DataOutputViewStreamWrapper(copied));
+ assertEquals(1, copied.toByteArray().length);
+ assertNull(deserialize(copied.toByteArray()));
+ }
+
+ @Test
+ public void objectCopyProducesEqualInstance() {
+ Box3D box = new Box3D(0.0, 0.0, 0.0, 1.0, 1.0, 1.0);
+ Box3D copy = serializer.copy(box);
+ assertEquals(box, copy);
+ assertNull(serializer.copy((Box3D) null));
+ }
+
+ @Test
+ public void serializerEqualityAndImmutability() {
+ assertEquals(serializer, new Box3DTypeSerializer());
+ assertEquals(serializer.hashCode(), new Box3DTypeSerializer().hashCode());
+ assertTrue(serializer.isImmutableType());
+ assertTrue(serializer.duplicate() == serializer);
+ }
+}
diff --git a/flink/src/test/java/org/apache/sedona/flink/ConstructorTest.java b/flink/src/test/java/org/apache/sedona/flink/ConstructorTest.java
index 4a48adc564..36ddf0b19f 100644
--- a/flink/src/test/java/org/apache/sedona/flink/ConstructorTest.java
+++ b/flink/src/test/java/org/apache/sedona/flink/ConstructorTest.java
@@ -35,6 +35,7 @@ import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Table;
import org.apache.flink.types.Row;
import org.apache.sedona.common.geometryObjects.Box2D;
+import org.apache.sedona.common.geometryObjects.Box3D;
import org.apache.sedona.common.jts2geojson.GeoJSONReader;
import org.apache.sedona.flink.expressions.Constructors;
import org.apache.sedona.flink.expressions.Functions;
@@ -380,6 +381,21 @@ public class ConstructorTest extends TestBase {
assertEquals(5.0, bbox.getYMax(), 0.0);
}
+ @Test
+ public void testMake3DBox() {
+ Box3D bbox =
+ (Box3D)
+ last(tableEnv.sqlQuery(
+ "SELECT ST_3DMakeBox(ST_PointZ(1.0, 2.0, 3.0), ST_PointZ(4.0, 5.0, 6.0))"))
+ .getField(0);
+ assertEquals(1.0, bbox.getXMin(), 0.0);
+ assertEquals(2.0, bbox.getYMin(), 0.0);
+ assertEquals(3.0, bbox.getZMin(), 0.0);
+ assertEquals(4.0, bbox.getXMax(), 0.0);
+ assertEquals(5.0, bbox.getYMax(), 0.0);
+ assertEquals(6.0, bbox.getZMax(), 0.0);
+ }
+
@Test
public void testGeomFromBox2D() {
String wkt =
diff --git a/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java b/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java
index c8d7edff9a..f565b591d5 100644
--- a/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java
+++ b/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java
@@ -31,6 +31,7 @@ import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.table.api.Table;
import org.apache.flink.types.Row;
import org.apache.sedona.common.geometryObjects.Box2D;
+import org.apache.sedona.common.geometryObjects.Box3D;
import org.apache.sedona.flink.expressions.Functions;
import org.apache.sedona.flink.expressions.FunctionsProj4;
import org.datasyslab.proj4sedona.core.Proj;
@@ -460,6 +461,36 @@ public class FunctionTest extends TestBase {
assertNull(row.getField(1));
}
+ @Test
+ public void testBox3D() {
+ Table t =
+ tableEnv.sqlQuery(
+ "SELECT ST_Box3D(ST_GeomFromText('LINESTRING Z(0 0 -3, 5 10 7)')) AS bbox");
+ Box3D bbox = (Box3D) first(t).getField(0);
+ assertEquals(0.0, bbox.getXMin(), 0.0);
+ assertEquals(0.0, bbox.getYMin(), 0.0);
+ assertEquals(-3.0, bbox.getZMin(), 0.0);
+ assertEquals(5.0, bbox.getXMax(), 0.0);
+ assertEquals(10.0, bbox.getYMax(), 0.0);
+ assertEquals(7.0, bbox.getZMax(), 0.0);
+
+ // XY-only geometry folds Z to 0 per PostGIS.
+ Table tXY =
+ tableEnv.sqlQuery("SELECT ST_Box3D(ST_GeomFromText('LINESTRING(0 0, 5 10)')) AS bbox");
+ Box3D xy = (Box3D) first(tXY).getField(0);
+ assertEquals(0.0, xy.getZMin(), 0.0);
+ assertEquals(0.0, xy.getZMax(), 0.0);
+
+ // null and empty inputs propagate to NULL.
+ Table tNull =
+ tableEnv.sqlQuery(
+ "SELECT ST_Box3D(ST_GeomFromText('LINESTRING EMPTY')) AS empty_bbox,"
+ + " ST_Box3D(ST_GeomFromText(CAST(NULL AS STRING))) AS null_bbox");
+ Row row = first(tNull);
+ assertNull(row.getField(0));
+ assertNull(row.getField(1));
+ }
+
@Test
public void testBox2DAsTextAndAccessors() {
Table t =