SiriusTableFunctionData() = default;

shared_ptr<::sirius::sirius_prepared_statement_data> gpu_prepared;

unique_ptr<QueryResult> res;

unique_ptr<Connection> conn;

unique_ptr<::sirius::sirius_interface> sirius_iface;

string query;

bool enable_optimizer;

bool finished = false;

bool plan_error = false;

//! Original options from the connection

ClientConfig original_config;

set<OptimizerType> original_disabled_optimizers;

void PrepareConnection(ClientContext& context)

{

// First collect original options

original_config = context.config;

original_disabled_optimizers = DBConfig::GetConfig(context).options.disabled_optimizers;

// The user might want to disable the optimizer of the new connection

context.config.enable_optimizer = enable_optimizer;

// We want for sure to disable the internal compression optimizations.

// These are DuckDB specific, no other system implements these. Also,

// respect the user's settings if they chose to disable any specific optimizers.

//

// The InClauseRewriter optimization converts large `IN` clauses to a

// "mark join" against a `ColumnDataCollection`, which may not make

// sense in other systems and would complicate the conversion to Substrait.

set<OptimizerType> disabled_optimizers =

DBConfig::GetConfig(context).options.disabled_optimizers;

disabled_optimizers.insert(OptimizerType::IN_CLAUSE);

disabled_optimizers.insert(OptimizerType::COMPRESSED_MATERIALIZATION);

// STATISTICS_PROPAGATION is now enabled: cpu_source_task handles the

// COLUMN_DATA_SCAN / EXPRESSION_GET / DUMMY_SCAN sources that this

// optimizer produces (e.g. folding count(*), MIN, MAX to constants).

#ifdef DEBUG

disabled_optimizers.insert(OptimizerType::COLUMN_LIFETIME);

#endif

// disabled_optimizers.insert(OptimizerType::MATERIALIZED_CTE);

// If error(varchar) gets implemented in substrait this can be removed

// context.config.scalar_subquery_error_on_multiple_rows = false;

DBConfig::GetConfig(context).options.disabled_optimizers = disabled_optimizers;

}

// Reset configuration

void CleanupConnection(ClientContext& context) const

{

DBConfig::GetConfig(context).options.disabled_optimizers = original_disabled_optimizers;

context.config = original_config;

}

unique_ptr<LogicalOperator> ExtractPlan(ClientContext& context)

{

PrepareConnection(context);

unique_ptr<LogicalOperator> plan;

try {

Parser parser(context.GetParserOptions());

parser.ParseQuery(query);

Planner planner(context);

planner.CreatePlan(std::move(parser.statements[0]));

D_ASSERT(planner.plan);

plan = std::move(planner.plan);

if (context.config.enable_optimizer) {

Optimizer optimizer(*planner.binder, context);

plan = optimizer.Optimize(std::move(plan));

}

// After optimization, refresh types before column binding resolution

// to ensure types are consistent (some optimizers may have set stale types)

plan->ResolveOperatorTypes();

ColumnBindingResolver resolver;

ColumnBindingResolver::Verify(*plan);

resolver.VisitOperator(*plan);

} catch (...) {

CleanupConnection(context);

throw;

}

CleanupConnection(context);

return plan;

}

GPUTableFunctionData() = default;

shared_ptr<Relation> plan;

shared_ptr<GPUPreparedStatementData> gpu_prepared;

unique_ptr<QueryResult> res;

unique_ptr<Connection> conn;

unique_ptr<GPUContext> gpu_context;

string query;

bool enable_optimizer;

bool finished = false;

bool plan_error = false;

//! Original options from the connection

ClientConfig original_config;

set<OptimizerType> original_disabled_optimizers;

void PrepareConnection(ClientContext& context)

