Advanced data verification and plan verification tests for RuntimeFilter (#514)

* Advanced data verification and plan verification tests for RuntimeFilter

Runtime filter was added to DRILL as a part of DRILL-6385, DRILL-6731 and DRILL-6792.

This PR contains the following changes

- Data verification tests for broadcast and hash joins with join on different data types.
- Plan verification tests to verify Runtime Filter is applied on probe side scan, wherever applicable.
- The queries are run against TPC-H SF 100 data

* Minor fix, remove tsv header for nested broadcast scenario

* Incorporated review comments

- Changed .q and .q.explain extensions to .sql and .sql.explain.
- Explictly enable / disable broadcast join and set the threshold so that change in default values for these parameters will not affect the tests.
- Round decimals to 3 digits to have more stable test case verification.
- Remove "alter session" to conform with standard for setting / resetting session level parameters.
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.e_tsv
new file mode 100644
index 0000000..b712b1f
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.e_tsv
@@ -0,0 +1 @@
+109307880
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.plan
new file mode 100644
index 0000000..9abfa02
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.plan
@@ -0,0 +1,6 @@
+.*01-03\s+HashJoin.*
+01-05\s+SelectionVectorRemover
+01-06\s+RuntimeFilter
+01-07\s+Scan\(table=\[\[dfs, drilltestdirtpch100parquet, lineitem\]\].*columns=\[`l_suppkey`\]\]\]\)
+01-04\s+BroadcastExchange
+.*02-03\s+Scan\(table=\[\[dfs, drilltestdirtpch100parquet, supplier\]\].*columns=\[`s_suppkey`, `s_acctbal`\]\]\]\)
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.sql
new file mode 100644
index 0000000..f3c7aca
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.sql
@@ -0,0 +1,20 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+select
+  count(*) as row_count
+from (
+  select *
+  from
+    lineitem l,
+    supplier s
+  where
+    s.s_suppkey = l.l_suppkey
+    and s.s_acctbal < 1000);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.sql.explain
new file mode 100644
index 0000000..ab55c9f
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_broadcast.sql.explain
@@ -0,0 +1,21 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+explain plan for
+select
+  count(*) as row_count
+from (
+  select *
+  from
+    lineitem l,
+    supplier s
+  where
+    s.s_suppkey = l.l_suppkey
+    and s.s_acctbal < 1000);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.e_tsv
new file mode 100644
index 0000000..1b0ad5e
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.e_tsv
@@ -0,0 +1 @@
+96176495
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.plan
new file mode 100644
index 0000000..541d30d
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.plan
@@ -0,0 +1,15 @@
+.*01-03\s+HashJoin.*
+01-05\s+Project.*
+01-07\s+HashToRandomExchange.*
+02-01\s+UnorderedMuxExchange
+04-01\s+Project.*
+04-02\s+SelectionVectorRemover
+04-03\s+RuntimeFilter
+04-04\s+Scan\(table=\[\[dfs, drilltestdirtpch100parquet, lineitem\]\].*columns=\[`l_orderkey`\]\]\]\)
+01-04\s+Project.*
+01-06\s+HashToRandomExchange.*
+03-01\s+UnorderedMuxExchange
+05-01\s+Project.*
+05-02\s+SelectionVectorRemover
+05-03\s+Filter.*
+05-04\s+Scan\(table=\[\[dfs, drilltestdirtpch100parquet, orders\]\].*columns=\[`o_orderkey`, `o_totalprice`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.sql
new file mode 100644
index 0000000..5b1e352
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.sql
@@ -0,0 +1,18 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+select
+  count(*) as row_count
+from (
+  select *
+  from
+    lineitem l,
+    orders o
+  where
+    o.o_orderkey = l.l_orderkey
+    and o.o_totalprice < 100000);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.sql.explain
new file mode 100644
index 0000000..13741e6
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/basic_hash.sql.explain
@@ -0,0 +1,19 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+explain plan for
+select
+  count(*) as row_count
+from (
+  select *
+  from
+    lineitem l,
+    orders o
+  where
+    o.o_orderkey = l.l_orderkey
+    and o.o_totalprice < 100000);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.e_tsv
new file mode 100644
index 0000000..c8ab573
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.e_tsv
@@ -0,0 +1,10 @@
+1992-05-22	6409608.0
+1994-12-19	6407722.0
+1995-09-18	6406494.0
+1994-08-07	6402739.0
+1997-10-12	6402156.0
+1995-01-04	6401099.0
+1993-05-11	6400985.0
+1997-10-15	6400105.0
+1995-11-17	6399570.0
+1993-02-04	6397672.0
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.plan
new file mode 100644
index 0000000..4ce445d
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.plan
@@ -0,0 +1,11 @@
+.*03-03\s+HashJoin.*
+03-05\s+HashToRandomExchange.*
+04-01\s+RuntimeFilter
+04-02\s+Scan\(table=\[\[dfs, drilltestdirtpch100parquet, lineitem\]\].*columns=\[`l_shipdate`, `l_quantity`\]\]\]\)
+03-04\s+HashAgg.*
+03-06\s+HashToRandomExchange.*
+05-01\s+HashAgg.*
+05-02\s+Project.*
+05-03\s+SelectionVectorRemover
+05-04\s+Filter.*
+05-05\s+Scan\(table=\[\[dfs, drilltestdirtpch100parquet, orders\]\].*columns=\[`o_orderpriority`, `o_orderdate`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.sql
new file mode 100644
index 0000000..8d69838
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.sql
@@ -0,0 +1,26 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+select
+  l.l_shipdate,
+  sum(l.l_quantity) as total_quantity
+from
+  lineitem l
+where
+  l.l_shipdate in (
+    select
+      distinct(o.o_orderdate)
+    from
+      orders o
+    where
+      o.o_orderpriority = '1-URGENT')
+group by
+  l.l_shipdate
+order by
+  total_quantity desc
+limit 10;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.sql.explain
new file mode 100644
index 0000000..3f727b3
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/blocking_op_build_side.sql.explain
@@ -0,0 +1,27 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+explain plan for
+select
+  l.l_shipdate,
+  sum(l.l_quantity) as total_quantity
+from
+  lineitem l
+where
+  l.l_shipdate in (
+    select
+      distinct(o.o_orderdate)
+    from
+      orders o
+    where
+      o.o_orderpriority = '1-URGENT')
+group by
+  l.l_shipdate
+order by
+  total_quantity desc
+limit 10;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.e_tsv
new file mode 100644
index 0000000..54473b8
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.e_tsv
@@ -0,0 +1,10 @@
+356846178	515224.1	1995-03-19	0
+501322081	496480.359	1995-02-04	0
+37911232	477743.003	1995-03-21	0
+191691776	472636.195	1995-03-14	0
+493528132	468488.906	1995-02-26	0
+174824647	466075.314	1995-01-21	0
+217865287	465172.779	1995-03-19	0
+103463138	465076.595	1995-03-20	0
+271880324	464526.735	1995-03-02	0
+353610375	462081.866	1995-02-17	0
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.plan
new file mode 100644
index 0000000..a2aae12
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.plan
@@ -0,0 +1,15 @@
+03-05\s+HashJoin.*
+03-07\s+HashToRandomExchange.*
+04-01\s+Filter.*
+04-02\s+RuntimeFilter
+04-03\s+Scan.*lineitem.*columns=\[`l_orderkey`, `l_shipdate`, `l_extendedprice`, `l_discount`\]\]\]\)
+03-06\s+HashToRandomExchange.*
+05-01\s+HashJoin.*
+05-03\s+SelectionVectorRemover
+05-04\s+Filter.*
+05-05\s+RuntimeFilter
+05-06\s+Scan.*orders.*columns=\[`o_custkey`, `o_orderkey`, `o_orderdate`, `o_shippriority`\]\]\]\)
+05-02\s+BroadcastExchange
+06-01\s+SelectionVectorRemover
+06-02\s+Filter.*
+06-03\s+Scan.*customer.*columns=\[`c_mktsegment`, `c_custkey`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.sql
new file mode 100644
index 0000000..faf9951
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.sql
@@ -0,0 +1,33 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 10000000;
+
+select
+  l.l_orderkey,
+  round(sum(l.l_extendedprice * (1 - l.l_discount)), 3) as revenue,
+  o.o_orderdate,
+  o.o_shippriority
+from
+  customer c,
+  orders o,
+  lineitem l
+where
+  c.c_mktsegment = 'HOUSEHOLD'
+  and c.c_custkey = o.o_custkey
+  and l.l_orderkey = o.o_orderkey
+  and o.o_orderdate < date '1995-03-25'
+  and l.l_shipdate > date '1995-03-25'
+group by
+  l.l_orderkey,
+  o.o_orderdate,
+  o.o_shippriority
+order by
+  revenue desc,
+  o.o_orderdate
+limit 10;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.sql.explain
new file mode 100644
index 0000000..3aef024
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_01.sql.explain
@@ -0,0 +1,34 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 10000000;
+
+explain plan for
+select
+  l.l_orderkey,
+  sum(l.l_extendedprice * (1 - l.l_discount)) as revenue,
+  o.o_orderdate,
+  o.o_shippriority
+from
+  customer c,
+  orders o,
+  lineitem l
+where
+  c.c_mktsegment = 'HOUSEHOLD'
+  and c.c_custkey = o.o_custkey
+  and l.l_orderkey = o.o_orderkey
+  and o.o_orderdate < date '1995-03-25'
+  and l.l_shipdate > date '1995-03-25'
+group by
+  l.l_orderkey,
+  o.o_orderdate,
+  o.o_shippriority
+order by
+  revenue desc,
+  o.o_orderdate
+limit 10;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.e_tsv
new file mode 100644
index 0000000..58eb761
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.e_tsv
@@ -0,0 +1,4 @@
+EGYPT	UNITED STATES	1995	5.282512709079098E9
+EGYPT	UNITED STATES	1996	5.2735809932832E9
+UNITED STATES	EGYPT	1995	5.3213732978949E9
+UNITED STATES	EGYPT	1996	5.320488357330402E9
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.plan
new file mode 100644
index 0000000..ba081c2
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.plan
@@ -0,0 +1,26 @@
+03-06\s+HashJoin.*
+03-07\s+Project.*
+03-09\s+HashToRandomExchange.*
+05-01\s+HashJoin.*
+05-03\s+HashToRandomExchange.*
+07-01\s+RuntimeFilter
+07-02\s+Scan.*orders.*columns=\[`o_orderkey`, `o_custkey`\]\]\]\)
+05-02\s+HashToRandomExchange.*
+08-01\s+HashJoin.*
+08-03\s+SelectionVectorRemover
+08-04\s+RuntimeFilter
+08-05\s+Scan.*customer.*columns=\[`c_custkey`, `c_nationkey`\]\]\]\)
+08-02\s+BroadcastExchange
+09-01\s+Scan.*nation.*columns=\[`n_nationkey`, `n_name`\]\]\]\)
+03-08\s+HashToRandomExchange.*
+04-01\s+HashJoin.*
+04-03\s+SelectionVectorRemover
+04-04\s+Filter.*
+04-05\s+RuntimeFilter
+04-06\s+SelectionVectorRemover
+04-07\s+RuntimeFilter
+04-08\s+Scan.*lineitem.*columns=\[`l_suppkey`, `l_orderkey`, `l_shipdate`, `l_extendedprice`, `l_discount`\]\]\]\)
+04-02\s+BroadcastExchange
+06-01\s+HashJoin.*
+06-03\s+Scan.*supplier.*columns=\[`s_suppkey`, `s_nationkey`\]\]\]\)
+06-02\s+Scan.*nation.*columns=\[`n_nationkey`, `n_name`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.sql
new file mode 100644
index 0000000..73f81c0
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.sql
@@ -0,0 +1,50 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 10000000;
+
+select
+  supp_nation,
+  cust_nation,
+  l_year,
+  sum(volume) as revenue
+from
+  (
+    select
+      n1.n_name as supp_nation,
+      n2.n_name as cust_nation,
+      extract(year from l.l_shipdate) as l_year,
+      l.l_extendedprice * (1 - l.l_discount) as volume
+    from
+      supplier s,
+      lineitem l,
+      orders o,
+      customer c,
+      nation n1,
+      nation n2
+    where
+      s.s_suppkey = l.l_suppkey
+      and o.o_orderkey = l.l_orderkey
+      and c.c_custkey = o.o_custkey
+      and s.s_nationkey = n1.n_nationkey
+      and c.c_nationkey = n2.n_nationkey
+      and (
+        (n1.n_name = 'EGYPT' and n2.n_name = 'UNITED STATES')
+        or (n1.n_name = 'UNITED STATES' and n2.n_name = 'EGYPT')
+      )
+      and l.l_shipdate between date '1995-01-01' and date '1996-12-31'
+  ) as shipping
+group by
+  supp_nation,
+  cust_nation,
+  l_year
+order by
+  supp_nation,
+  cust_nation,
+  l_year;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
+
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.sql.explain
new file mode 100644
index 0000000..a114ea2
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_hash_02.sql.explain
@@ -0,0 +1,50 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 10000000;
+
+explain plan for
+select
+  supp_nation,
+  cust_nation,
+  l_year,
+  sum(volume) as revenue
+from
+  (
+    select
+      n1.n_name as supp_nation,
+      n2.n_name as cust_nation,
+      extract(year from l.l_shipdate) as l_year,
+      l.l_extendedprice * (1 - l.l_discount) as volume
+    from
+      supplier s,
+      lineitem l,
+      orders o,
+      customer c,
+      nation n1,
+      nation n2
+    where
+      s.s_suppkey = l.l_suppkey
+      and o.o_orderkey = l.l_orderkey
+      and c.c_custkey = o.o_custkey
+      and s.s_nationkey = n1.n_nationkey
+      and c.c_nationkey = n2.n_nationkey
+      and (
+        (n1.n_name = 'EGYPT' and n2.n_name = 'UNITED STATES')
+        or (n1.n_name = 'UNITED STATES' and n2.n_name = 'EGYPT')
+      )
+      and l.l_shipdate between date '1995-01-01' and date '1996-12-31'
+  ) as shipping
+group by
+  supp_nation,
+  cust_nation,
+  l_year
+order by
+  supp_nation,
+  cust_nation,
+  l_year;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.e_tsv
new file mode 100644
index 0000000..477528e
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.e_tsv
@@ -0,0 +1 @@
+113802
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.plan
new file mode 100644
index 0000000..848051b
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.plan
@@ -0,0 +1,20 @@
+.*01-03\s+HashJoin.*
+01-05\s+SelectionVectorRemover
+01-07\s+RuntimeFilter
+01-09\s+Scan.*lineitem.*columns=\[`l_shipdate`\]\]\]\)
+01-04\s+SelectionVectorRemover
+01-06\s+Sort.*
+01-08\s+BroadcastExchange
+02-01\s+SelectionVectorRemover
+02-02\s+Limit.*
+02-03\s+SingleMergeExchange.*
+03-01\s+OrderedMuxExchange.*
+04-01\s+SelectionVectorRemover
+04-02\s+Limit.*
+04-03\s+SelectionVectorRemover
+04-04\s+TopN.*
+04-05\s+HashAgg.*
+04-06\s+HashToRandomExchange.*
+05-01\s+HashAgg.*
+05-02\s+Project\(distinctdates=\[CAST\(\$0\)\:DATE\]\)
+05-03\s+Scan.*lineitem.*columns=\[`l_shipdate`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.sql
new file mode 100644
index 0000000..2ad7f09
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.sql
@@ -0,0 +1,23 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+select
+  count(*) as row_count
+from
+  lineitem l1
+where
+  l1.l_shipdate IN (
+    select
+      distinct(cast(l2.l_shipdate as date)) as distinctdates
+    from
+      lineitem l2
+    order by
+      distinctdates desc
+    limit 10);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.sql.explain
new file mode 100644
index 0000000..6663dc4
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_date.sql.explain
@@ -0,0 +1,24 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+explain plan for
+select
+  count(*) as row_count
+from
+  lineitem l1
+where
+  l1.l_shipdate IN (
+    select
+      distinct(cast(l2.l_shipdate as date)) as distinctdates
+    from
+      lineitem l2
+    order by
+      distinctdates desc
+    limit 10);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.e_tsv
new file mode 100644
index 0000000..62f9457
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.e_tsv
@@ -0,0 +1 @@
+6
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.plan
new file mode 100644
index 0000000..6443f4b
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.plan
@@ -0,0 +1,9 @@
+.*01-03\s+HashJoin.*
+01-05\s+SelectionVectorRemover
+01-06\s+RuntimeFilter
+01-07\s+Scan.*orders.*columns=\[`o_totalprice`\]\]\]\)
+01-04\s+BroadcastExchange
+02-01\s+StreamAgg.*
+02-02\s+UnionExchange
+03-01\s+StreamAgg.*
+03-02\s+Scan.*lineitem.*columns=\[`l_extendedprice`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.sql
new file mode 100644
index 0000000..37805b5
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.sql
@@ -0,0 +1,20 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+select
+  count(*) as row_count
+from
+  orders o
+where
+  o.o_totalprice = (
+    select
+      max(l.l_extendedprice)
+    from
+      lineitem l);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.sql.explain
new file mode 100644
index 0000000..cc76a2e
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_double.sql.explain
@@ -0,0 +1,21 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+explain plan for
+select
+  count(*) as row_count
+from
+  orders o
+where
+  o.o_totalprice = (
+    select
+      max(l.l_extendedprice)
+    from
+      lineitem l);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.e_tsv
new file mode 100644
index 0000000..d90c867
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.e_tsv
@@ -0,0 +1 @@
+38240.384
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.plan
new file mode 100644
index 0000000..43fcb47
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.plan
@@ -0,0 +1,10 @@
+.*01-03\s+HashJoin.*
+01-05\s+SelectionVectorRemover
+01-06\s+RuntimeFilter
+01-07\s+Scan.*lineitem.*columns=\[`l_shipmode`, `l_extendedprice`\]\]\]\)
+01-04\s+BroadcastExchange.*
+05-01\s+Project\(least_used_ship_mode=\[CAST\(\$0\)\:VARCHAR\(65535\).*
+05-02\s+HashAgg.*
+05-03\s+HashToRandomExchange.*
+06-01\s+HashAgg.*
+06-02\s+Scan.*lineitem.*columns=\[`l_shipmode`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.sql
new file mode 100644
index 0000000..f6a1c6b
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.sql
@@ -0,0 +1,26 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+select
+  round(avg(l1.l_extendedprice), 3) as avg_cost_for_mode
+from
+  lineitem l1,
+  (select
+    cast(l2.l_shipmode as varchar) as least_used_ship_mode,
+    count(*) as num_items
+  from
+    lineitem l2
+  group by
+    l2.l_shipmode
+  order by
+    num_items
+  limit 1) as l3
+where
+  l1.l_shipmode = l3.least_used_ship_mode;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.sql.explain
new file mode 100644
index 0000000..dd21b9a
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/broadcast_join_varchar.sql.explain
@@ -0,0 +1,27 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+explain plan for
+select
+  round(avg(l1.l_extendedprice), 3) as avg_cost_for_mode
+from
+  lineitem l1,
+  (select
+    cast(l2.l_shipmode as varchar) as least_used_ship_mode,
+    count(*) as num_items
+  from
+    lineitem l2
+  group by
+    l2.l_shipmode
+  order by
+    num_items
+  limit 1) as l3
+where
+  l1.l_shipmode = l3.least_used_ship_mode;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.e_tsv
new file mode 100644
index 0000000..5c293e8
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.e_tsv
@@ -0,0 +1 @@
+600037902
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.plan
new file mode 100644
index 0000000..c112462
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.plan
@@ -0,0 +1,16 @@
+.*01-03\s+HashJoin.*
+01-05\s+Project.*
+01-07\s+HashToRandomExchange.*
+02-01\s+UnorderedMuxExchange
+04-01\s+Project.*
+04-02\s+SelectionVectorRemover
+04-03\s+RuntimeFilter
+04-04\s+Scan.*lineitem.*columns=\[`l_shipdate`\]\]\]\)
+01-04\s+HashAgg.*
+01-06\s+Project.*
+01-08\s+HashToRandomExchange.*
+03-01\s+UnorderedMuxExchange
+05-01\s+Project.*
+05-02\s+HashAgg.*
+05-03\s+Project\(EXPR\$0=\[CAST\(\$0\)\:\DATE\]\)
+05-04\s+Scan.*lineitem.*columns=\[`l_shipdate`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.sql
new file mode 100644
index 0000000..b6c0156
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.sql
@@ -0,0 +1,18 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+select
+  count(*) as row_count
+from
+  lineitem l1
+where
+  l1.l_shipdate IN (
+    select
+      distinct(cast(l2.l_shipdate as date))
+    from
+      lineitem l2);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.sql.explain
new file mode 100644
index 0000000..c5bb2c1
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_date.sql.explain
@@ -0,0 +1,19 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+explain plan for
+select
+  count(*) as row_count
+from
+  lineitem l1
+where
+  l1.l_shipdate IN (
+    select
+      distinct(cast(l2.l_shipdate as date))
+    from
+      lineitem l2);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.e_tsv
new file mode 100644
index 0000000..5c293e8
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.e_tsv
@@ -0,0 +1 @@
+600037902
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.plan
new file mode 100644
index 0000000..cdf4227
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.plan
@@ -0,0 +1,16 @@
+.*01-03\s+HashJoin.*
+01-05\s+Project.*
+01-07\s+HashToRandomExchange.*
+02-01\s+UnorderedMuxExchange
+04-01\s+Project.*
+04-02\s+SelectionVectorRemover
+04-03\s+RuntimeFilter
+04-04\s+Scan.*lineitem.*columns=\[`l_discount`\]\]\]\)
+01-04\s+HashAgg.*
+01-06\s+Project.*
+01-08\s+HashToRandomExchange.*
+03-01\s+UnorderedMuxExchange
+05-01\s+Project.*
+05-02\s+HashAgg.*
+05-03\s+Project\(EXPR\$0=\[CAST\(\$0\)\:DOUBLE\]\)
+05-04\s+Scan.*lineitem.*columns=\[`l_discount`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.sql
new file mode 100644
index 0000000..07bdfe3
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.sql
@@ -0,0 +1,18 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+select
+  count(*)
+from
+  lineitem l1
+where
+  l1.l_discount IN (
+    select
+      distinct(cast(l2.l_discount as double))
+    from
+      lineitem l2);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.sql.explain
new file mode 100644
index 0000000..f6a4f33
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_double.sql.explain
@@ -0,0 +1,19 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+explain plan for
+select
+  count(*)
+from
+  lineitem l1
+where
+  l1.l_discount IN (
+    select
+      distinct(cast(l2.l_discount as double))
+    from
+      lineitem l2);
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.e_tsv
new file mode 100644
index 0000000..3c0e795
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.e_tsv
@@ -0,0 +1 @@
+38236.704
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.plan
new file mode 100644
index 0000000..6692890
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.plan
@@ -0,0 +1,23 @@
+.*01-03\s+HashJoin.*
+01-05\s+Project.*
+01-07\s+HashToRandomExchange.*
+02-01\s+UnorderedMuxExchange
+04-01\s+Project.*
+04-02\s+SelectionVectorRemover
+04-03\s+RuntimeFilter
+04-04\s+Scan.*lineitem.*columns=\[`l_shipmode`, `l_extendedprice`\]\]\]\)
+01-04\s+Project.*
+01-06\s+HashToRandomExchange.*
+03-01\s+UnorderedMuxExchange
+05-01\s+Project.*
+05-02\s+Project.*
+05-03\s+SingleMergeExchange.*
+06-01\s+OrderedMuxExchange.*
+07-01\s+SelectionVectorRemover
+07-02\s+Sort.*
+07-03\s+HashToRandomExchange.*
+08-01\s+Project.*
+08-02\s+HashAgg.*
+08-03\s+HashToRandomExchange.*
+09-01\s+HashAgg.*
+09-02\s+Scan.*lineitem.*columns=\[`l_shipmode`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.sql
new file mode 100644
index 0000000..8bcc12e
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.sql
@@ -0,0 +1,23 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+select
+  round(avg(l1.l_extendedprice), 3) as avg_cost_for_mode
+from
+  lineitem l1,
+  (select
+    cast(l2.l_shipmode as varchar) as least_used_ship_mode,
+    count(*) as num_items
+  from
+    lineitem l2
+  group by
+    l2.l_shipmode
+  order by
+    num_items) as l3
+where
+  l1.l_shipmode = l3.least_used_ship_mode;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.sql.explain
new file mode 100644
index 0000000..626050a
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/hash_join_varchar.sql.explain
@@ -0,0 +1,24 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = false;
+
+explain plan for
+select
+  round(avg(l1.l_extendedprice), 3) as avg_cost_for_mode
+from
+  lineitem l1,
+  (select
+    cast(l2.l_shipmode as varchar) as least_used_ship_mode,
+    count(*) as num_items
+  from
+    lineitem l2
+  group by
+    l2.l_shipmode
+  order by
+    num_items) as l3
+where
+  l1.l_shipmode = l3.least_used_ship_mode;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.e_tsv b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.e_tsv
new file mode 100644
index 0000000..2f1b144
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.e_tsv
@@ -0,0 +1,150 @@
+Manufacturer#1	ECONOMY ANODIZED STEEL	4448
+Manufacturer#1	ECONOMY BRUSHED STEEL	4308
+Manufacturer#1	ECONOMY BURNISHED STEEL	4268
+Manufacturer#1	ECONOMY PLATED STEEL	4266
+Manufacturer#1	ECONOMY POLISHED STEEL	4302
+Manufacturer#1	LARGE ANODIZED STEEL	4389
+Manufacturer#1	LARGE BRUSHED STEEL	4254
+Manufacturer#1	LARGE BURNISHED STEEL	4276
+Manufacturer#1	LARGE PLATED STEEL	4351
+Manufacturer#1	LARGE POLISHED STEEL	4281
+Manufacturer#1	MEDIUM ANODIZED STEEL	4230
+Manufacturer#1	MEDIUM BRUSHED STEEL	4247
+Manufacturer#1	MEDIUM BURNISHED STEEL	4286
+Manufacturer#1	MEDIUM PLATED STEEL	4269
+Manufacturer#1	MEDIUM POLISHED STEEL	4274
+Manufacturer#1	PROMO ANODIZED STEEL	4283
+Manufacturer#1	PROMO BRUSHED STEEL	4221
+Manufacturer#1	PROMO BURNISHED STEEL	4315
+Manufacturer#1	PROMO PLATED STEEL	4361
+Manufacturer#1	PROMO POLISHED STEEL	4213
+Manufacturer#1	SMALL ANODIZED STEEL	4354
+Manufacturer#1	SMALL BRUSHED STEEL	4159
+Manufacturer#1	SMALL BURNISHED STEEL	4222
+Manufacturer#1	SMALL PLATED STEEL	4311
+Manufacturer#1	SMALL POLISHED STEEL	4268
+Manufacturer#1	STANDARD ANODIZED STEEL	4362
+Manufacturer#1	STANDARD BRUSHED STEEL	4250
+Manufacturer#1	STANDARD BURNISHED STEEL	4240
+Manufacturer#1	STANDARD PLATED STEEL	4314
+Manufacturer#1	STANDARD POLISHED STEEL	4207
+Manufacturer#2	ECONOMY ANODIZED STEEL	4317
+Manufacturer#2	ECONOMY BRUSHED STEEL	4142
+Manufacturer#2	ECONOMY BURNISHED STEEL	4267
+Manufacturer#2	ECONOMY PLATED STEEL	4229
+Manufacturer#2	ECONOMY POLISHED STEEL	4270
+Manufacturer#2	LARGE ANODIZED STEEL	4314
+Manufacturer#2	LARGE BRUSHED STEEL	4326
+Manufacturer#2	LARGE BURNISHED STEEL	4281
+Manufacturer#2	LARGE PLATED STEEL	4326
+Manufacturer#2	LARGE POLISHED STEEL	4244
+Manufacturer#2	MEDIUM ANODIZED STEEL	4268
+Manufacturer#2	MEDIUM BRUSHED STEEL	4212
+Manufacturer#2	MEDIUM BURNISHED STEEL	4360
+Manufacturer#2	MEDIUM PLATED STEEL	4326
+Manufacturer#2	MEDIUM POLISHED STEEL	4269
+Manufacturer#2	PROMO ANODIZED STEEL	4385
+Manufacturer#2	PROMO BRUSHED STEEL	4347
+Manufacturer#2	PROMO BURNISHED STEEL	4276
+Manufacturer#2	PROMO PLATED STEEL	4144
+Manufacturer#2	PROMO POLISHED STEEL	4388
+Manufacturer#2	SMALL ANODIZED STEEL	4174
+Manufacturer#2	SMALL BRUSHED STEEL	4346
+Manufacturer#2	SMALL BURNISHED STEEL	4375
+Manufacturer#2	SMALL PLATED STEEL	4375
+Manufacturer#2	SMALL POLISHED STEEL	4295
+Manufacturer#2	STANDARD ANODIZED STEEL	4342
+Manufacturer#2	STANDARD BRUSHED STEEL	4247
+Manufacturer#2	STANDARD BURNISHED STEEL	4176
+Manufacturer#2	STANDARD PLATED STEEL	4337
+Manufacturer#2	STANDARD POLISHED STEEL	4183
+Manufacturer#3	ECONOMY ANODIZED STEEL	4314
+Manufacturer#3	ECONOMY BRUSHED STEEL	4298
+Manufacturer#3	ECONOMY BURNISHED STEEL	4254
+Manufacturer#3	ECONOMY PLATED STEEL	4497
+Manufacturer#3	ECONOMY POLISHED STEEL	4083
+Manufacturer#3	LARGE ANODIZED STEEL	4252
+Manufacturer#3	LARGE BRUSHED STEEL	4399
+Manufacturer#3	LARGE BURNISHED STEEL	4181
+Manufacturer#3	LARGE PLATED STEEL	4399
+Manufacturer#3	LARGE POLISHED STEEL	4301
+Manufacturer#3	MEDIUM ANODIZED STEEL	4421
+Manufacturer#3	MEDIUM BRUSHED STEEL	4292
+Manufacturer#3	MEDIUM BURNISHED STEEL	4269
+Manufacturer#3	MEDIUM PLATED STEEL	4320
+Manufacturer#3	MEDIUM POLISHED STEEL	4341
+Manufacturer#3	PROMO ANODIZED STEEL	4216
+Manufacturer#3	PROMO BRUSHED STEEL	4290
+Manufacturer#3	PROMO BURNISHED STEEL	4198
+Manufacturer#3	PROMO PLATED STEEL	4338
+Manufacturer#3	PROMO POLISHED STEEL	4167
+Manufacturer#3	SMALL ANODIZED STEEL	4325
+Manufacturer#3	SMALL BRUSHED STEEL	4227
+Manufacturer#3	SMALL BURNISHED STEEL	4256
+Manufacturer#3	SMALL PLATED STEEL	4311
+Manufacturer#3	SMALL POLISHED STEEL	4197
+Manufacturer#3	STANDARD ANODIZED STEEL	4217
+Manufacturer#3	STANDARD BRUSHED STEEL	4209
+Manufacturer#3	STANDARD BURNISHED STEEL	4394
+Manufacturer#3	STANDARD PLATED STEEL	4300
+Manufacturer#3	STANDARD POLISHED STEEL	4360
+Manufacturer#4	ECONOMY ANODIZED STEEL	4339
+Manufacturer#4	ECONOMY BRUSHED STEEL	4226
+Manufacturer#4	ECONOMY BURNISHED STEEL	4383
+Manufacturer#4	ECONOMY PLATED STEEL	4290
+Manufacturer#4	ECONOMY POLISHED STEEL	4338
+Manufacturer#4	LARGE ANODIZED STEEL	4190
+Manufacturer#4	LARGE BRUSHED STEEL	4188
+Manufacturer#4	LARGE BURNISHED STEEL	4242
+Manufacturer#4	LARGE PLATED STEEL	4246
+Manufacturer#4	LARGE POLISHED STEEL	4220
+Manufacturer#4	MEDIUM ANODIZED STEEL	4289
+Manufacturer#4	MEDIUM BRUSHED STEEL	4227
+Manufacturer#4	MEDIUM BURNISHED STEEL	4429
+Manufacturer#4	MEDIUM PLATED STEEL	4345
+Manufacturer#4	MEDIUM POLISHED STEEL	4287
+Manufacturer#4	PROMO ANODIZED STEEL	4274
+Manufacturer#4	PROMO BRUSHED STEEL	4227
+Manufacturer#4	PROMO BURNISHED STEEL	4292
+Manufacturer#4	PROMO PLATED STEEL	4421
+Manufacturer#4	PROMO POLISHED STEEL	4215
+Manufacturer#4	SMALL ANODIZED STEEL	4254
+Manufacturer#4	SMALL BRUSHED STEEL	4331
+Manufacturer#4	SMALL BURNISHED STEEL	4299
+Manufacturer#4	SMALL PLATED STEEL	4225
+Manufacturer#4	SMALL POLISHED STEEL	4308
+Manufacturer#4	STANDARD ANODIZED STEEL	4361
+Manufacturer#4	STANDARD BRUSHED STEEL	4303
+Manufacturer#4	STANDARD BURNISHED STEEL	4325
+Manufacturer#4	STANDARD PLATED STEEL	4357
+Manufacturer#4	STANDARD POLISHED STEEL	4311
+Manufacturer#5	ECONOMY ANODIZED STEEL	4258
+Manufacturer#5	ECONOMY BRUSHED STEEL	4269
+Manufacturer#5	ECONOMY BURNISHED STEEL	4172
+Manufacturer#5	ECONOMY PLATED STEEL	4325
+Manufacturer#5	ECONOMY POLISHED STEEL	4295
+Manufacturer#5	LARGE ANODIZED STEEL	4169
+Manufacturer#5	LARGE BRUSHED STEEL	4174
+Manufacturer#5	LARGE BURNISHED STEEL	4266
+Manufacturer#5	LARGE PLATED STEEL	4412
+Manufacturer#5	LARGE POLISHED STEEL	4317
+Manufacturer#5	MEDIUM ANODIZED STEEL	4349
+Manufacturer#5	MEDIUM BRUSHED STEEL	4241
+Manufacturer#5	MEDIUM BURNISHED STEEL	4314
+Manufacturer#5	MEDIUM PLATED STEEL	4359
+Manufacturer#5	MEDIUM POLISHED STEEL	4228
+Manufacturer#5	PROMO ANODIZED STEEL	4267
+Manufacturer#5	PROMO BRUSHED STEEL	4403
+Manufacturer#5	PROMO BURNISHED STEEL	4265
+Manufacturer#5	PROMO PLATED STEEL	4308
+Manufacturer#5	PROMO POLISHED STEEL	4203
+Manufacturer#5	SMALL ANODIZED STEEL	4334
+Manufacturer#5	SMALL BRUSHED STEEL	4247
+Manufacturer#5	SMALL BURNISHED STEEL	4301
+Manufacturer#5	SMALL PLATED STEEL	4364
+Manufacturer#5	SMALL POLISHED STEEL	4226
+Manufacturer#5	STANDARD ANODIZED STEEL	4182
+Manufacturer#5	STANDARD BRUSHED STEEL	4299
+Manufacturer#5	STANDARD BURNISHED STEEL	4225
+Manufacturer#5	STANDARD PLATED STEEL	4341
+Manufacturer#5	STANDARD POLISHED STEEL	4300
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.plan b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.plan
new file mode 100644
index 0000000..8aa78f8
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.plan
@@ -0,0 +1,19 @@
+.*03-04\s+HashJoin.*
+03-06\s+HashJoin.*
+03-08\s+SelectionVectorRemover
+03-09\s+RuntimeFilter
+03-10\s+SelectionVectorRemover
+03-11\s+RuntimeFilter
+03-12\s+Scan.*partsupp.*columns=\[`ps_suppkey`, `ps_partkey`\]\]\]\)
+03-07\s+BroadcastExchange
+04-01\s+HashJoin.*
+04-03\s+SelectionVectorRemover
+04-05\s+Filter.*
+04-07\s+Scan.*supplier.*columns=\[`s_nationkey`, `s_suppkey`\]\]\]\)
+04-02\s+SelectionVectorRemover
+04-04\s+Filter.*
+04-06\s+Scan.*nation.*columns=\[`n_nationkey`\]\]\]\)
+03-05\s+BroadcastExchange
+05-01\s+SelectionVectorRemover
+05-02\s+Filter.*
+05-03\s+Scan.*part.*columns=\[`p_partkey`, `p_type`, `p_mfgr`\]\]\]\)
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.sql b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.sql
new file mode 100644
index 0000000..577b803
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.sql
@@ -0,0 +1,31 @@
+set `exec.hashjoin.enable.runtime_filter` = true;

