TASK 10: users need explicit sql functions to create geography values

[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 avoid sharing WKBReader/WKBWriter as static singletons here? JTS documents these classes as reusable but not thread-safe, and scalar functions may be evaluated concurrently by different tasks/threads in the same JVM. A per-call instance or ThreadLocal would make the constructor functions safe under parallel execution. We can apply the same pattern to WKTReader for consistency.

[davidchaava]

Could we make the WKB validation boundary stricter here? fromWkb() validates by reading through JTS but returns the original WKB bytes. JTS WKBReader repairs some structurally invalid inputs such as LineStrings/LinearRings with too few points or non-closed rings, so those inputs may validate successfully while we still store the unrepaired raw WKB. Since ST_GEOGFROMWKB is the constructor boundary, I think we should either add explicit strict checks for those WKB invariants before returning the raw bytes, or add tests that confirm such inputs are rejected.

fromWkb() validates by reading with JTS, but then returns the original geography bytes. JTS WKBReader is lenient: its source says it repairs structurally invalid LineStrings/LinearRings, e.g. too few points or non-closed rings. That means invalid WKB can pass validation while Flink stores the original invalid bytes.

[davidchaava]

Could we reject non-finite coordinates before the CRS84 range checks? JTS WKT supports NaN, and Java comparisons with NaN return false, so values such as POINT (NaN 0) can bypass the current longitude/latitude checks. We should probably allow explicit empty geometries, but reject NaN/infinite ordinates in non-empty coordinate sequences.

The CRS84 range checks don’t reject NaN. JTS WKTReader explicitly supports NaN numbers, and comparisons like longitude < -180D / longitude > 180D are false for NaN, so POINT (NaN 0)-style input can slip through. POINT EMPTY is fine, but non-empty non-finite coordinates should not become valid GEOGRAPHY.