[FLINK-40232][table-runtime] Handle null rowtime in OutputConversionOperator

[imankhadiv]

What is the purpose of the change

OutputConversionOperator.processElement dereferences the result of RowData#getTimestamp(...) without a null check, so converting a table with a null rowtime attribute into a DataStream fails the job with a NullPointerException. Both rowtime paths are affected: the rowtime metadata path and the rowtime column path.

Adding null checks alone is not enough. outRecord is a single StreamRecord allocated once in open() and reused for every element, and StreamRecord#replace(Object) leaves the timestamp untouched. Merely skipping setTimestamp() would leave the previous record's timestamp on the record carrying the null rowtime, i.e. it would trade the NullPointerException for a silently wrong timestamp. The null case therefore erases the timestamp explicitly.

Brief change log

• Both rowtime lookups in OutputConversionOperator go through a new private updateRowtime(RowData, int) helper
• The helper sets the output timestamp when the rowtime is present, and calls StreamRecord#eraseTimestamp() when it is null, so a record with a null rowtime cannot inherit the preceding record's timestamp

Verifying this change

This change added tests and can be verified as follows:

• Added OutputConversionOperatorTest with four cases — a null rowtime is emitted without a timestamp, and a null rowtime does not inherit the timestamp of the preceding record — each covering the rowtime metadata path and the rowtime column path
• All four tests were watched failing before the fix: first with the reported NullPointerException, and the two "does not inherit" cases again against a null-check-only variant of the fix, which is what motivated the explicit eraseTimestamp()
• flink-table-runtime module test suite passes (1824 tests)
• DataStreamJavaITCase passes (38 tests); it covers the non-null rowtime path end to end via toDataStream with event-time windows and a downstream assertion on ctx.timestamp()

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

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes — one additional isNullAt check per record, in an operator that already read the rowtime field
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

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

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Claude Opus 5)

[flinkbot]

CI report:

• a0142a2 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[imankhadiv]

@nateab thanks for assigning me FLINK-40232 — Azure is green on this one, so it's ready for review.

Both rowtime lookups in OutputConversionOperator now go through a small updateRowtime(RowData, int) helper that null-checks the rowtime. The one part that goes beyond the null checks suggested in the ticket is the else branch: outRecord is a single StreamRecord allocated in open() and reused for every element, and StreamRecord#replace(Object) leaves the timestamp untouched — so only skipping setTimestamp() would leave the previous record's timestamp on the record with the null rowtime, turning the NPE into a silently wrong timestamp. Hence the explicit eraseTimestamp().

Test coverage in OutputConversionOperatorTest follows that split: for both the rowtime metadata path and the rowtime column path, one case asserts the record is emitted without a timestamp, and one asserts it does not inherit the preceding record's timestamp. The latter two were confirmed failing against a null-checks-only version of the fix, which is what motivated the erase. Locally flink-table-runtime (1824 tests) and DataStreamJavaITCase (38 tests, covers the non-null rowtime path end to end) pass.

Could you take a look, or point me to the right committer for table-runtime? Happy to rebase or adjust anything.

[davidradl]

nit: you could move updateRowtime call outside the if else and set the index in the if else.

[imankhadiv]

Applied in a0142a2 — the position is now computed in the if/else and updateRowtime is called once, with the two type comments kept on their respective branches.

[imankhadiv]

Azure is green on a0142a2. Since the nit is addressed and the change is confined to OutputConversionOperator, would you know which committer is the right one to tag for a table-runtime review?