[Caideyipi]

Description

Table subscriptions can select databases, tables, and output columns, but they cannot filter individual rows by TAG value. This PR adds the tag-filter topic attribute for table-model subscriptions. Its default value is true, so existing topics keep their current behavior.

Filter semantics

• Parse a restricted SQL predicate over TAG columns. The supported forms are AND, OR, NOT, =, !=, IN, LIKE, REGEXP, and IS NULL, evaluated with SQL three-valued logic.
• CREATE/ALTER validates the topic attributes and filter expression syntax. Each DataNode binds referenced identifiers against its current table schema when it refreshes the matcher. Missing schema, stale bindings, unknown fields, and non-TAG fields fail closed instead of leaking unfiltered rows.
• Filter Tablet rows while preserving row order, data types, timestamps, and null bitmaps. TAG filtering runs before column-filter projection.
• Preserve quotes when formatting CREATE/ALTER TOPIC SQL and recreating topics: double each inner " as "" and each outer SQL ' as ''. The Subscription Session API accepts the raw expression.

Delivery paths and configuration changes

• Apply the filter to Record/Tablet and TsFile delivery, including normal subscription queues, consensus subscription queues, and Tree View projection.
• Snapshot filter configuration per event batch so ALTER TOPIC takes effect on subsequent batches without changing the meaning of an in-flight batch.
• Rebind matchers when table schema versions change, and clean up temporary TsFile rewrite resources on empty output and failure paths.
• Append TagFilterStatus and TagFilterMessage to table-model SHOW TOPICS. The status reports MATCH_ALL, MATCH_NONE, ACTIVE, or ERROR for the DataNode executing the statement; the message contains the local matcher failure when the status is ERROR. Tree-model output remains unchanged.

Validation

• Spotless and Checkstyle: passed for the changed modules.
• Quote handling regression: 17 tests passed (TagFilterParserTest, TopicStatementTest, and ColumnFilterParserTest), covering single/consecutive/mixed quotes, CREATE/ALTER formatting, and SHOW CREATE round trips.
• Focused DataNode tests: 14 tests passed (ShowTopicsTaskTest, TabletTagFilterTest, and SubscriptionTopicAgentTest).
• TopicConfigTest: 10 tests passed.
• Chinese-locale DataNode test compilation: passed (mvn test-compile -pl iotdb-core/datanode -P with-zh-locale -DskipTests).
• git diff --check: passed.

Performance comparison

Benchmarked commit 5e9f4c07cd7 with the subscription feature gate enabled in the test build; the TAG filtering implementation was unchanged. iot-benchmark generated a fixed table with 1,000,000 rows, 200 devices, 10 random DOUBLE fields, and one uniformly distributed TAG (tag_0, 10 values). All columns were retained.

Snapshot subscriptions; one warm-up and three measured runs per case, with shuffled order. The table reports medians. Consumption time runs from the start of subscribe() until all expected field cells are delivered and the last data batch is acknowledged. Topic creation, the later automatic snapshot-termination notification, and subsequent TsFile content verification are excluded. Input throughput uses all 1,000,000 source rows for every case.

Delivery	TAG predicate	Delivered rows	Consumption median [min–max] (s)	Time vs. no filter	Input throughput (M rows/s)	DataNode CPU (s)
Tablet/Record	Unset (default true)	1,000,000	4.056 [3.633–4.511]	baseline	0.247	5.63
Tablet/Record	tag_0 IS NOT NULL (100%)	1,000,000	5.073 [4.489–5.125]	+25.1%	0.197	6.47
Tablet/Record	tag_0 IN ("value_0", "value_1", "value_2", "value_3", "value_4") (50%)	500,000	3.244 [3.125–3.400]	-20.0%	0.308	3.98
Tablet/Record	tag_0 = "value_0" (10%)	100,000	2.378 [2.058–2.465]	-41.4%	0.420	2.57
TsFile	Unset (default true)	1,000,000	0.887 [0.489–0.988]	baseline	1.127	0.09
TsFile	tag_0 IS NOT NULL (100%)	1,000,000	92.360 [92.098–92.664]	104.1×	0.011	79.69
TsFile	tag_0 IN ("value_0", "value_1", "value_2", "value_3", "value_4") (50%)	500,000	47.003 [46.898–47.107]	53.0×	0.021	39.65
TsFile	tag_0 = "value_0" (10%)	100,000	10.480 [10.200–10.775]	11.8×	0.095	8.55

