feat(bigquery): add ArrowDeserializer helper utility

[jinseopkim0]

Stacked PR 2 of 3: Adds the ArrowDeserializer class which handles decoding serialized Arrow schemas and record batches into standard FieldValueList rows.

[gemini-code-assist]

Code Review

This pull request introduces the ArrowDeserializer class to convert Apache Arrow schemas and record batches into BigQuery schemas and FieldValueList rows. Feedback on the implementation highlights several critical issues: a potential native memory leak if vector creation fails, a bug in repeated record deserialization where subfields are omitted, missing defensive checks for schema and size mismatches, and an inefficient timestamp conversion that should cast to TimeStampVector and handle different time units.

[gemini-code-assist]

If an exception occurs during the creation of FieldVectors in the loop, any vectors already created will not be closed, leading to a native memory leak in the RootAllocator. Wrapping this in a try-catch block to close already created vectors in reverse order (LIFO) ensures exception safety.

      List<FieldVector> vectors = new ArrayList<>();
      try {
        for (Field field : arrowSchema.getFields()) {
          vectors.add(field.createVector(allocator));
        }
      } catch (Throwable t) {
        for (int i = vectors.size() - 1; i >= 0; i--) {
          try {
            vectors.get(i).close();
          } catch (Exception e) {
            t.addSuppressed(e);
          }
        }
        throw t;
      }

References

1. When managing a collection of closeable resources (e.g., scopes), ensure they are closed in the reverse order of their creation (LIFO). The implementation must be exception-safe to prevent resource leaks, meaning all opened resources should be closed even if exceptions occur during their creation or closing.

[gemini-code-assist]

Critical Bug: When deserializing repeated record fields (arrays of structs), elementBqField is constructed without copying the subfields of the record. Consequently, when arrowVectorToFieldValue is called recursively, bqField.getSubFields() will be empty or null, throwing a NullPointerException or IndexOutOfBoundsException when accessing subfields. We must set the subfields on the builder if the type is RECORD.

Suggested change

	com.google.cloud.bigquery.Field elementBqField =
	com.google.cloud.bigquery.Field.newBuilder(bqField.getName(), bqField.getType())
	.setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE)
	.build();
	com.google.cloud.bigquery.Field.Builder elementBuilder =
	com.google.cloud.bigquery.Field.newBuilder(bqField.getName(), bqField.getType());
	if (bqField.getType() == LegacySQLTypeName.RECORD && bqField.getSubFields() != null) {
	elementBuilder.setType(LegacySQLTypeName.RECORD, bqField.getSubFields());
	}
	com.google.cloud.bigquery.Field elementBqField =
	elementBuilder.setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE).build();

[gemini-code-assist]

Defensive programming check: If an Arrow List field is malformed or programmatically constructed without children, calling getChildren().get(0) will throw an IndexOutOfBoundsException. Adding a check ensures safety.

Suggested change

	if (type instanceof ArrowType.List) {
	Field innerField = arrowField.getChildren().get(0);
	if (type instanceof ArrowType.List) {
	if (arrowField.getChildren().isEmpty()) {
	throw new IllegalArgumentException("Arrow List field must have at least one child field: " + name);
	}
	Field innerField = arrowField.getChildren().get(0);

[gemini-code-assist]

Defensive programming check: If the Arrow vector count does not match the BigQuery schema field count, a cryptic IndexOutOfBoundsException will be thrown. Adding an explicit validation check with a clear error message improves robustness.

  static FieldValueList arrowRootToFieldValueList(
      VectorSchemaRoot root, int rowIndex, Schema schema) {
    if (root.getFieldVectors().size() != schema.getFields().size()) {
      throw new IllegalArgumentException(
          String.format(
              "Schema mismatch: Arrow vector count (%d) does not match BigQuery schema field count (%d)",
              root.getFieldVectors().size(), schema.getFields().size()));
    }
    List<FieldValue> fieldValues = new ArrayList<>();
    for (int colIndex = 0; colIndex < root.getFieldVectors().size(); colIndex++) {

[gemini-code-assist]

Defensive programming check: If the Arrow struct size does not match the BigQuery subfields size, a cryptic IndexOutOfBoundsException will be thrown. Adding an explicit validation check with a clear error message improves robustness.

    // Handle RECORD/STRUCT fields
    if (bqField.getType() == LegacySQLTypeName.RECORD) {
      StructVector structVector = (StructVector) vector;
      if (structVector.size() != bqField.getSubFields().size()) {
        throw new IllegalArgumentException(
            String.format(
                "Schema mismatch for field '%s': Arrow struct size (%d) does not match BigQuery subfields size (%d)",
                bqField.getName(), structVector.size(), bqField.getSubFields().size()));
      }
      List<FieldValue> elements = new ArrayList<>(structVector.size());
      for (int colIndex = 0; colIndex < structVector.size(); colIndex++) {

[gemini-code-assist]

Performance and Correctness: Calling vector.getObject(rowIndex) is highly inefficient as it allocates objects and can throw a ClassCastException if the returned object is a LocalDateTime or Instant instead of a Long (depending on the Arrow version and timezone settings). Casting the vector to TimeStampVector and calling get(rowIndex) avoids object allocation and is completely safe. Additionally, handling different Arrow time units (seconds, milliseconds, microseconds, nanoseconds) ensures robustness.

    } else if (bqField.getType() == LegacySQLTypeName.TIMESTAMP) {
      // Arrow timestamps are long values representing epoch micro/milli/nano seconds.
      // Standard BigQuery JSON returns timestamps as string of epoch seconds with micro precision
      // (e.g. "1408452095.220000").
      org.apache.arrow.vector.TimeStampVector tsVector = (org.apache.arrow.vector.TimeStampVector) vector;
      long rawVal = tsVector.get(rowIndex);
      ArrowType.Timestamp tsType = (ArrowType.Timestamp) vector.getField().getType();
      long micros;
      switch (tsType.getUnit()) {
        case SECOND:
          micros = rawVal * 1000000L;
          break;
        case MILLISECOND:
          micros = rawVal * 1000L;
          break;
        case MICROSECOND:
          micros = rawVal;
          break;
        case NANOSECOND:
          micros = rawVal / 1000L;
          break;
        default:
          micros = rawVal;
      }
      // Convert to seconds with 6 decimal places of precision
      stringVal = String.format(Locale.US, "%.6f", micros / 1000000.0);
    } else {