WITH query AS + SELECT query JOIN other throws invalid type

The first one of the three following queries fails, despite queries being equivalent:
```

Pipeline p = Pipeline.create();

Schema schemaA =
Schema.of(
Schema.Field.of("id",
Schema.FieldType.BYTES),
Schema.Field.of("fA1", Schema.FieldType.STRING));

Schema schemaB
=
Schema.of(
Schema.Field.of("id", Schema.FieldType.STRING),
Schema.Field.of("fB1",
Schema.FieldType.STRING));

PCollection inputA =
p.apply(Create.of(ImmutableList.of()).withCoder(SchemaCoder.of(schemaA)));

PCollection
inputB =
p.apply(Create.of(ImmutableList.of()).withCoder(SchemaCoder.of(schemaB)));

//
Fails
String query1 =
"WITH query AS "
+ "( "
+ " SELECT id, fA1, fA1 AS fA1_2
"
+ " FROM tblA"
+ ") "
+ "SELECT fA1, fB1, fA1_2 "
+ "FROM query
"
+ "JOIN tblB ON (TO_HEX(query.id) = tblB.id)";

// Ok
String query2 =
"WITH query
AS "
+ "( "
+ " SELECT fA1, fB1, fA1 AS fA1_2 "
+ " FROM tblA "

+ " JOIN tblB "
+ " ON (TO_HEX(tblA.id) = tblB.id) "
+ ")"
+ "SELECT fA1,
fB1, fA1_2 "
+ "FROM query ";

// Ok
String query3 =
"WITH query AS "
+ "( "

+ " SELECT TO_HEX(id) AS id, fA1, fA1 AS fA1_2 "
+ " FROM tblA"
+ ") "
+ "SELECT fA1,
fB1, fA1_2 "
+ "FROM query "
+ "JOIN tblB ON (query.id = tblB.id)";

Schema transform3 =

PCollectionTuple.of("tblA", inputA)
.and("tblB", inputB)
.apply(SqlTransform.query(query3))

.getSchema();
System.out.println(transform3);

Schema transform2 =
PCollectionTuple.of("tblA",
inputA)
.and("tblB", inputB)
.apply(SqlTransform.query(query2))
.getSchema();
System.out.println(transform2);

Schema
transform1 =
PCollectionTuple.of("tblA", inputA)
.and("tblB", inputB)
.apply(SqlTransform.query(query1))

.getSchema();
System.out.println(transform1);

```

 

The error is:
```

Exception in thread "main" java.lang.AssertionError: Field ordinal 2 is invalid for  type 'RecordType(VARBINARY
id, VARCHAR fA1)'Exception in thread "main" java.lang.AssertionError: Field ordinal 2 is invalid for 
type 'RecordType(VARBINARY id, VARCHAR fA1)' at org.apache.beam.repackaged.sql.org.apache.calcite.rex.RexBuilder.makeFieldAccess(RexBuilder.java:197)
```

 

If I change `schemaB.id` to `BYTES` (while also avoid using `TO_HEX`), all queries work fine. 

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