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 14326aea [FLINK-28128] Add e2e test for spark reader
14326aea is described below
commit 14326aea8e1ceb439932e032cf9407106e023e52
Author: shammon <[email protected]>
AuthorDate: Thu Oct 13 10:50:20 2022 +0800
[FLINK-28128] Add e2e test for spark reader
This closes #314
---
flink-table-store-e2e-tests/pom.xml | 16 +++
.../flink/table/store/tests/E2eQueryExecutor.java | 24 ++++
.../flink/table/store/tests/E2eReaderTestBase.java | 109 ++++++++++++++++
.../flink/table/store/tests/E2eTestBase.java | 17 +++
.../flink/table/store/tests/HiveE2eTest.java | 140 ++++++---------------
.../flink/table/store/tests/SparkE2eTest.java | 84 +++++++++++++
.../test/resources-filtered/docker-compose.yaml | 29 +++++
7 files changed, 315 insertions(+), 104 deletions(-)
diff --git a/flink-table-store-e2e-tests/pom.xml
b/flink-table-store-e2e-tests/pom.xml
index 178b96b8..07bf3a21 100644
--- a/flink-table-store-e2e-tests/pom.xml
+++ b/flink-table-store-e2e-tests/pom.xml
@@ -54,6 +54,12 @@ under the License.
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-store-spark</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<!-- flink dependencies -->
<dependency>
@@ -133,6 +139,16 @@ under the License.
<outputDirectory>/tmp/flink-table-store-e2e-tests-jars
</outputDirectory>
</artifactItem>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-store-spark</artifactId>
+ <version>${project.version}</version>
+
<destFileName>flink-table-store-spark.jar</destFileName>
+ <type>jar</type>
+ <overWrite>true</overWrite>
+
<outputDirectory>/tmp/flink-table-store-e2e-tests-jars
+ </outputDirectory>
+ </artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-hadoop-2-uber</artifactId>
diff --git
a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eQueryExecutor.java
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eQueryExecutor.java
new file mode 100644
index 00000000..e6b3786f
--- /dev/null
+++
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eQueryExecutor.java
@@ -0,0 +1,24 @@
+/*
+ * 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.tests;
+
+/** E2e query executor for testing. */
+public interface E2eQueryExecutor {
+ String execute(String sql) throws Exception;
+}
diff --git
a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eReaderTestBase.java
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eReaderTestBase.java
new file mode 100644
index 00000000..6507e08f
--- /dev/null
+++
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eReaderTestBase.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.tests;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Base e2e test class for hive and spark readers. */
+public abstract class E2eReaderTestBase extends E2eTestBase {
+ public E2eReaderTestBase(boolean withKafka, boolean withHive, boolean
withSpark) {
+ super(withKafka, withHive, withSpark);
+ }
+
+ protected String createCatalogSql(String catalog, String warehouse,
String... options) {
+ return String.join(
+ "\n",
+ String.format("CREATE CATALOG %s WITH (", catalog),
+ " 'type' = 'table-store',",
+ options.length > 0 ? String.join(",", options) + "," : "",
+ String.format(" 'warehouse' = '%s'", warehouse),
+ ");",
+ "",
+ String.format("USE CATALOG %s;", catalog));
+ }
+
+ protected String createTableSql(String tableName) {
+ return String.format(
+ "CREATE TABLE %s ("
+ + " a int,"
+ + " b bigint,"
+ + " c string"
+ + ") WITH ("
+ + " 'bucket' = '2'"
+ + ");",
+ tableName);
+ }
+
+ protected String createInsertSql(String tableName) {
+ return String.format(
+ "INSERT INTO %s VALUES "
+ + "(1, 10, 'Hi'), "
+ + "(1, 100, 'Hi Again'), "
+ + "(2, 20, 'Hello'), "
+ + "(3, 30, 'Table'), "
+ + "(4, 40, 'Store');",
+ tableName);
+ }
+
+ protected void checkQueryResults(String tableName, E2eQueryExecutor
executor) throws Exception {
+ checkQueryResults(tableName, executor, "");
+ }
+
+ protected void checkQueryResults(String tableName, E2eQueryExecutor
executor, String ddl)
+ throws Exception {
+ checkQueryResult(
+ executor,
+ ddl + String.format("SELECT * FROM %s ORDER BY b;", tableName),
+ "1\t10\tHi\n"
+ + "2\t20\tHello\n"
+ + "3\t30\tTable\n"
+ + "4\t40\tStore\n"
+ + "1\t100\tHi Again\n");
+ checkQueryResult(
+ executor,
+ ddl + String.format("SELECT b, a FROM %s ORDER BY b;",
tableName),
+ "10\t1\n" + "20\t2\n" + "30\t3\n" + "40\t4\n" + "100\t1\n");
+ checkQueryResult(
+ executor,
+ ddl + String.format("SELECT * FROM %s WHERE a > 1 ORDER BY
b;", tableName),
+ "2\t20\tHello\n" + "3\t30\tTable\n" + "4\t40\tStore\n");
+ checkQueryResult(
+ executor,
+ ddl
+ + String.format(
+ "SELECT a, SUM(b), MIN(c) FROM %s GROUP BY a
ORDER BY a;",
+ tableName),
+ "1\t110\tHi\n" + "2\t20\tHello\n" + "3\t30\tTable\n" +
"4\t40\tStore\n");
+ checkQueryResult(
+ executor,
+ ddl
+ + String.format(
+ "SELECT T1.a, T1.b, T2.b FROM %s T1 JOIN %s T2
"
+ + "ON T1.a = T2.a WHERE T1.a <= 2
ORDER BY T1.a, T1.b, T2.b;",
+ tableName, tableName),
+ "1\t10\t10\n" + "1\t10\t100\n" + "1\t100\t10\n" +
"1\t100\t100\n" + "2\t20\t20\n");
+ }
+
+ private void checkQueryResult(E2eQueryExecutor executor, String query,
String result)
+ throws Exception {
+ final String file = "pk.hql";
+ writeSharedFile(file, query);
+ assertEquals(result, executor.execute(file));
+ }
+}
diff --git
a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java
index bcaf2945..fe09cdb1 100644
---
a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java
+++
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java
@@ -52,14 +52,20 @@ public abstract class E2eTestBase {
private final boolean withKafka;
private final boolean withHive;
+ private final boolean withSpark;
protected E2eTestBase() {
this(false, false);
}
protected E2eTestBase(boolean withKafka, boolean withHive) {
+ this(withKafka, withHive, false);
+ }
+
+ protected E2eTestBase(boolean withKafka, boolean withHive, boolean
withSpark) {
this.withKafka = withKafka;
this.withHive = withHive;
+ this.withSpark = withSpark;
}
protected static final String TEST_DATA_DIR = "/test-data";
@@ -109,6 +115,17 @@ public abstract class E2eTestBase {
environment.waitingFor(
"hive-server_1", Wait.forLogMessage(".*Starting
HiveServer2.*", 1));
}
+ if (withSpark) {
+ List<String> sparkServices = Arrays.asList("spark-master",
"spark-worker");
+ services.addAll(sparkServices);
+ for (String s : sparkServices) {
+ environment.withLogConsumer(s + "_1", new
Slf4jLogConsumer(LOG));
+ }
+ environment.waitingFor(
+ "spark-master_1",
+ Wait.forLogMessage(
+ ".*Master: I have been elected leader! New state:
ALIVE.*", 1));
+ }
environment.withServices(services.toArray(new
String[0])).withLocalCompose(true);
environment.start();
diff --git
a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/HiveE2eTest.java
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/HiveE2eTest.java
index f1b1994b..0f5ebace 100644
---
a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/HiveE2eTest.java
+++
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/HiveE2eTest.java
@@ -26,8 +26,6 @@ import org.testcontainers.containers.ContainerState;
import java.util.UUID;
-import static org.assertj.core.api.Assertions.assertThat;
-
/**
* Tests for reading table store from Hive.
*
@@ -35,13 +33,13 @@ import static org.assertj.core.api.Assertions.assertThat;
* you're running this test locally, make sure that the memory limit of your
Docker is at least 8GB.
*/
@DisabledIfSystemProperty(named = "flink.version", matches = "1.14.*")
-public class HiveE2eTest extends E2eTestBase {
+public class HiveE2eTest extends E2eReaderTestBase {
private static final String TABLE_STORE_HIVE_CONNECTOR_JAR_NAME =
"flink-table-store-hive-connector.jar";
public HiveE2eTest() {
- super(false, true);
+ super(false, true, false);
}
@BeforeEach
@@ -59,116 +57,50 @@ public class HiveE2eTest extends E2eTestBase {
@Test
public void testReadExternalTable() throws Exception {
- String tableStorePkDdl =
- "CREATE TABLE IF NOT EXISTS table_store_pk (\n"
- + " a int,\n"
- + " b bigint,\n"
- + " c string,\n"
- + " PRIMARY KEY (a, b) NOT ENFORCED\n"
- + ") WITH (\n"
- + " 'bucket' = '2',\n"
- + " 'root-path' = '%s'\n"
- + ");";
+ final String table = "table_store_pk";
String tableStorePkPath = HDFS_ROOT + "/" +
UUID.randomUUID().toString() + ".store";
- tableStorePkDdl = String.format(tableStorePkDdl, tableStorePkPath);
- runSql(
- "INSERT INTO table_store_pk VALUES "
- + "(1, 10, 'Hi'), "
- + "(1, 100, 'Hi Again'), "
- + "(2, 20, 'Hello'), "
- + "(3, 30, 'Table'), "
- + "(4, 40, 'Store');",
- tableStorePkDdl);
+ String tableStorePkDdl =
+ String.format(
+ "CREATE TABLE IF NOT EXISTS %s (\n"
+ + " a int,\n"
+ + " b bigint,\n"
+ + " c string,\n"
+ + " PRIMARY KEY (a, b) NOT ENFORCED\n"
+ + ") WITH (\n"
+ + " 'bucket' = '2',\n"
+ + " 'root-path' = '%s'\n"
+ + ");",
+ table, tableStorePkPath);
+ runSql(createInsertSql(table), tableStorePkDdl);
String externalTablePkDdl =
- "CREATE EXTERNAL TABLE IF NOT EXISTS table_store_pk\n"
- + "STORED BY
'org.apache.flink.table.store.hive.TableStoreHiveStorageHandler'\n"
- + "LOCATION '"
- + tableStorePkPath
- +
"/default_catalog.catalog/default_database.db/table_store_pk';\n";
+ String.format(
+ "CREATE EXTERNAL TABLE IF NOT EXISTS %s\n"
+ + "STORED BY
'org.apache.flink.table.store.hive.TableStoreHiveStorageHandler'\n"
+ + "LOCATION
'%s/default_catalog.catalog/default_database.db/%s';\n",
+ table, tableStorePkPath, table);
- checkQueryResult(
- externalTablePkDdl + "SELECT * FROM table_store_pk ORDER BY
b;",
- "1\t10\tHi\n"
- + "2\t20\tHello\n"
- + "3\t30\tTable\n"
- + "4\t40\tStore\n"
- + "1\t100\tHi Again\n");
- checkQueryResult(
- externalTablePkDdl + "SELECT b, a FROM table_store_pk ORDER BY
b;",
- "10\t1\n" + "20\t2\n" + "30\t3\n" + "40\t4\n" + "100\t1\n");
- checkQueryResult(
- externalTablePkDdl + "SELECT * FROM table_store_pk WHERE a > 1
ORDER BY b;",
- "2\t20\tHello\n" + "3\t30\tTable\n" + "4\t40\tStore\n");
- checkQueryResult(
- externalTablePkDdl
- + "SELECT a, SUM(b), MIN(c) FROM table_store_pk GROUP
BY a ORDER BY a;",
- "1\t110\tHi\n" + "2\t20\tHello\n" + "3\t30\tTable\n" +
"4\t40\tStore\n");
- checkQueryResult(
- externalTablePkDdl
- + "SELECT T1.a, T1.b, T2.b FROM table_store_pk T1 JOIN
table_store_pk T2 "
- + "ON T1.a = T2.a WHERE T1.a <= 2 ORDER BY T1.a, T1.b,
T2.b;",
- "1\t10\t10\n" + "1\t10\t100\n" + "1\t100\t10\n" +
"1\t100\t100\n" + "2\t20\t20\n");
+ checkQueryResults(table, this::executeQuery, externalTablePkDdl);
}
@Test
public void testFlinkWriteAndHiveRead() throws Exception {
- String sql =
+ final String warehouse = HDFS_ROOT + "/" +
UUID.randomUUID().toString() + ".warehouse";
+ final String table = "T";
+ runSql(
String.join(
"\n",
- "CREATE CATALOG my_hive WITH (",
- " 'type' = 'table-store',",
- " 'metastore' = 'hive',",
- " 'uri' = 'thrift://hive-metastore:9083',",
- " 'warehouse' = '"
- + HDFS_ROOT
- + "/"
- + UUID.randomUUID().toString()
- + ".warehouse'",
- ");",
- "",
- "USE CATALOG my_hive;",
- "",
- "CREATE TABLE T (",
- " a int,",
- " b bigint,",
- " c string",
- ") WITH (",
- " 'bucket' = '2'",
- ");",
- "",
- "INSERT INTO T VALUES "
- + "(1, 10, 'Hi'), "
- + "(1, 100, 'Hi Again'), "
- + "(2, 20, 'Hello'), "
- + "(3, 30, 'Table'), "
- + "(4, 40, 'Store');");
- runSql(sql);
-
- checkQueryResult(
- "SELECT * FROM t ORDER BY b;",
- "1\t10\tHi\n"
- + "2\t20\tHello\n"
- + "3\t30\tTable\n"
- + "4\t40\tStore\n"
- + "1\t100\tHi Again\n");
- checkQueryResult(
- "SELECT b, a FROM t ORDER BY b;",
- "10\t1\n" + "20\t2\n" + "30\t3\n" + "40\t4\n" + "100\t1\n");
- checkQueryResult(
- "SELECT * FROM t WHERE a > 1 ORDER BY b;",
- "2\t20\tHello\n" + "3\t30\tTable\n" + "4\t40\tStore\n");
- checkQueryResult(
- "SELECT a, SUM(b), MIN(c) FROM t GROUP BY a ORDER BY a;",
- "1\t110\tHi\n" + "2\t20\tHello\n" + "3\t30\tTable\n" +
"4\t40\tStore\n");
- checkQueryResult(
- "SELECT T1.a, T1.b, T2.b FROM t T1 JOIN t T2 "
- + "ON T1.a = T2.a WHERE T1.a <= 2 ORDER BY T1.a, T1.b,
T2.b;",
- "1\t10\t10\n" + "1\t10\t100\n" + "1\t100\t10\n" +
"1\t100\t100\n" + "2\t20\t20\n");
+ createCatalogSql(
+ "my_hive",
+ warehouse,
+ "'metastore' = 'hive'",
+ "'uri' = 'thrift://hive-metastore:9083'"),
+ createTableSql(table),
+ createInsertSql(table)));
+ checkQueryResults(table, this::executeQuery);
}
- private void checkQueryResult(String query, String expected) throws
Exception {
- writeSharedFile("pk.hql", query);
+ private String executeQuery(String sql) throws Exception {
Container.ExecResult execResult =
getHive()
.execInContainer(
@@ -176,11 +108,11 @@ public class HiveE2eTest extends E2eTestBase {
"--hiveconf",
"hive.root.logger=INFO,console",
"-f",
- TEST_DATA_DIR + "/pk.hql");
+ TEST_DATA_DIR + "/" + sql);
if (execResult.getExitCode() != 0) {
throw new AssertionError("Failed when running hive sql.");
}
- assertThat(execResult.getStdout()).isEqualTo(expected);
+ return execResult.getStdout();
}
private ContainerState getHive() {
diff --git
a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/SparkE2eTest.java
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/SparkE2eTest.java
new file mode 100644
index 00000000..bf29a8fc
--- /dev/null
+++
b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/SparkE2eTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.tests;
+
+import org.apache.commons.lang3.StringUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.ContainerState;
+
+import java.util.Arrays;
+import java.util.UUID;
+
+/** Tests for reading table store from Spark3. */
+@DisabledIfSystemProperty(named = "flink.version", matches = "1.14.*")
+public class SparkE2eTest extends E2eReaderTestBase {
+
+ public SparkE2eTest() {
+ super(false, false, true);
+ }
+
+ @Test
+ public void testFlinkWriteAndSparkRead() throws Exception {
+ String warehousePath = TEST_DATA_DIR + "/" +
UUID.randomUUID().toString() + "_warehouse";
+ final String table = "T";
+ final String sparkTable = String.format("tablestore.default.%s",
table);
+ runSql(
+ String.join(
+ "\n",
+ createCatalogSql("my_spark", warehousePath),
+ createTableSql(table),
+ createInsertSql(table)));
+ checkQueryResults(
+ sparkTable,
+ sql -> {
+ Container.ExecResult execResult =
+ getSpark()
+ .execInContainer(
+ "/spark/bin/spark-sql",
+ "--master",
+ "spark://spark-master:7077",
+ "--conf",
+
"spark.sql.catalog.tablestore=org.apache.flink.table.store.spark.SparkCatalog",
+ "--conf",
+
"spark.sql.catalog.tablestore.warehouse=file:"
+ + warehousePath,
+ "-f",
+ TEST_DATA_DIR + "/" + sql);
+ if (execResult.getExitCode() != 0) {
+ throw new AssertionError("Failed when running spark
sql.");
+ }
+
+ return filterLog(execResult.getStdout());
+ });
+ }
+
+ private String filterLog(String result) {
+ return StringUtils.join(
+ Arrays.stream(StringUtils.splitByWholeSeparator(result, "\n"))
+ .filter(v -> !StringUtils.contains(v, " WARN "))
+ .toArray(),
+ "\n");
+ }
+
+ private ContainerState getSpark() {
+ return environment.getContainerByServiceName("spark-master_1").get();
+ }
+}
diff --git
a/flink-table-store-e2e-tests/src/test/resources-filtered/docker-compose.yaml
b/flink-table-store-e2e-tests/src/test/resources-filtered/docker-compose.yaml
index 5726c70d..15fa2389 100644
---
a/flink-table-store-e2e-tests/src/test/resources-filtered/docker-compose.yaml
+++
b/flink-table-store-e2e-tests/src/test/resources-filtered/docker-compose.yaml
@@ -170,6 +170,35 @@ services:
aliases:
- hive-metastore-postgresql
+ # ----------------------------------------
+ # Spark services, copied and modified from
https://github.com/big-data-europe/docker-spark
+ # ----------------------------------------
+
+ spark-master:
+ image: bde2020/spark-master:3.3.0-hadoop3.3
+ volumes:
+ - testdata:/test-data
+ - /tmp/flink-table-store-e2e-tests-jars:/jars
+ ports:
+ - "8080:8080"
+ - "7077:7077"
+ entrypoint: /bin/bash -c "cp /jars/flink-table-store.jar
/jars/flink-table-store-spark.jar /spark/jars/ && sh /master.sh"
+ environment:
+ - INIT_DAEMON_STEP=setup_spark
+
+ spark-worker:
+ image: bde2020/spark-worker:3.3.0-hadoop3.3
+ depends_on:
+ - spark-master
+ volumes:
+ - testdata:/test-data
+ - /tmp/flink-table-store-e2e-tests-jars:/jars
+ ports:
+ - "8081:8081"
+ entrypoint: /bin/bash -c "cp /jars/flink-table-store.jar
/jars/flink-table-store-spark.jar /spark/jars/ && sh /worker.sh"
+ environment:
+ - "SPARK_MASTER=spark://spark-master:7077"
+
volumes:
testdata:
namenode:
