[FLINK-39904][python] Add PyFlink GEOGRAPHY 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-python/pyflink/fn_execution/coder_impl_fast.pxd

@@ -126,6 +126,9 @@ cdef class DoubleCoderImpl(FieldCoderImpl):
 cdef class BinaryCoderImpl(FieldCoderImpl):
     pass
 
+cdef class GeographyCoderImpl(FieldCoderImpl):
+    pass
+
 cdef class CharCoderImpl(FieldCoderImpl):
     pass
 

flink-python/pyflink/fn_execution/coder_impl_fast.pyx

@@ -585,6 +585,25 @@ cdef class BinaryCoderImpl(FieldCoderImpl):
     cpdef decode_from_stream(self, InputStream in_stream, size_t size):
         return in_stream.read_bytes()
 
+
+cdef class GeographyCoderImpl(FieldCoderImpl):
+    """
+    A coder compatible with GeographyTypeSerializer's versioned WKB format.
+    """
+
+    cpdef encode_to_stream(self, value, OutputStream out_stream):
+        out_stream.write_byte(1)
+        out_stream.write_bytes(value, len(value))
+
+    cpdef decode_from_stream(self, InputStream in_stream, size_t size):
+        cdef long format_version = in_stream.read_byte()
+        if format_version != 1:
+            raise ValueError(
+                "Unsupported GEOGRAPHY serializer format version %d. Expected 1."
+                % format_version)
+        return in_stream.read_bytes()
+
+
 cdef class CharCoderImpl(FieldCoderImpl):
     """
     A coder for a str value.

flink-python/pyflink/fn_execution/coder_impl_slow.py

@@ -441,6 +441,26 @@ def decode_from_stream(self, in_stream: InputStream, length=0):
         return in_stream.read_bytes()
 
 
+class GeographyCoderImpl(FieldCoderImpl):
+    """
+    A coder compatible with GeographyTypeSerializer's versioned WKB format.
+    """
+
+    FORMAT_VERSION = 1
+
+    def encode_to_stream(self, value, out_stream: OutputStream):
+        out_stream.write_byte(self.FORMAT_VERSION)
+        out_stream.write_bytes(value, len(value))
+
+    def decode_from_stream(self, in_stream: InputStream, length=0):
+        format_version = in_stream.read_byte()
+        if format_version != self.FORMAT_VERSION:
+            raise ValueError(
+                "Unsupported GEOGRAPHY serializer format version %d. Expected %d."
+                % (format_version, self.FORMAT_VERSION))
+        return in_stream.read_bytes()
+
+
 class CharCoderImpl(FieldCoderImpl):
     """
     A coder for a str value.

flink-python/pyflink/fn_execution/coders.py

@@ -37,12 +37,13 @@
 from pyflink.table.types import TinyIntType, SmallIntType, IntType, BigIntType, BooleanType, \
     FloatType, DoubleType, VarCharType, VarBinaryType, DecimalType, DateType, TimeType, \
     LocalZonedTimestampType, RowType, RowField, to_arrow_type, TimestampType, ArrayType, MapType, \
-    BinaryType, NullType, CharType
+    BinaryType, GeographyType, NullType, CharType
 
 __all__ = ['FlattenRowCoder', 'RowCoder', 'BigIntCoder', 'TinyIntCoder', 'BooleanCoder',
-           'SmallIntCoder', 'IntCoder', 'FloatCoder', 'DoubleCoder', 'BinaryCoder', 'CharCoder',
-           'DateCoder', 'TimeCoder', 'TimestampCoder', 'LocalZonedTimestampCoder', 'InstantCoder',
-           'GenericArrayCoder', 'PrimitiveArrayCoder', 'MapCoder', 'DecimalCoder',
+           'SmallIntCoder', 'IntCoder', 'FloatCoder', 'DoubleCoder', 'BinaryCoder',
+           'GeographyCoder', 'CharCoder', 'DateCoder', 'TimeCoder', 'TimestampCoder',
+           'LocalZonedTimestampCoder', 'InstantCoder', 'GenericArrayCoder',
+           'PrimitiveArrayCoder', 'MapCoder', 'DecimalCoder',
            'BigDecimalCoder', 'TupleCoder', 'TimeWindowCoder', 'CountWindowCoder',
            'PickleCoder', 'CloudPickleCoder', 'DataViewFilterCoder']
 
