[SPARK-39552][SQL] Unify v1 and v2 DESCRIBE TABLE

### What changes were proposed in this pull request?
In the PR, I propose to change output of v2 `DESCRIBE TABLE`, and make it the same as v1. In particular:
1. Return NULL instead of empty strings when any comment doesn't exist in the schema info.
2. When a v2 table has identity transformations (partition by) only, output the partitioning info in v1 style. For instance:
```sql
> CREATE TABLE tbl (id bigint, data string) USING _;
> DESCRIBE TABLE tbl;
+-----------------------+---------+-------+
|col_name               |data_type|comment|
+-----------------------+---------+-------+
|id                     |bigint   |null   |
|data                   |string   |null   |
|# Partition Information|         |       |
|# col_name             |data_type|comment|
|id                     |bigint   |null   |
+-----------------------+---------+-------+
```

Also the PR moves/adds some tests to the base traits:
- "DESCRIBE TABLE of a non-partitioned table"
- "DESCRIBE TABLE of a partitioned table"

and addresses review comments in apache/spark#35265.

### Why are the changes needed?
The changes unify outputs of v1 and v2 implementations, and make the migration process from the version v1 to v2 easier for Spark SQL users.

### Does this PR introduce _any_ user-facing change?
Yes, it changes outputs of v2 `DESCRIBE TABLE`.

### How was this patch tested?
By running the `DESCRIBE TABLE` test suites:
```
$ build/sbt -Phive-2.3 -Phive-thriftserver "test:testOnly *DescribeTableSuite"
```
and related test suites:
```
$ build/sbt "test:testOnly *DataSourceV2SQLSuite"
```

Closes #36946 from MaxGekk/unify-v1-v2-describe-table.

Authored-by: Max Gekk <max.gekk@gmail.com>
Signed-off-by: Max Gekk <max.gekk@gmail.com>

