diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java index 1bbd88146c8f..0616d1134951 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java @@ -21,7 +21,6 @@ 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 @@ private FlinkReadOptions() {} public static final ConfigOption 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 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 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 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 STREAMING_OPTION = diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java index 1c6644238c3d..c63bafe3e73b 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/ScanContext.java @@ -29,6 +29,7 @@ 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 static class Builder { 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.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java new file mode 100644 index 000000000000..250dde092d9f --- /dev/null +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkReadConf.java @@ -0,0 +1,90 @@ +/* + * 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 static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +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.Test; + +class TestFlinkReadConf { + + @Test + void splitSizePrecedence() { + Table table = mock(Table.class); + when(table.properties()).thenReturn(ImmutableMap.of(TableProperties.SPLIT_SIZE, "111")); + + // 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); + + // default is used when neither is set + when(table.properties()).thenReturn(ImmutableMap.of()); + conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration()); + assertThat(conf.splitSize()).isEqualTo(TableProperties.SPLIT_SIZE_DEFAULT); + } + + @Test + void splitLookbackPrecedence() { + Table table = mock(Table.class); + when(table.properties()).thenReturn(ImmutableMap.of(TableProperties.SPLIT_LOOKBACK, "5")); + + 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); + + when(table.properties()).thenReturn(ImmutableMap.of()); + conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration()); + assertThat(conf.splitLookback()).isEqualTo(TableProperties.SPLIT_LOOKBACK_DEFAULT); + } + + @Test + void splitOpenFileCostPrecedence() { + Table table = mock(Table.class); + when(table.properties()) + .thenReturn(ImmutableMap.of(TableProperties.SPLIT_OPEN_FILE_COST, "333")); + + 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); + + when(table.properties()).thenReturn(ImmutableMap.of()); + conf = new FlinkReadConf(table, ImmutableMap.of(), new Configuration()); + assertThat(conf.splitFileOpenCost()).isEqualTo(TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT); + } +}