[FLINK-16860][orc] Distinguish empty filters from no pushed down in expainSource

This closes #11594

flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java

@@ -218,7 +218,9 @@ public class OrcTableSource
 	}
 
 	private String predicateString() {
-		if (predicates == null || predicates.length == 0) {
+		if (predicates == null) {
+			return "NULL";
+		} else if (predicates.length == 0) {
 			return "TRUE";
 		} else {
 			return "AND(" + Arrays.toString(predicates) + ")";

flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java

@@ -44,12 +44,14 @@ import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
@@ -184,18 +186,6 @@ public class OrcTableSourceTest {
 		Expression pred2 = new EqualTo(
 			new PlannerResolvedFieldReference("string1", Types.STRING),
 			new Literal("hello", Types.STRING));
-		// unsupported predicate
-		Expression unsupportedPred = new EqualTo(
-			new GetCompositeField(
-				new ItemAt(
-					new PlannerResolvedFieldReference(
-						"list",
-						ObjectArrayTypeInfo.getInfoFor(
-							Types.ROW_NAMED(new String[] {"int1", "string1"}, Types.INT, Types.STRING))),
-					new Literal(1, Types.INT)),
-				"int1"),
-			new Literal(1, Types.INT)
-			);
 		// invalid predicate
 		Expression invalidPred = new EqualTo(
 			new PlannerResolvedFieldReference("long1", Types.LONG),
@@ -206,7 +196,7 @@ public class OrcTableSourceTest {
 		ArrayList<Expression> preds = new ArrayList<>();
 		preds.add(pred1);
 		preds.add(pred2);
-		preds.add(unsupportedPred);
+		preds.add(unsupportedPred());
 		preds.add(invalidPred);
 
 		// apply predicates on TableSource
@@ -244,6 +234,34 @@ public class OrcTableSourceTest {
 		assertFalse(orc.isFilterPushedDown());
 	}
 
+	private Expression unsupportedPred() {
+		return new EqualTo(
+			new GetCompositeField(
+				new ItemAt(
+					new PlannerResolvedFieldReference(
+						"list",
+						ObjectArrayTypeInfo.getInfoFor(
+							Types.ROW_NAMED(new String[] {"int1", "string1"}, Types.INT, Types.STRING))),
+					new Literal(1, Types.INT)),
+				"int1"),
+			new Literal(1, Types.INT)
+		);
+	}
+
+	@Test
+	public void testUnsupportedPredOnly() {
+		OrcTableSource orc = OrcTableSource.builder()
+				.path(getPath(TEST_FILE_NESTED))
+				.forOrcSchema(TEST_SCHEMA_NESTED)
+				.build();
+
+		// apply predicates on TableSource
+		OrcTableSource projected = (OrcTableSource) orc.applyPredicate(
+				Collections.singletonList(unsupportedPred()));
+
+		assertNotEquals(orc.explainSource(), projected.explainSource());
+	}
+
 	@Test
 	public void testBuilder() throws Exception {