derrickaw commented on code in PR #35435:
URL: https://github.com/apache/beam/pull/35435#discussion_r2211704997


##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteSchemaTransformProvider.java:
##########
@@ -135,18 +140,197 @@ public PCollectionRowTuple expand(PCollectionRowTuple 
input) {
           String.format(
               "Could not find expected input [%s] to %s.", INPUT_TAG, 
getClass().getSimpleName()));
 
-      PCollection<Row> beamRowMutations = input.get(INPUT_TAG);
-      PCollection<KV<ByteString, Iterable<Mutation>>> bigtableMutations =
-          beamRowMutations.apply(MapElements.via(new 
GetMutationsFromBeamRow()));
+      Schema testOriginialSchema =
+          Schema.builder()
+              .addByteArrayField("key")
+              .addArrayField(
+                  "mutations",
+                  Schema.FieldType.map(Schema.FieldType.STRING, 
Schema.FieldType.BYTES))
+              .build();
 
-      bigtableMutations.apply(
-          BigtableIO.write()
-              .withTableId(configuration.getTableId())
-              .withInstanceId(configuration.getInstanceId())
-              .withProjectId(configuration.getProjectId()));
+      Schema inputSchema = input.getSinglePCollection().getSchema();
 
+      System.out.println("Input Schema for BigTableMutations: " + inputSchema);
+
+      PCollection<KV<ByteString, Iterable<Mutation>>> bigtableMutations = null;
+      if (inputSchema.equals(testOriginialSchema)) {
+        PCollection<Row> beamRowMutations = input.get(INPUT_TAG);
+        bigtableMutations =
+            beamRowMutations.apply(
+                // Original schema inputs gets sent out to the original 
transform provider mutations
+                // function
+                MapElements.via(
+                    new 
BigtableWriteSchemaTransformProvider.GetMutationsFromBeamRow()));
+      } else if (inputSchema.hasField("type")) {
+        //        // validate early doesn't work for all mutations IT test but 
it does help

Review Comment:
   do we still need this?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscr...@beam.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Reply via email to