reuvenlax commented on code in PR #40791:
URL: https://github.com/apache/beam/pull/40791#discussion_r4397221436
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java:
##########
@@ -19,204 +19,174 @@
import java.math.BigDecimal;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.beam.sdk.schemas.FieldAccessDescriptor;
import org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor;
-import
org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor.ListQualifier;
-import
org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor.MapQualifier;
-import
org.apache.beam.sdk.schemas.FieldAccessDescriptor.FieldDescriptor.Qualifier;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.schemas.Schema.LogicalType;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.Nullable;
-import org.joda.time.Instant;
+import org.checkerframework.checker.nullness.qual.PolyNull;
import org.joda.time.base.AbstractInstant;
@SuppressWarnings({
"nullness", // TODO(https://github.com/apache/beam/issues/20497)
"rawtypes"
})
class RowUtils {
- static class RowPosition {
- FieldAccessDescriptor descriptor;
- List<FieldAccessDescriptor.FieldDescriptor.Qualifier> qualifiers;
-
- RowPosition(FieldAccessDescriptor descriptor) {
- this(descriptor, Collections.emptyList());
- }
-
- RowPosition(FieldAccessDescriptor descriptor, List<Qualifier> qualifiers) {
- this.descriptor = descriptor;
- this.qualifiers = qualifiers;
- }
-
- RowPosition withArrayQualifier() {
- List<Qualifier> newQualifiers =
Lists.newArrayListWithCapacity(qualifiers.size() + 1);
- newQualifiers.addAll(qualifiers);
- newQualifiers.add(Qualifier.of(ListQualifier.ALL));
- return new RowPosition(descriptor, newQualifiers);
- }
-
- RowPosition withMapQualifier() {
- List<Qualifier> newQualifiers =
Lists.newArrayListWithCapacity(qualifiers.size() + 1);
- newQualifiers.addAll(qualifiers);
- newQualifiers.add(Qualifier.of(MapQualifier.ALL));
- return new RowPosition(descriptor, newQualifiers);
- }
+ static Row captureRow(Schema schema, List<@Nullable Object> rawValues) {
+ int fieldCount = schema.getFieldCount();
+ List<@Nullable Object> capturedValues = new ArrayList<>(fieldCount);
+ for (int i = 0; i < fieldCount; i++) {
+ Schema.Field field = schema.getField(i);
+ capturedValues.add(captureValue(field.getType(), rawValues.get(i),
field.getName()));
+ }
+ return new RowWithStorage(schema, capturedValues);
}
- // Subclasses of this interface implement process methods for each schema
type. Each process
- // method is invoked as
- // a RowFieldMatcher walks down the schema tree. The FieldAccessDescriptor
passed into each method
- // identifies the
- // current element of the schema being processed.
- interface RowCases {
- Row processRow(RowPosition rowPosition, Schema schema, Row value,
RowFieldMatcher matcher);
-
- Collection<Object> processArray(
- RowPosition rowPosition,
- FieldType collectionElementType,
- Collection<Object> values,
- RowFieldMatcher matcher);
-
- Iterable<Object> processIterable(
- RowPosition rowPosition,
- FieldType collectionElementType,
- Iterable<Object> values,
- RowFieldMatcher matcher);
-
- Map<Object, Object> processMap(
- RowPosition rowPosition,
- FieldType keyType,
- FieldType valueType,
- Map<Object, Object> valueMap,
- RowFieldMatcher matcher);
-
- Object processLogicalType(
- RowPosition rowPosition, LogicalType logicalType, Object baseType,
RowFieldMatcher matcher);
-
- Instant processDateTime(
- RowPosition rowPosition, AbstractInstant instant, RowFieldMatcher
matcher);
-
- Byte processByte(RowPosition rowPosition, Byte value, RowFieldMatcher
matcher);
-
- Short processInt16(RowPosition rowPosition, Short value, RowFieldMatcher
matcher);
-
- Integer processInt32(RowPosition rowPosition, Integer value,
RowFieldMatcher matcher);
-
- Long processInt64(RowPosition rowPosition, Long value, RowFieldMatcher
matcher);
-
- BigDecimal processDecimal(RowPosition rowPosition, BigDecimal value,
RowFieldMatcher matcher);
-
- Float processFloat(RowPosition rowPosition, Float value, RowFieldMatcher
matcher);
-
- Double processDouble(RowPosition rowPosition, Double value,
RowFieldMatcher matcher);
-
- String processString(RowPosition rowPosition, String value,
RowFieldMatcher matcher);
-
- Boolean processBoolean(RowPosition rowPosition, Boolean value,
RowFieldMatcher matcher);
-
- byte[] processBytes(RowPosition rowPosition, byte[] value, RowFieldMatcher
matcher);
+ @SuppressWarnings("unchecked")
+ static @PolyNull Object captureValue(
+ FieldType fieldType, @PolyNull Object value, String fieldName) {
+ if (value == null) {
+ if (!fieldType.getNullable()) {
+ throw new IllegalArgumentException(
+ String.format("%s is not nullable in field %s", fieldType,
fieldName));
+ }
+ return null;
+ }
+ Object processedValue;
+ switch (fieldType.getTypeName()) {
+ case ARRAY:
Review Comment:
Aren't these two case statements the same? Should we just fall through?
--
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]
