[FLINK-xxxxx] sanitizing tests and coverage

docs/content/docs/sql/functions/built-in-functions.md

@@ -69,6 +69,44 @@ The scalar functions take zero, one or more values as the input and return a sin
 
 {{< sql_functions "collection" >}}
 
+### Geography Functions
+
+The v1 geography surface is intentionally small and focuses on schema declaration plus portable WKT
+and WKB boundaries.
+
+| Function | Description |
+|:---------|:------------|
+| `ST_GEOGFROMTEXT(wkt)` | Parses 2D WKT text into a `GEOGRAPHY` value. Returns `NULL` for `NULL` input. |
+| `ST_GEOGFROMWKB(wkb)` | Parses ISO/OGC WKB bytes into a `GEOGRAPHY` value. Returns `NULL` for `NULL` input. |
+| `ST_ASTEXT(geography)` | Serializes a `GEOGRAPHY` value to WKT text. Returns `NULL` for `NULL` input. |
+| `ST_ASWKB(geography)` | Returns the raw WKB bytes stored in a `GEOGRAPHY` value. Returns `NULL` for `NULL` input. |
+
+Example:
+
+```sql
+SELECT
+  ST_ASTEXT(ST_GEOGFROMTEXT('POINT (1 2)')),
+  ST_ASWKB(ST_GEOGFROMTEXT('POINT (1 2)')),
+  ST_ASTEXT(
+    ST_GEOGFROMWKB(
+      ST_ASWKB(ST_GEOGFROMTEXT('LINESTRING (0 0, 1 1)'))));
+```
+
+Notes:
+
+- `GEOGRAPHY` coordinates are validated as 2D CRS84 longitude/latitude values. Longitudes must be in `[-180, 180]`; latitudes must be in `[-90, 90]`.
+- `ST_ASWKB` returns the raw WKB bytes stored in the value. `ST_GEOGFROMWKB` followed by `ST_ASWKB` therefore preserves user-provided bytes after validation.
+- Values created from `ST_GEOGFROMTEXT` are stored as 2D WKB without SRID metadata.
+- PyFlink uses `bytes` for WKB boundaries, so no extra Python geospatial dependency is required for round-trip handling.
+
+#### Current Limitations And Follow-up Work
+
+- Spatial predicates and measurements such as `ST_INTERSECTS`, `ST_WITHIN`, `ST_LENGTH`, and `ST_DISTANCE` are not part of v1 yet.
+- `GEOMETRY`, typed geography literals, and a broader geography function set remain follow-up work.
+- Unsupported connector mappings fail explicitly instead of silently degrading `GEOGRAPHY` to `VARBINARY`; broader mappings for Iceberg, Avro, JDBC, Debezium, and additional external systems remain follow-up work.
+- Spatial pruning, spatial joins, and related optimizer/runtime work are intentionally deferred.
+- Published-version savepoint restore tests remain a follow-up because `GEOGRAPHY` does not yet have a released serializer baseline. Current coverage is limited to local SQL/runtime validation plus serializer versioning tests.
+
 ### JSON Functions
 
 JSON functions make use of JSON path expressions as described in ISO/IEC TR 19075-6 of the SQL
