You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ap...@apache.org on 2019/01/28 23:56:27 UTC
[beam] branch master updated: [BEAM-6513] Fix NPE in
BeamEnumerableConveter.
This is an automated email from the ASF dual-hosted git repository.
apilloud pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 9b6165c [BEAM-6513] Fix NPE in BeamEnumerableConveter.
new 26c2272 Merge pull request #7636 from amaliujia/rui_wang-NPE
9b6165c is described below
commit 9b6165c4db2ba833a5b38d4fd0ef465727e1a886
Author: amaliujia <am...@163.com>
AuthorDate: Fri Jan 25 22:03:36 2019 -0800
[BEAM-6513] Fix NPE in BeamEnumerableConveter.
---
.../sql/impl/rel/BeamEnumerableConverter.java | 37 ++++++++++++++--------
.../extensions/sql/impl/utils/CalciteUtils.java | 3 +-
.../sql/impl/rel/BeamEnumerableConverterTest.java | 22 +++++++++++++
3 files changed, 47 insertions(+), 15 deletions(-)
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java
index 408922b..fedc8c3 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java
@@ -143,7 +143,7 @@ public class BeamEnumerableConverter extends ConverterImpl implements Enumerable
PipelineOptions options,
BeamRelNode node,
DoFn<Row, Void> doFn,
- Queue<Object> values,
+ Queue<Object[]> values,
int limitCount) {
options.as(DirectOptions.class).setBlockOnRun(false);
Pipeline pipeline = Pipeline.create(options);
@@ -176,7 +176,7 @@ public class BeamEnumerableConverter extends ConverterImpl implements Enumerable
private static Enumerable<Object> collect(PipelineOptions options, BeamRelNode node) {
long id = options.getOptionsId();
- Queue<Object> values = new ConcurrentLinkedQueue<>();
+ Queue<Object[]> values = new ConcurrentLinkedQueue<>();
checkArgument(
options
@@ -195,12 +195,12 @@ public class BeamEnumerableConverter extends ConverterImpl implements Enumerable
Collector.globalValues.remove(id);
- return Linq4j.asEnumerable(values);
+ return Linq4j.asEnumerable(unboxValues(values));
}
private static Enumerable<Object> limitCollect(PipelineOptions options, BeamRelNode node) {
long id = options.getOptionsId();
- Queue<Object> values = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object[]> values = new ConcurrentLinkedQueue<>();
checkArgument(
options
@@ -220,15 +220,15 @@ public class BeamEnumerableConverter extends ConverterImpl implements Enumerable
values.remove();
}
- return Linq4j.asEnumerable(values);
+ return Linq4j.asEnumerable(unboxValues(values));
}
private static class Collector extends DoFn<Row, Void> {
// This will only work on the direct runner.
- private static final Map<Long, Queue<Object>> globalValues = new ConcurrentHashMap<>();
+ private static final Map<Long, Queue<Object[]>> globalValues = new ConcurrentHashMap<>();
- @Nullable private volatile Queue<Object> values;
+ @Nullable private volatile Queue<Object[]> values;
@StartBundle
public void startBundle(StartBundleContext context) {
@@ -238,15 +238,26 @@ public class BeamEnumerableConverter extends ConverterImpl implements Enumerable
@ProcessElement
public void processElement(ProcessContext context) {
- Object[] avaticaRow = rowToAvatica(context.element());
- if (avaticaRow.length == 1) {
- values.add(avaticaRow[0]);
- } else {
- values.add(avaticaRow);
- }
+ values.add(rowToAvatica(context.element()));
}
}
+ private static List<Object> unboxValues(Queue<Object[]> values) {
+ return values.stream()
+ .map(
+ objects -> {
+ if (objects.length == 1) {
+ // if objects.length == 1, that means input Row contains only 1 column/element,
+ // then an Object instead of Object[] should be returned because of
+ // CalciteResultSet's behaviour that tries to convert one column row to an Object.
+ return objects[0];
+ } else {
+ return objects;
+ }
+ })
+ .collect(Collectors.toList());
+ }
+
private static Object[] rowToAvatica(Row row) {
Schema schema = row.getSchema();
Object[] convertedColumns = new Object[schema.getFields().size()];
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/utils/CalciteUtils.java b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/utils/CalciteUtils.java
index cbc23f0..9c94065 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/utils/CalciteUtils.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/utils/CalciteUtils.java
@@ -167,8 +167,7 @@ public class CalciteUtils {
return builder.build();
}
- private static RelDataType toRelDataType(
- RelDataTypeFactory dataTypeFactory, FieldType fieldType) {
+ public static RelDataType toRelDataType(RelDataTypeFactory dataTypeFactory, FieldType fieldType) {
switch (fieldType.getTypeName()) {
case ARRAY:
return dataTypeFactory.createArrayType(
diff --git a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverterTest.java b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverterTest.java
index 2098803..5011aa3 100644
--- a/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverterTest.java
+++ b/sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverterTest.java
@@ -28,6 +28,7 @@ import org.apache.beam.sdk.extensions.sql.impl.utils.CalciteUtils;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PBegin;
@@ -95,6 +96,27 @@ public class BeamEnumerableConverterTest {
enumerator.close();
}
+ @Test
+ public void testToEnumerable_collectNullValue() {
+ Schema schema = Schema.builder().addNullableField("id", FieldType.INT64).build();
+ RelDataType type = CalciteUtils.toCalciteRowType(schema, TYPE_FACTORY);
+ ImmutableList<ImmutableList<RexLiteral>> tuples =
+ ImmutableList.of(
+ ImmutableList.of(
+ rexBuilder.makeNullLiteral(
+ CalciteUtils.toRelDataType(TYPE_FACTORY, FieldType.INT64))));
+ BeamRelNode node = new BeamValuesRel(cluster, type, tuples, null);
+
+ Enumerable<Object> enumerable = BeamEnumerableConverter.toEnumerable(options, node);
+ Enumerator<Object> enumerator = enumerable.enumerator();
+
+ assertTrue(enumerator.moveNext());
+ Object row = enumerator.current();
+ assertEquals(null, row);
+ assertFalse(enumerator.moveNext());
+ enumerator.close();
+ }
+
private static class FakeTable extends BaseBeamTable {
public FakeTable() {
super(null);