Address comments by HyukjinKwon · Pull Request #12 · MaxGekk/spark

HyukjinKwon · 2018-10-16T04:07:28Z

No description provided.

### What changes were proposed in this pull request? This PR proposes to make `PythonFunction` holds `Seq[Byte]` instead of `Array[Byte]` to be able to compare if the byte array has the same values for the cache manager. ### Why are the changes needed? Currently the cache manager doesn't use the cache for `udf` if the `udf` is created again even if the functions is the same. ```py >>> func = lambda x: x >>> df = spark.range(1) >>> df.select(udf(func)("id")).cache() ``` ```py >>> df.select(udf(func)("id")).explain() == Physical Plan == *(2) Project [pythonUDF0#14 AS <lambda>(id)#12] +- BatchEvalPython [<lambda>(id#0L)], [pythonUDF0#14] +- *(1) Range (0, 1, step=1, splits=12) ``` This is because `PythonFunction` holds `Array[Byte]`, and `equals` method of array equals only when the both array is the same instance. ### Does this PR introduce _any_ user-facing change? Yes, if the user reuse the Python function for the UDF, the cache manager will detect the same function and use the cache for it. ### How was this patch tested? I added a test case and manually. ```py >>> df.select(udf(func)("id")).explain() == Physical Plan == InMemoryTableScan [<lambda>(id)#12] +- InMemoryRelation [<lambda>(id)#12], StorageLevel(disk, memory, deserialized, 1 replicas) +- *(2) Project [pythonUDF0#5 AS <lambda>(id)#3] +- BatchEvalPython [<lambda>(id#0L)], [pythonUDF0#5] +- *(1) Range (0, 1, step=1, splits=12) ``` Closes apache#28774 from ueshin/issues/SPARK-31945/udf_cache. Authored-by: Takuya UESHIN <ueshin@databricks.com> Signed-off-by: HyukjinKwon <gurwls223@apache.org>

…n properly ### What changes were proposed in this pull request? Make `ResolveRelations` handle plan id properly ### Why are the changes needed? bug fix for Spark Connect, it won't affect classic Spark SQL before this PR: ``` from pyspark.sql import functions as sf spark.range(10).withColumn("value_1", sf.lit(1)).write.saveAsTable("test_table_1") spark.range(10).withColumnRenamed("id", "index").withColumn("value_2", sf.lit(2)).write.saveAsTable("test_table_2") df1 = spark.read.table("test_table_1") df2 = spark.read.table("test_table_2") df3 = spark.read.table("test_table_1") join1 = df1.join(df2, on=df1.id==df2.index).select(df2.index, df2.value_2) join2 = df3.join(join1, how="left", on=join1.index==df3.id) join2.schema ``` fails with ``` AnalysisException: [CANNOT_RESOLVE_DATAFRAME_COLUMN] Cannot resolve dataframe column "id". It's probably because of illegal references like `df1.select(df2.col("a"))`. SQLSTATE: 42704 ``` That is due to existing plan caching in `ResolveRelations` doesn't work with Spark Connect ``` === Applying Rule org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveRelations === '[#12]Join LeftOuter, '`==`('index, 'id) '[#12]Join LeftOuter, '`==`('index, 'id) !:- '[#9]UnresolvedRelation [test_table_1], [], false :- '[#9]SubqueryAlias spark_catalog.default.test_table_1 !+- '[#11]Project ['index, 'value_2] : +- 'UnresolvedCatalogRelation `spark_catalog`.`default`.`test_table_1`, [], false ! +- '[#10]Join Inner, '`==`('id, 'index) +- '[#11]Project ['index, 'value_2] ! :- '[#7]UnresolvedRelation [test_table_1], [], false +- '[#10]Join Inner, '`==`('id, 'index) ! +- '[#8]UnresolvedRelation [test_table_2], [], false :- '[#9]SubqueryAlias spark_catalog.default.test_table_1 ! : +- 'UnresolvedCatalogRelation `spark_catalog`.`default`.`test_table_1`, [], false ! +- '[#8]SubqueryAlias spark_catalog.default.test_table_2 ! +- 'UnresolvedCatalogRelation `spark_catalog`.`default`.`test_table_2`, [], false Can not resolve 'id with plan 7 ``` `[#7]UnresolvedRelation [test_table_1], [], false` was wrongly resolved to the cached one ``` :- '[#9]SubqueryAlias spark_catalog.default.test_table_1 +- 'UnresolvedCatalogRelation `spark_catalog`.`default`.`test_table_1`, [], false ``` ### Does this PR introduce _any_ user-facing change? yes, bug fix ### How was this patch tested? added ut ### Was this patch authored or co-authored using generative AI tooling? ci Closes apache#45214 from zhengruifeng/connect_fix_read_join. Authored-by: Ruifeng Zheng <ruifengz@apache.org> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>