@@ -294,3 +332,8 @@ table.select(
 {{< /tabs >}}
 
 {{< top >}}
+
+
+
+
+

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

@@ -121,6 +121,7 @@ For vectorized Python UDFs, the input types and output type are `pandas.Series`.
 | `DOUBLE` | `float` | `numpy.float64` |
 | `VARCHAR` | `str` | `str` |
 | `VARBINARY` | `bytes` | `bytes` |
+| `GEOGRAPHY` | `bytes` | `bytes` |
 | `DECIMAL` | `decimal.Decimal` | `decimal.Decimal` |
 | `DATE` | `datetime.date` | `datetime.date` |
 | `TIME` | `datetime.time` | `datetime.time` |
@@ -215,6 +216,7 @@ The default planner supports the following set of SQL types:
 | `MULTISET`       |                                                    |
 | `MAP`            |                                                    |
 | `ROW`            |                                                    |
+| `GEOGRAPHY`      | 2D CRS84 longitude/latitude geospatial type.      |
 | `RAW`            |                                                    |
 | Structured types | Only exposed in user-defined functions yet.        |
 | `VARIANT`        |                                                    |
@@ -1591,6 +1593,94 @@ DataTypes.BITMAP()
 {{< /tab >}}
 {{< /tabs >}}
 
+#### `GEOGRAPHY`
+
+Data type of 2D geospatial values in CRS84 longitude/latitude coordinates.
+
+`GEOGRAPHY` is a user-facing type for SQL and Table API schemas. In v1, values are created and
+accessed through the built-in geography functions. `ST_GEOGFROMTEXT` parses WKT text,
+`ST_GEOGFROMWKB` parses WKB bytes, `ST_ASTEXT` serializes to WKT, and `ST_ASWKB` exposes WKB
+bytes.
+
+**Declaration**
+
+{{< tabs "geography-data-type" >}}
+{{< tab "SQL" >}}
+```text
+GEOGRAPHY
+```
+{{< /tab >}}
+{{< tab "Java/Scala" >}}
+```java
+DataTypes.GEOGRAPHY()
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+DataTypes.GEOGRAPHY()
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< tabs "geography-notes" >}}
+{{< tab "SQL/Java/Scala/Python" >}}
+Coordinates are validated as 2D CRS84 longitude/latitude values. Longitudes must be in `[-180, 180]`
+and latitudes in `[-90, 90]`. Z and M coordinates are rejected.
+
+`ST_ASWKB` returns the raw ISO/OGC WKB bytes stored in the value. When a value originates from
+`ST_GEOGFROMWKB`, the bytes round-trip unchanged after validation. When a value originates from
+`ST_GEOGFROMTEXT`, Flink stores a 2D WKB representation without SRID metadata.
+
+`NULL` inputs to `ST_GEOGFROMTEXT`, `ST_GEOGFROMWKB`, `ST_ASTEXT`, and `ST_ASWKB` return `NULL`.
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+
+Schema schema =
+        Schema.newBuilder()
+                .column("id", DataTypes.BIGINT())
+                .column("location", DataTypes.GEOGRAPHY())
+                .build();
+
+// Use SQL functions to construct or access GEOGRAPHY values.
+tableEnv.executeSql(
+        "INSERT INTO sink "
+                + "SELECT id, ST_GEOGFROMTEXT(wkt) "
+                + "FROM source");
+```
+{{< /tab >}}
+{{< tab "Python" >}}
+```python
+from pyflink.table import Schema
+from pyflink.table.types import DataTypes
+
+schema = (
+    Schema.new_builder()
+        .column("id", DataTypes.BIGINT())
+        .column("location", DataTypes.GEOGRAPHY())
+        .build()
+)
+
+# PyFlink uses bytes at WKB boundaries.
+result = table_env.execute_sql(
+    "SELECT ST_ASWKB(ST_GEOGFROMTEXT('POINT (1 2)'))")
+wkb = next(result.collect())[0]
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+Migration from `VARBINARY` plus custom WKB UDFs to `GEOGRAPHY` is usually a boundary refactor:
+
+```sql
+INSERT INTO new_places
+SELECT id, ST_GEOGFROMWKB(old_location_wkb)
+FROM old_places;
+```
+
+Use `ST_ASWKB(location)` when a sink or downstream system still expects WKB bytes. Unsupported
+connector mappings fail explicitly instead of silently degrading `GEOGRAPHY` to `VARBINARY`.
 #### `RAW`
 
 Data type of an arbitrary serialized type. This type is a black box within the table ecosystem
@@ -1897,3 +1987,8 @@ Not supported.
 {{< /tabs >}}
 
 {{< top >}}
+
+
+
+
+

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-avro/src/test/java/org/apache/flink/formats/avro/AvroFormatFactoryTest.java

@@ -129,6 +129,34 @@ void testNewSeDeNewSchema() {
         testSeDeSchema(NEW_ROW_TYPE, NEW_SCHEMA, false);
     }
 
+    @Test
+    void testGeographyTypeIsRejected() {
+        ResolvedSchema geographySchema =
+                ResolvedSchema.of(Column.physical("g", DataTypes.GEOGRAPHY()));
+
+        DynamicTableSource source =
+                FactoryMocks.createTableSource(geographySchema, getAllOptions(true));
+        assertThat(source).isInstanceOf(TestDynamicTableFactory.DynamicTableSourceMock.class);
+        TestDynamicTableFactory.DynamicTableSourceMock sourceMock =
+                (TestDynamicTableFactory.DynamicTableSourceMock) source;
+        assertThatThrownBy(
+                        () ->
+                                sourceMock.valueFormat.createRuntimeDecoder(
+                                        ScanRuntimeProviderContext.INSTANCE,
+                                        geographySchema.toPhysicalRowDataType()))
+                .hasMessageContaining("Unsupported to derive Schema for type: GEOGRAPHY");
+
+        DynamicTableSink sink = FactoryMocks.createTableSink(geographySchema, getAllOptions(true));
+        assertThat(sink).isInstanceOf(TestDynamicTableFactory.DynamicTableSinkMock.class);
+        TestDynamicTableFactory.DynamicTableSinkMock sinkMock =
+                (TestDynamicTableFactory.DynamicTableSinkMock) sink;
+        assertThatThrownBy(
+                        () ->
+                                sinkMock.valueFormat.createRuntimeEncoder(
+                                        null, geographySchema.toPhysicalRowDataType()))
+                .hasMessageContaining("Unsupported to derive Schema for type: GEOGRAPHY");
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     void testSeDeSchema(boolean legacyTimestampMapping) {

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