This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit b70989e0c0efe4ce3ce49286384d59dea3428c88 Author: Sergey Nuyanzin <[email protected]> AuthorDate: Sun Nov 1 15:26:20 2026 +0100 [hotfix][table] Treat missing predicates in MetadataFilterPushDownSpec as empty --- .../source/MetadataFilterPushDownSpec.java | 8 ++++++-- .../exec/serde/DynamicTableSourceSpecSerdeTest.java | 21 +++++++++++++++++++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/MetadataFilterPushDownSpec.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/MetadataFilterPushDownSpec.java index d1cb3303b80..0c738469d1a 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/MetadataFilterPushDownSpec.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/MetadataFilterPushDownSpec.java @@ -30,11 +30,14 @@ import org.apache.flink.table.types.logical.RowType; import org.apache.flink.shaded.guava33.com.google.common.collect.Sets; 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.JsonInclude; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName; import org.apache.calcite.rex.RexNode; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.List; import java.util.Objects; @@ -57,6 +60,7 @@ public final class MetadataFilterPushDownSpec extends SourceAbilitySpecBase { public static final String FIELD_NAME_PREDICATE_ROW_TYPE = "predicateRowType"; @JsonProperty(FIELD_NAME_PREDICATES) + @JsonInclude(JsonInclude.Include.NON_EMPTY) private final List<RexNode> predicates; /** @@ -68,9 +72,9 @@ public final class MetadataFilterPushDownSpec extends SourceAbilitySpecBase { @JsonCreator public MetadataFilterPushDownSpec( - @JsonProperty(FIELD_NAME_PREDICATES) List<RexNode> predicates, + @Nullable @JsonProperty(FIELD_NAME_PREDICATES) List<RexNode> predicates, @JsonProperty(FIELD_NAME_PREDICATE_ROW_TYPE) RowType predicateRowType) { - this.predicates = new ArrayList<>(checkNotNull(predicates)); + this.predicates = predicates == null ? List.of() : new ArrayList<>(predicates); this.predicateRowType = checkNotNull(predicateRowType); } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java index 478c91eedcd..086b96666c0 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java @@ -46,6 +46,7 @@ import org.apache.flink.table.planner.plan.abilities.source.MetadataFilterPushDo import org.apache.flink.table.planner.plan.abilities.source.PartitionPushDownSpec; import org.apache.flink.table.planner.plan.abilities.source.ProjectPushDownSpec; import org.apache.flink.table.planner.plan.abilities.source.ReadingMetadataSpec; +import org.apache.flink.table.planner.plan.abilities.source.SourceAbilitySpec; import org.apache.flink.table.planner.plan.abilities.source.SourceWatermarkSpec; import org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec; import org.apache.flink.table.planner.plan.nodes.exec.spec.DynamicTableSourceSpec; @@ -77,6 +78,7 @@ import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.stream.Stream; @@ -380,6 +382,25 @@ public class DynamicTableSourceSpecSerdeTest { .isNotNull(); } + @Test + void testMetadataFilterPushDownSpecWithEmptyPredicates() throws IOException { + SerdeContext serdeCtx = configuredSerdeContext(); + MetadataFilterPushDownSpec expected = + new MetadataFilterPushDownSpec( + List.of(), + RowType.of(new LogicalType[] {new BigIntType()}, new String[] {"offset"})); + + String json = toJson(serdeCtx, expected); + assertThat(json) + .doesNotContain("\"" + MetadataFilterPushDownSpec.FIELD_NAME_PREDICATES + "\""); + assertThat(toObject(serdeCtx, json, SourceAbilitySpec.class)).isEqualTo(expected); + + String jsonWithEmpty = + json.replaceFirst( + "\\{", "{\"" + MetadataFilterPushDownSpec.FIELD_NAME_PREDICATES + "\":[],"); + assertThat(toObject(serdeCtx, jsonWithEmpty, SourceAbilitySpec.class)).isEqualTo(expected); + } + @Test void testDynamicTableSourceSpecSerdeWithEnrichmentOptions() throws Exception { // Test model
