beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From reuven...@apache.org
Subject [beam] branch master updated: Merge pull request #7175: [BEAM-5884] Move the nullable attribute onto FieldType.
Date Sat, 01 Dec 2018 06:21:53 GMT
This is an automated email from the ASF dual-hosted git repository.

reuvenlax 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 a9dff65  Merge pull request #7175: [BEAM-5884] Move the nullable attribute onto FieldType.
a9dff65 is described below

commit a9dff652c43202d2563f38f352c72c8cd748d4fb
Author: reuvenlax <relax@google.com>
AuthorDate: Fri Nov 30 22:21:46 2018 -0800

    Merge pull request #7175: [BEAM-5884] Move the nullable attribute onto FieldType.
---
 .../apache/beam/sdk/coders/RowCoderGenerator.java  |  4 +-
 .../java/org/apache/beam/sdk/schemas/Schema.java   | 95 ++++++++++------------
 .../apache/beam/sdk/schemas/transforms/Cast.java   |  2 +-
 .../apache/beam/sdk/schemas/transforms/Select.java |  7 +-
 .../main/java/org/apache/beam/sdk/values/Row.java  | 29 ++-----
 .../org/apache/beam/sdk/values/RowWithGetters.java |  2 +-
 .../beam/sdk/schemas/utils/JavaBeanUtilsTest.java  |  4 +-
 .../beam/sdk/schemas/utils/POJOUtilsTest.java      |  4 +-
 .../sql/impl/parser/SqlCreateExternalTable.java    |  2 +-
 .../extensions/sql/impl/schema/BeamTableUtils.java |  2 +-
 .../extensions/sql/impl/utils/CalciteUtils.java    |  4 +-
 .../sql/meta/provider/text/TextTableProvider.java  |  5 +-
 .../beam/sdk/io/gcp/bigquery/BigQueryUtils.java    |  2 +-
 13 files changed, 69 insertions(+), 93 deletions(-)

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
index 50be370..112ffe3 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
@@ -51,6 +51,7 @@ import net.bytebuddy.implementation.bytecode.member.MethodVariableAccess;
 import net.bytebuddy.matcher.ElementMatchers;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.Schema.Field;
+import org.apache.beam.sdk.schemas.Schema.FieldType;
 import org.apache.beam.sdk.schemas.Schema.TypeName;
 import org.apache.beam.sdk.values.Row;
 
