apache · heguanhui · Jan 28, 2026
diff --git a/be/src/exec/scan/meta_scanner.cpp b/be/src/exec/scan/meta_scanner.cpp
@@ -265,6 +265,9 @@ Status MetaScanner::_fetch_metadata(const TMetaScanRange& meta_scan_range) {
     case TMetadataType::PARTITION_VALUES:
         RETURN_IF_ERROR(_build_partition_values_metadata_request(meta_scan_range, &request));
         break;
+    case TMetadataType::BROKERS:
+        RETURN_IF_ERROR(_build_brokers_metadata_request(meta_scan_range, &request));
+        break;
     default:
         _meta_eos = true;
         return Status::OK();
@@ -506,6 +509,24 @@ Status MetaScanner::_build_partition_values_metadata_request(
     request->__set_metada_table_params(metadata_table_params);
     return Status::OK();
 }
+Status MetaScanner::_build_brokers_metadata_request(const TMetaScanRange& meta_scan_range,
+                                                    TFetchSchemaTableDataRequest* request) {
+    VLOG_CRITICAL << "MetaScanner::_build_brokers_metadata_request";
+    if (!meta_scan_range.__isset.brokers_params) {
+        return Status::InternalError("Can not find TBrokersMetadataParams from meta_scan_range.");
+    }
+    // create request
+    request->__set_cluster_name(meta_scan_range.brokers_params.cluster_name);
+    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
+
+    // create TMetadataTableRequestParams
+    TMetadataTableRequestParams metadata_table_params;
+    metadata_table_params.__set_metadata_type(TMetadataType::BROKERS);
+    metadata_table_params.__set_brokers_metadata_params(meta_scan_range.brokers_params);
+
+    request->__set_metada_table_params(metadata_table_params);
+    return Status::OK();
+}
 
 Status MetaScanner::close(RuntimeState* state) {
     VLOG_CRITICAL << "MetaScanner::close";

diff --git a/be/src/exec/scan/meta_scanner.h b/be/src/exec/scan/meta_scanner.h
@@ -85,6 +85,8 @@ class MetaScanner : public Scanner {
                                            TFetchSchemaTableDataRequest* request);
     Status _build_partition_values_metadata_request(const TMetaScanRange& meta_scan_range,
                                                     TFetchSchemaTableDataRequest* request);
+    Status _build_brokers_metadata_request(const TMetaScanRange& meta_scan_range,
+                                           TFetchSchemaTableDataRequest* request);
     bool _meta_eos;
     TupleId _tuple_id;
     TUserIdentity _user_identity;

diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableValuedFunctions.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableValuedFunctions.java
@@ -19,6 +19,7 @@
 
 import org.apache.doris.nereids.trees.expressions.functions.table.Backends;
 import org.apache.doris.nereids.trees.expressions.functions.table.Binlog;
+import org.apache.doris.nereids.trees.expressions.functions.table.Brokers;
 import org.apache.doris.nereids.trees.expressions.functions.table.Catalogs;
 import org.apache.doris.nereids.trees.expressions.functions.table.CdcStream;
 import org.apache.doris.nereids.trees.expressions.functions.table.File;
@@ -42,6 +43,7 @@
 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;
+import org.apache.doris.tablefunction.BrokersTableValuedFunction;
 
 import com.google.common.collect.ImmutableList;
 
@@ -77,7 +79,8 @@ public class BuiltinTableValuedFunctions implements FunctionHelper {
             tableValued(ParquetFileMetadata.class, "parquet_file_metadata"),
             tableValued(ParquetKvMetadata.class, "parquet_kv_metadata"),
             tableValued(ParquetBloomProbe.class, "parquet_bloom_probe"),
-            tableValued(CdcStream.class, "cdc_stream")
+            tableValued(CdcStream.class, "cdc_stream"),
+            tableValued(Brokers.class, BrokersTableValuedFunction.NAME)
     );
 
     public static final BuiltinTableValuedFunctions INSTANCE = new BuiltinTableValuedFunctions();

@@ -0,0 +1,57 @@
+// 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.BrokersTableValuedFunction;
+import org.apache.doris.tablefunction.TableValuedFunctionIf;
+
+import java.util.Map;
+
+/** brokers */
+public class Brokers extends TableValuedFunction {
+
+    public Brokers(Properties properties) {
+        super("brokers", 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 BrokersTableValuedFunction(arguments);
+        } catch (Throwable t) {
+            throw new AnalysisException("Can not build BrokersTableValuedFunction by "
+                + this + ": " + t.getMessage(), t);
+        }
+    }
+
+    @Override
+    public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+        return visitor.visitBrokers(this, context);
+    }
+}
@@ -19,6 +19,7 @@
 
 import org.apache.doris.nereids.trees.expressions.functions.table.Backends;
 import org.apache.doris.nereids.trees.expressions.functions.table.Binlog;
+import org.apache.doris.nereids.trees.expressions.functions.table.Brokers;
 import org.apache.doris.nereids.trees.expressions.functions.table.Catalogs;
 import org.apache.doris.nereids.trees.expressions.functions.table.CdcStream;
 import org.apache.doris.nereids.trees.expressions.functions.table.File;
@@ -132,4 +133,8 @@ default R visitS3(S3 s3, C context) {
     default R visitQuery(Query query, C context) {
         return visitTableValuedFunction(query, context);
     }
+
+    default R visitBrokers(Brokers brokers, C context) {
+        return visitTableValuedFunction(brokers, context);
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/BrokersTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/BrokersTableValuedFunction.java
@@ -0,0 +1,136 @@
+// 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.tablefunction;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TBrokersMetadataParams;
+import org.apache.doris.thrift.TMetaScanRange;
+import org.apache.doris.thrift.TMetadataType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The Implement of table valued function
+ * brokers().
+ */
+public class BrokersTableValuedFunction extends MetadataTableValuedFunction {
+
+    public static final String NAME = "brokers";
+    private static final String CLUSTER_NAME = "cluster_name";
+    private static final ImmutableList<Column> SCHEMA = ImmutableList.of(
+            new Column("Name", ScalarType.createStringType()),
+            new Column("Host", ScalarType.createStringType()),
+            new Column("Port", ScalarType.createStringType()),
+            new Column("Alive", ScalarType.createStringType()),
+            new Column("LastStartTime", ScalarType.createStringType()),
+            new Column("LastUpdateTime", ScalarType.createStringType()),
+            new Column("ErrMsg", ScalarType.createStringType())
+    );
+    private static final ImmutableMap<String, Integer> COLUMN_TO_INDEX;
+
+    static {
+        ImmutableMap.Builder<String, Integer> builder = new ImmutableMap.Builder();
+        for (int i = 0; i < SCHEMA.size(); i++) {
+            builder.put(SCHEMA.get(i).getName().toLowerCase(), i);
+        }
+        COLUMN_TO_INDEX = builder.build();
+    }
+
+    private final String clusterName;
+
+    /**
+     * constructor for BrokersTableValuedFunction
+     */
+    public BrokersTableValuedFunction(Map<String, String> params) throws AnalysisException {
+        if (params == null) {
+            throw new AnalysisException("params cannot be null");
+        }
+        // 1. Normalize the property map: convert all keys to lowercase
+        Map<String, String> normalizedParams = Maps.newHashMap();
+        for (Map.Entry<String, String> entry : params.entrySet()) {
+            if (entry.getKey() == null) {
+                throw new AnalysisException("Property key cannot be null");
+            }
+            normalizedParams.put(entry.getKey().toLowerCase(), entry.getValue());
+        }
+        // 2. Lookup value using the normalized map
+        String originalClusterName = normalizedParams.get(CLUSTER_NAME);
+        if (originalClusterName != null) {
+            if (StringUtils.isBlank(originalClusterName)) {
+                throw new AnalysisException("Invalid brokers param value: " + originalClusterName);
+            }
+            clusterName = originalClusterName.trim();
+        } else {
+            clusterName = null;
+        }
+        // 3. Validate keys against the normalized map
+        for (String key : normalizedParams.keySet()) {
+            if (!CLUSTER_NAME.equals(key)) {
+                throw new AnalysisException("'" + key + "' is invalid property,"
+                    + " only support property [" + CLUSTER_NAME + "]");
+            }
+        }
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)
+                && !Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(),
+                PrivPredicate.OPERATOR)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN/OPERATOR");
+        }
+    }
+
+    public static Integer getColumnIndexFromColumnName(String columnName) {
+        return COLUMN_TO_INDEX.get(columnName.toLowerCase());
+    }
+
+    @Override
+    public TMetadataType getMetadataType() {
+        return TMetadataType.BROKERS;
+    }
+
+    @Override
+    public TMetaScanRange getMetaScanRange(List<String> requiredFields) {
+        TMetaScanRange metaScanRange = new TMetaScanRange();
+        metaScanRange.setMetadataType(TMetadataType.BROKERS);
+        TBrokersMetadataParams brokersMetadataParams = new TBrokersMetadataParams();
+        brokersMetadataParams.setClusterName(clusterName);
+        metaScanRange.setBrokersParams(brokersMetadataParams);
+        return metaScanRange;
+    }
+
+    @Override
+    public String getTableName() {
+        return "BrokersTableValuedFunction";
+    }
+
+    @Override
+    public List<Column> getTableColumns() throws AnalysisException {
+        return SCHEMA;
+    }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
@@ -122,6 +122,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.gson.Gson;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.logging.log4j.LogManager;
@@ -327,6 +328,9 @@ public static TFetchSchemaTableDataResult getMetadataTable(TFetchSchemaTableData
             case PARTITION_VALUES:
                 result = partitionValuesMetadataResult(params);
                 break;
+            case BROKERS:
+                result = brokersMetadataResult(params);
+                break;
             default:
                 return errorResult("Metadata table params is not set.");
         }
@@ -2194,4 +2198,29 @@ private static TFetchSchemaTableDataResult streamConsumptionMetadataResult(TSche
         result.setStatus(new TStatus(TStatusCode.OK));
         return result;
     }
+
+    private static TFetchSchemaTableDataResult brokersMetadataResult(TMetadataTableRequestParams params) {
+        if (!params.isSetBrokersMetadataParams()) {
+            return errorResult("brokers metadata param is not set.");
+        }
+        TFetchSchemaTableDataResult result = new TFetchSchemaTableDataResult();
+        List<TRow> dataBatch = Lists.newArrayList();
+        String clusterName = params.getBrokersMetadataParams().getClusterName();
+        boolean needFilter = StringUtils.isNotBlank(clusterName);
+        List<List<String>> infos = Env.getCurrentEnv().getBrokerMgr().getBrokersInfo();
+        for (List<String> info : infos) {
+            if (needFilter && !clusterName.equalsIgnoreCase(info.get(0))) {
+                continue;
+            }
+            TRow trow = new TRow();
+            for (String item : info) {
+                trow.addToColumnValue(new TCell().setStringVal(item));
+            }
+            dataBatch.add(trow);
+        }
+        result.setDataBatch(dataBatch);
+        result.setStatus(new TStatus(TStatusCode.OK));
+        return result;
+    }
+
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataTableValuedFunction.java
@@ -54,6 +54,8 @@ public static Integer getColumnIndexFromColumnName(TMetadataType type, String co
                 return JobsTableValuedFunction.getColumnIndexFromColumnName(columnName, params);
             case TASKS:
                 return TasksTableValuedFunction.getColumnIndexFromColumnName(columnName, params);
+            case BROKERS:
+                return BrokersTableValuedFunction.getColumnIndexFromColumnName(columnName);
             default:
                 throw new AnalysisException("Unknown Metadata TableValuedFunction type");
         }

diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java
@@ -104,6 +104,8 @@ public static TableValuedFunctionIf getTableFunction(String funcName, Map<String
                 return new HttpTableValuedFunction(params);
             case TableBinlogFunction.NAME:
                 return new TableBinlogFunction(params);
+            case BrokersTableValuedFunction.NAME:
+                return new BrokersTableValuedFunction(params);
             default:
                 throw new AnalysisException("Could not find table function " + funcName);
         }