Apache Flink Adds UUID Type And Runtime Fixes

Apache Flink is a distributed stream and batch engine. Recent master activity on apache/flink landed 27 commits across 177 files (7599 insertions, 2161 deletions). The operator facing pieces are a new UUID Table type, native S3 writer and plugin packaging fixes, and two runtime bugs that stall watermarks or undercount checkpoint acks during failover.

Master introduced a UUID logical type under LogicalTypeRoot.UUID. The new UuidType.java stores any 128 bit UUID as the canonical 16 byte big endian encoding from RFC 9562. Version and variant bits are kept as is and never validated. Default conversion is java.util.UUID. The serializable name is UUID.

DataTypes.java exposes DataTypes.UUID(). SQL Gateway JSON serde and the generated REST schemas (rest_v1 through rest_v4) accept the new root. The planner maps it to Calcite SqlTypeName.UUID. Runtime TypeCheckUtils treats UUID as not comparable, same bucket as VARIANT and BITMAP. ORDER BY and comparison joins are out unless the value is cast.

public static DataType UUID() {
    return new AtomicDataType(new UuidType());
}

This is type system plumbing, marked @PublicEvolving. Connector and function coverage will lag. Jobs that stuffed UUIDs into CHAR or STRING can start modeling the value correctly, but this commit does not migrate existing tables.

The same window also pushed VARIANT further. The raw format now accepts a single VARIANT column, decoding bytes with raw.charset like PARSE_JSON (duplicate keys rejected) and writing via Variant#toJson. The round trip is value lossless, not byte lossless: whitespace is dropped and object keys are ordered. raw.endianness does not apply. Pair it with raw.line-delimiter for newline delimited JSON. VARIANT also works in user defined functions and process table functions.

The SeaweedFS change is mostly test infrastructure. In memory mocks for the native S3 writer were replaced with a Testcontainers harness against SeaweedFS, plus high availability job and application run tests. That is not a new production filesystem. The one production edit in NativeS3RecoverableFsDataOutputStream.java stores incomplete multipart parts under .incomplete/<uploadId>/... instead of nesting them under the destination object key. Lifecycle rules keyed on the old prefix will miss leftovers.

Two related writer fixes landed around the same files. Temp file cleanup is now idempotent: Files.delete became Files.deleteIfExists, so a missing part file no longer fails close or persist. The flink-s3-fs-native shade plugin now includes *:* instead of a hand maintained AWS SDK / Netty / HTTP list. Extra plugin dependencies stop being dropped on the floor. The shaded jar gets larger. That is the tradeoff.

Local temp files elsewhere picked up a permissions fix. A CWE-378 change replaced File.createTempFile with Files.createTempFile in PackagedProgram, the DFS changelog cache, YarnClusterDescriptor job graph and config files, and one Table example. Files.createTempFile uses tighter defaults than the older File API. Yarn session startup and jar entry extraction are the blast radius worth caring about.

StatusWatermarkValve lost watermarks and could stall after a subpartition went idle, resumed, and only partly caught up. Two invariants broke. The all idle flush skipped unless the last idle input held the current min, so the final watermark depended on idle order. The idle to active path added a caught up subpartition back to the aligned set without deriving the min again. With no other aligned input, that watermark sat until a larger one arrived.

The fix always flushes the max watermark once every subpartition is idle, and derives the min again after a realign (after emitting ACTIVE so downstream does not drop it while the input still looks idle). Jobs with idle detection on keyed inputs are the ones to retest. Combined watermark status already flushed unconditionally since an earlier change. The valve now matches that behavior.

Checkpoint planning had a quieter bug. DefaultCheckpointPlanCalculator treated any terminal execution as finished, because Execution.isFinished() meant state.isTerminal(). FAILED and CANCELED tasks were dropped from tasksToWaitFor during failover, so the plan expected fewer acks than it should. The calculator now requires every task to be RUNNING or genuinely FINISHED and otherwise aborts with a CheckpointException. The check is the same for streaming and batch. Extra aborted checkpoints during failover are the cost. A checkpoint that completed on a partial graph was the old failure mode.

Interval join now emits and retracts early fire results. The planner already had earlyFireDelay. The operator was not receiving it (-1 disabled the path). Speculative outer join pads go out after the delay, then retract with UPDATE_BEFORE / UPDATE_AFTER when a match arrives. Savepoints taken before this feature restore with empty early fire state. Jobs that set an early fire delay and saw late outer pads were waiting on dead code.

MAP_FROM_ENTRIES is a new built in. Input is an array of rows with two fields. Last write wins on duplicate keys. NULL keys collapse to one entry. A NULL array or a NULL entry returns NULL.

-- {1=uno, 2=two}
MAP_FROM_ENTRIES(ARRAY[ROW(1, 'one'), ROW(2, 'two'), ROW(1, 'uno')])

The parser now accepts LATERAL fn(args) without TABLE, matching CALCITE-7183. Docs prefer the shorter form. Existing LATERAL TABLE(...) still parses.

Python UDF calls that appear in both a WHERE and a SELECT were evaluated twice. RexProgram already shared them via RexLocalRef, then Python calc translation expanded the refs back into separate trees. Two optimizer rules now keep one evaluation for deterministic calls. Nondeterministic calls stay independent. PyFlink SQL jobs that paid for the same remote UDF twice in one calc should see one call. The DataFrame API also gained drop_duplicates (keep first or last, optional order_by).

SHOW CREATE printed FROM_NOW as a future timestamp. Display only. It did not change which rows were read.

Source builds now pin Maven 3.9.16 (up from 3.8.6) in pom.xml. Plugin log lines switched from artifactId based to goal prefix based. CI parsers were updated to accept both. Apache Parent POM is 39. Local builds still on system Maven 3.8.x will fail the enforcer.

USING CONNECTION is now stored on CatalogTable / CreateTableOperation and round trips under connection.identifier (FLIP-529). That is catalog plumbing, not a finished connector feature.

UUID is an extension type. Do not assume every format and catalog can persist it yet. Retest idle watermark alignment and failover checkpoints on master snapshots. Those two runtime fixes are the ones that change production graphs.