diff --git a/be/src/exec/scan/meta_scanner.cpp b/be/src/exec/scan/meta_scanner.cpp index 52892882f7bcbb..3e3adeae91b0cc 100644 --- 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 index 3d38295cd5135b..9c7d42321b9e55 100644 --- 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 index 973a3dcb09a377..cbb1b730d6863b 100644 --- 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(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/Brokers.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/Brokers.java new file mode 100644 index 00000000000000..55f7b246a99d6f --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/Brokers.java @@ -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 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 accept(ExpressionVisitor visitor, C context) { + return visitor.visitBrokers(this, context); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java index 34bcbe4d6dc372..6fc9ce2759291d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.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; @@ -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 new file mode 100644 index 00000000000000..70e8de2578ea78 --- /dev/null +++ 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 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 COLUMN_TO_INDEX; + + static { + ImmutableMap.Builder 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 params) throws AnalysisException { + if (params == null) { + throw new AnalysisException("params cannot be null"); + } + // 1. Normalize the property map: convert all keys to lowercase + Map normalizedParams = Maps.newHashMap(); + for (Map.Entry 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 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 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 index 1838085cb1f4f2..20ae5529fc0280 100644 --- 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 dataBatch = Lists.newArrayList(); + String clusterName = params.getBrokersMetadataParams().getClusterName(); + boolean needFilter = StringUtils.isNotBlank(clusterName); + List> infos = Env.getCurrentEnv().getBrokerMgr().getBrokersInfo(); + for (List 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 index 39fde6a5615a60..70d204620e444e 100644 --- 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 index a36c289aca182c..1decfcbbb7f644 100644 --- 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 mockedEnvStatic; + private MockedStatic mockedCtxStatic; + + @After + public void tearDown() { + if (mockedCtxStatic != null) { + mockedCtxStatic.close(); + mockedCtxStatic = null; + } + if (mockedEnvStatic != null) { + mockedEnvStatic.close(); + mockedEnvStatic = null; + } + } + + private void mockContext() { + mockedEnvStatic = Mockito.mockStatic(Env.class); + mockedEnvStatic.when(Env::getCurrentEnv).thenReturn(env); + + mockedCtxStatic = Mockito.mockStatic(ConnectContext.class); + mockedCtxStatic.when(ConnectContext::get).thenReturn(ctx); + + Mockito.when(env.getAccessManager()).thenReturn(accessControllerManager); + // Mock ADMIN privilege to true to ensure permission check passes + Mockito.when(accessControllerManager.checkGlobalPriv(ctx, PrivPredicate.ADMIN)).thenReturn(true); + } + + @Test + public void testConstructorWithEmptyParams() throws Exception { + mockContext(); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(new HashMap<>()); + Assert.assertEquals(TMetadataType.BROKERS, tvf.getMetadataType()); + Assert.assertNull(tvf.getMetaScanRange(Collections.emptyList()).getBrokersParams().getClusterName()); + } + + @Test + public void testConstructorWithClusterName() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + params.put("cluster_name", "test_cluster"); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(params); + + TMetaScanRange metaScanRange = tvf.getMetaScanRange(Collections.emptyList()); + Assert.assertNotNull(metaScanRange); + Assert.assertTrue(metaScanRange.isSetBrokersParams()); + Assert.assertEquals("test_cluster", metaScanRange.getBrokersParams().getClusterName()); + } + + @Test + public void testConstructorWithUpperCaseClusterNameKey() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + // Test uppercase key + params.put("CLUSTER_NAME", "test_cluster"); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(params); + + TMetaScanRange metaScanRange = tvf.getMetaScanRange(Collections.emptyList()); + Assert.assertNotNull(metaScanRange); + Assert.assertTrue(metaScanRange.isSetBrokersParams()); + Assert.assertEquals("test_cluster", metaScanRange.getBrokersParams().getClusterName()); + } + + @Test + public void testConstructorWithMixedCaseClusterNameKey() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + // Test mixed case key + params.put("Cluster_Name", "test_cluster"); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(params); + + TMetaScanRange metaScanRange = tvf.getMetaScanRange(Collections.emptyList()); + Assert.assertNotNull(metaScanRange); + Assert.assertTrue(metaScanRange.isSetBrokersParams()); + Assert.assertEquals("test_cluster", metaScanRange.getBrokersParams().getClusterName()); + } + + @Test + public void testConstructorWithUpperCaseClusterNameKeyNotExist() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + // Test uppercase key with a non-existent cluster name + params.put("CLUSTER_NAME", "non_existent_cluster"); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(params); + + TMetaScanRange metaScanRange = tvf.getMetaScanRange(Collections.emptyList()); + Assert.assertNotNull(metaScanRange); + Assert.assertTrue(metaScanRange.isSetBrokersParams()); + // Ensure the cluster name is correctly passed down and not null + Assert.assertEquals("non_existent_cluster", metaScanRange.getBrokersParams().getClusterName()); + } + + @Test + public void testConstructorWithMixedCaseClusterNameKeyNotExist() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + // Test mixed case key with a non-existent cluster name + params.put("Cluster_Name", "non_existent_cluster"); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(params); + + TMetaScanRange metaScanRange = tvf.getMetaScanRange(Collections.emptyList()); + Assert.assertNotNull(metaScanRange); + Assert.assertTrue(metaScanRange.isSetBrokersParams()); + // Ensure the cluster name is correctly passed down and not null + Assert.assertEquals("non_existent_cluster", metaScanRange.getBrokersParams().getClusterName()); + } + + @Test(expected = AnalysisException.class) + public void testConstructorWithInvalidParamKey() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + params.put("invalid_param", "value"); + new BrokersTableValuedFunction(params); + } + + @Test(expected = AnalysisException.class) + public void testConstructorWithEmptyClusterName() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + params.put("cluster_name", ""); + new BrokersTableValuedFunction(params); + } + + @Test(expected = AnalysisException.class) + public void testConstructorWithBlankClusterName() throws Exception { + mockContext(); + HashMap params = new HashMap<>(); + params.put("cluster_name", " "); + new BrokersTableValuedFunction(params); + } + + @Test(expected = AnalysisException.class) + public void testConstructorWithNullParams() throws Exception { + mockContext(); + new BrokersTableValuedFunction(null); + } + + @Test(expected = AnalysisException.class) + public void testConstructorWithoutPrivilege() throws Exception { + mockedEnvStatic = Mockito.mockStatic(Env.class); + mockedEnvStatic.when(Env::getCurrentEnv).thenReturn(env); + + mockedCtxStatic = Mockito.mockStatic(ConnectContext.class); + mockedCtxStatic.when(ConnectContext::get).thenReturn(ctx); + + Mockito.when(env.getAccessManager()).thenReturn(accessControllerManager); + // Mock both ADMIN and OPERATOR privileges to false to test insufficient permission + Mockito.when(accessControllerManager.checkGlobalPriv(ctx, PrivPredicate.ADMIN)).thenReturn(false); + Mockito.when(accessControllerManager.checkGlobalPriv(ctx, PrivPredicate.OPERATOR)).thenReturn(false); + QueryState mockState = Mockito.mock(QueryState.class); + Mockito.when(ctx.getState()).thenReturn(mockState); + new BrokersTableValuedFunction(new HashMap<>()); + } + + @Test + public void testGetTableName() throws Exception { + mockContext(); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(new HashMap<>()); + Assert.assertEquals("BrokersTableValuedFunction", tvf.getTableName()); + } + + @Test + public void testGetTableColumns() throws Exception { + mockContext(); + BrokersTableValuedFunction tvf = new BrokersTableValuedFunction(new HashMap<>()); + Assert.assertEquals(7, tvf.getTableColumns().size()); + } + + @Test + public void testGetColumnIndexFromColumnName() { + Assert.assertEquals(0, BrokersTableValuedFunction.getColumnIndexFromColumnName("Name").intValue()); + Assert.assertEquals(1, BrokersTableValuedFunction.getColumnIndexFromColumnName("Host").intValue()); + Assert.assertEquals(2, BrokersTableValuedFunction.getColumnIndexFromColumnName("Port").intValue()); + Assert.assertEquals(3, BrokersTableValuedFunction.getColumnIndexFromColumnName("Alive").intValue()); + Assert.assertEquals(4, + BrokersTableValuedFunction.getColumnIndexFromColumnName("LastStartTime").intValue()); + Assert.assertEquals(5, + BrokersTableValuedFunction.getColumnIndexFromColumnName("LastUpdateTime").intValue()); + Assert.assertEquals(6, BrokersTableValuedFunction.getColumnIndexFromColumnName("ErrMsg").intValue()); + } +} diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index d87d7df56edd92..8bb42063c5d518 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -934,6 +934,7 @@ struct TMetadataTableRequestParams { // Reserved for downstream field `current_roles` to keep thrift field ids // wire-compatible across maintained branches. Do not reuse this id. 15: optional set reserved_field_15 + 16: optional PlanNodes.TBrokersMetadataParams brokers_metadata_params } struct TSchemaTableRequestParams { diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index c2dd42bd522a16..3646a6ddb747d3 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -759,6 +759,10 @@ struct TPartitionBoundary { 6: optional bool range_end_inclusive = false } +struct TBrokersMetadataParams { + 1: optional string cluster_name +} + struct TMetaScanRange { 1: optional Types.TMetadataType metadata_type 2: optional TIcebergMetadataParams iceberg_params // deprecated @@ -779,6 +783,7 @@ struct TMetaScanRange { 15: optional string serialized_table; 16: optional list serialized_splits; 17: optional TParquetMetadataParams parquet_params; + 18: optional TBrokersMetadataParams brokers_params; } // Specification of an individual data range which is held in its entirety diff --git a/gensrc/thrift/Types.thrift b/gensrc/thrift/Types.thrift index c6b9c705307380..f51c1c7d06a7e4 100644 --- a/gensrc/thrift/Types.thrift +++ b/gensrc/thrift/Types.thrift @@ -764,6 +764,7 @@ enum TMetadataType { PAIMON = 12, PARQUET = 13, STREAMS = 14, + BROKERS = 15, } // deprecated diff --git a/regression-test/suites/external_table_p0/tvf/test_brokers_tvf.groovy b/regression-test/suites/external_table_p0/tvf/test_brokers_tvf.groovy new file mode 100644 index 00000000000000..e621f884b57f40 --- /dev/null +++ b/regression-test/suites/external_table_p0/tvf/test_brokers_tvf.groovy @@ -0,0 +1,196 @@ +// 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. + +// This suit test the `brokers` tvf +suite("test_brokers_tvf","p0") { + def brokerClusterA = "brokers_tvf_cluster_a" + def brokerClusterB = "brokers_tvf_cluster_b" + + try { + // 0. Clean up possible residual test clusters + try { sql """ALTER SYSTEM DROP ALL BROKER ${brokerClusterA};""" } catch (Exception e) { log.info("Ignored: ${e.getMessage()}") } + try { sql """ALTER SYSTEM DROP ALL BROKER ${brokerClusterB};""" } catch (Exception e) { log.info("Ignored: ${e.getMessage()}") } + + def address1 = "172.18.0.1" + def address2 = "172.18.0.2" + def address3 = "172.18.0.3" + def address4 = "172.18.0.4" + def port = 18310 + + // 1. Get column metadata and build expected column expression + def columns = new ArrayList() + def columnMeta = sql """describe function brokers();""" + def size = columnMeta.size() + for (i in 0.. row.join("") }.join("\n") + assertTrue(explainStr.contains("BrokersTableValuedFunction")) + for (i in 0.. row[0] in [brokerClusterA, brokerClusterB] }.sort { it[1] } + def filteredShowProc = showBrokerProcResult.findAll { row -> row[0] in [brokerClusterA, brokerClusterB] }.sort { it[1] } + + assertTrue(filteredShowBroker.size() == 4) + assertTrue(filteredShowProc.size() == 4) + + // Strictly compare column values (excluding last 3 volatile columns: LastStartTime, LastUpdateTime, ErrMsg) + for (i in 0..