TASK 4: add planner validation for unsupported conversions

[gkalashyan-akv]

What is the purpose of the change

(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)

Brief change log

(for example:)

• The TaskInfo is stored in the blob store on job creation time as a persistent artifact
• Deployments RPC transmits only the blob storage reference
• TaskManagers retrieve the TaskInfo from the blob cache

Verifying this change

Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (100MB)
• Extended integration test for recovery after master (JobManager) failure
• Added test that validates that TaskInfo is transferred only once across recoveries
• Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
• The serializers: (yes / no / don't know)
• The runtime per-record code paths (performance sensitive): (yes / no / don't know)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
• The S3 file system connector: (yes / no / don't know)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

[davidchaava]

Could we clarify the intended versioning contract for GeographyDataSerializer? The FLIP describes the Flink state/shuffle serializer format as version byte + length + ISO WKB payload, while this implementation writes length + payload only. I understand Flink serializers can also evolve through TypeSerializerSnapshot, and existing serializers such as Bitmap/Variant do not necessarily put a version byte into every value. It would be good to either align this serializer with the FLIP or explicitly update the design to say that format evolution is handled only through the serializer snapshot. The Parquet/Iceberg payload should of course remain raw WKB without any Flink prefix.

https://docs.google.com/document/d/1rpOTETT_Ui3TlEGioUr2NKJ1p1dlxjJQudXHndxBpO0/edit?tab=t.0
On-disk format: one-byte version header 0x01, followed by 4-byte big-endian length, followed by the ISO WKB payload.

[gkalashyan-akv]

Good catch. This PR still uses the older internal serializer shape, but the mismatch is already fixed in the TASK 6 follow-up branch (task/users-and-connectors-need-portable-output-formats): the internal GEOGRAPHY serializer now uses version byte + length + ISO WKB payload, with tests covering the versioned envelope and unsupported-version handling. External connector/file payloads such as Parquet/Iceberg remain raw WKB without any Flink prefix.

[davidchaava]

Could we clarify the validation boundary for BinaryGeographyData? Right now constructing the binary wrapper fully walks the WKB payload via validateGeometry(), and this path is also used by BinaryRowData.getGeography(). My reading of the FLIP is that runtime/pass-through paths should keep GEOGRAPHY as raw ISO WKB and only support cheap header inspection, while full validation belongs to constructors such as ST_GEOGFROMWKB or connector ingestion. If the intended invariant is that every GeographyData instance is fully validated, could we document that trade-off and add a test/performance rationale?

BinaryGeographyData fully walks and validates the WKB structure on construction/read, including BinaryRowData.getGeography(). Proposal says runtime/pass-through paths should store raw ISO WKB and not decode; cheap header inspection only. This may add repeated O(n) validation on normal row access and connector pass-through.