updating fork

.asf.yaml

@@ -35,6 +35,9 @@ github:
 
   protected_branches:
     main:
+      required_pull_request_reviews:
+        required_approving_review_count: 1
+
       required_linear_history: true
 
   features:

.baseline/checkstyle/checkstyle.xml

@@ -145,7 +145,9 @@
                 org.apache.spark.sql.functions.*,
                 org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command.*,
                 org.apache.spark.sql.connector.write.RowLevelOperation.Command.*,
-                org.junit.Assert.*"/>
+                org.junit.Assert.*,
+                org.assertj.core.api.Assertions.*,
+                org.assertj.core.api.Assumptions.*"/>
         </module>
         <module name="ClassTypeParameterName"> <!-- Java Style Guide: Type variable names -->
             <property name="format" value="(^[A-Z][0-9]?)$|([A-Z][a-zA-Z0-9]*[T]$)"/>

.github/ISSUE_TEMPLATE/iceberg_bug_report.yml

@@ -28,7 +28,9 @@ body:
       description: What Apache Iceberg version are you using?
       multiple: false
       options:
-        - "1.5.0 (latest release)"
+        - "1.5.2 (latest release)"
+        - "1.5.1"
+        - "1.5.0"
         - "1.4.3"
         - "1.4.2"
         - "1.4.1"

.github/workflows/flink-ci.yml

@@ -72,7 +72,7 @@ jobs:
     strategy:
       matrix:
         jvm: [8, 11]
-        flink: ['1.16', '1.17', '1.18']
+        flink: ['1.17', '1.18', '1.19']
     env:
       SPARK_LOCAL_IP: localhost
     steps:

.github/workflows/site-ci.yml

@@ -22,6 +22,7 @@ on:
     branches:
       - main
     paths:
