BeamSQL might fail on test_distinct_bytes after migrated to Sets transform

This query breaks Java code import. Need to diagnose the root cause and have a fix properly.

```

public void test_distinct_bytes() {
String sql = "SELECT DISTINCT val.BYTES\n"
+ "from
(select b\"1\" BYTES union all\n"
+ " select cast(NULL as bytes) union all\n"

+ " select b\"-1\" union all\n"
+ " select b\"1\" union all\n"
+ "
select cast(NULL as bytes)) val";

ZetaSQLQueryPlanner zetaSQLQueryPlanner = new ZetaSQLQueryPlanner(config);

BeamRelNode beamRelNode = zetaSQLQueryPlanner.convertToBeamRel(sql);
PCollection stream
= BeamSqlRelUtils.toPCollection(pipeline, beamRelNode);

Schema singleField = Schema.builder().addByteArrayField("field1").build();

PAssert.that(stream)
.containsInAnyOrder(
Row.withSchema(singleField).addValues("123".getBytes(UTF_8)).build());

pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));

}

```

This query will throws

```

Forbidden IOException when reading from InputStream
java.lang.IllegalArgumentException: Forbidden
IOException when reading from InputStream
at org.apache.beam.sdk.util.CoderUtils.decodeFromSafeStream(CoderUtils.java:118)
at
org.apache.beam.sdk.util.CoderUtils.decodeFromByteArray(CoderUtils.java:98)
at org.apache.beam.sdk.util.CoderUtils.decodeFromByteArray(CoderUtils.java:92)
at
org.apache.beam.sdk.util.CoderUtils.clone(CoderUtils.java:141)

```

Imported from Jira [BEAM-10226](https://issues.apache.org/jira/browse/BEAM-10226). Original Jira may contain additional context.
Reported by: amaliujia.