[BEAM-8307] NPE in Calcite dialect when input PCollection has logical…

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedBytes.java

@@ -19,19 +19,18 @@
 
 import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
 
-import java.util.Arrays;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
-import org.apache.beam.sdk.schemas.Schema.FieldType;
-import org.apache.beam.sdk.schemas.Schema.LogicalType;
+import org.apache.beam.sdk.schemas.Schema;
 
 /** A LogicalType representing a fixed-size byte array. */
 @Experimental(Kind.SCHEMAS)
-public class FixedBytes implements LogicalType<byte[], byte[]> {
-  public static final String IDENTIFIER = "FixedBytes";
+public class FixedBytes extends IdenticalBaseTAndInputTLogicalType<byte[]> {
+  public static final String IDENTIFIER = "beam:logical_type:fixed_length_bytes:v1";
   private final int byteArraySize;
 
   private FixedBytes(int byteArraySize) {
+    super(IDENTIFIER, Schema.FieldType.INT32, byteArraySize, Schema.FieldType.BYTES);
     this.byteArraySize = byteArraySize;
   }
 
@@ -43,44 +42,15 @@ public int getLength() {
     return byteArraySize;
   }
 
-  @Override
-  public String getIdentifier() {
-    return IDENTIFIER;
-  }
-
-  @Override
-  public FieldType getArgumentType() {
-    return FieldType.INT32;
-  }
-
-  @Override
-  public Integer getArgument() {
-    return byteArraySize;
-  }
-
-  @Override
-  public FieldType getBaseType() {
-    return FieldType.BYTES;
-  }
-
   @Override
   public byte[] toBaseType(byte[] input) {
-    checkArgument(input.length == byteArraySize);
+    checkArgument(input == null || input.length == byteArraySize);
     return input;
   }
 
   @Override
   public byte[] toInputType(byte[] base) {
-    checkArgument(base.length <= byteArraySize);
-    if (base.length == byteArraySize) {
-      return base;
-    } else {
-      return Arrays.copyOf(base, byteArraySize);
-    }
-  }
-
-  @Override
-  public String toString() {
-    return "FixedBytes: " + byteArraySize;
+    checkArgument(base == null || base.length == byteArraySize);
+    return base;
   }
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedLengthString.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.schemas.logicaltypes;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.schemas.Schema;
+
+/** A LogicalType representing a fixed-length string. */
+@Experimental(Experimental.Kind.SCHEMAS)
+public class FixedLengthString extends IdenticalBaseTAndInputTLogicalType<String> {
+  public static final String IDENTIFIER = "beam:logical_type:fixed_length_string:v1";
+  private final int length;
+
+  private FixedLengthString(int length) {
+    super(IDENTIFIER, Schema.FieldType.INT32, length, Schema.FieldType.STRING);
+    this.length = length;
+  }
+
+  public int getLength() {
+    return length;
+  }
+
+  public static FixedLengthString of(int length) {
+    return new FixedLengthString(length);
+  }
+
+  @Override
+  public String toBaseType(String input) {
+    checkArgument(input == null || input.length() == length);
+    return input;
+  }
+
+  @Override
+  public String toInputType(String base) {
+    checkArgument(base == null || base.length() == length);
+    return base;
+  }
+}

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/IdenticalBaseTAndInputTLogicalType.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.schemas.logicaltypes;
+
+import java.util.Objects;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.schemas.Schema;
+
+/** A base class for LogicalTypes that use the same input type as the underlying base type. */
+@Experimental(Experimental.Kind.SCHEMAS)
+public abstract class IdenticalBaseTAndInputTLogicalType<T> implements Schema.LogicalType<T, T> {
+  protected final String identifier;
+  protected final Schema.FieldType argumentType;
+  protected final Object argument;
+  protected final Schema.FieldType baseType;
+
+  protected IdenticalBaseTAndInputTLogicalType(
+      String identifier,
+      Schema.FieldType argumentType,
+      Object argument,
+      Schema.FieldType baseType) {
+    this.identifier = identifier;
+    this.argumentType = argumentType;
+    this.argument = argument;
+    this.baseType = baseType;
+  }
+
+  @Override
+  public String getIdentifier() {
+    return identifier;
+  }
+
+  @Override
+  public Schema.FieldType getArgumentType() {
+    return argumentType;
+  }
+
+  @Override
+  @SuppressWarnings("TypeParameterUnusedInFormals")
+  public <ArgumentT> ArgumentT getArgument() {
+    return (ArgumentT) argument;
+  }
+
+  @Override
+  public Schema.FieldType getBaseType() {
+    return baseType;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    IdenticalBaseTAndInputTLogicalType<?> that = (IdenticalBaseTAndInputTLogicalType<?>) o;
+    return identifier.equals(that.identifier)
+        && argument.equals(that.argument)
+        && baseType.equals(that.baseType);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(identifier, argument, baseType);
+  }
+}

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalDecimal.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.schemas.logicaltypes;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.math.BigDecimal;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.values.Row;
+
+/** A LogicalType representing a Decimal type with custom precision and scale. */
+@Experimental(Experimental.Kind.SCHEMAS)
+public class LogicalDecimal extends IdenticalBaseTAndInputTLogicalType<BigDecimal> {
+  public static final String IDENTIFIER = "beam:logical_type:decimal:v1";
+  private static final Schema schema =
+      Schema.builder().addInt32Field("precision").addInt32Field("scale").build();
+  private final int precision;
+  private final int scale;
+
+  private LogicalDecimal(int precision, int scale) {
+    super(
+        IDENTIFIER,
+        Schema.FieldType.row(schema),
+        Row.withSchema(schema).addValues(precision, scale).build(),
+        Schema.FieldType.DECIMAL);
+    this.precision = precision;
+    this.scale = scale;
+  }
+
+  public static LogicalDecimal of(int precision, int scale) {
+    return new LogicalDecimal(precision, scale);
+  }
+
+  public int getPrecision() {
+    return precision;
+  }
+
+  public int getScale() {
+    return scale;
+  }
+
+  @Override
+  public BigDecimal toBaseType(BigDecimal input) {
+    checkArgument(input == null || (input.precision() == precision && input.scale() == scale));
+    return input;
+  }
+
+  @Override
+  public BigDecimal toInputType(BigDecimal base) {
+    checkArgument(base == null || (base.precision() == precision && base.scale() == scale));
+    return base;
+  }
+}

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/PassThroughLogicalType.java

@@ -20,43 +20,14 @@
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
 import org.apache.beam.sdk.schemas.Schema.FieldType;
-import org.apache.beam.sdk.schemas.Schema.LogicalType;
 
-/** A base class for LogicalTypes that use the same Java type as the underlying base type. */
+/** A base class for LogicalTypes that use the same value as the underlying base value. */
 @Experimental(Kind.SCHEMAS)
-public abstract class PassThroughLogicalType<T> implements LogicalType<T, T> {
-  private final String identifier;
-  private final FieldType argumentType;
-  private final Object argument;
-  private final FieldType fieldType;
+public abstract class PassThroughLogicalType<T> extends IdenticalBaseTAndInputTLogicalType<T> {
 
   protected PassThroughLogicalType(
       String identifier, FieldType argumentType, Object argument, FieldType fieldType) {
-    this.identifier = identifier;
-    this.argumentType = argumentType;
-    this.argument = argument;
-    this.fieldType = fieldType;
-  }
-
-  @Override
-  public String getIdentifier() {
-    return identifier;
-  }
-
-  @Override
-  public FieldType getArgumentType() {
-    return argumentType;
-  }
-
-  @Override
-  @SuppressWarnings("TypeParameterUnusedInFormals")
-  public <ArgumentT> ArgumentT getArgument() {
-    return (ArgumentT) argument;
-  }
-
-  @Override
-  public FieldType getBaseType() {
-    return fieldType;
+    super(identifier, argumentType, argument, fieldType);
   }
 
   @Override

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/VariableLengthBytes.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.schemas.logicaltypes;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.schemas.Schema;
+
+/** A LogicalType representing a variable-size byte array. */
+@Experimental(Experimental.Kind.SCHEMAS)
+public class VariableLengthBytes extends IdenticalBaseTAndInputTLogicalType<byte[]> {
+  public static final String IDENTIFIER = "beam:logical_type:variable_length_bytes:v1";
+  private final int byteArraySize;
+
+  private VariableLengthBytes(int byteArraySize) {
+    super(IDENTIFIER, Schema.FieldType.INT32, byteArraySize, Schema.FieldType.BYTES);
+    this.byteArraySize = byteArraySize;
+  }
+
+  public static VariableLengthBytes of(int byteArraySize) {
+    return new VariableLengthBytes(byteArraySize);
+  }
+
+  public int getLength() {
+    return byteArraySize;
+  }
+
+  @Override
+  public byte[] toBaseType(byte[] input) {
+    checkArgument(input == null || input.length <= byteArraySize);
+    return input;
+  }
+
+  @Override
+  public byte[] toInputType(byte[] base) {
+    checkArgument(base == null || base.length <= byteArraySize);
+    return base;
+  }
+}