You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2024/11/15 21:49:24 UTC
(beam) 01/01: Revert "Merge pull request #32705: fix schema inference for parameterized types"
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch revert-32705-schema_inference_parameterized_types
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 1ed5273833bc1d1dbe87b3627897125910693b2d
Author: Yi Hu <hu...@gmail.com>
AuthorDate: Fri Nov 15 16:49:17 2024 -0500
Revert "Merge pull request #32705: fix schema inference for parameterized types"
This reverts commit c243491254896e039e912662e4cfbe4bd38c766f.
---
.../main/java/org/apache/beam/sdk/Pipeline.java | 2 +-
.../org/apache/beam/sdk/coders/CoderRegistry.java | 44 ++++++----------------
.../beam/sdk/schemas/SchemaRegistryTest.java | 20 ----------
3 files changed, 12 insertions(+), 54 deletions(-)
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java
index 9006035279f..d3b58dd26bd 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java
@@ -335,7 +335,7 @@ public class Pipeline {
/** Returns the {@link CoderRegistry} that this {@link Pipeline} uses. */
public CoderRegistry getCoderRegistry() {
if (coderRegistry == null) {
- coderRegistry = CoderRegistry.createDefault(getSchemaRegistry());
+ coderRegistry = CoderRegistry.createDefault();
}
return coderRegistry;
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
index e404665e4f6..df64789ac3d 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
@@ -42,8 +42,6 @@ import org.apache.beam.sdk.io.fs.MatchResult.Metadata;
import org.apache.beam.sdk.io.fs.MetadataCoder;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.io.fs.ResourceIdCoder;
-import org.apache.beam.sdk.schemas.NoSuchSchemaException;
-import org.apache.beam.sdk.schemas.SchemaRegistry;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.util.CoderUtils;
@@ -197,17 +195,11 @@ public class CoderRegistry {
* the lexicographically smallest {@link Class#getName() class name} being used.
* </ul>
*/
- public static CoderRegistry createDefault(@Nullable SchemaRegistry schemaRegistry) {
- return new CoderRegistry(schemaRegistry);
- }
-
- /** Backwards compatible version of createDefault. */
public static CoderRegistry createDefault() {
- return new CoderRegistry(null);
+ return new CoderRegistry();
}
- private CoderRegistry(@Nullable SchemaRegistry schemaRegistry) {
- this.schemaRegistry = schemaRegistry;
+ private CoderRegistry() {
coderProviders = new ArrayDeque<>(REGISTERED_CODER_FACTORIES);
}
@@ -598,8 +590,6 @@ public class CoderRegistry {
/** The list of {@link CoderProvider coder providers} to use to provide Coders. */
private ArrayDeque<CoderProvider> coderProviders;
- private final @Nullable SchemaRegistry schemaRegistry;
-
/**
* Returns a {@link Coder} to use for values of the given type, in a context where the given types
* use the given coders.
@@ -660,28 +650,16 @@ public class CoderRegistry {
List<Coder<?>> typeArgumentCoders = new ArrayList<>();
for (Type typeArgument : type.getActualTypeArguments()) {
- Coder<?> typeArgumentCoder = null;
- if (schemaRegistry != null) {
- TypeDescriptor<?> typeDescriptor = TypeDescriptor.of(typeArgument);
- try {
- typeArgumentCoder = schemaRegistry.getSchemaCoder(typeDescriptor);
- } catch (NoSuchSchemaException e) {
- // No schema.
- }
- }
-
- if (typeArgumentCoder == null) {
- try {
- typeArgumentCoder =
- getCoderFromTypeDescriptor(TypeDescriptor.of(typeArgument), typeCoderBindings);
- } catch (CannotProvideCoderException exc) {
- throw new CannotProvideCoderException(
- String.format(
- "Cannot provide coder for parameterized type %s: %s", type, exc.getMessage()),
- exc);
- }
+ try {
+ Coder<?> typeArgumentCoder =
+ getCoderFromTypeDescriptor(TypeDescriptor.of(typeArgument), typeCoderBindings);
+ typeArgumentCoders.add(typeArgumentCoder);
+ } catch (CannotProvideCoderException exc) {
+ throw new CannotProvideCoderException(
+ String.format(
+ "Cannot provide coder for parameterized type %s: %s", type, exc.getMessage()),
+ exc);
}
- typeArgumentCoders.add(typeArgumentCoder);
}
return getCoderFromFactories(TypeDescriptor.of(type), typeArgumentCoders);
}
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaRegistryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaRegistryTest.java
index 54c80747b13..55a16e9faf3 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaRegistryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaRegistryTest.java
@@ -26,10 +26,6 @@ import static org.junit.Assert.assertTrue;
import com.google.auto.service.AutoService;
import com.google.auto.value.AutoValue;
import java.util.List;
-import org.apache.beam.sdk.coders.CannotProvideCoderException;
-import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.coders.CoderRegistry;
-import org.apache.beam.sdk.coders.IterableCoder;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.apache.beam.sdk.schemas.utils.TestJavaBeans.SimpleBean;
import org.apache.beam.sdk.schemas.utils.TestPOJOs.SimplePOJO;
@@ -227,22 +223,6 @@ public class SchemaRegistryTest {
assertTrue(SIMPLE_POJO_SCHEMA.equivalent(schema));
}
- @Test
- public void testSchemaTypeParameterInsideCoder() throws CannotProvideCoderException {
- SchemaRegistry schemaRegistry = SchemaRegistry.createDefault();
- schemaRegistry.registerPOJO(SimplePOJO.class);
-
- CoderRegistry coderRegistry = CoderRegistry.createDefault(schemaRegistry);
- Coder<Iterable<SimplePOJO>> coder =
- coderRegistry.getCoder(TypeDescriptors.iterables(TypeDescriptor.of(SimplePOJO.class)));
- assertTrue(coder instanceof IterableCoder);
- assertEquals(1, coder.getCoderArguments().size());
- assertTrue(coder.getCoderArguments().get(0) instanceof SchemaCoder);
- assertTrue(
- SIMPLE_POJO_SCHEMA.equivalent(
- ((SchemaCoder<SimplePOJO>) coder.getCoderArguments().get(0)).getSchema()));
- }
-
@Test
public void testRegisterJavaBean() throws NoSuchSchemaException {
SchemaRegistry registry = SchemaRegistry.createDefault();