[SPARK-52166] [SDP] Add support for PipelineEvents

.github/workflows/build_and_test.yml

@@ -246,7 +246,7 @@ jobs:
           - >-
             api, catalyst, hive-thriftserver
           - >-
-            mllib-local, mllib, graphx, profiler
+            mllib-local, mllib, graphx, profiler, pipelines
           - >-
             streaming, sql-kafka-0-10, streaming-kafka-0-10, streaming-kinesis-asl,
             kubernetes, hadoop-cloud, spark-ganglia-lgpl, protobuf, connect

sql/pipelines/pom.xml

@@ -31,8 +31,17 @@
     <packaging>jar</packaging>
     <name>Spark Project Declarative Pipelines Library</name>
     <url>https://spark.apache.org/</url>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
     <build>
-        <outputDirectory>core/target/scala-2.13/classes</outputDirectory>
-        <testOutputDirectory>core/target/scala-2.13/test-classes</testOutputDirectory>
+        <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
+        <testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory>
     </build>
 </project>

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/common/GraphStates.scala

@@ -0,0 +1,53 @@
+/*
+ * 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.pipelines.common
+
+// The status of the flow.
+sealed trait FlowStatus
+object FlowStatus {
+  // Flow is queued and will be started after all its dependencies have been updated.
+  case object QUEUED extends FlowStatus
+  // Flow is in the process of starting.
+  case object STARTING extends FlowStatus
+  // A task is currently running an update for this flow.
+  case object RUNNING extends FlowStatus
+  // This flow's update has completed successfully.
+  case object COMPLETED extends FlowStatus
+  // This flow's update has failed. Additional information about the failure is present in the error
+  // field.
+  case object FAILED extends FlowStatus
+  // This flow's update was skipped because an upstream dependency failed.
+  case object SKIPPED extends FlowStatus
+  // This flow's query was stopped or canceled by a user action.
+  case object STOPPED extends FlowStatus
+  // Flow is in the process of planning.
+  case object PLANNING extends FlowStatus
+  // This flow is excluded if it's not selected in the partial graph update API call.
+  case object EXCLUDED extends FlowStatus
+  // This flow is idle because there are no updates to be made because all available data has
+  // already been processed.
+  case object IDLE extends FlowStatus
+}
+
+// The type of the dataset.
+sealed trait DatasetType
+object DatasetType {
+  // Dataset is a materialized view.
+  case object MATERIALIZED_VIEW extends DatasetType
+  // Dataset is a streaming table.
+  case object STREAMING_TABLE extends DatasetType
+}

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/logging/ConstructPipelineEvent.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.pipelines.logging
+
+import java.sql.Timestamp
+import java.time.Instant
+import java.util.UUID
+
+/**
+ * A factory object that is used to construct [[PipelineEvent]]s with common fields
+ * automatically filled in. Developers should always use this factory rather than construct
+ * an event directly from an empty proto.
+ */
+object ConstructPipelineEvent {
+
+  /**
+   * Converts an exception that was thrown during a pipeline run to a more structured and standard
+   * internal representation.
+   */
+  private[pipelines] def serializeException(t: Throwable): Seq[SerializedException] = {
+    val className = t.getClass.getName
+    val stacks = Option(t.getStackTrace).map(_.toSeq).getOrElse(Nil).map { f =>
+      StackFrame(declaringClass = f.getClassName, methodName = f.getMethodName)
+    }
+    SerializedException(className = className, message = t.getMessage, stack = stacks) +:
+    Option(t.getCause).map(serializeException).getOrElse(Nil)
+  }
+
+  def constructErrorDetails(t: Throwable): ErrorDetail = ErrorDetail(serializeException(t))
+
+  /**
+   * Returns a new event with the current or provided timestamp and the given origin/message.
+   */
+  def apply(
+      origin: PipelineEventOrigin,
+      message: String,
+      details: EventDetails,
+      exception: Throwable = null,
+      eventTimestamp: Option[Timestamp] = None
+  ): PipelineEvent = {
+    ConstructPipelineEvent(
+      origin = origin,
+      message = message,
+      details = details,
+      errorDetails = Option(exception).map(constructErrorDetails),
+      eventTimestamp = eventTimestamp
+    )
+  }
+
+  /**
+   * Returns a new event with the current or given timestamp and the given origin / message.
+   */
+  def apply(
+      origin: PipelineEventOrigin,
+      message: String,
+      details: EventDetails,
+      errorDetails: Option[ErrorDetail],
+      eventTimestamp: Option[Timestamp]
+  ): PipelineEvent = synchronized {
+
+    val eventUUID = UUID.randomUUID()
+    val timestamp = Timestamp.from(Instant.now())
+
+    PipelineEvent(
+      id = eventUUID.toString,
+      timestamp = EventHelpers.formatTimestamp(eventTimestamp.getOrElse(timestamp)),
+      message = message,
+      details = details,
+      error = errorDetails,
+      origin = origin
+    )
+  }
+}

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/logging/EventHelpers.scala

@@ -0,0 +1,52 @@
+/*
+ * 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.pipelines.logging
+
+import java.sql.Timestamp
+import java.time.{Instant, ZoneId}
+import java.time.format.DateTimeFormatter
+
+/** Contains helpers and implicits for working with [[PipelineEvent]]s. */
+object EventHelpers {
+
+  /** A format string that defines how timestamps are serialized in a [[PipelineEvent]]. */
+  private val timestampFormat: String = "yyyy-MM-dd'T'HH:mm:ss.SSSXX"
+  // Currently only the UTC timezone is supported. Eventually we want to allow the user to specify
+  // the timezone as a pipeline level setting using the SESSION_LOCAL_TIMEZONE key, and it should
+  // not be possible to change this setting during a pipeline run.
+  private val zoneId: ZoneId = ZoneId.of("UTC")
+
+  private val formatter: DateTimeFormatter = DateTimeFormatter
+    .ofPattern(timestampFormat)
+    .withZone(zoneId)
+
+  /** Converts a timestamp to a string in ISO 8601 format. */
+  def formatTimestamp(ts: Timestamp): String = {
+    val instant = Instant.ofEpochMilli(ts.getTime)
+    formatter.format(instant)
+  }
+
+  /** Converts an ISO 8601 formatted timestamp to a {@link java.sql.Timestamp}.  */
+  def parseTimestamp(timeString: String): Timestamp = {
+    if (timeString.isEmpty) {
+      new Timestamp(0L)
+    } else {
+      val instant = Instant.from(formatter.parse(timeString))
+      new Timestamp(instant.toEpochMilli)
+    }
+  }
+}

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/logging/PipelineEvent.scala

@@ -0,0 +1,80 @@
+/*
+ * 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.pipelines.logging
+
+import org.apache.spark.sql.pipelines.common.FlowStatus
+
+/**
+ * An internal event that is emitted during the run of a pipeline.
+ * @param id A globally unique id
+ * @param timestamp The time of the event
+ * @param origin Where the event originated from
+ * @param message A user friendly description of the event
+ * @param details The details of the event
+ * @param error An error that occurred during the event
+ */
+case class PipelineEvent(
+    id: String,
+    timestamp: String,
+    origin: PipelineEventOrigin,
+    message: String,
+    details: EventDetails,
+    error: Option[ErrorDetail]
+)
+
+/**
+ * Describes where the event originated from
+ * @param datasetName The name of the dataset
+ * @param flowName The name of the flow
+ * @param sourceCodeLocation The location of the source code
+ */
+case class PipelineEventOrigin(
+    datasetName: Option[String],
+    flowName: Option[String],
+    sourceCodeLocation: Option[SourceCodeLocation]
+)
+
+/**
+ * Describes the location of the source code
+ * @param path The path to the source code
+ * @param lineNumber The line number of the source code
+ * @param columnNumber The column number of the source code
+ * @param endingLineNumber The ending line number of the source code
+ * @param endingColumnNumber The ending column number of the source code
+ */
+case class SourceCodeLocation(
+    path: Option[String],
+    lineNumber: Option[Int],
+    columnNumber: Option[Int],
+    endingLineNumber: Option[Int],
+    endingColumnNumber: Option[Int]
+)
+
+// Additional details about the PipelineEvent
+trait EventDetails
+
+// An event indicating that a flow has made progress and transitioned to a different state
+case class FlowProgress(status: FlowStatus) extends EventDetails
+
+// Additional details about the error that occurred during the event
+case class ErrorDetail(exceptions: Seq[SerializedException])
+
+// An exception that was thrown during a pipeline run
+case class SerializedException(className: String, message: String, stack: Seq[StackFrame])
+
+// A stack frame of an exception
+case class StackFrame(declaringClass: String, methodName: String)