[FEA] Per-row schema-mismatch diagnostic for read_json_with_diagnostics

[wjxiz1992]

Background

spark-rapids-jni's from_json_to_structs (see #4536) replicates Spark's depth-1 parent-NULL policy by reading the side-channel json_reader_diagnostics.top_level_columns_with_schema_mismatch returned by read_json_with_diagnostics and applying ALL_NULL to any top-level column flagged there.

Problem

The diagnostic is column-level, not row-level. cuDF reports that column X has SOME row(s) with a schema mismatch, but not which rows. Our workaround nulls the entire column, which over-nulls valid rows in a mixed-row batch:

Input rows for schema (data: struct<c1: int, c2: array<struct<c3, c4>>>):

{"data": {"c2": [{"c3": 19, "c4": "x"}], "c1": 1}}  // valid
{"data": {"c2": [19], "c1": 2}}                       // c2 is int, schema says struct

Expected (Spark): row 1 = Row(c1=1, c2=Array(Row(19,"x"))), row 2 = null
Actual (cuDF + current spark-rapids-jni workaround): both rows null.

Proposal

Add a row-level field to json_reader_diagnostics, e.g.

struct json_reader_diagnostics {
  std::vector<std::string> top_level_columns_with_schema_mismatch;        // existing
  std::unordered_map<std::string, std::vector<cudf::size_type>>
    top_level_columns_with_schema_mismatch_rows;                          // new
};

mapping column name → sorted vector of mismatched row indices for that column.

With this, the spark-rapids-jni loop can null only the affected rows (e.g. via cudf::detail::valid_if on a boolean column-of-rows mask), matching Spark's per-row depth-1 behavior exactly.

Notes

• Column-level diagnostic stays in place for backward compatibility.
• Row indices are already maintained inside the JSON reader during the schema-category check; surfacing them in the diagnostic struct should be a localized change.

Affected

• spark-rapids-jni#4536 — Spark parity workaround that this would close out.
• Any downstream consumer that needs Spark/Jackson-style isRoot=false nested-error-to-NULL propagation.

[wjxiz1992]

@ttnghia for vis, hope to get more advice on this!

[jihoonson]

The overall approach seems reasonable to me. Regarding the new std::unordered_map<std::string, std::vector<cudf::size_type>> top_level_columns_with_schema_mismatch_rows; map, I guess the idea is that the spark task will iterate each column and nullify invalid rows. We may use some cuDF API like cudf::detail::valid_if or cudf::detail::apply_mask for nullifying. Is this right? Then, why don't we return a boolean mask for each column in the first place and store them in the map instead of storing row indices?

Also, do we still need the existing top_level_columns_with_schema_mismatch vector even after we add the new map? the keys of the map will contain the same columns as the vector.

[wjxiz1992]

Thanks for the direction @ttnghia @jihoonson . I implemented this as a GPU-side row-level diagnostic path instead of CPU fallback.

Current split:

• cuDF draft PR: Add row-level JSON schema mismatch diagnostics rapidsai/cudf#22915 adds row indices to read_json_with_diagnostics for schema mismatches, while keeping plain read_json on the no-diagnostics path.
• JNI draft PR: from_json: null only schema-mismatched rows #4728 consumes those row diagnostics and nullifies only the affected row's depth-1 parent field, preserving sibling fields in the same row.

The JNI PR is draft until rapidsai/cudf#22915 lands, because the submodule pointer needs to reference a commit available from rapidsai/cudf for CI checkout.