Fix Float and Decimal coercion

Before the change, Float* and Decimal* would coerce to a decimal type.
However, decimal cannot store all the float values: different range,
NaN, and infinity. Coercing to floating point is more desirable and also
what other systems typically do.

Author: findepi

datafusion/core/tests/parquet/mod.rs

@@ -184,7 +184,13 @@ impl TestOutput {
 /// and the appropriate scenario
 impl ContextWithParquet {
     async fn new(scenario: Scenario, unit: Unit) -> Self {
-        Self::with_config(scenario, unit, SessionConfig::new()).await
+        let mut session_config = SessionConfig::new();
+        // TODO (https://github.com/apache/datafusion/issues/12817) once this is the default behavior, remove from here
+        session_config
+            .options_mut()
+            .sql_parser
+            .parse_float_as_decimal = true;
+        Self::with_config(scenario, unit, session_config).await
     }
 
     async fn with_config(

datafusion/expr-common/src/type_coercion/binary.rs

@@ -928,9 +928,6 @@ fn coerce_numeric_type_to_decimal(numeric_type: &DataType) -> Option<DataType> {
         Int16 | UInt16 => Some(Decimal128(5, 0)),
         Int32 | UInt32 => Some(Decimal128(10, 0)),
         Int64 | UInt64 => Some(Decimal128(20, 0)),
-        // TODO if we convert the floating-point data to the decimal type, it maybe overflow.
-        Float32 => Some(Decimal128(14, 7)),
-        Float64 => Some(Decimal128(30, 15)),
         _ => None,
     }
 }
@@ -946,9 +943,6 @@ fn coerce_numeric_type_to_decimal256(numeric_type: &DataType) -> Option<DataType
         Int16 | UInt16 => Some(Decimal256(5, 0)),
         Int32 | UInt32 => Some(Decimal256(10, 0)),
         Int64 | UInt64 => Some(Decimal256(20, 0)),
-        // TODO if we convert the floating-point data to the decimal type, it maybe overflow.
-        Float32 => Some(Decimal256(14, 7)),
-        Float64 => Some(Decimal256(30, 15)),
         _ => None,
     }
 }
@@ -1494,8 +1488,8 @@ mod tests {
             DataType::Decimal128(20, 3),
             DataType::Decimal128(20, 3),
             DataType::Decimal128(23, 3),
-            DataType::Decimal128(24, 7),
-            DataType::Decimal128(32, 15),
+            DataType::Float32,
+            DataType::Float64,
             DataType::Decimal128(38, 10),
             DataType::Decimal128(25, 8),
             DataType::Decimal128(20, 3),
@@ -1541,14 +1535,8 @@ mod tests {
             coerce_numeric_type_to_decimal(&DataType::Int64).unwrap(),
             DataType::Decimal128(20, 0)
         );
-        assert_eq!(
-            coerce_numeric_type_to_decimal(&DataType::Float32).unwrap(),
-            DataType::Decimal128(14, 7)
-        );
-        assert_eq!(
-            coerce_numeric_type_to_decimal(&DataType::Float64).unwrap(),
-            DataType::Decimal128(30, 15)
-        );
+        assert_eq!(coerce_numeric_type_to_decimal(&DataType::Float32), None);
+        assert_eq!(coerce_numeric_type_to_decimal(&DataType::Float64), None);
     }
 
     #[test]
@@ -2013,7 +2001,7 @@ mod tests {
             DataType::Float64,
             DataType::Decimal128(10, 3),
             Operator::Gt,
-            DataType::Decimal128(30, 15)
+            DataType::Float64
         );
         test_coercion_binary_rule!(
             DataType::Int64,

datafusion/sqllogictest/test_files/math.slt

@@ -694,3 +694,9 @@ select FACTORIAL(350943270);
 
 statement ok
 drop table signed_integers
+
+# Should not fail. The operands should coerce to float
+query B
+SELECT '1'::decimal(10,0) = '1e40'::double;
+----
+false

datafusion/sqllogictest/test_files/tpch/plans/q1.slt.part

@@ -57,4 +57,4 @@ physical_plan
 08)--------------ProjectionExec: expr=[l_extendedprice@1 * (Some(1),20,0 - l_discount@2) as __common_expr_1, l_quantity@0 as l_quantity, l_extendedprice@1 as l_extendedprice, l_discount@2 as l_discount, l_tax@3 as l_tax, l_returnflag@4 as l_returnflag, l_linestatus@5 as l_linestatus]
 09)----------------CoalesceBatchesExec: target_batch_size=8192
 10)------------------FilterExec: l_shipdate@6 <= 1998-09-02, projection=[l_quantity@0, l_extendedprice@1, l_discount@2, l_tax@3, l_returnflag@4, l_linestatus@5]
