[feature](plugin) Add information_schema.plugins for kernel plugin metadata

be/src/information_schema/schema_plugins_scanner.cpp

@@ -0,0 +1,150 @@
+// 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.
+
+#include "information_schema/schema_plugins_scanner.h"
+
+#include <utility>
+
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/string_ref.h"
+#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_state.h"
+#include "util/client_cache.h"
+#include "util/thrift_rpc_helper.h"
+
+namespace doris {
+
+std::vector<SchemaScanner::ColumnDesc> SchemaPluginsScanner::_s_tbls_columns = {
+        {"PLUGIN_NAME", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"PLUGIN_TYPE", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"PLUGIN_VERSION", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"SOURCE", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"DESCRIPTION", TYPE_STRING, sizeof(StringRef), true},
+};
+
+SchemaPluginsScanner::SchemaPluginsScanner()
+        : SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_PLUGINS) {}
+
+Status SchemaPluginsScanner::start(RuntimeState* state) {
+    if (!_is_init) {
+        return Status::InternalError("used before initialized.");
+    }
+    _block_rows_limit = state->batch_size();
+    _rpc_timeout_ms = state->execution_timeout() * 1000;
+    // Plugins are per-FE local state: ask the FE this session is connected to.
+    _fe_addr = state->get_query_ctx()->current_connect_fe;
+    return Status::OK();
+}
+
+Status SchemaPluginsScanner::_get_plugins_block_from_fe() {
+    TSchemaTableRequestParams schema_table_request_params;
+    for (int i = 0; i < _s_tbls_columns.size(); i++) {
+        schema_table_request_params.__isset.columns_name = true;
+        schema_table_request_params.columns_name.emplace_back(_s_tbls_columns[i].name);
+    }
+    schema_table_request_params.__set_current_user_ident(*_param->common_param->current_user_ident);
+
+    TFetchSchemaTableDataRequest request;
+    request.__set_schema_table_name(TSchemaTableName::PLUGINS);
+    request.__set_schema_table_params(schema_table_request_params);
+
+    TFetchSchemaTableDataResult result;
+    RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
+            _fe_addr.hostname, _fe_addr.port,
+            [&request, &result](FrontendServiceConnection& client) {
+                client->fetchSchemaTableData(result, request);
+            },
+            _rpc_timeout_ms));
+
+    Status status(Status::create(result.status));
+    if (!status.ok()) {
+        LOG(WARNING) << "fetch plugins from FE(" << _fe_addr.hostname
+                     << ") failed, errmsg=" << status;
+        return status;
+    }
+
+    _plugins_block = Block::create_unique();
+    for (int i = 0; i < _s_tbls_columns.size(); ++i) {
+        auto data_type =
+                DataTypeFactory::instance().create_data_type(_s_tbls_columns[i].type, true);
+        _plugins_block->insert(ColumnWithTypeAndName(data_type->create_column(), data_type,
+                                                     _s_tbls_columns[i].name));
+    }
+    _plugins_block->reserve(_block_rows_limit);
+
+    std::vector<TRow> result_data = std::move(result.data_batch);
+    if (!result_data.empty()) {
+        auto col_size = result_data[0].column_value.size();
+        if (col_size != _s_tbls_columns.size()) {
+            return Status::InternalError<false>("plugins schema is not match for FE and BE");
+        }
+    }
+
+    for (int i = 0; i < result_data.size(); i++) {
+        const TRow& row = result_data[i];
+        for (int j = 0; j < _s_tbls_columns.size(); j++) {
+            const TCell& cell = row.column_value[j];
+            // An unset string cell means SQL NULL (e.g. unknown PLUGIN_VERSION);
+            // insert_block_column would materialize it as an empty string.
+            if (!cell.__isset.stringVal) {
+                MutableColumnPtr mutable_col_ptr =
+                        IColumn::mutate(std::move(_plugins_block->get_by_position(j).column));
+                assert_cast<ColumnNullable*>(mutable_col_ptr.get())->insert_data(nullptr, 0);
+                _plugins_block->replace_by_position(j, std::move(mutable_col_ptr));
+                continue;
+            }
+            RETURN_IF_ERROR(
+                    insert_block_column(cell, j, _plugins_block.get(), _s_tbls_columns[j].type));
+        }
+    }
+    return Status::OK();
+}
+
+Status SchemaPluginsScanner::get_next_block_internal(Block* block, bool* eos) {
+    if (!_is_init) {
+        return Status::InternalError("Used before initialized.");
+    }
+
+    if (nullptr == block || nullptr == eos) {
+        return Status::InternalError("input pointer is nullptr.");
+    }
+
+    if (_plugins_block == nullptr) {
+        RETURN_IF_ERROR(_get_plugins_block_from_fe());
+        _total_rows = static_cast<int>(_plugins_block->rows());
+    }
+
+    if (_row_idx == _total_rows) {
+        *eos = true;
+        return Status::OK();
+    }
+
+    int current_batch_rows = std::min(_block_rows_limit, _total_rows - _row_idx);
+    ScopedMutableBlock scoped_mblock(block);
+    auto& mblock = scoped_mblock.mutable_block();
+    RETURN_IF_ERROR(mblock.add_rows(_plugins_block.get(), _row_idx, current_batch_rows));
+    _row_idx += current_batch_rows;
+
+    *eos = _row_idx == _total_rows;
+    return Status::OK();
+}
+
+} // namespace doris

