OrcTableSource implements BatchTableSource<Row>, ProjectableTableSource<Row>, FilterableTableSource<Row> { @Override public TableSource<Row> applyPredicate(List<Expression> predicates) { ArrayList<Predicate> orcPredicates = new ArrayList<>(); for (Expression pred : predicates) { Predicate orcPred = toOrcPredicate(pred); if (orcPred != null) { LOG.info("Predicate [{}] converted into OrcPredicate [{}] and pushed into OrcTableSource for path {}.", pred, orcPred, path); orcPredicates.add(orcPred); } else { LOG.info("Predicate [{}] could not be pushed into OrcTableSource for path {}.", pred, path); } } return new OrcTableSource(path, orcSchema, orcConfig, batchSize, recursiveEnumeration, selectedFields, orcPredicates.toArray(new Predicate[]{})); } private  OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig, int batchSize, boolean recursiveEnumeration); private  OrcTableSource(String path, TypeDescription orcSchema, Configuration orcConfig,
							int batchSize, boolean recursiveEnumeration,
							int[] selectedFields, Predicate[] predicates); @Override DataSet<Row> getDataSet(ExecutionEnvironment execEnv); @Override TypeInformation<Row> getReturnType(); @Override TableSchema getTableSchema(); @Override TableSource<Row> projectFields(int[] selectedFields); @Override TableSource<Row> applyPredicate(List<Expression> predicates); @Override boolean isFilterPushedDown(); @Override String explainSource(); static Builder builder();  }
@Test public void testApplyPredicate() throws Exception { OrcTableSource orc = OrcTableSource.builder() .path(getPath(TEST_FILE_NESTED)) .forOrcSchema(TEST_SCHEMA_NESTED) .build(); 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)); 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) ); Expression invalidPred = new EqualTo( new ResolvedFieldReference("long1", Types.LONG), new Literal(new OrcTableSourceTest(), Types.LONG) ); ArrayList<Expression> preds = new ArrayList<>(); preds.add(pred1); preds.add(pred2); preds.add(unsupportedPred); preds.add(invalidPred); OrcTableSource projected = (OrcTableSource) orc.applyPredicate(preds); assertTrue(orc != projected); assertEquals(orc.getTableSchema(), projected.getTableSchema()); assertEquals( Types.ROW_NAMED(getNestedFieldNames(), getNestedFieldTypes()), projected.getReturnType()); 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())); assertTrue(spyTS.isFilterPushedDown()); assertFalse(orc.isFilterPushedDown()); }