{

// First collect original options

original_config = context.config;

original_disabled_optimizers = DBConfig::GetConfig(context).options.disabled_optimizers;

// The user might want to disable the optimizer of the new connection

context.config.enable_optimizer = enable_optimizer;

// We want for sure to disable the internal compression optimizations.

// These are DuckDB specific, no other system implements these. Also,

// respect the user's settings if they chose to disable any specific optimizers.

//

// The InClauseRewriter optimization converts large `IN` clauses to a

// "mark join" against a `ColumnDataCollection`, which may not make

// sense in other systems and would complicate the conversion to Substrait.

set<OptimizerType> disabled_optimizers =

DBConfig::GetConfig(context).options.disabled_optimizers;

disabled_optimizers.insert(OptimizerType::IN_CLAUSE);

disabled_optimizers.insert(OptimizerType::COMPRESSED_MATERIALIZATION);

// STATISTICS_PROPAGATION folds ungrouped MIN/MAX aggregates into constant

// expressions using partition statistics, producing EXPRESSION_GET + DUMMY_SCAN.

// The GPU pipeline cannot schedule COLUMN_DATA_SCAN sources, so disable this

// to keep the query on the scan -> aggregate path where the GPU can execute it.

disabled_optimizers.insert(OptimizerType::STATISTICS_PROPAGATION);

#ifdef DEBUG

disabled_optimizers.insert(OptimizerType::COLUMN_LIFETIME);

#endif

// disabled_optimizers.insert(OptimizerType::MATERIALIZED_CTE);

// If error(varchar) gets implemented in substrait this can be removed

// context.config.scalar_subquery_error_on_multiple_rows = false;

DBConfig::GetConfig(context).options.disabled_optimizers = disabled_optimizers;

}

// Reset configuration

void CleanupConnection(ClientContext& context) const

{

DBConfig::GetConfig(context).options.disabled_optimizers = original_disabled_optimizers;

context.config = original_config;

}

unique_ptr<LogicalOperator> ExtractPlan(ClientContext& context)

{

PrepareConnection(context);

unique_ptr<LogicalOperator> plan;

try {

Parser parser(context.GetParserOptions());

parser.ParseQuery(query);

Planner planner(context);

planner.CreatePlan(std::move(parser.statements[0]));

D_ASSERT(planner.plan);

plan = std::move(planner.plan);

if (context.config.enable_optimizer) {

Optimizer optimizer(*planner.binder, context);

plan = optimizer.Optimize(std::move(plan));

}

// After optimization, refresh types before column binding resolution

// to ensure types are consistent (some optimizers may have set stale types)

plan->ResolveOperatorTypes();

ColumnBindingResolver resolver;

ColumnBindingResolver::Verify(*plan);

resolver.VisitOperator(*plan);

} catch (...) {

CleanupConnection(context);

throw;

}

CleanupConnection(context);

return plan;

}

ClientContext& context,

GPUContext& gpu_context,

unique_ptr<LogicalOperator>& logical_plan,

Connection& new_conn)

{

GPUPhysicalPlanGenerator physical_planner = GPUPhysicalPlanGenerator(context, gpu_context);

auto physical_plan = physical_planner.CreatePlan(std::move(logical_plan));

return physical_plan;

TableFunctionBindInput& input,

vector<LogicalType>& return_types,

vector<string>& names)

