hanzoai/datastore
2
0
Fork 0
mirror of https://github.com/hanzoai/datastore.git synced 2026-09-26 13:55:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into add-executable-udf-drivers

Browse Source
This commit is contained in:
Alexey Milovidov
2026-06-10 13:44:46 +00:00
parent 3d1839f7e1 b0e989e861
commit ee08129f75
442 changed files with 20650 additions and 1176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
programs/server/Server.cpp
+87 -2
View File
@@ -47,6 +47,7 @@
#include <Common/getExecutablePath.h>
#include <Common/ProfileEvents.h>
#include <Common/Scheduler/IResourceManager.h>
#include <Common/ThreadGroupSwitcher.h>
#include <Common/ThreadProfileEvents.h>
#include <Common/ThreadStatus.h>
#include <Common/getMappedArea.h>
@@ -117,8 +118,11 @@
#include <Compression/CompressionCodecEncrypted.h>
#include <Parsers/ASTAlterQuery.h>
#include <Server/CloudPlacementInfo.h>
#include <Server/DistributedQuery/ExchangeConnections.h>
#include <Server/DistributedQuery/ExchangeServer.h>
#include <Server/HTTP/HTTPServer.h>
#include <Server/HTTP/HTTPServerConnectionFactory.h>
#include <Server/StatelessWorker/StatelessWorkerEndpoint.h>
#include <Server/MySQLHandlerFactory.h>
#include <Server/PostgreSQLHandlerFactory.h>
#include <Server/ProtocolServerAdapter.h>
@@ -1074,7 +1078,8 @@ void loadStartupScripts(const Poco::Util::AbstractConfiguration & config, const
startup_context->setCurrentQueryId("");
{
auto query_scope = QueryScope::create(startup_context);
auto thread_group = ThreadGroup::createForQuery(startup_context);
ThreadGroupSwitcher switcher(thread_group, /*allow_existing_group=*/ true);
executeQuery(condition_read_buffer, condition_write_buffer, startup_context, callback, QueryFlags{ .internal = true }, std::nullopt, {});
}
@@ -1107,7 +1112,8 @@ void loadStartupScripts(const Poco::Util::AbstractConfiguration & config, const
startup_context->setQueryKind(ClientInfo::QueryKind::INITIAL_QUERY);
startup_context->setCurrentQueryId("");
auto query_scope = QueryScope::create(startup_context);
auto thread_group = ThreadGroup::createForQuery(startup_context);
ThreadGroupSwitcher switcher(thread_group, /*allow_existing_group=*/ true);
executeQuery(read_buffer, write_buffer, startup_context, callback, QueryFlags{ .internal = true }, std::nullopt, {});
}
@@ -2043,6 +2049,82 @@ try
LOG_DEBUG(log, "Initializing interserver credentials.");
global_context->updateInterserverCredentials(config());
std::shared_ptr<StatelessWorkerEndpoint> stateless_worker_endpoint_ptr{nullptr};
String stateless_worker_endpoint_name;
if (config().getBool("stateless_worker_server.enabled", false))
{
String stateless_worker_endpoint = config().getString("stateless_worker_server.endpoint", "localhost");
stateless_worker_endpoint_ptr = std::make_shared<StatelessWorkerEndpoint>();
stateless_worker_endpoint_name = stateless_worker_endpoint_ptr->getId(stateless_worker_endpoint);
global_context->getInterserverIOHandler().addEndpoint(stateless_worker_endpoint_name, stateless_worker_endpoint_ptr);
LOG_DEBUG(log, "Added stateless worker endpoint '{}'.", stateless_worker_endpoint_name);
}
SCOPE_EXIT({
if (stateless_worker_endpoint_ptr)
{
/// Remove the same endpoint that was registered (the configured name may differ from "localhost").
LOG_DEBUG(log, "Shutting down stateless worker endpoint '{}'.", stateless_worker_endpoint_name);
global_context->getInterserverIOHandler().removeEndpointIfExists(stateless_worker_endpoint_name);
stateless_worker_endpoint_ptr->blocker.cancelForever();
stateless_worker_endpoint_ptr->shutdown();
/// Acquire the lock to wait for all in-flight requests to finish.
std::lock_guard lock(stateless_worker_endpoint_ptr->rwlock);
}
stateless_worker_endpoint_ptr.reset();
});
#ifdef OS_LINUX
ExchangeConnectionsPtr exchange_connections_ptr = ExchangeConnections::instance();
std::vector<std::shared_ptr<ExchangeServer>> exchange_servers;
if (auto streaming_exchange_port = config().getUInt("distributed_query.streaming_exchange_port", 0))
{
if (streaming_exchange_port > 65535)
throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER,
"`distributed_query.streaming_exchange_port` must be in range 1..65535, got {}", streaming_exchange_port);
/// The exchange handshake is unauthenticated, so the listener is never bound to all interfaces
/// implicitly: the streaming exchange is enabled only when explicit listen host(s) are given.
Strings exchange_listen_hosts = DB::getMultipleValuesFromConfig(config(), "distributed_query", "streaming_exchange_listen_host");
if (exchange_listen_hosts.empty())
{
LOG_ERROR(log, "`distributed_query.streaming_exchange_port` is set but no "
"`distributed_query.streaming_exchange_listen_host` is configured; the streaming exchange "
"server is not started. Specify a listen host to enable it.");
}
else
{
for (const auto & listen_host : exchange_listen_hosts)
{
try
{
exchange_servers.emplace_back(std::make_shared<ExchangeServer>(listen_host, streaming_exchange_port, exchange_connections_ptr));
exchange_servers.back()->start();
}
catch (Poco::Exception & e)
{
LOG_INFO(log, "Failed to start exchange server on {}:{}: {}",
listen_host, streaming_exchange_port, e.displayText());
}
}
if (exchange_servers.empty())
throw Exception(ErrorCodes::NETWORK_ERROR, "Failed to start ExchangeServer on port {}", streaming_exchange_port);
}
}
SCOPE_EXIT({
for (auto & exchange_server_ptr : exchange_servers)
{
exchange_server_ptr->stop();
exchange_server_ptr.reset();
}
});
#else
if (config().getUInt("distributed_query.streaming_exchange_port", 0))
throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "ExchangeServer is not supported on non-linux platform");
#endif
/// Set up caches.
JemallocCacheArena::setEnabled(server_settings[ServerSetting::use_separate_cache_arena]);
@@ -2571,6 +2653,9 @@ try
std::lock_guard lock(servers_lock);
updateServers(config(), new_server_settings, server_pool, *async_metrics, servers, servers_to_start_before_tables);
}
if (config().has("startup_scripts"))
loadStartupScripts(config(), new_server_settings, global_context, log);
}
global_context->updateStorageConfiguration(config());