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-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 9844ed49 [FLINK-28289] Introduce Spark2 Reader for table store
9844ed49 is described below
commit 9844ed49085c6729e34403050d4e857ea0b6a9aa
Author: Nicholas Jiang <[email protected]>
AuthorDate: Thu Jul 14 15:17:33 2022 +0800
[FLINK-28289] Introduce Spark2 Reader for table store
This closes #213
---
docs/content/docs/engines/build.md | 4 +
docs/content/docs/engines/spark2.md | 68 ++++++
docs/content/docs/engines/trino.md | 2 +-
flink-table-store-spark2/pom.xml | 110 ++++++++++
.../flink/table/store/spark/SparkArrayData.java | 171 +++++++++++++++
.../table/store/spark/SparkDataSourceReader.java | 139 ++++++++++++
.../table/store/spark/SparkFilterConverter.java | 128 +++++++++++
.../table/store/spark/SparkInputPartition.java | 127 +++++++++++
.../flink/table/store/spark/SparkInternalRow.java | 240 +++++++++++++++++++++
.../flink/table/store/spark/SparkSource.java | 50 +++++
.../flink/table/store/spark/SparkTypeUtils.java | 185 ++++++++++++++++
.../store/spark/SpecializedGettersReader.java | 106 +++++++++
...org.apache.spark.sql.sources.DataSourceRegister | 16 ++
.../table/store/spark/SimpleTableTestHelper.java | 70 ++++++
.../store/spark/SparkFilterConverterTest.java | 176 +++++++++++++++
.../table/store/spark/SparkInternalRowTest.java | 109 ++++++++++
.../flink/table/store/spark/SparkReadITCase.java | 130 +++++++++++
.../flink/table/store/spark/SparkTypeTest.java | 99 +++++++++
.../src/test/resources/log4j2-test.properties | 38 ++++
pom.xml | 1 +
20 files changed, 1968 insertions(+), 1 deletion(-)
diff --git a/docs/content/docs/engines/build.md
b/docs/content/docs/engines/build.md
index 107020c6..b221f07b 100644
--- a/docs/content/docs/engines/build.md
+++ b/docs/content/docs/engines/build.md
@@ -50,6 +50,10 @@ You can find Hive bundled jar in
`./flink-table-store-hive/flink-table-store-hiv
You can find Spark bundled jar in
`./flink-table-store-spark/target/flink-table-store-spark-{{< version >}}.jar`.
+## Spark2
+
+You can find Spark2 bundled jar in
`./flink-table-store-spark2/target/flink-table-store-spark2-{{< version
>}}.jar`.
+
## Flink 1.14
Running:
diff --git a/docs/content/docs/engines/spark2.md
b/docs/content/docs/engines/spark2.md
new file mode 100644
index 00000000..c8976b99
--- /dev/null
+++ b/docs/content/docs/engines/spark2.md
@@ -0,0 +1,68 @@
+---
+title: "Spark2"
+weight: 4
+type: docs
+aliases:
+- /engines/spark2.html
+---
+<!--
+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
+
+ http://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.
+-->
+
+# Spark2
+
+Table Store supports reading table store tables through Spark.
+
+## Version
+
+Table Store supports Spark 2.4+. It is highly recommended to use Spark 2.4+
version with many improvements.
+
+## Install
+
+{{< stable >}}
+Download [flink-table-store-spark2-{{< version
>}}.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-table-store-spark2/{{<
version >}}/flink-table-store-spark2-{{< version >}}.jar).
+{{< /stable >}}
+{{< unstable >}}
+You are using an unreleased version of Table Store, you need to manually
[Build Spark Bundled Jar]({{< ref "docs/engines/build" >}}) from the source
code.
+{{< /unstable >}}
+
+Use `--jars` in spark-sql:
+```bash
+spark-sql ... --jars flink-table-store-spark2-{{< version >}}.jar
+```
+
+Alternatively, you can copy `flink-table-store-spark2-{{< version >}}.jar`
under `spark/jars` in your Spark installation.
+
+## Create Temporary View
+
+Use the `CREATE TEMPORARY VIEW` command to create a Spark mapping table on top
of
+an existing Table Store table.
+
+```sql
+CREATE TEMPORARY VIEW myTable
+USING tablestore
+OPTIONS (
+ path "file:/tmp/warehouse/default.db/myTable"
+)
+```
+
+## Query Table
+
+```sql
+SELECT * FROM table_store.default.myTable;
+```
diff --git a/docs/content/docs/engines/trino.md
b/docs/content/docs/engines/trino.md
index c2c308b9..50bae94a 100644
--- a/docs/content/docs/engines/trino.md
+++ b/docs/content/docs/engines/trino.md
@@ -1,6 +1,6 @@
---
title: "Trino"
-weight: 4
+weight: 5
type: docs
aliases:
- /engines/trino.html
diff --git a/flink-table-store-spark2/pom.xml b/flink-table-store-spark2/pom.xml
new file mode 100644
index 00000000..4ee55da0
--- /dev/null
+++ b/flink-table-store-spark2/pom.xml
@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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
+
+ http://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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>flink-table-store-parent</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>0.2-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-table-store-spark2</artifactId>
+ <name>Flink Table Store : Spark2</name>
+
+ <packaging>jar</packaging>
+
+ <properties>
+ <spark2.version>2.4.8</spark2.version>
+ <jackson.version>2.13.3</jackson.version>
+ </properties>
+
+ <dependencies>
+ <!-- Flink All dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-store-shade</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_2.11</artifactId>
+ <version>${spark2.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ </exclusion>
+ <!-- The Jackson dependencies of Spark 2.4.8 have
vulnerabilities. -->
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>jackson-module-scala_2.11</artifactId>
+ <version>${jackson.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <includes combine.children="append">
+
<include>org.apache.flink:flink-table-store-shade</include>
+ </includes>
+ </artifactSet>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkArrayData.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkArrayData.java
new file mode 100644
index 00000000..969691b2
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkArrayData.java
@@ -0,0 +1,171 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.store.utils.RowDataUtils;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import static org.apache.flink.table.store.spark.SparkInternalRow.fromFlink;
+import static org.apache.flink.table.store.utils.RowDataUtils.copyArray;
+import static org.apache.flink.table.store.utils.TypeUtils.timestampPrecision;
+
+/** Spark {@link ArrayData} to wrap Flink {@code ArrayData}. */
+public class SparkArrayData extends ArrayData {
+
+ private final LogicalType elementType;
+
+ private org.apache.flink.table.data.ArrayData array;
+
+ public SparkArrayData(LogicalType elementType) {
+ this.elementType = elementType;
+ }
+
+ public SparkArrayData replace(org.apache.flink.table.data.ArrayData array)
{
+ this.array = array;
+ return this;
+ }
+
+ @Override
+ public int numElements() {
+ return array.size();
+ }
+
+ @Override
+ public ArrayData copy() {
+ return new SparkArrayData(elementType).replace(copyArray(array,
elementType));
+ }
+
+ @Override
+ public Object[] array() {
+ Object[] objects = new Object[numElements()];
+ for (int i = 0; i < objects.length; i++) {
+ objects[i] = fromFlink(RowDataUtils.get(array, i, elementType),
elementType);
+ }
+ return objects;
+ }
+
+ @Override
+ public void setNullAt(int i) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void update(int i, Object value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isNullAt(int ordinal) {
+ return array.isNullAt(ordinal);
+ }
+
+ @Override
+ public boolean getBoolean(int ordinal) {
+ return array.getBoolean(ordinal);
+ }
+
+ @Override
+ public byte getByte(int ordinal) {
+ return array.getByte(ordinal);
+ }
+
+ @Override
+ public short getShort(int ordinal) {
+ return array.getShort(ordinal);
+ }
+
+ @Override
+ public int getInt(int ordinal) {
+ return array.getInt(ordinal);
+ }
+
+ @Override
+ public long getLong(int ordinal) {
+ if (elementType instanceof BigIntType) {
+ return array.getLong(ordinal);
+ }
+
+ return getTimestampMicros(ordinal);
+ }
+
+ private long getTimestampMicros(int ordinal) {
+ return fromFlink(array.getTimestamp(ordinal,
timestampPrecision(elementType)));
+ }
+
+ @Override
+ public float getFloat(int ordinal) {
+ return array.getFloat(ordinal);
+ }
+
+ @Override
+ public double getDouble(int ordinal) {
+ return array.getDouble(ordinal);
+ }
+
+ @Override
+ public Decimal getDecimal(int ordinal, int precision, int scale) {
+ return fromFlink(array.getDecimal(ordinal, precision, scale));
+ }
+
+ @Override
+ public UTF8String getUTF8String(int ordinal) {
+ return fromFlink(array.getString(ordinal));
+ }
+
+ @Override
+ public byte[] getBinary(int ordinal) {
+ return array.getBinary(ordinal);
+ }
+
+ @Override
+ public CalendarInterval getInterval(int ordinal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InternalRow getStruct(int ordinal, int numFields) {
+ return fromFlink(array.getRow(ordinal, numFields), (RowType)
elementType);
+ }
+
+ @Override
+ public ArrayData getArray(int ordinal) {
+ return fromFlink(array.getArray(ordinal), (ArrayType) elementType);
+ }
+
+ @Override
+ public MapData getMap(int ordinal) {
+ return fromFlink(array.getMap(ordinal), elementType);
+ }
+
+ @Override
+ public Object get(int ordinal, DataType dataType) {
+ return SpecializedGettersReader.read(this, ordinal, dataType);
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkDataSourceReader.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkDataSourceReader.java
new file mode 100644
index 00000000..900e0405
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkDataSourceReader.java
@@ -0,0 +1,139 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.store.file.data.DataFileMeta;
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.table.FileStoreTable;
+import org.apache.flink.table.store.table.source.Split;
+import org.apache.flink.table.store.utils.TypeUtils;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
+import org.apache.spark.sql.sources.v2.reader.InputPartition;
+import org.apache.spark.sql.sources.v2.reader.Statistics;
+import org.apache.spark.sql.sources.v2.reader.SupportsPushDownFilters;
+import org.apache.spark.sql.sources.v2.reader.SupportsPushDownRequiredColumns;
+import org.apache.spark.sql.sources.v2.reader.SupportsReportStatistics;
+import org.apache.spark.sql.types.StructType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.OptionalLong;
+import java.util.stream.Collectors;
+
+/** A Spark {@link DataSourceReader} for table store. */
+public class SparkDataSourceReader
+ implements SupportsPushDownFilters,
+ SupportsPushDownRequiredColumns,
+ SupportsReportStatistics {
+
+ private final FileStoreTable table;
+
+ private List<Predicate> predicates = new ArrayList<>();
+ private Filter[] pushedFilters;
+ private int[] projectedFields;
+ private List<Split> splits;
+
+ public SparkDataSourceReader(FileStoreTable table) {
+ this.table = table;
+ }
+
+ @Override
+ public Filter[] pushFilters(Filter[] filters) {
+ SparkFilterConverter converter = new
SparkFilterConverter(table.schema().logicalRowType());
+ List<Predicate> predicates = new ArrayList<>();
+ List<Filter> pushed = new ArrayList<>();
+ for (Filter filter : filters) {
+ try {
+ predicates.add(converter.convert(filter));
+ pushed.add(filter);
+ } catch (UnsupportedOperationException ignore) {
+ }
+ }
+ this.predicates = predicates;
+ this.pushedFilters = pushed.toArray(new Filter[0]);
+ return filters;
+ }
+
+ @Override
+ public Filter[] pushedFilters() {
+ return pushedFilters;
+ }
+
+ @Override
+ public void pruneColumns(StructType requiredSchema) {
+ String[] pruneFields = requiredSchema.fieldNames();
+ List<String> fieldNames = table.schema().fieldNames();
+ int[] projected = new int[pruneFields.length];
+ for (int i = 0; i < projected.length; i++) {
+ projected[i] = fieldNames.indexOf(pruneFields[i]);
+ }
+ this.projectedFields = projected;
+ }
+
+ @Override
+ public Statistics estimateStatistics() {
+ long rowCount = 0L;
+
+ for (Split split : splits()) {
+ for (DataFileMeta file : split.files()) {
+ rowCount += file.rowCount();
+ }
+ }
+
+ final long numRows = rowCount;
+ final long sizeInBytes = readSchema().defaultSize() * numRows;
+
+ return new Statistics() {
+ @Override
+ public OptionalLong sizeInBytes() {
+ return OptionalLong.of(sizeInBytes);
+ }
+
+ @Override
+ public OptionalLong numRows() {
+ return OptionalLong.of(numRows);
+ }
+ };
+ }
+
+ @Override
+ public StructType readSchema() {
+ RowType rowType = table.schema().logicalRowType();
+ return SparkTypeUtils.fromFlinkRowType(
+ projectedFields == null ? rowType : TypeUtils.project(rowType,
projectedFields));
+ }
+
+ @Override
+ public List<InputPartition<InternalRow>> planInputPartitions() {
+ return splits().stream()
+ .map(split -> new SparkInputPartition(table, projectedFields,
predicates, split))
+ .collect(Collectors.toList());
+ }
+
+ protected List<Split> splits() {
+ if (splits == null) {
+ this.splits = table.newScan().withFilter(predicates).plan().splits;
+ }
+ return splits;
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkFilterConverter.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkFilterConverter.java
new file mode 100644
index 00000000..285911bb
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkFilterConverter.java
@@ -0,0 +1,128 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.predicate.PredicateBuilder;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.spark.sql.sources.And;
+import org.apache.spark.sql.sources.EqualTo;
+import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.sources.GreaterThan;
+import org.apache.spark.sql.sources.GreaterThanOrEqual;
+import org.apache.spark.sql.sources.In;
+import org.apache.spark.sql.sources.IsNotNull;
+import org.apache.spark.sql.sources.IsNull;
+import org.apache.spark.sql.sources.LessThan;
+import org.apache.spark.sql.sources.LessThanOrEqual;
+import org.apache.spark.sql.sources.Not;
+import org.apache.spark.sql.sources.Or;
+import org.apache.spark.sql.sources.StringStartsWith;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+import static
org.apache.flink.table.store.file.predicate.PredicateBuilder.convertJavaObject;
+
+/** Conversion from {@link Filter} to {@link Predicate}. */
+public class SparkFilterConverter {
+
+ private final RowType rowType;
+ private final PredicateBuilder builder;
+
+ public SparkFilterConverter(RowType rowType) {
+ this.rowType = rowType;
+ this.builder = new PredicateBuilder(rowType);
+ }
+
+ public Predicate convert(Filter filter) {
+ if (filter instanceof EqualTo) {
+ EqualTo eq = (EqualTo) filter;
+ // TODO deal with isNaN
+ int index = fieldIndex(eq.attribute());
+ Object literal = convertLiteral(index, eq.value());
+ return builder.equal(index, literal);
+ } else if (filter instanceof GreaterThan) {
+ GreaterThan gt = (GreaterThan) filter;
+ int index = fieldIndex(gt.attribute());
+ Object literal = convertLiteral(index, gt.value());
+ return builder.greaterThan(index, literal);
+ } else if (filter instanceof GreaterThanOrEqual) {
+ GreaterThanOrEqual gt = (GreaterThanOrEqual) filter;
+ int index = fieldIndex(gt.attribute());
+ Object literal = convertLiteral(index, gt.value());
+ return builder.greaterOrEqual(index, literal);
+ } else if (filter instanceof LessThan) {
+ LessThan lt = (LessThan) filter;
+ int index = fieldIndex(lt.attribute());
+ Object literal = convertLiteral(index, lt.value());
+ return builder.lessThan(index, literal);
+ } else if (filter instanceof LessThanOrEqual) {
+ LessThanOrEqual lt = (LessThanOrEqual) filter;
+ int index = fieldIndex(lt.attribute());
+ Object literal = convertLiteral(index, lt.value());
+ return builder.lessOrEqual(index, literal);
+ } else if (filter instanceof In) {
+ In in = (In) filter;
+ int index = fieldIndex(in.attribute());
+ return builder.in(
+ index,
+ Arrays.stream(in.values())
+ .map(v -> convertLiteral(index, v))
+ .collect(Collectors.toList()));
+ } else if (filter instanceof IsNull) {
+ return builder.isNull(fieldIndex(((IsNull) filter).attribute()));
+ } else if (filter instanceof IsNotNull) {
+ return builder.isNotNull(fieldIndex(((IsNotNull)
filter).attribute()));
+ } else if (filter instanceof And) {
+ And and = (And) filter;
+ return PredicateBuilder.and(convert(and.left()),
convert(and.right()));
+ } else if (filter instanceof Or) {
+ Or or = (Or) filter;
+ return PredicateBuilder.or(convert(or.left()),
convert(or.right()));
+ } else if (filter instanceof Not) {
+ Not not = (Not) filter;
+ return
convert(not.child()).negate().orElseThrow(UnsupportedOperationException::new);
+ } else if (filter instanceof StringStartsWith) {
+ StringStartsWith startsWith = (StringStartsWith) filter;
+ int index = fieldIndex(startsWith.attribute());
+ Object literal = convertLiteral(index, startsWith.value());
+ return builder.startsWith(index, literal);
+ }
+
+ // TODO: In, NotIn, AlwaysTrue, AlwaysFalse, EqualNullSafe
+ throw new UnsupportedOperationException();
+ }
+
+ private int fieldIndex(String field) {
+ int index = rowType.getFieldIndex(field);
+ // TODO: support nested field
+ if (index == -1) {
+ throw new UnsupportedOperationException();
+ }
+ return index;
+ }
+
+ private Object convertLiteral(int index, Object value) {
+ LogicalType type = rowType.getTypeAt(index);
+ return convertJavaObject(type, value);
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkInputPartition.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkInputPartition.java
new file mode 100644
index 00000000..f4c5432a
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkInputPartition.java
@@ -0,0 +1,127 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.utils.RecordReader;
+import org.apache.flink.table.store.file.utils.RecordReaderIterator;
+import org.apache.flink.table.store.table.FileStoreTable;
+import org.apache.flink.table.store.table.source.Split;
+import org.apache.flink.table.store.table.source.TableRead;
+import org.apache.flink.table.store.utils.TypeUtils;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.sources.v2.reader.InputPartition;
+import org.apache.spark.sql.sources.v2.reader.InputPartitionReader;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.UncheckedIOException;
+import java.util.List;
+
+import static org.apache.flink.table.store.file.predicate.PredicateBuilder.and;
+
+/** A Spark {@link InputPartition} for table store. */
+public class SparkInputPartition implements InputPartition<InternalRow> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FileStoreTable table;
+ private final int[] projectedFields;
+ private final List<Predicate> predicates;
+
+ private transient Split split;
+
+ public SparkInputPartition(
+ FileStoreTable table, int[] projectedFields, List<Predicate>
predicates, Split split) {
+ this.table = table;
+ this.projectedFields = projectedFields;
+ this.predicates = predicates;
+ this.split = split;
+ }
+
+ @Override
+ public InputPartitionReader<InternalRow> createPartitionReader() {
+ RecordReader<RowData> recordReader;
+ try {
+ TableRead tableRead = table.newRead();
+ if (projectedFields != null) {
+ tableRead.withProjection(projectedFields);
+ }
+ if (predicates.size() > 0) {
+ tableRead.withFilter(and(predicates));
+ }
+ recordReader = tableRead.createReader(split);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ RecordReaderIterator<RowData> iterator = new
RecordReaderIterator<>(recordReader);
+ SparkInternalRow row = new SparkInternalRow(readRowType());
+ return new InputPartitionReader<InternalRow>() {
+
+ @Override
+ public boolean next() {
+ if (iterator.hasNext()) {
+ row.replace(iterator.next());
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public InternalRow get() {
+ return row;
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ iterator.close();
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+ };
+ }
+
+ @Override
+ public String[] preferredLocations() {
+ return new String[0];
+ }
+
+ private RowType readRowType() {
+ RowType rowType = table.schema().logicalRowType();
+ return projectedFields == null ? rowType : TypeUtils.project(rowType,
projectedFields);
+ }
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.defaultWriteObject();
+ split.serialize(new DataOutputViewStreamWrapper(out));
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
+ in.defaultReadObject();
+ split = Split.deserialize(new DataInputViewStreamWrapper(in));
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkInternalRow.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkInternalRow.java
new file mode 100644
index 00000000..c48e921f
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkInternalRow.java
@@ -0,0 +1,240 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.DateTimeUtils;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import static org.apache.flink.table.store.utils.RowDataUtils.copyRowData;
+import static org.apache.flink.table.store.utils.TypeUtils.timestampPrecision;
+
+/** Spark {@link InternalRow} to wrap {@link RowData}. */
+public class SparkInternalRow extends InternalRow {
+
+ private final RowType rowType;
+
+ private RowData row;
+
+ public SparkInternalRow(RowType rowType) {
+ this.rowType = rowType;
+ }
+
+ public SparkInternalRow replace(RowData row) {
+ this.row = row;
+ return this;
+ }
+
+ @Override
+ public int numFields() {
+ return row.getArity();
+ }
+
+ @Override
+ public void setNullAt(int i) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void update(int i, Object value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InternalRow copy() {
+ return new SparkInternalRow(rowType).replace(copyRowData(row,
rowType));
+ }
+
+ @Override
+ public boolean isNullAt(int ordinal) {
+ return row.isNullAt(ordinal);
+ }
+
+ @Override
+ public boolean getBoolean(int ordinal) {
+ return row.getBoolean(ordinal);
+ }
+
+ @Override
+ public byte getByte(int ordinal) {
+ return row.getByte(ordinal);
+ }
+
+ @Override
+ public short getShort(int ordinal) {
+ return row.getShort(ordinal);
+ }
+
+ @Override
+ public int getInt(int ordinal) {
+ return row.getInt(ordinal);
+ }
+
+ @Override
+ public long getLong(int ordinal) {
+ if (rowType.getTypeAt(ordinal) instanceof BigIntType) {
+ return row.getLong(ordinal);
+ }
+
+ return getTimestampMicros(ordinal);
+ }
+
+ private long getTimestampMicros(int ordinal) {
+ LogicalType type = rowType.getTypeAt(ordinal);
+ return fromFlink(row.getTimestamp(ordinal, timestampPrecision(type)));
+ }
+
+ @Override
+ public float getFloat(int ordinal) {
+ return row.getFloat(ordinal);
+ }
+
+ @Override
+ public double getDouble(int ordinal) {
+ return row.getDouble(ordinal);
+ }
+
+ @Override
+ public Decimal getDecimal(int ordinal, int precision, int scale) {
+ DecimalData decimal = row.getDecimal(ordinal, precision, scale);
+ return fromFlink(decimal);
+ }
+
+ @Override
+ public UTF8String getUTF8String(int ordinal) {
+ return fromFlink(row.getString(ordinal));
+ }
+
+ @Override
+ public byte[] getBinary(int ordinal) {
+ return row.getBinary(ordinal);
+ }
+
+ @Override
+ public CalendarInterval getInterval(int ordinal) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InternalRow getStruct(int ordinal, int numFields) {
+ return fromFlink(row.getRow(ordinal, numFields), (RowType)
rowType.getTypeAt(ordinal));
+ }
+
+ @Override
+ public ArrayData getArray(int ordinal) {
+ return fromFlink(row.getArray(ordinal), (ArrayType)
rowType.getTypeAt(ordinal));
+ }
+
+ @Override
+ public MapData getMap(int ordinal) {
+ return fromFlink(row.getMap(ordinal), rowType.getTypeAt(ordinal));
+ }
+
+ @Override
+ public Object get(int ordinal, DataType dataType) {
+ return SpecializedGettersReader.read(this, ordinal, dataType);
+ }
+
+ public static Object fromFlink(Object o, LogicalType type) {
+ if (o == null) {
+ return null;
+ }
+ switch (type.getTypeRoot()) {
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return fromFlink((TimestampData) o);
+ case CHAR:
+ case VARCHAR:
+ return fromFlink((StringData) o);
+ case DECIMAL:
+ return fromFlink((DecimalData) o);
+ case ARRAY:
+ return fromFlink((org.apache.flink.table.data.ArrayData) o,
(ArrayType) type);
+ case MAP:
+ case MULTISET:
+ return fromFlink((org.apache.flink.table.data.MapData) o,
type);
+ case ROW:
+ return fromFlink((RowData) o, (RowType) type);
+ default:
+ return o;
+ }
+ }
+
+ public static UTF8String fromFlink(StringData string) {
+ return UTF8String.fromBytes(string.toBytes());
+ }
+
+ public static Decimal fromFlink(DecimalData decimal) {
+ return Decimal.apply(decimal.toBigDecimal());
+ }
+
+ public static InternalRow fromFlink(RowData row, RowType rowType) {
+ return new SparkInternalRow(rowType).replace(row);
+ }
+
+ public static long fromFlink(TimestampData timestamp) {
+ return DateTimeUtils.fromJavaTimestamp(timestamp.toTimestamp());
+ }
+
+ public static ArrayData fromFlink(
+ org.apache.flink.table.data.ArrayData array, ArrayType arrayType) {
+ return fromFlinkArrayElementType(array, arrayType.getElementType());
+ }
+
+ private static ArrayData fromFlinkArrayElementType(
+ org.apache.flink.table.data.ArrayData array, LogicalType
elementType) {
+ return new SparkArrayData(elementType).replace(array);
+ }
+
+ public static MapData fromFlink(org.apache.flink.table.data.MapData map,
LogicalType mapType) {
+ LogicalType keyType;
+ LogicalType valueType;
+ if (mapType instanceof MapType) {
+ keyType = ((MapType) mapType).getKeyType();
+ valueType = ((MapType) mapType).getValueType();
+ } else if (mapType instanceof MultisetType) {
+ keyType = ((MultisetType) mapType).getElementType();
+ valueType = new IntType();
+ } else {
+ throw new UnsupportedOperationException("Unsupported type: " +
mapType);
+ }
+
+ return new ArrayBasedMapData(
+ fromFlinkArrayElementType(map.keyArray(), keyType),
+ fromFlinkArrayElementType(map.valueArray(), valueType));
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkSource.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkSource.java
new file mode 100644
index 00000000..bd2fde5b
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkSource.java
@@ -0,0 +1,50 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.store.table.FileStoreTable;
+import org.apache.flink.table.store.table.FileStoreTableFactory;
+
+import org.apache.spark.sql.sources.DataSourceRegister;
+import org.apache.spark.sql.sources.v2.DataSourceOptions;
+import org.apache.spark.sql.sources.v2.ReadSupport;
+import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
+import org.apache.spark.sql.types.StructType;
+
+/** The Spark source for table store. */
+public class SparkSource implements DataSourceRegister, ReadSupport {
+
+ @Override
+ public String shortName() {
+ // Not use 'table-store' here, the '-' is not allowed in SQL
+ return "tablestore";
+ }
+
+ @Override
+ public DataSourceReader createReader(StructType schema, DataSourceOptions
options) {
+ return createReader(options);
+ }
+
+ @Override
+ public DataSourceReader createReader(DataSourceOptions options) {
+ FileStoreTable table =
FileStoreTableFactory.create(Configuration.fromMap(options.asMap()));
+ return new SparkDataSourceReader(table);
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkTypeUtils.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkTypeUtils.java
new file mode 100644
index 00000000..e60fa5a3
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SparkTypeUtils.java
@@ -0,0 +1,185 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BinaryType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LocalZonedTimestampType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.RowType.RowField;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarBinaryType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** Utils for Spark {@link DataType}. */
+public class SparkTypeUtils {
+
+ private SparkTypeUtils() {}
+
+ public static StructType fromFlinkRowType(RowType type) {
+ return (StructType) fromFlinkType(type);
+ }
+
+ public static DataType fromFlinkType(LogicalType type) {
+ return type.accept(FlinkToSparkTypeVisitor.INSTANCE);
+ }
+
+ private static class FlinkToSparkTypeVisitor extends
LogicalTypeDefaultVisitor<DataType> {
+
+ private static final FlinkToSparkTypeVisitor INSTANCE = new
FlinkToSparkTypeVisitor();
+
+ @Override
+ public DataType visit(CharType charType) {
+ return DataTypes.StringType;
+ }
+
+ @Override
+ public DataType visit(VarCharType varCharType) {
+ return DataTypes.StringType;
+ }
+
+ @Override
+ public DataType visit(BooleanType booleanType) {
+ return DataTypes.BooleanType;
+ }
+
+ @Override
+ public DataType visit(BinaryType binaryType) {
+ return DataTypes.BinaryType;
+ }
+
+ @Override
+ public DataType visit(VarBinaryType varBinaryType) {
+ return DataTypes.BinaryType;
+ }
+
+ @Override
+ public DataType visit(DecimalType decimalType) {
+ return DataTypes.createDecimalType(decimalType.getPrecision(),
decimalType.getScale());
+ }
+
+ @Override
+ public DataType visit(TinyIntType tinyIntType) {
+ return DataTypes.ByteType;
+ }
+
+ @Override
+ public DataType visit(SmallIntType smallIntType) {
+ return DataTypes.ShortType;
+ }
+
+ @Override
+ public DataType visit(IntType intType) {
+ return DataTypes.IntegerType;
+ }
+
+ @Override
+ public DataType visit(BigIntType bigIntType) {
+ return DataTypes.LongType;
+ }
+
+ @Override
+ public DataType visit(FloatType floatType) {
+ return DataTypes.FloatType;
+ }
+
+ @Override
+ public DataType visit(DoubleType doubleType) {
+ return DataTypes.DoubleType;
+ }
+
+ @Override
+ public DataType visit(DateType dateType) {
+ return DataTypes.DateType;
+ }
+
+ @Override
+ public DataType visit(TimestampType timestampType) {
+ return DataTypes.TimestampType;
+ }
+
+ @Override
+ public DataType visit(LocalZonedTimestampType localZonedTimestampType)
{
+ return DataTypes.TimestampType;
+ }
+
+ @Override
+ public DataType visit(ArrayType arrayType) {
+ LogicalType elementType = arrayType.getElementType();
+ return DataTypes.createArrayType(elementType.accept(this),
elementType.isNullable());
+ }
+
+ @Override
+ public DataType visit(MultisetType multisetType) {
+ return DataTypes.createMapType(
+ multisetType.getElementType().accept(this),
DataTypes.IntegerType, false);
+ }
+
+ @Override
+ public DataType visit(MapType mapType) {
+ return DataTypes.createMapType(
+ mapType.getKeyType().accept(this),
+ mapType.getValueType().accept(this),
+ mapType.getValueType().isNullable());
+ }
+
+ @Override
+ public DataType visit(RowType rowType) {
+ List<StructField> fields = new
ArrayList<>(rowType.getFieldCount());
+ for (RowField field : rowType.getFields()) {
+ StructField structField =
+ DataTypes.createStructField(
+ field.getName(),
+ field.getType().accept(this),
+ field.getType().isNullable());
+ structField =
+
field.getDescription().map(structField::withComment).orElse(structField);
+ fields.add(structField);
+ }
+ return DataTypes.createStructType(fields);
+ }
+
+ @Override
+ protected DataType defaultMethod(LogicalType logicalType) {
+ throw new UnsupportedOperationException("Unsupported type: " +
logicalType);
+ }
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SpecializedGettersReader.java
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SpecializedGettersReader.java
new file mode 100644
index 00000000..5eaf1738
--- /dev/null
+++
b/flink-table-store-spark2/src/main/java/org/apache/flink/table/store/spark/SpecializedGettersReader.java
@@ -0,0 +1,106 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.BinaryType;
+import org.apache.spark.sql.types.BooleanType;
+import org.apache.spark.sql.types.ByteType;
+import org.apache.spark.sql.types.CalendarIntervalType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DateType;
+import org.apache.spark.sql.types.DecimalType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.FloatType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.sql.types.LongType;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.NullType;
+import org.apache.spark.sql.types.ShortType;
+import org.apache.spark.sql.types.StringType;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.types.TimestampType;
+import org.apache.spark.sql.types.UserDefinedType;
+
+/** Reader of Spark {@link SpecializedGetters}. */
+public final class SpecializedGettersReader {
+
+ private SpecializedGettersReader() {}
+
+ public static Object read(SpecializedGetters obj, int ordinal, DataType
dataType) {
+ if (obj.isNullAt(ordinal) || dataType instanceof NullType) {
+ return null;
+ }
+ if (dataType instanceof BooleanType) {
+ return obj.getBoolean(ordinal);
+ }
+ if (dataType instanceof ByteType) {
+ return obj.getByte(ordinal);
+ }
+ if (dataType instanceof ShortType) {
+ return obj.getShort(ordinal);
+ }
+ if (dataType instanceof IntegerType) {
+ return obj.getInt(ordinal);
+ }
+ if (dataType instanceof LongType) {
+ return obj.getLong(ordinal);
+ }
+ if (dataType instanceof FloatType) {
+ return obj.getFloat(ordinal);
+ }
+ if (dataType instanceof DoubleType) {
+ return obj.getDouble(ordinal);
+ }
+ if (dataType instanceof StringType) {
+ return obj.getUTF8String(ordinal);
+ }
+ if (dataType instanceof DecimalType) {
+ DecimalType dt = (DecimalType) dataType;
+ return obj.getDecimal(ordinal, dt.precision(), dt.scale());
+ }
+ if (dataType instanceof DateType) {
+ return obj.getInt(ordinal);
+ }
+ if (dataType instanceof TimestampType) {
+ return obj.getLong(ordinal);
+ }
+ if (dataType instanceof CalendarIntervalType) {
+ return obj.getInterval(ordinal);
+ }
+ if (dataType instanceof BinaryType) {
+ return obj.getBinary(ordinal);
+ }
+ if (dataType instanceof StructType) {
+ return obj.getStruct(ordinal, ((StructType) dataType).size());
+ }
+ if (dataType instanceof ArrayType) {
+ return obj.getArray(ordinal);
+ }
+ if (dataType instanceof MapType) {
+ return obj.getMap(ordinal);
+ }
+ if (dataType instanceof UserDefinedType) {
+ return obj.get(ordinal, ((UserDefinedType) dataType).sqlType());
+ }
+
+ throw new UnsupportedOperationException("Unsupported data type " +
dataType.simpleString());
+ }
+}
diff --git
a/flink-table-store-spark2/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
b/flink-table-store-spark2/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
new file mode 100644
index 00000000..e0626103
--- /dev/null
+++
b/flink-table-store-spark2/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
@@ -0,0 +1,16 @@
+# 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
+#
+# http://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.
+
+org.apache.flink.table.store.spark.SparkSource
diff --git
a/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
new file mode 100644
index 00000000..705a961a
--- /dev/null
+++
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SimpleTableTestHelper.java
@@ -0,0 +1,70 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.file.schema.SchemaManager;
+import org.apache.flink.table.store.file.schema.UpdateSchema;
+import org.apache.flink.table.store.table.FileStoreTable;
+import org.apache.flink.table.store.table.FileStoreTableFactory;
+import org.apache.flink.table.store.table.sink.TableCommit;
+import org.apache.flink.table.store.table.sink.TableWrite;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+/** A simple table test helper to write and commit. */
+public class SimpleTableTestHelper {
+
+ private final TableWrite writer;
+ private final TableCommit commit;
+
+ public SimpleTableTestHelper(Path path, RowType rowType) throws Exception {
+ Map<String, String> options = new HashMap<>();
+ // orc is shaded, can not find shaded classes in ide
+ options.put(CoreOptions.FILE_FORMAT.key(), "avro");
+ new SchemaManager(path)
+ .commitNewVersion(
+ new UpdateSchema(
+ rowType,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options,
+ ""));
+ Configuration conf = Configuration.fromMap(options);
+ conf.setString("path", path.toString());
+ FileStoreTable table = FileStoreTableFactory.create(conf);
+ this.writer = table.newWrite();
+ this.commit = table.newCommit("user");
+ }
+
+ public void write(RowData row) throws Exception {
+ writer.write(row);
+ }
+
+ public void commit() throws Exception {
+ commit.commit(UUID.randomUUID().toString(),
writer.prepareCommit(true));
+ }
+}
diff --git
a/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkFilterConverterTest.java
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkFilterConverterTest.java
new file mode 100644
index 00000000..7180f3d6
--- /dev/null
+++
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkFilterConverterTest.java
@@ -0,0 +1,176 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.predicate.PredicateBuilder;
+import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
+
+import org.apache.spark.sql.sources.EqualTo;
+import org.apache.spark.sql.sources.GreaterThan;
+import org.apache.spark.sql.sources.GreaterThanOrEqual;
+import org.apache.spark.sql.sources.In;
+import org.apache.spark.sql.sources.IsNotNull;
+import org.apache.spark.sql.sources.IsNull;
+import org.apache.spark.sql.sources.LessThan;
+import org.apache.spark.sql.sources.LessThanOrEqual;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link SparkFilterConverter}. */
+public class SparkFilterConverterTest {
+
+ @Test
+ public void testAll() {
+ RowType rowType =
+ new RowType(Collections.singletonList(new
RowType.RowField("id", new IntType())));
+ SparkFilterConverter converter = new SparkFilterConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ String field = "id";
+ IsNull isNull = IsNull.apply(field);
+ Predicate expectedIsNull = builder.isNull(0);
+ Predicate actualIsNull = converter.convert(isNull);
+ assertThat(actualIsNull).isEqualTo(expectedIsNull);
+
+ IsNotNull isNotNull = IsNotNull.apply(field);
+ Predicate expectedIsNotNull = builder.isNotNull(0);
+ Predicate actualIsNotNull = converter.convert(isNotNull);
+ assertThat(actualIsNotNull).isEqualTo(expectedIsNotNull);
+
+ LessThan lt = LessThan.apply(field, 1);
+ Predicate expectedLt = builder.lessThan(0, 1);
+ Predicate actualLt = converter.convert(lt);
+ assertThat(actualLt).isEqualTo(expectedLt);
+
+ LessThan ltNull = LessThan.apply(field, null);
+ Predicate expectedLtNull = builder.lessThan(0, null);
+ Predicate actualLtNull = converter.convert(ltNull);
+ assertThat(actualLtNull).isEqualTo(expectedLtNull);
+
+ LessThanOrEqual ltEq = LessThanOrEqual.apply(field, 1);
+ Predicate expectedLtEq = builder.lessOrEqual(0, 1);
+ Predicate actualLtEq = converter.convert(ltEq);
+ assertThat(actualLtEq).isEqualTo(expectedLtEq);
+
+ LessThanOrEqual ltEqNull = LessThanOrEqual.apply(field, null);
+ Predicate expectedLtEqNull = builder.lessOrEqual(0, null);
+ Predicate actualLtEqNull = converter.convert(ltEqNull);
+ assertThat(actualLtEqNull).isEqualTo(expectedLtEqNull);
+
+ GreaterThan gt = GreaterThan.apply(field, 1);
+ Predicate expectedGt = builder.greaterThan(0, 1);
+ Predicate actualGt = converter.convert(gt);
+ assertThat(actualGt).isEqualTo(expectedGt);
+
+ GreaterThan gtNull = GreaterThan.apply(field, null);
+ Predicate expectedGtNull = builder.greaterThan(0, null);
+ Predicate actualGtNull = converter.convert(gtNull);
+ assertThat(actualGtNull).isEqualTo(expectedGtNull);
+
+ GreaterThanOrEqual gtEq = GreaterThanOrEqual.apply(field, 1);
+ Predicate expectedGtEq = builder.greaterOrEqual(0, 1);
+ Predicate actualGtEq = converter.convert(gtEq);
+ assertThat(actualGtEq).isEqualTo(expectedGtEq);
+
+ GreaterThanOrEqual gtEqNull = GreaterThanOrEqual.apply(field, null);
+ Predicate expectedGtEqNull = builder.greaterOrEqual(0, null);
+ Predicate actualGtEqNull = converter.convert(gtEqNull);
+ assertThat(actualGtEqNull).isEqualTo(expectedGtEqNull);
+
+ EqualTo eq = EqualTo.apply(field, 1);
+ Predicate expectedEq = builder.equal(0, 1);
+ Predicate actualEq = converter.convert(eq);
+ assertThat(actualEq).isEqualTo(expectedEq);
+
+ EqualTo eqNull = EqualTo.apply(field, null);
+ Predicate expectedEqNull = builder.equal(0, null);
+ Predicate actualEqNull = converter.convert(eqNull);
+ assertThat(actualEqNull).isEqualTo(expectedEqNull);
+
+ In in = In.apply(field, new Object[] {1, null, 2});
+ Predicate expectedIn = builder.in(0, Arrays.asList(1, null, 2));
+ Predicate actualIn = converter.convert(in);
+ assertThat(actualIn).isEqualTo(expectedIn);
+
+ Object[] literals = new Object[30];
+ literals[0] = null;
+ for (int i = 1; i < literals.length; i++) {
+ literals[i] = i * 100;
+ }
+ In largeIn = In.apply(field, literals);
+ Predicate expectedLargeIn = builder.in(0, Arrays.asList(literals));
+ Predicate actualLargeIn = converter.convert(largeIn);
+ assertThat(actualLargeIn).isEqualTo(expectedLargeIn);
+ }
+
+ @Test
+ public void testTimestamp() {
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(new RowType.RowField("x",
new TimestampType())));
+ SparkFilterConverter converter = new SparkFilterConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ Timestamp timestamp = Timestamp.valueOf("2018-10-18 00:00:57.907");
+ LocalDateTime localDateTime =
LocalDateTime.parse("2018-10-18T00:00:57.907");
+ Instant instant = localDateTime.toInstant(ZoneOffset.UTC);
+
+ Predicate instantExpression = converter.convert(GreaterThan.apply("x",
instant));
+ Predicate timestampExpression =
converter.convert(GreaterThan.apply("x", timestamp));
+ Predicate rawExpression =
+ builder.greaterThan(0,
TimestampData.fromLocalDateTime(localDateTime));
+
+ assertThat(timestampExpression).isEqualTo(rawExpression);
+ assertThat(instantExpression).isEqualTo(rawExpression);
+ }
+
+ @Test
+ public void testDate() {
+ RowType rowType =
+ new RowType(Collections.singletonList(new
RowType.RowField("x", new DateType())));
+ SparkFilterConverter converter = new SparkFilterConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ LocalDate localDate = LocalDate.parse("2018-10-18");
+ Date date = Date.valueOf(localDate);
+ int epochDay = (int) localDate.toEpochDay();
+
+ Predicate localDateExpression =
converter.convert(GreaterThan.apply("x", localDate));
+ Predicate dateExpression = converter.convert(GreaterThan.apply("x",
date));
+ Predicate rawExpression = builder.greaterThan(0, epochDay);
+
+ assertThat(dateExpression).isEqualTo(rawExpression);
+ assertThat(localDateExpression).isEqualTo(rawExpression);
+ }
+}
diff --git
a/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkInternalRowTest.java
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkInternalRowTest.java
new file mode 100644
index 00000000..4bfb58ab
--- /dev/null
+++
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkInternalRowTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.conversion.RowRowConverter;
+import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.types.Row;
+
+import org.apache.spark.sql.catalyst.CatalystTypeConverters;
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.AbstractMap;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import scala.Function1;
+
+import static org.apache.flink.table.store.spark.SparkTypeTest.ALL_TYPES;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link SparkInternalRow}. */
+public class SparkInternalRowTest {
+
+ @Test
+ public void test() {
+ Row row =
+ Row.of(
+ 1,
+ "jingsong",
+ 22.2,
+ Stream.of(
+ new AbstractMap.SimpleEntry<>("key1",
Row.of(1.2, 2.3)),
+ new AbstractMap.SimpleEntry<>("key2",
Row.of(2.4, 3.5)))
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue)),
+ new String[] {"v1", "v5"},
+ new Integer[] {10, 30},
+ true,
+ (byte) 22,
+ (short) 356,
+ 23567222L,
+ "varbinary_v".getBytes(StandardCharsets.UTF_8),
+ LocalDateTime.parse("2007-12-03T10:15:30"),
+ LocalDate.parse("2022-05-02"),
+ BigDecimal.valueOf(0.21),
+ BigDecimal.valueOf(65782123123.01),
+ BigDecimal.valueOf(62123123.5));
+
+ RowRowConverter flinkConverter =
+
RowRowConverter.create(TypeConversions.fromLogicalToDataType(ALL_TYPES));
+ flinkConverter.open(Thread.currentThread().getContextClassLoader());
+ RowData rowData = flinkConverter.toInternal(row);
+
+ Function1<Object, Object> sparkConverter =
+ CatalystTypeConverters.createToScalaConverter(
+ SparkTypeUtils.fromFlinkType(ALL_TYPES));
+ org.apache.spark.sql.Row sparkRow =
+ (org.apache.spark.sql.Row)
+ sparkConverter.apply(new
SparkInternalRow(ALL_TYPES).replace(rowData));
+
+ assertThat(sparkRow.getInt(0)).isEqualTo(1);
+ assertThat(sparkRow.getString(1)).isEqualTo("jingsong");
+ assertThat(sparkRow.getDouble(2)).isEqualTo(22.2);
+ assertThat(((GenericRowWithSchema)
sparkRow.getJavaMap(3).get("key1")).values())
+ .isEqualTo(new Object[] {1.2, 2.3});
+ assertThat(((GenericRowWithSchema)
sparkRow.getJavaMap(3).get("key2")).values())
+ .isEqualTo(new Object[] {2.4, 3.5});
+ assertThat(sparkRow.getList(4)).isEqualTo(Arrays.asList("v1", "v5"));
+ assertThat(sparkRow.getList(5)).isEqualTo(Arrays.asList(10, 30));
+ assertThat(sparkRow.getBoolean(6)).isEqualTo(true);
+ assertThat(sparkRow.getByte(7)).isEqualTo((byte) 22);
+ assertThat(sparkRow.getShort(8)).isEqualTo((short) 356);
+ assertThat(sparkRow.getLong(9)).isEqualTo(23567222L);
+ assertThat((byte[]) sparkRow.get(10))
+ .isEqualTo("varbinary_v".getBytes(StandardCharsets.UTF_8));
+ assertThat(sparkRow.getTimestamp(11).getTime())
+
.isEqualTo(Timestamp.valueOf(LocalDateTime.parse("2007-12-03T10:15:30")).getTime());
+ assertThat(sparkRow.getDate(12).getTime())
+
.isEqualTo(Date.valueOf(LocalDate.parse("2022-05-02")).getTime());
+
assertThat(sparkRow.getDecimal(13)).isEqualTo(BigDecimal.valueOf(0.21));
+
assertThat(sparkRow.getDecimal(14)).isEqualTo(BigDecimal.valueOf(65782123123.01));
+
assertThat(sparkRow.getDecimal(15)).isEqualTo(BigDecimal.valueOf(62123123.5));
+ }
+}
diff --git
a/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkReadITCase.java
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkReadITCase.java
new file mode 100644
index 00000000..86427f09
--- /dev/null
+++
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkReadITCase.java
@@ -0,0 +1,130 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.RowKind;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** ITCase for Spark reader. */
+public class SparkReadITCase {
+
+ private static File warehouse = null;
+
+ private static SparkSession spark = null;
+
+ private static Path tablePath1;
+
+ private static Path tablePath2;
+
+ @BeforeAll
+ public static void startSpark() throws Exception {
+ warehouse = File.createTempFile("warehouse", null);
+ assertThat(warehouse.delete()).isTrue();
+ Path warehousePath = new Path("file:" + warehouse);
+ spark = SparkSession.builder().master("local[2]").getOrCreate();
+
+ // Flink sink
+ tablePath1 = new Path(warehousePath, "default.db/t1");
+ SimpleTableTestHelper testHelper1 = createTestHelper(tablePath1);
+ testHelper1.write(GenericRowData.of(1, 2L,
StringData.fromString("1")));
+ testHelper1.write(GenericRowData.of(3, 4L,
StringData.fromString("2")));
+ testHelper1.write(GenericRowData.of(5, 6L,
StringData.fromString("3")));
+ testHelper1.write(GenericRowData.ofKind(RowKind.DELETE, 3, 4L,
StringData.fromString("2")));
+ testHelper1.commit();
+
+ tablePath2 = new Path(warehousePath, "default.db/t2");
+ SimpleTableTestHelper testHelper2 = createTestHelper(tablePath2);
+ testHelper2.write(GenericRowData.of(1, 2L,
StringData.fromString("1")));
+ testHelper2.write(GenericRowData.of(3, 4L,
StringData.fromString("2")));
+ testHelper2.commit();
+ testHelper2.write(GenericRowData.of(5, 6L,
StringData.fromString("3")));
+ testHelper2.write(GenericRowData.of(7, 8L,
StringData.fromString("4")));
+ testHelper2.commit();
+ }
+
+ private static SimpleTableTestHelper createTestHelper(Path tablePath)
throws Exception {
+ RowType rowType =
+ new RowType(
+ Arrays.asList(
+ new RowType.RowField("a", new IntType()),
+ new RowType.RowField("b", new BigIntType()),
+ new RowType.RowField("c", new VarCharType())));
+ return new SimpleTableTestHelper(tablePath, rowType);
+ }
+
+ @AfterAll
+ public static void stopSpark() throws IOException {
+ if (warehouse != null && warehouse.exists()) {
+ FileUtils.deleteDirectory(warehouse);
+ }
+ if (spark != null) {
+ spark.stop();
+ spark = null;
+ }
+ }
+
+ @Test
+ public void testNormal() {
+ innerTestNormal(
+ spark.read().format("tablestore").option("path",
tablePath1.toString()).load());
+ }
+
+ @Test
+ public void testFilterPushDown() {
+ innerTestFilterPushDown(
+ spark.read().format("tablestore").option("path",
tablePath2.toString()).load());
+ }
+
+ private void innerTestNormal(Dataset<Row> dataset) {
+ List<Row> results = dataset.collectAsList();
+ assertThat(results.toString()).isEqualTo("[[1,2,1], [5,6,3]]");
+
+ results = dataset.select("a", "c").collectAsList();
+ assertThat(results.toString()).isEqualTo("[[1,1], [5,3]]");
+
+ results = dataset.groupBy().sum("b").collectAsList();
+ assertThat(results.toString()).isEqualTo("[[8]]");
+ }
+
+ private void innerTestFilterPushDown(Dataset<Row> dataset) {
+ List<Row> results = dataset.filter("a < 4").select("a",
"c").collectAsList();
+ assertThat(results.toString()).isEqualTo("[[1,1], [3,2]]");
+ }
+}
diff --git
a/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkTypeTest.java
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkTypeTest.java
new file mode 100644
index 00000000..6a71c6af
--- /dev/null
+++
b/flink-table-store-spark2/src/test/java/org/apache/flink/table/store/spark/SparkTypeTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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
+ *
+ * http://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.store.spark;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+
+import static
org.apache.flink.table.store.spark.SparkTypeUtils.fromFlinkRowType;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link SparkTypeUtils}. */
+public class SparkTypeTest {
+
+ public static final RowType ALL_TYPES =
+ (RowType)
+ TableSchema.builder()
+ .field("id", DataTypes.INT().notNull())
+ .field("name", DataTypes.STRING()) /* optional by
default */
+ .field("salary", DataTypes.DOUBLE().notNull())
+ .field(
+ "locations",
+ DataTypes.MAP(
+ DataTypes.STRING().notNull(),
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ "posX",
+
DataTypes.DOUBLE().notNull(),
+ "X field"),
+ DataTypes.FIELD(
+ "posY",
+
DataTypes.DOUBLE().notNull(),
+ "Y field"))))
+ .field("strArray",
DataTypes.ARRAY(DataTypes.STRING()).nullable())
+ .field("intArray",
DataTypes.ARRAY(DataTypes.INT()).nullable())
+ .field("boolean", DataTypes.BOOLEAN().nullable())
+ .field("tinyint", DataTypes.TINYINT())
+ .field("smallint", DataTypes.SMALLINT())
+ .field("bigint", DataTypes.BIGINT())
+ .field("bytes", DataTypes.BYTES())
+ .field("timestamp", DataTypes.TIMESTAMP())
+ .field("date", DataTypes.DATE())
+ .field("decimal", DataTypes.DECIMAL(2, 2))
+ .field("decimal2", DataTypes.DECIMAL(38, 2))
+ .field("decimal3", DataTypes.DECIMAL(10, 1))
+ .build()
+ .toRowDataType()
+ .getLogicalType()
+ .copy(true);
+
+ @Test
+ public void testAllTypes() {
+ String nestedRowMapType =
+ "StructField(locations,MapType("
+ + "StringType,"
+ + "StructType(StructField(posX,DoubleType,false),
StructField(posY,DoubleType,false)),true),true)";
+ String expected =
+ "StructType("
+ + "StructField(id,IntegerType,false), "
+ + "StructField(name,StringType,true), "
+ + "StructField(salary,DoubleType,false), "
+ + nestedRowMapType
+ + ", "
+ +
"StructField(strArray,ArrayType(StringType,true),true), "
+ +
"StructField(intArray,ArrayType(IntegerType,true),true), "
+ + "StructField(boolean,BooleanType,true), "
+ + "StructField(tinyint,ByteType,true), "
+ + "StructField(smallint,ShortType,true), "
+ + "StructField(bigint,LongType,true), "
+ + "StructField(bytes,BinaryType,true), "
+ + "StructField(timestamp,TimestampType,true), "
+ + "StructField(date,DateType,true), "
+ + "StructField(decimal,DecimalType(2,2),true), "
+ + "StructField(decimal2,DecimalType(38,2),true), "
+ + "StructField(decimal3,DecimalType(10,1),true))";
+
+ StructType sparkType = fromFlinkRowType(ALL_TYPES);
+ assertThat(sparkType.toString()).isEqualTo(expected);
+ }
+}
diff --git a/flink-table-store-spark2/src/test/resources/log4j2-test.properties
b/flink-table-store-spark2/src/test/resources/log4j2-test.properties
new file mode 100644
index 00000000..863665cf
--- /dev/null
+++ b/flink-table-store-spark2/src/test/resources/log4j2-test.properties
@@ -0,0 +1,38 @@
+################################################################################
+# 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
+#
+# http://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.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+
+logger.kafka.name = kafka
+logger.kafka.level = OFF
+logger.kafka2.name = state.change
+logger.kafka2.level = OFF
+
+logger.zookeeper.name = org.apache.zookeeper
+logger.zookeeper.level = OFF
+logger.I0Itec.name = org.I0Itec
+logger.I0Itec.level = OFF
diff --git a/pom.xml b/pom.xml
index b2408b37..674bcfe9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -65,6 +65,7 @@ under the License.
<module>flink-table-store-hive</module>
<module>flink-table-store-kafka</module>
<module>flink-table-store-spark</module>
+ <module>flink-table-store-spark2</module>
</modules>
<properties>
