From 74883ee33a4cef1ac0e2f3649de05d464142ddad Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Wed, 13 Jul 2022 23:09:10 -0700 Subject: [PATCH 01/17] WIP: enable nested types with collect_set Signed-off-by: Navin Kumar --- integration_tests/src/main/python/data_gen.py | 2 ++ integration_tests/src/main/python/hash_aggregate_test.py | 4 ++-- .../src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala | 4 ++-- 3 files changed, 6 insertions(+), 4 deletions(-) diff --git a/integration_tests/src/main/python/data_gen.py b/integration_tests/src/main/python/data_gen.py index e9057b77d6b..6e314b01814 100644 --- a/integration_tests/src/main/python/data_gen.py +++ b/integration_tests/src/main/python/data_gen.py @@ -949,6 +949,8 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False): # all of the basic types in a single struct all_basic_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens)]) +all_basic_plus_array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens + single_level_array_gens)]) + # Some struct gens, but not all because of nesting nonempty_struct_gens_sample = [all_basic_struct_gen, StructGen([['child0', byte_gen], ['child1', all_basic_struct_gen]]), diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 43948645574..24c8dbd038d 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -570,9 +570,9 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_list_op] _repeat_agg_column_for_collect_set_op = [ - RepeatSeqGen(all_basic_struct_gen, length=15), + RepeatSeqGen(all_basic_plus_array_struct_gen, length=15), RepeatSeqGen(StructGen([ - ['c0', all_basic_struct_gen], ['c1', int_gen]]), length=15)] + ['c0', all_basic_plus_array_struct_gen], ['c1', int_gen]]), length=15)] _gen_data_for_collect_set_op = [[ ('a', RepeatSeqGen(LongGen(), length=20)), diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala index a13e3f69cd2..d3f9228f3f4 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala @@ -3332,11 +3332,11 @@ object GpuOverrides extends Logging { "Collect a set of unique elements, not supported in reduction", ExprChecks.fullAgg( TypeSig.ARRAY.nested(TypeSig.commonCudfTypes + TypeSig.DECIMAL_128 + - TypeSig.NULL + TypeSig.STRUCT), + TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY + TypeSig.MAP), TypeSig.ARRAY.nested(TypeSig.all), Seq(ParamCheck("input", (TypeSig.commonCudfTypes + TypeSig.DECIMAL_128 + - TypeSig.NULL + TypeSig.STRUCT).nested(), + TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY + TypeSig.MAP).nested(), TypeSig.all))), (c, conf, p, r) => new TypedImperativeAggExprMeta[CollectSet](c, conf, p, r) { override def convertToGpu(childExprs: Seq[Expression]): GpuExpression = From a0fac741253f27c07e8979442dca2edeeff3f021 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Fri, 15 Jul 2022 11:49:15 -0700 Subject: [PATCH 02/17] tests for collect_set() with nested array types Signed-off-by: Navin Kumar --- integration_tests/src/main/python/data_gen.py | 2 ++ .../src/main/python/hash_aggregate_test.py | 33 +++++++++++++++++-- 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/integration_tests/src/main/python/data_gen.py b/integration_tests/src/main/python/data_gen.py index 6e314b01814..69e20b7b831 100644 --- a/integration_tests/src/main/python/data_gen.py +++ b/integration_tests/src/main/python/data_gen.py @@ -949,6 +949,8 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False): # all of the basic types in a single struct all_basic_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens)]) +array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens)]) + all_basic_plus_array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens + single_level_array_gens)]) # Some struct gens, but not all because of nesting diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 24c8dbd038d..bcd752a3b49 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -570,14 +570,23 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_list_op] _repeat_agg_column_for_collect_set_op = [ - RepeatSeqGen(all_basic_plus_array_struct_gen, length=15), + RepeatSeqGen(all_basic_struct_gen, length=15), RepeatSeqGen(StructGen([ - ['c0', all_basic_plus_array_struct_gen], ['c1', int_gen]]), length=15)] + ['c0', all_basic_struct_gen], ['c1', int_gen]]), length=15)] + +_repeat_agg_column_for_collect_set_op_nested = [ + RepeatSeqGen(array_struct_gen, length=15), + RepeatSeqGen(StructGen([ + ['c0', array_struct_gen], ['c1', int_gen]]), length=15)] _gen_data_for_collect_set_op = [[ ('a', RepeatSeqGen(LongGen(), length=20)), ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_set_op] +_gen_data_for_collect_set_op_nested = [[ + ('a', RepeatSeqGen(LongGen(), length=20)), + ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_set_op_nested] + # very simple test for just a count on decimals 128 values until we can support more with them @ignore_order(local=True) @pytest.mark.parametrize('data_gen', [decimal_gen_128bit], ids=idfn) @@ -642,6 +651,26 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): .groupby('a') .agg(f.sort_array(f.collect_set('b')))) + +# Note, using size() here instead of sort_array(), because sort_array() does not yet +# support sorting arbitrary types +# See https://github.com/NVIDIA/spark-rapids/issues/3715 +# and https://github.com/rapidsai/cudf/issues/11222 +@approximate_float +@ignore_order(local=True) +@incompat +@pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) +def test_hash_groupby_collect_set_on_nested_array_type(data_gen): + conf = { + "spark.rapids.sql.castFloatToString.enabled": "true", + "spark.rapids.sql.castDecimalToString.enabled": "true" + } + + assert_gpu_and_cpu_are_equal_collect( + lambda spark: gen_df(spark, data_gen, length=100) + .groupby('a') + .agg(f.size(f.collect_set('b'))), conf=conf) + @approximate_float @ignore_order(local=True) @incompat From 2aa4d1fc19b38bb5d57eae3f7117682eca708882 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Thu, 21 Jul 2022 19:42:47 -0700 Subject: [PATCH 03/17] preparing to switch to local sorting Signed-off-by: Navin Kumar --- integration_tests/src/main/python/hash_aggregate_test.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index bcd752a3b49..16923a19f16 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -652,7 +652,7 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): .agg(f.sort_array(f.collect_set('b')))) -# Note, using size() here instead of sort_array(), because sort_array() does not yet +# Note, using size() here instead of sort_array(), because sort_array() does not yet # support sorting arbitrary types # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 @@ -661,7 +661,7 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): @incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) def test_hash_groupby_collect_set_on_nested_array_type(data_gen): - conf = { + conf = { "spark.rapids.sql.castFloatToString.enabled": "true", "spark.rapids.sql.castDecimalToString.enabled": "true" } @@ -669,7 +669,7 @@ def test_hash_groupby_collect_set_on_nested_array_type(data_gen): assert_gpu_and_cpu_are_equal_collect( lambda spark: gen_df(spark, data_gen, length=100) .groupby('a') - .agg(f.size(f.collect_set('b'))), conf=conf) + .agg(f.collect_set('b')), conf=conf) @approximate_float @ignore_order(local=True) From 86ca609752982b4c38dbbbb81af323260b4cfc42 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Sat, 23 Jul 2022 15:12:32 -0700 Subject: [PATCH 04/17] Run sort_array on the CPU in order to test collect_set() with Struct[ArrayType] Signed-off-by: Navin Kumar --- .../src/main/python/hash_aggregate_test.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 16923a19f16..3aba6b969ee 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -652,24 +652,28 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): .agg(f.sort_array(f.collect_set('b')))) -# Note, using size() here instead of sort_array(), because sort_array() does not yet -# support sorting arbitrary types +# Note, using sort_array() on the CPU, because sort_array() does not yet +# support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 @approximate_float @ignore_order(local=True) @incompat +@allow_non_gpu("ProjectExec", "SortArray") @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) def test_hash_groupby_collect_set_on_nested_array_type(data_gen): conf = { "spark.rapids.sql.castFloatToString.enabled": "true", - "spark.rapids.sql.castDecimalToString.enabled": "true" + "spark.rapids.sql.castDecimalToString.enabled": "true", + "spark.rapids.sql.expression.SortArray": "false" } assert_gpu_and_cpu_are_equal_collect( lambda spark: gen_df(spark, data_gen, length=100) .groupby('a') - .agg(f.collect_set('b')), conf=conf) + .agg(f.collect_set('b').alias("collect_set")) + .selectExpr("sort_array(collect_set)") + , conf=conf) @approximate_float @ignore_order(local=True) From 4ced7109aaa123852c05d2a9f41e956def2981bb Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Mon, 25 Jul 2022 10:11:11 -0700 Subject: [PATCH 05/17] Test cleanup and right now just ensure that Struct[Array] works with collect set Signed-off-by: Navin Kumar --- integration_tests/src/main/python/data_gen.py | 4 ---- integration_tests/src/main/python/hash_aggregate_test.py | 3 +++ .../src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala | 4 ++-- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/integration_tests/src/main/python/data_gen.py b/integration_tests/src/main/python/data_gen.py index 69e20b7b831..e9057b77d6b 100644 --- a/integration_tests/src/main/python/data_gen.py +++ b/integration_tests/src/main/python/data_gen.py @@ -949,10 +949,6 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False): # all of the basic types in a single struct all_basic_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens)]) -array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens)]) - -all_basic_plus_array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens + single_level_array_gens)]) - # Some struct gens, but not all because of nesting nonempty_struct_gens_sample = [all_basic_struct_gen, StructGen([['child0', byte_gen], ['child1', all_basic_struct_gen]]), diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 3aba6b969ee..90c87411a70 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -569,6 +569,9 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): ('a', RepeatSeqGen(LongGen(), length=20)), ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_list_op] +# data generating for collect_set based-nested Struct[Array] types +array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) + _repeat_agg_column_for_collect_set_op = [ RepeatSeqGen(all_basic_struct_gen, length=15), RepeatSeqGen(StructGen([ diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala index d3f9228f3f4..944f4407311 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala @@ -3332,11 +3332,11 @@ object GpuOverrides extends Logging { "Collect a set of unique elements, not supported in reduction", ExprChecks.fullAgg( TypeSig.ARRAY.nested(TypeSig.commonCudfTypes + TypeSig.DECIMAL_128 + - TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY + TypeSig.MAP), + TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY), TypeSig.ARRAY.nested(TypeSig.all), Seq(ParamCheck("input", (TypeSig.commonCudfTypes + TypeSig.DECIMAL_128 + - TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY + TypeSig.MAP).nested(), + TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY).nested(), TypeSig.all))), (c, conf, p, r) => new TypedImperativeAggExprMeta[CollectSet](c, conf, p, r) { override def convertToGpu(childExprs: Seq[Expression]): GpuExpression = From 0abc00a616adbec86bfe5c8386bd8cedb7b19a99 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Mon, 25 Jul 2022 11:43:01 -0700 Subject: [PATCH 06/17] Updated docs Signed-off-by: Navin Kumar --- docs/supported_ops.md | 18 +++++++++--------- tools/src/main/resources/supportedExprs.csv | 6 +++--- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/docs/supported_ops.md b/docs/supported_ops.md index 645162b647d..9e20f6867c0 100644 --- a/docs/supported_ops.md +++ b/docs/supported_ops.md @@ -15459,9 +15459,9 @@ are limited. S NS NS +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS -NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, ARRAY, MAP, UDT +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS @@ -15480,7 +15480,7 @@ are limited. -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, ARRAY, MAP, UDT +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT @@ -15502,9 +15502,9 @@ are limited. S NS NS +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS -NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, ARRAY, MAP, UDT +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS @@ -15523,7 +15523,7 @@ are limited. -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, ARRAY, MAP, UDT +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT @@ -15545,9 +15545,9 @@ are limited. S NS NS +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS -NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, ARRAY, MAP, UDT +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS @@ -15566,7 +15566,7 @@ are limited. -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, ARRAY, MAP, UDT +PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT diff --git a/tools/src/main/resources/supportedExprs.csv b/tools/src/main/resources/supportedExprs.csv index 1ae14023b5d..caabbaa7883 100644 --- a/tools/src/main/resources/supportedExprs.csv +++ b/tools/src/main/resources/supportedExprs.csv @@ -581,11 +581,11 @@ CollectList,S,`collect_list`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS CollectList,S,`collect_list`,None,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA CollectList,S,`collect_list`,None,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,PS,PS,NS CollectList,S,`collect_list`,None,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,None,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS +CollectSet,S,`collect_set`,None,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS CollectSet,S,`collect_set`,None,aggregation,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS +CollectSet,S,`collect_set`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS CollectSet,S,`collect_set`,None,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,None,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,NS,NS,PS,NS +CollectSet,S,`collect_set`,None,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS CollectSet,S,`collect_set`,None,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA Count,S,`count`,None,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,S,S,PS,PS,PS,S Count,S,`count`,None,aggregation,result,NA,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA From f764d30265548be67955f98fc14e1b0f078e314d Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Mon, 25 Jul 2022 15:28:21 -0700 Subject: [PATCH 07/17] For integration tests, add array of struct case and array of array case Signed-off-by: Navin Kumar --- integration_tests/src/main/python/hash_aggregate_test.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 90c87411a70..2a7c481a06f 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -577,10 +577,15 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): RepeatSeqGen(StructGen([ ['c0', all_basic_struct_gen], ['c1', int_gen]]), length=15)] +all_basic_struct_gen_no_nan = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens_no_nan)]) + _repeat_agg_column_for_collect_set_op_nested = [ RepeatSeqGen(array_struct_gen, length=15), RepeatSeqGen(StructGen([ - ['c0', array_struct_gen], ['c1', int_gen]]), length=15)] + ['c0', array_struct_gen], ['c1', int_gen]]), length=15), + RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)] + +_array_of_array_gen = [RepeatSeqGen(ArrayGen(sub_gen), length=15) for sub_gen in single_level_array_gens_no_nan] _gen_data_for_collect_set_op = [[ ('a', RepeatSeqGen(LongGen(), length=20)), @@ -588,7 +593,7 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): _gen_data_for_collect_set_op_nested = [[ ('a', RepeatSeqGen(LongGen(), length=20)), - ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_set_op_nested] + ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_set_op_nested + _array_of_array_gen] # very simple test for just a count on decimals 128 values until we can support more with them @ignore_order(local=True) From 54d08a8b6df76b3cf91c922428cc9dce347675fa Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Wed, 27 Jul 2022 19:22:28 -0700 Subject: [PATCH 08/17] Fix window aggregation configuration, and add integration tests for collect_set on nested array types Signed-off-by: Navin Kumar --- integration_tests/src/main/python/data_gen.py | 4 + .../src/main/python/hash_aggregate_test.py | 6 +- .../src/main/python/window_function_test.py | 98 +++++++++++++++++++ .../spark/sql/rapids/AggregateFunctions.scala | 3 +- 4 files changed, 105 insertions(+), 6 deletions(-) diff --git a/integration_tests/src/main/python/data_gen.py b/integration_tests/src/main/python/data_gen.py index e9057b77d6b..4533be193aa 100644 --- a/integration_tests/src/main/python/data_gen.py +++ b/integration_tests/src/main/python/data_gen.py @@ -949,6 +949,10 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False): # all of the basic types in a single struct all_basic_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens)]) +all_basic_struct_gen_no_nan = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens_no_nan)]) + +array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) + # Some struct gens, but not all because of nesting nonempty_struct_gens_sample = [all_basic_struct_gen, StructGen([['child0', byte_gen], ['child1', all_basic_struct_gen]]), diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 2a7c481a06f..fba167f54a5 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -569,16 +569,12 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): ('a', RepeatSeqGen(LongGen(), length=20)), ('b', value_gen)] for value_gen in _repeat_agg_column_for_collect_list_op] -# data generating for collect_set based-nested Struct[Array] types -array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) - _repeat_agg_column_for_collect_set_op = [ RepeatSeqGen(all_basic_struct_gen, length=15), RepeatSeqGen(StructGen([ ['c0', all_basic_struct_gen], ['c1', int_gen]]), length=15)] -all_basic_struct_gen_no_nan = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens_no_nan)]) - +# data generating for collect_set based-nested Struct[Array] types _repeat_agg_column_for_collect_set_op_nested = [ RepeatSeqGen(array_struct_gen, length=15), RepeatSeqGen(StructGen([ diff --git a/integration_tests/src/main/python/window_function_test.py b/integration_tests/src/main/python/window_function_test.py index 313e4f57b66..ab2c7688f5c 100644 --- a/integration_tests/src/main/python/window_function_test.py +++ b/integration_tests/src/main/python/window_function_test.py @@ -920,6 +920,28 @@ def test_running_window_function_exec_for_all_aggs(): ('c_fp_nan', RepeatSeqGen(FloatGen().with_special_case(math.nan, 200.0), length=5)), ] +_gen_data_for_collect_set_nested = [ + ('a', RepeatSeqGen(LongGen(), length=20)), + ('b', LongRangeGen()), + ('c_int', RepeatSeqGen(IntegerGen(), length=15)), + ('c_struct_array_1', RepeatSeqGen(array_struct_gen, length=15)), + ('c_struct_array_2', RepeatSeqGen(StructGen([ + ['c0', array_struct_gen], ['c1', int_gen]]), length=14)), + ('c_array_struct', RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)), + ('c_array_array_bool', RepeatSeqGen(ArrayGen(ArrayGen(BooleanGen())), length=15)), + ('c_array_array_int', RepeatSeqGen(ArrayGen(ArrayGen(IntegerGen())), length=15)), + ('c_array_array_long', RepeatSeqGen(ArrayGen(ArrayGen(LongGen())), length=15)), + ('c_array_array_short', RepeatSeqGen(ArrayGen(ArrayGen(ShortGen())), length=15)), + ('c_array_array_date', RepeatSeqGen(ArrayGen(ArrayGen(DateGen())), length=15)), + ('c_array_array_timestamp', RepeatSeqGen(ArrayGen(ArrayGen(TimestampGen())), length=15)), + ('c_array_array_byte', RepeatSeqGen(ArrayGen(ArrayGen(ByteGen())), length=15)), + ('c_array_array_string', RepeatSeqGen(ArrayGen(ArrayGen(StringGen())), length=15)), + ('c_array_array_float', RepeatSeqGen(ArrayGen(ArrayGen(FloatGen(no_nans=True))), length=15)), + ('c_array_array_double', RepeatSeqGen(ArrayGen(ArrayGen(DoubleGen(no_nans=True))), length=15)), + ('c_array_array_decimal_32', RepeatSeqGen(ArrayGen(ArrayGen(DecimalGen(precision=8, scale=3))), length=15)), + ('c_array_array_decimal_64', RepeatSeqGen(ArrayGen(ArrayGen(decimal_gen_64bit)), length=15)), + ('c_array_array_decimal_128', RepeatSeqGen(ArrayGen(ArrayGen(decimal_gen_128bit)), length=15)), +] # SortExec does not support array type, so sort the result locally. @ignore_order(local=True) @@ -977,6 +999,82 @@ def test_window_aggs_for_rows_collect_set(): ) t ''') + +# Note, using sort_array() on the CPU, because sort_array() does not yet +# support sorting certain nested/arbitrary types on the GPU +# See https://github.com/NVIDIA/spark-rapids/issues/3715 +# and https://github.com/rapidsai/cudf/issues/11222 +@approximate_float +@ignore_order(local=True) +@incompat +@allow_non_gpu("ProjectExec", "SortArray") +def test_window_aggs_for_rows_collect_set_nested_array(): + conf = { + "spark.rapids.sql.castFloatToString.enabled": "true", + "spark.rapids.sql.castDecimalToString.enabled": "true", + "spark.rapids.sql.expression.SortArray": "false" + } + + assert_gpu_and_cpu_are_equal_sql( + lambda spark: gen_df(spark, _gen_data_for_collect_set_nested, length=1024), + "window_collect_table", + ''' + select a, b, + sort_array(cc_struct_array_1), + sort_array(cc_struct_array_2), + sort_array(cc_array_struct), + sort_array(cc_array_array_bool), + sort_array(cc_array_array_int), + sort_array(cc_array_array_long), + sort_array(cc_array_array_short), + sort_array(cc_array_array_date), + sort_array(cc_array_array_ts), + sort_array(cc_array_array_byte), + sort_array(cc_array_array_str), + sort_array(cc_array_array_float), + sort_array(cc_array_array_double), + sort_array(cc_array_array_decimal_32), + sort_array(cc_array_array_decimal_64), + sort_array(cc_array_array_decimal_128) + from ( + select a, b, + collect_set(c_struct_array_1) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_struct_array_1, + collect_set(c_struct_array_2) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_struct_array_2, + collect_set(c_array_struct) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_struct, + collect_set(c_array_array_bool) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_bool, + collect_set(c_array_array_int) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_int, + collect_set(c_array_array_long) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_long, + collect_set(c_array_array_short) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_short, + collect_set(c_array_array_date) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_date, + collect_set(c_array_array_timestamp) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_ts, + collect_set(c_array_array_byte) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_byte, + collect_set(c_array_array_string) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_str, + collect_set(c_array_array_float) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_float, + collect_set(c_array_array_double) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_double, + collect_set(c_array_array_decimal_32) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_decimal_32, + collect_set(c_array_array_decimal_64) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_decimal_64, + collect_set(c_array_array_decimal_128) over + (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_decimal_128 + from window_collect_table + ) t + ''', conf=conf) + + # In a distributed setup the order of the partitions returned might be different, so we must ignore the order # but small batch sizes can make sort very slow, so do the final order by locally @ignore_order(local=True) diff --git a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/AggregateFunctions.scala b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/AggregateFunctions.scala index 5c7701f4833..94e76a3ffb6 100644 --- a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/AggregateFunctions.scala +++ b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/AggregateFunctions.scala @@ -1691,7 +1691,8 @@ case class GpuCollectSet( override def windowAggregation( inputs: Seq[(ColumnVector, Int)]): RollingAggregationOnColumn = - RollingAggregation.collectSet().onColumn(inputs.head._2) + RollingAggregation.collectSet(NullPolicy.EXCLUDE, NullEquality.EQUAL, + NaNEquality.UNEQUAL).onColumn(inputs.head._2) } trait CpuToGpuAggregateBufferConverter { From f984ce415ecd469b4dc30ba6bf7fadd86de7a808 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Thu, 28 Jul 2022 07:22:22 -0700 Subject: [PATCH 09/17] Add reduction integration tests Signed-off-by: Navin Kumar --- .../src/main/python/hash_aggregate_test.py | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index fba167f54a5..388dbc9781a 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -679,6 +679,47 @@ def test_hash_groupby_collect_set_on_nested_array_type(data_gen): .selectExpr("sort_array(collect_set)") , conf=conf) +@approximate_float +@ignore_order(local=True) +@incompat +@pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) +def test_hash_reduction_collect_set(data_gen): + assert_gpu_and_cpu_are_equal_collect( + lambda spark: gen_df(spark, data_gen, length=100) + .agg(f.sort_array(f.collect_set('b')), f.count('b'))) + +@approximate_float +@ignore_order(local=True) +@incompat +@pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op, ids=idfn) +def test_hash_reduction_collect_set_on_nested_type(data_gen): + assert_gpu_and_cpu_are_equal_collect( + lambda spark: gen_df(spark, data_gen, length=100) + .agg(f.sort_array(f.collect_set('b')))) + + +# Note, using sort_array() on the CPU, because sort_array() does not yet +# support sorting certain nested/arbitrary types on the GPU +# See https://github.com/NVIDIA/spark-rapids/issues/3715 +# and https://github.com/rapidsai/cudf/issues/11222 +@approximate_float +@ignore_order(local=True) +@incompat +@allow_non_gpu("ProjectExec", "SortArray") +@pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) +def test_hash_reduction_collect_set_on_nested_array_type(data_gen): + conf = { + "spark.rapids.sql.castFloatToString.enabled": "true", + "spark.rapids.sql.castDecimalToString.enabled": "true", + "spark.rapids.sql.expression.SortArray": "false" + } + + assert_gpu_and_cpu_are_equal_collect( + lambda spark: gen_df(spark, data_gen, length=100) + .agg(f.collect_set('b').alias("collect_set")) + .selectExpr("sort_array(collect_set)") + , conf=conf) + @approximate_float @ignore_order(local=True) @incompat From 2dfe8ba50b0b1fd1c3073966dfec84fc07d5b921 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Mon, 1 Aug 2022 21:24:26 -0700 Subject: [PATCH 10/17] Add incompat documentation, and update tests to remove approximate float and incompat flag to remaining test Signed-off-by: Navin Kumar --- docs/configs.md | 2 +- docs/supported_ops.md | 4 ++-- integration_tests/src/main/python/data_gen.py | 2 +- .../src/main/python/hash_aggregate_test.py | 14 +++----------- .../src/main/python/window_function_test.py | 6 +++--- .../com/nvidia/spark/rapids/GpuOverrides.scala | 5 +++-- tools/src/main/resources/supportedExprs.csv | 12 ++++++------ 7 files changed, 19 insertions(+), 26 deletions(-) diff --git a/docs/configs.md b/docs/configs.md index e9f202beac7..0916f6ccee0 100644 --- a/docs/configs.md +++ b/docs/configs.md @@ -344,7 +344,7 @@ Name | SQL Function(s) | Description | Default Value | Notes spark.rapids.sql.expression.ApproximatePercentile|`percentile_approx`, `approx_percentile`|Approximate percentile|true|This is not 100% compatible with the Spark version because the GPU implementation of approx_percentile is not bit-for-bit compatible with Apache Spark| spark.rapids.sql.expression.Average|`avg`, `mean`|Average aggregate operator|true|None| spark.rapids.sql.expression.CollectList|`collect_list`|Collect a list of non-unique elements, not supported in reduction|true|None| -spark.rapids.sql.expression.CollectSet|`collect_set`|Collect a set of unique elements, not supported in reduction|true|None| +spark.rapids.sql.expression.CollectSet|`collect_set`|Collect a set of unique elements|true|This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays, NaN equality is not respected on the GPU but is respected on the CPU| spark.rapids.sql.expression.Count|`count`|Count aggregate operator|true|None| spark.rapids.sql.expression.First|`first_value`, `first`|first aggregate operator|true|None| spark.rapids.sql.expression.Last|`last`, `last_value`|last aggregate operator|true|None| diff --git a/docs/supported_ops.md b/docs/supported_ops.md index 9e20f6867c0..3a95ab08a0a 100644 --- a/docs/supported_ops.md +++ b/docs/supported_ops.md @@ -15441,8 +15441,8 @@ are limited. CollectSet `collect_set` -Collect a set of unique elements, not supported in reduction -None +Collect a set of unique elements +This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays, NaN equality is not respected on the GPU but is respected on the CPU aggregation input S diff --git a/integration_tests/src/main/python/data_gen.py b/integration_tests/src/main/python/data_gen.py index 4533be193aa..2e41927d14b 100644 --- a/integration_tests/src/main/python/data_gen.py +++ b/integration_tests/src/main/python/data_gen.py @@ -951,7 +951,7 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False): all_basic_struct_gen_no_nan = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens_no_nan)]) -array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) +struct_array_gen_no_nans = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) # Some struct gens, but not all because of nesting nonempty_struct_gens_sample = [all_basic_struct_gen, diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 388dbc9781a..58f5463e9a3 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -576,9 +576,9 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): # data generating for collect_set based-nested Struct[Array] types _repeat_agg_column_for_collect_set_op_nested = [ - RepeatSeqGen(array_struct_gen, length=15), + RepeatSeqGen(struct_array_gen_no_nans, length=15), RepeatSeqGen(StructGen([ - ['c0', array_struct_gen], ['c1', int_gen]]), length=15), + ['c0', struct_array_gen_no_nans], ['c1', int_gen]]), length=15), RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)] _array_of_array_gen = [RepeatSeqGen(ArrayGen(sub_gen), length=15) for sub_gen in single_level_array_gens_no_nan] @@ -635,7 +635,6 @@ def test_hash_groupby_collect_list(data_gen, use_obj_hash_agg): conf={'spark.sql.execution.useObjectHashAggregateExec': str(use_obj_hash_agg).lower(), 'spark.sql.shuffle.partitions': '1'}) -@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) @@ -645,7 +644,6 @@ def test_hash_groupby_collect_set(data_gen): .groupby('a') .agg(f.sort_array(f.collect_set('b')), f.count('b'))) -@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op, ids=idfn) @@ -660,7 +658,6 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 -@approximate_float @ignore_order(local=True) @incompat @allow_non_gpu("ProjectExec", "SortArray") @@ -679,7 +676,6 @@ def test_hash_groupby_collect_set_on_nested_array_type(data_gen): .selectExpr("sort_array(collect_set)") , conf=conf) -@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) @@ -688,7 +684,6 @@ def test_hash_reduction_collect_set(data_gen): lambda spark: gen_df(spark, data_gen, length=100) .agg(f.sort_array(f.collect_set('b')), f.count('b'))) -@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op, ids=idfn) @@ -702,7 +697,6 @@ def test_hash_reduction_collect_set_on_nested_type(data_gen): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 -@approximate_float @ignore_order(local=True) @incompat @allow_non_gpu("ProjectExec", "SortArray") @@ -720,7 +714,6 @@ def test_hash_reduction_collect_set_on_nested_array_type(data_gen): .selectExpr("sort_array(collect_set)") , conf=conf) -@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) @@ -734,7 +727,6 @@ def test_hash_groupby_collect_with_single_distinct(data_gen): f.countDistinct('c'), f.count('c'))) -@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_op, ids=idfn) @@ -759,7 +751,6 @@ def test_hash_groupby_single_distinct_collect(data_gen): df_fun=lambda spark: gen_df(spark, data_gen, length=100), table_name="tbl", sql=sql) -@approximate_float @ignore_order(local=True) @pytest.mark.parametrize('data_gen', _gen_data_for_collect_op, ids=idfn) def test_hash_groupby_collect_with_multi_distinct(data_gen): @@ -1157,6 +1148,7 @@ def test_collect_list_reductions(data_gen): _struct_only_nested_gens = [all_basic_struct_gen, StructGen([['child0', byte_gen], ['child1', all_basic_struct_gen]]), StructGen([])] +@incompat @pytest.mark.parametrize('data_gen', non_nan_all_basic_gens + decimal_gens + _struct_only_nested_gens, ids=idfn) diff --git a/integration_tests/src/main/python/window_function_test.py b/integration_tests/src/main/python/window_function_test.py index ab2c7688f5c..134d65f944e 100644 --- a/integration_tests/src/main/python/window_function_test.py +++ b/integration_tests/src/main/python/window_function_test.py @@ -924,9 +924,9 @@ def test_running_window_function_exec_for_all_aggs(): ('a', RepeatSeqGen(LongGen(), length=20)), ('b', LongRangeGen()), ('c_int', RepeatSeqGen(IntegerGen(), length=15)), - ('c_struct_array_1', RepeatSeqGen(array_struct_gen, length=15)), + ('c_struct_array_1', RepeatSeqGen(struct_array_gen_no_nans, length=15)), ('c_struct_array_2', RepeatSeqGen(StructGen([ - ['c0', array_struct_gen], ['c1', int_gen]]), length=14)), + ['c0', struct_array_gen_no_nans], ['c1', int_gen]]), length=14)), ('c_array_struct', RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)), ('c_array_array_bool', RepeatSeqGen(ArrayGen(ArrayGen(BooleanGen())), length=15)), ('c_array_array_int', RepeatSeqGen(ArrayGen(ArrayGen(IntegerGen())), length=15)), @@ -944,6 +944,7 @@ def test_running_window_function_exec_for_all_aggs(): ] # SortExec does not support array type, so sort the result locally. +@incompat @ignore_order(local=True) def test_window_aggs_for_rows_collect_set(): assert_gpu_and_cpu_are_equal_sql( @@ -1004,7 +1005,6 @@ def test_window_aggs_for_rows_collect_set(): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 -@approximate_float @ignore_order(local=True) @incompat @allow_non_gpu("ProjectExec", "SortArray") diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala index 944f4407311..ae48c1e12e4 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala @@ -3329,7 +3329,7 @@ object GpuOverrides extends Logging { override val needsAnsiCheck: Boolean = false }), expr[CollectSet]( - "Collect a set of unique elements, not supported in reduction", + "Collect a set of unique elements", ExprChecks.fullAgg( TypeSig.ARRAY.nested(TypeSig.commonCudfTypes + TypeSig.DECIMAL_128 + TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY), @@ -3357,7 +3357,8 @@ object GpuOverrides extends Logging { // Last does not overflow, so it doesn't need the ANSI check override val needsAnsiCheck: Boolean = false - }), + }).incompat("when using CollectSet with nested types that include arrays, NaN equality is " + + "not respected on the GPU but is respected on the CPU"), expr[StddevPop]( "Aggregation computing population standard deviation", ExprChecks.groupByOnly( diff --git a/tools/src/main/resources/supportedExprs.csv b/tools/src/main/resources/supportedExprs.csv index caabbaa7883..d44bfac38e7 100644 --- a/tools/src/main/resources/supportedExprs.csv +++ b/tools/src/main/resources/supportedExprs.csv @@ -581,12 +581,12 @@ CollectList,S,`collect_list`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS CollectList,S,`collect_list`,None,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA CollectList,S,`collect_list`,None,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,PS,PS,NS CollectList,S,`collect_list`,None,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,None,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS -CollectSet,S,`collect_set`,None,aggregation,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS -CollectSet,S,`collect_set`,None,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,None,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS -CollectSet,S,`collect_set`,None,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA +CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS +CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,aggregation,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA +CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS +CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA +CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS +CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA Count,S,`count`,None,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,S,S,PS,PS,PS,S Count,S,`count`,None,aggregation,result,NA,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA Count,S,`count`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,S,S,PS,PS,PS,S From 753bfe85baca1b66d185344b0f1be09feab1adaf Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Tue, 2 Aug 2022 13:04:43 -0700 Subject: [PATCH 11/17] Revert "Add incompat documentation, and update tests to remove approximate float and incompat flag to remaining test" This reverts commit 2dfe8ba50b0b1fd1c3073966dfec84fc07d5b921. Signed-off-by: Navin Kumar --- docs/configs.md | 2 +- docs/supported_ops.md | 4 ++-- integration_tests/src/main/python/data_gen.py | 2 +- .../src/main/python/hash_aggregate_test.py | 14 +++++++++++--- .../src/main/python/window_function_test.py | 6 +++--- .../com/nvidia/spark/rapids/GpuOverrides.scala | 5 ++--- tools/src/main/resources/supportedExprs.csv | 12 ++++++------ 7 files changed, 26 insertions(+), 19 deletions(-) diff --git a/docs/configs.md b/docs/configs.md index 0916f6ccee0..e9f202beac7 100644 --- a/docs/configs.md +++ b/docs/configs.md @@ -344,7 +344,7 @@ Name | SQL Function(s) | Description | Default Value | Notes spark.rapids.sql.expression.ApproximatePercentile|`percentile_approx`, `approx_percentile`|Approximate percentile|true|This is not 100% compatible with the Spark version because the GPU implementation of approx_percentile is not bit-for-bit compatible with Apache Spark| spark.rapids.sql.expression.Average|`avg`, `mean`|Average aggregate operator|true|None| spark.rapids.sql.expression.CollectList|`collect_list`|Collect a list of non-unique elements, not supported in reduction|true|None| -spark.rapids.sql.expression.CollectSet|`collect_set`|Collect a set of unique elements|true|This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays, NaN equality is not respected on the GPU but is respected on the CPU| +spark.rapids.sql.expression.CollectSet|`collect_set`|Collect a set of unique elements, not supported in reduction|true|None| spark.rapids.sql.expression.Count|`count`|Count aggregate operator|true|None| spark.rapids.sql.expression.First|`first_value`, `first`|first aggregate operator|true|None| spark.rapids.sql.expression.Last|`last`, `last_value`|last aggregate operator|true|None| diff --git a/docs/supported_ops.md b/docs/supported_ops.md index 3a95ab08a0a..9e20f6867c0 100644 --- a/docs/supported_ops.md +++ b/docs/supported_ops.md @@ -15441,8 +15441,8 @@ are limited. CollectSet `collect_set` -Collect a set of unique elements -This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays, NaN equality is not respected on the GPU but is respected on the CPU +Collect a set of unique elements, not supported in reduction +None aggregation input S diff --git a/integration_tests/src/main/python/data_gen.py b/integration_tests/src/main/python/data_gen.py index 2e41927d14b..4533be193aa 100644 --- a/integration_tests/src/main/python/data_gen.py +++ b/integration_tests/src/main/python/data_gen.py @@ -951,7 +951,7 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False): all_basic_struct_gen_no_nan = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens_no_nan)]) -struct_array_gen_no_nans = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) +array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) # Some struct gens, but not all because of nesting nonempty_struct_gens_sample = [all_basic_struct_gen, diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 58f5463e9a3..388dbc9781a 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -576,9 +576,9 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): # data generating for collect_set based-nested Struct[Array] types _repeat_agg_column_for_collect_set_op_nested = [ - RepeatSeqGen(struct_array_gen_no_nans, length=15), + RepeatSeqGen(array_struct_gen, length=15), RepeatSeqGen(StructGen([ - ['c0', struct_array_gen_no_nans], ['c1', int_gen]]), length=15), + ['c0', array_struct_gen], ['c1', int_gen]]), length=15), RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)] _array_of_array_gen = [RepeatSeqGen(ArrayGen(sub_gen), length=15) for sub_gen in single_level_array_gens_no_nan] @@ -635,6 +635,7 @@ def test_hash_groupby_collect_list(data_gen, use_obj_hash_agg): conf={'spark.sql.execution.useObjectHashAggregateExec': str(use_obj_hash_agg).lower(), 'spark.sql.shuffle.partitions': '1'}) +@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) @@ -644,6 +645,7 @@ def test_hash_groupby_collect_set(data_gen): .groupby('a') .agg(f.sort_array(f.collect_set('b')), f.count('b'))) +@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op, ids=idfn) @@ -658,6 +660,7 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 +@approximate_float @ignore_order(local=True) @incompat @allow_non_gpu("ProjectExec", "SortArray") @@ -676,6 +679,7 @@ def test_hash_groupby_collect_set_on_nested_array_type(data_gen): .selectExpr("sort_array(collect_set)") , conf=conf) +@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) @@ -684,6 +688,7 @@ def test_hash_reduction_collect_set(data_gen): lambda spark: gen_df(spark, data_gen, length=100) .agg(f.sort_array(f.collect_set('b')), f.count('b'))) +@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op, ids=idfn) @@ -697,6 +702,7 @@ def test_hash_reduction_collect_set_on_nested_type(data_gen): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 +@approximate_float @ignore_order(local=True) @incompat @allow_non_gpu("ProjectExec", "SortArray") @@ -714,6 +720,7 @@ def test_hash_reduction_collect_set_on_nested_array_type(data_gen): .selectExpr("sort_array(collect_set)") , conf=conf) +@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) @@ -727,6 +734,7 @@ def test_hash_groupby_collect_with_single_distinct(data_gen): f.countDistinct('c'), f.count('c'))) +@approximate_float @ignore_order(local=True) @incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_op, ids=idfn) @@ -751,6 +759,7 @@ def test_hash_groupby_single_distinct_collect(data_gen): df_fun=lambda spark: gen_df(spark, data_gen, length=100), table_name="tbl", sql=sql) +@approximate_float @ignore_order(local=True) @pytest.mark.parametrize('data_gen', _gen_data_for_collect_op, ids=idfn) def test_hash_groupby_collect_with_multi_distinct(data_gen): @@ -1148,7 +1157,6 @@ def test_collect_list_reductions(data_gen): _struct_only_nested_gens = [all_basic_struct_gen, StructGen([['child0', byte_gen], ['child1', all_basic_struct_gen]]), StructGen([])] -@incompat @pytest.mark.parametrize('data_gen', non_nan_all_basic_gens + decimal_gens + _struct_only_nested_gens, ids=idfn) diff --git a/integration_tests/src/main/python/window_function_test.py b/integration_tests/src/main/python/window_function_test.py index 134d65f944e..ab2c7688f5c 100644 --- a/integration_tests/src/main/python/window_function_test.py +++ b/integration_tests/src/main/python/window_function_test.py @@ -924,9 +924,9 @@ def test_running_window_function_exec_for_all_aggs(): ('a', RepeatSeqGen(LongGen(), length=20)), ('b', LongRangeGen()), ('c_int', RepeatSeqGen(IntegerGen(), length=15)), - ('c_struct_array_1', RepeatSeqGen(struct_array_gen_no_nans, length=15)), + ('c_struct_array_1', RepeatSeqGen(array_struct_gen, length=15)), ('c_struct_array_2', RepeatSeqGen(StructGen([ - ['c0', struct_array_gen_no_nans], ['c1', int_gen]]), length=14)), + ['c0', array_struct_gen], ['c1', int_gen]]), length=14)), ('c_array_struct', RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)), ('c_array_array_bool', RepeatSeqGen(ArrayGen(ArrayGen(BooleanGen())), length=15)), ('c_array_array_int', RepeatSeqGen(ArrayGen(ArrayGen(IntegerGen())), length=15)), @@ -944,7 +944,6 @@ def test_running_window_function_exec_for_all_aggs(): ] # SortExec does not support array type, so sort the result locally. -@incompat @ignore_order(local=True) def test_window_aggs_for_rows_collect_set(): assert_gpu_and_cpu_are_equal_sql( @@ -1005,6 +1004,7 @@ def test_window_aggs_for_rows_collect_set(): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 +@approximate_float @ignore_order(local=True) @incompat @allow_non_gpu("ProjectExec", "SortArray") diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala index ae48c1e12e4..944f4407311 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala @@ -3329,7 +3329,7 @@ object GpuOverrides extends Logging { override val needsAnsiCheck: Boolean = false }), expr[CollectSet]( - "Collect a set of unique elements", + "Collect a set of unique elements, not supported in reduction", ExprChecks.fullAgg( TypeSig.ARRAY.nested(TypeSig.commonCudfTypes + TypeSig.DECIMAL_128 + TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY), @@ -3357,8 +3357,7 @@ object GpuOverrides extends Logging { // Last does not overflow, so it doesn't need the ANSI check override val needsAnsiCheck: Boolean = false - }).incompat("when using CollectSet with nested types that include arrays, NaN equality is " + - "not respected on the GPU but is respected on the CPU"), + }), expr[StddevPop]( "Aggregation computing population standard deviation", ExprChecks.groupByOnly( diff --git a/tools/src/main/resources/supportedExprs.csv b/tools/src/main/resources/supportedExprs.csv index d44bfac38e7..caabbaa7883 100644 --- a/tools/src/main/resources/supportedExprs.csv +++ b/tools/src/main/resources/supportedExprs.csv @@ -581,12 +581,12 @@ CollectList,S,`collect_list`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS CollectList,S,`collect_list`,None,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA CollectList,S,`collect_list`,None,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,PS,PS,NS CollectList,S,`collect_list`,None,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS -CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,aggregation,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS -CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA -CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS -CollectSet,S,`collect_set`,This is not 100% compatible with the Spark version because when using CollectSet with nested types that include arrays; NaN equality is not respected on the GPU but is respected on the CPU,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA +CollectSet,S,`collect_set`,None,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS +CollectSet,S,`collect_set`,None,aggregation,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA +CollectSet,S,`collect_set`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS +CollectSet,S,`collect_set`,None,reduction,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA +CollectSet,S,`collect_set`,None,window,input,S,S,S,S,S,S,S,S,PS,S,S,S,NS,NS,PS,NS,PS,NS +CollectSet,S,`collect_set`,None,window,result,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,PS,NA,NA,NA Count,S,`count`,None,aggregation,input,S,S,S,S,S,S,S,S,PS,S,S,S,S,S,PS,PS,PS,S Count,S,`count`,None,aggregation,result,NA,NA,NA,NA,S,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA,NA Count,S,`count`,None,reduction,input,S,S,S,S,S,S,S,S,PS,S,S,S,S,S,PS,PS,PS,S From bd227fabdb774d6557e5acdb0319278c21fb541c Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Tue, 2 Aug 2022 14:15:51 -0700 Subject: [PATCH 12/17] Cleanup test code, rename data_gen var, and remove unnecessary decorators Signed-off-by: Navin Kumar --- integration_tests/src/main/python/data_gen.py | 2 +- .../src/main/python/hash_aggregate_test.py | 21 ++----------------- .../src/main/python/window_function_test.py | 6 ++---- 3 files changed, 5 insertions(+), 24 deletions(-) diff --git a/integration_tests/src/main/python/data_gen.py b/integration_tests/src/main/python/data_gen.py index 4533be193aa..2e41927d14b 100644 --- a/integration_tests/src/main/python/data_gen.py +++ b/integration_tests/src/main/python/data_gen.py @@ -951,7 +951,7 @@ def gen_scalars_for_sql(data_gen, count, seed=0, force_no_nulls=False): all_basic_struct_gen_no_nan = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(all_basic_gens_no_nan)]) -array_struct_gen = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) +struct_array_gen_no_nans = StructGen([['child'+str(ind), sub_gen] for ind, sub_gen in enumerate(single_level_array_gens_no_nan)]) # Some struct gens, but not all because of nesting nonempty_struct_gens_sample = [all_basic_struct_gen, diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 388dbc9781a..e3bdf423bb8 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -576,9 +576,9 @@ def test_hash_pivot_groupby_duplicates_fallback(data_gen): # data generating for collect_set based-nested Struct[Array] types _repeat_agg_column_for_collect_set_op_nested = [ - RepeatSeqGen(array_struct_gen, length=15), + RepeatSeqGen(struct_array_gen_no_nans, length=15), RepeatSeqGen(StructGen([ - ['c0', array_struct_gen], ['c1', int_gen]]), length=15), + ['c0', struct_array_gen_no_nans], ['c1', int_gen]]), length=15), RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)] _array_of_array_gen = [RepeatSeqGen(ArrayGen(sub_gen), length=15) for sub_gen in single_level_array_gens_no_nan] @@ -635,9 +635,7 @@ def test_hash_groupby_collect_list(data_gen, use_obj_hash_agg): conf={'spark.sql.execution.useObjectHashAggregateExec': str(use_obj_hash_agg).lower(), 'spark.sql.shuffle.partitions': '1'}) -@approximate_float @ignore_order(local=True) -@incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) def test_hash_groupby_collect_set(data_gen): assert_gpu_and_cpu_are_equal_collect( @@ -645,9 +643,7 @@ def test_hash_groupby_collect_set(data_gen): .groupby('a') .agg(f.sort_array(f.collect_set('b')), f.count('b'))) -@approximate_float @ignore_order(local=True) -@incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op, ids=idfn) def test_hash_groupby_collect_set_on_nested_type(data_gen): assert_gpu_and_cpu_are_equal_collect( @@ -660,9 +656,7 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 -@approximate_float @ignore_order(local=True) -@incompat @allow_non_gpu("ProjectExec", "SortArray") @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) def test_hash_groupby_collect_set_on_nested_array_type(data_gen): @@ -679,18 +673,14 @@ def test_hash_groupby_collect_set_on_nested_array_type(data_gen): .selectExpr("sort_array(collect_set)") , conf=conf) -@approximate_float @ignore_order(local=True) -@incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) def test_hash_reduction_collect_set(data_gen): assert_gpu_and_cpu_are_equal_collect( lambda spark: gen_df(spark, data_gen, length=100) .agg(f.sort_array(f.collect_set('b')), f.count('b'))) -@approximate_float @ignore_order(local=True) -@incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op, ids=idfn) def test_hash_reduction_collect_set_on_nested_type(data_gen): assert_gpu_and_cpu_are_equal_collect( @@ -702,9 +692,7 @@ def test_hash_reduction_collect_set_on_nested_type(data_gen): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 -@approximate_float @ignore_order(local=True) -@incompat @allow_non_gpu("ProjectExec", "SortArray") @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) def test_hash_reduction_collect_set_on_nested_array_type(data_gen): @@ -720,9 +708,7 @@ def test_hash_reduction_collect_set_on_nested_array_type(data_gen): .selectExpr("sort_array(collect_set)") , conf=conf) -@approximate_float @ignore_order(local=True) -@incompat @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) def test_hash_groupby_collect_with_single_distinct(data_gen): # test collect_ops with other distinct aggregations @@ -734,9 +720,7 @@ def test_hash_groupby_collect_with_single_distinct(data_gen): f.countDistinct('c'), f.count('c'))) -@approximate_float @ignore_order(local=True) -@incompat @pytest.mark.parametrize('data_gen', _gen_data_for_collect_op, ids=idfn) def test_hash_groupby_single_distinct_collect(data_gen): # test distinct collect @@ -759,7 +743,6 @@ def test_hash_groupby_single_distinct_collect(data_gen): df_fun=lambda spark: gen_df(spark, data_gen, length=100), table_name="tbl", sql=sql) -@approximate_float @ignore_order(local=True) @pytest.mark.parametrize('data_gen', _gen_data_for_collect_op, ids=idfn) def test_hash_groupby_collect_with_multi_distinct(data_gen): diff --git a/integration_tests/src/main/python/window_function_test.py b/integration_tests/src/main/python/window_function_test.py index ab2c7688f5c..7def8c2160e 100644 --- a/integration_tests/src/main/python/window_function_test.py +++ b/integration_tests/src/main/python/window_function_test.py @@ -924,9 +924,9 @@ def test_running_window_function_exec_for_all_aggs(): ('a', RepeatSeqGen(LongGen(), length=20)), ('b', LongRangeGen()), ('c_int', RepeatSeqGen(IntegerGen(), length=15)), - ('c_struct_array_1', RepeatSeqGen(array_struct_gen, length=15)), + ('c_struct_array_1', RepeatSeqGen(struct_array_gen_no_nans, length=15)), ('c_struct_array_2', RepeatSeqGen(StructGen([ - ['c0', array_struct_gen], ['c1', int_gen]]), length=14)), + ['c0', struct_array_gen_no_nans], ['c1', int_gen]]), length=14)), ('c_array_struct', RepeatSeqGen(ArrayGen(all_basic_struct_gen_no_nan), length=15)), ('c_array_array_bool', RepeatSeqGen(ArrayGen(ArrayGen(BooleanGen())), length=15)), ('c_array_array_int', RepeatSeqGen(ArrayGen(ArrayGen(IntegerGen())), length=15)), @@ -1004,9 +1004,7 @@ def test_window_aggs_for_rows_collect_set(): # support sorting certain nested/arbitrary types on the GPU # See https://github.com/NVIDIA/spark-rapids/issues/3715 # and https://github.com/rapidsai/cudf/issues/11222 -@approximate_float @ignore_order(local=True) -@incompat @allow_non_gpu("ProjectExec", "SortArray") def test_window_aggs_for_rows_collect_set_nested_array(): conf = { From 4cca48aee0a15a62164802740ab23f089310ece8 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Tue, 2 Aug 2022 15:58:53 -0700 Subject: [PATCH 13/17] Update CollectSet with limited hasNans requirement and update tests Signed-off-by: Navin Kumar --- .../src/main/python/hash_aggregate_test.py | 8 ++++---- .../src/main/python/window_function_test.py | 9 +++++++-- .../com/nvidia/spark/rapids/GpuOverrides.scala | 14 ++++++++++++++ 3 files changed, 25 insertions(+), 6 deletions(-) diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index e3bdf423bb8..0e77c6c12b0 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -660,11 +660,11 @@ def test_hash_groupby_collect_set_on_nested_type(data_gen): @allow_non_gpu("ProjectExec", "SortArray") @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) def test_hash_groupby_collect_set_on_nested_array_type(data_gen): - conf = { + conf = copy_and_update(_no_nans_float_conf, { "spark.rapids.sql.castFloatToString.enabled": "true", "spark.rapids.sql.castDecimalToString.enabled": "true", "spark.rapids.sql.expression.SortArray": "false" - } + }) assert_gpu_and_cpu_are_equal_collect( lambda spark: gen_df(spark, data_gen, length=100) @@ -696,11 +696,11 @@ def test_hash_reduction_collect_set_on_nested_type(data_gen): @allow_non_gpu("ProjectExec", "SortArray") @pytest.mark.parametrize('data_gen', _gen_data_for_collect_set_op_nested, ids=idfn) def test_hash_reduction_collect_set_on_nested_array_type(data_gen): - conf = { + conf = copy_and_update(_no_nans_float_conf, { "spark.rapids.sql.castFloatToString.enabled": "true", "spark.rapids.sql.castDecimalToString.enabled": "true", "spark.rapids.sql.expression.SortArray": "false" - } + }) assert_gpu_and_cpu_are_equal_collect( lambda spark: gen_df(spark, data_gen, length=100) diff --git a/integration_tests/src/main/python/window_function_test.py b/integration_tests/src/main/python/window_function_test.py index 7def8c2160e..0a3d0075d72 100644 --- a/integration_tests/src/main/python/window_function_test.py +++ b/integration_tests/src/main/python/window_function_test.py @@ -139,6 +139,11 @@ FloatGen(no_nans=True, special_cases=[]), DoubleGen(no_nans=True, special_cases=[]), string_gen, boolean_gen, date_gen, timestamp_gen, null_gen] +_no_nans_float_conf = {'spark.rapids.sql.variableFloatAgg.enabled': 'true', + 'spark.rapids.sql.hasNans': 'false', + 'spark.rapids.sql.castStringToFloat.enabled': 'true' + } + @ignore_order @pytest.mark.parametrize('data_gen', [decimal_gen_128bit], ids=idfn) def test_decimal128_count_window(data_gen): @@ -1007,11 +1012,11 @@ def test_window_aggs_for_rows_collect_set(): @ignore_order(local=True) @allow_non_gpu("ProjectExec", "SortArray") def test_window_aggs_for_rows_collect_set_nested_array(): - conf = { + conf = copy_and_update(_no_nans_float_conf, { "spark.rapids.sql.castFloatToString.enabled": "true", "spark.rapids.sql.castDecimalToString.enabled": "true", "spark.rapids.sql.expression.SortArray": "false" - } + }) assert_gpu_and_cpu_are_equal_sql( lambda spark: gen_df(spark, _gen_data_for_collect_set_nested, length=1024), diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala index 944f4407311..a4ed5db7fe1 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala @@ -3339,6 +3339,20 @@ object GpuOverrides extends Logging { TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY).nested(), TypeSig.all))), (c, conf, p, r) => new TypedImperativeAggExprMeta[CollectSet](c, conf, p, r) { + + private def isNestedArrayType(dt: DataType): Boolean = { + dt match { + case StructType(fields) => fields.exists(_.dataType.isInstanceOf[ArrayType]) + case ArrayType(et, _) => et.isInstanceOf[ArrayType] || et.isInstanceOf[StructType] + case _ => false + } + } + + override def tagAggForGpu(): Unit = { + if (isNestedArrayType(c.child.dataType)) { + checkAndTagFloatNanAgg("CollectSet", c.child.dataType, conf, this) + } + } override def convertToGpu(childExprs: Seq[Expression]): GpuExpression = GpuCollectSet(childExprs.head, c.mutableAggBufferOffset, c.inputAggBufferOffset) From 0ef4784e122e1874734607d7f327016b749fde4f Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Tue, 2 Aug 2022 16:31:07 -0700 Subject: [PATCH 14/17] Update partial support docs for CollectSet Signed-off-by: Navin Kumar --- docs/supported_ops.md | 12 ++++++------ .../scala/com/nvidia/spark/rapids/GpuOverrides.scala | 6 +++++- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/docs/supported_ops.md b/docs/supported_ops.md index 9e20f6867c0..0d33620ecb3 100644 --- a/docs/supported_ops.md +++ b/docs/supported_ops.md @@ -15459,9 +15459,9 @@ are limited. S NS NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT +PS
Support for arrays of arrays of floats/doubles requires spark.rapids.sql.hasNans to be set to false;
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT +PS
Support for structs containing float/double array columns requires spark.rapids.sql.hasNans to be set to false;
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS @@ -15502,9 +15502,9 @@ are limited. S NS NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT +PS
Support for arrays of arrays of floats/doubles requires spark.rapids.sql.hasNans to be set to false;
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT +PS
Support for structs containing float/double array columns requires spark.rapids.sql.hasNans to be set to false;
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS @@ -15545,9 +15545,9 @@ are limited. S NS NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT +PS
Support for arrays of arrays of floats/doubles requires spark.rapids.sql.hasNans to be set to false;
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS -PS
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT +PS
Support for structs containing float/double array columns requires spark.rapids.sql.hasNans to be set to false;
UTC is only supported TZ for child TIMESTAMP;
unsupported child types BINARY, CALENDAR, MAP, UDT NS diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala index a4ed5db7fe1..deac7898de1 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala @@ -3336,7 +3336,11 @@ object GpuOverrides extends Logging { TypeSig.ARRAY.nested(TypeSig.all), Seq(ParamCheck("input", (TypeSig.commonCudfTypes + TypeSig.DECIMAL_128 + - TypeSig.NULL + TypeSig.STRUCT + TypeSig.ARRAY).nested(), + TypeSig.NULL + + TypeSig.STRUCT.withPsNote(TypeEnum.STRUCT, "Support for structs containing " + + s"float/double array columns requires ${RapidsConf.HAS_NANS} to be set to false") + + TypeSig.ARRAY.withPsNote(TypeEnum.ARRAY, "Support for arrays of arrays of " + + s"floats/doubles requires ${RapidsConf.HAS_NANS} to be set to false")).nested(), TypeSig.all))), (c, conf, p, r) => new TypedImperativeAggExprMeta[CollectSet](c, conf, p, r) { From 14b6cedee7caac69615556b56ab1d4ffb2d40e95 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Wed, 3 Aug 2022 07:29:13 -0700 Subject: [PATCH 15/17] Make this check recursive to handle structs that contain structs of arrays Signed-off-by: Navin Kumar --- .../scala/com/nvidia/spark/rapids/GpuOverrides.scala | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala index deac7898de1..9299d3de487 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala @@ -3346,7 +3346,14 @@ object GpuOverrides extends Logging { private def isNestedArrayType(dt: DataType): Boolean = { dt match { - case StructType(fields) => fields.exists(_.dataType.isInstanceOf[ArrayType]) + case StructType(fields) => + fields.exists { field => + field.dataType match { + case sdt: StructType => isNestedArrayType(sdt) + case _: ArrayType => true + case _ => false + } + } case ArrayType(et, _) => et.isInstanceOf[ArrayType] || et.isInstanceOf[StructType] case _ => false } From 7f928c6c143fae5b39b1c99aa74f0b76d3d7843d Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Thu, 4 Aug 2022 17:28:40 -0700 Subject: [PATCH 16/17] Update integration test for collect_set on nested array types to handle Spark 3.3.0+ optimization Signed-off-by: Navin Kumar --- .../src/main/python/hash_aggregate_test.py | 25 +++++++++++++------ 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/integration_tests/src/main/python/hash_aggregate_test.py b/integration_tests/src/main/python/hash_aggregate_test.py index 0e77c6c12b0..09e4abd89a5 100644 --- a/integration_tests/src/main/python/hash_aggregate_test.py +++ b/integration_tests/src/main/python/hash_aggregate_test.py @@ -666,12 +666,18 @@ def test_hash_groupby_collect_set_on_nested_array_type(data_gen): "spark.rapids.sql.expression.SortArray": "false" }) - assert_gpu_and_cpu_are_equal_collect( - lambda spark: gen_df(spark, data_gen, length=100) - .groupby('a') + def do_it(spark): + df = gen_df(spark, data_gen, length=100)\ + .groupby('a')\ .agg(f.collect_set('b').alias("collect_set")) + # pull out the rdd and schema and create a new dataframe to run SortArray + # to handle Spark 3.3.0+ optimization that moves SortArray from ProjectExec + # to ObjectHashAggregateExec + return spark.createDataFrame(df.rdd, schema=df.schema)\ .selectExpr("sort_array(collect_set)") - , conf=conf) + + assert_gpu_and_cpu_are_equal_collect(do_it, conf=conf) + @ignore_order(local=True) @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) @@ -702,11 +708,16 @@ def test_hash_reduction_collect_set_on_nested_array_type(data_gen): "spark.rapids.sql.expression.SortArray": "false" }) - assert_gpu_and_cpu_are_equal_collect( - lambda spark: gen_df(spark, data_gen, length=100) + def do_it(spark): + df = gen_df(spark, data_gen, length=100)\ .agg(f.collect_set('b').alias("collect_set")) + # pull out the rdd and schema and create a new dataframe to run SortArray + # to handle Spark 3.3.0+ optimization that moves SortArray from ProjectExec + # to ObjectHashAggregateExec + return spark.createDataFrame(df.rdd, schema=df.schema)\ .selectExpr("sort_array(collect_set)") - , conf=conf) + + assert_gpu_and_cpu_are_equal_collect(do_it, conf=conf) @ignore_order(local=True) @pytest.mark.parametrize('data_gen', _full_gen_data_for_collect_op, ids=idfn) From 6f1e0dabf1b8e5b1f60a0db54af708aa0d6b7897 Mon Sep 17 00:00:00 2001 From: Navin Kumar Date: Thu, 4 Aug 2022 20:02:37 -0700 Subject: [PATCH 17/17] Update window_function_test to handle optimization performed by Databricks Signed-off-by: Navin Kumar --- .../src/main/python/window_function_test.py | 58 ++++++++++--------- 1 file changed, 32 insertions(+), 26 deletions(-) diff --git a/integration_tests/src/main/python/window_function_test.py b/integration_tests/src/main/python/window_function_test.py index 0a3d0075d72..b18fa095c5c 100644 --- a/integration_tests/src/main/python/window_function_test.py +++ b/integration_tests/src/main/python/window_function_test.py @@ -1018,29 +1018,11 @@ def test_window_aggs_for_rows_collect_set_nested_array(): "spark.rapids.sql.expression.SortArray": "false" }) - assert_gpu_and_cpu_are_equal_sql( - lambda spark: gen_df(spark, _gen_data_for_collect_set_nested, length=1024), - "window_collect_table", - ''' - select a, b, - sort_array(cc_struct_array_1), - sort_array(cc_struct_array_2), - sort_array(cc_array_struct), - sort_array(cc_array_array_bool), - sort_array(cc_array_array_int), - sort_array(cc_array_array_long), - sort_array(cc_array_array_short), - sort_array(cc_array_array_date), - sort_array(cc_array_array_ts), - sort_array(cc_array_array_byte), - sort_array(cc_array_array_str), - sort_array(cc_array_array_float), - sort_array(cc_array_array_double), - sort_array(cc_array_array_decimal_32), - sort_array(cc_array_array_decimal_64), - sort_array(cc_array_array_decimal_128) - from ( - select a, b, + def do_it(spark): + df = gen_df(spark, _gen_data_for_collect_set_nested, length=512) + df.createOrReplaceTempView("window_collect_table") + df = spark.sql( + """select a, b, collect_set(c_struct_array_1) over (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_struct_array_1, collect_set(c_struct_array_2) over @@ -1073,9 +1055,33 @@ def test_window_aggs_for_rows_collect_set_nested_array(): (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_decimal_64, collect_set(c_array_array_decimal_128) over (partition by a order by b,c_int rows between CURRENT ROW and UNBOUNDED FOLLOWING) as cc_array_array_decimal_128 - from window_collect_table - ) t - ''', conf=conf) + from window_collect_table + """) + df = spark.createDataFrame(df.rdd, schema=df.schema) + # pull out the rdd and schema and create a new dataframe to run SortArray + # to handle Databricks 10.4+ optimization that moves SortArray from ProjectExec + # to ObjectHashAggregateExec + df.createOrReplaceTempView("window_collect_table_2") + return spark.sql("""select a, b, + sort_array(cc_struct_array_1), + sort_array(cc_struct_array_2), + sort_array(cc_array_struct), + sort_array(cc_array_array_bool), + sort_array(cc_array_array_int), + sort_array(cc_array_array_long), + sort_array(cc_array_array_short), + sort_array(cc_array_array_date), + sort_array(cc_array_array_ts), + sort_array(cc_array_array_byte), + sort_array(cc_array_array_str), + sort_array(cc_array_array_float), + sort_array(cc_array_array_double), + sort_array(cc_array_array_decimal_32), + sort_array(cc_array_array_decimal_64), + sort_array(cc_array_array_decimal_128) + from window_collect_table_2 + """) + assert_gpu_and_cpu_are_equal_collect(do_it, conf=conf) # In a distributed setup the order of the partitions returned might be different, so we must ignore the order