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

docs/content/docs/sql/functions/built-in-functions.md

@@ -69,6 +69,44 @@ The scalar functions take zero, one or more values as the input and return a sin
 
 {{< sql_functions "collection" >}}
 
+### Geography Functions
+
+The v1 geography surface is intentionally small and focuses on schema declaration plus portable WKT
+and WKB boundaries.
+
+| Function | Description |
+|:---------|:------------|
+| `ST_GEOGFROMTEXT(wkt)` | Parses 2D WKT text into a `GEOGRAPHY` value. Returns `NULL` for `NULL` input. |
+| `ST_GEOGFROMWKB(wkb)` | Parses ISO/OGC WKB bytes into a `GEOGRAPHY` value. Returns `NULL` for `NULL` input. |
+| `ST_ASTEXT(geography)` | Serializes a `GEOGRAPHY` value to WKT text. Returns `NULL` for `NULL` input. |
+| `ST_ASWKB(geography)` | Returns the raw WKB bytes stored in a `GEOGRAPHY` value. Returns `NULL` for `NULL` input. |
+
+Example:
+
+```sql
+SELECT
+  ST_ASTEXT(ST_GEOGFROMTEXT('POINT (1 2)')),
+  ST_ASWKB(ST_GEOGFROMTEXT('POINT (1 2)')),
+  ST_ASTEXT(
+    ST_GEOGFROMWKB(
+      ST_ASWKB(ST_GEOGFROMTEXT('LINESTRING (0 0, 1 1)'))));
+```
+
+Notes:
+
+- `GEOGRAPHY` coordinates are validated as 2D CRS84 longitude/latitude values. Longitudes must be in `[-180, 180]`; latitudes must be in `[-90, 90]`.
+- `ST_ASWKB` returns the raw WKB bytes stored in the value. `ST_GEOGFROMWKB` followed by `ST_ASWKB` therefore preserves user-provided bytes after validation.
+- Values created from `ST_GEOGFROMTEXT` are stored as 2D WKB without SRID metadata.
+- PyFlink uses `bytes` for WKB boundaries, so no extra Python geospatial dependency is required for round-trip handling.
+
+#### Current Limitations And Follow-up Work
+
+- Spatial predicates and measurements such as `ST_INTERSECTS`, `ST_WITHIN`, `ST_LENGTH`, and `ST_DISTANCE` are not part of v1 yet.
+- `GEOMETRY`, typed geography literals, and a broader geography function set remain follow-up work.
+- Unsupported connector mappings fail explicitly instead of silently degrading `GEOGRAPHY` to `VARBINARY`; broader mappings for Iceberg, Avro, JDBC, Debezium, and additional external systems remain follow-up work.
+- Spatial pruning, spatial joins, and related optimizer/runtime work are intentionally deferred.
+- Published-version savepoint restore tests remain a follow-up because `GEOGRAPHY` does not yet have a released serializer baseline. Current coverage is limited to local SQL/runtime validation plus serializer versioning tests.
+
 ### JSON Functions
 
 JSON functions make use of JSON path expressions as described in ISO/IEC TR 19075-6 of the SQL