+      - docs/**
       - site/**
   workflow_dispatch:
 jobs:

.github/workflows/stale.yml

@@ -17,7 +17,7 @@
 # under the License.
 #
 
-name: "Close Stale Issues"
+name: "Close Stale Issues and PRs"
 on:
   schedule:
     - cron: '0 0 * * *'
@@ -33,12 +33,11 @@ jobs:
     steps:
       - uses: actions/stale@v9.0.0
         with:
+          # stale issues
           stale-issue-label: 'stale'
           exempt-issue-labels: 'not-stale'
           days-before-issue-stale: 180
           days-before-issue-close: 14
-          # Only close stale issues, leave PRs alone
-          days-before-pr-stale: -1
           stale-issue-message: >
             This issue has been automatically marked as stale because it has been open for 180 days
             with no activity. It will be closed in next 14 days if no further activity occurs. To
@@ -47,5 +46,12 @@ jobs:
           close-issue-message: >
             This issue has been closed because it has not received any activity in the last 14 days
             since being marked as 'stale'
+          # stale PRs
+          stale-pr-label: 'stale'
+          exempt-pr-labels: 'not-stale,security'
+          stale-pr-message: 'This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.'
+          close-pr-message: 'This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.'
+          days-before-pr-stale: 30
+          days-before-pr-close: 7
           ascending: true
           operations-per-run: 100

.gitignore

@@ -24,6 +24,7 @@ gradle/wrapper/gradle-wrapper.jar
 lib/
 
 # web site build
+docs/site/
 site/site/
 site/docs/docs/
 site/docs/.asf.yaml

api/src/main/java/org/apache/iceberg/FileFormat.java

@@ -32,6 +32,8 @@ public enum FileFormat {
   private final String ext;
   private final boolean splittable;
 
+  private static final FileFormat[] VALUES = values();
+
   FileFormat(String ext, boolean splittable) {
     this.ext = "." + ext;
     this.splittable = splittable;
@@ -55,7 +57,7 @@ public String addExtension(String filename) {
   }
 
   public static FileFormat fromFileName(CharSequence filename) {
-    for (FileFormat format : FileFormat.values()) {
+    for (FileFormat format : VALUES) {
       int extStart = filename.length() - format.ext.length();
       if (Comparators.charSequences()
               .compare(format.ext, filename.subSequence(extStart, filename.length()))

api/src/main/java/org/apache/iceberg/PartitionSpec.java

@@ -189,7 +189,7 @@ public String partitionToPath(StructLike data) {
       if (i > 0) {
         sb.append("/");
       }
-      sb.append(field.name()).append("=").append(escape(valueString));
+      sb.append(escape(field.name())).append("=").append(escape(valueString));
     }
     return sb.toString();
   }

api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java

@@ -52,6 +52,13 @@ public interface RewriteDataFiles
 
   int PARTIAL_PROGRESS_MAX_COMMITS_DEFAULT = 10;
 
+  /**
+   * The maximum amount of failed commits that this rewrite is allowed if partial progress is
+   * enabled. By default, all commits are allowed to fail. This setting has no effect if partial
+   * progress is disabled.
+   */
+  String PARTIAL_PROGRESS_MAX_FAILED_COMMITS = "partial-progress.max-failed-commits";
+
   /**
    * The entire rewrite operation is broken down into pieces based on partitioning and within
    * partitions based on size into groups. These sub-units of the rewrite are referred to as file

api/src/main/java/org/apache/iceberg/expressions/BoundAggregate.java

@@ -87,6 +87,13 @@ public String describe() {
     }
   }
 
+  <V> boolean safeContainsKey(Map<Integer, V> map, int key) {
+    if (map == null) {
+      return false;
+    }
+    return map.containsKey(key);
+  }
+
   <V> V safeGet(Map<Integer, V> map, int key) {
     return safeGet(map, key, null);
   }

api/src/main/java/org/apache/iceberg/expressions/CountNonNull.java

@@ -39,7 +39,8 @@ protected Long countFor(StructLike row) {
 
   @Override
   protected boolean hasValue(DataFile file) {
-    return file.valueCounts().containsKey(fieldId) && file.nullValueCounts().containsKey(fieldId);
+    return safeContainsKey(file.valueCounts(), fieldId)
+        && file.nullValueCounts().containsKey(fieldId);
   }
 
   @Override

api/src/main/java/org/apache/iceberg/expressions/MaxAggregate.java

@@ -40,7 +40,7 @@ protected MaxAggregate(BoundTerm<T> term) {
 
   @Override
   protected boolean hasValue(DataFile file) {
-    boolean hasBound = file.upperBounds().containsKey(fieldId);
+    boolean hasBound = safeContainsKey(file.upperBounds(), fieldId);
     Long valueCount = safeGet(file.valueCounts(), fieldId);
     Long nullCount = safeGet(file.nullValueCounts(), fieldId);
     boolean boundAllNull =

api/src/main/java/org/apache/iceberg/expressions/MinAggregate.java

@@ -40,7 +40,7 @@ protected MinAggregate(BoundTerm<T> term) {
 
   @Override
   protected boolean hasValue(DataFile file) {
-    boolean hasBound = file.lowerBounds().containsKey(fieldId);
+    boolean hasBound = safeContainsKey(file.lowerBounds(), fieldId);
     Long valueCount = safeGet(file.valueCounts(), fieldId);
     Long nullCount = safeGet(file.nullValueCounts(), fieldId);
     boolean boundAllNull =

api/src/main/java/org/apache/iceberg/io/FileIO.java

@@ -49,25 +49,25 @@ default InputFile newInputFile(String path, long length) {
   default InputFile newInputFile(DataFile file) {
     Preconditions.checkArgument(
         file.keyMetadata() == null,
-        "Cannot decrypt data file: {} (use EncryptingFileIO)",
+        "Cannot decrypt data file: %s (use EncryptingFileIO)",
         file.path());
-    return newInputFile(file.path().toString());
+    return newInputFile(file.path().toString(), file.fileSizeInBytes());
   }
 
   default InputFile newInputFile(DeleteFile file) {
     Preconditions.checkArgument(
         file.keyMetadata() == null,
-        "Cannot decrypt delete file: {} (use EncryptingFileIO)",
+        "Cannot decrypt delete file: %s (use EncryptingFileIO)",
         file.path());
-    return newInputFile(file.path().toString());
+    return newInputFile(file.path().toString(), file.fileSizeInBytes());
   }
 
   default InputFile newInputFile(ManifestFile manifest) {
     Preconditions.checkArgument(
         manifest.keyMetadata() == null,
-        "Cannot decrypt manifest: {} (use EncryptingFileIO)",
+        "Cannot decrypt manifest: %s (use EncryptingFileIO)",
         manifest.path());
-    return newInputFile(manifest.path());
+    return newInputFile(manifest.path(), manifest.length());
   }
 
   /** Get a {@link OutputFile} instance to write bytes to the file at the given path. */

