Fixing issue with row update not working for PK table:

[vamossagar12]

Purpose

Linked issue: close #2397

Brief change log

[FLUSS-2397] Fix UPDATE failure for primary key tables with computed columns

This commit fixes a bug where UPDATE statements failed when target tables
contained computed columns (e.g., PROCTIME). The root cause was that the
SinkWriter was using the table's physical RowType to initialize the
serialization schema, rather than the "consumed" RowType which includes
computed/metadata columns.

Key changes:

• Introduced 'consumedRowType' in FlinkSinkWriter, AppendSinkWriter, and
  UpsertSinkWriter to properly handle schema projection during serialization.
• Updated FlinkTableSink to ensure the correct row type is passed down to
  the writer.
• Added a regression test in FlinkTableSinkITCase that reproduces the
  reported failure (Bug Row UPDATE not working (PK Table) #2397).
• Updated internal builders and tests to accommodate the schema context.

Tests

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java

API and Format

N/A

Documentation

N/A

[vamossagar12]

@polyzos , sorry for the delay- I got busy with other stuff. I created this PR. Please take a look!

[vamossagar12]

@polyzos ptal whenever you get the chance. Thanks!

[vamossagar12]

@loserwang1024 , @wuchong can you PTAL at this PR when you have some time. Thanks!

[vamossagar12]

@loserwang1024 , @wuchong gentle reminder.

[polyzos]

Thanks a lot for tackling this @vamossagar12

The runtime fix in RowDataSerializationSchema is a nice, low-risk approach. Before we merge, I had a few suggestions I'd love your thoughts on:

1. The consumedRowType plumbing looks like it isn't doing anything yet

Tracing the field, it's always assigned tableRowType no code path populates it with a different value. The runtime check in RowDataSerializationSchema.serialize() already detects the extra arity, so the plumbing seems redundant today.

Could we either:

• drop the consumedRowType parameter from these constructors and rely solely on the runtime detection, or
• actually source it from Flink's Context / physical row type in getSinkRuntimeProvider (or from applyRowLevelUpdate)
  Either is fine

2. Identity projection assumes computed columns are trailing

RowDataSerializationSchema builds identityMapping[i] = i, which works when computed columns sit at the end (as in the test). If a user defines a INT, ptime AS PROCTIME(), b STRING, the trailing physical column would land at the wrong index. Could we either handle the mid-schema case, or add a comment + test documenting the trailing-only assumption?

3. The new test doesn't assert the update happened

Nit: consumedRowType in FlinkTableSink:95 is non-final only because copy() writes to it post-construction, if we keep the field, passing it through the constructor and marking it final would be cleaner.

Let me know your thoughts, and thanks again for the fix. 🙏 cc @wuchong

[vamossagar12]

@polyzos i made the requested changes.

[vamossagar12]

@polyzos , looking at the logs, it is a bit hard for me to say why the Java 11 tests are failing. Would it be possible for someone to rerun the tests?