[CARMEL-7220] Backport Integration with admin and metadata cache notification (apache#48)

Co-authored-by: Luan, Xuedong <xuluan@ebay.com>

Author: 2 people

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

@@ -80,6 +80,11 @@ private[spark] class LiveListenerBus(conf: SparkConf) {
     addToQueue(listener, EVENT_LOG_QUEUE)
   }
 
+  /** Add a listener to the external catalog queue. */
+  def addToExternalCatalogQueue(listener: SparkListenerInterface): Unit = {
+    addToQueue(listener, EXTERNAL_CATALOG_QUEUE)
+  }
+
   /**
    * Add a listener to a specific queue, creating a new queue if needed. Queues are independent
    * of each other (each one uses a separate thread for delivering events), allowing slower
@@ -158,6 +163,17 @@ private[spark] class LiveListenerBus(conf: SparkConf) {
     }
   }
 
+  def postToQueue(event: SparkListenerEvent, queue: String): Unit = {
+    if (stopped.get()) {
+      return
+    }
+    queues.asScala.find(_.name == queue) match {
+      case Some(queue) =>
+        queue.post(event)
+      case None =>
+    }
+  }
+
   /**
    * Start sending events to attached listeners.
    *
@@ -257,6 +273,8 @@ private[spark] object LiveListenerBus {
   private[scheduler] val EXECUTOR_MANAGEMENT_QUEUE = "executorManagement"
 
   private[scheduler] val EVENT_LOG_QUEUE = "eventLog"
+
+  private[scheduler] val EXTERNAL_CATALOG_QUEUE = "externalCatalog"
 }
 
 private[spark] class LiveListenerBusMetrics(conf: SparkConf)

core/src/main/scala/org/apache/spark/status/api/v1/ApiRootResource.scala

@@ -99,7 +99,7 @@ private[spark] trait UIRoot {
   def checkUIViewPermissions(appId: String, attemptId: Option[String], user: String): Boolean
 }
 
-private[v1] object UIRootFromServletContext {
+private[spark] object UIRootFromServletContext {
 
   private val attribute = getClass.getCanonicalName
 
@@ -112,7 +112,7 @@ private[v1] object UIRootFromServletContext {
   }
 }
 
-private[v1] trait ApiRequestContext {
+private[spark] trait ApiRequestContext {
   @Context
   protected var servletContext: ServletContext = _
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -33,6 +33,7 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.SCRATCH_DIR
+import org.apache.spark.scheduler.{LiveListenerBus, SparkListener, SparkListenerEvent}
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst._
 import org.apache.spark.sql.catalyst.analysis._
@@ -73,7 +74,8 @@ class SessionCatalog(
     cacheSize: Int = SQLConf.get.tableRelationCacheSize,
     cacheTTL: Long = SQLConf.get.metadataCacheTTL,
     defaultDatabase: String = SQLConf.get.defaultDatabase,
-    scratchSessionPath: Option[Path] = None) extends SQLConfHelper with Logging {
+    scratchSessionPath: Option[Path] = None,
+    liveBus: Option[LiveListenerBus] = None) extends SQLConfHelper with Logging {
   import SessionCatalog._
   import CatalogTypes.TablePartitionSpec
 
@@ -219,6 +221,9 @@ class SessionCatalog(
     builder.build[QualifiedTableName, LogicalPlan]()
   }
 
+  private val listener = new ExternalCatalogListener(this)
+  liveBus.map(_.addToExternalCatalogQueue(listener))
+
   /** This method provides a way to get a cached plan. */
   def getCachedPlan(t: QualifiedTableName, c: Callable[LogicalPlan]): LogicalPlan = {
     tableRelationCache.get(t, c)
@@ -1999,3 +2004,17 @@ class SessionCatalog(
     }
   }
 }
