[FLINK-39904][table] Add GEOGRAPHY logical type 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
 
@@ -1735,6 +1736,53 @@ DataTypes.BITMAP()
 {{< /tab >}}
 {{< /tabs >}}
 
+#### `GEOGRAPHY`
+
+Data type of geography data.
+
+`GEOGRAPHY` represents geospatial values in the [OGC:CRS84](https://www.opengis.net/def/crs/OGC/1.3/CRS84)
+coordinate reference system. Here, CRS means *Coordinate Reference System* and SRID means
+*Spatial Reference Identifier*. OGC:CRS84 identifies the coordinate reference system and axis
+order used by this type; CRS and SRID metadata are not stored in the WKB payload.
+
+Flink represents geography payloads as [ISO WKB](https://www.ogc.org/standards/sfa), where WKB
+means *Well-Known Binary*. The v1 contract supports the standard 2D WKB geometry type codes.
+[EWKB](https://postgis.net/docs/using_postgis_dbmanagement.html#EWKB_EWKT), or *Extended
+Well-Known Binary*, is not part of this contract. CRS validation, CRS transformation, and
+EWKB/SRID handling belong to constructors, functions, or connector-specific schema mapping.
+
+The type itself does not define SQL constructors, accessors, or spatial predicate functions.
+Those functions are expected to be added separately.
+
+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

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

docs/static/generated/rest_v2_sql_gateway.yml

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

docs/static/generated/rest_v3_sql_gateway.yml

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

docs/static/generated/rest_v4_sql_gateway.yml

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

flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonDeserializer.java

@@ -32,6 +32,7 @@
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.DoubleType;
 import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
 import org.apache.flink.table.types.logical.IntType;
 import org.apache.flink.table.types.logical.LocalZonedTimestampType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -168,6 +169,8 @@ private LogicalType deserializeInternal(JsonNode logicalTypeNode) {
                 return new BitmapType(isNullable);
             case UUID:
                 return new UuidType(isNullable);
+            case GEOGRAPHY:
+                return new GeographyType(isNullable);
             default:
                 throw new UnsupportedOperationException(
                         String.format(

flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerializer.java

@@ -131,6 +131,7 @@ private void serializeInternal(LogicalType logicalType, JsonGenerator jsonGenera
             case VARIANT:
             case BITMAP:
             case UUID:
+            case GEOGRAPHY:
                 break;
             case CHAR:
                 jsonGenerator.writeNumberField(

flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/serde/LogicalTypeJsonSerDeTest.java

@@ -34,6 +34,7 @@
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.DoubleType;
 import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
 import org.apache.flink.table.types.logical.IntType;
 import org.apache.flink.table.types.logical.LocalZonedTimestampType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -199,6 +200,7 @@ private static List<LogicalType> generateTestData() {
                         new MultisetType(VarBinaryType.ofEmptyLiteral()),
                         new BitmapType(),
                         new UuidType(),
+                        new GeographyType(),
                         RowType.of(new BigIntType(), new IntType(false), new VarCharType(200)),
                         RowType.of(
                                 new LogicalType[] {
@@ -207,6 +209,7 @@ private static List<LogicalType> generateTestData() {
                                 new String[] {"f1", "f2", "f3"}),
                         RowType.of(
                                 new TimestampType(false, 3), new LocalZonedTimestampType(false, 3)),
+                        RowType.of(new GeographyType(false), new ArrayType(new GeographyType())),
                         RowType.of(
                                 CharType.ofEmptyLiteral(),
                                 VarCharType.ofEmptyLiteral(),

flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd

@@ -173,6 +173,7 @@
     "org.apache.flink.sql.parser.type.ExtendedSqlCollectionTypeNameSpec"
     "org.apache.flink.sql.parser.type.ExtendedSqlRowTypeNameSpec"
     "org.apache.flink.sql.parser.type.SqlBitmapTypeNameSpec"
+    "org.apache.flink.sql.parser.type.SqlGeographyTypeNameSpec"
     "org.apache.flink.sql.parser.type.SqlRawTypeNameSpec"
     "org.apache.flink.sql.parser.type.SqlStructuredTypeNameSpec"
     "org.apache.flink.sql.parser.type.SqlTimestampLtzTypeNameSpec"
@@ -225,6 +226,7 @@
     "FROM_TIMESTAMP"
     "FUNCTIONS"
     "FRESHNESS"
+    "GEOGRAPHY"
     "HASH"
     "IF"
     "JSON_EXECUTION_PLAN"
@@ -719,6 +721,7 @@
     "ExtendedSqlRowTypeName()"
     "SqlStructuredTypeName()"
     "SqlBitmapTypeName()"
+    "SqlGeographyTypeName()"
   ]
 
   # List of methods for parsing builtin function calls.

flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl

@@ -2666,6 +2666,17 @@ SqlTypeNameSpec SqlBitmapTypeName() :
     }
 }
 
+/** Parses GEOGRAPHY type. */
+SqlTypeNameSpec SqlGeographyTypeName() :
+{
+}
+{
+    <GEOGRAPHY>
+    {
+        return new SqlGeographyTypeNameSpec(getPos());
+    }
+}
+
 /**
 * Parse a "name1 type1 [ NULL | NOT NULL] [ comment ]
 * [, name2 type2 [ NULL | NOT NULL] [ comment ] ]* ..." list.

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/type/SqlGeographyTypeNameSpec.java

@@ -0,0 +1,59 @@
+/*
+ * 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.sql.parser.type;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.calcite.ExtendedRelTypeFactory;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlTypeNameSpec;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.validate.SqlValidator;
+import org.apache.calcite.util.Litmus;
+
+/** Represents the GEOGRAPHY data type. */
+@Internal
+public final class SqlGeographyTypeNameSpec extends SqlTypeNameSpec {
+
+    private static final String GEOGRAPHY_TYPE_NAME = "GEOGRAPHY";
+
+    public SqlGeographyTypeNameSpec(SqlParserPos pos) {
+        super(new SqlIdentifier(GEOGRAPHY_TYPE_NAME, pos), pos);
+    }
+
+    @Override
+    public RelDataType deriveType(SqlValidator validator) {
+        return ((ExtendedRelTypeFactory) validator.getTypeFactory()).createGeographyType();
+    }
+
+    @Override
+    public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+        writer.keyword(GEOGRAPHY_TYPE_NAME);
+    }
+
+    @Override
+    public boolean equalsDeep(SqlTypeNameSpec spec, Litmus litmus) {
+        if (!(spec instanceof SqlGeographyTypeNameSpec)) {
+            return litmus.fail("{} != {}", this, spec);
+        }
+        return litmus.succeed();
+    }
+}

flink-table/flink-sql-parser/src/main/java/org/apache/flink/table/calcite/ExtendedRelTypeFactory.java

@@ -44,4 +44,7 @@ RelDataType createStructuredType(
 
     /** Creates a BITMAP type. */
     RelDataType createBitmapType();
+
+    /** Creates a GEOGRAPHY type. */
+    RelDataType createGeographyType();
 }

flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java

@@ -4214,4 +4214,23 @@ void testBitmapType() {
         sql("CREATE TABLE t (\n" + "^bitmap^ INT" + "\n)")
                 .fails("(?s).*Encountered \"bitmap\" at line 2, column 1.\n.*");
     }
+
+    @Test
+    void testGeographyType() {
+        sql("CREATE TABLE t (\n" + "g geography" + "\n)")
+                .ok("CREATE TABLE `T` (\n" + "  `G` GEOGRAPHY\n" + ")");
+
+        sql("CREATE TABLE t (\n" + "g geography NOT NULL" + "\n)")
+                .ok("CREATE TABLE `T` (\n" + "  `G` GEOGRAPHY NOT NULL\n" + ")");
+
+        // GEOGRAPHY takes no parameters
+        sql("CREATE TABLE t (\n" + "g geography^(^1)" + "\n)")
+                .fails("(?s).*Encountered \"\\(\" at line 2, column 12.\n.*");
+        sql("CREATE TABLE t (\n" + "g geography^(^4326)" + "\n)")
+                .fails("(?s).*Encountered \"\\(\" at line 2, column 12.\n.*");
+
+        // GEOGRAPHY is a reserved keyword and cannot be used as an identifier without escaping
+        sql("CREATE TABLE t (\n" + "^geography^ INT" + "\n)")
+                .fails("(?s).*Encountered \"geography\" at line 2, column 1.\n.*");
+    }
 }

flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/TestRelDataTypeFactory.java

@@ -53,6 +53,11 @@ public RelDataType createBitmapType() {
         return canonize(new DummyBitmapType());
     }
 
+    @Override
+    public RelDataType createGeographyType() {
+        return canonize(new DummyGeographyType());
+    }
+
     private static class DummyRawType extends RelDataTypeImpl {
 
         private final String className;
@@ -118,4 +123,16 @@ protected void generateTypeString(StringBuilder sb, boolean withDetail) {
             sb.append("BITMAP");
         }
     }
+
+    private static class DummyGeographyType extends RelDataTypeImpl {
+
+        DummyGeographyType() {
+            computeDigest();
+        }
+
+        @Override
+        protected void generateTypeString(StringBuilder sb, boolean withDetail) {
+            sb.append("GEOGRAPHY");
+        }
+    }
 }

flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/DataTypes.java

@@ -44,6 +44,7 @@
 import org.apache.flink.table.types.logical.DescriptorType;
 import org.apache.flink.table.types.logical.DoubleType;
 import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.GeographyType;
 import org.apache.flink.table.types.logical.IntType;
 import org.apache.flink.table.types.logical.LocalZonedTimestampType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -1087,6 +1088,15 @@ public static DataType UUID() {
         return new AtomicDataType(new UuidType());
     }
 
+    /**
+     * Data type of geography data.
+     *
+     * @see GeographyType
+     */
+    public static DataType GEOGRAPHY() {
+        return new AtomicDataType(new GeographyType());
+    }
+
     // --------------------------------------------------------------------------------------------
     // Helper functions
     // --------------------------------------------------------------------------------------------

flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/ArrayData.java

@@ -123,6 +123,12 @@ default Bitmap getBitmap(int pos) {
                 "This ArrayData implementation does not support Bitmap type.");
     }
 
+    /** Returns the geography value at the given position. */
+    default GeographyData getGeography(int pos) {
+        throw new UnsupportedOperationException(
+                "This ArrayData implementation does not support Geography type.");
+    }
+
     // ------------------------------------------------------------------------------------------
     // Conversion Utilities
     // ------------------------------------------------------------------------------------------
@@ -226,6 +232,9 @@ static ElementGetter createElementGetter(LogicalType elementType) {
             case BITMAP:
                 elementGetter = ArrayData::getBitmap;
                 break;
+            case GEOGRAPHY:
+                elementGetter = ArrayData::getGeography;
+                break;
             case NULL:
             case SYMBOL:
             case UNRESOLVED:

flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericArrayData.java

@@ -265,6 +265,11 @@ public Bitmap getBitmap(int pos) {
         return (Bitmap) getObject(pos);
     }
 
+    @Override
+    public GeographyData getGeography(int pos) {
+        return (GeographyData) getObject(pos);
+    }
+
     private Object getObject(int pos) {
         return ((Object[]) array)[pos];
     }

flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/GenericRowData.java

@@ -192,6 +192,11 @@ public byte[] getBinary(int pos) {
         return (byte[]) this.fields[pos];
     }
 
+    @Override
+    public GeographyData getGeography(int pos) {
+        return (GeographyData) this.fields[pos];
+    }
+
     @Override
     public ArrayData getArray(int pos) {
         return (ArrayData) this.fields[pos];