ordered partition keys

Author: xiaochen-zhou

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/util/CommonRpcMessageUtils.java

@@ -49,7 +49,9 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
@@ -230,4 +232,38 @@ public static ResolvedPartitionSpec toResolvedPartitionSpec(PbPartitionSpec pbPa
         }
         return new ResolvedPartitionSpec(partitionKeys, partitionValues);
     }
+
+    /**
+     * Convert PbPartitionSpec to ResolvedPartitionSpec with partition key ordering. This ensures
+     * the partition values are ordered according to the table's partition key sequence.
+     */
+    public static ResolvedPartitionSpec toResolvedPartitionSpec(
+            PbPartitionSpec pbPartitionSpec, List<String> orderedPartitionKeys) {
+
+        List<PbKeyValue> partitionKeyValuesList = pbPartitionSpec.getPartitionKeyValuesList();
+        if (partitionKeyValuesList.size() != orderedPartitionKeys.size()) {
+            return toResolvedPartitionSpec(pbPartitionSpec);
+        }
+
+        Map<String, String> pbkeyValueMap = new HashMap<>();
+        for (PbKeyValue pbKeyValue : partitionKeyValuesList) {
+            if (!orderedPartitionKeys.contains(pbKeyValue.getKey())) {
+                return toResolvedPartitionSpec(pbPartitionSpec);
+            }
+            pbkeyValueMap.put(pbKeyValue.getKey(), pbKeyValue.getValue());
+        }
+
+        List<String> partitionKeys = new ArrayList<>();
+        List<String> orderedPartitionValues = new ArrayList<>();
+
+        for (String orderedKey : orderedPartitionKeys) {
+            String value = pbkeyValueMap.get(orderedKey);
+            if (value != null) {
+                partitionKeys.add(orderedKey);
+                orderedPartitionValues.add(value);
+            }
+        }
+
+        return new ResolvedPartitionSpec(partitionKeys, orderedPartitionValues);
+    }
 }

fluss-server/src/main/java/com/alibaba/fluss/server/RpcServiceBase.java

@@ -404,17 +404,18 @@ public CompletableFuture<GetFileSystemSecurityTokenResponse> getFileSystemSecuri
     public CompletableFuture<ListPartitionInfosResponse> listPartitionInfos(
             ListPartitionInfosRequest request) {
         TablePath tablePath = toTablePath(request.getTablePath());
+        TableInfo tableInfo = metadataManager.getTable(tablePath);
+        List<String> partitionKeys = tableInfo.getPartitionKeys();
+
         Map<String, Long> partitionNameAndIds;
         if (request.hasPartialPartitionSpec()) {
             ResolvedPartitionSpec partitionSpecFromRequest =
-                    toResolvedPartitionSpec(request.getPartialPartitionSpec());
+                    toResolvedPartitionSpec(request.getPartialPartitionSpec(), partitionKeys);
             partitionNameAndIds =
                     metadataManager.listPartitions(tablePath, partitionSpecFromRequest);
         } else {
             partitionNameAndIds = metadataManager.listPartitions(tablePath);
         }
-        TableInfo tableInfo = metadataManager.getTable(tablePath);
-        List<String> partitionKeys = tableInfo.getPartitionKeys();
         return CompletableFuture.completedFuture(
                 toListPartitionInfosResponse(partitionKeys, partitionNameAndIds));
     }