Skip to content

[feature](cache) Add session variable to keep INSERT source scans in the normal file cache queue #66804

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
raghav-reglobe wants to merge 1 commit into
base: master
Choose a base branch
from
+19 −2
Open

[feature](cache) Add session variable to keep INSERT source scans in the normal file cache queue #66804

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -500,8 +500,15 @@ private void setForArrowFlight(CoordinatorContext coordinatorContext, PipelineDi
private void setForInsert(long jobId) {
JobProcessor jobProc = new LoadProcessor(this.coordinatorContext, jobId);
this.coordinatorContext.setJobProcessor(jobProc);
// Set this field to true to avoid data entering the normal cache LRU queue
this.coordinatorContext.queryOptions.setDisableFileCache(true);
// Set this field to true to avoid data entering the normal cache LRU queue.
// enable_file_cache_for_insert_source opts OUT of that protection: recurring
// INSERT ... SELECT / CTAS whose source working set is shared with interactive
// queries (and re-read every cycle) benefits from normal-queue retention.
boolean keepInsertSourceInCache = this.coordinatorContext.connectContext != null
&& this.coordinatorContext.connectContext.getSessionVariable().enableFileCacheForInsertSource;
if (!keepInsertSourceInCache) {
this.coordinatorContext.queryOptions.setDisableFileCache(true);
}
this.coordinatorContext.queryOptions.setNewVersionUnixTimestamp(true);
this.coordinatorContext.queryOptions.setNewVersionPercentile(true);
this.coordinatorContext.queryOptions.setNewVersionBitmapOpCount(true);
Expand Down
10 changes: 10 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,7 @@ public class SessionVariable implements Serializable, Writable {
public static final String MAX_SCANNERS_CONCURRENCY = "max_scanners_concurrency";
public static final String MAX_FILE_SCANNERS_CONCURRENCY = "max_file_scanners_concurrency";
public static final String ENABLE_FILE_SCANNER_V2 = "enable_file_scanner_v2";
public static final String ENABLE_FILE_CACHE_FOR_INSERT_SOURCE = "enable_file_cache_for_insert_source";
public static final String MIN_SCANNERS_CONCURRENCY = "min_scanners_concurrency";
public static final String MIN_FILE_SCANNERS_CONCURRENCY = "min_file_scanners_concurrency";
public static final String MIN_SCAN_SCHEDULER_CONCURRENCY = "min_scan_scheduler_concurrency";
Expand Down Expand Up @@ -1162,6 +1163,15 @@ public static double getHotValueThreshold() {
+ "FileScanNode uses FileScannerV2 for supported query scans. Enabled by default.")
public boolean enableFileScannerV2 = true;

@VarAttrDef.VarAttr(name = ENABLE_FILE_CACHE_FOR_INSERT_SOURCE, needForward = true, description = {
"开启后 INSERT ... SELECT / CTAS 的源表扫描使用普通文件缓存队列（不再强制 disposable 队列）。"
+ "适用于源扫描工作集与交互查询共享、且周期性重复读取的 ETL 场景。默认关闭（保持原有行为）",
"When enabled, source scans of INSERT ... SELECT / CTAS use the normal file cache queue "
+ "instead of being forced onto the disposable queue. Useful for recurring ETL whose "
+ "source working set is shared with interactive queries and re-read every cycle. "
+ "Disabled by default (preserves existing behavior)."})
public boolean enableFileCacheForInsertSource = false;

@VarAttrDef.VarAttr(name = LOCAL_EXCHANGE_FREE_BLOCKS_LIMIT)
public int localExchangeFreeBlocksLimit = 4;

Expand Down
Loading