diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index a98663a..45dceba 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -24,24 +24,30 @@ on: jobs: test: - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 strategy: fail-fast: false matrix: - scala: [2.12.18, 2.13.11] - name: Scala ${{matrix.scala}} + # sbt per project (Scala, Spark) row of the projectMatrix in build.sbt, Spark 4 with Java 17 + include: + - { scala: 2.12.20, spark: 3.5.7, java: '11', project: standardization2_12Spark35 } + - { scala: 2.12.20, spark: 3.5.7, java: '17', project: standardization2_12Spark35 } + - { scala: 2.13.17, spark: 3.5.7, java: '11', project: standardizationSpark35 } + - { scala: 2.13.17, spark: 3.5.7, java: '17', project: standardizationSpark35 } + - { scala: 2.13.17, spark: 4.1.2, java: '17', project: standardizationSpark41 } + name: Scala ${{matrix.scala}}, Spark ${{matrix.spark}}, Java ${{matrix.java}} steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 with: persist-credentials: false - - uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d - - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11" + java-version: ${{ matrix.java }} + distribution: temurin + cache: "sbt" - name: Build and run tests - run: sbt ++${{matrix.scala}} test doc + run: sbt ${{matrix.project}}/test ${{matrix.project}}/doc diff --git a/.github/workflows/format_check.yml b/.github/workflows/format_check.yml index 6ea1db9..a3ba658 100644 --- a/.github/workflows/format_check.yml +++ b/.github/workflows/format_check.yml @@ -22,7 +22,7 @@ on: jobs: scalafmt: - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 name: Scalafmt Check steps: - name: Checkout code @@ -34,10 +34,11 @@ jobs: - uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11" + distribution: temurin + java-version: '17' - name: Run scalafmt And Print Diff continue-on-error: true diff --git a/.github/workflows/jacoco_check.yml b/.github/workflows/jacoco_check.yml index b0070e5..fc1211a 100644 --- a/.github/workflows/jacoco_check.yml +++ b/.github/workflows/jacoco_check.yml @@ -23,21 +23,34 @@ on: jobs: test: - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 strategy: + fail-fast: false matrix: include: - # The project supports Scala 2.12, 2.13 - # The CI runs all tests suites for all supported Scala versions at build.yml - # The codebase for all Scala versions is the same, so the coverage is calculated only once - # Scala 2.12 is chosen since it is supported by the most wide range of Spark versions and - # vendor distributions. + # The project is built for (Scala 2.12, Spark 3.5), (Scala 2.13, Spark 3.5) and (Scala 2.13, Spark 4.1) + # The CI runs all tests suites for all of them at build.yml + # The codebase is shared but the runtime paths differ per Spark major version ( ANSI mode is on by default + # in Spark 4) so coverage is checked once per Spark major version: + # - Scala 2.12 / Spark 3.5: supported by the wide range of Spark versions and vendor distributions + # - Scala 2.13 / Spark 4.1: the only Spark 4 row - scala: 2.12.20 scalaShort: "2.12" + spark: 3.5.7 + project: standardization2_12Spark35 overall: 80.0 changed: 80.0 per-changed-file: 60.0 + - scala: 2.13.17 + scalaShort: "2.13" + spark: 4.1.2 + project: standardizationSpark41 + overall: 75.0 + changed: 80.0 + per-changed-file: 60.0 + name: JaCoCo - Scala ${{matrix.scala}}, Spark ${{matrix.spark}} + steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 @@ -46,10 +59,11 @@ jobs: - uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11" + distribution: temurin + java-version: '17' - name: Setup Python uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 @@ -57,7 +71,7 @@ jobs: python-version: "3.14" - name: Build and run tests - run: sbt ++${{matrix.scala}} jacoco + run: sbt "project ${{matrix.project}}" jacoco - name: Add JaCoCo Report in PR comments id: jacoco @@ -68,7 +82,7 @@ jobs: token: ${{ secrets.GITHUB_TOKEN }} global-thresholds: '${{ matrix.overall }}*${{ matrix.changed }}' report-thresholds-default: '${{ matrix.overall }}*${{ matrix.changed }}*${{ matrix.per-changed-file }}' - title: JaCoCo code coverage report - scala ${{ matrix.scala }} + title: JaCoCo code coverage report - scala ${{ matrix.scala }}, spark ${{ matrix.spark }} update-comment: true - name: Get the Coverage info diff --git a/.github/workflows/licence_check.yml b/.github/workflows/licence_check.yml index afbc4d3..d1fe69a 100644 --- a/.github/workflows/licence_check.yml +++ b/.github/workflows/licence_check.yml @@ -24,17 +24,18 @@ on: jobs: license-test: - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 with: persist-credentials: false - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + - name: Setup JDK + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11" + distribution: temurin + java-version: '17' # note, that task "headerCheck" defaults to just "compile:headerCheck" - see https://github.com/sbt/sbt-header/issues/14 - name: SBT src licence header check diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 1718f16..8d96e72 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -20,7 +20,7 @@ on: jobs: publish: - runs-on: ubuntu-latest + runs-on: ubuntu-22.04 steps: - name: Checkout code uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 @@ -28,14 +28,33 @@ jobs: fetch-depth: 0 persist-credentials: false - - name: Setup Scala - uses: olafurpg/setup-scala@32ffa16635ff8f19cc21ea253a987f0fdf29844c + # Spark 4.1 requires Java 17, the Spark 3.5 rows stay Java 11 compatible + # build.sbt pins -release 11 / -source 11 -target 11. + - name: Setup JDK + uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1 with: - java-version: "adopt@1.11" + distribution: temurin + java-version: '17' + cache: sbt - - run: sbt ci-release + # Making the release auditable: printing every module ci-release will publish, + # ensuring the Spark 4.1 / Scala 2.13 artifact are visible in the log. + # Expected: spark-data-standardization-spark3.5_2.12, + # spark-data-standardization-spark3.5_2.13, + # spark-data-standardization-spark4.1_2.13 + - name: List modules to be published + run: | + sbt -batch -no-colors \ + 'show standardization2_12Spark35 / moduleName' \ + 'show standardizationSpark35 / moduleName' \ + 'show standardizationSpark41 / moduleName' \ + 'show root / publish / skip' + + - name: Publish all cross-build artifacts + run: sbt ci-release env: PGP_PASSPHRASE: ${{ secrets.PGP_PASSPHRASE }} PGP_SECRET: ${{ secrets.PGP_SECRET }} SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }} SONATYPE_USERNAME: ${{ secrets.SONATYPE_USERNAME }} + \ No newline at end of file diff --git a/README.md b/README.md index 976891d..0b1b9ea 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,7 @@ [![License](http://img.shields.io/:license-apache-blue.svg)](http://www.apache.org/licenses/LICENSE-2.0.html) [![Release](https://github.com/AbsaOSS/spark-data-standardization/actions/workflows/release.yml/badge.svg)](https://github.com/AbsaOSS/spark-data-standardization/actions/workflows/release.yml) ![Java 11](https://img.shields.io/badge/Java_11-ED8B00?style=flat&logo=openjdk&logoColor=black) +![Java 17](https://img.shields.io/badge/Java_17-ED8B00?style=flat&logo=openjdk&logoColor=black) - Dataframe in - Standardized Dataframe out @@ -16,50 +17,99 @@ The library needs following dependencies to be included in your project ```sbt "org.apache.spark" %% "spark-core" % SPARK_VERSION, "org.apache.spark" %% "spark-sql" % SPARK_VERSION, -"za.co.absa" %% s"spark-commons-spark${SPARK_MAJOR}.${SPARK_MINOR}" % "0.6.3", +"za.co.absa" %% s"spark-commons-spark${SPARK_MAJOR}.${SPARK_MINOR}" % "1.0.0", ``` ### Usage in SBT: + +The artifact name carries the Spark `major.minor` version it is built against, pick the one matching your Spark: + ```sbt -"za.co.absa" %% "spark-data-standardization" % VERSION +"za.co.absa" %% s"spark-data-standardization-spark${SPARK_MAJOR}.${SPARK_MINOR}" % VERSION ``` +e.g. `"za.co.absa" %% "spark-data-standardization-spark4.1" % VERSION` for Spark 4.1.x on Scala 2.13. ### Usage in Maven -### Scala 2.12 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization_2.12) +### Spark 3.5, Scala 2.12 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization-spark3.5_2.12/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization-spark3.5_2.12) + +```xml + + za.co.absa + spark-data-standardization-spark3.5_2.12 + ${latest_version} + +``` + +### Spark 3.5, Scala 2.13 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization-spark3.5_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization-spark3.5_2.13) ```xml za.co.absa - spark-data-standardization_2.12 + spark-data-standardization-spark3.5_2.13 ${latest_version} ``` -### Scala 2.13 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization_2.13) +### Spark 4.1, Scala 2.13 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization-spark4.1_2.13/badge.svg)](https://maven-badges.herokuapp.com/maven-central/za.co.absa/spark-data-standardization-spark4.1_2.13) ```xml za.co.absa - spark-data-standardization_2.13 + spark-data-standardization-spark4.1_2.13 ${latest_version} ``` +> **Upgrading from versions without the Spark suffix:** the previous `spark-data-standardization_2.12` (Spark 3.5) +> and `spark-data-standardization_2.13` artifacts are replaced by the `-spark.` artifacts above, +> so update the artifact id in your dependency coordinates. + Spark and Scala compatibility >| | Scala 2.12 | Scala 2.13 | >|---|---|---| ->|Spark| 3.5.x | 3.5.x | +>|Spark| 3.5.x | 3.5.x / 4.1.x | + +Java compatibility + +>| | Scala 2.12 (Spark 3.5.x) | Scala 2.13 (Spark 3.5.x) | Scala 2.13 (Spark 4.1.x) | +>|---|---|---|---| +>|Java| 11 / 17 | 11 / 17 | 17 | +### ANSI mode + +Standardization reports values that can't be cast (unparseable, overflowing, invalid dates, ...) as `null` plus an +entry in the error column, never as a failing query. On Spark 4, where `spark.sql.ansi.enabled` defaults to `true`, +`standardize` therefore turns ANSI mode off on the session while the standardized DataFrame is being built (Spark +fixes a cast's evaluation mode when the plan is analysed, this is enough also for actions run later) and restores +the original value afterwards, even with concurrent `standardize` calls on the same session, see [ANSI handling](src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.md). ## How to Release Please see [this file](RELEASE.md) for more details. +## How to build and test + +The build is an [sbt-projectmatrix](https://github.com/sbt/sbt-projectmatrix), with one sbt project per supported +(Scala, Spark) combination: + +| sbt project | Scala | Spark | Artifact | +|------------------------------|-------|-------|---------------------------------------------| +| `standardization2_12Spark35` | 2.12 | 3.5.x | `spark-data-standardization-spark3.5_2.12` | +| `standardizationSpark35` | 2.13 | 3.5.x | `spark-data-standardization-spark3.5_2.13` | +| `standardizationSpark41` | 2.13 | 4.1.x | `spark-data-standardization-spark4.1_2.13` | + +```sbt +sbt test # all combinations +sbt standardizationSpark41/test # single combination +``` +Spark 4 requires Java 17, run sbt on Java 17 to test the Spark 4.1 project. + ## How to generate Code coverage report ```sbt -sbt ++ jacoco +sbt "project " jacoco ``` Code coverage will be generated on path: ``` -{project-root}/target/scala-{scala_version}/jacoco/report/html +{project-root}/target/jvm-{scala_version}-spark{spark_version}/jacoco-report ``` +e.g. `target/jvm-2.13-spark4.1/jacoco-report` for `standardizationSpark41`. diff --git a/build.sbt b/build.sbt index d9b5374..d9b1ae4 100644 --- a/build.sbt +++ b/build.sbt @@ -15,16 +15,17 @@ */ import sys.process._ -import Dependencies._ +import SparkVersionAxis._ ThisBuild / name := "spark-data-standardization" ThisBuild / organization := "za.co.absa" lazy val scala212 = "2.12.20" -lazy val scala213 = "2.13.16" +lazy val scala213 = "2.13.17" +lazy val spark35 = "3.5.7" +lazy val spark41 = "4.1.2" -ThisBuild / crossScalaVersions := Seq(scala212, scala213) -ThisBuild / scalaVersion := scala212 +ThisBuild / scalaVersion := scala213 ThisBuild / versionScheme := Some("early-semver") @@ -32,18 +33,53 @@ ThisBuild / versionScheme := Some("early-semver") ThisBuild / javacOptions ++= Seq("-source", "11", "-target", "11") ThisBuild / scalacOptions ++= Seq("-release", "11") -libraryDependencies ++= dependencyList(scalaVersion.value) - lazy val printSparkScalaVersion = taskKey[Unit]("Print Spark and Scala versions for standardization") -ThisBuild / printSparkScalaVersion := { - val log = streams.value.log - val scalaVers = scalaVersion.value - log.info(s"Building with Spark ${getSparkVersion(scalaVers)}, Scala ${scalaVers}") + +// Major version of the JVM the tests are forked into (Test / javaHome if its set, else the JVM running sbt) +def javaMajorVersion(javaHome: Option[File]): Int = { + val version = javaHome + .flatMap(home => IO.readLines(home / "release").find(_.startsWith("JAVA_VERSION="))) + .map(_.stripPrefix("JAVA_VERSION=").replace("\"", "")) + .getOrElse(sys.props("java.specification.version")) + version.stripPrefix("1.").takeWhile(_.isDigit).toInt } -Test / parallelExecution := false -Test / logBuffered := false -Test / fork := true +// Providing Spark required access to encapsulated JDK internals when running on Java 17+. +// to avoid failures IllegalAccessError (e.g. sun.nio.ch.DirectBuffer) & IllegalAccessException +// (e.g. sun.util.calendar.ZoneInfo) , They don't affect Java 11. +lazy val testJavaOptions = Seq( + "-XX:+IgnoreUnrecognizedVMOptions", + // keep messages of implicit JVM exceptions (e.g. ClassCastException) that tests assert on, even once JIT-compiled + "-XX:-OmitStackTraceInFastThrow", + "-Xmx2048m", + "--add-opens=java.base/java.lang=ALL-UNNAMED", + "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED", + "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED", + "--add-opens=java.base/java.io=ALL-UNNAMED", + "--add-opens=java.base/java.net=ALL-UNNAMED", + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-opens=java.base/java.util=ALL-UNNAMED", + "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED", + "--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED", + "--add-opens=java.base/jdk.internal.ref=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED", + "--add-exports=java.base/sun.nio.ch=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.cs=ALL-UNNAMED", + "--add-opens=java.base/sun.security.action=ALL-UNNAMED", + "--add-opens=java.base/sun.util.calendar=ALL-UNNAMED", + "--add-opens=java.base/sun.net.www.protocol.jar=ALL-UNNAMED", + "--add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED", + "-Djdk.reflect.useDirectMethodHandle=false", + "-Dio.netty.tryReflectionSetAccessible=true", + "-Dio.netty.allocator.type=pooled", + "-Dio.netty.handler.ssl.defaultEndpointVerificationAlgorithm=NONE" +) + +// preventing Java 11 JVM from starting (unknown module / option) +lazy val java17TestJavaOptions = Seq( + "--add-modules=jdk.incubator.vector", + "--enable-native-access=ALL-UNNAMED" +) // Only apply scalafmt to files that differ from master (i.e. files changed in the feature branch or so; n/a on Windows) lazy val fmtFilterExpression: String = System.getProperty("os.name").toLowerCase match { @@ -59,16 +95,47 @@ lazy val fmtFilterExpression: String = System.getProperty("os.name").toLowerCase ).!!.trim s"diff-ref=${baseBranchName}" } -scalafmtFilter.withRank(KeyRanks.Invisible) := fmtFilterExpression + +lazy val commonSettings = Seq( + name := "spark-data-standardization", + printSparkScalaVersion := { + streams.value.log.info(s"Building with Spark ${sparkVersion.value}, Scala ${scalaVersion.value}") + }, + Test / parallelExecution := false, + Test / logBuffered := false, + Test / fork := true, + Test / javaOptions ++= testJavaOptions ++ + (if (javaMajorVersion((Test / javaHome).value) >= 17) java17TestJavaOptions else Seq.empty), + scalafmtFilter.withRank(KeyRanks.Invisible) := fmtFilterExpression, + // JaCoCo Method Filter Plugin + jacocoReportName := s"spark-data-standardization Jacoco Report - spark:${sparkVersion.value} scala:${scalaVersion.value}", + jacocoReportFormats := Set("html", "xml") + // jacocoExcludes := Seq("za/co/absa/standardization/udf/UDFBuilder*", "za/co/absa/standardization/udf/UDFNames") +) + +// Aggregates all the build rows, has no sources of its own and not published +lazy val root = (project in file(".")) + .aggregate(standardization.projectRefs: _*) + .settings( + name := "spark-data-standardization-parent", + publish / skip := true, + crossScalaVersions := Nil, + Compile / unmanagedSourceDirectories := Nil, + Compile / unmanagedResourceDirectories := Nil, + Test / unmanagedSourceDirectories := Nil, + Test / unmanagedResourceDirectories := Nil + ) + +// One row per supported Spark version, each published e.g: spark-data-standardization-spark_ +lazy val standardization = (projectMatrix in file(".")) + .settings(commonSettings: _*) + .sparkRow(SparkVersionAxis(spark35), scalaVersions = Seq(scala212, scala213)) + .sparkRow(SparkVersionAxis(spark41), scalaVersions = Seq(scala213)) + .enablePlugins(JacocoFilterPlugin) // linting Global / excludeLintKeys += ThisBuild / name // will be used in publish, todo #3 - confirm if lint ignore is still needed -// JaCoCo Method Filter Plugin -enablePlugins(JacocoFilterPlugin) -jacocoReportName := s"spark-data-standardization Jacoco Report - scala:${scalaVersion.value}" -jacocoReportFormats := Set("html", "xml") -// jacocoExcludes := Seq("za/co/absa/standardization/udf/UDFBuilder*", "za/co/absa/standardization/udf/UDFNames") // Command aliases for JaCoCo coverage workflow addCommandAlias("jacoco", "; jacocoOn; clean; test; jacocoReportAll; jacocoOff") addCommandAlias("jacocoOn", "; set every jacocoPluginEnabled := true") diff --git a/project/Dependencies.scala b/project/Dependencies.scala index 7fb67b4..69ad413 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -17,27 +17,25 @@ import sbt._ object Dependencies { - private def getSparkVersionUpToMinor(sparkVersion: String): String = { + def getSparkVersionUpToMinor(sparkVersion: String): String = { val pattern = "([0-9]+)\\.([0-9]+)\\.([0-9]+)".r val pattern(major, minor, patch) = sparkVersion s"$major.$minor" } - private def sparkFastTestsVersion(scalaVersion: String): String = "1.1.0" + private val sparkFastTestsVersion: String = "1.1.0" - def getSparkVersion(scalaVersion: String): String = "3.5.7" - - def dependencyList(scalaVersion: String): Seq[ModuleID] = { - val sparkVersion = getSparkVersion(scalaVersion) + /** Dependencies of one build row, the Spark version is provided by the row's [[SparkVersionAxis]]. */ + def dependencyList(sparkVersion: String): Seq[ModuleID] = { val sparkVersionUpToMinor = getSparkVersionUpToMinor(sparkVersion) List( "org.apache.spark" %% "spark-core" % sparkVersion % Provided, "org.apache.spark" %% "spark-sql" % sparkVersion % Provided, - "za.co.absa" %% s"spark-commons-spark$sparkVersionUpToMinor" % "0.6.3" % Provided, - "za.co.absa" %% "spark-commons-test" % "0.6.3" % Test, + "za.co.absa" %% s"spark-commons-spark$sparkVersionUpToMinor" % "1.0.0" % Provided, + "za.co.absa" %% "spark-commons-test" % "1.0.0" % Test, "com.typesafe" % "config" % "1.4.2", - "com.github.mrpowers" %% "spark-fast-tests" % sparkFastTestsVersion(scalaVersion) % Test, - "org.scalatest" %% "scalatest" % "3.2.15" % Test + "com.github.mrpowers" %% "spark-fast-tests" % sparkFastTestsVersion % Test, + "org.scalatest" %% "scalatest" % "3.2.2" % Test ) } } diff --git a/project/SparkVersionAxis.scala b/project/SparkVersionAxis.scala new file mode 100644 index 0000000..9269a45 --- /dev/null +++ b/project/SparkVersionAxis.scala @@ -0,0 +1,53 @@ +/* + * Copyright 2022 ABSA Group Limited + * + * Licensed 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. + */ + +import sbt._ +import sbt.Keys._ +import sbt.internal.ProjectMatrix + +/** + * Spark version as an sbt-project matrix axis (inspired by spark-commons). + * + * Each row of the matrix gets a unique project id e.g. standardization2_12Spark35, its own target directory + * and an artifact name carrying the minor version, e.g. spark-data-standardization-spark3.5_2.12 and + * spark-data-standardization-spark4.1_2.13, the builds for different Spark versions don't collide. + */ +case class SparkVersionAxis(sparkVersion: String) extends VirtualAxis.WeakAxis { + val sparkVersionMinor: String = Dependencies.getSparkVersionUpToMinor(sparkVersion) + + override val directorySuffix: String = s"spark$sparkVersionMinor" + override val idSuffix: String = s"Spark${sparkVersionMinor.replace(".", "")}" + // Scala version suffix e.g: standardization2_12Spark35 and not standardizationSpark352_12 + override val suffixOrder: Int = 200 +} + +object SparkVersionAxis { + val sparkVersion = settingKey[String]("Spark version of the build row") + + implicit class ProjectExtension(val matrix: ProjectMatrix) extends AnyVal { + /** Adds JVM row for the given Scala versions, built and published against the Spark version. */ + def sparkRow(axis: SparkVersionAxis, scalaVersions: Seq[String], settings: Def.Setting[_]*): ProjectMatrix = + matrix.jvmPlatform( + axisValues = Seq(axis), + scalaVersions = scalaVersions, + settings = Seq( + sparkVersion := axis.sparkVersion, + moduleName := s"${name.value}-spark${axis.sparkVersionMinor}", + libraryDependencies ++= Dependencies.dependencyList(axis.sparkVersion) + ) ++ settings + ) + } +} diff --git a/project/plugins.sbt b/project/plugins.sbt index ab827fb..789754b 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -21,6 +21,9 @@ addSbtPlugin("de.heikoseeberger" % "sbt-header" % "5.7.0") addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.5") addSbtPlugin("io.github.moranaapps" % "jacoco-method-filter-sbt" % "2.2.0") + +addSbtPlugin("com.eed3si9n" % "sbt-projectmatrix" % "0.11.0") + lazy val ow2Version = "9.5" def ow2Url(artifactName: String): String = s"https://repo1.maven.org/maven2/org/ow2/asm/$artifactName/$ow2Version/$artifactName-$ow2Version.jar" diff --git a/src/main/scala/za/co/absa/standardization/Standardization.scala b/src/main/scala/za/co/absa/standardization/Standardization.scala index fb8a70e..0f180e3 100644 --- a/src/main/scala/za/co/absa/standardization/Standardization.scala +++ b/src/main/scala/za/co/absa/standardization/Standardization.scala @@ -39,27 +39,79 @@ object Standardization { udfLib.register(sparkSession) implicit val defaults: TypeDefaults = standardizationConfig.typeDefaults - logger.info(s"Step 1: Schema validation") - validateSchemaAgainstSelfInconsistencies(schema) - - logger.info(s"Step 2: Standardization") - val std = standardizeDataset(df, schema, standardizationConfig) - - logger.info(s"Step 3: Clean the final error column") - val cleanedStd = cleanTheFinalErrorColumn(std) - - val idedStd = if (cleanedStd.schema.fieldExists(standardizationConfig.metadataColumns.recordId)) { - cleanedStd // no new id regeneration - } else { - RecordIdGeneration.addRecordIdColumnByStrategy( - cleanedStd, - standardizationConfig.metadataColumns.recordId, - standardizationConfig.metadataColumns.recordIdStrategy - ) + withNonAnsiCasts(sparkSession) { + logger.info(s"Step 1: Schema validation") + validateSchemaAgainstSelfInconsistencies(schema) + + logger.info(s"Step 2: Standardization") + val std = standardizeDataset(df, schema, standardizationConfig) + + logger.info(s"Step 3: Clean the final error column") + val cleanedStd = cleanTheFinalErrorColumn(std) + + val idedStd = if (cleanedStd.schema.fieldExists(standardizationConfig.metadataColumns.recordId)) { + cleanedStd // no new id regeneration + } else { + RecordIdGeneration.addRecordIdColumnByStrategy( + cleanedStd, + standardizationConfig.metadataColumns.recordId, + standardizationConfig.metadataColumns.recordIdStrategy + ) + } + + logger.info(s"Standardization process finished, returning to the application...") + idedStd } + } + + /** + * Standardization treats cast failures (invalid values, overflows and unparseable dates) as error + * records relying on Spark returning `null` from a failed casts. Spark 4 upgrade enables ANSI SQL mode + * by default , this throws errors and aborts the job. This helper temporarily disables ANSI mode during + * plan construction and analysis to preserve the same behavior (null-on-failure) for both Spark 3.5 and + * Spark 4. The original value is restored after. + * See ADR.002-ansi-handling for more details: + * + * - DataFrame analysis is eager and expressions `Cast`, `to_date`, `to_timestamp`, and arithmetic operations + * capture their evaluation mode during analysis, so the DataFrame retains non-ANSI behavior even if + * execution occurs later with ANSI mode enabled. + * + * - Concurrent calls on the same session are reference counted: the first one saves the original value and + * disables ANSI, the last one restores it and overlapping calls never see ANSI re-enabled midway ,the + * original value is always restored. + */ + private def withNonAnsiCasts[T](spark: SparkSession)(body: => T): T = { + NonAnsiScope.enter(spark) + try body + finally NonAnsiScope.exit(spark) + } + + private object NonAnsiScope { + private val AnsiKey = "spark.sql.ansi.enabled" - logger.info(s"Standardization process finished, returning to the application...") - idedStd + private final class State(val previous: Option[String]) { + var activeCalls: Int = 1 + } + + private val states = new java.util.IdentityHashMap[SparkSession, State]() + + def enter(spark: SparkSession): Unit = states.synchronized { + Option(states.get(spark)) match { + case Some(state) => state.activeCalls += 1 + case None => + states.put(spark, new State(spark.conf.getOption(AnsiKey))) + spark.conf.set(AnsiKey, "false") + } + } + + def exit(spark: SparkSession): Unit = states.synchronized { + val state = states.get(spark) + state.activeCalls -= 1 + if (state.activeCalls == 0) { + states.remove(spark) + state.previous.fold(spark.conf.unset(AnsiKey))(spark.conf.set(AnsiKey, _)) + } + } } diff --git a/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.drawio b/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.drawio new file mode 100644 index 0000000..b94a05c --- /dev/null +++ b/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.drawio @@ -0,0 +1,85 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.md b/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.md new file mode 100644 index 0000000..cc6a778 --- /dev/null +++ b/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.md @@ -0,0 +1,130 @@ +# Context + +Standardization library converts raw input into a target schema. Its central rule is **a bad value must not +fail the job**. It becomes `null` in the output column plus an entry in `errCol`: + +``` +stdCastError, E00000, "Cast from 'string' to 'integer'", num, [abc] +``` + +The library implements this by casting and then checking for `null` in the result (see `TypeParser`). That only works +if Spark's casts and parse functions **return `null` on failure**, which is Spark's *legacy* (non-ANSI) behaviour. + +the library emits these ANSI-sensitive constructs on (`src/main/scala/.../stages/TypeParser.scala`, +`InfinitySupportIso.scala`): + +| Construct | Example in library | Spark expression | Where the ANSI flag lives | +|---|---|---|---| +| `Column.cast(dt)` | every type conversion in `TypeParser` | `Cast` | `Cast.evalMode` | +| `to_date(col, pattern)` | `TypeParser.scala:628,630,636,662` | `ParseToDate` → `Cast(GetTimestamp(...))` | `ParseToDate.ansiEnabled` | +| `to_timestamp(col, pattern)` | `TypeParser.scala:684,723` | `ParseToTimestamp` / `GetTimestamp` | `failOnError` | +| `unix_timestamp(col, pattern)` | `TypeParser.scala:699` | `UnixTimestamp` | `failOnError` | +| `col % 1.0` | `TypeParser.scala:399` | `Remainder` | `evalContext` | + + +## What changed in Spark 4: ANSI is on by default + +`spark.sql.ansi.enabled` default value: + +| Version | Source | Default | +|---|---|---| +| 3.5.7 | `spark-catalyst` `org/apache/spark/sql/internal/SQLConf.scala:3198-3207` | `sys.env.get("SPARK_ANSI_SQL_MODE").contains("true")` → **false** unless env var set | +| 4.1.2 | `spark-catalyst` `org/apache/spark/sql/internal/SQLConf.scala:4529-4538` | `!sys.env.get("SPARK_ANSI_SQL_MODE").contains("false")` → **true** unless env var set | + +The extract from config doc https://spark.apache.org/docs/4.1.2/sql-ref-ansi-compliance.html: +> *"When true ... Spark will throw an exception at runtime instead of returning null results when the inputs to a SQL +> operator/function are invalid."* + + +### Observed effect (with ANSI on) + +`standardize(...).collect()` gives the +following : + +| Input | Spark 4.1.2 | Spark 3.5.7 (ANSI explicitly on) | +|---|---|---| +| `"abc"` → int | `SparkNumberFormatException: [CAST_INVALID_INPUT] The value 'abc' of the type "STRING" cannot be cast to "INT" because it is malformed. ... SQLSTATE: 22018` | same error class | +| `"2147483648"` → int (overflow) | `SparkNumberFormatException: [CAST_INVALID_INPUT] The value '2147483648' ...` | same | +| `"not-a-date"` → date (`yyyy-MM-dd`) | `SparkDateTimeException: [CANNOT_PARSE_TIMESTAMP] Text 'not-a-date' could not be parsed at index 0. Use \`try_to_date\` ... SQLSTATE: 22007` | same, but the message ends with *"If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error."* | + + +**Test data :** +``` +=== standardizationSpark41 +spark=4.1.2 case=invalid -> org.apache.spark.SparkNumberFormatException: [CAST_INVALID_INPUT] The value 'abc' of the + type "STRING" cannot be cast to "INT" because it is malformed. Correct the value as per the + syntax, or change its target type. Use `try_cast` to tolerate malformed input and return + NULL instead. SQLSTATE: 22018 +spark=4.1.2 case=overflow -> org.apache.spark.SparkNumberFormatException: [CAST_INVALID_INPUT] The value '2147483648' of + the type "STRING" cannot be cast to "INT" because it is malformed. ... +spark=4.1.2 case=baddate -> org.apache.spark.SparkDateTimeException: [CANNOT_PARSE_TIMESTAMP] Text 'not-a-date' could + not be parsed at index 0. Use `try_to_date` to tolerate invalid input string and return + NULL instead. SQLSTATE: 22007 +- invalid values, overflows and unparseable dates become errors ... *** FAILED *** + org.apache.spark.SparkNumberFormatException: [CAST_INVALID_INPUT] The value 'abc' ... +- overlapping standardizations on a shared session ... *** FAILED *** + java.util.concurrent.ExecutionException: org.apache.spark.SparkNumberFormatException: [CAST_INVALID_INPUT] ... + +=== standardization2_12Spark35 +spark=3.5.7 case=invalid -> org.apache.spark.SparkNumberFormatException: [CAST_INVALID_INPUT] The value 'abc' ... + ... If necessary set "spark.sql.ansi.enabled" to "false" to bypass this error. +spark=3.5.7 case=overflow -> org.apache.spark.SparkNumberFormatException: [CAST_INVALID_INPUT] The value '2147483648' ... +spark=3.5.7 case=baddate -> org.apache.spark.SparkDateTimeException: [CANNOT_PARSE_TIMESTAMP] Text 'not-a-date' could + not be parsed at index 0. If necessary set "spark.sql.ansi.enabled" to "false" to bypass + this error. +- invalid values, overflows and unparseable dates become errors ... *** FAILED *** +- overlapping standardizations on a shared session ... *** FAILED *** +``` +**See file:** `src/test/scala/za/co/absa/standardization/StandardizationAnsiModeSuite.scala` + +Without any handling, **upgrading to Spark 4 with default settings would see the first malformed record +kill the job** instead of being reported in `errCol` . Spark 3.5's error output recommends turning ANSI off. Spark 4 +dropped the hint, but the mechanism is the same. + +--- +# Solution design: `withNonAnsiCasts` / `NonAnsiScope` + +**See file:** `src/main/scala/za/co/absa/standardization/Standardization.scala` +(`standardize` at line 34, wrapper call at line 42, `withNonAnsiCasts` at line 83, `NonAnsiScope` at lines 89-115) + +## 1. How its applied + +```scala +def standardize(df, schema, config)(implicit sparkSession): DataFrame = { + udfLib.register(sparkSession) + ... + withNonAnsiCasts(sparkSession) { + validateSchemaAgainstSelfInconsistencies(schema) // step 1 + val std = standardizeDataset(df, schema, config) // step 2: builds every cast / to_date / % ... (df.select) + val cleanedStd = cleanTheFinalErrorColumn(std) // step 3 + ... RecordIdGeneration ... // optional id column + idedStd // ← the returned DataFrame: plan already analysed + } +} +``` + +This covers **all** plan construction done by the library, from the first `Column` built to the last +`Dataset` created. Each `select` / `withColumn` creates a new Dataset, and in Spark Classic that eagerly analyses +it. + +## 2. Implementation +| Design | Reason | +|----------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Change the **session** conf (`spark.conf.set`) | Found as the only place Spark 4.1 Classic reads the cast mode from (`SparkSession.scala:882`, `columnNodeSupport.scala:47`). | +| **Reference count** per session | Overlapping calls share one ANSI-off window, doesn't restore while another call is still planning, and the original value is captured exactly once. | +| `IdentityHashMap[SparkSession, State]` | Sessions are keyed by **identity**, so two distinct sessions sharing a `SparkContext` get independent windows. The entry is removed as soon as the count hits 0, so sessions are never retained (avoiding potetial memory leak). | +| One `synchronized` monitor, held only for `enter`/`exit` | Planning (`body`) runs **outside** the lock, so concurrent standardizations still plan in parallel. | +| `getOption` + `unset` when absent | If the caller never set the key, we restore "not set", so Spark keeps using its built-in default (and any future default change). We don't freeze an explicit value. | +| `try/finally` | Restores even if schema validation throws `ValidationException` or plan analysis fails. | +| Private, no public API change | Callers don't need to change anything. | + + + +# Outcome + +The returned DataFrame produces null + errCol for invalid / overflowing / unparseable values, even when the action runs after standardize returned with ANSI on , the concurrent calls on a shared session produce correct results. +The session's ANSI setting is restored to the caller's original value after the last overlapping call. + + +![Architecture diagram!](ansi-handling.png) + diff --git a/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.png b/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.png new file mode 100644 index 0000000..b34b53f Binary files /dev/null and b/src/main/scala/za/co/absa/standardization/adr/002-ansi-handling/ansi-handling.png differ diff --git a/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala b/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala index 2914131..7384df8 100644 --- a/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala +++ b/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala @@ -25,7 +25,6 @@ import za.co.absa.spark.commons.implicits.ColumnImplicits.ColumnEnhancements import za.co.absa.spark.commons.implicits.StructFieldImplicits.StructFieldMetadataEnhancements import za.co.absa.spark.commons.implicits.StructTypeImplicits.StructTypeEnhancements import za.co.absa.spark.commons.utils.SchemaUtils -import za.co.absa.spark.hofs.transform import za.co.absa.standardization.config.StandardizationConfig import za.co.absa.standardization.implicits.StdColumnImplicits.StdColumnEnhancements import za.co.absa.standardization.schema.StdSchemaUtils.FieldWithSource @@ -214,9 +213,9 @@ object TypeParser { val finalErrs = when(nullErrCond, array(typedLit(StandardizationErrorMessage.stdNullErr(inputFullPathName)(stdConfig.errorCodes)))) .otherwise( - typedLit(flatten(transform(column, lambdaErrCols, lambdaVariableName))) + typedLit(flatten(transform(column, lambdaErrCols))) ) - val stdCols = transform(column, lambdaStdCols, lambdaVariableName) + val stdCols = transform(column, lambdaStdCols) logger.info(s"Finished standardization plan for Array $inputFullPathName") ParseOutput(stdCols as (fieldOutputName, metadata), finalErrs) } diff --git a/src/test/scala/za/co/absa/standardization/StandardizationAnsiModeSuite.scala b/src/test/scala/za/co/absa/standardization/StandardizationAnsiModeSuite.scala new file mode 100644 index 0000000..bdf8da4 --- /dev/null +++ b/src/test/scala/za/co/absa/standardization/StandardizationAnsiModeSuite.scala @@ -0,0 +1,122 @@ +/* + * Copyright 2021 ABSA Group Limited + * + * Licensed 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 za.co.absa.standardization + +import org.apache.spark.sql.types._ +import org.scalatest.funsuite.AnyFunSuite +import za.co.absa.spark.commons.test.SparkTestBase +import za.co.absa.standardization.RecordIdGeneration.IdType.NoId +import za.co.absa.standardization.StandardizationAnsiModeSuite.AnsiRow +import za.co.absa.standardization.config.{BasicMetadataColumnsConfig, BasicStandardizationConfig, ErrorCodesConfig} +import za.co.absa.standardization.schema.MetadataKeys + +import java.sql.Date +import java.util.concurrent.{Callable, CyclicBarrier, Executors, TimeUnit} +import scala.collection.JavaConverters._ + +/** + * Standardization relies on failed casts producing `null` (which is then turned into an error-column entry). + * Spark 4 enables ANSI mode by default, where invalid input, numeric overflow and unparseable dates throw. + * These tests run with ANSI explicitly enabled on the session and materialize the result only after `standardize` + * has returned, as evidence the non-ANSI cast semantics are fixed in the plan at analysis time and do not depend on the + * session setting when the action executes. They also check that overlapping calls on a shared session are safe + * and that the original session setting is restored. + */ +class StandardizationAnsiModeSuite extends AnyFunSuite with SparkTestBase { + + import spark.implicits._ + + private val AnsiKey = "spark.sql.ansi.enabled" + + private val stdConfig = BasicStandardizationConfig + .fromDefault() + .copy(metadataColumns = BasicMetadataColumnsConfig.fromDefault().copy(recordIdStrategy = NoId)) + private implicit val errorCodes: ErrorCodesConfig = stdConfig.errorCodes + + private val schema = StructType(Seq( + StructField("id", StringType, nullable = false), + StructField("num", IntegerType, nullable = true), + StructField("dt", DateType, nullable = true, + new MetadataBuilder().putString(MetadataKeys.Pattern, "yyyy-MM-dd").build()) + )) + + private def source = Seq( + ("1-valid", "42", "2020-01-15"), + ("2-invalid", "abc", "2020-01-15"), + ("3-overflow", "2147483648", "2020-01-15"), + ("4-baddate", "7", "not-a-date") + ).toDF("id", "num", "dt") + + private val expected = Seq( + AnsiRow("1-valid", Some(42), Some(Date.valueOf("2020-01-15")), Seq.empty), + AnsiRow("2-invalid", None, Some(Date.valueOf("2020-01-15")), + Seq(StandardizationErrorMessage.stdCastErr("num", "abc", "string", "integer", None))), + AnsiRow("3-overflow", None, Some(Date.valueOf("2020-01-15")), + Seq(StandardizationErrorMessage.stdCastErr("num", "2147483648", "string", "integer", None))), + AnsiRow("4-baddate", Some(7), None, + Seq(StandardizationErrorMessage.stdCastErr("dt", "not-a-date", "string", "date", Some("yyyy-MM-dd")))) + ) + + private def withAnsiEnabled[T](body: => T): T = { + val previous = spark.conf.getOption(AnsiKey) + spark.conf.set(AnsiKey, "true") + try body + finally previous.fold(spark.conf.unset(AnsiKey))(spark.conf.set(AnsiKey, _)) + } + + private def standardizeAndCollect(): List[AnsiRow] = { + val std = Standardization.standardize(source, schema, stdConfig) + // the action runs after standardize returned, i.e. with the session still in ANSI mode + std.as[AnsiRow].collect().sortBy(_.id).toList + } + + test("invalid values, overflows and unparseable dates become errors, not exceptions, with ANSI mode enabled") { + withAnsiEnabled { + assertResult(expected)(standardizeAndCollect()) + assertResult("true")(spark.conf.get(AnsiKey)) + } + } + + test("overlapping standardizations on a shared session all keep non-ANSI casts and restore the setting") { + withAnsiEnabled { + val threads = 8 + val rounds = 5 + val pool = Executors.newFixedThreadPool(threads) + try { + (1 to rounds).foreach { _ => + val startTogether = new CyclicBarrier(threads) + val tasks = (1 to threads).map(_ => new Callable[List[AnsiRow]] { + override def call(): List[AnsiRow] = { + startTogether.await() + standardizeAndCollect() + } + }) + val results = pool.invokeAll(tasks.asJava).asScala.map(_.get(5, TimeUnit.MINUTES)) + + results.foreach(result => assertResult(expected)(result)) + assertResult("true")(spark.conf.get(AnsiKey)) + } + } finally { + pool.shutdown() + } + } + } +} + +object StandardizationAnsiModeSuite { + case class AnsiRow(id: String, num: Option[Int], dt: Option[Date], errCol: Seq[ErrorMessage]) +} diff --git a/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuite.scala b/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuite.scala index 6d13274..e85b9f0 100644 --- a/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuite.scala +++ b/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuite.scala @@ -19,10 +19,11 @@ package za.co.absa.standardization.interpreter.stages import org.apache.spark.sql.types._ import org.scalatest.funsuite.AnyFunSuite import za.co.absa.spark.commons.test.SparkTestBase +import za.co.absa.spark.commons.utils.ColUtils import za.co.absa.standardization.RecordIdGeneration.IdType.NoId -import za.co.absa.standardization.config.{BasicMetadataColumnsConfig, BasicStandardizationConfig, StandardizationConfig} +import za.co.absa.standardization.config.{BasicMetadataColumnsConfig, BasicStandardizationConfig} import za.co.absa.standardization.stages.TypeParser -import za.co.absa.standardization.types.{TypeDefaults, CommonTypeDefaults} +import za.co.absa.standardization.types.{CommonTypeDefaults, TypeDefaults} import za.co.absa.standardization.udf.UDFLibrary class TypeParserSuite extends AnyFunSuite with SparkTestBase { @@ -44,21 +45,21 @@ class TypeParserSuite extends AnyFunSuite with SparkTestBase { val schema = StructType(Array(structFieldNoMetadata, structFieldWithMetadataNotSourceColumn, structFieldWithMetadataSourceColumn)) //Just Testing field name override val parseOutputStructFieldNoMetadata = TypeParser.standardize(structFieldNoMetadata, "path", schema, stdConfig) - assertResult(true)(parseOutputStructFieldNoMetadata.stdCol.expr.toString().contains("path.a")) - assertResult(false)(parseOutputStructFieldNoMetadata.stdCol.expr.toString().replaceAll("path.a", "").contains("path")) - assertResult(true)(parseOutputStructFieldNoMetadata.errors.expr.toString().contains("path.a")) - assertResult(false)(parseOutputStructFieldNoMetadata.errors.expr.toString().replaceAll("path.a", "").contains("path")) + assertResult(true)(ColUtils.col2Expr(parseOutputStructFieldNoMetadata.stdCol).toString().contains("path.a")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldNoMetadata.stdCol).toString().replaceAll("path.a", "").contains("path")) + assertResult(true)(ColUtils.col2Expr(parseOutputStructFieldNoMetadata.errors).toString().contains("path.a")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldNoMetadata.errors).toString().replaceAll("path.a", "").contains("path")) val parseOutputStructFieldWithMetadataNotSourceColumn = TypeParser.standardize(structFieldWithMetadataNotSourceColumn, "path", schema, stdConfig) - assertResult(true)(parseOutputStructFieldWithMetadataNotSourceColumn.stdCol.expr.toString().contains("path.b")) - assertResult(false)(parseOutputStructFieldWithMetadataNotSourceColumn.stdCol.expr.toString().replaceAll("path.b", "").contains("path")) - assertResult(true)(parseOutputStructFieldWithMetadataNotSourceColumn.errors.expr.toString().contains("path.b")) - assertResult(false)(parseOutputStructFieldWithMetadataNotSourceColumn.errors.expr.toString().replaceAll("path.b", "").contains("path")) + assertResult(true)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataNotSourceColumn.stdCol).toString().contains("path.b")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataNotSourceColumn.stdCol).toString().replaceAll("path.b", "").contains("path")) + assertResult(true)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataNotSourceColumn.errors).toString().contains("path.b")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataNotSourceColumn.errors).toString().replaceAll("path.b", "").contains("path")) val parseOutputStructFieldWithMetadataSourceColumn = TypeParser.standardize(structFieldWithMetadataSourceColumn, "path", schema, stdConfig) - assertResult(false)(parseOutputStructFieldWithMetadataSourceColumn.stdCol.expr.toString().contains("path.c")) - assertResult(true)(parseOutputStructFieldWithMetadataSourceColumn.stdCol.expr.toString().contains("path.override_c")) - assertResult(false)(parseOutputStructFieldWithMetadataSourceColumn.stdCol.expr.toString().replaceAll("path.override_c", "").contains("path")) - assertResult(false)(parseOutputStructFieldWithMetadataSourceColumn.errors.expr.toString().contains("path.c")) - assertResult(true)(parseOutputStructFieldWithMetadataSourceColumn.errors.expr.toString().contains("path.override_c")) - assertResult(false)(parseOutputStructFieldWithMetadataSourceColumn.errors.expr.toString().replaceAll("path.override_c", "").contains("path")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataSourceColumn.stdCol).toString().contains("path.c")) + assertResult(true)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataSourceColumn.stdCol).toString().contains("path.override_c")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataSourceColumn.stdCol).toString().replaceAll("path.override_c", "").contains("path")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataSourceColumn.errors).toString().contains("path.c")) + assertResult(true)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataSourceColumn.errors).toString().contains("path.override_c")) + assertResult(false)(ColUtils.col2Expr(parseOutputStructFieldWithMetadataSourceColumn.errors).toString().replaceAll("path.override_c", "").contains("path")) } } diff --git a/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuiteTemplate.scala b/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuiteTemplate.scala index c9b5bd2..a57022a 100644 --- a/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuiteTemplate.scala +++ b/src/test/scala/za/co/absa/standardization/interpreter/stages/TypeParserSuiteTemplate.scala @@ -302,14 +302,132 @@ trait TypeParserSuiteTemplate extends AnyFunSuite with SparkTestBase { } private def doAssert(expectedExpression: String, actualExpression: String, method: String): Unit = { - if (actualExpression != expectedExpression) { + val (expected, actual) = + if (SPARK_VERSION.startsWith("4.")) (normalizeExpr(expectedExpression), normalizeExpr(actualExpression)) + else (expectedExpression, actualExpression) + + if (actual != expected) { // the expressions tend to be rather long, the assert most often cuts the beginning and/or end of the string // showing just the vicinity of the difference, so we log the output of the whole strings log.error(s"Method: $method") - log.error(s"Expected: $expectedExpression") - log.error(s"Actual : $actualExpression") - assert(actualExpression == expectedExpression) + log.error(s"Expected: $expected") + log.error(s"Actual : $actual") + assert(actual == expected) + } + } + + /** + * Spark 4 changed how `Column.toString` renders expressions compared to Spark 3 and older versions. + * This normalizer converts both expected and actual expressions to the same form , it is only applied for Spark 4. + */ + private def normalizeExpr(expr: String): String = { + transformPrefixOps(expr) + .replace("isNaN(", "isnan(") + .replace("List()", "[]") + .replace("ARRAY()", "[]") + .replace("`", "") + .replace("'", "") + .replaceAll("(DATE|TIMESTAMP) (?=\\d{4})", "") + .replaceAll("(\\d\\d:\\d\\d:\\d\\d)\\.0", "$1") + .replaceAll("(\\d)L", "$1") + } + + /** Spark 4 prefix operators, longer names come first so `isNotNull` is not mistaken for `isNull`, `>=` for `>`. */ + private val prefixOps = List("isNotNull", "isNull", "and", "or", "in", ">=", "<=", "=", ">", "<", "/", "%") + + /** + * Converts Spark 4 prefix operators to Spark 3 infix syntax while leaving other expressions (e.g. `CAST`, `to_date`, + * `array`, literals) unchanged. + * + * An operator is only recognised at the start of the string or right after `(`, `,` or a space, preventing function + * names that contain operator names from being rewritten. + * + * Examples: + * {{{ + * isNull(x) => (x IS NULL) + * and(isNotNull(a), >(a, 5)) => ((a IS NOT NULL) AND (a > 5)) + * or(isNull(c), !(=(%(d, 1.0), 0.0))) => ((c IS NULL) OR (NOT ((d % 1.0) = 0.0))) + * in(CAST(d AS DOUBLE), Infinity, -Infinity) => (CAST(d AS DOUBLE) IN (Infinity, -Infinity)) + * }}} + */ + private def transformPrefixOps(s: String): String = { + val sb = new StringBuilder + var i = 0 + val n = s.length + while (i < n) { + val c = s.charAt(i) + val boundary = i == 0 || { val p = s.charAt(i - 1); p == '(' || p == ',' || p == ' ' } + if (c == '!' && i + 1 < n && s.charAt(i + 1) == '(') { + val close = matchParen(s, i + 1) + val inner = s.substring(i + 2, close) + sb.append("(NOT ").append(transformPrefixOps(inner)).append(")") + i = close + 1 + } else if (boundary && matchOpAt(s, i).isDefined) { + val op = matchOpAt(s, i).get + val openIdx = i + op.length + val close = matchParen(s, openIdx) + val args = splitArgs(s.substring(openIdx + 1, close)).map(a => transformPrefixOps(a.trim)) + sb.append(renderInfix(op, args)) + i = close + 1 + } else { + sb.append(c) + i += 1 + } + } + sb.toString + } + + /** Returns the prefix operator starting at position `i` (operator name immediately followed by `(`). */ + private def matchOpAt(s: String, i: Int): Option[String] = + prefixOps.find(op => s.startsWith(op + "(", i)) + + /** Returns the index of the `)` matching the `(` at `openIdx`, e.g. `matchParen("f(a(b))", 1) == 6`. */ + private def matchParen(s: String, openIdx: Int): Int = { + var depth = 0 + var i = openIdx + while (i < s.length) { + s.charAt(i) match { + case '(' => depth += 1 + case ')' => depth -= 1; if (depth == 0) return i + case _ => + } + i += 1 + } + throw new IllegalArgumentException(s"Unbalanced parentheses in expression: $s") + } + + /** Splits a comma separated argument list on top-level commas only, e.g. `"a, f(b, c)"` => `List("a", " f(b, c)")`. */ + private def splitArgs(inner: String): List[String] = { + val args = scala.collection.mutable.ListBuffer.empty[String] + val sb = new StringBuilder + var depth = 0 + inner.foreach { + case '(' => depth += 1; sb.append('(') + case ')' => depth -= 1; sb.append(')') + case ',' if depth == 0 => args += sb.toString; sb.setLength(0) + case ch => sb.append(ch) } + if (sb.nonEmpty || args.nonEmpty) args += sb.toString + args.toList + } + + /** + * Renders one prefix operator with its normalized arguments in Spark 3 infix form: + * {{{ + * renderInfix("isNull", List("x")) => (x IS NULL) + * renderInfix("isNotNull", List("x")) => (x IS NOT NULL) + * renderInfix("and", List("a", "b")) => (a AND b) + * renderInfix("in", List("x", "1", "2")) => (x IN (1, 2)) + * renderInfix(">", List("length(s)", "8")) => (length(s) > 8) + * }}} + */ + private def renderInfix(op: String, args: List[String]): String = op match { + case "isNull" => s"(${args.head} IS NULL)" + case "isNotNull" => s"(${args.head} IS NOT NULL)" + case "and" => s"(${args.head} AND ${args(1)})" + case "or" => s"(${args.head} OR ${args(1)})" + case "in" => s"(${args.head} IN (${args.tail.mkString(", ")}))" + case symbol => s"(${args.head} $symbol ${args(1)})" } } diff --git a/src/test/scala/za/co/absa/standardization/types/parsers/IntegralParser_RadixIntegralParserSuite.scala b/src/test/scala/za/co/absa/standardization/types/parsers/IntegralParser_RadixIntegralParserSuite.scala index 0c94bdd..200a0e1 100644 --- a/src/test/scala/za/co/absa/standardization/types/parsers/IntegralParser_RadixIntegralParserSuite.scala +++ b/src/test/scala/za/co/absa/standardization/types/parsers/IntegralParser_RadixIntegralParserSuite.scala @@ -80,7 +80,7 @@ class IntegralParser_RadixIntegralParserSuite extends AnyFunSuite { val wrong = "g" val fail = parser.parse(wrong).failed.get assert(fail.isInstanceOf[NumberFormatException]) - assert(fail.getMessage == """For input string: "g"""") + assert(fail.getMessage.startsWith("""For input string: "g"""")) } test("base 2 parsing succeeds") { @@ -103,7 +103,7 @@ class IntegralParser_RadixIntegralParserSuite extends AnyFunSuite { val wrong = "3" val fail = parser.parse(wrong).failed.get assert(fail.isInstanceOf[NumberFormatException]) - assert(fail.getMessage == """For input string: """" + wrong + """"""") + assert(fail.getMessage.startsWith("""For input string: """" + wrong + """"""")) } @@ -127,7 +127,7 @@ class IntegralParser_RadixIntegralParserSuite extends AnyFunSuite { val wrong = "__" val fail = parser.parse(wrong).failed.get assert(fail.isInstanceOf[NumberFormatException]) - assert(fail.getMessage == """For input string: """" + wrong + """"""") + assert(fail.getMessage.startsWith("""For input string: """" + wrong + """"""")) } test("string base inputs") {