{

auto result = make_uniq<GPUTableFunctionData>();

result->conn = make_uniq<Connection>(*context.db);

result->query = input.inputs[0].ToString();

result->enable_optimizer = true;

result->gpu_context = make_uniq<GPUContext>(context);

if (input.inputs[0].IsNull()) {

throw BinderException("gpu_processing cannot be called with a NULL parameter");

}

// Parse the query just to get the result type information and to create preparedstatmement data

auto statements = result->conn->context->ParseStatements(result->query);

Planner planner(context);

auto statement_type = statements[0]->type;

planner.CreatePlan(std::move(statements[0]));

D_ASSERT(planner.plan);

auto prepared = make_shared_ptr<PreparedStatementData>(statement_type);

prepared->names = planner.names;

prepared->types = planner.types;

prepared->value_map = std::move(planner.value_map);

// generate physical plan from the logical plan

unique_ptr<LogicalOperator> query_plan = result->ExtractPlan(context);

SIRIUS_LOG_DEBUG("Query plan:\n{}", query_plan->ToString());

if (buffer_is_initialized) {

try {

auto gpu_physical_plan =

GPUGeneratePhysicalPlan(context, *result->gpu_context, query_plan, *result->conn);

auto gpu_prepared = make_shared_ptr<GPUPreparedStatementData>(std::move(prepared),

std::move(gpu_physical_plan));

result->gpu_prepared = gpu_prepared;

} catch (std::exception& e) {

ErrorData error(e);

SIRIUS_LOG_ERROR("Error in GPUGeneratePhysicalPlan: {}", error.RawMessage());

result->plan_error = true;

}

} else {

result->gpu_prepared = nullptr;

}

for (auto& column : planner.names) {

names.emplace_back(column);

}

for (auto& type : planner.types) {

return_types.emplace_back(type);

}

return std::move(result);

TableFunctionInput& data_p,

DataChunk& output)

{

auto& data = (GPUTableFunctionData&)*data_p.bind_data;

if (data.finished) { return; }

if (!data.res) {

auto start = std::chrono::high_resolution_clock::now();

if (!buffer_is_initialized) {

printf("\033[1;31m");

printf("GPUBufferManager not initialized, please call gpu_buffer_init first\n");

printf("\033[0m");

printf(

"=============================================\nError in GPUExecuteQuery, fallback to "

"DuckDB\n=============================================\n");

data.res = data.conn->Query(data.query);

} else if (data.plan_error) {

printf(

"=============================================\nError in GPUExecuteQuery, fallback to "

"DuckDB\n=============================================\n");

data.res = data.conn->Query(data.query);

} else {

data.res = data.gpu_context->GPUExecuteQuery(context, data.query, data.gpu_prepared, {});

if (data.res->HasError()) {

printf(

"=============================================\nError in GPUExecuteQuery, fallback to "

"DuckDB\n=============================================\n");

data.res = data.conn->Query(data.query);

}

}

auto end = std::chrono::high_resolution_clock::now();

auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);

SIRIUS_LOG_INFO("Execute query time: {:.2f} ms", duration.count() / 1000.0);

}

auto result_chunk = data.res->Fetch();

if (result_chunk == nullptr) {

output.SetCardinality(0);

return;

}

output.Reference(*result_chunk);

return;

{

TableFunction gpu_processing("gpu_processing",

{LogicalType::VARCHAR},

SiriusExtension::GPUProcessingFunction,

SiriusExtension::GPUProcessingBind);

gpu_processing.named_parameters["enable_optimizer"] = LogicalType::BOOLEAN;

CreateTableFunctionInfo gpu_processing_info(gpu_processing);

catalog.CreateTableFunction(transaction, gpu_processing_info);

ClientContext& context, unique_ptr<LogicalOperator>& logical_plan)

{

sirius::planner::sirius_physical_plan_generator physical_planner =

sirius::planner::sirius_physical_plan_generator(context);

auto physical_plan = physical_planner.create_plan(std::move(logical_plan));

return physical_plan;

TableFunctionBindInput& input,

vector<LogicalType>& return_types,

vector<string>& names)

