From e8b62c407ec16e10c3959b2cb54f3464265d5709 Mon Sep 17 00:00:00 2001 From: stevelordbq Date: Fri, 13 Feb 2026 14:53:44 -0800 Subject: [PATCH 1/6] Write struct initial version --- .../spark/spanner/SpannerWriterUtils.java | 110 ++++++++++++------ 1 file changed, 74 insertions(+), 36 deletions(-) diff --git a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java index 46e80821..6d3b53c1 100644 --- a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java +++ b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java @@ -17,6 +17,8 @@ import com.google.cloud.Date; import com.google.cloud.Timestamp; import com.google.cloud.spanner.Mutation; +import com.google.cloud.spanner.Struct; +import com.google.cloud.spanner.Type; import com.google.cloud.spanner.Value; import java.math.BigDecimal; import java.time.LocalDate; @@ -98,65 +100,55 @@ private interface FieldConverter { // Long array ARRAY_TYPE_CONVERTERS.put( DataTypes.LongType, - (row, i, type) -> { - return arrayConverter(row, i, ArrayData::getLong, Value::int64Array); - }); + (row, i, type) -> arrayConverter(row, i, ArrayData::getLong, Value::int64Array)); // String array ARRAY_TYPE_CONVERTERS.put( DataTypes.StringType, - (row, i, type) -> { - return arrayConverter( - row, i, (a, idx) -> a.getUTF8String(idx).toString(), Value::stringArray); - }); + (row, i, type) -> + arrayConverter( + row, i, (a, idx) -> a.getUTF8String(idx).toString(), Value::stringArray)); // Boolean ARRAY_TYPE_CONVERTERS.put( DataTypes.BooleanType, - (row, i, type) -> { - return arrayConverter(row, i, ArrayData::getBoolean, Value::boolArray); - }); + (row, i, type) -> arrayConverter(row, i, ArrayData::getBoolean, Value::boolArray)); // Double ARRAY_TYPE_CONVERTERS.put( DataTypes.DoubleType, - (row, i, type) -> { - return arrayConverter(row, i, ArrayData::getDouble, Value::float64Array); - }); + (row, i, type) -> arrayConverter(row, i, ArrayData::getDouble, Value::float64Array)); // Binary ARRAY_TYPE_CONVERTERS.put( DataTypes.BinaryType, - (row, i, type) -> { - return arrayConverter( - row, i, (a, idx) -> ByteArray.copyFrom(a.getBinary(idx)), Value::bytesArray); - }); + (row, i, type) -> + arrayConverter( + row, i, (a, idx) -> ByteArray.copyFrom(a.getBinary(idx)), Value::bytesArray)); // Timestamp ARRAY_TYPE_CONVERTERS.put( DataTypes.TimestampType, - (row, i, type) -> { - return arrayConverter( - row, - i, - (a, idx) -> Timestamp.ofTimeMicroseconds(a.getLong(idx)), - Value::timestampArray); - }); + (row, i, type) -> + arrayConverter( + row, + i, + (a, idx) -> Timestamp.ofTimeMicroseconds(a.getLong(idx)), + Value::timestampArray)); // Date ARRAY_TYPE_CONVERTERS.put( DataTypes.DateType, - (row, i, type) -> { - return arrayConverter( - row, - i, - (a, idx) -> { - LocalDate localDate = LocalDate.ofEpochDay(a.getInt(idx)); - return Date.fromYearMonthDay( - localDate.getYear(), localDate.getMonthValue(), localDate.getDayOfMonth()); - }, - Value::dateArray); - }); + (row, i, type) -> + arrayConverter( + row, + i, + (a, idx) -> { + LocalDate localDate = LocalDate.ofEpochDay(a.getInt(idx)); + return Date.fromYearMonthDay( + localDate.getYear(), localDate.getMonthValue(), localDate.getDayOfMonth()); + }, + Value::dateArray)); // Note: DecimalType is handled dynamically in the method below // because it relies on instanceof checks rather than strict equality. @@ -211,6 +203,21 @@ public static Mutation internalRowToMutation( return builder.build(); } + private static Struct internalRowToStruct(InternalRow record, StructType schema) { + final Struct.Builder builder = Struct.newBuilder(); + + for (int i = 0; i < schema.length(); i++) { + final StructField field = schema.fields()[i]; + final DataType fieldType = field.dataType(); + final String fieldName = field.name(); + + Value spannerValue = convertField(record, i, fieldType); + builder.set(fieldName).to(spannerValue); + } + + return builder.build(); + } + // Helper method to resolve the strategy private static Value convertField(InternalRow row, int index, DataType type) { // Check exact matches @@ -229,6 +236,12 @@ private static Value convertField(InternalRow row, int index, DataType type) { return Value.numeric(bd); } + if (type instanceof StructType) { + StructType structType = (StructType) type; + return Value.struct( + internalRowToStruct(row.getStruct(index, structType.length()), structType)); + } + if (type instanceof ArrayType) { ArrayType arrayType = (ArrayType) type; DataType elementType = arrayType.elementType(); @@ -245,8 +258,33 @@ private static Value convertField(InternalRow row, int index, DataType type) { (ad, i) -> ad.getDecimal(i, dt.precision(), dt.scale()).toJavaBigDecimal(), Value::numericArray); } + + if (elementType instanceof StructType) { + StructType structType = (StructType) elementType; + + if (row.isNullAt(index)) { + return Value.struct(Type.struct(), null); // TODO: this may not be correct type. + } + + final ArrayData arrayData = row.getArray(index); + final int numElements = arrayData.numElements(); + List convertedList = new ArrayList<>(numElements); + + for (int j = 0; j < numElements; j++) { + if (arrayData.isNullAt(j)) { + convertedList.add(null); + } else { + // Uses specialized Spark getters (getLong, getDouble, etc.) + convertedList.add( + internalRowToStruct(row.getStruct(index, structType.length()), structType)); + } + } + + return Value.structArray( + Type.struct(), convertedList); // TODO: very dodgy code for spanner type + } } - // TODO handle Struct here. + // unsupported type throw new UnsupportedOperationException("Unsupported Spark DataType: " + type); } From 69765bf08eafb177f5ce149888c3e066ac7e1a3b Mon Sep 17 00:00:00 2001 From: stevelordbq Date: Tue, 17 Feb 2026 10:44:56 -0800 Subject: [PATCH 2/6] Write struct add unit test for struct array --- .../spark/spanner/SpannerWriterUtils.java | 8 ++-- .../spark/spanner/SpannerWriterUtilsTest.java | 41 ++++++++++++++++++- 2 files changed, 44 insertions(+), 5 deletions(-) diff --git a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java index 6d3b53c1..7d394c85 100644 --- a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java +++ b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java @@ -263,7 +263,7 @@ private static Value convertField(InternalRow row, int index, DataType type) { StructType structType = (StructType) elementType; if (row.isNullAt(index)) { - return Value.struct(Type.struct(), null); // TODO: this may not be correct type. + return Value.structArray(Type.struct(), null); } final ArrayData arrayData = row.getArray(index); @@ -276,12 +276,14 @@ private static Value convertField(InternalRow row, int index, DataType type) { } else { // Uses specialized Spark getters (getLong, getDouble, etc.) convertedList.add( - internalRowToStruct(row.getStruct(index, structType.length()), structType)); + internalRowToStruct(arrayData.getStruct(j, structType.length()), structType)); } } return Value.structArray( - Type.struct(), convertedList); // TODO: very dodgy code for spanner type + Type.struct(Type.StructField.of("field", Type.string())), + convertedList); // TODO: the struct element type should represent the schema of the + // struct. } } diff --git a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java index 18446827..a657a08a 100644 --- a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java +++ b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java @@ -21,6 +21,8 @@ import com.google.cloud.Date; import com.google.cloud.Timestamp; import com.google.cloud.spanner.Mutation; +import com.google.cloud.spanner.Struct; +import com.google.cloud.spanner.Type; import com.google.cloud.spanner.Value; import java.math.MathContext; import java.math.RoundingMode; @@ -143,6 +145,7 @@ public static Collection data() { {"timestamp", DataTypes.TimestampType, Value.timestamp(null)}, {"date", DataTypes.DateType, Value.date(null)}, {"decimal", new DecimalType(38, 9), Value.numeric(null)}, + {"struct", new StructType(), Value.struct(Struct.newBuilder().build())}, { "long_array", DataTypes.createArrayType(DataTypes.LongType), @@ -174,6 +177,11 @@ public static Collection data() { "decimal_array", DataTypes.createArrayType(new DecimalType(38, 9)), Value.numericArray(null) + }, + { + "struct_array", + DataTypes.createArrayType(new StructType()), + Value.structArray(Type.struct(), null) } }); } @@ -341,6 +349,30 @@ public static Collection data() { ad, (Decimal[]) d, (i, val) -> when(ad.getDecimal(i, 38, 9)).thenReturn((Decimal) val)) + }, + + // 9. Struct Array + { + "struct_array", + new StructType() + .add( + "field", + DataTypes.StringType), // Add at least one field to match getStruct(i, 1) + new InternalRow[] { + mock(InternalRow.class) + }, // Use InternalRow instead of Spanner Struct + Value.structArray( + Type.struct(Type.StructField.of("field", Type.string())), + Collections.singletonList(Struct.newBuilder().set("field").to("value").build())), + (BiConsumer) + (ad, d) -> + setupArrayMock( + ad, + (InternalRow[]) d, // Cast to InternalRow array + (i, val) -> { + when(ad.getStruct(i, 1)).thenReturn((InternalRow) val); + when(((InternalRow) val).getString(0)).thenReturn("value"); + }) } }); } @@ -365,10 +397,10 @@ public void testArrayConversion() { InternalRow row = mock(InternalRow.class); ArrayData arrayData = mock(ArrayData.class); - when(row.isNullAt(0)).thenReturn(false); + when(row.isNullAt(0)).thenReturn(false).thenReturn(false); when(row.getArray(0)).thenReturn(arrayData); - // Executes the specific stubbing (e.g., ad.toLongArray() or ad.toIntArray()) + // Executes the specific stubbing (e.g., ad.toLongArray(), ad.toIntArray(), ad.to) mockSetup.accept(arrayData, inputData); Mutation mutation = SpannerWriterUtils.internalRowToMutation(TABLE_NAME, row, schema); @@ -377,4 +409,9 @@ public void testArrayConversion() { "Failure on type: " + colName, expectedValue, mutation.asMap().get(colName)); } } + + @Test + public void testStructConversion() { + // TODO: add test + } } From d15abeda54ee44b3b922e6bd6d1e35cd6e5ce22c Mon Sep 17 00:00:00 2001 From: stevelordbq Date: Tue, 17 Feb 2026 14:37:18 -0800 Subject: [PATCH 3/6] Write struct extend unit test for struct array --- .../spark/spanner/SpannerWriterUtils.java | 9 +++---- .../spark/spanner/SpannerWriterUtilsTest.java | 24 ++++++++++++------- 2 files changed, 21 insertions(+), 12 deletions(-) diff --git a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java index 7d394c85..be060da0 100644 --- a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java +++ b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java @@ -237,6 +237,9 @@ private static Value convertField(InternalRow row, int index, DataType type) { } if (type instanceof StructType) { + if (row.isNullAt(index)) { + return Value.struct(Struct.newBuilder().build()); + } StructType structType = (StructType) type; return Value.struct( internalRowToStruct(row.getStruct(index, structType.length()), structType)); @@ -268,6 +271,7 @@ private static Value convertField(InternalRow row, int index, DataType type) { final ArrayData arrayData = row.getArray(index); final int numElements = arrayData.numElements(); + if (numElements == 0) return Value.structArray(Type.struct(), null); List convertedList = new ArrayList<>(numElements); for (int j = 0; j < numElements; j++) { @@ -280,10 +284,7 @@ private static Value convertField(InternalRow row, int index, DataType type) { } } - return Value.structArray( - Type.struct(Type.StructField.of("field", Type.string())), - convertedList); // TODO: the struct element type should represent the schema of the - // struct. + return Value.structArray(convertedList.get(0).getType(), convertedList); } } diff --git a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java index a657a08a..b4992fe2 100644 --- a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java +++ b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java @@ -355,23 +355,31 @@ public static Collection data() { { "struct_array", new StructType() - .add( - "field", - DataTypes.StringType), // Add at least one field to match getStruct(i, 1) + .add("str_field", DataTypes.StringType) + .add("bool_field", DataTypes.BooleanType), new InternalRow[] { mock(InternalRow.class) }, // Use InternalRow instead of Spanner Struct Value.structArray( - Type.struct(Type.StructField.of("field", Type.string())), - Collections.singletonList(Struct.newBuilder().set("field").to("value").build())), + Type.struct( + Type.StructField.of("str_field", Type.string()), + Type.StructField.of("bool_field", Type.bool())), + Collections.singletonList( + Struct.newBuilder() + .set("str_field") + .to("str_value") + .set("bool_field") + .to(true) + .build())), (BiConsumer) (ad, d) -> setupArrayMock( ad, (InternalRow[]) d, // Cast to InternalRow array (i, val) -> { - when(ad.getStruct(i, 1)).thenReturn((InternalRow) val); - when(((InternalRow) val).getString(0)).thenReturn("value"); + when(ad.getStruct(i, 2)).thenReturn((InternalRow) val); + when(((InternalRow) val).getString(0)).thenReturn("str_value"); + when(((InternalRow) val).getBoolean(1)).thenReturn(true); }) } }); @@ -397,7 +405,7 @@ public void testArrayConversion() { InternalRow row = mock(InternalRow.class); ArrayData arrayData = mock(ArrayData.class); - when(row.isNullAt(0)).thenReturn(false).thenReturn(false); + when(row.isNullAt(0)).thenReturn(false); when(row.getArray(0)).thenReturn(arrayData); // Executes the specific stubbing (e.g., ad.toLongArray(), ad.toIntArray(), ad.to) From 547383f178dbfd00efaffa4defe0176da7264698 Mon Sep 17 00:00:00 2001 From: stevelordbq Date: Tue, 17 Feb 2026 16:22:12 -0800 Subject: [PATCH 4/6] Spanner Write - assume Spark stores strings at UTF8String instead of String, as per Read. --- .../spark/spanner/SpannerWriterUtils.java | 3 +- .../spark/spanner/SpannerWriterUtilsTest.java | 31 ++++++++++++++++--- 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java index be060da0..ea38d16b 100644 --- a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java +++ b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java @@ -57,7 +57,8 @@ private interface FieldConverter { // String TYPE_CONVERTERS.put( DataTypes.StringType, - (row, i, type) -> row.isNullAt(i) ? Value.string(null) : Value.string(row.getString(i))); + (row, i, type) -> + row.isNullAt(i) ? Value.string(null) : Value.string(row.getUTF8String(i).toString())); // Boolean TYPE_CONVERTERS.put( diff --git a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java index b4992fe2..f8fcbe54 100644 --- a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java +++ b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java @@ -31,6 +31,7 @@ import java.util.Collections; import java.util.function.BiConsumer; import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; import org.apache.spark.sql.catalyst.util.ArrayData; import org.apache.spark.sql.types.DataType; import org.apache.spark.sql.types.DataTypes; @@ -58,6 +59,8 @@ public class SpannerWriterUtilsTest { @RunWith(Parameterized.class) public static class ScalarTests { private static final byte[] BYTE_DATA = {95, -10, 127}; + private static final Object[] structValues = + new Object[] {UTF8String.fromString("str_value"), true}; // Parameters for each test case: [ColumnName, DataType, MockValue, ExpectedSpannerValue] @Parameters(name = "Testing {0}") @@ -65,7 +68,7 @@ public static Collection data() { return Arrays.asList( new Object[][] { {"long", DataTypes.LongType, 100L, Value.int64(100L)}, - {"string", DataTypes.StringType, "Hello", Value.string("Hello")}, + {"string", DataTypes.StringType, UTF8String.fromString("Hello"), Value.string("Hello")}, {"boolean", DataTypes.BooleanType, true, Value.bool(true)}, {"double", DataTypes.DoubleType, 95.5, Value.float64(95.5)}, {"binary", DataTypes.BinaryType, BYTE_DATA, Value.bytes(ByteArray.copyFrom(BYTE_DATA))}, @@ -76,7 +79,24 @@ public static Collection data() { Value.timestamp(Timestamp.ofTimeMicroseconds(1704067200000000L)) }, {"dt", DataTypes.DateType, 19723, Value.date(Date.fromYearMonthDay(2024, 1, 1))}, - {"decimal", new DecimalType(38, 9), decimal, Value.numeric(jbd)} + {"decimal", new DecimalType(38, 9), decimal, Value.numeric(jbd)}, + { + "struct", + new StructType() + .add("str_field", DataTypes.StringType) + .add("bool_field", DataTypes.BooleanType), + new GenericInternalRow(structValues), + Value.struct( + Type.struct( + Type.StructField.of("str_field", Type.string()), + Type.StructField.of("bool_field", Type.bool())), + Struct.newBuilder() + .set("str_field") + .to("str_value") + .set("bool_field") + .to(true) + .build()) + } }); } @@ -104,7 +124,7 @@ public void testScalarConversion() { // Setup specific getter based on type if (sparkType == DataTypes.LongType) when(row.getLong(0)).thenReturn((Long) mockValue); else if (sparkType == DataTypes.StringType) - when(row.getString(0)).thenReturn((String) mockValue); + when(row.getUTF8String(0)).thenReturn((UTF8String) mockValue); else if (sparkType == DataTypes.BooleanType) when(row.getBoolean(0)).thenReturn((Boolean) mockValue); else if (sparkType == DataTypes.DoubleType) @@ -116,6 +136,8 @@ else if (sparkType == DataTypes.TimestampType) else if (sparkType == DataTypes.DateType) when(row.getInt(0)).thenReturn((int) mockValue); else if (sparkType instanceof DecimalType) when(row.getDecimal(0, 38, 9)).thenReturn((Decimal) mockValue); + else if (sparkType instanceof StructType) + when(row.getStruct(0, 2)).thenReturn((InternalRow) mockValue); // 3. Execute com.google.cloud.spanner.Mutation mutation = @@ -378,7 +400,8 @@ public static Collection data() { (InternalRow[]) d, // Cast to InternalRow array (i, val) -> { when(ad.getStruct(i, 2)).thenReturn((InternalRow) val); - when(((InternalRow) val).getString(0)).thenReturn("str_value"); + when(((InternalRow) val).getUTF8String(0)) + .thenReturn(UTF8String.fromString("str_value")); when(((InternalRow) val).getBoolean(1)).thenReturn(true); }) } From 0e7d5b89ae9702c72f49f8f45a4bcc982932f9e9 Mon Sep 17 00:00:00 2001 From: stevelordbq Date: Wed, 18 Feb 2026 09:16:41 -0800 Subject: [PATCH 5/6] Spanner Write - add types to struct test --- .../spark/spanner/SpannerWriterUtilsTest.java | 41 +++++++++++++++++-- 1 file changed, 37 insertions(+), 4 deletions(-) diff --git a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java index f8fcbe54..2de146b2 100644 --- a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java +++ b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java @@ -60,7 +60,16 @@ public class SpannerWriterUtilsTest { public static class ScalarTests { private static final byte[] BYTE_DATA = {95, -10, 127}; private static final Object[] structValues = - new Object[] {UTF8String.fromString("str_value"), true}; + new Object[] { + 100L, + UTF8String.fromString("str_value"), + true, + 95.5, + BYTE_DATA, + 1704067200000000L, + 19723, + decimal + }; // Parameters for each test case: [ColumnName, DataType, MockValue, ExpectedSpannerValue] @Parameters(name = "Testing {0}") @@ -83,18 +92,42 @@ public static Collection data() { { "struct", new StructType() + .add("long_field", DataTypes.LongType) .add("str_field", DataTypes.StringType) - .add("bool_field", DataTypes.BooleanType), + .add("bool_field", DataTypes.BooleanType) + .add("double_field", DataTypes.DoubleType) + .add("binary_field", DataTypes.BinaryType) + .add("ts_field", DataTypes.TimestampType) + .add("dt_field", DataTypes.DateType) + .add("decimal_field", new DecimalType(38, 9)), // TODO: Add Struct new GenericInternalRow(structValues), Value.struct( Type.struct( + Type.StructField.of("long_field", Type.int64()), Type.StructField.of("str_field", Type.string()), - Type.StructField.of("bool_field", Type.bool())), + Type.StructField.of("bool_field", Type.bool()), + Type.StructField.of("double_field", Type.float64()), + Type.StructField.of("binary_field", Type.bytes()), + Type.StructField.of("ts_field", Type.timestamp()), + Type.StructField.of("dt_field", Type.date()), + Type.StructField.of("decimal_field", Type.numeric())), Struct.newBuilder() + .set("long_field") + .to(100L) .set("str_field") .to("str_value") .set("bool_field") .to(true) + .set("double_field") + .to(95.5) + .set("binary_field") + .to(ByteArray.copyFrom(BYTE_DATA)) + .set("ts_field") + .to(Timestamp.ofTimeMicroseconds(1704067200000000L)) + .set("dt_field") + .to(Date.fromYearMonthDay(2024, 1, 1)) + .set("decimal_field") + .to(jbd) .build()) } }); @@ -137,7 +170,7 @@ else if (sparkType == DataTypes.TimestampType) else if (sparkType instanceof DecimalType) when(row.getDecimal(0, 38, 9)).thenReturn((Decimal) mockValue); else if (sparkType instanceof StructType) - when(row.getStruct(0, 2)).thenReturn((InternalRow) mockValue); + when(row.getStruct(0, 8)).thenReturn((InternalRow) mockValue); // 3. Execute com.google.cloud.spanner.Mutation mutation = From e893f55ae83248aeb14d2bbd3af3e67c8541c7c3 Mon Sep 17 00:00:00 2001 From: stevelordbq Date: Wed, 18 Feb 2026 11:59:01 -0800 Subject: [PATCH 6/6] Spanner Write - PR comments --- .../spark/spanner/SpannerWriterUtils.java | 16 +- .../spark/spanner/SpannerWriterUtilsTest.java | 142 ++++++++---------- 2 files changed, 74 insertions(+), 84 deletions(-) diff --git a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java index ea38d16b..6b1bf851 100644 --- a/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java +++ b/spark-3.1-spanner-lib/src/main/java/com/google/cloud/spark/spanner/SpannerWriterUtils.java @@ -26,6 +26,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.function.Function; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.util.ArrayData; @@ -238,9 +239,11 @@ private static Value convertField(InternalRow row, int index, DataType type) { } if (type instanceof StructType) { + // A Struct instance in Spanner always represents a non-NULL value. if (row.isNullAt(index)) { - return Value.struct(Struct.newBuilder().build()); + return Value.struct(Struct.newBuilder().build()); // TODO: How should this be handled. } + StructType structType = (StructType) type; return Value.struct( internalRowToStruct(row.getStruct(index, structType.length()), structType)); @@ -272,7 +275,7 @@ private static Value convertField(InternalRow row, int index, DataType type) { final ArrayData arrayData = row.getArray(index); final int numElements = arrayData.numElements(); - if (numElements == 0) return Value.structArray(Type.struct(), null); + if (numElements == 0) return Value.structArray(Type.struct(), new ArrayList<>()); List convertedList = new ArrayList<>(numElements); for (int j = 0; j < numElements; j++) { @@ -285,7 +288,14 @@ private static Value convertField(InternalRow row, int index, DataType type) { } } - return Value.structArray(convertedList.get(0).getType(), convertedList); + // Find Spanner schema from first non-null struct array element. + Type itemType = + convertedList.stream() + .filter(Objects::nonNull) + .findFirst() + .map(Struct::getType) + .orElse(Type.struct()); + return Value.structArray(itemType, convertedList); } } diff --git a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java index 2de146b2..aeb3f902 100644 --- a/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java +++ b/spark-3.1-spanner-lib/src/test/java/com/google/cloud/spark/spanner/SpannerWriterUtilsTest.java @@ -55,21 +55,61 @@ public class SpannerWriterUtilsTest { new java.math.BigDecimal("135.791", mc).setScale(9, RoundingMode.HALF_UP); private static final scala.math.BigDecimal bd = scala.math.BigDecimal$.MODULE$.apply(jbd); private static final Decimal decimal = new Decimal().set(bd); + private static final byte[] BYTE_DATA = {95, -10, 127}; + private static final StructType structType = + new StructType() + .add("long_field", DataTypes.LongType) + .add("str_field", DataTypes.StringType) + .add("bool_field", DataTypes.BooleanType) + .add("double_field", DataTypes.DoubleType) + .add("binary_field", DataTypes.BinaryType) + .add("ts_field", DataTypes.TimestampType) + .add("dt_field", DataTypes.DateType) + .add("decimal_field", new DecimalType(38, 9)); + private static final Type typeStruct = + Type.struct( + Type.StructField.of("long_field", Type.int64()), + Type.StructField.of("str_field", Type.string()), + Type.StructField.of("bool_field", Type.bool()), + Type.StructField.of("double_field", Type.float64()), + Type.StructField.of("binary_field", Type.bytes()), + Type.StructField.of("ts_field", Type.timestamp()), + Type.StructField.of("dt_field", Type.date()), + Type.StructField.of("decimal_field", Type.numeric())); + private static final Struct testStruct = + Struct.newBuilder() + .set("long_field") + .to(100L) + .set("str_field") + .to("str_value") + .set("bool_field") + .to(true) + .set("double_field") + .to(95.5) + .set("binary_field") + .to(ByteArray.copyFrom(BYTE_DATA)) + .set("ts_field") + .to(Timestamp.ofTimeMicroseconds(1704067200000000L)) + .set("dt_field") + .to(Date.fromYearMonthDay(2024, 1, 1)) + .set("decimal_field") + .to(jbd) + .build(); + private static final Object[] structValues = + new Object[] { + 100L, + UTF8String.fromString("str_value"), + true, + 95.5, + BYTE_DATA, + 1704067200000000L, + 19723, + decimal + }; + private static final InternalRow testInternalRow = new GenericInternalRow(structValues); @RunWith(Parameterized.class) public static class ScalarTests { - private static final byte[] BYTE_DATA = {95, -10, 127}; - private static final Object[] structValues = - new Object[] { - 100L, - UTF8String.fromString("str_value"), - true, - 95.5, - BYTE_DATA, - 1704067200000000L, - 19723, - decimal - }; // Parameters for each test case: [ColumnName, DataType, MockValue, ExpectedSpannerValue] @Parameters(name = "Testing {0}") @@ -89,47 +129,7 @@ public static Collection data() { }, {"dt", DataTypes.DateType, 19723, Value.date(Date.fromYearMonthDay(2024, 1, 1))}, {"decimal", new DecimalType(38, 9), decimal, Value.numeric(jbd)}, - { - "struct", - new StructType() - .add("long_field", DataTypes.LongType) - .add("str_field", DataTypes.StringType) - .add("bool_field", DataTypes.BooleanType) - .add("double_field", DataTypes.DoubleType) - .add("binary_field", DataTypes.BinaryType) - .add("ts_field", DataTypes.TimestampType) - .add("dt_field", DataTypes.DateType) - .add("decimal_field", new DecimalType(38, 9)), // TODO: Add Struct - new GenericInternalRow(structValues), - Value.struct( - Type.struct( - Type.StructField.of("long_field", Type.int64()), - Type.StructField.of("str_field", Type.string()), - Type.StructField.of("bool_field", Type.bool()), - Type.StructField.of("double_field", Type.float64()), - Type.StructField.of("binary_field", Type.bytes()), - Type.StructField.of("ts_field", Type.timestamp()), - Type.StructField.of("dt_field", Type.date()), - Type.StructField.of("decimal_field", Type.numeric())), - Struct.newBuilder() - .set("long_field") - .to(100L) - .set("str_field") - .to("str_value") - .set("bool_field") - .to(true) - .set("double_field") - .to(95.5) - .set("binary_field") - .to(ByteArray.copyFrom(BYTE_DATA)) - .set("ts_field") - .to(Timestamp.ofTimeMicroseconds(1704067200000000L)) - .set("dt_field") - .to(Date.fromYearMonthDay(2024, 1, 1)) - .set("decimal_field") - .to(jbd) - .build()) - } + {"struct", structType, testInternalRow, Value.struct(typeStruct, testStruct)} }); } @@ -170,7 +170,8 @@ else if (sparkType == DataTypes.TimestampType) else if (sparkType instanceof DecimalType) when(row.getDecimal(0, 38, 9)).thenReturn((Decimal) mockValue); else if (sparkType instanceof StructType) - when(row.getStruct(0, 8)).thenReturn((InternalRow) mockValue); + when(row.getStruct(0, ((StructType) sparkType).length())) + .thenReturn((InternalRow) mockValue); // 3. Execute com.google.cloud.spanner.Mutation mutation = @@ -200,7 +201,7 @@ public static Collection data() { {"timestamp", DataTypes.TimestampType, Value.timestamp(null)}, {"date", DataTypes.DateType, Value.date(null)}, {"decimal", new DecimalType(38, 9), Value.numeric(null)}, - {"struct", new StructType(), Value.struct(Struct.newBuilder().build())}, + {"struct", new StructType(), Value.struct(Struct.newBuilder().build())}, // { { "long_array", DataTypes.createArrayType(DataTypes.LongType), @@ -409,33 +410,17 @@ public static Collection data() { // 9. Struct Array { "struct_array", - new StructType() - .add("str_field", DataTypes.StringType) - .add("bool_field", DataTypes.BooleanType), - new InternalRow[] { - mock(InternalRow.class) - }, // Use InternalRow instead of Spanner Struct - Value.structArray( - Type.struct( - Type.StructField.of("str_field", Type.string()), - Type.StructField.of("bool_field", Type.bool())), - Collections.singletonList( - Struct.newBuilder() - .set("str_field") - .to("str_value") - .set("bool_field") - .to(true) - .build())), + structType, + new InternalRow[] {testInternalRow}, // Use InternalRow instead of Spanner Struct + Value.structArray(typeStruct, Collections.singletonList(testStruct)), (BiConsumer) (ad, d) -> setupArrayMock( ad, (InternalRow[]) d, // Cast to InternalRow array (i, val) -> { - when(ad.getStruct(i, 2)).thenReturn((InternalRow) val); - when(((InternalRow) val).getUTF8String(0)) - .thenReturn(UTF8String.fromString("str_value")); - when(((InternalRow) val).getBoolean(1)).thenReturn(true); + when(ad.getStruct(i, structType.length())) + .thenReturn((InternalRow) val); }) } }); @@ -473,9 +458,4 @@ public void testArrayConversion() { "Failure on type: " + colName, expectedValue, mutation.asMap().get(colName)); } } - - @Test - public void testStructConversion() { - // TODO: add test - } }