[FLINK-39904][table] Add GEOGRAPHY logical type support

[davidchaava]

What is the purpose of the change

This pull request adds GEOGRAPHY as a logical type in the Table & SQL API.

The change introduces the core type plumbing required for Flink to recognize,
parse, serialize, validate, and carry GEOGRAPHY values through the table runtime.
This PR intentionally does not add geography SQL functions or format-specific
support. Those parts are planned as follow-up changes.

Jira: https://issues.apache.org/jira/browse/FLINK-39904
Discussion: https://lists.apache.org/thread/flbj8dfdfgs26klrxt7xch3r9785ky67

Brief change log

• Added GEOGRAPHY as a new logical type root and introduced GeographyType.
• Added DataTypes.GEOGRAPHY() for declaring GEOGRAPHY columns in the Java API.
• Added SQL parser, logical type parser, and JSON serialization/deserialization support.
• Added planner and Calcite type conversion support for GEOGRAPHY.
• Added internal runtime support for carrying GEOGRAPHY values through row data.
• Added binary row, row data, writer, and code generation integration.
• Added GeographyTypeSerializer and serializer snapshot support.
• Added validation for unsupported GEOGRAPHY conversions.

Verifying this change

This change added tests and can be verified as follows:

• Added tests for DataTypes.GEOGRAPHY() and logical type parsing.
• Added tests for logical type JSON serialization/deserialization.
• Added tests for planner type conversion and validation behavior.
• Added tests for binary GEOGRAPHY data handling.
• Added serializer compatibility tests through SerializerTestBase.
• Extended row data and row serializer tests to cover GEOGRAPHY values.
• Extended type serializer coverage checks for the new serializer.

Does this pull request potentially affect one of the following parts:

• Dependencies: no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes
• The serializers: yes
• The runtime per-record code paths: yes
• Anything that affects deployment or recovery: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs and follow-up documentation PR

Follow-up changes

The following parts are intentionally left out of this PR:

• GEOGRAPHY SQL constructor/accessor functions such as ST_*.
• PyFlink exposure for GEOGRAPHY schemas.
• Format-specific support, including Parquet integration.

[flinkbot]

CI report:

• d105ec2 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[gkalashyan-akv]

@flinkbot run azure

[gkalashyan-akv]

@flinkbot run azure

[davidradl]

Please could you add documentation so it is obvious what the externals are .

[davidchaava]

Thanks, added docs for the GEOGRAPHY type and its external representation.

[talatuyarer]

@mxm Could you help us to review this PR ?

[dalelane]

I'm curious about the decision here around the performance and risk of eager validation: every getGeography() call on a BinaryRowData or BinaryArrayData traverses the entire WKB payload during construction, whereas other fromAddress-based types (BinaryStringData, BinaryRawValueData, etc.) can be copied lazily and defer any expensive work.

For a GEOGRAPHY field read in a hot loop (projection, filter, serialisation), the full WKB parse fires every single time rather than once on ingestion.

If the WKB is somehow malformed in a way the ingestion path didn't catch (e.g. written directly by a connector that bypasses fromBytes), the TableRuntimeException is thrown mid-operator rather than at the boundary. Whichever call site happens to invoke getGeography() first would cause the exception, making it perhaps difficult to attribute and recover from, maybe even non-deterministic (relative to where the bad bytes were introduced)

Should WKB structural validation happen once at construction/deserialization time (e.g. when the row arrives from the source/deserializer) and be treated as trusted thereafter, with fromAddress on a row just doing a cheap wrap like BinaryStringData?