{

auto result = make_uniq<SiriusTableFunctionData>();

result->conn = make_uniq<Connection>(*context.db);

result->query = input.inputs[0].ToString();

result->enable_optimizer = true;

result->sirius_iface = make_uniq<::sirius::sirius_interface>(context);

if (input.inputs[0].IsNull()) {

throw BinderException("gpu_execution cannot be called with a NULL parameter");

}

// Parse the query just to get the result type information and to create preparedstatmement data

Parser parser(context.GetParserOptions());

parser.ParseQuery(result->query);

Planner planner(context);

auto statement_type = parser.statements[0]->type;

planner.CreatePlan(std::move(parser.statements[0]));

D_ASSERT(planner.plan);

auto prepared = make_shared_ptr<PreparedStatementData>(statement_type);

prepared->names = planner.names;

prepared->types = planner.types;

prepared->value_map = std::move(planner.value_map);

// generate physical plan from the logical plan

unique_ptr<LogicalOperator> query_plan = result->ExtractPlan(context);

SIRIUS_LOG_DEBUG("Query plan:\n{}", query_plan->ToString());

try {

auto sirius_physical_plan = SiriusGeneratePhysicalPlan(context, query_plan);

SIRIUS_LOG_DEBUG("Done generating sirius physical plan");

auto gpu_prepared = make_shared_ptr<::sirius::sirius_prepared_statement_data>(

std::move(prepared), std::move(sirius_physical_plan));

result->gpu_prepared = gpu_prepared;

} catch (std::exception& e) {

ErrorData error(e);

SIRIUS_LOG_ERROR("Error in SiriusGeneratePhysicalPlan: {}", error.RawMessage());

if (Config::ENABLE_DUCKDB_FALLBACK) {

result->plan_error = true;

} else {

throw std::runtime_error("Error in SiriusGeneratePhysicalPlan: " + error.RawMessage());

return nullptr;

}

}

for (auto& column : planner.names) {

names.emplace_back(column);

}

for (auto& type : planner.types) {

return_types.emplace_back(type);

}

return std::move(result);

TableFunctionInput& data_p,

DataChunk& output)

{

auto& data = (SiriusTableFunctionData&)*data_p.bind_data;

if (data.finished) { return; }

if (!data.res) {

auto start = std::chrono::high_resolution_clock::now();

if (data.plan_error) {

printf(

"=============================================\nError in SiriusExecuteQuery, fallback to "

"DuckDB\n=============================================\n");

data.res = data.conn->Query(data.query);

} else {

data.res =

data.sirius_iface->sirius_execute_query(context, data.query, data.gpu_prepared, {});

if (data.res->HasError()) {

if (Config::ENABLE_DUCKDB_FALLBACK) {

SIRIUS_LOG_ERROR("SiriusExecuteQuery error: {}", data.res->GetError());

printf(

"=============================================\nError in SiriusExecuteQuery, fallback "

"to DuckDB\n=============================================\n");

data.res = data.conn->Query(data.query);

} else {

throw std::runtime_error("SiriusExecuteQuery error: " + data.res->GetError());

return;

}

}

}

auto end = std::chrono::high_resolution_clock::now();

auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start);

SIRIUS_LOG_INFO("Execute query time: {:.2f} ms", duration.count() / 1000.0);

}

auto result_chunk = data.res->Fetch();

if (result_chunk == nullptr) {

output.SetCardinality(0);

return;

}

output.Reference(*result_chunk);

return;

Planner& planner,

Connection& new_conn)

{

unique_ptr<LogicalOperator> plan;

plan = std::move(planner.plan);

Optimizer optimizer(*planner.binder, context);

plan = optimizer.Optimize(std::move(plan));

SIRIUS_LOG_DEBUG("Query plan:\n{}", plan->ToString());

ColumnBindingResolver resolver;

resolver.Verify(*plan);

resolver.VisitOperator(*plan);

plan->ResolveOperatorTypes();

return plan;

GPUBufferInitFunctionData() {}

bool finished = false;

size_t cache_size;

size_t processing_size;

size_t pinned_memory_size;

TableFunctionBindInput& input,

vector<LogicalType>& return_types,

vector<string>& names)

