diff --git a/Cargo.lock b/Cargo.lock index 06a20b7dc33..c80f9d761a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 013ec37f4fa..1e327fc30d4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" } diff --git a/src/common/query/Cargo.toml b/src/common/query/Cargo.toml index 4acdfafcb6c..9cdd46c6aa5 100644 --- a/src/common/query/Cargo.toml +++ b/src/common/query/Cargo.toml @@ -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 diff --git a/src/common/query/src/error.rs b/src/common/query/src/error.rs index 511f0a82db3..f5f880e8ced 100644 --- a/src/common/query/src/error.rs +++ b/src/common/query/src/error.rs @@ -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 = std::result::Result; @@ -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::(&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 { diff --git a/src/common/query/src/request.rs b/src/common/query/src/request.rs index c682cf81e02..2d68133d283 100644 --- a/src/common/query/src/request.rs +++ b/src/common/query/src/request.rs @@ -14,17 +14,21 @@ mod base64_serde; mod initial_remote_dyn_filter_reg; +#[cfg(any(test, feature = "testing"))] +pub mod remote_membership_test_support; use std::cell::Cell; use std::sync::Arc; use api::v1::region::RegionRequestHeader; +use datafusion::arrow::array::{Array, ArrayRef, StringArray}; use datafusion::execution::TaskContext; -use datafusion::physical_expr::expressions::{Column, InListExpr, lit}; +use datafusion::physical_expr::expressions::{Column, InListExpr, Literal}; +use datafusion::physical_expr::{conjunction, split_conjunction}; use datafusion::physical_plan::joins::HashTableLookupExpr; use datafusion::physical_plan::{ExecutionPlan, PhysicalExpr}; -use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; -use datafusion_common::{DataFusionError, Result as DataFusionResult}; +use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; +use datafusion_common::{DataFusionError, Result as DataFusionResult, ScalarValue}; use datafusion_expr::LogicalPlan; use datafusion_proto::physical_plan::from_proto::parse_physical_expr; use datafusion_proto::physical_plan::to_proto::{ @@ -34,7 +38,8 @@ use datafusion_proto::physical_plan::{ DefaultPhysicalExtensionCodec, DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext, PhysicalProtoConverterExtension, }; -use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode}; +use datafusion_proto::protobuf::physical_expr_node::ExprType as PhysicalExprNodeType; +use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalInListNode, PhysicalPlanNode}; use prost::Message; use serde::{Deserialize, Serialize}; use snafu::ensure; @@ -45,12 +50,46 @@ pub use self::initial_remote_dyn_filter_reg::{ INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, InitialDynFilterReg, InitialDynFilterRegs, InitialDynFilterSnapshot, }; -use crate::error::{DynFilterPayloadTooLargeSnafu, Error as CommonQueryError}; +use crate::error::{ + DynFilterPayloadTooLargeSnafu, DynFilterUnsupportedSnafu, Error as CommonQueryError, +}; pub const DYN_FILTER_PROTOCOL_VERSION: u32 = 1; /// Shared raw protobuf byte budget for remote dynamic filter payloads. pub const REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES: usize = 512 * 1024; +/// Smallest number of encoded bytes a single `IN` list literal can take. +/// +/// Every list entry is a `PhysicalExprNode` whose `literal` arm holds the +/// [`ScalarValue`] message itself, so the oneof field tag plus its length prefix are +/// already 2 bytes even for an empty `Utf8` value (whose `ScalarValue` body may encode to +/// nothing at all). Protobuf sizes add up across nesting, so a payload of `B` bytes can +/// never hold more than `B / 2` literals. This is the floor the value bound is derived +/// from; it is not an encoder policy constant. +const MIN_ENCODED_LITERAL_BYTES: usize = 2; + +/// Upper bound on how many `IN` list literals `max_payload_bytes` can hold. +/// +/// The count bound is derived from the transport budget with +/// [`MIN_ENCODED_LITERAL_BYTES`] rather than being an arbitrary cap: a larger budget +/// admits proportionally more build keys, and a small budget refuses the membership +/// before a single literal is built. +fn max_membership_values(max_payload_bytes: usize) -> usize { + max_payload_bytes / MIN_ENCODED_LITERAL_BYTES +} + +/// Frame size reported for a payload refused because it does not fit the budget. +/// +/// A refused payload is never materialized, so there is no measured frame length. The +/// error reports the largest lower bound that is known instead: the price of the literals +/// the refused membership would have written (`lower_bound`), or - when nothing about the +/// shape is known - one byte over the budget. Either way the number is genuinely above +/// `max_payload_bytes`; it is a bound, never a measurement, and it is only used for +/// diagnostics. +fn refused_payload_size(lower_bound: usize, max_payload_bytes: usize) -> usize { + lower_bound.max(max_payload_bytes.saturating_add(1)) +} + /// Serialized predicate payload for remote dynamic filter updates. /// /// The payload is tagged in JSON so receivers can reject unsupported encodings @@ -72,20 +111,98 @@ pub enum DynFilterPayload { impl DynFilterPayload { /// Encodes a DataFusion physical expression into a bounded dynamic filter payload. /// - /// Runtime-only hash lookup predicates are degraded to `true` before encoding so - /// serializable min/max bounds around them can still be shipped to remote scans. - /// If the full serializable predicate is still larger than `max_payload_bytes`, large - /// membership predicates (`IN (...)`) are also degraded to `true` as a bounds-only fallback. + /// # Narrow contract: a key list used as a positive pruning hint + /// + /// A runtime [`HashTableLookupExpr`] is written as the existing + /// `PhysicalInListNode` — `key IN (build key, ...)` — built from the lookup's + /// published build-key list, so no new wire variant is needed. The expression tree + /// around it is preserved node for node: `IS NULL`, `NOT`, `CASE`, `AND`, `OR` and + /// every other parent are serialized as-is, and a `NOT` over a lookup therefore + /// becomes `NOT IN`. + /// + /// The membership is a **hint for pruning only**, not a re-statement of the join + /// semantics. For supported positive pruning, the list preserves actual build-key + /// matches; it is not universal Boolean equivalence. `IN`/`NOT IN` propagate + /// `NULL` while a hash lookup can be definite on a `NULL` probe key, and hash + /// presence can admit collisions, so a predicate that inspects the *negation* of a + /// lookup is not universally equivalent. + /// + /// The conversion is decided by the lookup alone (single-column `Utf8`, non-empty, + /// `NULL`-free build keys — see [`HashTableLookupExpr::build_key_membership`]). + /// Exactly one node changes: every [`HashTableLookupExpr`] is written as its key + /// list, wherever it sits in the tree (`NOT lookup` included, which is why it + /// becomes `NOT IN`). No node is ever replaced by a *weaker stand-in* — in + /// particular never by `true` — and no membership is dropped from inside a + /// predicate: when a lookup cannot be written at all, the whole top-level `AND` + /// conjunct that holds it is dropped (see below) or the payload is rejected. + /// + /// # Top-level conjunctive retention + /// + /// If a lookup has no such key list (accessor limitation) or the serialized predicate + /// does not fit `max_payload_bytes`, [`split_conjunction`] splits the original + /// expression and the whole conjuncts that cannot be shipped are dropped. Dropping an + /// `AND` term can only widen a predicate, so the result never drops a row the producer + /// would keep. Which conjuncts go depends on why the payload was refused: + /// + /// - a lookup has no wire form: only conjuncts holding a [`HashTableLookupExpr`] are + /// dropped. A static [`InListExpr`] is representable, so `lookup AND id IN (...)` + /// keeps the list and stays `id IN (...)`; + /// - the payload is over budget: conjuncts holding a [`HashTableLookupExpr`] *or* an + /// [`InListExpr`] are dropped, because a static `IN` list is the node that usually + /// blows the budget. `bounds AND oversized IN (...)` becomes `bounds`. + /// + /// Either way a conjunct is kept or dropped as a whole: a membership nested inside + /// `NOT`, `CASE` or `OR` drops that entire conjunct instead of being rewritten. + /// Retained conjuncts are encoded again under the same budget, and when the first two + /// rules still leave an over-budget payload the second one is applied as well, so a + /// static `IN` list that is too large for the budget is dropped last. + /// + /// # Encoder-level no-substitution rule + /// + /// This encoder never replaces a node in the producer expression with `true`: it + /// serializes the expression, retries after dropping whole top-level conjuncts under + /// the retention rules above, or returns an error. A caller decides what a refused + /// generation means for its receiver. Initial capture keeps the parent filter local; + /// the long-lived query fanout path + /// (`remote_dyn_filter_registry::fanout_snapshot_for_query`) publishes an explicit + /// `true` reset to invalidate a stale region-side predicate. That reset is not a + /// representation of the refused producer expression. + /// + /// The two refusals stay distinct so callers can tell a shape problem from a size + /// problem: + /// + /// - [`CommonQueryError::DynFilterUnsupported`]: no conjunct can be retained because + /// every one of them holds a lookup with no UTF-8 key list (`NOT lookup`, `CASE` / + /// `OR` over such a membership). The wire has no form for the predicate. + /// - [`CommonQueryError::DynFilterPayloadTooLarge`]: the predicate is representable but + /// does not fit `max_payload_bytes` - either a key list itself exceeds the budget, or + /// the retained conjuncts still do not fit. An over-budget key list is never reported + /// as unsupported; it is a size refusal and uses + /// [`ConjunctRetention::OverBudget`]. + /// + /// These refusals do not fail the query in either caller: initial capture leaves the + /// parent filter local, while fanout sends an explicit `true` reset instead of leaving + /// a stale narrower predicate installed. pub fn from_datafusion_expr( expr: &Arc, max_payload_bytes: usize, ) -> DataFusionResult { - match encode_remote_dyn_filter_expr(expr, max_payload_bytes, false) { + match encode_bounded_physical_expr_to_bytes(expr, max_payload_bytes) { Ok(bytes) => Ok(Self::Datafusion(bytes)), + // The predicate cannot be shipped as it stands: either a lookup has no key + // list, or the frame does not fit the budget. Keep the widest *safe* part of + // the predicate instead of publishing an always-true filter. + Err(CommonQueryError::DynFilterUnsupported { .. }) => retain_membership_free_conjuncts( + expr, + max_payload_bytes, + ConjunctRetention::UnrepresentableLookup, + ), Err(CommonQueryError::DynFilterPayloadTooLarge { .. }) => { - encode_remote_dyn_filter_expr(expr, max_payload_bytes, true) - .map(Self::Datafusion) - .map_err(DataFusionError::from) + retain_membership_free_conjuncts( + expr, + max_payload_bytes, + ConjunctRetention::OverBudget, + ) } Err(error) => Err(DataFusionError::from(error)), } @@ -96,6 +213,22 @@ impl DynFilterPayload { /// The decoded expression is validated to ensure column indexes stay within the receiver-side /// schema, column names are consistent (defensive check), and the payload stays within /// `max_payload_bytes`. + /// + /// # Precondition: pruning only, never row filtering + /// + /// The decoded predicate is only a **pruning hint**. The sole production consumer — + /// `Predicate::dyn_filters` (in the `table` crate) and its pruning path, reached by + /// mito2's row-group pruner — uses it to skip containers whose statistics prove that + /// no row can match, so a predicate that is merely *wider* than the producer's costs + /// nothing but a wasted scan. + /// + /// The payload is not a re-statement of the producer predicate: the membership is the + /// join's build-key list, so `IN`/`NOT IN` `NULL` propagation and hash collisions can + /// make it disagree with a runtime lookup, and the encoder drops whole top-level + /// conjuncts it cannot ship (see [`DynFilterPayload::from_datafusion_expr`]). A + /// consumer that *filters rows* would turn every such widening into lost data, so row + /// filtering must not consume this payload unconditionally: it would need to prove the + /// predicate is equivalent for the rows it drops, which this format does not carry. pub fn decode_datafusion_expr( &self, task_ctx: &TaskContext, @@ -129,6 +262,141 @@ fn encode_physical_expr_to_bytes(expr: &Arc) -> DataFusionResu struct BoundedPhysicalExprEncoder { max_payload_bytes: usize, encoded_bytes: Cell, + /// Set when the encoding aborted on a runtime lookup that has no serializable key + /// list. + /// + /// The abort travels out as an ordinary [`DataFusionError`], which would lose the + /// typed dynamic filter error on the way, so the flag is what lets + /// `encode_bounded_physical_expr_to_bytes` rebuild + /// [`CommonQueryError::DynFilterUnsupported`]. No placeholder is ever encoded in its + /// place. + unsupported: Cell, + /// Set to the reported frame size when the encoding aborted because a build-key + /// membership does not fit `max_payload_bytes`. + /// + /// This is a size refusal, not a shape one, so + /// `encode_bounded_physical_expr_to_bytes` rebuilds + /// [`CommonQueryError::DynFilterPayloadTooLarge`] from it instead of reporting the + /// predicate as having no wire form. + over_budget: Cell>, +} + +impl BoundedPhysicalExprEncoder { + fn new(max_payload_bytes: usize) -> Self { + Self { + max_payload_bytes, + encoded_bytes: Cell::new(0), + unsupported: Cell::new(false), + over_budget: Cell::new(None), + } + } + + /// Charges `proto` (including its encoded children) against the payload budget. + /// + /// The accounting is additive over siblings and exact for the top-level node, so + /// no branch can push the payload over budget unnoticed. + fn finish( + &self, + before_children: usize, + proto: PhysicalExprNode, + ) -> DataFusionResult { + let encoded_bytes = before_children.saturating_add(proto.encoded_len()); + self.encoded_bytes.set(encoded_bytes); + validate_payload_size(encoded_bytes, self.max_payload_bytes) + .map_err(DataFusionError::from)?; + Ok(proto) + } + + /// Encodes a lookup as its build-key membership, charging every list literal and + /// node against the payload budget. + /// + /// The decision only looks at the lookup itself: if DataFusion can hand out the + /// single-column `Utf8` build-key list it is written as `key IN (...)`; otherwise the + /// whole payload aborts. A lookup is a real predicate that DataFusion only degrades to + /// `true` in its own proto encoder, so this transport must not write any substitute for + /// it, and the abort keeps the two reasons apart: no key list at all (unsupported) is + /// not the same thing as a key list that exceeds the budget (too large). + fn membership_to_proto( + &self, + lookup: &HashTableLookupExpr, + codec: &dyn PhysicalExtensionCodec, + ) -> DataFusionResult { + let membership = match RemoteMembership::from_lookup(lookup, self.max_payload_bytes) { + Ok(membership) => membership, + // A lookup without a key list has no wire form at all: that is a shape + // refusal, not a size one. + Err(MembershipRefusal::NoKeyList) => return Err(self.abort_unsupported_lookup()), + // The key list exists but does not fit the budget: report it as a size + // refusal so it is not counted as "no wire form". + Err(MembershipRefusal::OverBudget { build_keys }) => { + return Err(self.abort_over_budget_membership(build_keys)); + } + }; + // The count bound is enforced once more before any literal exists, so a build side + // larger than the budget can hold is never materialized. This cannot fail for any + // other reason here: the accessor already guaranteed a non-empty, `NULL`-free + // `Utf8` list, so a refusal is again a size refusal. + let Some(literals) = utf8_list_literals( + &membership.values, + max_membership_values(self.max_payload_bytes), + ) else { + return Err(self.abort_over_budget_membership(membership.values.len())); + }; + + let before_children = self.encoded_bytes.get(); + let mut list = Vec::with_capacity(literals.len()); + for literal in literals { + list.push(self.physical_expr_to_proto(&literal, codec)?); + } + + // The `InList` proto is built by hand instead of going through the generic + // serializer: that path needs an `InListExpr`, whose only constructors + // (`InListExpr::try_new` / `try_new_from_array`) require an input `Schema` to + // resolve the key expression's data type, and this encoder converts a bare + // expression without one. The key is known to be `Utf8` (see + // `HashTableLookupExpr::build_key_membership`), which is exactly what the list + // below writes, so no type check is lost. + let proto = PhysicalExprNode { + expr_id: None, + expr_type: Some(PhysicalExprNodeType::InList(Box::new(PhysicalInListNode { + expr: Some(Box::new( + self.physical_expr_to_proto(&membership.key, codec)?, + )), + list, + negated: false, + }))), + }; + + self.finish(before_children, proto) + } + + /// Aborts the payload because the wire has no form for `lookup`. + /// + /// The returned error only carries the abort; [`Self::unsupported`] is what the caller + /// reads to rebuild the typed dynamic filter error. + fn abort_unsupported_lookup(&self) -> DataFusionError { + self.unsupported.set(true); + DataFusionError::NotImplemented( + "hash table lookup has no UTF-8 build-key membership to encode".to_string(), + ) + } + + /// Aborts the payload because a build-key membership does not fit `max_payload_bytes`. + /// + /// This is a *size* refusal, never a "no wire form" one, so the caller rebuilds + /// [`CommonQueryError::DynFilterPayloadTooLarge`] from [`Self::over_budget`] instead of + /// [`CommonQueryError::DynFilterUnsupported`]. `build_keys` is the number of build keys + /// the refused list would have written; every one of them costs at least + /// [`MIN_ENCODED_LITERAL_BYTES`], so that product is a real lower bound of the frame. + fn abort_over_budget_membership(&self, build_keys: usize) -> DataFusionError { + self.over_budget.set(Some(refused_payload_size( + build_keys.saturating_mul(MIN_ENCODED_LITERAL_BYTES), + self.max_payload_bytes, + ))); + DataFusionError::NotImplemented( + "hash table lookup build-key membership exceeds the payload budget".to_string(), + ) + } } impl PhysicalProtoConverterExtension for BoundedPhysicalExprEncoder { @@ -162,55 +430,268 @@ impl PhysicalProtoConverterExtension for BoundedPhysicalExprEncoder { expr: &Arc, codec: &dyn PhysicalExtensionCodec, ) -> DataFusionResult { + // A real `HashTableLookupExpr` has no wire form: DataFusion's own + // `try_to_proto` degrades it to `true`. Intercept it so the build-key list can + // be written with the existing `InList` node instead. The decision is made by + // the lookup alone; every other node, including `IS NULL`, `NOT`, `CASE`, + // `AND` and `OR`, is serialized exactly as it is. + if let Some(lookup) = expr.downcast_ref::() { + return self.membership_to_proto(lookup, codec); + } + let before_children = self.encoded_bytes.get(); let proto = serialize_physical_expr_with_converter(expr, codec, self)?; // Replace the children's contribution with the whole subtree, including protobuf framing. // Checking each child stops large IN lists before materializing all their Arrow IPC literals. - let encoded_bytes = before_children.saturating_add(proto.encoded_len()); - self.encoded_bytes.set(encoded_bytes); - validate_payload_size(encoded_bytes, self.max_payload_bytes) - .map_err(DataFusionError::from)?; - Ok(proto) + self.finish(before_children, proto) } } +/// Encodes `expr` for the remote wire under `max_payload_bytes`. +/// +/// Returns the typed [`CommonQueryError::DynFilterUnsupported`] when the predicate holds a +/// lookup whose key list cannot be serialized, and +/// [`CommonQueryError::DynFilterPayloadTooLarge`] when the frame (or a build-key +/// membership) does not fit the budget. The two refusals stay distinct: an over-budget +/// membership is a size refusal and is never reported as a predicate without a wire form. +/// The caller answers both with [`retain_membership_free_conjuncts`]; neither is ever +/// replaced with `true`. fn encode_bounded_physical_expr_to_bytes( expr: &Arc, max_payload_bytes: usize, ) -> Result, CommonQueryError> { let codec = DefaultPhysicalExtensionCodec {}; - let encoder = BoundedPhysicalExprEncoder { - max_payload_bytes, - encoded_bytes: Cell::new(0), - }; + let encoder = BoundedPhysicalExprEncoder::new(max_payload_bytes); let proto = encoder.physical_expr_to_proto(expr, &codec); - // Preserve the size error so the caller can retry with bounds only. + // A build-key membership that exceeds the budget is a size refusal even though the + // frame that would have held it is never materialized. + if let Some(payload_size_bytes) = encoder.over_budget.get() { + return Err(DynFilterPayloadTooLargeSnafu { + payload_size_bytes, + max_payload_bytes, + } + .build()); + } + let unsupported = encoder.unsupported.get(); + // The encoder stops at the node that went over budget, so this reports the size + // error unwrapped instead of the wrapped DataFusion error. validate_payload_size(encoder.encoded_bytes.get(), max_payload_bytes)?; + if unsupported { + return Err(DynFilterUnsupportedSnafu { + reason: "the predicate holds a hash table lookup without a UTF-8 build-key \ + membership", + } + .build()); + } Ok(proto.map_err(CommonQueryError::from)?.encode_to_vec()) } -fn encode_remote_dyn_filter_expr( - expr: &Arc, - max_payload_bytes: usize, - bounds_only: bool, -) -> Result, CommonQueryError> { - let expr = portable_remote_dyn_filter_expr(Arc::clone(expr), bounds_only) - .map_err(CommonQueryError::from)?; - encode_bounded_physical_expr_to_bytes(&expr, max_payload_bytes) +/// Which top-level conjuncts [`retain_membership_free_conjuncts`] has to drop. +#[derive(Clone, Copy, Debug)] +enum ConjunctRetention { + /// The payload holds a [`HashTableLookupExpr`] without a wire form, so such a lookup + /// is the only unshippable node: a static [`InListExpr`] is representable and stays. + UnrepresentableLookup, + /// The payload exceeds its budget. A static [`InListExpr`] is the node most likely to + /// be the oversized one, so it is dropped with the lookups. + OverBudget, } -fn portable_remote_dyn_filter_expr( - expr: Arc, - bounds_only: bool, -) -> DataFusionResult> { - expr.transform_up(|node| { - if node.is::() || (bounds_only && node.is::()) { - Ok(Transformed::yes(lit(true))) - } else { - Ok(Transformed::no(node)) +/// Keeps the top-level conjuncts of a predicate that cannot be shipped as it stands. +/// +/// Dropping an `AND` term can only widen a predicate, so the retained payload still never +/// drops a row the producer would keep. `retention` decides which conjuncts have to go. It +/// only widens as far as it has to: when the narrow retention leaves conjuncts that still do +/// not fit `max_payload_bytes`, the [`InListExpr`]-bearing ones are dropped as well, because +/// a static `IN` list is the node most likely to be the oversized one. +/// +/// The refusal keeps the two reasons apart instead of folding them into "no wire form": +/// nothing retained under [`ConjunctRetention::UnrepresentableLookup`] means the wire +/// really has no form for the predicate +/// ([`CommonQueryError::DynFilterUnsupported`]), while nothing retained under +/// [`ConjunctRetention::OverBudget`] (or a retained payload that still does not fit) is a +/// size refusal ([`CommonQueryError::DynFilterPayloadTooLarge`]). Transport never +/// substitutes `true` for either. +fn retain_membership_free_conjuncts( + expr: &Arc, + max_payload_bytes: usize, + retention: ConjunctRetention, +) -> DataFusionResult { + match encode_retained_conjuncts_to_bytes(expr, max_payload_bytes, retention) { + Ok(Some(bytes)) => Ok(DynFilterPayload::Datafusion(bytes)), + // Nothing could be retained. Which refusal that is depends on *why* the conjuncts + // had to go, and the two must not be mixed up: an over-budget predicate is + // representable, it only does not fit. + Ok(None) => Err(match retention { + ConjunctRetention::UnrepresentableLookup => DynFilterUnsupportedSnafu { + reason: "every top-level conjunct of the predicate holds a hash table \ + lookup without a UTF-8 build-key membership, so the wire has no \ + form for it", + } + .build() + .into(), + ConjunctRetention::OverBudget => DynFilterPayloadTooLargeSnafu { + payload_size_bytes: refused_payload_size(0, max_payload_bytes), + max_payload_bytes, + } + .build() + .into(), + }), + Err(CommonQueryError::DynFilterPayloadTooLarge { .. }) + if matches!(retention, ConjunctRetention::UnrepresentableLookup) => + { + retain_membership_free_conjuncts(expr, max_payload_bytes, ConjunctRetention::OverBudget) } - }) - .map(|transformed| transformed.data) + Err(error) => Err(error.into()), + } +} + +/// Encodes the top-level conjuncts of `expr` that `retention` allows to keep. +/// +/// [`split_conjunction`] hands out the top-level `AND` terms of the original +/// expression; a term is dropped as a whole when its subtree holds a node the retention +/// cannot ship (see [`ConjunctRetention`]). Membership nested inside `NOT`, `CASE` or `OR` +/// therefore drops that whole conjunct instead of being rewritten in place, so the payload +/// only ever loses top-level `AND` terms, which can only widen a predicate. +/// +/// The retained terms are encoded again under the same `max_payload_bytes`, so a caller +/// supplied budget still bounds the frame. Returns `Ok(None)` when no conjunct can be +/// retained; the caller decides which typed refusal that is from `retention`, because an +/// empty retention caused by an unrepresentable lookup is not the same thing as an empty +/// retention caused by the budget. +fn encode_retained_conjuncts_to_bytes( + expr: &Arc, + max_payload_bytes: usize, + retention: ConjunctRetention, +) -> Result>, CommonQueryError> { + let mut retained = Vec::new(); + for conjunct in split_conjunction(expr) { + if !retention.drops_conjunct(conjunct)? { + retained.push(Arc::clone(conjunct)); + } + } + if retained.is_empty() { + return Ok(None); + } + encode_bounded_physical_expr_to_bytes(&conjunction(retained), max_payload_bytes).map(Some) +} + +impl ConjunctRetention { + /// Whether the whole subtree of `expr` holds a node this retention cannot ship. + /// + /// This is a "whole conjunct" test on purpose: a conjunct that reaches such a node + /// anywhere is dropped by the retention instead of being rewritten around it. + fn drops_conjunct(self, expr: &Arc) -> DataFusionResult { + let mut found = false; + expr.apply(|node| { + let unshippable = match self { + Self::UnrepresentableLookup => node.is::(), + Self::OverBudget => node.is::() || node.is::(), + }; + if unshippable { + found = true; + Ok(TreeNodeRecursion::Stop) + } else { + Ok(TreeNodeRecursion::Continue) + } + })?; + Ok(found) + } +} + +/// Build-key list of one hash lookup, written as `probe_key IN (value, ...)`. +/// +/// The membership reuses the existing `PhysicalInListNode`, so no new wire variant +/// and no in-payload version tag are needed, and it is charged against the normal +/// payload budget. Only the build-key list itself is consulted here; the wire form +/// never depends on where the lookup sits in the predicate tree. +struct RemoteMembership { + /// Probe-side key expression, reused as the `InList` operand. + key: Arc, + /// Published build key values, tied to the lookup's build lease. + values: ArrayRef, +} + +impl RemoteMembership { + /// Key-list membership of `lookup`, or the reason it cannot be written. + /// + /// The count bound handed to DataFusion is derived from the byte budget (see + /// [`max_membership_values`]) instead of being a policy constant, and the byte bound + /// is the raw payload budget itself, which every build-key buffer has to fit before it + /// can be written as literals. Both are conservative early guards, not the exact + /// serialized frame length: the encoder still charges every literal against the budget + /// node by node. + /// + /// The accessor keeps the value-level limitations — it returns `None` for lookups + /// without a build lease (e.g. a test-only lookup), composite keys, non-`Utf8` keys + /// (integer/float/decimal/dictionary/unknown), and `NULL`-containing or empty build + /// keys — and [`utf8_list_literals`] re-checks everything it materializes. Those are + /// accessor and materialization limitations, not encoder policy. + /// + /// The accessor folds its shape and size checks into one `None`, so a key list that + /// exists but exceeds the bounds is told apart by asking again with the bounds lifted: + /// only a list that exists at all can be refused for its size, and a size refusal must + /// not be reported as a predicate without a wire form (it is + /// [`CommonQueryError::DynFilterPayloadTooLarge`], not + /// [`CommonQueryError::DynFilterUnsupported`]). + fn from_lookup( + lookup: &HashTableLookupExpr, + max_payload_bytes: usize, + ) -> Result { + match lookup + .build_key_membership(max_membership_values(max_payload_bytes), max_payload_bytes) + { + Some(membership) => Ok(Self { + key: Arc::clone(membership.key()), + values: Arc::clone(membership.values()), + }), + None => Err(match lookup.build_key_membership(usize::MAX, usize::MAX) { + Some(unbounded) => MembershipRefusal::OverBudget { + build_keys: unbounded.values().len(), + }, + None => MembershipRefusal::NoKeyList, + }), + } + } +} + +/// Why a lookup's build-key membership cannot be written as `probe_key IN (...)`. +/// +/// The two reasons are answered differently: [`Self::NoKeyList`] means the wire has no +/// form for the node, while [`Self::OverBudget`] means the membership is real but too +/// large for the payload budget, which the encoder reports as a size refusal. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum MembershipRefusal { + /// The lookup has no single-column `Utf8` build-key list at all: no build lease, + /// composite key, unsupported key type, or an empty / `NULL`-containing build side. + NoKeyList, + /// The build-key list exists but exceeds the caller's bounds. `build_keys` is how many + /// literals it would have written, measured from the unbounded accessor. + OverBudget { build_keys: usize }, +} + +/// Materializes the `IN` list literals for a build-key membership. +/// +/// This is the authoritative value check: the list is only built for a non-empty, +/// `NULL`-free, non-dictionary `Utf8` array of at most `max_values` entries. `max_values` +/// is checked *before* a [`Literal`] is created for every build key, so an over-large +/// build side is refused instead of being materialized and then measured. The caller has +/// already been through [`RemoteMembership::from_lookup`], so a refusal here can only be +/// the count bound and is reported as a size refusal, not as a missing wire form. +fn utf8_list_literals(values: &ArrayRef, max_values: usize) -> Option>> { + let values = values.as_any().downcast_ref::()?; + if values.is_empty() || values.null_count() != 0 || values.len() > max_values { + return None; + } + values + .iter() + .map(|value| { + value.map(|value| { + Arc::new(Literal::new(ScalarValue::Utf8(Some(value.to_owned())))) + as Arc + }) + }) + .collect() } pub(crate) fn decode_physical_expr_from_bytes( @@ -246,6 +727,78 @@ fn validate_payload_size( Ok(()) } +/// Asserts the predicate is reported as having no wire form instead of becoming `true`. +#[cfg(test)] +fn assert_unsupported(error: DataFusionError) { + assert_typed_dyn_filter_error(error, "DynFilterUnsupported", |error| { + matches!(error, CommonQueryError::DynFilterUnsupported { .. }) + }); +} + +/// Asserts the retained payload still exceeds the caller-supplied budget. +#[cfg(test)] +fn assert_too_large(error: DataFusionError) { + assert_typed_dyn_filter_error(error, "DynFilterPayloadTooLarge", |error| { + matches!(error, CommonQueryError::DynFilterPayloadTooLarge { .. }) + }); +} + +#[cfg(test)] +fn assert_typed_dyn_filter_error( + error: DataFusionError, + expected: &str, + matches_error: impl Fn(&CommonQueryError) -> bool, +) { + let DataFusionError::External(error) = error else { + panic!("expected external common query error, got: {error:?}"); + }; + let typed = error + .downcast_ref::() + .unwrap_or_else(|| panic!("expected external {expected}, got: {error:?}")); + assert!(matches_error(typed), "expected {expected}, got: {error:?}"); +} + +/// Whether `error` reports a dynamic filter predicate the wire cannot represent. +/// +/// Transport never rewrites a producer predicate to `true`, so a producer answers this +/// error by not shipping the producer predicate and keeping the query running. Callers +/// that keep a long-lived region-side predicate treat it as an invalidation trigger: the +/// stale predicate is replaced by an explicit `true` reset (see +/// `remote_dyn_filter_registry`), so a producer that later widens its filter cannot be +/// pruned away by it. +pub fn is_dyn_filter_unsupported(error: &DataFusionError) -> bool { + matches!( + error, + DataFusionError::External(source) + if source + .downcast_ref::() + .is_some_and(CommonQueryError::is_dyn_filter_unsupported) + ) +} + +/// Whether `error` reports a dynamic filter payload that exceeded `max_payload_bytes`. +/// +/// This is the other refusal [`DynFilterPayload::from_datafusion_expr`] reports, next to +/// [`is_dyn_filter_unsupported`]: the predicate exists and is representable, but even the +/// retained top-level conjuncts do not fit the budget. It is *not* a "no wire form" case +/// and must not be reported as one. Like the unsupported case it is answered by not +/// shipping the producer predicate (the query keeps running) and never by narrowing or +/// rewriting it; a caller that keeps a long-lived region predicate publishes the explicit +/// `true` reset instead, so a stale, narrower predicate cannot prune rows the producer +/// still keeps. +pub fn is_dyn_filter_payload_too_large(error: &DataFusionError) -> bool { + matches!( + error, + DataFusionError::External(source) + if source + .downcast_ref::() + .is_some_and(|error| matches!( + error, + CommonQueryError::DynFilterPayloadTooLarge { .. } + )) + ) +} + fn validate_supported_payload_expr(expr: &Arc) -> DataFusionResult<()> { expr.apply(|node| { if node.is::() { @@ -352,14 +905,51 @@ pub struct QueryRequest { pub plan: LogicalPlan, } +/// First node of type `T` anywhere in `expr`. +#[cfg(test)] +fn find_expr(expr: &Arc) -> Option> { + let mut found = None; + expr.apply(|node| { + if node.is::() { + found = Some(Arc::clone(node)); + Ok(TreeNodeRecursion::Stop) + } else { + Ok(TreeNodeRecursion::Continue) + } + }) + .unwrap(); + found +} + +/// Whether `expr` holds a node of type `T` anywhere. +#[cfg(test)] +fn has_expr(expr: &Arc) -> bool { + find_expr::(expr).is_some() +} + +/// Encodes `expr` at `max_payload_bytes` and decodes it against `schema`. This is the +/// round trip a receiver performs, so the tests never assert on the encoder alone. +#[cfg(test)] +fn decode( + expr: &Arc, + schema: &Arc, + max_payload_bytes: usize, +) -> Arc { + DynFilterPayload::from_datafusion_expr(expr, max_payload_bytes) + .unwrap() + .decode_datafusion_expr(&TaskContext::default(), schema, max_payload_bytes) + .unwrap() +} + #[cfg(test)] mod tests { use std::sync::Arc; use base64::Engine; use base64::prelude::BASE64_STANDARD; + use datafusion::arrow::array::{ArrayRef, Int32Array, StringArray}; use datafusion::arrow::datatypes::{DataType, Field, Schema}; - use datafusion::physical_expr::expressions::{BinaryExpr, Column, InListExpr, lit}; + use datafusion::physical_expr::expressions::{BinaryExpr, Column, InListExpr, NotExpr, lit}; use datafusion::physical_plan::expressions::col; use datafusion::physical_plan::joins::join_hash_map::JoinHashMapU32; use datafusion::physical_plan::joins::{HashTableLookupExpr, Map, SeededRandomState}; @@ -507,8 +1097,23 @@ mod tests { assert!(matches!(err, DataFusionError::Plan(_))); } + /// A hand-built lookup has no build lease, hence no key list: the wire cannot + /// represent it, so the encoder aborts unless a membership-free conjunct can be kept. + fn unrepresentable_lookup(key: &Arc) -> Arc { + Arc::new(HashTableLookupExpr::new( + vec![Arc::clone(key)], + SeededRandomState::with_seed(0), + Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(0)))), + "hash_lookup".to_string(), + None, + )) as Arc + } + #[test] - fn dyn_filter_payload_hash_lookup_fallback_preserves_bounds() { + fn dyn_filter_payload_unconvertible_lookup_keeps_integer_bounds() { + // `bounds AND lookup` where the lookup has no key list: only the + // membership-bearing conjunct is dropped, so the integer bounds still travel as a + // regular payload instead of an always-true filter or an error. let schema = Arc::new(Schema::new(vec![Field::new( "device_id", DataType::Int32, @@ -520,30 +1125,224 @@ mod tests { Operator::GtEq, lit(10i32), )) as Arc; - let lookup = Arc::new(HashTableLookupExpr::new( - vec![Arc::clone(&device_id)], - SeededRandomState::with_seed(0), - Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(0)))), - "hash_lookup".to_string(), - None, + let upper_bound = Arc::new(BinaryExpr::new( + Arc::clone(&device_id), + Operator::LtEq, + lit(20i32), + )) as Arc; + let bounds = Arc::new(BinaryExpr::new(lower_bound, Operator::And, upper_bound)) + as Arc; + let expr = Arc::new(BinaryExpr::new( + Arc::clone(&bounds), + Operator::And, + unrepresentable_lookup(&device_id), )) as Arc; - let expr = - Arc::new(BinaryExpr::new(lower_bound, Operator::And, lookup)) as Arc; - let payload = DynFilterPayload::from_datafusion_expr(&expr, 1024).unwrap(); - let decoded = payload - .decode_datafusion_expr(&TaskContext::default(), &schema, 1024) - .unwrap(); + // The predicate itself has no wire form: the lookup has no key list. + assert!(matches!( + encode_bounded_physical_expr_to_bytes(&expr, 1024), + Err(CommonQueryError::DynFilterUnsupported { .. }) + )); - assert!(!contains_expr::(&decoded)); - let decoded_display = decoded.to_string(); - assert!(decoded_display.contains("device_id")); - assert!(decoded_display.contains(">=")); - assert!(!decoded_display.contains("hash_lookup")); + // Keeping the membership-free top-level conjuncts turns it into a normal payload. + let decoded = decode(&expr, &schema, 1024); + let display = decoded.to_string(); + assert!( + display.contains(">= 10"), + "lower bound must survive: {display}" + ); + assert!( + display.contains("<= 20"), + "upper bound must survive: {display}" + ); + assert!(!has_expr::(&decoded), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + // Both bounds are copied verbatim: no `AND` term was rewritten in place. + assert_eq!( + encode_physical_expr_to_bytes(&decoded).unwrap(), + encode_physical_expr_to_bytes(&bounds).unwrap() + ); } #[test] - fn dyn_filter_payload_oversized_inlist_falls_back_to_bounds() { + fn unrepresentable_lookup_drops_only_the_lookup_conjunct() { + // One unsupported lookup (no key list) between the bounds and a static `IN` list + // that is itself perfectly representable: only the lookup-bearing conjunct is + // dropped, so the list survives as a whole and the payload stays as narrow as the + // wire allows. + let schema = Arc::new(Schema::new(vec![Field::new( + "device_id", + DataType::Int32, + false, + )])); + let device_id = col("device_id", &schema).unwrap(); + let lower_bound = Arc::new(BinaryExpr::new( + Arc::clone(&device_id), + Operator::GtEq, + lit(10i32), + )) as Arc; + let upper_bound = Arc::new(BinaryExpr::new( + Arc::clone(&device_id), + Operator::LtEq, + lit(20i32), + )) as Arc; + let bounds = Arc::new(BinaryExpr::new( + Arc::clone(&lower_bound), + Operator::And, + Arc::clone(&upper_bound), + )) as Arc; + let in_list = Arc::new( + InListExpr::try_new( + Arc::clone(&device_id), + (10..20).map(lit).collect(), + false, + &schema, + ) + .unwrap(), + ) as Arc; + let expr = Arc::new(BinaryExpr::new( + Arc::new(BinaryExpr::new( + Arc::clone(&bounds), + Operator::And, + unrepresentable_lookup(&device_id), + )) as Arc, + Operator::And, + Arc::clone(&in_list), + )) as Arc; + + let decoded = decode(&expr, &schema, REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES); + let display = decoded.to_string(); + assert!(display.contains(">= 10"), "{display}"); + assert!(display.contains("<= 20"), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + // The static `IN` list is representable, so it is kept verbatim. + assert!(has_expr::(&decoded), "{display}"); + assert_eq!( + split_conjunction(&decoded).len(), + 3, + "the two bounds and the static `IN` list survive: {display}" + ); + // The retained conjuncts are the original terms verbatim: the list was not + // rewritten, truncated or moved. + let retained_in_list = find_expr::(&decoded).expect("static IN survives"); + let retained_in_list = retained_in_list.downcast_ref::().unwrap(); + assert!(!retained_in_list.negated()); + assert_eq!(retained_in_list.list().len(), 10); + let retained = conjunction(vec![ + Arc::clone(&lower_bound), + Arc::clone(&upper_bound), + in_list, + ]); + assert_eq!( + encode_physical_expr_to_bytes(&decoded).unwrap(), + encode_physical_expr_to_bytes(&retained).unwrap() + ); + } + + #[test] + fn over_budget_membership_drops_the_oversized_in_list_too() { + // The same shape under a budget that only fits the bounds: keeping the + // representative-but-large `IN` list is not enough, so the retention widens once + // more and drops the `InListExpr`-bearing conjuncts as well. + let schema = Arc::new(Schema::new(vec![Field::new( + "device_id", + DataType::Int32, + false, + )])); + let device_id = col("device_id", &schema).unwrap(); + let lower_bound = Arc::new(BinaryExpr::new( + Arc::clone(&device_id), + Operator::GtEq, + lit(10i32), + )) as Arc; + let upper_bound = Arc::new(BinaryExpr::new( + Arc::clone(&device_id), + Operator::LtEq, + lit(20i32), + )) as Arc; + let bounds = Arc::new(BinaryExpr::new( + Arc::clone(&lower_bound), + Operator::And, + Arc::clone(&upper_bound), + )) as Arc; + let in_list = Arc::new( + InListExpr::try_new( + Arc::clone(&device_id), + (10..20).map(lit).collect(), + false, + &schema, + ) + .unwrap(), + ) as Arc; + let expr = Arc::new(BinaryExpr::new( + Arc::new(BinaryExpr::new( + Arc::clone(&bounds), + Operator::And, + unrepresentable_lookup(&device_id), + )) as Arc, + Operator::And, + in_list, + )) as Arc; + + let budget = encode_physical_expr_to_bytes(&bounds).unwrap().len(); + assert!(encode_bounded_physical_expr_to_bytes(&expr, budget).is_err()); + + let decoded = decode(&expr, &schema, budget); + let display = decoded.to_string(); + assert!(display.contains(">= 10"), "{display}"); + assert!(display.contains("<= 20"), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + assert_eq!( + encode_physical_expr_to_bytes(&decoded).unwrap(), + encode_physical_expr_to_bytes(&bounds).unwrap() + ); + } + + #[test] + fn dyn_filter_payload_negated_unconvertible_lookup_is_unsupported() { + // `NOT lookup` is a single top-level conjunct that carries the membership, so + // there is nothing safe to keep. Transport must not encode `NOT true` (or `true`) + // in its place: the predicate is reported as unsupported. + let schema = Arc::new(Schema::new(vec![Field::new( + "device_id", + DataType::Int32, + false, + )])); + let device_id = col("device_id", &schema).unwrap(); + let not_lookup = + Arc::new(NotExpr::new(unrepresentable_lookup(&device_id))) as Arc; + + assert_unsupported(DynFilterPayload::from_datafusion_expr(¬_lookup, 1024).unwrap_err()); + } + + #[test] + fn dyn_filter_payload_membership_under_or_is_unsupported() { + // `bound OR lookup` is a single top-level conjunct that carries the membership: + // rewriting it to `bound OR true` would be nothing but `true`, and keeping only + // the bound would *narrow* the predicate, so it is reported as unsupported. + let schema = Arc::new(Schema::new(vec![Field::new( + "device_id", + DataType::Int32, + false, + )])); + let device_id = col("device_id", &schema).unwrap(); + let lower_bound = Arc::new(BinaryExpr::new( + Arc::clone(&device_id), + Operator::GtEq, + lit(10i32), + )) as Arc; + let expr = Arc::new(BinaryExpr::new( + lower_bound, + Operator::Or, + unrepresentable_lookup(&device_id), + )) as Arc; + + assert_unsupported(DynFilterPayload::from_datafusion_expr(&expr, 1024).unwrap_err()); + } + + #[test] + fn dyn_filter_payload_over_budget_keeps_bounds() { let schema = Arc::new(Schema::new(vec![Field::new( "device_id", DataType::Int32, @@ -560,9 +1359,12 @@ mod tests { Operator::LtEq, lit(8255i32), )) as Arc; - let bounds = Arc::new(BinaryExpr::new(lower_bound, Operator::And, upper_bound)) - as Arc; - let in_list = Arc::new( + let bounds = Arc::new(BinaryExpr::new( + Arc::clone(&lower_bound), + Operator::And, + Arc::clone(&upper_bound), + )) as Arc; + let oversized_in = Arc::new( InListExpr::try_new( Arc::clone(&device_id), (8192..8256).map(lit).collect(), @@ -571,23 +1373,33 @@ mod tests { ) .unwrap(), ) as Arc; - let expr = Arc::new(BinaryExpr::new(Arc::clone(&bounds), Operator::And, in_list)) - as Arc; - let bounds_only = portable_remote_dyn_filter_expr(Arc::clone(&expr), true).unwrap(); - let bounds_only_size = encode_physical_expr_to_bytes(&bounds_only).unwrap().len(); - let full_size = encode_physical_expr_to_bytes(&expr).unwrap().len(); - assert!(full_size > bounds_only_size); + let expr = Arc::new(BinaryExpr::new( + Arc::clone(&bounds), + Operator::And, + oversized_in, + )) as Arc; - let payload = DynFilterPayload::from_datafusion_expr(&expr, bounds_only_size).unwrap(); - let decoded = payload - .decode_datafusion_expr(&TaskContext::default(), &schema, bounds_only_size) - .unwrap(); + // A caller-supplied limit that fits the bounds but not the oversized `IN` list: + // the membership conjunct is dropped and the bounds are encoded under the same + // budget. + let bounds_size = encode_physical_expr_to_bytes(&bounds).unwrap().len(); + assert!(encode_bounded_physical_expr_to_bytes(&expr, bounds_size).is_err()); - assert!(!contains_expr::(&decoded)); - let decoded_display = decoded.to_string(); - assert!(decoded_display.contains("device_id")); - assert!(decoded_display.contains(">=")); - assert!(decoded_display.contains("<=")); + let decoded = decode(&expr, &schema, bounds_size); + let display = decoded.to_string(); + assert!(display.contains(">= 8192"), "{display}"); + assert!(display.contains("<= 8255"), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + assert_eq!( + encode_physical_expr_to_bytes(&decoded).unwrap(), + encode_physical_expr_to_bytes(&bounds).unwrap() + ); + + // A budget that does not even fit the retained bounds is reported as too large: + // nothing that cannot be shipped may be replaced with `true`. + let true_size = encode_physical_expr_to_bytes(&lit(true)).unwrap().len(); + assert!(true_size < bounds_size); + assert_too_large(DynFilterPayload::from_datafusion_expr(&expr, true_size).unwrap_err()); } #[test] @@ -661,17 +1473,1069 @@ mod tests { )); } - fn contains_expr(expr: &Arc) -> bool { - let mut found = false; + #[test] + fn membership_value_bound_really_bounds_what_the_byte_budget_holds() { + // The count bound has to be *real*, not a restatement of its own formula: a + // payload of `B` bytes can never hold more than `B / MIN_ENCODED_LITERAL_BYTES` + // list literals. Prove it by encoding one literal more than the bound and checking + // that the resulting frame really exceeds `B`. + let budget = 512; + let bound = max_membership_values(budget); + assert!(bound > 1, "the bound must leave room for a real list"); + let schema = Schema::new(vec![Field::new("host", DataType::Utf8, true)]); + let minimal_in_list = |values: usize| { + Arc::new( + InListExpr::try_new( + Arc::new(Column::new("host", 0)), + (0..values).map(|_| lit("")).collect(), + false, + &schema, + ) + .unwrap(), + ) as Arc + }; + + let over_bound = encode_physical_expr_to_bytes(&minimal_in_list(bound + 1)).unwrap(); + assert!( + over_bound.len() > budget, + "{} literals must not fit a {budget}-byte budget, but the frame is {} bytes", + bound + 1, + over_bound.len() + ); + // The same payload is refused by the bounded encoder, so the derived bound and the + // encoder agree on where the budget ends. + assert!(matches!( + encode_bounded_physical_expr_to_bytes(&minimal_in_list(bound + 1), budget), + Err(CommonQueryError::DynFilterPayloadTooLarge { .. }) + )); + // The floor the bound is derived from must hold for the cheapest literal there is. + let cheapest = encode_physical_expr_to_bytes(&lit("")).unwrap().len(); + assert!( + cheapest >= MIN_ENCODED_LITERAL_BYTES, + "floor must hold for the cheapest literal, got {cheapest} bytes" + ); + } + + #[test] + fn utf8_list_literals_is_the_authoritative_value_check() { + let values: ArrayRef = Arc::new(StringArray::from(vec!["a", "b", "c"])); + assert!(utf8_list_literals(&values, 3).is_some()); + // The count is bounded before a single `Literal` is built. + assert!(utf8_list_literals(&values, 2).is_none()); + assert!(utf8_list_literals(&values, 0).is_none()); + // NULL-containing, empty and non-`Utf8` arrays are refused here, not only by the + // `HashTableLookupExpr` accessor. + let with_null: ArrayRef = Arc::new(StringArray::from(vec![Some("a"), None])); + assert!(utf8_list_literals(&with_null, 8).is_none()); + let empty: ArrayRef = Arc::new(StringArray::from(Vec::<&str>::new())); + assert!(utf8_list_literals(&empty, 8).is_none()); + let not_utf8: ArrayRef = Arc::new(Int32Array::from(vec![1, 2])); + assert!(utf8_list_literals(¬_utf8, 8).is_none()); + } +} + +/// Remote dynamic filter membership recovered from *real* DataFusion build leases. +/// +/// The tests above build `HashTableLookupExpr` by hand, which always has +/// `build_lease = None`. These tests run an actual `HashJoinExec` so DataFusion +/// publishes lookups backed by a real build lease, which is the only way to prove that +/// the key list (and the membership-free conjunct retention) actually works. +/// +/// Most remote predicates are checked twice: structurally (what the payload contains) +/// and by **evaluating** the producer predicate and the decoded predicate on the same +/// `NULL`-free rows. Evaluation pins the contract that applies here: the converted +/// membership is the build key list, so on `NULL`-free keys only, the decoded predicate +/// keeps exactly the rows the producer kept. It is a pruning hint, not a universal +/// equivalence — see `not_lookup_becomes_not_in` for the `NULL` caveat. +#[cfg(test)] +mod remote_membership_tests { + use std::collections::BTreeMap; + + use datafusion::arrow::array::{ + ArrayRef, BooleanArray, Float64Array, Int32Array, StringArray, StringDictionaryBuilder, + }; + use datafusion::arrow::datatypes::{DataType, Field, Int32Type, Schema}; + use datafusion::arrow::record_batch::RecordBatch; + use datafusion::physical_expr::expressions::{ + BinaryExpr, CaseExpr, Column, InListExpr, IsNullExpr, Literal, NotExpr, lit, + }; + use datafusion::physical_plan::PhysicalExpr; + use datafusion::physical_plan::joins::{HashExpr, HashTableLookupExpr}; + use datafusion_expr::Operator; + + use super::remote_membership_test_support::{ + RealDynFilter, key, one_column_schema, partitions, run_real_hash_join, utf8, utf8_join, + }; + use super::*; + + fn in_lists(expr: &Arc) -> Vec> { + let mut found = Vec::new(); expr.apply(|node| { - if node.is::() { - found = true; + if node.is::() { + found.push(Arc::clone(node)); + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + found + } + + fn in_list_values(expr: &Arc) -> Vec { + let in_list = expr.downcast_ref::().unwrap(); + assert!(!in_list.negated()); + in_list + .list() + .iter() + .map(|literal| { + literal + .downcast_ref::() + .unwrap() + .value() + .to_string() + .trim_matches('\'') + .to_string() + }) + .collect() + } + + fn assert_has_bounds(expr: &Arc) { + let display = expr.to_string(); + assert!(display.contains(">="), "missing lower bound: {display}"); + assert!(display.contains("<="), "missing upper bound: {display}"); + } + + fn utf8_batch(schema: &Arc, values: &[&str]) -> RecordBatch { + RecordBatch::try_new(Arc::clone(schema), vec![utf8(values)]).unwrap() + } + + fn selected(expr: &Arc, batch: &RecordBatch) -> Vec { + let array = expr + .evaluate(batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + let booleans = array.as_any().downcast_ref::().unwrap(); + (0..batch.num_rows()) + .map(|index| booleans.value(index)) + .collect() + } + + /// Widening check: the decoded predicate keeps every row the producer kept. + /// + /// Only used on `NULL`-free rows, where the build-key membership is exact for the + /// converted node. + fn assert_fail_open( + original: &Arc, + decoded: &Arc, + batch: &RecordBatch, + ) -> (Vec, Vec) { + let original_selected = selected(original, batch); + let decoded_selected = selected(decoded, batch); + for (row, (kept, kept_after)) in original_selected + .iter() + .zip(decoded_selected.iter()) + .enumerate() + { + assert!( + kept_after >= kept, + "decoded predicate dropped row {row}: producer={kept} decoded={kept_after} \ + (decoded predicate: {decoded})" + ); + } + (original_selected, decoded_selected) + } + + /// Branch chosen by the CASE routing expression for every row. + fn routed_branch(case: &CaseExpr, batch: &RecordBatch) -> Vec { + let routing = case.expr().expect("hash routing operand"); + let array = routing + .evaluate(batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + let branches = array + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|index| branches.value(index)) + .collect() + } + + fn hash_seed(expr: &Arc) -> u64 { + let mut seed = None; + expr.apply(|node| { + if let Some(hash) = node.downcast_ref::() { + seed = Some(hash.seed()); Ok(TreeNodeRecursion::Stop) } else { Ok(TreeNodeRecursion::Continue) } }) .unwrap(); - found + seed.expect("hash routing expression") + } + + /// `trace_id` values used to evaluate recovered filters. + fn evaluation_batch(schema: &Arc, values: &[String]) -> RecordBatch { + let borrowed = values.iter().map(String::as_str).collect::>(); + utf8_batch(schema, &borrowed) + } + + /// UTF-8 batch that can contain SQL NULLs, for wrapper semantics. + fn nullable_evaluation_batch(schema: &Arc, values: &[Option]) -> RecordBatch { + let array = StringArray::from(values.to_vec()); + RecordBatch::try_new(Arc::clone(schema), vec![Arc::new(array) as ArrayRef]).unwrap() + } + + #[tokio::test] + async fn membership_round_trips_real_utf8_build_lease() { + let schema = one_column_schema("trace_id", DataType::Utf8); + let (build, probe, values) = utf8_join(1); + let trace_id = key("trace_id", 0); + let filter = run_real_hash_join( + 1, + Arc::clone(&schema), + Arc::clone(&schema), + build, + probe, + vec![(Arc::clone(&trace_id), Arc::clone(&trace_id))], + ) + .await; + + // The producer predicate really holds a runtime-only lookup... + assert!(has_expr::(&filter.expr)); + + // ...and it is serialized as exact build membership, not as `true`. + let decoded = decode( + &filter.expr, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + assert!(!has_expr::(&decoded)); + let lists = in_lists(&decoded); + assert_eq!(lists.len(), 1, "expected one recovered InList: {decoded}"); + assert_eq!(in_list_values(&lists[0]), values); + assert_has_bounds(&decoded); + + // Members are selected, non-members are not, nothing else changes. + let mut rows = values.clone(); + rows.push("miss-not-a-member".to_string()); + let batch = evaluation_batch(&schema, &rows); + let (original_selected, decoded_selected) = + assert_fail_open(&filter.expr, &decoded, &batch); + for row in 0..values.len() { + assert!(original_selected[row], "member row {row} must be selected"); + assert!(decoded_selected[row], "member row {row} must stay selected"); + } + assert!( + !decoded_selected[values.len()], + "non-member must not be selected" + ); + } + + #[tokio::test] + async fn membership_preserves_three_branch_case_structure() { + let schema = one_column_schema("trace_id", DataType::Utf8); + let (build, probe, values) = utf8_join(3); + let trace_id = key("trace_id", 0); + let filter = run_real_hash_join( + 3, + Arc::clone(&schema), + Arc::clone(&schema), + build, + probe, + vec![(Arc::clone(&trace_id), Arc::clone(&trace_id))], + ) + .await; + + // Three hash-routed partitions produce a three-branch CASE. + let original_case = filter + .expr + .downcast_ref::() + .expect("real three-branch CASE"); + assert_eq!(original_case.when_then_expr().len(), 3); + + let decoded = decode( + &filter.expr, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + let case = decoded.downcast_ref::().expect("CASE preserved"); + assert_eq!(case.when_then_expr().len(), 3); + assert!(case.else_expr().is_some()); + + // Routing is untouched: same operand expression and same `HashExpr` seed, so + // the receiver selects the same branch for every row. + assert_eq!( + original_case.expr().unwrap().to_string(), + case.expr().unwrap().to_string() + ); + assert_eq!( + hash_seed(original_case.expr().unwrap()), + hash_seed(case.expr().unwrap()) + ); + + // Branch-specific membership: the WHEN literal names the branch, so the values + // recovered for each branch can be checked individually. + let mut branch_values: BTreeMap> = BTreeMap::new(); + for (when, then) in case.when_then_expr() { + let branch = match when.downcast_ref::().unwrap().value() { + ScalarValue::UInt64(Some(branch)) => *branch, + other => panic!("unexpected CASE routing literal: {other:?}"), + }; + let lists = in_lists(then); + assert_eq!(lists.len(), 1, "each branch keeps its membership: {then}"); + let mut branch_list = in_list_values(&lists[0]); + branch_list.sort(); + assert!( + branch_values.insert(branch, branch_list).is_none(), + "branch {branch} appears twice" + ); + assert_has_bounds(then); + } + assert_eq!(branch_values.len(), 3, "three distinct branches"); + + // Every build key is recovered exactly once, in its own branch. + let mut recovered = branch_values + .values() + .flatten() + .cloned() + .collect::>(); + recovered.sort(); + assert_eq!(recovered, values); + + // Evaluation proves branch-specific routing: a row is selected iff its key is in + // the membership of the branch the routing sends it to. + let mut rows = values.clone(); + rows.extend((0..4).map(|i| format!("miss-{i:02}"))); + let batch = evaluation_batch(&schema, &rows); + let (original_selected, decoded_selected) = + assert_fail_open(&filter.expr, &decoded, &batch); + let branches = routed_branch(case, &batch); + for (row, value) in rows.iter().enumerate() { + let branch = branches[row]; + let expected = branch_values + .get(&branch) + .is_some_and(|members| members.binary_search(value).is_ok()); + assert_eq!( + decoded_selected[row], expected, + "row {value} routed to branch {branch} must be selected iff it is a \ + member of that branch's list" + ); + // Build members must keep their decision; misses only ever widen. + if expected { + assert!( + original_selected[row], + "producer must select its own build member {value}" + ); + assert!(decoded_selected[row]); + } + } + assert!( + decoded_selected + .iter() + .filter(|selected| **selected) + .count() + >= values.len(), + "every build member must still be selected" + ); + } + + /// Runs a real one-partition `Utf8` hash join over `key_count` distinct keys. + async fn real_join_with_keys(key_count: usize) -> (Arc, RealDynFilter, Vec) { + let schema = one_column_schema("trace_id", DataType::Utf8); + let trace_id = key("trace_id", 0); + let values = (0..key_count) + .map(|i| format!("trace-{i:04}")) + .collect::>(); + let borrowed = values.iter().map(String::as_str).collect::>(); + let build = partitions(&schema, vec![vec![utf8(&borrowed)]]); + let probe = partitions(&schema, vec![vec![utf8(&borrowed)]]); + let filter = run_real_hash_join( + 1, + Arc::clone(&schema), + Arc::clone(&schema), + build, + probe, + vec![(Arc::clone(&trace_id), Arc::clone(&trace_id))], + ) + .await; + (schema, filter, values) + } + + #[tokio::test] + async fn bounds_survive_unrepresentable_integer_lookup() { + // Integer keys are not a single-column `Utf8` membership, so this real lookup has + // no wire form. Retention keeps the top-level bounds and drops the lookup + // conjunct, so the decoded predicate only ever gets wider. + let schema = one_column_schema("k", DataType::Int32); + let k = key("k", 0); + let filter = run_real_hash_join( + 1, + Arc::clone(&schema), + Arc::clone(&schema), + partitions( + &schema, + vec![vec![Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef]], + ), + partitions( + &schema, + vec![vec![ + Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 10])) as ArrayRef + ]], + ), + vec![(Arc::clone(&k), Arc::clone(&k))], + ) + .await; + let lookup = find_expr::(&filter.expr).expect("real lookup"); + + let lower_bound = Arc::new(BinaryExpr::new(Arc::clone(&k), Operator::GtEq, lit(2i32))) + as Arc; + let upper_bound = Arc::new(BinaryExpr::new(Arc::clone(&k), Operator::LtEq, lit(4i32))) + as Arc; + let bounds = Arc::new(BinaryExpr::new(lower_bound, Operator::And, upper_bound)) + as Arc; + let expr = Arc::new(BinaryExpr::new(Arc::clone(&bounds), Operator::And, lookup)) + as Arc; + + let decoded = decode( + &expr, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + let display = decoded.to_string(); + assert!( + display.contains(">= 2"), + "lower bound must survive: {display}" + ); + assert!( + display.contains("<= 4"), + "upper bound must survive: {display}" + ); + assert!(!has_expr::(&decoded), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + + // The producer and the decoded predicate agree on every row the producer kept and + // only diverge on the non-member `4`, where the retained payload widens. + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 10])) as ArrayRef], + ) + .unwrap(); + let (original_selected, decoded_selected) = assert_fail_open(&expr, &decoded, &batch); + assert_eq!( + original_selected, + vec![false, true, true, false, false, false] + ); + assert_eq!( + decoded_selected, + vec![false, true, true, true, false, false] + ); + } + + #[tokio::test] + async fn small_budget_keeps_bounds_and_drops_real_membership() { + // A real `Utf8` lookup next to integer bounds, shipped under a caller-supplied + // limit that fits only the bounds: retention drops the membership conjunct and + // still sends the bounds, which can only widen the producer predicate. + let probe_schema = Arc::new(Schema::new(vec![ + Field::new("trace_id", DataType::Utf8, true), + Field::new("k", DataType::Int32, true), + ])); + let build_schema = one_column_schema("trace_id", DataType::Utf8); + let long = "t".repeat(256); + let values = (0..24) + .map(|i| format!("{long}-{i:02}")) + .collect::>(); + let borrowed = values.iter().map(String::as_str).collect::>(); + let build_key = key("trace_id", 0); + let probe_key = key("trace_id", 0); + let filter = run_real_hash_join( + 1, + Arc::clone(&build_schema), + Arc::clone(&probe_schema), + partitions(&build_schema, vec![vec![utf8(&borrowed)]]), + partitions( + &probe_schema, + vec![vec![ + utf8(&borrowed), + Arc::new(Int32Array::from(vec![0i32; values.len()])) as ArrayRef, + ]], + ), + vec![(Arc::clone(&build_key), Arc::clone(&probe_key))], + ) + .await; + let lookup = find_expr::(&filter.expr).expect("real lookup"); + + let k = key("k", 1); + let lower_bound = Arc::new(BinaryExpr::new(Arc::clone(&k), Operator::GtEq, lit(2i32))) + as Arc; + let upper_bound = Arc::new(BinaryExpr::new(Arc::clone(&k), Operator::LtEq, lit(5i32))) + as Arc; + let bounds = Arc::new(BinaryExpr::new(lower_bound, Operator::And, upper_bound)) + as Arc; + let expr = Arc::new(BinaryExpr::new(Arc::clone(&bounds), Operator::And, lookup)) + as Arc; + + // Only the bounds fit: the membership of 24 long keys is far beyond this budget. + let bounds_size = encode_physical_expr_to_bytes(&bounds).unwrap().len(); + let decoded = decode(&expr, &filter.probe_schema, bounds_size); + let display = decoded.to_string(); + assert!( + display.contains(">= 2"), + "lower bound must survive: {display}" + ); + assert!( + display.contains("<= 5"), + "upper bound must survive: {display}" + ); + assert!(!has_expr::(&decoded), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + + // Evaluation: a member inside the bounds is kept by both, a member outside the + // bounds is dropped by both, and only the non-member is widened. + let trace_ids = vec![ + values[0].as_str(), + values[1].as_str(), + "miss-not-a-member", + values[2].as_str(), + ]; + let batch = RecordBatch::try_new( + Arc::clone(&probe_schema), + vec![ + utf8(&trace_ids), + Arc::new(Int32Array::from(vec![2i32, 6, 3, 5])) as ArrayRef, + ], + ) + .unwrap(); + let (original_selected, decoded_selected) = assert_fail_open(&expr, &decoded, &batch); + assert_eq!(original_selected, vec![true, false, false, true]); + assert_eq!(decoded_selected, vec![true, false, true, true]); + } + + #[tokio::test] + async fn real_642_key_hash_join_becomes_exact_642_literal_in() { + let (schema, filter, values) = real_join_with_keys(642).await; + assert_eq!(values.len(), 642); + // The producer predicate really holds a runtime-only lookup... + assert!(has_expr::(&filter.expr)); + + // ...and it is written as one literal per build key within the payload budget. + let decoded = decode( + &filter.expr, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + assert!(!has_expr::(&decoded)); + let lists = in_lists(&decoded); + assert_eq!(lists.len(), 1, "expected one converted InList: {decoded}"); + let mut recovered = in_list_values(&lists[0]); + assert_eq!(recovered.len(), 642, "one literal per build key"); + recovered.sort(); + assert_eq!(recovered, values); + + // The membership selects exactly the build keys. + let mut rows = values.clone(); + rows.push("miss-not-a-member".to_string()); + let batch = evaluation_batch(&schema, &rows); + let (original_selected, decoded_selected) = + assert_fail_open(&filter.expr, &decoded, &batch); + for row in 0..values.len() { + assert!( + original_selected[row], + "build key row {row} must be selected" + ); + assert!( + decoded_selected[row], + "build key row {row} must stay selected" + ); + } + assert!( + !decoded_selected[values.len()], + "non-member must not be selected" + ); + } + + #[tokio::test] + async fn large_build_side_is_refused_before_materializing_literals() { + // A build side far larger than a small budget can hold must be refused by the + // derived count bound *before* the encoder creates one `Literal` per build key. + let (_, filter, values) = real_join_with_keys(1024).await; + let lookup = find_expr::(&filter.expr).expect("real lookup"); + let lookup = lookup.downcast_ref::().unwrap(); + + // DataFusion itself would hand out every build key ... + let unbounded = lookup + .build_key_membership(usize::MAX, usize::MAX) + .expect("the build lease holds the keys"); + assert_eq!(unbounded.values().len(), values.len()); + + // ... but a budget that only fits the producer's own bounds leaves room for a few + // literals, so the count bound derived from it refuses the list outright. + let bounds = conjunction( + split_conjunction(&filter.expr) + .into_iter() + .filter(|conjunct| !has_expr::(conjunct)) + .cloned() + .collect::>(), + ); + let budget = encode_physical_expr_to_bytes(&bounds).unwrap().len(); + assert!( + max_membership_values(budget) < values.len(), + "the derived count bound must stay below the build side size" + ); + // The refusal is reported as a *size* refusal, not as a predicate without a wire + // form: the key list exists, it just does not fit this budget. + assert!( + matches!( + RemoteMembership::from_lookup(lookup, budget), + Err(MembershipRefusal::OverBudget { build_keys }) if build_keys == values.len() + ), + "an over-large build key list must be refused as over budget" + ); + + // Nothing of the 1024-key membership reaches the payload. + let decoded = decode(&filter.expr, &filter.probe_schema, budget); + assert!(!has_expr::(&decoded), "{decoded}"); + assert!(!has_expr::(&decoded), "{decoded}"); + assert_has_bounds(&decoded); + } + + #[tokio::test] + async fn lookup_or_key_is_null_preserves_wrapper_and_becomes_in_list() { + // The `key IS NULL OR ` wrapper of a null-aware join is an ordinary + // node: it stays in the payload and only the lookup inside it is converted. + let (schema, filter, values) = real_join_with_keys(8).await; + let lookup = find_expr::(&filter.expr).expect("real lookup"); + let trace_id = key("trace_id", 0); + let wrapped = Arc::new(BinaryExpr::new( + Arc::clone(&lookup), + Operator::Or, + Arc::new(IsNullExpr::new(Arc::clone(&trace_id))), + )) as Arc; + + let decoded = decode( + &wrapped, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + let display = decoded.to_string(); + assert!( + display.contains("IS NULL"), + "IS NULL wrapper must survive: {display}" + ); + assert!(!display.contains("hash_lookup"), "{display}"); + let lists = in_lists(&decoded); + assert_eq!(lists.len(), 1, "expected one converted InList: {display}"); + assert_eq!(in_list_values(&lists[0]), values); + + let mut rows = values + .iter() + .map(|value| Some(value.clone())) + .collect::>(); + rows.push(Some("miss-not-a-member".to_string())); + rows.push(None); + let batch = nullable_evaluation_batch(&schema, &rows); + let (original_selected, decoded_selected) = assert_fail_open(&wrapped, &decoded, &batch); + for row in 0..values.len() { + assert!(original_selected[row]); + assert!(decoded_selected[row]); + } + assert!( + original_selected[rows.len() - 1], + "producer `lookup OR key IS NULL` keeps a NULL probe" + ); + assert!( + decoded_selected[rows.len() - 1], + "decoded `IN ... OR key IS NULL` must also keep a NULL probe" + ); + } + + #[tokio::test] + async fn null_build_keys_keep_only_the_safe_bounds() { + // The key-list accessor requires `NULL`-free build keys, so a build side with a + // `NULL` key has no membership. This producer shape is `bounds AND lookup`: the + // bounds DataFusion derived from the non-`NULL` keys are retained, and the lookup + // is dropped instead of the predicate being replaced with `true`. + let schema = one_column_schema("trace_id", DataType::Utf8); + let trace_id = key("trace_id", 0); + let build_values: ArrayRef = Arc::new(StringArray::from(vec![ + Some("trace-a"), + None, + Some("trace-b"), + ])); + let filter = run_real_hash_join( + 1, + Arc::clone(&schema), + Arc::clone(&schema), + partitions(&schema, vec![vec![build_values]]), + partitions(&schema, vec![vec![utf8(&["trace-a", "trace-b", "miss"])]]), + vec![(Arc::clone(&trace_id), Arc::clone(&trace_id))], + ) + .await; + assert!(has_expr::(&filter.expr)); + let producer = filter.expr.to_string(); + assert!(producer.contains(">= trace-a"), "{producer}"); + assert!(producer.contains("<= trace-b"), "{producer}"); + + let decoded = decode( + &filter.expr, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + let display = decoded.to_string(); + assert!(!has_expr::(&decoded), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + assert!(display.contains(">= trace-a"), "{display}"); + assert!(display.contains("<= trace-b"), "{display}"); + assert_ne!(display, "true", "transport must not widen to `true`"); + + // The retained conjuncts are the producer bounds verbatim. + let bounds = split_conjunction(&filter.expr) + .into_iter() + .filter(|conjunct| !has_expr::(conjunct)) + .cloned() + .collect::>(); + assert_eq!( + encode_physical_expr_to_bytes(&decoded).unwrap(), + encode_physical_expr_to_bytes(&conjunction(bounds)).unwrap() + ); + + // Widening only: every build key row the producer kept is still kept. + let rows = ["trace-a", "trace-b", "miss"] + .map(ToString::to_string) + .to_vec(); + let batch = evaluation_batch(&schema, &rows); + let (original_selected, decoded_selected) = + assert_fail_open(&filter.expr, &decoded, &batch); + assert_eq!(original_selected, vec![true, true, false]); + assert!(decoded_selected[0], "build key row must stay selected"); + assert!(decoded_selected[1], "build key row must stay selected"); + } + + /// `NOT lookup` becomes `NOT (key IN (...))`. + /// + /// This narrow contract is about **positive pruning**: the membership is the build + /// key list of the join, so converting the node in place structurally preserves the + /// negation instead of substituting `true` into it. It is *not* a claim of + /// universal equivalence with the join: `IN`/`NOT IN` are `NULL`-propagating while + /// a hash lookup can be definite on a `NULL` probe key, so a `NULL` key that the + /// lookup keeps can still be dropped by `NOT IN`. The rows below are non-`NULL`, + /// which is the shape the hint is meant for. + #[tokio::test] + async fn not_lookup_becomes_not_in() { + let (schema, filter, values) = real_join_with_keys(8).await; + let lookup = find_expr::(&filter.expr).expect("real lookup"); + let not_lookup = Arc::new(NotExpr::new(Arc::clone(&lookup))) as Arc; + + let decoded = decode( + ¬_lookup, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + let display = decoded.to_string(); + assert!( + display.starts_with("NOT "), + "NOT must survive the wire: {display}" + ); + let lists = in_lists(&decoded); + assert_eq!(lists.len(), 1, "expected one converted InList: {display}"); + assert_eq!(in_list_values(&lists[0]), values); + + let mut rows = values.clone(); + rows.push("miss-not-a-member".to_string()); + let batch = evaluation_batch(&schema, &rows); + let (original_selected, decoded_selected) = assert_fail_open(¬_lookup, &decoded, &batch); + assert!(original_selected[..values.len()].iter().all(|kept| !*kept)); + assert!( + original_selected[values.len()], + "producer `NOT lookup` keeps the non-member" + ); + assert!(decoded_selected[..values.len()].iter().all(|kept| !*kept)); + assert!( + decoded_selected[values.len()], + "decoded `NOT IN` keeps the non-member" + ); + } + + #[tokio::test] + async fn unsupported_float_composite_and_dictionary_keys_keep_only_safe_conjuncts() { + // Float64 keys: the key-list accessor only exposes a single `Utf8` column. + let float_schema = one_column_schema("k", DataType::Float64); + let float_on = key("k", 0); + let float_filter = run_real_hash_join( + 2, + Arc::clone(&float_schema), + Arc::clone(&float_schema), + partitions( + &float_schema, + vec![ + vec![Arc::new(Float64Array::from(vec![0.0, -0.0])) as ArrayRef], + vec![Arc::new(Float64Array::from(vec![1.5])) as ArrayRef], + ], + ), + partitions( + &float_schema, + vec![ + vec![Arc::new(Float64Array::from(vec![0.0, 2.5])) as ArrayRef], + vec![Arc::new(Float64Array::from(vec![1.5])) as ArrayRef], + ], + ), + vec![(Arc::clone(&float_on), Arc::clone(&float_on))], + ) + .await; + assert!(has_expr::(&float_filter.expr)); + // The partitioned producer publishes a per-partition `CASE`, so the only + // top-level conjunct carries the membership and there is nothing safe to keep: + // the predicate is reported as unsupported instead of becoming `true`. + assert_unsupported( + DynFilterPayload::from_datafusion_expr( + &float_filter.expr, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ) + .unwrap_err(), + ); + + // Composite (two-column) keys: not a single-column membership. Both columns + // take part in the hash partitioning, exactly like a real two-key join. + let composite_schema = Arc::new(Schema::new(vec![ + Field::new("trace_id", DataType::Utf8, true), + Field::new("host", DataType::Utf8, true), + ])); + let composite_on = vec![ + (key("trace_id", 0), key("trace_id", 0)), + (key("host", 1), key("host", 1)), + ]; + let composite_filter = run_real_hash_join( + 2, + Arc::clone(&composite_schema), + Arc::clone(&composite_schema), + partitions( + &composite_schema, + vec![ + vec![utf8(&["trace-0", "trace-1"]), utf8(&["host-a", "host-a"])], + vec![utf8(&["trace-2"]), utf8(&["host-b"])], + ], + ), + partitions( + &composite_schema, + vec![ + vec![utf8(&["trace-0", "trace-9"]), utf8(&["host-a", "host-z"])], + vec![utf8(&["trace-2"]), utf8(&["host-b"])], + ], + ), + composite_on, + ) + .await; + assert!(has_expr::(&composite_filter.expr)); + // Same per-partition `CASE` shape as the float case: unsupported, not `true`. + assert_unsupported( + DynFilterPayload::from_datafusion_expr( + &composite_filter.expr, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ) + .unwrap_err(), + ); + + // Dictionary keys: not a plain `Utf8` key column. + let dict_type = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)); + let dict_schema = one_column_schema("trace_id", dict_type); + let dict_on = key("trace_id", 0); + let mut dict_builder = StringDictionaryBuilder::::new(); + dict_builder.append_value("trace-0"); + dict_builder.append_value("trace-1"); + let dict_values: ArrayRef = Arc::new(dict_builder.finish()); + let dict_filter = run_real_hash_join( + 2, + Arc::clone(&dict_schema), + Arc::clone(&dict_schema), + partitions(&dict_schema, vec![vec![Arc::clone(&dict_values)]]), + partitions(&dict_schema, vec![vec![dict_values]]), + vec![(Arc::clone(&dict_on), Arc::clone(&dict_on))], + ) + .await; + assert!(has_expr::(&dict_filter.expr)); + let decoded = decode( + &dict_filter.expr, + &dict_filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + // Dictionary keys are not a plain `Utf8` key column either. This producer shape + // is a flat `bounds AND lookup`, so retention keeps the bounds conjunct that + // DataFusion itself published and drops only the membership. + let display = decoded.to_string(); + assert!(!has_expr::(&decoded), "{display}"); + assert!(!has_expr::(&decoded), "{display}"); + assert!( + display.contains(">="), + "lower bound must survive: {display}" + ); + assert!( + display.contains("<="), + "upper bound must survive: {display}" + ); + } + + #[tokio::test] + async fn producer_paths_encode_150_and_1024_keys() { + // The two producer shapes that can carry a build key list: the static `IN` list + // pushdown (150 and 1024 are the pushdown limits of the runtime experiment) and + // the hash lookup. Neither is capped by an encoder-side value/byte policy: the + // only bound is the payload budget itself. + for key_count in [150, 1024] { + let (schema, filter, values) = real_join_with_keys(key_count).await; + assert!(has_expr::(&filter.expr)); + let decoded = decode( + &filter.expr, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + let lists = in_lists(&decoded); + assert_eq!(lists.len(), 1, "{key_count} keys: {decoded}"); + assert_eq!(in_list_values(&lists[0]), values); + + // The static `IN` list pushdown of the same size encodes as well. + let in_list = Arc::new( + InListExpr::try_new( + key("trace_id", 0), + values.iter().map(|value| lit(value.clone())).collect(), + false, + &schema, + ) + .unwrap(), + ) as Arc; + let decoded = decode( + &in_list, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + let lists = in_lists(&decoded); + assert_eq!(lists.len(), 1, "{key_count} static keys: {decoded}"); + assert_eq!(in_list_values(&lists[0]), values); + } + } + + #[tokio::test] + async fn over_budget_membership_case_is_a_size_refusal() { + let schema = one_column_schema("trace_id", DataType::Utf8); + // Long values so the membership lists dominate the payload. + let long = "t".repeat(256); + let values = (0..24) + .map(|i| format!("{long}-{i:02}")) + .collect::>(); + let borrowed = values.iter().map(String::as_str).collect::>(); + let build = partitions(&schema, vec![vec![utf8(&borrowed)]]); + let probe = partitions(&schema, vec![vec![utf8(&borrowed)]]); + let trace_id = key("trace_id", 0); + let filter = run_real_hash_join( + 3, + Arc::clone(&schema), + Arc::clone(&schema), + build, + probe, + vec![(Arc::clone(&trace_id), Arc::clone(&trace_id))], + ) + .await; + assert!(has_expr::(&filter.expr)); + + // The real payload budget carries all three branch memberships exactly. + let decoded = decode( + &filter.expr, + &filter.probe_schema, + REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES, + ); + assert_eq!(in_lists(&decoded).len(), 3); + let case = decoded.downcast_ref::().expect("CASE preserved"); + assert_eq!(case.when_then_expr().len(), 3); + + // The whole predicate is one top-level `CASE` conjunct that carries the + // membership, so a budget that only fits `true` has nothing safe to keep: + // membership is never rewritten inside `CASE` and the payload is an explicit + // error instead of an always-true filter. The predicates *are* representable, they + // only do not fit, so the refusal is a size one and not "no wire form". + let true_size = encode_physical_expr_to_bytes(&lit(true)).unwrap().len(); + assert!(true_size < REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES); + assert_too_large( + DynFilterPayload::from_datafusion_expr(&filter.expr, true_size).unwrap_err(), + ); + } + + #[test] + fn over_budget_negation_is_a_size_refusal() { + let schema = Arc::new(Schema::new(vec![Field::new("host", DataType::Utf8, false)])); + let host = Arc::new(Column::new("host", 0)) as Arc; + let long = "h".repeat(256); + let in_list = Arc::new( + InListExpr::try_new( + Arc::clone(&host), + (0..16).map(|i| lit(format!("{long}-{i:02}"))).collect(), + false, + &schema, + ) + .unwrap(), + ) as Arc; + // `NOT (host IN (...))`: dropping the list in place would evaluate `NOT true` + // = false and drop every row, so the whole conjunct is dropped. Nothing is left + // to ship and the predicate itself is representable, so the refusal is a size one. + let not_in = Arc::new(NotExpr::new(Arc::clone(&in_list))) as Arc; + let true_size = encode_physical_expr_to_bytes(&lit(true)).unwrap().len(); + + assert_too_large(DynFilterPayload::from_datafusion_expr(¬_in, true_size).unwrap_err()); + } + + #[tokio::test] + async fn recovered_payload_uses_only_existing_wire_variants() { + let schema = one_column_schema("trace_id", DataType::Utf8); + let (build, probe, values) = utf8_join(3); + let trace_id = key("trace_id", 0); + let filter = run_real_hash_join( + 3, + Arc::clone(&schema), + Arc::clone(&schema), + build, + probe, + vec![(Arc::clone(&trace_id), Arc::clone(&trace_id))], + ) + .await; + let max = REMOTE_DYN_FILTER_PAYLOAD_MAX_BYTES; + + // The recovered payload decodes with the default extension codec, which rejects + // every expression variant it does not know: a new wire variant would fail here. + let payload = DynFilterPayload::from_datafusion_expr(&filter.expr, max).unwrap(); + let bytes = match &payload { + DynFilterPayload::Datafusion(bytes) => bytes.clone(), + }; + let decoded = payload + .decode_datafusion_expr(&TaskContext::default(), &filter.probe_schema, max) + .unwrap(); + assert_eq!(in_lists(&decoded).len(), 3); + assert_eq!( + hash_seed( + filter + .expr + .downcast_ref::() + .unwrap() + .expr() + .unwrap() + ), + hash_seed(decoded.downcast_ref::().unwrap().expr().unwrap()), + "HashExpr routing seed must survive the wire" + ); + + // Receivers still accept a plain `true` payload: an older sender could publish one, + // and a producer may legitimately publish `true` itself. This transport no longer + // synthesizes `true` as an encoding fallback, so the decode-side compatibility stays + // while the encode-side fallback is gone. + let legacy_bytes = encode_physical_expr_to_bytes(&lit(true)).unwrap(); + assert!(legacy_bytes.len() < bytes.len()); + let legacy_decoded = DynFilterPayload::Datafusion(legacy_bytes) + .decode_datafusion_expr(&TaskContext::default(), &filter.probe_schema, max) + .unwrap(); + assert_eq!(legacy_decoded.to_string(), "true"); + let mut rows = values.clone(); + rows.push("miss-not-a-member".to_string()); + let batch = evaluation_batch(&schema, &rows); + assert_fail_open(&filter.expr, &legacy_decoded, &batch); } } diff --git a/src/common/query/src/request/remote_membership_test_support.rs b/src/common/query/src/request/remote_membership_test_support.rs new file mode 100644 index 00000000000..6bcc16d9b49 --- /dev/null +++ b/src/common/query/src/request/remote_membership_test_support.rs @@ -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, + /// Schema of the probe side the predicate applies to. + pub probe_schema: Arc, +} + +/// Schema with a single nullable column. +pub fn one_column_schema(name: &str, data_type: DataType) -> Arc { + 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, columns: Vec>) -> Vec> { + 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 { + 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, + right_schema: Arc, + build_partitions: Vec>, + probe_partitions: Vec>, + on: Vec<(Arc, Arc)>, +) -> RealDynFilter { + assert!(partition_count >= 1); + + let left_scan: Arc = + MemorySourceConfig::try_new_exec(&build_partitions, Arc::clone(&left_schema), None) + .unwrap(); + let right_scan: Arc = + 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, + Arc::new(RepartitionExec::try_new(right_scan, right_partitioning).unwrap()) + as Arc, + ) + } 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 = Arc::clone(&probe_filter) as _; + // The probe subtree must reference the produced filter, otherwise the join never + // publishes it. + let right: Arc = + 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 = 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::>().await.unwrap(); + } + }) + .collect::>(); + 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>, Vec) { + let schema = one_column_schema("trace_id", DataType::Utf8); + let values = (0..partition_count * 8) + .map(|i| format!("trace-{i:02}")) + .collect::>(); + let borrowed = values.iter().map(String::as_str).collect::>(); + 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::>(); + probe_values.extend(misses.iter().map(String::as_str)); + let probe = partitions(&schema, vec![vec![utf8(&probe_values)]]); + (build, probe, values) +} diff --git a/src/query/src/dist_plan/dyn_filter_bridge.rs b/src/query/src/dist_plan/dyn_filter_bridge.rs index fb49839b304..0ebd0dcbf3e 100644 --- a/src/query/src/dist_plan/dyn_filter_bridge.rs +++ b/src/query/src/dist_plan/dyn_filter_bridge.rs @@ -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) -> bool { let mut has_scalar_subquery = false; let mut supported = true; let _ = filter.apply(|expr| { if expr.is::() { has_scalar_subquery = true; - } else if expr.is::() - || (expr.children().is_empty() && !expr.is::() && !expr.is::()) - { + } else if expr.children().is_empty() && !expr.is::() && !expr.is::() { 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 { @@ -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 { + let column = Arc::new(Column::new(column_name, column_index)) as Arc; + 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; + 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; @@ -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]; + + 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::()); + + // 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( diff --git a/src/query/src/dist_plan/remote_dyn_filter_receiver.rs b/src/query/src/dist_plan/remote_dyn_filter_receiver.rs index 562cfd3c77c..71eeca16198 100644 --- a/src/query/src/dist_plan/remote_dyn_filter_receiver.rs +++ b/src/query/src/dist_plan/remote_dyn_filter_receiver.rs @@ -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) + .collect::>(); + 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::() + .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::().unwrap(); + assert_eq!( + (0..batch.num_rows()) + .map(|row| selected.value(row)) + .collect::>(), + vec![true, true, false] + ); + } } diff --git a/src/query/src/dist_plan/remote_dyn_filter_registry.rs b/src/query/src/dist_plan/remote_dyn_filter_registry.rs index 6a44db70513..42a60d6d653 100644 --- a/src/query/src/dist_plan/remote_dyn_filter_registry.rs +++ b/src/query/src/dist_plan/remote_dyn_filter_registry.rs @@ -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) -> Option> { + 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> { + let reset = lit(true) as Arc; + 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 { + let column = Arc::new(Column::new("host", 0)) as Arc; + 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; + + Arc::new(NotExpr::new(lookup)) as Arc + } + + fn host_schema() -> Arc { + 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, + ) -> Arc { + 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); diff --git a/src/table/src/predicate.rs b/src/table/src/predicate.rs index 8dc85b610f4..d3c66430045 100644 --- a/src/table/src/predicate.rs +++ b/src/table/src/predicate.rs @@ -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>> { 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>> { self.dyn_filters .load()