[core] fix NPE and ArrayIndexOutOfBoundsException for PartitionExpire

Author: LsomeYeah

paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java

@@ -185,7 +185,8 @@ private List<Map<String, String>> convertToPartitionString(
         return expiredPartValues.stream()
                 .map(values -> String.join(DELIMITER, values))
                 .sorted()
-                .map(s -> s.split(DELIMITER))
+                // Use split(DELIMITER, -1) to preserve trailing empty strings
+                .map(s -> s.split(DELIMITER, -1))
                 .map(strategy::toPartitionString)
                 .limit(Math.min(expiredPartValues.size(), maxExpireNum))
                 .collect(Collectors.toList());

paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java

@@ -39,11 +39,13 @@
 public abstract class PartitionExpireStrategy {
 
     protected final List<String> partitionKeys;
+    protected final String partitionDefaultName;
     private final RowDataToObjectArrayConverter toObjectArrayConverter;
 
-    public PartitionExpireStrategy(RowType partitionType) {
+    public PartitionExpireStrategy(RowType partitionType, String partitionDefaultName) {
         this.toObjectArrayConverter = new RowDataToObjectArrayConverter(partitionType);
         this.partitionKeys = partitionType.getFieldNames();
+        this.partitionDefaultName = partitionDefaultName;
     }
 
     public Map<String, String> toPartitionString(Object[] array) {
@@ -57,7 +59,11 @@ public Map<String, String> toPartitionString(Object[] array) {
     public List<String> toPartitionValue(Object[] array) {
         List<String> list = new ArrayList<>(partitionKeys.size());
         for (int i = 0; i < partitionKeys.size(); i++) {
-            list.add(array[i].toString());
+            if (array[i] != null) {
+                list.add(array[i].toString());
+            } else {
+                list.add(partitionDefaultName);
+            }
         }
         return list;
     }
@@ -76,13 +82,13 @@ public static PartitionExpireStrategy createPartitionExpireStrategy(
             @Nullable Identifier identifier) {
         switch (options.partitionExpireStrategy()) {
             case UPDATE_TIME:
-                return new PartitionUpdateTimeExpireStrategy(partitionType);
+                return new PartitionUpdateTimeExpireStrategy(options, partitionType);
             case VALUES_TIME:
                 return new PartitionValuesTimeExpireStrategy(options, partitionType);
             case CUSTOM:
                 return PartitionExpireStrategyFactory.INSTANCE
                         .get()
-                        .create(catalogLoader, identifier, partitionType);
+                        .create(catalogLoader, identifier, options, partitionType);
             default:
                 throw new IllegalArgumentException(
                         "Unknown partitionExpireStrategy: " + options.partitionExpireStrategy());

paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategyFactory.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.partition;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.factories.FactoryUtil;
@@ -30,7 +31,10 @@
 public interface PartitionExpireStrategyFactory {
 
     PartitionExpireStrategy create(
-            CatalogLoader catalogLoader, Identifier identifier, RowType partitionType);
+            CatalogLoader catalogLoader,
+            Identifier identifier,
+            CoreOptions options,
+            RowType partitionType);
 
     Supplier<PartitionExpireStrategyFactory> INSTANCE =
             Suppliers.memoize(

paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.partition;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.operation.FileStoreScan;
 import org.apache.paimon.types.RowType;
@@ -33,8 +34,8 @@
  */
 public class PartitionUpdateTimeExpireStrategy extends PartitionExpireStrategy {
 
-    public PartitionUpdateTimeExpireStrategy(RowType partitionType) {
-        super(partitionType);
+    public PartitionUpdateTimeExpireStrategy(CoreOptions options, RowType partitionType) {
+        super(partitionType, options.partitionDefaultName());
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java

@@ -48,7 +48,7 @@ public class PartitionValuesTimeExpireStrategy extends PartitionExpireStrategy {
     private final PartitionTimeExtractor timeExtractor;
 
     public PartitionValuesTimeExpireStrategy(CoreOptions options, RowType partitionType) {
-        super(partitionType);
+        super(partitionType, options.partitionDefaultName());
         String timePattern = options.partitionTimestampPattern();
         String timeFormatter = options.partitionTimestampFormatter();
         this.timeExtractor = new PartitionTimeExtractor(timePattern, timeFormatter);

paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java

@@ -61,6 +61,7 @@
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -217,6 +218,34 @@ public void testBatchExpire() throws Exception {
         assertThat(overwriteSnapshotCnt).isEqualTo(3L);
     }
 
+    @Test
+    public void testExpireWithNullOrEmptyPartition() throws Exception {
+        SchemaManager schemaManager = new SchemaManager(LocalFileIO.create(), path);
+        schemaManager.createTable(
+                new Schema(
+                        RowType.of(VarCharType.STRING_TYPE, VarCharType.STRING_TYPE).getFields(),
+                        Arrays.asList("f0", "f1"),
+                        emptyList(),
+                        Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), "true"),
+                        ""));
+        newTable();
+        write("20230101", "11");
+        write("20230101", "12");
+        // sub partition is null
+        write("20230101", null);
+        // sub partition is empty string
+        write("20230103", "");
+        write("20230103", "32");
+        write("20230105", "51");
+
+        PartitionExpire expire = newExpire();
+        expire.setLastCheck(date(1));
+        Assertions.assertDoesNotThrow(() -> expire.expire(date(6), Long.MAX_VALUE));
+
+        // null partition and empty string partition should be expired
+        assertThat(read()).containsExactlyInAnyOrder("20230105:51");
+    }
+
     @Test
     public void test() throws Exception {
         SchemaManager schemaManager = new SchemaManager(LocalFileIO.create(), path);

paimon-core/src/test/java/org/apache/paimon/partition/CustomPartitionExpirationFactory.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.partition;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
@@ -38,8 +39,11 @@ public class CustomPartitionExpirationFactory implements PartitionExpireStrategy
 
     @Override
     public PartitionExpireStrategy create(
-            CatalogLoader catalogLoader, Identifier identifier, RowType partitionType) {
-        return new PartitionExpireStrategy(partitionType) {
+            CatalogLoader catalogLoader,
+            Identifier identifier,
+            CoreOptions options,
+            RowType partitionType) {
+        return new PartitionExpireStrategy(partitionType, options.partitionDefaultName()) {
             @Override
             public List<PartitionEntry> selectExpiredPartitions(
                     FileStoreScan scan, LocalDateTime expirationTime) {