taozhiyi13 opened a new issue, #12923: URL: https://github.com/apache/seatunnel/issues/12923
### Search before asking - [x] I had searched in the [feature](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Feature%22) and found no similar feature requirement. ### Description Related issue: https://github.com/apache/seatunnel/issues/12832 ## Background SeaTunnel cannot currently synchronize MySQL spatial columns end to end through these paths: ```text MySQL JDBC Source ----+ +--> SeaTunnel --> Elasticsearch Sink --> Elasticsearch / OpenSearch MySQL-CDC / Debezium --+ ``` MySQL has eight spatial column types: `GEOMETRY`, `POINT`, `LINESTRING`, `POLYGON`, `MULTIPOINT`, `MULTILINESTRING`, `MULTIPOLYGON`, and `GEOMETRYCOLLECTION`. SeaTunnel has no logical type that preserves their kind, SRID, and value across connector boundaries. Generic MySQL `GEOMETRY` is treated as `BYTES`, concrete types are not handled consistently, and Debezium geometry structures cannot be decoded by the existing `BYTES` converter. The Elasticsearch Sink also cannot derive `geo_point` or `geo_shape` from `BYTES`. Target dynamic mapping treats longitude/latitude objects and GeoJSON as ordinary objects, not geo fields. A local prototype in commits `ce7ffb6f5` and `185298f99` validates the main data path. It is not the normative design and will be adjusted and split after this STIP is accepted. ## Motivation Applications store locations, routes, areas, and geofences in MySQL and need to synchronize both existing rows and binlog changes to Elasticsearch or OpenSearch. This requires one explicit contract across: - MySQL JDBC Source; - MySQL-CDC snapshot and binlog decoding; - SeaTunnel schema and row transport; - Elasticsearch/OpenSearch mapping and serialization. The source must preserve database semantics without depending on the target. Target-specific restrictions belong at the sink boundary. Although the first implementation connects MySQL sources to the Elasticsearch Sink, the runtime representation is source- and sink-neutral. Future connectors can explicitly adopt the same contract without introducing another intermediate geometry representation. ## Goals - Define a minimal, reusable SeaTunnel carrier for spatial columns. - Produce the same value from MySQL JDBC Source and MySQL-CDC. - Preserve geometry kind, unsigned 32-bit SRID, and standard 2D WKB. - Support MySQL-CDC snapshot and binlog `INSERT`, `UPDATE`, and `DELETE`. - Map `POINT` to `geo_point` and other kinds to `geo_shape`. - Create or validate geo mappings before source data is consumed. - Enforce the Elasticsearch/OpenSearch SRID policy without changing coordinates. - Define schema serialization and deterministic behavior for unsupported engines. ## Non-Goals - Coordinate transformation or reprojection. - Spatial functions, predicates, or a general GIS framework. - MySQL JDBC Sink geometry support. - Elasticsearch/OpenSearch Source geometry support. - PostgreSQL/PostGIS or other spatial sources. - Runtime spatial schema evolution. - Geometry execution support in Flink or Spark in the first implementation. - Automatic support in unrelated sinks, formats, or transforms. ## Design The proposal makes four primary decisions: 1. A schema carries `GeometryType(kind)`; each row carries `Geometry(srid, wkb)`. 2. MySQL sources preserve every valid SRID and never transform coordinates. 3. The Elasticsearch Sink maps the geometry and accepts only SRID `4326`. 4. Zeta supports the value; Flink and Spark reject it deterministically during planning. No new user-facing connector option is introduced. The runtime carrier contains no MySQL internal format, Elasticsearch field type, GeoJSON, or target-specific SRID policy. Those concerns remain in the corresponding connector adapters. ### 1. API, schema, and engine contract The proposal adds: ```text SqlType.GEOMETRY GeometryType kind: GeometryKind GeometryKind GEOMETRY, POINT, LINESTRING, POLYGON, MULTIPOINT, MULTILINESTRING, MULTIPOLYGON, GEOMETRYCOLLECTION Geometry srid: long wkb: byte[] ``` `GeometryType.kind` is the subtype declared by the source column. `Geometry.srid` is the row-level unsigned 32-bit SRID, represented as a Java `long` from `0` through `4294967295`. `Geometry.wkb` is standard 2D WKB. Source adapters remove any vendor-specific framing before constructing the value. The value is immutable from callers' perspective. MySQL 8 column-level `SRS_ID` is stored as: ```text column.options["srid"] = Long ``` This option is absent when the column does not declare an SRID or the source cannot report it, including MySQL 5.7. It enables early sink validation but does not replace the row-level SRID. `Geometry`, `GeometryType`, `CatalogTable`, and the column metadata must round-trip without losing kind, SRID, WKB, or existing column attributes. Serialization must not depend on `SqlType` ordinals. Manual `GEOMETRY(...)` schema declarations remain unsupported in the first implementation and fail during configuration parsing. Catalog-discovered geometry schemas use the object contract above. #### Engine behavior - **Zeta:** supports `Geometry` in `SeaTunnelRow`. - **Flink:** rejects a geometry schema while translating the job plan. - **Spark:** rejects a geometry schema while translating it to Spark types. Flink and Spark errors must identify the engine, table, column, and geometry kind. The job must fail before data is consumed, not later in a serializer or sink. No Connector API/SPI method signature changes. Components only receive `GEOMETRY` when an upstream schema contains a spatial column; unsupported components keep their existing unsupported-type behavior. ### 2. MySQL source contract Both source paths produce: ```text Geometry(unsignedSrid, standardWkb) ``` They preserve every MySQL-valid unsigned 32-bit SRID and apply no target-specific restriction. #### MySQL-CDC Debezium emits MySQL `POINT` as Debezium `Point` and other spatial kinds as Debezium `Geometry`. Both structures contain WKB and an `INT32` SRID. SeaTunnel converts: - SQL `NULL` to Java `null`; - Debezium `srid = null` to MySQL SRID `0`; - other SRIDs with unsigned 32-bit semantics; - Debezium WKB to the unchanged standard WKB payload. Snapshot and binlog records use the same converter and produce the same SeaTunnel value for the same source value. #### MySQL JDBC Source MySQL JDBC returns: ```text 4-byte little-endian SRID + standard WKB ``` The JDBC converter restores the unsigned SRID and removes the four-byte prefix. SQL `NULL` remains `null`, and a value without both the prefix and WKB is rejected. JDBC Source is included because it shares `MySqlCatalog` and `MySqlTypeConverter` with MySQL-CDC. Changing the shared spatial type mapping without adapting JDBC Source would break that source. JDBC Sink remains out of scope. #### Coordinate order No axis swapping is performed. MySQL internally stores geographic longitude in X and latitude in Y. Its WKT/WKB functions may display or parse SRS-defined axis order, but Debezium and JDBC expose the internal X/Y representation used here. The sink writes: ```text X -> longitude Y -> latitude ``` Fixtures use explicit longitude-latitude input and verify `ST_Longitude`/`ST_Latitude` against the target coordinates. ### 3. Elasticsearch/OpenSearch sink contract The mapping is: | Source kind | Target type | |---|---| | `POINT` | `geo_point` | | all other kinds | `geo_shape` | A generic `GEOMETRY` column uses `geo_shape` because different rows may contain different kinds. `geo_point` is serialized as `{ "lon": x, "lat": y }`. `geo_shape` is serialized as GeoJSON converted from WKB. #### Index mapping - **New index:** SeaTunnel creates explicit mappings for all geometry columns before writing. - **Existing index:** SeaTunnel reads the mapping and verifies every geometry field. - **Compatible field:** the job proceeds. - **Missing or incompatible field:** sink creation fails with the index, field, expected type, and actual or missing type. The sink does not mutate an incompatible existing field. Users must provide a compatible index or use a new index. Dynamic index names cannot be validated before data is read and are not supported for automatic geometry mapping in the first implementation. #### SRID Elasticsearch and OpenSearch geo fields do not store an SRID and interpret coordinates as WGS84. The sink policy therefore accepts only SRID `4326`. | Source schema/value | Result | |---|---| | column declares `4326` | sink creation succeeds | | column declares another SRID | sink creation fails before source data is read | | column has no declared SRID | each non-null value is checked when written | | row value has `4326` | value is written | | row value has another SRID | writing fails with column and actual SRID | The sink never changes an SRID, swaps axes, or transforms coordinates. #### Invalid geometry The sink accepts standard 2D WKB types 1 through 7 in either byte order. It rejects malformed or unsupported WKB before sending a document, with an error that identifies the source column. Structurally valid but target-invalid shapes are rejected by Elasticsearch/OpenSearch, and the target error is preserved with index and column context. Elasticsearch/OpenSearch cannot represent an empty geometry collection. The sink rejects it instead of silently converting it to SQL/JSON `null`. A row-level conversion or SRID error fails the job. Documents already flushed before the failing row are not rolled back. ### 4. Delivery slices After acceptance, the implementation is submitted as one atomic PR organized into three dependency-ordered, individually buildable commits: 1. **API/SPI:** carrier, schema serialization tests, Zeta support, and Flink/Spark plan-time rejection. 2. **MySQL sources:** catalog metadata, JDBC decoding, and Debezium snapshot/binlog decoding. 3. **Elasticsearch Sink:** mapping, existing-index validation, SRID/WKB enforcement, E2E, and documentation. Each commit is independently reviewable and depends only on the preceding commit. The complete feature is merged atomically after all three slices are ready. ## Compatibility, Deprecation, and Migration Plan No existing API, SPI method, or connector option is deprecated. Jobs without MySQL spatial columns are unchanged. Previous MySQL-CDC versions could not process non-null MySQL spatial values, so there is no working historical CDC spatial contract or checkpoint migration to preserve. MySQL JDBC Source could expose generic `GEOMETRY` as `BYTES`. It now produces `GEOMETRY(GEOMETRY)`. A JDBC job that still requires raw WKB can keep the previous binary result by selecting: ```sql ST_AsWKB(column_name) ``` Components receive the new type only when an upstream schema contains a spatial column. A component that has not explicitly added `GEOMETRY` support rejects that schema through its unsupported-type path. ## Test Plan ### API and engine contract - Validate SRID range, WKB immutability, equality, and row-size accounting. - Round-trip `Geometry`, `GeometryType`, and `CatalogTable` with column-level SRID metadata. - Verify Zeta carries geometry rows. - Verify Flink and Spark reject geometry during plan translation with required context. ### MySQL source contract - Map all eight MySQL types to the expected kind. - Verify declared, missing, and NULL `SRS_ID`. - Verify JDBC little-endian and unsigned SRID decoding. - Verify Debezium `Point`/`Geometry`, SQL `NULL`, WKB forms, and unsigned SRIDs. - Verify snapshot and binlog produce equal values. - Use real MySQL fixtures for all eight spatial types. ### Sink contract - Verify `POINT -> geo_point` and other kinds to `geo_shape`. - Verify all standard 2D WKB kinds and both byte orders. - Reject malformed WKB and empty geometry collections. - Create correct mappings for a new index. - Accept a correctly mapped existing index and reject missing or incompatible mappings. - Verify column-level and row-level SRID failures. - Verify longitude/latitude order and representative geo queries. ### End-to-end matrix - MySQL-CDC snapshot and binlog `INSERT`, `UPDATE`, and `DELETE` with all eight types. - MySQL JDBC Source with all eight types. - Elasticsearch 8.9.0. - OpenSearch 2.19.1. Tested versions are documented separately from versions expected to be protocol-compatible. ## Alternatives Considered ### `BYTES` or `STRING` Rejected because they do not preserve a complete kind/SRID contract across connectors. `STRING` also forces target-oriented WKT or GeoJSON serialization into the source. ### Connector-local geometry class Rejected because MySQL sources and Elasticsearch Sink are separate plugins and require a stable schema and value contract across classloader boundaries. A connector-local class would also force future spatial sources and sinks to introduce another intermediate representation instead of explicitly adopting the shared carrier. ### Coordinate reprojection in the sink Rejected because it mutates source data and requires a coordinate reference system library. The sink validates compatibility and never transforms coordinates. ### Flink and Spark support in the first implementation Rejected to keep the first change focused. Deterministic plan-time rejection provides a safe boundary until a later proposal adds engine translation support. ## References - Feature request: https://github.com/apache/seatunnel/issues/12832 - Previous BYTES-based issue: https://github.com/apache/seatunnel/issues/8548 - SeaTunnel STIP guide: https://seatunnel.apache.org/community/contribution_guide/STIP - MySQL spatial reference systems: https://dev.mysql.com/doc/refman/8.0/en/spatial-reference-systems.html - MySQL geometry format conversion: https://dev.mysql.com/doc/refman/8.0/en/gis-format-conversion-functions.html - OpenSearch `geo_point`: https://docs.opensearch.org/latest/field-types/supported-field-types/geo-point/ - OpenSearch `geo_shape`: https://docs.opensearch.org/latest/field-types/supported-field-types/geo-shape/ ### Usage Scenario _No response_ ### Related issues _No response_ ### Are you willing to submit a PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
