[FLINK-38352][table-planner] Make source specs forward compatible (like sink specs)

Author: twalthr

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/AggregatePushDownSpec.java

@@ -35,6 +35,7 @@
 import org.apache.flink.table.types.utils.TypeConversions;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -56,6 +57,7 @@
  * A sub-class of {@link SourceAbilitySpec} that can not only serialize/deserialize the aggregation
  * to/from JSON, but also can push the local aggregate into a {@link SupportsAggregatePushDown}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("AggregatePushDown")
 public final class AggregatePushDownSpec extends SourceAbilitySpecBase {
 

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/FilterPushDownSpec.java

@@ -31,6 +31,7 @@
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -51,6 +52,7 @@
  * A sub-class of {@link SourceAbilitySpec} that can not only serialize/deserialize the filter
  * to/from JSON, but also can push the filter into a {@link SupportsFilterPushDown}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("FilterPushDown")
 public final class FilterPushDownSpec extends SourceAbilitySpecBase {
     public static final String FIELD_NAME_PREDICATES = "predicates";

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/LimitPushDownSpec.java

@@ -23,6 +23,7 @@
 import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -32,6 +33,7 @@
  * A sub-class of {@link SourceAbilitySpec} that can not only serialize/deserialize the limit value
  * to/from JSON, but also can push the limit value into a {@link LimitPushDownSpec}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("LimitPushDown")
 public final class LimitPushDownSpec extends SourceAbilitySpecBase {
     public static final String FIELD_NAME_LIMIT = "limit";

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/PartitionPushDownSpec.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.planner.utils.PartitionUtils;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -38,6 +39,7 @@
  * A sub-class of {@link SourceAbilitySpec} that can not only serialize/deserialize the partitions
  * to/from JSON, but also can push the partitions into a {@link SupportsPartitionPushDown}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("PartitionPushDown")
 public final class PartitionPushDownSpec extends SourceAbilitySpecBase {
     public static final String FIELD_NAME_PARTITIONS = "partitions";

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/ProjectPushDownSpec.java

@@ -25,6 +25,7 @@
 import org.apache.flink.table.types.logical.RowType;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -37,6 +38,7 @@
  * A sub-class of {@link SourceAbilitySpec} that can not only serialize/deserialize the projection
  * to/from JSON, but also can push the projection into a {@link SupportsProjectionPushDown}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("ProjectPushDown")
 public final class ProjectPushDownSpec extends SourceAbilitySpecBase {
     public static final String FIELD_NAME_PROJECTED_FIELDS = "projectedFields";

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/ReadingMetadataSpec.java

@@ -26,6 +26,7 @@
 import org.apache.flink.table.types.utils.TypeConversions;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -41,6 +42,7 @@
  * columns to/from JSON, but also can read the metadata columns from {@link
  * SupportsReadingMetadata}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("ReadingMetadata")
 public final class ReadingMetadataSpec extends SourceAbilitySpecBase {
     public static final String FIELD_NAME_METADATA_KEYS = "metadataKeys";

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/SourceWatermarkSpec.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.types.logical.RowType;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -34,6 +35,7 @@
  * whether a source watermark should be used to/from JSON, but can also apply it to {@link
  * SupportsSourceWatermark}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("SourceWatermark")
 public final class SourceWatermarkSpec extends SourceAbilitySpecBase {
     public static final String FIELD_NAME_SOURCE_WATERMARK_ENABLED = "sourceWatermarkEnabled";

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/WatermarkPushDownSpec.java

@@ -34,6 +34,7 @@
 import org.apache.flink.table.watermark.WatermarkParams;
 
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
 
@@ -53,6 +54,7 @@
  * A sub-class of {@link SourceAbilitySpec} that can not only serialize/deserialize the watermark
  * to/from JSON, but also can push the watermark into a {@link SupportsWatermarkPushDown}.
  */
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeName("WatermarkPushDown")
 public final class WatermarkPushDownSpec extends SourceAbilitySpecBase {
     public static final String FIELD_NAME_WATERMARK_EXPR = "watermarkExpr";