TheNeuralBit commented on code in PR #23014:
URL: https://github.com/apache/beam/pull/23014#discussion_r984059164
##########
model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto:
##########
@@ -139,6 +139,14 @@ message LogicalTypes {
// corresponds to chronological order.
MILLIS_INSTANT = 2 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:logical_type:millis_instant:v1"];
+
+ // A URN for Decimal type
+ // - Representation type: BYTES
+ // - A decimal number with specified scale and value. Its BYTES
Review Comment:
IIUC this type actually has _unspecified_ scale (while fixed_decimal has
specified scale). Is that right?
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalTypesTest.java:
##########
@@ -130,4 +133,51 @@ public void testSchema() {
schemaValue,
new SchemaLogicalType().toInputType(new
SchemaLogicalType().toBaseType(schemaValue)));
}
+
+ @Test
+ public void testFixedPrecisionNumeric() {
+ final int precision = 10;
+ final int scale = 2;
+ Random random = new Random();
+
+ Schema argumentSchema =
+
Schema.builder().addInt32Field("precision").addInt32Field("scale").build();
+
+ // invalid schema
+ final Schema invalidArgumentSchema =
+
Schema.builder().addInt32Field("invalid").addInt32Field("schema").build();
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ FixedPrecisionNumeric.of(
+ Row.withSchema(invalidArgumentSchema).addValues(precision,
scale).build()));
+
+ // FixedPrecisionNumeric specified precision and scale
+ FixedPrecisionNumeric numeric =
+ FixedPrecisionNumeric.of(
+ Row.withSchema(argumentSchema).addValues(precision,
scale).build());
+ Schema schema = Schema.builder().addLogicalTypeField("decimal",
numeric).build();
+
+ // check argument valid case
+ BigDecimal decimal = BigDecimal.valueOf(random.nextInt(), scale);
+ Row row = Row.withSchema(schema).addValues(decimal).build();
+ assertEquals(decimal, row.getLogicalTypeValue(0, BigDecimal.class));
+
+ // check argument invalid case
+ decimal = BigDecimal.valueOf(random.nextLong() + 100_000_000_000L, scale);
+ assertThrows(IllegalArgumentException.class,
Row.withSchema(schema).addValues(decimal)::build);
Review Comment:
Nice, I like how concise this is :)
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalTypesTest.java:
##########
@@ -130,4 +133,51 @@ public void testSchema() {
schemaValue,
new SchemaLogicalType().toInputType(new
SchemaLogicalType().toBaseType(schemaValue)));
}
+
+ @Test
+ public void testFixedPrecisionNumeric() {
+ final int precision = 10;
+ final int scale = 2;
+ Random random = new Random();
+
+ Schema argumentSchema =
+
Schema.builder().addInt32Field("precision").addInt32Field("scale").build();
+
+ // invalid schema
+ final Schema invalidArgumentSchema =
+
Schema.builder().addInt32Field("invalid").addInt32Field("schema").build();
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ FixedPrecisionNumeric.of(
+ Row.withSchema(invalidArgumentSchema).addValues(precision,
scale).build()));
+
+ // FixedPrecisionNumeric specified precision and scale
+ FixedPrecisionNumeric numeric =
+ FixedPrecisionNumeric.of(
+ Row.withSchema(argumentSchema).addValues(precision,
scale).build());
+ Schema schema = Schema.builder().addLogicalTypeField("decimal",
numeric).build();
+
+ // check argument valid case
+ BigDecimal decimal = BigDecimal.valueOf(random.nextInt(), scale);
+ Row row = Row.withSchema(schema).addValues(decimal).build();
+ assertEquals(decimal, row.getLogicalTypeValue(0, BigDecimal.class));
+
+ // check argument invalid case
+ decimal = BigDecimal.valueOf(random.nextLong() + 100_000_000_000L, scale);
+ assertThrows(IllegalArgumentException.class,
Row.withSchema(schema).addValues(decimal)::build);
+
+ // FixedPrecisionNumeric without specifying precision
+ numeric = FixedPrecisionNumeric.of(scale);
+ schema = Schema.builder().addLogicalTypeField("decimal", numeric).build();
+
+ // check argument always valid
+ decimal = BigDecimal.valueOf(random.nextInt(), scale);
+ row = Row.withSchema(schema).addValues(decimal).build();
+ assertEquals(decimal, row.getLogicalTypeValue(0, BigDecimal.class));
+
+ decimal = BigDecimal.valueOf(random.nextLong() + 100_000_000_000L, scale);
Review Comment:
nit: I'm a little nervous that the random component can make this flaky. Can
we just select some constants istead?
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalTypesTest.java:
##########
@@ -130,4 +133,51 @@ public void testSchema() {
schemaValue,
new SchemaLogicalType().toInputType(new
SchemaLogicalType().toBaseType(schemaValue)));
}
+
+ @Test
+ public void testFixedPrecisionNumeric() {
+ final int precision = 10;
+ final int scale = 2;
+ Random random = new Random();
+
+ Schema argumentSchema =
+
Schema.builder().addInt32Field("precision").addInt32Field("scale").build();
+
+ // invalid schema
+ final Schema invalidArgumentSchema =
+
Schema.builder().addInt32Field("invalid").addInt32Field("schema").build();
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ FixedPrecisionNumeric.of(
+ Row.withSchema(invalidArgumentSchema).addValues(precision,
scale).build()));
Review Comment:
nit: I think unit testing best practice would be to put each of these test
cases in a different test method. On the other hand this pattern is consistent
with the rest of the file, so I'm fine if you want to leave it as-is.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedPrecisionNumeric.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.MoreObjects.firstNonNull;
+import static
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.SchemaApi;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.values.Row;
+
+/** Fixed precision numeric types used to represent jdbc NUMERIC and DECIMAL
types. */
+public class FixedPrecisionNumeric extends PassThroughLogicalType<BigDecimal> {
+ public static final String IDENTIFIER = "beam:logical_type:fixed_decimal:v1";
Review Comment:
Since this is being used cross-language can we define the URN in
schema.proto and document it there as well?
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedPrecisionNumeric.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.MoreObjects.firstNonNull;
+import static
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.values.Row;
+
+/** Fixed precision numeric types used to represent jdbc NUMERIC and DECIMAL
types. */
+public class FixedPrecisionNumeric extends PassThroughLogicalType<BigDecimal> {
+ public static final String IDENTIFIER = "beam:logical_type:fixed_decimal:v1";
+
+ // TODO(https://github.com/apache/beam/issues/19817) implement
beam:logical_type:decimal:v1 as
+ // CoderLogicalType (once CoderLogicalType is implemented).
+ /**
+ * Identifier of the unspecified precision numeric type. It corresponds to
Java SDK's {@link
+ * FieldType#DECIMAL}. It is the underlying representation type of
FixedPrecisionNumeric logical
+ * type in order to be compatible with existing Java field types.
+ */
+ public static final String BASE_IDENTIFIER = "beam:logical_type:decimal:v1";
+
+ private final int precision;
+ private final int scale;
+
+ /**
+ * Create a FixedPrecisionNumeric instance with specified precision and
scale. ``precision=-1``
+ * indicates unspecified precision.
+ */
+ public static FixedPrecisionNumeric of(int precision, int scale) {
+ Schema schema =
Schema.builder().addInt32Field("precision").addInt32Field("scale").build();
+ return new FixedPrecisionNumeric(schema, precision, scale);
+ }
+
+ /** Create a FixedPrecisionNumeric instance with specified scale and
unspecified precision. */
+ public static FixedPrecisionNumeric of(int scale) {
+ return of(-1, scale);
+ }
+
+ /** Create a FixedPrecisionNumeric instance with specified argument row. */
+ public static FixedPrecisionNumeric of(Row row) {
+ final Integer precision = row.getInt32("precision");
+ final Integer scale = row.getInt32("scale");
+ checkArgument(
+ precision != null && scale != null,
+ "precision and scale cannot be null for FixedPrecisionNumeric
arguments.");
+ // firstNonNull is used to cast precision and scale to @NonNull input
+ return of(firstNonNull(precision, -1), firstNonNull(scale, 0));
+ }
+
+ private FixedPrecisionNumeric(Schema schema, int precision, int scale) {
+ super(
+ IDENTIFIER,
+ FieldType.row(schema),
+ Row.withSchema(schema).addValues(precision, scale).build(),
+ FieldType.DECIMAL);
+ this.precision = precision;
+ this.scale = scale;
+ }
+
+ @Override
+ public BigDecimal toInputType(BigDecimal base) {
+ checkArgument(
+ base == null
+ || (base.precision() <= precision && base.scale() <= scale)
+ // for cases when received values can be safely coerced to the
schema
+ || base.round(new MathContext(precision)).compareTo(base) == 0,
+ "Expected BigDecimal base to be null or have precision <= %s (was %s),
scale <= %s (was %s)",
+ precision,
+ (base == null) ? null : base.precision(),
+ scale,
+ (base == null) ? null : base.scale());
+ return base;
Review Comment:
Ah sorry about that, thank you
##########
sdks/python/apache_beam/typehints/schemas.py:
##########
@@ -751,3 +765,87 @@ def to_representation_type(self, value):
def to_language_type(self, value):
# type: (str) -> PythonCallableWithSource
return PythonCallableWithSource(value)
+
+
+FixedPrecisionDecimalArgumentRepresentation = NamedTuple(
+ 'FixedPrecisionDecimalArgumentRepresentation', [('precision', np.int32),
+ ('scale', np.int32)])
+
+
+class DecimalLogicalType(NoArgumentLogicalType[decimal.Decimal, bytes]):
+ """A logical type for decimal objects handling values consistent with that
+ encoded by ``BigDecimalCoder`` in the Java SDK.
+ """
+ @classmethod
+ def urn(cls):
+ return common_urns.decimal.urn
+
+ @classmethod
+ def representation_type(cls):
+ # type: () -> type
+ return bytes
+
+ @classmethod
+ def language_type(cls):
+ return decimal.Decimal
+
+ def to_representation_type(self, value):
+ # type: (decimal.Decimal) -> bytes
+ return str(value).encode()
+
+ def to_language_type(self, value):
+ # type: (bytes) -> decimal.Decimal
+ return decimal.Decimal(value.decode())
+
+
[email protected]_logical_type
+class FixedPrecisionDecimalLogicalType(
+ LogicalType[decimal.Decimal,
+ DecimalLogicalType,
+ FixedPrecisionDecimalArgumentRepresentation]):
+ """A wrapper of DecimalLogicalType that contains the precision value.
+ """
+ def __init__(self, precision=-1, scale=0):
+ self.precision = precision
+ self.scale = scale
+
+ @classmethod
+ def urn(cls):
+ return "beam:logical_type:fixed_decimal:v1"
+
+ @classmethod
+ def representation_type(cls):
+ # type: () -> type
+ return DecimalLogicalType
+
+ @classmethod
+ def language_type(cls):
+ return decimal.Decimal
+
+ def to_representation_type(self, value):
+ # type: (decimal.Decimal) -> bytes
+
+ return DecimalLogicalType().to_representation_type(value)
+
+ def to_language_type(self, value):
+ # type: (bytes) -> decimal.Decimal
+
+ return DecimalLogicalType().to_language_type(value)
+
+ @classmethod
+ def argument_type(cls):
+ return FixedPrecisionDecimalArgumentRepresentation
+
+ def argument(self):
+ return FixedPrecisionDecimalArgumentRepresentation(
+ precision=self.precision, scale=self.scale)
+
+ @classmethod
+ def _from_typing(cls, typ):
+ print(typ)
Review Comment:
This looks like a debug statement that was left in?
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedPrecisionNumeric.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.MoreObjects.firstNonNull;
+import static
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.values.Row;
+
+/** Fixed precision numeric types used to represent jdbc NUMERIC and DECIMAL
types. */
+public class FixedPrecisionNumeric extends PassThroughLogicalType<BigDecimal> {
+ public static final String IDENTIFIER = "beam:logical_type:fixed_decimal:v1";
+
+ // TODO(https://github.com/apache/beam/issues/19817) implement
beam:logical_type:decimal:v1 as
+ // CoderLogicalType (once CoderLogicalType is implemented).
Review Comment:
Thank you!
--
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]
