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 2018/08/29 11:17:47 UTC
[flink] branch release-1.6 updated: [FLINK-8686] [sql-client] Limit
result size for prototyping modes
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.6 by this push:
new 0f80283 [FLINK-8686] [sql-client] Limit result size for prototyping modes
0f80283 is described below
commit 0f8028378fffceecebaa023e1b17753dbfa280bc
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Aug 27 15:02:59 2018 +0200
[FLINK-8686] [sql-client] Limit result size for prototyping modes
This makes the SQL Client more robust by limiting the result size for
both changelog and table result modes. In changelog mode the
responsiveness of the CLI is the limiting factor. This adds a hard
limit for changelog mode. In table mode only the main memory is the
limiting factor and the configurable maximum row count. This adds a
configurable limit for table mode.
This closes #6621.
---
docs/dev/table/sqlClient.md | 6 +-
.../flink-sql-client/conf/sql-client-defaults.yaml | 4 +-
.../table/client/cli/CliChangelogResultView.java | 12 +-
.../flink/table/client/cli/CliResultView.java | 14 +-
.../flink/table/client/config/Execution.java | 4 +
.../flink/table/client/config/PropertyStrings.java | 2 +
.../table/client/gateway/local/ResultStore.java | 7 +-
.../result/MaterializedCollectStreamResult.java | 170 ++++++++++++++++++---
.../client/gateway/local/LocalExecutorITCase.java | 73 ++++++---
.../MaterializedCollectStreamResultTest.java | 87 ++++++++++-
.../test/resources/test-sql-client-defaults.yaml | 1 +
11 files changed, 319 insertions(+), 61 deletions(-)
diff --git a/docs/dev/table/sqlClient.md b/docs/dev/table/sqlClient.md
index d35aa59..ed36354 100644
--- a/docs/dev/table/sqlClient.md
+++ b/docs/dev/table/sqlClient.md
@@ -106,7 +106,7 @@ Alice, 1
Greg, 1
{% endhighlight %}
-Both result modes can be useful during the prototyping of SQL queries.
+Both result modes can be useful during the prototyping of SQL queries. In both modes, results are stored in the Java heap memory of the SQL Client. In order to keep the CLI interface responsive, the changelog mode only shows the latest 1000 changes. The table mode allows for navigating through bigger results that are only limited by the available main memory and the configured [maximum number of rows](sqlClient.html#configuration) (`max-table-result-rows`).
After a query is defined, it can be submitted to the cluster as a long-running, detached Flink job. For this, a target system that stores the results needs to be specified using the [INSERT INTO statement](sqlClient.html#detached-sql-queries). The [configuration section](sqlClient.html#configuration) explains how to declare table sources for reading data, how to declare table sinks for writing data, and how to configure other table program properties.
@@ -165,6 +165,7 @@ Every environment file is a regular [YAML file](http://yaml.org/). An example of
tables:
- name: MyTableSource
type: source
+ update-mode: append
connector:
type: filesystem
path: "/path/to/something.csv"
@@ -198,6 +199,9 @@ functions:
execution:
type: streaming # required: execution mode either 'batch' or 'streaming'
result-mode: table # required: either 'table' or 'changelog'
+ max-table-result-rows: 1000000 # optional: maximum number of maintained rows in
+ # 'table' mode (1000000 by default, smaller 1 means unlimited)
+ # (from Flink 1.6.1)
time-characteristic: event-time # optional: 'processing-time' or 'event-time' (default)
parallelism: 1 # optional: Flink's parallelism (1 by default)
periodic-watermarks-interval: 200 # optional: interval for periodic watermarks (200 ms by default)
diff --git a/flink-libraries/flink-sql-client/conf/sql-client-defaults.yaml b/flink-libraries/flink-sql-client/conf/sql-client-defaults.yaml
index 51e6e95..6dc5485 100644
--- a/flink-libraries/flink-sql-client/conf/sql-client-defaults.yaml
+++ b/flink-libraries/flink-sql-client/conf/sql-client-defaults.yaml
@@ -65,7 +65,9 @@ execution:
# interval in ms for emitting periodic watermarks
periodic-watermarks-interval: 200
# 'changelog' or 'table' presentation of results
- result-mode: changelog
+ result-mode: table
+ # maximum number of maintained rows in 'table' presentation of results
+ max-table-result-rows: 1000000
# parallelism of the program
parallelism: 1
# maximum parallelism
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java
index cf17933..d70af80 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliChangelogResultView.java
@@ -34,6 +34,7 @@ import java.time.LocalTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.LinkedList;
import java.util.List;
import static org.apache.flink.table.client.cli.CliUtils.TIME_FORMATTER;
@@ -50,6 +51,7 @@ import static org.jline.keymap.KeyMap.key;
*/
public class CliChangelogResultView extends CliResultView<CliChangelogResultView.ResultChangelogOperation> {
+ private static final int DEFAULT_MAX_ROW_COUNT = 1000;
private static final int DEFAULT_REFRESH_INTERVAL = 0; // as fast as possible
private static final int DEFAULT_REFRESH_INTERVAL_PLAIN = 3; // every 1s
private static final int MIN_REFRESH_INTERVAL = 0; // every 100ms
@@ -66,7 +68,8 @@ public class CliChangelogResultView extends CliResultView<CliChangelogResultView
refreshInterval = DEFAULT_REFRESH_INTERVAL;
}
previousResults = null;
- results = new ArrayList<>();
+ // rows are always appended at the tail and deleted from the head of the list
+ results = new LinkedList<>();
}
// --------------------------------------------------------------------------------------------
@@ -133,6 +136,13 @@ public class CliChangelogResultView extends CliResultView<CliChangelogResultView
}
// update results
+
+ // formatting and printing of rows is expensive in the current implementation,
+ // therefore we limit the maximum number of lines shown in changelog mode to
+ // keep the CLI responsive
+ if (results.size() >= DEFAULT_MAX_ROW_COUNT) {
+ results.remove(0);
+ }
results.add(changeRow);
scrolling++;
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliResultView.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliResultView.java
index df42edd..78f3a56 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliResultView.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/cli/CliResultView.java
@@ -28,7 +28,6 @@ import org.jline.utils.AttributedStyle;
import java.util.ArrayList;
import java.util.List;
-import java.util.stream.IntStream;
import static org.apache.flink.table.client.cli.CliUtils.normalizeColumn;
@@ -182,9 +181,8 @@ public abstract class CliResultView<O extends Enum<O>> extends CliView<O, Void>
protected List<AttributedString> computeMainLines() {
final List<AttributedString> lines = new ArrayList<>();
- IntStream.range(0, results.size()).forEach(lineIdx -> {
- final String[] line = results.get(lineIdx);
-
+ int lineIdx = 0;
+ for (String[] line : results) {
final AttributedStringBuilder row = new AttributedStringBuilder();
// highlight selected row
@@ -192,7 +190,7 @@ public abstract class CliResultView<O extends Enum<O>> extends CliView<O, Void>
row.style(AttributedStyle.DEFAULT.inverse());
}
- IntStream.range(0, line.length).forEach(colIdx -> {
+ for (int colIdx = 0; colIdx < line.length; colIdx++) {
final String col = line[colIdx];
final int columnWidth = computeColumnWidth(colIdx);
@@ -208,9 +206,11 @@ public abstract class CliResultView<O extends Enum<O>> extends CliView<O, Void>
} else {
normalizeColumn(row, col, columnWidth);
}
- });
+ }
lines.add(row.toAttributedString());
- });
+
+ lineIdx++;
+ }
return lines;
}
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/Execution.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/Execution.java
index 0d6e6dd..5bf49fe 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/Execution.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/Execution.java
@@ -90,6 +90,10 @@ public class Execution {
return Integer.parseInt(properties.getOrDefault(PropertyStrings.EXECUTION_MAX_PARALLELISM, Integer.toString(128)));
}
+ public int getMaxTableResultRows() {
+ return Integer.parseInt(properties.getOrDefault(PropertyStrings.EXECUTION_MAX_TABLE_RESULT_ROWS, Integer.toString(1_000_000)));
+ }
+
public boolean isChangelogMode() {
return Objects.equals(
properties.get(PropertyStrings.EXECUTION_RESULT_MODE),
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/PropertyStrings.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/PropertyStrings.java
index 76e52de..9f9eb72 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/PropertyStrings.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/config/PropertyStrings.java
@@ -57,6 +57,8 @@ public final class PropertyStrings {
public static final String EXECUTION_RESULT_MODE_VALUE_TABLE = "table";
+ public static final String EXECUTION_MAX_TABLE_RESULT_ROWS = "max-table-result-rows";
+
public static final String DEPLOYMENT = "deployment";
public static final String DEPLOYMENT_TYPE = "type";
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java
index a54160f..93fdd48 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ResultStore.java
@@ -71,7 +71,12 @@ public class ResultStore {
if (env.getExecution().isChangelogMode()) {
return new ChangelogCollectStreamResult<>(outputType, config, gatewayAddress, gatewayPort);
} else {
- return new MaterializedCollectStreamResult<>(outputType, config, gatewayAddress, gatewayPort);
+ return new MaterializedCollectStreamResult<>(
+ outputType,
+ config,
+ gatewayAddress,
+ gatewayPort,
+ env.getExecution().getMaxTableResultRows());
}
} else {
diff --git a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java
index 45c4f75..0beabe9 100644
--- a/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java
+++ b/flink-libraries/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResult.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.client.gateway.local.result;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -38,6 +39,31 @@ import java.util.Map;
*/
public class MaterializedCollectStreamResult<C> extends CollectStreamResult<C> implements MaterializedResult<C> {
+ /** Maximum initial capacity of the materialized table. */
+ public static final int MATERIALIZED_TABLE_MAX_INITIAL_CAPACITY = 1_000_000;
+
+ /** Maximum overcommitment of the materialized table. */
+ public static final int MATERIALIZED_TABLE_MAX_OVERCOMMIT = 1_000_000;
+
+ /** Factor for the initial capacity of the materialized table. */
+ public static final double MATERIALIZED_TABLE_CAPACITY_FACTOR = 0.05;
+
+ /** Factor for cleaning up deleted rows in the materialized table. */
+ public static final double MATERIALIZED_TABLE_OVERCOMMIT_FACTOR = 0.01;
+
+ /**
+ * Maximum number of materialized rows to be stored. After the count is reached, oldest
+ * rows are dropped.
+ */
+ private final int maxRowCount;
+
+ /** Threshold for cleaning up deleted rows in the materialized table. */
+ private final int overcommitThreshold;
+
+ /**
+ * Materialized table that is continuously updated by inserts and deletes. Deletes at
+ * the beginning are lazily cleaned up when the threshold is reached.
+ */
private final List<Row> materializedTable;
/**
@@ -47,26 +73,65 @@ public class MaterializedCollectStreamResult<C> extends CollectStreamResult<C> i
*/
private final Map<Row, Integer> rowPositionCache;
+ /** Current snapshot of the materialized table. */
private final List<Row> snapshot;
+ /** Counter for deleted rows to be deleted at the beginning of the materialized table. */
+ private int validRowPosition;
+
+ /** Page count of the snapshot (always >= 1). */
private int pageCount;
+ /** Page size of the snapshot (always >= 1). */
private int pageSize;
+ /** Indicator that this is the last snapshot possible (EOS afterwards). */
private boolean isLastSnapshot;
- public MaterializedCollectStreamResult(TypeInformation<Row> outputType, ExecutionConfig config,
- InetAddress gatewayAddress, int gatewayPort) {
+ @VisibleForTesting
+ public MaterializedCollectStreamResult(
+ TypeInformation<Row> outputType,
+ ExecutionConfig config,
+ InetAddress gatewayAddress,
+ int gatewayPort,
+ int maxRowCount,
+ int overcommitThreshold) {
super(outputType, config, gatewayAddress, gatewayPort);
+ if (maxRowCount <= 0) {
+ this.maxRowCount = Integer.MAX_VALUE;
+ } else {
+ this.maxRowCount = maxRowCount;
+ }
+
+ this.overcommitThreshold = overcommitThreshold;
+
// prepare for materialization
- materializedTable = new ArrayList<>();
- rowPositionCache = new HashMap<>();
+ final int initialCapacity = computeMaterializedTableCapacity(maxRowCount); // avoid frequent resizing
+ materializedTable = new ArrayList<>(initialCapacity);
+ rowPositionCache = new HashMap<>(initialCapacity);
snapshot = new ArrayList<>();
+ validRowPosition = 0;
isLastSnapshot = false;
pageCount = 0;
}
+ public MaterializedCollectStreamResult(
+ TypeInformation<Row> outputType,
+ ExecutionConfig config,
+ InetAddress gatewayAddress,
+ int gatewayPort,
+ int maxRowCount) {
+
+ this(
+ outputType,
+ config,
+ gatewayAddress,
+ gatewayPort,
+ maxRowCount,
+ computeMaterializedTableOvercommit(maxRowCount));
+ }
+
@Override
public boolean isMaterialized() {
return true;
@@ -74,6 +139,10 @@ public class MaterializedCollectStreamResult<C> extends CollectStreamResult<C> i
@Override
public TypedResult<Integer> snapshot(int pageSize) {
+ if (pageSize < 1) {
+ throw new SqlExecutionException("Page size must be greater than 0.");
+ }
+
synchronized (resultLock) {
// retrieval thread is dead and there are no results anymore
// or program failed
@@ -87,7 +156,9 @@ public class MaterializedCollectStreamResult<C> extends CollectStreamResult<C> i
this.pageSize = pageSize;
snapshot.clear();
- snapshot.addAll(materializedTable);
+ for (int i = validRowPosition; i < materializedTable.size(); i++) {
+ snapshot.add(materializedTable.get(i));
+ }
// at least one page
pageCount = Math.max(1, (int) Math.ceil(((double) snapshot.size() / pageSize)));
@@ -112,31 +183,82 @@ public class MaterializedCollectStreamResult<C> extends CollectStreamResult<C> i
@Override
protected void processRecord(Tuple2<Boolean, Row> change) {
synchronized (resultLock) {
- final Row row = change.f1;
// insert
if (change.f0) {
- materializedTable.add(row);
- rowPositionCache.put(row, materializedTable.size() - 1);
+ processInsert(change.f1);
}
// delete
else {
- // delete the newest record first to minimize per-page changes
- final Integer cachedPos = rowPositionCache.get(row);
- final int startSearchPos;
- if (cachedPos != null) {
- startSearchPos = Math.min(cachedPos, materializedTable.size() - 1);
- } else {
- startSearchPos = materializedTable.size() - 1;
- }
-
- for (int i = startSearchPos; i >= 0; i--) {
- if (materializedTable.get(i).equals(row)) {
- materializedTable.remove(i);
- rowPositionCache.remove(row);
- break;
- }
- }
+ processDelete(change.f1);
}
}
}
+
+ @VisibleForTesting
+ protected List<Row> getMaterializedTable() {
+ return materializedTable;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ private void processInsert(Row row) {
+ // limit the materialized table
+ if (materializedTable.size() - validRowPosition >= maxRowCount) {
+ cleanUp();
+ }
+ materializedTable.add(row);
+ rowPositionCache.put(row, materializedTable.size() - 1);
+ }
+
+ private void processDelete(Row row) {
+ // delete the newest record first to minimize per-page changes
+ final Integer cachedPos = rowPositionCache.get(row);
+ final int startSearchPos;
+ if (cachedPos != null) {
+ startSearchPos = Math.min(cachedPos, materializedTable.size() - 1);
+ } else {
+ startSearchPos = materializedTable.size() - 1;
+ }
+
+ for (int i = startSearchPos; i >= validRowPosition; i--) {
+ if (materializedTable.get(i).equals(row)) {
+ materializedTable.remove(i);
+ rowPositionCache.remove(row);
+ break;
+ }
+ }
+ }
+
+ private void cleanUp() {
+ // invalidate row
+ final Row 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
+ rowPositionCache.remove(deleteRow);
+ }
+ materializedTable.set(validRowPosition, null);
+
+ validRowPosition++;
+
+ // perform clean up in batches
+ if (validRowPosition >= overcommitThreshold) {
+ materializedTable.subList(0, validRowPosition).clear();
+ // adjust all cached indexes
+ rowPositionCache.replaceAll((k, v) -> v - validRowPosition);
+ validRowPosition = 0;
+ }
+ }
+
+ private static int computeMaterializedTableCapacity(int maxRowCount) {
+ return Math.min(
+ MATERIALIZED_TABLE_MAX_INITIAL_CAPACITY,
+ Math.max(1, (int) (maxRowCount * MATERIALIZED_TABLE_CAPACITY_FACTOR)));
+ }
+
+ private static int computeMaterializedTableOvercommit(int maxRowCount) {
+ return Math.min(
+ MATERIALIZED_TABLE_MAX_OVERCOMMIT,
+ (int) (maxRowCount * MATERIALIZED_TABLE_OVERCOMMIT_FACTOR));
+ }
}
diff --git a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
index d8452e4..c5c0f7d 100644
--- a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
+++ b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
@@ -149,6 +149,7 @@ public class LocalExecutorITCase extends TestLogger {
expectedProperties.put("execution.max-idle-state-retention", "0");
expectedProperties.put("execution.min-idle-state-retention", "0");
expectedProperties.put("execution.result-mode", "table");
+ expectedProperties.put("execution.max-table-result-rows", "100");
expectedProperties.put("deployment.response-timeout", "5000");
assertEquals(expectedProperties, actualProperties);
@@ -211,38 +212,47 @@ public class LocalExecutorITCase extends TestLogger {
public void testStreamQueryExecutionTable() throws Exception {
final URL url = getClass().getClassLoader().getResource("test-data.csv");
Objects.requireNonNull(url);
+
final Map<String, String> replaceVars = new HashMap<>();
replaceVars.put("$VAR_0", url.getPath());
replaceVars.put("$VAR_1", "/");
replaceVars.put("$VAR_2", "streaming");
replaceVars.put("$VAR_3", "table");
replaceVars.put("$VAR_UPDATE_MODE", "update-mode: append");
+ replaceVars.put("$VAR_MAX_ROWS", "100");
- final Executor executor = createModifiedExecutor(clusterClient, replaceVars);
- final SessionContext session = new SessionContext("test-session", new Environment());
+ final String query = "SELECT scalarUDF(IntegerField1), StringField1 FROM TableNumber1";
- try {
- // start job and retrieval
- final ResultDescriptor desc = executor.executeQuery(
- session,
- "SELECT scalarUDF(IntegerField1), StringField1 FROM TableNumber1");
+ final List<String> expectedResults = new ArrayList<>();
+ 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!!!!");
+
+ executeStreamQueryTable(replaceVars, query, expectedResults);
+ }
- assertTrue(desc.isMaterialized());
+ @Test(timeout = 30_000L)
+ public void testStreamQueryExecutionLimitedTable() throws Exception {
+ final URL url = getClass().getClassLoader().getResource("test-data.csv");
+ Objects.requireNonNull(url);
- final List<String> actualResults = retrieveTableResult(executor, session, desc.getResultId());
+ final Map<String, String> replaceVars = new HashMap<>();
+ replaceVars.put("$VAR_0", url.getPath());
+ replaceVars.put("$VAR_1", "/");
+ replaceVars.put("$VAR_2", "streaming");
+ replaceVars.put("$VAR_3", "table");
+ replaceVars.put("$VAR_UPDATE_MODE", "update-mode: append");
+ replaceVars.put("$VAR_MAX_ROWS", "1");
- final List<String> expectedResults = new ArrayList<>();
- 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!!!!");
+ final String query = "SELECT COUNT(*), StringField1 FROM TableNumber1 GROUP BY StringField1";
- TestBaseUtils.compareResultCollections(expectedResults, actualResults, Comparator.naturalOrder());
- } finally {
- executor.stop(session);
- }
+ final List<String> expectedResults = new ArrayList<>();
+ expectedResults.add("1,Hello World!!!!");
+
+ executeStreamQueryTable(replaceVars, query, expectedResults);
}
@Test(timeout = 30_000L)
@@ -322,6 +332,28 @@ public class LocalExecutorITCase extends TestLogger {
}
}
+ private void executeStreamQueryTable(
+ Map<String, String> replaceVars,
+ String query,
+ List<String> expectedResults) throws Exception {
+
+ final Executor executor = createModifiedExecutor(clusterClient, replaceVars);
+ final SessionContext session = new SessionContext("test-session", new Environment());
+
+ try {
+ // start job and retrieval
+ final ResultDescriptor desc = executor.executeQuery(session, query);
+
+ assertTrue(desc.isMaterialized());
+
+ final List<String> actualResults = retrieveTableResult(executor, session, desc.getResultId());
+
+ TestBaseUtils.compareResultCollections(expectedResults, actualResults, Comparator.naturalOrder());
+ } finally {
+ executor.stop(session);
+ }
+ }
+
private void verifySinkResult(String path) throws IOException {
final List<String> actualResults = new ArrayList<>();
TestBaseUtils.readAllResultLines(actualResults, path);
@@ -339,6 +371,7 @@ public class LocalExecutorITCase extends TestLogger {
final Map<String, String> replaceVars = new HashMap<>();
replaceVars.put("$VAR_2", "batch");
replaceVars.put("$VAR_UPDATE_MODE", "");
+ replaceVars.put("$VAR_MAX_ROWS", "100");
return new LocalExecutor(
EnvironmentFileUtil.parseModified(DEFAULTS_ENVIRONMENT_FILE, replaceVars),
Collections.emptyList(),
diff --git a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java
index c7e41ff..ba8c924 100644
--- a/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java
+++ b/flink-libraries/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectStreamResultTest.java
@@ -29,7 +29,9 @@ import org.junit.Test;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import static org.junit.Assert.assertEquals;
@@ -42,13 +44,14 @@ public class MaterializedCollectStreamResultTest {
public void testSnapshot() throws UnknownHostException {
final TypeInformation<Row> type = Types.ROW(Types.STRING, Types.LONG);
- TestMaterializedCollectStreamResult result = null;
+ TestMaterializedCollectStreamResult<?> result = null;
try {
- result = new TestMaterializedCollectStreamResult(
+ result = new TestMaterializedCollectStreamResult<>(
type,
new ExecutionConfig(),
InetAddress.getLocalHost(),
- 0);
+ 0,
+ Integer.MAX_VALUE);
result.isRetrieving = true;
@@ -85,11 +88,59 @@ public class MaterializedCollectStreamResultTest {
}
}
+ @Test
+ public void testLimitedSnapshot() throws UnknownHostException {
+ final TypeInformation<Row> type = Types.ROW(Types.STRING, Types.LONG);
+
+ TestMaterializedCollectStreamResult<?> result = null;
+ try {
+ result = new TestMaterializedCollectStreamResult<>(
+ type,
+ new ExecutionConfig(),
+ InetAddress.getLocalHost(),
+ 0,
+ 2, // limit the materialized table to 2 rows
+ 3); // with 3 rows overcommitment
+
+ result.isRetrieving = true;
+
+ result.processRecord(Tuple2.of(true, Row.of("D", 1)));
+ result.processRecord(Tuple2.of(true, Row.of("A", 1)));
+ result.processRecord(Tuple2.of(true, Row.of("B", 1)));
+ result.processRecord(Tuple2.of(true, Row.of("A", 1)));
+
+ assertEquals(
+ Arrays.asList(null, null, Row.of("B", 1), Row.of("A", 1)), // two over-committed rows
+ result.getMaterializedTable());
+
+ 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));
+
+ result.processRecord(Tuple2.of(true, Row.of("C", 1)));
+
+ assertEquals(
+ Arrays.asList(Row.of("A", 1), Row.of("C", 1)), // limit clean up has taken place
+ result.getMaterializedTable());
+
+ result.processRecord(Tuple2.of(false, Row.of("A", 1)));
+
+ assertEquals(
+ Collections.singletonList(Row.of("C", 1)), // regular clean up has taken place
+ result.getMaterializedTable());
+ } finally {
+ if (result != null) {
+ result.close();
+ }
+ }
+ }
+
// --------------------------------------------------------------------------------------------
// Helper classes
// --------------------------------------------------------------------------------------------
- private static class TestMaterializedCollectStreamResult extends MaterializedCollectStreamResult {
+ private static class TestMaterializedCollectStreamResult<T> extends MaterializedCollectStreamResult<T> {
public boolean isRetrieving;
@@ -97,18 +148,42 @@ public class MaterializedCollectStreamResultTest {
TypeInformation<Row> outputType,
ExecutionConfig config,
InetAddress gatewayAddress,
- int gatewayPort) {
+ int gatewayPort,
+ int maxRowCount,
+ int overcommitThreshold) {
super(
outputType,
config,
gatewayAddress,
- gatewayPort);
+ gatewayPort,
+ maxRowCount,
+ overcommitThreshold);
+ }
+
+ public TestMaterializedCollectStreamResult(
+ TypeInformation<Row> outputType,
+ ExecutionConfig config,
+ InetAddress gatewayAddress,
+ int gatewayPort,
+ int maxRowCount) {
+
+ super(
+ outputType,
+ config,
+ gatewayAddress,
+ gatewayPort,
+ maxRowCount);
}
@Override
protected boolean isRetrieving() {
return isRetrieving;
}
+
+ @Override
+ public List<Row> getMaterializedTable() {
+ return super.getMaterializedTable();
+ }
}
}
diff --git a/flink-libraries/flink-sql-client/src/test/resources/test-sql-client-defaults.yaml b/flink-libraries/flink-sql-client/src/test/resources/test-sql-client-defaults.yaml
index b759874..83a27a3 100644
--- a/flink-libraries/flink-sql-client/src/test/resources/test-sql-client-defaults.yaml
+++ b/flink-libraries/flink-sql-client/src/test/resources/test-sql-client-defaults.yaml
@@ -117,6 +117,7 @@ execution:
min-idle-state-retention: 0
max-idle-state-retention: 0
result-mode: "$VAR_3"
+ max-table-result-rows: "$VAR_MAX_ROWS"
deployment:
response-timeout: 5000