Remove primaryKeyColumns options

Author: Amar3tto

sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java

@@ -23,7 +23,6 @@
 import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.stream.Collectors;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.MapCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
@@ -319,23 +318,14 @@ protected Schema getRecordSchema() {
       SourceRecord sampledRecord =
           fn.getOneRecord(getConnectorConfiguration().getConfigurationMap());
       fn.reset();
-      Schema keySchema =
-          sampledRecord.keySchema() != null
-              ? KafkaConnectUtils.beamSchemaFromKafkaConnectSchema(sampledRecord.keySchema())
-              : Schema.builder().build();
       Schema valueSchema =
           KafkaConnectUtils.beamSchemaFromKafkaConnectSchema(sampledRecord.valueSchema());
 
       return Schema.builder()
           .addFields(valueSchema.getFields())
-          .setOptions(
-              Schema.Options.builder()
-                  .setOption(
-                      "primaryKeyColumns",
-                      Schema.FieldType.array(Schema.FieldType.STRING),
-                      keySchema.getFields().stream()
-                          .map(Schema.Field::getName)
-                          .collect(Collectors.toList())))
+        // TODO(https://github.com/apache/beam/issues/39557):
+        // Restore 'primaryKeyColumns' once Python can decode ARRAY<STRING>
+        // schema options across the Java/Python cross-language boundary.
           .build();
     }
 

sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java

@@ -74,10 +74,6 @@ protected DebeziumReadSchemaTransformProvider(
     this.testLimitMilliseconds = timeLimitMs;
   }
 
-  private static Schema withoutOptions(Schema schema) {
-    return Schema.builder().addFields(schema.getFields()).build();
-  }
-
   private static Connectors parseConnector(String value) {
     try {
       return Connectors.valueOf(value);
@@ -162,7 +158,7 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
         }
 
         // TODO(pabloem): Database connection issues can be debugged here.
-        Schema recordSchema = withoutOptions(readTransform.getRecordSchema());
+        Schema recordSchema = readTransform.getRecordSchema();
         LOG.info(
             "Computed schema for table {} from {}: {}",
             configuration.getTable(),