-11)--------------------CsvExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:0..18561749], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:18561749..37123498], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:37123498..55685247], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:55685247..74246996]]}, projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], has_header=false
+11)--------------------CsvExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:0..18561749], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:18561749..37123498], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:37123498..55685247], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:55685247..74246996]]}, projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], has_header=false

datafusion/sqllogictest/test_files/tpch/plans/q11.slt.part

@@ -49,7 +49,7 @@ limit 10;
 logical_plan
 01)Sort: value DESC NULLS FIRST, fetch=10
 02)--Projection: partsupp.ps_partkey, sum(partsupp.ps_supplycost * partsupp.ps_availqty) AS value
-03)----Inner Join:  Filter: CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty) AS Decimal128(38, 15)) > __scalar_sq_1.sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)
+03)----Inner Join:  Filter: CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty) AS Float64) > __scalar_sq_1.sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)
 04)------Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[sum(partsupp.ps_supplycost * CAST(partsupp.ps_availqty AS Decimal128(10, 0)))]]
 05)--------Projection: partsupp.ps_partkey, partsupp.ps_availqty, partsupp.ps_supplycost
 06)----------Inner Join: supplier.s_nationkey = nation.n_nationkey
@@ -61,7 +61,7 @@ logical_plan
 12)--------------Filter: nation.n_name = Utf8("GERMANY")
 13)----------------TableScan: nation projection=[n_nationkey, n_name], partial_filters=[nation.n_name = Utf8("GERMANY")]
 14)------SubqueryAlias: __scalar_sq_1
-15)--------Projection: CAST(CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty) AS Float64) * Float64(0.0001) AS Decimal128(38, 15))
+15)--------Projection: CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty) AS Float64) * Float64(0.0001)
 16)----------Aggregate: groupBy=[[]], aggr=[[sum(partsupp.ps_supplycost * CAST(partsupp.ps_availqty AS Decimal128(10, 0)))]]
 17)------------Projection: partsupp.ps_availqty, partsupp.ps_supplycost
 18)--------------Inner Join: supplier.s_nationkey = nation.n_nationkey
@@ -75,7 +75,7 @@ logical_plan
 physical_plan
 01)SortExec: TopK(fetch=10), expr=[value@1 DESC], preserve_partitioning=[false]
 02)--ProjectionExec: expr=[ps_partkey@0 as ps_partkey, sum(partsupp.ps_supplycost * partsupp.ps_availqty)@1 as value]
-03)----NestedLoopJoinExec: join_type=Inner, filter=CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty)@0 AS Decimal128(38, 15)) > sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)@1, projection=[ps_partkey@0, sum(partsupp.ps_supplycost * partsupp.ps_availqty)@1]
+03)----NestedLoopJoinExec: join_type=Inner, filter=CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty)@0 AS Float64) > sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)@1, projection=[ps_partkey@0, sum(partsupp.ps_supplycost * partsupp.ps_availqty)@1]
 04)------CoalescePartitionsExec
 05)--------AggregateExec: mode=FinalPartitioned, gby=[ps_partkey@0 as ps_partkey], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
 06)----------CoalesceBatchesExec: target_batch_size=8192
@@ -100,7 +100,7 @@ physical_plan
 25)--------------------------FilterExec: n_name@1 = GERMANY, projection=[n_nationkey@0]
 26)----------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 27)------------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/nation.tbl]]}, projection=[n_nationkey, n_name], has_header=false
