Repository: influxdata/influxdb
File: influxdb3/src/commands/serve.rs (Rust)

Code before:
        )
        .with_jaeger_debug_name(config.tracing_config.traces_jaeger_debug_name);

    let persister = Arc::new(Persister::new(
        Arc::clone(&object_store),
        config.node_identifier_prefix.as_str(),
        Arc::clone(&time_provider) as _,
    ));
    let wal_config = WalConfig {
        gen1_duration: config.gen1_duration,
        max_write_buffer_size: config.wal_max_write_buffer_size,
        flush_interval: config.wal_flush_interval.into(),
        snapshot_size: config.wal_snapshot_size,
    };

    let catalog = Catalog::new_with_shutdown(
        config.node_identifier_prefix.as_str(),
        Arc::clone(&object_store),
        Arc::<SystemProvider>::clone(&time_provider),
        Arc::clone(&metrics),
        shutdown_manager.register(),
    )
    .await?;
    info!(catalog_uuid = ?catalog.catalog_uuid(), "catalog initialized");

    let _ = catalog
        .register_node(
            &config.node_identifier_prefix,
            num_cpus as u64,
            vec![influxdb3_catalog::log::NodeMode::Core],
        )
        .await?;
    let node_def = catalog
        .node(&config.node_identifier_prefix)
        .expect("node should be registered in catalog");
    info!(instance_id = ?node_def.instance_id(), "catalog initialized");

    let last_cache = LastCacheProvider::new_from_catalog_with_background_eviction(
        Arc::clone(&catalog) as _,
        config.last_cache_eviction_interval.into(),
    )
    .await
    .map_err(Error::InitializeLastCache)?;

    let distinct_cache = DistinctCacheProvider::new_from_catalog_with_background_eviction(
        Arc::clone(&time_provider) as _,
        Arc::clone(&catalog),
        config.distinct_cache_eviction_interval.into(),
    )
    .await
    .map_err(Error::InitializeDistinctCache)?;

Review comment (suggestion):
Yep, these came in from `iox` in the most recent sync. They aren't used anywhere in our code yet. I opened https://github.com/influxdata/influxdb/issues/26460 to address using them.

Code after:
                .builder()
                .map_err(Error::TokioRuntime)?,
            Arc::clone(&metrics),
        ),
    ));

    // Note: using same metrics registry causes runtime panic.
    let write_path_executor = Arc::new(Executor::new_with_config_and_executor(
        ExecutorConfig {
            // should this be divided? or should this contend for threads with executor that's
            // setup for querying only
            target_query_partitions: tokio_datafusion_config.num_threads.unwrap(),
            object_stores: [&parquet_store]
                .into_iter()
                .map(|store| (store.id(), Arc::clone(store.object_store())))
                .collect(),
            metric_registry: Arc::clone(&write_path_metrics),
            // use as much memory for persistence, can this be UnboundedMemoryPool?
            mem_pool_size: usize::MAX,
            // These are new additions, just skimming through the code it does not look like we can
            // achieve the same effect as having a separate executor. It looks like it's for "all"
            // queries, it'd be nice to have a filter to say when the query matches this pattern
            // apply these limits. If that's possible maybe we could avoid creating a separate
            // executor.
            per_query_mem_pool_config: PerQueryMemoryPoolConfig::Disabled,
            heap_memory_limit: None,
        },
        DedicatedExecutor::new(
            "datafusion_write_path",
            tokio_datafusion_config
                .builder()
                .map_err(Error::TokioRuntime)?,
            Arc::clone(&write_path_metrics),
        ),
    ));

    let trace_header_parser = TraceHeaderParser::new()
        .with_jaeger_trace_context_header_name(
            config
                .tracing_config
                .traces_jaeger_trace_context_header_name,
        )
        .with_jaeger_debug_name(config.tracing_config.traces_jaeger_debug_name);

    let persister = Arc::new(Persister::new(
        Arc::clone(&object_store),
        config.node_identifier_prefix.as_str(),
        Arc::clone(&time_provider) as _,
    ));
    let wal_config = WalConfig {
        gen1_duration: config.gen1_duration,