TsFile TAG predicate	Delivered files	Delivered bytes (MiB)
Unset (default true)	1	1.775
tag_0 IS NOT NULL (100%)	900	2.995
tag_0 IN ("value_0", "value_1", "value_2", "value_3", "value_4") (50%)	452	1.486
tag_0 = "value_0" (10%)	94	0.310

All 32 runs (including warm-ups) passed TAG-value, per-field row completeness, and duplicate-cell verification. Expected outputs were 1,000,000, 1,000,000, 500,000, and 100,000 logical rows with all 10 fields. Column-split Tablet/TsFile payloads were verified by device, timestamp, and field. TsFile contents were read after the measured delivery interval.

• Tablet/Record: the 100%-matching predicate increased consumption time by 25.1%. Selecting 50% and 10% of rows reduced total time by 20.0% and 41.4%, respectively.
• TsFile: nontrivial TAG predicates require parsing/filtering/rebuilding instead of forwarding the original file. The 100%-matching predicate produced 900 files and took 104.1× the baseline time. Even the 10%-matching predicate produced 94 files and took 11.8× the baseline. The resulting file fragmentation and delivery overhead are a substantial optimization target.
• These results describe this flushed snapshot workload and do not measure incremental/consensus subscription throughput.

Single-device matching (2026-10-20)

Added device_id = "d_0", selecting exactly one of the 200 devices: 5,000 logical rows and 50,000 non-null field cells (0.5% of the source rows), with all 10 fields retained.

This additional experiment uses the same test build at commit 5e9f4c07cd7, with the subscription feature gate enabled, and a copy of the original flushed data verified by SHA-256. It does not measure the current PR HEAD. All five selections were rerun for both formats, with one warm-up and three measured runs per case. Timing and input-throughput definitions are the same as above; the no-filter values below come from this rerun, independently of the earlier table.

Environment: Intel Core i7-11700 (8 cores / 16 threads), 31 GiB RAM, Java 17.0.9; one ConfigNode and one DataNode with 1 GiB and 6 GiB heaps, respectively; consumer heap 512 MiB–2 GiB.

Delivery	TAG predicate	Delivered rows	Consumption median [min–max] (s)	No-filter median [min–max] in this rerun (s)	Input throughput (M rows/s)	DataNode CPU (s)
Tablet/Record	device_id = "d_0"	5,000	0.679 [0.484–0.874]	4.242 [4.153–4.319]	1.473	0.63
TsFile	device_id = "d_0"	5,000	1.023 [0.927–1.214]	0.396 [0.160–0.886]	0.978	0.37

Tablet/Record consumption time decreased by 84.0% relative to its no-filter median in this rerun. TsFile took 2.58× its no-filter median and delivered 5 files (0.013 MiB), compared with one forwarded source file without filtering. Even this small selection still incurs TsFile parsing, filtering, rewriting, and delivery overhead.

All 40/40 runs passed exact device/timestamp/field coverage and duplicate-cell verification, including all 8/8 single-device runs (both formats, warm-ups and measured runs). Each single-device run delivered only d_0, all 5,000 timestamps, and all 10 fields. TsFile content verification remained outside the measured interval.

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added comments explaining the intent where it would not be obvious to an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths.
• added integration tests.

---

Key changed/added classes (or packages if there are too many classes) in this PR

• org.apache.iotdb.db.subscription.tagfilter
• SubscriptionFilterSnapshot
• SubscriptionTreeViewProjector
• SubscriptionBrokerAgent
• ConsensusLogToTabletConverter
• PipeTabletEventTsFileBatch
• ShowTopicsTask
• TopicConfig and TopicMeta