You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by st...@apache.org on 2024/08/19 22:29:52 UTC
(iceberg) branch main updated: Flink: FLIP-27 IcebergSource builder missed a couple of configs compared to old FlinkSource: expose locality and plan parallelism (#10957)
This is an automated email from the ASF dual-hosted git repository.
stevenwu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 43bbf08adc Flink: FLIP-27 IcebergSource builder missed a couple of configs compared to old FlinkSource: expose locality and plan parallelism (#10957)
43bbf08adc is described below
commit 43bbf08adc63c0a5a5edb26934f62d9994c25af1
Author: Steven Zhen Wu <st...@gmail.com>
AuthorDate: Mon Aug 19 15:29:47 2024 -0700
Flink: FLIP-27 IcebergSource builder missed a couple of configs compared to old FlinkSource: expose locality and plan parallelism (#10957)
---
.../src/main/java/org/apache/iceberg/flink/source/IcebergSource.java | 4 ++++
.../src/main/java/org/apache/iceberg/flink/source/IcebergSource.java | 4 ++++
.../src/main/java/org/apache/iceberg/flink/source/IcebergSource.java | 4 ++++
3 files changed, 12 insertions(+)
diff --git a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
index ccbd0d9997..e629cc19bb 100644
--- a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
+++ b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
@@ -482,6 +482,10 @@ public class IcebergSource<T> implements Source<T, IcebergSourceSplit, IcebergEn
}
contextBuilder.resolveConfig(table, readOptions, flinkConfig);
+ contextBuilder.exposeLocality(
+ SourceUtil.isLocalityEnabled(table, flinkConfig, exposeLocality));
+ contextBuilder.planParallelism(
+ flinkConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE));
Schema icebergSchema = table.schema();
if (projectedFlinkSchema != null) {
contextBuilder.project(FlinkSchemaUtil.convert(icebergSchema, projectedFlinkSchema));
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
index ccbd0d9997..e629cc19bb 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
@@ -482,6 +482,10 @@ public class IcebergSource<T> implements Source<T, IcebergSourceSplit, IcebergEn
}
contextBuilder.resolveConfig(table, readOptions, flinkConfig);
+ contextBuilder.exposeLocality(
+ SourceUtil.isLocalityEnabled(table, flinkConfig, exposeLocality));
+ contextBuilder.planParallelism(
+ flinkConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE));
Schema icebergSchema = table.schema();
if (projectedFlinkSchema != null) {
contextBuilder.project(FlinkSchemaUtil.convert(icebergSchema, projectedFlinkSchema));
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
index ccbd0d9997..e629cc19bb 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergSource.java
@@ -482,6 +482,10 @@ public class IcebergSource<T> implements Source<T, IcebergSourceSplit, IcebergEn
}
contextBuilder.resolveConfig(table, readOptions, flinkConfig);
+ contextBuilder.exposeLocality(
+ SourceUtil.isLocalityEnabled(table, flinkConfig, exposeLocality));
+ contextBuilder.planParallelism(
+ flinkConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE));
Schema icebergSchema = table.schema();
if (projectedFlinkSchema != null) {
contextBuilder.project(FlinkSchemaUtil.convert(icebergSchema, projectedFlinkSchema));