mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-15 07:17:01 +00:00
feat(query): serialize remote dynamic membership (#10341)
* feat(query): serialize remote dynamic membership Recover single-column UTF-8 build-key membership from InList and HashTableLookupExpr producers and send it to remote datanodes using the existing PhysicalInListNode. Non-monotone or unsupported subtrees widen to true; the receiver keeps the live filter authoritative. Signed-off-by: discord9 <discord9@163.com> * refactor(query): send lookup membership as a key list Serialize a HashTableLookupExpr as the existing PhysicalInListNode and preserve the rest of the expression tree. Remove per-subtree monotone and true rewrites; only an unrepresentable lookup or an over-budget payload falls back at the whole-payload level. Signed-off-by: discord9 <discord9@163.com> * docs: tighten narrow membership contract Clarify that the key-list transport is a positive pruning hint, not universal Boolean equivalence, and test that a NULL probe survives the IS NULL wrapper. Signed-off-by: discord9 <discord9@163.com> * fix(query): retain safe bounds on membership fallback When a key-list predicate cannot be shipped, keep top-level conjuncts that contain no membership and re-encode their conjunction. Drop membership-bearing conjuncts whole; fall back to true only when no safe bounds remain. Signed-off-by: discord9 <discord9@163.com> * fix(query): skip unrepresentable dynamic filter snapshots Do not synthesize a true predicate when a runtime hash-table lookup has no wire representation. Capture and fanout skip the affected snapshot, keep the query running, and report PushedDown::No so the parent filter is retained. Safe membership-free conjuncts still travel, and a producer-supplied true remains encodable. Cover restrictive-to-true updates, unsupported initial registrations with later supported updates, final unsupported cleanup, and NULL build keys. Signed-off-by: discord9 <discord9@163.com> * docs(query): clarify 642-key payload budget comment Signed-off-by: discord9 <discord9@163.com> * refactor(query): tighten dynamic membership encoding Preserve static IN conjuncts when only a runtime lookup is unrepresentable, apply the IN-list exclusion only on the over-budget path, and derive the membership count guard from the byte budget before materializing literals. Document that decoded dynamic filters are pruning-only, fix the NOT/CASE contract wording, deduplicate test helpers, and justify the schema-less InList proto encoding. Signed-off-by: discord9 <discord9@163.com> * fix(query): reset stale dynamic filters on unsupported updates Publish an explicit true reset when a producer generation cannot be carried, instead of leaving an older narrower predicate that could prune rows after widening. Keep the unsupported expression unrewritten; the reset is a transport-built invalidation. Type over-budget build-key memberships as DynFilterPayloadTooLarge, accept scalar-subquery IN lists under the conjunct-retention contract, and clarify the no-initial-predicate wording. Signed-off-by: discord9 <discord9@163.com> * docs(query): scope no-true rule to the encoder Clarify that the encoder never substitutes true for a producer node, while the long-lived fanout path publishes an explicit true reset to invalidate a stale region predicate. Point at the fanout caller instead of claiming producers always skip refused generations. Signed-off-by: discord9 <discord9@163.com> * docs(query): correct reset wording Fix the inverted conjunct-retention sentence and describe initial capture versus ongoing fanout resets consistently in the error docs. Signed-off-by: discord9 <discord9@163.com> --------- Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
Generated
+36
-35
@@ -2814,6 +2814,7 @@ dependencies = [
|
||||
"common-time",
|
||||
"datafusion",
|
||||
"datafusion-common",
|
||||
"datafusion-datasource",
|
||||
"datafusion-expr",
|
||||
"datafusion-proto",
|
||||
"datatypes",
|
||||
@@ -3771,7 +3772,7 @@ checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
|
||||
[[package]]
|
||||
name = "datafusion"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-schema 59.2.0",
|
||||
@@ -3823,7 +3824,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-catalog"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"async-trait",
|
||||
@@ -3847,7 +3848,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-catalog-listing"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"async-trait",
|
||||
@@ -3870,7 +3871,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-common"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-ipc 59.2.0",
|
||||
@@ -3896,7 +3897,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-common-runtime"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"futures",
|
||||
"log",
|
||||
@@ -3906,7 +3907,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-datasource"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"async-compression",
|
||||
@@ -3942,7 +3943,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-datasource-arrow"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-ipc 59.2.0",
|
||||
@@ -3966,7 +3967,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-datasource-csv"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"async-trait",
|
||||
@@ -3989,7 +3990,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-datasource-json"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"async-trait",
|
||||
@@ -4012,7 +4013,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-datasource-parquet"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-schema 59.2.0",
|
||||
@@ -4044,12 +4045,12 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-doc"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
|
||||
[[package]]
|
||||
name = "datafusion-execution"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-buffer 59.2.0",
|
||||
@@ -4074,7 +4075,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-expr"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-schema 59.2.0",
|
||||
@@ -4098,7 +4099,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-expr-common"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4109,7 +4110,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-functions"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-buffer 59.2.0",
|
||||
@@ -4140,7 +4141,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-functions-aggregate"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4160,7 +4161,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-functions-aggregate-common"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4171,7 +4172,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-functions-nested"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-ord 59.2.0",
|
||||
@@ -4195,7 +4196,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-functions-table"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"async-trait",
|
||||
@@ -4210,7 +4211,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-functions-window"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4226,7 +4227,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-functions-window-common"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"datafusion-common",
|
||||
"datafusion-physical-expr-common",
|
||||
@@ -4235,7 +4236,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-macros"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"datafusion-doc",
|
||||
"quote",
|
||||
@@ -4245,7 +4246,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-optimizer"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"chrono",
|
||||
@@ -4293,7 +4294,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-physical-expr"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-schema 59.2.0",
|
||||
@@ -4316,7 +4317,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-physical-expr-adapter"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4330,7 +4331,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-physical-expr-common"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"chrono",
|
||||
@@ -4347,7 +4348,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-physical-optimizer"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4366,7 +4367,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-physical-plan"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"arrow-data 59.2.0",
|
||||
@@ -4403,7 +4404,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-proto"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-catalog",
|
||||
@@ -4429,7 +4430,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-proto-common"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4439,7 +4440,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-proto-models"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"datafusion-common",
|
||||
"datafusion-proto-common",
|
||||
@@ -4449,7 +4450,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-pruning"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"datafusion-common",
|
||||
@@ -4464,7 +4465,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-session"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow-schema 59.2.0",
|
||||
"async-trait",
|
||||
@@ -4478,7 +4479,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-sql"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"arrow 59.2.0",
|
||||
"bigdecimal 0.4.8",
|
||||
@@ -4497,7 +4498,7 @@ dependencies = [
|
||||
[[package]]
|
||||
name = "datafusion-substrait"
|
||||
version = "55.1.0"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=70f84c5bf2511d40ee4a4036fa086ba36f207a85#70f84c5bf2511d40ee4a4036fa086ba36f207a85"
|
||||
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=638a77fb504e86f85ef6053c552bbee1ed814fb9#638a77fb504e86f85ef6053c552bbee1ed814fb9"
|
||||
dependencies = [
|
||||
"async-recursion",
|
||||
"async-trait",
|
||||
|
||||
+16
-16
@@ -372,22 +372,22 @@ rev = "f743c3a5fe8c1f57363eac5866c356b7d91f9b39"
|
||||
|
||||
[patch.crates-io]
|
||||
substrait = { git = "https://github.com/GreptimeTeam/substrait-rs.git", rev = "91ec978b0649417ad3da8390e7a515baec723b1b" }
|
||||
datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-execution = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-proto = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "70f84c5bf2511d40ee4a4036fa086ba36f207a85" }
|
||||
datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-execution = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-proto = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "638a77fb504e86f85ef6053c552bbee1ed814fb9" }
|
||||
datafusion-pg-catalog = { git = "https://github.com/datafusion-contrib/datafusion-postgres.git", rev = "3ae0a6cb30b1c5131687f4637247ff4f84ac11e4" }
|
||||
arrow-pg = { git = "https://github.com/datafusion-contrib/datafusion-postgres.git", rev = "3ae0a6cb30b1c5131687f4637247ff4f84ac11e4" }
|
||||
sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "9e9019bb1c7040ed956f654e39378dd42ab17884" }
|
||||
|
||||
@@ -5,7 +5,9 @@ edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[features]
|
||||
testing = []
|
||||
# The only two optional dependencies, `datafusion-datasource` and `futures-util`,
|
||||
# are needed by `testing` alone: they build the real hash-join fixtures.
|
||||
testing = ["dep:datafusion-datasource", "dep:futures-util"]
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
@@ -22,9 +24,11 @@ common-recordbatch.workspace = true
|
||||
common-time.workspace = true
|
||||
datafusion.workspace = true
|
||||
datafusion-common.workspace = true
|
||||
datafusion-datasource = { workspace = true, optional = true }
|
||||
datafusion-expr.workspace = true
|
||||
datafusion-proto.workspace = true
|
||||
datatypes.workspace = true
|
||||
futures-util = { workspace = true, optional = true }
|
||||
once_cell.workspace = true
|
||||
prost.workspace = true
|
||||
serde.workspace = true
|
||||
@@ -36,6 +40,7 @@ store-api.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
common-base.workspace = true
|
||||
datafusion-datasource.workspace = true
|
||||
futures-util.workspace = true
|
||||
serde_json.workspace = true
|
||||
tokio.workspace = true
|
||||
|
||||
@@ -218,6 +218,27 @@ pub enum Error {
|
||||
#[snafu(implicit)]
|
||||
location: Location,
|
||||
},
|
||||
|
||||
#[snafu(display(
|
||||
"Remote dynamic filter predicate cannot be represented on the wire: {}",
|
||||
reason
|
||||
))]
|
||||
DynFilterUnsupported {
|
||||
reason: String,
|
||||
#[snafu(implicit)]
|
||||
location: Location,
|
||||
},
|
||||
}
|
||||
|
||||
impl Error {
|
||||
/// Whether the error reports a dynamic filter predicate with no wire form.
|
||||
///
|
||||
/// Initial capture omits the refused snapshot; ongoing fanout sends an explicit
|
||||
/// `true` reset to invalidate any stale region-side predicate. Neither path fails
|
||||
/// the query.
|
||||
pub fn is_dyn_filter_unsupported(&self) -> bool {
|
||||
matches!(self, Error::DynFilterUnsupported { .. })
|
||||
}
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, Error>;
|
||||
@@ -250,6 +271,9 @@ impl ErrorExt for Error {
|
||||
| Error::InvalidColumnPrefix { .. } => StatusCode::InvalidArguments,
|
||||
|
||||
Error::DynFilterPayloadTooLarge { .. } => StatusCode::PlanQuery,
|
||||
// An unrepresentable dynamic predicate is a transport limitation, not a
|
||||
// server failure: callers omit the initial snapshot or reset the remote predicate.
|
||||
Error::DynFilterUnsupported { .. } => StatusCode::Unsupported,
|
||||
|
||||
Error::ConvertDfRecordBatchStream { source, .. } => source.status_code(),
|
||||
|
||||
@@ -394,6 +418,15 @@ mod tests {
|
||||
StatusCode::PlanQuery
|
||||
);
|
||||
|
||||
let unsupported_error = DataFusionError::External(Box::new(Error::DynFilterUnsupported {
|
||||
reason: "no wire form".to_string(),
|
||||
location: Location::default(),
|
||||
}));
|
||||
assert_eq!(
|
||||
datafusion_status_code::<Error>(&unsupported_error, Some(StatusCode::Internal)),
|
||||
StatusCode::Unsupported
|
||||
);
|
||||
|
||||
let boundary_error =
|
||||
DataFusionError::Shared(Arc::new(DataFusionError::External(Box::new(
|
||||
BoxedError::new(common_recordbatch::error::Error::PhysicalExpr {
|
||||
|
||||
+1949
-85
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,194 @@
|
||||
// Copyright 2023 Greptime Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Test-only helpers that produce *real* DataFusion dynamic filters.
|
||||
//!
|
||||
//! A `HashTableLookupExpr` built by hand always has `build_lease = None`, so it cannot
|
||||
//! prove that build-side membership recovery works. These helpers run an actual
|
||||
//! `HashJoinExec`, which publishes lookups backed by real build leases, and return the
|
||||
//! code of the dynamic filter it produced.
|
||||
//!
|
||||
//! The helpers live behind `test`/the `testing` feature so both `common-query` and the
|
||||
//! distributed-plan tests can exercise the same fixture.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::arrow::array::{ArrayRef, StringArray};
|
||||
use datafusion::arrow::datatypes::{DataType, Field, Schema};
|
||||
use datafusion::arrow::record_batch::RecordBatch;
|
||||
use datafusion::execution::TaskContext;
|
||||
use datafusion::execution::config::SessionConfig;
|
||||
use datafusion::execution::session_state::SessionStateBuilder;
|
||||
use datafusion::physical_expr::expressions::{Column, DynamicFilterPhysicalExpr, lit};
|
||||
use datafusion::physical_plan::filter::FilterExec;
|
||||
use datafusion::physical_plan::joins::{HashJoinExec, PartitionMode};
|
||||
use datafusion::physical_plan::repartition::RepartitionExec;
|
||||
use datafusion::physical_plan::{ExecutionPlan, Partitioning, PhysicalExpr};
|
||||
use datafusion_common::config::ConfigOptions;
|
||||
use datafusion_common::{JoinType, NullEquality};
|
||||
use datafusion_datasource::memory::MemorySourceConfig;
|
||||
use futures_util::TryStreamExt;
|
||||
|
||||
/// A real hash-join dynamic filter and the probe schema it is bound to.
|
||||
pub struct RealDynFilter {
|
||||
/// The predicate DataFusion published for the join's probe side.
|
||||
pub expr: Arc<dyn PhysicalExpr>,
|
||||
/// Schema of the probe side the predicate applies to.
|
||||
pub probe_schema: Arc<Schema>,
|
||||
}
|
||||
|
||||
/// Schema with a single nullable column.
|
||||
pub fn one_column_schema(name: &str, data_type: DataType) -> Arc<Schema> {
|
||||
Arc::new(Schema::new(vec![Field::new(name, data_type, true)]))
|
||||
}
|
||||
|
||||
/// Turns one column vector per partition into one `RecordBatch` per partition.
|
||||
pub fn partitions(schema: &Arc<Schema>, columns: Vec<Vec<ArrayRef>>) -> Vec<Vec<RecordBatch>> {
|
||||
columns
|
||||
.into_iter()
|
||||
.map(|columns| vec![RecordBatch::try_new(Arc::clone(schema), columns).unwrap()])
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// A column reference by name and index.
|
||||
pub fn key(name: &str, index: usize) -> Arc<dyn PhysicalExpr> {
|
||||
Arc::new(Column::new(name, index))
|
||||
}
|
||||
|
||||
/// A `Utf8` array.
|
||||
pub fn utf8(values: &[&str]) -> ArrayRef {
|
||||
Arc::new(StringArray::from(values.to_vec()))
|
||||
}
|
||||
|
||||
/// Runs a real `HashJoinExec` and returns the dynamic filter it published.
|
||||
///
|
||||
/// The join is forced onto the `Map` pushdown strategy (the only strategy that
|
||||
/// produces a runtime lookup) by setting the `IN`-list pushdown limit to zero, and
|
||||
/// both children are hash-partitioned on the complete join-key vector, which is what
|
||||
/// DataFusion requires before it publishes a partitioned dynamic filter.
|
||||
pub async fn run_real_hash_join(
|
||||
partition_count: usize,
|
||||
left_schema: Arc<Schema>,
|
||||
right_schema: Arc<Schema>,
|
||||
build_partitions: Vec<Vec<RecordBatch>>,
|
||||
probe_partitions: Vec<Vec<RecordBatch>>,
|
||||
on: Vec<(Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)>,
|
||||
) -> RealDynFilter {
|
||||
assert!(partition_count >= 1);
|
||||
|
||||
let left_scan: Arc<dyn ExecutionPlan> =
|
||||
MemorySourceConfig::try_new_exec(&build_partitions, Arc::clone(&left_schema), None)
|
||||
.unwrap();
|
||||
let right_scan: Arc<dyn ExecutionPlan> =
|
||||
MemorySourceConfig::try_new_exec(&probe_partitions, Arc::clone(&right_schema), None)
|
||||
.unwrap();
|
||||
|
||||
let (left, right_scan) = if partition_count > 1 {
|
||||
// Hash on the complete join-key vector on both sides, like the real plan does.
|
||||
let left_partitioning = Partitioning::Hash(
|
||||
on.iter().map(|(build, _)| Arc::clone(build)).collect(),
|
||||
partition_count,
|
||||
);
|
||||
let right_partitioning = Partitioning::Hash(
|
||||
on.iter().map(|(_, probe)| Arc::clone(probe)).collect(),
|
||||
partition_count,
|
||||
);
|
||||
(
|
||||
Arc::new(RepartitionExec::try_new(left_scan, left_partitioning).unwrap())
|
||||
as Arc<dyn ExecutionPlan>,
|
||||
Arc::new(RepartitionExec::try_new(right_scan, right_partitioning).unwrap())
|
||||
as Arc<dyn ExecutionPlan>,
|
||||
)
|
||||
} else {
|
||||
(left_scan, right_scan)
|
||||
};
|
||||
|
||||
let probe_filter = Arc::new(DynamicFilterPhysicalExpr::new(
|
||||
on.iter().map(|(_, probe)| Arc::clone(probe)).collect(),
|
||||
lit(true),
|
||||
));
|
||||
let dyn_filter: Arc<dyn PhysicalExpr> = Arc::clone(&probe_filter) as _;
|
||||
// The probe subtree must reference the produced filter, otherwise the join never
|
||||
// publishes it.
|
||||
let right: Arc<dyn ExecutionPlan> =
|
||||
Arc::new(FilterExec::try_new(Arc::clone(&dyn_filter), right_scan).unwrap());
|
||||
|
||||
let join = HashJoinExec::try_new(
|
||||
left,
|
||||
right,
|
||||
on,
|
||||
None,
|
||||
&JoinType::Inner,
|
||||
None,
|
||||
PartitionMode::Partitioned,
|
||||
NullEquality::NullEqualsNothing,
|
||||
false,
|
||||
)
|
||||
.unwrap()
|
||||
.with_dynamic_filter_expr(Arc::clone(&probe_filter))
|
||||
.unwrap();
|
||||
let join: Arc<dyn ExecutionPlan> = Arc::new(join);
|
||||
|
||||
let mut options = ConfigOptions::default();
|
||||
options.optimizer.hash_join_inlist_pushdown_max_size = 0;
|
||||
let session_config = SessionConfig::from(options);
|
||||
let state = SessionStateBuilder::new()
|
||||
.with_config(session_config)
|
||||
.with_default_features()
|
||||
.build();
|
||||
let task_ctx = Arc::new(TaskContext::from(&state));
|
||||
|
||||
// Every build partition must report before the filter is finalized, so all
|
||||
// partition streams have to make progress concurrently.
|
||||
let streams = (0..partition_count)
|
||||
.map(|partition| {
|
||||
let join = Arc::clone(&join);
|
||||
let task_ctx = Arc::clone(&task_ctx);
|
||||
async move {
|
||||
let stream = join.execute(partition, task_ctx).unwrap();
|
||||
stream.try_collect::<Vec<_>>().await.unwrap();
|
||||
}
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
futures_util::future::join_all(streams).await;
|
||||
|
||||
RealDynFilter {
|
||||
expr: probe_filter.current().unwrap(),
|
||||
probe_schema: right_schema,
|
||||
}
|
||||
}
|
||||
|
||||
/// Build and probe batches for a `Utf8` join over distinct keys.
|
||||
///
|
||||
/// Both sides start as one partition; the fixture hash-repartitions them so every hash
|
||||
/// partition holds rows and DataFusion produces one filter branch per partition. Also
|
||||
/// returns every distinct key.
|
||||
pub fn utf8_join(
|
||||
partition_count: usize,
|
||||
) -> (Vec<Vec<RecordBatch>>, Vec<Vec<RecordBatch>>, Vec<String>) {
|
||||
let schema = one_column_schema("trace_id", DataType::Utf8);
|
||||
let values = (0..partition_count * 8)
|
||||
.map(|i| format!("trace-{i:02}"))
|
||||
.collect::<Vec<_>>();
|
||||
let borrowed = values.iter().map(String::as_str).collect::<Vec<_>>();
|
||||
let build = partitions(&schema, vec![vec![utf8(&borrowed)]]);
|
||||
// Probe rows include every member plus keys that belong to no build side.
|
||||
let mut probe_values = borrowed.clone();
|
||||
let misses = (0..partition_count * 4)
|
||||
.map(|i| format!("miss-{i:02}"))
|
||||
.collect::<Vec<_>>();
|
||||
probe_values.extend(misses.iter().map(String::as_str));
|
||||
let probe = partitions(&schema, vec![vec![utf8(&probe_values)]]);
|
||||
(build, probe, values)
|
||||
}
|
||||
@@ -18,15 +18,14 @@ use std::sync::Arc;
|
||||
use common_query::request::{
|
||||
DynFilterPayload, INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, InitialDynFilterReg,
|
||||
InitialDynFilterRegs, InitialDynFilterSnapshot, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
is_dyn_filter_payload_too_large, is_dyn_filter_unsupported,
|
||||
};
|
||||
use datafusion::physical_plan::filter_pushdown::PushedDown;
|
||||
use datafusion_common::Result;
|
||||
use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
|
||||
use datafusion_expr::ColumnarValue;
|
||||
use datafusion_physical_expr::PhysicalExpr;
|
||||
use datafusion_physical_expr::expressions::{
|
||||
Column, DynamicFilterPhysicalExpr, InListExpr, Literal, lit,
|
||||
};
|
||||
use datafusion_physical_expr::expressions::{Column, DynamicFilterPhysicalExpr, Literal, lit};
|
||||
use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr;
|
||||
use datafusion_physical_expr::utils::collect_columns;
|
||||
use session::context::{QueryContext, QueryContextRef};
|
||||
@@ -36,6 +35,7 @@ use crate::dist_plan::filter_id::build_remote_dyn_filter_id;
|
||||
use crate::dist_plan::{
|
||||
FilterId, QueryDynFilterRegistry, RemoteDynFilterProducerId, Subscriber, SubscriberRegistration,
|
||||
};
|
||||
use crate::metrics::REMOTE_DYN_FILTER_ENCODE_TOTAL;
|
||||
use crate::region_query::RegionQueryTarget;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -82,7 +82,7 @@ pub(crate) fn capture_remote_dyn_filters_for_pushdown(
|
||||
producer_local_ordinal,
|
||||
alive_dyn_filter,
|
||||
)
|
||||
.map(|captured| (captured, PushedDown::Yes))
|
||||
.map(pushdown_for_captured_filter)
|
||||
} else if is_scalar_subquery_filter(&filter) {
|
||||
// The filter is only forwarded to regions for pruning; the parent keeps
|
||||
// evaluating it, so report it as not pushed down.
|
||||
@@ -128,24 +128,43 @@ pub(crate) fn capture_remote_dyn_filters_for_pushdown(
|
||||
}
|
||||
}
|
||||
|
||||
/// A captured filter is only reported as pushed down when its initial snapshot could be
|
||||
/// shipped.
|
||||
///
|
||||
/// An unsupported (or oversized) snapshot is never rewritten to `true`, so a region
|
||||
/// receives **no initial predicate** from the initial registration. The registration
|
||||
/// itself is still sent - filter id plus children - and the receiver installs a `true`
|
||||
/// placeholder for it, which is why the region is not filtered yet but the later updates
|
||||
/// already have somewhere to land. The parent must therefore keep evaluating the filter
|
||||
/// itself, which is what [`PushedDown::No`] tells DataFusion.
|
||||
fn pushdown_for_captured_filter(captured: CapturedDynFilter) -> (CapturedDynFilter, PushedDown) {
|
||||
let pushed_down = if captured.initial_registration.initial_snapshot.is_some() {
|
||||
PushedDown::Yes
|
||||
} else {
|
||||
PushedDown::No
|
||||
};
|
||||
|
||||
(captured, pushed_down)
|
||||
}
|
||||
|
||||
/// Returns true if `filter` references at least one scalar subquery and is otherwise
|
||||
/// built only from columns and literals, so it becomes a plain predicate once the
|
||||
/// subqueries are resolved.
|
||||
///
|
||||
/// Filters containing `IN` lists are rejected: an oversized payload replaces `IN` with
|
||||
/// `true`, which only widens a conjunction but can narrow an arbitrary predicate such as
|
||||
/// `(host IN (...)) = false`.
|
||||
// TODO(dennis): make the bounds-only fallback in `DynFilterPayload` replace only `IN` lists
|
||||
// in conjunctive positions, then drop this restriction.
|
||||
/// `IN` lists are accepted: the retention contract in
|
||||
/// [`DynFilterPayload::from_datafusion_expr`] only ever drops whole top-level `AND`
|
||||
/// conjuncts and never substitutes `true` for a node, so an `IN` list inside an arbitrary
|
||||
/// predicate such as `(host IN (...)) = false` cannot be narrowed by transport. A payload
|
||||
/// that cannot be shipped as it stands is an explicit error, which the capture answers by
|
||||
/// only attaching no initial predicate (the region keeps its `true` placeholder) while the
|
||||
/// parent keeps evaluating the filter.
|
||||
fn is_scalar_subquery_filter(filter: &Arc<dyn PhysicalExpr>) -> bool {
|
||||
let mut has_scalar_subquery = false;
|
||||
let mut supported = true;
|
||||
let _ = filter.apply(|expr| {
|
||||
if expr.is::<ScalarSubqueryExpr>() {
|
||||
has_scalar_subquery = true;
|
||||
} else if expr.is::<InListExpr>()
|
||||
|| (expr.children().is_empty() && !expr.is::<Column>() && !expr.is::<Literal>())
|
||||
{
|
||||
} else if expr.children().is_empty() && !expr.is::<Column>() && !expr.is::<Literal>() {
|
||||
supported = false;
|
||||
return Ok(TreeNodeRecursion::Stop);
|
||||
}
|
||||
@@ -323,6 +342,15 @@ fn attach_initial_snapshot(
|
||||
initial_registration.with_initial_snapshot(initial_snapshot)
|
||||
}
|
||||
|
||||
/// Initial snapshot of a captured filter, or `None` when regions must not be filtered yet.
|
||||
///
|
||||
/// A predicate transport cannot represent (or that does not fit the payload budget) is
|
||||
/// *not* replaced with `true`: only the initial snapshot is dropped (warn + metric), so the
|
||||
/// region starts with no initial predicate but still receives the registration itself
|
||||
/// (filter id plus children) and installs its `true` placeholder. Later updates land in
|
||||
/// that placeholder - a representable payload, or the explicit `true` reset the registry
|
||||
/// publishes when a generation has no wire form. The capture reports `PushedDown::No` so
|
||||
/// the parent keeps evaluating the filter itself.
|
||||
fn initial_snapshot(
|
||||
alive_dyn_filter: &DynamicFilterPhysicalExpr,
|
||||
) -> Option<InitialDynFilterSnapshot> {
|
||||
@@ -340,6 +368,24 @@ fn initial_snapshot(
|
||||
REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
) {
|
||||
Ok(payload) => payload,
|
||||
Err(error) if is_dyn_filter_unsupported(&error) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["unsupported"])
|
||||
.inc();
|
||||
common_telemetry::warn!(
|
||||
error; "Remote dyn filter initial snapshot is dropped because the predicate has no wire form"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
Err(error) if is_dyn_filter_payload_too_large(&error) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["too_large"])
|
||||
.inc();
|
||||
common_telemetry::warn!(
|
||||
error; "Remote dyn filter initial snapshot is dropped because the predicate does not fit the payload budget"
|
||||
);
|
||||
return None;
|
||||
}
|
||||
Err(error) => {
|
||||
common_telemetry::warn!(error; "Failed to encode remote dyn filter initial snapshot");
|
||||
return None;
|
||||
@@ -453,11 +499,13 @@ mod tests {
|
||||
|
||||
use common_meta::peer::Peer;
|
||||
use datafusion::execution::TaskContext;
|
||||
use datafusion::physical_plan::joins::join_hash_map::JoinHashMapU32;
|
||||
use datafusion::physical_plan::joins::{HashTableLookupExpr, Map, SeededRandomState};
|
||||
use datafusion_common::ScalarValue;
|
||||
use datafusion_expr::ColumnarValue;
|
||||
use datafusion_expr::Operator;
|
||||
use datafusion_expr::physical_planning_context::{ScalarSubqueryResults, SubqueryIndex};
|
||||
use datafusion_physical_expr::expressions::BinaryExpr;
|
||||
use datafusion_physical_expr::expressions::{BinaryExpr, NotExpr};
|
||||
use session::query_id::QueryId;
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -557,6 +605,33 @@ mod tests {
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// A probe filter whose current predicate has no wire form.
|
||||
///
|
||||
/// A hand-built lookup has no build lease, hence no key list, and `NOT lookup` is a
|
||||
/// single top-level conjunct that carries the membership: neither a key list nor a
|
||||
/// membership-free conjunct exists.
|
||||
fn test_unsupported_snapshot_filter(
|
||||
column_name: &str,
|
||||
column_index: usize,
|
||||
) -> Arc<DynamicFilterPhysicalExpr> {
|
||||
let column = Arc::new(Column::new(column_name, column_index)) as Arc<dyn PhysicalExpr>;
|
||||
let lookup = Arc::new(HashTableLookupExpr::new(
|
||||
vec![Arc::clone(&column)],
|
||||
SeededRandomState::with_seed(0),
|
||||
Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(0)))),
|
||||
"hash_lookup".to_string(),
|
||||
None,
|
||||
)) as Arc<dyn PhysicalExpr>;
|
||||
let dyn_filter = Arc::new(DynamicFilterPhysicalExpr::new(
|
||||
vec![Arc::clone(&column) as _],
|
||||
lit(true) as _,
|
||||
));
|
||||
dyn_filter
|
||||
.update(Arc::new(NotExpr::new(lookup)) as _)
|
||||
.unwrap();
|
||||
dyn_filter
|
||||
}
|
||||
|
||||
fn test_dyn_filter_with_snapshot_payload(
|
||||
column_name: &str,
|
||||
column_index: usize,
|
||||
@@ -654,7 +729,12 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scalar_subquery_filter_with_in_list_is_left_to_parent() {
|
||||
fn scalar_subquery_filter_with_in_list_is_captured_and_never_narrowed() {
|
||||
// `(host IN ('a', 'b')) = (SELECT flag ...)` used to be rejected because an
|
||||
// oversized payload was thought to replace `IN` with `true`. That is not the
|
||||
// contract any more: transport never substitutes `true` for a node and only drops
|
||||
// whole top-level conjuncts, so a membership inside an arbitrary predicate can only
|
||||
// be kept or refused as a whole, never narrowed.
|
||||
let results = ScalarSubqueryResults::new(1);
|
||||
let schema = arrow_schema::Schema::new(vec![arrow_schema::Field::new(
|
||||
"host",
|
||||
@@ -668,7 +748,6 @@ mod tests {
|
||||
&schema,
|
||||
)
|
||||
.unwrap();
|
||||
// `(host IN ('a', 'b')) = (SELECT flag ...)`
|
||||
let filter = Arc::new(BinaryExpr::new(
|
||||
host_in_list,
|
||||
Operator::Eq,
|
||||
@@ -676,7 +755,7 @@ mod tests {
|
||||
arrow_schema::DataType::Boolean,
|
||||
true,
|
||||
SubqueryIndex::new(0),
|
||||
results,
|
||||
results.clone(),
|
||||
)),
|
||||
)) as Arc<dyn PhysicalExpr>;
|
||||
|
||||
@@ -687,8 +766,40 @@ mod tests {
|
||||
vec![filter],
|
||||
);
|
||||
|
||||
// Captured like any other scalar subquery filter, and still left to the parent.
|
||||
assert_eq!(pushed_down(&pushdown), vec![false]);
|
||||
assert!(pushdown.captured_dyn_filters.is_empty());
|
||||
assert_eq!(pushdown.captured_dyn_filters.len(), 1);
|
||||
let captured = &pushdown.captured_dyn_filters[0];
|
||||
|
||||
// Once resolved, the published predicate keeps the `IN` list structurally instead
|
||||
// of widening it to `true`: `(host IN (...)) = true`.
|
||||
results
|
||||
.set(SubqueryIndex::new(0), ScalarValue::Boolean(Some(true)))
|
||||
.unwrap();
|
||||
resolve_scalar_subquery_dyn_filters(std::slice::from_ref(captured));
|
||||
let snapshot = refreshed_regs(std::slice::from_ref(captured)).regs[0]
|
||||
.initial_snapshot
|
||||
.clone()
|
||||
.expect("the resolved predicate is representable");
|
||||
let display = decode_snapshot(&snapshot, "host");
|
||||
assert!(
|
||||
display.contains("IN"),
|
||||
"the IN list must survive the wire: {display}"
|
||||
);
|
||||
assert_ne!(display, "true", "transport must not widen to `true`");
|
||||
assert!(
|
||||
display.ends_with("= true"),
|
||||
"the equality over the IN list must survive: {display}"
|
||||
);
|
||||
|
||||
// A payload that does not fit is refused outright rather than narrowed: the whole
|
||||
// conjunct is dropped and nothing is left to ship.
|
||||
let current = captured.alive_dyn_filter.current().unwrap();
|
||||
let refused = DynFilterPayload::from_datafusion_expr(¤t, 8).unwrap_err();
|
||||
assert!(
|
||||
is_dyn_filter_payload_too_large(&refused),
|
||||
"an oversized payload must be refused as too large, got: {refused:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -851,6 +962,88 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capture_remote_dyn_filters_for_pushdown_skips_unsupported_initial_snapshot() {
|
||||
// The parent retention and the registration are checked end to end: the capture
|
||||
// still sends its registration (filter id plus children) but no initial predicate,
|
||||
// the region installs its `true` placeholder, and a later representable generation
|
||||
// still reaches it.
|
||||
let dyn_filter = test_unsupported_snapshot_filter("host", 0);
|
||||
let parent_filters =
|
||||
vec![Arc::clone(&dyn_filter) as Arc<dyn datafusion::physical_plan::PhysicalExpr>];
|
||||
|
||||
let pushdown = capture_remote_dyn_filters_for_pushdown(
|
||||
test_remote_dyn_filter_producer_id(42),
|
||||
0,
|
||||
&[],
|
||||
parent_filters,
|
||||
);
|
||||
|
||||
// No snapshot is published (and it is never replaced with `true`), so the parent
|
||||
// must keep evaluating the filter itself.
|
||||
assert_eq!(pushed_down(&pushdown), vec![false]);
|
||||
assert_eq!(pushdown.captured_dyn_filters.len(), 1);
|
||||
let captured = &pushdown.captured_dyn_filters[0];
|
||||
assert!(captured.initial_registration.initial_snapshot.is_none());
|
||||
assert!(
|
||||
!captured
|
||||
.initial_registration
|
||||
.child_exprs_datafusion_proto
|
||||
.is_empty(),
|
||||
"the registration must still describe the filter children"
|
||||
);
|
||||
|
||||
// The registration survives the wire: filter id and children are intact, and the
|
||||
// region installs the filter with a `true` placeholder and no initial predicate.
|
||||
let regs = build_initial_dyn_filter_regs_for_region(&pushdown.captured_dyn_filters);
|
||||
let serialized = regs.to_extension_value().unwrap();
|
||||
let received = InitialDynFilterRegs::from_extension_value(&serialized).unwrap();
|
||||
assert_eq!(received.regs.len(), 1);
|
||||
assert_eq!(
|
||||
received.regs[0].filter_id,
|
||||
captured.filter_id.to_string(),
|
||||
"the filter id must survive serialization"
|
||||
);
|
||||
assert!(received.regs[0].initial_snapshot.is_none());
|
||||
let schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
|
||||
"host",
|
||||
arrow_schema::DataType::Utf8,
|
||||
true,
|
||||
)]));
|
||||
let children = received.regs[0]
|
||||
.decode_children(
|
||||
&TaskContext::default(),
|
||||
&schema,
|
||||
REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(children.len(), 1);
|
||||
assert!(children[0].is::<Column>());
|
||||
|
||||
// A later generation transport can represent is delivered to the installed filter.
|
||||
dyn_filter.update(lit(false) as _).unwrap();
|
||||
let current = captured.alive_dyn_filter.current().unwrap();
|
||||
let payload =
|
||||
DynFilterPayload::from_datafusion_expr(¤t, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES)
|
||||
.unwrap();
|
||||
let decoded = payload
|
||||
.decode_datafusion_expr(
|
||||
&TaskContext::default(),
|
||||
&schema,
|
||||
REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(decoded.to_string(), "false");
|
||||
let installed = Arc::new(DynamicFilterPhysicalExpr::new(children, lit(true) as _));
|
||||
let generation = installed.snapshot_generation();
|
||||
installed.update(decoded).unwrap();
|
||||
assert!(
|
||||
installed.snapshot_generation() > generation,
|
||||
"the receiver generation must advance"
|
||||
);
|
||||
assert_eq!(installed.current().unwrap().to_string(), "false");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capture_remote_dyn_filters_for_pushdown_attaches_initial_snapshot_after_update() {
|
||||
let dyn_filter = Arc::new(DynamicFilterPhysicalExpr::new(
|
||||
|
||||
@@ -238,11 +238,21 @@ impl RemoteDynFilterReceiverExtensionPlanner {
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::arrow::datatypes::{DataType, Field, Schema};
|
||||
use datafusion::arrow::array::{
|
||||
BooleanArray, Float64Array, StringArray, TimestampMillisecondArray,
|
||||
};
|
||||
use datafusion::arrow::datatypes::{DataType, Field, Schema, TimeUnit};
|
||||
use datafusion::arrow::record_batch::RecordBatch;
|
||||
use datafusion::execution::TaskContext;
|
||||
use datafusion::physical_plan::expressions::{Column, DynamicFilterPhysicalExpr, lit};
|
||||
use datafusion_common::DFSchema;
|
||||
use datafusion_expr::{EmptyRelation, UserDefinedLogicalNodeCore};
|
||||
|
||||
use common_query::request::{
|
||||
DynFilterPayload, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
remote_membership_test_support::{key, one_column_schema, run_real_hash_join, utf8},
|
||||
};
|
||||
|
||||
use super::*;
|
||||
|
||||
fn empty_input() -> LogicalPlan {
|
||||
@@ -318,4 +328,118 @@ mod tests {
|
||||
assert_eq!(column.name(), "instance");
|
||||
assert_eq!(column.index(), 1);
|
||||
}
|
||||
|
||||
/// The receiver keeps the recovered payload's columns as the filter's children, so
|
||||
/// a name-based remap must also remap the recovered `InList` operand when the input
|
||||
/// is pruned to a different column order.
|
||||
fn pruned_input_with_trace_id() -> LogicalPlan {
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
Field::new(
|
||||
"ts",
|
||||
DataType::Timestamp(TimeUnit::Millisecond, None),
|
||||
false,
|
||||
),
|
||||
Field::new("value", DataType::Float64, true),
|
||||
Field::new("trace_id", DataType::Utf8, true),
|
||||
]));
|
||||
LogicalPlan::EmptyRelation(EmptyRelation {
|
||||
produce_one_row: false,
|
||||
schema: Arc::new(DFSchema::try_from(schema).unwrap()),
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recovered_membership_payload_survives_receiver_remap() {
|
||||
// Coordinator side: a real hash join over the trace id publishes a runtime
|
||||
// lookup that the GT encoder recovers as exact build membership.
|
||||
let producer_schema = one_column_schema("trace_id", DataType::Utf8);
|
||||
let trace_id = key("trace_id", 0);
|
||||
let members = ["trace-a", "trace-b"];
|
||||
let member_batch =
|
||||
RecordBatch::try_new(Arc::clone(&producer_schema), vec![utf8(&members)]).unwrap();
|
||||
let filter = run_real_hash_join(
|
||||
1,
|
||||
Arc::clone(&producer_schema),
|
||||
Arc::clone(&producer_schema),
|
||||
vec![vec![member_batch.clone()]],
|
||||
vec![vec![member_batch]],
|
||||
vec![(Arc::clone(&trace_id), Arc::clone(&trace_id))],
|
||||
)
|
||||
.await;
|
||||
let payload = DynFilterPayload::from_datafusion_expr(
|
||||
&filter.expr,
|
||||
REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
)
|
||||
.unwrap();
|
||||
let decoded = payload
|
||||
.decode_datafusion_expr(
|
||||
&TaskContext::default(),
|
||||
&producer_schema,
|
||||
REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// Receiver side: the payload's columns become the filter's children.
|
||||
let payload_columns = collect_columns(&decoded)
|
||||
.into_iter()
|
||||
.map(|column| Arc::new(column) as Arc<dyn PhysicalExpr>)
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(payload_columns.len(), 1);
|
||||
let dyn_filter = Arc::new(DynamicFilterPhysicalExpr::new(
|
||||
payload_columns,
|
||||
Arc::clone(&decoded),
|
||||
));
|
||||
|
||||
let plan =
|
||||
RemoteDynFilterReceiverLogicalPlan::new(pruned_input_with_trace_id(), vec![dyn_filter]);
|
||||
let remapped = UserDefinedLogicalNodeCore::with_exprs_and_inputs(
|
||||
&plan,
|
||||
vec![],
|
||||
vec![pruned_input_with_trace_id()],
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// The recovered membership is remapped by name to the pruned input ordinal.
|
||||
let receiver_filter = remapped.dyn_filters()[0]
|
||||
.downcast_ref::<DynamicFilterPhysicalExpr>()
|
||||
.expect("receiver keeps the dynamic filter");
|
||||
let current = receiver_filter.current().unwrap();
|
||||
let remapped_columns = collect_columns(¤t);
|
||||
assert_eq!(remapped_columns.len(), 1);
|
||||
let remapped_column = remapped_columns.iter().next().unwrap();
|
||||
assert_eq!(remapped_column.name(), "trace_id");
|
||||
assert_eq!(remapped_column.index(), 2);
|
||||
|
||||
// ...and still selects exactly the build keys on the pruned schema.
|
||||
let pruned_schema = Arc::new(Schema::new(vec![
|
||||
Field::new(
|
||||
"ts",
|
||||
DataType::Timestamp(TimeUnit::Millisecond, None),
|
||||
false,
|
||||
),
|
||||
Field::new("value", DataType::Float64, true),
|
||||
Field::new("trace_id", DataType::Utf8, true),
|
||||
]));
|
||||
let batch = RecordBatch::try_new(
|
||||
pruned_schema,
|
||||
vec![
|
||||
Arc::new(TimestampMillisecondArray::from(vec![0i64, 1, 2])),
|
||||
Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0])),
|
||||
Arc::new(StringArray::from(vec!["trace-a", "trace-b", "miss"])),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let selected = current
|
||||
.evaluate(&batch)
|
||||
.unwrap()
|
||||
.into_array(batch.num_rows())
|
||||
.unwrap();
|
||||
let selected = selected.as_any().downcast_ref::<BooleanArray>().unwrap();
|
||||
assert_eq!(
|
||||
(0..batch.num_rows())
|
||||
.map(|row| selected.value(row))
|
||||
.collect::<Vec<_>>(),
|
||||
vec![true, true, false]
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,11 +18,14 @@ use std::sync::{Arc, Mutex, RwLock, Weak};
|
||||
use std::time::Duration;
|
||||
|
||||
use api::v1::region::{RemoteDynFilterUnregister, RemoteDynFilterUpdate};
|
||||
use common_query::request::{DynFilterPayload, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES};
|
||||
use common_query::request::{
|
||||
DynFilterPayload, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, is_dyn_filter_payload_too_large,
|
||||
is_dyn_filter_unsupported,
|
||||
};
|
||||
use common_runtime::spawn_global;
|
||||
use common_telemetry::{debug, warn};
|
||||
use datafusion_physical_expr::PhysicalExpr;
|
||||
use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
|
||||
use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr, lit};
|
||||
use session::query_id::QueryId;
|
||||
use store_api::storage::RegionId;
|
||||
use tokio::sync::{Notify, watch};
|
||||
@@ -381,6 +384,20 @@ async fn run_entry_fanout(
|
||||
unregister_entry_once_for_query(®ion_query_handler, query_id, &entry).await;
|
||||
}
|
||||
|
||||
/// Publishes one producer snapshot to every subscriber of `entry`.
|
||||
///
|
||||
/// # Explicit `true` reset instead of a stale predicate
|
||||
///
|
||||
/// A generation the wire cannot carry (a predicate with no wire form, or one that stays
|
||||
/// over budget after top-level conjunct retention) is **not** skipped any more: the
|
||||
/// subscriber would keep the previous, usually narrower predicate, and a producer that
|
||||
/// then widens its filter would have rows pruned away by that stale predicate. Such a
|
||||
/// generation is instead published as a literal-`true` payload at the same generation, so
|
||||
/// the region drops the stale predicate.
|
||||
///
|
||||
/// This reset is an explicit invalidation, not a serialization fallback: the producer
|
||||
/// expression is never rewritten and no predicate is ever *narrowed* on the way out. The
|
||||
/// query keeps running in every case; nothing in here returns an error.
|
||||
async fn fanout_snapshot_for_query(
|
||||
query_id: QueryId,
|
||||
region_query_handler: &RegionQueryHandlerRef,
|
||||
@@ -405,31 +422,9 @@ async fn fanout_snapshot_for_query(
|
||||
let _ = entry.mark_generation_sent(generation);
|
||||
}
|
||||
|
||||
let payload =
|
||||
match DynFilterPayload::from_datafusion_expr(¤t, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES)
|
||||
{
|
||||
Ok(DynFilterPayload::Datafusion(payload)) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["success"])
|
||||
.inc();
|
||||
REMOTE_DYN_FILTER_PAYLOAD_BYTES.observe(payload.len() as f64);
|
||||
payload
|
||||
}
|
||||
Ok(_) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["unsupported"])
|
||||
.inc();
|
||||
warn!("Ignored unsupported remote dynamic filter producer payload");
|
||||
return true;
|
||||
}
|
||||
Err(error) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["error"])
|
||||
.inc();
|
||||
warn!(error; "Failed to encode remote dynamic filter producer snapshot");
|
||||
return true;
|
||||
}
|
||||
};
|
||||
let Some(payload) = snapshot_payload(¤t) else {
|
||||
return true;
|
||||
};
|
||||
|
||||
fanout_update_for_query(
|
||||
query_id,
|
||||
@@ -444,6 +439,100 @@ async fn fanout_snapshot_for_query(
|
||||
.await
|
||||
}
|
||||
|
||||
/// Payload to publish for one producer snapshot, or `None` when nothing can be published.
|
||||
///
|
||||
/// A representable snapshot is published exactly as the producer wrote it. A snapshot the
|
||||
/// wire cannot carry is **not** published as the producer wrote it (that is impossible) and
|
||||
/// is **not** skipped either: the stale region predicate it would leave behind could prune
|
||||
/// away rows the producer keeps after widening, so it is replaced by the explicit `true`
|
||||
/// reset described on [`fanout_snapshot_for_query`]. The producer expression itself is
|
||||
/// never rewritten.
|
||||
fn snapshot_payload(current: &Arc<dyn PhysicalExpr>) -> Option<Vec<u8>> {
|
||||
let max_payload_bytes = REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES;
|
||||
match DynFilterPayload::from_datafusion_expr(current, max_payload_bytes) {
|
||||
Ok(DynFilterPayload::Datafusion(payload)) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["success"])
|
||||
.inc();
|
||||
REMOTE_DYN_FILTER_PAYLOAD_BYTES.observe(payload.len() as f64);
|
||||
Some(payload)
|
||||
}
|
||||
Ok(_) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["unsupported"])
|
||||
.inc();
|
||||
warn!(
|
||||
"Remote dynamic filter producer payload has an unknown encoding; publishing an explicit `true` reset"
|
||||
);
|
||||
true_reset_payload()
|
||||
}
|
||||
Err(error) if is_dyn_filter_unsupported(&error) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["unsupported"])
|
||||
.inc();
|
||||
warn!(
|
||||
error;
|
||||
"Remote dynamic filter snapshot has no wire form; publishing an explicit `true` reset to invalidate the stale region predicate"
|
||||
);
|
||||
true_reset_payload()
|
||||
}
|
||||
Err(error) if is_dyn_filter_payload_too_large(&error) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["too_large"])
|
||||
.inc();
|
||||
warn!(
|
||||
error;
|
||||
"Remote dynamic filter snapshot stays over budget; publishing an explicit `true` reset to invalidate the stale region predicate"
|
||||
);
|
||||
true_reset_payload()
|
||||
}
|
||||
Err(error) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["error"])
|
||||
.inc();
|
||||
warn!(
|
||||
error;
|
||||
"Failed to encode remote dynamic filter producer snapshot; publishing an explicit `true` reset to invalidate the stale region predicate"
|
||||
);
|
||||
true_reset_payload()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Encodes the literal-`true` payload that invalidates a stale remote predicate.
|
||||
///
|
||||
/// This is the wire form of the explicit reset described on [`fanout_snapshot_for_query`]:
|
||||
/// it is a *payload the transport builds itself*, never a reuse of the producer's
|
||||
/// unrepresentable expression, so no predicate is rewritten to `true` on the way out. It
|
||||
/// has no budget problem by construction and can only fail on a transport defect, which
|
||||
/// is reported as `None` so best-effort fanout continues with the query running.
|
||||
fn true_reset_payload() -> Option<Vec<u8>> {
|
||||
let reset = lit(true) as Arc<dyn PhysicalExpr>;
|
||||
match DynFilterPayload::from_datafusion_expr(&reset, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES) {
|
||||
Ok(DynFilterPayload::Datafusion(payload)) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["reset"])
|
||||
.inc();
|
||||
REMOTE_DYN_FILTER_PAYLOAD_BYTES.observe(payload.len() as f64);
|
||||
Some(payload)
|
||||
}
|
||||
Ok(_) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["reset_error"])
|
||||
.inc();
|
||||
warn!("Ignored remote dynamic filter `true` reset with an unknown encoding");
|
||||
None
|
||||
}
|
||||
Err(error) => {
|
||||
REMOTE_DYN_FILTER_ENCODE_TOTAL
|
||||
.with_label_values(&["reset_error"])
|
||||
.inc();
|
||||
warn!(error; "Failed to encode the remote dynamic filter `true` reset");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn fanout_update_for_query(
|
||||
query_id: QueryId,
|
||||
@@ -828,7 +917,10 @@ mod tests {
|
||||
use async_trait::async_trait;
|
||||
use common_meta::peer::Peer;
|
||||
use common_query::request::QueryRequest;
|
||||
use datafusion_physical_expr::expressions::{Column, lit};
|
||||
use datafusion::execution::TaskContext;
|
||||
use datafusion::physical_plan::joins::join_hash_map::JoinHashMapU32;
|
||||
use datafusion::physical_plan::joins::{HashTableLookupExpr, Map, SeededRandomState};
|
||||
use datafusion_physical_expr::expressions::{Column, NotExpr, lit};
|
||||
use session::ReadPreference;
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -1009,6 +1101,42 @@ mod tests {
|
||||
Arc::new(DynamicFilterPhysicalExpr::new(children, lit(true) as _))
|
||||
}
|
||||
|
||||
/// `NOT lookup` over a hand-built lookup: no key list, and no membership-free conjunct.
|
||||
fn unsupported_lookup_predicate() -> Arc<dyn PhysicalExpr> {
|
||||
let column = Arc::new(Column::new("host", 0)) as Arc<dyn PhysicalExpr>;
|
||||
let lookup = Arc::new(HashTableLookupExpr::new(
|
||||
vec![Arc::clone(&column)],
|
||||
SeededRandomState::with_seed(0),
|
||||
Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(0)))),
|
||||
"hash_lookup".to_string(),
|
||||
None,
|
||||
)) as Arc<dyn PhysicalExpr>;
|
||||
|
||||
Arc::new(NotExpr::new(lookup)) as Arc<dyn PhysicalExpr>
|
||||
}
|
||||
|
||||
fn host_schema() -> Arc<arrow_schema::Schema> {
|
||||
Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
|
||||
"host",
|
||||
arrow_schema::DataType::Utf8,
|
||||
true,
|
||||
)]))
|
||||
}
|
||||
|
||||
/// Decodes a recorded update the way a region receiver does.
|
||||
fn decode_recorded_payload(
|
||||
update: &RecordedUpdate,
|
||||
schema: &Arc<arrow_schema::Schema>,
|
||||
) -> Arc<dyn PhysicalExpr> {
|
||||
DynFilterPayload::Datafusion(update.payload.clone())
|
||||
.decode_datafusion_expr(
|
||||
&TaskContext::default(),
|
||||
schema,
|
||||
REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES,
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn registry_manager_returns_same_registry_for_same_query() {
|
||||
let manager = Arc::new(DynFilterRegistryManager::default());
|
||||
@@ -1337,6 +1465,145 @@ mod tests {
|
||||
assert_eq!(entry.subscribers().len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fanout_resets_stale_predicate_when_a_wider_generation_has_no_wire_form() {
|
||||
let query_id = test_query_id(11);
|
||||
let registry = Arc::new(QueryDynFilterRegistry::new(query_id));
|
||||
let filter = test_dyn_filter(&["host"]);
|
||||
let filter_id = test_filter_id(1);
|
||||
let entry = match registry.register_remote_dyn_filter(filter_id.clone(), filter.clone()) {
|
||||
EntryRegistration::Inserted(entry) => entry,
|
||||
other => panic!("unexpected registration result: {other:?}"),
|
||||
};
|
||||
let subscriber = Subscriber::new(RegionId::new(1024, 7), test_target(1));
|
||||
assert_eq!(
|
||||
registry.register_subscriber(&filter_id, subscriber),
|
||||
SubscriberRegistration::Added
|
||||
);
|
||||
|
||||
let handler = Arc::new(RecordingRegionQueryHandler::default());
|
||||
let handler_ref = handler.clone() as RegionQueryHandlerRef;
|
||||
let schema = host_schema();
|
||||
|
||||
// A representable narrow generation reaches the region and is applied there.
|
||||
filter.update(lit(false) as _).unwrap();
|
||||
registry
|
||||
.fanout_snapshot(&handler_ref, &entry, filter.as_ref(), false)
|
||||
.await;
|
||||
let updates = handler.updates();
|
||||
assert_eq!(updates.len(), 1);
|
||||
let narrow = decode_recorded_payload(&updates[0], &schema);
|
||||
assert_eq!(narrow.to_string(), "false");
|
||||
let receiver = Arc::new(DynamicFilterPhysicalExpr::new(
|
||||
vec![Arc::new(Column::new("host", 0)) as _],
|
||||
lit(true) as _,
|
||||
));
|
||||
receiver.update(narrow).unwrap();
|
||||
let narrow_generation = receiver.snapshot_generation();
|
||||
assert_eq!(receiver.current().unwrap().to_string(), "false");
|
||||
|
||||
// `NOT lookup` has no wire form and no membership-free conjunct. The producer
|
||||
// expression is never rewritten, but leaving the stale `false` installed would let
|
||||
// it prune away the rows of a producer that widened, so the region gets an explicit
|
||||
// `true` reset at the new generation instead.
|
||||
filter.update(unsupported_lookup_predicate() as _).unwrap();
|
||||
registry
|
||||
.fanout_snapshot(&handler_ref, &entry, filter.as_ref(), false)
|
||||
.await;
|
||||
let updates = handler.updates();
|
||||
assert_eq!(
|
||||
updates.len(),
|
||||
2,
|
||||
"the unsupported generation must be published as a reset"
|
||||
);
|
||||
assert_eq!(updates[1].generation, filter.snapshot_generation());
|
||||
assert!(updates[1].generation > updates[0].generation);
|
||||
let reset = decode_recorded_payload(&updates[1], &schema);
|
||||
assert_eq!(reset.to_string(), "true");
|
||||
receiver.update(reset).unwrap();
|
||||
assert!(
|
||||
receiver.snapshot_generation() > narrow_generation,
|
||||
"the receiver generation must advance for the reset"
|
||||
);
|
||||
assert_eq!(receiver.current().unwrap().to_string(), "true");
|
||||
|
||||
// A representable generation after the reset is still published as the producer
|
||||
// wrote it, and the reset is not repeated for the same generation.
|
||||
registry
|
||||
.fanout_snapshot(&handler_ref, &entry, filter.as_ref(), false)
|
||||
.await;
|
||||
assert_eq!(handler.updates().len(), 2);
|
||||
filter.update(lit(false) as _).unwrap();
|
||||
registry
|
||||
.fanout_snapshot(&handler_ref, &entry, filter.as_ref(), false)
|
||||
.await;
|
||||
let updates = handler.updates();
|
||||
assert_eq!(updates.len(), 3);
|
||||
assert_eq!(updates[2].generation, filter.snapshot_generation());
|
||||
assert_eq!(
|
||||
decode_recorded_payload(&updates[2], &schema).to_string(),
|
||||
"false"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fanout_publishes_producer_supplied_true_after_restrictive_generation() {
|
||||
let query_id = test_query_id(12);
|
||||
let registry = Arc::new(QueryDynFilterRegistry::new(query_id));
|
||||
let filter = test_dyn_filter(&["host"]);
|
||||
let filter_id = test_filter_id(1);
|
||||
let entry = match registry.register_remote_dyn_filter(filter_id.clone(), filter.clone()) {
|
||||
EntryRegistration::Inserted(entry) => entry,
|
||||
other => panic!("unexpected registration result: {other:?}"),
|
||||
};
|
||||
let subscriber = Subscriber::new(RegionId::new(1024, 7), test_target(1));
|
||||
assert_eq!(
|
||||
registry.register_subscriber(&filter_id, subscriber),
|
||||
SubscriberRegistration::Added
|
||||
);
|
||||
|
||||
let handler = Arc::new(RecordingRegionQueryHandler::default());
|
||||
let handler_ref = handler.clone() as RegionQueryHandlerRef;
|
||||
let schema = host_schema();
|
||||
let receiver = Arc::new(DynamicFilterPhysicalExpr::new(
|
||||
vec![Arc::new(Column::new("host", 0)) as _],
|
||||
lit(true) as _,
|
||||
));
|
||||
|
||||
// A restrictive generation reaches the region and is applied there.
|
||||
filter.update(lit(false) as _).unwrap();
|
||||
registry
|
||||
.fanout_snapshot(&handler_ref, &entry, filter.as_ref(), false)
|
||||
.await;
|
||||
let updates = handler.updates();
|
||||
assert_eq!(updates.len(), 1);
|
||||
let restrictive = decode_recorded_payload(&updates[0], &schema);
|
||||
assert_eq!(restrictive.to_string(), "false");
|
||||
receiver.update(restrictive).unwrap();
|
||||
let restrictive_generation = receiver.snapshot_generation();
|
||||
assert_eq!(receiver.current().unwrap().to_string(), "false");
|
||||
|
||||
// A `true` the *producer* itself publishes is a real predicate and is published
|
||||
// normally - it is not the transport reset, which only exists for generations the
|
||||
// wire cannot carry, and it is not a synthesized fallback either.
|
||||
filter.update(lit(true) as _).unwrap();
|
||||
registry
|
||||
.fanout_snapshot(&handler_ref, &entry, filter.as_ref(), false)
|
||||
.await;
|
||||
let updates = handler.updates();
|
||||
assert_eq!(updates.len(), 2);
|
||||
assert_eq!(updates[1].generation, filter.snapshot_generation());
|
||||
assert!(updates[1].generation > updates[0].generation);
|
||||
let widened = decode_recorded_payload(&updates[1], &schema);
|
||||
assert_eq!(widened.to_string(), "true");
|
||||
receiver.update(widened).unwrap();
|
||||
assert!(
|
||||
receiver.snapshot_generation() > restrictive_generation,
|
||||
"the receiver generation must advance for a producer-supplied `true`"
|
||||
);
|
||||
assert_eq!(receiver.current().unwrap().to_string(), "true");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fanout_task_waits_for_dynamic_filter_notifications() {
|
||||
let query_id = test_query_id(3);
|
||||
@@ -1590,6 +1857,75 @@ mod tests {
|
||||
wait_for_registry_drop(registry_weak).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fanout_task_resets_unsupported_final_generation_and_cleans_up_once() {
|
||||
let query_id = test_query_id(13);
|
||||
let manager = Arc::new(DynFilterRegistryManager::default());
|
||||
let lease = manager.acquire_lease(query_id);
|
||||
let registry_weak = Arc::downgrade(lease.registry.as_ref().unwrap());
|
||||
let filter = test_dyn_filter(&["host"]);
|
||||
// A narrow, representable generation is the predicate the region starts with.
|
||||
filter.update(lit(false) as _).unwrap();
|
||||
let filter_id = test_filter_id(1);
|
||||
let _ = lease
|
||||
.registry()
|
||||
.register_remote_dyn_filter(filter_id.clone(), filter.clone());
|
||||
let subscriber = Subscriber::new(RegionId::new(1024, 7), test_target(1));
|
||||
assert_eq!(
|
||||
lease.registry().register_subscriber(&filter_id, subscriber),
|
||||
SubscriberRegistration::Added
|
||||
);
|
||||
|
||||
let handler = Arc::new(RecordingRegionQueryHandler::default());
|
||||
lease.ensure_fanout_task(handler.clone() as RegionQueryHandlerRef);
|
||||
|
||||
// The supported initial generation installs the narrow predicate on the region.
|
||||
handler.wait_for_update_count(1).await;
|
||||
assert_eq!(
|
||||
decode_recorded_payload(&handler.updates()[0], &host_schema()).to_string(),
|
||||
"false"
|
||||
);
|
||||
|
||||
// The final generation has no wire form, so the stale `false` must be invalidated
|
||||
// instead of being kept: an explicit `true` reset is published at that generation
|
||||
// (complete, if the completion was already visible), and the fanout still cleans up
|
||||
// exactly once. Whether the update and the completion are observed in one snapshot
|
||||
// or two is racy, so both shapes are accepted.
|
||||
filter.update(unsupported_lookup_predicate() as _).unwrap();
|
||||
filter.mark_complete();
|
||||
handler.wait_for_complete_update().await;
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
drop(lease);
|
||||
handler.wait_for_unregister_count(1).await;
|
||||
|
||||
let schema = host_schema();
|
||||
let updates = handler.updates();
|
||||
assert!(updates.len() >= 2, "the final generation must be reset");
|
||||
for reset in &updates[1..] {
|
||||
assert_eq!(
|
||||
decode_recorded_payload(reset, &schema).to_string(),
|
||||
"true",
|
||||
"every snapshot after the narrow one must be the true reset"
|
||||
);
|
||||
assert_eq!(
|
||||
reset.generation,
|
||||
filter.snapshot_generation(),
|
||||
"the reset must carry the current generation"
|
||||
);
|
||||
}
|
||||
assert!(
|
||||
updates.last().unwrap().is_complete,
|
||||
"the complete final generation must be marked complete"
|
||||
);
|
||||
assert_eq!(
|
||||
handler.unregisters().len(),
|
||||
1,
|
||||
"the unsupported final generation must still clean up exactly once"
|
||||
);
|
||||
wait_for_registry_drop(registry_weak).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fanout_task_unregisters_after_lifecycle_close_during_blocked_update() {
|
||||
let query_id = test_query_id(5);
|
||||
|
||||
@@ -121,12 +121,27 @@ impl Predicate {
|
||||
|
||||
/// Returns the dynamic filter physical exprs. Notice this return a live dynamic filters which
|
||||
/// can change during query execution.
|
||||
///
|
||||
/// # Precondition: pruning only, never row filtering
|
||||
///
|
||||
/// A consumer may only use these expressions to *prune*: to skip a region, a file or a
|
||||
/// row group whose statistics prove that no row can match, as
|
||||
/// [`Self::prune_with_stats`] does. A remote dynamic filter is a bounded pruning hint,
|
||||
/// not a copy of the producer's predicate — the coordinator ships the join's build-key
|
||||
/// list as `IN`/`NOT IN` and drops the top-level conjuncts it cannot ship, so the
|
||||
/// predicate can be *wider* than the one the producer evaluates (see
|
||||
/// `DynFilterPayload::decode_datafusion_expr` in `common-query`). Over-approximating is
|
||||
/// free for pruning but loses rows for filtering, so row filtering must not consume
|
||||
/// these expressions unconditionally.
|
||||
pub fn dyn_filters(&self) -> Arc<Vec<Arc<DynamicFilterPhysicalExpr>>> {
|
||||
self.dyn_filters.load_full()
|
||||
}
|
||||
|
||||
/// Returns the dynamic filter as physical exprs. Notice this return a "snapshot" of
|
||||
/// dynamic filters at the time of calling this method.
|
||||
///
|
||||
/// Pruning only: see [`Self::dyn_filters`] for why these expressions must not filter
|
||||
/// rows.
|
||||
pub fn dyn_filter_phy_exprs(&self) -> error::Result<Vec<Arc<dyn PhysicalExpr>>> {
|
||||
self.dyn_filters
|
||||
.load()
|
||||
|
||||
Reference in New Issue
Block a user