[fix](meta) show partitions with Limit for external HMS tables (27835)

Issue:apache#27834

Author: Nitin-Kashyap

fe/fe-core/src/main/java/org/apache/doris/analysis/ShowPartitionsStmt.java

@@ -58,7 +58,7 @@ public class ShowPartitionsStmt extends ShowStmt {
     private static final Logger LOG = LogManager.getLogger(ShowPartitionsStmt.class);
 
     private static final String FILTER_PARTITION_ID = "PartitionId";
-    private static final String FILTER_PARTITION_NAME = "PartitionName";
+    public static final String FILTER_PARTITION_NAME = "PartitionName";
     private static final String FILTER_STATE = "State";
     private static final String FILTER_BUCKETS = "Buckets";
     private static final String FILTER_REPLICATION_NUM = "ReplicationNum";
@@ -198,6 +198,11 @@ public void analyzeImpl(Analyzer analyzer) throws UserException {
                     throw new AnalysisException("Should order by column");
                 }
                 SlotRef slotRef = (SlotRef) orderByElement.getExpr();
+                if (catalog instanceof HMSExternalCatalog
+                        && !slotRef.getColumnName().equalsIgnoreCase(FILTER_PARTITION_NAME)) {
+                    throw new AnalysisException("External table only support Order By on PartitionName");
+                }
+
                 int index = PartitionsProcDir.analyzeColumn(slotRef.getColumnName());
                 OrderByPair orderByPair = new OrderByPair(index, !orderByElement.getIsAsc());
                 orderByPairs.add(orderByPair);
@@ -228,6 +233,10 @@ private void analyzeSubPredicate(Expr subExpr) throws AnalysisException {
         }
 
         String leftKey = ((SlotRef) subExpr.getChild(0)).getColumnName();
+        if (catalog instanceof HMSExternalCatalog && !leftKey.equalsIgnoreCase(FILTER_PARTITION_NAME)) {
+            throw new AnalysisException(String.format("Only %s column supported in where clause for this catalog",
+                FILTER_PARTITION_NAME));
+        }
         if (subExpr instanceof BinaryPredicate) {
             BinaryPredicate binaryPredicate = (BinaryPredicate) subExpr;
             if (leftKey.equalsIgnoreCase(FILTER_PARTITION_NAME) || leftKey.equalsIgnoreCase(FILTER_STATE)) {

fe/fe-core/src/main/java/org/apache/doris/common/proc/PartitionsProcDir.java

@@ -81,7 +81,8 @@ public PartitionsProcDir(Database db, OlapTable olapTable, boolean isTempPartiti
         this.isTempPartition = isTempPartition;
     }
 
-    public boolean filter(String columnName, Comparable element, Map<String, Expr> filterMap) throws AnalysisException {
+    public static boolean filter(String columnName, Comparable element, Map<String, Expr> filterMap)
+            throws AnalysisException {
         if (filterMap == null) {
             return true;
         }
@@ -142,7 +143,7 @@ public boolean filter(String columnName, Comparable element, Map<String, Expr> f
         return true;
     }
 
-    public boolean like(String str, String expr) {
+    public static boolean like(String str, String expr) {
         expr = expr.toLowerCase();
         expr = expr.replace(".", "\\.");
         expr = expr.replace("?", ".");

fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java

@@ -24,6 +24,7 @@
 import org.apache.doris.analysis.AdminShowReplicaStatusStmt;
 import org.apache.doris.analysis.AdminShowTabletStorageFormatStmt;
 import org.apache.doris.analysis.DescribeStmt;
+import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.HelpStmt;
 import org.apache.doris.analysis.LimitElement;
 import org.apache.doris.analysis.PartitionNames;
@@ -1691,28 +1692,54 @@ private void handleShowMaxComputeTablePartitions(ShowPartitionsStmt showStmt) {
         resultSet = new ShowResultSet(showStmt.getMetaData(), rows);
     }
 
-    private void handleShowHMSTablePartitions(ShowPartitionsStmt showStmt) {
+    private void handleShowHMSTablePartitions(ShowPartitionsStmt showStmt) throws AnalysisException {
         HMSExternalCatalog catalog = (HMSExternalCatalog) (showStmt.getCatalog());
         List<List<String>> rows = new ArrayList<>();
         String dbName = ClusterNamespace.getNameFromFullName(showStmt.getTableName().getDb());
 
         List<String> partitionNames;
         LimitElement limit = showStmt.getLimitElement();
-        if (limit != null && limit.hasLimit()) {
-            // only limit is valid on Hive
+        if (limit != null && limit.hasLimit()
+                && (!limit.hasOffset() || limit.getOffset() == 0)) {
+            // hmsClient returns unordered partition list, hence if offset > 0 cannot pass limit
             partitionNames = catalog.getClient()
-                    .listPartitionNames(dbName, showStmt.getTableName().getTbl(), limit.getLimit());
+                .listPartitionNames(dbName, showStmt.getTableName().getTbl(),  limit.getLimit());
         } else {
             partitionNames = catalog.getClient().listPartitionNames(dbName, showStmt.getTableName().getTbl());
         }
+
+        Map<String, Expr> filterMap = showStmt.getFilterMap();
+        List<OrderByPair> orderByPairs = showStmt.getOrderByPairs();
+        LimitElement limitElement = showStmt.getLimitElement();
+
+        /* Filter add rows */
         for (String partition : partitionNames) {
             List<String> list = new ArrayList<>();
+
+            if (filterMap != null && !filterMap.isEmpty()) {
+                if (!PartitionsProcDir.filter(ShowPartitionsStmt.FILTER_PARTITION_NAME, partition, filterMap)) {
+                    continue;
+                }
+            }
             list.add(partition);
             rows.add(list);
         }
 
         // sort by partition name
-        rows.sort(Comparator.comparing(x -> x.get(0)));
+        if (orderByPairs != null && orderByPairs.get(0).isDesc()) {
+            rows.sort(Comparator.comparing(x -> x.get(0), Comparator.reverseOrder()));
+        } else {
+            rows.sort(Comparator.comparing(x -> x.get(0)));
+        }
+
+        if (limitElement != null && limitElement.hasLimit()) {
+            int beginIndex = (int) limitElement.getOffset();
+            int endIndex = (int) (beginIndex + limitElement.getLimit());
+            if (endIndex > rows.size()) {
+                endIndex = rows.size();
+            }
+            rows = rows.subList(beginIndex, endIndex);
+        }
 
         resultSet = new ShowResultSet(showStmt.getMetaData(), rows);
     }