You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by bh...@apache.org on 2022/10/03 23:47:46 UTC
[beam] branch master updated: Support DECIMAL logical type in python SDK (#23014)
This is an automated email from the ASF dual-hosted git repository.
bhulette 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 2955954402f Support DECIMAL logical type in python SDK (#23014)
2955954402f is described below
commit 2955954402f8c10aecee6d4d2230847160d3fb40
Author: Yi Hu <hu...@gmail.com>
AuthorDate: Mon Oct 3 19:47:37 2022 -0400
Support DECIMAL logical type in python SDK (#23014)
* Support DECIMAL logical type in python sdk
* migrate FixedPrecisionNumeric jdbclogicaltype to portable
logical type
* Support DECIMAL type in python sdk xlang jdbc transform
* Support standard logical type with argument in java sdk
* proto support logical type's argument type in java sdk.
Support of logical type's argument value is still pending
* Implement BigIntegerCoder and DecimalCoder in python sdk
* Add standard coder and tests
* Add decimal logical type in schema.proto
* Add unit test for FixedPrecisionNumeric
* Assert FixedPrecisionNumeric argument row assinable
* Add decimal test case in standard_coders.yaml
* Fix static assignment causing NoClassDefFoundError when initializing from serialized
* Add comments regarding to fixed_decimal URN, fix typo and leftover
* Update model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto
fix incorrect documentation of Decimal logical type
Co-authored-by: Brian Hulette <hu...@gmail.com>
Co-authored-by: Brian Hulette <hu...@gmail.com>
---
.../beam/model/fnexecution/v1/standard_coders.yaml | 12 +++
.../org/apache/beam/model/pipeline/v1/schema.proto | 8 ++
.../runners/fnexecution/wire/CommonCoderTest.java | 5 +-
.../go/test/regression/coders/fromyaml/fromyaml.go | 1 +
.../apache/beam/sdk/schemas/SchemaTranslation.java | 119 ++++++++++++++-------
.../logicaltypes/FixedPrecisionNumeric.java | 116 ++++++++++++++++++++
.../sdk/schemas/logicaltypes/LogicalTypesTest.java | 48 +++++++++
.../java/org/apache/beam/sdk/io/jdbc/JdbcUtil.java | 7 ++
.../org/apache/beam/sdk/io/jdbc/LogicalTypes.java | 44 +-------
.../org/apache/beam/sdk/io/jdbc/SchemaUtil.java | 23 ++--
.../org/apache/beam/sdk/io/jdbc/JdbcIOTest.java | 11 +-
sdks/python/apache_beam/coders/coder_impl.pxd | 12 +++
sdks/python/apache_beam/coders/coder_impl.py | 45 ++++++++
sdks/python/apache_beam/coders/coders.py | 41 ++++++-
.../apache_beam/coders/coders_test_common.py | 23 ++++
sdks/python/apache_beam/coders/row_coder.py | 3 +
.../io/external/xlang_jdbcio_it_test.py | 26 +++--
sdks/python/apache_beam/portability/common_urns.py | 1 +
sdks/python/apache_beam/typehints/schemas.py | 112 +++++++++++++++++--
19 files changed, 543 insertions(+), 114 deletions(-)
diff --git a/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml b/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml
index ef06fe17967..ca914b6e6f8 100644
--- a/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml
+++ b/model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml
@@ -475,6 +475,18 @@ examples:
---
+coder:
+ urn: "beam:coder:row:v1"
+ # f_float: float32, f_decimal: logical(decimal)
+ payload: "\n\r\n\x07f_float\x1a\x02\x10\x05\n1\n\tf_decimal\x1a$:\"\n\x1cbeam:logical_type:decimal:v1\x1a\x02\x10\t\x12$800c44ae-a1b7-4def-bbf6-6217cca89ec4"
+examples:
+ "\x02\x00\x00\x00\x00\x00\x01\x01\x00": {f_float: "0.0", f_decimal: "0.0"}
+ "\x02\x00?\x80\x00\x00\x01\x01\n": {f_float: "1.0", f_decimal: "1.0"}
+ "\x02\x00@I\x0eV\x04\x02z\xb7": {f_float: "3.1415", f_decimal: "3.1415"}
+ "\x02\x00\xc2\xc8>\xfa\x03\x03\xfex\xe5": {f_float: "-100.123", f_decimal: "-100.123"}
+
+---
+
coder:
urn: "beam:coder:sharded_key:v1"
components: [{urn: "beam:coder:string_utf8:v1"}]
diff --git a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto
index 20de76d53da..fa626bc747f 100644
--- a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto
+++ b/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 variable scale. Its BYTES
+ // representation consists of an integer (INT32) scale followed by a
+ // two's complement encoded big integer.
+ DECIMAL = 3 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:logical_type:decimal:v1"];
}
}
diff --git a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/wire/CommonCoderTest.java b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/wire/CommonCoderTest.java
index 46131826ea8..b5e0251033a 100644
--- a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/wire/CommonCoderTest.java
+++ b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/wire/CommonCoderTest.java
@@ -40,6 +40,7 @@ import com.google.auto.value.AutoValue;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
+import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
@@ -425,6 +426,8 @@ public class CommonCoderTest {
case DATETIME:
// convert shifted millis to epoch millis as in InstantCoder
return new Instant((Long) value + -9223372036854775808L);
+ case DECIMAL:
+ return new BigDecimal((String) value);
case BYTES:
// extract String as byte[]
return ((String) value).getBytes(StandardCharsets.ISO_8859_1);
@@ -468,7 +471,7 @@ public class CommonCoderTest {
return fieldType
.getLogicalType()
.toInputType(parseField(value, fieldType.getLogicalType().getBaseType()));
- default: // DECIMAL
+ default:
throw new IllegalArgumentException("Unsupported type name: " + fieldType.getTypeName());
}
}
diff --git a/sdks/go/test/regression/coders/fromyaml/fromyaml.go b/sdks/go/test/regression/coders/fromyaml/fromyaml.go
index 2a8476fe65a..86d7969c2a0 100644
--- a/sdks/go/test/regression/coders/fromyaml/fromyaml.go
+++ b/sdks/go/test/regression/coders/fromyaml/fromyaml.go
@@ -52,6 +52,7 @@ var filteredCases = []struct{ filter, reason string }{
{"logical", "BEAM-9615: Support logical types"},
{"30ea5a25-dcd8-4cdb-abeb-5332d15ab4b9", "https://github.com/apache/beam/issues/21206: Support encoding position."},
{"80be749a-5700-4ede-89d8-dd9a4433a3f8", "https://github.com/apache/beam/issues/19817: Support millis_instant."},
+ {"800c44ae-a1b7-4def-bbf6-6217cca89ec4", "https://github.com/apache/beam/issues/19817: Support decimal."},
}
// Coder is a representation a serialized beam coder.
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java
index c1667cc9684..f38e50aea6f 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java
@@ -22,6 +22,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
+import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -43,6 +44,7 @@ import org.apache.beam.sdk.schemas.Schema.Field;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.schemas.Schema.LogicalType;
import org.apache.beam.sdk.schemas.Schema.TypeName;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
import org.apache.beam.sdk.schemas.logicaltypes.MicrosInstant;
import org.apache.beam.sdk.schemas.logicaltypes.PythonCallable;
import org.apache.beam.sdk.schemas.logicaltypes.SchemaLogicalType;
@@ -65,7 +67,7 @@ import org.checkerframework.checker.nullness.qual.Nullable;
})
public class SchemaTranslation {
- private static final String URN_BEAM_LOGICAL_DECIMAL = "beam:logical_type:decimal:v1";
+ private static final String URN_BEAM_LOGICAL_DECIMAL = FixedPrecisionNumeric.BASE_IDENTIFIER;
private static final String URN_BEAM_LOGICAL_JAVASDK = "beam:logical_type:javasdk:v1";
private static final String URN_BEAM_LOGICAL_MILLIS_INSTANT =
SchemaApi.LogicalTypes.Enum.MILLIS_INSTANT
@@ -79,6 +81,7 @@ public class SchemaTranslation {
private static final ImmutableMap<String, Class<? extends LogicalType<?, ?>>>
STANDARD_LOGICAL_TYPES =
ImmutableMap.<String, Class<? extends LogicalType<?, ?>>>builder()
+ .put(FixedPrecisionNumeric.IDENTIFIER, FixedPrecisionNumeric.class)
.put(MicrosInstant.IDENTIFIER, MicrosInstant.class)
.put(SchemaLogicalType.IDENTIFIER, SchemaLogicalType.class)
.put(PythonCallable.IDENTIFIER, PythonCallable.class)
@@ -147,17 +150,10 @@ public class SchemaTranslation {
case LOGICAL_TYPE:
LogicalType logicalType = fieldType.getLogicalType();
SchemaApi.LogicalType.Builder logicalTypeBuilder;
- if (STANDARD_LOGICAL_TYPES.containsKey(logicalType.getIdentifier())) {
- Preconditions.checkArgument(
- logicalType.getArgumentType() == null,
- "Logical type '%s' cannot be used as a logical type, it has a non-null argument type.",
- logicalType.getIdentifier());
- logicalTypeBuilder =
- SchemaApi.LogicalType.newBuilder()
- .setRepresentation(
- fieldTypeToProto(logicalType.getBaseType(), serializeLogicalType))
- .setUrn(logicalType.getIdentifier());
- } else if (logicalType instanceof UnknownLogicalType) {
+ String identifier = logicalType.getIdentifier();
+ boolean isStandard = STANDARD_LOGICAL_TYPES.containsKey(identifier);
+
+ if (!isStandard && logicalType instanceof UnknownLogicalType) {
logicalTypeBuilder =
SchemaApi.LogicalType.newBuilder()
.setUrn(logicalType.getIdentifier())
@@ -173,14 +169,16 @@ public class SchemaTranslation {
fieldValueToProto(logicalType.getArgumentType(), logicalType.getArgument()));
}
} else {
+ // TODO(https://github.com/apache/beam/issues/19715): "javasdk" types should only
+ // be a last resort. Types defined in Beam should have their own URN, and there
+ // should be a mechanism for users to register their own types by URN.
+ String urn =
+ identifier.startsWith("beam:logical_type:") ? identifier : URN_BEAM_LOGICAL_JAVASDK;
logicalTypeBuilder =
SchemaApi.LogicalType.newBuilder()
.setRepresentation(
fieldTypeToProto(logicalType.getBaseType(), serializeLogicalType))
- // TODO(https://github.com/apache/beam/issues/19715): "javasdk" types should only
- // be a last resort. Types defined in Beam should have their own URN, and there
- // should be a mechanism for users to register their own types by URN.
- .setUrn(URN_BEAM_LOGICAL_JAVASDK);
+ .setUrn(urn);
if (logicalType.getArgumentType() != null) {
logicalTypeBuilder =
logicalTypeBuilder
@@ -190,7 +188,8 @@ public class SchemaTranslation {
fieldValueToProto(
logicalType.getArgumentType(), logicalType.getArgument()));
}
- if (serializeLogicalType) {
+ // No need to embed serialized bytes to payload for standard (known) logical type
+ if (!isStandard && serializeLogicalType) {
logicalTypeBuilder =
logicalTypeBuilder.setPayload(
ByteString.copyFrom(SerializableUtils.serializeToByteArray(logicalType)));
@@ -208,6 +207,8 @@ public class SchemaTranslation {
.build());
break;
case DECIMAL:
+ // DECIMAL without precision specified. Used as the representation type of
+ // FixedPrecisionNumeric logical type.
builder.setLogicalType(
SchemaApi.LogicalType.newBuilder()
.setUrn(URN_BEAM_LOGICAL_DECIMAL)
@@ -337,28 +338,74 @@ public class SchemaTranslation {
fieldTypeFromProto(protoFieldType.getMapType().getKeyType()),
fieldTypeFromProto(protoFieldType.getMapType().getValueType()));
case LOGICAL_TYPE:
- String urn = protoFieldType.getLogicalType().getUrn();
SchemaApi.LogicalType logicalType = protoFieldType.getLogicalType();
+ String urn = logicalType.getUrn();
Class<? extends LogicalType<?, ?>> logicalTypeClass = STANDARD_LOGICAL_TYPES.get(urn);
if (logicalTypeClass != null) {
- try {
- return FieldType.logicalType(logicalTypeClass.getConstructor().newInstance());
- } catch (NoSuchMethodException e) {
- throw new RuntimeException(
- String.format(
- "Standard logical type '%s' does not have a zero-argument constructor.", urn),
- e);
- } catch (IllegalAccessException e) {
- throw new RuntimeException(
- String.format(
- "Standard logical type '%s' has a zero-argument constructor, but it is not accessible.",
- urn),
- e);
- } catch (ReflectiveOperationException e) {
- throw new RuntimeException(
- String.format(
- "Error instantiating logical type '%s' with zero-argument constructor.", urn),
- e);
+ boolean hasArgument = logicalType.hasArgument();
+ if (hasArgument) {
+ // Logical type with argument. Construct from compatible of() method with single
+ // argument type is either a primitive, List, Map, or Row.
+ FieldType fieldType = fieldTypeFromProto(logicalType.getArgumentType());
+ Object fieldValue =
+ Objects.requireNonNull(fieldValueFromProto(fieldType, logicalType.getArgument()));
+ Class clazz = fieldValue.getClass();
+ if (fieldValue instanceof List) {
+ // argument is ArrayValue or iterableValue
+ clazz = List.class;
+ }
+ if (fieldValue instanceof Map) {
+ // argument is Map
+ clazz = Map.class;
+ } else if (fieldValue instanceof Row) {
+ // argument is Row
+ clazz = Row.class;
+ }
+ String objectName = clazz.getName();
+ try {
+ return FieldType.logicalType(
+ logicalTypeClass.cast(
+ logicalTypeClass.getMethod("of", clazz).invoke(null, fieldValue)));
+ } catch (NoSuchMethodException e) {
+ throw new RuntimeException(
+ String.format(
+ "Standard logical type '%s' does not have a static of('%s') method.",
+ urn, objectName),
+ e);
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(
+ String.format(
+ "Standard logical type '%s' has an of('%s') method, but it is not accessible.",
+ urn, objectName),
+ e);
+ } catch (InvocationTargetException e) {
+ throw new RuntimeException(
+ String.format(
+ "Error instantiating logical type '%s' with of('%s') method.",
+ urn, objectName),
+ e);
+ }
+ } else {
+ // Logical type without argument. Construct from constructor without parameter
+ try {
+ return FieldType.logicalType(logicalTypeClass.getConstructor().newInstance());
+ } catch (NoSuchMethodException e) {
+ throw new RuntimeException(
+ String.format(
+ "Standard logical type '%s' does not have a zero-argument constructor.", urn),
+ e);
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(
+ String.format(
+ "Standard logical type '%s' has a zero-argument constructor, but it is not accessible.",
+ urn),
+ e);
+ } catch (ReflectiveOperationException e) {
+ throw new RuntimeException(
+ String.format(
+ "Error instantiating logical type '%s' with zero-argument constructor.", urn),
+ e);
+ }
}
}
// Special-case for DATETIME and DECIMAL which are logical types in portable representation,
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedPrecisionNumeric.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedPrecisionNumeric.java
new file mode 100644
index 00000000000..63be36ec1be
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedPrecisionNumeric.java
@@ -0,0 +1,116 @@
+/*
+ * 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> {
+ // TODO(https://github.com/apache/beam/issues/23373) promote this URN to schema.proto once logical
+ // types with arguments are fully supported and the implementation of this logical type can thus
+ // be considered standardized.
+ public static final String IDENTIFIER = "beam:logical_type:fixed_decimal:v1";
+
+ // TODO(https://github.com/apache/beam/issues/23374) 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;
+
+ private static final Schema SCHEMA;
+
+ static {
+ BASE_IDENTIFIER =
+ SchemaApi.LogicalTypes.Enum.DECIMAL
+ .getValueDescriptor()
+ .getOptions()
+ .getExtension(RunnerApi.beamUrn);
+ SCHEMA = Schema.builder().addInt32Field("precision").addInt32Field("scale").build();
+ }
+
+ 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) {
+ return new FixedPrecisionNumeric(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) {
+ checkArgument(
+ row.getSchema().assignableTo(SCHEMA),
+ "Row has an incompatible schema to construct the logical type object: %s",
+ row.getSchema());
+
+ 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(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) {
+ if (precision != -1) {
+ // check value not causing overflow when precision is fixed.
+ checkArgument(
+ base == null
+ || (base.precision() <= precision && base.scale() <= scale)
+ || 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;
+ }
+}
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalTypesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalTypesTest.java
index 8ee20f1ad16..2de9096d2b4 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalTypesTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/logicaltypes/LogicalTypesTest.java
@@ -18,7 +18,9 @@
package org.apache.beam.sdk.schemas.logicaltypes;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import java.math.BigDecimal;
import java.time.Duration;
import java.time.Instant;
import java.util.Map;
@@ -130,4 +132,50 @@ public class LogicalTypesTest {
schemaValue,
new SchemaLogicalType().toInputType(new SchemaLogicalType().toBaseType(schemaValue)));
}
+
+ @Test
+ public void testFixedPrecisionNumeric() {
+ final int precision = 10;
+ final int scale = 2;
+
+ 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(1_000_000_001, scale);
+ Row row = Row.withSchema(schema).addValues(decimal).build();
+ assertEquals(decimal, row.getLogicalTypeValue(0, BigDecimal.class));
+
+ // check argument invalid case (value out of precision limit)
+ decimal = BigDecimal.valueOf(100_000_000_001L, 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 (precision limit unspecified)
+ decimal = BigDecimal.valueOf(1_000_000_001, scale);
+ row = Row.withSchema(schema).addValues(decimal).build();
+ assertEquals(decimal, row.getLogicalTypeValue(0, BigDecimal.class));
+
+ decimal = BigDecimal.valueOf(100_000_000_001L, scale);
+ row = Row.withSchema(schema).addValues(decimal).build();
+ assertEquals(decimal, row.getLogicalTypeValue(0, BigDecimal.class));
+ }
}
diff --git a/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcUtil.java b/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcUtil.java
index 644d3d801a5..01894c621ca 100644
--- a/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcUtil.java
+++ b/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcUtil.java
@@ -39,6 +39,7 @@ import org.apache.beam.sdk.io.jdbc.JdbcIO.PreparedStatementSetter;
import org.apache.beam.sdk.io.jdbc.JdbcIO.ReadWithPartitions;
import org.apache.beam.sdk.io.jdbc.JdbcIO.RowMapper;
import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
import org.apache.beam.sdk.schemas.logicaltypes.MicrosInstant;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.values.KV;
@@ -191,6 +192,8 @@ class JdbcUtil {
String logicalTypeName = fieldType.getLogicalType().getIdentifier();
+ // Special case of Timestamp and Numeric which are logical types in Portable framework
+ // but has their own fieldType in Java.
if (logicalTypeName.equals(MicrosInstant.IDENTIFIER)) {
// Process timestamp of MicrosInstant kind, which should only be passed from other type
// systems such as SQL and other Beam SDKs.
@@ -200,6 +203,10 @@ class JdbcUtil {
element.getLogicalTypeValue(fieldWithIndex.getIndex(), java.time.Instant.class);
ps.setTimestamp(i + 1, value == null ? null : new Timestamp(value.toEpochMilli()));
};
+ } else if (logicalTypeName.equals(FixedPrecisionNumeric.IDENTIFIER)) {
+ return (element, ps, i, fieldWithIndex) -> {
+ ps.setBigDecimal(i + 1, element.getDecimal(fieldWithIndex.getIndex()));
+ };
}
JDBCType jdbcType = JDBCType.valueOf(logicalTypeName);
diff --git a/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/LogicalTypes.java b/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/LogicalTypes.java
index 9fb3677f485..e6129588959 100644
--- a/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/LogicalTypes.java
+++ b/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/LogicalTypes.java
@@ -19,8 +19,6 @@ package org.apache.beam.sdk.io.jdbc;
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 java.sql.JDBCType;
import java.time.Instant;
import java.util.Arrays;
@@ -30,9 +28,9 @@ import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
import org.apache.beam.sdk.schemas.logicaltypes.PassThroughLogicalType;
import org.apache.beam.sdk.schemas.logicaltypes.UuidLogicalType;
-import org.apache.beam.sdk.values.Row;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -100,8 +98,7 @@ class LogicalTypes {
@VisibleForTesting
static Schema.FieldType numeric(int precision, int scale) {
- return Schema.FieldType.logicalType(
- FixedPrecisionNumeric.of(JDBCType.NUMERIC.getName(), precision, scale));
+ return Schema.FieldType.logicalType(FixedPrecisionNumeric.of(precision, scale));
}
/** Base class for JDBC logical types. */
@@ -251,41 +248,4 @@ class LogicalTypes {
return base;
}
}
-
- /** Fixed precision numeric types such as NUMERIC. */
- static final class FixedPrecisionNumeric extends JdbcLogicalType<BigDecimal> {
- private final int precision;
- private final int scale;
-
- static FixedPrecisionNumeric of(String identifier, int precision, int scale) {
- Schema schema = Schema.builder().addInt32Field("precision").addInt32Field("scale").build();
- return new FixedPrecisionNumeric(schema, identifier, precision, scale);
- }
-
- private FixedPrecisionNumeric(
- Schema argumentSchema, String identifier, int precision, int scale) {
- super(
- identifier,
- FieldType.row(argumentSchema),
- Schema.FieldType.DECIMAL,
- Row.withSchema(argumentSchema).addValues(precision, scale).build());
- 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;
- }
- }
}
diff --git a/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/SchemaUtil.java b/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/SchemaUtil.java
index e2fbd9a5c70..a34ef7847ef 100644
--- a/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/SchemaUtil.java
+++ b/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/SchemaUtil.java
@@ -23,7 +23,6 @@ import static java.sql.JDBCType.LONGNVARCHAR;
import static java.sql.JDBCType.LONGVARBINARY;
import static java.sql.JDBCType.LONGVARCHAR;
import static java.sql.JDBCType.NCHAR;
-import static java.sql.JDBCType.NUMERIC;
import static java.sql.JDBCType.NVARCHAR;
import static java.sql.JDBCType.VARBINARY;
import static java.sql.JDBCType.VARCHAR;
@@ -53,6 +52,7 @@ import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
import org.joda.time.DateTime;
@@ -140,7 +140,7 @@ class SchemaUtil {
case NCHAR:
return beamLogicalField(NCHAR.getName(), LogicalTypes.FixedLengthString::of);
case NUMERIC:
- return beamLogicalNumericField(NUMERIC.getName());
+ return beamLogicalNumericField();
case NVARCHAR:
return beamLogicalField(NVARCHAR.getName(), LogicalTypes.VariableLengthString::of);
case REAL:
@@ -211,11 +211,11 @@ class SchemaUtil {
}
/**
- * Converts numeric fields with specified precision and scale to {@link
- * LogicalTypes.FixedPrecisionNumeric}. If a precision of numeric field is not specified, then
- * converts such field to {@link FieldType#DECIMAL}.
+ * Converts numeric fields with specified precision and scale to {@link FixedPrecisionNumeric}. If
+ * a precision of numeric field is not specified, then converts such field to {@link
+ * FieldType#DECIMAL}.
*/
- private static BeamFieldConverter beamLogicalNumericField(String identifier) {
+ private static BeamFieldConverter beamLogicalNumericField() {
return (index, md) -> {
int precision = md.getPrecision(index);
if (precision == Integer.MAX_VALUE || precision == -1) {
@@ -225,8 +225,7 @@ class SchemaUtil {
}
int scale = md.getScale(index);
Schema.FieldType fieldType =
- Schema.FieldType.logicalType(
- LogicalTypes.FixedPrecisionNumeric.of(identifier, precision, scale));
+ Schema.FieldType.logicalType(FixedPrecisionNumeric.of(precision, scale));
return beamFieldOfType(fieldType).create(index, md);
};
}
@@ -291,10 +290,16 @@ class SchemaUtil {
final Schema.LogicalType<InputT, BaseT> fieldType) {
String logicalTypeName = fieldType.getIdentifier();
+ JDBCType underlyingType;
+
if (Objects.equals(fieldType, LogicalTypes.JDBC_UUID_TYPE.getLogicalType())) {
return OBJECT_EXTRACTOR;
+ } else if (Objects.equals(logicalTypeName, FixedPrecisionNumeric.IDENTIFIER)) {
+ underlyingType = JDBCType.NUMERIC;
+ } else {
+ underlyingType = JDBCType.valueOf(logicalTypeName);
}
- JDBCType underlyingType = JDBCType.valueOf(logicalTypeName);
+
switch (underlyingType) {
case DATE:
return DATE_EXTRACTOR;
diff --git a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOTest.java b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOTest.java
index 6b3fb87d0cd..72fb670b2ac 100644
--- a/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOTest.java
+++ b/sdks/java/io/jdbc/src/test/java/org/apache/beam/sdk/io/jdbc/JdbcIOTest.java
@@ -18,7 +18,6 @@
package org.apache.beam.sdk.io.jdbc;
import static java.sql.JDBCType.NULL;
-import static java.sql.JDBCType.NUMERIC;
import static org.apache.beam.sdk.io.common.DatabaseTestHelper.assertRowCount;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.closeTo;
@@ -72,9 +71,9 @@ import org.apache.beam.sdk.io.common.TestRow;
import org.apache.beam.sdk.io.jdbc.JdbcIO.DataSourceConfiguration;
import org.apache.beam.sdk.io.jdbc.JdbcIO.PoolableDataSourceProvider;
import org.apache.beam.sdk.io.jdbc.JdbcUtil.PartitioningFn;
-import org.apache.beam.sdk.io.jdbc.LogicalTypes.FixedPrecisionNumeric;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
import org.apache.beam.sdk.schemas.transforms.Select;
import org.apache.beam.sdk.testing.ExpectedLogs;
import org.apache.beam.sdk.testing.PAssert;
@@ -324,9 +323,7 @@ public class JdbcIOTest implements Serializable {
Schema expectedSchema =
Schema.of(
Schema.Field.of(
- "T1",
- FieldType.logicalType(FixedPrecisionNumeric.of(NUMERIC.getName(), 1, 0))
- .withNullable(false)));
+ "T1", FieldType.logicalType(FixedPrecisionNumeric.of(1, 0)).withNullable(false)));
assertEquals(expectedSchema, rows.getSchema());
@@ -379,9 +376,7 @@ public class JdbcIOTest implements Serializable {
Schema expectedSchema =
Schema.of(
Schema.Field.of(
- "T1",
- FieldType.logicalType(FixedPrecisionNumeric.of(NUMERIC.getName(), 10, 2))
- .withNullable(false)));
+ "T1", FieldType.logicalType(FixedPrecisionNumeric.of(10, 2)).withNullable(false)));
assertEquals(expectedSchema, rows.getSchema());
diff --git a/sdks/python/apache_beam/coders/coder_impl.pxd b/sdks/python/apache_beam/coders/coder_impl.pxd
index f4cfb26ea7e..d6bee7537f9 100644
--- a/sdks/python/apache_beam/coders/coder_impl.pxd
+++ b/sdks/python/apache_beam/coders/coder_impl.pxd
@@ -110,6 +110,10 @@ cdef class BooleanCoderImpl(CoderImpl):
pass
+cdef class SinglePrecisionFloatCoderImpl(StreamCoderImpl):
+ pass
+
+
cdef class FloatCoderImpl(StreamCoderImpl):
pass
@@ -289,3 +293,11 @@ cdef class LogicalTypeCoderImpl(StreamCoderImpl):
cpdef decode_from_stream(self, InputStream stream, bint nested)
cpdef encode_to_stream(self, value, OutputStream stream, bint nested)
+
+
+cdef class BigIntegerCoderImpl(StreamCoderImpl):
+ pass
+
+
+cdef class DecimalCoderImpl(StreamCoderImpl):
+ pass
diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py
index 467ed5636ce..9e48a3a8f8e 100644
--- a/sdks/python/apache_beam/coders/coder_impl.py
+++ b/sdks/python/apache_beam/coders/coder_impl.py
@@ -32,6 +32,7 @@ For internal use only; no backwards-compatibility guarantees.
"""
# pytype: skip-file
+import decimal
import enum
import itertools
import json
@@ -1865,3 +1866,47 @@ class LogicalTypeCoderImpl(StreamCoderImpl):
def decode_from_stream(self, in_stream, nested):
return self.logical_type.to_language_type(
self.representation_coder.decode_from_stream(in_stream, nested))
+
+
+class BigIntegerCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ For interoperability with Java SDK, encoding needs to match that of the Java
+ SDK BigIntegerCoder."""
+ def encode_to_stream(self, value, out, nested):
+ # type: (int, create_OutputStream, bool) -> None
+ if value < 0:
+ byte_length = ((value + 1).bit_length() + 8) // 8
+ else:
+ byte_length = (value.bit_length() + 8) // 8
+ encoded_value = value.to_bytes(
+ length=byte_length, byteorder='big', signed=True)
+ out.write(encoded_value, nested)
+
+ def decode_from_stream(self, in_stream, nested):
+ # type: (create_InputStream, bool) -> int
+ encoded_value = in_stream.read_all(nested)
+ return int.from_bytes(encoded_value, byteorder='big', signed=True)
+
+
+class DecimalCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ For interoperability with Java SDK, encoding needs to match that of the Java
+ SDK BigDecimalCoder."""
+
+ BIG_INT_CODER_IMPL = BigIntegerCoderImpl()
+
+ def encode_to_stream(self, value, out, nested):
+ # type: (decimal.Decimal, create_OutputStream, bool) -> None
+ scale = -value.as_tuple().exponent
+ int_value = int(value.scaleb(scale))
+ out.write_var_int64(scale)
+ self.BIG_INT_CODER_IMPL.encode_to_stream(int_value, out, nested)
+
+ def decode_from_stream(self, in_stream, nested):
+ # type: (create_InputStream, bool) -> decimal.Decimal
+ scale = in_stream.read_var_int64()
+ int_value = self.BIG_INT_CODER_IMPL.decode_from_stream(in_stream, nested)
+ value = decimal.Decimal(int_value).scaleb(-scale)
+ return value
diff --git a/sdks/python/apache_beam/coders/coders.py b/sdks/python/apache_beam/coders/coders.py
index 84502024461..25fabc951c5 100644
--- a/sdks/python/apache_beam/coders/coders.py
+++ b/sdks/python/apache_beam/coders/coders.py
@@ -37,6 +37,7 @@ encoded.
# pytype: skip-file
import base64
+import decimal
import pickle
from functools import lru_cache
from typing import TYPE_CHECKING
@@ -110,7 +111,9 @@ __all__ = [
'TupleSequenceCoder',
'VarIntCoder',
'WindowedValueCoder',
- 'ParamWindowedValueCoder'
+ 'ParamWindowedValueCoder',
+ 'BigIntegerCoder',
+ 'DecimalCoder'
]
T = TypeVar('T')
@@ -1737,3 +1740,39 @@ class TimestampPrefixingWindowCoder(FastCoder):
Coder.register_structured_urn(
common_urns.coders.CUSTOM_WINDOW.urn, TimestampPrefixingWindowCoder)
+
+
+class BigIntegerCoder(FastCoder):
+ def _create_impl(self):
+ return coder_impl.BigIntegerCoderImpl()
+
+ def is_deterministic(self):
+ # type: () -> bool
+ return True
+
+ def to_type_hint(self):
+ return int
+
+ def __eq__(self, other):
+ return type(self) == type(other)
+
+ def __hash__(self):
+ return hash(type(self))
+
+
+class DecimalCoder(FastCoder):
+ def _create_impl(self):
+ return coder_impl.DecimalCoderImpl()
+
+ def is_deterministic(self):
+ # type: () -> bool
+ return True
+
+ def to_type_hint(self):
+ return decimal.Decimal
+
+ def __eq__(self, other):
+ return type(self) == type(other)
+
+ def __hash__(self):
+ return hash(type(self))
diff --git a/sdks/python/apache_beam/coders/coders_test_common.py b/sdks/python/apache_beam/coders/coders_test_common.py
index b8a4bcb8fea..a0bec891bdf 100644
--- a/sdks/python/apache_beam/coders/coders_test_common.py
+++ b/sdks/python/apache_beam/coders/coders_test_common.py
@@ -18,11 +18,13 @@
"""Tests common to all coder implementations."""
# pytype: skip-file
+import base64
import collections
import enum
import logging
import math
import unittest
+from decimal import Decimal
from typing import Any
from typing import List
from typing import NamedTuple
@@ -160,6 +162,7 @@ class CodersTest(unittest.TestCase):
coders.ProtoPlusCoder,
coders.SinglePrecisionFloatCoder,
coders.ToBytesCoder,
+ coders.BigIntegerCoder, # tested in DecimalCoder
])
cls.seen_nested -= set(
[coders.ProtoCoder, coders.ProtoPlusCoder, CustomCoder])
@@ -776,6 +779,26 @@ class CodersTest(unittest.TestCase):
coders.IntervalWindowCoder()), )),
(window.IntervalWindow(0, 10), ))
+ def test_decimal_coder(self):
+ test_coder = coders.DecimalCoder()
+
+ test_values = [
+ Decimal("-10.5"),
+ Decimal("-1"),
+ Decimal(),
+ Decimal("1"),
+ Decimal("13.258"),
+ ]
+
+ test_encodings = ("AZc", "AP8", "AAA", "AAE", "AzPK")
+
+ self.check_coder(test_coder, *test_values)
+
+ for idx, value in enumerate(test_values):
+ self.assertEqual(
+ test_encodings[idx],
+ base64.b64encode(test_coder.encode(value)).decode().rstrip("="))
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
diff --git a/sdks/python/apache_beam/coders/row_coder.py b/sdks/python/apache_beam/coders/row_coder.py
index d8c07306b9b..9dd4dcd9f63 100644
--- a/sdks/python/apache_beam/coders/row_coder.py
+++ b/sdks/python/apache_beam/coders/row_coder.py
@@ -25,6 +25,7 @@ from apache_beam.coders.coder_impl import RowCoderImpl
from apache_beam.coders.coders import BooleanCoder
from apache_beam.coders.coders import BytesCoder
from apache_beam.coders.coders import Coder
+from apache_beam.coders.coders import DecimalCoder
from apache_beam.coders.coders import FastCoder
from apache_beam.coders.coders import FloatCoder
from apache_beam.coders.coders import IterableCoder
@@ -178,6 +179,8 @@ def _nonnull_coder_from_type(field_type):
# millis Instant. It explicitly uses TimestampCoder which deals with fix
# length 8-bytes big-endian-long instead of VarInt coder.
return TimestampCoder()
+ elif field_type.logical_type.urn == 'beam:logical_type:decimal:v1':
+ return DecimalCoder()
logical_type = LogicalType.from_runner_api(field_type.logical_type)
return LogicalTypeCoder(
diff --git a/sdks/python/apache_beam/io/external/xlang_jdbcio_it_test.py b/sdks/python/apache_beam/io/external/xlang_jdbcio_it_test.py
index 62620122e59..75dad46124f 100644
--- a/sdks/python/apache_beam/io/external/xlang_jdbcio_it_test.py
+++ b/sdks/python/apache_beam/io/external/xlang_jdbcio_it_test.py
@@ -22,6 +22,7 @@ import logging
import time
import typing
import unittest
+from decimal import Decimal
from typing import Callable
from typing import Union
@@ -59,14 +60,14 @@ ROW_COUNT = 10
JdbcReadTestRow = typing.NamedTuple(
"JdbcReadTestRow",
- [("f_int", int), ("f_timestamp", Timestamp)],
+ [("f_int", int), ("f_timestamp", Timestamp), ("f_decimal", Decimal)],
)
coders.registry.register_coder(JdbcReadTestRow, coders.RowCoder)
JdbcWriteTestRow = typing.NamedTuple(
"JdbcWriteTestRow",
[("f_id", int), ("f_real", float), ("f_string", str),
- ("f_timestamp", Timestamp)],
+ ("f_timestamp", Timestamp), ("f_decimal", Decimal)],
)
coders.registry.register_coder(JdbcWriteTestRow, coders.RowCoder)
@@ -128,7 +129,7 @@ class CrossLanguageJdbcIOTest(unittest.TestCase):
self._setUpTestCase(container_init, db_string, driver)
table_name = 'jdbc_external_test_write'
self.engine.execute(
- "CREATE TABLE {}(f_id INTEGER, f_real FLOAT, f_string VARCHAR(100), f_timestamp TIMESTAMP(3))" # pylint: disable=line-too-long
+ "CREATE TABLE {}(f_id INTEGER, f_real FLOAT, f_string VARCHAR(100), f_timestamp TIMESTAMP(3), f_decimal DECIMAL(10, 2))" # pylint: disable=line-too-long
.format(table_name))
inserted_rows = [
JdbcWriteTestRow(
@@ -136,8 +137,8 @@ class CrossLanguageJdbcIOTest(unittest.TestCase):
i + 0.1,
'Test{}'.format(i),
# In alignment with Java Instant which supports milli precision.
- Timestamp.of(seconds=round(time.time(), 3)))
- for i in range(ROW_COUNT)
+ Timestamp.of(seconds=round(time.time(), 3)),
+ Decimal(f'{i-1}.23')) for i in range(ROW_COUNT)
]
with TestPipeline() as p:
@@ -162,8 +163,8 @@ class CrossLanguageJdbcIOTest(unittest.TestCase):
int(row[0]),
float(row[1]),
str(row[2]),
- Timestamp.from_utc_datetime(row[3].replace(tzinfo=pytz.UTC)))
- for row in fetched_data
+ Timestamp.from_utc_datetime(row[3].replace(tzinfo=pytz.UTC)),
+ Decimal(row[4])) for row in fetched_data
]
self.assertEqual(
@@ -179,8 +180,8 @@ class CrossLanguageJdbcIOTest(unittest.TestCase):
self._setUpTestCase(container_init, db_string, driver)
table_name = 'jdbc_external_test_read'
self.engine.execute(
- "CREATE TABLE {}(f_int INTEGER, f_timestamp TIMESTAMP)".format(
- table_name))
+ "CREATE TABLE {}(f_int INTEGER, f_timestamp TIMESTAMP, f_decimal DECIMAL(10,2))" # pylint: disable=line-too-long
+ .format(table_name))
all_timestamps = []
for i in range(ROW_COUNT):
@@ -189,10 +190,12 @@ class CrossLanguageJdbcIOTest(unittest.TestCase):
dttime = datetime.datetime.strptime(
strtime, '%Y-%m-%dT%H:%M:%S').replace(tzinfo=pytz.UTC)
all_timestamps.append(Timestamp.from_utc_datetime(dttime))
+ decimal_value = Decimal(f'{i-1}.23')
# write records using sqlalchemy engine
self.engine.execute(
- "INSERT INTO {} VALUES({},'{}')".format(table_name, i, strtime))
+ "INSERT INTO {} VALUES({},'{}','{}')".format(
+ table_name, i, strtime, decimal_value))
# Register MillisInstant logical type to override the mapping from Timestamp
# originally handled by MicrosInstant.
@@ -215,7 +218,8 @@ class CrossLanguageJdbcIOTest(unittest.TestCase):
assert_that(
result,
equal_to([
- JdbcReadTestRow(i, all_timestamps[i]) for i in range(ROW_COUNT)
+ JdbcReadTestRow(i, all_timestamps[i], Decimal(f'{i-1}.23'))
+ for i in range(ROW_COUNT)
]))
# Creating a container with testcontainers sometimes raises ReadTimeout
diff --git a/sdks/python/apache_beam/portability/common_urns.py b/sdks/python/apache_beam/portability/common_urns.py
index e5788b1ad21..199ce4d7058 100644
--- a/sdks/python/apache_beam/portability/common_urns.py
+++ b/sdks/python/apache_beam/portability/common_urns.py
@@ -79,6 +79,7 @@ displayData = StandardDisplayData.DisplayData
java_class_lookup = ExpansionMethods.Enum.JAVA_CLASS_LOOKUP
+decimal = LogicalTypes.Enum.DECIMAL
micros_instant = LogicalTypes.Enum.MICROS_INSTANT
millis_instant = LogicalTypes.Enum.MILLIS_INSTANT
python_callable = LogicalTypes.Enum.PYTHON_CALLABLE
diff --git a/sdks/python/apache_beam/typehints/schemas.py b/sdks/python/apache_beam/typehints/schemas.py
index 4b312d96ce9..4371ca2de0d 100644
--- a/sdks/python/apache_beam/typehints/schemas.py
+++ b/sdks/python/apache_beam/typehints/schemas.py
@@ -35,6 +35,7 @@ Imposes a mapping between common Python types and Beam portable schemas
ByteString ------> BYTES
Timestamp <-----> LogicalType(urn="beam:logical_type:micros_instant:v1")
Timestamp <------ LogicalType(urn="beam:logical_type:millis_instant:v1")
+ Decimal <-----> LogicalType(urn="beam:logical_type:fixed_decimal:v1")
Mapping <-----> MapType
Sequence <-----> ArrayType
NamedTuple <-----> RowType
@@ -55,6 +56,7 @@ any backwards-compatibility guarantee.
# pytype: skip-file
+import decimal
from typing import Any
from typing import ByteString
from typing import Dict
@@ -267,12 +269,24 @@ class SchemaTranslation(object):
logical_type=schema_pb2.LogicalType(urn=PYTHON_ANY_URN),
nullable=True)
else:
- # TODO(bhulette): Add support for logical types that require arguments
- return schema_pb2.FieldType(
- logical_type=schema_pb2.LogicalType(
- urn=logical_type.urn(),
- representation=self.typing_to_runner_api(
- logical_type.representation_type())))
+ if logical_type.argument_type() is None:
+ return schema_pb2.FieldType(
+ logical_type=schema_pb2.LogicalType(
+ urn=logical_type.urn(),
+ representation=self.typing_to_runner_api(
+ logical_type.representation_type())))
+ else:
+ # TODO(https://github.com/apache/beam/issues/23373): Complete support
+ # for logical types that require arguments.
+ # This include implement SchemaTranslation.value_to_runner_api (see Java
+ # SDK's SchemaTranslation.fieldValueToProto)
+ return schema_pb2.FieldType(
+ logical_type=schema_pb2.LogicalType(
+ urn=logical_type.urn(),
+ representation=self.typing_to_runner_api(
+ logical_type.representation_type()),
+ argument_type=self.typing_to_runner_api(
+ logical_type.argument_type())))
def option_from_runner_api(
self, option_proto: schema_pb2.Option) -> Tuple[str, Any]:
@@ -751,3 +765,89 @@ class PythonCallable(NoArgumentLogicalType[PythonCallableWithSource, str]):
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())
+
+
+@LogicalType.register_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):
+ # TODO(https://github.com/apache/beam/issues/23373) promote this URN to
+ # schema.proto once logical types with argument are fully supported and the
+ # implementation of this logical type can thus be considered standardized.
+ 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):
+ return cls()
+
+
+# TODO(yathu,BEAM-10722): Investigate and resolve conflicts in logical type
+# registration when more than one logical types sharing the same language type
+LogicalType.register_logical_type(DecimalLogicalType)