@Test
public void testApplyPredicate() throws Exception {

	OrcTableSource orc = OrcTableSource.builder()
		.path(getPath(TEST_FILE_NESTED))
		.forOrcSchema(TEST_SCHEMA_NESTED)
		.build();
	// expressions for supported predicates
	Expression pred1 = new GreaterThan(
		new ResolvedFieldReference("int1", Types.INT),
		new Literal(100, Types.INT));
	Expression pred2 = new EqualTo(
		new ResolvedFieldReference("string1", Types.STRING),
		new Literal("hello", Types.STRING));
	// unsupported predicate
	Expression unsupportedPred = new EqualTo(
		new GetCompositeField(
			new ArrayElementAt(
				new ResolvedFieldReference(
					"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 ResolvedFieldReference("long1", Types.LONG),
		// some invalid, non-serializable literal (here an object of this test class)
		new Literal(new OrcTableSourceTest(), Types.LONG)
	);
	ArrayList<Expression> preds = new ArrayList<>();
	preds.add(pred1);
	preds.add(pred2);
	preds.add(unsupportedPred);
	preds.add(invalidPred);
	// apply predicates on TableSource
	OrcTableSource projected = (OrcTableSource) orc.applyPredicate(preds);
	// ensure copy is returned
	assertTrue(orc != projected);
	// ensure table schema is identical
	assertEquals(orc.getTableSchema(), projected.getTableSchema());
	// ensure return type is identical
	assertEquals(
		Types.ROW_NAMED(getNestedFieldNames(), getNestedFieldTypes()),
		projected.getReturnType());
	// ensure IF is configured with valid/supported predicates
	OrcTableSource spyTS = spy(projected);
	OrcRowInputFormat mockIF = mock(OrcRowInputFormat.class);
	doReturn(mockIF).when(spyTS).buildOrcInputFormat();
	spyTS.getDataSet(mock(ExecutionEnvironment.class));
	ArgumentCaptor<OrcRowInputFormat.Predicate> arguments = ArgumentCaptor.forClass(OrcRowInputFormat.Predicate.class);
	verify(mockIF, times(2)).addPredicate(arguments.capture());
	List<String> values = arguments.getAllValues().stream().map(Object::toString).collect(Collectors.toList());
	assertTrue(values.contains(
		new OrcRowInputFormat.Not(new OrcRowInputFormat.LessThanEquals("int1", PredicateLeaf.Type.LONG, 100)).toString()));
	assertTrue(values.contains(
		new OrcRowInputFormat.Equals("string1", PredicateLeaf.Type.STRING, "hello").toString()));
	// ensure filter pushdown is correct
	assertTrue(spyTS.isFilterPushedDown());
	assertFalse(orc.isFilterPushedDown());
}