[SPARK-39479][SQL] DS V2 supports push down math functions(non ANSI)

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralScalarExpression.java

@@ -124,6 +124,24 @@
  *    <li>Since version: 3.4.0</li>
  *   </ul>
  *  </li>
+ *  <li>Name: <code>LOG</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>LOG(base, expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>LOG10</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>LOG10(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>LOG2</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>LOG2(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
  *  <li>Name: <code>LN</code>
  *   <ul>
  *    <li>SQL semantic: <code>LN(expr)</code></li>
@@ -160,6 +178,120 @@
  *    <li>Since version: 3.3.0</li>
  *   </ul>
  *  </li>
+ *  <li>Name: <code>ROUND</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ROUND(expr, [scale])</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>SIN</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>SIN(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>SINH</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>SINH(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>COS</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>COS(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>COSH</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>COSH(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>TAN</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>TAN(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>TANH</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>TANH(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>COT</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>COT(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>ASIN</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ASIN(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>ASINH</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ASINH(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>ACOS</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ACOS(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>ACOSH</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ACOSH(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>ATAN</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ATAN(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>ATANH</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ATANH(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>ATAN2</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>ATAN2(exprY, exprX)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>CBRT</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>CBRT(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>DEGREES</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>DEGREES(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>RADIANS</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>RADIANS(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
+ *  <li>Name: <code>SIGN</code>
+ *   <ul>
+ *    <li>SQL semantic: <code>SIGN(expr)</code></li>
+ *    <li>Since version: 3.4.0</li>
+ *   </ul>
+ *  </li>
  *  <li>Name: <code>WIDTH_BUCKET</code>
  *   <ul>
  *    <li>SQL semantic: <code>WIDTH_BUCKET(expr)</code></li>

sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java

@@ -100,12 +100,34 @@ public String build(Expression expr) {
         case "GREATEST":
         case "LEAST":
         case "RAND":
+        case "LOG":
+        case "LOG10":
+        case "LOG2":
         case "LN":
         case "EXP":
         case "POWER":
         case "SQRT":
         case "FLOOR":
         case "CEIL":
+        case "ROUND":
+        case "SIN":
+        case "SINH":
+        case "COS":
+        case "COSH":
+        case "TAN":
+        case "TANH":
+        case "COT":
+        case "ASIN":
+        case "ASINH":
+        case "ACOS":
+        case "ACOSH":
+        case "ATAN":
+        case "ATANH":
+        case "ATAN2":
+        case "CBRT":
+        case "DEGREES":
+        case "RADIANS":
+        case "SIGN":
         case "WIDTH_BUCKET":
         case "SUBSTRING":
         case "UPPER":

sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala

@@ -120,6 +120,18 @@ class V2ExpressionBuilder(e: Expression, isPredicate: Boolean = false) {
         generateExpression(child)
           .map(v => new GeneralScalarExpression("RAND", Array[V2Expression](v)))
       }
+    case log: Logarithm =>
+      val l = generateExpression(log.left)
+      val r = generateExpression(log.right)
+      if (l.isDefined && r.isDefined) {
+        Some(new GeneralScalarExpression("LOG", Array[V2Expression](l.get, r.get)))
+      } else {
+        None
+      }
+    case Log10(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("LOG10", Array[V2Expression](v)))
+    case Log2(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("LOG2", Array[V2Expression](v)))
     case Log(child) => generateExpression(child)
       .map(v => new GeneralScalarExpression("LN", Array[V2Expression](v)))
     case Exp(child) => generateExpression(child)
@@ -138,6 +150,56 @@ class V2ExpressionBuilder(e: Expression, isPredicate: Boolean = false) {
       .map(v => new GeneralScalarExpression("FLOOR", Array[V2Expression](v)))
     case Ceil(child) => generateExpression(child)
       .map(v => new GeneralScalarExpression("CEIL", Array[V2Expression](v)))
+    case round: Round =>
+      val l = generateExpression(round.left)
+      val r = generateExpression(round.right)
+      if (l.isDefined && r.isDefined) {
+        Some(new GeneralScalarExpression("ROUND", Array[V2Expression](l.get, r.get)))
+      } else {
+        None
+      }
+    case Sin(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("SIN", Array[V2Expression](v)))
+    case Sinh(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("SINH", Array[V2Expression](v)))
+    case Cos(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("COS", Array[V2Expression](v)))
+    case Cosh(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("COSH", Array[V2Expression](v)))
+    case Tan(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("TAN", Array[V2Expression](v)))
+    case Tanh(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("TANH", Array[V2Expression](v)))
+    case Cot(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("COT", Array[V2Expression](v)))
+    case Asin(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("ASIN", Array[V2Expression](v)))
+    case Asinh(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("ASINH", Array[V2Expression](v)))
+    case Acos(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("ACOS", Array[V2Expression](v)))
+    case Acosh(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("ACOSH", Array[V2Expression](v)))
+    case Atan(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("ATAN", Array[V2Expression](v)))
+    case Atanh(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("ATANH", Array[V2Expression](v)))
+    case atan2: Atan2 =>
+      val l = generateExpression(atan2.left)
+      val r = generateExpression(atan2.right)
+      if (l.isDefined && r.isDefined) {
+        Some(new GeneralScalarExpression("ATAN2", Array[V2Expression](l.get, r.get)))
+      } else {
+        None
+      }
+    case Cbrt(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("CBRT", Array[V2Expression](v)))
+    case ToDegrees(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("DEGREES", Array[V2Expression](v)))
+    case ToRadians(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("RADIANS", Array[V2Expression](v)))
+    case Signum(child) => generateExpression(child)
+      .map(v => new GeneralScalarExpression("SIGN", Array[V2Expression](v)))
     case wb: WidthBucket =>
       val childrenExpressions = wb.children.flatMap(generateExpression(_))
       if (childrenExpressions.length == wb.children.length) {

sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala

@@ -35,8 +35,10 @@ private[sql] object H2Dialect extends JdbcDialect {
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:h2")
 
   private val supportedFunctions =
-    Set("ABS", "COALESCE", "GREATEST", "LEAST", "RAND", "LN", "EXP", "POWER", "SQRT", "FLOOR",
-      "CEIL", "SUBSTRING", "UPPER", "LOWER", "TRANSLATE", "TRIM")
+    Set("ABS", "COALESCE", "GREATEST", "LEAST", "RAND", "LOG", "LOG10", "LN", "EXP",
+      "POWER", "SQRT", "FLOOR", "CEIL", "ROUND", "SIN", "SINH", "COS", "COSH", "TAN",
+      "TANH", "COT", "ASIN", "ACOS", "ATAN", "ATAN2", "DEGREES", "RADIANS", "SIGN",
+      "PI", "SUBSTRING", "UPPER", "LOWER", "TRANSLATE", "TRIM")
 
   override def isSupportedFunction(funcName: String): Boolean =
     supportedFunctions.contains(funcName)