You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by op...@apache.org on 2021/06/28 13:00:54 UTC
[iceberg] branch master updated: Flink: Rename FlinkTableOptions to
more generic FlinkConfigOptions
This is an automated email from the ASF dual-hosted git repository.
openinx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new bf2cbc3 Flink: Rename FlinkTableOptions to more generic FlinkConfigOptions
bf2cbc3 is described below
commit bf2cbc3d6cc4d8fbb261cc48de75592f1caa807a
Author: Steven Zhen Wu <st...@gmail.com>
AuthorDate: Mon Jun 28 06:00:35 2021 -0700
Flink: Rename FlinkTableOptions to more generic FlinkConfigOptions
---
.../{FlinkTableOptions.java => FlinkConfigOptions.java} | 4 ++--
.../org/apache/iceberg/flink/source/FlinkSource.java | 17 +++++++++--------
.../java/org/apache/iceberg/flink/FlinkTestBase.java | 2 +-
.../apache/iceberg/flink/source/TestFlinkScanSql.java | 6 +++---
4 files changed, 15 insertions(+), 14 deletions(-)
diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkTableOptions.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
similarity index 96%
rename from flink/src/main/java/org/apache/iceberg/flink/FlinkTableOptions.java
rename to flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
index 145190c..067abe8 100644
--- a/flink/src/main/java/org/apache/iceberg/flink/FlinkTableOptions.java
+++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
@@ -23,9 +23,9 @@ package org.apache.iceberg.flink;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
-public class FlinkTableOptions {
+public class FlinkConfigOptions {
- private FlinkTableOptions() {
+ private FlinkConfigOptions() {
}
public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM =
diff --git a/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java b/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
index 84507c4..a3263d2 100644
--- a/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
+++ b/flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java
@@ -36,8 +36,8 @@ import org.apache.iceberg.Table;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkConfigOptions;
import org.apache.iceberg.flink.FlinkSchemaUtil;
-import org.apache.iceberg.flink.FlinkTableOptions;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
import org.apache.iceberg.io.FileIO;
@@ -48,11 +48,11 @@ public class FlinkSource {
}
/**
- * Initialize a {@link Builder} to read the data from iceberg table. Equivalent to {@link TableScan}.
- * See more options in {@link ScanContext}.
+ * Initialize a {@link Builder} to read the data from iceberg table. Equivalent to {@link TableScan}. See more options
+ * in {@link ScanContext}.
* <p>
- * The Source can be read static data in bounded mode. It can also continuously check the arrival of new data and
- * read records incrementally.
+ * The Source can be read static data in bounded mode. It can also continuously check the arrival of new data and read
+ * records incrementally.
* <ul>
* <li>Without startSnapshotId: Bounded</li>
* <li>With startSnapshotId and with endSnapshotId: Bounded</li>
@@ -222,10 +222,11 @@ public class FlinkSource {
int inferParallelism(FlinkInputFormat format, ScanContext context) {
int parallelism = readableConfig.get(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM);
- if (readableConfig.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
- int maxInferParallelism = readableConfig.get(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
+ if (readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM)) {
+ int maxInferParallelism = readableConfig.get(FlinkConfigOptions
+ .TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX);
Preconditions.checkState(maxInferParallelism >= 1,
- FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
+ FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX.key() + " cannot be less than 1");
int splitNum;
try {
FlinkInputSplit[] splits = format.createInputSplits(0);
diff --git a/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java b/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
index 2810326..1ee000c 100644
--- a/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
+++ b/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
@@ -79,7 +79,7 @@ public abstract class FlinkTestBase extends TestBaseUtils {
.build();
TableEnvironment env = TableEnvironment.create(settings);
- env.getConfig().getConfiguration().set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ env.getConfig().getConfiguration().set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
tEnv = env;
}
}
diff --git a/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java b/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
index f99e4b6..9af1b7c 100644
--- a/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
+++ b/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScanSql.java
@@ -38,7 +38,7 @@ import org.apache.iceberg.data.RandomGenericData;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.flink.FlinkTableOptions;
+import org.apache.iceberg.flink.FlinkConfigOptions;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.TestFixtures;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -158,7 +158,7 @@ public class TestFlinkScanSql extends TestFlinkSource {
// 2 splits and max infer parallelism is 1 (max < splits num), the parallelism is 1
Configuration configuration = new Configuration();
- configuration.setInteger(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX, 1);
+ configuration.setInteger(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX, 1);
parallelism = FlinkSource.forRowData()
.flinkConf(configuration)
.inferParallelism(flinkInputFormat, ScanContext.builder().build());
@@ -171,7 +171,7 @@ public class TestFlinkScanSql extends TestFlinkSource {
Assert.assertEquals("Should produce the expected parallelism.", 1, parallelism);
// 2 splits, infer parallelism is disabled, the parallelism is flink default parallelism 1
- configuration.setBoolean(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ configuration.setBoolean(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
parallelism = FlinkSource.forRowData()
.flinkConf(configuration)
.inferParallelism(flinkInputFormat, ScanContext.builder().limit(3).build());