-28)------ProjectionExec: expr=[CAST(CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty)@0 AS Float64) * 0.0001 AS Decimal128(38, 15)) as sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)]
+28)------ProjectionExec: expr=[CAST(sum(partsupp.ps_supplycost * partsupp.ps_availqty)@0 AS Float64) * 0.0001 as sum(partsupp.ps_supplycost * partsupp.ps_availqty) * Float64(0.0001)]
 29)--------AggregateExec: mode=Final, gby=[], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]
 30)----------CoalescePartitionsExec
 31)------------AggregateExec: mode=Partial, gby=[], aggr=[sum(partsupp.ps_supplycost * partsupp.ps_availqty)]

datafusion/sqllogictest/test_files/tpch/plans/q17.slt.part

@@ -39,15 +39,15 @@ logical_plan
 01)Projection: CAST(sum(lineitem.l_extendedprice) AS Float64) / Float64(7) AS avg_yearly
 02)--Aggregate: groupBy=[[]], aggr=[[sum(lineitem.l_extendedprice)]]
 03)----Projection: lineitem.l_extendedprice
-04)------Inner Join: part.p_partkey = __scalar_sq_1.l_partkey Filter: CAST(lineitem.l_quantity AS Decimal128(30, 15)) < __scalar_sq_1.Float64(0.2) * avg(lineitem.l_quantity)
+04)------Inner Join: part.p_partkey = __scalar_sq_1.l_partkey Filter: CAST(lineitem.l_quantity AS Float64) < __scalar_sq_1.Float64(0.2) * avg(lineitem.l_quantity)
 05)--------Projection: lineitem.l_quantity, lineitem.l_extendedprice, part.p_partkey
 06)----------Inner Join: lineitem.l_partkey = part.p_partkey
 07)------------TableScan: lineitem projection=[l_partkey, l_quantity, l_extendedprice]
 08)------------Projection: part.p_partkey
 09)--------------Filter: part.p_brand = Utf8("Brand#23") AND part.p_container = Utf8("MED BOX")
 10)----------------TableScan: part projection=[p_partkey, p_brand, p_container], partial_filters=[part.p_brand = Utf8("Brand#23"), part.p_container = Utf8("MED BOX")]
 11)--------SubqueryAlias: __scalar_sq_1
-12)----------Projection: CAST(Float64(0.2) * CAST(avg(lineitem.l_quantity) AS Float64) AS Decimal128(30, 15)), lineitem.l_partkey
+12)----------Projection: Float64(0.2) * CAST(avg(lineitem.l_quantity) AS Float64), lineitem.l_partkey
 13)------------Aggregate: groupBy=[[lineitem.l_partkey]], aggr=[[avg(lineitem.l_quantity)]]
 14)--------------TableScan: lineitem projection=[l_partkey, l_quantity]
 physical_plan
@@ -56,7 +56,7 @@ physical_plan
 03)----CoalescePartitionsExec
 04)------AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.l_extendedprice)]
 05)--------CoalesceBatchesExec: target_batch_size=8192
-06)----------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(p_partkey@2, l_partkey@1)], filter=CAST(l_quantity@0 AS Decimal128(30, 15)) < Float64(0.2) * avg(lineitem.l_quantity)@1, projection=[l_extendedprice@1]
+06)----------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(p_partkey@2, l_partkey@1)], filter=CAST(l_quantity@0 AS Float64) < Float64(0.2) * avg(lineitem.l_quantity)@1, projection=[l_extendedprice@1]
 07)------------CoalesceBatchesExec: target_batch_size=8192
 08)--------------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(l_partkey@0, p_partkey@0)], projection=[l_quantity@1, l_extendedprice@2, p_partkey@3]
 09)----------------CoalesceBatchesExec: target_batch_size=8192
@@ -68,10 +68,9 @@ physical_plan
 15)----------------------FilterExec: p_brand@1 = Brand#23 AND p_container@2 = MED BOX, projection=[p_partkey@0]
 16)------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 17)--------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/part.tbl]]}, projection=[p_partkey, p_brand, p_container], has_header=false
