hequn8128 commented on a change in pull request #9977: [FLINK-14497][python]
Support primitive data types in Python user-defined functions
URL: https://github.com/apache/flink/pull/9977#discussion_r338553198
##########
File path:
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/BaseRowSerializer.java
##########
@@ -16,107 +16,75 @@
* limitations under the License.
*/
-package org.apache.flink.table.runtime.typeutils.coders;
+package org.apache.flink.table.runtime.typeutils.serializers;
-import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.table.dataformat.BaseRow;
import org.apache.flink.table.dataformat.GenericRow;
import org.apache.flink.table.dataformat.TypeGetterSetters;
import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.beam.sdk.coders.Coder;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Collections;
-import java.util.List;
import static
org.apache.flink.api.java.typeutils.runtime.NullMaskUtils.readIntoNullMask;
/**
- * A {@link Coder} for {@link BaseRow}. It should be noted that the header
will not be encoded.
+ * BaseRowSerializer for BasRow.
*/
-@Internal
-public class BaseRowCoder extends Coder<BaseRow> {
-
- private static final long serialVersionUID = 1L;
+public class BaseRowSerializer extends
org.apache.flink.table.runtime.typeutils.BaseRowSerializer {
- private final Coder<Object>[] fieldCoders;
private final LogicalType[] fieldTypes;
- private transient ReusableDataInputView reusableInputStream;
- private transient ReusableDataOutputView reusableOutputStream;
+ private final TypeSerializer[] fieldSerializers;
private transient boolean[] nullMask;
- @SuppressWarnings("unchecked")
- public BaseRowCoder(Coder<?>[] fieldCoders, LogicalType[] fieldTypes) {
- this.fieldCoders = (Coder<Object>[])
Preconditions.checkNotNull(fieldCoders);
- this.fieldTypes = Preconditions.checkNotNull(fieldTypes);
- this.reusableInputStream = new ReusableDataInputView();
- this.reusableOutputStream = new ReusableDataOutputView();
- this.nullMask = new boolean[fieldCoders.length];
- }
-
- public Coder<?>[] getFieldCoders() {
- return this.fieldCoders;
+ public BaseRowSerializer(LogicalType[] types, TypeSerializer[]
fieldSerializers) {
+ super(types, fieldSerializers);
+ this.fieldTypes = types;
+ this.fieldSerializers = fieldSerializers;
+ this.nullMask = new boolean[fieldTypes.length];
}
@Override
- public void encode(BaseRow row, OutputStream outStream) throws
IOException {
- int len = fieldCoders.length;
+ public void serialize(BaseRow row, DataOutputView target) throws
IOException {
+ int len = fieldSerializers.length;
if (row.getArity() != len) {
throw new RuntimeException("Row arity of input element
does not match coders.");
Review comment:
"does not match coders." => "does not match serializers."
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
