[GitHub] [flink] twalthr commented on a change in pull request #12688: [FLINK-18337][table] Introduce TableResult#await method to wait for data ready

[GitBox <gi...@apache.org>]

twalthr commented on a change in pull request #12688:
URL: https://github.com/apache/flink/pull/12688#discussion_r441519689



##########
File path: flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java
##########
@@ -135,7 +126,7 @@ public void testTableSourceFullScan() throws Exception {
 			"  h.family3.col3 " +
 			"FROM hTable AS h");
 

Review comment:
       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.

##########
File path: flink-python/pyflink/table/table_result.py
##########
@@ -48,6 +51,22 @@ def get_job_client(self):
         else:
             return None
 
+    def wait(self, timeout_ms=None):

Review comment:
       call it `await` to be consistent. how are the exceptions handled in Python?

##########
File path: flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java
##########
@@ -135,7 +126,7 @@ public void testTableSourceFullScan() throws Exception {
 			"  h.family3.col3 " +
 			"FROM hTable AS h");
 
-		List<Row> results = collectBatchResult(table);
+		List<Row> results = Lists.newArrayList(table.execute().collect());

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

##########
File path: flink-python/pyflink/table/table_result.py
##########
@@ -48,6 +51,22 @@ def get_job_client(self):
         else:
             return None
 
+    def wait(self, timeout_ms=None):
+        """
+        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.

Review comment:
       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.

##########
File path: flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java
##########
@@ -284,9 +284,13 @@ public void testStreamingAppend() throws Exception {
 			tEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
 			tEnv.useCatalog(hiveCatalog.getName());
 
-			TableEnvUtil.execInsertSqlAndWaitResult(
-					tEnv,
-					"insert into db1.sink_table select 6,'a','b','2020-05-03','12'");
+			try {
+				tEnv.executeSql(
+						"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);

Review comment:
       throw `AssertionError` in tests with `Assert.fail()`.

##########
File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
##########
@@ -1413,4 +1415,46 @@ protected TableImpl createTable(QueryOperation tableOperation) {
 			operationTreeBuilder,
 			functionCatalog.asLookup(parser::parseIdentifier));
 	}
+
+	/**
+	 * Iterator for insert operation result.
+	 */
+	private static final class InsertResultIterator implements CloseableIterator<Row> {

Review comment:
       `TableEnvironmentImpl` is already quite large. Move it to a separate class in default scope.

##########
File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
##########
@@ -73,6 +82,50 @@ private TableResultImpl(
 		return Optional.ofNullable(jobClient);
 	}
 
+	@Override
+	public void await() throws InterruptedException, ExecutionException {
+		try {
+			awaitInternal(-1, TimeUnit.MILLISECONDS);
+		} catch (TimeoutException e) {
+			// do nothing
+		}
+	}
+
+	@Override
+	public void await(long timeout, TimeUnit unit)
+			throws InterruptedException, ExecutionException, TimeoutException {
+		awaitInternal(timeout, unit);
+	}
+
+	private void awaitInternal(long timeout, TimeUnit unit)
+			throws InterruptedException, ExecutionException, TimeoutException {
+		if (jobClient == null) {
+			return;
+		}
+
+		ExecutorService executor = Executors.newFixedThreadPool(
+				1, new ThreadFactoryBuilder().setNameFormat("TableResult-await-thread").build());

Review comment:
       avoid guava

##########
File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableResult.java
##########
@@ -38,6 +41,33 @@
 	 */
 	Optional<JobClient> getJobClient();
 
+	/**
+	 * Wait if necessary until the data is ready.
+	 *
+	 * <p>For select operation, this method will wait unit the first row can be accessed in local.
+	 * 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

Review comment:
       `if this future completed exceptionally` it is not really a future




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org