Repository: NetEase/arctic (java)
Issue: [Bug]: when partition value contains spaces, optimize can't process correctly
### What happened?

1、When the partition value contains spaces, arctic writer will replace the spaces with + when creating a directory.



2、optimize will process the same file repeatly.



### Affects Versions

0.3.1/0.4.2/master

### What engines are you seeing the problem on?

Optimizer

### How to reproduce

Insert a record with a partition value containing spaces into a table.




### Relevant log output

_No response_

### Anything else

_No response_

### Code of Conduct

- [X] I agree to follow this project's Code of Conduct
Pull request description: [ARCTIC-1653] Support special characters in partition values
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about its use case.
  2. If you fix a bug, you can clarify why it is a bug.
-->

fix #1653 

## Brief change log

  - using `URLDecoder` when converting partition path to partition data

## How was this patch tested?
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (yes / **no**)
  - If yes, how is the feature documented? (not applicable / docs / JavaDocs / **not documented**)


Patch:
diff --git a/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java b/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
index 826f03e79..134eca771 100644
--- a/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
+++ b/core/src/main/java/com/netease/arctic/utils/ArcticDataFiles.java
@@ -8,7 +8,9 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 
+import java.io.UnsupportedEncodingException;
 import java.math.BigDecimal;
+import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -114,11 +116,22 @@ public class ArcticDataFiles {
               field.name().equals(parts[0]),
           "Invalid partition: %s", partitions[i]);
 
+      String value;
       if ("null".equals(parts[1])) {
-        parts[1] = null;
+        value = null;
+      } else {
+        try {
+          /*
+          Decode URL with UTF-8, because in org.apache.iceberg.PartitionSpec#escape,
+          the value is encoded to URL with UTF-8
+           */
+          value = URLDecoder.decode(parts[1], "UTF-8");
+        } catch (UnsupportedEncodingException e) {
+          throw new IllegalStateException(String.format("failed to decode %s of %s", parts[1], partitionPath), e);
+        }
       }
 
-      data.set(i, ArcticDataFiles.fromPartitionString(field, spec.partitionType().fieldType(parts[0]), parts[1]));
+      data.set(i, ArcticDataFiles.fromPartitionString(field, spec.partitionType().fieldType(parts[0]), value));
     }
 
     return data;


Test patch:
diff --git a/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java b/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
index 4f83a47f0..1af503c8f 100644
--- a/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
+++ b/core/src/test/java/com/netease/arctic/utils/TestArcticDataFiles.java
@@ -140,4 +140,28 @@ public class TestArcticDataFiles {
     StructLike partitionData = ArcticDataFiles.data(spec, partitionPath);
     Assert.assertNull(partitionData.get(0, Types.StringType.get().typeId().javaClass()));
   }
+
+  @Test
+  public void testSpecialCharactersPartition() {
+    Schema schema = new Schema(
+        Types.NestedField.required(1, "day", Types.StringType.get()),
+        Types.NestedField.required(2, "name", Types.StringType.get())
+    );
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("day").identity("name").build();
+    PartitionKey partitionKey = new PartitionKey(spec, schema);
+    GenericRecord record = GenericRecord.create(schema);
+    InternalRecordWrapper internalRecordWrapper = new InternalRecordWrapper(schema.asStruct());
+    partitionKey.partition(internalRecordWrapper.wrap(record.copy("day", "2023-01-01")
+        .copy("name", "AAA BBB/CCC=_*-\\%")));
+
+    String partitionToPath = spec.partitionToPath(partitionKey);
+    Assert.assertEquals("day=2023-01-01/name=AAA+BBB%2FCCC%3D_*-%5C%25", partitionToPath);
+    GenericRecord partitionData = ArcticDataFiles.data(spec, partitionToPath);
+    StructLikeWrapper p1 = StructLikeWrapper.forType(spec.partitionType());
+    p1.set(partitionKey);
+    StructLikeWrapper p2 = StructLikeWrapper.forType(spec.partitionType());
+    p2.set(partitionData);
+
+    Assert.assertEquals(p1, p2);
+  }
 }