You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/10/19 12:08:55 UTC
[flink] branch master updated: [FLINK-24461][table] Let PrintUtils
accept only internal data structures
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ca164d1 [FLINK-24461][table] Let PrintUtils accept only internal data structures
ca164d1 is described below
commit ca164d1cfbb77944ca87c62c5a07bb99e54c87cd
Author: slinkydeveloper <fr...@gmail.com>
AuthorDate: Fri Oct 8 18:12:49 2021 +0200
[FLINK-24461][table] Let PrintUtils accept only internal data structures
Signed-off-by: slinkydeveloper <fr...@gmail.com>
This closes #17441.
---
.../table/client/cli/CliChangelogResultView.java | 8 +-
.../apache/flink/table/client/cli/CliClient.java | 5 +-
.../flink/table/client/cli/CliTableResultView.java | 4 +-
.../table/client/cli/CliTableauResultView.java | 16 +-
.../flink/table/client/gateway/Executor.java | 11 +-
.../table/client/gateway/local/LocalExecutor.java | 17 +-
.../table/client/gateway/local/ResultStore.java | 4 +-
.../local/result/ChangelogCollectResult.java | 16 +-
.../gateway/local/result/ChangelogResult.java | 4 +-
.../gateway/local/result/CollectResultBase.java | 11 +-
.../result/MaterializedCollectBatchResult.java | 17 +-
.../result/MaterializedCollectResultBase.java | 18 +-
.../result/MaterializedCollectStreamResult.java | 28 ++-
.../gateway/local/result/MaterializedResult.java | 4 +-
.../flink/table/client/cli/CliClientTest.java | 11 +-
.../flink/table/client/cli/CliResultViewTest.java | 17 +-
.../table/client/cli/CliTableauResultViewTest.java | 198 +++++++++++----------
.../flink/table/client/cli/TestingExecutor.java | 23 +--
.../table/client/cli/TestingExecutorBuilder.java | 10 +-
.../table/client/cli/utils/TestTableResult.java | 20 ++-
.../client/gateway/local/LocalExecutorITCase.java | 120 +++++++------
.../local/result/BaseMaterializedResultTest.java | 54 ++++++
.../local/result/ChangelogCollectResultTest.java | 3 +-
.../result/MaterializedCollectBatchResultTest.java | 122 ++++++++++---
.../MaterializedCollectStreamResultTest.java | 123 ++++++++++---
.../table/api/internal/InsertResultIterator.java | 72 --------
.../table/api/internal/InsertResultProvider.java | 113 ++++++++++++
...lectResultProvider.java => ResultProvider.java} | 28 ++-
.../table/api/internal/StaticResultProvider.java | 93 ++++++++++
.../table/api/internal/TableEnvironmentImpl.java | 49 +++--
.../api/internal/TableEnvironmentInternal.java | 4 +-
.../flink/table/api/internal/TableResultImpl.java | 95 +++-------
...esultProvider.java => TableResultInternal.java} | 28 +--
.../table/operations/CollectModifyOperation.java | 8 +-
.../org/apache/flink/table/utils/PrintUtils.java | 54 ++++--
.../flink/table/utils/TimestampStringUtils.java | 8 +
.../apache/flink/table/utils/PrintUtilsTest.java | 178 +++++++++++-------
.../planner/connectors/CollectDynamicSink.java | 112 +++++++++---
.../table/planner/connectors/DynamicSinkUtils.java | 6 +-
.../table/planner/delegation/PlannerBase.scala | 7 +-
40 files changed, 1102 insertions(+), 617 deletions(-)
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java
index 33f64fd..266f5d3 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java
@@ -22,8 +22,8 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.utils.PrintUtils;
-import org.apache.flink.types.Row;
import org.jline.keymap.KeyMap;
import org.jline.utils.AttributedString;
@@ -108,7 +108,7 @@ public class CliChangelogResultView
@Override
protected void refresh() {
// retrieve change record
- final TypedResult<List<Row>> result;
+ final TypedResult<List<RowData>> result;
try {
result =
client.getExecutor()
@@ -129,9 +129,9 @@ public class CliChangelogResultView
stopRetrieval(false);
break;
default:
- List<Row> changes = result.getPayload();
+ List<RowData> changes = result.getPayload();
- for (Row change : changes) {
+ for (RowData change : changes) {
// convert row
final String[] row =
PrintUtils.rowToString(
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
index 12dacb8..f6b1bf5 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliClient.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.client.cli;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.client.SqlClientException;
import org.apache.flink.table.client.config.ResultMode;
import org.apache.flink.table.client.config.SqlClientOptions;
@@ -622,7 +623,7 @@ public class CliClient implements AutoCloseable {
}
private void executeOperation(Operation operation) {
- TableResult result = executor.executeOperation(sessionId, operation);
+ TableResultInternal result = executor.executeOperation(sessionId, operation);
if (TABLE_RESULT_OK == result) {
// print more meaningful message than tableau OK result
printInfo(MESSAGE_EXECUTE_STATEMENT);
@@ -630,7 +631,7 @@ public class CliClient implements AutoCloseable {
// print tableau if result has content
PrintUtils.printAsTableauForm(
result.getResolvedSchema(),
- result.collect(),
+ result.collectInternal(),
terminal.writer(),
Integer.MAX_VALUE,
"",
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableResultView.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableResultView.java
index 1fa3e94..68a108f 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableResultView.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableResultView.java
@@ -22,8 +22,8 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.utils.PrintUtils;
-import org.apache.flink.types.Row;
import org.jline.keymap.KeyMap;
import org.jline.utils.AttributedString;
@@ -296,7 +296,7 @@ public class CliTableResultView extends CliResultView<CliTableResultView.ResultT
private void updatePage() {
// retrieve page
final int retrievalPage = page == LAST_PAGE ? pageCount : page;
- final List<Row> rows;
+ final List<RowData> rows;
try {
rows =
client.getExecutor()
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java
index 99838a5..3360e1b 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliTableauResultView.java
@@ -23,8 +23,8 @@ import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.utils.PrintUtils;
-import org.apache.flink.types.Row;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.jline.terminal.Terminal;
@@ -124,7 +124,7 @@ public class CliTableauResultView implements AutoCloseable {
}
private void printBatchResults(AtomicInteger receivedRowCount) {
- final List<Row> resultRows = waitBatchResults();
+ final List<RowData> resultRows = waitBatchResults();
receivedRowCount.addAndGet(resultRows.size());
PrintUtils.printAsTableauForm(
resultDescriptor.getResultSchema(),
@@ -157,7 +157,7 @@ public class CliTableauResultView implements AutoCloseable {
terminal.flush();
while (true) {
- final TypedResult<List<Row>> result =
+ final TypedResult<List<RowData>> result =
sqlExecutor.retrieveResultChanges(sessionId, resultDescriptor.getResultId());
switch (result.getType()) {
@@ -184,8 +184,8 @@ public class CliTableauResultView implements AutoCloseable {
terminal.flush();
return;
case PAYLOAD:
- List<Row> changes = result.getPayload();
- for (Row change : changes) {
+ List<RowData> changes = result.getPayload();
+ for (RowData change : changes) {
final String[] row =
PrintUtils.rowToString(
change,
@@ -203,15 +203,15 @@ public class CliTableauResultView implements AutoCloseable {
}
}
- private List<Row> waitBatchResults() {
- List<Row> resultRows = new ArrayList<>();
+ private List<RowData> waitBatchResults() {
+ List<RowData> resultRows = new ArrayList<>();
do {
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
- TypedResult<List<Row>> result =
+ TypedResult<List<RowData>> result =
sqlExecutor.retrieveResultChanges(sessionId, resultDescriptor.getResultId());
if (result.getType() == TypedResult.ResultType.EOS) {
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
index 7a20b14..d82a949 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/Executor.java
@@ -20,10 +20,11 @@ package org.apache.flink.table.client.gateway;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
-import org.apache.flink.types.Row;
import javax.annotation.Nullable;
@@ -106,11 +107,11 @@ public interface Executor {
List<String> completeStatement(String sessionId, String statement, int position);
/** Executes an operation, and return {@link TableResult} as execution result. */
- TableResult executeOperation(String sessionId, Operation operation)
+ TableResultInternal executeOperation(String sessionId, Operation operation)
throws SqlExecutionException;
/** Executes modify operations, and return {@link TableResult} as execution result. */
- TableResult executeModifyOperations(String sessionId, List<ModifyOperation> operations)
+ TableResultInternal executeModifyOperations(String sessionId, List<ModifyOperation> operations)
throws SqlExecutionException;
/** Submits a Flink SQL query job (detached) and returns the result descriptor. */
@@ -118,7 +119,7 @@ public interface Executor {
throws SqlExecutionException;
/** Asks for the next changelog results (non-blocking). */
- TypedResult<List<Row>> retrieveResultChanges(String sessionId, String resultId)
+ TypedResult<List<RowData>> retrieveResultChanges(String sessionId, String resultId)
throws SqlExecutionException;
/**
@@ -132,7 +133,7 @@ public interface Executor {
* Returns the rows that are part of the current page or throws an exception if the snapshot has
* been expired.
*/
- List<Row> retrieveResultPage(String resultId, int page) throws SqlExecutionException;
+ List<RowData> retrieveResultPage(String resultId, int page) throws SqlExecutionException;
/**
* Cancels a table program and stops the result retrieval. Blocking until cancellation command
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
index 2306488..f125570 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java
@@ -21,8 +21,8 @@ package org.apache.flink.table.client.gateway.local;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
@@ -33,11 +33,11 @@ import org.apache.flink.table.client.gateway.context.SessionContext;
import org.apache.flink.table.client.gateway.local.result.ChangelogResult;
import org.apache.flink.table.client.gateway.local.result.DynamicResult;
import org.apache.flink.table.client.gateway.local.result.MaterializedResult;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.delegation.Parser;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
-import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -200,7 +200,7 @@ public class LocalExecutor implements Executor {
}
@Override
- public TableResult executeOperation(String sessionId, Operation operation)
+ public TableResultInternal executeOperation(String sessionId, Operation operation)
throws SqlExecutionException {
final ExecutionContext context = getExecutionContext(sessionId);
final TableEnvironmentInternal tEnv =
@@ -213,8 +213,8 @@ public class LocalExecutor implements Executor {
}
@Override
- public TableResult executeModifyOperations(String sessionId, List<ModifyOperation> operations)
- throws SqlExecutionException {
+ public TableResultInternal executeModifyOperations(
+ String sessionId, List<ModifyOperation> operations) throws SqlExecutionException {
final ExecutionContext context = getExecutionContext(sessionId);
final TableEnvironmentInternal tEnv =
(TableEnvironmentInternal) context.getTableEnvironment();
@@ -228,7 +228,7 @@ public class LocalExecutor implements Executor {
@Override
public ResultDescriptor executeQuery(String sessionId, QueryOperation query)
throws SqlExecutionException {
- final TableResult tableResult = executeOperation(sessionId, query);
+ final TableResultInternal tableResult = executeOperation(sessionId, query);
final SessionContext context = getSessionContext(sessionId);
final ReadableConfig config = context.getReadableConfig();
final DynamicResult result = resultStore.createResult(config, tableResult);
@@ -241,7 +241,7 @@ public class LocalExecutor implements Executor {
}
@Override
- public TypedResult<List<Row>> retrieveResultChanges(String sessionId, String resultId)
+ public TypedResult<List<RowData>> retrieveResultChanges(String sessionId, String resultId)
throws SqlExecutionException {
final DynamicResult result = resultStore.getResult(resultId);
if (result == null) {
@@ -269,7 +269,8 @@ public class LocalExecutor implements Executor {
}
@Override
- public List<Row> retrieveResultPage(String resultId, int page) throws SqlExecutionException {
+ public List<RowData> retrieveResultPage(String resultId, int page)
+ throws SqlExecutionException {
final DynamicResult result = resultStore.getResult(resultId);
if (result == null) {
throw new SqlExecutionException(
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java
index b3d3f4f..a44de18 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java
@@ -20,7 +20,7 @@ package org.apache.flink.table.client.gateway.local;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.local.result.ChangelogCollectResult;
import org.apache.flink.table.client.gateway.local.result.DynamicResult;
@@ -50,7 +50,7 @@ public class ResultStore {
* Creates a result. Might start threads or opens sockets so every created result must be
* closed.
*/
- public DynamicResult createResult(ReadableConfig config, TableResult tableResult) {
+ public DynamicResult createResult(ReadableConfig config, TableResultInternal tableResult) {
// validate
if (config.get(EXECUTION_RESULT_MODE).equals(CHANGELOG)
&& config.get(RUNTIME_MODE).equals(RuntimeExecutionMode.BATCH)) {
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResult.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResult.java
index 4101a06..a123995 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResult.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResult.java
@@ -19,9 +19,9 @@
package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.client.gateway.TypedResult;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.data.RowData;
import java.util.ArrayList;
import java.util.List;
@@ -29,10 +29,10 @@ import java.util.List;
/** Collects results and returns them as a changelog. */
public class ChangelogCollectResult extends CollectResultBase implements ChangelogResult {
- private final List<Row> changeRecordBuffer;
+ private final List<RowData> changeRecordBuffer;
@VisibleForTesting protected static final int CHANGE_RECORD_BUFFER_SIZE = 5_000;
- public ChangelogCollectResult(TableResult tableResult) {
+ public ChangelogCollectResult(TableResultInternal tableResult) {
super(tableResult);
// prepare for changelog
changeRecordBuffer = new ArrayList<>();
@@ -45,7 +45,7 @@ public class ChangelogCollectResult extends CollectResultBase implements Changel
}
@Override
- public TypedResult<List<Row>> retrieveChanges() {
+ public TypedResult<List<RowData>> retrieveChanges() {
synchronized (resultLock) {
// retrieval thread is alive return a record if available
// but the program must not have failed
@@ -53,7 +53,7 @@ public class ChangelogCollectResult extends CollectResultBase implements Changel
if (changeRecordBuffer.isEmpty()) {
return TypedResult.empty();
} else {
- final List<Row> change = new ArrayList<>(changeRecordBuffer);
+ final List<RowData> change = new ArrayList<>(changeRecordBuffer);
changeRecordBuffer.clear();
resultLock.notify();
return TypedResult.payload(change);
@@ -61,7 +61,7 @@ public class ChangelogCollectResult extends CollectResultBase implements Changel
}
// retrieval thread is dead but there is still a record to be delivered
else if (!isRetrieving() && !changeRecordBuffer.isEmpty()) {
- final List<Row> change = new ArrayList<>(changeRecordBuffer);
+ final List<RowData> change = new ArrayList<>(changeRecordBuffer);
changeRecordBuffer.clear();
return TypedResult.payload(change);
}
@@ -75,7 +75,7 @@ public class ChangelogCollectResult extends CollectResultBase implements Changel
// --------------------------------------------------------------------------------------------
@Override
- protected void processRecord(Row row) {
+ protected void processRecord(RowData row) {
synchronized (resultLock) {
// wait if the buffer is full
if (changeRecordBuffer.size() >= CHANGE_RECORD_BUFFER_SIZE) {
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogResult.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogResult.java
index 58c517f..934b677 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogResult.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/ChangelogResult.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.table.client.gateway.TypedResult;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.data.RowData;
import java.util.List;
@@ -27,5 +27,5 @@ import java.util.List;
public interface ChangelogResult extends DynamicResult {
/** Retrieves the available result records. */
- TypedResult<List<Row>> retrieveChanges();
+ TypedResult<List<RowData>> retrieveChanges();
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectResultBase.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectResultBase.java
index 0ad6930..fef970d 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectResultBase.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectResultBase.java
@@ -19,23 +19,24 @@
package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.util.CloseableIterator;
import java.util.concurrent.atomic.AtomicReference;
/** A result that works through {@link TableResult#collect()}. */
public abstract class CollectResultBase implements DynamicResult {
- private final CloseableIterator<Row> result;
+ private final CloseableIterator<RowData> result;
protected final Object resultLock;
protected AtomicReference<SqlExecutionException> executionException = new AtomicReference<>();
protected final ResultRetrievalThread retrievalThread;
- public CollectResultBase(TableResult tableResult) {
- result = tableResult.collect();
+ public CollectResultBase(TableResultInternal tableResult) {
+ result = tableResult.collectInternal();
resultLock = new Object();
retrievalThread = new ResultRetrievalThread();
}
@@ -56,7 +57,7 @@ public abstract class CollectResultBase implements DynamicResult {
return TypedResult.endOfStream();
}
- protected abstract void processRecord(Row row);
+ protected abstract void processRecord(RowData row);
protected boolean isRetrieving() {
return retrievalThread.isRunning;
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java
index 2e1a451..d0975c3 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java
@@ -19,33 +19,26 @@
package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.table.api.TableResult;
-import org.apache.flink.types.Row;
-
-import java.util.List;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.data.RowData;
/** Collects results and returns them as table snapshots. */
public class MaterializedCollectBatchResult extends MaterializedCollectResultBase {
@VisibleForTesting
public MaterializedCollectBatchResult(
- TableResult tableResult, int maxRowCount, int overcommitThreshold) {
+ TableResultInternal tableResult, int maxRowCount, int overcommitThreshold) {
super(tableResult, maxRowCount, overcommitThreshold);
// start listener thread
retrievalThread.start();
}
- public MaterializedCollectBatchResult(TableResult tableResult, int maxRowCount) {
+ public MaterializedCollectBatchResult(TableResultInternal tableResult, int maxRowCount) {
this(tableResult, maxRowCount, computeMaterializedTableOvercommit(maxRowCount));
}
- @VisibleForTesting
- protected List<Row> getMaterializedTable() {
- return materializedTable;
- }
-
@Override
- protected void processRecord(Row row) {
+ protected void processRecord(RowData row) {
// limit the materialized table
if (materializedTable.size() - validRowPosition >= maxRowCount) {
cleanUp();
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectResultBase.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectResultBase.java
index ef3924b..d49a001 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectResultBase.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectResultBase.java
@@ -18,10 +18,11 @@
package org.apache.flink.table.client.gateway.local.result;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.data.RowData;
import java.util.ArrayList;
import java.util.List;
@@ -55,13 +56,13 @@ public abstract class MaterializedCollectResultBase extends CollectResultBase
* Materialized table that is continuously updated by inserts and deletes. Deletes at the
* beginning are lazily cleaned up when the threshold is reached.
*/
- protected final List<Row> materializedTable;
+ protected final List<RowData> materializedTable;
/** Counter for deleted rows to be deleted at the beginning of the materialized table. */
protected int validRowPosition;
/** Current snapshot of the materialized table. */
- private final List<Row> snapshot;
+ private final List<RowData> snapshot;
/** Page count of the snapshot (always >= 1). */
private int pageCount;
@@ -73,7 +74,7 @@ public abstract class MaterializedCollectResultBase extends CollectResultBase
private boolean isLastSnapshot;
public MaterializedCollectResultBase(
- TableResult tableResult, int maxRowCount, int overcommitThreshold) {
+ TableResultInternal tableResult, int maxRowCount, int overcommitThreshold) {
super(tableResult);
if (maxRowCount <= 0) {
@@ -131,7 +132,7 @@ public abstract class MaterializedCollectResultBase extends CollectResultBase
}
@Override
- public List<Row> retrievePage(int page) {
+ public List<RowData> retrievePage(int page) {
synchronized (resultLock) {
if (page <= 0 || page > pageCount) {
throw new SqlExecutionException("Invalid page '" + page + "'.");
@@ -153,4 +154,9 @@ public abstract class MaterializedCollectResultBase extends CollectResultBase
MATERIALIZED_TABLE_MAX_OVERCOMMIT,
(int) (maxRowCount * MATERIALIZED_TABLE_OVERCOMMIT_FACTOR));
}
+
+ @VisibleForTesting
+ protected List<RowData> getMaterializedTable() {
+ return materializedTable;
+ }
}
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java
index 9e26659..86abd03 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java
@@ -19,12 +19,11 @@
package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.table.api.TableResult;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.api.internal.TableResultInternal;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
/** Collects results and returns them as table snapshots. */
@@ -35,11 +34,11 @@ public class MaterializedCollectStreamResult extends MaterializedCollectResultBa
* smaller position are deleted) nor complete (for deletes of duplicates). However, the cache
* narrows the search in the materialized table.
*/
- private final Map<Row, Integer> rowPositionCache;
+ private final Map<RowData, Integer> rowPositionCache;
@VisibleForTesting
public MaterializedCollectStreamResult(
- TableResult tableResult, int maxRowCount, int overcommitThreshold) {
+ TableResultInternal tableResult, int maxRowCount, int overcommitThreshold) {
super(tableResult, maxRowCount, overcommitThreshold);
final int initialCapacity =
@@ -49,20 +48,20 @@ public class MaterializedCollectStreamResult extends MaterializedCollectResultBa
retrievalThread.start();
}
- public MaterializedCollectStreamResult(TableResult tableResult, int maxRowCount) {
+ public MaterializedCollectStreamResult(TableResultInternal tableResult, int maxRowCount) {
this(tableResult, maxRowCount, computeMaterializedTableOvercommit(maxRowCount));
}
// --------------------------------------------------------------------------------------------
@Override
- protected void processRecord(Row row) {
+ protected void processRecord(RowData row) {
synchronized (resultLock) {
boolean isInsertOp =
- row.getKind() == RowKind.INSERT || row.getKind() == RowKind.UPDATE_AFTER;
+ row.getRowKind() == RowKind.INSERT || row.getRowKind() == RowKind.UPDATE_AFTER;
// Always set the RowKind to INSERT, so that we can compare rows correctly (RowKind will
// be ignored),
- row.setKind(RowKind.INSERT);
+ row.setRowKind(RowKind.INSERT);
// insert
if (isInsertOp) {
@@ -75,14 +74,9 @@ public class MaterializedCollectStreamResult extends MaterializedCollectResultBa
}
}
- @VisibleForTesting
- protected List<Row> getMaterializedTable() {
- return materializedTable;
- }
-
// --------------------------------------------------------------------------------------------
- private void processInsert(Row row) {
+ private void processInsert(RowData row) {
// limit the materialized table
if (materializedTable.size() - validRowPosition >= maxRowCount) {
cleanUp();
@@ -91,7 +85,7 @@ public class MaterializedCollectStreamResult extends MaterializedCollectResultBa
rowPositionCache.put(row, materializedTable.size() - 1);
}
- private void processDelete(Row row) {
+ private void processDelete(RowData row) {
// delete the newest record first to minimize per-page changes
final Integer cachedPos = rowPositionCache.get(row);
final int startSearchPos;
@@ -112,7 +106,7 @@ public class MaterializedCollectStreamResult extends MaterializedCollectResultBa
private void cleanUp() {
// invalidate row
- final Row deleteRow = materializedTable.get(validRowPosition);
+ final RowData deleteRow = materializedTable.get(validRowPosition);
if (rowPositionCache.get(deleteRow) == validRowPosition) {
// this row has no duplicates in the materialized table,
// it can be removed from the cache
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedResult.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedResult.java
index bb17996..ce3bf7f 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedResult.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedResult.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.table.client.gateway.TypedResult;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.data.RowData;
import java.util.List;
@@ -33,5 +33,5 @@ public interface MaterializedResult extends DynamicResult {
TypedResult<Integer> snapshot(int pageSize);
/** Retrieves a page of a snapshotted result. */
- List<Row> retrievePage(int page);
+ List<RowData> retrievePage(int page);
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
index 759433b..aea05de 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java
@@ -26,7 +26,7 @@ import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.streaming.environment.TestingJobClient;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.cli.utils.SqlParserHelper;
@@ -37,6 +37,7 @@ import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
import org.apache.flink.table.client.gateway.context.DefaultContext;
import org.apache.flink.table.client.gateway.context.SessionContext;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
@@ -465,7 +466,7 @@ public class CliClientTest extends TestLogger {
}
@Override
- public TableResult executeOperation(String sessionId, Operation operation)
+ public TableResultInternal executeOperation(String sessionId, Operation operation)
throws SqlExecutionException {
if (failExecution) {
throw new SqlExecutionException("Fail execution.");
@@ -490,7 +491,7 @@ public class CliClientTest extends TestLogger {
}
@Override
- public TableResult executeModifyOperations(
+ public TableResultInternal executeModifyOperations(
String sessionId, List<ModifyOperation> operations) throws SqlExecutionException {
if (failExecution) {
throw new SqlExecutionException("Fail execution.");
@@ -545,7 +546,7 @@ public class CliClientTest extends TestLogger {
}
@Override
- public TypedResult<List<Row>> retrieveResultChanges(String sessionId, String resultId)
+ public TypedResult<List<RowData>> retrieveResultChanges(String sessionId, String resultId)
throws SqlExecutionException {
return null;
}
@@ -557,7 +558,7 @@ public class CliClientTest extends TestLogger {
}
@Override
- public List<Row> retrieveResultPage(String resultId, int page)
+ public List<RowData> retrieveResultPage(String resultId, int page)
throws SqlExecutionException {
return null;
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
index eaa520c..3f19971 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliResultViewTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.config.ResultMode;
@@ -30,10 +30,11 @@ import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
-import org.apache.flink.types.Row;
import org.jline.reader.MaskingCallback;
import org.jline.terminal.Terminal;
@@ -191,13 +192,13 @@ public class CliResultViewTest {
}
@Override
- public TableResult executeOperation(String sessionId, Operation operation)
+ public TableResultInternal executeOperation(String sessionId, Operation operation)
throws SqlExecutionException {
return null;
}
@Override
- public TableResult executeModifyOperations(
+ public TableResultInternal executeModifyOperations(
String sessionId, List<ModifyOperation> operations) throws SqlExecutionException {
return null;
}
@@ -210,9 +211,9 @@ public class CliResultViewTest {
@Override
@SuppressWarnings("unchecked")
- public TypedResult<List<Row>> retrieveResultChanges(String sessionId, String resultId)
+ public TypedResult<List<RowData>> retrieveResultChanges(String sessionId, String resultId)
throws SqlExecutionException {
- return (TypedResult<List<Row>>) typedResult;
+ return (TypedResult<List<RowData>>) typedResult;
}
@Override
@@ -223,9 +224,9 @@ public class CliResultViewTest {
}
@Override
- public List<Row> retrieveResultPage(String resultId, int page)
+ public List<RowData> retrieveResultPage(String resultId, int page)
throws SqlExecutionException {
- return Collections.singletonList(new Row(1));
+ return Collections.singletonList(new GenericRowData(1));
}
@Override
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
index 9c69e48..99d2b6c 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliTableauResultViewTest.java
@@ -29,6 +29,9 @@ import org.apache.flink.table.client.config.ResultMode;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.conversion.DataStructureConverter;
+import org.apache.flink.table.data.conversion.DataStructureConverters;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
@@ -41,12 +44,13 @@ import java.io.ByteArrayOutputStream;
import java.math.BigDecimal;
import java.sql.Timestamp;
import java.time.Duration;
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE;
import static org.apache.flink.table.client.config.SqlClientOptions.EXECUTION_RESULT_MODE;
@@ -59,8 +63,8 @@ public class CliTableauResultViewTest {
private ByteArrayOutputStream terminalOutput;
private Terminal terminal;
private ResolvedSchema schema;
- private List<Row> data;
- private List<Row> streamingData;
+ private List<RowData> data;
+ private List<RowData> streamingData;
@Before
public void setUp() {
@@ -74,89 +78,87 @@ public class CliTableauResultViewTest {
Column.physical("bigint", DataTypes.BIGINT()),
Column.physical("varchar", DataTypes.STRING()),
Column.physical("decimal(10, 5)", DataTypes.DECIMAL(10, 5)),
- Column.physical("timestamp", DataTypes.TIMESTAMP(6)));
-
- data = new ArrayList<>();
- data.add(
- Row.ofKind(
- RowKind.INSERT,
- null,
- 1,
- 2,
- "abc",
- BigDecimal.valueOf(1.23),
- Timestamp.valueOf("2020-03-01 18:39:14")));
- data.add(
- Row.ofKind(
- RowKind.UPDATE_BEFORE,
- false,
- null,
- 0,
- "",
- BigDecimal.valueOf(1),
- Timestamp.valueOf("2020-03-01 18:39:14.1")));
- data.add(
- Row.ofKind(
- RowKind.UPDATE_AFTER,
- true,
- Integer.MAX_VALUE,
- null,
- "abcdefg",
- BigDecimal.valueOf(1234567890),
- Timestamp.valueOf("2020-03-01 18:39:14.12")));
- data.add(
- Row.ofKind(
- RowKind.DELETE,
- false,
- Integer.MIN_VALUE,
- Long.MAX_VALUE,
- null,
- BigDecimal.valueOf(12345.06789),
- Timestamp.valueOf("2020-03-01 18:39:14.123")));
- data.add(
- Row.ofKind(
- RowKind.INSERT,
- true,
- 100,
- Long.MIN_VALUE,
- "abcdefg111",
- null,
- Timestamp.valueOf("2020-03-01 18:39:14.123456")));
- data.add(
- Row.ofKind(
- RowKind.DELETE,
- null,
- -1,
- -1,
- "abcdefghijklmnopqrstuvwxyz",
- BigDecimal.valueOf(-12345.06789),
- null));
-
- data.add(
- Row.ofKind(
- RowKind.INSERT,
- null,
- -1,
- -1,
- "这是一段中文",
- BigDecimal.valueOf(-12345.06789),
- Timestamp.valueOf("2020-03-04 18:39:14")));
-
- data.add(
- Row.ofKind(
- RowKind.DELETE,
- null,
- -1,
- -1,
- "これは日本語をテストするための文です",
- BigDecimal.valueOf(-12345.06789),
- Timestamp.valueOf("2020-03-04 18:39:14")));
-
- streamingData = new ArrayList<>();
- for (Row datum : data) {
- Row row = Row.copy(datum);
- streamingData.add(row);
- }
+ Column.physical(
+ "timestamp", DataTypes.TIMESTAMP(6).bridgedTo(Timestamp.class)));
+
+ List<Row> rows =
+ Arrays.asList(
+ Row.ofKind(
+ RowKind.INSERT,
+ null,
+ 1,
+ 2L,
+ "abc",
+ BigDecimal.valueOf(1.23),
+ Timestamp.valueOf("2020-03-01 18:39:14")),
+ Row.ofKind(
+ RowKind.UPDATE_BEFORE,
+ false,
+ null,
+ 0L,
+ "",
+ BigDecimal.valueOf(1),
+ Timestamp.valueOf("2020-03-01 18:39:14.1")),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ true,
+ Integer.MAX_VALUE,
+ null,
+ "abcdefg",
+ BigDecimal.valueOf(12345),
+ Timestamp.valueOf("2020-03-01 18:39:14.12")),
+ Row.ofKind(
+ RowKind.DELETE,
+ false,
+ Integer.MIN_VALUE,
+ Long.MAX_VALUE,
+ null,
+ BigDecimal.valueOf(12345.06789),
+ Timestamp.valueOf("2020-03-01 18:39:14.123")),
+ Row.ofKind(
+ RowKind.INSERT,
+ true,
+ 100,
+ Long.MIN_VALUE,
+ "abcdefg111",
+ null,
+ Timestamp.valueOf("2020-03-01 18:39:14.123456")),
+ Row.ofKind(
+ RowKind.DELETE,
+ null,
+ -1,
+ -1L,
+ "abcdefghijklmnopqrstuvwxyz",
+ BigDecimal.valueOf(-12345.06789),
+ null),
+ Row.ofKind(
+ RowKind.INSERT,
+ null,
+ -1,
+ -1L,
+ "这是一段中文",
+ BigDecimal.valueOf(-12345.06789),
+ Timestamp.valueOf("2020-03-04 18:39:14")),
+ Row.ofKind(
+ RowKind.DELETE,
+ null,
+ -1,
+ -1L,
+ "これは日本語をテストするための文です",
+ BigDecimal.valueOf(-12345.06789),
+ Timestamp.valueOf("2020-03-04 18:39:14")));
+
+ final DataStructureConverter<Object, Object> dataStructureConverter =
+ DataStructureConverters.getConverter(schema.toPhysicalRowDataType());
+
+ data =
+ rows.stream()
+ .map(r -> (RowData) (dataStructureConverter.toInternal(r)))
+ .collect(Collectors.toList());
+ streamingData =
+ rows.stream()
+ .map(r -> (RowData) (dataStructureConverter.toInternal(r)))
+ .collect(Collectors.toList());
}
@Test
@@ -188,11 +190,11 @@ public class CliTableauResultViewTest {
+ System.lineSeparator()
+ "+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+"
+ System.lineSeparator()
- + "| (NULL) | 1 | 2 | abc | 1.23 | 2020-03-01 18:39:14.000000 |"
+ + "| (NULL) | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 |"
+ System.lineSeparator()
- + "| false | (NULL) | 0 | | 1 | 2020-03-01 18:39:14.100000 |"
+ + "| false | (NULL) | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 |"
+ System.lineSeparator()
- + "| true | 2147483647 | (NULL) | abcdefg | 1234567890 | 2020-03-01 18:39:14.120000 |"
+ + "| true | 2147483647 | (NULL) | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 |"
+ System.lineSeparator()
+ "| false | -2147483648 | 9223372036854775807 | (NULL) | 12345.06789 | 2020-03-01 18:39:14.123000 |"
+ System.lineSeparator()
@@ -349,11 +351,11 @@ public class CliTableauResultViewTest {
+ System.lineSeparator()
+ "+----+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+"
+ System.lineSeparator()
- + "| +I | (NULL) | 1 | 2 | abc | 1.23 | 2020-03-01 18:39:14.000000 |"
+ + "| +I | (NULL) | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 |"
+ System.lineSeparator()
- + "| -U | false | (NULL) | 0 | | 1 | 2020-03-01 18:39:14.100000 |"
+ + "| -U | false | (NULL) | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 |"
+ System.lineSeparator()
- + "| +U | true | 2147483647 | (NULL) | abcdefg | 1234567890 | 2020-03-01 18:39:14.120000 |"
+ + "| +U | true | 2147483647 | (NULL) | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 |"
+ System.lineSeparator()
+ "| -D | false | -2147483648 | 9223372036854775807 | (NULL) | 12345.06789 | 2020-03-01 18:39:14.123000 |"
+ System.lineSeparator()
@@ -445,11 +447,11 @@ public class CliTableauResultViewTest {
+ System.lineSeparator()
+ "+----+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+"
+ System.lineSeparator()
- + "| +I | (NULL) | 1 | 2 | abc | 1.23 | 2020-03-01 18:39:14.000000 |"
+ + "| +I | (NULL) | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 |"
+ System.lineSeparator()
- + "| -U | false | (NULL) | 0 | | 1 | 2020-03-01 18:39:14.100000 |"
+ + "| -U | false | (NULL) | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 |"
+ System.lineSeparator()
- + "| +U | true | 2147483647 | (NULL) | abcdefg | 1234567890 | 2020-03-01 18:39:14.120000 |"
+ + "| +U | true | 2147483647 | (NULL) | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 |"
+ System.lineSeparator()
+ "| -D | false | -2147483648 | 9223372036854775807 | (NULL) | 12345.06789 | 2020-03-01 18:39:14.123000 |"
+ System.lineSeparator()
@@ -496,11 +498,11 @@ public class CliTableauResultViewTest {
+ System.lineSeparator()
+ "+----+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+"
+ System.lineSeparator()
- + "| +I | (NULL) | 1 | 2 | abc | 1.23 | 2020-03-01 18:39:14.000000 |"
+ + "| +I | (NULL) | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 |"
+ System.lineSeparator()
- + "| -U | false | (NULL) | 0 | | 1 | 2020-03-01 18:39:14.100000 |"
+ + "| -U | false | (NULL) | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 |"
+ System.lineSeparator()
- + "| +U | true | 2147483647 | (NULL) | abcdefg | 1234567890 | 2020-03-01 18:39:14.120000 |"
+ + "| +U | true | 2147483647 | (NULL) | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 |"
+ System.lineSeparator()
+ "| -D | false | -2147483648 | 9223372036854775807 | (NULL) | 12345.06789 | 2020-03-01 18:39:14.123000 |"
+ System.lineSeparator(),
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
index 26542b6..759db8f 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutor.java
@@ -20,16 +20,16 @@ package org.apache.flink.table.client.cli;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.client.cli.utils.SqlParserHelper;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
-import org.apache.flink.types.Row;
import org.apache.flink.util.function.SupplierWithException;
import javax.annotation.Nullable;
@@ -44,18 +44,18 @@ class TestingExecutor implements Executor {
private int numCancelCalls = 0;
private int numRetrieveResultChancesCalls = 0;
- private final List<SupplierWithException<TypedResult<List<Row>>, SqlExecutionException>>
+ private final List<SupplierWithException<TypedResult<List<RowData>>, SqlExecutionException>>
resultChanges;
private int numRetrieveResultPageCalls = 0;
- private final List<SupplierWithException<List<Row>, SqlExecutionException>> resultPages;
+ private final List<SupplierWithException<List<RowData>, SqlExecutionException>> resultPages;
private final SqlParserHelper helper;
TestingExecutor(
- List<SupplierWithException<TypedResult<List<Row>>, SqlExecutionException>>
+ List<SupplierWithException<TypedResult<List<RowData>>, SqlExecutionException>>
resultChanges,
- List<SupplierWithException<List<Row>, SqlExecutionException>> resultPages) {
+ List<SupplierWithException<List<RowData>, SqlExecutionException>> resultPages) {
this.resultChanges = resultChanges;
this.resultPages = resultPages;
helper = new SqlParserHelper();
@@ -83,7 +83,7 @@ class TestingExecutor implements Executor {
}
@Override
- public TypedResult<List<Row>> retrieveResultChanges(String sessionId, String resultId)
+ public TypedResult<List<RowData>> retrieveResultChanges(String sessionId, String resultId)
throws SqlExecutionException {
return resultChanges
.get(Math.min(numRetrieveResultChancesCalls++, resultChanges.size() - 1))
@@ -91,7 +91,8 @@ class TestingExecutor implements Executor {
}
@Override
- public List<Row> retrieveResultPage(String resultId, int page) throws SqlExecutionException {
+ public List<RowData> retrieveResultPage(String resultId, int page)
+ throws SqlExecutionException {
return resultPages
.get(Math.min(numRetrieveResultPageCalls++, resultPages.size() - 1))
.get();
@@ -146,14 +147,14 @@ class TestingExecutor implements Executor {
}
@Override
- public TableResult executeOperation(String sessionId, Operation operation)
+ public TableResultInternal executeOperation(String sessionId, Operation operation)
throws SqlExecutionException {
throw new UnsupportedOperationException("Not implemented.");
}
@Override
- public TableResult executeModifyOperations(String sessionId, List<ModifyOperation> operations)
- throws SqlExecutionException {
+ public TableResultInternal executeModifyOperations(
+ String sessionId, List<ModifyOperation> operations) throws SqlExecutionException {
throw new UnsupportedOperationException("Not implemented.");
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutorBuilder.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutorBuilder.java
index 536dd76..f1fdc35 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutorBuilder.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/TestingExecutorBuilder.java
@@ -19,7 +19,7 @@ package org.apache.flink.table.client.cli;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.util.function.SupplierWithException;
import java.util.Arrays;
@@ -29,14 +29,14 @@ import java.util.List;
/** Builder for {@link TestingExecutor}. */
class TestingExecutorBuilder {
- private List<SupplierWithException<TypedResult<List<Row>>, SqlExecutionException>>
+ private List<SupplierWithException<TypedResult<List<RowData>>, SqlExecutionException>>
resultChangesSupplier = Collections.emptyList();
- private List<SupplierWithException<List<Row>, SqlExecutionException>> resultPagesSupplier =
+ private List<SupplierWithException<List<RowData>, SqlExecutionException>> resultPagesSupplier =
Collections.emptyList();
@SafeVarargs
public final TestingExecutorBuilder setResultChangesSupplier(
- SupplierWithException<TypedResult<List<Row>>, SqlExecutionException>...
+ SupplierWithException<TypedResult<List<RowData>>, SqlExecutionException>...
resultChangesSupplier) {
this.resultChangesSupplier = Arrays.asList(resultChangesSupplier);
return this;
@@ -44,7 +44,7 @@ class TestingExecutorBuilder {
@SafeVarargs
public final TestingExecutorBuilder setResultPageSupplier(
- SupplierWithException<List<Row>, SqlExecutionException>... resultPageSupplier) {
+ SupplierWithException<List<RowData>, SqlExecutionException>... resultPageSupplier) {
resultPagesSupplier = Arrays.asList(resultPageSupplier);
return this;
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/utils/TestTableResult.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/utils/TestTableResult.java
index 7f35c64..392acfe 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/utils/TestTableResult.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/utils/TestTableResult.java
@@ -22,17 +22,23 @@ import org.apache.flink.core.execution.JobClient;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.conversion.DataStructureConverter;
+import org.apache.flink.table.data.conversion.DataStructureConverters;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import java.util.Collections;
import java.util.Optional;
+import java.util.Spliterators;
import java.util.concurrent.TimeUnit;
+import java.util.stream.StreamSupport;
/** {@link TableResult} for testing. */
-public class TestTableResult implements TableResult {
+public class TestTableResult implements TableResultInternal {
private final JobClient jobClient;
private final ResolvedSchema resolvedSchema;
private final ResultKind resultKind;
@@ -97,4 +103,16 @@ public class TestTableResult implements TableResult {
public void print() {
// do nothing
}
+
+ @Override
+ public CloseableIterator<RowData> collectInternal() {
+ DataStructureConverter<Object, Object> converter =
+ DataStructureConverters.getConverter(resolvedSchema.toPhysicalRowDataType());
+ converter.open(TestTableResult.class.getClassLoader());
+ return CloseableIterator.adapterForIterator(
+ StreamSupport.stream(Spliterators.spliteratorUnknownSize(data, 0), false)
+ .map(row -> (RowData) converter.toInternalOrNull(row))
+ .iterator(),
+ data);
+ }
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
index 8e3cc18..0f53d3e 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
@@ -29,6 +29,7 @@ import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.config.ResultMode;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.table.client.gateway.ResultDescriptor;
@@ -36,13 +37,15 @@ import org.apache.flink.table.client.gateway.TypedResult;
import org.apache.flink.table.client.gateway.context.DefaultContext;
import org.apache.flink.table.client.gateway.utils.TestUserClassLoaderJar;
import org.apache.flink.table.client.gateway.utils.UserDefinedFunctions.TableUDF;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
+import org.apache.flink.table.utils.PrintUtils;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.test.util.TestBaseUtils;
-import org.apache.flink.types.Row;
+import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TestLogger;
import org.junit.BeforeClass;
@@ -55,6 +58,7 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
import java.net.URL;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -170,15 +174,16 @@ public class LocalExecutorITCase extends TestLogger {
assertFalse(desc.isMaterialized());
final List<String> actualResults =
- retrieveChangelogResult(executor, sessionId, desc.getResultId());
+ retrieveChangelogResult(
+ executor, sessionId, desc.getResultId(), desc.getResultSchema());
final List<String> expectedResults = new ArrayList<>();
- expectedResults.add("+I[47, Hello World, ABC]");
- expectedResults.add("+I[27, Hello World, ABC]");
- expectedResults.add("+I[37, Hello World, ABC]");
- expectedResults.add("+I[37, Hello World, ABC]");
- expectedResults.add("+I[47, Hello World, ABC]");
- expectedResults.add("+I[57, Hello World!!!!, ABC]");
+ expectedResults.add("[47, Hello World, ABC]");
+ expectedResults.add("[27, Hello World, ABC]");
+ expectedResults.add("[37, Hello World, ABC]");
+ expectedResults.add("[37, Hello World, ABC]");
+ expectedResults.add("[47, Hello World, ABC]");
+ expectedResults.add("[57, Hello World!!!!, ABC]");
TestBaseUtils.compareResultCollections(
expectedResults, actualResults, Comparator.naturalOrder());
@@ -202,12 +207,12 @@ public class LocalExecutorITCase extends TestLogger {
assertEquals("test-session", sessionId);
final List<String> expectedResults = new ArrayList<>();
- expectedResults.add("+I[47, Hello World]");
- expectedResults.add("+I[27, Hello World]");
- expectedResults.add("+I[37, Hello World]");
- expectedResults.add("+I[37, Hello World]");
- expectedResults.add("+I[47, Hello World]");
- expectedResults.add("+I[57, Hello World!!!!]");
+ expectedResults.add("[47, Hello World]");
+ expectedResults.add("[27, Hello World]");
+ expectedResults.add("[37, Hello World]");
+ expectedResults.add("[37, Hello World]");
+ expectedResults.add("[47, Hello World]");
+ expectedResults.add("[57, Hello World!!!!]");
initSession(executor, sessionId, replaceVars);
try {
@@ -222,7 +227,8 @@ public class LocalExecutorITCase extends TestLogger {
assertFalse(desc.isMaterialized());
final List<String> actualResults =
- retrieveChangelogResult(executor, sessionId, desc.getResultId());
+ retrieveChangelogResult(
+ executor, sessionId, desc.getResultId(), desc.getResultSchema());
TestBaseUtils.compareResultCollections(
expectedResults, actualResults, Comparator.naturalOrder());
@@ -247,12 +253,12 @@ public class LocalExecutorITCase extends TestLogger {
"SELECT scalarUDF(IntegerField1, 5), StringField1, 'ABC' FROM TableNumber1";
final List<String> expectedResults = new ArrayList<>();
- expectedResults.add("+I[47, Hello World, ABC]");
- expectedResults.add("+I[27, Hello World, ABC]");
- expectedResults.add("+I[37, Hello World, ABC]");
- expectedResults.add("+I[37, Hello World, ABC]");
- expectedResults.add("+I[47, Hello World, ABC]");
- expectedResults.add("+I[57, Hello World!!!!, ABC]");
+ expectedResults.add("[47, Hello World, ABC]");
+ expectedResults.add("[27, Hello World, ABC]");
+ expectedResults.add("[37, Hello World, ABC]");
+ expectedResults.add("[37, Hello World, ABC]");
+ expectedResults.add("[47, Hello World, ABC]");
+ expectedResults.add("[57, Hello World!!!!, ABC]");
executeStreamQueryTable(replaceVars, configMap, query, expectedResults);
}
@@ -271,12 +277,12 @@ public class LocalExecutorITCase extends TestLogger {
final String query = "SELECT scalarUDF(IntegerField1, 5), StringField1 FROM TableNumber1";
final List<String> expectedResults = new ArrayList<>();
- expectedResults.add("+I[47, Hello World]");
- expectedResults.add("+I[27, Hello World]");
- expectedResults.add("+I[37, Hello World]");
- expectedResults.add("+I[37, Hello World]");
- expectedResults.add("+I[47, Hello World]");
- expectedResults.add("+I[57, Hello World!!!!]");
+ expectedResults.add("[47, Hello World]");
+ expectedResults.add("[27, Hello World]");
+ expectedResults.add("[37, Hello World]");
+ expectedResults.add("[37, Hello World]");
+ expectedResults.add("[47, Hello World]");
+ expectedResults.add("[57, Hello World!!!!]");
for (int i = 0; i < 3; i++) {
executeStreamQueryTable(replaceVars, configMap, query, expectedResults);
@@ -299,7 +305,7 @@ public class LocalExecutorITCase extends TestLogger {
"SELECT COUNT(*), StringField1 FROM TableNumber1 GROUP BY StringField1";
final List<String> expectedResults = new ArrayList<>();
- expectedResults.add("+I[1, Hello World!!!!]");
+ expectedResults.add("[1, Hello World!!!!]");
executeStreamQueryTable(replaceVars, configMap, query, expectedResults);
}
@@ -329,15 +335,16 @@ public class LocalExecutorITCase extends TestLogger {
assertTrue(desc.isMaterialized());
final List<String> actualResults =
- retrieveTableResult(executor, sessionId, desc.getResultId());
+ retrieveTableResult(
+ executor, sessionId, desc.getResultId(), desc.getResultSchema());
final List<String> expectedResults = new ArrayList<>();
- expectedResults.add("+I[47, ABC]");
- expectedResults.add("+I[27, ABC]");
- expectedResults.add("+I[37, ABC]");
- expectedResults.add("+I[37, ABC]");
- expectedResults.add("+I[47, ABC]");
- expectedResults.add("+I[57, ABC]");
+ expectedResults.add("[47, ABC]");
+ expectedResults.add("[27, ABC]");
+ expectedResults.add("[37, ABC]");
+ expectedResults.add("[37, ABC]");
+ expectedResults.add("[47, ABC]");
+ expectedResults.add("[57, ABC]");
TestBaseUtils.compareResultCollections(
expectedResults, actualResults, Comparator.naturalOrder());
@@ -365,12 +372,12 @@ public class LocalExecutorITCase extends TestLogger {
initSession(executor, sessionId, replaceVars);
final List<String> expectedResults = new ArrayList<>();
- expectedResults.add("+I[47]");
- expectedResults.add("+I[27]");
- expectedResults.add("+I[37]");
- expectedResults.add("+I[37]");
- expectedResults.add("+I[47]");
- expectedResults.add("+I[57]");
+ expectedResults.add("[47]");
+ expectedResults.add("[27]");
+ expectedResults.add("[37]");
+ expectedResults.add("[37]");
+ expectedResults.add("[47]");
+ expectedResults.add("[57]");
try {
for (int i = 0; i < 3; i++) {
@@ -380,7 +387,8 @@ public class LocalExecutorITCase extends TestLogger {
assertTrue(desc.isMaterialized());
final List<String> actualResults =
- retrieveTableResult(executor, sessionId, desc.getResultId());
+ retrieveTableResult(
+ executor, sessionId, desc.getResultId(), desc.getResultSchema());
TestBaseUtils.compareResultCollections(
expectedResults, actualResults, Comparator.naturalOrder());
@@ -445,7 +453,8 @@ public class LocalExecutorITCase extends TestLogger {
assertTrue(desc.isMaterialized());
final List<String> actualResults =
- retrieveTableResult(executor, sessionId, desc.getResultId());
+ retrieveTableResult(
+ executor, sessionId, desc.getResultId(), desc.getResultSchema());
TestBaseUtils.compareResultCollections(
expectedResults, actualResults, Comparator.naturalOrder());
@@ -454,7 +463,8 @@ public class LocalExecutorITCase extends TestLogger {
}
}
- private List<String> retrieveTableResult(Executor executor, String sessionId, String resultID)
+ private List<String> retrieveTableResult(
+ Executor executor, String sessionId, String resultID, ResolvedSchema resultSchema)
throws InterruptedException {
final List<String> actualResults = new ArrayList<>();
@@ -466,8 +476,14 @@ public class LocalExecutorITCase extends TestLogger {
IntStream.rangeClosed(1, result.getPayload())
.forEach(
(page) -> {
- for (Row row : executor.retrieveResultPage(resultID, page)) {
- actualResults.add(row.toString());
+ for (RowData row :
+ executor.retrieveResultPage(resultID, page)) {
+ actualResults.add(
+ StringUtils.arrayAwareToString(
+ PrintUtils.rowToString(
+ row,
+ resultSchema,
+ ZoneId.systemDefault())));
}
});
} else if (result.getType() == TypedResult.ResultType.EOS) {
@@ -479,16 +495,20 @@ public class LocalExecutorITCase extends TestLogger {
}
private List<String> retrieveChangelogResult(
- Executor executor, String sessionId, String resultID) throws InterruptedException {
+ Executor executor, String sessionId, String resultID, ResolvedSchema resultSchema)
+ throws InterruptedException {
final List<String> actualResults = new ArrayList<>();
while (true) {
Thread.sleep(50); // slow the processing down
- final TypedResult<List<Row>> result =
+ final TypedResult<List<RowData>> result =
executor.retrieveResultChanges(sessionId, resultID);
if (result.getType() == TypedResult.ResultType.PAYLOAD) {
- for (Row row : result.getPayload()) {
- actualResults.add(row.toString());
+ for (RowData row : result.getPayload()) {
+ actualResults.add(
+ StringUtils.arrayAwareToString(
+ PrintUtils.rowToString(
+ row, resultSchema, ZoneId.systemDefault())));
}
} else if (result.getType() == TypedResult.ResultType.EOS) {
break;
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/BaseMaterializedResultTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/BaseMaterializedResultTest.java
new file mode 100644
index 0000000..8b5c9e1
--- /dev/null
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/BaseMaterializedResultTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.client.gateway.local.result;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.conversion.DataStructureConverter;
+import org.apache.flink.table.data.conversion.DataStructureConverters;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class BaseMaterializedResultTest {
+
+ static Function<Row, BinaryRowData> createInternalBinaryRowDataConverter(DataType dataType) {
+ DataStructureConverter<Object, Object> converter =
+ DataStructureConverters.getConverter(dataType);
+ RowDataSerializer serializer = new RowDataSerializer((RowType) dataType.getLogicalType());
+
+ return row -> serializer.toBinaryRow((RowData) converter.toInternalOrNull(row)).copy();
+ }
+
+ static void assertRowEquals(
+ List<Row> expected,
+ List<RowData> actual,
+ DataStructureConverter<RowData, Row> converter) {
+ assertEquals(
+ expected,
+ actual.stream().map(converter::toExternalOrNull).collect(Collectors.toList()));
+ }
+}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResultTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResultTest.java
index b5d3fd7..aafbbb7 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResultTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/ChangelogCollectResultTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.cli.utils.TestTableResult;
import org.apache.flink.table.client.gateway.SqlExecutionException;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
@@ -54,7 +55,7 @@ public class ChangelogCollectResultTest {
int count = 0;
boolean running = true;
while (running) {
- final TypedResult<List<Row>> result = changelogResult.retrieveChanges();
+ final TypedResult<List<RowData>> result = changelogResult.retrieveChanges();
Thread.sleep(100); // slow the processing down
switch (result.getType()) {
case EMPTY:
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResultTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResultTest.java
index c2facd5..25d63d7 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResultTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResultTest.java
@@ -20,10 +20,14 @@ package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.cli.utils.TestTableResult;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.conversion.DataStructureConverter;
+import org.apache.flink.table.data.conversion.DataStructureConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
@@ -31,23 +35,30 @@ import org.junit.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.function.Function;
import static org.junit.Assert.assertEquals;
/** Tests for {@link MaterializedCollectBatchResult}. */
-public class MaterializedCollectBatchResultTest {
+public class MaterializedCollectBatchResultTest extends BaseMaterializedResultTest {
@Test
public void testSnapshot() throws Exception {
- ResolvedSchema schema =
+ final ResolvedSchema schema =
ResolvedSchema.physical(
new String[] {"f0", "f1"},
- new DataType[] {DataTypes.STRING(), DataTypes.BIGINT()});
+ new DataType[] {DataTypes.STRING(), DataTypes.INT()});
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ final DataStructureConverter<RowData, Row> rowConverter =
+ (DataStructureConverter)
+ DataStructureConverters.getConverter(schema.toPhysicalRowDataType());
try (TestMaterializedCollectBatchResult result =
new TestMaterializedCollectBatchResult(
new TestTableResult(ResultKind.SUCCESS_WITH_CONTENT, schema),
- Integer.MAX_VALUE)) {
+ Integer.MAX_VALUE,
+ createInternalBinaryRowDataConverter(schema.toPhysicalRowDataType()))) {
result.isRetrieving = true;
@@ -58,35 +69,69 @@ public class MaterializedCollectBatchResultTest {
assertEquals(TypedResult.payload(4), result.snapshot(1));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(1));
- assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(2));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(3));
- assertEquals(Collections.singletonList(Row.of("C", 2)), result.retrievePage(4));
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(1),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("B", 1)),
+ result.retrievePage(2),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(3),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("C", 2)),
+ result.retrievePage(4),
+ rowConverter);
result.processRecord(Row.of("A", 1));
assertEquals(TypedResult.payload(5), result.snapshot(1));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(1));
- assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(2));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(3));
- assertEquals(Collections.singletonList(Row.of("C", 2)), result.retrievePage(4));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(5));
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(1),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("B", 1)),
+ result.retrievePage(2),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(3),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("C", 2)),
+ result.retrievePage(4),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(5),
+ rowConverter);
}
}
@Test
public void testLimitedSnapshot() throws Exception {
- ResolvedSchema schema =
+ final ResolvedSchema schema =
ResolvedSchema.physical(
new String[] {"f0", "f1"},
- new DataType[] {DataTypes.STRING(), DataTypes.BIGINT()});
+ new DataType[] {DataTypes.STRING(), DataTypes.INT()});
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ final DataStructureConverter<RowData, Row> rowConverter =
+ (DataStructureConverter)
+ DataStructureConverters.getConverter(schema.toPhysicalRowDataType());
try (TestMaterializedCollectBatchResult result =
new TestMaterializedCollectBatchResult(
new TestTableResult(ResultKind.SUCCESS_WITH_CONTENT, schema),
2, // limit the materialized table to 2 rows
- 3)) { // with 3 rows overcommitment
+ 3,
+ createInternalBinaryRowDataConverter(
+ schema.toPhysicalRowDataType()))) { // with 3 rows overcommitment
result.isRetrieving = true;
result.processRecord(Row.of("D", 1));
@@ -94,27 +139,36 @@ public class MaterializedCollectBatchResultTest {
result.processRecord(Row.of("B", 1));
result.processRecord(Row.of("A", 1));
- assertEquals(
+ assertRowEquals(
Arrays.asList(
null, null, Row.of("B", 1), Row.of("A", 1)), // two over-committed rows
- result.getMaterializedTable());
+ result.getMaterializedTable(),
+ rowConverter);
assertEquals(TypedResult.payload(2), result.snapshot(1));
- assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(1));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(2));
+ assertRowEquals(
+ Collections.singletonList(Row.of("B", 1)),
+ result.retrievePage(1),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(2),
+ rowConverter);
result.processRecord(Row.of("C", 1));
- assertEquals(
+ assertRowEquals(
Arrays.asList(Row.of("A", 1), Row.of("C", 1)), // limit clean up has taken place
- result.getMaterializedTable());
+ result.getMaterializedTable(),
+ rowConverter);
result.processRecord(Row.of("A", 1));
- assertEquals(
+ assertRowEquals(
Arrays.asList(null, Row.of("C", 1), Row.of("A", 1)),
- result.getMaterializedTable());
+ result.getMaterializedTable(),
+ rowConverter);
}
}
@@ -125,20 +179,34 @@ public class MaterializedCollectBatchResultTest {
private static class TestMaterializedCollectBatchResult extends MaterializedCollectBatchResult
implements AutoCloseable {
+ private final Function<Row, BinaryRowData> converter;
+
public boolean isRetrieving;
public TestMaterializedCollectBatchResult(
- TableResult tableResult, int maxRowCount, int overcommitThreshold) {
+ TableResultInternal tableResult,
+ int maxRowCount,
+ int overcommitThreshold,
+ Function<Row, BinaryRowData> converter) {
super(tableResult, maxRowCount, overcommitThreshold);
+ this.converter = converter;
}
- public TestMaterializedCollectBatchResult(TableResult tableResult, int maxRowCount) {
+ public TestMaterializedCollectBatchResult(
+ TableResultInternal tableResult,
+ int maxRowCount,
+ Function<Row, BinaryRowData> converter) {
super(tableResult, maxRowCount);
+ this.converter = converter;
}
@Override
protected boolean isRetrieving() {
return isRetrieving;
}
+
+ public void processRecord(Row row) {
+ processRecord(converter.apply(row));
+ }
}
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java
index c1e8538..cf8196c 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java
@@ -20,10 +20,14 @@ package org.apache.flink.table.client.gateway.local.result;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
-import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.client.cli.utils.TestTableResult;
import org.apache.flink.table.client.gateway.TypedResult;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.data.conversion.DataStructureConverter;
+import org.apache.flink.table.data.conversion.DataStructureConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
@@ -32,23 +36,30 @@ import org.junit.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.function.Function;
import static org.junit.Assert.assertEquals;
/** Tests for {@link MaterializedCollectStreamResult}. */
-public class MaterializedCollectStreamResultTest {
+public class MaterializedCollectStreamResultTest extends BaseMaterializedResultTest {
@Test
public void testSnapshot() throws Exception {
- ResolvedSchema schema =
+ final ResolvedSchema schema =
ResolvedSchema.physical(
new String[] {"f0", "f1"},
- new DataType[] {DataTypes.STRING(), DataTypes.BIGINT()});
+ new DataType[] {DataTypes.STRING(), DataTypes.INT()});
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ final DataStructureConverter<RowData, Row> rowConverter =
+ (DataStructureConverter)
+ DataStructureConverters.getConverter(schema.toPhysicalRowDataType());
try (TestMaterializedCollectStreamResult result =
new TestMaterializedCollectStreamResult(
new TestTableResult(ResultKind.SUCCESS_WITH_CONTENT, schema),
- Integer.MAX_VALUE)) {
+ Integer.MAX_VALUE,
+ createInternalBinaryRowDataConverter(schema.toPhysicalRowDataType()))) {
result.isRetrieving = true;
result.processRecord(Row.ofKind(RowKind.INSERT, "A", 1));
@@ -58,18 +69,39 @@ public class MaterializedCollectStreamResultTest {
assertEquals(TypedResult.payload(4), result.snapshot(1));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(1));
- assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(2));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(3));
- assertEquals(Collections.singletonList(Row.of("C", 2)), result.retrievePage(4));
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(1),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("B", 1)),
+ result.retrievePage(2),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(3),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("C", 2)),
+ result.retrievePage(4),
+ rowConverter);
result.processRecord(Row.ofKind(RowKind.UPDATE_BEFORE, "A", 1));
assertEquals(TypedResult.payload(3), result.snapshot(1));
- assertEquals(Collections.singletonList(Row.of("A", 1)), result.retrievePage(1));
- assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(2));
- assertEquals(Collections.singletonList(Row.of("C", 2)), result.retrievePage(3));
+ assertRowEquals(
+ Collections.singletonList(Row.of("A", 1)),
+ result.retrievePage(1),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("B", 1)),
+ result.retrievePage(2),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("C", 2)),
+ result.retrievePage(3),
+ rowConverter);
result.processRecord(Row.ofKind(RowKind.UPDATE_BEFORE, "C", 2));
result.processRecord(Row.ofKind(RowKind.UPDATE_BEFORE, "A", 1));
@@ -77,23 +109,37 @@ public class MaterializedCollectStreamResultTest {
assertEquals(TypedResult.payload(2), result.snapshot(1));
- assertEquals(Collections.singletonList(Row.of("B", 1)), result.retrievePage(1));
- assertEquals(Collections.singletonList(Row.of("D", 1)), result.retrievePage(2));
+ assertRowEquals(
+ Collections.singletonList(Row.of("B", 1)),
+ result.retrievePage(1),
+ rowConverter);
+ assertRowEquals(
+ Collections.singletonList(Row.of("D", 1)),
+ result.retrievePage(2),
+ rowConverter);
}
}
@Test
public void testLimitedSnapshot() throws Exception {
- ResolvedSchema schema =
+ final ResolvedSchema schema =
ResolvedSchema.physical(
new String[] {"f0", "f1"},
- new DataType[] {DataTypes.STRING(), DataTypes.BIGINT()});
+ new DataType[] {DataTypes.STRING(), DataTypes.INT()});
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ final DataStructureConverter<RowData, Row> rowConverter =
+ (DataStructureConverter)
+ DataStructureConverters.getConverter(schema.toPhysicalRowDataType());
// limit the materialized table to 2 rows
// with 3 rows overcommitment
try (TestMaterializedCollectStreamResult result =
new TestMaterializedCollectStreamResult(
- new TestTableResult(ResultKind.SUCCESS_WITH_CONTENT, schema), 2, 3)) {
+ new TestTableResult(ResultKind.SUCCESS_WITH_CONTENT, schema),
+ 2,
+ 3,
+ createInternalBinaryRowDataConverter(schema.toPhysicalRowDataType()))) {
result.isRetrieving = true;
@@ -102,37 +148,42 @@ public class MaterializedCollectStreamResultTest {
result.processRecord(Row.ofKind(RowKind.INSERT, "B", 1));
result.processRecord(Row.ofKind(RowKind.INSERT, "A", 1));
- assertEquals(
+ assertRowEquals(
Arrays.asList(
null,
null,
Row.ofKind(RowKind.INSERT, "B", 1),
Row.ofKind(RowKind.INSERT, "A", 1)), // two over-committed rows
- result.getMaterializedTable());
+ result.getMaterializedTable(),
+ rowConverter);
assertEquals(TypedResult.payload(2), result.snapshot(1));
- assertEquals(
+ assertRowEquals(
Collections.singletonList(Row.ofKind(RowKind.INSERT, "B", 1)),
- result.retrievePage(1));
- assertEquals(
+ result.retrievePage(1),
+ rowConverter);
+ assertRowEquals(
Collections.singletonList(Row.ofKind(RowKind.INSERT, "A", 1)),
- result.retrievePage(2));
+ result.retrievePage(2),
+ rowConverter);
result.processRecord(Row.ofKind(RowKind.INSERT, "C", 1));
- assertEquals(
+ assertRowEquals(
Arrays.asList(
Row.ofKind(RowKind.INSERT, "A", 1),
Row.ofKind(RowKind.INSERT, "C", 1)), // limit clean up has taken place
- result.getMaterializedTable());
+ result.getMaterializedTable(),
+ rowConverter);
result.processRecord(Row.ofKind(RowKind.DELETE, "A", 1));
- assertEquals(
+ assertRowEquals(
Collections.singletonList(
Row.ofKind(RowKind.INSERT, "C", 1)), // regular clean up has taken place
- result.getMaterializedTable());
+ result.getMaterializedTable(),
+ rowConverter);
}
}
@@ -143,20 +194,34 @@ public class MaterializedCollectStreamResultTest {
private static class TestMaterializedCollectStreamResult extends MaterializedCollectStreamResult
implements AutoCloseable {
+ private final Function<Row, BinaryRowData> converter;
+
public boolean isRetrieving;
public TestMaterializedCollectStreamResult(
- TableResult tableResult, int maxRowCount, int overcommitThreshold) {
+ TableResultInternal tableResult,
+ int maxRowCount,
+ int overcommitThreshold,
+ Function<Row, BinaryRowData> converter) {
super(tableResult, maxRowCount, overcommitThreshold);
+ this.converter = converter;
}
- public TestMaterializedCollectStreamResult(TableResult tableResult, int maxRowCount) {
+ public TestMaterializedCollectStreamResult(
+ TableResultInternal tableResult,
+ int maxRowCount,
+ Function<Row, BinaryRowData> converter) {
super(tableResult, maxRowCount);
+ this.converter = converter;
}
@Override
protected boolean isRetrieving() {
return isRetrieving;
}
+
+ public void processRecord(Row row) {
+ processRecord(converter.apply(row));
+ }
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultIterator.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultIterator.java
deleted file mode 100644
index 254e73a..0000000
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultIterator.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.api.internal;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.execution.JobClient;
-import org.apache.flink.table.api.TableException;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.CloseableIterator;
-
-import javax.annotation.Nullable;
-
-import java.util.NoSuchElementException;
-
-/** A {@link CloseableIterator} for insert operation result. */
-@Internal
-class InsertResultIterator implements CloseableIterator<Row> {
- private final JobClient jobClient;
- private final Row affectedRowCountsRow;
- private final ClassLoader classLoader;
- @Nullable private Boolean hasNext = null;
-
- InsertResultIterator(JobClient jobClient, Row affectedRowCountsRow, ClassLoader classLoader) {
- this.jobClient = jobClient;
- this.affectedRowCountsRow = affectedRowCountsRow;
- this.classLoader = classLoader;
- }
-
- @Override
- public void close() throws Exception {
- jobClient.cancel();
- }
-
- @Override
- public boolean hasNext() {
- if (hasNext == null) {
- try {
- jobClient.getJobExecutionResult().get();
- } catch (Exception e) {
- throw new TableException("Failed to wait job finish", e);
- }
- hasNext = true;
- }
- return hasNext;
- }
-
- @Override
- public Row next() {
- if (hasNext()) {
- hasNext = false;
- return affectedRowCountsRow;
- } else {
- throw new NoSuchElementException();
- }
- }
-}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultProvider.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultProvider.java
new file mode 100644
index 0000000..5792e11
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/InsertResultProvider.java
@@ -0,0 +1,113 @@
+/*
+ * 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.api.internal;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import javax.annotation.Nullable;
+
+import java.util.NoSuchElementException;
+import java.util.function.Supplier;
+
+/** A {@link ResultProvider} with custom wait logic for insert operation result. */
+@Internal
+class InsertResultProvider implements ResultProvider {
+
+ private final Long[] affectedRowCountsRow;
+
+ private @Nullable Boolean hasNext = null;
+ private JobClient jobClient;
+
+ InsertResultProvider(Long[] affectedRowCountsRow) {
+ this.affectedRowCountsRow = affectedRowCountsRow;
+ }
+
+ @Override
+ public InsertResultProvider setJobClient(JobClient jobClient) {
+ this.jobClient = jobClient;
+ return this;
+ }
+
+ @Override
+ public CloseableIterator<RowData> toInternalIterator() {
+ return new Iterator<>(() -> GenericRowData.of(((Object[]) affectedRowCountsRow)));
+ }
+
+ @Override
+ public CloseableIterator<Row> toExternalIterator() {
+ return new Iterator<>(() -> Row.of(((Object[]) affectedRowCountsRow)));
+ }
+
+ @Override
+ public boolean isFirstRowReady() {
+ return hasNext != null || hasNext();
+ }
+
+ private void close() throws Exception {
+ if (jobClient != null) {
+ jobClient.cancel();
+ }
+ }
+
+ private boolean hasNext() {
+ if (hasNext == null) {
+ try {
+ jobClient.getJobExecutionResult().get();
+ } catch (Exception e) {
+ throw new TableException("Failed to wait job finish", e);
+ }
+ hasNext = true;
+ }
+ return hasNext;
+ }
+
+ private class Iterator<T> implements CloseableIterator<T> {
+ private final Supplier<T> rowSupplier;
+
+ private Iterator(Supplier<T> rowSupplier) {
+ this.rowSupplier = rowSupplier;
+ }
+
+ @Override
+ public void close() throws Exception {
+ InsertResultProvider.this.close();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return InsertResultProvider.this.hasNext();
+ }
+
+ @Override
+ public T next() {
+ if (hasNext()) {
+ hasNext = false;
+ return rowSupplier.get();
+ } else {
+ throw new NoSuchElementException();
+ }
+ }
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/CollectResultProvider.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ResultProvider.java
similarity index 59%
copy from flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/CollectResultProvider.java
copy to flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ResultProvider.java
index e3ec02a..d7300fd 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/CollectResultProvider.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ResultProvider.java
@@ -21,19 +21,39 @@ package org.apache.flink.table.api.internal;
import org.apache.flink.annotation.Internal;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
/**
* An internal class which helps the client to get the execute result from a specific sink.
*
+ * <p>The two iterators are exclusive, that is you can use only one of the two at the same time.
+ *
* <p>This class is generated by specific sink and brings the result info to a {@link TableResult}.
*/
@Internal
-public interface CollectResultProvider {
+public interface ResultProvider {
/** Set the job client associated with the select job to retrieve the result. */
- void setJobClient(JobClient jobClient);
+ ResultProvider setJobClient(JobClient jobClient);
+
+ /**
+ * Returns the select result as row iterator using internal data types. You must not invoke this
+ * method after invoking {@link #toExternalIterator()}.
+ */
+ CloseableIterator<RowData> toInternalIterator();
+
+ /**
+ * Returns the select result as row iterator using external data types. You must not invoke this
+ * method after invoking {@link #toInternalIterator()}.
+ */
+ CloseableIterator<Row> toExternalIterator();
- /** Returns the select result as row iterator. */
- CloseableIterator<Row> getResultIterator();
+ /**
+ * Return true if the first row is ready.
+ *
+ * <p>The first row is ready when {@link CloseableIterator#hasNext} method returns true or
+ * {@link CloseableIterator#next()} method returns a row.
+ */
+ boolean isFirstRowReady();
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/StaticResultProvider.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/StaticResultProvider.java
new file mode 100644
index 0000000..97fc848
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/StaticResultProvider.java
@@ -0,0 +1,93 @@
+/*
+ * 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.api.internal;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.List;
+import java.util.function.Function;
+
+/** Create result provider from a static set of data using external types. */
+@Internal
+class StaticResultProvider implements ResultProvider {
+
+ private final List<Row> rows;
+ private final Function<Row, RowData> externalToInternalConverter;
+
+ public StaticResultProvider(List<Row> rows) {
+ this(rows, StaticResultProvider::rowToInternalRow);
+ }
+
+ public StaticResultProvider(
+ List<Row> rows, Function<Row, RowData> externalToInternalConverter) {
+ this.rows = rows;
+ this.externalToInternalConverter = externalToInternalConverter;
+ }
+
+ @Override
+ public StaticResultProvider setJobClient(JobClient jobClient) {
+ return this;
+ }
+
+ @Override
+ public CloseableIterator<RowData> toInternalIterator() {
+ return CloseableIterator.adapterForIterator(
+ this.rows.stream().map(this.externalToInternalConverter).iterator());
+ }
+
+ @Override
+ public CloseableIterator<Row> toExternalIterator() {
+ return CloseableIterator.adapterForIterator(this.rows.iterator());
+ }
+
+ @Override
+ public boolean isFirstRowReady() {
+ return true;
+ }
+
+ /** This function supports only String, long, int and boolean fields. */
+ @VisibleForTesting
+ static RowData rowToInternalRow(Row row) {
+ Object[] values = new Object[row.getArity()];
+ for (int i = 0; i < row.getArity(); i++) {
+ Object value = row.getField(i);
+ if (value == null) {
+ values[i] = null;
+ } else if (value instanceof String) {
+ values[i] = StringData.fromString((String) value);
+ } else if (value instanceof Boolean
+ || value instanceof Long
+ || value instanceof Integer) {
+ values[i] = value;
+ } else {
+ throw new TableException("Cannot convert row type");
+ }
+ }
+
+ return GenericRowData.of(values);
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 037843d..e13c182 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -751,10 +751,10 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
@Override
- public TableResult executeInternal(List<ModifyOperation> operations) {
+ public TableResultInternal executeInternal(List<ModifyOperation> operations) {
List<Transformation<?>> transformations = translate(operations);
List<String> sinkIdentifierNames = extractSinkIdentifierNames(operations);
- TableResult result = executeInternal(transformations, sinkIdentifierNames);
+ TableResultInternal result = executeInternal(transformations, sinkIdentifierNames);
if (tableConfig.getConfiguration().get(TABLE_DML_SYNC)) {
try {
result.await();
@@ -766,7 +766,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return result;
}
- private TableResult executeInternal(
+ private TableResultInternal executeInternal(
List<Transformation<?>> transformations, List<String> sinkIdentifierNames) {
final String defaultJobName = "insert-into_" + String.join(",", sinkIdentifierNames);
Pipeline pipeline =
@@ -775,7 +775,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
try {
JobClient jobClient = execEnv.executeAsync(pipeline);
final List<Column> columns = new ArrayList<>();
- Object[] affectedRowCounts = new Long[transformations.size()];
+ Long[] affectedRowCounts = new Long[transformations.size()];
for (int i = 0; i < transformations.size(); ++i) {
// use sink identifier name as field name
columns.add(Column.physical(sinkIdentifierNames.get(i), DataTypes.BIGINT()));
@@ -786,16 +786,15 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
.jobClient(jobClient)
.resultKind(ResultKind.SUCCESS_WITH_CONTENT)
.schema(ResolvedSchema.of(columns))
- .data(
- new InsertResultIterator(
- jobClient, Row.of(affectedRowCounts), userClassLoader))
+ .resultProvider(
+ new InsertResultProvider(affectedRowCounts).setJobClient(jobClient))
.build();
} catch (Exception e) {
throw new TableException("Failed to execute sql", e);
}
}
- private TableResult executeQueryOperation(QueryOperation operation) {
+ private TableResultInternal executeQueryOperation(QueryOperation operation) {
final UnresolvedIdentifier unresolvedIdentifier =
UnresolvedIdentifier.of(
"Unregistered_Collect_Sink_" + CollectModifyOperation.getUniqueId());
@@ -812,13 +811,13 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
transformations, tableConfig.getConfiguration(), defaultJobName);
try {
JobClient jobClient = execEnv.executeAsync(pipeline);
- CollectResultProvider resultProvider = sinkOperation.getSelectResultProvider();
+ ResultProvider resultProvider = sinkOperation.getSelectResultProvider();
resultProvider.setJobClient(jobClient);
return TableResultImpl.builder()
.jobClient(jobClient)
.resultKind(ResultKind.SUCCESS_WITH_CONTENT)
.schema(operation.getResolvedSchema())
- .data(resultProvider.getResultIterator())
+ .resultProvider(resultProvider)
.setPrintStyle(
TableResultImpl.PrintStyle.tableau(
PrintUtils.MAX_COLUMN_WIDTH,
@@ -870,7 +869,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
@Override
- public TableResult executeInternal(Operation operation) {
+ public TableResultInternal executeInternal(Operation operation) {
if (operation instanceof ModifyOperation) {
return executeInternal(Collections.singletonList((ModifyOperation) operation));
} else if (operation instanceof CreateTableOperation) {
@@ -1298,7 +1297,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult createCatalog(CreateCatalogOperation operation) {
+ private TableResultInternal createCatalog(CreateCatalogOperation operation) {
String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
try {
String catalogName = operation.getCatalogName();
@@ -1318,7 +1317,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult loadModule(LoadModuleOperation operation) {
+ private TableResultInternal loadModule(LoadModuleOperation operation) {
final String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
try {
final Module module =
@@ -1336,7 +1335,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult unloadModule(UnloadModuleOperation operation) {
+ private TableResultInternal unloadModule(UnloadModuleOperation operation) {
String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
try {
moduleManager.unloadModule(operation.getModuleName());
@@ -1346,7 +1345,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult useModules(UseModulesOperation operation) {
+ private TableResultInternal useModules(UseModulesOperation operation) {
String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
try {
moduleManager.useModules(operation.getModuleNames().toArray(new String[0]));
@@ -1356,7 +1355,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult buildShowResult(String columnName, String[] objects) {
+ private TableResultInternal buildShowResult(String columnName, String[] objects) {
return buildResult(
new String[] {columnName},
new DataType[] {DataTypes.STRING()},
@@ -1442,7 +1441,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return sb.toString();
}
- private TableResult buildDescribeResult(ResolvedSchema schema) {
+ private TableResultInternal buildDescribeResult(ResolvedSchema schema) {
Object[][] rows = buildTableColumns(schema);
return buildResult(generateTableColumnsNames(), generateTableColumnsDataTypes(), rows);
}
@@ -1462,7 +1461,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return new String[] {"name", "type", "null", "key", "extras", "watermark"};
}
- private TableResult buildShowColumnsResult(
+ private TableResultInternal buildShowColumnsResult(
ResolvedSchema schema, ShowColumnsOperation showColumnsOp) {
Object[][] rows = buildTableColumns(schema);
if (showColumnsOp.isUseLike()) {
@@ -1507,7 +1506,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
return sb.toString();
}
- private TableResult buildShowFullModulesResult(ModuleEntry[] moduleEntries) {
+ private TableResultInternal buildShowFullModulesResult(ModuleEntry[] moduleEntries) {
Object[][] rows =
Arrays.stream(moduleEntries)
.map(entry -> new Object[] {entry.name(), entry.used()})
@@ -1556,7 +1555,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
.toArray(Object[][]::new);
}
- private TableResult buildResult(String[] headers, DataType[] types, Object[][] rows) {
+ private TableResultInternal buildResult(String[] headers, DataType[] types, Object[][] rows) {
return TableResultImpl.builder()
.resultKind(ResultKind.SUCCESS_WITH_CONTENT)
.schema(ResolvedSchema.physical(headers, types))
@@ -1807,7 +1806,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
.map(CatalogManager.TableLookupResult::getTable);
}
- private TableResult createCatalogFunction(
+ private TableResultInternal createCatalogFunction(
CreateCatalogFunctionOperation createCatalogFunctionOperation) {
String exMsg = getDDLOpExecuteErrorMsg(createCatalogFunctionOperation.asSummaryString());
try {
@@ -1838,7 +1837,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult alterCatalogFunction(
+ private TableResultInternal alterCatalogFunction(
AlterCatalogFunctionOperation alterCatalogFunctionOperation) {
String exMsg = getDDLOpExecuteErrorMsg(alterCatalogFunctionOperation.asSummaryString());
try {
@@ -1866,7 +1865,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult dropCatalogFunction(
+ private TableResultInternal dropCatalogFunction(
DropCatalogFunctionOperation dropCatalogFunctionOperation) {
String exMsg = getDDLOpExecuteErrorMsg(dropCatalogFunctionOperation.asSummaryString());
try {
@@ -1895,7 +1894,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult createSystemFunction(CreateTempSystemFunctionOperation operation) {
+ private TableResultInternal createSystemFunction(CreateTempSystemFunctionOperation operation) {
String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
try {
functionCatalog.registerTemporarySystemFunction(
@@ -1910,7 +1909,7 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
}
- private TableResult dropSystemFunction(DropTempSystemFunctionOperation operation) {
+ private TableResultInternal dropSystemFunction(DropTempSystemFunctionOperation operation) {
try {
functionCatalog.dropTemporarySystemFunction(
operation.getFunctionName(), operation.isIfExists());
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java
index 63b4f5b..c921ff4 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentInternal.java
@@ -63,7 +63,7 @@ public interface TableEnvironmentInternal extends TableEnvironment {
* @param operations The operations to be executed.
* @return the affected row counts (-1 means unknown).
*/
- TableResult executeInternal(List<ModifyOperation> operations);
+ TableResultInternal executeInternal(List<ModifyOperation> operations);
/**
* Execute the given operation and return the execution result.
@@ -71,7 +71,7 @@ public interface TableEnvironmentInternal extends TableEnvironment {
* @param operation The operation to be executed.
* @return the content of the execution result.
*/
- TableResult executeInternal(Operation operation);
+ TableResultInternal executeInternal(Operation operation);
/**
* Returns the AST of this table and the execution plan to compute the result of this table.
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
index 2537f8a..c562ce2 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
@@ -20,12 +20,11 @@ package org.apache.flink.table.api.internal;
import org.apache.flink.annotation.Internal;
import org.apache.flink.core.execution.JobClient;
-import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.TableResult;
-import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.utils.PrintUtils;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
@@ -35,7 +34,6 @@ import javax.annotation.Nullable;
import java.io.PrintWriter;
import java.time.ZoneId;
-import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
@@ -48,18 +46,12 @@ import java.util.concurrent.TimeoutException;
/** Implementation for {@link TableResult}. */
@Internal
-public class TableResultImpl implements TableResult {
- public static final TableResult TABLE_RESULT_OK =
- TableResultImpl.builder()
- .resultKind(ResultKind.SUCCESS)
- .schema(ResolvedSchema.of(Column.physical("result", DataTypes.STRING())))
- .data(Collections.singletonList(Row.of("OK")))
- .build();
+public class TableResultImpl implements TableResultInternal {
private final JobClient jobClient;
private final ResolvedSchema resolvedSchema;
private final ResultKind resultKind;
- private final CloseableRowIteratorWrapper data;
+ private final ResultProvider resultProvider;
private final PrintStyle printStyle;
private final ZoneId sessionTimeZone;
@@ -67,15 +59,15 @@ public class TableResultImpl implements TableResult {
@Nullable JobClient jobClient,
ResolvedSchema resolvedSchema,
ResultKind resultKind,
- CloseableIterator<Row> data,
+ ResultProvider resultProvider,
PrintStyle printStyle,
ZoneId sessionTimeZone) {
this.jobClient = jobClient;
this.resolvedSchema =
Preconditions.checkNotNull(resolvedSchema, "resolvedSchema should not be null");
this.resultKind = Preconditions.checkNotNull(resultKind, "resultKind should not be null");
- Preconditions.checkNotNull(data, "data should not be null");
- this.data = new CloseableRowIteratorWrapper(data);
+ Preconditions.checkNotNull(resultProvider, "result provider should not be null");
+ this.resultProvider = resultProvider;
this.printStyle = Preconditions.checkNotNull(printStyle, "printStyle should not be null");
this.sessionTimeZone =
Preconditions.checkNotNull(sessionTimeZone, "sessionTimeZone should not be null");
@@ -113,7 +105,7 @@ public class TableResultImpl implements TableResult {
CompletableFuture<Void> future =
CompletableFuture.runAsync(
() -> {
- while (!data.isFirstRowReady()) {
+ while (!resultProvider.isFirstRowReady()) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
@@ -145,12 +137,17 @@ public class TableResultImpl implements TableResult {
@Override
public CloseableIterator<Row> collect() {
- return data;
+ return resultProvider.toExternalIterator();
+ }
+
+ @Override
+ public CloseableIterator<RowData> collectInternal() {
+ return resultProvider.toInternalIterator();
}
@Override
public void print() {
- Iterator<Row> it = collect();
+ Iterator<RowData> it = resultProvider.toInternalIterator();
if (printStyle instanceof TableauStyle) {
int maxColumnWidth = ((TableauStyle) printStyle).getMaxColumnWidth();
String nullColumn = ((TableauStyle) printStyle).getNullColumn();
@@ -188,7 +185,7 @@ public class TableResultImpl implements TableResult {
private JobClient jobClient = null;
private ResolvedSchema resolvedSchema = null;
private ResultKind resultKind = null;
- private CloseableIterator<Row> data = null;
+ private ResultProvider resultProvider = null;
private PrintStyle printStyle =
PrintStyle.tableau(Integer.MAX_VALUE, PrintUtils.NULL_COLUMN, false, false);
private ZoneId sessionTimeZone = ZoneId.of("UTC");
@@ -227,14 +224,9 @@ public class TableResultImpl implements TableResult {
return this;
}
- /**
- * Specifies an row iterator as the execution result.
- *
- * @param rowIterator a row iterator as the execution result.
- */
- public Builder data(CloseableIterator<Row> rowIterator) {
- Preconditions.checkNotNull(rowIterator, "rowIterator should not be null");
- this.data = rowIterator;
+ public Builder resultProvider(ResultProvider resultProvider) {
+ Preconditions.checkNotNull(resultProvider, "resultProvider should not be null");
+ this.resultProvider = resultProvider;
return this;
}
@@ -245,7 +237,7 @@ public class TableResultImpl implements TableResult {
*/
public Builder data(List<Row> rowList) {
Preconditions.checkNotNull(rowList, "listRows should not be null");
- this.data = CloseableIterator.adapterForIterator(rowList.iterator());
+ this.resultProvider = new StaticResultProvider(rowList);
return this;
}
@@ -264,9 +256,14 @@ public class TableResultImpl implements TableResult {
}
/** Returns a {@link TableResult} instance. */
- public TableResult build() {
+ public TableResultInternal build() {
return new TableResultImpl(
- jobClient, resolvedSchema, resultKind, data, printStyle, sessionTimeZone);
+ jobClient,
+ resolvedSchema,
+ resultKind,
+ resultProvider,
+ printStyle,
+ sessionTimeZone);
}
}
@@ -343,44 +340,4 @@ public class TableResultImpl implements TableResult {
* only print the result content as raw form. column delimiter is ",", row delimiter is "\n".
*/
private static final class RawContentStyle implements PrintStyle {}
-
- /**
- * A {@link CloseableIterator} wrapper class that can return whether the first row is ready.
- *
- * <p>The first row is ready when {@link #hasNext} method returns true or {@link #next()} method
- * returns a row. The execution order of {@link TableResult#collect} method and {@link
- * TableResult#await()} may be arbitrary, this class will record whether the first row is ready
- * (or accessed).
- */
- private static final class CloseableRowIteratorWrapper implements CloseableIterator<Row> {
- private final CloseableIterator<Row> iterator;
- private boolean isFirstRowReady = false;
-
- private CloseableRowIteratorWrapper(CloseableIterator<Row> iterator) {
- this.iterator = iterator;
- }
-
- @Override
- public void close() throws Exception {
- iterator.close();
- }
-
- @Override
- public boolean hasNext() {
- boolean hasNext = iterator.hasNext();
- isFirstRowReady = isFirstRowReady || hasNext;
- return hasNext;
- }
-
- @Override
- public Row next() {
- Row next = iterator.next();
- isFirstRowReady = true;
- return next;
- }
-
- public boolean isFirstRowReady() {
- return isFirstRowReady || hasNext();
- }
- }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/CollectResultProvider.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultInternal.java
similarity index 54%
rename from flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/CollectResultProvider.java
rename to flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultInternal.java
index e3ec02a..a7d9e55 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/CollectResultProvider.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultInternal.java
@@ -19,21 +19,27 @@
package org.apache.flink.table.api.internal;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ResultKind;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
-/**
- * An internal class which helps the client to get the execute result from a specific sink.
- *
- * <p>This class is generated by specific sink and brings the result info to a {@link TableResult}.
- */
+import java.util.Collections;
+
+/** Internal interface with internal methods for {@link TableResult}. */
@Internal
-public interface CollectResultProvider {
- /** Set the job client associated with the select job to retrieve the result. */
- void setJobClient(JobClient jobClient);
+public interface TableResultInternal extends TableResult {
+ TableResultInternal TABLE_RESULT_OK =
+ TableResultImpl.builder()
+ .resultKind(ResultKind.SUCCESS)
+ .schema(ResolvedSchema.of(Column.physical("result", DataTypes.STRING())))
+ .data(Collections.singletonList(Row.of("OK")))
+ .build();
- /** Returns the select result as row iterator. */
- CloseableIterator<Row> getResultIterator();
+ /** Returns an iterator that returns the iterator with the internal row data type. */
+ CloseableIterator<RowData> collectInternal();
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CollectModifyOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CollectModifyOperation.java
index 1d95b4e..2cd0909 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CollectModifyOperation.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/CollectModifyOperation.java
@@ -19,7 +19,7 @@
package org.apache.flink.table.operations;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.table.api.internal.CollectResultProvider;
+import org.apache.flink.table.api.internal.ResultProvider;
import org.apache.flink.table.catalog.ObjectIdentifier;
import java.util.Collections;
@@ -39,7 +39,7 @@ public final class CollectModifyOperation implements ModifyOperation {
private final QueryOperation child;
// help the client to get the execute result from a specific sink.
- private CollectResultProvider resultProvider;
+ private ResultProvider resultProvider;
public CollectModifyOperation(ObjectIdentifier tableIdentifier, QueryOperation child) {
this.tableIdentifier = tableIdentifier;
@@ -54,11 +54,11 @@ public final class CollectModifyOperation implements ModifyOperation {
return tableIdentifier;
}
- public void setSelectResultProvider(CollectResultProvider resultProvider) {
+ public void setSelectResultProvider(ResultProvider resultProvider) {
this.resultProvider = resultProvider;
}
- public CollectResultProvider getSelectResultProvider() {
+ public ResultProvider getSelectResultProvider() {
return resultProvider;
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java
index 891bfa0..fd6b71e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.utils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.ArrayData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.TimestampData;
@@ -64,6 +65,7 @@ import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getPr
import static org.apache.flink.table.utils.TimestampStringUtils.localTimeToUnixDate;
import static org.apache.flink.table.utils.TimestampStringUtils.timeToInternal;
import static org.apache.flink.table.utils.TimestampStringUtils.timestampToString;
+import static org.apache.flink.table.utils.TimestampStringUtils.unixDateToString;
import static org.apache.flink.table.utils.TimestampStringUtils.unixTimeToString;
/** Utilities for print formatting. */
@@ -96,7 +98,7 @@ public class PrintUtils {
*/
public static void printAsTableauForm(
ResolvedSchema resolvedSchema,
- Iterator<Row> it,
+ Iterator<RowData> it,
PrintWriter printWriter,
ZoneId sessionTimeZone) {
printAsTableauForm(
@@ -142,7 +144,7 @@ public class PrintUtils {
*/
public static void printAsTableauForm(
ResolvedSchema resolvedSchema,
- Iterator<Row> it,
+ Iterator<RowData> it,
PrintWriter printWriter,
int maxColumnWidth,
String nullColumn,
@@ -171,11 +173,11 @@ public class PrintUtils {
nullColumn,
printRowKind ? ROW_KIND_COLUMN : null);
} else {
- final List<Row> rows = new ArrayList<>();
+ final List<RowData> rows = new ArrayList<>();
final List<String[]> content = new ArrayList<>();
content.add(columnNames);
while (it.hasNext()) {
- Row row = it.next();
+ RowData row = it.next();
rows.add(row);
content.add(
rowToString(
@@ -212,12 +214,12 @@ public class PrintUtils {
}
public static String[] rowToString(
- Row row, ResolvedSchema resolvedSchema, ZoneId sessionTimeZone) {
+ RowData row, ResolvedSchema resolvedSchema, ZoneId sessionTimeZone) {
return rowToString(row, NULL_COLUMN, false, resolvedSchema, sessionTimeZone);
}
public static String[] rowToString(
- Row row,
+ RowData row,
String nullColumn,
boolean printRowKind,
ResolvedSchema resolvedSchema,
@@ -225,12 +227,12 @@ public class PrintUtils {
final int len = printRowKind ? row.getArity() + 1 : row.getArity();
final List<String> fields = new ArrayList<>(len);
if (printRowKind) {
- fields.add(row.getKind().shortString());
+ fields.add(row.getRowKind().shortString());
}
for (int i = 0; i < row.getArity(); i++) {
- final Object field = row.getField(i);
final LogicalType fieldType =
resolvedSchema.getColumnDataTypes().get(i).getLogicalType();
+ final Object field = RowData.createFieldGetter(fieldType, i).getFieldOrNull(row);
if (field == null) {
fields.add(nullColumn);
} else {
@@ -259,6 +261,8 @@ public class PrintUtils {
return formatTimestampField(field, fieldType, sessionTimeZone);
case TIME_WITHOUT_TIME_ZONE:
return formatTimeField(field);
+ case DATE:
+ return unixDateToString((int) field);
case ARRAY:
LogicalType elementType = ((ArrayType) fieldType).getElementType();
if (field instanceof List) {
@@ -269,6 +273,18 @@ public class PrintUtils {
formattedTimestamp(array.get(i), elementType, sessionTimeZone);
}
return formattedArray;
+ } else if (field instanceof ArrayData) {
+ ArrayData array = (ArrayData) field;
+ Object[] formattedArray = new Object[array.size()];
+ for (int i = 0; i < array.size(); i++) {
+ formattedArray[i] =
+ formattedTimestamp(
+ ArrayData.createElementGetter(elementType)
+ .getElementOrNull(array, i),
+ elementType,
+ sessionTimeZone);
+ }
+ return formattedArray;
} else if (field.getClass().isArray()) {
// primitive type
if (field.getClass() == byte[].class) {
@@ -391,14 +407,20 @@ public class PrintUtils {
} else if (fieldType instanceof MapType && field instanceof MapData) {
MapData map = ((MapData) field);
Map<Object, Object> formattedMap = new HashMap<>(map.size());
- Object[] keyArray =
- (Object[]) formattedTimestamp(map.keyArray(), keyType, sessionTimeZone);
- Object[] valueArray =
- (Object[])
- formattedTimestamp(
- map.valueArray(), valueType, sessionTimeZone);
- for (int i = 0; i < keyArray.length; i++) {
- formattedMap.put(keyArray[i], valueArray[i]);
+ ArrayData keyArray = map.keyArray();
+ ArrayData valueArray = map.valueArray();
+ for (int i = 0; i < keyArray.size(); i++) {
+ formattedMap.put(
+ formattedTimestamp(
+ ArrayData.createElementGetter(keyType)
+ .getElementOrNull(keyArray, i),
+ keyType,
+ sessionTimeZone),
+ formattedTimestamp(
+ ArrayData.createElementGetter(valueType)
+ .getElementOrNull(valueArray, i),
+ valueType,
+ sessionTimeZone));
}
return formattedMap;
} else {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java
index abde5b8..ec4c320 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/TimestampStringUtils.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.utils;
import org.apache.flink.annotation.Internal;
+import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.TimeZone;
@@ -124,6 +125,13 @@ public class TimestampStringUtils {
return buf.toString();
}
+ public static String unixDateToString(int time) {
+ final StringBuilder buf = new StringBuilder(8);
+ final LocalDate date = LocalDate.ofEpochDay(time);
+ ymd(buf, date.getYear(), date.getMonthValue(), date.getDayOfMonth());
+ return buf.toString();
+ }
+
private static void unixTimeToString(StringBuilder buf, int time, int precision) {
// we copy this method from Calcite DateTimeUtils but add the following changes
// time may be negative which means time milli seconds before 00:00:00
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java
index e883944..4cb31a4 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java
@@ -21,6 +21,12 @@ package org.apache.flink.table.utils;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+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.types.Row;
import org.apache.flink.types.RowKind;
@@ -40,6 +46,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.junit.Assert.assertEquals;
@@ -50,24 +57,41 @@ public class PrintUtilsTest {
@Test
public void testArrayToString() {
- Row row = new Row(7);
- row.setField(0, new int[] {1, 2});
- row.setField(1, new Integer[] {3, 4});
- row.setField(2, new Object[] {new int[] {5, 6}, new int[] {7, 8}});
- row.setField(3, new Integer[][] {new Integer[] {9, 10}, new Integer[] {11, 12}});
- row.setField(
- 4,
- new LocalDateTime[] {
- LocalDateTime.parse("2021-04-18T18:00:00.123456"),
- LocalDateTime.parse("2021-04-18T18:00:00.000001")
- });
- row.setField(
- 5,
- new Instant[][] {
- new Instant[] {Instant.ofEpochMilli(1), Instant.ofEpochMilli(10)},
- new Instant[] {Instant.ofEpochSecond(1), Instant.ofEpochSecond(10)}
- });
- row.setField(6, new int[] {1123, 2123});
+ RowData row =
+ GenericRowData.of(
+ new GenericArrayData(new int[] {1, 2}),
+ new GenericArrayData(new int[] {3, 4}),
+ new GenericArrayData(
+ new Object[] {
+ new GenericArrayData(new int[] {5, 6}),
+ new GenericArrayData(new int[] {7, 8})
+ }),
+ new GenericArrayData(
+ new Object[] {
+ new GenericArrayData(new int[] {9, 10}),
+ new GenericArrayData(new int[] {11, 12})
+ }),
+ new GenericArrayData(
+ new Object[] {
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.parse("2021-04-18T18:00:00.123456")),
+ TimestampData.fromLocalDateTime(
+ LocalDateTime.parse("2021-04-18T18:00:00.000001"))
+ }),
+ new GenericArrayData(
+ new Object[] {
+ new GenericArrayData(
+ new Object[] {
+ TimestampData.fromInstant(Instant.ofEpochMilli(1)),
+ TimestampData.fromInstant(Instant.ofEpochMilli(10))
+ }),
+ new GenericArrayData(
+ new Object[] {
+ TimestampData.fromInstant(Instant.ofEpochSecond(1)),
+ TimestampData.fromInstant(Instant.ofEpochSecond(10))
+ })
+ }),
+ new GenericArrayData(new int[] {1123, 2123}));
ResolvedSchema resolvedSchema =
ResolvedSchema.of(
@@ -95,26 +119,31 @@ public class PrintUtilsTest {
@Test
public void testNestedRowToString() {
- Row row = new Row(4);
- row.setField(0, new int[] {1, 2});
- Row row1 = new Row(4);
- row1.setField(0, "hello");
- row1.setField(1, new boolean[] {true, false});
- row1.setField(
- 2,
- new Timestamp[] {
- Timestamp.valueOf("2021-04-18 18:00:00.123456"),
- Timestamp.valueOf("2021-04-18 18:00:00.000001")
- });
- row1.setField(3, new Long[] {100L, 200L});
- row.setField(1, row1);
- row.setField(
- 2,
- new int[][] {
- new int[] {1, 10},
- new int[] {2, 20}
- });
- row.setField(3, new Integer[] {3000, 4000});
+ RowData row =
+ GenericRowData.of(
+ new GenericArrayData(new int[] {1, 2}),
+ GenericRowData.of(
+ StringData.fromString("hello"),
+ new GenericArrayData(new boolean[] {true, false}),
+ new GenericArrayData(
+ new Object[] {
+ TimestampData.fromTimestamp(
+ Timestamp.valueOf(
+ "2021-04-18 18:00:00.123456")),
+ TimestampData.fromTimestamp(
+ Timestamp.valueOf("2021-04-18 18:00:00.000001"))
+ }),
+ new GenericArrayData(
+ new Object[] {
+ TimestampData.fromEpochMillis(100L),
+ TimestampData.fromEpochMillis(200L)
+ })),
+ new GenericArrayData(
+ new Object[] {
+ new GenericArrayData(new int[] {1, 10}),
+ new GenericArrayData(new int[] {2, 20})
+ }),
+ new GenericArrayData(new int[] {3000, 4000}));
ResolvedSchema resolvedSchema =
ResolvedSchema.of(
@@ -140,15 +169,15 @@ public class PrintUtilsTest {
@Test
public void testNestedMapToString() {
- Row row = new Row(2);
- row.setField(0, new int[] {1, 2});
- Row row1 = new Row(2);
- row1.setField(0, "hello");
Map<TimestampData, TimestampData> map = new HashMap<>();
map.put(TimestampData.fromEpochMillis(1000), TimestampData.fromEpochMillis(2000));
map.put(TimestampData.fromEpochMillis(2000), TimestampData.fromEpochMillis(4000));
- row1.setField(1, map);
- row.setField(1, row1);
+
+ RowData row =
+ GenericRowData.of(
+ new GenericArrayData(new int[] {1, 2}),
+ GenericRowData.of(StringData.fromString("hello"), new GenericMapData(map)));
+
ResolvedSchema resolvedSchema =
ResolvedSchema.of(
Arrays.asList(
@@ -194,10 +223,7 @@ public class PrintUtilsTest {
@Test
public void testPrintWithEmptyResult() {
PrintUtils.printAsTableauForm(
- getSchema(),
- Collections.<Row>emptyList().iterator(),
- new PrintWriter(outContent),
- UTC_ZONE_ID);
+ getSchema(), Collections.emptyIterator(), new PrintWriter(outContent), UTC_ZONE_ID);
assertEquals("Empty set" + System.lineSeparator(), outContent.toString());
}
@@ -206,7 +232,7 @@ public class PrintUtilsTest {
public void testPrintWithEmptyResultAndRowKind() {
PrintUtils.printAsTableauForm(
getSchema(),
- Collections.<Row>emptyList().iterator(),
+ Collections.emptyIterator(),
new PrintWriter(outContent),
PrintUtils.MAX_COLUMN_WIDTH,
"",
@@ -221,7 +247,7 @@ public class PrintUtilsTest {
public void testPrintWithEmptyResultAndDeriveColumnWidthByContent() {
PrintUtils.printAsTableauForm(
getSchema(),
- Collections.<Row>emptyList().iterator(),
+ Collections.emptyIterator(),
new PrintWriter(outContent),
PrintUtils.MAX_COLUMN_WIDTH,
"",
@@ -252,11 +278,11 @@ public class PrintUtilsTest {
+ System.lineSeparator()
+ "+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+"
+ System.lineSeparator()
- + "| (NULL) | 1 | 2 | abc | 1.23 | 2020-03-01 18:39:14.000000 |"
+ + "| (NULL) | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 |"
+ System.lineSeparator()
- + "| false | (NULL) | 0 | | 1 | 2020-03-01 18:39:14.100000 |"
+ + "| false | (NULL) | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 |"
+ System.lineSeparator()
- + "| true | 2147483647 | (NULL) | abcdefg | 1234567890 | 2020-03-01 18:39:14.120000 |"
+ + "| true | 2147483647 | (NULL) | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 |"
+ System.lineSeparator()
+ "| false | -2147483648 | 9223372036854775807 | (NULL) | 12345.06789 | 2020-03-01 18:39:14.123000 |"
+ System.lineSeparator()
@@ -302,11 +328,11 @@ public class PrintUtilsTest {
+ System.lineSeparator()
+ "+----+---------+-------------+----------------------+--------------------------------+----------------+----------------------------+"
+ System.lineSeparator()
- + "| +I | | 1 | 2 | abc | 1.23 | 2020-03-01 18:39:14.000000 |"
+ + "| +I | | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 |"
+ System.lineSeparator()
- + "| +I | false | | 0 | | 1 | 2020-03-01 18:39:14.100000 |"
+ + "| +I | false | | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 |"
+ System.lineSeparator()
- + "| -D | true | 2147483647 | | abcdefg | 1234567890 | 2020-03-01 18:39:14.120000 |"
+ + "| -D | true | 2147483647 | | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 |"
+ System.lineSeparator()
+ "| +I | false | -2147483648 | 9223372036854775807 | | 12345.06789 | 2020-03-01 18:39:14.123000 |"
+ System.lineSeparator()
@@ -344,11 +370,11 @@ public class PrintUtilsTest {
+ System.lineSeparator()
+ "+----+---------+------------+--------+---------+----------------+----------------------------+"
+ System.lineSeparator()
- + "| +I | | 1 | 2 | abc | 1.23 | 2020-03-01 18:39:14.000000 |"
+ + "| +I | | 1 | 2 | abc | 1.23000 | 2020-03-01 18:39:14.000000 |"
+ System.lineSeparator()
- + "| +I | false | | 0 | | 1 | 2020-03-01 18:39:14.100000 |"
+ + "| +I | false | | 0 | | 1.00000 | 2020-03-01 18:39:14.100000 |"
+ System.lineSeparator()
- + "| -D | true | 2147483647 | | abcdefg | 1234567890 | 2020-03-01 18:39:14.120000 |"
+ + "| -D | true | 2147483647 | | abcdefg | 12345.00000 | 2020-03-01 18:39:14.120000 |"
+ System.lineSeparator()
+ "+----+---------+------------+--------+---------+----------------+----------------------------+"
+ System.lineSeparator()
@@ -367,14 +393,14 @@ public class PrintUtilsTest {
Column.physical("timestamp", DataTypes.TIMESTAMP(6)));
}
- private List<Row> getData() {
+ private List<RowData> getData() {
List<Row> data = new ArrayList<>();
data.add(
Row.ofKind(
RowKind.INSERT,
null,
1,
- 2,
+ 2L,
"abc",
BigDecimal.valueOf(1.23),
Timestamp.valueOf("2020-03-01 18:39:14")));
@@ -383,7 +409,7 @@ public class PrintUtilsTest {
RowKind.INSERT,
false,
null,
- 0,
+ 0L,
"",
BigDecimal.valueOf(1),
Timestamp.valueOf("2020-03-01 18:39:14.1")));
@@ -394,7 +420,7 @@ public class PrintUtilsTest {
Integer.MAX_VALUE,
null,
"abcdefg",
- BigDecimal.valueOf(1234567890),
+ BigDecimal.valueOf(12345),
Timestamp.valueOf("2020-03-01 18:39:14.12")));
data.add(
Row.ofKind(
@@ -419,7 +445,7 @@ public class PrintUtilsTest {
RowKind.UPDATE_BEFORE,
null,
-1,
- -1,
+ -1L,
"abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz",
BigDecimal.valueOf(-12345.06789),
null));
@@ -428,7 +454,7 @@ public class PrintUtilsTest {
RowKind.UPDATE_AFTER,
null,
-1,
- -1,
+ -1L,
"这是一段中文",
BigDecimal.valueOf(-12345.06789),
Timestamp.valueOf("2020-03-04 18:39:14")));
@@ -437,10 +463,28 @@ public class PrintUtilsTest {
RowKind.DELETE,
null,
-1,
- -1,
+ -1L,
"これは日本語をテストするための文です",
BigDecimal.valueOf(-12345.06789),
Timestamp.valueOf("2020-03-04 18:39:14")));
- return data;
+ return data.stream()
+ .map(
+ row ->
+ GenericRowData.ofKind(
+ row.getKind(),
+ row.getField(0),
+ row.getField(1),
+ row.getField(2),
+ (row.getField(3) == null)
+ ? null
+ : StringData.fromString(row.getFieldAs(3)),
+ (row.getField(4) == null)
+ ? null
+ : DecimalData.fromBigDecimal(
+ row.getFieldAs(4), 10, 5),
+ (row.getField(5) == null)
+ ? null
+ : TimestampData.fromTimestamp(row.getFieldAs(5))))
+ .collect(Collectors.toList());
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/CollectDynamicSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/CollectDynamicSink.java
index 8f34406..d9e65ce 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/CollectDynamicSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/CollectDynamicSink.java
@@ -29,18 +29,20 @@ import org.apache.flink.streaming.api.operators.collect.CollectSinkOperator;
import org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorFactory;
import org.apache.flink.streaming.api.operators.collect.CollectStreamSink;
import org.apache.flink.table.api.TableResult;
-import org.apache.flink.table.api.internal.CollectResultProvider;
+import org.apache.flink.table.api.internal.ResultProvider;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.RuntimeConverter;
import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.runtime.typeutils.ExternalTypeInfo;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import java.time.Duration;
+import java.util.function.Function;
/** Table sink for {@link TableResult#collect()}. */
@Internal
@@ -50,36 +52,27 @@ final class CollectDynamicSink implements DynamicTableSink {
private final DataType consumedDataType;
private final MemorySize maxBatchSize;
private final Duration socketTimeout;
+ private final ClassLoader classLoader;
// mutable attributes
-
private CollectResultIterator<RowData> iterator;
+ private DataStructureConverter converter;
CollectDynamicSink(
ObjectIdentifier tableIdentifier,
DataType consumedDataType,
MemorySize maxBatchSize,
- Duration socketTimeout) {
+ Duration socketTimeout,
+ ClassLoader classLoader) {
this.tableIdentifier = tableIdentifier;
this.consumedDataType = consumedDataType;
this.maxBatchSize = maxBatchSize;
this.socketTimeout = socketTimeout;
+ this.classLoader = classLoader;
}
- public CollectResultProvider getSelectResultProvider() {
- return new CollectResultProvider() {
- @Override
- public void setJobClient(JobClient jobClient) {
- iterator.setJobClient(jobClient);
- }
-
- @Override
- @SuppressWarnings({"unchecked", "rawtypes"})
- public CloseableIterator<Row> getResultIterator() {
- // Row after deserialization
- return (CloseableIterator) iterator;
- }
- };
+ public ResultProvider getSelectResultProvider() {
+ return new CollectResultProvider();
}
@Override
@@ -96,7 +89,7 @@ final class CollectDynamicSink implements DynamicTableSink {
final ExecutionConfig config = inputStream.getExecutionConfig();
final TypeSerializer<RowData> externalSerializer =
- ExternalTypeInfo.<RowData>of(consumedDataType, true)
+ InternalTypeInfo.<RowData>of(consumedDataType.getLogicalType())
.createSerializer(config);
final String accumulatorName = tableIdentifier.getObjectName();
@@ -115,6 +108,8 @@ final class CollectDynamicSink implements DynamicTableSink {
externalSerializer,
accumulatorName,
checkpointConfig);
+ this.converter = context.createDataStructureConverter(consumedDataType);
+ this.converter.open(RuntimeConverter.Context.create(classLoader));
final CollectStreamSink<RowData> sink =
new CollectStreamSink<>(inputStream, factory);
@@ -124,17 +119,82 @@ final class CollectDynamicSink implements DynamicTableSink {
@Override
public DynamicTableSink copy() {
- final CollectDynamicSink copy =
- new CollectDynamicSink(
- tableIdentifier, consumedDataType, maxBatchSize, socketTimeout);
- // kind of violates the contract of copy() but should not harm
- // as it is null during optimization anyway until physical translation
- copy.iterator = iterator;
- return copy;
+ return new CollectDynamicSink(
+ tableIdentifier, consumedDataType, maxBatchSize, socketTimeout, classLoader);
}
@Override
public String asSummaryString() {
return String.format("TableToCollect(type=%s)", consumedDataType);
}
+
+ private final class CollectResultProvider implements ResultProvider {
+
+ private CloseableRowIteratorWrapper<RowData> rowDataIterator;
+ private CloseableRowIteratorWrapper<Row> rowIterator;
+
+ @Override
+ public ResultProvider setJobClient(JobClient jobClient) {
+ iterator.setJobClient(jobClient);
+ return this;
+ }
+
+ @Override
+ public CloseableIterator<RowData> toInternalIterator() {
+ if (this.rowDataIterator == null) {
+ this.rowDataIterator =
+ new CloseableRowIteratorWrapper<>(iterator, Function.identity());
+ }
+ return this.rowDataIterator;
+ }
+
+ @Override
+ public CloseableIterator<Row> toExternalIterator() {
+ if (this.rowIterator == null) {
+ this.rowIterator =
+ new CloseableRowIteratorWrapper<>(
+ iterator, r -> (Row) converter.toExternal(r));
+ }
+ return this.rowIterator;
+ }
+
+ @Override
+ public boolean isFirstRowReady() {
+ return (this.rowDataIterator != null && this.rowDataIterator.firstRowProcessed)
+ || (this.rowIterator != null && this.rowIterator.firstRowProcessed)
+ || iterator.hasNext();
+ }
+ }
+
+ private static final class CloseableRowIteratorWrapper<T> implements CloseableIterator<T> {
+ private final CloseableIterator<RowData> iterator;
+ private final Function<RowData, T> mapper;
+
+ private boolean firstRowProcessed = false;
+
+ private CloseableRowIteratorWrapper(
+ CloseableIterator<RowData> iterator, Function<RowData, T> mapper) {
+ this.iterator = iterator;
+ this.mapper = mapper;
+ }
+
+ @Override
+ public void close() throws Exception {
+ iterator.close();
+ }
+
+ @Override
+ public boolean hasNext() {
+ boolean hasNext = iterator.hasNext();
+ firstRowProcessed = firstRowProcessed || hasNext;
+ return hasNext;
+ }
+
+ @Override
+ public T next() {
+ RowData next = iterator.next();
+ firstRowProcessed = true;
+ return mapper.apply(next);
+ }
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java
index f9d931e..f3eefed 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java
@@ -87,7 +87,8 @@ public final class DynamicSinkUtils {
FlinkRelBuilder relBuilder,
RelNode input,
CollectModifyOperation collectModifyOperation,
- Configuration configuration) {
+ Configuration configuration,
+ ClassLoader classLoader) {
final DataTypeFactory dataTypeFactory =
unwrapContext(relBuilder).getCatalogManager().getDataTypeFactory();
final ResolvedSchema childSchema = collectModifyOperation.getChild().getResolvedSchema();
@@ -104,7 +105,8 @@ public final class DynamicSinkUtils {
collectModifyOperation.getTableIdentifier(),
consumedDataType,
configuration.get(CollectSinkOperatorFactory.MAX_BATCH_SIZE),
- configuration.get(CollectSinkOperatorFactory.SOCKET_TIMEOUT));
+ configuration.get(CollectSinkOperatorFactory.SOCKET_TIMEOUT),
+ classLoader);
collectModifyOperation.setSelectResultProvider(tableSink.getSelectResultProvider());
return convertSinkToRel(
relBuilder,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
index ee038f5..a4e7351 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala
@@ -217,7 +217,12 @@ abstract class PlannerBase(
case collectModifyOperation: CollectModifyOperation =>
val input = getRelBuilder.queryOperation(modifyOperation.getChild).build()
DynamicSinkUtils.convertCollectToRel(
- getRelBuilder, input, collectModifyOperation, getTableConfig.getConfiguration)
+ getRelBuilder,
+ input,
+ collectModifyOperation,
+ getTableConfig.getConfiguration,
+ getClassLoader
+ )
case catalogSink: CatalogSinkModifyOperation =>
val input = getRelBuilder.queryOperation(modifyOperation.getChild).build()