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

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

@@ -219,6 +219,7 @@ The default planner supports the following set of SQL types:
 | Structured types | Only exposed in user-defined functions yet.        |
 | `VARIANT`        |                                                    |
 | `BITMAP`         |                                                    |
+| `GEOGRAPHY`      | Geography values in OGC:CRS84.                    |
 
 ### Character Strings
 
@@ -1597,6 +1598,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

docs/static/generated/rest_v1_sql_gateway.yml

@@ -403,6 +403,7 @@ components:
       - DESCRIPTOR
       - VARIANT
       - BITMAP
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

docs/static/generated/rest_v2_sql_gateway.yml

@@ -477,6 +477,7 @@ components:
       - DESCRIPTOR
       - VARIANT
       - BITMAP
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

docs/static/generated/rest_v3_sql_gateway.yml

@@ -506,6 +506,7 @@ components:
       - DESCRIPTOR
       - VARIANT
       - BITMAP
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

docs/static/generated/rest_v4_sql_gateway.yml

@@ -516,6 +516,7 @@ components:
       - DESCRIPTOR
       - VARIANT
       - BITMAP
+      - GEOGRAPHY
     OpenSessionRequestBody:
       type: object
       properties:

flink-formats/flink-parquet/pom.xml

@@ -80,6 +80,14 @@ under the License.
 			<optional>true</optional>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+			<optional>true</optional>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-hadoop-fs</artifactId>

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));
+            }
+        }
+    }
+}