[Bug] Low throughput of PostgreSQL CDC streaming job, single-core CPU bottleneck on BE write path

[gaogao110]

Search before asking

• I had searched in the issues and found no similar issues.

Version

4.1.3

What's Wrong?

When synchronizing PostgreSQL incremental data via Doris Streaming Job, the import throughput is extremely low and cannot reach business expectations.
Server overall resources are abundant: the machine has 60.7% idle CPU and plenty of free memory, no swap pressure, and IO wait is low. However, the doris_be process consistently occupies 100% single-core CPU, which becomes the strict bottleneck of CDC import.
The upstream Flink task CPU load is only 6.3%, which proves that the data source side is not the bottleneck. The performance bottleneck is completely inside the Doris BE write logic.

Top System Status
%Cpu(s): 24.3 us, 4.7 sy, 0.0 ni, 60.7 id, 6.9 wa, 0.9 hi, 2.5 si, 0.0 st
MiB Mem : 31301.9 total, 12756.9 free, 16703.9 used, 2284.1 buff/cache
MiB Swap: 0.0 total, 0.0 free, 0.0 used. 14598.0 avail Mem

PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND

3923736 root 20 0 31.1g 1.5g 256252 S 102.0 5.0 51:40.35 doris_be
3928325 root 20 0 16.5g 7.6g 19520 S 6.3 24.8 13:12.08 java

• BE single-core full load, multi-core idle, typical single-thread serialization bottleneck
• Upstream Flink Java task almost no pressure
• Memory is sufficient, no IO bottleneck

Perf Flame Graph Analysis
CPU hotspot is concentrated in the single-thread write path of FragmentMgrAsyn asynchronous thread pool:
FragmentMgrAsyn
└── doris::AsyncResultWriter::process_block
└── doris::VTabletWriter::write
└── doris::VRowDistribution::generate_rows_distribution
└── doris::VRowDistribution::__generate_rows_distribution_for_non_auto_partition
├── doris::OlapTableFinder::find_tables
└── doris::Block::dump_data
└── doris::ColumnWithTypeAndName::to_string
└── doris::IDataType::to_string
Two major CPU overheads:

1. Non-auto partition routing: For standard Range partition tables (not AUTO-RANGE), every row needs full tablet metadata lookup via find_tables, which causes heavy per-row computation.
2. Block serialization overhead: Frequent dump_data / to_string conversion consumes massive user CPU during CDC stream load writing.

What You Expected?

fix it

How to Reproduce?

To Reproduce

1. Create a standard Range partitioned table (non AUTO-RANGE)
2. Create PostgreSQL CDC streaming job to consume incremental data
3. Generate continuous PG incremental traffic
4. Observe low import throughput and 100% single-core BE CPU usage

Anything Else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Doris-Breakwater]

Breakwater-GitHub-Analysis-Slot: slot_b598caa888f8

Initial triage

Verdict: the supplied flame graph strongly indicates trace-logging overhead; it does not yet confirm a normal non-auto Range partition routing bug. In the profiled run, effective VLOG(10) logging was enabled for tablet_info (either through a matching verbose module or a global verbose level). The two large dump_data -> to_string branches in the graph match the trace statements in the 4.1.3 source exactly. Those calls are not evaluated with the default verbose configuration.

The issue currently has no labels. Suggested initial classification is performance/load plus needs-info; a BE routing bug label should wait for a clean-verbosity reproduction.

Verified from Doris 4.1.3

1. OlapTabletFinder::find_tablets() calls find_partition() once per row. For a Range-partitioned table, find_partition() uses std::map::upper_bound. This is a per-row ordered partition-key lookup, not a full scan of all tablet metadata. Its normal cost is still worth measuring, especially with many partitions, but the current profile is confounded by logging.
2. Immediately after that lookup, find_partition() has a VLOG_TRACE statement which dumps the complete input block and the partition-boundary block. The Range-key comparator also dumps both compared blocks on every comparison. These are the OlapTabletFinder and VOlapTablePartKeyComparator branches visible in the attached flame graph.
3. VLOG_TRACE is VLOG(10). Block::dump_data() formats every column and up to 100 rows by default, so enabling those statements inside a per-row lookup and its comparator can multiply full-block string conversion many times per input block. This explains the observed ColumnWithTypeAndName::to_string/IDataType::to_string CPU stacks.
4. The 4.1.3 defaults are an empty sys_log_verbose_modules and sys_log_verbose_flags_v = -1; sys_log_verbose_level = 10 does not enable tracing by itself. Doris also documents in the source that glog does not generate the message when its VLOG level is disabled. Therefore these dump_data() calls are not part of the normal default write path.
5. One busy FragmentMgrAsyn thread alone is not proof of a global serialization defect. Each AsyncResultWriter drains its own queue serially in one submitted task; aggregate concurrency depends on the number of sink instances/jobs and their distribution across BEs. The issue does not include those values. Likewise, low Flink CPU can be a consequence of sink backpressure, so it does not independently locate the original bottleneck.

The same expensive trace statements are still present on current master, so there is also a reasonable defensive code-cleanup opportunity, but that is separate from proving a default-configuration throughput regression.

Required validation

Please first repeat the same workload after restoring default BE verbosity and restarting the affected BE:

# leave sys_log_verbose_modules unset/empty
sys_log_verbose_flags_v = -1