api/src/main/java/org/apache/iceberg/types/JavaHashes.java

@@ -26,6 +26,10 @@ public class JavaHashes {
   private JavaHashes() {}
 
   public static int hashCode(CharSequence str) {
+    if (null == str) {
+      return 0;
+    }
+
     int result = 177;
     for (int i = 0; i < str.length(); i += 1) {
       char ch = str.charAt(i);

api/src/main/java/org/apache/iceberg/util/CharSequenceWrapper.java

@@ -29,13 +29,20 @@ public static CharSequenceWrapper wrap(CharSequence seq) {
   }
 
   private CharSequence wrapped;
+  // lazily computed & cached hashCode
+  private transient int hashCode = 0;
+  // tracks if the hash has been calculated as actually being zero to avoid re-calculating the hash.
+  // this follows the hashCode() implementation from java.lang.String
+  private transient boolean hashIsZero = false;
 
   private CharSequenceWrapper(CharSequence wrapped) {
     this.wrapped = wrapped;
   }
 
   public CharSequenceWrapper set(CharSequence newWrapped) {
     this.wrapped = newWrapped;
+    this.hashCode = 0;
+    this.hashIsZero = false;
     return this;
   }
 
@@ -58,6 +65,10 @@ public boolean equals(Object other) {
       return wrapped.equals(that.wrapped);
     }
 
+    if (null == wrapped && null == that.wrapped) {
+      return true;
+    }
+
     if (length() != that.length()) {
       return false;
     }
@@ -67,7 +78,19 @@ public boolean equals(Object other) {
 
   @Override
   public int hashCode() {
-    return JavaHashes.hashCode(wrapped);
+    int hash = hashCode;
+
+    // don't recalculate if the hash is actually 0
+    if (hash == 0 && !hashIsZero) {
+      hash = JavaHashes.hashCode(wrapped);
+      if (hash == 0) {
+        hashIsZero = true;
+      } else {
+        this.hashCode = hash;
+      }
+    }
+
+    return hash;
   }
 
   @Override

api/src/test/java/org/apache/iceberg/TestPartitionPaths.java

@@ -32,7 +32,8 @@ public class TestPartitionPaths {
       new Schema(
           Types.NestedField.required(1, "id", Types.IntegerType.get()),
           Types.NestedField.optional(2, "data", Types.StringType.get()),
-          Types.NestedField.optional(3, "ts", Types.TimestampType.withoutZone()));
+          Types.NestedField.optional(3, "ts", Types.TimestampType.withoutZone()),
+          Types.NestedField.optional(4, "\"esc\"#1", Types.StringType.get()));
 
   @Test
   public void testPartitionPath() {
@@ -62,4 +63,13 @@ public void testEscapedStrings() {
         .as("Should escape / as %2F")
         .isEqualTo("data=a%2Fb%2Fc%2Fd/data_trunc=a%2Fb%2Fc%2Fd");
   }
+
+  @Test
+  public void testEscapedFieldNames() {
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("\"esc\"#1").build();
+
+    assertThat(spec.partitionToPath(Row.of("a/b/c/d")))
+        .as("Should escape \" as %22 and # as %23")
+        .isEqualTo("%22esc%22%231=a%2Fb%2Fc%2Fd");
+  }
 }

api/src/test/java/org/apache/iceberg/expressions/TestAggregateEvaluator.java

@@ -95,6 +95,22 @@ public class TestAggregateEvaluator {
     FILE, MISSING_SOME_NULLS_STATS_1, MISSING_SOME_NULLS_STATS_2
   };
 
+  private static final DataFile MISSING_ALL_OPTIONAL_STATS =
+      new TestDataFile(
+          "file_null_stats.avro",
+          Row.of(),
+          20,
+          // any value counts, including nulls
+          null,
+          // null value counts
+          null,
+          // nan value counts
+          null,
+          // lower bounds
+          null,
+          // upper bounds
+          null);
+
   @Test
   public void testIntAggregate() {
     List<Expression> list =
@@ -173,6 +189,42 @@ public void testNoStats() {
     assertEvaluatorResult(result, expected);
   }
 
+  @Test
+  public void testIntAggregateAllMissingStats() {
+    List<Expression> list =
+        ImmutableList.of(
+            Expressions.countStar(),
+            Expressions.count("id"),
+            Expressions.max("id"),
+            Expressions.min("id"));
+    AggregateEvaluator aggregateEvaluator = AggregateEvaluator.create(SCHEMA, list);
+
+    aggregateEvaluator.update(MISSING_ALL_OPTIONAL_STATS);
+
+    assertThat(aggregateEvaluator.allAggregatorsValid()).isFalse();
+    StructLike result = aggregateEvaluator.result();
+    Object[] expected = {20L, null, null, null};
+    assertEvaluatorResult(result, expected);
+  }
+
+  @Test
+  public void testOptionalColAllMissingStats() {
+    List<Expression> list =
+        ImmutableList.of(
+            Expressions.countStar(),
+            Expressions.count("no_stats"),
+            Expressions.max("no_stats"),
+            Expressions.min("no_stats"));
+    AggregateEvaluator aggregateEvaluator = AggregateEvaluator.create(SCHEMA, list);
+
+    aggregateEvaluator.update(MISSING_ALL_OPTIONAL_STATS);
+
+    assertThat(aggregateEvaluator.allAggregatorsValid()).isFalse();
+    StructLike result = aggregateEvaluator.result();
+    Object[] expected = {20L, null, null, null};
+    assertEvaluatorResult(result, expected);
+  }
+
   private void assertEvaluatorResult(StructLike result, Object[] expected) {
     Object[] actual = new Object[result.size()];
     for (int i = 0; i < result.size(); i++) {

api/src/test/java/org/apache/iceberg/util/TestCharSequenceMap.java

@@ -25,10 +25,17 @@
 import java.util.concurrent.TimeUnit;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class TestCharSequenceMap {
 
+  @Test
+  public void nullString() {
+    Assertions.assertThat(CharSequenceMap.create()).doesNotContainKey((String) null);
+    Assertions.assertThat(CharSequenceMap.create()).doesNotContainValue((String) null);
+  }
+
   @Test
   public void testEmptyMap() {
     CharSequenceMap<String> map = CharSequenceMap.create();

api/src/test/java/org/apache/iceberg/util/TestCharSequenceSet.java

@@ -39,6 +39,12 @@ public void testSearchingInCharSequenceCollection() {
     Assertions.assertThat(set.contains("def")).isTrue();
   }
 
+  @Test
+  public void nullString() {
+    Assertions.assertThat(CharSequenceSet.of(Arrays.asList((String) null))).contains((String) null);
+    Assertions.assertThat(CharSequenceSet.empty()).doesNotContain((String) null);
+  }
+
   @Test
   public void testRetainAll() {
     CharSequenceSet set = CharSequenceSet.of(ImmutableList.of("123", "456"));