[enhance](mtmv)support partition tvf

be/src/vec/exec/scan/vmeta_scanner.cpp

@@ -243,6 +243,9 @@ Status VMetaScanner::_fetch_metadata(const TMetaScanRange& meta_scan_range) {
     case TMetadataType::MATERIALIZED_VIEWS:
         RETURN_IF_ERROR(_build_materialized_views_metadata_request(meta_scan_range, &request));
         break;
+    case TMetadataType::PARTITIONS:
+        RETURN_IF_ERROR(_build_partitions_metadata_request(meta_scan_range, &request));
+        break;
     case TMetadataType::JOBS:
         RETURN_IF_ERROR(_build_jobs_metadata_request(meta_scan_range, &request));
         break;
@@ -411,6 +414,26 @@ Status VMetaScanner::_build_materialized_views_metadata_request(
     return Status::OK();
 }
 
+Status VMetaScanner::_build_partitions_metadata_request(const TMetaScanRange& meta_scan_range,
+                                                        TFetchSchemaTableDataRequest* request) {
+    VLOG_CRITICAL << "VMetaScanner::_build_partitions_metadata_request";
+    if (!meta_scan_range.__isset.partitions_params) {
+        return Status::InternalError(
+                "Can not find TPartitionsMetadataParams from meta_scan_range.");
+    }
+
+    // create request
+    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
+
+    // create TMetadataTableRequestParams
+    TMetadataTableRequestParams metadata_table_params;
+    metadata_table_params.__set_metadata_type(TMetadataType::PARTITIONS);
+    metadata_table_params.__set_partitions_metadata_params(meta_scan_range.partitions_params);
+
+    request->__set_metada_table_params(metadata_table_params);
+    return Status::OK();
+}
+
 Status VMetaScanner::_build_jobs_metadata_request(const TMetaScanRange& meta_scan_range,
                                                   TFetchSchemaTableDataRequest* request) {
     VLOG_CRITICAL << "VMetaScanner::_build_jobs_metadata_request";

be/src/vec/exec/scan/vmeta_scanner.h

@@ -85,6 +85,8 @@ class VMetaScanner : public VScanner {
                                             TFetchSchemaTableDataRequest* request);
     Status _build_materialized_views_metadata_request(const TMetaScanRange& meta_scan_range,
                                                       TFetchSchemaTableDataRequest* request);
+    Status _build_partitions_metadata_request(const TMetaScanRange& meta_scan_range,
+                                              TFetchSchemaTableDataRequest* request);
     Status _build_jobs_metadata_request(const TMetaScanRange& meta_scan_range,
                                         TFetchSchemaTableDataRequest* request);
     Status _build_tasks_metadata_request(const TMetaScanRange& meta_scan_range,

fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableValuedFunctions.java

@@ -29,6 +29,7 @@
 import org.apache.doris.nereids.trees.expressions.functions.table.Local;
 import org.apache.doris.nereids.trees.expressions.functions.table.MvInfos;
 import org.apache.doris.nereids.trees.expressions.functions.table.Numbers;
+import org.apache.doris.nereids.trees.expressions.functions.table.Partitions;
 import org.apache.doris.nereids.trees.expressions.functions.table.Query;
 import org.apache.doris.nereids.trees.expressions.functions.table.S3;
 import org.apache.doris.nereids.trees.expressions.functions.table.Tasks;
@@ -55,6 +56,7 @@ public class BuiltinTableValuedFunctions implements FunctionHelper {
             tableValued(Numbers.class, "numbers"),
             tableValued(S3.class, "s3"),
             tableValued(MvInfos.class, "mv_infos"),
+            tableValued(Partitions.class, "partitions"),
             tableValued(Jobs.class, "jobs"),
             tableValued(Tasks.class, "tasks"),
             tableValued(Query.class, "query")

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

@@ -46,6 +46,8 @@
 import org.apache.doris.common.util.OrderByPair;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.mtmv.MTMVPartitionUtil;
+import org.apache.doris.thrift.TCell;
+import org.apache.doris.thrift.TRow;
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
@@ -222,12 +224,22 @@ public BaseProcResult getBasicProcResult(List<List<Comparable>> partitionInfos)
     }
 
     private List<List<Comparable>> getPartitionInfos() throws AnalysisException {
+        List<Pair<List<Comparable>, TRow>> partitionInfosInrernal = getPartitionInfosInrernal();
+        return partitionInfosInrernal.stream().map(pair -> pair.first).collect(Collectors.toList());
+    }
+
+    public List<TRow> getPartitionInfosForTvf() throws AnalysisException {
+        List<Pair<List<Comparable>, TRow>> partitionInfosInrernal = getPartitionInfosInrernal();
+        return partitionInfosInrernal.stream().map(pair -> pair.second).collect(Collectors.toList());
+    }
+
+    private List<Pair<List<Comparable>, TRow>> getPartitionInfosInrernal() throws AnalysisException {
         Preconditions.checkNotNull(db);
         Preconditions.checkNotNull(olapTable);
         Preconditions.checkState(olapTable.isManagedTable());
 
         // get info
-        List<List<Comparable>> partitionInfos = new ArrayList<List<Comparable>>();
+        List<Pair<List<Comparable>, TRow>> partitionInfos = new ArrayList<Pair<List<Comparable>, TRow>>();
         olapTable.readLock();
         try {
             List<Long> partitionIds;
@@ -259,25 +271,37 @@ private List<List<Comparable>> getPartitionInfos() throws AnalysisException {
                 Partition partition = olapTable.getPartition(partitionId);
 
                 List<Comparable> partitionInfo = new ArrayList<Comparable>();
+                TRow trow = new TRow();
                 String partitionName = partition.getName();
                 partitionInfo.add(partitionId);
+                trow.addToColumnValue(new TCell().setLongVal(partitionId));
                 partitionInfo.add(partitionName);
+                trow.addToColumnValue(new TCell().setStringVal(partitionName));
                 partitionInfo.add(partition.getVisibleVersion());
-                partitionInfo.add(TimeUtils.longToTimeString(partition.getVisibleVersionTime()));
+                trow.addToColumnValue(new TCell().setLongVal(partition.getVisibleVersion()));
+                String visibleTime = TimeUtils.longToTimeString(partition.getVisibleVersionTime());
+                partitionInfo.add(visibleTime);
+                trow.addToColumnValue(new TCell().setStringVal(visibleTime));
                 partitionInfo.add(partition.getState());
-
+                trow.addToColumnValue(new TCell().setStringVal(partition.getState().toString()));
                 if (tblPartitionInfo.getType() == PartitionType.RANGE
                         || tblPartitionInfo.getType() == PartitionType.LIST) {
                     List<Column> partitionColumns = tblPartitionInfo.getPartitionColumns();
                     List<String> colNames = new ArrayList<>();
                     for (Column column : partitionColumns) {
                         colNames.add(column.getName());
                     }
-                    partitionInfo.add(joiner.join(colNames));
-                    partitionInfo.add(tblPartitionInfo.getItem(partitionId).getItems().toString());
+                    String colNamesStr = joiner.join(colNames);
+                    partitionInfo.add(colNamesStr);
+                    trow.addToColumnValue(new TCell().setStringVal(colNamesStr));
+                    String itemStr = tblPartitionInfo.getItem(partitionId).getItems().toString();
+                    partitionInfo.add(itemStr);
+                    trow.addToColumnValue(new TCell().setStringVal(itemStr));
                 } else {
                     partitionInfo.add("");
+                    trow.addToColumnValue(new TCell().setStringVal(""));
                     partitionInfo.add("");
+                    trow.addToColumnValue(new TCell().setStringVal(""));
                 }
 
                 // distribution
@@ -293,47 +317,70 @@ private List<List<Comparable>> getPartitionInfos() throws AnalysisException {
                         sb.append(distributionColumns.get(i).getName());
                     }
                     partitionInfo.add(sb.toString());
+                    trow.addToColumnValue(new TCell().setStringVal(sb.toString()));
                 } else {
                     partitionInfo.add("RANDOM");
+                    trow.addToColumnValue(new TCell().setStringVal("RANDOM"));
                 }
 
                 partitionInfo.add(distributionInfo.getBucketNum());
+                trow.addToColumnValue(new TCell().setIntVal(distributionInfo.getBucketNum()));
                 // replica num
-                partitionInfo.add(tblPartitionInfo.getReplicaAllocation(partitionId).getTotalReplicaNum());
+                short totalReplicaNum = tblPartitionInfo.getReplicaAllocation(partitionId).getTotalReplicaNum();
+                partitionInfo.add(totalReplicaNum);
+                trow.addToColumnValue(new TCell().setIntVal(totalReplicaNum));
 
                 DataProperty dataProperty = tblPartitionInfo.getDataProperty(partitionId);
                 partitionInfo.add(dataProperty.getStorageMedium().name());
-                partitionInfo.add(TimeUtils.longToTimeString(dataProperty.getCooldownTimeMs()));
+                trow.addToColumnValue(new TCell().setStringVal(dataProperty.getStorageMedium().name()));
+                String cooldownTimeStr = TimeUtils.longToTimeString(dataProperty.getCooldownTimeMs());
+                partitionInfo.add(cooldownTimeStr);
+                trow.addToColumnValue(new TCell().setStringVal(cooldownTimeStr));
                 partitionInfo.add(dataProperty.getStoragePolicy());
-
-                partitionInfo.add(TimeUtils.longToTimeString(partition.getLastCheckTime()));
-
+                trow.addToColumnValue(new TCell().setStringVal(dataProperty.getStoragePolicy()));
+                String lastCheckTime = TimeUtils.longToTimeString(partition.getLastCheckTime());
+                partitionInfo.add(lastCheckTime);
+                trow.addToColumnValue(new TCell().setStringVal(lastCheckTime));
                 long dataSize = partition.getDataSize(false);
                 Pair<Double, String> sizePair = DebugUtil.getByteUint(dataSize);
                 String readableSize = DebugUtil.DECIMAL_FORMAT_SCALE_3.format(sizePair.first) + " "
                         + sizePair.second;
                 partitionInfo.add(readableSize);
-                partitionInfo.add(tblPartitionInfo.getIsInMemory(partitionId));
+                trow.addToColumnValue(new TCell().setStringVal(readableSize));
+                boolean isInMemory = tblPartitionInfo.getIsInMemory(partitionId);
+                partitionInfo.add(isInMemory);
+                trow.addToColumnValue(new TCell().setBoolVal(isInMemory));
                 // replica allocation
-                partitionInfo.add(tblPartitionInfo.getReplicaAllocation(partitionId).toCreateStmt());
+                String replica = tblPartitionInfo.getReplicaAllocation(partitionId).toCreateStmt();
+                partitionInfo.add(replica);
+                trow.addToColumnValue(new TCell().setStringVal(replica));
 
-                partitionInfo.add(tblPartitionInfo.getIsMutable(partitionId));
+                boolean isMutable = tblPartitionInfo.getIsMutable(partitionId);
+                partitionInfo.add(isMutable);
+                trow.addToColumnValue(new TCell().setBoolVal(isMutable));
                 if (olapTable instanceof MTMV) {
                     if (StringUtils.isEmpty(mtmvPartitionSyncErrorMsg)) {
                         List<String> partitionUnSyncTables = partitionsUnSyncTables.getOrDefault(partitionId,
                                 Lists.newArrayList());
-                        partitionInfo.add(CollectionUtils.isEmpty(partitionUnSyncTables));
+                        boolean isSync = CollectionUtils.isEmpty(partitionUnSyncTables);
+                        partitionInfo.add(isSync);
+                        trow.addToColumnValue(new TCell().setBoolVal(isSync));
                         partitionInfo.add(partitionUnSyncTables.toString());
+                        trow.addToColumnValue(new TCell().setStringVal(partitionUnSyncTables.toString()));
                     } else {
                         partitionInfo.add(false);
+                        trow.addToColumnValue(new TCell().setBoolVal(false));
                         partitionInfo.add(mtmvPartitionSyncErrorMsg);
+                        trow.addToColumnValue(new TCell().setStringVal(mtmvPartitionSyncErrorMsg));
                     }
                 } else {
                     partitionInfo.add(true);
+                    trow.addToColumnValue(new TCell().setBoolVal(true));
                     partitionInfo.add(FeConstants.null_string);
+                    trow.addToColumnValue(new TCell().setStringVal(FeConstants.null_string));
                 }
 
-                partitionInfos.add(partitionInfo);
+                partitionInfos.add(Pair.of(partitionInfo, trow));
             }
         } finally {
             olapTable.readUnlock();

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/Partitions.java

@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.expressions.functions.table;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Properties;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.coercion.AnyDataType;
+import org.apache.doris.tablefunction.PartitionsTableValuedFunction;
+import org.apache.doris.tablefunction.TableValuedFunctionIf;
+
+import java.util.Map;
+
+/**
+ * partitions
+ */
+public class Partitions extends TableValuedFunction {
+    public Partitions(Properties properties) {
+        super("partitions", properties);
+    }
+
+    @Override
+    public FunctionSignature customSignature() {
+        return FunctionSignature.of(AnyDataType.INSTANCE_WITHOUT_INDEX, getArgumentsTypes());
+    }
+
+    @Override
+    protected TableValuedFunctionIf toCatalogFunction() {
+        try {
+            Map<String, String> arguments = getTVFProperties().getMap();
+            return new PartitionsTableValuedFunction(arguments);
+        } catch (Throwable t) {
+            throw new AnalysisException("Can not build PartitionsTableValuedFunction by "
+                    + this + ": " + t.getMessage(), t);
+        }
+    }
+
+    @Override
+    public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+        return visitor.visitPartitions(this, context);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java

@@ -29,6 +29,7 @@
 import org.apache.doris.nereids.trees.expressions.functions.table.Local;
 import org.apache.doris.nereids.trees.expressions.functions.table.MvInfos;
 import org.apache.doris.nereids.trees.expressions.functions.table.Numbers;
+import org.apache.doris.nereids.trees.expressions.functions.table.Partitions;
 import org.apache.doris.nereids.trees.expressions.functions.table.Query;
 import org.apache.doris.nereids.trees.expressions.functions.table.S3;
 import org.apache.doris.nereids.trees.expressions.functions.table.TableValuedFunction;
@@ -54,6 +55,10 @@ default R visitMvInfos(MvInfos mvInfos, C context) {
         return visitTableValuedFunction(mvInfos, context);
     }
 
+    default R visitPartitions(Partitions partitions, C context) {
+        return visitTableValuedFunction(partitions, context);
+    }
+
     default R visitJobs(Jobs jobs, C context) {
         return visitTableValuedFunction(jobs, context);
     }