[FLINK-xxxxx][table] Add GEOGRAPHY logical type support

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];

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

@@ -0,0 +1,82 @@
+/*
+ * 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.table.data;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.data.binary.BinaryGeographyData;
+import org.apache.flink.table.types.logical.GeographyType;
+
+/** An internal data structure representing data of {@link GeographyType}. */
+@PublicEvolving
+public interface GeographyData {
+
+    /** ISO WKB subtype ID for Point geometries. */
+    int POINT = 1;
+
+    /** ISO WKB subtype ID for LineString geometries. */
+    int LINE_STRING = 2;
+
+    /** ISO WKB subtype ID for Polygon geometries. */
+    int POLYGON = 3;
+
+    /** ISO WKB subtype ID for MultiPoint geometries. */
+    int MULTI_POINT = 4;
+
+    /** ISO WKB subtype ID for MultiLineString geometries. */
+    int MULTI_LINE_STRING = 5;
+
+    /** ISO WKB subtype ID for MultiPolygon geometries. */
+    int MULTI_POLYGON = 6;
+
+    /** ISO WKB subtype ID for GeometryCollection geometries. */
+    int GEOMETRY_COLLECTION = 7;
+
+    /**
+     * Converts this {@link GeographyData} object to an ISO WKB byte array.
+     *
+     * <p>Note: The returned byte array may be reused.
+     */
+    byte[] toBytes();
+
+    /** Returns the ISO WKB subtype ID. */
+    int subtypeId();
+
+    /** Returns the size in bytes of the ISO WKB payload. */
+    int sizeInBytes();
+
+    // ------------------------------------------------------------------------------------------
+    // Construction Utilities
+    // ------------------------------------------------------------------------------------------
+
+    /**
+     * Creates an instance of {@link GeographyData} from the given ISO WKB byte array. Returns
+     * {@code null} if the input is {@code null}.
+     */
+    static GeographyData fromBytes(byte[] bytes) {
+        return BinaryGeographyData.fromBytes(bytes);
+    }
+
+    /**
+     * Creates an instance of {@link GeographyData} from the given ISO WKB byte range. Returns
+     * {@code null} if the input is {@code null}.
+     */
+    static GeographyData fromBytes(byte[] bytes, int offset, int numBytes) {
+        return BinaryGeographyData.fromBytes(bytes, offset, numBytes);
+    }
+}

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

@@ -111,6 +111,8 @@
  * +--------------------------------+-----------------------------------------+
  * | BITMAP                         | {@link Bitmap}                          |
  * +--------------------------------+-----------------------------------------+
+ * | GEOGRAPHY                      | {@link GeographyData}                   |
+ * +--------------------------------+-----------------------------------------+
  * </pre>
  *
  * <p>Nullability is always handled by the container data structure.
@@ -214,6 +216,12 @@ default Bitmap getBitmap(int pos) {
                 "This RowData implementation does not support Bitmap type.");
     }
 
+    /** Returns the geography value at the given position. */
+    default GeographyData getGeography(int pos) {
+        throw new UnsupportedOperationException(
+                "This RowData implementation does not support Geography type.");
+    }
+
     // ------------------------------------------------------------------------------------------
     // Access Utilities
     // ------------------------------------------------------------------------------------------
@@ -299,6 +307,9 @@ static FieldGetter createFieldGetter(LogicalType fieldType, int fieldPos) {
             case BITMAP:
                 fieldGetter = row -> row.getBitmap(fieldPos);
                 break;
+            case GEOGRAPHY:
+                fieldGetter = row -> row.getGeography(fieldPos);
+                break;
             case NULL:
             case SYMBOL:
             case UNRESOLVED: