utafrali commented on code in PR #40841:
URL: https://github.com/apache/beam/pull/40841#discussion_r4387613850
##########
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:
`isRowColumn` will throw `IndexOutOfBoundsException` if `index >=
outputSchema.getFieldCount()`. This can only happen when the element schema and
the output schema disagree on field count, which indicates a planning bug
rather than user error. A bounds guard with a clear message would make that
failure much easier to diagnose than a raw IOOBE:
```java
private boolean isRowColumn(int index) {
if (index >= outputSchema.getFieldCount()) {
throw new IllegalStateException(
"Element schema produces more fields than output schema at index " +
index);
}
return
outputSchema.getField(index).getType().getTypeName().equals(Schema.TypeName.ROW);
}
```
##########
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)) {
Review Comment:
`Objects.nonNull(element)` is equivalent to `element != null` but less
idiomatic in Java. Prefer the direct null check: `if (element != null)`. The
rest of the file uses direct null comparisons (e.g. `rawValues == null` on line
184).
##########
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:
The new JDBC test only exercises the non-null nested row path. The
null-nested-row fix in `addElementValues` (adding a null per flattened column)
is covered by `BeamSqlDslUnnestRowsTest#testUnnestArrayWithNullNestedRow`, but
that test runs through `SqlTransform`, not the JDBC planner. On the JDBC path
the row is kept as a single ROW column, so a null `nestedRow` would be added as
a single `null` value via `values.add(value)` at line 160, which is a different
branch. Consider adding a second entry in `tableProvider.addRows` with a `null`
nested row and asserting that the corresponding result row has `null` for the
nested field.
--
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]