@@ -134,6 +135,8 @@ def _to_data_type(cls, field_type):
             return BinaryType(field_type.binary_info.length, field_type.nullable)
         elif field_type.type_name == flink_fn_execution_pb2.Schema.VARBINARY:
             return VarBinaryType(field_type.var_binary_info.length, field_type.nullable)
+        elif field_type.type_name == flink_fn_execution_pb2.Schema.GEOGRAPHY:
+            return GeographyType(field_type.nullable)
         elif field_type.type_name == flink_fn_execution_pb2.Schema.DECIMAL:
             return DecimalType(field_type.decimal_info.precision,
                                field_type.decimal_info.scale,
@@ -476,6 +479,15 @@ def get_impl(self):
         return coder_impl.BinaryCoderImpl()
 
 
+class GeographyCoder(FieldCoder):
+    """
+    Coder for GEOGRAPHY values represented as WKB bytes.
+    """
+
+    def get_impl(self):
+        return coder_impl.GeographyCoderImpl()
+
+
 class CharCoder(FieldCoder):
     """
     Coder for Character String.
@@ -673,6 +685,7 @@ def from_proto(field_type):
         type_name.DOUBLE: DoubleCoder(),
         type_name.BINARY: BinaryCoder(),
         type_name.VARBINARY: BinaryCoder(),
+        type_name.GEOGRAPHY: GeographyCoder(),
         type_name.CHAR: CharCoder(),
         type_name.VARCHAR: CharCoder(),
         type_name.DATE: DateCoder(),

flink-python/pyflink/fn_execution/flink_fn_execution_pb2.pyi

@@ -256,6 +256,7 @@ class Schema(_message.Message):
         LOCAL_ZONED_TIMESTAMP: _ClassVar[Schema.TypeName]
         ZONED_TIMESTAMP: _ClassVar[Schema.TypeName]
         NULL: _ClassVar[Schema.TypeName]
+        GEOGRAPHY: _ClassVar[Schema.TypeName]
     ROW: Schema.TypeName
     TINYINT: Schema.TypeName
     SMALLINT: Schema.TypeName
@@ -278,6 +279,7 @@ class Schema(_message.Message):
     LOCAL_ZONED_TIMESTAMP: Schema.TypeName
     ZONED_TIMESTAMP: Schema.TypeName
     NULL: Schema.TypeName
+    GEOGRAPHY: Schema.TypeName
     class MapInfo(_message.Message):
         __slots__ = ("key_type", "value_type")
         KEY_TYPE_FIELD_NUMBER: _ClassVar[int]

flink-python/pyflink/fn_execution/tests/test_coders.py

@@ -25,8 +25,9 @@
     SmallIntCoder, IntCoder, FloatCoder, DoubleCoder, BinaryCoder, CharCoder, DateCoder, \
     TimeCoder, TimestampCoder, GenericArrayCoder, MapCoder, DecimalCoder, FlattenRowCoder, \
     RowCoder, LocalZonedTimestampCoder, BigDecimalCoder, TupleCoder, PrimitiveArrayCoder, \
-    TimeWindowCoder, CountWindowCoder, InstantCoder
+    TimeWindowCoder, CountWindowCoder, InstantCoder, GeographyCoder, from_proto
 from pyflink.datastream.window import TimeWindow, CountWindow
+from pyflink.fn_execution import flink_fn_execution_pb2
 from pyflink.testing.test_case_utils import PyFlinkTestCase
 
 
@@ -78,6 +79,15 @@ def test_binary_coder(self):
         coder = BinaryCoder()
         self.check_coder(coder, b'pyflink')
 
+    def test_geography_coder_from_proto(self):
+        field_type = flink_fn_execution_pb2.Schema.FieldType(
+            type_name=flink_fn_execution_pb2.Schema.GEOGRAPHY)
+        coder = from_proto(field_type)
+        self.assertIsInstance(coder, GeographyCoder)
+        self.assertEqual(
+            b'\x01\x00\x00\x00\tgeography', coder.get_impl().encode(b'geography'))
+        self.check_coder(coder, b'geography')
+
     def test_char_coder(self):
         coder = CharCoder()
         self.check_coder(coder, 'flink', '🐿')

flink-python/pyflink/proto/flink-fn-execution.proto

@@ -233,6 +233,7 @@ message Schema {
     LOCAL_ZONED_TIMESTAMP = 19;
     ZONED_TIMESTAMP = 20;
     NULL = 21;
+    GEOGRAPHY = 22;
   }
 
   message MapInfo {

flink-python/pyflink/table/tests/test_pandas_udf.py

@@ -381,6 +381,24 @@ def local_zoned_timestamp_func(local_zoned_timestamp_param):
         actual = source_sink_utils.results()
         self.assert_equals(actual, ["+I[1970-01-02T00:00:00.123Z]"])
 
+    def test_geography_type(self):
+        point_wkb = bytes([
+            1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xF0, 0x3F, 0, 0, 0, 0, 0, 0, 0, 0x40
+        ])
+
+        geography_identity = udf(
+            lambda geography: geography,
+            result_type=DataTypes.GEOGRAPHY(),
+            func_type="pandas")
+
+        table = self.t_env.from_elements(
+            [(point_wkb,)],
+            DataTypes.ROW([DataTypes.FIELD("g", DataTypes.GEOGRAPHY())]))
+        collected = list(table.select(geography_identity(table.g)).execute().collect())
+
+        self.assertEqual(1, len(collected))
+        self.assertEqual(point_wkb, collected[0][0])
+
 
 class BatchPandasUDFITTests(PandasUDFITTests,
                             PyFlinkBatchTableTestCase):

flink-python/pyflink/table/tests/test_schema.py

@@ -23,33 +23,40 @@
 
 class SchemaTest(PyFlinkTestCase):
     def test_schema_basic(self):
-        old_schema = Schema.new_builder() \
-            .from_row_data_type(DataTypes.ROW(
-                [DataTypes.FIELD("a", DataTypes.TINYINT()),
-                 DataTypes.FIELD("b", DataTypes.SMALLINT()),
-                 DataTypes.FIELD("c", DataTypes.INT())])) \
-            .from_fields(["d", "e"], [DataTypes.STRING(), DataTypes.BOOLEAN()]) \
-            .build()
-        self.schema = Schema.new_builder() \
-            .from_schema(old_schema) \
-            .primary_key_named("primary_constraint", "id") \
-            .column("id", DataTypes.INT().not_null()) \
-            .column("counter", DataTypes.INT().not_null()) \
-            .column("payload", "ROW<name STRING, age INT, flag BOOLEAN>") \
-            .column_by_metadata("topic", DataTypes.STRING(), None, True) \
-            .column_by_expression("ts", call_sql("orig_ts - INTERVAL '60' MINUTE")) \
-            .column_by_metadata("orig_ts", DataTypes.TIMESTAMP(3), "timestamp") \
-            .watermark("ts", "ts - INTERVAL '5' SECOND") \
-            .column_by_expression("proctime", "PROCTIME()") \
-            .build()
+        old_schema = (
+            Schema.new_builder()
+            .from_row_data_type(
+                DataTypes.ROW(
+                    [DataTypes.FIELD("a", DataTypes.TINYINT()),
+                     DataTypes.FIELD("b", DataTypes.SMALLINT()),
+                     DataTypes.FIELD("c", DataTypes.INT()),
+                     DataTypes.FIELD("f", DataTypes.GEOGRAPHY())]))
+            .from_fields(["d", "e"], [DataTypes.STRING(), DataTypes.BOOLEAN()])
+            .build())
+        self.schema = (
+            Schema.new_builder()
+            .from_schema(old_schema)
+            .primary_key_named("primary_constraint", "id")
+            .column("id", DataTypes.INT().not_null())
+            .column("counter", DataTypes.INT().not_null())
+            .column("location", DataTypes.GEOGRAPHY())
+            .column("payload", "ROW<name STRING, age INT, flag BOOLEAN>")
+            .column_by_metadata("topic", DataTypes.STRING(), None, True)
+            .column_by_expression("ts", call_sql("orig_ts - INTERVAL '60' MINUTE"))
+            .column_by_metadata("orig_ts", DataTypes.TIMESTAMP(3), "timestamp")
+            .watermark("ts", "ts - INTERVAL '5' SECOND")
+            .column_by_expression("proctime", "PROCTIME()")
+            .build())
         self.assertEqual("""(
   `a` TINYINT,
   `b` SMALLINT,
   `c` INT,
+  `f` GEOGRAPHY,
   `d` STRING,
   `e` BOOLEAN,
   `id` INT NOT NULL,
   `counter` INT NOT NULL,
+  `location` GEOGRAPHY,
   `payload` [ROW<name STRING, age INT, flag BOOLEAN>],
   `topic` STRING METADATA VIRTUAL,
   `ts` AS [orig_ts - INTERVAL '60' MINUTE],