Fix PinotDataWriter on UnsafeArrayData and null inputs#19067
Open
shuturmurgh wants to merge 1 commit into
Open
Fix PinotDataWriter on UnsafeArrayData and null inputs#19067shuturmurgh wants to merge 1 commit into
shuturmurgh wants to merge 1 commit into
Conversation
Spark's DataSourceV2 write path applies an UnsafeProjection immediately before invoking DataWriter.write(...), so every row reaching PinotDataWriter is an UnsafeRow whose array fields are UnsafeArrayData. The old ArrayType branch called record.getArray(idx).array.map(...), but UnsafeArrayData.array() throws UnsupportedOperationException, so every job projecting an ArrayType column failed on the first row and users had to flatten multi-value columns to comma-delimited strings in Spark SQL and rebuild them server-side via ingestionConfig.transformConfigs. Rewrite the ArrayType branch to use accessors valid on both UnsafeArrayData and GenericArrayData -- per-element iteration over the typed getXxx(i) methods -- and emit Object[] of boxed primitives (or String[] / byte[][]) because Pinot's segment-generation path requires that shape: PinotBufferedRecordReader.next(reuse) calls GenericRow.copy() which casts every array value to (Object[]), and the pinot-segment-local stats collectors all do the same. Returning a primitive int[] / long[] would ClassCastException at the first stats pass. Other fixes in the same write path: - Field-level isNullAt guards in internalRowToGenericRow so the scalar StringType branch no longer NPEs on a null UTF8String and primitive scalar branches no longer silently substitute 0 / false for a null. - Reject null elements within multi-value arrays with IllegalArgumentException naming the column and index, since Pinot's per-type stats collectors NPE on null elements anyway. - Guard the time-column tracking in write() with isNullAt so a null time value does not silently pull startTime to 0 and corrupt getSegmentName. - Switch the scalar StringType branch to getUTF8String(idx).toString to drop the dependency on Spark's test-only InternalRow.getString. Tests: - internalRowToGenericRow handles UnsafeArrayData and null fields: materializes a real UnsafeRow via UnsafeProjection, asserts the array fields are UnsafeArrayData, exercises every supported element type plus a null array field and null scalar fields. Verified to fail on the pre-fix writer with the exact NPE / UnsupportedOperationException the ticket described. - internalRowToGenericRow rejects null elements in multi-value array columns: covers all nine element types. - commit produces a valid segment from an UnsafeRow with multi-value array columns: drives the full SegmentIndexCreationDriverImpl pipeline, untars the segment, and asserts the metadata for INT / LONG / FLOAT / DOUBLE / STRING multi-value columns.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Spark's DataSourceV2 write path applies an UnsafeProjection immediately before invoking DataWriter.write(...), so every row reaching PinotDataWriter is an UnsafeRow whose array fields are UnsafeArrayData. The old ArrayType branch called record.getArray(idx).array.map(...), but UnsafeArrayData.array() throws UnsupportedOperationException, so every job projecting an ArrayType column failed on the first row and users had to flatten multi-value columns to comma-delimited strings in Spark SQL and rebuild them server-side via ingestionConfig.transformConfigs.
Rewrite the ArrayType branch to use accessors valid on both UnsafeArrayData and GenericArrayData -- per-element iteration over the typed getXxx(i) methods -- and emit Object[] of boxed primitives (or String[] / byte[][]) because Pinot's segment-generation path requires that shape: PinotBufferedRecordReader.next(reuse) calls GenericRow.copy() which casts every array value to (Object[]), and the pinot-segment-local stats collectors all do the same. Returning a primitive int[] / long[] would ClassCastException at the first stats pass.
Other fixes in the same write path: