[FLINK-21549][table-planner-blink] FileSystemTableSource should retur…

…n the accepted predicates when filter push-down

This close apache#15062

flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/filesystem/FileSystemTableSourceTest.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.filesystem;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.planner.utils.StreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/** Test for {@link FileSystemTableSource}. */
+public class FileSystemTableSourceTest extends TableTestBase {
+
+ private StreamTableTestUtil util;
+
+ @Before
+ public void setup() {
+ util = streamTestUtil(TableConfig.getDefault());
+ TableEnvironment tEnv = util.getTableEnv();
+
+ String srcTableDdl =
+ "CREATE TABLE MyTable (\n"
+ + " a bigint,\n"
+ + " b int,\n"
+ + " c varchar\n"
+ + ") with (\n"
+ + " 'connector' = 'filesystem',"
+ + " 'format' = 'testcsv',"
+ + " 'path' = '/tmp')";
+ tEnv.executeSql(srcTableDdl);
+
+ String sinkTableDdl =
+ "CREATE TABLE MySink (\n"
+ + " a bigint,\n"
+ + " b int,\n"
+ + " c varchar\n"
+ + ") with (\n"
+ + " 'connector' = 'values',\n"
+ + " 'table-sink-class' = 'DEFAULT')";
+ tEnv.executeSql(sinkTableDdl);
+ }
+
+ @Test
+ public void testFilterPushDown() {
+ util.verifyRelPlanInsert("insert into MySink select * from MyTable where a > 10");
+ }
+}

flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/filesystem/FileSystemTableSourceTest.xml

@@ -0,0 +1,39 @@
+<?xml version="1.0" ?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to you under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<Root>
+ <TestCase name="testFilterPushDown">
+ <Resource name="sql">
+ <![CDATA[insert into MySink select * from MyTable where a > 10]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.MySink], fields=[a, b, c])
++- LogicalProject(a=[$0], b=[$1], c=[$2])
+ +- LogicalFilter(condition=[>($0, 10)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Sink(table=[default_catalog.default_database.MySink], fields=[a, b, c])
++- Calc(select=[a, b, c], where=[>(a, 10)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, filter=[greaterThan(a, 10)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+</Root>

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableSource.java

@@ -219,8 +219,8 @@ public ChangelogMode getChangelogMode() {
 
  @Override
  public Result applyFilters(List<ResolvedExpression> filters) {
- this.filters = filters;
- return Result.of(Collections.emptyList(), filters);
+ this.filters = new ArrayList<>(filters);
+ return Result.of(new ArrayList<>(filters), new ArrayList<>(filters));
  }
 
  @Override