本文研究 duckdb 内置的 extensions 工作机制。
插件架构
在 duckdb 源码内,内置了一组原生插件,位于顶层 extension 目录下:
除此之外,还支持 Out-of-Tree Extension,简单说就是独立的插件,不是集成在源码树中。独立编译好后,可以 运行时载入到 duckdb 中。
这里有一个细节:为了安全,任何 Out-of-Tree Extension,必须通过 duckdb 官方签名后,才能被正确载入。这个可以一定程度上防范载入有安全风险的第三方插件。
插件原理
为了明白原理,先简单看几个插件的核心注册逻辑。以 dbgen、httpfs、parquet 三个插件为例。
- dbgen
// https://github.com/duckdb/duckdb/blob/main/extension/tpch/tpch_extension.cpp
namespace duckdb {static void LoadInternal(DuckDB &db) {auto &db_instance = *db.instance;TableFunction dbgen_func("dbgen", {}, DbgenFunction, DbgenBind);dbgen_func.named_parameters["sf"] = LogicalType::DOUBLE;dbgen_func.named_parameters["overwrite"] = LogicalType::BOOLEAN;dbgen_func.named_parameters["catalog"] = LogicalType::VARCHAR;dbgen_func.named_parameters["schema"] = LogicalType::VARCHAR;dbgen_func.named_parameters["suffix"] = LogicalType::VARCHAR;dbgen_func.named_parameters["children"] = LogicalType::UINTEGER;dbgen_func.named_parameters["step"] = LogicalType::UINTEGER;ExtensionUtil::RegisterFunction(db_instance, dbgen_func);// create the TPCH pragma that allows us to run the queryauto tpch_func = PragmaFunction::PragmaCall("tpch", PragmaTpchQuery, {LogicalType::BIGINT});ExtensionUtil::RegisterFunction(db_instance, tpch_func);// create the TPCH_QUERIES function that returns the queryTableFunction tpch_query_func("tpch_queries", {}, TPCHQueryFunction, TPCHQueryBind, TPCHInit);ExtensionUtil::RegisterFunction(db_instance, tpch_query_func);// create the TPCH_ANSWERS that returns the query resultTableFunction tpch_query_answer_func("tpch_answers", {}, TPCHQueryAnswerFunction, TPCHQueryAnswerBind, TPCHInit);ExtensionUtil::RegisterFunction(db_instance, tpch_query_answer_func);
}void TpchExtension::Load(DuckDB &db) {LoadInternal(db);
}std::string TpchExtension::GetQuery(int query) {return tpch::DBGenWrapper::GetQuery(query);
}std::string TpchExtension::GetAnswer(double sf, int query) {return tpch::DBGenWrapper::GetAnswer(sf, query);
}std::string TpchExtension::Name() {return "tpch";
}std::string TpchExtension::Version() const {
#ifdef EXT_VERSION_TPCHreturn EXT_VERSION_TPCH;
#elsereturn "";
#endif
}} // namespace duckdbextern "C" {DUCKDB_EXTENSION_API void tpch_init(duckdb::DatabaseInstance &db) {duckdb::DuckDB db_wrapper(db);duckdb::LoadInternal(db_wrapper);
}DUCKDB_EXTENSION_API const char *tpch_version() {return duckdb::DuckDB::LibraryVersion();
}
}
- httpfs
namespace duckdb {static void LoadInternal(DatabaseInstance &instance) {S3FileSystem::Verify(); // run some tests to see if all the hashes work outauto &fs = instance.GetFileSystem();fs.RegisterSubSystem(make_uniq<HTTPFileSystem>());fs.RegisterSubSystem(make_uniq<HuggingFaceFileSystem>());fs.RegisterSubSystem(make_uniq<S3FileSystem>(BufferManager::GetBufferManager(instance)));auto &config = DBConfig::GetConfig(instance);// Global HTTP config// Single timeout value is used for all 4 types of timeouts, we could split it into 4 if users need thatconfig.AddExtensionOption("http_timeout", "HTTP timeout read/write/connection/retry", LogicalType::UBIGINT,Value(30000));config.AddExtensionOption("http_retries", "HTTP retries on I/O error", LogicalType::UBIGINT, Value(3));config.AddExtensionOption("http_retry_wait_ms", "Time between retries", LogicalType::UBIGINT, Value(100));config.AddExtensionOption("force_download", "Forces upfront download of file", LogicalType::BOOLEAN, Value(false));// Reduces the number of requests made while waiting, for example retry_wait_ms of 50 and backoff factor of 2 will// result in wait times of 0 50 100 200 400...etc.config.AddExtensionOption("http_retry_backoff", "Backoff factor for exponentially increasing retry wait time",LogicalType::FLOAT, Value(4));config.AddExtensionOption("http_keep_alive","Keep alive connections. Setting this to false can help when running into connection failures",LogicalType::BOOLEAN, Value(true));config.AddExtensionOption("enable_server_cert_verification", "Enable server side certificate verification.",LogicalType::BOOLEAN, Value(false));config.AddExtensionOption("ca_cert_file", "Path to a custom certificate file for self-signed certificates.",LogicalType::VARCHAR, Value(""));// Global S3 configconfig.AddExtensionOption("s3_region", "S3 Region", LogicalType::VARCHAR, Value("us-east-1"));config.AddExtensionOption("s3_access_key_id", "S3 Access Key ID", LogicalType::VARCHAR);config.AddExtensionOption("s3_secret_access_key", "S3 Access Key", LogicalType::VARCHAR);config.AddExtensionOption("s3_session_token", "S3 Session Token", LogicalType::VARCHAR);config.AddExtensionOption("s3_endpoint", "S3 Endpoint", LogicalType::VARCHAR);config.AddExtensionOption("s3_url_style", "S3 URL style", LogicalType::VARCHAR, Value("vhost"));config.AddExtensionOption("s3_use_ssl", "S3 use SSL", LogicalType::BOOLEAN, Value(true));config.AddExtensionOption("s3_url_compatibility_mode", "Disable Globs and Query Parameters on S3 URLs",LogicalType::BOOLEAN, Value(false));// S3 Uploader configconfig.AddExtensionOption("s3_uploader_max_filesize", "S3 Uploader max filesize (between 50GB and 5TB)",LogicalType::VARCHAR, "800GB");config.AddExtensionOption("s3_uploader_max_parts_per_file", "S3 Uploader max parts per file (between 1 and 10000)",LogicalType::UBIGINT, Value(10000));config.AddExtensionOption("s3_uploader_thread_limit", "S3 Uploader global thread limit", LogicalType::UBIGINT,Value(50));// HuggingFace optionsconfig.AddExtensionOption("hf_max_per_page", "Debug option to limit number of items returned in list requests",LogicalType::UBIGINT, Value::UBIGINT(0));auto provider = make_uniq<AWSEnvironmentCredentialsProvider>(config);provider->SetAll();CreateS3SecretFunctions::Register(instance);CreateBearerTokenFunctions::Register(instance);
}void HttpfsExtension::Load(DuckDB &db) {LoadInternal(*db.instance);
}
std::string HttpfsExtension::Name() {return "httpfs";
}std::string HttpfsExtension::Version() const {
#ifdef EXT_VERSION_HTTPFSreturn EXT_VERSION_HTTPFS;
#elsereturn "";
#endif
}} // namespace duckdbextern "C" {DUCKDB_EXTENSION_API void httpfs_init(duckdb::DatabaseInstance &db) {LoadInternal(db);
}DUCKDB_EXTENSION_API const char *httpfs_version() {return duckdb::DuckDB::LibraryVersion();
}
}
- parquet
void ParquetExtension::Load(DuckDB &db) {auto &db_instance = *db.instance;auto &fs = db.GetFileSystem();fs.RegisterSubSystem(FileCompressionType::ZSTD, make_uniq<ZStdFileSystem>());auto scan_fun = ParquetScanFunction::GetFunctionSet();scan_fun.name = "read_parquet";ExtensionUtil::RegisterFunction(db_instance, scan_fun);scan_fun.name = "parquet_scan";ExtensionUtil::RegisterFunction(db_instance, scan_fun);// parquet_metadataParquetMetaDataFunction meta_fun;ExtensionUtil::RegisterFunction(db_instance, MultiFileReader::CreateFunctionSet(meta_fun));// parquet_schemaParquetSchemaFunction schema_fun;ExtensionUtil::RegisterFunction(db_instance, MultiFileReader::CreateFunctionSet(schema_fun));// parquet_key_value_metadataParquetKeyValueMetadataFunction kv_meta_fun;ExtensionUtil::RegisterFunction(db_instance, MultiFileReader::CreateFunctionSet(kv_meta_fun));// parquet_file_metadataParquetFileMetadataFunction file_meta_fun;ExtensionUtil::RegisterFunction(db_instance, MultiFileReader::CreateFunctionSet(file_meta_fun));CopyFunction function("parquet");function.copy_to_bind = ParquetWriteBind;function.copy_to_initialize_global = ParquetWriteInitializeGlobal;function.copy_to_initialize_local = ParquetWriteInitializeLocal;function.copy_to_sink = ParquetWriteSink;function.copy_to_combine = ParquetWriteCombine;function.copy_to_finalize = ParquetWriteFinalize;function.execution_mode = ParquetWriteExecutionMode;function.copy_from_bind = ParquetScanFunction::ParquetReadBind;function.copy_from_function = scan_fun.functions[0];function.prepare_batch = ParquetWritePrepareBatch;function.flush_batch = ParquetWriteFlushBatch;function.desired_batch_size = ParquetWriteDesiredBatchSize;function.file_size_bytes = ParquetWriteFileSize;function.serialize = ParquetCopySerialize;function.deserialize = ParquetCopyDeserialize;function.supports_type = ParquetWriter::TypeIsSupported;function.extension = "parquet";ExtensionUtil::RegisterFunction(db_instance, function);// parquet_keyauto parquet_key_fun = PragmaFunction::PragmaCall("add_parquet_key", ParquetCrypto::AddKey,{LogicalType::VARCHAR, LogicalType::VARCHAR});ExtensionUtil::RegisterFunction(db_instance, parquet_key_fun);auto &config = DBConfig::GetConfig(*db.instance);config.replacement_scans.emplace_back(ParquetScanReplacement);config.AddExtensionOption("binary_as_string", "In Parquet files, interpret binary data as a string.",LogicalType::BOOLEAN);
}std::string ParquetExtension::Name() {return "parquet";
}std::string ParquetExtension::Version() const {
#ifdef EXT_VERSION_PARQUETreturn EXT_VERSION_PARQUET;
#elsereturn "";
#endif
}} // namespace duckdb#ifdef DUCKDB_BUILD_LOADABLE_EXTENSION
extern "C" {DUCKDB_EXTENSION_API void parquet_init(duckdb::DatabaseInstance &db) { // NOLINTduckdb::DuckDB db_wrapper(db);db_wrapper.LoadExtension<duckdb::ParquetExtension>();
}DUCKDB_EXTENSION_API const char *parquet_version() { // NOLINTreturn duckdb::DuckDB::LibraryVersion();
}
}
#endif
上面三个插件里,分别出现了如下注册字眼:
// tpchTableFunction tpch_query_answer_func("tpch_answers", {}, TPCHQueryAnswerFunction, TPCHQueryAnswerBind, TPCHInit);ExtensionUtil::RegisterFunction(db_instance, tpch_query_answer_func);// parquetauto &db_instance = *db.instance;auto &fs = db.GetFileSystem();fs.RegisterSubSystem(FileCompressionType::ZSTD, make_uniq<ZStdFileSystem>());ExtensionUtil::RegisterFunction(db_instance, MultiFileReader::CreateFunctionSet(meta_fun));// httpfsauto &fs = instance.GetFileSystem();fs.RegisterSubSystem(make_uniq<HTTPFileSystem>());fs.RegisterSubSystem(make_uniq<HuggingFaceFileSystem>());fs.RegisterSubSystem(make_uniq<S3FileSystem>(BufferManager::GetBufferManager(instance)));ExtensionUtil::RegisterSecretType(instance, secret_type);
可以看到,ExtensionUtil 在这里面扮演了很重要的角色。ExtensionUtil 支持注册普通函数、聚合函数、Table 函数、PragmaFunction、Collation、Secret 等。对于 Duckdb 的框架,了解到这一层已经差不多了,至于 ExtensionUtil 的实现,大家可以根据自己系统的特性灵活处理,本质上无非就是针对每一个类型的扩展,设计一套可扩展、可查找、高性能的功能集合接口。
class ExtensionUtil {
public://! Register a new scalar function - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, ScalarFunction function);//! Register a new scalar function set - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, ScalarFunctionSet function);//! Register a new aggregate function - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, AggregateFunction function);//! Register a new aggregate function set - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, AggregateFunctionSet function);//! Register a new table function - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, TableFunction function);//! Register a new table function set - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, TableFunctionSet function);//! Register a new pragma function - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, PragmaFunction function);//! Register a new pragma function set - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, PragmaFunctionSet function);//! Register a CreateSecretFunctionDUCKDB_API static void RegisterFunction(DatabaseInstance &db, CreateSecretFunction function);//! Register a new copy function - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, CopyFunction function);//! Register a new macro function - throw an exception if the function already existsDUCKDB_API static void RegisterFunction(DatabaseInstance &db, CreateMacroInfo &info);//! Register a new collationDUCKDB_API static void RegisterCollation(DatabaseInstance &db, CreateCollationInfo &info);//! Returns a reference to the function in the catalog - throws an exception if it does not existDUCKDB_API static ScalarFunctionCatalogEntry &GetFunction(DatabaseInstance &db, const string &name);DUCKDB_API static TableFunctionCatalogEntry &GetTableFunction(DatabaseInstance &db, const string &name);//! Add a function overloadDUCKDB_API static void AddFunctionOverload(DatabaseInstance &db, ScalarFunction function);DUCKDB_API static void AddFunctionOverload(DatabaseInstance &db, ScalarFunctionSet function);DUCKDB_API static void AddFunctionOverload(DatabaseInstance &db, TableFunctionSet function);//! Registers a new typeDUCKDB_API static void RegisterType(DatabaseInstance &db, string type_name, LogicalType type,bind_type_modifiers_function_t bind_type_modifiers = nullptr);//! Registers a new secret typeDUCKDB_API static void RegisterSecretType(DatabaseInstance &db, SecretType secret_type);//! Registers a cast between two typesDUCKDB_API static void RegisterCastFunction(DatabaseInstance &db, const LogicalType &source,const LogicalType &target, BoundCastInfo function,int64_t implicit_cast_cost = -1);
};} // namespace duckdb
另外,还用到了 RegisterSubSystem
,这说明对于文件系统,Duckdb 也做了抽象,方便插件化。
instance.GetFileSystem().RegisterSubSystem(make_uniq<HTTPFileSystem>());
总结
插件机制是个好东西。为了你的系统能够获得这个好东西,最好是系统在第一天设计的时候就做好抽象,把每一类系统功能都抽象成集合,每一类功能都支持“运行时可寻址”。
在 OceanBase 中,最容易插件化的是系统函数,因为它数量众多,客观上从第一天起就逼迫 OceanBase 把它做成易扩展、易寻址的样子。
反面例子是聚合函数,OceanBase 在最初的时候,只支持很少的聚合函数,所以并没有对其做抽象,而是直接在 parsser 里就 hardcode 地抽取出了聚合函数信息,这导致在运行时几乎不可能对其做扩展。如下图所示:
src/parser/sql_parser_mysql_mode.y
至于 filesystem 支持、文件格式支持,OceanBase 更没有做好抽象,他们的插件化,也需要做大量改造才可能。