@@ -147,7 +148,8 @@ public abstract class RowCoderGenerator {
 
   private static DynamicType.Builder<Coder> implementMethods(
       Schema schema, DynamicType.Builder<Coder> builder) {
-    boolean hasNullableFields = schema.getFields().stream().anyMatch(Field::getNullable);
+    boolean hasNullableFields =
+        schema.getFields().stream().map(Field::getType).anyMatch(FieldType::getNullable);
     return builder
         .defineMethod("getSchema", Schema.class, Visibility.PRIVATE, Ownership.STATIC)
         .intercept(FixedValue.reference(schema))
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java
index 90ec371..a923c0e 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java
@@ -269,7 +269,7 @@ public class Schema implements Serializable {
     for (int i = 0; i < otherFields.size(); ++i) {
       Field otherField = otherFields.get(i);
       Field actualField = actualFields.get(i);
-      if (!otherField.equivalent(actualField, nullablePolicy)) {
+      if (!actualField.equivalent(otherField, nullablePolicy)) {
         return false;
       }
     }
@@ -410,14 +410,13 @@ public class Schema implements Serializable {
     // Returns the type of this field.
     public abstract TypeName getTypeName();
 
+    // Whether this type is nullable.
+    public abstract Boolean getNullable();
+
     // For container types (e.g. ARRAY), returns the type of the contained element.
     @Nullable
     public abstract FieldType getCollectionElementType();
 
-    // For container types (e.g. ARRAY), returns nullable of the type of the contained element.
-    @Nullable
-    public abstract Boolean getCollectionElementTypeNullable();
-
     // For MAP type, returns the type of the key element, it must be a primitive type;
     @Nullable
     public abstract FieldType getMapKeyType();
@@ -426,10 +425,6 @@ public class Schema implements Serializable {
     @Nullable
     public abstract FieldType getMapValueType();
 
-    // For MAP type, returns nullable of the type of the value element, it can be a nested
type;
-    @Nullable
-    public abstract Boolean getMapValueTypeNullable();
-
     // For ROW types, returns the schema for the row.
     @Nullable
     public abstract Schema getRowSchema();
@@ -442,7 +437,7 @@ public class Schema implements Serializable {
     abstract FieldType.Builder toBuilder();
 
     public static FieldType.Builder forTypeName(TypeName typeName) {
-      return new AutoValue_Schema_FieldType.Builder().setTypeName(typeName);
+      return new AutoValue_Schema_FieldType.Builder().setTypeName(typeName).setNullable(false);
     }
 
     @AutoValue.Builder
@@ -451,14 +446,12 @@ public class Schema implements Serializable {
 
       abstract Builder setCollectionElementType(@Nullable FieldType collectionElementType);
 
-      abstract Builder setCollectionElementTypeNullable(@Nullable Boolean nullable);
+      abstract Builder setNullable(Boolean nullable);
 
       abstract Builder setMapKeyType(@Nullable FieldType mapKeyType);
 
       abstract Builder setMapValueType(@Nullable FieldType mapValueType);
 
-      abstract Builder setMapValueTypeNullable(@Nullable Boolean nullable);
-
       abstract Builder setRowSchema(@Nullable Schema rowSchema);
 
       abstract Builder setMetadata(@Nullable byte[] metadata);
@@ -506,16 +499,12 @@ public class Schema implements Serializable {
 
     /** Create an array type for the given field type. */
     public static final FieldType array(FieldType elementType) {
-      return FieldType.forTypeName(TypeName.ARRAY)
-          .setCollectionElementType(elementType)
-          .setCollectionElementTypeNullable(false)
-          .build();
+      return FieldType.forTypeName(TypeName.ARRAY).setCollectionElementType(elementType).build();
     }
 
     public static final FieldType array(FieldType elementType, boolean nullable) {
       return FieldType.forTypeName(TypeName.ARRAY)
-          .setCollectionElementType(elementType)
-          .setCollectionElementTypeNullable(nullable)
+          .setCollectionElementType(elementType.withNullable(nullable))
           .build();
     }
 
@@ -524,7 +513,6 @@ public class Schema implements Serializable {
       return FieldType.forTypeName(TypeName.MAP)
           .setMapKeyType(keyType)
           .setMapValueType(valueType)
-          .setMapValueTypeNullable(false)
           .build();
     }
 
@@ -532,8 +520,7 @@ public class Schema implements Serializable {
         FieldType keyType, FieldType valueType, boolean valueTypeNullable) {
       return FieldType.forTypeName(TypeName.MAP)
           .setMapKeyType(keyType)
-          .setMapValueType(valueType)
-          .setMapValueTypeNullable(valueTypeNullable)
+          .setMapValueType(valueType.withNullable(valueTypeNullable))
           .build();
     }
 
@@ -552,6 +539,10 @@ public class Schema implements Serializable {
       return toBuilder().setMetadata(metadata.getBytes(StandardCharsets.UTF_8)).build();
     }
 
+    public FieldType withNullable(boolean nullable) {
+      return toBuilder().setNullable(nullable).build();
+    }
+
     @Override
     public boolean equals(Object o) {
       if (!(o instanceof FieldType)) {
@@ -559,6 +550,7 @@ public class Schema implements Serializable {
       }
       FieldType other = (FieldType) o;
       return Objects.equals(getTypeName(), other.getTypeName())
+          && Objects.equals(getNullable(), other.getNullable())
           && Objects.equals(getCollectionElementType(), other.getCollectionElementType())
           && Objects.equals(getMapKeyType(), other.getMapKeyType())
           && Objects.equals(getMapValueType(), other.getMapValueType())
@@ -589,29 +581,43 @@ public class Schema implements Serializable {
       return true;
     }
 
-    private boolean equivalent(FieldType other) {
-      if (!other.getTypeName().equals(getTypeName())) {
+    private boolean equivalent(FieldType other, EquivalenceNullablePolicy nullablePolicy)
{
+      if (nullablePolicy == EquivalenceNullablePolicy.SAME
+          && !other.getNullable().equals(getNullable())) {
         return false;
+      } else if (nullablePolicy == EquivalenceNullablePolicy.WEAKEN) {
+        if (getNullable() && !other.getNullable()) {
+          return false;
+        }
       }
+
+      if (!getTypeName().equals(other.getTypeName())) {
+        return false;
+      }
+      if (!Arrays.equals(getMetadata(), other.getMetadata())) {
+        return false;
+      }
+
       switch (getTypeName()) {
         case ROW:
-          if (!other.getRowSchema().equivalent(getRowSchema())) {
+          if (!getRowSchema().equivalent(other.getRowSchema(), nullablePolicy)) {
             return false;
           }
           break;
         case ARRAY:
-          if (!other.getCollectionElementType().equivalent(getCollectionElementType())) {
+          if (!getCollectionElementType()
+              .equivalent(other.getCollectionElementType(), nullablePolicy)) {
             return false;
           }
           break;
         case MAP:
-          if (!other.getMapKeyType().equivalent(getMapKeyType())
-              || !other.getMapValueType().equivalent(getMapValueType())) {
+          if (!getMapKeyType().equivalent(other.getMapKeyType(), nullablePolicy)
+              || !getMapValueType().equivalent(other.getMapValueType(), nullablePolicy))
{
             return false;
           }
           break;
         default:
-          return other.equals(this);
+          return true;
       }
       return true;
     }
@@ -621,6 +627,7 @@ public class Schema implements Serializable {
       return Arrays.deepHashCode(
           new Object[] {
             getTypeName(),
+            getNullable(),
             getCollectionElementType(),
             getMapKeyType(),
             getMapValueType(),
@@ -642,9 +649,6 @@ public class Schema implements Serializable {
     /** Returns the fields {@link FieldType}. */
     public abstract FieldType getType();
 
-    /** Returns whether the field supports null values. */
-    public abstract Boolean getNullable();
-
     public abstract Builder toBuilder();
 
     /** Builder for {@link Field}. */
@@ -656,8 +660,6 @@ public class Schema implements Serializable {
 
       public abstract Builder setType(FieldType fieldType);
 
-      public abstract Builder setNullable(Boolean nullable);
-
       public abstract Field build();
     }
 
@@ -667,7 +669,6 @@ public class Schema implements Serializable {
           .setName(name)
           .setDescription("")
           .setType(fieldType)
-          .setNullable(false) // By default fields are not nullable.
           .build();
     }
 
@@ -676,8 +677,7 @@ public class Schema implements Serializable {
       return new AutoValue_Schema_Field.Builder()
           .setName(name)
           .setDescription("")
-          .setType(fieldType)
-          .setNullable(true)
+          .setType(fieldType.withNullable(true))
           .build();
     }
 
@@ -698,7 +698,7 @@ public class Schema implements Serializable {
 
     /** Returns a copy of the Field with isNullable set. */
     public Field withNullable(boolean isNullable) {
-      return toBuilder().setNullable(isNullable).build();
+      return toBuilder().setType(getType().withNullable(isNullable)).build();
     }
 
     @Override
@@ -709,31 +709,22 @@ public class Schema implements Serializable {
       Field other = (Field) o;
       return Objects.equals(getName(), other.getName())
           && Objects.equals(getDescription(), other.getDescription())
-          && Objects.equals(getType(), other.getType())
-          && Objects.equals(getNullable(), other.getNullable());
+          && Objects.equals(getType(), other.getType());
     }
 
     /** Returns true if two fields are equal, ignoring name and description. */
     public boolean typesEqual(Field other) {
-      return getType().typesEqual(other.getType())
-          && Objects.equals(getNullable(), other.getNullable());
+      return getType().typesEqual(other.getType());
     }
 
     private boolean equivalent(Field otherField, EquivalenceNullablePolicy nullablePolicy)
{
-      if (nullablePolicy == EquivalenceNullablePolicy.SAME
-          && !otherField.getNullable().equals(getNullable())) {
-        return false;
-      } else if (nullablePolicy == EquivalenceNullablePolicy.WEAKEN) {
-        if (getNullable() && !otherField.getNullable()) {
-          return false;
-        }
-      }
-      return otherField.getName().equals(getName()) && getType().equivalent(otherField.getType());
+      return getName().equals(otherField.getName())
+          && getType().equivalent(otherField.getType(), nullablePolicy);
     }
 
     @Override
     public int hashCode() {
-      return Objects.hash(getName(), getDescription(), getType(), getNullable());
+      return Objects.hash(getName(), getDescription(), getType());
     }
   }
 
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Cast.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Cast.java
index 3048806..fc25214 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Cast.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Cast.java
@@ -148,7 +148,7 @@ public abstract class Cast<T> extends PTransform<PCollection<T>,
PCollection<Row
           return Collections.singletonList(
               CompatibilityError.create(context.path(), "Field is missing in output schema"));
         } else {
-          if (left.get().getNullable() && !right.get().getNullable()) {
+          if (left.get().getType().getNullable() && !right.get().getType().getNullable())
{
             return Collections.singletonList(
                 CompatibilityError.create(
                     context.path(), "Can't cast nullable field to non-nullable field"));
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Select.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Select.java
index 651cb16..fa49eda 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Select.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/Select.java
@@ -149,11 +149,8 @@ public class Select<T> extends PTransform<PCollection<T>,
PCollection<Row>> {
       FieldType nestedType =
           FieldType.row(getOutputSchema(field.getType().getRowSchema(), nestedDescriptor));
 
-      if (field.getNullable()) {
-        builder.addNullableField(field.getName(), nestedType);
-      } else {
-        builder.addField(field.getName(), nestedType);
-      }
+      nestedType = nestedType.withNullable(field.getType().getNullable());
+      builder.addField(field.getName(), nestedType);
     }
     return builder.build();
   }
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
index 23256a3..7f349c9 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
@@ -543,7 +543,7 @@ public abstract class Row implements Serializable {
         Object value = values.get(i);
         Schema.Field field = schema.getField(i);
         if (value == null) {
-          if (!field.getNullable()) {
+          if (!field.getType().getNullable()) {
             throw new IllegalArgumentException(
                 String.format("Field %s is not nullable", field.getName()));
           }
@@ -557,21 +557,11 @@ public abstract class Row implements Serializable {
 
     private Object verify(Object value, FieldType type, String fieldName) {
       if (TypeName.ARRAY.equals(type.getTypeName())) {
-        List<Object> arrayElements =
-            verifyArray(
-                value,
-                type.getCollectionElementType(),
-                type.getCollectionElementTypeNullable(),
-                fieldName);
+        List<Object> arrayElements = verifyArray(value, type.getCollectionElementType(),
fieldName);
         return arrayElements;
       } else if (TypeName.MAP.equals(type.getTypeName())) {
         Map<Object, Object> mapElements =
-            verifyMap(
-                value,
-                type.getMapKeyType().getTypeName(),
-                type.getMapValueType(),
-                type.getMapValueTypeNullable(),
-                fieldName);
+            verifyMap(value, type.getMapKeyType().getTypeName(), type.getMapValueType(),
fieldName);
         return mapElements;
       } else if (TypeName.ROW.equals(type.getTypeName())) {
         return verifyRow(value, fieldName);
@@ -581,10 +571,8 @@ public abstract class Row implements Serializable {
     }
 
     private List<Object> verifyArray(
-        Object value,
-        FieldType collectionElementType,
-        boolean collectionElementTypeNullable,
-        String fieldName) {
+        Object value, FieldType collectionElementType, String fieldName) {
+      boolean collectionElementTypeNullable = collectionElementType.getNullable();
       if (!(value instanceof List)) {
         throw new IllegalArgumentException(
             String.format(
@@ -610,11 +598,8 @@ public abstract class Row implements Serializable {
     }
 
     private Map<Object, Object> verifyMap(
-        Object value,
-        TypeName keyTypeName,
-        FieldType valueType,
-        boolean valueTypeNullable,
-        String fieldName) {
+        Object value, TypeName keyTypeName, FieldType valueType, String fieldName) {
+      boolean valueTypeNullable = valueType.getNullable();
       if (!(value instanceof Map)) {
         throw new IllegalArgumentException(
             String.format(
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java
index c0ea202..27599c2 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java
@@ -60,7 +60,7 @@ public class RowWithGetters extends Row {
     Field field = getSchema().getField(fieldIdx);
     FieldType type = field.getType();
     Object fieldValue = getters.get(fieldIdx).get(getterTarget);
-    if (fieldValue == null && !field.getNullable()) {
+    if (fieldValue == null && !field.getType().getNullable()) {
       throw new RuntimeException("Null value set on non-nullable field" + field);
     }
     return fieldValue != null ? getValue(type, fieldValue, fieldIdx) : null;
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/JavaBeanUtilsTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/JavaBeanUtilsTest.java
index 0f438b4..125be49 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/JavaBeanUtilsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/JavaBeanUtilsTest.java
@@ -61,8 +61,8 @@ public class JavaBeanUtilsTest {
   @Test
   public void testNullable() {
     Schema schema = JavaBeanUtils.schemaFromJavaBeanClass(NullableBean.class);
-    assertTrue(schema.getField("str").getNullable());
-    assertFalse(schema.getField("anInt").getNullable());
+    assertTrue(schema.getField("str").getType().getNullable());
+    assertFalse(schema.getField("anInt").getType().getNullable());
   }
 
   @Test
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/POJOUtilsTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/POJOUtilsTest.java
index 51ef8ff..1a0946d 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/POJOUtilsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/utils/POJOUtilsTest.java
@@ -63,8 +63,8 @@ public class POJOUtilsTest {
   @Test
   public void testNullables() {
     Schema schema = POJOUtils.schemaFromPojoClass(POJOWithNullables.class);
-    assertTrue(schema.getField("str").getNullable());
-    assertFalse(schema.getField("anInt").getNullable());
+    assertTrue(schema.getField("str").getType().getNullable());
+    assertFalse(schema.getField("anInt").getType().getNullable());
   }
 
   @Test
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlCreateExternalTable.java
b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlCreateExternalTable.java
index 3c779ca..e5d3de8 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlCreateExternalTable.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/parser/SqlCreateExternalTable.java
@@ -138,7 +138,7 @@ public class SqlCreateExternalTable extends SqlCreate implements SqlExecutableSt
     writer.identifier(column.getName());
     writer.identifier(CalciteUtils.toSqlTypeName(column.getType()).name());
 
-    if (column.getNullable() != null && !column.getNullable()) {
+    if (column.getType().getNullable() != null && !column.getType().getNullable())
{
       writer.keyword("NOT NULL");
     }
 
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java
b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java
index 3252082..9ddc642 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/schema/BeamTableUtils.java
@@ -92,7 +92,7 @@ public final class BeamTableUtils {
 
   public static Object autoCastField(Schema.Field field, Object rawObj) {
     if (rawObj == null) {
-      if (!field.getNullable()) {
+      if (!field.getType().getNullable()) {
         throw new IllegalArgumentException(String.format("Field %s not nullable", field.getName()));
       }
       return null;
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 5c22314..cbd0675 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
@@ -104,7 +104,7 @@ public class CalciteUtils {
       case MAP:
         return SqlTypeName.MAP;
       default:
-        SqlTypeName typeName = BEAM_TO_CALCITE_TYPE_MAPPING.get(type);
+        SqlTypeName typeName = BEAM_TO_CALCITE_TYPE_MAPPING.get(type.withNullable(false));
         if (typeName != null) {
           return typeName;
         } else {
@@ -191,7 +191,7 @@ public class CalciteUtils {
     Schema.Field field = schema.getField(fieldIndex);
     RelDataType type = toRelDataType(dataTypeFactory, field.getType());
 
-    return dataTypeFactory.createTypeWithNullability(type, field.getNullable());
+    return dataTypeFactory.createTypeWithNullability(type, field.getType().getNullable());
   }
 
   /**
diff --git a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java
b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java
index 547fe44..3185e91 100644
--- a/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java
+++ b/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/text/TextTableProvider.java
@@ -33,6 +33,7 @@ import org.apache.beam.sdk.extensions.sql.meta.Table;
 import org.apache.beam.sdk.extensions.sql.meta.provider.InMemoryMetaTableProvider;
 import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
 import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.Schema.TypeName;
 import org.apache.beam.sdk.transforms.FlatMapElements;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -95,9 +96,9 @@ public class TextTableProvider extends InMemoryMetaTableProvider {
       case "lines":
         checkArgument(
             schema.getFieldCount() == 1
-                && schema.getField(0).getType().equals(Schema.FieldType.STRING),
+                && schema.getField(0).getType().getTypeName().equals(TypeName.STRING),
             "Table with type 'text' and format 'lines' "
-                + "must have exactly one STRING/VARCHAR/CHAR column");
+                + "must have exactly one STRING/VARCHAR/CHAR column ");
         return new TextTable(
             schema, filePattern, new LinesReadConverter(), new LinesWriteConverter());
       default:
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java
index db86c21..649f8c8 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java
@@ -132,7 +132,7 @@ public class BigQueryUtils {
         field.setDescription(schemaField.getDescription());
       }
 
-      if (!schemaField.getNullable()) {
+      if (!schemaField.getType().getNullable()) {
         field.setMode(Mode.REQUIRED.toString());
       }
       if (TypeName.ARRAY == type.getTypeName()) {


Mime
View raw message