add micros_instant urn

tomstepp · May 12, 2022 · 514e325 · 514e325
1 parent 792baa2
commit 514e325
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 4 deletions.
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
@@ -117,12 +117,19 @@ message LogicalType {
 message LogicalTypes {
   enum Enum {
     // A URN for Python Callable logical type
-    // - Representation type: STRING
-    // - Language type: In Python SDK, PythonCallableWithSource.
+    //   - Representation type: STRING
+    //   - Language type: In Python SDK, PythonCallableWithSource.
     //     In any other SDKs, a wrapper object for a string which
     //     can be evaluated to a Python Callable object.
     PYTHON_CALLABLE = 0 [(org.apache.beam.model.pipeline.v1.beam_urn) =
         "beam:logical_type:python_callable:v1"];
+
+    // A URN for MicrosInstant type
+    //   - Representation type: ROW<seconds: INT64, micros: INT64>
+    //   - A timestamp without a timezone where seconds + micros represents the
+    //     amount of time since the epoch.
+    MICROS_INSTANT  = 1 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+        "beam:logical_type:micros_instant:v1"];
   }
 }
 

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/MicrosInstant.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/MicrosInstant.java
@@ -18,6 +18,8 @@
 package org.apache.beam.sdk.schemas.logicaltypes;
 
 import java.time.Instant;
+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.values.Row;
 
@@ -36,7 +38,11 @@
   "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
 })
 public class MicrosInstant implements Schema.LogicalType<Instant, Row> {
-  public static final String IDENTIFIER = "beam:logical_type:micros_instant:v1";
+  public static final String IDENTIFIER =
+      SchemaApi.LogicalTypes.Enum.MICROS_INSTANT
+          .getValueDescriptor()
+          .getOptions()
+          .getExtension(RunnerApi.beamUrn);
   // TODO(BEAM-10878): This should be a constant
   private final Schema schema;
 

diff --git a/sdks/python/apache_beam/portability/common_urns.py b/sdks/python/apache_beam/portability/common_urns.py
@@ -79,4 +79,5 @@
 
 java_class_lookup = ExpansionMethods.Enum.JAVA_CLASS_LOOKUP
 
+micros_instant = LogicalTypes.Enum.MICROS_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
@@ -542,7 +542,7 @@ class MicrosInstant(NoArgumentLogicalType[Timestamp,
                                           MicrosInstantRepresentation]):
   @classmethod
   def urn(cls):
-    return "beam:logical_type:micros_instant:v1"
+    return common_urns.micros_instant.urn
 
   @classmethod
   def representation_type(cls):
Original file line number	Diff line number	Diff line change
Expand Up		@@ -79,4 +79,5 @@

		java_class_lookup = ExpansionMethods.Enum.JAVA_CLASS_LOOKUP

		micros_instant = LogicalTypes.Enum.MICROS_INSTANT
		python_callable = LogicalTypes.Enum.PYTHON_CALLABLE