[FLINK-39904] [table] Add SQL constructor and accessor functions for GEOGRAPHY

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-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java

@@ -253,8 +253,11 @@ void testCheckpointDrain() throws Exception {
                             });
             ((AbstractAsyncRunnableStreamOperator<String>) testHarness.getOperator())
                     .postProcessElement();
-            assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(1);
-            unblockAsyncRequest.complete(null);
+            try {
+                assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(1);
+            } finally {
+                unblockAsyncRequest.complete(null);
+            }
             testHarness.drainAsyncRequests();
             assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(0);
         }

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

@@ -4074,4 +4074,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;
@@ -117,4 +122,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;
@@ -1075,6 +1076,15 @@ public static DataType BITMAP() {
         return new AtomicDataType(new BitmapType());
     }
 
+    /**
+     * 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
     // ------------------------------------------------------------------------------------------
@@ -225,6 +231,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];