[FLINK-18337][table] Introduce TableResult#await method to wait for data ready #12688

godfreyhe · 2020-06-16T17:43:18Z

What is the purpose of the change

TableEnvironment.executeSql() method for INSERT statement returns TableResult once the job is submitted. Users must use tableResult.getJobClient.get().getJobExecutionResult(Thread.currentThread().getContextClassLoader).get() to wait the job finish. This API looks very ugly. This pr aims to introduce TableResult#await method to make the api fluent

Brief change log

Introduce await in TableResult
remove TableEnvUtil and update the related tests
remove the usage of tableResult.getJobClient.get().getJobExecutionResult().get() in planner tests

Verifying this change

This change is already covered by existing tests*.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2020-06-16T17:45:39Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit a66d075 (Tue Jun 16 17:45:39 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-06-16T18:03:33Z

CI report:

9fc17e0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

godfreyhe · 2020-06-17T11:56:59Z

cc @twalthr

twalthr

Thanks for this change @godfreyhe, it makes the API much nicer and concise. I left some comments.

twalthr · 2020-06-17T12:47:09Z

...ink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java

 			"FROM hTable AS h");

-		List<Row> results = collectBatchResult(table);
+		List<Row> results = Lists.newArrayList(table.execute().collect());


avoid dependencies to other modules for simple util functions. Introduce an overloaded method org.apache.flink.util.CollectionUtil#iteratorToList in a hotfix commit

twalthr · 2020-06-17T12:51:10Z

...ink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java

 			"  h.family3.col2, " +
 			"  h.family3.col3 " +
 			"FROM hTable AS h");



For all locations that were updated in the previous commit the declared throws Exception is not necessary anymore. My IDE shows a lot of warnings Exception 'java.lang.Exception' is never thrown in the method which we can fix in this PR.

twalthr · 2020-06-17T12:53:17Z

...flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java

+						"insert into db1.sink_table select 6,'a','b','2020-05-03','12'")
+						.await();
+			} catch (Exception e) {
+				throw new TableException("Failed to execute sql", e);


throw AssertionError in tests with Assert.fail().

twalthr · 2020-06-17T13:12:30Z

flink-python/pyflink/table/table_result.py

+        """
+        Wait if necessary for at most the given time (milliseconds) for the data to be ready.
+
+        For select operation, this method will wait unit the first row can be accessed in local.


For a select operation, this method will wait until the first row can be accessed locally.
For an insert operation, this method will wait for the job to finish, because the result contains only one row.
For other operations, this method will return immediately, because the result is already available locally.

twalthr · 2020-06-17T13:26:02Z

flink-python/pyflink/table/table_result.py

        else:
            return None

+    def wait(self, timeout_ms=None):


call it await to be consistent. how are the exceptions handled in Python?

await is a keyword in python.
just like as method in Table, python Table uses alias.

now flink-python does not define Exceptions, java exception will be thrown directly. example: python Catalog class

twalthr · 2020-06-17T13:26:56Z

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableResult.java

+	 * For insert operation, this method will wait for the job to finish, because the result contains only one row.
+	 * For other operations, this method will return immediately, because the result is ready in local.
+	 *
+	 * @throws ExecutionException if this future completed exceptionally


if this future completed exceptionally it is not really a future

twalthr · 2020-06-17T13:31:03Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

+	/**
+	 * Iterator for insert operation result.
+	 */
+	private static final class InsertResultIterator implements CloseableIterator<Row> {


TableEnvironmentImpl is already quite large. Move it to a separate class in default scope.

twalthr · 2020-06-17T13:38:29Z

.../flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java

+		}
+
+		ExecutorService executor = Executors.newFixedThreadPool(
+				1, new ThreadFactoryBuilder().setNameFormat("TableResult-await-thread").build());


avoid guava

godfreyhe · 2020-06-18T05:44:46Z

@twalthr Thanks for the suggestion, I have updated the pr

twalthr

Thanks for the update @godfreyhe. I added my last batch of comments. Good to merge afterwards. Can you rebase the PR?

twalthr · 2020-08-31T12:34:39Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultIterator.java

+import java.util.Optional;
+
+/**
+ * A CloseableIterator for insert operation result.


nit: use {@link CloseableIterator} in such cases to make refactorings easier in the future

twalthr · 2020-08-31T12:36:36Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultIterator.java

+class InsertResultIterator implements CloseableIterator<Row> {
+	private final JobClient jobClient;
+	private final Row affectedRowCountsRow;
+	private Optional<Boolean> hasNext = Optional.empty();


Optional should not be used in fields: https://flink.apache.org/contributing/code-style-and-quality-java.html#java-optional

twalthr · 2020-08-31T12:39:14Z

...k-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultIterator.java

+	public boolean hasNext() {
+		if (!hasNext.isPresent()) {
+			try {
+				jobClient.getJobExecutionResult(Thread.currentThread().getContextClassLoader()).get();


Make the classloader configurable for this class and pass the context classloader higher in the stack. The TableEnvironment will have it soon (see FLINK-15635).

twalthr · 2020-08-31T12:45:44Z

...planner-blink/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala

        |GROUP BY TUMBLE(ts, INTERVAL '5' SECOND)
      """.stripMargin
-    execInsertSqlAndWaitResult(tableEnv, query)
+      tableEnv.executeSql(query).await()


nit: wrong indention

…ata ready

…of guava Lists when collecting result

godfreyhe · 2020-09-03T08:19:10Z

@twalthr I have updated the pr based on latest comments

rmetzger added the review=description? label Jun 16, 2020

rmetzger added the component=TableSQL/API label Jun 16, 2020

godfreyhe force-pushed the FLINK-18337 branch 4 times, most recently from 51d8c4c to 5faf259 Compare June 17, 2020 06:25

twalthr requested changes Jun 17, 2020

View reviewed changes

twalthr requested changes Aug 31, 2020

View reviewed changes

twalthr mentioned this pull request Sep 1, 2020

[FLINK-19049][table] Fix validation of table functions in projections #13285

Closed

godfreyhe added 5 commits September 3, 2020 06:58

[FLINK-18337][table] Introduce TableResult#await method to wait for d…

3e80ce0

…ata ready

fix test and update python api

bdb7a44

Address Timo's comments

76a3b45

address Timo's comments

64f7e87

[hotfix][table][connector] Use CollectionUtil#iteratorToList instead …

9fc17e0

…of guava Lists when collecting result

godfreyhe force-pushed the FLINK-18337 branch from 80045af to 9fc17e0 Compare September 3, 2020 08:10

twalthr closed this in 6f2a041 Sep 7, 2020

godfreyhe deleted the FLINK-18337 branch September 7, 2020 09:43

[FLINK-18337][table] Introduce TableResult#await method to wait for data ready #12688

[FLINK-18337][table] Introduce TableResult#await method to wait for data ready #12688

Uh oh!

Conversation

godfreyhe commented Jun 16, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jun 16, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Jun 16, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

godfreyhe commented Jun 17, 2020

Uh oh!

twalthr left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

godfreyhe Jun 18, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

godfreyhe commented Jun 18, 2020

Uh oh!

twalthr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

godfreyhe commented Sep 3, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Jun 16, 2020 •

edited

Loading

twalthr left a comment •

edited

Loading

godfreyhe Jun 18, 2020 •

edited

Loading