Use parse_url kernel for HOST parsing

docs/supported_ops.md

@@ -10770,7 +10770,7 @@ are limited.
 <td> </td>
 <td> </td>
 <td> </td>
-<td><em>PS<br/>only support partToExtract=PROTOCOL;<br/>Literal value only</em></td>
+<td><em>PS<br/>only support partToExtract = PROTOCOL | HOST;<br/>Literal value only</em></td>
 <td> </td>
 <td> </td>
 <td> </td>

integration_tests/src/main/python/url_test.py

@@ -145,23 +145,19 @@
 edge_cases_gen = SetValuesGen(StringType(), edge_cases)
 
 url_gen = StringGen(url_pattern)
+
+supported_parts = ['PROTOCOL', 'HOST']
+unsupported_parts = ['PATH', 'QUERY', 'REF', 'FILE', 'AUTHORITY', 'USERINFO']
 
 @pytest.mark.parametrize('data_gen', [url_gen, edge_cases_gen], ids=idfn)
-def test_parse_url_protocol(data_gen):
+@pytest.mark.parametrize('part', supported_parts, ids=idfn)
+def test_parse_url_supported(data_gen, part):
     assert_gpu_and_cpu_are_equal_collect(
-            lambda spark : unary_op_df(spark, data_gen).selectExpr(
-                "a",
-                "parse_url(a, 'PROTOCOL')"
-                ))
-
-unsupported_parts = ['HOST', 'PATH', 'QUERY', 'REF', 'FILE', 'AUTHORITY', 'USERINFO']
+        lambda spark: unary_op_df(spark, data_gen).selectExpr("a", "parse_url(a, '" + part + "')"))
 
 @allow_non_gpu('ProjectExec', 'ParseUrl')
 @pytest.mark.parametrize('part', unsupported_parts, ids=idfn)
-def test_parse_url_host_fallback(part):
+def test_parse_url_unsupported_fallback(part):
     assert_gpu_fallback_collect(
-            lambda spark : unary_op_df(spark, url_gen).selectExpr(
-                "a",
-                "parse_url(a, '" + part + "')"
-                ),
-            'ParseUrl')
+        lambda spark: unary_op_df(spark, url_gen).selectExpr("a", "parse_url(a, '" + part + "')"),
+        'ParseUrl')

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOverrides.scala

@@ -3231,7 +3231,7 @@ object GpuOverrides extends Logging {
       ExprChecks.projectOnly(TypeSig.STRING, TypeSig.STRING,
         Seq(ParamCheck("url", TypeSig.STRING, TypeSig.STRING),
           ParamCheck("partToExtract", TypeSig.lit(TypeEnum.STRING).withPsNote(
-            TypeEnum.STRING, "only support partToExtract=PROTOCOL"), TypeSig.STRING)),
+            TypeEnum.STRING, "only support partToExtract = PROTOCOL | HOST"), TypeSig.STRING)),
           // Should really be an OptionalParam
           Some(RepeatingParamCheck("key", TypeSig.lit(TypeEnum.STRING), TypeSig.STRING))),
       (a, conf, p, r) => new ExprMeta[ParseUrl](a, conf, p, r) {
@@ -3241,7 +3241,7 @@ object GpuOverrides extends Logging {
           }
 
           extractStringLit(a.children(1)).map(_.toUpperCase) match {
-            case Some(GpuParseUrl.PROTOCOL) =>
+            case Some(part) if GpuParseUrl.isSupportedPart(part) =>
             case Some(other) =>
               willNotWorkOnGpu(s"Part to extract $other is not supported on GPU")
             case None =>

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuParseUrl.scala

@@ -37,6 +37,15 @@ object GpuParseUrl {
   val FILE = "FILE"
   val AUTHORITY = "AUTHORITY"
   val USERINFO = "USERINFO"
+
+  def isSupportedPart(part: String): Boolean = {
+    part match {
+      case PROTOCOL | HOST =>
+        true
+      case _ =>
+        false
+    }
+  }
 }
 
 case class GpuParseUrl(children: Seq[Expression]) 
@@ -54,8 +63,11 @@ case class GpuParseUrl(children: Seq[Expression])
     part match {
       case PROTOCOL =>
         ParseURI.parseURIProtocol(url.getBase)
-      case HOST | PATH | QUERY | REF | FILE | AUTHORITY | USERINFO =>
-        throw new UnsupportedOperationException(s"$this is not supported partToExtract=$part")
+      case HOST =>
+        ParseURI.parseURIHost(url.getBase)
+      case PATH | QUERY | REF | FILE | AUTHORITY | USERINFO =>
+        throw new UnsupportedOperationException(s"$this is not supported partToExtract=$part. " +
+            s"Only PROTOCOL and HOST are supported")
       case _ =>
         throw new IllegalArgumentException(s"Invalid partToExtract: $partToExtract")
     }
@@ -67,7 +79,8 @@ case class GpuParseUrl(children: Seq[Expression])
       // return a null columnvector
       return ColumnVector.fromStrings(null, null)
     }
-    throw new UnsupportedOperationException(s"$this only supports partToExtract = PROTOCOL")
+    throw new UnsupportedOperationException(s"$this is not supported partToExtract=$part. " +
+        s"Only PROTOCOL and HOST are supported")
   }
 
   override def columnarEval(batch: ColumnarBatch): GpuColumnVector = {
@@ -79,8 +92,8 @@ case class GpuParseUrl(children: Seq[Expression])
             case partScalar: GpuScalar =>
               GpuColumnVector.from(doColumnar(urls, partScalar), dataType)
             case _ =>
-              throw new 
-                  UnsupportedOperationException(s"Cannot columnar evaluate expression: $this")
+              throw new UnsupportedOperationException(
+                s"Cannot columnar evaluate expression: $this")
           }
         }
       }