[FLINK-39904][table] Add GeoParquet-compatible GEOGRAPHY Parquet support

[davidchaava]

What is the purpose of the change

This PR adds Parquet read/write support for the GEOGRAPHY logical type using GeoParquet-compatible metadata on top of Flink's existing Parquet version.

GEOGRAPHY values are stored as Parquet BINARY values with WKB encoding. The writer emits GeoParquet geo key-value metadata with encoding=WKB and edges=spherical, allowing downstream systems such as BigQuery to recognize the column as GEOGRAPHY.

This keeps the implementation compatible with Flink's current Parquet dependency (parquet-mr 1.15.2) and does not require upgrading to the native Parquet GEOGRAPHY logical type.

Brief change log

• Map Flink GEOGRAPHY to Parquet BINARY.
• Emit GeoParquet 1.1.0 metadata for top-level GEOGRAPHY columns.
• Add GEOGRAPHY support to Parquet row writing.
• Add GEOGRAPHY support to Parquet vectorized reading, including nested values.
• Add Parquet writer and reader coverage for GEOGRAPHY, nested GEOGRAPHY, and mixed schemas.

Related specifications

This PR follows the GeoParquet 1.1.0 specification for file-level geo metadata.

GEOGRAPHY values are stored as Parquet BINARY with GeoParquet metadata using encoding=WKB and edges=spherical. This keeps the implementation compatible with Flink's current Parquet dependency (parquet-mr 1.15.2) instead of relying on the native Parquet GEOGRAPHY logical type.

Dependencies

This PR is stacked on top of:

• [FLINK-39904][table] Add GEOGRAPHY logical type support #28740
• [FLINK-39904] [table] Add SQL constructor and accessor functions for GEOGRAPHY #28788
• [FLINK-39904][python] Add PyFlink GEOGRAPHY support #28833

Only the Parquet-specific changes should be reviewed here. Earlier GEOGRAPHY logical type, SQL function, and PyFlink changes are included only because this is a stacked branch.

Related changes

The SQL ST functions are tracked separately in apache/flink#28788.

This PR focuses on Parquet format interoperability for GEOGRAPHY. The ST functions PR provides the SQL constructor/accessor functions used by user-facing SQL workflows.

Verifying this change

This change is covered by unit tests in flink-parquet.

I also validated a generated GeoParquet file manually with a local Flink pipeline:

• Wrote GEOGRAPHY values through the Flink filesystem connector with format='parquet'.

• Read the same Parquet data back through Flink SQL.

• Verified the generated file contains GeoParquet 1.1.0 metadata and was written with Flink's current Parquet version:

  {"version":"1.1.0","primary_column":"location","columns":{"location":{"encoding":"WKB","geometry_types":[],"edges":"spherical"}}}
  parquet-mr version 1.15.2
  

I additionally loaded the generated Parquet file into BigQuery and verified that BigQuery recognized the column as GEOGRAPHY.

BigQuery test table:

healthy-skill-501608-t7.akvelon.legacy_geography_flink_20260727

BigQuery schema:

case_id: integer
subtype: string
location: geography

Spatial query validation:

SELECT case_id, subtype, ST_ASTEXT(location) AS wkt
FROM `healthy-skill-501608-t7.akvelon.legacy_geography_flink_20260727`
ORDER BY case_id

Result:

1  POINT    POINT(-122.4194 37.7749)
2  POLYGON  POLYGON((-122.5 37.7, -122.4 37.7, -122.4 37.8, -122.5 37.8, -122.5 37.7))
3  NULL     NULL

Does this pull request potentially affect one of the following parts

• Dependencies: no
• The public API: yes, GEOGRAPHY can now be represented in Parquet format
• The serializers: no
• The runtime per-record code paths: yes, Parquet GEOGRAPHY read/write paths
• Anything that affects deployment or recovery: no
• The S3 file system connector: no

Documentation

This PR does not add end-user documentation. GEOGRAPHY SQL functions and broader user-facing SQL behavior are handled in the related ST functions PR.

[flinkbot]

CI report:

• 69e9770 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[davidchaava]

@flinkbot run azure