[WIP check create table diff] Backport "support decimal data type in …

…create table (apache#1431)"

* support decimal data type in create table

Can drop this after rebase on commit dc80c11, first released in 7.0.0

datafusion/src/sql/planner.rs

@@ -294,7 +294,27 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             SQLDataType::Char(_) | SQLDataType::Varchar(_) | SQLDataType::Text => {
                 Ok(DataType::Utf8)
             }
-            SQLDataType::Decimal(_, _) => Ok(DataType::Float64),
+            SQLDataType::Decimal(precision, scale) => {
+                match (precision, scale) {
+                    (None, _) | (_, None) => {
+                        return Err(DataFusionError::Internal(format!(
+                            "Invalid Decimal type ({:?}), precision or scale can't be empty.",
+                            sql_type
+                        )));
+                    }
+                    (Some(p), Some(s)) => {
+                        // TODO add bound checker in some utils file or function
+                        if *p > 38 || *s > *p {
+                            return Err(DataFusionError::Internal(format!(
+                                "Error Decimal Type ({:?}), precision must be less than or equal to 38 and scale can't be greater than precision",
+                                sql_type
+                            )));
+                        } else {
+                            Ok(DataType::Decimal(*p as usize, *s as usize))
+                        }
+                    }
+                }
+            }
             SQLDataType::Float(_) => Ok(DataType::Float32),
             SQLDataType::Real | SQLDataType::Double => Ok(DataType::Float64),
             SQLDataType::Boolean => Ok(DataType::Boolean),
@@ -1926,8 +1946,8 @@ fn extract_possible_join_keys(
 }
 
 /// Convert SQL data type to relational representation of data type
-pub fn convert_data_type(sql: &SQLDataType) -> Result<DataType> {
-    match sql {
+pub fn convert_data_type(sql_type: &SQLDataType) -> Result<DataType> {
+    match sql_type {
         SQLDataType::Boolean => Ok(DataType::Boolean),
         SQLDataType::SmallInt => Ok(DataType::Int16),
         SQLDataType::Int => Ok(DataType::Int32),
@@ -1937,6 +1957,27 @@ pub fn convert_data_type(sql: &SQLDataType) -> Result<DataType> {
         SQLDataType::Char(_) | SQLDataType::Varchar(_) => Ok(DataType::Utf8),
         SQLDataType::Timestamp => Ok(DataType::Timestamp(TimeUnit::Nanosecond, None)),
         SQLDataType::Date => Ok(DataType::Date32),
+        SQLDataType::Decimal(precision, scale) => {
+            match (precision, scale) {
+                (None, _) | (_, None) => {
+                    return Err(DataFusionError::Internal(format!(
+                        "Invalid Decimal type ({:?}), precision or scale can't be empty.",
+                        sql_type
+                    )));
+                }
+                (Some(p), Some(s)) => {
+                    // TODO add bound checker in some utils file or function
+                    if *p > 38 || *s > *p {
+                        return Err(DataFusionError::Internal(format!(
+                            "Error Decimal Type ({:?})",
+                            sql_type
+                        )));
+                    } else {
+                        Ok(DataType::Decimal(*p as usize, *s as usize))
+                    }
+                }
+            }
+        }
         other => Err(DataFusionError::NotImplemented(format!(
             "Unsupported SQL type {:?}",
             other