perf(bigquery): unwrap cell JSON during streaming

[lqiu96]

Summary

Eliminates reflective TableCell and ArrayMap allocations when parsing row responses in HttpBigQueryRpc (listTableData, listTableDataWithRowLimit, getQueryResults, getQueryResultsWithRowLimit, and queryRpc).

Background

In the BigQuery v2 REST API, each row is serialized as {"f": [{"v": ...}, {"v": ...}, ...]}. Previously, request.execute() used google-http-client's reflective JsonParser to instantiate a TableCell (extends GenericJson -> GenericData) for every cell:

• Each cell invoked Data.newMapInstance -> Types.newInstance(TableCell.class) -> Class.getDeclaredConstructor() (allocating a Class[0] array), plus EnumSet.noneOf(...), an ArrayMap for unknownFields, and the TableCell instance itself (4 object allocations per cell).
• Immediately after RPC deserialization, FieldValueList.fromPb and FieldValue.fromPb unwrapped cell.getV() and discarded the TableCell.
• Because TableCell implements Map<String, Object>, FieldValue.fromPb also fell into cellPb instanceof Map, triggering reflective ClassInfo/FieldInfo lookups on "f" and "v" for every cell before recursing on cell.getV().

Changes

1. HttpBigQueryRpc.java:
   • Adds executeAndParseRowResponse, parseRowResponse, parseTableRows, and parseCellValue to stream the "rows" array and unwrap single-key {"v": ...} cell objects directly into their inner representations (String for scalars, Data.nullOf(Object.class) for null, List<Object> for REPEATED arrays, and Collections.singletonMap("f", subFields) for RECORD structs).
   • Delegates all non-"rows" top-level fields (schema, pageToken, @JsonString 64-bit numbers such as totalRows and totalBytesProcessed, jobReference, errors, etc.) to ClassInfo and JsonParser for full metadata fidelity.
2. FieldValue.java:
   • Adds an explicit cellPb instanceof TableCell fast-path before cellPb instanceof Map and checks cellPb == null || cellPb == NULL_CELL_SENTINEL before Data.isNull(cellPb).

[gemini-code-assist]

Code Review

This pull request optimizes BigQuery REST response parsing by implementing a custom streaming JSON parser in HttpBigQueryRpc to bypass reflective TableCell allocations, significantly reducing memory overhead. It also updates FieldValue to support unwrapped cell representations and adds corresponding unit tests. The code review highlights critical robustness issues in the new parser: several loops parsing JSON arrays do not check for EOF (null tokens), which could cause infinite loops and OutOfMemoryError on truncated or malformed streams. Additionally, a potential UnsupportedOperationException was identified when attempting to mutate an immutable singleton map if unexpected extra keys are encountered after an 'f' field.

[gemini-code-assist]

If the JSON stream is truncated or malformed, parser.nextToken() can return null (representing EOF). Since null != JsonToken.END_ARRAY is true, the loop will continue. Inside the loop, parser.getCurrentToken() will be null, which is not START_OBJECT, so it calls parser.skipChildren(). Since skipChildren() does nothing when the current token is null, this results in an infinite loop consuming 100% CPU. Additionally, the inner loop reading cells will also loop infinitely adding JSON_NULL to the list, leading to an OutOfMemoryError.

To prevent this, ensure all while (parser.nextToken() != JsonToken.END_ARRAY) loops also check for null (EOF).

    JsonToken token;
    while ((token = parser.nextToken()) != JsonToken.END_ARRAY && token != null) {
      if (token != JsonToken.START_OBJECT) {
        parser.skipChildren();
        continue;
      }
      List<Object> cells = null;
      while (parser.nextToken() == JsonToken.FIELD_NAME) {
        String rowField = parser.getText();
        JsonToken valueToken = parser.nextToken();
        if ("f".equals(rowField) && valueToken == JsonToken.START_ARRAY) {
          // Pre-size each row's cell list from the previous row's column count to avoid
          // ArrayList resizing across rows of the same table/query schema.
          cells = new ArrayList<>(expectedCols);
          JsonToken cellToken;
          while ((cellToken = parser.nextToken()) != JsonToken.END_ARRAY && cellToken != null) {
            cells.add(parseCellValue(parser));
          }
          expectedCols = cells.size();
        } else {
          parser.skipChildren();
        }
      }
      TableRow row = new TableRow();
      if (cells != null) {
        row.setF((List) cells);
      }
      rows.add(row);
    }

[gemini-code-assist]

If the JSON stream is truncated or malformed, parser.nextToken() can return null (representing EOF). Since null != JsonToken.END_ARRAY is true, the loop will continue infinitely, adding JSON_NULL to subFields and eventually causing an OutOfMemoryError.

To prevent this, ensure the loop condition also checks for null (EOF).

Suggested change

	List<Object> subFields = new ArrayList<>();
	if (parser.getCurrentToken() == JsonToken.START_ARRAY) {
	while (parser.nextToken() != JsonToken.END_ARRAY) {
	subFields.add(parseCellValue(parser));
	if (parser.getCurrentToken() == JsonToken.START_ARRAY) {
	JsonToken subToken;
	while ((subToken = parser.nextToken()) != JsonToken.END_ARRAY && subToken != null) {
	subFields.add(parseCellValue(parser));
	}

[gemini-code-assist]

If the JSON stream is truncated or malformed, parser.nextToken() can return null (representing EOF). Since null != JsonToken.END_ARRAY is true, the loop will continue infinitely, adding JSON_NULL to the list and eventually causing an OutOfMemoryError.

To prevent this, ensure the loop condition also checks for null (EOF).

Suggested change

	if (token == JsonToken.START_ARRAY) {
	List<Object> list = new ArrayList<>();
	while (parser.nextToken() != JsonToken.END_ARRAY) {
	list.add(parseCellValue(parser));
	}
	return list;
	}
	if (token == JsonToken.START_ARRAY) {
	List<Object> list = new ArrayList<>();
	JsonToken arrayToken;
	while ((arrayToken = parser.nextToken()) != JsonToken.END_ARRAY && arrayToken != null) {
	list.add(parseCellValue(parser));
	}
	return list;
	}

[gemini-code-assist]

If the JSON object contains an 'f' field followed by unexpected extra keys, mapValue is initialized as an immutable singleton map via Collections.singletonMap("f", subFields). In the next iteration, attempting to call mapValue.put(key, ...) will throw an UnsupportedOperationException.

To prevent this, copy the map to a mutable HashMap if it is not already a HashMap before adding extra keys.

          if (mapValue == null) {
            mapValue = new HashMap<>();
            if (hasValue) {
              mapValue.put("v", unwrappedValue);
              hasValue = false;
            }
          } else if (!(mapValue instanceof HashMap)) {
            mapValue = new HashMap<>(mapValue);
          }
          mapValue.put(key, parseCellValue(parser));