be/src/information_schema/schema_plugins_scanner.h

@@ -0,0 +1,56 @@
+// 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.
+
+#pragma once
+
+#include <vector>
+
+#include "common/status.h"
+#include "information_schema/schema_scanner.h"
+
+namespace doris {
+class RuntimeState;
+class Block;
+
+// information_schema.plugins: kernel plugin inventory of the currently
+// connected FE (plugins are per-FE local state, so the RPC targets the
+// connected FE instead of the master).
+class SchemaPluginsScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaPluginsScanner);
+
+public:
+    SchemaPluginsScanner();
+    ~SchemaPluginsScanner() override = default;
+
+    Status start(RuntimeState* state) override;
+    Status get_next_block_internal(Block* block, bool* eos) override;
+
+    static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
+
+private:
+    Status _get_plugins_block_from_fe();
+
+    TNetworkAddress _fe_addr;
+
+    int _block_rows_limit = 4096;
+    int _row_idx = 0;
+    int _total_rows = 0;
+    int _rpc_timeout_ms = 3000;
+    std::unique_ptr<Block> _plugins_block = nullptr;
+};
+
+} // namespace doris

be/src/information_schema/schema_scanner.cpp

@@ -66,6 +66,7 @@
 #include "information_schema/schema_load_job_scanner.h"
 #include "information_schema/schema_metadata_name_ids_scanner.h"
 #include "information_schema/schema_partitions_scanner.h"
+#include "information_schema/schema_plugins_scanner.h"
 #include "information_schema/schema_processlist_scanner.h"
 #include "information_schema/schema_profiling_scanner.h"
 #include "information_schema/schema_role_mappings_scanner.h"
@@ -286,6 +287,8 @@ std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type
         return SchemaFileCacheInfoScanner::create_unique();
     case TSchemaTableType::SCH_AUTHENTICATION_INTEGRATIONS:
         return SchemaAuthenticationIntegrationsScanner::create_unique();
+    case TSchemaTableType::SCH_PLUGINS:
+        return SchemaPluginsScanner::create_unique();
     case TSchemaTableType::SCH_ROLE_MAPPINGS:
         return SchemaRoleMappingsScanner::create_unique();
     case TSchemaTableType::SCH_TABLE_STREAMS:

fe/fe-authentication/fe-authentication-handler/src/main/java/org/apache/doris/authentication/handler/AuthenticationPluginManager.java

@@ -26,12 +26,11 @@
 import org.apache.doris.extension.loader.LoadFailure;
 import org.apache.doris.extension.loader.LoadReport;
 import org.apache.doris.extension.loader.PluginHandle;