{

auto result = make_uniq<GPUBufferInitFunctionData>();

string gpu_cache_size = input.inputs[0].ToString();

string gpu_processing_size = input.inputs[1].ToString();

string pinned_memory_size("0 GB"); // Default size of pinned memory

if (input.named_parameters.find(PINNED_MEMORY_PARAM_KEY) != input.named_parameters.end()) {

// If the pinned memory size is specified in the arguments then use that

pinned_memory_size = input.named_parameters[PINNED_MEMORY_PARAM_KEY].ToString();

}

// parsing 2GB or 2GiB to size_t

// Function to parse size strings like "2GB" or "2GiB" to size_t

auto parse_size = [](const string& size_str) -> size_t {

size_t result = 0;

size_t multiplier = 1;

string num_part;

string unit_part;

size_t i = 0;

// Skip any whitespace between number and unit

while (i < size_str.length() && isspace(size_str[i])) {

i++;

}

// Find where the number ends and unit begins

while (i < size_str.length() && (isdigit(size_str[i]) || size_str[i] == '.')) {

num_part += size_str[i];

i++;

}

// Skip any whitespace between number and unit

while (i < size_str.length() && isspace(size_str[i])) {

i++;

}

// Extract unit part

unit_part = size_str.substr(i);

// Convert number part to double

double num_value = stod(num_part);

// Determine multiplier based on unit

if (unit_part == "B") {

multiplier = 1;

} else if (unit_part == "KB" || unit_part == "KiB") {

multiplier = 1024;

} else if (unit_part == "MB" || unit_part == "MiB") {

multiplier = 1024 * 1024;

} else if (unit_part == "GB" || unit_part == "GiB") {

multiplier = 1024 * 1024 * 1024;

} else if (unit_part == "TB" || unit_part == "TiB") {

multiplier = 1024ULL * 1024ULL * 1024ULL * 1024ULL;

} else {

throw InvalidInputException("Invalid format");

}

result = (size_t)(num_value * multiplier);

return result;

};

// Parse the input sizes

result->cache_size = parse_size(gpu_cache_size);

result->processing_size = parse_size(gpu_processing_size);

result->pinned_memory_size = parse_size(pinned_memory_size);

auto type = LogicalType(LogicalTypeId::BOOLEAN);

return_types.emplace_back(type);

names.emplace_back("Success");

return std::move(result);

TableFunctionInput& data_p,

DataChunk& output)

{

auto& data = data_p.bind_data->CastNoConst<GPUBufferInitFunctionData>();

if (data.finished) { return; }

size_t cache_size = data.cache_size;

size_t processing_size = data.processing_size;

size_t pinned_memory_size = data.pinned_memory_size;

if (pinned_memory_size == 0) { pinned_memory_size = std::max(cache_size, processing_size); }

if (!buffer_is_initialized) {

SIRIUS_LOG_DEBUG(

"GPU Buffer Manager initialized with args: Cache Size - {}, Processing Size - {}, Pinned Mem "

"Size - {}\n",

cache_size,

processing_size,

pinned_memory_size);

GPUBufferManager* gpuBufferManager =

&(GPUBufferManager::GetInstance(cache_size, processing_size, pinned_memory_size));

buffer_is_initialized = true;

} else {

SIRIUS_LOG_WARN("GPUBufferManager already initialized");

}

data.finished = true;

TableFunctionBindInput& input,

vector<LogicalType>& return_types,

vector<string>& names)

{

return_types.push_back(LogicalType::BOOLEAN);

names.push_back("ok");

return nullptr;

TableFunctionInitInput& input)

{

return make_uniq<ProfilerFunctionData>();

TableFunctionInput& data_p,

DataChunk& output)

{

auto& data = data_p.global_state->Cast<ProfilerFunctionData>();

if (data.finished) return;

cudaProfilerStart();

output.SetCardinality(1);

output.SetValue(0, 0, Value::BOOLEAN(true));

data.finished = true;

TableFunctionInput& data_p,

DataChunk& output)

