From 441fb29d2bcc08c445a937954f1616340914d281 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:16:18 +0200 Subject: [PATCH 01/13] Add reproducer tests for Row.Equals.deepHashCode on logical types and RowWithGetters equals/hashCode --- .../beam/sdk/schemas/JavaFieldSchemaTest.java | 31 ++++++++++++ .../org/apache/beam/sdk/values/RowTest.java | 47 +++++++++++++++++++ 2 files changed, 78 insertions(+) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java index d4f94f021c33..939ba45d99cd 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaFieldSchemaTest.java @@ -1071,4 +1071,35 @@ public void testGenericBeamSchemaArrayFromRow() throws Exception { assertEquals("v1", pojos[0].t); assertEquals("v2", pojos[1].t); } + + @Test + public void testRowWithGettersEqualsAndHashCodeContract() throws NoSuchSchemaException { + SchemaRegistry registry = SchemaRegistry.createDefault(); + SimplePOJO pojo1 = createSimple("string"); + SimplePOJO pojo2 = createSimple("string"); + Row rowWithGetters1 = registry.getToRowFunction(SimplePOJO.class).apply(pojo1); + Row rowWithGetters2 = registry.getToRowFunction(SimplePOJO.class).apply(pojo2); + Row rowWithStorage = + Row.withSchema(rowWithGetters1.getSchema()) + .addValues( + "string", + (byte) 1, + (short) 2, + 3, + 4L, + true, + DATE, + INSTANT, + BYTE_ARRAY, + BYTE_BUFFER.array(), + BigDecimal.ONE, + "stringbuilder") + .build(); + + assertEquals(rowWithGetters1, rowWithGetters2); + assertEquals(rowWithGetters1.hashCode(), rowWithGetters2.hashCode()); + assertEquals(rowWithStorage, rowWithGetters1); + assertEquals(rowWithGetters1, rowWithStorage); + assertEquals(rowWithStorage.hashCode(), rowWithGetters1.hashCode()); + } } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java index 95d43aed1703..d0a07322cb2a 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java @@ -896,4 +896,51 @@ public void testToCamelCase() { assertEquals(expectedCamelCaseInnerRow, row.toCamelCase().getRow("myRowField")); assertEquals(expectedCamelCaseRow, row.toCamelCase()); } + + @Test + public void testLogicalTypeBytesEqualsAndHashCode() { + Schema.LogicalType wrappedBytesType = + new Schema.LogicalType() { + @Override + public String getIdentifier() { + return "wrapped_bytes"; + } + + @Override + public FieldType getArgumentType() { + return FieldType.STRING; + } + + @Override + public String getArgument() { + return ""; + } + + @Override + public FieldType getBaseType() { + return FieldType.BYTES; + } + + @Override + public byte[] toBaseType(ByteBuffer input) { + byte[] bytes = new byte[input.remaining()]; + input.duplicate().get(bytes); + return bytes; + } + + @Override + public ByteBuffer toInputType(byte[] base) { + return ByteBuffer.wrap(base); + } + }; + + Schema schema = Schema.builder().addLogicalTypeField("f_bytes", wrappedBytesType).build(); + Row row1 = + Row.withSchema(schema).addValue(ByteBuffer.wrap(new byte[] {1, 2, 3, 4})).build(); + Row row2 = + Row.withSchema(schema).addValue(ByteBuffer.wrap(new byte[] {1, 2, 3, 4})).build(); + + assertEquals(row1, row2); + assertEquals(row1.hashCode(), row2.hashCode()); + } } From b6af710a99690643a66fb864d2913d05c64189ac Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:16:18 +0200 Subject: [PATCH 02/13] Fix Row.Equals.deepHashCode for logical types and RowWithGetters equals/hashCode contract --- .../src/main/java/org/apache/beam/sdk/values/Row.java | 5 ++++- .../org/apache/beam/sdk/values/RowWithGetters.java | 11 +++++++---- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java index 6eb063f84b67..7148c9f9abad 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java @@ -486,7 +486,10 @@ public static int deepHashCode(Object a, Schema.FieldType fieldType) { if (a == null) { return 0; } else if (fieldType.getTypeName() == TypeName.LOGICAL_TYPE) { - return deepHashCode(a, fieldType.getLogicalType().getBaseType()); + Schema.LogicalType logicalType = + (Schema.LogicalType) fieldType.getLogicalType(); + return deepHashCode( + SchemaUtils.toLogicalBaseType(logicalType, a), logicalType.getBaseType()); } else if (fieldType.getTypeName() == Schema.TypeName.BYTES) { return Arrays.hashCode((byte[]) a); } else if (fieldType.getTypeName().isCollectionType()) { diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java index 0cbc7bfb992a..17e4b7c2bca2 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java @@ -140,9 +140,12 @@ public boolean equals(@Nullable Object o) { } if (o instanceof RowWithGetters) { RowWithGetters other = (RowWithGetters) o; - return Objects.equals(getSchema(), other.getSchema()) - && Objects.equals(getterTarget, other.getterTarget); - } else if (o instanceof Row) { + if (Objects.equals(getSchema(), other.getSchema()) + && Objects.equals(getterTarget, other.getterTarget)) { + return true; + } + } + if (o instanceof Row) { return super.equals(o); } return false; @@ -150,6 +153,6 @@ public boolean equals(@Nullable Object o) { @Override public int hashCode() { - return Objects.hash(getSchema(), getterTarget); + return super.hashCode(); } } From 340f9ca9021427f03dad702b81856c1d91a5fa22 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:16:22 +0200 Subject: [PATCH 03/13] Add reproducer tests for nested Row schema validation in RowUtils --- .../org/apache/beam/sdk/values/RowTest.java | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java index d0a07322cb2a..a0366b6b4940 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java @@ -943,4 +943,27 @@ public ByteBuffer toInputType(byte[] base) { assertEquals(row1, row2); assertEquals(row1.hashCode(), row2.hashCode()); } + + @Test + public void testRejectsMismatchedNestedRowSchema() { + Schema expectedNestedSchema = Schema.builder().addStringField("f_str").build(); + Schema wrongNestedSchema = Schema.builder().addInt32Field("f_int").build(); + Schema parentSchema = Schema.builder().addRowField("nested", expectedNestedSchema).build(); + + Row wrongNestedRow = Row.withSchema(wrongNestedSchema).addValue(42).build(); + thrown.expect(IllegalArgumentException.class); + Row.withSchema(parentSchema).addValue(wrongNestedRow).build(); + } + + @Test + public void testRejectsMismatchedRowInArraySchema() { + Schema expectedNestedSchema = Schema.builder().addStringField("f_str").build(); + Schema wrongNestedSchema = Schema.builder().addInt32Field("f_int").build(); + Schema parentSchema = + Schema.builder().addArrayField("items", FieldType.row(expectedNestedSchema)).build(); + + Row wrongNestedRow = Row.withSchema(wrongNestedSchema).addValue(42).build(); + thrown.expect(IllegalArgumentException.class); + Row.withSchema(parentSchema).addArray(ImmutableList.of(wrongNestedRow)).build(); + } } From 57941fa5892d4854d97a5def700b3594b044d7fc Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:16:22 +0200 Subject: [PATCH 04/13] Validate nested Row schema when setting ROW fields in RowUtils --- .../src/main/java/org/apache/beam/sdk/values/RowUtils.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java index 978575e6b9db..95838b05dddc 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java @@ -423,6 +423,12 @@ public Row processRow( } retValue = new RowWithStorage(schema, values); } + if (retValue != null && !schema.equals(retValue.getSchema())) { + throw new IllegalArgumentException( + String.format( + "Expected schema %s for field %s, but got %s", + schema, rowPosition.descriptor, retValue.getSchema())); + } return retValue; } From 9c373a799b41bf96e1173e9b46912a5bf123847b Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:17:41 +0200 Subject: [PATCH 05/13] Add reproducer test for sliced and read-only ByteBuffers in RowUtils --- .../org/apache/beam/sdk/values/RowTest.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java index a0366b6b4940..a594284201d0 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java @@ -966,4 +966,22 @@ public void testRejectsMismatchedRowInArraySchema() { thrown.expect(IllegalArgumentException.class); Row.withSchema(parentSchema).addArray(ImmutableList.of(wrongNestedRow)).build(); } + + @Test + public void testSlicedAndReadOnlyByteBufferEquality() { + byte[] expected = new byte[] {2, 3, 4}; + ByteBuffer sliced = ByteBuffer.wrap(new byte[] {99, 2, 3, 4, 99}, 1, 3).slice(); + ByteBuffer readOnly = ByteBuffer.wrap(expected).asReadOnlyBuffer(); + + Schema schema = Schema.of(Schema.Field.of("bytes", Schema.FieldType.BYTES)); + + Row rowFromBytes = Row.withSchema(schema).addValue(expected).build(); + Row rowFromSlice = Row.withSchema(schema).addValue(sliced).build(); + Row rowFromReadOnly = Row.withSchema(schema).withFieldValue("bytes", readOnly).build(); + + assertEquals(rowFromBytes, rowFromSlice); + assertEquals(rowFromBytes, rowFromReadOnly); + assertTrue(Arrays.equals(expected, rowFromSlice.getBytes(0))); + assertTrue(Arrays.equals(expected, rowFromReadOnly.getBytes(0))); + } } From d5423768bcca883030ff0e39b7c1f5d27f021593 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:17:41 +0200 Subject: [PATCH 06/13] Preserve ByteBuffer position and limit when extracting bytes in RowUtils --- .../org/apache/beam/sdk/values/RowUtils.java | 27 ++++++++++++++----- 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java index 95838b05dddc..941de983afd5 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java @@ -173,11 +173,7 @@ public Object match( processedValue = cases.processByte(rowPosition, (Byte) value, this); break; case BYTES: - processedValue = - cases.processBytes( - rowPosition, - (byte[]) (value instanceof ByteBuffer ? ((ByteBuffer) value).array() : value), - this); + processedValue = cases.processBytes(rowPosition, toByteArray(value), this); break; case INT16: processedValue = cases.processInt16(rowPosition, (Short) value, this); @@ -611,7 +607,26 @@ public Boolean processBoolean(RowPosition rowPosition, Boolean value, RowFieldMa @Override public byte[] processBytes(RowPosition rowPosition, byte[] value, RowFieldMatcher matcher) { Object retValue = overrideOrReturn(rowPosition, value); - return (retValue instanceof ByteBuffer) ? ((ByteBuffer) retValue).array() : (byte[]) retValue; + return toByteArray(retValue); + } + } + + static byte @Nullable [] toByteArray(@Nullable Object value) { + if (value == null) { + return null; + } + if (value instanceof ByteBuffer) { + ByteBuffer buf = (ByteBuffer) value; + if (buf.hasArray() + && buf.arrayOffset() == 0 + && buf.position() == 0 + && buf.limit() == buf.capacity()) { + return buf.array(); + } + byte[] bytes = new byte[buf.remaining()]; + buf.duplicate().get(bytes); + return bytes; } + return (byte[]) value; } } From 2a4a0621f8ede6ac852d9ad29f0a788894f6b8b7 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:18:33 +0200 Subject: [PATCH 07/13] Add reproducer test for DATETIME-based logical types in SchemaCoderHelpers --- .../apache/beam/sdk/coders/RowCoderTest.java | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/RowCoderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/RowCoderTest.java index 7818cba9a818..1abd4cbcf764 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/RowCoderTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/RowCoderTest.java @@ -645,4 +645,47 @@ public void testStaticEncoding() throws Exception { CoderProperties.coderDecodeEncodeEqual(RowCoder.of(schema), row); } + + private static class MillisDateTimeLogicalType + implements LogicalType { + @Override + public String getIdentifier() { + return "millis_datetime"; + } + + @Override + public FieldType getArgumentType() { + return FieldType.STRING; + } + + @Override + public String getArgument() { + return ""; + } + + @Override + public FieldType getBaseType() { + return FieldType.DATETIME; + } + + @Override + public org.joda.time.ReadableInstant toBaseType(Long input) { + return new DateTime(input, DateTimeZone.UTC); + } + + @Override + public Long toInputType(org.joda.time.ReadableInstant base) { + return base.getMillis(); + } + } + + @Test + public void testDateTimeLogicalTypeCoderMethods() throws Exception { + Coder coder = + org.apache.beam.sdk.schemas.SchemaCoder.coderForFieldType( + FieldType.logicalType(new MillisDateTimeLogicalType())); + CoderProperties.coderDecodeEncodeEqual(coder, 123456789L); + CoderProperties.coderConsistentWithEquals(coder, 123456789L, 123456789L); + CoderProperties.structuralValueConsistentWithEquals(coder, 123456789L, 123456789L); + } } From 33b4bdd0bfdefa5a35335490928b2c7591ab0108 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:18:33 +0200 Subject: [PATCH 08/13] Handle null and structural values for DATETIME-based logical types in SchemaCoderHelpers --- .../beam/sdk/schemas/SchemaCoderHelpers.java | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaCoderHelpers.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaCoderHelpers.java index dfc0d82d2145..363372563913 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaCoderHelpers.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaCoderHelpers.java @@ -80,13 +80,20 @@ private static class LogicalTypeCoder extends Coder { this.isDateTime = logicalType.getBaseType().equals(FieldType.DATETIME); } - @Override - public void encode(InputT value, OutputStream outStream) throws CoderException, IOException { + private BaseT toBaseType(InputT value) { + if (value == null) { + return null; + } BaseT baseType = logicalType.toBaseType(value); - if (isDateTime) { + if (isDateTime && baseType != null) { baseType = (BaseT) ((ReadableInstant) baseType).toInstant(); } - baseTypeCoder.encode(baseType, outStream); + return baseType; + } + + @Override + public void encode(InputT value, OutputStream outStream) throws CoderException, IOException { + baseTypeCoder.encode(toBaseType(value), outStream); } @Override @@ -114,22 +121,23 @@ public boolean consistentWithEquals() { @Override public Object structuralValue(InputT value) { + BaseT baseValue = toBaseType(value); if (baseTypeCoder.consistentWithEquals()) { - return logicalType.toBaseType(value); + return baseValue; } else { - return baseTypeCoder.structuralValue(logicalType.toBaseType(value)); + return baseTypeCoder.structuralValue(baseValue); } } @Override public boolean isRegisterByteSizeObserverCheap(InputT value) { - return baseTypeCoder.isRegisterByteSizeObserverCheap(logicalType.toBaseType(value)); + return baseTypeCoder.isRegisterByteSizeObserverCheap(toBaseType(value)); } @Override public void registerByteSizeObserver(InputT value, ElementByteSizeObserver observer) throws Exception { - baseTypeCoder.registerByteSizeObserver(logicalType.toBaseType(value), observer); + baseTypeCoder.registerByteSizeObserver(toBaseType(value), observer); } } From ad142a8aa45eb0bca75250da9d18475b51a00965 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sun, 25 Oct 2026 15:32:36 +0100 Subject: [PATCH 09/13] Apply spotless formatting --- .../src/test/java/org/apache/beam/sdk/values/RowTest.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java index a594284201d0..88603ffcc1a2 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/RowTest.java @@ -935,10 +935,8 @@ public ByteBuffer toInputType(byte[] base) { }; Schema schema = Schema.builder().addLogicalTypeField("f_bytes", wrappedBytesType).build(); - Row row1 = - Row.withSchema(schema).addValue(ByteBuffer.wrap(new byte[] {1, 2, 3, 4})).build(); - Row row2 = - Row.withSchema(schema).addValue(ByteBuffer.wrap(new byte[] {1, 2, 3, 4})).build(); + Row row1 = Row.withSchema(schema).addValue(ByteBuffer.wrap(new byte[] {1, 2, 3, 4})).build(); + Row row2 = Row.withSchema(schema).addValue(ByteBuffer.wrap(new byte[] {1, 2, 3, 4})).build(); assertEquals(row1, row2); assertEquals(row1.hashCode(), row2.hashCode()); From 03bb446a2f741b17120474682e3895bebdaebaa8 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sun, 25 Oct 2026 23:31:48 +0100 Subject: [PATCH 10/13] Fix nested Row schema in RowStringInterpolatorTest --- .../org/apache/beam/sdk/util/RowStringInterpolatorTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowStringInterpolatorTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowStringInterpolatorTest.java index fed55cc4cb89..1b7f2383a90c 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowStringInterpolatorTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/RowStringInterpolatorTest.java @@ -102,7 +102,7 @@ public void testInvalidRowThrowsHelpfulErrorForDoublyNestedFields() { Schema nestedSchema = Schema.builder().addNullableRowField("nested_row", doublyNestedSchema).build(); Row invalidRow = - Row.withSchema(Schema.builder().addNullableRowField("row", doublyNestedSchema).build()) + Row.withSchema(Schema.builder().addNullableRowField("row", nestedSchema).build()) .addValue( Row.withSchema(nestedSchema).addValue(Row.nullRow(doublyNestedSchema)).build()) .build(); From 1af8a39b29e1f83cfcb66d13df132c28f61a2441 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:09:55 +0200 Subject: [PATCH 11/13] Add Row and RowCoder benchmark --- .../sdk/jmh/schemas/RowAndCoderBenchmark.java | 342 ++++++++++++++++++ 1 file changed, 342 insertions(+) create mode 100644 sdks/java/core/jmh/src/main/java/org/apache/beam/sdk/jmh/schemas/RowAndCoderBenchmark.java diff --git a/sdks/java/core/jmh/src/main/java/org/apache/beam/sdk/jmh/schemas/RowAndCoderBenchmark.java b/sdks/java/core/jmh/src/main/java/org/apache/beam/sdk/jmh/schemas/RowAndCoderBenchmark.java new file mode 100644 index 000000000000..c285a0e4caa9 --- /dev/null +++ b/sdks/java/core/jmh/src/main/java/org/apache/beam/sdk/jmh/schemas/RowAndCoderBenchmark.java @@ -0,0 +1,342 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.beam.sdk.jmh.schemas; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.math.BigDecimal; +import java.time.LocalDate; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.schemas.JavaFieldSchema; +import org.apache.beam.sdk.schemas.NoSuchSchemaException; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.schemas.Schema.FieldType; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes; +import org.apache.beam.sdk.util.ByteStringOutputStream; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; +import org.joda.time.Instant; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +/** Benchmarks for {@link Row}, {@link RowCoder}, and {@link SchemaCoder}. */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +@Warmup(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@Fork(1) +public class RowAndCoderBenchmark { + + public static class NestedPojo { + public String nestedStr = "inner"; + public long nestedLong = 999L; + public @Nullable String nestedNullable = null; + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (!(o instanceof NestedPojo)) { + return false; + } + NestedPojo that = (NestedPojo) o; + return nestedLong == that.nestedLong + && Objects.equals(nestedStr, that.nestedStr) + && Objects.equals(nestedNullable, that.nestedNullable); + } + + @Override + public int hashCode() { + return Objects.hash(nestedStr, nestedLong, nestedNullable); + } + } + + public static class ComplexPojo { + public String id = "user-12345"; + public int count = 42; + public double score = 98.6d; + public boolean active = true; + public @Nullable String optionalTag = null; + public List tags = Arrays.asList("alpha", "beta", "gamma", "delta"); + public Map metrics = ImmutableMap.of("k1", 10L, "k2", 20L, "k3", 30L); + public NestedPojo nested = new NestedPojo(); + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (!(o instanceof ComplexPojo)) { + return false; + } + ComplexPojo that = (ComplexPojo) o; + return count == that.count + && Double.compare(that.score, score) == 0 + && active == that.active + && Objects.equals(id, that.id) + && Objects.equals(optionalTag, that.optionalTag) + && Objects.equals(tags, that.tags) + && Objects.equals(metrics, that.metrics) + && Objects.equals(nested, that.nested); + } + + @Override + public int hashCode() { + return Objects.hash(id, count, score, active, optionalTag, tags, metrics, nested); + } + } + + @State(Scope.Benchmark) + public static class BenchmarkState { + Schema flatSchema; + Instant timestamp; + BigDecimal decimal; + Row flatRow; + Row flatRowCopy; + RowCoder flatRowCoder; + byte[] encodedFlatRow; + + Schema nestedSchema; + Schema complexSchema; + LocalDate localDate; + List tags; + Map metrics; + Row nestedRow; + Row complexRow; + Row complexRowCopy; + RowCoder complexRowCoder; + byte[] encodedComplexRow; + + ComplexPojo complexPojo; + SchemaCoder pojoSchemaCoder; + Row rowWithGetters; + byte[] encodedPojo; + + ByteStringOutputStream outStream; + + @Setup + public void setup() throws IOException, NoSuchSchemaException { + timestamp = new DateTime(2021, 11, 18, 8, 45, 38, DateTimeZone.UTC).toInstant(); + decimal = new BigDecimal("123.12"); + + flatSchema = + Schema.builder() + .addByteField("f_byte") + .addInt16Field("f_int16") + .addInt32Field("f_int32") + .addInt64Field("f_int64") + .addFloatField("f_float") + .addDoubleField("f_double") + .addBooleanField("f_boolean") + .addStringField("f_string") + .addDecimalField("f_decimal") + .addDateTimeField("f_timestamp") + .build(); + + flatRow = + Row.withSchema(flatSchema) + .addValues( + (byte) 12, + (short) 22, + 32, + 42L, + 1.02E5f, + 62.2d, + true, + "hello", + decimal, + timestamp) + .build(); + flatRowCopy = + Row.withSchema(flatSchema) + .addValues( + (byte) 12, + (short) 22, + 32, + 42L, + 1.02E5f, + 62.2d, + true, + "hello", + decimal, + timestamp) + .build(); + + flatRowCoder = RowCoder.of(flatSchema); + outStream = new ByteStringOutputStream(256); + flatRowCoder.encode(flatRow, outStream); + encodedFlatRow = outStream.toByteStringAndReset().toByteArray(); + + nestedSchema = + Schema.builder() + .addStringField("nested_str") + .addInt64Field("nested_long") + .addNullableField("nested_nullable", FieldType.STRING) + .build(); + + complexSchema = + Schema.builder() + .addStringField("id") + .addInt32Field("count") + .addDoubleField("score") + .addBooleanField("active") + .addNullableField("optional_tag", FieldType.STRING) + .addLogicalTypeField("date", SqlTypes.DATE) + .addArrayField("tags", FieldType.STRING) + .addMapField("metrics", FieldType.STRING, FieldType.INT64) + .addRowField("nested", nestedSchema) + .build(); + + localDate = LocalDate.of(2025, 10, 24); + tags = Arrays.asList("alpha", "beta", "gamma", "delta"); + metrics = ImmutableMap.of("k1", 10L, "k2", 20L, "k3", 30L); + nestedRow = Row.withSchema(nestedSchema).addValues("inner", 999L, null).build(); + + complexRow = + Row.withSchema(complexSchema) + .addValues( + "user-12345", 42, 98.6d, true, null, localDate, tags, metrics, nestedRow) + .build(); + complexRowCopy = + Row.withSchema(complexSchema) + .addValues( + "user-12345", 42, 98.6d, true, null, localDate, tags, metrics, nestedRow) + .build(); + + complexRowCoder = RowCoder.of(complexSchema); + complexRowCoder.encode(complexRow, outStream); + encodedComplexRow = outStream.toByteStringAndReset().toByteArray(); + + SchemaRegistry registry = SchemaRegistry.createDefault(); + registry.registerSchemaProvider(ComplexPojo.class, new JavaFieldSchema()); + registry.registerSchemaProvider(NestedPojo.class, new JavaFieldSchema()); + complexPojo = new ComplexPojo(); + pojoSchemaCoder = registry.getSchemaCoder(ComplexPojo.class); + rowWithGetters = pojoSchemaCoder.getToRowFunction().apply(complexPojo); + pojoSchemaCoder.encode(complexPojo, outStream); + encodedPojo = outStream.toByteStringAndReset().toByteArray(); + } + } + + @Benchmark + public Row buildFlatRow(BenchmarkState state) { + return Row.withSchema(state.flatSchema) + .addValues( + (byte) 12, + (short) 22, + 32, + 42L, + 1.02E5f, + 62.2d, + true, + "hello", + state.decimal, + state.timestamp) + .build(); + } + + @Benchmark + public Row buildComplexRow(BenchmarkState state) { + Row nested = Row.withSchema(state.nestedSchema).addValues("inner", 999L, null).build(); + return Row.withSchema(state.complexSchema) + .addValues( + "user-12345", + 42, + 98.6d, + true, + null, + state.localDate, + state.tags, + state.metrics, + nested) + .build(); + } + + @Benchmark + public int complexRowEqualsAndHashCode(BenchmarkState state) { + boolean eq = state.complexRow.equals(state.complexRowCopy); + return eq ? state.complexRow.hashCode() : 0; + } + + @Benchmark + public int readRowWithGettersRepeatedly(BenchmarkState state) { + Row r = state.rowWithGetters; + // Read nested row and collection/map fields multiple times on the same RowWithGetters + Row nested1 = r.getRow(7); + Row nested2 = r.getRow(7); + return r.getArray(5).size() + + r.getMap(6).size() + + nested1.getString(0).length() + + nested2.getString(0).length(); + } + + @Benchmark + public int encodeFlatRow(BenchmarkState state) throws IOException { + state.outStream.reset(); + state.flatRowCoder.encode(state.flatRow, state.outStream); + return state.outStream.size(); + } + + @Benchmark + public Row decodeFlatRow(BenchmarkState state) throws IOException { + return state.flatRowCoder.decode(new ByteArrayInputStream(state.encodedFlatRow)); + } + + @Benchmark + public int encodeComplexRow(BenchmarkState state) throws IOException { + state.outStream.reset(); + state.complexRowCoder.encode(state.complexRow, state.outStream); + return state.outStream.size(); + } + + @Benchmark + public Row decodeComplexRow(BenchmarkState state) throws IOException { + return state.complexRowCoder.decode(new ByteArrayInputStream(state.encodedComplexRow)); + } + + @Benchmark + public int encodePojoViaSchemaCoder(BenchmarkState state) throws IOException { + state.outStream.reset(); + state.pojoSchemaCoder.encode(state.complexPojo, state.outStream); + return state.outStream.size(); + } + + @Benchmark + public ComplexPojo decodePojoViaSchemaCoder(BenchmarkState state) throws IOException { + return state.pojoSchemaCoder.decode(new ByteArrayInputStream(state.encodedPojo)); + } +} From 2af3ce644fd03e463a10bd3aeade11583efe4923 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sat, 24 Oct 2026 12:20:47 +0200 Subject: [PATCH 12/13] Optimize Row construction, RowWithGetters field caching, and Row utilities --- .../apache/beam/sdk/schemas/SchemaUtils.java | 37 +- .../java/org/apache/beam/sdk/values/Row.java | 126 ++-- .../org/apache/beam/sdk/values/RowUtils.java | 644 +++++------------- .../beam/sdk/values/RowWithGetters.java | 52 +- .../beam/sdk/values/SchemaVerification.java | 14 +- 5 files changed, 258 insertions(+), 615 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaUtils.java index d45454be90fa..92363dfbae8c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaUtils.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaUtils.java @@ -382,32 +382,31 @@ static String toPrettySchemaString(Schema schema, String prefix) { } static String toPrettyRowString(Row row, String prefix) { - long nonNullFieldCount = row.getValues().stream().filter(Objects::nonNull).count(); - if (nonNullFieldCount == 0) { - return "{}"; - } - - String nextPrefix = prefix + INDENT; - StringBuilder sb = new StringBuilder(); - sb.append("{\n"); - long nonNullFieldIndex = 0; - for (Schema.Field field : row.getSchema().getFields()) { - String fieldName = field.getName(); - Object fieldValue = row.getValue(fieldName); + Schema schema = row.getSchema(); + int fieldCount = schema.getFieldCount(); + String nextPrefix = null; + StringBuilder sb = null; + for (int i = 0; i < fieldCount; i++) { + Object fieldValue = row.getValue(i); if (fieldValue == null) { continue; } + Schema.Field field = schema.getField(i); + if (sb == null) { + nextPrefix = prefix + INDENT; + sb = new StringBuilder("{\n"); + } else { + sb.append(",\n"); + } sb.append(nextPrefix) - .append(fieldName) + .append(field.getName()) .append(": ") .append(toPrettyFieldValueString(field.getType(), fieldValue, nextPrefix)); - if (nonNullFieldIndex++ < nonNullFieldCount - 1) { - sb.append(",\n"); - } else { - sb.append("\n"); - } } - sb.append(prefix).append("}"); + if (sb == null) { + return "{}"; + } + sb.append("\n").append(prefix).append("}"); return sb.toString(); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java index 7148c9f9abad..437a1d6e4585 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java @@ -33,7 +33,6 @@ import java.util.Objects; import java.util.stream.Collector; import java.util.stream.Collectors; -import java.util.stream.IntStream; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.schemas.Factory; import org.apache.beam.sdk.schemas.FieldAccessDescriptor; @@ -42,12 +41,8 @@ import org.apache.beam.sdk.schemas.Schema.FieldType; import org.apache.beam.sdk.schemas.Schema.TypeName; import org.apache.beam.sdk.schemas.SchemaUtils; -import org.apache.beam.sdk.values.RowUtils.CapturingRowCases; import org.apache.beam.sdk.values.RowUtils.FieldOverride; import org.apache.beam.sdk.values.RowUtils.FieldOverrides; -import org.apache.beam.sdk.values.RowUtils.RowFieldMatcher; -import org.apache.beam.sdk.values.RowUtils.RowPosition; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.DateTime; import org.joda.time.ReadableDateTime; @@ -108,9 +103,12 @@ public abstract class Row implements Serializable { /** Return a list of data values. Any LogicalType values are returned as base values. * */ public List getBaseValues() { - return IntStream.range(0, getFieldCount()) - .mapToObj(i -> getBaseValue(i)) - .collect(Collectors.toList()); + int fieldCount = getFieldCount(); + List baseValues = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + baseValues.add(getBaseValue(i)); + } + return baseValues; } /** Get value by field name, {@link ClassCastException} is thrown if type doesn't match. */ @@ -727,25 +725,18 @@ public FieldValueBuilder withFieldAccessDescriptors(Map values = Lists.newArrayList(); + private final List<@Nullable Object> values; private final Schema schema; Builder(Schema schema) { this.schema = schema; + this.values = new ArrayList<>(schema.getFieldCount()); } /** Return the schema for the row being built. */ @@ -800,8 +791,9 @@ public Builder addValues(List<@Nullable Object> values) { return this; } - public Builder addValues(Object... values) { - return addValues(Arrays.asList(values)); + public Builder addValues(@Nullable Object... values) { + Collections.addAll(this.values, values); + return this; } public Builder addArray(Collection values) { @@ -866,16 +858,7 @@ public Row build() { } if (!values.isEmpty()) { - FieldOverrides fieldOverrides = new FieldOverrides(schema, this.values); - if (!fieldOverrides.isEmpty()) { - return (Row) - new RowFieldMatcher() - .match( - new CapturingRowCases(schema, fieldOverrides), - FieldType.row(schema), - new RowPosition(FieldAccessDescriptor.create()), - null); - } + return RowUtils.captureRow(schema, this.values); } return new RowWithStorage(schema, Collections.emptyList()); } @@ -895,57 +878,62 @@ public static Collector, Row> toRow(Schema schema) { /** Creates a new record filled with nulls. */ public static Row nullRow(Schema schema) { - return Row.withSchema(schema) - .addValues(Collections.nCopies(schema.getFieldCount(), null)) - .build(); + int fieldCount = schema.getFieldCount(); + for (int i = 0; i < fieldCount; i++) { + Schema.Field field = schema.getField(i); + if (!field.getType().getNullable()) { + throw new IllegalArgumentException( + String.format("%s is not nullable in field %s", field.getType(), field.getName())); + } + } + return new RowWithStorage(schema, Collections.nCopies(fieldCount, null)); } /** Returns an equivalent {@link Row} with fields lexicographically sorted by their name. */ public Row sorted() { Schema sortedSchema = getSchema().sorted(); - return sortedSchema.getFields().stream() - .map( - field -> { - if (field.getType().getRowSchema() != null) { - Row innerRow = getValue(field.getName()); - if (innerRow != null) { - return innerRow.sorted(); - } - } - return (Object) getValue(field.getName()); - }) - .collect(Row.toRow(sortedSchema)); + int fieldCount = sortedSchema.getFieldCount(); + List<@Nullable Object> sortedValues = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + Schema.Field field = sortedSchema.getField(i); + Object value = getValue(field.getName()); + if (value != null && field.getType().getRowSchema() != null) { + value = ((Row) value).sorted(); + } + sortedValues.add(value); + } + return RowUtils.captureRow(sortedSchema, sortedValues); } /** Returns an equivalent {@link Row} with `snake_case` field names. */ public Row toSnakeCase() { - return getSchema().getFields().stream() - .map( - field -> { - if (field.getType().getRowSchema() != null) { - Row innerRow = getValue(field.getName()); - if (innerRow != null) { - return innerRow.toSnakeCase(); - } - } - return (Object) getValue(field.getName()); - }) - .collect(toRow(getSchema().toSnakeCase())); + Schema snakeCaseSchema = getSchema().toSnakeCase(); + int fieldCount = getSchema().getFieldCount(); + List<@Nullable Object> values = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + Schema.Field field = getSchema().getField(i); + Object value = getValue(i); + if (value != null && field.getType().getRowSchema() != null) { + value = ((Row) value).toSnakeCase(); + } + values.add(value); + } + return RowUtils.captureRow(snakeCaseSchema, values); } /** Returns an equivalent {@link Row} with `lowerCamelCase` field names. */ public Row toCamelCase() { - return getSchema().getFields().stream() - .map( - field -> { - if (field.getType().getRowSchema() != null) { - Row innerRow = getValue(field.getName()); - if (innerRow != null) { - return innerRow.toCamelCase(); - } - } - return (Object) getValue(field.getName()); - }) - .collect(toRow(getSchema().toCamelCase())); + Schema camelCaseSchema = getSchema().toCamelCase(); + int fieldCount = getSchema().getFieldCount(); + List<@Nullable Object> values = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + Schema.Field field = getSchema().getField(i); + Object value = getValue(i); + if (value != null && field.getType().getRowSchema() != null) { + value = ((Row) value).toCamelCase(); + } + values.add(value); + } + return RowUtils.captureRow(camelCaseSchema, values); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java index 941de983afd5..5c164d5a67f4 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java @@ -19,6 +19,7 @@ import java.math.BigDecimal; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.List; @@ -26,9 +27,6 @@ import java.util.Map.Entry; import org.apache.beam.sdk.schemas.FieldAccessDescriptor; import org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor; -import org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor.ListQualifier; -import org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor.MapQualifier; -import org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor.Qualifier; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.Schema.FieldType; import org.apache.beam.sdk.schemas.Schema.LogicalType; @@ -36,7 +34,6 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps; import org.checkerframework.checker.nullness.qual.Nullable; -import org.joda.time.Instant; import org.joda.time.base.AbstractInstant; @SuppressWarnings({ @@ -44,175 +41,151 @@ "rawtypes" }) class RowUtils { - static class RowPosition { - FieldAccessDescriptor descriptor; - List qualifiers; - - RowPosition(FieldAccessDescriptor descriptor) { - this(descriptor, Collections.emptyList()); - } + static Row captureRow(Schema schema, List<@Nullable Object> rawValues) { + int fieldCount = schema.getFieldCount(); + List<@Nullable Object> capturedValues = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + Schema.Field field = schema.getField(i); + capturedValues.add(captureValue(field.getType(), rawValues.get(i), field.getName())); + } + return new RowWithStorage(schema, capturedValues); + } - RowPosition(FieldAccessDescriptor descriptor, List qualifiers) { - this.descriptor = descriptor; - this.qualifiers = qualifiers; + @SuppressWarnings("unchecked") + static @Nullable Object captureValue( + FieldType fieldType, @Nullable Object value, String fieldName) { + if (value == null) { + if (!fieldType.getNullable()) { + throw new IllegalArgumentException( + String.format("%s is not nullable in field %s", fieldType, fieldName)); + } + return null; } - - RowPosition withArrayQualifier() { - List newQualifiers = Lists.newArrayListWithCapacity(qualifiers.size() + 1); - newQualifiers.addAll(qualifiers); - newQualifiers.add(Qualifier.of(ListQualifier.ALL)); - return new RowPosition(descriptor, newQualifiers); + Object processedValue; + switch (fieldType.getTypeName()) { + case ARRAY: + processedValue = + captureIterable( + fieldType.getCollectionElementType(), (Collection) value, fieldName); + break; + case ITERABLE: + processedValue = + captureIterable( + fieldType.getCollectionElementType(), (Iterable) value, fieldName); + break; + case MAP: + processedValue = + captureMap( + fieldType.getMapKeyType(), + fieldType.getMapValueType(), + (Map) value, + fieldName); + break; + case ROW: + Row rowValue = (Row) value; + Schema expectedSchema = fieldType.getRowSchema(); + if (!expectedSchema.equals(rowValue.getSchema())) { + throw new IllegalArgumentException( + String.format( + "Expected schema %s for field %s, but got %s", + expectedSchema, fieldName, rowValue.getSchema())); + } + processedValue = rowValue; + break; + case LOGICAL_TYPE: + LogicalType logicalType = fieldType.getLogicalType(); + processedValue = logicalType.toInputType(logicalType.toBaseType(value)); + break; + case DATETIME: + processedValue = ((AbstractInstant) value).toInstant(); + break; + case BYTE: + processedValue = (Byte) value; + break; + case BYTES: + processedValue = toByteArray(value); + break; + case INT16: + processedValue = (Short) value; + break; + case INT32: + processedValue = (Integer) value; + break; + case INT64: + processedValue = (Long) value; + break; + case DECIMAL: + processedValue = (BigDecimal) value; + break; + case FLOAT: + processedValue = (Float) value; + break; + case DOUBLE: + processedValue = (Double) value; + break; + case STRING: + processedValue = (String) value; + break; + case BOOLEAN: + processedValue = (Boolean) value; + break; + default: + throw new IllegalArgumentException( + String.format("Not a primitive type for field name %s: %s", fieldName, fieldType)); } - - RowPosition withMapQualifier() { - List newQualifiers = Lists.newArrayListWithCapacity(qualifiers.size() + 1); - newQualifiers.addAll(qualifiers); - newQualifiers.add(Qualifier.of(MapQualifier.ALL)); - return new RowPosition(descriptor, newQualifiers); + if (processedValue == null && !fieldType.getNullable()) { + throw new IllegalArgumentException( + String.format("%s is not nullable in field %s", fieldType, fieldName)); } + return processedValue; } - // Subclasses of this interface implement process methods for each schema type. Each process - // method is invoked as - // a RowFieldMatcher walks down the schema tree. The FieldAccessDescriptor passed into each method - // identifies the - // current element of the schema being processed. - interface RowCases { - Row processRow(RowPosition rowPosition, Schema schema, Row value, RowFieldMatcher matcher); - - Collection processArray( - RowPosition rowPosition, - FieldType collectionElementType, - Collection values, - RowFieldMatcher matcher); - - Iterable processIterable( - RowPosition rowPosition, - FieldType collectionElementType, - Iterable values, - RowFieldMatcher matcher); - - Map processMap( - RowPosition rowPosition, - FieldType keyType, - FieldType valueType, - Map valueMap, - RowFieldMatcher matcher); - - Object processLogicalType( - RowPosition rowPosition, LogicalType logicalType, Object baseType, RowFieldMatcher matcher); - - Instant processDateTime( - RowPosition rowPosition, AbstractInstant instant, RowFieldMatcher matcher); - - Byte processByte(RowPosition rowPosition, Byte value, RowFieldMatcher matcher); - - Short processInt16(RowPosition rowPosition, Short value, RowFieldMatcher matcher); - - Integer processInt32(RowPosition rowPosition, Integer value, RowFieldMatcher matcher); - - Long processInt64(RowPosition rowPosition, Long value, RowFieldMatcher matcher); - - BigDecimal processDecimal(RowPosition rowPosition, BigDecimal value, RowFieldMatcher matcher); - - Float processFloat(RowPosition rowPosition, Float value, RowFieldMatcher matcher); - - Double processDouble(RowPosition rowPosition, Double value, RowFieldMatcher matcher); - - String processString(RowPosition rowPosition, String value, RowFieldMatcher matcher); - - Boolean processBoolean(RowPosition rowPosition, Boolean value, RowFieldMatcher matcher); - - byte[] processBytes(RowPosition rowPosition, byte[] value, RowFieldMatcher matcher); - } - - // Given a Row field, delegates processing to the correct process method on the RowCases - // parameter. - static class RowFieldMatcher { - public Object match( - RowCases cases, FieldType fieldType, RowPosition rowPosition, Object value) { - Object processedValue = null; - switch (fieldType.getTypeName()) { - case ARRAY: - processedValue = - cases.processArray( - rowPosition, - fieldType.getCollectionElementType(), - (Collection) value, - this); - break; - case ITERABLE: - processedValue = - cases.processIterable( - rowPosition, - fieldType.getCollectionElementType(), - (Iterable) value, - this); - break; - case MAP: - processedValue = - cases.processMap( - rowPosition, - fieldType.getMapKeyType(), - fieldType.getMapValueType(), - (Map) value, - this); - break; - case ROW: - processedValue = - cases.processRow(rowPosition, fieldType.getRowSchema(), (Row) value, this); - break; - case LOGICAL_TYPE: - LogicalType logicalType = fieldType.getLogicalType(); - processedValue = cases.processLogicalType(rowPosition, logicalType, value, this); - break; - case DATETIME: - processedValue = cases.processDateTime(rowPosition, (AbstractInstant) value, this); - break; - case BYTE: - processedValue = cases.processByte(rowPosition, (Byte) value, this); - break; - case BYTES: - processedValue = cases.processBytes(rowPosition, toByteArray(value), this); - break; - case INT16: - processedValue = cases.processInt16(rowPosition, (Short) value, this); - break; - case INT32: - processedValue = cases.processInt32(rowPosition, (Integer) value, this); - break; - case INT64: - processedValue = cases.processInt64(rowPosition, (Long) value, this); - break; - case DECIMAL: - processedValue = cases.processDecimal(rowPosition, (BigDecimal) value, this); - break; - case FLOAT: - processedValue = cases.processFloat(rowPosition, (Float) value, this); - break; - case DOUBLE: - processedValue = cases.processDouble(rowPosition, (Double) value, this); - break; - case STRING: - processedValue = cases.processString(rowPosition, (String) value, this); - break; - case BOOLEAN: - processedValue = cases.processBoolean(rowPosition, (Boolean) value, this); - break; - default: - // Shouldn't actually get here, but we need this case to satisfy linters. + private static @Nullable Collection captureIterable( + FieldType collectionElementType, @Nullable Iterable values, String fieldName) { + if (values == null) { + return null; + } + List captured = + (values instanceof Collection) + ? new ArrayList<>(((Collection) values).size()) + : new ArrayList<>(); + for (Object listValue : values) { + if (listValue == null) { + if (!collectionElementType.getNullable()) { throw new IllegalArgumentException( String.format( - "Not a primitive type for field name %s: %s", rowPosition.descriptor, fieldType)); + "%s is not nullable in Array field %s", collectionElementType, fieldName)); + } + captured.add(null); + } else { + captured.add(captureValue(collectionElementType, listValue, fieldName)); } - if (processedValue == null) { - if (!fieldType.getNullable()) { + } + return captured; + } + + private static @Nullable Map captureMap( + FieldType keyType, + FieldType valueType, + @Nullable Map valueMap, + String fieldName) { + if (valueMap == null) { + return null; + } + Map retValue = Maps.newHashMapWithExpectedSize(valueMap.size()); + for (Entry kv : valueMap.entrySet()) { + Object capturedKey = captureValue(keyType, kv.getKey(), fieldName); + if (kv.getValue() == null) { + if (!valueType.getNullable()) { throw new IllegalArgumentException( - String.format("%s is not nullable in field %s", fieldType, rowPosition.descriptor)); + String.format("%s is not nullable in Map field %s", valueType, fieldName)); } + retValue.put(capturedKey, null); + } else { + retValue.put(capturedKey, captureValue(valueType, kv.getValue(), fieldName)); } - return processedValue; } + return retValue; } static class FieldOverride { @@ -228,64 +201,31 @@ Object getOverrideValue() { } static class FieldOverrides { - private FieldAccessNode topNode; - private Schema rootSchema; + private final FieldAccessNode topNode; + private final Schema rootSchema; FieldOverrides(Schema rootSchema) { this.topNode = new FieldAccessNode(rootSchema); this.rootSchema = rootSchema; } - FieldOverrides(Schema rootSchema, List overrides) { - this.topNode = new FieldAccessNode(rootSchema, overrides); - this.rootSchema = rootSchema; - } - - boolean isEmpty() { - return topNode.isEmpty(); - } - void addOverride(FieldAccessDescriptor fieldAccessDescriptor, FieldOverride fieldOverride) { topNode.addOverride(fieldAccessDescriptor, fieldOverride, rootSchema); } - void setOverrides(List values) { - List overrides = Lists.newArrayListWithExpectedSize(values.size()); - for (Object value : values) { - overrides.add(new FieldOverride(value)); - } - topNode.setOverrides(overrides); - } - - @Nullable FieldOverride getOverride(FieldAccessDescriptor fieldAccessDescriptor) { - return topNode.getOverride(fieldAccessDescriptor); - } - - boolean hasOverrideBelow(FieldAccessDescriptor fieldAccessDescriptor) { - return topNode.hasOverrideBelow(fieldAccessDescriptor); + Row apply(@Nullable Row sourceRow) { + return topNode.apply(rootSchema, sourceRow); } private static class FieldAccessNode { - List fieldOverrides; - List nestedAccess; + final List fieldOverrides; + final List nestedAccess; FieldAccessNode(Schema schema) { fieldOverrides = Lists.newArrayListWithExpectedSize(schema.getFieldCount()); nestedAccess = Lists.newArrayList(); } - FieldAccessNode(Schema schema, List overrides) { - fieldOverrides = Lists.newArrayListWithExpectedSize(schema.getFieldCount()); - for (Object value : overrides) { - fieldOverrides.add(new FieldOverride(value)); - } - nestedAccess = Lists.newArrayList(); - } - - boolean isEmpty() { - return fieldOverrides.isEmpty() && nestedAccess.isEmpty(); - } - void addOverride( FieldAccessDescriptor fieldAccessDescriptor, FieldOverride fieldOverride, @@ -317,297 +257,29 @@ void addOverride( } } - void setOverrides(List overrides) { - this.fieldOverrides = overrides; - } - - @Nullable FieldOverride getOverride(FieldAccessDescriptor fieldAccessDescriptor) { - FieldOverride override = null; - if (!fieldAccessDescriptor.getFieldsAccessed().isEmpty()) { - FieldDescriptor fieldDescriptor = - Iterables.getOnlyElement(fieldAccessDescriptor.getFieldsAccessed()); - if (fieldDescriptor.getFieldId() < fieldOverrides.size()) { - override = fieldOverrides.get(fieldDescriptor.getFieldId()); - } - } else if (!fieldAccessDescriptor.getNestedFieldsAccessed().isEmpty()) { - Map.Entry entry = - Iterables.getOnlyElement(fieldAccessDescriptor.getNestedFieldsAccessed().entrySet()); - if (entry.getKey().getFieldId() < nestedAccess.size()) { - FieldAccessNode node = nestedAccess.get(entry.getKey().getFieldId()); - if (node != null) { - override = node.getOverride(entry.getValue()); - } - } - } - return override; - } - - boolean hasOverrideBelow(FieldAccessDescriptor fieldAccessDescriptor) { - if (!fieldAccessDescriptor.getFieldsAccessed().isEmpty()) { - FieldDescriptor fieldDescriptor = - Iterables.getOnlyElement(fieldAccessDescriptor.getFieldsAccessed()); - return ((fieldDescriptor.getFieldId() < nestedAccess.size()) - && nestedAccess.get(fieldDescriptor.getFieldId()) != null); - } else if (!fieldAccessDescriptor.getNestedFieldsAccessed().isEmpty()) { - Map.Entry entry = - Iterables.getOnlyElement(fieldAccessDescriptor.getNestedFieldsAccessed().entrySet()); - if (entry.getKey().getFieldId() < nestedAccess.size()) { - FieldAccessNode node = nestedAccess.get(entry.getKey().getFieldId()); - if (node != null) { - return node.hasOverrideBelow(entry.getValue()); - } - } - } else { - return true; - } - return false; - } - } - } - - // This implementation of RowCases captures a Row into a new Row. It also has the effect of - // validating all the - // field parameters. - // A Map of field values can also be passed in, and those field values will be used to override - // the values in the - // passed-in row. - static class CapturingRowCases implements RowCases { - private final Schema topSchema; - private final FieldOverrides fieldOverrides; - - CapturingRowCases(Schema topSchema, FieldOverrides fieldOverrides) { - this.topSchema = topSchema; - this.fieldOverrides = fieldOverrides; - } - - private @Nullable FieldOverride override(RowPosition rowPosition) { - if (!rowPosition.qualifiers.isEmpty()) { - // Currently we only support overriding named schema fields. Individual array/map elements - // or nested collections - // cannot be overriden without overriding the entire schema fields. - return null; - } else { - return fieldOverrides.getOverride(rowPosition.descriptor); - } - } - - private T overrideOrReturn(RowPosition rowPosition, T value) { - FieldOverride fieldOverride = override(rowPosition); - // null return means the item isn't in the map. - return (fieldOverride != null) ? (T) fieldOverride.getOverrideValue() : value; - } - - @Override - public Row processRow( - RowPosition rowPosition, Schema schema, Row value, RowFieldMatcher matcher) { - FieldOverride override = override(rowPosition); - Row retValue = value; - if (override != null) { - retValue = (Row) override.getOverrideValue(); - } else if (fieldOverrides.hasOverrideBelow(rowPosition.descriptor)) { - List values = Lists.newArrayListWithCapacity(schema.getFieldCount()); - for (int i = 0; i < schema.getFieldCount(); ++i) { - FieldAccessDescriptor nestedDescriptor = - FieldAccessDescriptor.withFieldIds(rowPosition.descriptor, i).resolve(topSchema); - Object fieldValue = (value != null) ? value.getValue(i) : null; - values.add( - matcher.match( - this, - schema.getField(i).getType(), - new RowPosition(nestedDescriptor), - fieldValue)); - } - retValue = new RowWithStorage(schema, values); - } - if (retValue != null && !schema.equals(retValue.getSchema())) { - throw new IllegalArgumentException( - String.format( - "Expected schema %s for field %s, but got %s", - schema, rowPosition.descriptor, retValue.getSchema())); - } - return retValue; - } - - @Override - public Collection processArray( - RowPosition rowPosition, - FieldType collectionElementType, - Collection values, - RowFieldMatcher matcher) { - Collection retValue = null; - FieldOverride override = override(rowPosition); - if (override != null) { - retValue = - captureIterable( - rowPosition, - collectionElementType, - (Collection) override.getOverrideValue(), - matcher); - } else if (values != null) { - retValue = captureIterable(rowPosition, collectionElementType, values, matcher); - } - return retValue; - } - - @Override - public Iterable processIterable( - RowPosition rowPosition, - FieldType collectionElementType, - Iterable values, - RowFieldMatcher matcher) { - Iterable retValue = null; - FieldOverride override = override(rowPosition); - if (override != null) { - retValue = - captureIterable( - rowPosition, - collectionElementType, - (Iterable) override.getOverrideValue(), - matcher); - } else if (values != null) { - retValue = captureIterable(rowPosition, collectionElementType, values, matcher); - } - return retValue; - } - - private Collection captureIterable( - RowPosition rowPosition, - FieldType collectionElementType, - Iterable values, - RowFieldMatcher matcher) { - if (values == null) { - return null; - } - - List captured = Lists.newArrayListWithCapacity(Iterables.size(values)); - RowPosition elementPosition = rowPosition.withArrayQualifier(); - for (Object listValue : values) { - if (listValue == null) { - if (!collectionElementType.getNullable()) { - throw new IllegalArgumentException( - String.format( - "%s is not nullable in Array field %s", - collectionElementType, rowPosition.descriptor)); - } - captured.add(null); - } else { - Object capturedElement = - matcher.match(this, collectionElementType, elementPosition, listValue); - captured.add(capturedElement); - } - } - return captured; - } - - @Override - public Map processMap( - RowPosition rowPosition, - FieldType keyType, - FieldType valueType, - Map valueMap, - RowFieldMatcher matcher) { - Map retValue = null; - FieldOverride override = override(rowPosition); - if (override != null) { - valueMap = (Map) override.getOverrideValue(); - } - - if (valueMap != null) { - RowPosition elementPosition = rowPosition.withMapQualifier(); - - retValue = Maps.newHashMapWithExpectedSize(valueMap.size()); - for (Entry kv : valueMap.entrySet()) { - if (kv.getValue() == null) { - if (!valueType.getNullable()) { - throw new IllegalArgumentException( - String.format( - "%s is not nullable in Map field %s", valueType, rowPosition.descriptor)); - } - retValue.put(matcher.match(this, keyType, elementPosition, kv.getKey()), null); + Row apply(Schema schema, @Nullable Row sourceRow) { + int fieldCount = schema.getFieldCount(); + List<@Nullable Object> values = new ArrayList<>(fieldCount); + for (int i = 0; i < fieldCount; i++) { + Schema.Field field = schema.getField(i); + FieldType fieldType = field.getType(); + FieldOverride override = (i < fieldOverrides.size()) ? fieldOverrides.get(i) : null; + if (override != null) { + values.add(captureValue(fieldType, override.getOverrideValue(), field.getName())); } else { - retValue.put( - matcher.match(this, keyType, elementPosition, kv.getKey()), - matcher.match(this, valueType, elementPosition, kv.getValue())); + FieldAccessNode nestedNode = (i < nestedAccess.size()) ? nestedAccess.get(i) : null; + if (nestedNode != null) { + Row nestedSourceRow = (sourceRow != null) ? sourceRow.getValue(i) : null; + Row capturedRow = nestedNode.apply(fieldType.getRowSchema(), nestedSourceRow); + values.add(captureValue(fieldType, capturedRow, field.getName())); + } else { + Object fieldValue = (sourceRow != null) ? sourceRow.getValue(i) : null; + values.add(captureValue(fieldType, fieldValue, field.getName())); + } } } + return new RowWithStorage(schema, values); } - return retValue; - } - - @Override - public Object processLogicalType( - RowPosition rowPosition, LogicalType logicalType, Object value, RowFieldMatcher matcher) { - Object retValue = null; - FieldOverride override = override(rowPosition); - if (override != null) { - if (override.getOverrideValue() == null) { - // This value is supposed to be overridden with null - return null; - } - retValue = logicalType.toInputType(logicalType.toBaseType(override.getOverrideValue())); - } else if (value != null) { - retValue = logicalType.toInputType(logicalType.toBaseType(value)); - } - return retValue; - } - - @Override - public Instant processDateTime( - RowPosition rowPosition, AbstractInstant value, RowFieldMatcher matcher) { - AbstractInstant instantValue = overrideOrReturn(rowPosition, value); - return (instantValue != null) ? instantValue.toInstant() : null; - } - - @Override - public Byte processByte(RowPosition rowPosition, Byte value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public Short processInt16(RowPosition rowPosition, Short value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public Integer processInt32(RowPosition rowPosition, Integer value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public Long processInt64(RowPosition rowPosition, Long value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public BigDecimal processDecimal( - RowPosition rowPosition, BigDecimal value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public Float processFloat(RowPosition rowPosition, Float value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public Double processDouble(RowPosition rowPosition, Double value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public String processString(RowPosition rowPosition, String value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public Boolean processBoolean(RowPosition rowPosition, Boolean value, RowFieldMatcher matcher) { - return overrideOrReturn(rowPosition, value); - } - - @Override - public byte[] processBytes(RowPosition rowPosition, byte[] value, RowFieldMatcher matcher) { - Object retValue = overrideOrReturn(rowPosition, value); - return toByteArray(retValue); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java index 17e4b7c2bca2..c829a562d249 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java @@ -21,16 +21,12 @@ import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.Objects; -import java.util.TreeMap; -import java.util.function.Function; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.schemas.Factory; import org.apache.beam.sdk.schemas.FieldValueGetter; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.Schema.Field; -import org.apache.beam.sdk.schemas.Schema.TypeName; import org.checkerframework.checker.nullness.qual.NonNull; import org.checkerframework.checker.nullness.qual.Nullable; @@ -43,10 +39,12 @@ */ @SuppressWarnings("rawtypes") public class RowWithGetters extends Row { + private static final Object NULL_VALUE = new Object(); + private final T getterTarget; private final TypeDescriptor getterTargetType; private final List> getters; - private @Nullable Map cache = null; + private transient @Nullable Object @Nullable [] cache = null; RowWithGetters( Schema schema, @@ -63,28 +61,20 @@ public class RowWithGetters extends Row { @SuppressWarnings({"TypeParameterUnusedInFormals", "unchecked"}) public W getValue(int fieldIdx) { Field field = getSchema().getField(fieldIdx); - boolean cacheField = cacheFieldType(field); - - if (cacheField && cache == null) { - cache = new TreeMap<>(); - } - @Nullable Object fieldValue; - if (cacheField) { + if (cacheFieldType(field)) { if (cache == null) { - cache = new TreeMap<>(); + cache = new Object[getters.size()]; + } + Object cached = cache[fieldIdx]; + if (cached == null) { + FieldValueGetter getter = getters.get(fieldIdx); + checkStateNotNull(getter); + fieldValue = getter.get(getterTarget); + cache[fieldIdx] = (fieldValue == null) ? NULL_VALUE : fieldValue; + } else { + fieldValue = (cached == NULL_VALUE) ? null : cached; } - fieldValue = - cache.computeIfAbsent( - fieldIdx, - new Function() { - @Override - public @Nullable Object apply(Integer idx) { - FieldValueGetter getter = getters.get(idx); - checkStateNotNull(getter); - return getter.get(getterTarget); - } - }); } else { fieldValue = getters.get(fieldIdx).get(getterTarget); } @@ -100,10 +90,16 @@ public TypeDescriptor getGetterTargetType() { } private boolean cacheFieldType(Field field) { - TypeName typeName = field.getType().getTypeName(); - return typeName.equals(TypeName.MAP) - || typeName.equals(TypeName.ARRAY) - || typeName.equals(TypeName.ITERABLE); + switch (field.getType().getTypeName()) { + case MAP: + case ARRAY: + case ITERABLE: + case ROW: + case LOGICAL_TYPE: + return true; + default: + return false; + } } @Override diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java index 848a86d55425..fe635e20f3ce 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java @@ -18,22 +18,10 @@ package org.apache.beam.sdk.values; import java.io.Serializable; -import org.apache.beam.sdk.schemas.FieldAccessDescriptor; -import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.Schema.FieldType; -import org.apache.beam.sdk.values.RowUtils.CapturingRowCases; -import org.apache.beam.sdk.values.RowUtils.FieldOverrides; -import org.apache.beam.sdk.values.RowUtils.RowFieldMatcher; -import org.apache.beam.sdk.values.RowUtils.RowPosition; public abstract class SchemaVerification implements Serializable { public static Object verifyFieldValue(Object value, FieldType type, String fieldName) { - Schema schema = Schema.builder().addField(fieldName, type).build(); - return new RowFieldMatcher() - .match( - new CapturingRowCases(schema, new FieldOverrides(schema)), - type, - new RowPosition(FieldAccessDescriptor.withFieldIds(0)), - value); + return RowUtils.captureValue(type, value, fieldName); } } From 0b1047867c406a03201517c5f1cef87e46213d87 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Sun, 25 Oct 2026 23:24:10 +0100 Subject: [PATCH 13/13] Fix CheckerFramework nullness errors in SchemaVerification and RowWithGetters --- .../src/main/java/org/apache/beam/sdk/values/RowUtils.java | 5 +++-- .../main/java/org/apache/beam/sdk/values/RowWithGetters.java | 2 ++ .../java/org/apache/beam/sdk/values/SchemaVerification.java | 4 +++- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java index 5c164d5a67f4..74e5d89b494c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java @@ -34,6 +34,7 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps; import org.checkerframework.checker.nullness.qual.Nullable; +import org.checkerframework.checker.nullness.qual.PolyNull; import org.joda.time.base.AbstractInstant; @SuppressWarnings({ @@ -52,8 +53,8 @@ static Row captureRow(Schema schema, List<@Nullable Object> rawValues) { } @SuppressWarnings("unchecked") - static @Nullable Object captureValue( - FieldType fieldType, @Nullable Object value, String fieldName) { + static @PolyNull Object captureValue( + FieldType fieldType, @PolyNull Object value, String fieldName) { if (value == null) { if (!fieldType.getNullable()) { throw new IllegalArgumentException( diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java index c829a562d249..397a47dcf3d9 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java @@ -63,8 +63,10 @@ public W getValue(int fieldIdx) { Field field = getSchema().getField(fieldIdx); @Nullable Object fieldValue; if (cacheFieldType(field)) { + @Nullable Object[] cache = this.cache; if (cache == null) { cache = new Object[getters.size()]; + this.cache = cache; } Object cached = cache[fieldIdx]; if (cached == null) { diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java index fe635e20f3ce..f54be0f3c3ce 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/SchemaVerification.java @@ -19,9 +19,11 @@ import java.io.Serializable; import org.apache.beam.sdk.schemas.Schema.FieldType; +import org.checkerframework.checker.nullness.qual.PolyNull; public abstract class SchemaVerification implements Serializable { - public static Object verifyFieldValue(Object value, FieldType type, String fieldName) { + public static @PolyNull Object verifyFieldValue( + @PolyNull Object value, FieldType type, String fieldName) { return RowUtils.captureValue(type, value, fieldName); } }