+import org.apache.doris.extension.loader.PluginRegistry;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-import java.io.Closeable;
-import java.io.IOException;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -63,6 +62,9 @@ public class AuthenticationPluginManager {
     private static final List<String> AUTH_PARENT_FIRST_PREFIXES =
             Collections.singletonList("org.apache.doris.authentication.");
 
+    /** Family label in the process-wide {@link PluginRegistry}. */
+    private static final String PLUGIN_FAMILY = "AUTHENTICATION";
+
     /** Factories by plugin name (e.g., "ldap", "oidc", "password") */
     private final Map<String, AuthenticationPluginFactory> factories = new ConcurrentHashMap<>();
 
@@ -88,7 +90,23 @@ public AuthenticationPluginManager(DirectoryPluginRuntimeManager<AuthenticationP
                 : ClassLoadingPolicy.defaultPolicy();
 
         ServiceLoader.load(AuthenticationPluginFactory.class)
-                .forEach(factory -> factories.put(factory.name(), factory));
+                .forEach(factory -> {
+                    try {
+                        // Snapshot self-reported metadata before publishing the factory so
+                        // one throwing implementation is rejected cleanly. The registry is
+                        // first-wins on (family, name), so re-registration from another
+                        // manager instance is a quiet no-op. First registration also wins
+                        // here, keeping the executable factory and the inventory row the
+                        // same plugin.
+                        PluginRegistry.getInstance().registerBuiltin(PLUGIN_FAMILY, factory);
+                        if (factories.putIfAbsent(factory.name(), factory) != null) {
+                            LOG.warn("Skip duplicated built-in authentication plugin name: {}", factory.name());
+                        }
+                    } catch (RuntimeException e) {
+                        LOG.warn("Skip built-in authentication plugin factory {}: self-reported metadata failed",
+                                factory.getClass().getName(), e);
+                    }
+                });
     }
 
     /**
@@ -136,11 +154,14 @@ public void loadAll(List<Path> pluginRoots, ClassLoader parent) throws Authentic
             String pluginName = handle.getPluginName();
             AuthenticationPluginFactory existing = factories.putIfAbsent(pluginName, handle.getFactory());
             if (existing != null) {
-                closeClassLoaderQuietly(handle.getClassLoader());
+                // Remove the rejected handle from the runtime manager too, so its
+                // classloader is not retained for the FE lifetime.
+                runtimeManager.discard(pluginName);
                 LOG.warn("Skip duplicated plugin name: {} from directory {}", pluginName, handle.getPluginDir());
                 continue;
             }
             loadedPlugins++;
+            PluginRegistry.getInstance().registerExternal(PLUGIN_FAMILY, handle);
             LOG.info("Loaded external authentication plugin: name={}, pluginDir={}, jarCount={}",
                     pluginName, handle.getPluginDir(), handle.getResolvedJars().size());
         }
@@ -165,17 +186,6 @@ private static LoadFailure firstNonConflictFailure(List<LoadFailure> failures) {
         return null;
     }
 
-    private static void closeClassLoaderQuietly(ClassLoader classLoader) {
-        if (!(classLoader instanceof Closeable)) {
-            return;
-        }
-        try {
-            ((Closeable) classLoader).close();
-        } catch (IOException ignored) {
-            // Best effort close.
-        }
-    }
-
     /**
      * Get a factory by plugin name.
      *

fe/fe-authentication/fe-authentication-handler/src/test/java/org/apache/doris/authentication/handler/AuthenticationPluginManagerTest.java

@@ -521,6 +521,22 @@ public Optional<PluginHandle<AuthenticationPluginFactory>> get(String pluginName
             return Optional.empty();
         }
 
+        @Override
+        public void discard(String pluginName) {
+            // This stub bypasses the real loadAll bookkeeping, so honor the
+            // contract that discarding a reported success closes its classloader.
+            for (PluginHandle<AuthenticationPluginFactory> handle : report.getSuccesses()) {
+                if (handle.getPluginName().equals(pluginName)
+                        && handle.getClassLoader() instanceof Closeable) {
+                    try {
+                        ((Closeable) handle.getClassLoader()).close();
+                    } catch (IOException ignored) {
+                        // test stub: never expected
+                    }
+                }
+            }
+        }
+
         @Override
         public List<PluginHandle<AuthenticationPluginFactory>> list() {
             return Collections.emptyList();

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

@@ -862,6 +862,14 @@ public class SchemaTable extends Table {
                             .column("ALTER_USER", ScalarType.createStringType())
                             .column("MODIFY_TIME", ScalarType.createStringType())
                             .build()))
+            .put("plugins",
+                    new SchemaTable(SystemIdGenerator.getNextId(), "plugins", TableType.SCHEMA,
+                        builder().column("PLUGIN_NAME", ScalarType.createVarchar(64))
+                            .column("PLUGIN_TYPE", ScalarType.createVarchar(64))
+                            .column("PLUGIN_VERSION", ScalarType.createVarchar(128))
+                            .column("SOURCE", ScalarType.createVarchar(16))
+                            .column("DESCRIPTION", ScalarType.createStringType())
+                            .build()))
             .put("table_streams",
                     new SchemaTable(SystemIdGenerator.getNextId(), "table_streams", TableType.SCHEMA,
                             builder().column("DB_NAME", ScalarType.createVarchar(NAME_CHAR_LEN))