[FLINK-40528][table] Make codegen tolerant to partial deletes

[snuyanzin]

What is the purpose of the change

The PR makes codegen tolerant to partial deletes

Brief change log

codegen

Verifying this change

tests

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): (no)
• 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: [Tool Name and Version]

[flinkbot]

CI report:

• 589bd9f Azure: SUCCESS
• bdfa85d Azure: PENDING
• 749d1c4 UNKNOWN

Bot commands

The @flinkbot bot supports the following commands:

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

[gustavodemorais]

Thanks for looking into this and the fix, Sergey!

I like the fix. As I expected though, we still have more gaps:

1. generatePrimitiveArrayUpdateCode (ARRAY[...]/MAP[...] literals) still gates on elementType.isNullable only:

SELECT id, ROW(id, ARRAY[v, 99]) FROM source_t
-D[1, +I[1, [-1, 99]]]   <- should be null, writes -1

generateMap hits the same function for fixed-length primitive keys/values.

2. JsonGenerateUtils.createNullableNodeTerm has the same gate, also untouched

SELECT id, JSON_OBJECT('r' VALUE ROW(v)) FROM source_t
-D[1, {"r":{"EXPR$0":0}}]   <- should be null, writes 0

3. And separately, we also need to fix the ConstraintEnforcer. FLINK-40477 (NOT_NULL_ERROR_DELETE_BY_KEY - bare scalar passthrough, no composite at all) is a different bug in a different module (ConstraintEnforcerExecutor). Simple projections without Row fail still because we have partial deletes.

SELECT id, v FROM source_t   -- sink column v INT NOT NULL
  EnforcerException: Column 'v' is NOT NULL, however, a null value is being written into it.

In general, the partial deletes optimization is broken. I think these three above are doable so I'm +1 into fix forward: we fix them instead of forcing ChangelogNormalize back into the pipelines. I think 1. fits into this PR and 2. maybe, up to you. For 3, I think it makes sense to do it in another PR.

[gustavodemorais]

Can we extract this as a small helper to slightly improve readability?

/** True only when codegen proved this statically; a dynamic nullTerm may still be null at runtime. */
def isProvenNonNull: Boolean = nullTerm == NEVER_NULL
...
} else if (tpe.isNullable || !element.isProvenNonNull) {