feat(bigquery): add ArrowDeserializer helper utility - #13943
feat(bigquery): add ArrowDeserializer helper utility#13943jinseopkim0 wants to merge 14 commits into
Conversation
There was a problem hiding this comment.
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.
e149afc to
796b77f
Compare
|
@gemini-code-assist review |
There was a problem hiding this comment.
Code Review
This pull request introduces comprehensive documentation (including developer, user, and storage API guides) and adds support for Apache Arrow serialization and deserialization in the BigQuery client and JDBC driver. It also introduces a centralized type registry and temporal utilities for optimized type coercion. The review feedback highlights several critical improvements: optimizing memory allocation and preventing resource leaks in ArrowDeserializer by passing the BufferAllocator and using proper try-finally blocks; avoiding precision loss in timestamp formatting by using integer division; ensuring that the new Arrow configuration options are actually transmitted to the BigQuery backend by serializing them in QueryJobConfiguration.toPb(); and adhering to timezone handling rules by using legacy Calendar manipulation and LocalTime.ofNanoOfDay for java.sql.Time conversions.
|
@gemini-code-assist review |
There was a problem hiding this comment.
Code Review
This pull request introduces Apache Arrow serialization options and deserialization support to the BigQuery client library, alongside extensive documentation and optimized type registry utilities for the JDBC driver. The review feedback highlights critical issues where the new Arrow configuration fields in QueryJobConfiguration are not serialized or deserialized in toPb() and fromPb(), rendering them ineffective. Additionally, the reviewer recommends reusing a single BufferAllocator in ArrowDeserializer to avoid high allocation overhead, ensuring exception-safe LIFO cleanup of FieldVector instances to prevent memory leaks, and using a timezone-aware manual conversion when converting java.sql.Time to java.time.LocalTime to preserve millisecond precision.
9056c4c to
9ddd9d2
Compare
9ddd9d2 to
712b972
Compare
f139009 to
dd8bc5a
Compare
dd8bc5a to
a2486de
Compare
…adArrowRows streaming and pagination
| List<FieldVector> vectors = ArrowPojoUtils.createVectors(arrowSchema, allocator); | ||
| try { | ||
| return new VectorSchemaRoot(vectors); | ||
| } catch (Throwable t) { |
There was a problem hiding this comment.
IIUC, this may have some funky behavior with native memory. Is that why we need to catch Throwable here?
There was a problem hiding this comment.
Yes, this is to prevent leaks.
| * @throws IOException if deserialization of the Arrow schema fails | ||
| */ | ||
| static Object deserializeSchema(byte[] schemaBytes) throws IOException { | ||
| return MessageSerializer.deserializeSchema( |
There was a problem hiding this comment.
Do we need to close ReadChannel afterwards? Or do we expect this to be the caller's responsibility?
There was a problem hiding this comment.
Thanks for the questions. Closing ReadChannel isn't strictly necessary here because ByteArrayReadableSeekableByteChannel wraps a heap byte[] in memory. However, as a best practice, I've updated the code to wrap it in try-with-resources inside deserializeSchema so callers don't need to worry about channel lifecycle.
| com.google.cloud.bigquery.storage.v1.ArrowRecordBatch batch = | ||
| response.getArrowRecordBatch(); |
There was a problem hiding this comment.
qq, I'm not entirely sure how this works. Is the size of ArrowRecordBatch and rowBatch always equal? What happens if there is a difference in say pageSize and ArrowRecordBatch?
There was a problem hiding this comment.
Thanks for the questions.
Is the size of ArrowRecordBatch and rowBatch always equal?
No, they are not always equal.
ArrowRecordBatchis the server's streaming chunk size (typically 1k–4k rows per gRPC response).rowBatchis the client's requested page container (pageSize, default 10k rows).
What happens if there is a difference in say pageSize and ArrowRecordBatch?
loadArrowRows iterates through incoming responses and accumulates rows into rowBatch. If a batch fills the remaining capacity of rowBatch (or hits maxResults), it stops mid-batch, marks hasMore = true, and returns the page. The next page fetch will resume reading from the updated offset (totalRowsReturned + rowBatch.size())
We have unit tests in ArrowDeserializerTest covering both multi-batch accumulation and mid-batch stoppage (testLoadArrowRows_multiBatchStream, testLoadArrowRows_respectsPageSize, and testLoadArrowRows_unconsumedBatchRowsSignalHasMore).
| } | ||
| builder = | ||
| com.google.cloud.bigquery.Field.newBuilder( | ||
| name, LegacySQLTypeName.RECORD, FieldList.of(subFields)); |
There was a problem hiding this comment.
Any reason in particular for LegacySQLTypeName instead of StandardSQLTypeName?
There was a problem hiding this comment.
Thanks for the question. In the BigQuery Java Veneer library, the Field class has historically used LegacySQLTypeName for all table and query result schema definitions (Field.getType() returns LegacySQLTypeName, and Field.newBuilder(...) requires it).
| } | ||
| builder.setMode(Mode.REPEATED); | ||
| } else { | ||
| if (!arrowField.getChildren().isEmpty()) { |
There was a problem hiding this comment.
qq, what are the other types where the are child nodes if it's not a list?
There was a problem hiding this comment.
Thanks for the question. In Apache Arrow, the non-list type with child nodes is ArrowType.Struct (representing a BigQuery RECORD / STRUCT). Per the Apache Arrow Columnar Format Specification (Struct Layout) and Arrow Java's Field.java#getChildren(), struct fields store their nested sub-fields inside arrowField.getChildren().
…ycle and schema resolution
Stacked PR 2 of 3: Adds the ArrowDeserializer class which handles decoding serialized Arrow schemas and record batches into standard FieldValueList rows.