+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;

+set `planner.enable_broadcast_join` = true;

+set `planner.broadcast_threshold` = 2147483647;

+

+select

+  p.p_mfgr,

+  p.p_type,

+  count(*) as num_parts

+from

+  supplier s,

+  part p,

+  partsupp ps,

+  nation n

+where

+  n.n_nationkey = 15

+  and n.n_nationkey = s.s_nationkey

+  and s.s_suppkey = ps.ps_suppkey

+  and ps.ps_partkey = p.p_partkey

+  and p.p_type like '%STEEL%'

+group by

+  p.p_mfgr,

+  p.p_type

+order by

+  p.p_mfgr,

+  p.p_type;

+

+reset `exec.hashjoin.enable.runtime_filter`;

+reset `exec.hashjoin.runtime_filter.max.waiting.time`;

+reset `planner.enable_broadcast_join`;

+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.sql.explain b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.sql.explain
new file mode 100644
index 0000000..dfd3c26
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/nested_broadcast.sql.explain
@@ -0,0 +1,32 @@
+set `exec.hashjoin.enable.runtime_filter` = true;
+set `exec.hashjoin.runtime_filter.max.waiting.time` = 10000;
+set `planner.enable_broadcast_join` = true;
+set `planner.broadcast_threshold` = 2147483647;
+
+explain plan for
+select
+  p.p_mfgr,
+  p.p_type,
+  count(*) as num_parts
+from
+  supplier s,
+  part p,
+  partsupp ps,
+  nation n
+where
+  n.n_nationkey = 15
+  and n.n_nationkey = s.s_nationkey
+  and s.s_suppkey = ps.ps_suppkey
+  and ps.ps_partkey = p.p_partkey
+  and p.p_type like '%STEEL%'
+group by
+  p.p_mfgr,
+  p.p_type
+order by
+  p.p_mfgr,
+  p.p_type;
+
+reset `exec.hashjoin.enable.runtime_filter`;
+reset `exec.hashjoin.runtime_filter.max.waiting.time`;
+reset `planner.enable_broadcast_join`;
+reset `planner.broadcast_threshold`;
\ No newline at end of file
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/tpch_sf100_parquet.json b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/tpch_sf100_parquet.json
new file mode 100644
index 0000000..a6964ae
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/tpch_sf100_parquet.json
@@ -0,0 +1,21 @@
+{
+    "testId": "runtime_filter_tpch_sf100_parquet",
+    "type": "group",
+    "description": "Test Runtime Filter on TPC-H parquet data",
+    "categories": [
+        "advanced"
+    ],
+    "matrices": [
+        {
+            "query-file": ".*.sql",
+            "schema": "dfs.drillTestDirTpch100Parquet",
+            "output-format": "tsv",
+            "expected-file": ".*.e_tsv",
+            "verification-type": [
+                "in-memory"
+            ]
+        }
+    ],
+    "datasources": [
+    ]
+}
diff --git a/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/tpch_sf100_parquet_explain.json b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/tpch_sf100_parquet_explain.json
new file mode 100644
index 0000000..041c23a
--- /dev/null
+++ b/framework/resources/Advanced/runtimefilter/tpch_sf100/parquet/tpch_sf100_parquet_explain.json
@@ -0,0 +1,21 @@
+{
+  "testId": "runtime_filter_tpch_sf100_parquet_plan",
+  "type": "group",
+  "description": "Plan Validation for Runtime Filter on TPC-H parquet data",
+  "categories": [
+    "advanced"
+  ],
+  "matrices": [
+    {
+      "query-file": ".*.sql.explain",
+      "schema": "dfs.drillTestDirTpch100Parquet",
+      "output-format": "tsv",
+      "expected-file": ".*.plan",
+      "verification-type": [
+        "regex"
+      ]
+    }
+  ],
+  "datasources": [
+  ]
+}
\ No newline at end of file