@@ -312,3 +350,8 @@ table.select(
 {{< /tabs >}}
 
 {{< top >}}
+
+
+
+
+

docs/content/docs/sql/reference/data-types.md

@@ -121,6 +121,7 @@ For vectorized Python UDFs, the input types and output type are `pandas.Series`.
 | `DOUBLE` | `float` | `numpy.float64` |
 | `VARCHAR` | `str` | `str` |
 | `VARBINARY` | `bytes` | `bytes` |
+| `GEOGRAPHY` | `bytes` | `bytes` |
 | `DECIMAL` | `decimal.Decimal` | `decimal.Decimal` |
 | `DATE` | `datetime.date` | `datetime.date` |
 | `TIME` | `datetime.time` | `datetime.time` |
@@ -215,10 +216,12 @@ The default planner supports the following set of SQL types:
 | `MULTISET`       |                                                    |
 | `MAP`            |                                                    |
 | `ROW`            |                                                    |
+| `GEOGRAPHY`      | 2D CRS84 longitude/latitude geospatial type.      |
 | `RAW`            |                                                    |
 | Structured types | Only exposed in user-defined functions yet.        |
 | `VARIANT`        |                                                    |
 | `BITMAP`         |                                                    |
+| `GEOGRAPHY`      | Geography values in OGC:CRS84.                    |
 
 ### Character Strings
 
@@ -1664,6 +1667,47 @@ DataTypes.BITMAP()
 {{< /tab >}}
 {{< /tabs >}}
 
+#### `GEOGRAPHY`
+
+Data type of geography data.
+
+`GEOGRAPHY` represents geospatial values in the OGC:CRS84 coordinate reference system.
+The type itself does not define SQL constructors, accessors, or spatial predicate functions.
+Those functions are expected to be added separately.
+
+Flink represents geography payloads as ISO WKB bytes. ISO WKB does not encode CRS or SRID
+metadata, so CRS validation, CRS transformation, and EWKB/SRID handling belong to
+constructors, functions, or connector-specific schema mapping.
+
+The geography type is an extension to the SQL standard.
+
+**Declaration**
+
+{{< tabs "9fef5895-3fa0-4b9b-9f92-9c94ba96ef1a" >}}
+{{< tab "SQL" >}}
+```text
+GEOGRAPHY
+```
+
+{{< /tab >}}
+{{< tab "Java/Scala" >}}
+```java
+DataTypes.GEOGRAPHY()
+```
+
+**Bridging to JVM Types**
+
+| Java Type                                      | Input | Output | Remarks   |
+|:-----------------------------------------------|:-----:|:------:|:----------|
+| `org.apache.flink.table.data.GeographyData`    |   X   |   X    | *Default* |
+
+{{< /tab >}}
+{{< /tabs >}}
+
+`GEOGRAPHY` values cannot be constructed with `CAST` from character or binary string
+types. Likewise, `GEOGRAPHY` values cannot be cast to character or binary string types.
+Use explicit geography functions for those conversions once such functions are available.
+
 #### `RAW`
 
 Data type of an arbitrary serialized type. This type is a black box within the table ecosystem
@@ -1970,3 +2014,8 @@ Not supported.
 {{< /tabs >}}
 
 {{< top >}}
+
+
+
+
+

docs/static/generated/rest_v1_sql_gateway.yml

@@ -404,6 +404,7 @@ components:
       - VARIANT
       - BITMAP
       - UUID
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

docs/static/generated/rest_v2_sql_gateway.yml

@@ -478,6 +478,7 @@ components:
       - VARIANT
       - BITMAP
       - UUID
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

docs/static/generated/rest_v3_sql_gateway.yml

@@ -507,6 +507,7 @@ components:
       - VARIANT
       - BITMAP
       - UUID
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

docs/static/generated/rest_v4_sql_gateway.yml

@@ -517,6 +517,7 @@ components:
       - VARIANT
       - BITMAP
       - UUID
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroFormatFactoryTest.java

@@ -129,6 +129,34 @@ void testNewSeDeNewSchema() {
         testSeDeSchema(NEW_ROW_TYPE, NEW_SCHEMA, false);
     }
 
+    @Test
+    void testGeographyTypeIsRejected() {
+        ResolvedSchema geographySchema =
+                ResolvedSchema.of(Column.physical("g", DataTypes.GEOGRAPHY()));
+
+        DynamicTableSource source =
+                FactoryMocks.createTableSource(geographySchema, getAllOptions(true));
+        assertThat(source).isInstanceOf(TestDynamicTableFactory.DynamicTableSourceMock.class);
+        TestDynamicTableFactory.DynamicTableSourceMock sourceMock =
+                (TestDynamicTableFactory.DynamicTableSourceMock) source;
+        assertThatThrownBy(
+                        () ->
+                                sourceMock.valueFormat.createRuntimeDecoder(
+                                        ScanRuntimeProviderContext.INSTANCE,
+                                        geographySchema.toPhysicalRowDataType()))
+                .hasMessageContaining("Unsupported to derive Schema for type: GEOGRAPHY");
+
+        DynamicTableSink sink = FactoryMocks.createTableSink(geographySchema, getAllOptions(true));
+        assertThat(sink).isInstanceOf(TestDynamicTableFactory.DynamicTableSinkMock.class);
+        TestDynamicTableFactory.DynamicTableSinkMock sinkMock =
+                (TestDynamicTableFactory.DynamicTableSinkMock) sink;
+        assertThatThrownBy(
+                        () ->
+                                sinkMock.valueFormat.createRuntimeEncoder(
+                                        null, geographySchema.toPhysicalRowDataType()))
+                .hasMessageContaining("Unsupported to derive Schema for type: GEOGRAPHY");
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     void testSeDeSchema(boolean legacyTimestampMapping) {

flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/row/ParquetRowDataBuilder.java

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.serialization.BulkWriter;
 import org.apache.flink.formats.parquet.ParquetBuilder;
 import org.apache.flink.formats.parquet.ParquetWriterFactory;
+import org.apache.flink.formats.parquet.utils.GeoParquetMetadataUtil;
 import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.logical.RowType;
@@ -35,7 +36,7 @@
 import org.apache.parquet.schema.MessageType;
 
 import java.io.IOException;
-import java.util.HashMap;
+import java.util.Map;
 
 import static org.apache.flink.formats.parquet.utils.ParquetSchemaConverter.convertToParquetMessageType;
 import static org.apache.parquet.hadoop.ParquetOutputFormat.MAX_PADDING_BYTES;
@@ -81,7 +82,9 @@ private ParquetWriteSupport(Configuration conf) {
 
         @Override
         public WriteContext init(Configuration configuration) {
-            return new WriteContext(schema, new HashMap<>());
+            final Map<String, String> keyValueMetaData =
+                    GeoParquetMetadataUtil.createGeoParquetKeyValueMetaData(rowType);
+            return new WriteContext(schema, keyValueMetaData);
         }
 
         @Override

flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/row/ParquetRowDataWriter.java

@@ -119,6 +119,8 @@ private FieldWriter createWriter(LogicalType t, Type type) {
                 case BINARY:
                 case VARBINARY:
                     return new BinaryWriter();
+                case GEOGRAPHY:
+                    return new GeographyWriter();
                 case DECIMAL:
                     DecimalType decimalType = (DecimalType) t;
                     return createDecimalWriter(decimalType.getPrecision(), decimalType.getScale());
@@ -311,6 +313,23 @@ private void writeBinary(byte[] value) {
         }
     }
 
+    private class GeographyWriter implements FieldWriter {
+
+        @Override
+        public void write(RowData row, int ordinal) {
+            writeGeography(row.getGeography(ordinal).toBytes());
+        }
+
+        @Override
+        public void write(ArrayData arrayData, int ordinal) {
+            writeGeography(arrayData.getGeography(ordinal).toBytes());
+        }
+
+        private void writeGeography(byte[] value) {
+            recordConsumer.addBinary(Binary.fromReusedByteArray(value));
+        }
+    }
+
     private class IntWriter implements FieldWriter {
 
         @Override

flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/GeoParquetMetadataUtil.java

@@ -0,0 +1,132 @@
+/*
+ * 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.formats.parquet.utils;
+
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.apache.parquet.schema.MessageType;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+
+/** Utilities for emitting and validating GeoParquet metadata. */
+public final class GeoParquetMetadataUtil {
+
+    public static final String GEO_METADATA_KEY = "geo";
+    public static final String GEOPARQUET_VERSION = "1.1.0";
+
+    private static final String GEOGRAPHY_ENCODING = "WKB";
+    private static final String GEOGRAPHY_EDGES = "spherical";
+    private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper();
+
+    private GeoParquetMetadataUtil() {}
+
+    public static Map<String, String> createGeoParquetKeyValueMetaData(RowType rowType) {
+        final ObjectNode columnsNode = OBJECT_MAPPER.createObjectNode();
+        String primaryColumn = null;
+
+        for (int i = 0; i < rowType.getFieldCount(); i++) {
+            final LogicalType fieldType = rowType.getTypeAt(i);
+            if (fieldType.getTypeRoot() != LogicalTypeRoot.GEOGRAPHY) {
+                continue;
+            }
+
+            final String fieldName = rowType.getFieldNames().get(i);
+            if (primaryColumn == null) {
+                primaryColumn = fieldName;
+            }
+
+            final ObjectNode columnNode = columnsNode.putObject(fieldName);
+            columnNode.put("encoding", GEOGRAPHY_ENCODING);
+            columnNode.putArray("geometry_types");
+            columnNode.put("edges", GEOGRAPHY_EDGES);
+        }
+
+        if (primaryColumn == null) {
+            return Collections.emptyMap();
+        }
+
+        final ObjectNode rootNode = OBJECT_MAPPER.createObjectNode();
+        rootNode.put("version", GEOPARQUET_VERSION);
+        rootNode.put("primary_column", primaryColumn);
+        rootNode.set("columns", columnsNode);
+
+        try {
+            return Collections.singletonMap(
+                    GEO_METADATA_KEY, OBJECT_MAPPER.writeValueAsString(rootNode));
+        } catch (JsonProcessingException e) {
+            throw new IllegalStateException("Failed to serialize GeoParquet metadata.", e);
+        }
+    }
+
+    public static void validateGeoParquetMetadata(
+            Map<String, String> keyValueMetaData,
+            MessageType requestedSchema,
+            LogicalType[] selectedTypes)
+            throws IOException {
+        if (keyValueMetaData == null || !keyValueMetaData.containsKey(GEO_METADATA_KEY)) {
+            return;
+        }
+
+        final JsonNode geoNode = OBJECT_MAPPER.readTree(keyValueMetaData.get(GEO_METADATA_KEY));
+        final JsonNode columnsNode = geoNode.get("columns");
+        if (columnsNode == null || !columnsNode.isObject()) {
+            throw new IOException("Invalid GeoParquet metadata: missing 'columns' object.");
+        }
+
+        for (int i = 0; i < selectedTypes.length; i++) {
+            if (selectedTypes[i].getTypeRoot() != LogicalTypeRoot.GEOGRAPHY) {
+                continue;
+            }
+
+            final String columnName = requestedSchema.getFields().get(i).getName();
+            final JsonNode columnNode = columnsNode.get(columnName);
+            if (columnNode == null || !columnNode.isObject()) {
+                throw new IOException(
+                        String.format(
+                                "Invalid GeoParquet metadata: missing GEOGRAPHY column '%s'.",
+                                columnName));
+            }
+
+            final JsonNode encodingNode = columnNode.get("encoding");
+            if (encodingNode == null || !encodingNode.isTextual()) {
+                throw new IOException(
+                        String.format(
+                                "Invalid GeoParquet metadata: missing encoding for GEOGRAPHY column '%s'.",
+                                columnName));
+            }
+
+            if (!GEOGRAPHY_ENCODING.equals(encodingNode.asText())) {
+                throw new IOException(
+                        String.format(
+                                "Unsupported GeoParquet encoding '%s' for GEOGRAPHY column '%s'.",
+                                encodingNode.asText(), columnName));
+            }
+        }
+    }
+}