[core] Support geospatial data types

docs/docs/concepts/data-types.md

@@ -76,6 +76,23 @@ All data types supported by Paimon are as follows:
           <code>BYTES is a synonym for VARBINARY(2147483647).</code>
       </td>
     </tr>
+    <tr>
+      <td><code>GEOMETRY</code><br>
+          <code>GEOMETRY(crs)</code>
+      </td>
+      <td><code>Data type of a planar geometry encoded as OGC Well-Known Binary (WKB).</code><br><br>
+          <code>The optional crs identifies the coordinate reference system. The default is OGC:CRS84.</code>
+      </td>
+    </tr>
+    <tr>
+      <td><code>GEOGRAPHY</code><br>
+          <code>GEOGRAPHY(crs)</code><br>
+          <code>GEOGRAPHY(crs, algorithm)</code>
+      </td>
+      <td><code>Data type of a geography whose edges are interpolated on the surface of the coordinate reference system, encoded as OGC Well-Known Binary (WKB).</code><br><br>
+          <code>The default crs is OGC:CRS84 and the default edge interpolation algorithm is spherical. Supported algorithms are spherical, vincenty, thomas, andoyer, and karney.</code>
+      </td>
+    </tr>
     <tr>
       <td><code>DECIMAL</code><br>
           <code>DECIMAL(p)</code><br>
@@ -189,3 +206,11 @@ All data types supported by Paimon are as follows:
     </tr>
     </tbody>
 </table>
+
+:::note Geospatial type availability
+
+`GEOMETRY` and `GEOGRAPHY` columns require Parquet for data, per-level, and changelog files. They cannot be used as primary, partition, bucket, sequence, or clustering keys.
+
+The Paimon Java API supports geospatial columns. Spark 4.1 requires `spark.sql.geospatial.enabled=true`, supports CRSs recognized by Spark, and supports only the `spherical` geography edge algorithm. Flink SQL, Spark 3.x, and Spark 4.0 reject geospatial columns instead of exposing them as binary and losing the CRS or edge algorithm.
+
+:::

docs/docs/concepts/spec/fileformat.md

@@ -63,6 +63,16 @@ The following table lists the type mapping from Paimon type to Parquet type.
       <td>BINARY</td>
       <td></td>
     </tr>
+    <tr>
+      <td>GEOMETRY(crs)</td>
+      <td>BINARY</td>
+      <td>GEOMETRY(crs)</td>
+    </tr>
+    <tr>
+      <td>GEOGRAPHY(crs, algorithm)</td>
+      <td>BINARY</td>
+      <td>GEOGRAPHY(crs, algorithm)</td>
+    </tr>
     <tr>
       <td>DECIMAL(P, S)</td>
       <td>P <= 9: INT32, P <= 18: INT64, P > 18: FIXED_LEN_BYTE_ARRAY</td>
@@ -142,8 +152,10 @@ The following table lists the type mapping from Paimon type to Parquet type.
 </table>
 
 Limitations:
+
 1. [Parquet does not support nullable map keys](https://github.com/apache/parquet-format/blob/master/LogicalTypes#maps).
 2. Parquet TIMESTAMP type with precision 9 will use INT96, but this int96 is a time zone converted value and requires additional adjustments.
+3. Tables containing `GEOMETRY` or `GEOGRAPHY` columns must use Parquet for `file.format`, every entry in `file.format.per.level`, and `changelog-file.format` when configured.
 
 ## AVRO
 

docs/docs/flink/quick-start.mdx

@@ -31,6 +31,12 @@ under the License.
 
 This documentation is a guide for using Paimon in Flink.
 
+:::warning
+
+Flink SQL does not currently support Paimon `GEOMETRY` or `GEOGRAPHY` columns. Reading, writing, or copying a table whose schema contains either type fails explicitly instead of exposing the column as `VARBINARY` and losing its CRS or edge algorithm. Use the Paimon Java API or Spark 4.1 for geospatial columns.
+
+:::
+
 ## Jars
 
 Paimon currently supports Flink 2.2, 2.1, 2.0, 1.20, 1.19, 1.18, 1.17, 1.16. We recommend the latest Flink version for a better experience.

docs/docs/iceberg/index.md

@@ -98,6 +98,8 @@ Paimon Iceberg compatibility currently supports the following data types.
 | `TIMESTAMP_LTZ` (precision 3-6) | `timestamptz`     |
 | `TIMESTAMP` (precision 7-9)  | `timestamp_ns`    |
 | `TIMESTAMP_LTZ` (precision 7-9) | `timestamptz_ns`  |
+| `GEOMETRY(crs)` | `geometry(crs)` |
+| `GEOGRAPHY(crs, algorithm)` | `geography(crs, algorithm)` |
 | `ARRAY`        | `list`            |
 | `MAP`          | `map`             |
 | `ROW`          | `struct`          |
@@ -108,4 +110,12 @@ Paimon Iceberg compatibility currently supports the following data types.
 - `TIMESTAMP` and `TIMESTAMP_LTZ` types with precision from 3 to 6 are mapped to standard Iceberg timestamp types
 - `TIMESTAMP` and `TIMESTAMP_LTZ` types with precision from 7 to 9 use nanosecond precision and require Iceberg v3 format
 
+**Note on Geospatial Types:**
+- `GEOMETRY` and `GEOGRAPHY` values use OGC Well-Known Binary (WKB). The default CRS is `OGC:CRS84`, and the default geography edge algorithm is `spherical`.
+- Geospatial columns require Parquet for data, per-level, and changelog files. When Iceberg metadata is enabled, set `metadata.iceberg.format-version` to `3`.
+- Spark SQL supports geospatial columns in Spark 4.1 when `spark.sql.geospatial.enabled=true`, for CRSs recognized by Spark, with the `spherical` geography edge algorithm. Spark 3.x, Spark 4.0, and Flink SQL reject these columns instead of exposing them as binary and losing the CRS or edge algorithm.
+- When Iceberg metadata is enabled, a `GEOGRAPHY` CRS cannot contain a comma, including in nested columns, because Iceberg's geospatial type grammar uses commas to separate parameters.
+- Iceberg REST catalog publication does not yet support geospatial columns. Use `table-location`, `hadoop-catalog`, or `hive-catalog` metadata storage instead.
+- Geospatial columns cannot be primary, partition, bucket, sequence, or clustering keys. Paimon records null counts but does not publish byte-wise lower or upper bounds for WKB values.
+
 :::

docs/docs/iceberg/rest-catalog.mdx

@@ -35,6 +35,12 @@ You need to provide information about Rest Catalog by setting options prefixed w
 `'metadata.iceberg.rest.uri' = 'https://localhost/'`. Paimon will try to use these options to initialize an iceberg rest catalog, 
 and use this rest catalog to commit metadata.
 
+:::warning
+
+Tables containing `GEOMETRY` or `GEOGRAPHY` columns cannot use `'metadata.iceberg.storage' = 'rest-catalog'` because the bundled Iceberg REST client cannot parse Iceberg v3 geospatial types. Use `table-location`, `hadoop-catalog`, or `hive-catalog` metadata storage instead.
+
+:::
+
 **Dependency:**
 
 This feature needs dependency:
@@ -128,4 +134,3 @@ There are some cases when committing to iceberg rest catalog:
 1. table not exists in iceberg rest-catalog. It'll create the table in rest catalog first, and commit metadata.
 2. table exists in iceberg rest-catalog and is compatible with the base metadata stored in the separate directory. It'll directly get the table and commit metadata. 
 3. table exists, and isn't compatible with the base metadata stored in the separate directory. It'll **drop the table and recreate the table**, then commit metadata.
-

docs/docs/primary-key-table/sequence-rowkind.mdx

@@ -55,7 +55,7 @@ CREATE TABLE my_table (
 </Tabs>
 
 The record with the largest `sequence.field` value will be the last to merge, if the values are the same, the input
-order will be used to determine which one is the last one. `sequence.field` supports fields of all data types.
+order will be used to determine which one is the last one. `sequence.field` does not support `GEOMETRY`, `GEOGRAPHY`, or managed `BLOB` fields.
 
 You can define multiple fields for `sequence.field`, for example `'update_time,flag'`, multiple fields will be compared in order.
 

docs/docs/program-api/java-api.mdx

@@ -459,6 +459,24 @@ public class StreamWriteTable {
 | map          | org.apache.paimon.data.InternalMap   |
 | InternalRow  | org.apache.paimon.data.InternalRow   |
 
+### Geospatial Types
+
+Use the public type factories to declare geospatial columns:
+
+```java
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.EdgeAlgorithm;
+
+DataType defaultGeometry = DataTypes.GEOMETRY();
+DataType projectedGeometry = DataTypes.GEOMETRY("EPSG:3857");
+DataType defaultGeography = DataTypes.GEOGRAPHY();
+DataType karneyGeography =
+        DataTypes.GEOGRAPHY("OGC:CRS84", EdgeAlgorithm.KARNEY);
+```
+
+`GEOMETRY` and `GEOGRAPHY` values are represented as OGC Well-Known Binary (WKB) byte arrays in Paimon's internal row API. The default CRS is `OGC:CRS84`, and the default geography edge interpolation algorithm is `EdgeAlgorithm.SPHERICAL`.
+
 ## Predicate Types
 
 | SQL Predicate | Paimon Predicate                                            | 

docs/docs/spark/quick-start.mdx

@@ -374,6 +374,16 @@ All Spark's data types are available in package `org.apache.spark.sql.types`.
       <td><code>VarBinaryType</code>, <code>BinaryType</code></td>
       <td>true</td>
     </tr>
+    <tr>
+      <td><code>GeometryType (Spark 4.1)</code></td>
+      <td><code>GeometryType</code></td>
+      <td>true</td>
+    </tr>
+    <tr>
+      <td><code>GeographyType (Spark 4.1)</code></td>
+      <td><code>GeographyType</code></td>
+      <td>true</td>
+    </tr>
     <tr>
       <td><code>VariantType(Spark4.0+)</code></td>
       <td><code>VariantType</code></td>
@@ -384,10 +394,16 @@ All Spark's data types are available in package `org.apache.spark.sql.types`.
 
 :::warning
 
+Native `GeometryType` and `GeographyType` conversion is supported only in Spark 4.1 and only for CRSs recognized by Spark. Enable it explicitly in production with `--conf spark.sql.geospatial.enabled=true`; Spark enables it automatically only in its test environment. Spark 4.1 supports only the `spherical` geography edge algorithm, so Paimon geography types using `vincenty`, `thomas`, `andoyer`, or `karney` cannot be converted. Spark 3.x and Spark 4.0 reject Paimon geospatial columns instead of exposing them as `BinaryType`, which would lose the CRS or edge algorithm. Paimon does not support Spark geospatial types with mixed SRIDs.
+
+:::
+
+:::warning
+
 Due to the previous design, in Spark3.3 and below, Paimon will map both Paimon's TimestampType and LocalZonedTimestamp to Spark's TimestampType, and only correctly handle with TimestampType.
 
 Therefore, when using Spark3.3 and below, reads Paimon table with LocalZonedTimestamp type written by other engines, such as Flink, the query result of LocalZonedTimestamp type will have time zone offset, which needs to be adjusted manually.
 
 When using Spark3.4 and above, all timestamp types can be parsed correctly.
 
-:::
+:::

paimon-api/src/main/java/org/apache/paimon/types/DataTypeCasts.java

@@ -211,6 +211,11 @@ public static boolean supportsCompatibleCast(DataType sourceType, DataType targe
             return true;
         }
 
+        if (sourceType.isAnyOf(DataTypeRoot.GEOMETRY, DataTypeRoot.GEOGRAPHY)
+                || targetType.isAnyOf(DataTypeRoot.GEOMETRY, DataTypeRoot.GEOGRAPHY)) {
+            return false;
+        }
+
         return compatibleCastingRules
                 .get(targetType.getTypeRoot())
                 .contains(sourceType.getTypeRoot());
@@ -230,6 +235,11 @@ private static boolean supportsCasting(
             return true;
         }
 
+        if (sourceType.isAnyOf(DataTypeRoot.GEOMETRY, DataTypeRoot.GEOGRAPHY)
+                || targetType.isAnyOf(DataTypeRoot.GEOMETRY, DataTypeRoot.GEOGRAPHY)) {
+            return false;
+        }
+
         final DataTypeRoot sourceRoot = sourceType.getTypeRoot();
         final DataTypeRoot targetRoot = targetType.getTypeRoot();
 

paimon-api/src/main/java/org/apache/paimon/types/DataTypeDefaultVisitor.java

@@ -119,6 +119,16 @@ public R visit(BlobType blobType) {
         return defaultMethod(blobType);
     }
 
+    @Override
+    public R visit(GeometryType geometryType) {
+        return defaultMethod(geometryType);
+    }
+
+    @Override
+    public R visit(GeographyType geographyType) {
+        return defaultMethod(geographyType);
+    }
+
     @Override
     public R visit(ArrayType arrayType) {
         return defaultMethod(arrayType);

paimon-api/src/main/java/org/apache/paimon/types/DataTypeJsonParser.java

@@ -331,6 +331,8 @@ private enum Keyword {
         LEGACY,
         VARIANT,
         BLOB,
+        GEOMETRY,
+        GEOGRAPHY,
         NOT
     }
 
@@ -549,6 +551,10 @@ private DataType parseTypeByKeyword() {
                     return new VariantType();
                 case BLOB:
                     return new BlobType();
+                case GEOMETRY:
+                    return parseGeometryType();
+                case GEOGRAPHY:
+                    return parseGeographyType();
                 case VECTOR:
                     return parseVectorType();
                 default:
@@ -683,5 +689,40 @@ private DataType parseVectorType() {
             nextToken(TokenType.END_SUBTYPE);
             return DataTypes.VECTOR(length, elementType);
         }
+
+        private DataType parseGeometryType() {
+            if (!hasNextToken(TokenType.BEGIN_PARAMETER)) {
+                return DataTypes.GEOMETRY();
+            }
+            nextToken(TokenType.BEGIN_PARAMETER);
+            String crs = parseGeospatialParameter();
+            nextToken(TokenType.END_PARAMETER);
+            return DataTypes.GEOMETRY(crs);
+        }
+
+        private DataType parseGeographyType() {
+            if (!hasNextToken(TokenType.BEGIN_PARAMETER)) {
+                return DataTypes.GEOGRAPHY();
+            }
+            nextToken(TokenType.BEGIN_PARAMETER);
+            String crs = parseGeospatialParameter();
+            EdgeAlgorithm algorithm = GeographyType.DEFAULT_ALGORITHM;
+            if (hasNextToken(TokenType.LIST_SEPARATOR)) {
+                nextToken(TokenType.LIST_SEPARATOR);
+                algorithm = EdgeAlgorithm.fromName(parseGeospatialParameter());
+            }
+            nextToken(TokenType.END_PARAMETER);
+            return DataTypes.GEOGRAPHY(crs, algorithm);
+        }
+
+        private String parseGeospatialParameter() {
+            nextToken();
+            if (token().type != TokenType.IDENTIFIER
+                    && token().type != TokenType.LITERAL_STRING
+                    && token().type != TokenType.KEYWORD) {
+                throw parsingError("Geospatial type parameter expected.");
+            }
+            return token().value;
+        }
     }
 }

paimon-api/src/main/java/org/apache/paimon/types/DataTypeRoot.java

@@ -104,6 +104,10 @@ public enum DataTypeRoot {
 
     BLOB(DataTypeFamily.PREDEFINED),
 
+    GEOMETRY(DataTypeFamily.PREDEFINED),
+
+    GEOGRAPHY(DataTypeFamily.PREDEFINED),
+
     ARRAY(DataTypeFamily.CONSTRUCTED, DataTypeFamily.COLLECTION),
 
     VECTOR(DataTypeFamily.CONSTRUCTED, DataTypeFamily.COLLECTION),

paimon-api/src/main/java/org/apache/paimon/types/DataTypeVisitor.java

@@ -66,6 +66,14 @@ public interface DataTypeVisitor<R> {
 
     R visit(BlobType blobType);
 
+    default R visit(GeometryType geometryType) {
+        throw new UnsupportedOperationException("Unsupported type: " + geometryType);
+    }
+
+    default R visit(GeographyType geographyType) {
+        throw new UnsupportedOperationException("Unsupported type: " + geographyType);
+    }
+
     R visit(ArrayType arrayType);
 
     R visit(VectorType vectorType);

paimon-api/src/main/java/org/apache/paimon/types/DataTypes.java

@@ -163,6 +163,26 @@ public static BlobType BLOB() {
         return new BlobType();
     }
 
+    public static GeometryType GEOMETRY() {
+        return new GeometryType();
+    }
+
+    public static GeometryType GEOMETRY(String crs) {
+        return new GeometryType(crs);
+    }
+
+    public static GeographyType GEOGRAPHY() {
+        return new GeographyType();
+    }
+
+    public static GeographyType GEOGRAPHY(String crs) {
+        return new GeographyType(crs);
+    }
+
+    public static GeographyType GEOGRAPHY(String crs, EdgeAlgorithm algorithm) {
+        return new GeographyType(crs, algorithm);
+    }
+
     public static OptionalInt getPrecision(DataType dataType) {
         return dataType.accept(PRECISION_EXTRACTOR);
     }

paimon-api/src/main/java/org/apache/paimon/types/EdgeAlgorithm.java

@@ -0,0 +1,51 @@
+/*
+ * 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.paimon.types;
+
+import org.apache.paimon.annotation.Public;
+
+import java.util.Locale;
+
+/** Algorithm used to interpolate geography edges. */
+@Public
+public enum EdgeAlgorithm {
+    SPHERICAL,
+    VINCENTY,
+    THOMAS,
+    ANDOYER,
+    KARNEY;
+
+    public static EdgeAlgorithm fromName(String algorithmName) {
+        if (algorithmName == null) {
+            throw new IllegalArgumentException("Invalid edge interpolation algorithm: null");
+        }
+
+        try {
+            return valueOf(algorithmName.toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException e) {
+            throw new IllegalArgumentException(
+                    "Invalid edge interpolation algorithm: " + algorithmName, e);
+        }
+    }
+
+    @Override
+    public String toString() {
+        return name().toLowerCase(Locale.ROOT);
+    }
+}