Flink: Fix read split table properties being ignored

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<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 =

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());

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);
+  }
+}