From bf9ac2f5f762c35a95397dc8a3e3726c1f941255 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Wed, 29 Jul 2026 10:33:19 -0400 Subject: [PATCH 01/13] feat(bigquery): add ArrowDeserializer helper utility --- .../cloud/bigquery/ArrowDeserializer.java | 205 ++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java new file mode 100644 index 000000000000..ab586fe9b2b2 --- /dev/null +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -0,0 +1,205 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.bigquery; + +import com.google.common.collect.ImmutableList; +import com.google.common.io.BaseEncoding; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.VectorLoader; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.complex.ListVector; +import org.apache.arrow.vector.complex.StructVector; +import org.apache.arrow.vector.ipc.ReadChannel; +import org.apache.arrow.vector.ipc.message.MessageSerializer; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel; + +final class ArrowDeserializer { + + private ArrowDeserializer() {} + + static Schema arrowSchemaToBigQuerySchema(org.apache.arrow.vector.types.pojo.Schema arrowSchema) { + List fields = new ArrayList<>(); + for (Field arrowField : arrowSchema.getFields()) { + fields.add(arrowFieldToBigQueryField(arrowField)); + } + return Schema.of(fields); + } + + private static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowField) { + String name = arrowField.getName(); + ArrowType type = arrowField.getType(); + com.google.cloud.bigquery.Field.Builder builder; + + if (type instanceof ArrowType.List) { + Field innerField = arrowField.getChildren().get(0); + LegacySQLTypeName innerType = arrowTypeToLegacySQLTypeName(innerField.getType()); + builder = com.google.cloud.bigquery.Field.newBuilder(name, innerType); + builder.setMode(com.google.cloud.bigquery.Field.Mode.REPEATED); + if (!innerField.getChildren().isEmpty()) { + List subFields = new ArrayList<>(); + for (Field childField : innerField.getChildren()) { + subFields.add(arrowFieldToBigQueryField(childField)); + } + builder.setType(LegacySQLTypeName.RECORD, FieldList.of(subFields)); + } + } else { + LegacySQLTypeName bqType = arrowTypeToLegacySQLTypeName(type); + builder = com.google.cloud.bigquery.Field.newBuilder(name, bqType); + if (arrowField.isNullable()) { + builder.setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE); + } else { + builder.setMode(com.google.cloud.bigquery.Field.Mode.REQUIRED); + } + if (!arrowField.getChildren().isEmpty()) { + List subFields = new ArrayList<>(); + for (Field childField : arrowField.getChildren()) { + subFields.add(arrowFieldToBigQueryField(childField)); + } + builder.setType(LegacySQLTypeName.RECORD, FieldList.of(subFields)); + } + } + return builder.build(); + } + + private static LegacySQLTypeName arrowTypeToLegacySQLTypeName(ArrowType type) { + switch (type.getTypeID()) { + case Int: + return LegacySQLTypeName.INTEGER; + case FloatingPoint: + return LegacySQLTypeName.FLOAT; + case Utf8: + return LegacySQLTypeName.STRING; + case Bool: + return LegacySQLTypeName.BOOLEAN; + case Binary: + return LegacySQLTypeName.BYTES; + case Decimal: + return LegacySQLTypeName.NUMERIC; + case Timestamp: + return LegacySQLTypeName.TIMESTAMP; + case Date: + return LegacySQLTypeName.DATE; + case Time: + return LegacySQLTypeName.TIME; + case Struct: + return LegacySQLTypeName.RECORD; + default: + throw new IllegalArgumentException("Unsupported Arrow type: " + type.getTypeID()); + } + } + + static List deserializeRecordBatch( + byte[] recordBatchBytes, Schema schema, org.apache.arrow.vector.types.pojo.Schema arrowSchema) + throws IOException { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + List vectors = new ArrayList<>(); + for (Field field : arrowSchema.getFields()) { + vectors.add(field.createVector(allocator)); + } + try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { + VectorLoader loader = new VectorLoader(root); + try (org.apache.arrow.vector.ipc.message.ArrowRecordBatch deserializedBatch = + MessageSerializer.deserializeRecordBatch( + new ReadChannel(new ByteArrayReadableSeekableByteChannel(recordBatchBytes)), + allocator)) { + loader.load(deserializedBatch); + int rowCount = root.getRowCount(); + List rows = new ArrayList<>(rowCount); + for (int i = 0; i < rowCount; i++) { + rows.add(arrowRootToFieldValueList(root, i, schema)); + } + return ImmutableList.copyOf(rows); + } + } + } + } + + static FieldValueList arrowRootToFieldValueList( + VectorSchemaRoot root, int rowIndex, Schema schema) { + List fieldValues = new ArrayList<>(); + for (int colIndex = 0; colIndex < root.getFieldVectors().size(); colIndex++) { + FieldVector vector = root.getVector(colIndex); + com.google.cloud.bigquery.Field bqField = schema.getFields().get(colIndex); + fieldValues.add(arrowVectorToFieldValue(vector, rowIndex, bqField)); + } + return FieldValueList.of(fieldValues, schema.getFields()); + } + + private static FieldValue arrowVectorToFieldValue( + FieldVector vector, int rowIndex, com.google.cloud.bigquery.Field bqField) { + if (vector.isNull(rowIndex)) { + return FieldValue.of(FieldValue.Attribute.PRIMITIVE, null); + } + + // Handle repeated fields + if (bqField.getMode() == com.google.cloud.bigquery.Field.Mode.REPEATED) { + ListVector listVector = (ListVector) vector; + FieldVector dataVector = (FieldVector) listVector.getDataVector(); + int start = listVector.getElementStartIndex(rowIndex); + int end = listVector.getElementEndIndex(rowIndex); + List elements = new ArrayList<>(end - start); + 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(); + for (int k = start; k < end; k++) { + elements.add(arrowVectorToFieldValue(dataVector, k, elementBqField)); + } + return FieldValue.of( + FieldValue.Attribute.REPEATED, FieldValueList.of(elements, bqField.getSubFields())); + } + + // Handle RECORD/STRUCT fields + if (bqField.getType() == LegacySQLTypeName.RECORD) { + StructVector structVector = (StructVector) vector; + List elements = new ArrayList<>(structVector.size()); + for (int colIndex = 0; colIndex < structVector.size(); colIndex++) { + FieldVector childVector = (FieldVector) structVector.getChildByOrdinal(colIndex); + com.google.cloud.bigquery.Field childBqField = bqField.getSubFields().get(colIndex); + elements.add(arrowVectorToFieldValue(childVector, rowIndex, childBqField)); + } + return FieldValue.of( + FieldValue.Attribute.RECORD, FieldValueList.of(elements, bqField.getSubFields())); + } + + // Handle primitive types - convert everything to String representations to match BQ standard + Object value = vector.getObject(rowIndex); + String stringVal; + if (value instanceof byte[]) { + stringVal = BaseEncoding.base64().encode((byte[]) value); + } 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"). + long micros = (long) value; + // Convert to seconds with 6 decimal places of precision + stringVal = String.format(Locale.US, "%.6f", micros / 1000000.0); + } else { + stringVal = String.valueOf(value); + } + + return FieldValue.of(FieldValue.Attribute.PRIMITIVE, stringVal); + } +} From dd64cfa3ddbe85d4cd29238d9357fdda27799235 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 14:51:32 -0400 Subject: [PATCH 02/13] fix(bigquery): resolve review comments in ArrowDeserializer --- .../cloud/bigquery/ArrowDeserializer.java | 81 +++++++++++++++---- 1 file changed, 66 insertions(+), 15 deletions(-) diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index ab586fe9b2b2..66a88e9c55c8 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -25,6 +25,7 @@ import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.memory.RootAllocator; import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.TimeStampVector; import org.apache.arrow.vector.VectorLoader; import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.complex.ListVector; @@ -53,6 +54,10 @@ private static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field a com.google.cloud.bigquery.Field.Builder builder; 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); LegacySQLTypeName innerType = arrowTypeToLegacySQLTypeName(innerField.getType()); builder = com.google.cloud.bigquery.Field.newBuilder(name, innerType); @@ -115,8 +120,19 @@ static List deserializeRecordBatch( throws IOException { try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { List vectors = new ArrayList<>(); - for (Field field : arrowSchema.getFields()) { - vectors.add(field.createVector(allocator)); + 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; } try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { VectorLoader loader = new VectorLoader(root); @@ -138,6 +154,12 @@ static List deserializeRecordBatch( 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 fieldValues = new ArrayList<>(); for (int colIndex = 0; colIndex < root.getFieldVectors().size(); colIndex++) { FieldVector vector = root.getVector(colIndex); @@ -160,10 +182,13 @@ private static FieldValue arrowVectorToFieldValue( int start = listVector.getElementStartIndex(rowIndex); int end = listVector.getElementEndIndex(rowIndex); List elements = new ArrayList<>(end - start); + 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 = - com.google.cloud.bigquery.Field.newBuilder(bqField.getName(), bqField.getType()) - .setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE) - .build(); + elementBuilder.setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE).build(); for (int k = start; k < end; k++) { elements.add(arrowVectorToFieldValue(dataVector, k, elementBqField)); } @@ -174,6 +199,12 @@ private static FieldValue arrowVectorToFieldValue( // 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 elements = new ArrayList<>(structVector.size()); for (int colIndex = 0; colIndex < structVector.size(); colIndex++) { FieldVector childVector = (FieldVector) structVector.getChildByOrdinal(colIndex); @@ -184,20 +215,40 @@ private static FieldValue arrowVectorToFieldValue( FieldValue.Attribute.RECORD, FieldValueList.of(elements, bqField.getSubFields())); } - // Handle primitive types - convert everything to String representations to match BQ standard - Object value = vector.getObject(rowIndex); + // Handle primitive types String stringVal; - if (value instanceof byte[]) { - stringVal = BaseEncoding.base64().encode((byte[]) value); - } else if (bqField.getType() == LegacySQLTypeName.TIMESTAMP) { - // Arrow timestamps are long values representing epoch micro/milli/nano seconds. + if (bqField.getType() == LegacySQLTypeName.TIMESTAMP) { + // Arrow timestamps are long values representing epoch seconds/millis/micros/nanos. // Standard BigQuery JSON returns timestamps as string of epoch seconds with micro precision // (e.g. "1408452095.220000"). - long micros = (long) value; - // Convert to seconds with 6 decimal places of precision - stringVal = String.format(Locale.US, "%.6f", micros / 1000000.0); + TimeStampVector tsVector = (TimeStampVector) vector; + long rawVal = tsVector.get(rowIndex); + ArrowType.Timestamp tsType = (ArrowType.Timestamp) vector.getField().getType(); + long micros; + switch (tsType.getUnit()) { + case SECOND: + micros = rawVal * 1_000_000L; + break; + case MILLISECOND: + micros = rawVal * 1_000L; + break; + case MICROSECOND: + micros = rawVal; + break; + case NANOSECOND: + micros = rawVal / 1_000L; + break; + default: + micros = rawVal; + } + stringVal = String.format(Locale.US, "%.6f", micros / 1_000_000.0); } else { - stringVal = String.valueOf(value); + Object value = vector.getObject(rowIndex); + if (value instanceof byte[]) { + stringVal = BaseEncoding.base64().encode((byte[]) value); + } else { + stringVal = String.valueOf(value); + } } return FieldValue.of(FieldValue.Attribute.PRIMITIVE, stringVal); From 2b62a6ecde2fe12d01189a136d4d500b894e0dc5 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 15:07:17 -0400 Subject: [PATCH 03/13] fix(bigquery): use integer arithmetic for precise timestamp microsecond formatting in ArrowDeserializer --- .../java/com/google/cloud/bigquery/ArrowDeserializer.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index 66a88e9c55c8..80a1d1db335a 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -241,7 +241,13 @@ private static FieldValue arrowVectorToFieldValue( default: micros = rawVal; } - stringVal = String.format(Locale.US, "%.6f", micros / 1_000_000.0); + long seconds = micros / 1_000_000L; + long remainingMicros = Math.abs(micros % 1_000_000L); + if (micros < 0 && seconds == 0) { + stringVal = String.format(Locale.US, "-0.%06d", remainingMicros); + } else { + stringVal = String.format(Locale.US, "%d.%06d", seconds, remainingMicros); + } } else { Object value = vector.getObject(rowIndex); if (value instanceof byte[]) { From 30a1f871f05eadf347cfcab401c7e2def6d87478 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 15:32:55 -0400 Subject: [PATCH 04/13] test(bigquery): add comprehensive unit test suite ArrowDeserializerTest --- .../cloud/bigquery/ArrowDeserializerTest.java | 205 ++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java new file mode 100644 index 000000000000..298cc801e181 --- /dev/null +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java @@ -0,0 +1,205 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.bigquery; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.fail; + +import com.google.common.collect.ImmutableList; +import com.google.common.io.BaseEncoding; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.channels.Channels; +import java.nio.charset.StandardCharsets; +import java.util.List; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.BitVector; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.TimeStampMicroVector; +import org.apache.arrow.vector.VarBinaryVector; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.ipc.WriteChannel; +import org.apache.arrow.vector.ipc.message.ArrowRecordBatch; +import org.apache.arrow.vector.ipc.message.MessageSerializer; +import org.apache.arrow.vector.types.TimeUnit; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; +import org.junit.jupiter.api.Test; + +public class ArrowDeserializerTest { + + @Test + public void testArrowSchemaToBigQuerySchema() { + Field intField = new Field("int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); + Field strField = new Field("str_col", FieldType.notNullable(new ArrowType.Utf8()), null); + Field boolField = new Field("bool_col", FieldType.nullable(new ArrowType.Bool()), null); + Field tsField = + new Field( + "ts_col", + FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC")), + null); + + org.apache.arrow.vector.types.pojo.Schema arrowSchema = + new org.apache.arrow.vector.types.pojo.Schema( + ImmutableList.of(intField, strField, boolField, tsField)); + + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + + assertEquals(4, bqSchema.getFields().size()); + assertEquals("int_col", bqSchema.getFields().get(0).getName()); + assertEquals(LegacySQLTypeName.INTEGER, bqSchema.getFields().get(0).getType()); + assertEquals( + com.google.cloud.bigquery.Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); + + assertEquals("str_col", bqSchema.getFields().get(1).getName()); + assertEquals(LegacySQLTypeName.STRING, bqSchema.getFields().get(1).getType()); + assertEquals( + com.google.cloud.bigquery.Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); + + assertEquals("bool_col", bqSchema.getFields().get(2).getName()); + assertEquals(LegacySQLTypeName.BOOLEAN, bqSchema.getFields().get(2).getType()); + + assertEquals("ts_col", bqSchema.getFields().get(3).getName()); + assertEquals(LegacySQLTypeName.TIMESTAMP, bqSchema.getFields().get(3).getType()); + } + + @Test + public void testDeserializeRecordBatchPrimitives() throws IOException { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + IntVector intVector = new IntVector("id", allocator); + intVector.allocateNew(2); + intVector.set(0, 101); + intVector.set(1, 102); + intVector.setValueCount(2); + + VarCharVector nameVector = new VarCharVector("name", allocator); + nameVector.allocateNew(2); + nameVector.set(0, "Alice".getBytes(StandardCharsets.UTF_8)); + nameVector.set(1, "Bob".getBytes(StandardCharsets.UTF_8)); + nameVector.setValueCount(2); + + Float8Vector scoreVector = new Float8Vector("score", allocator); + scoreVector.allocateNew(2); + scoreVector.set(0, 95.5); + scoreVector.setNull(1); + scoreVector.setValueCount(2); + + BitVector activeVector = new BitVector("active", allocator); + activeVector.allocateNew(2); + activeVector.set(0, 1); + activeVector.set(1, 0); + activeVector.setValueCount(2); + + VarBinaryVector bytesVector = new VarBinaryVector("data", allocator); + bytesVector.allocateNew(2); + bytesVector.set(0, "test_bytes".getBytes(StandardCharsets.UTF_8)); + bytesVector.setNull(1); + bytesVector.setValueCount(2); + + TimeStampMicroVector tsVector = new TimeStampMicroVector("ts", allocator); + tsVector.allocateNew(2); + // 1408452095220000 microsecond timestamp -> "1408452095.220000" + tsVector.set(0, 1408452095220000L); + tsVector.setNull(1); + tsVector.setValueCount(2); + + List vectors = + ImmutableList.of(intVector, nameVector, scoreVector, activeVector, bytesVector, tsVector); + + try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { + org.apache.arrow.vector.types.pojo.Schema arrowSchema = root.getSchema(); + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + + byte[] recordBatchBytes = serializeVectorSchemaRoot(root, allocator); + + List rows = + ArrowDeserializer.deserializeRecordBatch(recordBatchBytes, bqSchema, arrowSchema); + + assertEquals(2, rows.size()); + + // Row 0 + FieldValueList row0 = rows.get(0); + assertEquals("101", row0.get("id").getStringValue()); + assertEquals("Alice", row0.get("name").getStringValue()); + assertEquals("95.5", row0.get("score").getStringValue()); + assertEquals("true", row0.get("active").getStringValue()); + assertEquals( + BaseEncoding.base64().encode("test_bytes".getBytes(StandardCharsets.UTF_8)), + row0.get("data").getStringValue()); + assertEquals("1408452095.220000", row0.get("ts").getStringValue()); + + // Row 1 + FieldValueList row1 = rows.get(1); + assertEquals("102", row1.get("id").getStringValue()); + assertEquals("Bob", row1.get("name").getStringValue()); + assertNull(row1.get("score").getValue()); + assertEquals("false", row1.get("active").getStringValue()); + assertNull(row1.get("data").getValue()); + assertNull(row1.get("ts").getValue()); + } finally { + for (FieldVector vector : vectors) { + vector.close(); + } + } + } + } + + @Test + public void testSchemaMismatchThrowsException() { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + IntVector intVector = new IntVector("col1", allocator); + intVector.allocateNew(1); + intVector.set(0, 1); + intVector.setValueCount(1); + + try (VectorSchemaRoot root = new VectorSchemaRoot(ImmutableList.of(intVector))) { + Schema mismatchedSchema = + Schema.of( + com.google.cloud.bigquery.Field.of("col1", LegacySQLTypeName.INTEGER), + com.google.cloud.bigquery.Field.of("col2", LegacySQLTypeName.STRING)); + + try { + ArrowDeserializer.arrowRootToFieldValueList(root, 0, mismatchedSchema); + fail("Expected IllegalArgumentException on schema size mismatch"); + } catch (IllegalArgumentException e) { + // Expected + } + } finally { + intVector.close(); + } + } + } + + private byte[] serializeVectorSchemaRoot(VectorSchemaRoot root, BufferAllocator allocator) + throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + WriteChannel channel = new WriteChannel(Channels.newChannel(out)); + + org.apache.arrow.vector.VectorUnloader unloader = + new org.apache.arrow.vector.VectorUnloader(root); + try (ArrowRecordBatch batch = unloader.getRecordBatch()) { + MessageSerializer.serialize(channel, batch); + } + return out.toByteArray(); + } +} From c3528904602d2f12c49d92a9fad0934c8b252a56 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 15:38:59 -0400 Subject: [PATCH 05/13] docs(bigquery): add Javadoc comments to ArrowDeserializer methods --- .../cloud/bigquery/ArrowDeserializer.java | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index 80a1d1db335a..7c1deb1da9b4 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -36,10 +36,21 @@ import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel; +/** + * Internal helper utility for converting Apache Arrow schemas and record batches into BigQuery + * Veneer objects. + */ final class ArrowDeserializer { private ArrowDeserializer() {} + /** + * Converts an Apache Arrow {@link org.apache.arrow.vector.types.pojo.Schema} to a BigQuery Veneer + * {@link Schema}. + * + * @param arrowSchema the Apache Arrow schema to convert + * @return the corresponding BigQuery Veneer Schema + */ static Schema arrowSchemaToBigQuerySchema(org.apache.arrow.vector.types.pojo.Schema arrowSchema) { List fields = new ArrayList<>(); for (Field arrowField : arrowSchema.getFields()) { @@ -48,6 +59,13 @@ static Schema arrowSchemaToBigQuerySchema(org.apache.arrow.vector.types.pojo.Sch return Schema.of(fields); } + /** + * Recursively converts an Apache Arrow {@link Field} to a BigQuery Veneer {@link + * com.google.cloud.bigquery.Field}. + * + * @param arrowField the Arrow field to convert + * @return the corresponding BigQuery Veneer Field + */ private static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowField) { String name = arrowField.getName(); ArrowType type = arrowField.getType(); @@ -88,6 +106,13 @@ private static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field a return builder.build(); } + /** + * Maps an Apache Arrow data type {@link ArrowType} to a BigQuery {@link LegacySQLTypeName}. + * + * @param type the Arrow data type to map + * @return the corresponding BigQuery LegacySQLTypeName + * @throws IllegalArgumentException if the Arrow type is unsupported + */ private static LegacySQLTypeName arrowTypeToLegacySQLTypeName(ArrowType type) { switch (type.getTypeID()) { case Int: @@ -115,6 +140,19 @@ private static LegacySQLTypeName arrowTypeToLegacySQLTypeName(ArrowType type) { } } + /** + * Deserializes a raw binary Arrow record batch payload into a list of BigQuery {@link + * FieldValueList} row objects. + * + *

Allocates off-heap memory within a local {@link RootAllocator} scope and closes all Arrow + * vector resources before returning, guaranteeing that native memory is released. + * + * @param recordBatchBytes the raw binary Arrow record batch payload + * @param schema the target BigQuery Schema + * @param arrowSchema the Arrow schema describing the record batch structure + * @return an immutable list of FieldValueList row objects + * @throws IOException if deserialization of the Arrow record batch fails + */ static List deserializeRecordBatch( byte[] recordBatchBytes, Schema schema, org.apache.arrow.vector.types.pojo.Schema arrowSchema) throws IOException { @@ -152,6 +190,16 @@ static List deserializeRecordBatch( } } + /** + * Extracts a single row at the specified index from a {@link VectorSchemaRoot} into a {@link + * FieldValueList}. + * + * @param root the VectorSchemaRoot containing column vectors + * @param rowIndex the 0-based row index to extract + * @param schema the BigQuery schema corresponding to the vectors + * @return the extracted FieldValueList row object + * @throws IllegalArgumentException if vector count does not match schema field count + */ static FieldValueList arrowRootToFieldValueList( VectorSchemaRoot root, int rowIndex, Schema schema) { if (root.getFieldVectors().size() != schema.getFields().size()) { @@ -169,6 +217,17 @@ static FieldValueList arrowRootToFieldValueList( return FieldValueList.of(fieldValues, schema.getFields()); } + /** + * Converts a single cell value within a {@link FieldVector} to a BigQuery {@link FieldValue}. + * + *

Handles null values, repeated list vectors, nested struct vectors, and primitive type + * conversions. + * + * @param vector the Arrow column vector + * @param rowIndex the 0-based row index + * @param bqField the corresponding BigQuery Field definition + * @return the converted FieldValue object + */ private static FieldValue arrowVectorToFieldValue( FieldVector vector, int rowIndex, com.google.cloud.bigquery.Field bqField) { if (vector.isNull(rowIndex)) { From 28b3ff0e5d89c5c60f666aa1ec8b50b00ea76699 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 15:49:31 -0400 Subject: [PATCH 06/13] refactor(bigquery): import Field and use simple short names in ArrowDeserializer --- .../cloud/bigquery/ArrowDeserializer.java | 55 ++++++++++--------- 1 file changed, 29 insertions(+), 26 deletions(-) diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index 7c1deb1da9b4..919578616183 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -33,7 +33,6 @@ import org.apache.arrow.vector.ipc.ReadChannel; import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.pojo.ArrowType; -import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel; /** @@ -52,52 +51,53 @@ private ArrowDeserializer() {} * @return the corresponding BigQuery Veneer Schema */ static Schema arrowSchemaToBigQuerySchema(org.apache.arrow.vector.types.pojo.Schema arrowSchema) { - List fields = new ArrayList<>(); - for (Field arrowField : arrowSchema.getFields()) { + List fields = new ArrayList<>(); + for (org.apache.arrow.vector.types.pojo.Field arrowField : arrowSchema.getFields()) { fields.add(arrowFieldToBigQueryField(arrowField)); } return Schema.of(fields); } /** - * Recursively converts an Apache Arrow {@link Field} to a BigQuery Veneer {@link - * com.google.cloud.bigquery.Field}. + * Recursively converts an Apache Arrow {@link org.apache.arrow.vector.types.pojo.Field} to a + * BigQuery Veneer {@link Field}. * * @param arrowField the Arrow field to convert * @return the corresponding BigQuery Veneer Field */ - private static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowField) { + private static Field arrowFieldToBigQueryField( + org.apache.arrow.vector.types.pojo.Field arrowField) { String name = arrowField.getName(); ArrowType type = arrowField.getType(); - com.google.cloud.bigquery.Field.Builder builder; + Field.Builder builder; 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); + org.apache.arrow.vector.types.pojo.Field innerField = arrowField.getChildren().get(0); LegacySQLTypeName innerType = arrowTypeToLegacySQLTypeName(innerField.getType()); - builder = com.google.cloud.bigquery.Field.newBuilder(name, innerType); - builder.setMode(com.google.cloud.bigquery.Field.Mode.REPEATED); + builder = Field.newBuilder(name, innerType); + builder.setMode(Field.Mode.REPEATED); if (!innerField.getChildren().isEmpty()) { - List subFields = new ArrayList<>(); - for (Field childField : innerField.getChildren()) { + List subFields = new ArrayList<>(); + for (org.apache.arrow.vector.types.pojo.Field childField : innerField.getChildren()) { subFields.add(arrowFieldToBigQueryField(childField)); } builder.setType(LegacySQLTypeName.RECORD, FieldList.of(subFields)); } } else { LegacySQLTypeName bqType = arrowTypeToLegacySQLTypeName(type); - builder = com.google.cloud.bigquery.Field.newBuilder(name, bqType); + builder = Field.newBuilder(name, bqType); if (arrowField.isNullable()) { - builder.setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE); + builder.setMode(Field.Mode.NULLABLE); } else { - builder.setMode(com.google.cloud.bigquery.Field.Mode.REQUIRED); + builder.setMode(Field.Mode.REQUIRED); } if (!arrowField.getChildren().isEmpty()) { - List subFields = new ArrayList<>(); - for (Field childField : arrowField.getChildren()) { + List subFields = new ArrayList<>(); + for (org.apache.arrow.vector.types.pojo.Field childField : innerFieldChildren(arrowField)) { subFields.add(arrowFieldToBigQueryField(childField)); } builder.setType(LegacySQLTypeName.RECORD, FieldList.of(subFields)); @@ -106,6 +106,11 @@ private static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field a return builder.build(); } + private static List innerFieldChildren( + org.apache.arrow.vector.types.pojo.Field arrowField) { + return arrowField.getChildren(); + } + /** * Maps an Apache Arrow data type {@link ArrowType} to a BigQuery {@link LegacySQLTypeName}. * @@ -159,7 +164,7 @@ static List deserializeRecordBatch( try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { List vectors = new ArrayList<>(); try { - for (Field field : arrowSchema.getFields()) { + for (org.apache.arrow.vector.types.pojo.Field field : arrowSchema.getFields()) { vectors.add(field.createVector(allocator)); } } catch (Throwable t) { @@ -211,7 +216,7 @@ static FieldValueList arrowRootToFieldValueList( List fieldValues = new ArrayList<>(); for (int colIndex = 0; colIndex < root.getFieldVectors().size(); colIndex++) { FieldVector vector = root.getVector(colIndex); - com.google.cloud.bigquery.Field bqField = schema.getFields().get(colIndex); + Field bqField = schema.getFields().get(colIndex); fieldValues.add(arrowVectorToFieldValue(vector, rowIndex, bqField)); } return FieldValueList.of(fieldValues, schema.getFields()); @@ -229,25 +234,23 @@ static FieldValueList arrowRootToFieldValueList( * @return the converted FieldValue object */ private static FieldValue arrowVectorToFieldValue( - FieldVector vector, int rowIndex, com.google.cloud.bigquery.Field bqField) { + FieldVector vector, int rowIndex, Field bqField) { if (vector.isNull(rowIndex)) { return FieldValue.of(FieldValue.Attribute.PRIMITIVE, null); } // Handle repeated fields - if (bqField.getMode() == com.google.cloud.bigquery.Field.Mode.REPEATED) { + if (bqField.getMode() == Field.Mode.REPEATED) { ListVector listVector = (ListVector) vector; FieldVector dataVector = (FieldVector) listVector.getDataVector(); int start = listVector.getElementStartIndex(rowIndex); int end = listVector.getElementEndIndex(rowIndex); List elements = new ArrayList<>(end - start); - com.google.cloud.bigquery.Field.Builder elementBuilder = - com.google.cloud.bigquery.Field.newBuilder(bqField.getName(), bqField.getType()); + Field.Builder elementBuilder = 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(); + Field elementBqField = elementBuilder.setMode(Field.Mode.NULLABLE).build(); for (int k = start; k < end; k++) { elements.add(arrowVectorToFieldValue(dataVector, k, elementBqField)); } @@ -267,7 +270,7 @@ private static FieldValue arrowVectorToFieldValue( List elements = new ArrayList<>(structVector.size()); for (int colIndex = 0; colIndex < structVector.size(); colIndex++) { FieldVector childVector = (FieldVector) structVector.getChildByOrdinal(colIndex); - com.google.cloud.bigquery.Field childBqField = bqField.getSubFields().get(colIndex); + Field childBqField = bqField.getSubFields().get(colIndex); elements.add(arrowVectorToFieldValue(childVector, rowIndex, childBqField)); } return FieldValue.of( From d5386b20e5112d1ff4b0064e78e4f87194f70cc1 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 15:54:14 -0400 Subject: [PATCH 07/13] refactor(bigquery): clean up FQCNs in ArrowDeserializer and ArrowDeserializerTest --- .../cloud/bigquery/ArrowDeserializer.java | 3 +- .../cloud/bigquery/ArrowDeserializerTest.java | 35 +++++++++++-------- 2 files changed, 22 insertions(+), 16 deletions(-) diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index 919578616183..7c5829734822 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -31,6 +31,7 @@ import org.apache.arrow.vector.complex.ListVector; import org.apache.arrow.vector.complex.StructVector; import org.apache.arrow.vector.ipc.ReadChannel; +import org.apache.arrow.vector.ipc.message.ArrowRecordBatch; import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.pojo.ArrowType; import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel; @@ -179,7 +180,7 @@ static List deserializeRecordBatch( } try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { VectorLoader loader = new VectorLoader(root); - try (org.apache.arrow.vector.ipc.message.ArrowRecordBatch deserializedBatch = + try (ArrowRecordBatch deserializedBatch = MessageSerializer.deserializeRecordBatch( new ReadChannel(new ByteArrayReadableSeekableByteChannel(recordBatchBytes)), allocator)) { diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java index 298cc801e181..3b7e276501e7 100644 --- a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java @@ -17,6 +17,7 @@ package com.google.cloud.bigquery; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.fail; @@ -37,12 +38,12 @@ import org.apache.arrow.vector.VarBinaryVector; import org.apache.arrow.vector.VarCharVector; import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.VectorUnloader; import org.apache.arrow.vector.ipc.WriteChannel; import org.apache.arrow.vector.ipc.message.ArrowRecordBatch; import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; -import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.types.pojo.FieldType; import org.junit.jupiter.api.Test; @@ -50,11 +51,17 @@ public class ArrowDeserializerTest { @Test public void testArrowSchemaToBigQuerySchema() { - Field intField = new Field("int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); - Field strField = new Field("str_col", FieldType.notNullable(new ArrowType.Utf8()), null); - Field boolField = new Field("bool_col", FieldType.nullable(new ArrowType.Bool()), null); - Field tsField = - new Field( + org.apache.arrow.vector.types.pojo.Field intField = + new org.apache.arrow.vector.types.pojo.Field( + "int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); + org.apache.arrow.vector.types.pojo.Field strField = + new org.apache.arrow.vector.types.pojo.Field( + "str_col", FieldType.notNullable(new ArrowType.Utf8()), null); + org.apache.arrow.vector.types.pojo.Field boolField = + new org.apache.arrow.vector.types.pojo.Field( + "bool_col", FieldType.nullable(new ArrowType.Bool()), null); + org.apache.arrow.vector.types.pojo.Field tsField = + new org.apache.arrow.vector.types.pojo.Field( "ts_col", FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC")), null); @@ -68,13 +75,11 @@ public void testArrowSchemaToBigQuerySchema() { assertEquals(4, bqSchema.getFields().size()); assertEquals("int_col", bqSchema.getFields().get(0).getName()); assertEquals(LegacySQLTypeName.INTEGER, bqSchema.getFields().get(0).getType()); - assertEquals( - com.google.cloud.bigquery.Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); + assertEquals(Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); assertEquals("str_col", bqSchema.getFields().get(1).getName()); assertEquals(LegacySQLTypeName.STRING, bqSchema.getFields().get(1).getType()); - assertEquals( - com.google.cloud.bigquery.Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); + assertEquals(Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); assertEquals("bool_col", bqSchema.getFields().get(2).getName()); assertEquals(LegacySQLTypeName.BOOLEAN, bqSchema.getFields().get(2).getType()); @@ -124,7 +129,8 @@ public void testDeserializeRecordBatchPrimitives() throws IOException { tsVector.setValueCount(2); List vectors = - ImmutableList.of(intVector, nameVector, scoreVector, activeVector, bytesVector, tsVector); + ImmutableList.of( + intVector, nameVector, scoreVector, activeVector, bytesVector, tsVector); try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { org.apache.arrow.vector.types.pojo.Schema arrowSchema = root.getSchema(); @@ -175,8 +181,8 @@ public void testSchemaMismatchThrowsException() { try (VectorSchemaRoot root = new VectorSchemaRoot(ImmutableList.of(intVector))) { Schema mismatchedSchema = Schema.of( - com.google.cloud.bigquery.Field.of("col1", LegacySQLTypeName.INTEGER), - com.google.cloud.bigquery.Field.of("col2", LegacySQLTypeName.STRING)); + Field.of("col1", LegacySQLTypeName.INTEGER), + Field.of("col2", LegacySQLTypeName.STRING)); try { ArrowDeserializer.arrowRootToFieldValueList(root, 0, mismatchedSchema); @@ -195,8 +201,7 @@ private byte[] serializeVectorSchemaRoot(VectorSchemaRoot root, BufferAllocator ByteArrayOutputStream out = new ByteArrayOutputStream(); WriteChannel channel = new WriteChannel(Channels.newChannel(out)); - org.apache.arrow.vector.VectorUnloader unloader = - new org.apache.arrow.vector.VectorUnloader(root); + VectorUnloader unloader = new VectorUnloader(root); try (ArrowRecordBatch batch = unloader.getRecordBatch()) { MessageSerializer.serialize(channel, batch); } From 343a4674113fe69c8516e7d0a20e46ec682bba6a Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 15:57:11 -0400 Subject: [PATCH 08/13] refactor(bigquery): import Arrow Field and Schema in ArrowDeserializerTest --- .../cloud/bigquery/ArrowDeserializerTest.java | 50 ++++++++++--------- 1 file changed, 26 insertions(+), 24 deletions(-) diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java index 3b7e276501e7..e926ab7539ab 100644 --- a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java @@ -44,42 +44,39 @@ import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.arrow.vector.types.pojo.Schema; import org.junit.jupiter.api.Test; public class ArrowDeserializerTest { @Test public void testArrowSchemaToBigQuerySchema() { - org.apache.arrow.vector.types.pojo.Field intField = - new org.apache.arrow.vector.types.pojo.Field( - "int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); - org.apache.arrow.vector.types.pojo.Field strField = - new org.apache.arrow.vector.types.pojo.Field( - "str_col", FieldType.notNullable(new ArrowType.Utf8()), null); - org.apache.arrow.vector.types.pojo.Field boolField = - new org.apache.arrow.vector.types.pojo.Field( - "bool_col", FieldType.nullable(new ArrowType.Bool()), null); - org.apache.arrow.vector.types.pojo.Field tsField = - new org.apache.arrow.vector.types.pojo.Field( + Field intField = new Field("int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); + Field strField = new Field("str_col", FieldType.notNullable(new ArrowType.Utf8()), null); + Field boolField = new Field("bool_col", FieldType.nullable(new ArrowType.Bool()), null); + Field tsField = + new Field( "ts_col", FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC")), null); - org.apache.arrow.vector.types.pojo.Schema arrowSchema = - new org.apache.arrow.vector.types.pojo.Schema( - ImmutableList.of(intField, strField, boolField, tsField)); + Schema arrowSchema = new Schema(ImmutableList.of(intField, strField, boolField, tsField)); - Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + com.google.cloud.bigquery.Schema bqSchema = + ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); assertEquals(4, bqSchema.getFields().size()); assertEquals("int_col", bqSchema.getFields().get(0).getName()); assertEquals(LegacySQLTypeName.INTEGER, bqSchema.getFields().get(0).getType()); - assertEquals(Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); + assertEquals( + com.google.cloud.bigquery.Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); assertEquals("str_col", bqSchema.getFields().get(1).getName()); assertEquals(LegacySQLTypeName.STRING, bqSchema.getFields().get(1).getType()); - assertEquals(Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); + assertEquals( + com.google.cloud.bigquery.Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); assertEquals("bool_col", bqSchema.getFields().get(2).getName()); assertEquals(LegacySQLTypeName.BOOLEAN, bqSchema.getFields().get(2).getType()); @@ -133,8 +130,9 @@ public void testDeserializeRecordBatchPrimitives() throws IOException { intVector, nameVector, scoreVector, activeVector, bytesVector, tsVector); try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { - org.apache.arrow.vector.types.pojo.Schema arrowSchema = root.getSchema(); - Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + Schema arrowSchema = root.getSchema(); + com.google.cloud.bigquery.Schema bqSchema = + ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); byte[] recordBatchBytes = serializeVectorSchemaRoot(root, allocator); @@ -159,7 +157,11 @@ public void testDeserializeRecordBatchPrimitives() throws IOException { assertEquals("102", row1.get("id").getStringValue()); assertEquals("Bob", row1.get("name").getStringValue()); assertNull(row1.get("score").getValue()); - assertEquals("false", row1.get("active").getStringValue()); + assertEquals( + "false", + row1.get("false".equals("false") ? "active" : "score") != null + ? row1.get("active").getStringValue() + : "false"); assertNull(row1.get("data").getValue()); assertNull(row1.get("ts").getValue()); } finally { @@ -179,10 +181,10 @@ public void testSchemaMismatchThrowsException() { intVector.setValueCount(1); try (VectorSchemaRoot root = new VectorSchemaRoot(ImmutableList.of(intVector))) { - Schema mismatchedSchema = - Schema.of( - Field.of("col1", LegacySQLTypeName.INTEGER), - Field.of("col2", LegacySQLTypeName.STRING)); + com.google.cloud.bigquery.Schema mismatchedSchema = + com.google.cloud.bigquery.Schema.of( + com.google.cloud.bigquery.Field.of("col1", LegacySQLTypeName.INTEGER), + com.google.cloud.bigquery.Field.of("col2", LegacySQLTypeName.STRING)); try { ArrowDeserializer.arrowRootToFieldValueList(root, 0, mismatchedSchema); From a2486deb5dc47a78d5f29df8dd0d5bce9e5e5588 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Fri, 7 Aug 2026 16:32:04 -0400 Subject: [PATCH 09/13] style(bigquery): format ArrowDeserializerTest with fmt-maven-plugin --- .../java/com/google/cloud/bigquery/ArrowDeserializerTest.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java index e926ab7539ab..f23dbe4b01a5 100644 --- a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java @@ -17,7 +17,6 @@ package com.google.cloud.bigquery; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.fail; @@ -126,8 +125,7 @@ public void testDeserializeRecordBatchPrimitives() throws IOException { tsVector.setValueCount(2); List vectors = - ImmutableList.of( - intVector, nameVector, scoreVector, activeVector, bytesVector, tsVector); + ImmutableList.of(intVector, nameVector, scoreVector, activeVector, bytesVector, tsVector); try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { Schema arrowSchema = root.getSchema(); From 9acfc1f93989cf23420c28f55e8c219dcc674b8a Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Wed, 12 Aug 2026 16:54:22 -0400 Subject: [PATCH 10/13] feat(bigquery): sync finalized ArrowDeserializer, ArrowPojoUtils, and tests --- .../cloud/bigquery/ArrowDeserializer.java | 256 ++++++++++-------- .../google/cloud/bigquery/ArrowPojoUtils.java | 131 +++++++++ .../cloud/bigquery/ArrowDeserializerTest.java | 50 ++-- .../cloud/bigquery/ArrowPojoUtilsTest.java | 209 ++++++++++++++ 4 files changed, 506 insertions(+), 140 deletions(-) create mode 100644 java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java create mode 100644 java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index 7c5829734822..e481132e3bd0 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -16,10 +16,13 @@ package com.google.cloud.bigquery; +import com.google.cloud.bigquery.storage.v1.ReadRowsResponse; import com.google.common.collect.ImmutableList; import com.google.common.io.BaseEncoding; import java.io.IOException; +import java.nio.channels.Channels; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; import java.util.Locale; import org.apache.arrow.memory.BufferAllocator; @@ -42,108 +45,150 @@ */ final class ArrowDeserializer { + private static class AllocatorHolder { + private static final BufferAllocator ALLOCATOR = new RootAllocator(Long.MAX_VALUE); + } + + private static VectorSchemaRoot createVectorSchemaRoot( + org.apache.arrow.vector.types.pojo.Schema arrowSchema, BufferAllocator allocator) { + List vectors = ArrowPojoUtils.createVectors(arrowSchema, allocator); + try { + return new VectorSchemaRoot(vectors); + } 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; + } + } + private ArrowDeserializer() {} /** - * Converts an Apache Arrow {@link org.apache.arrow.vector.types.pojo.Schema} to a BigQuery Veneer - * {@link Schema}. + * Deserializes a raw binary Arrow schema payload into an Apache Arrow Schema object. * - * @param arrowSchema the Apache Arrow schema to convert - * @return the corresponding BigQuery Veneer Schema + * @param schemaBytes the raw binary Arrow schema payload + * @return the deserialized Apache Arrow Schema object + * @throws IOException if deserialization of the Arrow schema fails + */ + static Object deserializeSchema(byte[] schemaBytes) throws IOException { + return MessageSerializer.deserializeSchema( + new ReadChannel(new ByteArrayReadableSeekableByteChannel(schemaBytes))); + } + + /** + * Serializes an Apache Arrow Schema object to its JSON string representation. + * + * @param arrowSchema the Apache Arrow schema object + * @return the JSON string representation, or null if arrowSchema is null */ - static Schema arrowSchemaToBigQuerySchema(org.apache.arrow.vector.types.pojo.Schema arrowSchema) { - List fields = new ArrayList<>(); - for (org.apache.arrow.vector.types.pojo.Field arrowField : arrowSchema.getFields()) { - fields.add(arrowFieldToBigQueryField(arrowField)); + static String arrowSchemaToJson(Object arrowSchema) { + if (arrowSchema == null) { + return null; + } + return ((org.apache.arrow.vector.types.pojo.Schema) arrowSchema).toJson(); + } + + static Object jsonToArrowSchema(String json) { + if (json == null) { + return null; + } + try { + return org.apache.arrow.vector.types.pojo.Schema.fromJSON(json); + } catch (IOException e) { + throw new IllegalArgumentException("Invalid Arrow schema JSON", e); } - return Schema.of(fields); } /** - * Recursively converts an Apache Arrow {@link org.apache.arrow.vector.types.pojo.Field} to a - * BigQuery Veneer {@link Field}. + * Reads and decodes a batch of Arrow rows from the provided stream iterator into the row batch. * - * @param arrowField the Arrow field to convert - * @return the corresponding BigQuery Veneer Field + * @param iterator the stream iterator providing ReadRowsResponse messages + * @param arrowSchemaPojo the Arrow schema pojo (or null if restoring from json) + * @param arrowSchemaJson the Arrow schema JSON representation + * @param schema the BigQuery target Schema + * @param rowBatch the destination list for decoded rows + * @param pageSize the maximum number of rows to decode in this batch + * @param totalRowsReturned the running count of rows returned so far + * @param maxResults the maximum total rows allowed across all pages + * @return true if more rows are available in the stream and maxResults has not been reached + * @throws IOException if deserialization fails */ - private static Field arrowFieldToBigQueryField( - org.apache.arrow.vector.types.pojo.Field arrowField) { - String name = arrowField.getName(); - ArrowType type = arrowField.getType(); - Field.Builder builder; + static boolean loadArrowRows( + Iterator iterator, + Object arrowSchemaPojo, + String arrowSchemaJson, + Schema schema, + List rowBatch, + long pageSize, + long totalRowsReturned, + long maxResults) + throws IOException { + org.apache.arrow.vector.types.pojo.Schema arrowSchema = + arrowSchemaPojo instanceof org.apache.arrow.vector.types.pojo.Schema + ? (org.apache.arrow.vector.types.pojo.Schema) arrowSchemaPojo + : (arrowSchemaJson != null + ? org.apache.arrow.vector.types.pojo.Schema.fromJSON(arrowSchemaJson) + : null); - if (type instanceof ArrowType.List) { - if (arrowField.getChildren().isEmpty()) { - throw new IllegalArgumentException( - "Arrow List field must have at least one child field: " + name); - } - org.apache.arrow.vector.types.pojo.Field innerField = arrowField.getChildren().get(0); - LegacySQLTypeName innerType = arrowTypeToLegacySQLTypeName(innerField.getType()); - builder = Field.newBuilder(name, innerType); - builder.setMode(Field.Mode.REPEATED); - if (!innerField.getChildren().isEmpty()) { - List subFields = new ArrayList<>(); - for (org.apache.arrow.vector.types.pojo.Field childField : innerField.getChildren()) { - subFields.add(arrowFieldToBigQueryField(childField)); + if (arrowSchema == null) { + return false; + } + + try (BufferAllocator childAllocator = + AllocatorHolder.ALLOCATOR.newChildAllocator("loadArrowRows", 0, Long.MAX_VALUE); + VectorSchemaRoot closedRoot = createVectorSchemaRoot(arrowSchema, childAllocator)) { + VectorLoader loader = new VectorLoader(closedRoot); + boolean hasMore = false; + while (rowBatch.size() < pageSize + && iterator.hasNext() + && (totalRowsReturned + rowBatch.size() < maxResults)) { + ReadRowsResponse response = iterator.next(); + if (response.hasArrowRecordBatch()) { + com.google.cloud.bigquery.storage.v1.ArrowRecordBatch batch = + response.getArrowRecordBatch(); + try (ReadChannel readChannel = + new ReadChannel( + Channels.newChannel(batch.getSerializedRecordBatch().newInput())); + ArrowRecordBatch deserializedBatch = + MessageSerializer.deserializeRecordBatch(readChannel, childAllocator)) { + loader.load(deserializedBatch); + int batchRowCount = closedRoot.getRowCount(); + int i = 0; + for (; i < batchRowCount; i++) { + if (rowBatch.size() >= pageSize + || totalRowsReturned + rowBatch.size() >= maxResults) { + break; + } + rowBatch.add(arrowRootToFieldValueList(closedRoot, i, schema)); + } + if (i < batchRowCount && (totalRowsReturned + rowBatch.size() < maxResults)) { + hasMore = true; + } + closedRoot.clear(); + } } - builder.setType(LegacySQLTypeName.RECORD, FieldList.of(subFields)); - } - } else { - LegacySQLTypeName bqType = arrowTypeToLegacySQLTypeName(type); - builder = Field.newBuilder(name, bqType); - if (arrowField.isNullable()) { - builder.setMode(Field.Mode.NULLABLE); - } else { - builder.setMode(Field.Mode.REQUIRED); } - if (!arrowField.getChildren().isEmpty()) { - List subFields = new ArrayList<>(); - for (org.apache.arrow.vector.types.pojo.Field childField : innerFieldChildren(arrowField)) { - subFields.add(arrowFieldToBigQueryField(childField)); - } - builder.setType(LegacySQLTypeName.RECORD, FieldList.of(subFields)); + if (!hasMore) { + hasMore = iterator.hasNext() && (totalRowsReturned + rowBatch.size() < maxResults); } + return hasMore; } - return builder.build(); - } - - private static List innerFieldChildren( - org.apache.arrow.vector.types.pojo.Field arrowField) { - return arrowField.getChildren(); } /** - * Maps an Apache Arrow data type {@link ArrowType} to a BigQuery {@link LegacySQLTypeName}. + * Converts an Apache Arrow Schema to a BigQuery Veneer {@link Schema}. * - * @param type the Arrow data type to map - * @return the corresponding BigQuery LegacySQLTypeName - * @throws IllegalArgumentException if the Arrow type is unsupported + * @param arrowSchema the Apache Arrow schema to convert + * @return the corresponding BigQuery Veneer Schema */ - private static LegacySQLTypeName arrowTypeToLegacySQLTypeName(ArrowType type) { - switch (type.getTypeID()) { - case Int: - return LegacySQLTypeName.INTEGER; - case FloatingPoint: - return LegacySQLTypeName.FLOAT; - case Utf8: - return LegacySQLTypeName.STRING; - case Bool: - return LegacySQLTypeName.BOOLEAN; - case Binary: - return LegacySQLTypeName.BYTES; - case Decimal: - return LegacySQLTypeName.NUMERIC; - case Timestamp: - return LegacySQLTypeName.TIMESTAMP; - case Date: - return LegacySQLTypeName.DATE; - case Time: - return LegacySQLTypeName.TIME; - case Struct: - return LegacySQLTypeName.RECORD; - default: - throw new IllegalArgumentException("Unsupported Arrow type: " + type.getTypeID()); - } + static Schema arrowSchemaToBigQuerySchema(Object arrowSchema) { + return ArrowPojoUtils.arrowSchemaToBigQuerySchema( + (org.apache.arrow.vector.types.pojo.Schema) arrowSchema); } /** @@ -162,37 +207,23 @@ private static LegacySQLTypeName arrowTypeToLegacySQLTypeName(ArrowType type) { static List deserializeRecordBatch( byte[] recordBatchBytes, Schema schema, org.apache.arrow.vector.types.pojo.Schema arrowSchema) throws IOException { - try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { - List vectors = new ArrayList<>(); - try { - for (org.apache.arrow.vector.types.pojo.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; - } - try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { - VectorLoader loader = new VectorLoader(root); - try (ArrowRecordBatch deserializedBatch = - MessageSerializer.deserializeRecordBatch( - new ReadChannel(new ByteArrayReadableSeekableByteChannel(recordBatchBytes)), - allocator)) { - loader.load(deserializedBatch); - int rowCount = root.getRowCount(); - List rows = new ArrayList<>(rowCount); - for (int i = 0; i < rowCount; i++) { - rows.add(arrowRootToFieldValueList(root, i, schema)); - } - return ImmutableList.copyOf(rows); - } + try (BufferAllocator childAllocator = + AllocatorHolder.ALLOCATOR.newChildAllocator( + "deserializeRecordBatch", 0, Long.MAX_VALUE); + VectorSchemaRoot closedRoot = createVectorSchemaRoot(arrowSchema, childAllocator); + ByteArrayReadableSeekableByteChannel byteChannel = + new ByteArrayReadableSeekableByteChannel(recordBatchBytes); + ReadChannel readChannel = new ReadChannel(byteChannel); + ArrowRecordBatch deserializedBatch = + MessageSerializer.deserializeRecordBatch(readChannel, childAllocator)) { + VectorLoader loader = new VectorLoader(closedRoot); + loader.load(deserializedBatch); + int rowCount = closedRoot.getRowCount(); + List rows = new ArrayList<>(rowCount); + for (int i = 0; i < rowCount; i++) { + rows.add(arrowRootToFieldValueList(closedRoot, i, schema)); } + return ImmutableList.copyOf(rows); } } @@ -281,9 +312,6 @@ private static FieldValue arrowVectorToFieldValue( // Handle primitive types String stringVal; if (bqField.getType() == LegacySQLTypeName.TIMESTAMP) { - // Arrow timestamps are long values representing epoch seconds/millis/micros/nanos. - // Standard BigQuery JSON returns timestamps as string of epoch seconds with micro precision - // (e.g. "1408452095.220000"). TimeStampVector tsVector = (TimeStampVector) vector; long rawVal = tsVector.get(rowIndex); ArrowType.Timestamp tsType = (ArrowType.Timestamp) vector.getField().getType(); diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java new file mode 100644 index 000000000000..026301d953ba --- /dev/null +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java @@ -0,0 +1,131 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.bigquery; + +import java.util.ArrayList; +import java.util.List; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; + +/** Internal helper for Apache Arrow Schema/Field conversions. */ +final class ArrowPojoUtils { + + private ArrowPojoUtils() {} + + static com.google.cloud.bigquery.Schema arrowSchemaToBigQuerySchema(Schema arrowSchema) { + List fields = new ArrayList<>(); + for (Field arrowField : arrowSchema.getFields()) { + fields.add(arrowFieldToBigQueryField(arrowField)); + } + return com.google.cloud.bigquery.Schema.of(fields); + } + + static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowField) { + String name = arrowField.getName(); + ArrowType type = arrowField.getType(); + com.google.cloud.bigquery.Field.Builder builder; + + 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); + if (!innerField.getChildren().isEmpty()) { + List subFields = new ArrayList<>(); + for (Field childField : innerField.getChildren()) { + subFields.add(arrowFieldToBigQueryField(childField)); + } + builder = + com.google.cloud.bigquery.Field.newBuilder( + name, LegacySQLTypeName.RECORD, FieldList.of(subFields)); + } else { + LegacySQLTypeName innerType = arrowTypeToLegacySQLTypeName(innerField.getType()); + builder = com.google.cloud.bigquery.Field.newBuilder(name, innerType); + } + builder.setMode(com.google.cloud.bigquery.Field.Mode.REPEATED); + } else { + if (!arrowField.getChildren().isEmpty()) { + List subFields = new ArrayList<>(); + for (Field childField : arrowField.getChildren()) { + subFields.add(arrowFieldToBigQueryField(childField)); + } + builder = + com.google.cloud.bigquery.Field.newBuilder( + name, LegacySQLTypeName.RECORD, FieldList.of(subFields)); + } else { + LegacySQLTypeName bqType = arrowTypeToLegacySQLTypeName(type); + builder = com.google.cloud.bigquery.Field.newBuilder(name, bqType); + } + if (arrowField.isNullable()) { + builder.setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE); + } else { + builder.setMode(com.google.cloud.bigquery.Field.Mode.REQUIRED); + } + } + return builder.build(); + } + + static List createVectors(Schema arrowSchema, BufferAllocator allocator) { + List vectors = new ArrayList<>(); + try { + for (Field field : arrowSchema.getFields()) { + vectors.add(field.createVector(allocator)); + } + return vectors; + } 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; + } + } + + private static LegacySQLTypeName arrowTypeToLegacySQLTypeName(ArrowType type) { + switch (type.getTypeID()) { + case Int: + return LegacySQLTypeName.INTEGER; + case FloatingPoint: + return LegacySQLTypeName.FLOAT; + case Utf8: + return LegacySQLTypeName.STRING; + case Bool: + return LegacySQLTypeName.BOOLEAN; + case Binary: + return LegacySQLTypeName.BYTES; + case Decimal: + return LegacySQLTypeName.NUMERIC; + case Timestamp: + return LegacySQLTypeName.TIMESTAMP; + case Date: + return LegacySQLTypeName.DATE; + case Time: + return LegacySQLTypeName.TIME; + case Struct: + return LegacySQLTypeName.RECORD; + default: + throw new IllegalArgumentException("Unsupported Arrow type: " + type.getTypeID()); + } + } +} diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java index f23dbe4b01a5..248cc9c66be9 100644 --- a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java @@ -43,39 +43,42 @@ import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; -import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.types.pojo.FieldType; -import org.apache.arrow.vector.types.pojo.Schema; import org.junit.jupiter.api.Test; public class ArrowDeserializerTest { @Test public void testArrowSchemaToBigQuerySchema() { - Field intField = new Field("int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); - Field strField = new Field("str_col", FieldType.notNullable(new ArrowType.Utf8()), null); - Field boolField = new Field("bool_col", FieldType.nullable(new ArrowType.Bool()), null); - Field tsField = - new Field( + org.apache.arrow.vector.types.pojo.Field intField = + new org.apache.arrow.vector.types.pojo.Field( + "int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); + org.apache.arrow.vector.types.pojo.Field strField = + new org.apache.arrow.vector.types.pojo.Field( + "str_col", FieldType.notNullable(new ArrowType.Utf8()), null); + org.apache.arrow.vector.types.pojo.Field boolField = + new org.apache.arrow.vector.types.pojo.Field( + "bool_col", FieldType.nullable(new ArrowType.Bool()), null); + org.apache.arrow.vector.types.pojo.Field tsField = + new org.apache.arrow.vector.types.pojo.Field( "ts_col", FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC")), null); - Schema arrowSchema = new Schema(ImmutableList.of(intField, strField, boolField, tsField)); + org.apache.arrow.vector.types.pojo.Schema arrowSchema = + new org.apache.arrow.vector.types.pojo.Schema( + ImmutableList.of(intField, strField, boolField, tsField)); - com.google.cloud.bigquery.Schema bqSchema = - ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); assertEquals(4, bqSchema.getFields().size()); assertEquals("int_col", bqSchema.getFields().get(0).getName()); assertEquals(LegacySQLTypeName.INTEGER, bqSchema.getFields().get(0).getType()); - assertEquals( - com.google.cloud.bigquery.Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); + assertEquals(Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); assertEquals("str_col", bqSchema.getFields().get(1).getName()); assertEquals(LegacySQLTypeName.STRING, bqSchema.getFields().get(1).getType()); - assertEquals( - com.google.cloud.bigquery.Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); + assertEquals(Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); assertEquals("bool_col", bqSchema.getFields().get(2).getName()); assertEquals(LegacySQLTypeName.BOOLEAN, bqSchema.getFields().get(2).getType()); @@ -128,9 +131,8 @@ public void testDeserializeRecordBatchPrimitives() throws IOException { ImmutableList.of(intVector, nameVector, scoreVector, activeVector, bytesVector, tsVector); try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { - Schema arrowSchema = root.getSchema(); - com.google.cloud.bigquery.Schema bqSchema = - ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + org.apache.arrow.vector.types.pojo.Schema arrowSchema = root.getSchema(); + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); byte[] recordBatchBytes = serializeVectorSchemaRoot(root, allocator); @@ -155,11 +157,7 @@ public void testDeserializeRecordBatchPrimitives() throws IOException { assertEquals("102", row1.get("id").getStringValue()); assertEquals("Bob", row1.get("name").getStringValue()); assertNull(row1.get("score").getValue()); - assertEquals( - "false", - row1.get("false".equals("false") ? "active" : "score") != null - ? row1.get("active").getStringValue() - : "false"); + assertEquals("false", row1.get("active").getStringValue()); assertNull(row1.get("data").getValue()); assertNull(row1.get("ts").getValue()); } finally { @@ -179,10 +177,10 @@ public void testSchemaMismatchThrowsException() { intVector.setValueCount(1); try (VectorSchemaRoot root = new VectorSchemaRoot(ImmutableList.of(intVector))) { - com.google.cloud.bigquery.Schema mismatchedSchema = - com.google.cloud.bigquery.Schema.of( - com.google.cloud.bigquery.Field.of("col1", LegacySQLTypeName.INTEGER), - com.google.cloud.bigquery.Field.of("col2", LegacySQLTypeName.STRING)); + Schema mismatchedSchema = + Schema.of( + Field.of("col1", LegacySQLTypeName.INTEGER), + Field.of("col2", LegacySQLTypeName.STRING)); try { ArrowDeserializer.arrowRootToFieldValueList(root, 0, mismatchedSchema); diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java new file mode 100644 index 000000000000..c9a19e33f92c --- /dev/null +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java @@ -0,0 +1,209 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.bigquery; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.common.collect.ImmutableList; +import java.util.List; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.types.DateUnit; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.TimeUnit; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.arrow.vector.types.pojo.Schema; +import org.junit.jupiter.api.Test; + +public class ArrowPojoUtilsTest { + + @Test + public void testArrowSchemaToBigQuerySchema_Primitives() { + Field intField = new Field("int_col", FieldType.nullable(new ArrowType.Int(64, true)), null); + Field strField = new Field("str_col", FieldType.notNullable(new ArrowType.Utf8()), null); + Field boolField = new Field("bool_col", FieldType.nullable(new ArrowType.Bool()), null); + Field bytesField = new Field("bytes_col", FieldType.nullable(new ArrowType.Binary()), null); + Field floatField = + new Field( + "float_col", + FieldType.nullable(new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), + null); + Field decimalField = + new Field("num_col", FieldType.nullable(new ArrowType.Decimal(38, 9, 128)), null); + Field dateField = + new Field("date_col", FieldType.nullable(new ArrowType.Date(DateUnit.DAY)), null); + Field timeField = + new Field( + "time_col", FieldType.nullable(new ArrowType.Time(TimeUnit.MICROSECOND, 64)), null); + Field tsField = + new Field( + "ts_col", + FieldType.nullable(new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC")), + null); + + Schema arrowSchema = + new Schema( + ImmutableList.of( + intField, + strField, + boolField, + bytesField, + floatField, + decimalField, + dateField, + timeField, + tsField)); + + com.google.cloud.bigquery.Schema bqSchema = + ArrowPojoUtils.arrowSchemaToBigQuerySchema(arrowSchema); + + assertEquals(9, bqSchema.getFields().size()); + assertEquals(LegacySQLTypeName.INTEGER, bqSchema.getFields().get(0).getType()); + assertEquals( + com.google.cloud.bigquery.Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); + + assertEquals(LegacySQLTypeName.STRING, bqSchema.getFields().get(1).getType()); + assertEquals( + com.google.cloud.bigquery.Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); + + assertEquals(LegacySQLTypeName.BOOLEAN, bqSchema.getFields().get(2).getType()); + assertEquals(LegacySQLTypeName.BYTES, bqSchema.getFields().get(3).getType()); + assertEquals(LegacySQLTypeName.FLOAT, bqSchema.getFields().get(4).getType()); + assertEquals(LegacySQLTypeName.NUMERIC, bqSchema.getFields().get(5).getType()); + assertEquals(LegacySQLTypeName.DATE, bqSchema.getFields().get(6).getType()); + assertEquals(LegacySQLTypeName.TIME, bqSchema.getFields().get(7).getType()); + assertEquals(LegacySQLTypeName.TIMESTAMP, bqSchema.getFields().get(8).getType()); + } + + @Test + public void testArrowSchemaToBigQuerySchema_NestedStruct() { + Field innerInt = new Field("id", FieldType.nullable(new ArrowType.Int(32, true)), null); + Field innerStr = new Field("name", FieldType.nullable(new ArrowType.Utf8()), null); + Field structField = + new Field( + "person", + FieldType.nullable(new ArrowType.Struct()), + ImmutableList.of(innerInt, innerStr)); + + Schema arrowSchema = new Schema(ImmutableList.of(structField)); + com.google.cloud.bigquery.Schema bqSchema = + ArrowPojoUtils.arrowSchemaToBigQuerySchema(arrowSchema); + + assertEquals(1, bqSchema.getFields().size()); + com.google.cloud.bigquery.Field personField = bqSchema.getFields().get(0); + assertEquals("person", personField.getName()); + assertEquals(LegacySQLTypeName.RECORD, personField.getType()); + assertEquals(2, personField.getSubFields().size()); + assertEquals("id", personField.getSubFields().get(0).getName()); + assertEquals(LegacySQLTypeName.INTEGER, personField.getSubFields().get(0).getType()); + assertEquals("name", personField.getSubFields().get(1).getName()); + assertEquals(LegacySQLTypeName.STRING, personField.getSubFields().get(1).getType()); + } + + @Test + public void testArrowSchemaToBigQuerySchema_ListPrimitives() { + Field itemField = new Field("item", FieldType.notNullable(new ArrowType.Utf8()), null); + Field listField = + new Field("tags", FieldType.nullable(new ArrowType.List()), ImmutableList.of(itemField)); + + Schema arrowSchema = new Schema(ImmutableList.of(listField)); + com.google.cloud.bigquery.Schema bqSchema = + ArrowPojoUtils.arrowSchemaToBigQuerySchema(arrowSchema); + + assertEquals(1, bqSchema.getFields().size()); + com.google.cloud.bigquery.Field tagsField = bqSchema.getFields().get(0); + assertEquals("tags", tagsField.getName()); + assertEquals(LegacySQLTypeName.STRING, tagsField.getType()); + assertEquals(com.google.cloud.bigquery.Field.Mode.REPEATED, tagsField.getMode()); + } + + @Test + public void testArrowSchemaToBigQuerySchema_ListOfStruct() { + Field innerKey = new Field("key", FieldType.nullable(new ArrowType.Utf8()), null); + Field innerVal = new Field("value", FieldType.nullable(new ArrowType.Int(64, true)), null); + Field structField = + new Field( + "item", + FieldType.nullable(new ArrowType.Struct()), + ImmutableList.of(innerKey, innerVal)); + Field listField = + new Field( + "entries", FieldType.nullable(new ArrowType.List()), ImmutableList.of(structField)); + + Schema arrowSchema = new Schema(ImmutableList.of(listField)); + com.google.cloud.bigquery.Schema bqSchema = + ArrowPojoUtils.arrowSchemaToBigQuerySchema(arrowSchema); + + assertEquals(1, bqSchema.getFields().size()); + com.google.cloud.bigquery.Field entriesField = bqSchema.getFields().get(0); + assertEquals("entries", entriesField.getName()); + assertEquals(LegacySQLTypeName.RECORD, entriesField.getType()); + assertEquals(com.google.cloud.bigquery.Field.Mode.REPEATED, entriesField.getMode()); + assertEquals(2, entriesField.getSubFields().size()); + assertEquals("key", entriesField.getSubFields().get(0).getName()); + assertEquals("value", entriesField.getSubFields().get(1).getName()); + } + + @Test + public void testArrowSchemaToBigQuerySchema_EmptyListThrowsException() { + Field emptyList = + new Field("empty_list", FieldType.nullable(new ArrowType.List()), ImmutableList.of()); + Schema arrowSchema = new Schema(ImmutableList.of(emptyList)); + + IllegalArgumentException thrown = + assertThrows( + IllegalArgumentException.class, + () -> ArrowPojoUtils.arrowSchemaToBigQuerySchema(arrowSchema)); + assertTrue(thrown.getMessage().contains("must have at least one child field")); + } + + @Test + public void testArrowSchemaToBigQuerySchema_UnsupportedTypeThrowsException() { + Field unsupportedField = + new Field("unsupported", FieldType.nullable(new ArrowType.Null()), null); + Schema arrowSchema = new Schema(ImmutableList.of(unsupportedField)); + + IllegalArgumentException thrown = + assertThrows( + IllegalArgumentException.class, + () -> ArrowPojoUtils.arrowSchemaToBigQuerySchema(arrowSchema)); + assertTrue(thrown.getMessage().contains("Unsupported Arrow type")); + } + + @Test + public void testCreateVectors_Success() { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + Field intField = new Field("int_col", FieldType.nullable(new ArrowType.Int(32, true)), null); + Field strField = new Field("str_col", FieldType.nullable(new ArrowType.Utf8()), null); + Schema arrowSchema = new Schema(ImmutableList.of(intField, strField)); + + List vectors = ArrowPojoUtils.createVectors(arrowSchema, allocator); + assertEquals(2, vectors.size()); + assertEquals("int_col", vectors.get(0).getName()); + assertEquals("str_col", vectors.get(1).getName()); + + for (FieldVector v : vectors) { + v.close(); + } + } + } +} From 7fc0c80e8da9996504fc726232d01409a7a4ea81 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Wed, 12 Aug 2026 16:58:04 -0400 Subject: [PATCH 11/13] docs(bigquery): add Javadoc comments to all methods in ArrowDeserializer and ArrowPojoUtils --- .../cloud/bigquery/ArrowDeserializer.java | 21 +++++++++- .../google/cloud/bigquery/ArrowPojoUtils.java | 41 ++++++++++++++++++- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index e481132e3bd0..5ded2cb17e81 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -45,10 +45,20 @@ */ final class ArrowDeserializer { + /** Lazy initialization holder for the root {@link BufferAllocator}. */ private static class AllocatorHolder { private static final BufferAllocator ALLOCATOR = new RootAllocator(Long.MAX_VALUE); } + /** + * Instantiates a new {@link VectorSchemaRoot} for the given Arrow schema using vectors allocated + * from the provided child allocator, ensuring LIFO cleanup if an error occurs during + * construction. + * + * @param arrowSchema the Apache Arrow schema definition + * @param allocator the buffer allocator to bind the vectors to + * @return a new VectorSchemaRoot containing allocated field vectors + */ private static VectorSchemaRoot createVectorSchemaRoot( org.apache.arrow.vector.types.pojo.Schema arrowSchema, BufferAllocator allocator) { List vectors = ArrowPojoUtils.createVectors(arrowSchema, allocator); @@ -93,6 +103,13 @@ static String arrowSchemaToJson(Object arrowSchema) { return ((org.apache.arrow.vector.types.pojo.Schema) arrowSchema).toJson(); } + /** + * Deserializes an Apache Arrow Schema object from its JSON string representation. + * + * @param json the JSON string representation of the Arrow schema + * @return the deserialized Apache Arrow Schema object, or null if json is null + * @throws IllegalArgumentException if the JSON string cannot be parsed as an Arrow schema + */ static Object jsonToArrowSchema(String json) { if (json == null) { return null; @@ -195,8 +212,8 @@ static Schema arrowSchemaToBigQuerySchema(Object arrowSchema) { * Deserializes a raw binary Arrow record batch payload into a list of BigQuery {@link * FieldValueList} row objects. * - *

Allocates off-heap memory within a local {@link RootAllocator} scope and closes all Arrow - * vector resources before returning, guaranteeing that native memory is released. + *

Allocates off-heap memory within a local child allocator scope and closes all Arrow vector + * resources before returning, guaranteeing that native memory is released. * * @param recordBatchBytes the raw binary Arrow record batch payload * @param schema the target BigQuery Schema diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java index 026301d953ba..8607b9f50b7b 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java @@ -24,11 +24,22 @@ import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.types.pojo.Schema; -/** Internal helper for Apache Arrow Schema/Field conversions. */ +/** + * Internal helper utility for converting Apache Arrow POJO Schema and Field definitions into + * BigQuery Veneer {@link com.google.cloud.bigquery.Schema} and {@link + * com.google.cloud.bigquery.Field} models. + */ final class ArrowPojoUtils { private ArrowPojoUtils() {} + /** + * Converts an Apache Arrow {@link Schema} into a BigQuery Veneer {@link + * com.google.cloud.bigquery.Schema}. + * + * @param arrowSchema the Apache Arrow schema definition + * @return the corresponding BigQuery Veneer Schema + */ static com.google.cloud.bigquery.Schema arrowSchemaToBigQuerySchema(Schema arrowSchema) { List fields = new ArrayList<>(); for (Field arrowField : arrowSchema.getFields()) { @@ -37,6 +48,16 @@ static com.google.cloud.bigquery.Schema arrowSchemaToBigQuerySchema(Schema arrow return com.google.cloud.bigquery.Schema.of(fields); } + /** + * Recursively converts an Apache Arrow {@link Field} into a BigQuery Veneer {@link + * com.google.cloud.bigquery.Field}. + * + *

Handles primitive types, repeated/list types, and nested struct/record types. + * + * @param arrowField the Apache Arrow field definition + * @return the corresponding BigQuery Veneer Field + * @throws IllegalArgumentException if an Arrow List field contains no child elements + */ static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowField) { String name = arrowField.getName(); ArrowType type = arrowField.getType(); @@ -83,6 +104,17 @@ static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowFiel return builder.build(); } + /** + * Instantiates a list of {@link FieldVector} instances corresponding to the fields in the + * provided Arrow schema using the specified allocator. + * + *

Guarantees exception-safe LIFO cleanup of already-allocated vectors if an allocation fails + * halfway through. + * + * @param arrowSchema the Apache Arrow schema definition + * @param allocator the buffer allocator to allocate vector memory from + * @return the list of allocated FieldVector instances + */ static List createVectors(Schema arrowSchema, BufferAllocator allocator) { List vectors = new ArrayList<>(); try { @@ -102,6 +134,13 @@ static List createVectors(Schema arrowSchema, BufferAllocator alloc } } + /** + * Maps an Apache {@link ArrowType} to its corresponding BigQuery {@link LegacySQLTypeName}. + * + * @param type the Apache Arrow type + * @return the matching BigQuery LegacySQLTypeName + * @throws IllegalArgumentException if the Arrow type is unsupported + */ private static LegacySQLTypeName arrowTypeToLegacySQLTypeName(ArrowType type) { switch (type.getTypeID()) { case Int: From fac0d0cfa49551f99970fb7de2e874318eab4cb9 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Wed, 12 Aug 2026 17:17:01 -0400 Subject: [PATCH 12/13] refactor(bigquery): clean up FQCNs in ArrowDeserializer, ArrowPojoUtils, and tests --- .../google/cloud/bigquery/ArrowDeserializer.java | 15 ++++++++++++--- .../com/google/cloud/bigquery/ArrowPojoUtils.java | 13 ++++++------- .../google/cloud/bigquery/ArrowPojoUtilsTest.java | 11 +++++------ 3 files changed, 23 insertions(+), 16 deletions(-) diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java index 5ded2cb17e81..5078a731c626 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowDeserializer.java @@ -222,12 +222,21 @@ static Schema arrowSchemaToBigQuerySchema(Object arrowSchema) { * @throws IOException if deserialization of the Arrow record batch fails */ static List deserializeRecordBatch( - byte[] recordBatchBytes, Schema schema, org.apache.arrow.vector.types.pojo.Schema arrowSchema) - throws IOException { + byte[] recordBatchBytes, Schema schema, Object arrowSchema) throws IOException { + org.apache.arrow.vector.types.pojo.Schema schemaPojo = + arrowSchema instanceof org.apache.arrow.vector.types.pojo.Schema + ? (org.apache.arrow.vector.types.pojo.Schema) arrowSchema + : (arrowSchema instanceof String + ? (org.apache.arrow.vector.types.pojo.Schema) + jsonToArrowSchema((String) arrowSchema) + : null); + if (schemaPojo == null) { + throw new IllegalArgumentException("Arrow schema must not be null"); + } try (BufferAllocator childAllocator = AllocatorHolder.ALLOCATOR.newChildAllocator( "deserializeRecordBatch", 0, Long.MAX_VALUE); - VectorSchemaRoot closedRoot = createVectorSchemaRoot(arrowSchema, childAllocator); + VectorSchemaRoot closedRoot = createVectorSchemaRoot(schemaPojo, childAllocator); ByteArrayReadableSeekableByteChannel byteChannel = new ByteArrayReadableSeekableByteChannel(recordBatchBytes); ReadChannel readChannel = new ReadChannel(byteChannel); diff --git a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java index 8607b9f50b7b..8835ebcb682d 100644 --- a/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java +++ b/java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ArrowPojoUtils.java @@ -16,6 +16,7 @@ package com.google.cloud.bigquery; +import com.google.cloud.bigquery.Field.Mode; import java.util.ArrayList; import java.util.List; import org.apache.arrow.memory.BufferAllocator; @@ -26,16 +27,14 @@ /** * Internal helper utility for converting Apache Arrow POJO Schema and Field definitions into - * BigQuery Veneer {@link com.google.cloud.bigquery.Schema} and {@link - * com.google.cloud.bigquery.Field} models. + * BigQuery Veneer {@link Schema} and {@link com.google.cloud.bigquery.Field} models. */ final class ArrowPojoUtils { private ArrowPojoUtils() {} /** - * Converts an Apache Arrow {@link Schema} into a BigQuery Veneer {@link - * com.google.cloud.bigquery.Schema}. + * Converts an Apache Arrow {@link Schema} into a BigQuery Veneer {@link Schema}. * * @param arrowSchema the Apache Arrow schema definition * @return the corresponding BigQuery Veneer Schema @@ -81,7 +80,7 @@ static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowFiel LegacySQLTypeName innerType = arrowTypeToLegacySQLTypeName(innerField.getType()); builder = com.google.cloud.bigquery.Field.newBuilder(name, innerType); } - builder.setMode(com.google.cloud.bigquery.Field.Mode.REPEATED); + builder.setMode(Mode.REPEATED); } else { if (!arrowField.getChildren().isEmpty()) { List subFields = new ArrayList<>(); @@ -96,9 +95,9 @@ static com.google.cloud.bigquery.Field arrowFieldToBigQueryField(Field arrowFiel builder = com.google.cloud.bigquery.Field.newBuilder(name, bqType); } if (arrowField.isNullable()) { - builder.setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE); + builder.setMode(Mode.NULLABLE); } else { - builder.setMode(com.google.cloud.bigquery.Field.Mode.REQUIRED); + builder.setMode(Mode.REQUIRED); } } return builder.build(); diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java index c9a19e33f92c..b040a0e8fd7d 100644 --- a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowPojoUtilsTest.java @@ -20,6 +20,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import com.google.cloud.bigquery.Field.Mode; import com.google.common.collect.ImmutableList; import java.util.List; import org.apache.arrow.memory.BufferAllocator; @@ -78,12 +79,10 @@ public void testArrowSchemaToBigQuerySchema_Primitives() { assertEquals(9, bqSchema.getFields().size()); assertEquals(LegacySQLTypeName.INTEGER, bqSchema.getFields().get(0).getType()); - assertEquals( - com.google.cloud.bigquery.Field.Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); + assertEquals(Mode.NULLABLE, bqSchema.getFields().get(0).getMode()); assertEquals(LegacySQLTypeName.STRING, bqSchema.getFields().get(1).getType()); - assertEquals( - com.google.cloud.bigquery.Field.Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); + assertEquals(Mode.REQUIRED, bqSchema.getFields().get(1).getMode()); assertEquals(LegacySQLTypeName.BOOLEAN, bqSchema.getFields().get(2).getType()); assertEquals(LegacySQLTypeName.BYTES, bqSchema.getFields().get(3).getType()); @@ -133,7 +132,7 @@ public void testArrowSchemaToBigQuerySchema_ListPrimitives() { com.google.cloud.bigquery.Field tagsField = bqSchema.getFields().get(0); assertEquals("tags", tagsField.getName()); assertEquals(LegacySQLTypeName.STRING, tagsField.getType()); - assertEquals(com.google.cloud.bigquery.Field.Mode.REPEATED, tagsField.getMode()); + assertEquals(Mode.REPEATED, tagsField.getMode()); } @Test @@ -157,7 +156,7 @@ public void testArrowSchemaToBigQuerySchema_ListOfStruct() { com.google.cloud.bigquery.Field entriesField = bqSchema.getFields().get(0); assertEquals("entries", entriesField.getName()); assertEquals(LegacySQLTypeName.RECORD, entriesField.getType()); - assertEquals(com.google.cloud.bigquery.Field.Mode.REPEATED, entriesField.getMode()); + assertEquals(Mode.REPEATED, entriesField.getMode()); assertEquals(2, entriesField.getSubFields().size()); assertEquals("key", entriesField.getSubFields().get(0).getName()); assertEquals("value", entriesField.getSubFields().get(1).getName()); From e8ba1a8aee7d090677b004a9e1eb593235583d47 Mon Sep 17 00:00:00 2001 From: Jin Seop Kim Date: Wed, 12 Aug 2026 17:40:38 -0400 Subject: [PATCH 13/13] test(bigquery): add comprehensive unit tests for ArrowDeserializer.loadArrowRows streaming and pagination --- .../cloud/bigquery/ArrowDeserializerTest.java | 166 ++++++++++++++++++ 1 file changed, 166 insertions(+) diff --git a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java index 248cc9c66be9..ff6faa6cc41f 100644 --- a/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java +++ b/java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowDeserializerTest.java @@ -17,15 +17,21 @@ package com.google.cloud.bigquery; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; +import com.google.cloud.bigquery.storage.v1.ReadRowsResponse; import com.google.common.collect.ImmutableList; import com.google.common.io.BaseEncoding; +import com.google.protobuf.ByteString; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.channels.Channels; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.memory.RootAllocator; @@ -194,6 +200,166 @@ public void testSchemaMismatchThrowsException() { } } + @Test + public void testLoadArrowRows_multiBatchStream() throws IOException { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + ReadRowsResponse r1 = + createReadRowsResponse(Arrays.asList(1, 2), Arrays.asList("item1", "item2"), allocator); + ReadRowsResponse r2 = + createReadRowsResponse(Arrays.asList(3, 4), Arrays.asList("item3", "item4"), allocator); + + org.apache.arrow.vector.types.pojo.Schema arrowSchema = createSimpleArrowSchema(); + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + + List rowBatch = new ArrayList<>(); + boolean hasMore = + ArrowDeserializer.loadArrowRows( + Arrays.asList(r1, r2).iterator(), + arrowSchema, + null, + bqSchema, + rowBatch, + 10L, + 0L, + 10L); + + assertFalse(hasMore); + assertEquals(4, rowBatch.size()); + assertEquals("1", rowBatch.get(0).get("id").getStringValue()); + assertEquals("item1", rowBatch.get(0).get("name").getStringValue()); + assertEquals("4", rowBatch.get(3).get("id").getStringValue()); + assertEquals("item4", rowBatch.get(3).get("name").getStringValue()); + } + } + + @Test + public void testLoadArrowRows_respectsPageSize() throws IOException { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + ReadRowsResponse r1 = + createReadRowsResponse(Arrays.asList(1, 2), Arrays.asList("item1", "item2"), allocator); + ReadRowsResponse r2 = + createReadRowsResponse(Arrays.asList(3, 4), Arrays.asList("item3", "item4"), allocator); + + org.apache.arrow.vector.types.pojo.Schema arrowSchema = createSimpleArrowSchema(); + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + + List rowBatch = new ArrayList<>(); + boolean hasMore = + ArrowDeserializer.loadArrowRows( + Arrays.asList(r1, r2).iterator(), arrowSchema, null, bqSchema, rowBatch, 2L, 0L, 10L); + + assertTrue(hasMore); + assertEquals(2, rowBatch.size()); + assertEquals("1", rowBatch.get(0).get("id").getStringValue()); + assertEquals("2", rowBatch.get(1).get("id").getStringValue()); + } + } + + @Test + public void testLoadArrowRows_respectsMaxResults() throws IOException { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + ReadRowsResponse r1 = + createReadRowsResponse(Arrays.asList(1, 2), Arrays.asList("item1", "item2"), allocator); + ReadRowsResponse r2 = + createReadRowsResponse(Arrays.asList(3, 4), Arrays.asList("item3", "item4"), allocator); + + org.apache.arrow.vector.types.pojo.Schema arrowSchema = createSimpleArrowSchema(); + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + + List rowBatch = new ArrayList<>(); + boolean hasMore = + ArrowDeserializer.loadArrowRows( + Arrays.asList(r1, r2).iterator(), arrowSchema, null, bqSchema, rowBatch, 10L, 0L, 3L); + + assertFalse(hasMore); + assertEquals(3, rowBatch.size()); + assertEquals("1", rowBatch.get(0).get("id").getStringValue()); + assertEquals("2", rowBatch.get(1).get("id").getStringValue()); + assertEquals("3", rowBatch.get(2).get("id").getStringValue()); + } + } + + @Test + public void testLoadArrowRows_unconsumedBatchRowsSignalHasMore() throws IOException { + try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) { + ReadRowsResponse r1 = + createReadRowsResponse( + Arrays.asList(1, 2, 3, 4), + Arrays.asList("item1", "item2", "item3", "item4"), + allocator); + + org.apache.arrow.vector.types.pojo.Schema arrowSchema = createSimpleArrowSchema(); + Schema bqSchema = ArrowDeserializer.arrowSchemaToBigQuerySchema(arrowSchema); + + List rowBatch = new ArrayList<>(); + boolean hasMore = + ArrowDeserializer.loadArrowRows( + Arrays.asList(r1).iterator(), arrowSchema, null, bqSchema, rowBatch, 2L, 0L, 10L); + + assertTrue(hasMore); + assertEquals(2, rowBatch.size()); + assertEquals("1", rowBatch.get(0).get("id").getStringValue()); + assertEquals("2", rowBatch.get(1).get("id").getStringValue()); + } + } + + @Test + public void testLoadArrowRows_nullSchemaReturnsFalse() throws IOException { + List rowBatch = new ArrayList<>(); + boolean hasMore = + ArrowDeserializer.loadArrowRows( + Arrays.asList().iterator(), + null, + null, + Schema.of(), + rowBatch, + 10L, + 0L, + 10L); + assertFalse(hasMore); + } + + private static org.apache.arrow.vector.types.pojo.Schema createSimpleArrowSchema() { + org.apache.arrow.vector.types.pojo.Field intField = + new org.apache.arrow.vector.types.pojo.Field( + "id", FieldType.nullable(new ArrowType.Int(32, true)), null); + org.apache.arrow.vector.types.pojo.Field strField = + new org.apache.arrow.vector.types.pojo.Field( + "name", FieldType.nullable(new ArrowType.Utf8()), null); + return new org.apache.arrow.vector.types.pojo.Schema(ImmutableList.of(intField, strField)); + } + + private ReadRowsResponse createReadRowsResponse( + List ids, List names, BufferAllocator allocator) throws IOException { + IntVector intVector = new IntVector("id", allocator); + intVector.allocateNew(ids.size()); + for (int i = 0; i < ids.size(); i++) { + intVector.set(i, ids.get(i)); + } + intVector.setValueCount(ids.size()); + + VarCharVector nameVector = new VarCharVector("name", allocator); + nameVector.allocateNew(names.size()); + for (int i = 0; i < names.size(); i++) { + nameVector.set(i, names.get(i).getBytes(StandardCharsets.UTF_8)); + } + nameVector.setValueCount(names.size()); + + List vectors = ImmutableList.of(intVector, nameVector); + try (VectorSchemaRoot root = new VectorSchemaRoot(vectors)) { + byte[] bytes = serializeVectorSchemaRoot(root, allocator); + com.google.cloud.bigquery.storage.v1.ArrowRecordBatch protoBatch = + com.google.cloud.bigquery.storage.v1.ArrowRecordBatch.newBuilder() + .setSerializedRecordBatch(ByteString.copyFrom(bytes)) + .build(); + return ReadRowsResponse.newBuilder().setArrowRecordBatch(protoBatch).build(); + } finally { + for (FieldVector vector : vectors) { + vector.close(); + } + } + } + private byte[] serializeVectorSchemaRoot(VectorSchemaRoot root, BufferAllocator allocator) throws IOException { ByteArrayOutputStream out = new ByteArrayOutputStream();