-18)------------ProjectionExec: expr=[CAST(0.2 * CAST(avg(lineitem.l_quantity)@1 AS Float64) AS Decimal128(30, 15)) as Float64(0.2) * avg(lineitem.l_quantity), l_partkey@0 as l_partkey]
+18)------------ProjectionExec: expr=[0.2 * CAST(avg(lineitem.l_quantity)@1 AS Float64) as Float64(0.2) * avg(lineitem.l_quantity), l_partkey@0 as l_partkey]
 19)--------------AggregateExec: mode=FinalPartitioned, gby=[l_partkey@0 as l_partkey], aggr=[avg(lineitem.l_quantity)]
 20)----------------CoalesceBatchesExec: target_batch_size=8192
 21)------------------RepartitionExec: partitioning=Hash([l_partkey@0], 4), input_partitions=4
 22)--------------------AggregateExec: mode=Partial, gby=[l_partkey@0 as l_partkey], aggr=[avg(lineitem.l_quantity)]
 23)----------------------CsvExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:0..18561749], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:18561749..37123498], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:37123498..55685247], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:55685247..74246996]]}, projection=[l_partkey, l_quantity], has_header=false
-

datafusion/sqllogictest/test_files/tpch/plans/q22.slt.part

@@ -71,8 +71,8 @@ logical_plan
 13)------------SubqueryAlias: __scalar_sq_2
 14)--------------Aggregate: groupBy=[[]], aggr=[[avg(customer.c_acctbal)]]
 15)----------------Projection: customer.c_acctbal
-16)------------------Filter: customer.c_acctbal > Decimal128(Some(0),15,2) AND substr(customer.c_phone, Int64(1), Int64(2)) IN ([Utf8("13"), Utf8("31"), Utf8("23"), Utf8("29"), Utf8("30"), Utf8("18"), Utf8("17")])
-17)--------------------TableScan: customer projection=[c_phone, c_acctbal], partial_filters=[customer.c_acctbal > Decimal128(Some(0),15,2), substr(customer.c_phone, Int64(1), Int64(2)) IN ([Utf8("13"), Utf8("31"), Utf8("23"), Utf8("29"), Utf8("30"), Utf8("18"), Utf8("17")])]
+16)------------------Filter: CAST(customer.c_acctbal AS Float64) > Float64(0) AND substr(customer.c_phone, Int64(1), Int64(2)) IN ([Utf8("13"), Utf8("31"), Utf8("23"), Utf8("29"), Utf8("30"), Utf8("18"), Utf8("17")])
+17)--------------------TableScan: customer projection=[c_phone, c_acctbal], partial_filters=[CAST(customer.c_acctbal AS Float64) > Float64(0), substr(customer.c_phone, Int64(1), Int64(2)) IN ([Utf8("13"), Utf8("31"), Utf8("23"), Utf8("29"), Utf8("30"), Utf8("18"), Utf8("17")])]
 physical_plan
 01)SortPreservingMergeExec: [cntrycode@0 ASC NULLS LAST]
 02)--SortExec: expr=[cntrycode@0 ASC NULLS LAST], preserve_partitioning=[true]
@@ -100,6 +100,6 @@ physical_plan
 24)----------------------CoalescePartitionsExec
 25)------------------------AggregateExec: mode=Partial, gby=[], aggr=[avg(customer.c_acctbal)]
 26)--------------------------CoalesceBatchesExec: target_batch_size=8192
-27)----------------------------FilterExec: c_acctbal@1 > Some(0),15,2 AND Use substr(c_phone@0, 1, 2) IN (SET) ([Literal { value: Utf8("13") }, Literal { value: Utf8("31") }, Literal { value: Utf8("23") }, Literal { value: Utf8("29") }, Literal { value: Utf8("30") }, Literal { value: Utf8("18") }, Literal { value: Utf8("17") }]), projection=[c_acctbal@1]
+27)----------------------------FilterExec: CAST(c_acctbal@1 AS Float64) > 0 AND Use substr(c_phone@0, 1, 2) IN (SET) ([Literal { value: Utf8("13") }, Literal { value: Utf8("31") }, Literal { value: Utf8("23") }, Literal { value: Utf8("29") }, Literal { value: Utf8("30") }, Literal { value: Utf8("18") }, Literal { value: Utf8("17") }]), projection=[c_acctbal@1]
 28)------------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
 29)--------------------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/customer.tbl]]}, projection=[c_phone, c_acctbal], has_header=false

datafusion/sqllogictest/test_files/tpch/plans/q6.slt.part

