[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:

• 894a8da Azure: FAILURE

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) {

[gustavodemorais]

Thanks for the improvements, @snuyanzin! We caught many more cases. Now added only some nits

[gustavodemorais]

test is inner join, which one did you want?

Suggested change

	* A LEFT JOIN whose probe (left) side produces a delete-by-key tombstone carrying null for a
	* A INNER JOIN whose probe (left) side produces a delete-by-key tombstone carrying null for a

[gustavodemorais]

nit: these are a lot of tests. I'm ok with it if they're really necessary. Maybe we can reduce the number of tests without losing coverage?

For example, I think these two were passing before the change as well. Is there value in keeping them?
INSERT_SELECT_DELETE_BY_KEY_WITH_NOT_NULL_CAST and INSERT_SELECT_DELETE_BY_KEY_WITH_NESTED_NOT_NULL_STRING

[snuyanzin]

INSERT_SELECT_DELETE_BY_KEY_WITH_NOT_NULL_CAST

this is required
it highlights another finding flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala

[fhueske]

Thanks for the fix @snuyanzin.

The changes look good to me.
Would be nice if you could fix the incorrect comment that Gustavo pointed out.

Thanks, Fabian

[fhueske]

Which expressions are proven not null at runtime if we do not trust the types?
It would only be literals and expressions on literals, no?