You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2023/01/10 06:00:51 UTC
[GitHub] [flink] masteryhx opened a new pull request, #21635: Improve resolving schema compatibility -- Milestone one
masteryhx opened a new pull request, #21635:
URL: https://github.com/apache/flink/pull/21635
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This pr reverses the direction of resolving schema compatibility could improve the usability of schema evolution.
See FLIP-263 for more details.
## Brief change log
In the milestone one, we should:
1. Add an extra method (TypeserializeSnapshotr#resolveSchemaCompatibility(TypeSerializerSnapshot<T> oldSerializerSnapshot)) in TypeSerializerSnapshot.java as above, and return INCOMPATIBLE as default.
2. Mark the original method as deprecated and it will use new method to resolve as default.
3. Implement the new method for all built-in TypeserializerSnapshots.
## Verifying this change
This change is already covered by existing tests, such as *TypeSerializerUpgradeTestBase*, *PojoSerializerSnapshotTest*.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: yes
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? docs
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "llkj1 (via GitHub)" <gi...@apache.org>.
llkj1 commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879977455
感谢您与我们联系。(Thank you for contacting us.)
------------------------------------------------------------
我们收到了您的请求，并会在一至两个工作日内给您回复。您的案例编号是 102195425319。
想了解更多与开发相关主题的信息，请访问 开发者支持网页 (https://developer.apple.com/support/)。
Apple Developer Program Support
We’ve received your support request and will get back to you in one to two business days. Your case number is 102195425319.
For additional information on development-related topics, visit:
https://developer.apple.com/support/
Best regards,
Apple Developer Program Support
------------------------------------------------------------
Copyright (c) 2024 Apple Inc. All rights reserved.
Contact Us
https://developer.apple.com/contact/
Developer
https://developer.apple.com/
My Apple ID
https://appleid.apple.com
Privacy Policy
https://www.apple.com/privacy/
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1803157270
rebased.
@Zakelly Could you also help to take a look ? Thanks a lot!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1857189693
> Hi @masteryhx , sorry for the late reply!
>
> After reading your PR, I finally figured out why the default implementation of old method and new method should call each other. I think we could do something tricky to check whether one of the old and new methods has been implemented in specific `TypeSerializerSnapshot` before it is called. https://stackoverflow.com/a/2315467 this may provide some ideas. WDYT?
Thanks for the advice.
It's a good idea to avoid infinite loop when users don't implement both methods.
I will update the pr to support this.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1221419790
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
Thanks a lot for the detailed explanation.
I rethinked it and aggreed that option 2 should be a better one currently.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1383361255
@gaoyunhaii @dawidwys
Could you help to take a review ?
I'd like to make it usable in 1.17. Thanks a lot!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081541284
##########
flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/ScalaEnumSerializerSnapshot.scala:
##########
@@ -80,25 +80,26 @@ class ScalaEnumSerializerSnapshot[E <: Enumeration] extends TypeSerializerSnapsh
new EnumValueSerializer(enumObject)
}
- override def resolveSchemaCompatibility(
- newSerializer: TypeSerializer[E#Value]): TypeSerializerSchemaCompatibility[E#Value] = {
+ override def resolveSchemaCompatibility(oldSerializerSnapshot: TypeSerializerSnapshot[E#Value])
+ : TypeSerializerSchemaCompatibility[E#Value] = {
Preconditions.checkState(enumClass != null)
- Preconditions.checkState(previousEnumConstants != null)
+ Preconditions.checkState(enumConstants != null)
- if (!newSerializer.isInstanceOf[EnumValueSerializer[E]]) {
+ if (!oldSerializerSnapshot.isInstanceOf[ScalaEnumSerializerSnapshot[E]]) {
return TypeSerializerSchemaCompatibility.incompatible()
}
- val newEnumSerializer = newSerializer.asInstanceOf[EnumValueSerializer[E]]
- if (!enumClass.equals(newEnumSerializer.enum.getClass)) {
+ val oldEnumSerializerSnapshot =
+ oldSerializerSnapshot.asInstanceOf[ScalaEnumSerializerSnapshot[E]]
+ if (!enumClass.equals(oldEnumSerializerSnapshot.enumClass)) {
return TypeSerializerSchemaCompatibility.incompatible()
}
- for ((previousEnumName, index) <- previousEnumConstants) {
+ for ((enumName, index) <- oldEnumSerializerSnapshot.enumConstants) {
try {
- val newEnumName = newEnumSerializer.enum(index).toString
- if (previousEnumName != newEnumName) {
+ val oldEnumName = enumConstants(index)._1
Review Comment:
```suggestion
val enumName = enumConstants(index)._1
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1407691028
@dawidwys Could you help to take a review again?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1426110489
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
Review Comment:
Yes, you're right.
These are all root classes call the `resolveSchemaCompatibility` in state module.
So I'd like to prepare a seprate PR to adjust them as I mentioned before.
It should not affect the correctness of usage.
WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "llkj1 (via GitHub)" <gi...@apache.org>.
llkj1 commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879975977
感谢您与我们联系。(Thank you for contacting us.)
------------------------------------------------------------
我们收到了您的请求，并会在一至两个工作日内给您回复。您的案例编号是 102195421822。
想了解更多与开发相关主题的信息，请访问 开发者支持网页 (https://developer.apple.com/support/)。
Apple Developer Program Support
We’ve received your support request and will get back to you in one to two business days. Your case number is 102195421822.
For additional information on development-related topics, visit:
https://developer.apple.com/support/
Best regards,
Apple Developer Program Support
------------------------------------------------------------
Copyright (c) 2024 Apple Inc. All rights reserved.
Contact Us
https://developer.apple.com/contact/
Developer
https://developer.apple.com/
My Apple ID
https://appleid.apple.com
Privacy Policy
https://www.apple.com/privacy/
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879934580
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085369657
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -290,26 +332,26 @@ protected void readOuterSnapshot(
throws IOException {}
/**
- * Checks whether the outer snapshot is compatible with a given new serializer.
+ * Checks the schema compatibility of the given old serializer snapshot based on the outer
+ * snapshot.
*
- * <p>The base implementation of this method just returns {@code true}, i.e. it assumes that the
- * outer serializer only has nested serializers and no extra information, and therefore the
- * result of the check must always be true. Otherwise, if the outer serializer contains some
+ * <p>The base implementation of this method assumes that the outer serializer only has nested
+ * serializers and no extra information, and therefore the result of the check is {@link
+ * OuterSchemaCompatibility#COMPATIBLE_AS_IS}. Otherwise, if the outer serializer contains some
* extra information that has been persisted as part of the serializer snapshot, this must be
* overridden. Note that this method and the corresponding methods {@link
* #writeOuterSnapshot(DataOutputView)}, {@link #readOuterSnapshot(int, DataInputView,
* ClassLoader)} needs to be implemented.
*
- * @param newSerializer the new serializer, which contains the new outer information to check
- * against.
- * @return a flag indicating whether or not the new serializer's outer information is compatible
- * with the one written in this snapshot.
- * @deprecated this method is deprecated, and will be removed in the future. Please implement
- * {@link #resolveOuterSchemaCompatibility(TypeSerializer)} instead.
+ * @param oldSerializerSnapshot the old serializer snapshot, which contains the old outer
+ * information to check against.
+ * @return a {@link OuterSchemaCompatibility} indicating whether or the new serializer's outer
+ * information is compatible, requires migration, or incompatible with the one written in
+ * this snapshot.
*/
- @Deprecated
- protected boolean isOuterSnapshotCompatible(S newSerializer) {
Review Comment:
When is this method called now? Could you please explain the flow how `CompositeTypeSerializerSnapshot` works? I still think there is something off here.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx closed pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
URL: https://github.com/apache/flink/pull/21635
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1827061892
> Thanks @masteryhx for the contribution!
>
> I didn't finish the review, and left some minor comments. Please take a look in your free time, thanks~
Thanks a lot for your time!
I have updated the pr by adding two commits.
I will rebase some commits after the pr is approved.
Please take a review again in your free time. Thanks a lot again!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081536771
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java:
##########
@@ -328,26 +327,27 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
@Override
public RowDataSerializer restoreSerializer() {
return new RowDataSerializer(
- previousTypes,
- nestedSerializersSnapshotDelegate.getRestoredNestedSerializers());
+ types, nestedSerializersSnapshotDelegate.getRestoredNestedSerializers());
}
@Override
public TypeSerializerSchemaCompatibility<RowData> resolveSchemaCompatibility(
- TypeSerializer<RowData> newSerializer) {
- if (!(newSerializer instanceof RowDataSerializer)) {
+ TypeSerializerSnapshot<RowData> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof RowDataSerializerSnapshot)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
- RowDataSerializer newRowSerializer = (RowDataSerializer) newSerializer;
- if (!Arrays.equals(previousTypes, newRowSerializer.types)) {
+ RowDataSerializerSnapshot oldRowDataSerializerSnapshot =
+ (RowDataSerializerSnapshot) oldSerializerSnapshot;
+ if (!Arrays.equals(types, oldRowDataSerializerSnapshot.types)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData>
intermediateResult =
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
- newRowSerializer.fieldSerializers,
+ oldRowDataSerializerSnapshot.nestedSerializersSnapshotDelegate
Review Comment:
shouldn't the order of serializers be reversed here?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1090379604
##########
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V1TestTypeSerializerSnapshot.java:
##########
@@ -36,16 +36,14 @@ public int getCurrentVersion() {
@Override
public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(
- TypeSerializer<TestType> newSerializer) {
- if (newSerializer instanceof TestType.V1TestTypeSerializer) {
+ TypeSerializerSnapshot<TestType> oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof V1TestTypeSerializerSnapshot) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
- } else if (newSerializer instanceof TestType.V2TestTypeSerializer) {
- return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
- } else if (newSerializer instanceof TestType.ReconfigurationRequiringTestTypeSerializer) {
- // we mimic the reconfiguration by just re-instantiating the correct serializer
- return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
- new TestType.V1TestTypeSerializer());
- } else if (newSerializer instanceof TestType.IncompatibleTestTypeSerializer) {
+ } else if (oldSerializerSnapshot
+ instanceof TestType.ReconfigurationRequiringTestTypeSerializer
Review Comment:
This is always false, you compare serializer with snapshot.
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/SimpleTypeSerializerSnapshot.java:
##########
@@ -74,9 +74,14 @@ public TypeSerializer<T> restoreSerializer() {
@Override
public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer) {
-
- return newSerializer.getClass() == serializerSupplier.get().getClass()
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof SimpleTypeSerializerSnapshot)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ SimpleTypeSerializerSnapshot<T> oldSimpleTypeSerializerSnapshot =
+ (SimpleTypeSerializerSnapshot<T>) oldSerializerSnapshot;
+ return oldSimpleTypeSerializerSnapshot.serializerSupplier.get().getClass()
Review Comment:
Let's use `restoreSerializer` here.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V2TestTypeSerializerSnapshot.java:
##########
@@ -36,18 +36,14 @@ public int getCurrentVersion() {
@Override
public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(
- TypeSerializer<TestType> newSerializer) {
- if (newSerializer instanceof TestType.V2TestTypeSerializer) {
+ TypeSerializerSnapshot<TestType> oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof V2TestTypeSerializerSnapshot) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
- } else if (newSerializer instanceof TestType.ReconfigurationRequiringTestTypeSerializer) {
- // we mimic the reconfiguration by just re-instantiating the correct serializer
- return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
- new TestType.V2TestTypeSerializer());
- } else if (
- // migrating from V2 -> V1 is not supported
- newSerializer instanceof TestType.V1TestTypeSerializer
- || newSerializer instanceof TestType.IncompatibleTestTypeSerializer) {
-
+ } else if (oldSerializerSnapshot instanceof V1TestTypeSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ } else if (oldSerializerSnapshot
+ instanceof TestType.ReconfigurationRequiringTestTypeSerializer
+ || oldSerializerSnapshot instanceof TestType.IncompatibleTestTypeSerializer) {
Review Comment:
ditto, also always false, you compare snapshot with a serializer
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
What happens if a user has a custom serializer that embeds e.g. a `PojoSerializer` and want to migrate to the new method. A situation like:
```
class CustomCollectionSerializerSnapshot<T> extends TypeSerializerSnapshot<Collection<T>>{
private final TypeSerializerSnapshot innerSnapshot;
public ... resolveSchemaCompatibility(TypeSerializerSnapshot oldSnapshot) {
...
return innerSnapshot.resolveSchemaCompatibility(oldSnapshot.innerSnapshot);
}
}
CustomCollectionSerializer<? extends POJO> serializer
```
This would fail, wouldn't it?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java:
##########
@@ -200,8 +237,34 @@ public TestType deserialize(DataInputView source) throws IOException {
@Override
public TypeSerializerSnapshot<TestType> snapshotConfiguration() {
- throw new UnsupportedOperationException(
- "This is an incompatible serializer; shouldn't be used.");
+ return new TypeSerializerSnapshot<TestType>() {
+ @Override
+ public int getCurrentVersion() {
+ return 0;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) {
+ // do nothing
+ }
+
+ @Override
+ public void readSnapshot(
+ int readVersion, DataInputView in, ClassLoader userCodeClassLoader) {
+ // do nothing
+ }
+
+ @Override
+ public TypeSerializer<TestType> restoreSerializer() {
+ return new IncompatibleTestTypeSerializer();
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<TestType> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ };
Review Comment:
We shouldn't do that. Let's implement the original intent in a proper way.
If I understand the issue (why the code needs to be changed in the first place) is that you need to perform `newSerializer.snapshotConfiguration().resolveSchemaCompatibility`. Let's implement that properly.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/state/InternalPriorityQueueTestBase.java:
##########
@@ -546,13 +546,13 @@ public TypeSerializer<TestElement> restoreSerializer() {
@Override
public TypeSerializerSchemaCompatibility<TestElement> resolveSchemaCompatibility(
- TypeSerializer<TestElement> newSerializer) {
- if (!(newSerializer instanceof TestElementSerializer)) {
+ TypeSerializerSnapshot<TestElement> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof Snapshot)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
- TestElementSerializer testElementSerializer = (TestElementSerializer) newSerializer;
- return (revision <= testElementSerializer.getRevision())
+ Snapshot snapshot = (Snapshot) oldSerializerSnapshot;
+ return (revision <= snapshot.getRevision())
Review Comment:
That should be the other way round, shouldn't it? Old revision should be lower than the new one.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1089650991
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
I think the real reason why previous implementation could not run as we want is that my codes not meet "all callers use the old method" (Composite TypeSerializers will call the new method for their internal TypeSerializers).
I removed migration codes in PojoSerializerSnapshot, CompositeTypeSerializerSnapshot who rely on internal TypeSerializers so that "all callers use the old method".
these classes will be migrated in milestone 2.
Do you think it could work?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1402925601
##########
docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -442,4 +445,23 @@ migrate from the old abstractions. The steps to do this is as follows:
`TypeSerializerConfigSnapshot` implementation as will as the
`TypeSerializer#ensureCompatibility(TypeSerializerConfigSnapshot)` from the serializer).
+## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.18
Review Comment:
```suggestion
## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.19
```
And this part has a series of 1.18
##########
docs/content/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -345,12 +345,15 @@ public final class GenericArraySerializerSnapshot<C> extends CompositeTypeSerial
this.componentClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
}
- @Override
- protected boolean resolveOuterSchemaCompatibility(GenericArraySerializer newSerializer) {
- return (this.componentClass == newSerializer.getComponentClass())
- ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
- : OuterSchemaCompatibility.INCOMPATIBLE;
- }
+ @Override
+ protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(
+ TypeSerializerSnapshot<C[]> oldSerializerSnapshot) {
+ GenericArraySerializerSnapshot<C[]> oldGenericArraySerializerSnapshot =
+ (GenericArraySerializerSnapshot<C[]>) oldSerializerSnapshot;
+ return (this.componentClass == oldGenericArraySerializerSnapshot.componentClass)
+ ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
+ : OuterSchemaCompatibility.INCOMPATIBLE;
+ }
Review Comment:
Same comment: we should use 4 spaces instead of tab.
##########
docs/content.zh/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -342,12 +342,15 @@ public final class GenericArraySerializerSnapshot<C> extends CompositeTypeSerial
this.componentClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
}
- @Override
- protected boolean resolveOuterSchemaCompatibility(GenericArraySerializer newSerializer) {
- return (this.componentClass == newSerializer.getComponentClass())
- ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
- : OuterSchemaCompatibility.INCOMPATIBLE;
- }
+ @Override
+ protected OuterSchemaCompatibility resolveOuterSchemaCompatibility(
+ TypeSerializerSnapshot<C[]> oldSerializerSnapshot) {
+ GenericArraySerializerSnapshot<C[]> oldGenericArraySerializerSnapshot =
+ (GenericArraySerializerSnapshot<C[]>) oldSerializerSnapshot;
+ return (this.componentClass == oldGenericArraySerializerSnapshot.componentClass)
+ ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
+ : OuterSchemaCompatibility.INCOMPATIBLE;
+ }
Review Comment:
keep the code style are same. We should use 4 spaces instead of tab.
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java:
##########
@@ -34,21 +34,21 @@ public class CompositeTypeSerializerUtil {
* can be used by legacy snapshot classes, which have a newer implementation implemented as a
* {@link CompositeTypeSerializerSnapshot}.
*
- * @param newSerializer the new serializer to check for compatibility.
+ * @param legacySerializerSnapshot the legacy serializer snapshot to check for compatibility.
* @param newCompositeSnapshot an instance of the new snapshot class to delegate compatibility
* checks to. This instance should already contain the outer snapshot information.
* @param legacyNestedSnapshots the nested serializer snapshots of the legacy composite
* snapshot.
* @return the result compatibility.
*/
public static <T> TypeSerializerSchemaCompatibility<T> delegateCompatibilityCheckToNewSnapshot(
- TypeSerializer<T> newSerializer,
- CompositeTypeSerializerSnapshot<T, ? extends TypeSerializer> newCompositeSnapshot,
+ TypeSerializerSnapshot<T> legacySerializerSnapshot,
Review Comment:
It use the `legacy` as the name prefix, and the caller uses `old` as the prefix.
I see most of your changes use `old` or `previous` as the prefix, should we unify them?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1422192659
##########
flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java:
##########
@@ -1244,8 +1244,8 @@ void testStateMigrationAfterChangingTTLFromDisablingToEnabling() {
testKeyedValueStateUpgrade(
initialAccessDescriptor, newAccessDescriptorAfterRestore))
.satisfiesAnyOf(
- e -> assertThat(e).isInstanceOf(IllegalStateException.class),
- e -> assertThat(e).hasCauseInstanceOf(IllegalStateException.class));
+ e -> assertThat(e).isInstanceOf(StateMigrationException.class),
+ e -> assertThat(e).hasCauseInstanceOf(StateMigrationException.class));
}
Review Comment:
Would you mind adding one test to check all state backends call the new `resolveSchemaCompatibility(TypeSerializerSnapshot)` instead of old method?
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -289,6 +292,38 @@ protected void readOuterSnapshot(
int readOuterSnapshotVersion, DataInputView in, ClassLoader userCodeClassLoader)
throws IOException {}
+ /**
+ * Checks the schema compatibility of the given old serializer snapshot based on the outer
+ * snapshot.
+ *
+ * <p>The base implementation of this method assumes that the outer serializer only has nested
+ * serializers and no extra information, and therefore the result of the check is {@link
+ * OuterSchemaCompatibility#COMPATIBLE_AS_IS}. Otherwise, if the outer serializer contains some
+ * extra information that has been persisted as part of the serializer snapshot, this must be
+ * overridden. Note that this method and the corresponding methods {@link
+ * #writeOuterSnapshot(DataOutputView)}, {@link #readOuterSnapshot(int, DataInputView,
+ * ClassLoader)} needs to be implemented.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot, which contains the old outer
+ * information to check against.
+ * @return a {@link OuterSchemaCompatibility} indicating whether or the new serializer's outer
Review Comment:
```suggestion
* @return a {@link OuterSchemaCompatibility} indicating whether the new serializer's outer
```
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return oldSerializerSnapshot.resolveSchemaCompatibility(restoreSerializer());
Review Comment:
I'm not sure should we add a default implementation for the new `resolveSchemaCompatibility` method. I'm worried that providing a default implementation may not work well in some scenarios. We have 2 `resolveSchemaCompatibility` methods for now, and they call each other by default. This behavior may be a bug, for example:
- The `TypeSerializerSnapshot` of `data1` has old `resolveSchemaCompatibility(TypeSerializer)` method.
- The `TypeSerializerSnapshot` of `data2` has new `resolveSchemaCompatibility(TypeSerializerSnapshot)` method.
When flink users upgrade the data from data2 to data1, and flink will call the `resolveSchemaCompatibility(TypeSerializerSnapshot)` of data1, it will call the default implementation, so data1 will call the data2's `resolveSchemaCompatibility(TypeSerializer)`, and then it will call the default implementation. So call `resolveSchemaCompatibility(TypeSerializerSnapshot)` of data1 again.
It's a dead loop.
Based on this, I suggest all types inside of flink project should implement new `resolveSchemaCompatibility(TypeSerializerSnapshot)`. Currently, the `EitherSerializerSnapshot` and `GenericArraySerializerConfigSnapshot` don't implement it. WDYT?
Can we not provide a default implementation for the new `resolveSchemaCompatibility(TypeSerializerSnapshot)`? The default implementation is provided because the user may have a custom serializer, right?
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
Review Comment:
After this change, should all production callers call the new `resolveSchemaCompatibility` instead of old one?
For example:
<img width="1366" alt="image" src="https://github.com/apache/flink/assets/38427477/9e21ad30-c2cc-4c0d-8a00-e5eaf082c011">
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "llkj1 (via GitHub)" <gi...@apache.org>.
llkj1 commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879978170
感谢您与我们联系。(Thank you for contacting us.)
------------------------------------------------------------
我们收到了您的请求，并会在一至两个工作日内给您回复。您的案例编号是 102195427714。
想了解更多与开发相关主题的信息，请访问 开发者支持网页 (https://developer.apple.com/support/)。
Apple Developer Program Support
We’ve received your support request and will get back to you in one to two business days. Your case number is 102195427714.
For additional information on development-related topics, visit:
https://developer.apple.com/support/
Best regards,
Apple Developer Program Support
------------------------------------------------------------
Copyright (c) 2024 Apple Inc. All rights reserved.
Contact Us
https://developer.apple.com/contact/
Developer
https://developer.apple.com/
My Apple ID
https://appleid.apple.com
Privacy Policy
https://www.apple.com/privacy/
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081540937
##########
flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/ScalaEnumSerializerSnapshot.scala:
##########
@@ -80,25 +80,26 @@ class ScalaEnumSerializerSnapshot[E <: Enumeration] extends TypeSerializerSnapsh
new EnumValueSerializer(enumObject)
}
- override def resolveSchemaCompatibility(
- newSerializer: TypeSerializer[E#Value]): TypeSerializerSchemaCompatibility[E#Value] = {
+ override def resolveSchemaCompatibility(oldSerializerSnapshot: TypeSerializerSnapshot[E#Value])
+ : TypeSerializerSchemaCompatibility[E#Value] = {
Preconditions.checkState(enumClass != null)
- Preconditions.checkState(previousEnumConstants != null)
+ Preconditions.checkState(enumConstants != null)
- if (!newSerializer.isInstanceOf[EnumValueSerializer[E]]) {
+ if (!oldSerializerSnapshot.isInstanceOf[ScalaEnumSerializerSnapshot[E]]) {
return TypeSerializerSchemaCompatibility.incompatible()
}
- val newEnumSerializer = newSerializer.asInstanceOf[EnumValueSerializer[E]]
- if (!enumClass.equals(newEnumSerializer.enum.getClass)) {
+ val oldEnumSerializerSnapshot =
+ oldSerializerSnapshot.asInstanceOf[ScalaEnumSerializerSnapshot[E]]
+ if (!enumClass.equals(oldEnumSerializerSnapshot.enumClass)) {
return TypeSerializerSchemaCompatibility.incompatible()
}
- for ((previousEnumName, index) <- previousEnumConstants) {
+ for ((enumName, index) <- oldEnumSerializerSnapshot.enumConstants) {
Review Comment:
```suggestion
for ((oldEnumName, index) <- oldEnumSerializerSnapshot.enumConstants) {
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085579491
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
In milestone 1, the caller still use the old method.
So if a user has a snapshot implements the old method, it will not call the new method.
All behavior will be like before, right ?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1459246416
> Hey @masteryhx, do you plan to work on that for 1.18?
Sure, I'd like to. Thanks a lot for the reminder.
Big sorry for the delayed update due to my personal business last month.
I will rework on the pr this month.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085568912
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -290,26 +332,26 @@ protected void readOuterSnapshot(
throws IOException {}
/**
- * Checks whether the outer snapshot is compatible with a given new serializer.
+ * Checks the schema compatibility of the given old serializer snapshot based on the outer
+ * snapshot.
*
- * <p>The base implementation of this method just returns {@code true}, i.e. it assumes that the
- * outer serializer only has nested serializers and no extra information, and therefore the
- * result of the check must always be true. Otherwise, if the outer serializer contains some
+ * <p>The base implementation of this method assumes that the outer serializer only has nested
+ * serializers and no extra information, and therefore the result of the check is {@link
+ * OuterSchemaCompatibility#COMPATIBLE_AS_IS}. Otherwise, if the outer serializer contains some
* extra information that has been persisted as part of the serializer snapshot, this must be
* overridden. Note that this method and the corresponding methods {@link
* #writeOuterSnapshot(DataOutputView)}, {@link #readOuterSnapshot(int, DataInputView,
* ClassLoader)} needs to be implemented.
*
- * @param newSerializer the new serializer, which contains the new outer information to check
- * against.
- * @return a flag indicating whether or not the new serializer's outer information is compatible
- * with the one written in this snapshot.
- * @deprecated this method is deprecated, and will be removed in the future. Please implement
- * {@link #resolveOuterSchemaCompatibility(TypeSerializer)} instead.
+ * @param oldSerializerSnapshot the old serializer snapshot, which contains the old outer
+ * information to check against.
+ * @return a {@link OuterSchemaCompatibility} indicating whether or the new serializer's outer
+ * information is compatible, requires migration, or incompatible with the one written in
+ * this snapshot.
*/
- @Deprecated
- protected boolean isOuterSnapshotCompatible(S newSerializer) {
Review Comment:
Actually, `isOuterSnapshotCompatible` will not be used in current implementation.
Currently, the old method called `OuterSchemaCompatibility resolveOuterSchemaCompatibility(S newSerializer)` behaves like the old method in TypeSerializer -- it calls the new method as its default implementation.
Users could make its logic work without modifying any codes.
Also if they need to migrate, they could implement the new method called `OuterSchemaCompatibility resolveOuterSchemaCompatibility(TypeSerializerSnapshot<T> oldSerializerSnapshot)` and remove the old method.
In this logic, I think maybe you're right. Users will find its original `isOuterSnapshotCompatible` could not work which will confused them.
Maybe we still need to remove it and add descripation in the release note, WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1086625857
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
I don't think this will work. Take this for example:
```
@Test
public void testCompatibleWithNonMigratedSerializer() throws IOException {
TypeSerializer<?>[] testNestedSerializers = {
new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS),
new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),
};
TypeSerializerSchemaCompatibility<String> compatibility =
snapshotCompositeSerializerAndGetSchemaCompatibilityAfterRestore(
testNestedSerializers,
testNestedSerializers,
OuterSchemaCompatibility.COMPATIBLE_AS_IS);
Assert.assertTrue(compatibility.isCompatibleAsIs());
}
```
where the `new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),` returns a snapshot that still implements `resolveSchemaCompatibility(TypeSerializer<String> newSerializer)` instead of the new method.
This should work, but as it stands it returns `incompatible` forcing user to implement the new method.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1605893295
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21635: Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1376775464
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f03e78346937fd7085d0d03b59610a14cd8180f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8f03e78346937fd7085d0d03b59610a14cd8180f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f03e78346937fd7085d0d03b59610a14cd8180f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1092823807
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
Thanks for the detail case.
It seems it's the most critical problem so I think we could resolve this firstly.
For resolving this, there are some solutions:
1. implement all internal serializers as before pr but not remove the old deprecated method for composite serializers:
a. Benefit: From users' perspective, previous existing custon serializers and new serializers could work no matter whether users migrate to the new method:
ⅰ. For some previous existing custom serializers that don't migrate, all old code paths will be accessed as before. Users' codes could run without any modifying.
ⅱ. For some previous existing custom serializers that want to migrate, they could just implement the new method.
1. If users don't remove the old method, it could also work. Of course, we recomend users to remove it.
2. If users has used the internal serializers, the new method could just be called.
ⅲ. For new defined custom serializers:
1. If users don't implement any methods and restore from the snapshot. INCOMPATIBLE will be returned and the job will fail. Of course, we highly recomend users to implement the new method.
2. If users implement the new method, it's same as [b].
b. Cost: there are many deprecated method remained.
2. Also implement the new method to call the old method:
a. Benefit: Then we could just migrate all internal serializers and remove old methods.
b. Cost: It requires users to implement at least one method else it will stuck in an infinite loop. Although Option 1 also have requirement for users, this is a worse case obviously.
3. Just remove the old method in the interface:
a. Benefit: it will make code path clear and we could migrate all serializers and make all caller use the new method
b. Cost: it will beak change so all users have to migrate their serializers to the new method. Else their jobs could not compile.
I think Option 1 looks good to me. WDYT? Or any other suggestions ?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1884161683
Thanks all for the detailed review.
Please let me know if any other comments.
I will merge it if no other comments beyond two days from now on.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1855044542
> Hi @masteryhx , thanks for the contribution and update!
>
> I have reviewed this PR and left some comments. Please take a look in your free time, thanks~
>
> Also, `The Schema Compatibility` related code is Flink’s infrastructure and is used by many components. I didn't change and read related code before this PR, so I'm not professional in this area. I have reviewed, but it would be great to have more committers who are familiar with this area to help review. It will be helpful for the bug-free.
Thanks a lot for the detailed review.
We may have some key problems remaing in the comment to discuss.
I'll update the pr after we are on the same way about these problems.
Also ping @Zakelly @fredia here to help to review and discuss together, Thanks a lot for your time!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081243028
##########
flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java:
##########
@@ -56,6 +55,9 @@
/** Contains the actual content for the serializer snapshot. */
private PojoSerializerSnapshotData<T> snapshotData;
+ /** Configuration of the current execution. */
+ private ExecutionConfig executionConfig;
Review Comment:
This is a serious issue :(
We should not do that ever! This gives an impression the `executionConfig` is part of the serialized snapshot which it **must** not!
If I understand correctly this is necessary, because the method `resolveSchemaCompatibility(TypeSerializerSnapshot<T> oldSerializerSnapshot` is called at a point where the originating `newSerializer` has not been used yet and thus has not registeredSubclassSerializers.
We need to fix it in a different way. The current approach makes it even messier than it already is and we can not go with that approach.
I'll give it a thought.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081544400
##########
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/RowDataSerializer.java:
##########
@@ -194,31 +193,33 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
@Override
public RowDataSerializer restoreSerializer() {
return new RowDataSerializer(
- previousTypes,
- nestedSerializersSnapshotDelegate.getRestoredNestedSerializers());
+ types, nestedSerializersSnapshotDelegate.getRestoredNestedSerializers());
}
@Override
public TypeSerializerSchemaCompatibility<RowData> resolveSchemaCompatibility(
- TypeSerializer<RowData> newSerializer) {
- if (!(newSerializer instanceof RowDataSerializer)) {
+ TypeSerializerSnapshot<RowData> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof RowDataSerializerSnapshot)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
- RowDataSerializer newRowSerializer = (RowDataSerializer) newSerializer;
- if (!Arrays.equals(previousTypes, newRowSerializer.fieldTypes)) {
+ RowDataSerializerSnapshot oldRowDataSerializerSnapshot =
+ (RowDataSerializerSnapshot) oldSerializerSnapshot;
+ if (!Arrays.equals(types, oldRowDataSerializerSnapshot.types)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
CompositeTypeSerializerUtil.IntermediateCompatibilityResult<RowData>
intermediateResult =
CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult(
- newRowSerializer.fieldSerializers,
+ oldRowDataSerializerSnapshot.nestedSerializersSnapshotDelegate
+ .getNestedSerializerSnapshots(),
nestedSerializersSnapshotDelegate
.getNestedSerializerSnapshots());
if (intermediateResult.isCompatibleWithReconfiguredSerializer()) {
- RowDataSerializer reconfiguredCompositeSerializer = restoreSerializer();
+ org.apache.flink.table.runtime.typeutils.RowDataSerializer
Review Comment:
Is that correct? This looks wrong.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1856242365
Hi @masteryhx , sorry for the late reply!
After reading your PR, I finally figured out why the default implementation of old method and new method should call each other. I think we could do something tricky to check whether one of the old and new methods has been implemented in specific `TypeSerializerSnapshot` before it is called. https://stackoverflow.com/a/2315467 this may provide some ideas. WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879974970
@1996fanrui @Zakelly Thanks for the review. I have updated remaining minor comments.
The failed CI about python occurs from yesterday (see other [failed cases](https://dev.azure.com/apache-flink/apache-flink/_build?definitionId=2&_a=summary)) which is not related to this pr.
@curcur Could you also help to take a review ? Thanks a lot!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1426108037
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return oldSerializerSnapshot.resolveSchemaCompatibility(restoreSerializer());
Review Comment:
Thanks for the detailed review.
IIUC, Only If users has implemented one of the methods (old one or new one), it should work well in current PR.
Sorry maybe I missed your point. The case you described is an nested serializer, right ?
Whichever method `data1` and `data2` implement, the call should be onesided, otherwise the logic should be error even if we don't have this pr.
The reason why we provide two default methods is that users could not use mixed migrated method and non-migrated method as you could saw the discussion [#21635(comment)](https://github.com/apache/flink/pull/21635#discussion_r1086625857)
Just cc @Zakelly as we also talked about this problem offline.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1090310149
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
Could you please add changes after reviews as separate commits? This makes it easier to track what has changed since the last review.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1086625857
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
I don't think this will work. Take this for example:
```
@Test
public void testCompatibleWithNonMigratedSerializer() throws IOException {
TypeSerializer<?>[] testNestedSerializers = {
new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),
new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),
};
TypeSerializerSchemaCompatibility<String> compatibility =
snapshotCompositeSerializerAndGetSchemaCompatibilityAfterRestore(
testNestedSerializers,
testNestedSerializers,
OuterSchemaCompatibility.COMPATIBLE_AS_IS);
Assert.assertTrue(compatibility.isCompatibleAsIs());
}
```
where the `new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),` returns a snapshot that still implements `resolveSchemaCompatibility(TypeSerializer<String> newSerializer)` instead of the new method.
This should work, but as it stands it returns `incompatible` forcing user to implement the new method.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085539860
##########
flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshotTest.java:
##########
@@ -232,14 +232,11 @@ public void testRestoreCompositeTypeSerializer() throws IOException {
// the restored nested serializer should be a RestoredNestedSerializer
testSerializer = (TestCompositeTypeSerializer) testSerializerSnapshot.restoreSerializer();
Assert.assertTrue(
- testSerializer.getNestedSerializers()[0].getClass()
Review Comment:
Yes, it's updated incorrectly.
I have restored the code.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "llkj1 (via GitHub)" <gi...@apache.org>.
llkj1 commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879976379
感谢您与我们联系。(Thank you for contacting us.)
------------------------------------------------------------
我们收到了您的请求，并会在一至两个工作日内给您回复。您的案例编号是 102195422930。
想了解更多与开发相关主题的信息，请访问 开发者支持网页 (https://developer.apple.com/support/)。
Apple Developer Program Support
We’ve received your support request and will get back to you in one to two business days. Your case number is 102195422930.
For additional information on development-related topics, visit:
https://developer.apple.com/support/
Best regards,
Apple Developer Program Support
------------------------------------------------------------
Copyright (c) 2024 Apple Inc. All rights reserved.
Contact Us
https://developer.apple.com/contact/
Developer
https://developer.apple.com/
My Apple ID
https://appleid.apple.com
Privacy Policy
https://www.apple.com/privacy/
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "llkj1 (via GitHub)" <gi...@apache.org>.
llkj1 commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879977820
感谢您与我们联系。(Thank you for contacting us.)
------------------------------------------------------------
我们收到了您的请求，并会在一至两个工作日内给您回复。您的案例编号是 102195426596。
想了解更多与开发相关主题的信息，请访问 开发者支持网页 (https://developer.apple.com/support/)。
Apple Developer Program Support
We’ve received your support request and will get back to you in one to two business days. Your case number is 102195426596.
For additional information on development-related topics, visit:
https://developer.apple.com/support/
Best regards,
Apple Developer Program Support
------------------------------------------------------------
Copyright (c) 2024 Apple Inc. All rights reserved.
Contact Us
https://developer.apple.com/contact/
Developer
https://developer.apple.com/
My Apple ID
https://appleid.apple.com
Privacy Policy
https://www.apple.com/privacy/
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1093196114
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
I disagree with statements in Option 1. Option 1 is basically the current state of the PR and thus have all the problems listed in this thread. In Option 1 one cannot mix migrated and non-migrated serializers, it simply does not work. You'd have to force users to always implement both methods and we'd never use the new path.
TBH, when reviewing the FLIP, I thought option 2 is the suggested one, which was obviously my mistake. Because of the con you listed I was advocating for immediately dropping the old method instead of deprecating it (option 3).
There was a strong preference for keeping compatibility with old user programs in the ML, thus I'd suggest to go with option 2. The con affects only newly written serializers, which is usually done carefully, so I am not worried it's gonna be too much of a hassle. We took that approach for much more commonly used interfaces such as e.g. `SinkFunction#invoke`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1093196114
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
I disagree with statements in Option 1. Option 1 is basically the current state of the PR and thus have all the problems listed in this thread. In Option 1 one cannot mix migrated and non-migrated serializers, it simply does not work. You'd have to force users to always implement both methods.
TBH, when reviewing the FLIP, I thought option 2 is the suggested one, which was obviously my mistake. Because of the con you listed I was advocating for immediately dropping the old method instead of deprecating it (option 3).
There was a strong preference for keeping compatibility with old user programs in the ML, thus I'd suggest to go with option 2. The con affects only newly written serializers, which is usually done carefully, so I am not worried it's gonna be too much of a hassle. We took that approach for much more commonly used interfaces such as e.g. `SinkFunction#invoke`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1407693515
Sure, I will during my working days. Also to be clear it's definitely too late now for it to make it to 1.17.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1084775911
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -290,26 +332,26 @@ protected void readOuterSnapshot(
throws IOException {}
/**
- * Checks whether the outer snapshot is compatible with a given new serializer.
+ * Checks the schema compatibility of the given old serializer snapshot based on the outer
+ * snapshot.
*
- * <p>The base implementation of this method just returns {@code true}, i.e. it assumes that the
- * outer serializer only has nested serializers and no extra information, and therefore the
- * result of the check must always be true. Otherwise, if the outer serializer contains some
+ * <p>The base implementation of this method assumes that the outer serializer only has nested
+ * serializers and no extra information, and therefore the result of the check is {@link
+ * OuterSchemaCompatibility#COMPATIBLE_AS_IS}. Otherwise, if the outer serializer contains some
* extra information that has been persisted as part of the serializer snapshot, this must be
* overridden. Note that this method and the corresponding methods {@link
* #writeOuterSnapshot(DataOutputView)}, {@link #readOuterSnapshot(int, DataInputView,
* ClassLoader)} needs to be implemented.
*
- * @param newSerializer the new serializer, which contains the new outer information to check
- * against.
- * @return a flag indicating whether or not the new serializer's outer information is compatible
- * with the one written in this snapshot.
- * @deprecated this method is deprecated, and will be removed in the future. Please implement
- * {@link #resolveOuterSchemaCompatibility(TypeSerializer)} instead.
+ * @param oldSerializerSnapshot the old serializer snapshot, which contains the old outer
+ * information to check against.
+ * @return a {@link OuterSchemaCompatibility} indicating whether or the new serializer's outer
+ * information is compatible, requires migration, or incompatible with the one written in
+ * this snapshot.
*/
- @Deprecated
- protected boolean isOuterSnapshotCompatible(S newSerializer) {
Review Comment:
I thinked again and added it back to avoid breaking change.
I think it's better to remove these deprecated methods in milestone 2.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081546445
##########
flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/MapDataSerializer.java:
##########
@@ -227,25 +225,22 @@ public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCode
@Override
public TypeSerializer<MapData> restoreSerializer() {
- return new MapDataSerializer(
- previousKeyType,
- previousValueType,
- previousKeySerializer,
- previousValueSerializer);
+ return new MapDataSerializer(keyType, valueType, keySerializer, valueSerializer);
}
@Override
public TypeSerializerSchemaCompatibility<MapData> resolveSchemaCompatibility(
- TypeSerializer<MapData> newSerializer) {
- if (!(newSerializer instanceof MapDataSerializer)) {
+ TypeSerializerSnapshot<MapData> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof BaseMapSerializerSnapshot)) {
Review Comment:
Why has the type changed in the check?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1594300571
@Zakelly @reswqa Could you also help to take a look ? Thanks a lot!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1405573786
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerUtil.java:
##########
@@ -34,21 +34,21 @@ public class CompositeTypeSerializerUtil {
* can be used by legacy snapshot classes, which have a newer implementation implemented as a
* {@link CompositeTypeSerializerSnapshot}.
*
- * @param newSerializer the new serializer to check for compatibility.
+ * @param legacySerializerSnapshot the legacy serializer snapshot to check for compatibility.
* @param newCompositeSnapshot an instance of the new snapshot class to delegate compatibility
* checks to. This instance should already contain the outer snapshot information.
* @param legacyNestedSnapshots the nested serializer snapshots of the legacy composite
* snapshot.
* @return the result compatibility.
*/
public static <T> TypeSerializerSchemaCompatibility<T> delegateCompatibilityCheckToNewSnapshot(
- TypeSerializer<T> newSerializer,
- CompositeTypeSerializerSnapshot<T, ? extends TypeSerializer> newCompositeSnapshot,
+ TypeSerializerSnapshot<T> legacySerializerSnapshot,
Review Comment:
The `legacy` means it's deprecated and the code is just used for temporary compatibility check as before.
It's different from regular compatibility check from old/previous to new one.
I have migrated `previous` to `old` for some classes.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1580620144
@dawidwys
Hi, Sorry for the late update due to personal business.
I have updated some previous commits so just also splitted them into different commits for the convinence of review.
I will continue to update after your every review in the upcoming time.
BTW, The caller could call the new method instead of the old one currently, I will prepare a new pr for it after this pr.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1221424165
##########
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java:
##########
@@ -200,8 +237,34 @@ public TestType deserialize(DataInputView source) throws IOException {
@Override
public TypeSerializerSnapshot<TestType> snapshotConfiguration() {
- throw new UnsupportedOperationException(
- "This is an incompatible serializer; shouldn't be used.");
+ return new TypeSerializerSnapshot<TestType>() {
+ @Override
+ public int getCurrentVersion() {
+ return 0;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) {
+ // do nothing
+ }
+
+ @Override
+ public void readSnapshot(
+ int readVersion, DataInputView in, ClassLoader userCodeClassLoader) {
+ // do nothing
+ }
+
+ @Override
+ public TypeSerializer<TestType> restoreSerializer() {
+ return new IncompatibleTestTypeSerializer();
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<TestType> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<TestType> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ };
Review Comment:
Sorry, maybe I missed something.
You mean that we should not suppose the method will be supported by all TypeSerializers, some exceptions should be handled in the caller logic ?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1383357691
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1084777905
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerSnapshot.java:
##########
@@ -76,10 +80,26 @@ protected void readOuterSnapshot(
this.componentClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
}
+ @Override
+ public TypeSerializerSchemaCompatibility<C[]> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<C[]> oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof GenericArraySerializerConfigSnapshot) {
Review Comment:
You are right.
I have added CompositeTypeSerializerUpgradeTest#testUpgradeFromDeprecatedSnapshot to test it.
BTW, I have a minor problem.
TypeSerializerUpgradeTestBase seems just to test upgradation from Flink Version 1.11 ?
IIUC, TypeSerializer should supports Flink Version from 1.8 currently.
Should we need to make it test from 1.8 ?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1401341853
> I started the review, but since I started finding issues or potential issues in the migrated methods, would you be so kind and split migrating serializers/snapshots to separate commits? That would help me tremendously to do a proper review.
Thanks a lot for the review.
I have splitted them as clear as possible.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085560912
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
> It's implemented just as we talked in https://lists.apache.org/thread/4zgpd4cw232xcjp3cnj2hlokf6zvm3qc
Maybe I missed something there.
Looking at the code. Imagine a situation a user has a snapshot that implements custom logic in `TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<T> newSerializer)`. Our updated code goes through `TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializerSnapshot<T> oldSerializerSnapshot)`, which in this case will return `incompatible`. In result there is no real compatibility and the user is forced to implement the new method. We may as well simply drop the old method, which will even give you a nicer output as the program won't even compile instead of failing during runtime in (sic!) restore.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1090310149
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
Could you please add changes after reviews as separate commits? This makes it easier to track what has changed since the last review.
BTW, don't need to split it now, but I'd appreciate that for the future.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1092823807
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
Thanks for the detail case.
It seems it's the most critical problem so I think we could resolve this firstly.
For resolving this, there are some solutions:
1. implement all internal serializers as before pr but not remove the old deprecated method for composite serializers:
a. Benefit: From users' perspective, previous existing custon serializers and new serializers could work no matter whether users migrate to the new method:
ⅰ. For some previous existing custom serializers that don't migrate, all old code paths will be accessed as before. Users' codes could run without any modifying.
ⅱ. For some previous existing custom serializers that want to migrate, they could just implement the new method.
1. If users don't remove the old method, it could also work. Of course, we recomend users to remove it.
2. If users has used the internal serializers, the new method could just be called.
ⅲ. For new defined custom serializers:
1. If users don't implement any methods and restore from the snapshot. INCOMPATIBLE will be returned and the job will fail. Of course, we highly recomend users to implement the new method.
2. If users implement the new method, it's same as [ⅱ].
b. Cost: there are many deprecated method remained.
2. Also implement the new method to call the old method:
a. Benefit: Then we could just migrate all internal serializers and remove old methods.
b. Cost: It requires users to implement at least one method else it will stuck in an infinite loop. Although Option 1 also have requirement for users, this is a worse case obviously.
3. Just remove the old method in the interface:
a. Benefit: it will make code path clear and we could migrate all serializers and make all caller use the new method
b. Cost: it will beak change so all users have to migrate their serializers to the new method. Else their jobs could not compile.
I think Option 1 looks good to me. WDYT? Or any other suggestions ?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1427497404
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return oldSerializerSnapshot.resolveSchemaCompatibility(restoreSerializer());
Review Comment:
> IIUC, Flink just calls the new resolveSchemaCompatibility(TypeSerializerSnapshot) method in the future, right?
Yes, Flink could make sure this.
But IIUC, `TypeSerializerSnapshot` and `TypeSerializer` are public API which means they maybe also used in the user code. Then users still have to modify their codes which may break changes.
> BTW, the code comment should guide users or developers to implement the new resolveSchemaCompatibility(TypeSerializerSnapshot) method in the future.
Thanks for the advice. That's a good idea.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1427798414
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
Review Comment:
Sounds make sense!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "llkj1 (via GitHub)" <gi...@apache.org>.
llkj1 commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879975416
感谢您与我们联系。(Thank you for contacting us.)
------------------------------------------------------------
我们收到了您的请求，并会在一至两个工作日内给您回复。您的案例编号是 102195420550。
想了解更多与开发相关主题的信息，请访问 开发者支持网页 (https://developer.apple.com/support/)。
Apple Developer Program Support
We’ve received your support request and will get back to you in one to two business days. Your case number is 102195420550.
For additional information on development-related topics, visit:
https://developer.apple.com/support/
Best regards,
Apple Developer Program Support
------------------------------------------------------------
Copyright (c) 2024 Apple Inc. All rights reserved.
Contact Us
https://developer.apple.com/contact/
Developer
https://developer.apple.com/
My Apple ID
https://appleid.apple.com
Privacy Policy
https://www.apple.com/privacy/
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "llkj1 (via GitHub)" <gi...@apache.org>.
llkj1 commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1879977057
感谢您与我们联系。(Thank you for contacting us.)
------------------------------------------------------------
我们收到了您的请求，并会在一至两个工作日内给您回复。您的案例编号是 102195424272。
想了解更多与开发相关主题的信息，请访问 开发者支持网页 (https://developer.apple.com/support/)。
Apple Developer Program Support
We’ve received your support request and will get back to you in one to two business days. Your case number is 102195424272.
For additional information on development-related topics, visit:
https://developer.apple.com/support/
Best regards,
Apple Developer Program Support
------------------------------------------------------------
Copyright (c) 2024 Apple Inc. All rights reserved.
Contact Us
https://developer.apple.com/contact/
Developer
https://developer.apple.com/
My Apple ID
https://appleid.apple.com
Privacy Policy
https://www.apple.com/privacy/
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1093196114
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
I disagree with statements in Option 1. Option 1 is basically the current state of the PR and thus have all the problems listed in this thread. In Option 1 one cannot mix migrated and non-migrated serializers, it simply does not work.
TBH, when reviewing the FLIP, I thought option 2 is the suggested one, which was obviously my mistake. Because of the con you listed I was advocating for immediately dropping the old method instead of deprecating it (option 3).
There was a strong preference for keeping compatibility with old user programs in the ML, thus I'd suggest to go with option 2. The con affects only newly written serializers, which is usually done carefully, so I am not worried it's gonna be too much of a hassle. We took that approach for much more commonly used interfaces such as e.g. `SinkFunction#invoke`
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1072181884
##########
docs/content/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -345,12 +345,15 @@ public final class GenericArraySerializerSnapshot<C> extends CompositeTypeSerial
this.componentClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
}
- @Override
- protected boolean resolveOuterSchemaCompatibility(GenericArraySerializer newSerializer) {
- return (this.componentClass == newSerializer.getComponentClass())
- ? OuterSchemaCompatibility.COMPATIBLE_AS_IS
- : OuterSchemaCompatibility.INCOMPATIBLE;
- }
+ @Override
Review Comment:
please fix the indentation
##########
docs/content/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -446,4 +449,23 @@ migrate from the old abstractions. The steps to do this is as follows:
`TypeSerializerConfigSnapshot` implementation as will as the
`TypeSerializer#ensureCompatibility(TypeSerializerConfigSnapshot)` from the serializer).
+## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.16
+
+This section is a guide for method migration from serializer snapshots that existed before Flink 1.16.
Review Comment:
```suggestion
This section is a guide for a method migration from the serializer snapshots that existed before Flink 1.17.
```
##########
docs/content/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -446,4 +449,23 @@ migrate from the old abstractions. The steps to do this is as follows:
`TypeSerializerConfigSnapshot` implementation as will as the
`TypeSerializer#ensureCompatibility(TypeSerializerConfigSnapshot)` from the serializer).
+## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.16
Review Comment:
```suggestion
## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.17
```
##########
docs/content/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -446,4 +449,23 @@ migrate from the old abstractions. The steps to do this is as follows:
`TypeSerializerConfigSnapshot` implementation as will as the
`TypeSerializer#ensureCompatibility(TypeSerializerConfigSnapshot)` from the serializer).
+## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.16
+
+This section is a guide for method migration from serializer snapshots that existed before Flink 1.16.
+
+Before Flink 1.16, when using a customized serializer to process data, the schema compatibility in the old serializer
Review Comment:
```suggestion
Before Flink 1.17, when using a customized serializer to process data, the schema compatibility in the old serializer
```
##########
docs/content/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -446,4 +449,23 @@ migrate from the old abstractions. The steps to do this is as follows:
`TypeSerializerConfigSnapshot` implementation as will as the
`TypeSerializer#ensureCompatibility(TypeSerializerConfigSnapshot)` from the serializer).
+## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.16
+
+This section is a guide for method migration from serializer snapshots that existed before Flink 1.16.
+
+Before Flink 1.16, when using a customized serializer to process data, the schema compatibility in the old serializer
+(maybe in Flink library) has to meet the future need.
+Or else TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer<T> newSerializer) of the old serializer has to be modified.
+There are no ways to specify the compatibility with the old serializer in the new serializer, which also makes scheme evolution
+not supported in some scenarios.
+
+So from Flink 1.17, the direction of resolving schema compatibility has been reversed. The old method
+`TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` has been marked as deprecated
+and will be removed in the future. it is highly recommended to migrate from the old one to
+`TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializerSnapshot oldSerializerSnapshot)`. The steps to do this is as follows:
Review Comment:
```suggestion
`TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializerSnapshot oldSerializerSnapshot)`. The steps to do this are as follows:
```
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -369,13 +417,13 @@ private void legacyInternalReadOuterSnapshot(
}
private TypeSerializerSchemaCompatibility<T> constructFinalSchemaCompatibilityResult(
- TypeSerializer<?>[] newNestedSerializers,
- TypeSerializerSnapshot<?>[] nestedSerializerSnapshots,
+ TypeSerializerSnapshot<?>[] newNestedSerializerSnapshot,
Review Comment:
`newNestedSerializerSnapshot` -> `newNestedSerializerSnapshots`
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -290,26 +332,26 @@ protected void readOuterSnapshot(
throws IOException {}
/**
- * Checks whether the outer snapshot is compatible with a given new serializer.
+ * Checks the schema compatibility of the given old serializer snapshot based on the outer
+ * snapshot.
*
- * <p>The base implementation of this method just returns {@code true}, i.e. it assumes that the
- * outer serializer only has nested serializers and no extra information, and therefore the
- * result of the check must always be true. Otherwise, if the outer serializer contains some
+ * <p>The base implementation of this method assumes that the outer serializer only has nested
+ * serializers and no extra information, and therefore the result of the check is {@link
+ * OuterSchemaCompatibility#COMPATIBLE_AS_IS}. Otherwise, if the outer serializer contains some
* extra information that has been persisted as part of the serializer snapshot, this must be
* overridden. Note that this method and the corresponding methods {@link
* #writeOuterSnapshot(DataOutputView)}, {@link #readOuterSnapshot(int, DataInputView,
* ClassLoader)} needs to be implemented.
*
- * @param newSerializer the new serializer, which contains the new outer information to check
- * against.
- * @return a flag indicating whether or not the new serializer's outer information is compatible
- * with the one written in this snapshot.
- * @deprecated this method is deprecated, and will be removed in the future. Please implement
- * {@link #resolveOuterSchemaCompatibility(TypeSerializer)} instead.
+ * @param oldSerializerSnapshot the old serializer snapshot, which contains the old outer
+ * information to check against.
+ * @return a {@link OuterSchemaCompatibility} indicating whether or the new serializer's outer
+ * information is compatible, requires migration, or incompatible with the one written in
+ * this snapshot.
*/
- @Deprecated
- protected boolean isOuterSnapshotCompatible(S newSerializer) {
Review Comment:
Do I see it correctly that you've removed this deprecated method? If so , please make sure this ends up in the release notes.
##########
docs/content/docs/dev/datastream/fault-tolerance/serialization/custom_serialization.md:
##########
@@ -446,4 +449,23 @@ migrate from the old abstractions. The steps to do this is as follows:
`TypeSerializerConfigSnapshot` implementation as will as the
`TypeSerializer#ensureCompatibility(TypeSerializerConfigSnapshot)` from the serializer).
+## Migrating from deprecated `TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer newSerializer)` before Flink 1.16
Review Comment:
@alpinegizmo Do you mind taking a look at the docs part?
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializerSnapshot.java:
##########
@@ -76,10 +80,26 @@ protected void readOuterSnapshot(
this.componentClass = InstantiationUtil.resolveClassByName(in, userCodeClassLoader);
}
+ @Override
+ public TypeSerializerSchemaCompatibility<C[]> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<C[]> oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof GenericArraySerializerConfigSnapshot) {
Review Comment:
imo, there is something wrong with this branch. If the `oldSerializerSnapshot` is `GenericArraySerializerConfigSnapshot` we then call:
`CompositeTypeSerializerUtil.delegateCompatibilityCheckToNewSnapshot` -> `CompositeTypeSerializerSnapshot#internalResolveSchemaCompatibility` ->
`GenericArraySerializerSnapshot#resolveOuterSchemaCompatibility` where we try to cast the initial
`GenericArraySerializerConfigSnapshot` to `GenericArraySerializerSnapshot` which will fail.
Could you please verify that, preferably with a test? Or if it works point me to a test that checks that?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081243028
##########
flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java:
##########
@@ -56,6 +55,9 @@
/** Contains the actual content for the serializer snapshot. */
private PojoSerializerSnapshotData<T> snapshotData;
+ /** Configuration of the current execution. */
+ private ExecutionConfig executionConfig;
Review Comment:
This is a serious issue :(
This gives an impression the `executionConfig` is part of the serialized snapshot which it **must** not!
If I understand correctly this is necessary, because the method `resolveSchemaCompatibility(TypeSerializerSnapshot<T> oldSerializerSnapshot` is called at a point where the originating `newSerializer` has not been used yet and thus has not registeredSubclassSerializers.
I'll give it a thought.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by GitBox <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1081388117
##########
flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java:
##########
@@ -56,6 +55,9 @@
/** Contains the actual content for the serializer snapshot. */
private PojoSerializerSnapshotData<T> snapshotData;
+ /** Configuration of the current execution. */
+ private ExecutionConfig executionConfig;
Review Comment:
After thinking over it for a while I somewhat softened my opinion here.
Yes, it is messy, but I think we're forced to do it that way :( Do you mind adding a big, strong warning next to the `executionConfig` that it should never, under any circumstances end up serialized?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085267360
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
Shouldn't this use the old method instead?
What is the point of keeping `resolveSchemaCompatibility(TypeSerializer<T> newSerializer` if users need to migrate their serializers already or they won't work?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1579956303
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1427799403
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return oldSerializerSnapshot.resolveSchemaCompatibility(restoreSerializer());
Review Comment:
> `TypeSerializerSnapshot and TypeSerializer` are public API which means they maybe also used in the user code. Then users still have to modify their codes which may break changes.
Thanks for the clarification!
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1426111500
##########
flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendMigrationTestBase.java:
##########
@@ -1244,8 +1244,8 @@ void testStateMigrationAfterChangingTTLFromDisablingToEnabling() {
testKeyedValueStateUpgrade(
initialAccessDescriptor, newAccessDescriptorAfterRestore))
.satisfiesAnyOf(
- e -> assertThat(e).isInstanceOf(IllegalStateException.class),
- e -> assertThat(e).hasCauseInstanceOf(IllegalStateException.class));
+ e -> assertThat(e).isInstanceOf(StateMigrationException.class),
+ e -> assertThat(e).hasCauseInstanceOf(StateMigrationException.class));
}
Review Comment:
As discussed in [#21635(comment)](https://github.com/apache/flink/pull/21635#discussion_r1426110489), I'd like to also add more state level test after adjusting in a seprate pr. WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30613] Improve resolving schema compatibility -- Milestone one [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1427479694
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return oldSerializerSnapshot.resolveSchemaCompatibility(restoreSerializer());
Review Comment:
After thinking more about it, I'm wondering if we don't need a compatibility implementation for the old `resolveSchemaCompatibility(TypeSerializer)` method.
IIUC, Flink just calls the new `resolveSchemaCompatibility(TypeSerializerSnapshot)` method in the future, right?
If so, we can add a default implementation for the old `resolveSchemaCompatibility(TypeSerializer)` method, the default implementation throwes `UnsupportedException`. We need to add a default implementation because users or developers don't need to implement it in the future.
For the new `resolveSchemaCompatibility(TypeSerializerSnapshot)` method, you current implementation is fine. All old implementation class can be supported if `resolveSchemaCompatibility(TypeSerializerSnapshot)` calls the old one.
BTW, the code comment should guide users or developers to implement the new `resolveSchemaCompatibility(TypeSerializerSnapshot)` method in the future.
WDYT?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1221418449
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
Thanks a lot for the detailed explanation.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1579768878
@flinkbot run azure
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on PR #21635:
URL: https://github.com/apache/flink/pull/21635#issuecomment-1455840478
Hey @masteryhx, do you plan to work on that for 1.18?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1084778811
##########
flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerSnapshot.java:
##########
@@ -56,6 +55,9 @@
/** Contains the actual content for the serializer snapshot. */
private PojoSerializerSnapshotData<T> snapshotData;
+ /** Configuration of the current execution. */
+ private ExecutionConfig executionConfig;
Review Comment:
Thanks for the reminder.
I have added a comment above the field.
Please help me check again whether it fits your idea, Thanks.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085369657
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -290,26 +332,26 @@ protected void readOuterSnapshot(
throws IOException {}
/**
- * Checks whether the outer snapshot is compatible with a given new serializer.
+ * Checks the schema compatibility of the given old serializer snapshot based on the outer
+ * snapshot.
*
- * <p>The base implementation of this method just returns {@code true}, i.e. it assumes that the
- * outer serializer only has nested serializers and no extra information, and therefore the
- * result of the check must always be true. Otherwise, if the outer serializer contains some
+ * <p>The base implementation of this method assumes that the outer serializer only has nested
+ * serializers and no extra information, and therefore the result of the check is {@link
+ * OuterSchemaCompatibility#COMPATIBLE_AS_IS}. Otherwise, if the outer serializer contains some
* extra information that has been persisted as part of the serializer snapshot, this must be
* overridden. Note that this method and the corresponding methods {@link
* #writeOuterSnapshot(DataOutputView)}, {@link #readOuterSnapshot(int, DataInputView,
* ClassLoader)} needs to be implemented.
*
- * @param newSerializer the new serializer, which contains the new outer information to check
- * against.
- * @return a flag indicating whether or not the new serializer's outer information is compatible
- * with the one written in this snapshot.
- * @deprecated this method is deprecated, and will be removed in the future. Please implement
- * {@link #resolveOuterSchemaCompatibility(TypeSerializer)} instead.
+ * @param oldSerializerSnapshot the old serializer snapshot, which contains the old outer
+ * information to check against.
+ * @return a {@link OuterSchemaCompatibility} indicating whether or the new serializer's outer
+ * information is compatible, requires migration, or incompatible with the one written in
+ * this snapshot.
*/
- @Deprecated
- protected boolean isOuterSnapshotCompatible(S newSerializer) {
Review Comment:
When is this method called now? Could you please explain the flow how `CompositeTypeSerializerSnapshot` works? What is the flow for existing custom serializers that implement:
1. `protected boolean isOuterSnapshotCompatible(S newSerializer)`
2. `OuterSchemaCompatibility resolveOuterSchemaCompatibility(S newSerializer)`
I still think there is something off here.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085396914
##########
flink-core/src/test/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshotTest.java:
##########
@@ -232,14 +232,11 @@ public void testRestoreCompositeTypeSerializer() throws IOException {
// the restored nested serializer should be a RestoredNestedSerializer
testSerializer = (TestCompositeTypeSerializer) testSerializerSnapshot.restoreSerializer();
Assert.assertTrue(
- testSerializer.getNestedSerializers()[0].getClass()
Review Comment:
That is wrong, isn't it? We should've restored the nested serializers as well. Why haven't we?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1086625857
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
I don't think this will work. Take this for example:
```
@Test
public void testCompatibleWithNonMigratedSerializer() throws IOException {
TypeSerializer<?>[] testNestedSerializers = {
new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),
new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),
};
TypeSerializerSchemaCompatibility<String> compatibility =
snapshotCompositeSerializerAndGetSchemaCompatibilityAfterRestore(
testNestedSerializers,
testNestedSerializers,
OuterSchemaCompatibility.COMPATIBLE_AS_IS);
Assert.assertTrue(compatibility.isCompatibleAsIs());
}
```
where the `new NestedSerializer(TargetCompatibility.COMPATIBLE_AS_IS, NestedSerializer.SnapshotType.NOT_MIGRATED),` returns a snapshot that still implements `resolveSchemaCompatibility(TypeSerializer<String> newSerializer)` instead of the new method.
This should work, but as it stands it returns `incompatible` forcing user to implement the new method.
The reason is that once you enter the new path you never go back to the old one.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1086628860
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeSerializerSnapshot.java:
##########
@@ -290,26 +332,26 @@ protected void readOuterSnapshot(
throws IOException {}
/**
- * Checks whether the outer snapshot is compatible with a given new serializer.
+ * Checks the schema compatibility of the given old serializer snapshot based on the outer
+ * snapshot.
*
- * <p>The base implementation of this method just returns {@code true}, i.e. it assumes that the
- * outer serializer only has nested serializers and no extra information, and therefore the
- * result of the check must always be true. Otherwise, if the outer serializer contains some
+ * <p>The base implementation of this method assumes that the outer serializer only has nested
+ * serializers and no extra information, and therefore the result of the check is {@link
+ * OuterSchemaCompatibility#COMPATIBLE_AS_IS}. Otherwise, if the outer serializer contains some
* extra information that has been persisted as part of the serializer snapshot, this must be
* overridden. Note that this method and the corresponding methods {@link
* #writeOuterSnapshot(DataOutputView)}, {@link #readOuterSnapshot(int, DataInputView,
* ClassLoader)} needs to be implemented.
*
- * @param newSerializer the new serializer, which contains the new outer information to check
- * against.
- * @return a flag indicating whether or not the new serializer's outer information is compatible
- * with the one written in this snapshot.
- * @deprecated this method is deprecated, and will be removed in the future. Please implement
- * {@link #resolveOuterSchemaCompatibility(TypeSerializer)} instead.
+ * @param oldSerializerSnapshot the old serializer snapshot, which contains the old outer
+ * information to check against.
+ * @return a {@link OuterSchemaCompatibility} indicating whether or the new serializer's outer
+ * information is compatible, requires migration, or incompatible with the one written in
+ * this snapshot.
*/
- @Deprecated
- protected boolean isOuterSnapshotCompatible(S newSerializer) {
Review Comment:
Let's first deal with https://github.com/apache/flink/pull/21635#discussion_r1086625857 as I believe it is related.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #21635: [FLINK-30613] Improve resolving schema compatibility -- Milestone one
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1085553601
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome can be that the
+ * serialization format is compatible, that the program's serializer needs to reconfigure itself
+ * (meaning to incorporate some information from the TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed. In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data, and the restoring
+ * program's serializer re-serializes the data, thus converting the format during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
For users' existing TypeSerializer, old method must have been implemented.
So it could also works even if user don't modify their codes.
If users need to migrate, they could just implement the new method and remove the old method just as `custom_serialization.md` said.
It's implemented just as we talked in https://lists.apache.org/thread/4zgpd4cw232xcjp3cnj2hlokf6zvm3qc
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org