+ {
+ return new SqlGeographyTypeNameSpec(getPos());
+ }
+}
+
/**
* Parse a "name1 type1 [ NULL | NOT NULL] [ comment ]
* [, name2 type2 [ NULL | NOT NULL] [ comment ] ]* ..." list.
diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlGeographyTypeNameSpec.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlGeographyTypeNameSpec.java
new file mode 100644
index 00000000000000..55a3d7ff4fd7d6
--- /dev/null
+++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlGeographyTypeNameSpec.java
@@ -0,0 +1,59 @@
+/*
+ * 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.flink.sql.parser.type;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.calcite.ExtendedRelTypeFactory;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlTypeNameSpec;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.validate.SqlValidator;
+import org.apache.calcite.util.Litmus;
+
+/** Represents the GEOGRAPHY data type. */
+@Internal
+public final class SqlGeographyTypeNameSpec extends SqlTypeNameSpec {
+
+ private static final String GEOGRAPHY_TYPE_NAME = "GEOGRAPHY";
+
+ public SqlGeographyTypeNameSpec(SqlParserPos pos) {
+ super(new SqlIdentifier(GEOGRAPHY_TYPE_NAME, pos), pos);
+ }
+
+ @Override
+ public RelDataType deriveType(SqlValidator validator) {
+ return ((ExtendedRelTypeFactory) validator.getTypeFactory()).createGeographyType();
+ }
+
+ @Override
+ public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+ writer.keyword(GEOGRAPHY_TYPE_NAME);
+ }
+
+ @Override
+ public boolean equalsDeep(SqlTypeNameSpec spec, Litmus litmus) {
+ if (!(spec instanceof SqlGeographyTypeNameSpec)) {
+ return litmus.fail("{} != {}", this, spec);
+ }
+ return litmus.succeed();
+ }
+}
diff --git a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java
index 57b9bc615ec930..57be54e4cb6bd4 100644
--- a/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java
+++ b/flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java
@@ -44,4 +44,7 @@ RelDataType createStructuredType(
/** Creates a BITMAP type. */
RelDataType createBitmapType();
+
+ /** Creates a GEOGRAPHY type. */
+ RelDataType createGeographyType();
}
diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
index 92f8e34c770a00..ee33ea2e37daf2 100644
--- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
+++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
@@ -4074,4 +4074,23 @@ void testBitmapType() {
sql("CREATE TABLE t (\n" + "^bitmap^ INT" + "\n)")
.fails("(?s).*Encountered \"bitmap\" at line 2, column 1.\n.*");
}
+
+ @Test
+ void testGeographyType() {
+ sql("CREATE TABLE t (\n" + "g geography" + "\n)")
+ .ok("CREATE TABLE `T` (\n" + " `G` GEOGRAPHY\n" + ")");
+
+ sql("CREATE TABLE t (\n" + "g geography NOT NULL" + "\n)")
+ .ok("CREATE TABLE `T` (\n" + " `G` GEOGRAPHY NOT NULL\n" + ")");
+
+ // GEOGRAPHY takes no parameters
+ sql("CREATE TABLE t (\n" + "g geography^(^1)" + "\n)")
+ .fails("(?s).*Encountered \"\\(\" at line 2, column 12.\n.*");
+ sql("CREATE TABLE t (\n" + "g geography^(^4326)" + "\n)")
+ .fails("(?s).*Encountered \"\\(\" at line 2, column 12.\n.*");
+
+ // GEOGRAPHY is a reserved keyword and cannot be used as an identifier without escaping
+ sql("CREATE TABLE t (\n" + "^geography^ INT" + "\n)")
+ .fails("(?s).*Encountered \"geography\" at line 2, column 1.\n.*");
+ }
}
diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java
index 34fe9926079b3c..070c77905a335f 100644
--- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java
+++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java
@@ -53,6 +53,11 @@ public RelDataType createBitmapType() {
return canonize(new DummyBitmapType());
}
+ @Override
+ public RelDataType createGeographyType() {
+ return canonize(new DummyGeographyType());
+ }
+
private static class DummyRawType extends RelDataTypeImpl {
private final String className;
@@ -117,4 +122,16 @@ protected void generateTypeString(StringBuilder sb, boolean withDetail) {
sb.append("BITMAP");
}
}
+
+ private static class DummyGeographyType extends RelDataTypeImpl {
+
+ DummyGeographyType() {
+ computeDigest();
+ }
+
+ @Override
+ protected void generateTypeString(StringBuilder sb, boolean withDetail) {
+ sb.append("GEOGRAPHY");
+ }
+ }
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java
index af1e718d90e04c..3912c178328526 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java
@@ -44,6 +44,7 @@
import org.apache.flink.table.types.logical.DescriptorType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -1075,6 +1076,15 @@ public static DataType BITMAP() {
return new AtomicDataType(new BitmapType());
}
+ /**
+ * Data type of geography data.
+ *
+ * @see GeographyType
+ */
+ public static DataType GEOGRAPHY() {
+ return new AtomicDataType(new GeographyType());
+ }
+
// --------------------------------------------------------------------------------------------
// Helper functions
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java
index 811c9be70e6b87..728d3c228f05da 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java
@@ -123,6 +123,12 @@ default Bitmap getBitmap(int pos) {
"This ArrayData implementation does not support Bitmap type.");
}
+ /** Returns the geography value at the given position. */
+ default GeographyData getGeography(int pos) {
+ throw new UnsupportedOperationException(
+ "This ArrayData implementation does not support Geography type.");
+ }
+
// ------------------------------------------------------------------------------------------
// Conversion Utilities
// ------------------------------------------------------------------------------------------
@@ -225,6 +231,9 @@ static ElementGetter createElementGetter(LogicalType elementType) {
case BITMAP:
elementGetter = ArrayData::getBitmap;
break;
+ case GEOGRAPHY:
+ elementGetter = ArrayData::getGeography;
+ break;
case NULL:
case SYMBOL:
case UNRESOLVED:
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java
index f609c2f22f1311..08856cdb0ceb1e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java
@@ -265,6 +265,11 @@ public Bitmap getBitmap(int pos) {
return (Bitmap) getObject(pos);
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ return (GeographyData) getObject(pos);
+ }
+
private Object getObject(int pos) {
return ((Object[]) array)[pos];
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java
index b234edef62d05b..98c77bca28f06d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java
@@ -192,6 +192,11 @@ public byte[] getBinary(int pos) {
return (byte[]) this.fields[pos];
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ return (GeographyData) this.fields[pos];
+ }
+
@Override
public ArrayData getArray(int pos) {
return (ArrayData) this.fields[pos];
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GeographyData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GeographyData.java
new file mode 100644
index 00000000000000..f65075be637c80
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GeographyData.java
@@ -0,0 +1,82 @@
+/*
+ * 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.flink.table.data;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.data.binary.BinaryGeographyData;
+import org.apache.flink.table.types.logical.GeographyType;
+
+/** An internal data structure representing data of {@link GeographyType}. */
+@PublicEvolving
+public interface GeographyData {
+
+ /** ISO WKB subtype ID for Point geometries. */
+ int POINT = 1;
+
+ /** ISO WKB subtype ID for LineString geometries. */
+ int LINE_STRING = 2;
+
+ /** ISO WKB subtype ID for Polygon geometries. */
+ int POLYGON = 3;
+
+ /** ISO WKB subtype ID for MultiPoint geometries. */
+ int MULTI_POINT = 4;
+
+ /** ISO WKB subtype ID for MultiLineString geometries. */
+ int MULTI_LINE_STRING = 5;
+
+ /** ISO WKB subtype ID for MultiPolygon geometries. */
+ int MULTI_POLYGON = 6;
+
+ /** ISO WKB subtype ID for GeometryCollection geometries. */
+ int GEOMETRY_COLLECTION = 7;
+
+ /**
+ * Converts this {@link GeographyData} object to an ISO WKB byte array.
+ *
+ * Note: The returned byte array may be reused.
+ */
+ byte[] toBytes();
+
+ /** Returns the ISO WKB subtype ID. */
+ int subtypeId();
+
+ /** Returns the size in bytes of the ISO WKB payload. */
+ int sizeInBytes();
+
+ // ------------------------------------------------------------------------------------------
+ // Construction Utilities
+ // ------------------------------------------------------------------------------------------
+
+ /**
+ * Creates an instance of {@link GeographyData} from the given ISO WKB byte array. Returns
+ * {@code null} if the input is {@code null}.
+ */
+ static GeographyData fromBytes(byte[] bytes) {
+ return BinaryGeographyData.fromBytes(bytes);
+ }
+
+ /**
+ * Creates an instance of {@link GeographyData} from the given ISO WKB byte range. Returns
+ * {@code null} if the input is {@code null}.
+ */
+ static GeographyData fromBytes(byte[] bytes, int offset, int numBytes) {
+ return BinaryGeographyData.fromBytes(bytes, offset, numBytes);
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java
index ca43f1608f99ef..a783ecb572a698 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/RowData.java
@@ -111,6 +111,8 @@
* +--------------------------------+-----------------------------------------+
* | BITMAP | {@link Bitmap} |
* +--------------------------------+-----------------------------------------+
+ * | GEOGRAPHY | {@link GeographyData} |
+ * +--------------------------------+-----------------------------------------+
*
*
*
Nullability is always handled by the container data structure.
@@ -214,6 +216,12 @@ default Bitmap getBitmap(int pos) {
"This RowData implementation does not support Bitmap type.");
}
+ /** Returns the geography value at the given position. */
+ default GeographyData getGeography(int pos) {
+ throw new UnsupportedOperationException(
+ "This RowData implementation does not support Geography type.");
+ }
+
// ------------------------------------------------------------------------------------------
// Access Utilities
// ------------------------------------------------------------------------------------------
@@ -299,6 +307,9 @@ static FieldGetter createFieldGetter(LogicalType fieldType, int fieldPos) {
case BITMAP:
fieldGetter = row -> row.getBitmap(fieldPos);
break;
+ case GEOGRAPHY:
+ fieldGetter = row -> row.getGeography(fieldPos);
+ break;
case NULL:
case SYMBOL:
case UNRESOLVED:
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
index 10a8b3e6ef71c7..f09f07d25996c1 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryArrayData.java
@@ -23,6 +23,7 @@
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -95,6 +96,7 @@ public static int calculateFixLengthPartSize(LogicalType type) {
case RAW:
case VARIANT:
case BITMAP:
+ case GEOGRAPHY:
// long and double are 8 bytes;
// otherwise it stores the length and offset of the variable-length part for types
// such as is string, map, etc.
@@ -269,6 +271,14 @@ public byte[] getBinary(int pos) {
return BinarySegmentUtils.readBinary(segments, offset, fieldOffset, offsetAndSize);
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ assertIndexIsValid(pos);
+ int fieldOffset = getElementOffset(pos, 8);
+ final long offsetAndSize = BinarySegmentUtils.getLong(segments, fieldOffset);
+ return BinarySegmentUtils.readGeographyData(segments, offset, fieldOffset, offsetAndSize);
+ }
+
@Override
public ArrayData getArray(int pos) {
assertIndexIsValid(pos);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryGeographyData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryGeographyData.java
new file mode 100644
index 00000000000000..4256f786a86816
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryGeographyData.java
@@ -0,0 +1,269 @@
+/*
+ * 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.flink.table.data.binary;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.GeographyData;
+
+import java.util.Arrays;
+
+/**
+ * A binary implementation of {@link GeographyData} backed by raw ISO WKB bytes.
+ *
+ *
GEOGRAPHY uses OGC:CRS84 by contract, but ISO WKB does not encode CRS or SRID metadata. This
+ * container stores the raw ISO WKB payload only; CRS validation, CRS transformation, and EWKB/SRID
+ * handling belong to constructors, functions, and connector schema mapping.
+ */
+@Internal
+public final class BinaryGeographyData extends BinarySection implements GeographyData {
+
+ private static final int MIN_WKB_HEADER_SIZE = 5;
+ private static final int WKB_COUNT_SIZE = 4;
+ private static final int WKB_POINT_COORDINATE_SIZE = 16;
+ private static final int BIG_ENDIAN = 0;
+ private static final int LITTLE_ENDIAN = 1;
+
+ private final int subtypeId;
+
+ private BinaryGeographyData(MemorySegment[] segments, int offset, int sizeInBytes) {
+ super(segments, offset, sizeInBytes);
+ this.subtypeId = readSubtypeId(segments, offset, sizeInBytes);
+ }
+
+ /** Creates a {@link BinaryGeographyData} instance from the given address and length. */
+ public static BinaryGeographyData fromAddress(
+ MemorySegment[] segments, int offset, int numBytes) {
+ return new BinaryGeographyData(segments, offset, numBytes);
+ }
+
+ /** Creates a {@link BinaryGeographyData} instance from the given ISO WKB bytes. */
+ public static BinaryGeographyData fromBytes(byte[] bytes) {
+ return bytes == null ? null : fromBytes(bytes, 0, bytes.length);
+ }
+
+ /** Creates a {@link BinaryGeographyData} instance from the given ISO WKB byte range. */
+ public static BinaryGeographyData fromBytes(byte[] bytes, int offset, int numBytes) {
+ if (bytes == null) {
+ return null;
+ }
+ checkRange(bytes, offset, numBytes);
+ byte[] copy = Arrays.copyOfRange(bytes, offset, offset + numBytes);
+ return fromAddress(new MemorySegment[] {MemorySegmentFactory.wrap(copy)}, 0, copy.length);
+ }
+
+ @Override
+ public byte[] toBytes() {
+ return BinarySegmentUtils.copyToBytes(segments, offset, sizeInBytes);
+ }
+
+ @Override
+ public int subtypeId() {
+ return subtypeId;
+ }
+
+ @Override
+ public int sizeInBytes() {
+ return sizeInBytes;
+ }
+
+ private static void checkRange(byte[] bytes, int offset, int numBytes) {
+ if (offset < 0 || numBytes < 0 || offset > bytes.length - numBytes) {
+ throw new TableRuntimeException(
+ String.format(
+ "Invalid ISO WKB byte range: offset %d, length %d, array length %d.",
+ offset, numBytes, bytes.length));
+ }
+ }
+
+ private static int readSubtypeId(MemorySegment[] segments, int offset, int sizeInBytes) {
+ final long endOffset = (long) offset + sizeInBytes;
+ final GeometryHeader header = readHeader(segments, offset, endOffset);
+ final long consumedOffset = validateGeometry(segments, offset, endOffset);
+ if (consumedOffset != endOffset) {
+ throw new TableRuntimeException(
+ String.format(
+ "Malformed ISO WKB payload. Found %d trailing byte(s).",
+ endOffset - consumedOffset));
+ }
+ return header.subtypeId;
+ }
+
+ private static long validateGeometry(
+ MemorySegment[] segments, long geometryOffset, long endOffset) {
+ final GeometryHeader header = readHeader(segments, geometryOffset, endOffset);
+ long cursor = geometryOffset + MIN_WKB_HEADER_SIZE;
+
+ switch (header.subtypeId) {
+ case GeographyData.POINT:
+ return requireBytes(
+ cursor, WKB_POINT_COORDINATE_SIZE, endOffset, "POINT coordinates")
+ + WKB_POINT_COORDINATE_SIZE;
+ case GeographyData.LINE_STRING:
+ return skipCoordinateSequence(segments, cursor, endOffset, header.byteOrder);
+ case GeographyData.POLYGON:
+ return skipPolygon(segments, cursor, endOffset, header.byteOrder);
+ case GeographyData.MULTI_POINT:
+ return skipTypedGeometryCollection(
+ segments, cursor, endOffset, header.byteOrder, GeographyData.POINT);
+ case GeographyData.MULTI_LINE_STRING:
+ return skipTypedGeometryCollection(
+ segments, cursor, endOffset, header.byteOrder, GeographyData.LINE_STRING);
+ case GeographyData.MULTI_POLYGON:
+ return skipTypedGeometryCollection(
+ segments, cursor, endOffset, header.byteOrder, GeographyData.POLYGON);
+ case GeographyData.GEOMETRY_COLLECTION:
+ return skipGeometryCollection(segments, cursor, endOffset, header.byteOrder);
+ default:
+ throw new TableRuntimeException(
+ String.format(
+ "Malformed ISO WKB payload. Unsupported geography subtype ID %d.",
+ header.subtypeId));
+ }
+ }
+
+ private static long skipCoordinateSequence(
+ MemorySegment[] segments, long offset, long endOffset, int byteOrder) {
+ final long numPoints =
+ readUnsignedInt(segments, offset, endOffset, byteOrder, "point count");
+ long cursor = offset + WKB_COUNT_SIZE;
+ final long coordinateSequenceSize =
+ multiplyExact(numPoints, WKB_POINT_COORDINATE_SIZE, "coordinate sequence");
+ return requireBytes(cursor, coordinateSequenceSize, endOffset, "coordinate sequence")
+ + coordinateSequenceSize;
+ }
+
+ private static long skipPolygon(
+ MemorySegment[] segments, long offset, long endOffset, int byteOrder) {
+ final long numRings = readUnsignedInt(segments, offset, endOffset, byteOrder, "ring count");
+ long cursor = offset + WKB_COUNT_SIZE;
+ for (long i = 0; i < numRings; i++) {
+ cursor = skipCoordinateSequence(segments, cursor, endOffset, byteOrder);
+ }
+ return cursor;
+ }
+
+ private static long skipTypedGeometryCollection(
+ MemorySegment[] segments,
+ long offset,
+ long endOffset,
+ int byteOrder,
+ int expectedSubtypeId) {
+ final long numGeometries =
+ readUnsignedInt(segments, offset, endOffset, byteOrder, "geometry count");
+ long cursor = offset + WKB_COUNT_SIZE;
+ for (long i = 0; i < numGeometries; i++) {
+ final GeometryHeader nestedHeader = readHeader(segments, cursor, endOffset);
+ if (nestedHeader.subtypeId != expectedSubtypeId) {
+ throw new TableRuntimeException(
+ String.format(
+ "Malformed ISO WKB payload. Expected nested subtype ID %d but found %d.",
+ expectedSubtypeId, nestedHeader.subtypeId));
+ }
+ cursor = validateGeometry(segments, cursor, endOffset);
+ }
+ return cursor;
+ }
+
+ private static long skipGeometryCollection(
+ MemorySegment[] segments, long offset, long endOffset, int byteOrder) {
+ final long numGeometries =
+ readUnsignedInt(segments, offset, endOffset, byteOrder, "geometry count");
+ long cursor = offset + WKB_COUNT_SIZE;
+ for (long i = 0; i < numGeometries; i++) {
+ cursor = validateGeometry(segments, cursor, endOffset);
+ }
+ return cursor;
+ }
+
+ private static GeometryHeader readHeader(
+ MemorySegment[] segments, long offset, long endOffset) {
+ requireBytes(offset, MIN_WKB_HEADER_SIZE, endOffset, "WKB header");
+
+ final int byteOrder = BinarySegmentUtils.getByte(segments, (int) offset) & 0xFF;
+ if (byteOrder != BIG_ENDIAN && byteOrder != LITTLE_ENDIAN) {
+ throw new TableRuntimeException(
+ String.format(
+ "Malformed ISO WKB payload. Unsupported byte order %d.", byteOrder));
+ }
+
+ final long subtypeId =
+ readUnsignedInt(segments, offset + 1, endOffset, byteOrder, "subtype ID");
+
+ if (subtypeId < GeographyData.POINT || subtypeId > GeographyData.GEOMETRY_COLLECTION) {
+ throw new TableRuntimeException(
+ String.format(
+ "Malformed ISO WKB payload. Unsupported geography subtype ID %d.",
+ subtypeId));
+ }
+ return new GeometryHeader(byteOrder, (int) subtypeId);
+ }
+
+ private static long readUnsignedInt(
+ MemorySegment[] segments,
+ long offset,
+ long endOffset,
+ int byteOrder,
+ String fieldName) {
+ requireBytes(offset, WKB_COUNT_SIZE, endOffset, fieldName);
+
+ if (byteOrder == LITTLE_ENDIAN) {
+ return (BinarySegmentUtils.getByte(segments, (int) offset) & 0xFFL)
+ | ((BinarySegmentUtils.getByte(segments, (int) offset + 1) & 0xFFL) << 8)
+ | ((BinarySegmentUtils.getByte(segments, (int) offset + 2) & 0xFFL) << 16)
+ | ((BinarySegmentUtils.getByte(segments, (int) offset + 3) & 0xFFL) << 24);
+ }
+ return ((BinarySegmentUtils.getByte(segments, (int) offset) & 0xFFL) << 24)
+ | ((BinarySegmentUtils.getByte(segments, (int) offset + 1) & 0xFFL) << 16)
+ | ((BinarySegmentUtils.getByte(segments, (int) offset + 2) & 0xFFL) << 8)
+ | (BinarySegmentUtils.getByte(segments, (int) offset + 3) & 0xFFL);
+ }
+
+ private static long requireBytes(
+ long offset, long numBytes, long endOffset, String componentName) {
+ if (offset > endOffset || endOffset - offset < numBytes) {
+ throw new TableRuntimeException(
+ String.format(
+ "Malformed ISO WKB payload. Incomplete %s: expected %d byte(s) but found %d.",
+ componentName, numBytes, Math.max(0, endOffset - offset)));
+ }
+ return offset;
+ }
+
+ private static long multiplyExact(long value, int factor, String componentName) {
+ final long result = value * factor;
+ if (value != 0 && result / value != factor) {
+ throw new TableRuntimeException(
+ String.format("Malformed ISO WKB payload. %s size overflows.", componentName));
+ }
+ return result;
+ }
+
+ private static final class GeometryHeader {
+ private final int byteOrder;
+ private final int subtypeId;
+
+ private GeometryHeader(int byteOrder, int subtypeId) {
+ this.byteOrder = byteOrder;
+ this.subtypeId = subtypeId;
+ }
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java
index 45bcf7cd3f1128..04c9a0e50ed43d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinaryRowData.java
@@ -22,6 +22,7 @@
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -372,6 +373,14 @@ public byte[] getBinary(int pos) {
return BinarySegmentUtils.readBinary(segments, offset, fieldOffset, offsetAndLen);
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ assertIndexIsValid(pos);
+ int fieldOffset = getFieldOffset(pos);
+ final long offsetAndLen = segments[0].getLong(fieldOffset);
+ return BinarySegmentUtils.readGeographyData(segments, offset, fieldOffset, offsetAndLen);
+ }
+
@Override
public ArrayData getArray(int pos) {
assertIndexIsValid(pos);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java
index 653fc559df9bce..2dd12978d6bf5f 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/BinarySegmentUtils.java
@@ -1056,6 +1056,34 @@ public static byte[] readBinary(
}
}
+ /**
+ * Get geography data, if len less than 8, it will be included in variablePartOffsetAndLen.
+ *
+ * @param baseOffset base offset of composite binary format.
+ * @param fieldOffset absolute start offset of 'variablePartOffsetAndLen'.
+ * @param variablePartOffsetAndLen a long value, real data or offset and len.
+ */
+ public static BinaryGeographyData readGeographyData(
+ MemorySegment[] segments,
+ int baseOffset,
+ int fieldOffset,
+ long variablePartOffsetAndLen) {
+ long mark = variablePartOffsetAndLen & HIGHEST_FIRST_BIT;
+ if (mark == 0) {
+ final int subOffset = (int) (variablePartOffsetAndLen >> 32);
+ final int len = (int) variablePartOffsetAndLen;
+ return BinaryGeographyData.fromAddress(segments, baseOffset + subOffset, len);
+ } else {
+ int len = (int) ((variablePartOffsetAndLen & HIGHEST_SECOND_TO_EIGHTH_BIT) >>> 56);
+ if (BinarySegmentUtils.LITTLE_ENDIAN) {
+ return BinaryGeographyData.fromAddress(segments, fieldOffset, len);
+ } else {
+ // fieldOffset + 1 to skip header.
+ return BinaryGeographyData.fromAddress(segments, fieldOffset + 1, len);
+ }
+ }
+ }
+
/**
* Get binary string, if len less than 8, will be include in variablePartOffsetAndLen.
*
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java
index 27085b487ae50d..225d3e103fe103 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/NestedRowData.java
@@ -21,6 +21,7 @@
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -294,6 +295,14 @@ public byte[] getBinary(int pos) {
return BinarySegmentUtils.readBinary(segments, offset, fieldOffset, offsetAndLen);
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ assertIndexIsValid(pos);
+ int fieldOffset = getFieldOffset(pos);
+ final long offsetAndLen = BinarySegmentUtils.getLong(segments, fieldOffset);
+ return BinarySegmentUtils.readGeographyData(segments, offset, fieldOffset, offsetAndLen);
+ }
+
@Override
public RowData getRow(int pos, int numFields) {
assertIndexIsValid(pos);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java
index a3b8e7dd56daf9..9c6e6ad8c317d0 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarArrayData.java
@@ -21,6 +21,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -147,6 +148,12 @@ public byte[] getBinary(int pos) {
}
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ BytesColumnVector.Bytes byteArray = getByteArray(pos);
+ return GeographyData.fromBytes(byteArray.data, byteArray.offset, byteArray.len);
+ }
+
@Override
public ArrayData getArray(int pos) {
return ((ArrayColumnVector) data).getArray(offset + pos);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java
index bd3dce1edf6bae..0413ac980f737a 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/ColumnarRowData.java
@@ -21,6 +21,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -152,6 +153,12 @@ public byte[] getBinary(int pos) {
}
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ Bytes byteArray = vectorizedColumnBatch.getByteArray(rowId, pos);
+ return GeographyData.fromBytes(byteArray.data, byteArray.offset, byteArray.len);
+ }
+
@Override
public RowData getRow(int pos, int numFields) {
return vectorizedColumnBatch.getRow(rowId, pos);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java
index b9eff05a6c388f..d1420debfdb07a 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/JoinedRowData.java
@@ -21,6 +21,7 @@
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -269,6 +270,15 @@ public Bitmap getBitmap(int pos) {
}
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ if (pos < row1.getArity()) {
+ return row1.getGeography(pos);
+ } else {
+ return row2.getGeography(pos - row1.getArity());
+ }
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java
index 7f29ca5e613236..e6a173bb4cd991 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ProjectedRowData.java
@@ -22,6 +22,7 @@
import org.apache.flink.table.connector.Projection;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -178,6 +179,11 @@ public Bitmap getBitmap(int pos) {
return row.getBitmap(indexMapping[pos]);
}
+ @Override
+ public GeographyData getGeography(int pos) {
+ return row.getGeography(indexMapping[pos]);
+ }
+
@Override
public boolean equals(Object o) {
throw new UnsupportedOperationException("Projected row data cannot be compared");
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java
index 72a2534b64fdb1..4427b1be21993b 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/expressions/ValueLiteralExpression.java
@@ -21,6 +21,7 @@
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.logical.DayTimeIntervalType;
@@ -273,6 +274,11 @@ public String asSerializableString(SqlFactory sqlFactory) {
case BINARY:
case VARBINARY:
return String.format("X'%s'", StringUtils.byteToHexString((byte[]) value));
+ case GEOGRAPHY:
+ return String.format(
+ "ST_GEOGFROMWKB(X'%s')",
+ StringUtils.byteToHexString(
+ getValueAs(GeographyData.class).get().toBytes()));
case DATE:
return String.format("DATE '%s'", getValueAs(LocalDate.class).get());
case TIME_WITHOUT_TIME_ZONE:
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index dae5f976da5fd0..b95f6f6910688d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -3116,6 +3116,63 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
"org.apache.flink.table.runtime.functions.scalar.TryParseJsonFunction")
.build();
+ // --------------------------------------------------------------------------------------------
+ // Geography functions
+ // --------------------------------------------------------------------------------------------
+
+ public static final BuiltInFunctionDefinition ST_GEOGFROMTEXT =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("ST_GEOGFROMTEXT")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Collections.singletonList("text"),
+ Collections.singletonList(
+ logical(LogicalTypeFamily.CHARACTER_STRING))))
+ .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.GEOGRAPHY())))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.StGeogFromTextFunction")
+ .build();
+
+ public static final BuiltInFunctionDefinition ST_GEOGFROMWKB =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("ST_GEOGFROMWKB")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Collections.singletonList("bytes"),
+ Collections.singletonList(
+ logical(LogicalTypeFamily.BINARY_STRING))))
+ .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.GEOGRAPHY())))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.StGeogFromWkbFunction")
+ .build();
+
+ public static final BuiltInFunctionDefinition ST_ASTEXT =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("ST_ASTEXT")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Collections.singletonList("geography"),
+ Collections.singletonList(logical(LogicalTypeRoot.GEOGRAPHY))))
+ .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.STRING())))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.StAsTextFunction")
+ .build();
+
+ public static final BuiltInFunctionDefinition ST_ASWKB =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("ST_ASWKB")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Collections.singletonList("geography"),
+ Collections.singletonList(logical(LogicalTypeRoot.GEOGRAPHY))))
+ .outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BYTES())))
+ .runtimeClass("org.apache.flink.table.runtime.functions.scalar.StAsWkbFunction")
+ .build();
+
// --------------------------------------------------------------------------------------------
// Bitmap functions
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/GeographyType.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/GeographyType.java
new file mode 100644
index 00000000000000..403ee88e0cfccd
--- /dev/null
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/GeographyType.java
@@ -0,0 +1,83 @@
+/*
+ * 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.flink.table.types.logical;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.data.GeographyData;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Data type of geography data.
+ *
+ *
The serializable string representation of this type is {@code GEOGRAPHY}.
+ */
+@PublicEvolving
+public final class GeographyType extends LogicalType {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String FORMAT = "GEOGRAPHY";
+
+ private static final Class INPUT_OUTPUT_CONVERSION = GeographyData.class;
+
+ public GeographyType(boolean isNullable) {
+ super(isNullable, LogicalTypeRoot.GEOGRAPHY);
+ }
+
+ public GeographyType() {
+ this(true);
+ }
+
+ @Override
+ public LogicalType copy(boolean isNullable) {
+ return new GeographyType(isNullable);
+ }
+
+ @Override
+ public String asSerializableString() {
+ return withNullability(FORMAT);
+ }
+
+ @Override
+ public boolean supportsInputConversion(Class clazz) {
+ return INPUT_OUTPUT_CONVERSION.isAssignableFrom(clazz);
+ }
+
+ @Override
+ public boolean supportsOutputConversion(Class clazz) {
+ return INPUT_OUTPUT_CONVERSION.isAssignableFrom(clazz);
+ }
+
+ @Override
+ public Class getDefaultConversion() {
+ return INPUT_OUTPUT_CONVERSION;
+ }
+
+ @Override
+ public List getChildren() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public R accept(LogicalTypeVisitor visitor) {
+ return visitor.visit(this);
+ }
+}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java
index 6c823add433bfb..c21abce4af7461 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeRoot.java
@@ -145,7 +145,9 @@ public enum LogicalTypeRoot {
VARIANT(LogicalTypeFamily.EXTENSION),
- BITMAP(LogicalTypeFamily.EXTENSION);
+ BITMAP(LogicalTypeFamily.EXTENSION),
+
+ GEOGRAPHY(LogicalTypeFamily.EXTENSION);
private final Set families;
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
index 6a0e5614466d11..b35ffe05541f44 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/LogicalTypeVisitor.java
@@ -101,5 +101,9 @@ default R visit(BitmapType bitmapType) {
return visit((LogicalType) bitmapType);
}
+ default R visit(GeographyType geographyType) {
+ return visit((LogicalType) geographyType);
+ }
+
R visit(LogicalType other);
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java
index 056ec5953ccdc3..43237c443f56e9 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeDefaultVisitor.java
@@ -31,6 +31,7 @@
import org.apache.flink.table.types.logical.DistinctType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -203,6 +204,11 @@ public R visit(BitmapType bitmapType) {
return defaultMethod(bitmapType);
}
+ @Override
+ public R visit(GeographyType geographyType) {
+ return defaultMethod(geographyType);
+ }
+
@Override
public R visit(LogicalType other) {
return defaultMethod(other);
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java
index 3711ac76f4fd2a..dfa2e7cf358147 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeParser.java
@@ -36,6 +36,7 @@
import org.apache.flink.table.types.logical.DescriptorType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
@@ -336,7 +337,8 @@ private enum Keyword {
DESCRIPTOR,
STRUCTURED,
VARIANT,
- BITMAP
+ BITMAP,
+ GEOGRAPHY
}
private static final Set KEYWORDS =
@@ -586,6 +588,8 @@ private LogicalType parseTypeByKeyword() {
return new VariantType();
case BITMAP:
return new BitmapType();
+ case GEOGRAPHY:
+ return new GeographyType();
default:
throw parsingError("Unsupported type: " + token().value);
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java
index 98629b15fc1770..fa994ed160f3c7 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeUtils.java
@@ -21,6 +21,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RawValueData;
import org.apache.flink.table.data.RowData;
@@ -115,6 +116,8 @@ public static Class toInternalConversionClass(LogicalType type) {
return Variant.class;
case BITMAP:
return Bitmap.class;
+ case GEOGRAPHY:
+ return GeographyData.class;
case SYMBOL:
case UNRESOLVED:
default:
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java
index 9aa83f3cbee1d3..929aaec786f490 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/LogicalTypeDataTypeConverter.java
@@ -38,6 +38,7 @@
import org.apache.flink.table.types.logical.DistinctType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -265,6 +266,11 @@ public DataType visit(BitmapType bitmapType) {
return new AtomicDataType(bitmapType);
}
+ @Override
+ public DataType visit(GeographyType geographyType) {
+ return new AtomicDataType(geographyType);
+ }
+
@Override
public DataType visit(LogicalType other) {
if (other.is(LogicalTypeRoot.UNRESOLVED)) {
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java
index 7b54e40f387935..7b4aef14cc3a80 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java
@@ -51,5 +51,21 @@ void testFromResolvedSchema() {
assertThat(newSchema.resolve(new TestSchemaResolver())).isEqualTo(originalSchema);
}
+
+ @Test
+ void testGeographyColumn() {
+ final Schema schema =
+ Schema.newBuilder()
+ .column("location", DataTypes.GEOGRAPHY())
+ .column("required_location", DataTypes.GEOGRAPHY().notNull())
+ .build();
+
+ assertThat(schema.resolve(new TestSchemaResolver()))
+ .isEqualTo(
+ ResolvedSchema.of(
+ Column.physical("location", DataTypes.GEOGRAPHY()),
+ Column.physical(
+ "required_location", DataTypes.GEOGRAPHY().notNull())));
+ }
}
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/BinaryGeographyDataTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/BinaryGeographyDataTest.java
new file mode 100644
index 00000000000000..4e45d2d28549aa
--- /dev/null
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/BinaryGeographyDataTest.java
@@ -0,0 +1,331 @@
+/*
+ * 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.flink.table.data.binary;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.GeographyData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.GeographyType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Arrays;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link BinaryGeographyData}. */
+class BinaryGeographyDataTest {
+
+ private static final int BIG_ENDIAN = 0;
+ private static final int LITTLE_ENDIAN = 1;
+ private static final byte[] POINT_WKB = pointWkb(LITTLE_ENDIAN);
+
+ @Test
+ void testValidWkbRoundTrip() {
+ final GeographyData geography = GeographyData.fromBytes(POINT_WKB);
+
+ assertThat(geography).isInstanceOf(BinaryGeographyData.class);
+ assertThat(geography.toBytes()).isEqualTo(POINT_WKB);
+ assertThat(geography.subtypeId()).isEqualTo(GeographyData.POINT);
+ assertThat(geography.sizeInBytes()).isEqualTo(POINT_WKB.length);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("supportedSubtypePayloads")
+ void testSupported2dSubtypePayloads(String name, int subtypeId, byte[] wkb) {
+ final BinaryGeographyData geography = BinaryGeographyData.fromBytes(wkb);
+
+ assertThat(geography.subtypeId()).isEqualTo(subtypeId);
+ assertThat(geography.toBytes()).isEqualTo(wkb);
+ assertThat(geography.sizeInBytes()).isEqualTo(wkb.length);
+ }
+
+ static Stream supportedSubtypePayloads() {
+ return Stream.of(
+ Arguments.of("Point", GeographyData.POINT, pointWkb(LITTLE_ENDIAN)),
+ Arguments.of("LineString", GeographyData.LINE_STRING, lineStringWkb(LITTLE_ENDIAN)),
+ Arguments.of("Polygon", GeographyData.POLYGON, polygonWkb(LITTLE_ENDIAN)),
+ Arguments.of("MultiPoint", GeographyData.MULTI_POINT, multiPointWkb(LITTLE_ENDIAN)),
+ Arguments.of(
+ "MultiLineString",
+ GeographyData.MULTI_LINE_STRING,
+ multiLineStringWkb(LITTLE_ENDIAN)),
+ Arguments.of(
+ "MultiPolygon",
+ GeographyData.MULTI_POLYGON,
+ multiPolygonWkb(LITTLE_ENDIAN)),
+ Arguments.of(
+ "GeometryCollection",
+ GeographyData.GEOMETRY_COLLECTION,
+ geometryCollectionWkb(LITTLE_ENDIAN)));
+ }
+
+ @Test
+ void testBigEndianSubtypeExtraction() {
+ final BinaryGeographyData point = BinaryGeographyData.fromBytes(pointWkb(BIG_ENDIAN));
+ final BinaryGeographyData lineString =
+ BinaryGeographyData.fromBytes(lineStringWkb(BIG_ENDIAN));
+
+ assertThat(point.subtypeId()).isEqualTo(GeographyData.POINT);
+ assertThat(lineString.subtypeId()).isEqualTo(GeographyData.LINE_STRING);
+ }
+
+ @Test
+ void testNullHandling() {
+ assertThat(GeographyData.fromBytes((byte[]) null)).isNull();
+ assertThat(GeographyData.fromBytes(null, 0, 0)).isNull();
+ assertThat(BinaryGeographyData.fromBytes((byte[]) null)).isNull();
+ assertThat(BinaryGeographyData.fromBytes(null, 0, 0)).isNull();
+ }
+
+ @Test
+ void testGenericRowDataNullPath() {
+ final RowData.FieldGetter getter = RowData.createFieldGetter(new GeographyType(), 0);
+ final GenericRowData nullRow = GenericRowData.of((Object) null);
+ final GeographyData geography = GeographyData.fromBytes(POINT_WKB);
+ final GenericRowData valueRow = GenericRowData.of(geography);
+
+ assertThat(getter.getFieldOrNull(nullRow)).isNull();
+ assertThat(getter.getFieldOrNull(valueRow)).isSameAs(geography);
+ }
+
+ @Test
+ void testBinaryRowDataNullPath() {
+ final RowData.FieldGetter getter = RowData.createFieldGetter(new GeographyType(), 0);
+ final int fixedLength = BinaryRowData.calculateFixPartSizeInBytes(1);
+ final BinaryRowData nullRow = new BinaryRowData(1);
+ nullRow.pointTo(MemorySegmentFactory.wrap(new byte[fixedLength]), 0, fixedLength);
+ nullRow.setNullAt(0);
+
+ assertThat(getter.getFieldOrNull(nullRow)).isNull();
+ }
+
+ @Test
+ void testBinaryRowDataValuePath() {
+ final BinaryRowData row = binaryRowWithGeography(POINT_WKB);
+
+ assertThat(row.getGeography(0).toBytes()).isEqualTo(POINT_WKB);
+ assertThat(row.getGeography(0).subtypeId()).isEqualTo(GeographyData.POINT);
+ }
+
+ @Test
+ void testByteRangeCopiesOnlySelectedPayload() {
+ final byte[] bytes = concat(bytes(42), POINT_WKB, bytes(99));
+ final BinaryGeographyData geography =
+ BinaryGeographyData.fromBytes(bytes, 1, POINT_WKB.length);
+
+ bytes[1] = 0;
+
+ assertThat(geography.toBytes()).isEqualTo(POINT_WKB);
+ assertThat(geography.subtypeId()).isEqualTo(GeographyData.POINT);
+ }
+
+ @Test
+ void testFromAddressSupportsSegmentBoundary() {
+ final byte[] first = concat(new byte[14], Arrays.copyOfRange(POINT_WKB, 0, 2));
+ final byte[] second = Arrays.copyOfRange(POINT_WKB, 2, 18);
+ final byte[] third =
+ concat(Arrays.copyOfRange(POINT_WKB, 18, POINT_WKB.length), new byte[13]);
+ final MemorySegment[] segments =
+ new MemorySegment[] {
+ MemorySegmentFactory.wrap(first),
+ MemorySegmentFactory.wrap(second),
+ MemorySegmentFactory.wrap(third)
+ };
+
+ final BinaryGeographyData geography =
+ BinaryGeographyData.fromAddress(segments, 14, POINT_WKB.length);
+
+ assertThat(geography.toBytes()).isEqualTo(POINT_WKB);
+ assertThat(geography.subtypeId()).isEqualTo(GeographyData.POINT);
+ }
+
+ @Test
+ void testMalformedPayloadBoundaries() {
+ assertThatThrownBy(() -> BinaryGeographyData.fromBytes(bytes()))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Incomplete WKB header");
+
+ assertThatThrownBy(() -> BinaryGeographyData.fromBytes(bytes(1, 1, 0, 0)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Incomplete WKB header");
+
+ assertThatThrownBy(() -> BinaryGeographyData.fromBytes(bytes(2, 1, 0, 0, 0)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Unsupported byte order 2");
+
+ assertThatThrownBy(() -> BinaryGeographyData.fromBytes(header(LITTLE_ENDIAN, 0)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Unsupported geography subtype ID 0");
+
+ assertThatThrownBy(() -> BinaryGeographyData.fromBytes(header(LITTLE_ENDIAN, 8)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Unsupported geography subtype ID 8");
+ }
+
+ @Test
+ void testStructurallyIncompletePayloads() {
+ assertThatThrownBy(
+ () ->
+ BinaryGeographyData.fromBytes(
+ header(LITTLE_ENDIAN, GeographyData.POINT)))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Incomplete POINT coordinates");
+
+ assertThatThrownBy(
+ () ->
+ BinaryGeographyData.fromBytes(
+ concat(
+ header(LITTLE_ENDIAN, GeographyData.LINE_STRING),
+ unsignedInt(LITTLE_ENDIAN, 1),
+ new byte[15])))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Incomplete coordinate sequence");
+
+ assertThatThrownBy(
+ () ->
+ BinaryGeographyData.fromBytes(
+ concat(
+ header(LITTLE_ENDIAN, GeographyData.MULTI_POINT),
+ unsignedInt(LITTLE_ENDIAN, 1),
+ lineStringWkb(LITTLE_ENDIAN))))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Expected nested subtype ID 1 but found 2");
+
+ assertThatThrownBy(
+ () ->
+ BinaryGeographyData.fromBytes(
+ concat(pointWkb(LITTLE_ENDIAN), bytes(42))))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("trailing byte");
+ }
+
+ @Test
+ void testInvalidByteRanges() {
+ assertThatThrownBy(() -> BinaryGeographyData.fromBytes(POINT_WKB, -1, POINT_WKB.length))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Invalid ISO WKB byte range");
+
+ assertThatThrownBy(() -> BinaryGeographyData.fromBytes(POINT_WKB, 0, POINT_WKB.length + 1))
+ .isInstanceOf(TableRuntimeException.class)
+ .hasMessageContaining("Invalid ISO WKB byte range");
+ }
+
+ private static BinaryRowData binaryRowWithGeography(byte[] wkb) {
+ final int fixedLength = BinaryRowData.calculateFixPartSizeInBytes(1);
+ final byte[] rowBytes = new byte[fixedLength + wkb.length];
+ final MemorySegment segment = MemorySegmentFactory.wrap(rowBytes);
+ segment.putLong(8, ((long) fixedLength << 32) | wkb.length);
+ segment.put(fixedLength, wkb, 0, wkb.length);
+
+ final BinaryRowData row = new BinaryRowData(1);
+ row.pointTo(segment, 0, rowBytes.length);
+ return row;
+ }
+
+ private static byte[] pointWkb(int byteOrder) {
+ return concat(header(byteOrder, GeographyData.POINT), new byte[16]);
+ }
+
+ private static byte[] lineStringWkb(int byteOrder) {
+ return concat(
+ header(byteOrder, GeographyData.LINE_STRING),
+ unsignedInt(byteOrder, 2),
+ new byte[32]);
+ }
+
+ private static byte[] polygonWkb(int byteOrder) {
+ return concat(
+ header(byteOrder, GeographyData.POLYGON),
+ unsignedInt(byteOrder, 1),
+ unsignedInt(byteOrder, 4),
+ new byte[64]);
+ }
+
+ private static byte[] multiPointWkb(int byteOrder) {
+ return concat(
+ header(byteOrder, GeographyData.MULTI_POINT),
+ unsignedInt(byteOrder, 1),
+ pointWkb(byteOrder));
+ }
+
+ private static byte[] multiLineStringWkb(int byteOrder) {
+ return concat(
+ header(byteOrder, GeographyData.MULTI_LINE_STRING),
+ unsignedInt(byteOrder, 1),
+ lineStringWkb(byteOrder));
+ }
+
+ private static byte[] multiPolygonWkb(int byteOrder) {
+ return concat(
+ header(byteOrder, GeographyData.MULTI_POLYGON),
+ unsignedInt(byteOrder, 1),
+ polygonWkb(byteOrder));
+ }
+
+ private static byte[] geometryCollectionWkb(int byteOrder) {
+ return concat(
+ header(byteOrder, GeographyData.GEOMETRY_COLLECTION),
+ unsignedInt(byteOrder, 2),
+ pointWkb(byteOrder),
+ lineStringWkb(byteOrder));
+ }
+
+ private static byte[] header(int byteOrder, int subtypeId) {
+ return concat(bytes(byteOrder), unsignedInt(byteOrder, subtypeId));
+ }
+
+ private static byte[] unsignedInt(int byteOrder, int value) {
+ if (byteOrder == LITTLE_ENDIAN) {
+ return bytes(value, value >>> 8, value >>> 16, value >>> 24);
+ }
+ return bytes(value >>> 24, value >>> 16, value >>> 8, value);
+ }
+
+ private static byte[] concat(byte[]... values) {
+ int length = 0;
+ for (byte[] value : values) {
+ length += value.length;
+ }
+
+ final byte[] result = new byte[length];
+ int offset = 0;
+ for (byte[] value : values) {
+ System.arraycopy(value, 0, result, offset, value.length);
+ offset += value.length;
+ }
+ return result;
+ }
+
+ private static byte[] bytes(int... values) {
+ final byte[] result = new byte[values.length];
+ for (int i = 0; i < values.length; i++) {
+ result[i] = (byte) values[i];
+ }
+ return result;
+ }
+}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java
index 8c274ed719137e..3c0bf063631459 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/DataTypesTest.java
@@ -21,6 +21,7 @@
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GeographyData;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BinaryType;
@@ -32,6 +33,7 @@
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -82,6 +84,7 @@
import static org.apache.flink.table.api.DataTypes.DOUBLE;
import static org.apache.flink.table.api.DataTypes.FIELD;
import static org.apache.flink.table.api.DataTypes.FLOAT;
+import static org.apache.flink.table.api.DataTypes.GEOGRAPHY;
import static org.apache.flink.table.api.DataTypes.INT;
import static org.apache.flink.table.api.DataTypes.INTERVAL;
import static org.apache.flink.table.api.DataTypes.MAP;
@@ -231,6 +234,9 @@ private static Stream testData() {
TestSpec.forDataType(BITMAP())
.expectLogicalType(new BitmapType())
.expectConversionClass(Bitmap.class),
+ TestSpec.forDataType(GEOGRAPHY())
+ .expectLogicalType(new GeographyType())
+ .expectConversionClass(GeographyData.class),
TestSpec.forUnresolvedDataType(RAW(Types.VOID))
.expectUnresolvedString("[RAW('java.lang.Void', '?')]")
.lookupReturns(dummyRaw(Void.class))
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
index 6fb3574b1f59f6..89cc1341dfe2a8 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeCastsTest.java
@@ -31,6 +31,7 @@
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -295,7 +296,13 @@ private static Stream testData() {
new VariantType(),
new MapType(new IntType(), new CharType()),
false,
- false));
+ false),
+
+ // GEOGRAPHY construction and serialization require explicit functions.
+ Arguments.of(new GeographyType(), VarCharType.STRING_TYPE, false, false),
+ Arguments.of(VarCharType.STRING_TYPE, new GeographyType(), false, false),
+ Arguments.of(new GeographyType(), new VarBinaryType(), false, false),
+ Arguments.of(new VarBinaryType(), new GeographyType(), false, false));
}
@ParameterizedTest(name = "{index}: [From: {0}, To: {1}, Implicit: {2}, Explicit: {3}]")
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java
index 258b397c871878..e1c96b1fbc9e7c 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypeParserTest.java
@@ -39,6 +39,7 @@
import org.apache.flink.table.types.logical.DescriptorType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LegacyTypeInformationType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
@@ -309,6 +310,8 @@ private static Stream testData() {
TestSpec.forString("VARIANT NOT NULL").expectType(new VariantType(false)),
TestSpec.forString("BITMAP").expectType(new BitmapType()),
TestSpec.forString("BITMAP NOT NULL").expectType(new BitmapType(false)),
+ TestSpec.forString("GEOGRAPHY").expectType(new GeographyType()),
+ TestSpec.forString("GEOGRAPHY NOT NULL").expectType(new GeographyType(false)),
// error message testing
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
index ee40463cf0b92d..b608d923cc3b6a 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/LogicalTypesTest.java
@@ -26,6 +26,8 @@
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.UnresolvedIdentifier;
+import org.apache.flink.table.data.GeographyData;
+import org.apache.flink.table.data.binary.BinaryGeographyData;
import org.apache.flink.table.expressions.TimeIntervalUnit;
import org.apache.flink.table.legacy.types.logical.TypeInformationRawType;
import org.apache.flink.table.types.logical.ArrayType;
@@ -40,6 +42,7 @@
import org.apache.flink.table.types.logical.DistinctType;
import org.apache.flink.table.types.logical.DoubleType;
import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -629,6 +632,20 @@ void testBitmapType() {
new BitmapType(false)));
}
+ @Test
+ void testGeographyType() {
+ assertThat(new GeographyType())
+ .isJavaSerializable()
+ .satisfies(
+ baseAssertions(
+ "GEOGRAPHY",
+ "GEOGRAPHY",
+ new Class[] {GeographyData.class, BinaryGeographyData.class},
+ new Class[] {GeographyData.class, BinaryGeographyData.class},
+ new LogicalType[] {},
+ new GeographyType(false)));
+ }
+
@Test
void testTypeInformationRawType() {
final TypeInformationRawType rawType =
diff --git a/flink-table/flink-table-planner/pom.xml b/flink-table/flink-table-planner/pom.xml
index 450dc204ec69ed..584b5aa31f22e8 100644
--- a/flink-table/flink-table-planner/pom.xml
+++ b/flink-table/flink-table-planner/pom.xml
@@ -189,6 +189,19 @@ under the License.
test