flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-5790) Use list types when ListStateDescriptor extends StateDescriptor
Date Tue, 14 Feb 2017 16:09:42 GMT

    [ https://issues.apache.org/jira/browse/FLINK-5790?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15866032#comment-15866032
] 

ASF GitHub Bot commented on FLINK-5790:
---------------------------------------

Github user tillrohrmann commented on a diff in the pull request:

    https://github.com/apache/flink/pull/3305#discussion_r101070139
  
    --- Diff: flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java
---
    @@ -0,0 +1,131 @@
    +/*
    + * 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.flink.api.common.typeutils.base;
    +
    +import org.apache.flink.api.common.typeutils.TypeSerializer;
    +import org.apache.flink.core.memory.DataInputView;
    +import org.apache.flink.core.memory.DataOutputView;
    +
    +import java.io.IOException;
    +import java.util.List;
    +import java.util.ArrayList;
    +
    +@SuppressWarnings("ForLoopReplaceableByForEach")
    +public class ListSerializer<T> extends TypeSerializer<List<T>> {
    +
    +	private static final long serialVersionUID = 1119562170939152304L;
    +
    +	private final TypeSerializer<T> elementSerializer;
    +
    +	public ListSerializer(TypeSerializer<T> elementSerializer) {
    +		this.elementSerializer = elementSerializer;
    +	}
    +
    +	public TypeSerializer<T> getElementSerializer() {
    +		return elementSerializer;
    +	}
    +
    +	@Override
    +	public boolean isImmutableType() {
    +		return false;
    +	}
    +
    +	@Override
    +	public TypeSerializer<List<T>> duplicate() {
    +		TypeSerializer<T> duplicateElement = elementSerializer.duplicate();
    +		return duplicateElement == elementSerializer ? this : new ListSerializer<T>(duplicateElement);
    +	}
    +
    +	@Override
    +	public List<T> createInstance() {
    +		return new ArrayList<>();
    +	}
    +
    +	@Override
    +	public List<T> copy(List<T> from) {
    +		List<T> newList = new ArrayList<>(from.size());
    +		for (int i = 0; i < from.size(); i++) {
    +			newList.add(elementSerializer.copy(from.get(i)));
    +		}
    +		return newList;
    +	}
    +
    +	@Override
    +	public List<T> copy(List<T> from, List<T> reuse) {
    +		return copy(from);
    +	}
    +
    +	@Override
    +	public int getLength() {
    +		return -1; // var length
    +	}
    +
    +	@Override
    +	public void serialize(List<T> list, DataOutputView target) throws IOException
{
    +		final int size = list.size();
    +		target.writeInt(size);
    +		for (int i = 0; i < size; i++) {
    +			elementSerializer.serialize(list.get(i), target);
    +		}
    +	}
    +
    +	@Override
    +	public List<T> deserialize(DataInputView source) throws IOException {
    +		final int size = source.readInt();
    +		final List<T> list = new ArrayList<>(size);
    +		for (int i = 0; i < size; i++) {
    +			list.add(elementSerializer.deserialize(source));
    +		}
    +		return list;
    +	}
    +
    +	@Override
    +	public List<T> deserialize(List<T> reuse, DataInputView source) throws IOException
{
    +		return deserialize(source);
    +	}
    +
    +	@Override
    +	public void copy(DataInputView source, DataOutputView target) throws IOException {
    +		// copy number of elements
    +		final int num = source.readInt();
    +		target.writeInt(num);
    +		for (int i = 0; i < num; i++) {
    +			elementSerializer.copy(source, target);
    +		}
    +	}
    +
    +	// --------------------------------------------------------------------
    +
    +	@Override
    +	public boolean equals(Object obj) {
    +		return obj == this ||
    +				(obj != null && obj.getClass() == getClass() &&
    +						elementSerializer.equals(((ListSerializer<?>) obj).elementSerializer));
    +	}
    +
    +	@Override
    +	public boolean canEqual(Object obj) {
    +		return true;
    +	}
    --- End diff --
    
    This is not the contract of `canEqual`. Given that this class is not final this method
should define which objects can be equal to this type. So in this case it would be `obj instanceOf
ListSerializer`. Furthermore the check has to be incorporated into the `equals` method where
you check `((ListSerializer) obj).canEqual(this))`. Otherwise the symmetry of equals is violated.


> Use list types when ListStateDescriptor extends StateDescriptor
> ---------------------------------------------------------------
>
>                 Key: FLINK-5790
>                 URL: https://issues.apache.org/jira/browse/FLINK-5790
>             Project: Flink
>          Issue Type: Improvement
>            Reporter: Xiaogang Shi
>            Assignee: Xiaogang Shi
>
> Flink keeps the state serializer in {{StateDescriptor}}, but it's the serializer of list
elements  that is put in {{ListStateDescriptor}}. The implementation is a little confusing.
Some backends need to construct the state serializer with the element serializer by themselves.
> We should use an {{ArrayListSerializer}}, which is composed of the serializer of the
element, in the {{ListStateDescriptor}}. It helps the backend to avoid constructing the state
serializer.
> If a backend needs customized serialization of the state (e.g. {{RocksDBStateBackend}}),
it still can obtain the element serializer from the {{ArrayListSerializer}}.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Mime
View raw message