[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:

• fccf312 Azure: SUCCESS

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?