@@ -31,13 +31,13 @@ logical_plan
 01)Projection: sum(lineitem.l_extendedprice * lineitem.l_discount) AS revenue
 02)--Aggregate: groupBy=[[]], aggr=[[sum(lineitem.l_extendedprice * lineitem.l_discount)]]
 03)----Projection: lineitem.l_extendedprice, lineitem.l_discount
-04)------Filter: lineitem.l_shipdate >= Date32("1994-01-01") AND lineitem.l_shipdate < Date32("1995-01-01") AND lineitem.l_discount >= Decimal128(Some(5),15,2) AND lineitem.l_discount <= Decimal128(Some(7),15,2) AND lineitem.l_quantity < Decimal128(Some(2400),15,2)
-05)--------TableScan: lineitem projection=[l_quantity, l_extendedprice, l_discount, l_shipdate], partial_filters=[lineitem.l_shipdate >= Date32("1994-01-01"), lineitem.l_shipdate < Date32("1995-01-01"), lineitem.l_discount >= Decimal128(Some(5),15,2), lineitem.l_discount <= Decimal128(Some(7),15,2), lineitem.l_quantity < Decimal128(Some(2400),15,2)]
+04)------Filter: lineitem.l_shipdate >= Date32("1994-01-01") AND lineitem.l_shipdate < Date32("1995-01-01") AND CAST(lineitem.l_discount AS Float64) >= Float64(0.049999999999999996) AND CAST(lineitem.l_discount AS Float64) <= Float64(0.06999999999999999) AND lineitem.l_quantity < Decimal128(Some(2400),15,2)
+05)--------TableScan: lineitem projection=[l_quantity, l_extendedprice, l_discount, l_shipdate], partial_filters=[lineitem.l_shipdate >= Date32("1994-01-01"), lineitem.l_shipdate < Date32("1995-01-01"), CAST(lineitem.l_discount AS Float64) >= Float64(0.049999999999999996), CAST(lineitem.l_discount AS Float64) <= Float64(0.06999999999999999), lineitem.l_quantity < Decimal128(Some(2400),15,2)]
 physical_plan
 01)ProjectionExec: expr=[sum(lineitem.l_extendedprice * lineitem.l_discount)@0 as revenue]
 02)--AggregateExec: mode=Final, gby=[], aggr=[sum(lineitem.l_extendedprice * lineitem.l_discount)]
 03)----CoalescePartitionsExec
 04)------AggregateExec: mode=Partial, gby=[], aggr=[sum(lineitem.l_extendedprice * lineitem.l_discount)]
 05)--------CoalesceBatchesExec: target_batch_size=8192
-06)----------FilterExec: l_shipdate@3 >= 1994-01-01 AND l_shipdate@3 < 1995-01-01 AND l_discount@2 >= Some(5),15,2 AND l_discount@2 <= Some(7),15,2 AND l_quantity@0 < Some(2400),15,2, projection=[l_extendedprice@1, l_discount@2]
+06)----------FilterExec: l_shipdate@3 >= 1994-01-01 AND l_shipdate@3 < 1995-01-01 AND CAST(l_discount@2 AS Float64) >= 0.049999999999999996 AND CAST(l_discount@2 AS Float64) <= 0.06999999999999999 AND l_quantity@0 < Some(2400),15,2, projection=[l_extendedprice@1, l_discount@2]
 07)------------CsvExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:0..18561749], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:18561749..37123498], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:37123498..55685247], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/tpch/data/lineitem.tbl:55685247..74246996]]}, projection=[l_quantity, l_extendedprice, l_discount, l_shipdate], has_header=false

datafusion/sqllogictest/test_files/union.slt

@@ -836,3 +836,18 @@ physical_plan
 # Clean up after the test
 statement ok
 drop table aggregate_test_100;
+
+query T
+SELECT DISTINCT arrow_typeof(a) FROM (SELECT '1'::float UNION ALL SELECT '1'::decimal(10)) t(a)
+----
+Float32
+
+query T
+SELECT DISTINCT arrow_typeof(a) FROM (SELECT '1'::decimal(10) UNION ALL SELECT '1'::float ) t(a)
+----
+Float32
+
+query T
+SELECT DISTINCT arrow_typeof(a) FROM (SELECT '1'::decimal(10) UNION ALL SELECT '1'::double) t(a)
+----
+Float64