You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2020/05/12 12:19:07 UTC
[flink] branch master updated: [FLINK-17614][table] Add project it
case for partition table in filesystem connector
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6def38d [FLINK-17614][table] Add project it case for partition table in filesystem connector
6def38d is described below
commit 6def38d836323d4c8ce72296f27da57b65c530ee
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue May 12 20:18:51 2020 +0800
[FLINK-17614][table] Add project it case for partition table in filesystem connector
This closes #12091
---
.../formats/parquet/vector/ParquetSplitReaderUtil.java | 2 +-
.../flink/table/planner/runtime/FileSystemITCaseBase.scala | 14 ++++++++++++++
2 files changed, 15 insertions(+), 1 deletion(-)
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
index 8be1e2f..280cfab 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
@@ -112,7 +112,7 @@ public class ParquetSplitReaderUtil {
LogicalType type = fullFieldTypes[selectedFields[i]].getLogicalType();
vectors[i] = partitionSpec.containsKey(name) ?
createVectorFromConstant(type, partitionSpec.get(name), batchSize) :
- readVectors[i];
+ readVectors[selNonPartNames.indexOf(name)];
}
return new VectorizedColumnBatch(vectors);
};
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala
index 2c69b07..f303228 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/FileSystemITCaseBase.scala
@@ -226,6 +226,20 @@ trait FileSystemITCaseBase {
"select x, y from nonPartitionedTable where a=10086",
Seq())
}
+
+ @Test
+ def testProjectPushDown(): Unit = {
+ tableEnv.sqlUpdate("insert into partitionedTable select x, y, a, b from originalT")
+ tableEnv.execute("test")
+
+ check(
+ "select y, b, x from partitionedTable where a=3",
+ Seq(
+ row(17, 1, "x17"),
+ row(18, 2, "x18"),
+ row(19, 3, "x19")
+ ))
+ }
}
object FileSystemITCaseBase {