Or is re-validation on every read intentional (e.g. defense against corrupted/unsafe memory since BinarySection doesn't own the segments)?

[dalelane]

BinaryGeographyData inherits a hashCode() method by extending BinarySection, so hashing will be based on the binary representation of the geography.

Will geographies that are semantically equal serialize to identical bytes? (e.g. same geometry encoded as big-endian vs little-endian - they would hash differently right? )

Is the assumption that users need to normalize to some canonical byte order up-front so comparisons in Flink work as expected?

[davidchaava]

Thanks, good catch. I changed fromAddress to follow the lazy binary-data pattern: row/array access now only wraps the existing segments and does not validate the full WKB payload. Explicit construction from byte arrays still validates the full payload, so malformed input is rejected at the conversion boundary. I added row/array tests for this path as well.

On equality/hashCode: this layer intentionally uses byte-representation equality, not topological equality. We keep it byte-based to preserve the cheap and predictable behavior of Flink's binary runtime values. Different valid WKB encodings of the same geometry may therefore compare/hash differently unless they are normalized before being stored. Full topological equality would require parsing and normalization on comparison/hash paths, which is more expensive and needs separate geospatial semantics. We also aligned the proposal wording to make this explicit.

[davidchaava]

@mxm Could you help us to review this PR ?

Hi @mxm - bumping this in case it slipped by. Would you have bandwidth to review when you get a chance?

[mxm]

Thanks @davidchaava for the PR! I’ll have a look, but perhaps @twalthr could review as well? 🙏

[davidradl]

nit: I think we should provide a references to the new acronyms EWKB, CRS, SRID, OGC:CRS84 and ISO WKB somewhere in the wording. So the user can find the meanings of these acronyms and how they should use them here.

[davidchaava]

Addressed. I expanded the GEOGRAPHY documentation in
docs/content/docs/sql/reference/data-types.md.

It now explains CRS, SRID, ISO WKB, and EWKB, and includes references for OGC:CRS84, OGC Simple Feature Access, and EWKB. I also clarified that the v1 contract uses standard 2D ISO WKB, while EWKB is not supported and CRS/SRID metadata is not stored in the WKB payload.

Thanks for pointing this out.

[davidradl]

My AI says :
Overall: The structural approach is solid — it correctly follows the BITMAP precedent at every layer (logical type → parser → planner → binary runtime → serializer). But there are issues worth blocking on:

The one I'd block merge on
(int) cast truncation in BinaryGeographyData.readUnsignedInt() — the validator traverses the WKB tree with a long cursor, but every call to BinarySegmentUtils.getByte(segments, (int) offset) silently truncates the offset. For any geometry whose internal offset exceeds Integer.MAX_VALUE in a native memory segment, this either reads the wrong byte or throws an out-of-bounds exception. Either document a hard 2 GB limit and enforce it in checkRange, or pass the long through to a 64-bit accessor.

Other notable findings
writeGeography() in AbstractBinaryWriter calls geography.toBytes() which always allocates a new heap array — even for a BinaryGeographyData that already has a segment. The reader path correctly uses zero-copy fromAddress; the writer should too.
WKB subtype constants on @PublicEvolving GeographyData — hardcoding POINT=1…GEOMETRY_COLLECTION=7 on a public interface is a maintenance trap; ISO WKB extended types (3D, 4D) reuse these same IDs with different coordinate packing.
isFlinkExtensionType() in FlinkTypeFactory is a growing instanceof chain (RAW, BITMAP, now GEOGRAPHY) with no marker interface — a guaranteed maintenance hazard for every future extension type.
Columnar getGeography() in ColumnarArrayData/ColumnarRowData calls GeographyData.fromBytes() (full WKB validation) on every row access, where the binary counterparts use zero-copy fromAddress.
skipTypedGeometryCollection() receives a byteOrder param it never uses — dead parameter that misleads readers.
createInstance()'s raw byte literal for the empty GEOMETRYCOLLECTION is fragile — no test validates it; compare to BitmapSerializer which builds its default via the public API.
The unrelated async test fix in AbstractAsyncRunnableStreamOperatorTest should be a separate PR.

[dalelane]

This would end up advertising GEOGRAPHY to REST clients that isn't supported by the gateway yet.

I know your PR description says that follow-up changes are coming, but do you think it's worth making updates to LogicalTypeJsonSerializer.java and LogicalTypeJsonDeserializer.java in flink-table/flink-sql-gateway in this PR?

It probably just needs a new case in the switch statements in serializeInternal and deserializeInternal

[davidchaava]

Yes, agreed. Since the generated REST schema advertises GEOGRAPHY, the SQL Gateway should support the logical type in the same PR.

We added GEOGRAPHY cases to LogicalTypeJsonSerializer and LogicalTypeJsonDeserializer, together with JSON round-trip test coverage. This allows REST clients to serialize and deserialize GEOGRAPHY type metadata consistently.

The SQL constructors, accessors, and spatial functions are implemented separately in the follow-up PR #28788.

[dalelane]

do you think we need a new datatype converter for geography in https://github.com/akvelon/flink/blob/fccf3126db72d8fbbb3829f81c4f123d1f005df4/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/util/DataFormatConverters.java#L115-L165 ?

[davidchaava]

Yes, we did need a dedicated converter for GEOGRAPHY, so we added one following the existing identity-converter pattern.

DataFormatConverters now registers GeographyConverter for both GeographyData and BinaryGeographyData. The converter preserves values as-is and reads row fields through RowData.getGeography().

We also added coverage for converter lookup, nullable and non-nullable types, WKB round-trip, row extraction, null handling, BinaryGeographyData bridging, and nested ARRAY<GEOGRAPHY> conversion.

Thanks for pointing this out.

[dalelane]

@davidradl #28740 (review)

The one I'd block merge on
(int) cast truncation in BinaryGeographyData.readUnsignedInt() — the validator traverses the WKB tree with a long cursor, but every call to BinarySegmentUtils.getByte(segments, (int) offset) silently truncates the offset. For any geometry whose internal offset exceeds Integer.MAX_VALUE in a native memory segment, this either reads the wrong byte or throws an out-of-bounds exception. Either document a hard 2 GB limit and enforce it in checkRange, or pass the long through to a 64-bit accessor.

I think the use of a long is intentional - it's about protecting against arithmetic overflow if the data was corrupted somehow. WKB data contains counts. A LineString declares how many points it has, and we scale that up to work out how many bytes are needed. If we did use an int for that, we really would run the risk of an overflow.

If you imagine malformed data claiming that it had 4 billion points (which is about as high as it could claim as an unsigned 32-bit number) in a 100-byte payload:

1. numPoints = 4,000,000,000 (read as a long)
2. 4,000,000,000 points × 16 = 64,000,000,000 bytes (again, computed as a long, so theres no wraparound)
3. requireBytes compares that against the 100 bytes actually available, and can throw the exception
4. getByte is never called, so no cast to an int never happens

The point is that the huge number gets rejected while it's still a long. Only values confirmed to fit inside the real payload will reach the cast.

I don't think it's reasonable to describe this as silently truncating, as each cast is preceded by a bounds check.

At any rate, offset points into a Flink binary row and Flink's binary format is int-addressed (the offset parameter in getByte is an int). All the binary types have the same 2gb ceiling, so I don't think a 64-bit accessor is an available option here.

@davidchaava - maybe it's worth adding a comment to help the next reader not tripping over this though?

.

isFlinkExtensionType() in FlinkTypeFactory is a growing instanceof chain (RAW, BITMAP, now GEOGRAPHY) with no marker interface — a guaranteed maintenance hazard for every future extension type.

It's only three items, so this wouldn't worry me overly, but perhaps the more interesting aspect is that this would change how leastRestrictive is handled for RAW and BITMAP.

Before this change, they would fall through to Calcite's super.leastRestrictive. But with this change, they'll go to LogicalTypeMerging.findCommonType instead. @davidchaava - is that intentional?

.

skipTypedGeometryCollection() receives a byteOrder param it never uses — dead parameter that misleads readers.

It does use it. In the first line, it calls readUnsignedInt with it, which uses it to switch on how to use BinarySegmentUtils.getByte

.

createInstance()'s raw byte literal for the empty GEOMETRYCOLLECTION is fragile — no test validates it; compare to BitmapSerializer which builds its default via the public API.

testCreateInstanceReturnsValidGeographyData in GeographyTypeSerializerTest has a test that validates it

And because it's included in getTestData(), it'll have the entire SerializerTestBase suite run against it.

(Feels like it's low risk at any rate, because if the literal was wrong, it'd throw an exception the first time it was used)

[davidradl]

@dalelane thanks for your feedback on the AI review. That makes sense. @davidchaava I think a comment to describe this would be good as suggested by Dale. Also a unit test driving the code with a large long value and showing it failing would be good to confirm this code path works as expected.

[davidchaava]

@davidradl @dalelane thanks for the detailed review. We went through the findings and updated the PR accordingly.

The validator traverses the WKB tree with a long cursor, but the accessor truncates the offset to int.

Addressed. We added checked address conversion and range validation in BinaryGeographyData.

Following David’s follow-up comment, we also added an explanatory comment describing why WKB counts and offsets are handled with long arithmetic, while Flink’s binary memory access remains int-addressed and is performed only after range validation.

The test uses a large unsigned 32-bit WKB count (0xFFFFFFFF) and verifies that the malformed payload is rejected based on the validated long range before any truncated memory access can occur.

writeGeography() ... always allocates a new heap array.

Addressed. AbstractBinaryWriter now uses a zero-copy path for BinaryGeographyData and keeps the existing byte-array fallback for other implementations.

getGeography() ... calls GeographyData.fromBytes() on every row access.

Addressed. Columnar GEOGRAPHY access now uses lazy trusted binary views.

WKB subtype constants ... need clarification.

Addressed. The supported 2D ISO WKB subtype contract is now documented, including the distinction from EWKB and extended dimensional encodings.

createInstance()'s raw byte literal for the empty GEOMETRYCOLLECTION is fragile.

Addressed. We added a test confirming that the default instance is a valid GEOMETRYCOLLECTION.

The unrelated async test fix ... should be a separate PR.

The unrelated change has been removed from this PR.

We also expanded the GEOGRAPHY documentation with references for CRS, SRID, OGC:CRS84, ISO WKB, and EWKB.

The SQL constructors, accessors, and spatial functions are being handled separately in follow-up PR #28788.

The PR should now be ready for another review round. Thanks again for the thorough feedback.