mxtymoshyk commented on code in PR #40841:
URL: https://github.com/apache/beam/pull/40841#discussion_r4387819785
##########
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamUnnestRel.java:
##########
@@ -136,32 +133,41 @@ private UnnestFn(Schema outputSchema, List<Integer>
unnestIndices) {
}
/**
- * This is recursive call to get all the values of the nested rows. The
recursion is bounded by
- * the amount of nesting with in the data. This mirrors the unnest
behavior of calcite towards
- * schema. *
+ * Appends the fields of an unnested array element to {@code values}, laid
out the way {@code
+ * outputSchema} expects them. The recursion is bounded by the nesting
depth of the element
+ * schema.
+ *
+ * <p>Whether a nested row field of the element arrives as one ROW column
or as one column per
+ * leaf field depends on which Calcite planner produced the plan. For an
element {@code ROW<id,
+ * author ROW<name, email>>}, the planner behind {@code SqlTransform}
flattens the element into
+ * {@code id, name, email}, while the planner behind the JDBC driver and
the SQL shell keeps
+ * {@code id, author}. Building the values from the element schema alone
matches only one of
+ * them, and the other then fails in {@link Row.Builder#build()} with a
field count mismatch. So
+ * each nested row is checked against the output column at the same
position: a ROW column takes
+ * the row as is, any other column means the row was flattened.
*/
- private List<Object> getNestedRowBaseValues(Row nestedRow) {
- return IntStream.range(0, nestedRow.getFieldCount())
- .mapToObj(
- (i) -> {
- List<Object> values = new ArrayList<>();
- Schema.FieldType fieldType =
nestedRow.getSchema().getField(i).getType();
- if (fieldType.getTypeName().equals(Schema.TypeName.ROW)) {
- @Nullable Row row = nestedRow.getBaseValue(i, Row.class);
- if (row == null) {
- return Stream.builder().build();
- }
- List<Object> rowValues = getNestedRowBaseValues(row);
- if (null != rowValues) {
- values.addAll(rowValues);
- }
- } else {
- values.add(nestedRow.getBaseValue(i));
- }
- return values.stream();
- })
- .flatMap(Function.identity())
- .collect(Collectors.toList());
+ private void addElementValues(
+ Schema elementSchema, @Nullable Row element, List<Object> values) {
+ for (int i = 0; i < elementSchema.getFieldCount(); i++) {
+ Schema.FieldType fieldType = elementSchema.getField(i).getType();
+ Object value = null;
+ if (Objects.nonNull(element)) {
+ value = element.getBaseValue(i);
+ }
+
+ boolean isRow = fieldType.getTypeName().equals(Schema.TypeName.ROW);
+ if (!isRow || isRowColumn(values.size())) {
+ values.add(value);
+ continue;
+ }
+
+ // a null nested row still fills one null per flattened column
+ addElementValues(fieldType.getRowSchema(), (Row) value, values);
+ }
+ }
+
+ private boolean isRowColumn(int index) {
Review Comment:
Added a guard in 7a2a631. Instead of throwing, `isRowColumn` returns false
past the last output column. The extra values then reach `Row.Builder#build()`,
which fails with the same `Row expected N fields ... initialized with M fields`
error as in #26911. That message lists the output schema fields, so a planner
mismatch is easier to trace. Happy to switch to an `IllegalStateException` if
you would rather have that.
##########
sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/JdbcDriverTest.java:
##########
@@ -351,6 +351,59 @@ public void testSelectsFromExistingComplexTable() throws
Exception {
assertThat(resultRows, containsInAnyOrder(row(1L, "aaa"), row(2L, "bbb")));
}
+ @Test
Review Comment:
Added in 7a2a631. The JDBC test now has a third element whose `nestedRow` is
null, and the query returns null for `item.nestedRow.name` on that row.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