{

auto& data = data_p.global_state->Cast<ProfilerFunctionData>();

if (data.finished) return;

cudaProfilerStop();

output.SetCardinality(1);

output.SetValue(0, 0, Value::BOOLEAN(true));

data.finished = true;

{

auto transaction = CatalogTransaction::GetSystemTransaction(instance);

auto& catalog = Catalog::GetSystemCatalog(instance);

TableFunction gpu_buffer_init("gpu_buffer_init",

{LogicalType::VARCHAR, LogicalType::VARCHAR},

GPUBufferInitFunction,

GPUBufferInitBind);

gpu_buffer_init.named_parameters[PINNED_MEMORY_PARAM_KEY] = LogicalType::VARCHAR;

CreateTableFunctionInfo gpu_buffer_init_info(gpu_buffer_init);

catalog.CreateTableFunction(transaction, gpu_buffer_init_info);

#ifdef SIRIUS_ENABLE_LEGACY

RegisterLegacyGPUFunctions(transaction, catalog);

#endif

TableFunction gpu_execution("gpu_execution",

{LogicalType::VARCHAR},

GPUExecutionFunction,

SiriusExtension::GPUExecutionBind);

gpu_execution.named_parameters["enable_optimizer"] = LogicalType::BOOLEAN;

CreateTableFunctionInfo gpu_execution_info(gpu_execution);

catalog.CreateTableFunction(transaction, gpu_execution_info);

// Profiler control functions for nsys --capture-range=cudaProfilerApi

TableFunction profiler_start(

"profiler_start", {}, ProfilerStartFunction, ProfilerBind, ProfilerInit);

CreateTableFunctionInfo profiler_start_info(profiler_start);

catalog.CreateTableFunction(transaction, profiler_start_info);

TableFunction profiler_stop(

"profiler_stop", {}, ProfilerStopFunction, ProfilerBind, ProfilerInit);

CreateTableFunctionInfo profiler_stop_info(profiler_stop);

catalog.CreateTableFunction(transaction, profiler_stop_info);

{

Config::USE_PIN_MEM_FOR_CPU_PROCESSING = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config USE_PIN_MEM_FOR_CPU_PROCESSING to {}",

Config::USE_PIN_MEM_FOR_CPU_PROCESSING);

{

Config::USE_PIN_MEM_FOR_CACHING = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config USE_PIN_MEM_FOR_CACHING to {}", Config::USE_PIN_MEM_FOR_CACHING);

{

Config::USE_CUDF_EXPR = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config USE_CUDF_EXPR to {}", Config::USE_CUDF_EXPR);

{

auto value = StringValue::Get(parameter);

if (value != "materialize" && value != "ast_interpret" && value != "ast_jit") {

throw InvalidInputException(

"Invalid expression_executor_strategy '{}'. Valid values: materialize, ast_interpret, "

"ast_jit",

value);

}

Config::EXPRESSION_EXECUTOR_STRATEGY = std::move(value);

SIRIUS_LOG_DEBUG("Updated config EXPRESSION_EXECUTOR_STRATEGY to {}",

Config::EXPRESSION_EXECUTOR_STRATEGY);

{

Config::USE_CUSTOM_TOP_N = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config USE_CUSTOM_TOP_N to {}", Config::USE_CUSTOM_TOP_N);

{

Config::USE_OPT_TABLE_SCAN = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config USE_OPT_TABLE_SCAN to {}", Config::USE_OPT_TABLE_SCAN);

{

Config::OPT_TABLE_SCAN_NUM_CUDA_STREAMS = IntegerValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config OPT_TABLE_SCAN_NUM_CUDA_STREAMS to {}",

Config::OPT_TABLE_SCAN_NUM_CUDA_STREAMS);

{

Config::OPT_TABLE_SCAN_CUDA_MEMCPY_SIZE = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config OPT_TABLE_SCAN_CUDA_MEMCPY_SIZE to {}",

Config::OPT_TABLE_SCAN_CUDA_MEMCPY_SIZE);

{

Config::PRINT_GPU_TABLE_MAX_ROWS = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config PRINT_GPU_TABLE_MAX_ROWS to {}",

Config::PRINT_GPU_TABLE_MAX_ROWS);

{

Config::ENABLE_FALLBACK_CHECK = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config ENABLE_FALLBACK_CHECK to {}", Config::ENABLE_FALLBACK_CHECK);

{

Config::ENABLE_DUCKDB_FALLBACK = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config ENABLE_DUCKDB_FALLBACK to {}", Config::ENABLE_DUCKDB_FALLBACK);

{

Config::ENABLE_REGEX_JIT_IMPL = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config ENABLE_REGEX_JIT_IMPL to {}", Config::ENABLE_REGEX_JIT_IMPL);

{

Config::MODIFIED_PIPELINE = BooleanValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config MODIFIED_PIPELINE to {}", Config::MODIFIED_PIPELINE);

{

auto sirius_ctx = context.registered_state->Get<duckdb::SiriusContext>("sirius_state");

if (sirius_ctx == nullptr) {

SIRIUS_LOG_DEBUG("SiriusContext not available; cache_scan_level SET ignored");

return;

}

auto level_str = StringValue::Get(parameter);

sirius::op::scan::cache_level level;

if (!sirius::op::scan::string_to_enum(level_str, level)) {

throw InvalidInputException(

"Invalid cache_scan_level '{}'. Valid values: none, table_gpu, table_host, parquet",

level_str);

}

auto& cfg = sirius_ctx->get_config();

cfg.set_cache_level(level);

SIRIUS_LOG_DEBUG("Updated config cache_scan_level to {}", level_str);

{

auto* params = get_operator_params(context);

if (!params) { return; }

params->scan_task_batch_size = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config SCAN_TASK_BATCH_SIZE to {}", params->scan_task_batch_size);

{

auto* params = get_operator_params(context);

if (!params) { return; }

params->default_scan_task_varchar_size = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config DEFAULT_SCAN_TASK_VARCHAR_SIZE to {}",

params->default_scan_task_varchar_size);

{

auto* params = get_operator_params(context);

if (!params) { return; }

params->max_sort_partition_bytes = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config MAX_SORT_PARTITION_BYTES to {}",

params->max_sort_partition_bytes);

{

auto* params = get_operator_params(context);

if (!params) { return; }

params->hash_partition_bytes = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config HASH_PARTITION_BYTES to {}", params->hash_partition_bytes);

{

auto* params = get_operator_params(context);

if (!params) { return; }

params->concat_batch_bytes = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config CONCAT_BATCH_BYTES to {}", params->concat_batch_bytes);

{

Config::LOG_LEVEL = StringValue::Get(parameter);

SetGlobalLogLevel(Config::LOG_LEVEL);

SIRIUS_LOG_DEBUG("Updated config LOG_LEVEL to {}", Config::LOG_LEVEL);

{

Config::LOG_DIR = StringValue::Get(parameter);

InitGlobalLogger(Config::LOG_LEVEL, Config::LOG_DIR, Config::LOG_FLUSH_SECONDS);

SIRIUS_LOG_DEBUG("Updated config LOG_DIR to {}", Config::LOG_DIR);

{

Config::LOG_FLUSH_SECONDS = IntegerValue::Get(parameter);

SetGlobalLogFlush(Config::LOG_FLUSH_SECONDS);

SIRIUS_LOG_DEBUG("Updated config LOG_FLUSH_SECONDS to {}", Config::LOG_FLUSH_SECONDS);

{

auto* params = get_operator_params(context);

if (!params) { return; }

params->max_build_hash_table_bytes = UBigIntValue::Get(parameter);

SIRIUS_LOG_DEBUG("Updated config MAX_BUILD_HASH_TABLE_BYTES to {}",

params->max_build_hash_table_bytes);

{

// Add in config option for gpu buffer manager

config.AddExtensionOption("use_pin_memory",

"Whether or not the buffer manager is initialized with pinned memory",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::USE_PIN_MEM_FOR_CPU_PROCESSING),

SetUsePinMemory);

config.AddExtensionOption(

"use_pin_memory_for_caching",

"Whether or not the cache buffer is allocated with pinned host memory instead of GPU memory",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::USE_PIN_MEM_FOR_CACHING),

SetUsePinMemoryForCaching);

// Add in config option for expression executor

config.AddExtensionOption("use_cudf_expr",

"Whether or not cudf is used to evaluate expressions",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::USE_CUDF_EXPR),

SetUseCudfExpr);

config.AddExtensionOption(

"expression_executor_strategy",

"Strategy for the experimental gpu_expression_executor: 'materialize', 'ast_interpret', or "

"'ast_jit'",

LogicalType::VARCHAR,

Value(Config::EXPRESSION_EXECUTOR_STRATEGY),

SetExpressionExecutorStrategy);

// Add in config option for top-N

config.AddExtensionOption("use_custom_top_n",

"Whether or not custom kernel is used to evalaute top n",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::USE_CUSTOM_TOP_N),

SetUseCustomTopN);

// Add in config options for custom table scan

config.AddExtensionOption("use_opt_table_scan",

"Whether or not the optional table scan is used",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::USE_OPT_TABLE_SCAN),

SetUseOptTableScan);

config.AddExtensionOption("opt_table_scan_num_streams",

"The number of cuda streams to use in the optional table scan",

LogicalType::INTEGER,

Value::INTEGER(Config::OPT_TABLE_SCAN_NUM_CUDA_STREAMS),

SetOptTableScanNumStreams);

config.AddExtensionOption("opt_table_scan_memcpy_size",

"The memcpy size (in bytes) used by the optional table scan",

LogicalType::UBIGINT,

Value::UBIGINT(Config::OPT_TABLE_SCAN_CUDA_MEMCPY_SIZE),

SetOptTableScanMemcpySize);

// Add in config options for printing gpu table

config.AddExtensionOption("print_gpu_table_max_rows",

"Maximal amount of rows to render when printing gpu table",

LogicalType::UBIGINT,

Value::UBIGINT(Config::PRINT_GPU_TABLE_MAX_ROWS),

SetPrintGPUTableMaxRows);

// Add in config options for duckdb fallback checking

config.AddExtensionOption("enable_fallback_check",

"Whether to enable fallback checking",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::ENABLE_FALLBACK_CHECK),

SetEnableFallbackCheck);

config.AddExtensionOption(

"enable_duckdb_fallback",

"Whether to enable fallback to duckdb execution after an error is detected",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::ENABLE_DUCKDB_FALLBACK),

SetEnableDuckdbFallback);

// Add in config options for special JIT implementation for regex

config.AddExtensionOption(

"enable_regex_jit_impl",

"Whether to use special JIT implementation for particular regex evaluation",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::ENABLE_REGEX_JIT_IMPL),

SetEnableRegexJitImpl);

// Add in config options for modified pipeline

config.AddExtensionOption("modified_pipeline",

"Whether to use modified pipeline for GPU execution",

LogicalType::BOOLEAN,

Value::BOOLEAN(Config::MODIFIED_PIPELINE),

SetModifiedPipeline);

// Add in config options for duckdb scan task

// Default batch size

config.AddExtensionOption("scan_task_batch_size",

"The default batch size for a duckdb scan task",

LogicalType::UBIGINT,

Value::UBIGINT(sirius::operator_params{}.scan_task_batch_size),

SetDefaultScanTaskBatchSize);

// Default varchar size for estimating rows per batch

config.AddExtensionOption(

"default_scan_task_varchar_size",

"The default varchar size for estimating rows per batch in a duckdb scan task",

LogicalType::UBIGINT,

Value::UBIGINT(sirius::operator_params{}.default_scan_task_varchar_size),

SetDefaultScanTaskVarcharSize);

// Add in config option for sort partition size

config.AddExtensionOption("max_sort_partition_bytes",

"Maximum bytes per sort partition (0 = auto based on 33% GPU memory)",

LogicalType::UBIGINT,

Value::UBIGINT(sirius::operator_params{}.max_sort_partition_bytes),

SetMaxSortPartitionBytes);