taozhiyi13 commented on issue #12923: URL: https://github.com/apache/seatunnel/issues/12923#issuecomment-6413704084
> 4\. Classification: D / API and Connector-V2 cross-connector spatial data contract. > Thank you for converting [#12832](https://github.com/apache/seatunnel/issues/12832) into a concrete STIP. I rechecked current `dev`, the complete open-PR set, and recent merged work: SeaTunnel has MySQL spatial parser/type references and Elasticsearch/OpenSearch `geo_point`/`geo_shape` constants, but no API `GEOMETRY` SQL type or shared geometry carrier, and there is no competing implementation PR. The proposal correctly treats this as an end-to-end data contract rather than a sink-only mapping. > This remains **STIP in review**; please do not open the implementation PR yet. Before endorsement, update the design with an explicit compatibility inventory for every persisted/transported schema representation (not only `CatalogTable`): the type encoding must be stable by name rather than ordinal, preserve existing column attributes, and have deterministic behavior when an older or unsupported component sees the new type. Also identify the exact Zeta serialization boundary and the Flink/Spark planning entry points, with tests proving rejection happens before source data is consumed. > The source/sink contract should also state the authoritative handling for column-level SRID metadata when it is absent or stale relative to a row value, and make the existing-index/mapping validation failure messages part of the acceptance criteria. After those API, upgrade, and unsupported-engine boundaries are made testable, we can evaluate the proposal for acceptance and split implementation into reviewable, dependency-ordered work. Thanks for the review. The core design remains unchanged. The following additions clarify the requested compatibility and failure contracts. ## 1. Persisted and transported representations This proposal uses SeaTunnel's existing serialization paths and does not introduce a separate geometry wire format. The relevant lifecycle areas are listed below. ### 1.1 Schema establishment | Item | Existing path | Behavior in the first version | | --- | --- | --- | | Manual HOCON schema | `SeaTunnelDataTypeConvertorUtil` parses manually declared field types | Manually declared `GEOMETRY` or `GEOMETRY(...)` fields are rejected during configuration parsing. Spatial columns must be discovered from a MySQL source. | | MySQL catalog discovery | The MySQL catalog reads the source table and builds an in-memory `CatalogTable` | The schema contains `GeometryType` and, when MySQL reports `SRS_ID`, the column-level `srid` option. No serialization occurs at this point. | MySQL catalog discovery is the only supported way to establish a spatial schema in the first version. ### 1.2 Job-plan distribution and engine boundary | Engine | Boundary | Behavior in the first version | | --- | --- | --- | | Zeta | Hazelcast serializes the job action and plugin object graph, including `CatalogTable` | Geometry kind, column-level `srid`, and existing column attributes are preserved. `GeometryKind` is serialized by symbolic name, not ordinal. | | Spark | Plan properties carry catalog tables through the existing Base64-encoded Java serialization path and convert them through `TypeConverterUtils` | The current conversion reaches the unsupported-type path. The implementation will provide a deterministic error before source data is read. | | Flink | `FactoryUtil.createAndPrepareSource` creates the source and discovers its `CatalogTable` in `AbstractSourceExecuteProcessor.initializePlugins` | The implementation will validate the discovered schema immediately afterward, before `execute` registers the source with `fromSource` and before reader creation. | ### 1.3 Runtime data transfer `RecordSerializer` supports the following record types. Only `SeaTunnelRow` can carry a geometry value; the other records are listed to make the transport inventory complete. | Record | Encoding | Geometry behavior | | --- | --- | --- | | `SeaTunnelRow` | Record type tag, `tableId`, `rowKind`, arity, and each field through Hazelcast `writeObject` | `Geometry` is serialized as a field object. Its row-level SRID and WKB are preserved. | | `CheckpointBarrier` | Barrier ID, timestamp, checkpoint type, and task lists | Carries no row or geometry value. | | `FlushSignal` | Job ID, task ID, and timestamp | Carries no row or geometry value. | `SchemaChangeEvent` is not one of the cross-worker record types supported by `RecordSerializer`. Runtime spatial schema evolution is therefore outside the first version. ### 1.4 Checkpoint and restore | Item | Existing path | Geometry behavior | | --- | --- | --- | | Checkpoint envelope | ProtoStuff serializes `CompletedCheckpoint`, including opaque task-state bytes | It does not interpret or introduce a separate encoding for geometry. | | Current CDC checkpoint | `IncrementalSplit.checkpointTables` stores the full `CatalogTable` through the existing `DefaultSerializer` Java serialization path | Geometry kind, column-level `srid`, and existing column attributes are preserved. | | Legacy CDC checkpoint | Only `SeaTunnelRowType` is stored | Column options are absent, so it cannot provide a declared SRID after restore. | | Elasticsearch/OpenSearch sink state | Existing sink checkpoint state does not store the full `CatalogTable` | It does not carry the spatial schema contract. | `checkpointTables` stores schema metadata, not row values. It therefore does not contain the row-level SRID or WKB held by `Geometry`. ### 1.5 Debezium schema history (CDC only) | Item | Existing path | Geometry behavior | | --- | --- | --- | | Debezium schema history | Debezium records MySQL-native table changes in Connect JSON. The history bytes are stored alongside the CDC split checkpoint state. | It does not store SeaTunnel `GeometryType` or the column-level `srid` option and is not authoritative for the SeaTunnel spatial schema. | ## 2. Stable type encoding and column attributes Previous SeaTunnel versions did not have an end-to-end spatial type contract. The compatibility requirement here is therefore for the persisted identity of the new type, rather than migration of a historical spatial schema. The new schema identity is `GeometryType(GeometryKind)`. `GeometryType` is serialized as an object, and Java enum serialization stores `GeometryKind` by symbolic name, such as `POINT`, rather than ordinal. No spatial path persists `SqlType.GEOMETRY` or `GeometryKind` as an ordinal. The byte tags used by `RecordSerializer` identify record kinds and are unrelated to SQL types. A `CatalogTable` round trip must preserve: - the `GeometryKind`; - the column-level `options["srid"]`; - the existing column name, length, scale, nullability, default value, and comment; - `sourceType`, `sinkType`, and other existing column options. Row-level `Geometry.srid` and WKB are value data, not column attributes. They are preserved through `SeaTunnelRow` and `RecordSerializer`, not through `CatalogTable`. Tests will round-trip `GeometryType`, column-level `srid`, and non-default column attributes through Zeta job-plan serialization and CDC `checkpointTables`. They will also round-trip row-level SRID and WKB through `RecordSerializer`. Preserving `GeometryKind` across these boundaries allows the Elasticsearch/OpenSearch sink to map `POINT` to `geo_point` and all other geometry kinds to `geo_shape`. ## 3. Older or unsupported components This change is backward compatible for existing non-spatial jobs, but a new spatial job is not supported on a mixed-version Zeta cluster. Existing non-spatial jobs do not contain SeaTunnel `GeometryType` or `Geometry` objects. Their schema, row, and checkpoint representations remain unchanged after the upgrade. One JDBC compatibility case requires attention. Generic MySQL `GEOMETRY` was previously exposed as `BYTES`; it is now exposed as `Geometry`. A job that depends on the previous raw-byte representation must explicitly select `ST_AsWKB(column)`. Concrete spatial types such as `POINT` previously failed schema conversion, and non-null CDC spatial values had no supported end-to-end contract. Therefore, there is no supported CDC spatial checkpoint to migrate. Before submitting a spatial job, the Zeta client, master, and all workers must be upgraded to a version containing `GeometryType` and `Geometry`. An older node cannot deserialize the new schema or row value, so the spatial job fails rather than being downgraded. Mixed-version execution of a spatial job is unsupported. After the upgrade, an engine or connector that does not implement `GEOMETRY` support must reject the type through its unsupported-type path. It must not silently reinterpret `GEOMETRY` as `BYTES` or `STRING`. ## 4. Zeta, Flink, and Spark boundaries ### Zeta `JobImmutableInformation` serializes each `LogicalVertex` through Hazelcast. `LogicalVertex.writeData` then serializes its action, whose object graph contains the `CatalogTable`. Rows transferred between workers use `RecordSerializer`. `Geometry` is serialized as a `SeaTunnelRow` field, preserving its SRID and WKB. Zeta supports this schema and value path. ### Flink After `FactoryUtil.createAndPrepareSource` discovers the source `CatalogTable`, the implementation validates it in `AbstractSourceExecuteProcessor.initializePlugins`. A geometry schema is rejected before `execute` registers the source with `fromSource` and before the Flink source reader is created. ### Spark Spark builds its schema through `MultiTableManager.getTableSchema` and `TypeConverterUtils.parcel/convert`. The implementation rejects `GEOMETRY` at this planning boundary, before a Spark source reader consumes data. For both Flink and Spark, rejection identifies the engine, table, column, and geometry kind and occurs before source data is consumed. Tests will verify that no source reader is created and no row-reading JDBC `ResultSet` is opened. ## 5. Column-level and row-level SRID The MySQL sources preserve any valid unsigned 32-bit SRID from `0` to `4294967295`. The Elasticsearch/OpenSearch sink accepts only SRID `4326`. `Geometry.srid` is authoritative for write-time validation. Column-level `options["srid"]` is used only for fail-fast validation when the sink is created. - Absent metadata does not imply `4326`; every row is still validated. - Metadata declaring `4326` does not bypass row validation. A non-4326 row fails when written. - Metadata declaring a non-4326 SRID fails during sink creation, before source data is consumed. - Stale non-4326 metadata may conservatively reject a valid job, but it cannot allow incompatible data to be written. The first version does not refresh or reconcile stale catalog or checkpoint SRID metadata against the current source schema. Row-level validation is always retained. ## 6. Existing-index validation For a static existing index, the implementation will validate its mapping before source data is consumed. - `POINT` requires `geo_point`. - Every other geometry kind requires `geo_shape`. - A compatible mapping allows the job to proceed. - A missing or incompatible field causes sink creation to fail. - The sink does not modify an incompatible existing field. The error must include the index, field, expected mapping type, and actual mapping type or `missing`. For example: ```text Index [places] field [location] expected type [geo_point], but actual type is [text] ``` Tests will cover compatible, missing, and incompatible mappings and verify that failure occurs before source data is consumed. Automatic geometry mapping for dynamic or placeholder index names is outside the first version. ## 7. Implementation organization No implementation PR will be opened before the STIP is endorsed. After endorsement, the implementation will be submitted as one PR organized into three dependency-ordered, individually buildable commits: 1. Geometry API, serialization tests, Zeta support, and Flink/Spark rejection. 2. MySQL JDBC and CDC source support. 3. Elasticsearch/OpenSearch mapping, existing-index validation, SRID enforcement, E2E tests, and documentation. Please let me know whether these additions resolve the review points. -- 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]
