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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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/10] 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();