[FLINK-40346][model-triton] Preserve null array elements when deserializing Triton responses

[SEPURI-SAI-KRISHNA]

What is the purpose of the change

TritonTypeMapper#deserializeArrayFromJson silently corrupts null elements in array-typed
Triton responses. A JSON null inside an array is read back as 0, false, or the literal
four-character string "null", with isNullAt() reporting false at that position — so the
substituted value is indistinguishable from a real prediction downstream.

Two things combine to cause it:

1. Each branch builds a primitive backing array (int[], double[], boolean[], ...) and
   fills it with Jackson's coercing accessors. NullNode.asInt() returns 0,
   asBoolean() returns false, asDouble() returns 0.0, and asText() returns the string
   "null". No branch checks isNull().
2. Even a check would not have been enough: GenericArrayData#isNullAt is
   return !isPrimitiveArray && ((Object[]) array)[pos] == null;, so a primitive-backed
   GenericArrayData reports isNullAt() == false at every position by construction. A null
   cannot be represented in that shape at all.

Measured on unpatched master:

Declared type	JSON payload	Read back as
ARRAY<STRING>	["a", null, "b"]	isNullAt(1) == false, getString(1) == "null" (4 chars)
ARRAY<INT>	[1, null, 3]	isNullAt(1) == false, getInt(1) == 0
ARRAY<DOUBLE>	[1.5, null]	getDouble(1) == 0.0
ARRAY<BOOLEAN>	[true, null]	getBoolean(1) == false

This is a correctness bug rather than a crash, which is what makes it worth fixing: an inference
result of 0 or false looks like a legitimate model output, so the corruption propagates
silently into user queries.

It is also an asymmetry within the same class. serializeArrayToJsonArray already emits
addNull() for a null element, and the scalar path deserializeFromJson already returns null
for a null node. Only the array deserialization path drops the information, so a null does not
survive a serialize/deserialize round trip.

Brief change log

• deserializeArrayFromJson scans for a null element first. When there is none — the common case
  — the existing primitive fast path runs completely unchanged.
• When a null is present, deserializeNullableArrayFromJson builds a boxed Object[] and
  delegates each element to the existing deserializeFromJson, which already maps a null node to
  a Java null and already covers every element type the primitive path supports, including the
  FloatType isNumber() special case.
• Nested array element types are rejected on the nullable path as well, so the presence of a null
  element cannot change which element types are accepted.
• A null against a NOT NULL element type is rejected with a clear message rather than written
  into an array whose declared type forbids it.

Verifying this change

This change adds tests and can be verified as follows:

• TritonTypeMapperTest#testDeserializeArrayWithNullStringElement
• TritonTypeMapperTest#testDeserializeArrayWithNullNumericElements
• TritonTypeMapperTest#testDeserializeArrayWithNullBooleanElement
• TritonTypeMapperTest#testNullElementSurvivesSerializeDeserializeRoundTrip
• TritonTypeMapperTest#testDeserializeArrayRejectsNullForNotNullElementType
• TritonTypeMapperTest#testDeserializeArrayWithoutNullsIsUnchanged

Red/green verified against this exact commit:

• Without the production change: Tests run: 14, Failures: 5 — the five null cases fail,
  reading back exactly the 0 / false / "null" values tabulated above.
• With it: TritonTypeMapperTest 14/14, and the whole flink-model-triton module
  Tests run: 99, Failures: 0, Errors: 0, Skipped: 0.

testDeserializeArrayWithoutNullsIsUnchanged passes in both directions by design — it pins the
primitive fast path so a future change cannot quietly route null-free payloads through the boxed
branch.

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, marginally. Null-free
  payloads keep the existing primitive arrays and gain one isNull() pass over a node list that
  is iterated immediately afterwards. Payloads containing a null take a boxed array, which is
  the only shape that can represent the value correctly.
• 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

[SEPURI-SAI-KRISHNA]

@dianfu thanks for the assignment, PR is up. Red/green verified against this commit; details in the description.

[flinkbot]

CI report:

• 74bfd43 Azure: SUCCESS
• 936c053 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

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

[dianfu]

Allocating Object[] violates GenericArrayData's requirement that boxed primitive arrays retain their concrete component type. For example, nullable ARRAY<INT> advertises Integer[], but Flink's external conversion fast path returns this underlying Object[]; generated or user code casting it to Integer[] can then fail with ClassCastException.

[dianfu]

It's also good to call toTritonDataType(elementType) before the loop to reject unsupported types.

[SEPURI-SAI-KRISHNA]

Thanks @dianfu — you're right on both counts, and the first one is reachable.

I reproduced it: with an Object[] backing array, ArrayObjectArrayConverter#toExternal takes the hasInternalElements fast path and returns the backing array directly to the caller, which fails with

java.lang.ClassCastException: class [Ljava.lang.Object; cannot be cast to class [Ljava.lang.Integer;

The GenericArrayData Javadoc states the requirement explicitly too ("Object arrays that contain boxed types ... MUST be boxed arrays"), so the patch was violating a documented contract. Good catch.

Fixed in 936c053 by allocating the concrete component type, matching the idiom already used in ArrayObjectArrayConverter and ArrayListConverter:

Object[] array =
        (Object[])
                Array.newInstance(
                        LogicalTypeUtils.toInternalConversionClass(elementType), size);

Two tests cover it:

• testNullableArrayKeepsConcreteComponentType — pins the component type for all eight supported element types (Integer[], Long[], Byte[], Short[], Float[], Double[], Boolean[], StringData[])
• testNullableArraySurvivesExternalConversion — drives the real DataStructureConverters path and reproduces the ClassCastException above without the fix

Your second suggestion is adopted as well — toTritonDataType(elementType) is now called before the loop. One note: it recurses into ArrayType rather than rejecting it, so it doesn't reject a nested array on its own. I kept the explicit ArrayType guard alongside it so the nullable path accepts exactly the same element types as the primitive path.

Also rebased onto master and pushed the fix as a separate commit, so the delta against your review is isolated in 936c053.

flink-model-triton is green locally at Tests run: 101, Failures: 0, Errors: 0, Skipped: 0 (checkstyle and spotless included); waiting on Azure for the new head. PR description updated to match.

[dianfu]

@SEPURI-SAI-KRISHNA Thanks for the update! LGTM.

[SEPURI-SAI-KRISHNA]

@dianfu Thanks for your time and review