Also provide the current effective values of sys_log_verbose_modules, sys_log_verbose_level, and sys_log_verbose_flags_v. Confirm that be.INFO no longer contains high-volume messages beginning with find row ... of, then report actual rows/s and CPU before/after and attach a new 60-second per-thread flame graph. This A/B test should be done before changing partition type or job parallelism.

If throughput is still low with trace logging disabled, the following information is needed to isolate the remaining path:

• Exact SHOW CREATE TABLE, total partition count, key model, distribution columns/bucket count, indexes, and the number/specification of BEs.
• Complete Streaming Job/Flink sink definition and connector build, source and sink parallelism, batch row/byte size, flush interval, and actual versus expected rows/s.
• The relevant load/execution profile and job metrics showing sink instance count, input block sizes, queue/backpressure time, filtered rows, and per-BE row distribution.
• A clean-verbosity CPU profile (SVG plus folded stacks or perf report) and the matching BE log interval, including any no partition for this tuple or immutable-partition filtering messages.
• Confirmation that all FE and BE nodes run the same exact 4.1.3 build/commit.

Maintainer next steps

1. Treat trace verbosity as the leading explanation for this capture and request the A/B result above; do not infer a normal-path partition-routing regression from the current graph.
2. Independently consider removing/rate-limiting the full-block VLOG_TRACE calls in VOlapTablePartKeyComparator and find_partition(), or logging only the current partition key row. Trace mode should not repeatedly stringify whole blocks from a hot comparator.
3. Only if the clean profile remains dominated by compare_at/upper_bound rather than dump_data, benchmark routing versus partition count and block size and then evaluate a vectorized or cached Range-routing improvement.
4. Evaluate sink parallelism separately from routing cost using the job profile; one writer task being single-threaded is expected, while insufficient writer-instance parallelism may be a separate tuning or implementation issue.

No code was changed and no runtime reproduction was performed for this initial triage.

[gaogao110]

thank you repair @Doris-Breakwater

sys_log_custom_date_time_format=%Y-%m-%d %H:%M:%S
sys_log_custom_date_time_ms_format=,{:03d}
sys_log_dir=
sys_log_enable_custom_date_time_format=0
sys_log_level=WARNING
sys_log_roll_mode=SIZE-MB-50
sys_log_roll_num=10
sys_log_verbose_flags_v=-1
sys_log_verbose_level=10
sys_log_verbose_modules=*

[gaogao110]

CREATE TABLE xxxx (
id bigint NOT NULL COMMENT "ID",
a bigint NOT NULL COMMENT "xxxx ",
b bigint NOT NULL COMMENT "xxxx",
c bigint NULL COMMENT "xxxx",
d bigint NULL COMMENT "xxxx",
c1 bigint NULL COMMENT "xxxx",
c2 text NOT NULL COMMENT "xxxx",
c3 int NULL COMMENT "xxxx",
c4 int NOT NULL COMMENT "xxxx",
c4 int NOT NULL COMMENT "xxxx",
c4 int NOT NULL COMMENT "xxxx",
c4 int NOT NULL COMMENT "xxxxx",
c4 datetime(6) NULL COMMENT "xxxx",
c4 datetime(6) NULL COMMENT "xxxx",
c4 bigint NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 bigint NULL COMMENT "xxxx",
c4 bigint NULL COMMENT "xxxx",
c4 bigint NULL COMMENT "xxxx",
c4 text NULL COMMENT "操作设备",
c4 int NULL COMMENT " xxxx",
c4 text NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 datetime(6) NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 datetime(6) NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 bigint NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 bigint NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 bigint NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 text NULL COMMENT "",
c4 text NULL COMMENT "",
c4 int NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 text NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 date NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 int NULL COMMENT "xxxx",
c4 datetime(6) NULL COMMENT "xxxx"
) ENGINE=OLAP
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS AUTO
PROPERTIES (
"replication_allocation" = "tag.location.default: 1",
"min_load_replica_num" = "-1",
"is_being_synced" = "false",
"storage_medium" = "hdd",
"storage_format" = "V2",
"inverted_index_storage_format" = "V3",
"enable_unique_key_merge_on_write" = "true",
"light_schema_change" = "true",
"disable_auto_compaction" = "false",
"group_commit_interval_ms" = "10000",
"group_commit_data_bytes" = "134217728",
"enable_mow_light_delete" = "false"
);

[gaogao110]

CREATE JOB doris_postgres_job_xxx
ON STREAMING
FROM POSTGRES (
"jdbc_url" = "jdbc:postgresql://xxx:5432/xxxx",
"driver_url" = "postgresql-42.7.8.jar",
"driver_class" = "org.postgresql.Driver",
"user" = "xxxxx",
"password" = "xxxxx",
"database" = "xxxxx",
"schema" = "public",
"include_tables" = "cm_xxxx",
"offset" = "initial",
"snapshot_split_size" = "8096",
"snapshot_parallelism" = "1",
"publication_name"="dbz_publication"
)
TO DATABASE k12cloud_consume (
"table.create.properties.replication_num" = "1" -- 单BE部署时需要设置为1
)

[gaogao110]

Found the problem. It's caused by the following parameter!
Abnormal:
sys_log_verbose_modules=*
Normal:
sys_log_verbose_modules=