+
+private class ExternalCatalogListener(catalog: SessionCatalog) extends SparkListener {
+
+  override def onOtherEvent(event: SparkListenerEvent): Unit = {
+    event match {
+      case e: AlterTableEvent => onAlterTable(e)
+      case _ => // Ignore
+    }
+  }
+
+  private def onAlterTable(e: AlterTableEvent): Unit = {
+    catalog.refreshTable(TableIdentifier(e.name, Some(e.database)))
+  }
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/ui/ExternalCatalogResource.scala

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.ui
+
+import javax.ws.rs._
+import javax.ws.rs.core.MediaType
+
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.glassfish.jersey.server.ServerProperties
+import org.glassfish.jersey.servlet.ServletContainer
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.catalog.{AlterTableEvent, AlterTableKind, DropDatabaseEvent, DropTableEvent}
+import org.apache.spark.sql.execution.ui.EventType.EventType
+import org.apache.spark.status.api.v1.{ApiRequestContext, UIRoot, UIRootFromServletContext}
+import org.apache.spark.ui.SparkUI
+
+object EventType extends Enumeration {
+  type EventType = Value
+  val ALTER_TABLE, DROP_TABLE, DROP_DATABASE = Value
+}
+
+case class CatalogEvent(eventType: EventType, database: String, name: String)
+
+case class RespondMessage(success: Boolean, msg: String)
+
+@Path("/api")
+private[spark] class ExternalCatalogResource extends ApiRequestContext with Logging {
+
+  private val EXTERNAL_CATALOG_QUEUE = "externalCatalog"
+
+  @POST
+  @Path("events")
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @Consumes(Array(MediaType.APPLICATION_JSON))
+  def postEvents(@FormParam("eventList") eventList: Seq[CatalogEvent]): RespondMessage = {
+    try {
+      uiRoot.asInstanceOf[SparkUI].sc.map { sparkContext =>
+        val listener = sparkContext.listenerBus
+        eventList.map { event =>
+          event.eventType match {
+            case EventType.ALTER_TABLE =>
+              val sparkEvent = AlterTableEvent(event.database, event.name, AlterTableKind.TABLE)
+              listener.postToQueue(sparkEvent, EXTERNAL_CATALOG_QUEUE)
+            case EventType.DROP_TABLE =>
+              listener.postToQueue(
+                DropTableEvent(event.database, event.name), EXTERNAL_CATALOG_QUEUE)
+            case EventType.DROP_DATABASE =>
+              listener.postToQueue(
+                DropDatabaseEvent(event.database), EXTERNAL_CATALOG_QUEUE)
+            case _ =>
+          }
+        }
+      }
+      RespondMessage(success = true, "Successful to post catalog events")
+    } catch {
+      case e: Exception =>
+        logWarning("Failed to post catalog events", e)
+        RespondMessage(success = false, "Failed to post catalog events")
+    }
+  }
+}
+
+private[spark] object ExternalCatalogResource {
+  def getServletHandler(uiRoot: UIRoot): ServletContextHandler = {
+    val jerseyContext = new ServletContextHandler(ServletContextHandler.NO_SESSIONS)
+    jerseyContext.setContextPath("/catalog")
+    val holder: ServletHolder = new ServletHolder(classOf[ServletContainer])
+    holder.setInitParameter(ServerProperties.PROVIDER_PACKAGES, "org.apache.spark.status.api.v1")
+    UIRootFromServletContext.setUiRoot(jerseyContext, uiRoot)
+    jerseyContext.addServlet(holder, "/*")
+    jerseyContext
+  }
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLTab.scala

@@ -35,6 +35,7 @@ class SQLTab(val sqlStore: SQLAppStatusStore, sparkUI: SparkUI)
   parent.attachTab(this)
 
   parent.addStaticHandler(SQLTab.STATIC_RESOURCE_DIR, "/static/sql")
+  parent.attachHandler(ExternalCatalogResource.getServletHandler(parent))
 
   override def displayOrder: Int = 0
 }

sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala

@@ -160,7 +160,8 @@ abstract class BaseSessionStateBuilder(
       sqlParser,
       resourceLoader,
       new SparkUDFExpressionBuilder,
-      scratchSessionPath = session.scratchSessionPath)
+      scratchSessionPath = session.scratchSessionPath,
+      liveBus = Some(session.sparkContext.listenerBus))
     parentState.foreach(_.catalog.copyStateTo(catalog))
     catalog
   }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionCatalog.scala

@@ -20,8 +20,10 @@ package org.apache.spark.sql.hive
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 
+import org.apache.spark.scheduler.LiveListenerBus
 import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, TableFunctionRegistry}
 import org.apache.spark.sql.catalyst.catalog._
+import org.apache.spark.sql.catalyst.catalog.{ExternalCatalog, FunctionResourceLoader, GlobalTempViewManager, SessionCatalog}
 import org.apache.spark.sql.catalyst.parser.ParserInterface
 
 private[sql] class HiveSessionCatalog(
@@ -34,7 +36,8 @@ private[sql] class HiveSessionCatalog(
     parser: ParserInterface,
     functionResourceLoader: FunctionResourceLoader,
     functionExpressionBuilder: FunctionExpressionBuilder,
-    scratchSessionPath: Option[Path])
+    scratchSessionPath: Option[Path],
+    liveBus: Option[LiveListenerBus] = None)
   extends SessionCatalog(
     externalCatalogBuilder,
     globalTempViewManagerBuilder,
@@ -44,5 +47,6 @@ private[sql] class HiveSessionCatalog(
     parser,
     functionResourceLoader,
     functionExpressionBuilder,
-    scratchSessionPath = scratchSessionPath) {
+    scratchSessionPath = scratchSessionPath,
+    liveBus = liveBus) {
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionStateBuilder.scala

@@ -77,7 +77,8 @@ class HiveSessionStateBuilder(
       sqlParser,
       resourceLoader,
       HiveUDFExpressionBuilder,
-      scratchSessionPath = session.scratchSessionPath)
+      scratchSessionPath = session.scratchSessionPath,
+      liveBus = Some(session.sparkContext.listenerBus))
     parentState.foreach(_.catalog.copyStateTo(catalog))
     catalog
   }