Perform joins tests on same inputs and verify outputs are equal `join_tests` - identifies what join types to test if `debug` flag is set the test will save randomly generated inputs and outputs to user folders, so it is easy to debug a test on top of the failed data
(&self, join_tests: &[JoinTestType], debug: bool)
| 879 | /// if `debug` flag is set the test will save randomly generated inputs and outputs to user folders, |
| 880 | /// so it is easy to debug a test on top of the failed data |
| 881 | async fn run_test(&self, join_tests: &[JoinTestType], debug: bool) { |
| 882 | for batch_size in self.batch_sizes { |
| 883 | let session_config = SessionConfig::new().with_batch_size(*batch_size); |
| 884 | let ctx = SessionContext::new_with_config(session_config); |
| 885 | let task_ctx = ctx.task_ctx(); |
| 886 | |
| 887 | let hj = self.hash_join(); |
| 888 | let hj_collected = collect(hj, task_ctx.clone()).await.unwrap(); |
| 889 | |
| 890 | let smj = self.sort_merge_join(); |
| 891 | let smj_collected = collect(smj, task_ctx.clone()).await.unwrap(); |
| 892 | |
| 893 | let nlj = self.nested_loop_join(); |
| 894 | let nlj_collected = collect(nlj, task_ctx.clone()).await.unwrap(); |
| 895 | |
| 896 | // Get actual row counts(without formatting overhead) for HJ and SMJ |
| 897 | let hj_rows = hj_collected.iter().fold(0, |acc, b| acc + b.num_rows()); |
| 898 | let smj_rows = smj_collected.iter().fold(0, |acc, b| acc + b.num_rows()); |
| 899 | let nlj_rows = nlj_collected.iter().fold(0, |acc, b| acc + b.num_rows()); |
| 900 | |
| 901 | // compare |
| 902 | let smj_formatted = |
| 903 | pretty_format_batches(&smj_collected).unwrap().to_string(); |
| 904 | let hj_formatted = pretty_format_batches(&hj_collected).unwrap().to_string(); |
| 905 | let nlj_formatted = |
| 906 | pretty_format_batches(&nlj_collected).unwrap().to_string(); |
| 907 | |
| 908 | let mut smj_formatted_sorted: Vec<&str> = |
| 909 | smj_formatted.trim().lines().collect(); |
| 910 | smj_formatted_sorted.sort_unstable(); |
| 911 | |
| 912 | let mut hj_formatted_sorted: Vec<&str> = |
| 913 | hj_formatted.trim().lines().collect(); |
| 914 | hj_formatted_sorted.sort_unstable(); |
| 915 | |
| 916 | let mut nlj_formatted_sorted: Vec<&str> = |
| 917 | nlj_formatted.trim().lines().collect(); |
| 918 | nlj_formatted_sorted.sort_unstable(); |
| 919 | |
| 920 | if debug |
| 921 | && ((join_tests.contains(&NljHj) && nlj_rows != hj_rows) |
| 922 | || (join_tests.contains(&HjSmj) && smj_rows != hj_rows)) |
| 923 | { |
| 924 | let fuzz_debug = "fuzz_test_debug"; |
| 925 | std::fs::remove_dir_all(fuzz_debug).unwrap_or(()); |
| 926 | std::fs::create_dir_all(fuzz_debug).unwrap(); |
| 927 | let out_dir_name = &format!("{fuzz_debug}/batch_size_{batch_size}"); |
| 928 | println!( |
| 929 | "Test result data mismatch found. HJ rows {hj_rows}, SMJ rows {smj_rows}, NLJ rows {nlj_rows}" |
| 930 | ); |
| 931 | println!("The debug is ON. Input data will be saved to {out_dir_name}"); |
| 932 | |
| 933 | Self::save_partitioned_batches_as_parquet( |
| 934 | &self.input1, |
| 935 | out_dir_name, |
| 936 | "input1", |
| 937 | ); |
| 938 | Self::save_partitioned_batches_as_parquet( |
no test coverage detected