forked from apache/beam
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Migrate BINARY, VARBINARY, CHAR, VARCHAR jdbc logical types to portab…
…le (apache#23548) * Migrate BINARY, VARBINARY, CHAR, VARCHAR jdbc logical types to portable * Move jdbc logical type to portable logical types in Java * Create portable logical types in Python * Support value_from_runner_api and value_to_runner_api in Python SchemaTransform (currently only support atomic type values) Fix nullable/test/leftovers * Fix typos * Add standard coder test * Fix RowCoderImpl cannot encode bytes column in cython compiled * Set coder_impl.is_compiled=True when running on compiled stream module * Add docstring, add todo and warnings for unsupported
- Loading branch information
1 parent
e69bd6e
commit 635c41c
Showing
17 changed files
with
900 additions
and
431 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
77 changes: 77 additions & 0 deletions
77
sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/FixedString.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * 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.model.pipeline.v1.RunnerApi; | ||
| import org.apache.beam.model.pipeline.v1.SchemaApi; | ||
| import org.apache.beam.sdk.schemas.Schema.FieldType; | ||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.checkerframework.checker.nullness.qual.Nullable; | ||
|
|
||
| /** A LogicalType representing a fixed-length string. */ | ||
| public class FixedString extends PassThroughLogicalType<String> { | ||
| public static final String IDENTIFIER = | ||
| SchemaApi.LogicalTypes.Enum.FIXED_CHAR | ||
| .getValueDescriptor() | ||
| .getOptions() | ||
| .getExtension(RunnerApi.beamUrn); | ||
| private final @Nullable String name; | ||
| private final int stringLength; | ||
|
|
||
| /** | ||
| * Return an instance of FixedString with specified string length. | ||
| * | ||
| * <p>The name, if set, refers to the TYPE name in the underlying database, for example, CHAR. | ||
| */ | ||
| public static FixedString of(@Nullable String name, int stringLength) { | ||
| return new FixedString(name, stringLength); | ||
| } | ||
|
|
||
| /** Return an instance of FixedString with specified string length. */ | ||
| public static FixedString of(int stringLength) { | ||
| return new FixedString(null, stringLength); | ||
| } | ||
|
|
||
| private FixedString(@Nullable String name, int stringLength) { | ||
| super(IDENTIFIER, FieldType.INT32, stringLength, FieldType.STRING); | ||
| this.name = name; | ||
| this.stringLength = stringLength; | ||
| } | ||
|
|
||
| public int getLength() { | ||
| return stringLength; | ||
| } | ||
|
|
||
| public @Nullable String getName() { | ||
| return name; | ||
| } | ||
|
|
||
| @Override | ||
| public String toInputType(String base) { | ||
| checkArgument(base.length() <= stringLength); | ||
|
|
||
| return StringUtils.rightPad(base, stringLength); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "FixedString: " + stringLength; | ||
| } | ||
| } |
76 changes: 76 additions & 0 deletions
76
sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/logicaltypes/VariableBytes.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /* | ||
| * 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.model.pipeline.v1.RunnerApi; | ||
| import org.apache.beam.model.pipeline.v1.SchemaApi; | ||
| import org.apache.beam.sdk.schemas.Schema.FieldType; | ||
| import org.checkerframework.checker.nullness.qual.Nullable; | ||
|
|
||
| /** A LogicalType representing a variable-length byte array with specified maximum length. */ | ||
| public class VariableBytes extends PassThroughLogicalType<byte[]> { | ||
| public static final String IDENTIFIER = | ||
| SchemaApi.LogicalTypes.Enum.VAR_BYTES | ||
| .getValueDescriptor() | ||
| .getOptions() | ||
| .getExtension(RunnerApi.beamUrn); | ||
| private final @Nullable String name; | ||
| private final int maxByteArrayLength; | ||
|
|
||
| /** | ||
| * Return an instance of VariableBytes with specified max byte array length. | ||
| * | ||
| * <p>The name, if set, refers to the TYPE name in the underlying database, for example, VARBINARY | ||
| * and LONGVARBINARY. | ||
| */ | ||
| public static VariableBytes of(@Nullable String name, int maxByteArrayLength) { | ||
| return new VariableBytes(name, maxByteArrayLength); | ||
| } | ||
|
|
||
| /** Return an instance of VariableBytes with specified max byte array length. */ | ||
| public static VariableBytes of(int maxByteArrayLength) { | ||
| return of(null, maxByteArrayLength); | ||
| } | ||
|
|
||
| private VariableBytes(@Nullable String name, int maxByteArrayLength) { | ||
| super(IDENTIFIER, FieldType.INT32, maxByteArrayLength, FieldType.BYTES); | ||
| this.name = name; | ||
| this.maxByteArrayLength = maxByteArrayLength; | ||
| } | ||
|
|
||
| public int getMaxLength() { | ||
| return maxByteArrayLength; | ||
| } | ||
|
|
||
| public @Nullable String getName() { | ||
| return name; | ||
| } | ||
|
|
||
| @Override | ||
| public byte[] toInputType(byte[] base) { | ||
| checkArgument(base.length <= maxByteArrayLength); | ||
| return base; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "VariableBytes: " + maxByteArrayLength; | ||
| } | ||
| } |
Oops, something went wrong.