…onicalized expressions ### What changes were proposed in this pull request? Make PullOutNonDeterministic use canonicalized expressions to dedup group and aggregate expressions. This affects pyspark udfs in particular. Example: ``` from pyspark.sql.functions import col, avg, udf pythonUDF = udf(lambda x: x).asNondeterministic() spark.range(10)\ .selectExpr("id", "id % 3 as value")\ .groupBy(pythonUDF(col("value")))\ .agg(avg("id"), pythonUDF(col("value")))\ .explain(extended=True) ``` Currently results in a plan like this: ``` Aggregate [_nondeterministic#15](#15), [_nondeterministic#15 AS dummyNondeterministicUDF(value)#12, avg(id#0L) AS avg(id)#13, dummyNondeterministicUDF(value#6L)#8 AS dummyNondeterministicUDF(value)#14](#15%20AS%20dummyNondeterministicUDF(value)#12,%20avg(id#0L)%20AS%20avg(id)#13,%20dummyNondeterministicUDF(value#6L)#8%20AS%20dummyNondeterministicUDF(value)#14) +- Project [id#0L, value#6L, dummyNondeterministicUDF(value#6L)#7 AS _nondeterministic#15](#0L,%20value#6L,%20dummyNondeterministicUDF(value#6L)#7%20AS%20_nondeterministic#15) +- Project [id#0L, (id#0L % cast(3 as bigint)) AS value#6L](#0L,%20(id#0L%20%%20cast(3%20as%20bigint))%20AS%20value#6L) +- Range (0, 10, step=1, splits=Some(2)) ``` and then it throws: ``` [[MISSING_AGGREGATION] The non-aggregating expression "value" is based on columns which are not participating in the GROUP BY clause. Add the columns or the expression to the GROUP BY, aggregate the expression, or use "any_value(value)" if you do not care which of the values within a group is returned. SQLSTATE: 42803 ``` - how canonicalized fixes this: - nondeterministic PythonUDF expressions always have distinct resultIds per udf - The fix is to canonicalize the expressions when matching. Canonicalized means that we're setting the resultIds to -1, allowing us to dedup the PythonUDF expressions. - for deterministic UDFs, this rule does not apply and "Post Analysis" batch extracts and deduplicates the expressions, as expected ### Why are the changes needed? - the output of the query with the fix applied still makes sense - the nondeterministic UDF is invoked only once, in the project. ### Does this PR introduce _any_ user-facing change? Yes, it's additive, it enables queries to run that previously threw errors. ### How was this patch tested? - added unit test ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#52061 from benrobby/adhoc-fix-pull-out-nondeterministic. Authored-by: Ben Hurdelhey <ben.hurdelhey@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

### What changes were proposed in this pull request? Two CI fixes, both pre-existing on `master`. **1. Build the Varka engine before every Spark build.** `sql/catalyst/pom.xml` and `sql/core/pom.xml` declare a test-scope dependency on `org.apache.spark.varka:varka-engine:0.1.0-SNAPSHOT`, but `sql/varka/engine` is a standalone Maven module outside the reactor, so no Spark build produces it. This PR adds a composite action, `.github/actions/install-varka-engine`, that installs the engine into the local Maven repository, and invokes it from every job that builds Spark, immediately after that job's JDK is set up: | Workflow | Jobs | | :--- | :--- | | `build_and_test.yml` | build, precompile, pyspark, sparkr, lint, java25, docs, tpcds-1g, docker-integration-tests, k8s-integration-tests | | `maven_test.yml` | both build jobs | | `python_hosted_runner_test.yml` | both build jobs | Jobs that only need the jar run with `-Dmaven.test.skip=true`. A new `varka-engine` job runs the engine's own suites with tests enabled - nothing in CI ran them before. Four jobs are deliberately left untouched, because they build no Spark code: `precondition` (change detection), `infra-image` (docker image), `buf` (protobuf lint plus a Python codegen check, no JDK at all) and `ui` (Node/Jest only). **2. Run the push build on JDK 25.** `build_main.yml` called `build_and_test.yml` with no `with:` block, so every push build took the workflow's default of `java: 17`. It now passes `java: 25`. ### Why are the changes needed? Every CI job that builds Spark dies before compiling anything: ``` sbt.librarymanagement.ResolveException: Error downloading org.apache.spark.varka:varka-engine:0.1.0-SNAPSHOT not found: https://repo1.maven.org/maven2/.../varka-engine-0.1.0-SNAPSHOT.pom not found: /home/runner/.m2/repository/.../varka-engine-0.1.0-SNAPSHOT.pom (catalyst / update) ``` and under Maven: ``` [ERROR] Failed to execute goal on project spark-catalyst_2.13: Could not resolve dependencies for project org.apache.spark:spark-catalyst_2.13 [ERROR] dependency: org.apache.spark.varka:varka-engine:jar:0.1.0-SNAPSHOT (test) ``` The dependency resolves on a developer machine only because sbt's resolver list includes `Resolver.mavenLocal` (`SparkBuild.scala:319`) and the engine gets installed by hand with `./build/mvn -f sql/varka/engine/pom.xml install`. A clean runner has no such copy, and the artifact is not published anywhere. On `master` this shows up as one red job, `Run / Precompile Spark`, with Linters / Java 25 / Documentation *skipped* behind it. On a branch push those three run independently and each hits the same wall, so four jobs fail. Resolution happens before compilation, so no Spark or Varka source is ever reached. The JDK change is a second, independent breakage found while fixing the first. The root pom sets `<java.version>25</java.version>`, which feeds both `maven.compiler.release` and the enforcer's `requireJavaVersion` rule, so a Spark build on JDK 17 cannot succeed - the resolution failure simply came first. It also matters directly here: the engine sets `maven.compiler.release=25`, so installing it on a Java 17 job would fail on its own. ### Does this PR introduce _any_ user-facing change? No. CI configuration only. ### How was this patch tested? Both commands the composite action runs were verified locally on JDK 25: ``` $ ./build/mvn -B -f sql/varka/engine/pom.xml -Dmaven.test.skip=true install [INFO] Installing .../varka-engine-0.1.0-SNAPSHOT.jar to ~/.m2/repository/... [INFO] BUILD SUCCESS $ ./build/mvn -B -f sql/varka/engine/pom.xml install [INFO] Tests run: 24, Failures: 0, Errors: 0, Skipped: 0 [INFO] BUILD SUCCESS ``` All five touched YAML files parse, and a script over the parsed `build_and_test.yml` job graph confirmed the engine step is present in exactly the ten jobs that build Spark and absent from the four that do not. Checked that the new step ordering does not disturb the `lint` job: it now builds the engine before `dev/check-license`, and `target` is the first entry in `dev/.rat-excludes`, so the generated tree is not scanned. CI on this branch is the real test - the whole point is that this cannot be verified locally. **Not attempted here:** the structural fix, making `sql/varka/engine` a real reactor module. sbt's `PomBuild` would then create a project for it and compile it without `--add-modules jdk.incubator.vector`, because Maven plugin configuration is not translated into sbt `javacOptions`. That needs `SparkBuild.scala` changes which cannot be validated without a full Spark build. Tracked as finding 4 in `sql/varka/ISSUES.md`. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5)