Author: a0x8o

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -407,6 +407,7 @@ class SparkContext(config: SparkConf) extends Logging {
     SparkContext.fillMissingMagicCommitterConfsIfNeeded(_conf)
 
     SparkContext.supplementJavaModuleOptions(_conf)
+    SparkContext.supplementJavaIPv6Options(_conf)
 
     _driverLogger = DriverLogger(_conf)
 
@@ -3049,6 +3050,18 @@ object SparkContext extends Logging {
     supplement(DRIVER_JAVA_OPTIONS)
     supplement(EXECUTOR_JAVA_OPTIONS)
   }
+
+  private def supplementJavaIPv6Options(conf: SparkConf): Unit = {
+    def supplement(key: OptionalConfigEntry[String]): Unit = {
+      val v = conf.get(key) match {
+        case Some(opts) => s"-Djava.net.preferIPv6Addresses=${Utils.preferIPv6} $opts"
+        case None => s"-Djava.net.preferIPv6Addresses=${Utils.preferIPv6}"
+      }
+      conf.set(key.key, v)
+    }
+    supplement(DRIVER_JAVA_OPTIONS)
+    supplement(EXECUTOR_JAVA_OPTIONS)
+  }
 }
 
 /**

core/src/main/scala/org/apache/spark/ui/JettyUtils.scala

@@ -246,6 +246,7 @@ private[spark] object JettyUtils extends Logging {
       serverName: String = "",
       poolSize: Int = 200): ServerInfo = {
 
+    logInfo(s"Start Jetty $hostName:$port for $serverName")
     // Start the server first, with no connectors.
     val pool = new QueuedThreadPool(poolSize)
     if (serverName.nonEmpty) {

core/src/main/scala/org/apache/spark/ui/WebUI.scala

@@ -140,20 +140,19 @@ private[spark] abstract class WebUI(
   def initialize(): Unit
 
   def initServer(): ServerInfo = {
-    val host = Option(conf.getenv("SPARK_LOCAL_IP")).getOrElse("0.0.0.0")
-    val server = startJettyServer(host, port, sslOptions, conf, name, poolSize)
+    val hostName = Utils.localHostNameForURI()
+    val server = startJettyServer(hostName, port, sslOptions, conf, name, poolSize)
     server
   }
 
   /** Binds to the HTTP server behind this web interface. */
   def bind(): Unit = {
     assert(serverInfo.isEmpty, s"Attempted to bind $className more than once!")
     try {
-      val host = Option(conf.getenv("SPARK_LOCAL_IP")).getOrElse("0.0.0.0")
       val server = initServer()
       handlers.foreach(server.addHandler(_, securityManager))
       serverInfo = Some(server)
-      logInfo(s"Bound $className to $host, and started at $webUrl")
+      logInfo(s"Bound $className to ${Utils.localHostNameForURI()}, and started at $webUrl")
     } catch {
       case e: Exception =>
         logError(s"Failed to bind $className", e)

core/src/test/scala/org/apache/spark/ui/UISuite.scala

@@ -38,6 +38,8 @@ import org.apache.spark.util.Utils
 
 class UISuite extends SparkFunSuite {
 
+  val localhost = Utils.localHostNameForURI()
+
   /**
    * Create a test SparkContext with the SparkUI enabled.
    * It is safe to `get` the SparkUI directly from the SparkContext returned here.
@@ -91,7 +93,7 @@ class UISuite extends SparkFunSuite {
     withSpark(newSparkContext()) { sc =>
       // test if visible from http://localhost:4040
       eventually(timeout(10.seconds), interval(50.milliseconds)) {
-        val html = Utils.tryWithResource(Source.fromURL("http://localhost:4040"))(_.mkString)
+        val html = Utils.tryWithResource(Source.fromURL(s"http://$localhost:4040"))(_.mkString)
         assert(html.toLowerCase(Locale.ROOT).contains("stages"))
       }
     }
@@ -201,41 +203,41 @@ class UISuite extends SparkFunSuite {
 
   test("verify proxy rewrittenURI") {
     val prefix = "/worker-id"
-    val target = "http://localhost:8081"
+    val target = s"http://$localhost:8081"
     val path = "/worker-id/json"
     var rewrittenURI = JettyUtils.createProxyURI(prefix, target, path, null)
-    assert(rewrittenURI.toString() === "http://localhost:8081/json")
+    assert(rewrittenURI.toString() === s"http://$localhost:8081/json")
     rewrittenURI = JettyUtils.createProxyURI(prefix, target, path, "test=done")
-    assert(rewrittenURI.toString() === "http://localhost:8081/json?test=done")
+    assert(rewrittenURI.toString() === s"http://$localhost:8081/json?test=done")
     rewrittenURI = JettyUtils.createProxyURI(prefix, target, "/worker-id", null)
-    assert(rewrittenURI.toString() === "http://localhost:8081")
+    assert(rewrittenURI.toString() === s"http://$localhost:8081")
     rewrittenURI = JettyUtils.createProxyURI(prefix, target, "/worker-id/test%2F", null)
-    assert(rewrittenURI.toString() === "http://localhost:8081/test%2F")
+    assert(rewrittenURI.toString() === s"http://$localhost:8081/test%2F")
     rewrittenURI = JettyUtils.createProxyURI(prefix, target, "/worker-id/%F0%9F%98%84", null)
-    assert(rewrittenURI.toString() === "http://localhost:8081/%F0%9F%98%84")
+    assert(rewrittenURI.toString() === s"http://$localhost:8081/%F0%9F%98%84")
     rewrittenURI = JettyUtils.createProxyURI(prefix, target, "/worker-noid/json", null)
     assert(rewrittenURI === null)
   }
 
   test("SPARK-33611: Avoid encoding twice on the query parameter of proxy rewrittenURI") {
     val prefix = "/worker-id"
-    val target = "http://localhost:8081"
+    val target = s"http://$localhost:8081"
     val path = "/worker-id/json"
     val rewrittenURI =
       JettyUtils.createProxyURI(prefix, target, path, "order%5B0%5D%5Bcolumn%5D=0")
-    assert(rewrittenURI.toString === "http://localhost:8081/json?order%5B0%5D%5Bcolumn%5D=0")
+    assert(rewrittenURI.toString === s"http://$localhost:8081/json?order%5B0%5D%5Bcolumn%5D=0")
   }
 
   test("verify rewriting location header for reverse proxy") {
     val clientRequest = mock(classOf[HttpServletRequest])
-    var headerValue = "http://localhost:4040/jobs"
-    val targetUri = URI.create("http://localhost:4040")
+    var headerValue = s"http://$localhost:4040/jobs"
+    val targetUri = URI.create(s"http://$localhost:4040")
     when(clientRequest.getScheme()).thenReturn("http")
-    when(clientRequest.getHeader("host")).thenReturn("localhost:8080")
+    when(clientRequest.getHeader("host")).thenReturn(s"$localhost:8080")
     when(clientRequest.getPathInfo()).thenReturn("/proxy/worker-id/jobs")
     var newHeader = JettyUtils.createProxyLocationHeader(headerValue, clientRequest, targetUri)
-    assert(newHeader.toString() === "http://localhost:8080/proxy/worker-id/jobs")
-    headerValue = "http://localhost:4041/jobs"
+    assert(newHeader.toString() === s"http://$localhost:8080/proxy/worker-id/jobs")
+    headerValue = s"http://$localhost:4041/jobs"
     newHeader = JettyUtils.createProxyLocationHeader(headerValue, clientRequest, targetUri)
     assert(newHeader === null)
   }
@@ -249,7 +251,7 @@ class UISuite extends SparkFunSuite {
     val serverInfo = JettyUtils.startJettyServer("0.0.0.0", 0, sslOptions, conf)
     try {
       val path = "/test"
-      val url = new URL(s"http://localhost:${serverInfo.boundPort}$path/root")
+      val url = new URL(s"http://$localhost:${serverInfo.boundPort}$path/root")
 
       assert(TestUtils.httpResponseCode(url) === HttpServletResponse.SC_NOT_FOUND)
 
@@ -260,7 +262,7 @@ class UISuite extends SparkFunSuite {
       // Try a request with bad content in a parameter to make sure the security filter
       // is being added to new handlers.
       val badRequest = new URL(
-        s"http://localhost:${serverInfo.boundPort}$path/root?bypass&invalid<=foo")
+        s"http://$localhost:${serverInfo.boundPort}$path/root?bypass&invalid<=foo")
       assert(TestUtils.httpResponseCode(badRequest) === HttpServletResponse.SC_OK)
       assert(servlet.lastRequest.getParameter("invalid<") === null)
       assert(servlet.lastRequest.getParameter("invalid&lt;") !== null)
@@ -276,7 +278,7 @@ class UISuite extends SparkFunSuite {
     val (conf, securityMgr, sslOptions) = sslEnabledConf()
     val serverInfo = JettyUtils.startJettyServer("0.0.0.0", 0, sslOptions, conf)
     try {
-      val serverAddr = s"http://localhost:${serverInfo.boundPort}"
+      val serverAddr = s"http://$localhost:${serverInfo.boundPort}"
 
       val (_, ctx) = newContext("/ctx1")
       serverInfo.addHandler(ctx, securityMgr)
@@ -285,7 +287,7 @@ class UISuite extends SparkFunSuite {
         assert(conn.getResponseCode() === HttpServletResponse.SC_FOUND)
         val location = Option(conn.getHeaderFields().get("Location"))
           .map(_.get(0)).orNull
-        val expectedLocation = s"https://localhost:${serverInfo.securePort.get}/ctx(1)?a[0]=b"
+        val expectedLocation = s"https://$localhost:${serverInfo.securePort.get}/ctx(1)?a[0]=b"
         assert(location == expectedLocation)
       }
     } finally {
@@ -313,9 +315,9 @@ class UISuite extends SparkFunSuite {
 
       tests.foreach { case (scheme, port, expected) =>
         val urls = Seq(
-          s"$scheme://localhost:$port/root",
-          s"$scheme://localhost:$port/test1/root",
-          s"$scheme://localhost:$port/test2/root")
+          s"$scheme://$localhost:$port/root",
+          s"$scheme://$localhost:$port/test1/root",
+          s"$scheme://$localhost:$port/test2/root")
         urls.foreach { url =>
           val rc = TestUtils.httpResponseCode(new URL(url))
           assert(rc === expected, s"Unexpected status $rc for $url")
@@ -355,7 +357,7 @@ class UISuite extends SparkFunSuite {
 
     val serverInfo = JettyUtils.startJettyServer("0.0.0.0", 0, sslOptions, conf)
     try {
-      val serverAddr = s"http://localhost:${serverInfo.boundPort}"
+      val serverAddr = s"http://$localhost:${serverInfo.boundPort}"
 
       val redirect = JettyUtils.createRedirectHandler("/src", "/dst")
       serverInfo.addHandler(redirect, securityMgr)
@@ -395,7 +397,7 @@ class UISuite extends SparkFunSuite {
     try {
       val (_, ctx) = newContext("/ctx")
       serverInfo.addHandler(ctx, securityMgr)
-      val urlStr = s"http://localhost:${serverInfo.boundPort}/ctx"
+      val urlStr = s"http://$localhost:${serverInfo.boundPort}/ctx"
 
       assert(TestUtils.httpResponseCode(new URL(urlStr + "/")) === HttpServletResponse.SC_OK)
 

sql/catalyst/src/main/java/org/apache/spark/sql/util/CaseInsensitiveStringMap.java

@@ -41,7 +41,7 @@
  */
 @Experimental
 public class CaseInsensitiveStringMap implements Map<String, String> {
-  private final Logger logger = LoggerFactory.getLogger(CaseInsensitiveStringMap.class);
+  private static final Logger logger = LoggerFactory.getLogger(CaseInsensitiveStringMap.class);
 
   private String unsupportedOperationMsg = "CaseInsensitiveStringMap is read-only.";
 

sql/catalyst/src/main/scala-2.13/org/apache/spark/sql/catalyst/expressions/ExpressionSet.scala

@@ -132,6 +132,12 @@ class ExpressionSet protected(
     newSet
   }
 
+  override def concat(that: IterableOnce[Expression]): ExpressionSet = {
+    val newSet = clone()
+    that.iterator.foreach(newSet.add)
+    newSet
+  }
+
   override def --(that: IterableOnce[Expression]): ExpressionSet = {
     val newSet = clone()
     that.iterator.foreach(newSet.remove)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -451,7 +451,7 @@ class Analyzer(override val catalogManager: CatalogManager)
    * Substitute child plan with WindowSpecDefinitions.
    */
   object WindowsSubstitution extends Rule[LogicalPlan] {
-    def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsDownWithPruning(
+    def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
       _.containsAnyPattern(WITH_WINDOW_DEFINITION, UNRESOLVED_WINDOW_EXPRESSION), ruleId) {
       // Lookup WindowSpecDefinitions. This rule works with unresolved children.
       case WithWindowDefinition(windowDefinitions, child) => child.resolveExpressions {
@@ -460,14 +460,6 @@ class Analyzer(override val catalogManager: CatalogManager)
             throw QueryCompilationErrors.windowSpecificationNotDefinedError(windowName))
           WindowExpression(c, windowSpecDefinition)
       }
-
-      case p @ Project(projectList, _) =>
-        projectList.foreach(_.transformDownWithPruning(
-          _.containsPattern(UNRESOLVED_WINDOW_EXPRESSION), ruleId) {
-          case UnresolvedWindowExpression(_, windowSpec) =>
-            throw QueryCompilationErrors.windowSpecificationNotDefinedError(windowSpec.name)
-        })
-        p
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.optimizer.{BooleanSimplification, Decorrela
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_WINDOW_EXPRESSION
 import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, StringUtils, TypeUtils}
 import org.apache.spark.sql.connector.catalog.{LookupCatalog, SupportsPartitionManagement}
 import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -231,7 +232,7 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
             failAnalysis("grouping_id() can only be used with GroupingSets/Cube/Rollup")
 
           case e: Expression if e.children.exists(_.isInstanceOf[WindowFunction]) &&
-              !e.isInstanceOf[WindowExpression] =>
+              !e.isInstanceOf[WindowExpression] && e.resolved =>
             val w = e.children.find(_.isInstanceOf[WindowFunction]).get
             failAnalysis(s"Window function $w requires an OVER clause.")
 
@@ -542,6 +543,13 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog {
               s"""Only a single table generating function is allowed in a SELECT clause, found:
                  | ${exprs.map(_.sql).mkString(",")}""".stripMargin)
 
+          case p @ Project(projectList, _) =>
+            projectList.foreach(_.transformDownWithPruning(
+              _.containsPattern(UNRESOLVED_WINDOW_EXPRESSION)) {
+              case UnresolvedWindowExpression(_, windowSpec) =>
+                throw QueryCompilationErrors.windowSpecificationNotDefinedError(windowSpec.name)
+            })
+
           case j: Join if !j.duplicateResolved =>
             val conflictingAttributes = j.left.outputSet.intersect(j.right.outputSet)
             failAnalysis(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushDownThroughWindow.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.optimizer
 
-import org.apache.spark.sql.catalyst.expressions.{Alias, CurrentRow, IntegerLiteral, NamedExpression, RankLike, RowFrame, RowNumberLike, SpecifiedWindowFrame, UnboundedPreceding, WindowExpression, WindowSpecDefinition}
+import org.apache.spark.sql.catalyst.expressions.{Alias, CurrentRow, DenseRank, IntegerLiteral, NamedExpression, NTile, Rank, RowFrame, RowNumber, SpecifiedWindowFrame, UnboundedPreceding, WindowExpression, WindowSpecDefinition}
 import org.apache.spark.sql.catalyst.plans.logical.{Limit, LocalLimit, LogicalPlan, Project, Sort, Window}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.{LIMIT, WINDOW}
@@ -33,7 +33,8 @@ object LimitPushDownThroughWindow extends Rule[LogicalPlan] {
   // The window frame of RankLike and RowNumberLike can only be UNBOUNDED PRECEDING to CURRENT ROW.
   private def supportsPushdownThroughWindow(
       windowExpressions: Seq[NamedExpression]): Boolean = windowExpressions.forall {
-    case Alias(WindowExpression(_: RankLike | _: RowNumberLike, WindowSpecDefinition(Nil, _,
+    case Alias(WindowExpression(_: Rank | _: DenseRank | _: NTile | _: RowNumber,
+        WindowSpecDefinition(Nil, _,
         SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow))), _) => true
     case _ => false
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushdownThroughWindowSuite.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.optimizer
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{CurrentRow, Rank, RowFrame, RowNumber, SpecifiedWindowFrame, UnboundedPreceding}
+import org.apache.spark.sql.catalyst.expressions.{CurrentRow, PercentRank, Rank, RowFrame, RowNumber, SpecifiedWindowFrame, UnboundedPreceding}
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules._
@@ -187,4 +187,15 @@ class LimitPushdownThroughWindowSuite extends PlanTest {
       Optimize.execute(originalQuery.analyze),
       WithoutOptimize.execute(originalQuery.analyze))
   }
+
+  test("SPARK-38614: Should not push through percent_rank window function") {
+    val originalQuery = testRelation
+      .select(a, b, c,
+        windowExpr(new PercentRank(), windowSpec(Nil, c.desc :: Nil, windowFrame)).as("rn"))
+      .limit(2)
+
+    comparePlans(
+      Optimize.execute(originalQuery.analyze),
+      WithoutOptimize.execute(originalQuery.analyze))
+  }
 }