You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by mx...@apache.org on 2026/08/10 15:11:15 UTC
(iceberg) branch main updated: Flink: Backport read split table properties fix to 2.0 and 1.20 (#17445) (#17591)
This is an automated email from the ASF dual-hosted git repository.
mxm 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 89e2f88749 Flink: Backport read split table properties fix to 2.0 and 1.20 (#17445) (#17591)
89e2f88749 is described below
commit 89e2f887491c1b5fa9f8b9de81b3aa8b31fa6974
Author: vishnu prakash <nv...@gmail.com>
AuthorDate: Mon Aug 10 20:41:09 2026 +0530
Flink: Backport read split table properties fix to 2.0 and 1.20 (#17445) (#17591)
---
.../org/apache/iceberg/flink/FlinkReadOptions.java | 15 +--
.../apache/iceberg/flink/source/ScanContext.java | 7 +-
.../apache/iceberg/flink/TestFlinkReadConf.java | 116 +++++++++++++++++++++
.../org/apache/iceberg/flink/FlinkReadOptions.java | 15 +--
.../apache/iceberg/flink/source/ScanContext.java | 7 +-
.../apache/iceberg/flink/TestFlinkReadConf.java | 116 +++++++++++++++++++++
6 files changed, 250 insertions(+), 26 deletions(-)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java
index 1bbd88146c..0616d11349 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java
@@ -21,7 +21,6 @@ package org.apache.iceberg.flink;
import java.util.concurrent.TimeUnit;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
-import org.apache.iceberg.TableProperties;
import org.apache.iceberg.flink.source.StreamingStartingStrategy;
/** Flink source read options */
@@ -67,23 +66,19 @@ public class FlinkReadOptions {
public static final ConfigOption<Long> END_SNAPSHOT_ID =
ConfigOptions.key("end-snapshot-id").longType().defaultValue(null);
+ // No defaults for the split options: a declared default would mask the table property fallback
+ // in FlinkReadConf, which applies the default instead
public static final String SPLIT_SIZE = "split-size";
public static final ConfigOption<Long> SPLIT_SIZE_OPTION =
- ConfigOptions.key(PREFIX + SPLIT_SIZE)
- .longType()
- .defaultValue(TableProperties.SPLIT_SIZE_DEFAULT);
+ ConfigOptions.key(PREFIX + SPLIT_SIZE).longType().noDefaultValue();
public static final String SPLIT_LOOKBACK = "split-lookback";
public static final ConfigOption<Integer> SPLIT_LOOKBACK_OPTION =
- ConfigOptions.key(PREFIX + SPLIT_LOOKBACK)
- .intType()
- .defaultValue(TableProperties.SPLIT_LOOKBACK_DEFAULT);
+ ConfigOptions.key(PREFIX + SPLIT_LOOKBACK).intType().noDefaultValue();
public static final String SPLIT_FILE_OPEN_COST = "split-file-open-cost";
public static final ConfigOption<Long> SPLIT_FILE_OPEN_COST_OPTION =
- ConfigOptions.key(PREFIX + SPLIT_FILE_OPEN_COST)
- .longType()
- .defaultValue(TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT);
+ ConfigOptions.key(PREFIX + SPLIT_FILE_OPEN_COST).longType().noDefaultValue();
public static final String STREAMING = "streaming";
public static final ConfigOption<Boolean> STREAMING_OPTION =
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
index 1c6644238c..c63bafe3e7 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
@@ -29,6 +29,7 @@ import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.util.TimeUtils;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.flink.FlinkConfigOptions;
import org.apache.iceberg.flink.FlinkReadConf;
@@ -369,9 +370,9 @@ public class ScanContext implements Serializable {
private Long startSnapshotId = FlinkReadOptions.START_SNAPSHOT_ID.defaultValue();
private Long endSnapshotId = FlinkReadOptions.END_SNAPSHOT_ID.defaultValue();
private Long asOfTimestamp = FlinkReadOptions.AS_OF_TIMESTAMP.defaultValue();
- private Long splitSize = FlinkReadOptions.SPLIT_SIZE_OPTION.defaultValue();
- private Integer splitLookback = FlinkReadOptions.SPLIT_LOOKBACK_OPTION.defaultValue();
- private Long splitOpenFileCost = FlinkReadOptions.SPLIT_FILE_OPEN_COST_OPTION.defaultValue();
+ private Long splitSize = TableProperties.SPLIT_SIZE_DEFAULT;
+ private Integer splitLookback = TableProperties.SPLIT_LOOKBACK_DEFAULT;
+ private Long splitOpenFileCost = TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT;
private boolean isStreaming = FlinkReadOptions.STREAMING_OPTION.defaultValue();
private Duration monitorInterval =
TimeUtils.parseDuration(FlinkReadOptions.MONITOR_INTERVAL_OPTION.defaultValue());
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java
new file mode 100644
index 0000000000..8f5395fbf6
--- /dev/null
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java
@@ -0,0 +1,116 @@
+/*
+ * 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.iceberg.flink;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import org.apache.flink.configuration.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestFlinkReadConf {
+
+ @TempDir private Path temporaryFolder;
+
+ private Table table;
+ private Table tableWithoutProperties;
+
+ @BeforeEach
+ void before() throws IOException {
+ table =
+ createTable(
+ ImmutableMap.of(
+ TableProperties.SPLIT_SIZE, "111",
+ TableProperties.SPLIT_LOOKBACK, "5",
+ TableProperties.SPLIT_OPEN_FILE_COST, "333"));
+ tableWithoutProperties = createTable(ImmutableMap.of());
+ }
+
+ @Test
+ void splitSizePrecedence() {
+ // table property is used when the flink config does not set the option
+ FlinkReadConf conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitSize()).isEqualTo(111L);
+
+ // flink config takes precedence over the table property
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(FlinkReadOptions.SPLIT_SIZE_OPTION, 222L);
+ conf = new FlinkReadConf(table, ImmutableMap.of(), flinkConf);
+ assertThat(conf.splitSize()).isEqualTo(222L);
+
+ // read options take precedence over the flink config and the table property
+ conf = new FlinkReadConf(table, ImmutableMap.of(FlinkReadOptions.SPLIT_SIZE, "999"), flinkConf);
+ assertThat(conf.splitSize()).isEqualTo(999L);
+
+ // default is used when neither is set
+ conf = new FlinkReadConf(tableWithoutProperties, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitSize()).isEqualTo(TableProperties.SPLIT_SIZE_DEFAULT);
+ }
+
+ @Test
+ void splitLookbackPrecedence() {
+ FlinkReadConf conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitLookback()).isEqualTo(5);
+
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(FlinkReadOptions.SPLIT_LOOKBACK_OPTION, 7);
+ conf = new FlinkReadConf(table, ImmutableMap.of(), flinkConf);
+ assertThat(conf.splitLookback()).isEqualTo(7);
+
+ conf =
+ new FlinkReadConf(table, ImmutableMap.of(FlinkReadOptions.SPLIT_LOOKBACK, "9"), flinkConf);
+ assertThat(conf.splitLookback()).isEqualTo(9);
+
+ conf = new FlinkReadConf(tableWithoutProperties, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitLookback()).isEqualTo(TableProperties.SPLIT_LOOKBACK_DEFAULT);
+ }
+
+ @Test
+ void splitOpenFileCostPrecedence() {
+ FlinkReadConf conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitFileOpenCost()).isEqualTo(333L);
+
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(FlinkReadOptions.SPLIT_FILE_OPEN_COST_OPTION, 444L);
+ conf = new FlinkReadConf(table, ImmutableMap.of(), flinkConf);
+ assertThat(conf.splitFileOpenCost()).isEqualTo(444L);
+
+ conf =
+ new FlinkReadConf(
+ table, ImmutableMap.of(FlinkReadOptions.SPLIT_FILE_OPEN_COST, "999"), flinkConf);
+ assertThat(conf.splitFileOpenCost()).isEqualTo(999L);
+
+ conf = new FlinkReadConf(tableWithoutProperties, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitFileOpenCost()).isEqualTo(TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT);
+ }
+
+ private Table createTable(Map<String, String> properties) throws IOException {
+ File folder = Files.createTempDirectory(temporaryFolder, "junit").toFile();
+ return SimpleDataUtil.createTable(folder.getAbsolutePath(), properties, false);
+ }
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java
index 1bbd88146c..0616d11349 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java
@@ -21,7 +21,6 @@ package org.apache.iceberg.flink;
import java.util.concurrent.TimeUnit;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
-import org.apache.iceberg.TableProperties;
import org.apache.iceberg.flink.source.StreamingStartingStrategy;
/** Flink source read options */
@@ -67,23 +66,19 @@ public class FlinkReadOptions {
public static final ConfigOption<Long> END_SNAPSHOT_ID =
ConfigOptions.key("end-snapshot-id").longType().defaultValue(null);
+ // No defaults for the split options: a declared default would mask the table property fallback
+ // in FlinkReadConf, which applies the default instead
public static final String SPLIT_SIZE = "split-size";
public static final ConfigOption<Long> SPLIT_SIZE_OPTION =
- ConfigOptions.key(PREFIX + SPLIT_SIZE)
- .longType()
- .defaultValue(TableProperties.SPLIT_SIZE_DEFAULT);
+ ConfigOptions.key(PREFIX + SPLIT_SIZE).longType().noDefaultValue();
public static final String SPLIT_LOOKBACK = "split-lookback";
public static final ConfigOption<Integer> SPLIT_LOOKBACK_OPTION =
- ConfigOptions.key(PREFIX + SPLIT_LOOKBACK)
- .intType()
- .defaultValue(TableProperties.SPLIT_LOOKBACK_DEFAULT);
+ ConfigOptions.key(PREFIX + SPLIT_LOOKBACK).intType().noDefaultValue();
public static final String SPLIT_FILE_OPEN_COST = "split-file-open-cost";
public static final ConfigOption<Long> SPLIT_FILE_OPEN_COST_OPTION =
- ConfigOptions.key(PREFIX + SPLIT_FILE_OPEN_COST)
- .longType()
- .defaultValue(TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT);
+ ConfigOptions.key(PREFIX + SPLIT_FILE_OPEN_COST).longType().noDefaultValue();
public static final String STREAMING = "streaming";
public static final ConfigOption<Boolean> STREAMING_OPTION =
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
index 1c6644238c..c63bafe3e7 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java
@@ -29,6 +29,7 @@ import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.util.TimeUtils;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.flink.FlinkConfigOptions;
import org.apache.iceberg.flink.FlinkReadConf;
@@ -369,9 +370,9 @@ public class ScanContext implements Serializable {
private Long startSnapshotId = FlinkReadOptions.START_SNAPSHOT_ID.defaultValue();
private Long endSnapshotId = FlinkReadOptions.END_SNAPSHOT_ID.defaultValue();
private Long asOfTimestamp = FlinkReadOptions.AS_OF_TIMESTAMP.defaultValue();
- private Long splitSize = FlinkReadOptions.SPLIT_SIZE_OPTION.defaultValue();
- private Integer splitLookback = FlinkReadOptions.SPLIT_LOOKBACK_OPTION.defaultValue();
- private Long splitOpenFileCost = FlinkReadOptions.SPLIT_FILE_OPEN_COST_OPTION.defaultValue();
+ private Long splitSize = TableProperties.SPLIT_SIZE_DEFAULT;
+ private Integer splitLookback = TableProperties.SPLIT_LOOKBACK_DEFAULT;
+ private Long splitOpenFileCost = TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT;
private boolean isStreaming = FlinkReadOptions.STREAMING_OPTION.defaultValue();
private Duration monitorInterval =
TimeUtils.parseDuration(FlinkReadOptions.MONITOR_INTERVAL_OPTION.defaultValue());
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java
new file mode 100644
index 0000000000..8f5395fbf6
--- /dev/null
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java
@@ -0,0 +1,116 @@
+/*
+ * 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.iceberg.flink;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import org.apache.flink.configuration.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestFlinkReadConf {
+
+ @TempDir private Path temporaryFolder;
+
+ private Table table;
+ private Table tableWithoutProperties;
+
+ @BeforeEach
+ void before() throws IOException {
+ table =
+ createTable(
+ ImmutableMap.of(
+ TableProperties.SPLIT_SIZE, "111",
+ TableProperties.SPLIT_LOOKBACK, "5",
+ TableProperties.SPLIT_OPEN_FILE_COST, "333"));
+ tableWithoutProperties = createTable(ImmutableMap.of());
+ }
+
+ @Test
+ void splitSizePrecedence() {
+ // table property is used when the flink config does not set the option
+ FlinkReadConf conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitSize()).isEqualTo(111L);
+
+ // flink config takes precedence over the table property
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(FlinkReadOptions.SPLIT_SIZE_OPTION, 222L);
+ conf = new FlinkReadConf(table, ImmutableMap.of(), flinkConf);
+ assertThat(conf.splitSize()).isEqualTo(222L);
+
+ // read options take precedence over the flink config and the table property
+ conf = new FlinkReadConf(table, ImmutableMap.of(FlinkReadOptions.SPLIT_SIZE, "999"), flinkConf);
+ assertThat(conf.splitSize()).isEqualTo(999L);
+
+ // default is used when neither is set
+ conf = new FlinkReadConf(tableWithoutProperties, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitSize()).isEqualTo(TableProperties.SPLIT_SIZE_DEFAULT);
+ }
+
+ @Test
+ void splitLookbackPrecedence() {
+ FlinkReadConf conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitLookback()).isEqualTo(5);
+
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(FlinkReadOptions.SPLIT_LOOKBACK_OPTION, 7);
+ conf = new FlinkReadConf(table, ImmutableMap.of(), flinkConf);
+ assertThat(conf.splitLookback()).isEqualTo(7);
+
+ conf =
+ new FlinkReadConf(table, ImmutableMap.of(FlinkReadOptions.SPLIT_LOOKBACK, "9"), flinkConf);
+ assertThat(conf.splitLookback()).isEqualTo(9);
+
+ conf = new FlinkReadConf(tableWithoutProperties, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitLookback()).isEqualTo(TableProperties.SPLIT_LOOKBACK_DEFAULT);
+ }
+
+ @Test
+ void splitOpenFileCostPrecedence() {
+ FlinkReadConf conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitFileOpenCost()).isEqualTo(333L);
+
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(FlinkReadOptions.SPLIT_FILE_OPEN_COST_OPTION, 444L);
+ conf = new FlinkReadConf(table, ImmutableMap.of(), flinkConf);
+ assertThat(conf.splitFileOpenCost()).isEqualTo(444L);
+
+ conf =
+ new FlinkReadConf(
+ table, ImmutableMap.of(FlinkReadOptions.SPLIT_FILE_OPEN_COST, "999"), flinkConf);
+ assertThat(conf.splitFileOpenCost()).isEqualTo(999L);
+
+ conf = new FlinkReadConf(tableWithoutProperties, ImmutableMap.of(), new Configuration());
+ assertThat(conf.splitFileOpenCost()).isEqualTo(TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT);
+ }
+
+ private Table createTable(Map<String, String> properties) throws IOException {
+ File folder = Files.createTempDirectory(temporaryFolder, "junit").toFile();
+ return SimpleDataUtil.createTable(folder.getAbsolutePath(), properties, false);
+ }
+}