…rnels ### What changes were proposed in this pull request? The residual half of finding 4 in `sql/varka/ISSUES.md`, which is the last thing left open in that review. #12 got the engine built and tested in CI; two parts stayed open. The engine was not a Maven reactor module, so a Maven build outside CI still depended on a manual `-f sql/varka/engine/pom.xml install`. And every runner was x86_64, so the four-lane kernel path that finding 1 fixed was never actually executed - the finding said so itself: "an aarch64 runner is what would actually execute the kernels at 4 lanes." **`sql/varka/engine` is now a module of the root pom**, listed before `sql/api`. Nothing had to be told the order: catalyst declares the dependency, so Maven puts the engine ahead of it on its own - ``` [INFO] Building Spark Project SQL API 5.0.0-SNAPSHOT [17/19] [INFO] Building Varka Engine 0.1.0-SNAPSHOT [18/19] [INFO] Building Spark Project Catalyst 5.0.0-SNAPSHOT [19/19] ``` so `./build/mvn install` builds and installs it before anything that needs it, and no manual step is left in a Maven build. It keeps its own pom rather than inheriting `spark-parent`. Its sources need the incubator Vector API and its tests need native-access flags the Spark build does not set, and inheriting would also apply the Scala, shading and packaging plugins it has no use for. So it still builds standalone, which is how CI puts it into the local Maven repository - **sbt** resolves it from there, not from the reactor, so the composite action stays. Its description said the engine was "outside the Spark reactor"; it now says what the step is actually still for. **Being in the reactor exposed one cost of having no parent.** `dev/lint-java` runs `checkstyle:check` over every module, and the plugin resolves a relative `configLocation` against the *parent* project's directory - which a module with no parent does not have. So the engine was checked against the plugin's default sun_checks: 294 violations, of the `MagicNumber` / `FinalParameters` / 80-column variety. The engine pom now configures checkstyle itself, at `${maven.multiModuleProjectDirectory}/dev/checkstyle.xml`, which is the repository root whether the reactor or that pom alone is built. Against Spark's actual rules the engine had two violations, both files missing a trailing newline. The engine's Java is linted by `dev/lint-java` from here on, which it never was before. **The four-lane risk is covered from two directions**, and both are needed. The engine's own build now runs its suite twice. The second execution adds `-XX:MaxVectorSize=16`, which makes `IntVector.SPECIES_PREFERRED` four lanes - the shape of a 128-bit NEON machine, and the narrow-species case the kernels were once wrong on. That covers the four-lane *code path* on every runner, x86_64 included. It could fail silently, though: if a JVM ignored the flag the run would just repeat the first one and the coverage would evaporate with nothing to show for it. So the narrow execution also passes `-Dvarka.expected.int.lanes=4`, and `DateVectorOpsTest.preferredSpeciesIsTheWidthTheRunAskedFor` asserts the width it got is the width it asked for. The default execution does not set the property and skips that test. Independently, the `varka-engine` job became a matrix over `ubuntu-latest` and `ubuntu-24.04-arm`, with `fail-fast: false`. That covers the four-lane *machine*: real NEON codegen and a weaker memory model, which a capped width on x86 does not reproduce. The prose that described the engine as living outside the reactor is updated in the places it appears: the engine pom's header, the composite action, `dev/check_varka_arrow_version.py`'s docstring, `VarkaGeneratedClassLoader`'s javadoc, `docs/sql-varka.md` and the layout diagram in `IMPLEMENTATION_PLAN.md`. The `arrow.version` duplication and its guard script stay as they are: the reason for them was never the reactor, it was the absent parent pom, and that has not changed. ### Why are the changes needed? Two gaps the review left explicitly open. A contributor cloning the fork and running `./build/mvn install` got a resolution failure on `org.apache.spark.varka:varka-engine` before anything compiled, with nothing in the build to say why. And the kernels' narrow-species path - the one that produced finding 1, a wrong-answer bug - was covered only by `DateVectorOpsBitmapTest` walking species widths explicitly; the kernels themselves, which run at `SPECIES_PREFERRED`, had never executed at four lanes anywhere. ### Does this PR introduce _any_ user-facing change? No. No Spark behaviour changes. What changes is the build: `./build/mvn install` now also builds `varka-engine`, and `dev/lint-java` now also checks it. ### How was this patch tested? The reactor claim, directly: `./build/mvn -pl sql/catalyst -am validate` prints the order quoted above, and `./build/mvn -pl sql/varka/engine -am install` builds and installs the jar through the reactor. The standalone path CI depends on still works: `./build/mvn -f sql/varka/engine/pom.xml install`. `dev/lint-java` was run three times, and the sequence is the evidence: with the module added and no checkstyle configuration it failed with 294 sun_checks violations; with `configLocation` set to `dev/checkstyle.xml` it failed differently, unable to resolve the config at all, which is what proves the parent-directory resolution claim above; with `${maven.multiModuleProjectDirectory}` it reported the two missing newlines, and after fixing those, `Checkstyle checks passed.` The narrow-vector run was verified in both directions. That the flag does what is claimed, on this machine: ``` $ java --add-modules jdk.incubator.vector Lanes.java preferred int lanes = 16 $ java -XX:MaxVectorSize=16 --add-modules jdk.incubator.vector Lanes.java preferred int lanes = 4 ``` `./build/mvn -f sql/varka/engine/pom.xml test` runs 28 tests twice - `Skipped: 1` in the default execution, where the width guard does not apply, and 28 of 28 in the narrow one. And the guard is not decorative: with the cap temporarily set to 32 instead of 16, it fails with ``` expected: <4> but was: <8> the JVM did not honour the requested vector width, so this run duplicates the default one instead of exercising the narrow-species path ``` `dev/test-dependencies.sh` was run in full, since adding a reactor module is exactly what could disturb it, and it exits 0 with an empty diff against `dev/deps/spark-deps-hadoop-3-hive-2.3` - the manifest is unchanged, as it should be, `varka-engine` being a test-scope dependency of catalyst and test scope not being transitive. Its `versions:set` step also leaves `varka-engine` at `0.1.0-SNAPSHOT` (a different groupId and version from Spark's), so catalyst's dependency on it keeps resolving while that script runs with temporary versions. `catalyst/scalastyle` reports 0 errors and the edited workflow parses as YAML. This branch test-merges cleanly against #21. One thing this run did not establish: **the aarch64 job has never executed** - CI is its first run. If the kernels are wrong at four lanes on real hardware, that job is where it will show, which is the entire point of adding it; the capped-width run passing locally says the code path is right, not that the machine is. Note on `ISSUES.md`: finding 4's section and the two entries that cross-reference it are updated here. The preamble is deliberately untouched, because #21 rewrites it; a test merge of this branch against #21 is clean. It can be swept once both land. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5)

Address comments

1b3089a

HyukjinKwon merged commit cb23bd7 into MaxGekk:from_csv Oct 16, 2018

HyukjinKwon deleted the address-from_csv_2 branch October 16, 2018 12:40

Navigation Menu

Repository navigation

Address comments - #12

Address comments#12
HyukjinKwon merged 1 commit into
MaxGekk:from_csvfrom
HyukjinKwon:address-from_csv_2

HyukjinKwon commented Oct 16, 2018

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

HyukjinKwon commented Oct 16, 2018

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant