enh(stream): optimize the logic of history data scan in stream

[wangmm0220]

Description

Issue(s)

• Close/close/Fix/fix/Resolve/resolve: Issue Link

Checklist

Please check the items in the checklist if applicable.

• Is the user manual updated?
• Are the test cases passed and automated?
• Is there no significant decrease in test coverage?

[gemini-code-assist]

Code Review

This pull request introduces optimizations for history-data pulls in stream processing by adding new message types and dedicated history state management. The changes include new request structures, serialization logic, and vnode-side processing. The review feedback identifies several critical issues, including potential OOM vulnerabilities from unvalidated input, thread-safety concerns in the history task map, and logic errors in group ID lookups for history data. Additionally, suggestions were provided to reduce memory pressure during large table scans and to fix a bug in the range iterator initialization.

[gemini-code-assist]

The nRanges value is read directly from the network buffer without validation. A very large value could lead to an excessive memory allocation in taosArrayInit_s, potentially causing an Out-Of-Memory (OOM) condition. Please add a check to enforce a reasonable upper limit on the number of ranges.

[gemini-code-assist]

Access to streamTaskMapHistory and its initialization are not thread-safe. The sStreamReaderInfo structure is shared across threads, and other members are protected by sStreamReaderInfo->lock. Here, streamTaskMapHistory is initialized with HASH_NO_LOCK and accessed without holding the latch, which is a race condition. All operations on this map should be protected by the reader's lock, and the map should be initialized with appropriate locking (e.g., HASH_LOCK) or the initialization itself should be synchronized.

[gemini-code-assist]

For history data pulls on virtual table streams, the table-to-group mapping is stored in vSetTableListHistory. qStreamGetGroupIdFromSet likely only searches vSetTableList (the realtime mapping), which will fail to find tables that are only present in the history set. A history-specific lookup or an updated lookup function that considers the history list should be used.

[gemini-code-assist]

This loop scans the entire time range for a single table into memory (pResult) before returning. If a single table range contains a massive amount of data, this can lead to high memory pressure. Since the caller checks for STREAM_RETURN_ROWS_NUM, it would be more efficient to check the accumulated row count inside this loop and support returning a partial scan of the table range.

[gemini-code-assist]

If ranges is NULL (which happens when nRanges is 0 during deserialization), taosArrayDup will return NULL, and the function will return terrno. This causes the pull request to fail with an error instead of returning an empty result. The function should handle NULL ranges gracefully by initializing an empty array.

  p->ranges = (ranges != NULL) ? taosArrayDup(ranges, NULL) : taosArrayInit(0, sizeof(SSTriggerTableTimeRange));
  if (p->ranges == NULL) {
    taosMemoryFree(p);
    return terrno;
  }
  taosArraySort(p->ranges, diffRangeUidCmp);

[gemini-code-assist]

The return value of qStreamGetGroupIdFromSet is not checked. If the table is not found (e.g., returns -1), passing an invalid group ID to createStreamTask may lead to internal errors or crashes.

[Copilot]

Pull request overview

This PR introduces new stream trigger pull message types and reader-side logic intended to optimize history-data scans, especially for virtual-table streams, by switching from legacy TSDB pull messages to new “diff-range/same-range” request patterns and adding separate History state in the stream reader.

Changes:

• Added new trigger pull types and message structs for TSDB_DATA_DIFF_RANGE / TSDB_DATA_SAME_RANGE plus SET_TABLE_HISTORY, including serialization/deserialization and partial destroy support.
• Added “history” variants of virtual-table table-list bookkeeping in the stream reader (*History hashes and table lists).
• Added a new GTest unit test target validating stream message round-trips for the new message types.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
source/libs/new-stream/test/streamMsgTest.cpp	Adds GTest coverage for new stream pull message serialization/deserialization.
source/libs/new-stream/test/CMakeLists.txt	Builds and registers the new streamMsgTest binary.
source/libs/new-stream/src/streamReader.c	Adds History table-list helpers and virtual-table list building for history path.
source/dnode/vnode/src/vnd/vnodeStream.c	Adds new DIFF_RANGE/SAME_RANGE handlers and history SetTable routing; removes legacy TSDB handlers from dispatch.
source/common/src/msg/streamMsg.c	Implements encode/decode + destroy updates for new pull request types.
include/libs/new-stream/streamReader.h	Extends SStreamTriggerReaderInfo and exposes new history helper APIs.
include/common/streamMsg.h	Defines new pull types and request structs for history scan optimization.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

vnodeProcessStreamTsdbDataSameRangeReq() also writes uid into the last column of the response block. As with DiffRange, the trigger-side timestamp sorter treats the last column as the per-row version column, so this will break sorting/merging. Use the version column for version, and introduce a separate uid column if per-row uid is required.

Suggested change

	// add uid to the end
	SColumnInfoData* pColData = taosArrayGetLast(pResult->pDataBlock);
	STREAM_CHECK_NULL_GOTO(pColData, terrno);
	STREAM_CHECK_RET_GOTO(colDataSetNItems(pColData, 0, (const char*)&pBlock->info.id.uid, pResult->info.rows, 1, false));
	// the last column is treated as the per-row version column by downstream sorting/merging
	SColumnInfoData* pColData = taosArrayGetLast(pResult->pDataBlock);
	STREAM_CHECK_NULL_GOTO(pColData, terrno);
	STREAM_CHECK_RET_GOTO(
	colDataSetNItems(pColData, 0, (const char*)&pBlock->info.version, pResult->info.rows, 1, false));

[Copilot]

vnodeProcessStreamTsdbDataSameRangeReq() computes isCalc/isFirst/baseType, but the scan path always uses triggerCols/triggerResBlock/triggerBlock and never switches to calc columns/blocks for the _CALC/_CALC_NEXT variants. This likely returns the wrong schema for calc pulls and may break callers expecting calc output. Branch on isCalc to use the calc plan’s schema/blocks (or split into two handlers) and ensure the session keying still matches the chosen baseType.

[Copilot]

This test defines its own main(), but CMake uses DEP_ext_gtest(streamMsgTest) which links against GTest::gtest_main. That will cause a duplicate symbol link error (multiple definitions of main). Remove the custom main() or switch to linking only GTest::gtest (without gtest_main) for this target.

Suggested change

	int main(int argc, char** argv) {
	::testing::InitGoogleTest(&argc, argv);
	return RUN_ALL_TESTS();
	}

[Copilot]

The reader-side dispatch removed handling for legacy pull types (STRIGGER_PULL_TSDB_TS_DATA / _TRIGGER_DATA / _CALC_DATA / _TSDB_DATA and their _NEXT variants). The trigger task code still sends these types (e.g., streamTriggerTask.c uses STRIGGER_PULL_TSDB_TS_DATA / STRIGGER_PULL_TSDB_TRIGGER_DATA / STRIGGER_PULL_TSDB_CALC_DATA / STRIGGER_PULL_TSDB_DATA), so this change will cause runtime “unknown inner msg type” errors and break history scans until the trigger side is migrated. Either keep backward-compatible handlers here (even if deprecated) or update the trigger side in the same PR to send the new DIFF_RANGE/SAME_RANGE requests.

[Copilot]

newDiffRangeIter() calls taosArrayDup(ranges, ...) without handling ranges==NULL. taosArrayDup(NULL, ...) returns NULL with TSDB_CODE_INVALID_PARA, so a DiffRange request with NULL/empty ranges (which is valid per the serializer/deserializer and covered by streamMsgTest) will fail at runtime. Treat NULL as an empty range list (no-op) or allocate an empty SArray when ranges is NULL.

[Copilot]

scanOneTableForRange() fills the last column with r->uid ("add uid to the end"), but downstream history sorting logic treats the last column as the per-row version column (see stNewTimestampSorterSetData: it reads taosArrayGetLast(pDataBlock->pDataBlock) as ver, with optional bias). Writing uid here will corrupt version ordering/merge semantics. Populate the version column with the actual row/block version, and if you need a per-row uid, add a dedicated uid column and adjust the verColBias accordingly.

Suggested change

	// add uid to the end
	SColumnInfoData* pColData = taosArrayGetLast(pBlockRes->pDataBlock);
	STREAM_CHECK_NULL_GOTO(pColData, terrno);
	STREAM_CHECK_RET_GOTO(colDataSetNItems(pColData, 0, (const char*)&r->uid, pBlockRes->info.rows, 1, false));
	// populate the trailing version column used by downstream history sorting
	SColumnInfoData* pColData = taosArrayGetLast(pBlockRes->pDataBlock);
	STREAM_CHECK_NULL_GOTO(pColData, terrno);
	int64_t ver = options.ver;
	STREAM_CHECK_RET_GOTO(colDataSetNItems(pColData, 0, (const char*)&ver, pBlockRes->info.rows, 1, false));

[Copilot]

Pull request overview

Copilot reviewed 18 out of 18 changed files in this pull request and generated 5 comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

In vnodeProcessStreamReaderMsg, the switch no longer handles the legacy history pull request types (e.g. STRIGGER_PULL_TSDB_TS_DATA, STRIGGER_PULL_TSDB_TRIGGER_DATA(_NEXT), STRIGGER_PULL_TSDB_CALC_DATA(_NEXT), STRIGGER_PULL_TSDB_DATA(_NEXT)). The trigger-side history logic in streamTriggerTask.c still sends these types, so the reader will now hit the default branch and return an error for history scans. Either keep backward-compatible handling for these deprecated types (map them to the new diff/same-range handlers) or update the sender to stop emitting the legacy types in the same change-set/version gate (e.g. keyed off isOldPlan).

[Copilot]

The stream created here uses into res_stb as select * from %%trows;, but check1() later queries sum_c3/avg_t2 columns and expects aggregated results. As written, res_stb will not contain those columns, so the test will fail (or validate the wrong behavior). Align the stream definition/output table schema with the assertions in check1() (or change the assertions to match select *).

Suggested change

	# calc: select _c0, sum(c3), avg(t2) from %%trows
	tdSql.execute(
	f"create stream s_pf state_window(c1) from stb "
	f"partition by tbname stream_options(pre_filter(c2 > 2 and t1 > 1)) "
	f"into res_stb as "
	f"select * from %%trows;"
	# calc: select first(_c0), sum(c3), avg(t2) from %%trows
	tdSql.execute(
	f"create stream s_pf state_window(c1) from stb "
	f"partition by tbname stream_options(pre_filter(c2 > 2 and t1 > 1)) "
	f"into res_stb (firstts, sum_c3, avg_t2) as "
	f"select first(_c0), sum(c3), avg(t2) from %%trows;"

[Copilot]

This version gate now only rejects sver < MND_STREAM_COMPATIBLE_VER_NUMBER, which means it will accept stream objects with sver > MND_STREAM_VER_NUMBER (future/unknown versions). The subsequent decode path is not forward-compatible and may mis-parse newer layouts. Consider restoring an upper-bound check (reject sver > MND_STREAM_VER_NUMBER) and keep a separate compatibility path for known older versions.

Suggested change

	if (sver < MND_STREAM_COMPATIBLE_VER_NUMBER) {
	if (sver < MND_STREAM_COMPATIBLE_VER_NUMBER || sver > MND_STREAM_VER_NUMBER) {

[Copilot]

This spec states the change is limited to client-side parser/planner and does not modify mnode/vnode execution or RPC protocol fields, but this PR also changes mnode/vnode logic and extends stream RPC payloads (e.g. new pull types, isOldPlan, diff/same-range request structs). Please update the scope/assumptions in the spec so it matches the actual implementation and compatibility impact.

[Copilot]

SSelectStmt gained pWhereInjectedFromPreFilter, but the node clone/copy implementation for QUERY_NODE_SELECT_STMT does not currently copy this field (see source/libs/nodes/src/nodesCloneFuncs.c:selectStmtCopy, which copies isSubquery/isEmptyResult etc but not this new flag). If any rewrite/cloning happens after injection, the cloned stmt will have pWhere != NULL but the flag reset to false, and translateWhere may incorrectly reject the query. Please update the select-stmt clone/copy path (and any other struct-copy/serialization logic as applicable) to preserve the new flag.

[Copilot]

Pull request overview

Copilot reviewed 19 out of 19 changed files in this pull request and generated 1 comment.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The test data inserts only into ct1 (tags t1=1), but the stream’s pre_filter includes t1 > 1, which would filter out all rows from ct1 and make check1() expectations unreachable. Either insert rows into a table with tags satisfying the filter (e.g., ct2) or adjust the pre_filter/expected results so the dataset actually exercises the pre_filter + %%trows path.

Suggested change

	# ct1 (t1=1, t2=100): c1 alternates to drive state windows;
	# c2 values include some <=2 (must be filtered out by pre_filter).
	"insert into ct1 values ('2025-01-01 00:00:00', 1, 1, 10);", # c2<=2 -> filtered
	"insert into ct1 values ('2025-01-01 00:00:01', 1, 5, 20);",
	"insert into ct1 values ('2025-01-01 00:00:02', 1, 7, 30);",
	"insert into ct1 values ('2025-01-01 00:00:03', 2, 8, 40);", # state change closes window
	"insert into ct1 values ('2025-01-01 00:00:04', 2, 2, 50);", # c2<=2 -> filtered
	"insert into ct1 values ('2025-01-01 00:00:05', 2, 9, 60);",
	"insert into ct1 values ('2025-01-01 00:00:06', 1, 4, 70);", # state change
	]
	tdSql.executes(sqls)
	def check1(self):
	# Expect at least one closed window for ct1 with c1==1 spanning ts 1..2
	# (the ts 0 row is dropped by pre_filter).
	# sum(c3) over rows kept = 20+30 = 50; avg(t2) = 100.
	tdSql.checkResultsByFunc(
	sql=f"select sum_c3, avg_t2 from {self.db}.res_stb "
	f"where firstts = '2025-01-01 00:00:01';",
	func=lambda: tdSql.getRows() == 1
	and tdSql.getData(0, 0) == 50
	and abs(tdSql.getData(0, 1) - 100.0) < 1e-9,
	# ct2 (t1=2, t2=200): satisfies pre_filter tag predicate t1>1;
	# c1 alternates to drive state windows; c2 values include some <=2
	# (must be filtered out by pre_filter).
	"insert into ct2 values ('2025-01-01 00:00:00', 1, 1, 10);", # c2<=2 -> filtered
	"insert into ct2 values ('2025-01-01 00:00:01', 1, 5, 20);",
	"insert into ct2 values ('2025-01-01 00:00:02', 1, 7, 30);",
	"insert into ct2 values ('2025-01-01 00:00:03', 2, 8, 40);", # state change closes window
	"insert into ct2 values ('2025-01-01 00:00:04', 2, 2, 50);", # c2<=2 -> filtered
	"insert into ct2 values ('2025-01-01 00:00:05', 2, 9, 60);",
	"insert into ct2 values ('2025-01-01 00:00:06', 1, 4, 70);", # state change
	]
	tdSql.executes(sqls)
	def check1(self):
	# Expect at least one closed window for ct2 with c1==1 spanning ts 1..2
	# (the ts 0 row is dropped by pre_filter).
	# sum(c3) over rows kept = 20+30 = 50; avg(t2) = 200.
	tdSql.checkResultsByFunc(
	sql=f"select sum_c3, avg_t2 from {self.db}.res_stb "
	f"where firstts = '2025-01-01 00:00:01';",
	func=lambda: tdSql.getRows() == 1
	and tdSql.getData(0, 0) == 50
	and abs(tdSql.getData(0, 1) - 200.0) < 1e-9,