spark git commit: [SPARK-21694][MESOS] Support Mesos CNI network labels
Repository: spark Updated Branches: refs/heads/master 43cbfad99 -> ce0d3bb37 [SPARK-21694][MESOS] Support Mesos CNI network labels JIRA ticket: https://issues.apache.org/jira/browse/SPARK-21694 ## What changes were proposed in this pull request? Spark already supports launching containers attached to a given CNI network by specifying it via the config `spark.mesos.network.name`. This PR adds support to pass in network labels to CNI plugins via a new config option `spark.mesos.network.labels`. These network labels are key-value pairs that are set in the `NetworkInfo` of both the driver and executor tasks. More details in the related Mesos documentation: http://mesos.apache.org/documentation/latest/cni/#mesos-meta-data-to-cni-plugins ## How was this patch tested? Unit tests, for both driver and executor tasks. Manual integration test to submit a job with the `spark.mesos.network.labels` option, hit the mesos/state.json endpoint, and check that the labels are set in the driver and executor tasks. ArtRand skonto Author: Susan X. Huynh Closes #18910 from susanxhuynh/sh-mesos-cni-labels. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/ce0d3bb3 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/ce0d3bb3 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/ce0d3bb3 Branch: refs/heads/master Commit: ce0d3bb377766bdf4df7852272557ae846408877 Parents: 43cbfad Author: Susan X. Huynh Authored: Thu Aug 24 10:05:38 2017 +0100 Committer: Sean Owen Committed: Thu Aug 24 10:05:38 2017 +0100 -- docs/running-on-mesos.md | 14 ++ .../org/apache/spark/deploy/mesos/config.scala | 19 +-- .../MesosCoarseGrainedSchedulerBackend.scala | 2 +- .../mesos/MesosSchedulerBackendUtil.scala| 9 +++-- .../mesos/MesosClusterSchedulerSuite.scala | 9 +++-- ...MesosCoarseGrainedSchedulerBackendSuite.scala | 9 +++-- 6 files changed, 53 insertions(+), 9 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/ce0d3bb3/docs/running-on-mesos.md -- diff --git a/docs/running-on-mesos.md b/docs/running-on-mesos.md index ae38550..0e5a20c 100644 --- a/docs/running-on-mesos.md +++ b/docs/running-on-mesos.md @@ -538,6 +538,20 @@ See the [configuration page](configuration.html) for information on Spark config + spark.mesos.network.labels + (none) + +Pass network labels to CNI plugins. This is a comma-separated list +of key-value pairs, where each key-value pair has the format key:value. +Example: + +key1:val1,key2:val2 +See +http://mesos.apache.org/documentation/latest/cni/#mesos-meta-data-to-cni-plugins";>the Mesos CNI docs +for more details. + + + spark.mesos.fetcherCache.enable false http://git-wip-us.apache.org/repos/asf/spark/blob/ce0d3bb3/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala -- diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala index 6c8619e..a5015b9 100644 --- a/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala +++ b/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala @@ -56,7 +56,7 @@ package object config { .stringConf .createOptional - private [spark] val DRIVER_LABELS = + private[spark] val DRIVER_LABELS = ConfigBuilder("spark.mesos.driver.labels") .doc("Mesos labels to add to the driver. Labels are free-form key-value pairs. Key-value " + "pairs should be separated by a colon, and commas used to list more than one." + @@ -64,10 +64,25 @@ package object config { .stringConf .createOptional - private [spark] val DRIVER_FAILOVER_TIMEOUT = + private[spark] val DRIVER_FAILOVER_TIMEOUT = ConfigBuilder("spark.mesos.driver.failoverTimeout") .doc("Amount of time in seconds that the master will wait to hear from the driver, " + "during a temporary disconnection, before tearing down all the executors.") .doubleConf .createWithDefault(0.0) + + private[spark] val NETWORK_NAME = +ConfigBuilder("spark.mesos.network.name") + .doc("Attach containers to the given named network. If this job is launched " + +"in cluster mode, also launch the driver in the given named network.") + .stringConf + .createOptional + + private[spark] val NETWORK_LABELS = +ConfigBuilder("spark.mesos.network.labels") + .doc("Network labels to pass to CNI plugins. This is a comma-separated
spark git commit: [MINOR][SQL] The comment of Class ExchangeCoordinator exist a typing and context error
Repository: spark Updated Branches: refs/heads/master ce0d3bb37 -> 846bc61cf [MINOR][SQL] The comment of Class ExchangeCoordinator exist a typing and context error ## What changes were proposed in this pull request? The given example in the comment of Class ExchangeCoordinator is exist four post-shuffle partitions,but the current comment is âthreeâ. ## How was this patch tested? Author: lufei Closes #19028 from figo77/SPARK-21816. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/846bc61c Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/846bc61c Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/846bc61c Branch: refs/heads/master Commit: 846bc61cf5aa522dc755d50359ef3856ef2b17bf Parents: ce0d3bb Author: lufei Authored: Thu Aug 24 10:07:27 2017 +0100 Committer: Sean Owen Committed: Thu Aug 24 10:07:27 2017 +0100 -- .../apache/spark/sql/execution/exchange/ExchangeCoordinator.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/846bc61c/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ExchangeCoordinator.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ExchangeCoordinator.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ExchangeCoordinator.scala index deb2c24..9fc4ffb 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ExchangeCoordinator.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/ExchangeCoordinator.scala @@ -75,7 +75,7 @@ import org.apache.spark.sql.execution.{ShuffledRowRDD, SparkPlan} * For example, we have two stages with the following pre-shuffle partition size statistics: * stage 1: [100 MB, 20 MB, 100 MB, 10MB, 30 MB] * stage 2: [10 MB, 10 MB, 70 MB, 5 MB, 5 MB] - * assuming the target input size is 128 MB, we will have three post-shuffle partitions, + * assuming the target input size is 128 MB, we will have four post-shuffle partitions, * which are: * - post-shuffle partition 0: pre-shuffle partition 0 (size 110 MB) * - post-shuffle partition 1: pre-shuffle partition 1 (size 30 MB) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-21804][SQL] json_tuple returns null values within repeated columns except the first one
Repository: spark Updated Branches: refs/heads/master 846bc61cf -> 95713eb4f [SPARK-21804][SQL] json_tuple returns null values within repeated columns except the first one ## What changes were proposed in this pull request? When json_tuple in extracting values from JSON it returns null values within repeated columns except the first one as below: ``` scala scala> spark.sql("""SELECT json_tuple('{"a":1, "b":2}', 'a', 'b', 'a')""").show() +---+---++ | c0| c1| c2| +---+---++ | 1| 2|null| +---+---++ ``` I think this should be consistent with Hive's implementation: ``` hive> SELECT json_tuple('{"a": 1, "b": 2}', 'a', 'a'); ... 11 ``` In this PR, we located all the matched indices in `fieldNames` instead of returning the first matched index, i.e., indexOf. ## How was this patch tested? Added test in JsonExpressionsSuite. Author: Jen-Ming Chung Closes #19017 from jmchung/SPARK-21804. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/95713eb4 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/95713eb4 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/95713eb4 Branch: refs/heads/master Commit: 95713eb4f22de4e16617a605f74a1d6373ed270b Parents: 846bc61 Author: Jen-Ming Chung Authored: Thu Aug 24 19:24:00 2017 +0900 Committer: hyukjinkwon Committed: Thu Aug 24 19:24:00 2017 +0900 -- .../sql/catalyst/expressions/jsonExpressions.scala | 12 ++-- .../sql/catalyst/expressions/JsonExpressionsSuite.scala | 10 ++ 2 files changed, 20 insertions(+), 2 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/95713eb4/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala index c375737..ee5da1a 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala @@ -436,7 +436,8 @@ case class JsonTuple(children: Seq[Expression]) while (parser.nextToken() != JsonToken.END_OBJECT) { if (parser.getCurrentToken == JsonToken.FIELD_NAME) { // check to see if this field is desired in the output -val idx = fieldNames.indexOf(parser.getCurrentName) +val jsonField = parser.getCurrentName +var idx = fieldNames.indexOf(jsonField) if (idx >= 0) { // it is, copy the child tree to the correct location in the output row val output = new ByteArrayOutputStream() @@ -447,7 +448,14 @@ case class JsonTuple(children: Seq[Expression]) generator => copyCurrentStructure(generator, parser) } -row(idx) = UTF8String.fromBytes(output.toByteArray) +val jsonValue = UTF8String.fromBytes(output.toByteArray) + +// SPARK-21804: json_tuple returns null values within repeated columns +// except the first one; so that we need to check the remaining fields. +do { + row(idx) = jsonValue + idx = fieldNames.indexOf(jsonField, idx + 1) +} while (idx >= 0) } } } http://git-wip-us.apache.org/repos/asf/spark/blob/95713eb4/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala -- diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala index 1cd2b4f..9991bda 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala @@ -373,6 +373,16 @@ class JsonExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper { InternalRow(UTF8String.fromString("1"), null, UTF8String.fromString("2"))) } + test("SPARK-21804: json_tuple returns null values within repeated columns except the first one") { +checkJsonTuple( + JsonTuple(Literal("""{"f1": 1, "f2": 2}""") :: +NonFoldableLiteral("f1") :: +NonFoldableLiteral("cast(NULL AS STRING)") :: +NonFoldableLiteral("f1") :: +Nil), + InternalRow(UTF8String.fromString("1"), null, UTF8String.fromString("1"))) + } + val gmtId = Option(DateTimeUtils.TimeZoneGMT.getID) test("from_json") { -
spark git commit: [SPARK-19165][PYTHON][SQL] PySpark APIs using columns as arguments should validate input types for column
Repository: spark Updated Branches: refs/heads/master 95713eb4f -> dc5d34d8d [SPARK-19165][PYTHON][SQL] PySpark APIs using columns as arguments should validate input types for column ## What changes were proposed in this pull request? While preparing to take over https://github.com/apache/spark/pull/16537, I realised a (I think) better approach to make the exception handling in one point. This PR proposes to fix `_to_java_column` in `pyspark.sql.column`, which most of functions in `functions.py` and some other APIs use. This `_to_java_column` basically looks not working with other types than `pyspark.sql.column.Column` or string (`str` and `unicode`). If this is not `Column`, then it calls `_create_column_from_name` which calls `functions.col` within JVM: https://github.com/apache/spark/blob/42b9eda80e975d970c3e8da4047b318b83dd269f/sql/core/src/main/scala/org/apache/spark/sql/functions.scala#L76 And it looks we only have `String` one with `col`. So, these should work: ```python >>> from pyspark.sql.column import _to_java_column, Column >>> _to_java_column("a") JavaObject id=o28 >>> _to_java_column(u"a") JavaObject id=o29 >>> _to_java_column(spark.range(1).id) JavaObject id=o33 ``` whereas these do not: ```python >>> _to_java_column(1) ``` ``` ... py4j.protocol.Py4JError: An error occurred while calling z:org.apache.spark.sql.functions.col. Trace: py4j.Py4JException: Method col([class java.lang.Integer]) does not exist ... ``` ```python >>> _to_java_column([]) ``` ``` ... py4j.protocol.Py4JError: An error occurred while calling z:org.apache.spark.sql.functions.col. Trace: py4j.Py4JException: Method col([class java.util.ArrayList]) does not exist ... ``` ```python >>> class A(): pass >>> _to_java_column(A()) ``` ``` ... AttributeError: 'A' object has no attribute '_get_object_id' ``` Meaning most of functions using `_to_java_column` such as `udf` or `to_json` or some other APIs throw an exception as below: ```python >>> from pyspark.sql.functions import udf >>> udf(lambda x: x)(None) ``` ``` ... py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.sql.functions.col. : java.lang.NullPointerException ... ``` ```python >>> from pyspark.sql.functions import to_json >>> to_json(None) ``` ``` ... py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.sql.functions.col. : java.lang.NullPointerException ... ``` **After this PR**: ```python >>> from pyspark.sql.functions import udf >>> udf(lambda x: x)(None) ... ``` ``` TypeError: Invalid argument, not a string or column: None of type . For column literals, use 'lit', 'array', 'struct' or 'create_map' functions. ``` ```python >>> from pyspark.sql.functions import to_json >>> to_json(None) ``` ``` ... TypeError: Invalid argument, not a string or column: None of type . For column literals, use 'lit', 'array', 'struct' or 'create_map' functions. ``` ## How was this patch tested? Unit tests added in `python/pyspark/sql/tests.py` and manual tests. Author: hyukjinkwon Author: zero323 Closes #19027 from HyukjinKwon/SPARK-19165. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/dc5d34d8 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/dc5d34d8 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/dc5d34d8 Branch: refs/heads/master Commit: dc5d34d8dcd6526d1dfdac8606661561c7576a62 Parents: 95713eb Author: hyukjinkwon Authored: Thu Aug 24 20:29:03 2017 +0900 Committer: hyukjinkwon Committed: Thu Aug 24 20:29:03 2017 +0900 -- python/pyspark/sql/column.py | 8 +++- python/pyspark/sql/tests.py | 25 + 2 files changed, 32 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/dc5d34d8/python/pyspark/sql/column.py -- diff --git a/python/pyspark/sql/column.py b/python/pyspark/sql/column.py index b172f38..43b38a2 100644 --- a/python/pyspark/sql/column.py +++ b/python/pyspark/sql/column.py @@ -44,8 +44,14 @@ def _create_column_from_name(name): def _to_java_column(col): if isinstance(col, Column): jcol = col._jc -else: +elif isinstance(col, basestring): jcol = _create_column_from_name(col) +else: +raise TypeError( +"Invalid argument, not a string or column: " +"{0} of type {1}. " +"For column literals, use 'lit', 'array', 'struct' or 'create_map' " +"function.".format(col, type(col))) return jcol http://git-wip-us.apache.org/repos/asf/spark/blob/dc5d34d8/python/pyspark/sql/tests.py -- diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py index 45a3f
[2/2] spark git commit: [SPARK-21745][SQL] Refactor ColumnVector hierarchy to make ColumnVector read-only and to introduce WritableColumnVector.
[SPARK-21745][SQL] Refactor ColumnVector hierarchy to make ColumnVector read-only and to introduce WritableColumnVector. ## What changes were proposed in this pull request? This is a refactoring of `ColumnVector` hierarchy and related classes. 1. make `ColumnVector` read-only 2. introduce `WritableColumnVector` with write interface 3. remove `ReadOnlyColumnVector` ## How was this patch tested? Existing tests. Author: Takuya UESHIN Closes #18958 from ueshin/issues/SPARK-21745. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/9e33954d Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/9e33954d Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/9e33954d Branch: refs/heads/master Commit: 9e33954ddfe1148f69e523c89827feb76ba892c9 Parents: dc5d34d Author: Takuya UESHIN Authored: Thu Aug 24 21:13:44 2017 +0800 Committer: Wenchen Fan Committed: Thu Aug 24 21:13:44 2017 +0800 -- .../expressions/codegen/CodeGenerator.scala | 28 +- .../parquet/VectorizedColumnReader.java | 31 +- .../parquet/VectorizedParquetRecordReader.java | 23 +- .../parquet/VectorizedPlainValuesReader.java| 16 +- .../parquet/VectorizedRleValuesReader.java | 87 ++- .../parquet/VectorizedValuesReader.java | 16 +- .../execution/vectorized/AggregateHashMap.java | 10 +- .../execution/vectorized/ArrowColumnVector.java | 45 +- .../sql/execution/vectorized/ColumnVector.java | 632 + .../execution/vectorized/ColumnVectorUtils.java | 18 +- .../sql/execution/vectorized/ColumnarBatch.java | 106 +-- .../vectorized/OffHeapColumnVector.java | 34 +- .../vectorized/OnHeapColumnVector.java | 35 +- .../vectorized/ReadOnlyColumnVector.java| 251 --- .../vectorized/WritableColumnVector.java| 674 +++ .../aggregate/VectorizedHashMapGenerator.scala | 39 +- .../vectorized/ColumnarBatchBenchmark.scala | 23 +- .../vectorized/ColumnarBatchSuite.scala | 109 +-- 18 files changed, 1078 insertions(+), 1099 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/9e33954d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala index 807765c..3853863 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala @@ -464,14 +464,13 @@ class CodegenContext { /** * Returns the specialized code to set a given value in a column vector for a given `DataType`. */ - def setValue(batch: String, row: String, dataType: DataType, ordinal: Int, - value: String): String = { + def setValue(vector: String, rowId: String, dataType: DataType, value: String): String = { val jt = javaType(dataType) dataType match { case _ if isPrimitiveType(jt) => -s"$batch.column($ordinal).put${primitiveTypeName(jt)}($row, $value);" - case t: DecimalType => s"$batch.column($ordinal).putDecimal($row, $value, ${t.precision});" - case t: StringType => s"$batch.column($ordinal).putByteArray($row, $value.getBytes());" +s"$vector.put${primitiveTypeName(jt)}($rowId, $value);" + case t: DecimalType => s"$vector.putDecimal($rowId, $value, ${t.precision});" + case t: StringType => s"$vector.putByteArray($rowId, $value.getBytes());" case _ => throw new IllegalArgumentException(s"cannot generate code for unsupported type: $dataType") } @@ -482,37 +481,36 @@ class CodegenContext { * that could potentially be nullable. */ def updateColumn( - batch: String, - row: String, + vector: String, + rowId: String, dataType: DataType, - ordinal: Int, ev: ExprCode, nullable: Boolean): String = { if (nullable) { s""" if (!${ev.isNull}) { - ${setValue(batch, row, dataType, ordinal, ev.value)} + ${setValue(vector, rowId, dataType, ev.value)} } else { - $batch.column($ordinal).putNull($row); + $vector.putNull($rowId); } """ } else { - s"""${setValue(batch, row, dataType, ordinal, ev.value)};""" + s"""${setValue(vector, rowId, dataType, ev.value)};""" } } /** * Returns the specialized code to access a value from a column vector for a given `DataType`. */ - def getValue(batch: String, row: String, dataType: DataType, o
[1/2] spark git commit: [SPARK-21745][SQL] Refactor ColumnVector hierarchy to make ColumnVector read-only and to introduce WritableColumnVector.
Repository: spark Updated Branches: refs/heads/master dc5d34d8d -> 9e33954dd http://git-wip-us.apache.org/repos/asf/spark/blob/9e33954d/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java -- diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java new file mode 100644 index 000..b4f753c --- /dev/null +++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java @@ -0,0 +1,674 @@ +/* + * 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.vectorized; + +import java.math.BigDecimal; +import java.math.BigInteger; + +import com.google.common.annotations.VisibleForTesting; + +import org.apache.spark.sql.internal.SQLConf; +import org.apache.spark.sql.types.*; +import org.apache.spark.unsafe.types.UTF8String; + +/** + * This class adds write APIs to ColumnVector. + * It supports all the types and contains put APIs as well as their batched versions. + * The batched versions are preferable whenever possible. + * + * Capacity: The data stored is dense but the arrays are not fixed capacity. It is the + * responsibility of the caller to call reserve() to ensure there is enough room before adding + * elements. This means that the put() APIs do not check as in common cases (i.e. flat schemas), + * the lengths are known up front. + * + * A ColumnVector should be considered immutable once originally created. In other words, it is not + * valid to call put APIs after reads until reset() is called. + */ +public abstract class WritableColumnVector extends ColumnVector { + + /** + * Resets this column for writing. The currently stored values are no longer accessible. + */ + public void reset() { +if (isConstant) return; + +if (childColumns != null) { + for (ColumnVector c: childColumns) { +((WritableColumnVector) c).reset(); + } +} +numNulls = 0; +elementsAppended = 0; +if (anyNullsSet) { + putNotNulls(0, capacity); + anyNullsSet = false; +} + } + + public void reserve(int requiredCapacity) { +if (requiredCapacity > capacity) { + int newCapacity = (int) Math.min(MAX_CAPACITY, requiredCapacity * 2L); + if (requiredCapacity <= newCapacity) { +try { + reserveInternal(newCapacity); +} catch (OutOfMemoryError outOfMemoryError) { + throwUnsupportedException(requiredCapacity, outOfMemoryError); +} + } else { +throwUnsupportedException(requiredCapacity, null); + } +} + } + + private void throwUnsupportedException(int requiredCapacity, Throwable cause) { +String message = "Cannot reserve additional contiguous bytes in the vectorized reader " + +"(requested = " + requiredCapacity + " bytes). As a workaround, you can disable the " + +"vectorized reader by setting " + SQLConf.PARQUET_VECTORIZED_READER_ENABLED().key() + +" to false."; +throw new RuntimeException(message, cause); + } + + @Override + public int numNulls() { return numNulls; } + + @Override + public boolean anyNullsSet() { return anyNullsSet; } + + /** + * Ensures that there is enough storage to store capacity elements. That is, the put() APIs + * must work for all rowIds < capacity. + */ + protected abstract void reserveInternal(int capacity); + + /** + * Sets the value at rowId to null/not null. + */ + public abstract void putNotNull(int rowId); + public abstract void putNull(int rowId); + + /** + * Sets the values from [rowId, rowId + count) to null/not null. + */ + public abstract void putNulls(int rowId, int count); + public abstract void putNotNulls(int rowId, int count); + + /** + * Sets the value at rowId to `value`. + */ + public abstract void putBoolean(int rowId, boolean value); + + /** + * Sets values from [rowId, rowId + count) to value. + */ + public abstract void putBooleans(int rowId, int count, boolean value); + + /** + * Sets the value at rowId to `value`. + */
spark git commit: [SPARK-21759][SQL] In.checkInputDataTypes should not wrongly report unresolved plans for IN correlated subquery
Repository: spark Updated Branches: refs/heads/master 9e33954dd -> 183d4cb71 [SPARK-21759][SQL] In.checkInputDataTypes should not wrongly report unresolved plans for IN correlated subquery ## What changes were proposed in this pull request? With the check for structural integrity proposed in SPARK-21726, it is found that the optimization rule `PullupCorrelatedPredicates` can produce unresolved plans. For a correlated IN query looks like: SELECT t1.a FROM t1 WHERE t1.a IN (SELECT t2.c FROM t2 WHERE t1.b < t2.d); The query plan might look like: Project [a#0] +- Filter a#0 IN (list#4 [b#1]) : +- Project [c#2] : +- Filter (outer(b#1) < d#3) :+- LocalRelation , [c#2, d#3] +- LocalRelation , [a#0, b#1] After `PullupCorrelatedPredicates`, it produces query plan like: 'Project [a#0] +- 'Filter a#0 IN (list#4 [(b#1 < d#3)]) : +- Project [c#2, d#3] : +- LocalRelation , [c#2, d#3] +- LocalRelation , [a#0, b#1] Because the correlated predicate involves another attribute `d#3` in subquery, it has been pulled out and added into the `Project` on the top of the subquery. When `list` in `In` contains just one `ListQuery`, `In.checkInputDataTypes` checks if the size of `value` expressions matches the output size of subquery. In the above example, there is only `value` expression and the subquery output has two attributes `c#2, d#3`, so it fails the check and `In.resolved` returns `false`. We should not let `In.checkInputDataTypes` wrongly report unresolved plans to fail the structural integrity check. ## How was this patch tested? Added test. Author: Liang-Chi Hsieh Closes #18968 from viirya/SPARK-21759. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/183d4cb7 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/183d4cb7 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/183d4cb7 Branch: refs/heads/master Commit: 183d4cb71fbcbf484fc85d8621e1fe04cbbc8195 Parents: 9e33954 Author: Liang-Chi Hsieh Authored: Thu Aug 24 21:46:58 2017 +0800 Committer: Wenchen Fan Committed: Thu Aug 24 21:46:58 2017 +0800 -- .../spark/sql/catalyst/analysis/Analyzer.scala | 6 +- .../sql/catalyst/analysis/TypeCoercion.scala| 5 +- .../sql/catalyst/expressions/predicates.scala | 65 +--- .../sql/catalyst/expressions/subquery.scala | 13 +++- .../spark/sql/catalyst/optimizer/subquery.scala | 10 +-- .../PullupCorrelatedPredicatesSuite.scala | 52 .../negative-cases/subq-input-typecheck.sql.out | 6 +- 7 files changed, 106 insertions(+), 51 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/183d4cb7/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala index 70a3885..1e934d0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala @@ -1286,8 +1286,10 @@ class Analyzer( resolveSubQuery(s, plans)(ScalarSubquery(_, _, exprId)) case e @ Exists(sub, _, exprId) if !sub.resolved => resolveSubQuery(e, plans)(Exists(_, _, exprId)) -case In(value, Seq(l @ ListQuery(sub, _, exprId))) if value.resolved && !sub.resolved => - val expr = resolveSubQuery(l, plans)(ListQuery(_, _, exprId)) +case In(value, Seq(l @ ListQuery(sub, _, exprId, _))) if value.resolved && !l.resolved => + val expr = resolveSubQuery(l, plans)((plan, exprs) => { +ListQuery(plan, exprs, exprId, plan.output) + }) In(value, Seq(expr)) } } http://git-wip-us.apache.org/repos/asf/spark/blob/183d4cb7/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala index 06d8350..9ffe646 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala @@ -402,7 +402,7 @@ object TypeCoercion { // Handle type casting required between value expression and subquery output // in IN subquery. - case i @ In(a, Seq(ListQuery(sub, children, exprI
spark git commit: [SPARK-21826][SQL] outer broadcast hash join should not throw NPE
Repository: spark Updated Branches: refs/heads/master 183d4cb71 -> 2dd37d827 [SPARK-21826][SQL] outer broadcast hash join should not throw NPE ## What changes were proposed in this pull request? This is a bug introduced by https://github.com/apache/spark/pull/11274/files#diff-7adb688cbfa583b5711801f196a074bbL274 . Non-equal join condition should only be applied when the equal-join condition matches. ## How was this patch tested? regression test Author: Wenchen Fan Closes #19036 from cloud-fan/bug. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2dd37d82 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2dd37d82 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2dd37d82 Branch: refs/heads/master Commit: 2dd37d827f2e443dcb3eaf8a95437d179130d55c Parents: 183d4cb Author: Wenchen Fan Authored: Thu Aug 24 16:44:12 2017 +0200 Committer: Herman van Hovell Committed: Thu Aug 24 16:44:12 2017 +0200 -- .../execution/joins/BroadcastHashJoinExec.scala | 2 +- .../scala/org/apache/spark/sql/JoinSuite.scala | 20 2 files changed, 21 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/2dd37d82/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala index bfa1e9d..2f52a08 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala @@ -283,8 +283,8 @@ case class BroadcastHashJoinExec( s""" |boolean $conditionPassed = true; |${eval.trim} - |${ev.code} |if ($matched != null) { + | ${ev.code} | $conditionPassed = !${ev.isNull} && ${ev.value}; |} """.stripMargin http://git-wip-us.apache.org/repos/asf/spark/blob/2dd37d82/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala index 86fe09b..453052a 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql +import scala.collection.JavaConverters._ import scala.collection.mutable.ListBuffer import scala.language.existentials @@ -26,6 +27,7 @@ import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation import org.apache.spark.sql.execution.joins._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSQLContext +import org.apache.spark.sql.types.StructType class JoinSuite extends QueryTest with SharedSQLContext { import testImplicits._ @@ -767,4 +769,22 @@ class JoinSuite extends QueryTest with SharedSQLContext { } } } + + test("outer broadcast hash join should not throw NPE") { +withTempView("v1", "v2") { + withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") { +Seq(2 -> 2).toDF("x", "y").createTempView("v1") + +spark.createDataFrame( + Seq(Row(1, "a")).asJava, + new StructType().add("i", "int", nullable = false).add("j", "string", nullable = false) +).createTempView("v2") + +checkAnswer( + sql("select x, y, i, j from v1 left join v2 on x = i and y < length(j)"), + Row(2, 2, null, null) +) + } +} + } } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-21826][SQL] outer broadcast hash join should not throw NPE
Repository: spark Updated Branches: refs/heads/branch-2.2 236b2f4d5 -> a58536741 [SPARK-21826][SQL] outer broadcast hash join should not throw NPE This is a bug introduced by https://github.com/apache/spark/pull/11274/files#diff-7adb688cbfa583b5711801f196a074bbL274 . Non-equal join condition should only be applied when the equal-join condition matches. regression test Author: Wenchen Fan Closes #19036 from cloud-fan/bug. (cherry picked from commit 2dd37d827f2e443dcb3eaf8a95437d179130d55c) Signed-off-by: Herman van Hovell Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/a5853674 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/a5853674 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/a5853674 Branch: refs/heads/branch-2.2 Commit: a58536741f8365bb3fff01b588f3b42b219d11e5 Parents: 236b2f4 Author: Wenchen Fan Authored: Thu Aug 24 16:44:12 2017 +0200 Committer: Herman van Hovell Committed: Thu Aug 24 16:49:49 2017 +0200 -- .../execution/joins/BroadcastHashJoinExec.scala | 2 +- .../scala/org/apache/spark/sql/JoinSuite.scala | 20 2 files changed, 21 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/a5853674/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala index 0bc261d..69715ab 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala @@ -257,8 +257,8 @@ case class BroadcastHashJoinExec( s""" |boolean $conditionPassed = true; |${eval.trim} - |${ev.code} |if ($matched != null) { + | ${ev.code} | $conditionPassed = !${ev.isNull} && ${ev.value}; |} """.stripMargin http://git-wip-us.apache.org/repos/asf/spark/blob/a5853674/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala index 40bc1e9..95dc147 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql +import scala.collection.JavaConverters._ import scala.collection.mutable.ListBuffer import scala.language.existentials @@ -25,6 +26,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.execution.joins._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSQLContext +import org.apache.spark.sql.types.StructType import org.apache.spark.TestUtils.{assertNotSpilled, assertSpilled} class JoinSuite extends QueryTest with SharedSQLContext { @@ -739,4 +741,22 @@ class JoinSuite extends QueryTest with SharedSQLContext { } } } + + test("outer broadcast hash join should not throw NPE") { +withTempView("v1", "v2") { + withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") { +Seq(2 -> 2).toDF("x", "y").createTempView("v1") + +spark.createDataFrame( + Seq(Row(1, "a")).asJava, + new StructType().add("i", "int", nullable = false).add("j", "string", nullable = false) +).createTempView("v2") + +checkAnswer( + sql("select x, y, i, j from v1 left join v2 on x = i and y < length(j)"), + Row(2, 2, null, null) +) + } +} + } } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark-website git commit: Remove old sentence about 4-month minor releases
Repository: spark-website Updated Branches: refs/heads/asf-site 1a2e57670 -> cca972e7f Remove old sentence about 4-month minor releases Project: http://git-wip-us.apache.org/repos/asf/spark-website/repo Commit: http://git-wip-us.apache.org/repos/asf/spark-website/commit/cca972e7 Tree: http://git-wip-us.apache.org/repos/asf/spark-website/tree/cca972e7 Diff: http://git-wip-us.apache.org/repos/asf/spark-website/diff/cca972e7 Branch: refs/heads/asf-site Commit: cca972e7fbf866df211076dcd5afb5f93af51994 Parents: 1a2e576 Author: Sean Owen Authored: Thu Aug 24 17:58:57 2017 +0100 Committer: Sean Owen Committed: Thu Aug 24 17:58:57 2017 +0100 -- site/versioning-policy.html | 1 - versioning-policy.md| 1 - 2 files changed, 2 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark-website/blob/cca972e7/site/versioning-policy.html -- diff --git a/site/versioning-policy.html b/site/versioning-policy.html index e0afa7c..82e39d9 100644 --- a/site/versioning-policy.html +++ b/site/versioning-policy.html @@ -209,7 +209,6 @@ These small differences account for Spark’s nature as a multi-module proje Major version numbers will remain stable over long periods of time. For instance, 1.X.Y may last 1 year or more. FEATURE: Feature releases will typically contain new features, improvements, and bug fixes. -The target frequency for feature releases is every 4 months. Each feature release will have a merge window where new patches can be merged, a QA window when only fixes can be merged, then a final period where voting occurs on release candidates. These windows will be announced immediately after the previous feature release to give people plenty http://git-wip-us.apache.org/repos/asf/spark-website/blob/cca972e7/versioning-policy.md -- diff --git a/versioning-policy.md b/versioning-policy.md index 509b595..a6c47d2 100644 --- a/versioning-policy.md +++ b/versioning-policy.md @@ -21,7 +21,6 @@ Each Spark release will be versioned: `[MAJOR].[FEATURE].[MAINTENANCE]` Major version numbers will remain stable over long periods of time. For instance, 1.X.Y may last 1 year or more. - **FEATURE**: Feature releases will typically contain new features, improvements, and bug fixes. -The target frequency for feature releases is every 4 months. Each feature release will have a merge window where new patches can be merged, a QA window when only fixes can be merged, then a final period where voting occurs on release candidates. These windows will be announced immediately after the previous feature release to give people plenty - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-21681][ML] fix bug of MLOR do not work correctly when featureStd contains zero (backport PR for 2.2)
Repository: spark Updated Branches: refs/heads/branch-2.2 a58536741 -> 2b4bd7910 [SPARK-21681][ML] fix bug of MLOR do not work correctly when featureStd contains zero (backport PR for 2.2) ## What changes were proposed in this pull request? This is backport PR of https://github.com/apache/spark/pull/18896 fix bug of MLOR do not work correctly when featureStd contains zero We can reproduce the bug through such dataset (features including zero variance), will generate wrong result (all coefficients becomes 0) ``` val multinomialDatasetWithZeroVar = { val nPoints = 100 val coefficients = Array( -0.57997, 0.912083, -0.371077, -0.16624, -0.84355, -0.048509) val xMean = Array(5.843, 3.0) val xVariance = Array(0.6856, 0.0) // including zero variance val testData = generateMultinomialLogisticInput( coefficients, xMean, xVariance, addIntercept = true, nPoints, seed) val df = sc.parallelize(testData, 4).toDF().withColumn("weight", lit(1.0)) df.cache() df } ``` ## How was this patch tested? testcase added. Author: WeichenXu Closes #19026 from WeichenXu123/fix_mlor_zero_var_bug_2_2. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2b4bd791 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2b4bd791 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2b4bd791 Branch: refs/heads/branch-2.2 Commit: 2b4bd7910fecc8b7b41c7d4388d2a8204c1901e8 Parents: a585367 Author: Weichen Xu Authored: Thu Aug 24 10:18:56 2017 -0700 Committer: Joseph K. Bradley Committed: Thu Aug 24 10:18:56 2017 -0700 -- .../ml/classification/LogisticRegression.scala | 12 ++-- .../LogisticRegressionSuite.scala | 75 2 files changed, 82 insertions(+), 5 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/2b4bd791/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala -- diff --git a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala index 567af04..1de2373 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala @@ -1727,11 +1727,13 @@ private class LogisticAggregator( val margins = new Array[Double](numClasses) features.foreachActive { (index, value) => - val stdValue = value / localFeaturesStd(index) - var j = 0 - while (j < numClasses) { -margins(j) += localCoefficients(index * numClasses + j) * stdValue -j += 1 + if (localFeaturesStd(index) != 0.0 && value != 0.0) { +val stdValue = value / localFeaturesStd(index) +var j = 0 +while (j < numClasses) { + margins(j) += localCoefficients(index * numClasses + j) * stdValue + j += 1 +} } } var i = 0 http://git-wip-us.apache.org/repos/asf/spark/blob/2b4bd791/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala -- diff --git a/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala index 1ffd8dc..8461d64 100644 --- a/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala +++ b/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala @@ -45,6 +45,7 @@ class LogisticRegressionSuite @transient var smallMultinomialDataset: Dataset[_] = _ @transient var binaryDataset: Dataset[_] = _ @transient var multinomialDataset: Dataset[_] = _ + @transient var multinomialDatasetWithZeroVar: Dataset[_] = _ private val eps: Double = 1e-5 override def beforeAll(): Unit = { @@ -98,6 +99,23 @@ class LogisticRegressionSuite df.cache() df } + +multinomialDatasetWithZeroVar = { + val nPoints = 100 + val coefficients = Array( +-0.57997, 0.912083, -0.371077, +-0.16624, -0.84355, -0.048509) + + val xMean = Array(5.843, 3.0) + val xVariance = Array(0.6856, 0.0) + + val testData = generateMultinomialLogisticInput( +coefficients, xMean, xVariance, addIntercept = true, nPoints, seed) + + val df = sc.parallelize(testData, 4).toDF().withColumn("weight", lit(1.0)) + df.cache() + df +} } /** @@ -111,6 +129,11 @@ class LogisticRegressionSuite multinomialDataset.rdd.map { case Row(label: Double, features: Vector, weight: Double) => l
spark git commit: [SPARK-21788][SS] Handle more exceptions when stopping a streaming query
Repository: spark Updated Branches: refs/heads/master 2dd37d827 -> d3abb3699 [SPARK-21788][SS] Handle more exceptions when stopping a streaming query ## What changes were proposed in this pull request? Add more cases we should view as a normal query stop rather than a failure. ## How was this patch tested? The new unit tests. Author: Shixiong Zhu Closes #18997 from zsxwing/SPARK-21788. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/d3abb369 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/d3abb369 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/d3abb369 Branch: refs/heads/master Commit: d3abb36990d928a8445a8c69ddebeabdfeb1484d Parents: 2dd37d8 Author: Shixiong Zhu Authored: Thu Aug 24 10:23:59 2017 -0700 Committer: Tathagata Das Committed: Thu Aug 24 10:23:59 2017 -0700 -- .../execution/streaming/StreamExecution.scala | 34 ++- .../spark/sql/streaming/StreamSuite.scala | 60 +++- 2 files changed, 89 insertions(+), 5 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/d3abb369/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala index 432b2d4..c224f2f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala @@ -17,9 +17,10 @@ package org.apache.spark.sql.execution.streaming -import java.io.{InterruptedIOException, IOException} +import java.io.{InterruptedIOException, IOException, UncheckedIOException} +import java.nio.channels.ClosedByInterruptException import java.util.UUID -import java.util.concurrent.{CountDownLatch, TimeUnit} +import java.util.concurrent.{CountDownLatch, ExecutionException, TimeUnit} import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock @@ -27,6 +28,7 @@ import scala.collection.mutable.{Map => MutableMap} import scala.collection.mutable.ArrayBuffer import scala.util.control.NonFatal +import com.google.common.util.concurrent.UncheckedExecutionException import org.apache.hadoop.fs.Path import org.apache.spark.internal.Logging @@ -335,7 +337,7 @@ class StreamExecution( // `stop()` is already called. Let `finally` finish the cleanup. } } catch { - case _: InterruptedException | _: InterruptedIOException if state.get == TERMINATED => + case e if isInterruptedByStop(e) => // interrupted by stop() updateStatusMessage("Stopped") case e: IOException if e.getMessage != null @@ -407,6 +409,32 @@ class StreamExecution( } } + private def isInterruptedByStop(e: Throwable): Boolean = { +if (state.get == TERMINATED) { + e match { +// InterruptedIOException - thrown when an I/O operation is interrupted +// ClosedByInterruptException - thrown when an I/O operation upon a channel is interrupted +case _: InterruptedException | _: InterruptedIOException | _: ClosedByInterruptException => + true +// The cause of the following exceptions may be one of the above exceptions: +// +// UncheckedIOException - thrown by codes that cannot throw a checked IOException, such as +//BiFunction.apply +// ExecutionException - thrown by codes running in a thread pool and these codes throw an +// exception +// UncheckedExecutionException - thrown by codes that cannot throw a checked +// ExecutionException, such as BiFunction.apply +case e2 @ (_: UncheckedIOException | _: ExecutionException | _: UncheckedExecutionException) + if e2.getCause != null => + isInterruptedByStop(e2.getCause) +case _ => + false + } +} else { + false +} + } + /** * Populate the start offsets to start the execution at the current offsets stored in the sink * (i.e. avoid reprocessing data that we have already processed). This function must be called http://git-wip-us.apache.org/repos/asf/spark/blob/d3abb369/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala index 012cccf..d0b2041 100644 --- a/sql/core/sr
spark git commit: [SPARK-21826][SQL][2.1][2.0] outer broadcast hash join should not throw NPE
Repository: spark Updated Branches: refs/heads/branch-2.1 3d3be4dca -> 576975356 [SPARK-21826][SQL][2.1][2.0] outer broadcast hash join should not throw NPE backport https://github.com/apache/spark/pull/19036 to branch 2.1 and 2.0 Author: Wenchen Fan Closes #19040 from cloud-fan/bug. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/57697535 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/57697535 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/57697535 Branch: refs/heads/branch-2.1 Commit: 576975356357ead203e452d0d794794349ba4578 Parents: 3d3be4d Author: Wenchen Fan Authored: Thu Aug 24 10:36:37 2017 -0700 Committer: gatorsmile Committed: Thu Aug 24 10:36:37 2017 -0700 -- .../execution/joins/BroadcastHashJoinExec.scala | 2 +- .../scala/org/apache/spark/sql/JoinSuite.scala | 20 2 files changed, 21 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/57697535/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala index 0bc261d..69715ab 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala @@ -257,8 +257,8 @@ case class BroadcastHashJoinExec( s""" |boolean $conditionPassed = true; |${eval.trim} - |${ev.code} |if ($matched != null) { + | ${ev.code} | $conditionPassed = !${ev.isNull} && ${ev.value}; |} """.stripMargin http://git-wip-us.apache.org/repos/asf/spark/blob/57697535/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala index 913b2ae..3416532 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql +import scala.collection.JavaConverters._ import scala.language.existentials import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation @@ -24,6 +25,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.execution.joins._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSQLContext +import org.apache.spark.sql.types.StructType class JoinSuite extends QueryTest with SharedSQLContext { @@ -604,4 +606,22 @@ class JoinSuite extends QueryTest with SharedSQLContext { cartesianQueries.foreach(checkCartesianDetection) } + + test("outer broadcast hash join should not throw NPE") { +withTempView("v1", "v2") { + withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") { +Seq(2 -> 2).toDF("x", "y").createTempView("v1") + +spark.createDataFrame( + Seq(Row(1, "a")).asJava, + new StructType().add("i", "int", nullable = false).add("j", "string", nullable = false) +).createTempView("v2") + +checkAnswer( + sql("select x, y, i, j from v1 left join v2 on x = i and y < length(j)"), + Row(2, 2, null, null) +) + } +} + } } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-21826][SQL][2.1][2.0] outer broadcast hash join should not throw NPE
Repository: spark Updated Branches: refs/heads/branch-2.0 9f670ce5d -> bf1f30d7d [SPARK-21826][SQL][2.1][2.0] outer broadcast hash join should not throw NPE backport https://github.com/apache/spark/pull/19036 to branch 2.1 and 2.0 Author: Wenchen Fan Closes #19040 from cloud-fan/bug. (cherry picked from commit 576975356357ead203e452d0d794794349ba4578) Signed-off-by: gatorsmile Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/bf1f30d7 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/bf1f30d7 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/bf1f30d7 Branch: refs/heads/branch-2.0 Commit: bf1f30d7dcce81df6826d2630decfc8a93f5fa01 Parents: 9f670ce Author: Wenchen Fan Authored: Thu Aug 24 10:36:37 2017 -0700 Committer: gatorsmile Committed: Thu Aug 24 10:39:40 2017 -0700 -- .../execution/joins/BroadcastHashJoinExec.scala | 2 +- .../scala/org/apache/spark/sql/JoinSuite.scala | 20 2 files changed, 21 insertions(+), 1 deletion(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/bf1f30d7/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala index 0f24baa..20b531b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala @@ -257,8 +257,8 @@ case class BroadcastHashJoinExec( s""" |boolean $conditionPassed = true; |${eval.trim} - |${ev.code} |if ($matched != null) { + | ${ev.code} | $conditionPassed = !${ev.isNull} && ${ev.value}; |} """.stripMargin http://git-wip-us.apache.org/repos/asf/spark/blob/bf1f30d7/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala -- diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala index 44889d9..5e7c9f7 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala @@ -17,6 +17,7 @@ package org.apache.spark.sql +import scala.collection.JavaConverters._ import scala.language.existentials import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation @@ -24,6 +25,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.execution.joins._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSQLContext +import org.apache.spark.sql.types.StructType class JoinSuite extends QueryTest with SharedSQLContext { @@ -573,4 +575,22 @@ class JoinSuite extends QueryTest with SharedSQLContext { Row(3, 1) :: Row(3, 2) :: Nil) } + + test("outer broadcast hash join should not throw NPE") { +withTempView("v1", "v2") { + withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") { +Seq(2 -> 2).toDF("x", "y").createTempView("v1") + +spark.createDataFrame( + Seq(Row(1, "a")).asJava, + new StructType().add("i", "int", nullable = false).add("j", "string", nullable = false) +).createTempView("v2") + +checkAnswer( + sql("select x, y, i, j from v1 left join v2 on x = i and y < length(j)"), + Row(2, 2, null, null) +) + } +} + } } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-21701][CORE] Enable RPC client to use ` SO_RCVBUF` and ` SO_SNDBUF` in SparkConf.
Repository: spark Updated Branches: refs/heads/master d3abb3699 -> 763b83ee8 [SPARK-21701][CORE] Enable RPC client to use ` SO_RCVBUF` and ` SO_SNDBUF` in SparkConf. ## What changes were proposed in this pull request? TCP parameters like SO_RCVBUF and SO_SNDBUF can be set in SparkConf, and `org.apache.spark.network.server.TransportServe`r can use those parameters to build server by leveraging netty. But for TransportClientFactory, there is no such way to set those parameters from SparkConf. This could be inconsistent in server and client side when people set parameters in SparkConf. So this PR make RPC client to be enable to use those TCP parameters as well. ## How was this patch tested? Existing tests. Author: xu.zhang Closes #18964 from neoremind/add_client_param. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/763b83ee Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/763b83ee Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/763b83ee Branch: refs/heads/master Commit: 763b83ee84cbb6f263218c471dd9198dd6bee411 Parents: d3abb36 Author: xu.zhang Authored: Thu Aug 24 14:27:52 2017 -0700 Committer: Shixiong Zhu Committed: Thu Aug 24 14:27:52 2017 -0700 -- .../apache/spark/network/client/TransportClientFactory.java | 8 1 file changed, 8 insertions(+) -- http://git-wip-us.apache.org/repos/asf/spark/blob/763b83ee/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java -- diff --git a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java index b50e043..8add4e1 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java +++ b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java @@ -210,6 +210,14 @@ public class TransportClientFactory implements Closeable { .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, conf.connectionTimeoutMs()) .option(ChannelOption.ALLOCATOR, pooledAllocator); +if (conf.receiveBuf() > 0) { + bootstrap.option(ChannelOption.SO_RCVBUF, conf.receiveBuf()); +} + +if (conf.sendBuf() > 0) { + bootstrap.option(ChannelOption.SO_SNDBUF, conf.sendBuf()); +} + final AtomicReference clientRef = new AtomicReference<>(); final AtomicReference channelRef = new AtomicReference<>(); - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-21830][SQL] Bump ANTLR version and fix a few issues.
Repository: spark Updated Branches: refs/heads/master 763b83ee8 -> 05af2de0f [SPARK-21830][SQL] Bump ANTLR version and fix a few issues. ## What changes were proposed in this pull request? This PR bumps the ANTLR version to 4.7, and fixes a number of small parser related issues uncovered by the bump. The main reason for upgrading is that in some cases the current version of ANTLR (4.5) can exhibit exponential slowdowns if it needs to parse boolean predicates. For example the following query will take forever to parse: ```sql SELECT * FROM RANGE(1000) WHERE TRUE AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' AND NOT upper(DESCRIPTION) LIKE '%FOO%' ``` This is caused by a know bug in ANTLR (https://github.com/antlr/antlr4/issues/994), which was fixed in version 4.6. ## How was this patch tested? Existing tests. Author: Herman van Hovell Closes #19042 from hvanhovell/SPARK-21830. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/05af2de0 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/05af2de0 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/05af2de0 Branch: refs/heads/master Commit: 05af2de0fdce625041b99908adc320c576bac116 Parents: 763b83e Author: Herman van Hovell Authored: Thu Aug 24 16:33:55 2017 -0700 Committer: gatorsmile Committed: Thu Aug 24 16:33:55 2017 -0700 -- dev/deps/spark-deps-hadoop-2.6| 2 +- dev/deps/spark-deps-hadoop-2.7| 2 +- pom.xml | 2 +- project/SparkBuild.scala | 1 + .../org/apache/spark/sql/catalyst/parser/SqlBase.g4 | 6 +- .../apache/spark/sql/catalyst/parser/AstBuilder.scala | 4 .../spark/sql/catalyst/parser/ParseDriver.scala | 2 +- .../sql/catalyst/parser/TableSchemaParserSuite.scala | 14 -- .../resources/sql-tests/results/show-tables.sql.out | 4 ++-- .../org/apache/spark/sql/jdbc/JDBCWriteSuite.scala| 2 +- 10 files changed, 25 insertions(+), 14 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/05af2de0/dev/deps/spark-deps-hadoop-2.6 -- diff --git a/dev/deps/spark-deps-hadoop-2.6 b/dev/deps/spark-deps-hadoop-2.6 index 01af2c7..de17507 100644 --- a/dev/deps/spark-deps-hadoop-2.6 +++ b/dev/deps/spark-deps-hadoop-2.6 @@ -5,7 +5,7 @@ activation-1.1.1.jar aircompressor-0.3.jar antlr-2.7.7.jar antlr-runtime-3.4.jar -antlr4-runtime-4.5.3.jar +antlr4-runtime-4.7.jar aopalliance-1.0.jar aopalliance-repackaged-2.4.0-b34.jar apache-log4j-extras-1.2.17.jar http://git-wip-us.apache.org/repos/asf/spark/blob/05af2de0/dev/deps/spark-deps-hadoop-2.7 -- diff --git a/dev/deps/spark-deps-hadoop-2.7 b/dev/deps/spark-deps-hadoop-2.7 index 69f3a4b..da826a7 100644 --- a/dev/deps/spark-deps-hadoop-2.7 +++ b/dev/deps/spark-deps-hadoop-2.7 @@ -5,7 +5,7 @@ activation-1.1.1.jar aircompressor-0.3.jar antlr-2.7.7.jar antlr-runtime-3.4.jar -antlr4-runtime-4.5.3.jar +antlr4-runtime-4.7.jar aopalliance-1.0.jar aopalliance-repackaged-2.4.0-b34.jar apache-log4j-extras-1.2.17.jar http://git-wip-us.apache.org/repos/asf/spark/blob/05af2de0/pom.xml -- diff --git a/pom.xml b/pom.xml index c0df3ef..8b4a6c5 100644 --- a/pom.xml +++ b/pom.xml @@ -178,7 +178,7 @@ 3.5.2 1.3.9 0.9.3 -4.5.3 +4.7 1.1 2.52.0 2.6 http://git-wip-us.apache.org/repos/asf/spark/blob/05af2de0/project/SparkBuild.scala -- diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala index 7565e14..18059ad 100644 --- a/project/SparkBuild.scala +++ b/project/SparkBuild.scala @@ -474,6 +474,7 @@ object OldDeps { object Catalyst { lazy val settings = antlr4Settings ++ Seq( +antlr4Version in Antlr4 := "4.7", antlr4PackageName in Antlr4 := Some("org.apache.spark.sql.catalyst.parser"), antlr4GenListener in Antlr4 := true, antlr4GenVisitor in Antlr4 := true http://g
spark git commit: [SPARK-21108][ML] convert LinearSVC to aggregator framework
Repository: spark Updated Branches: refs/heads/master 05af2de0f -> f3676d639 [SPARK-21108][ML] convert LinearSVC to aggregator framework ## What changes were proposed in this pull request? convert LinearSVC to new aggregator framework ## How was this patch tested? existing unit test. Author: Yuhao Yang Closes #18315 from hhbyyh/svcAggregator. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/f3676d63 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/f3676d63 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/f3676d63 Branch: refs/heads/master Commit: f3676d63913e0706e071b71e1742b8d57b102fba Parents: 05af2de Author: Yuhao Yang Authored: Fri Aug 25 10:22:27 2017 +0800 Committer: Yanbo Liang Committed: Fri Aug 25 10:22:27 2017 +0800 -- .../spark/ml/classification/LinearSVC.scala | 204 ++- .../ml/optim/aggregator/HingeAggregator.scala | 105 ++ .../ml/classification/LinearSVCSuite.scala | 7 +- .../optim/aggregator/HingeAggregatorSuite.scala | 163 +++ .../aggregator/LogisticAggregatorSuite.scala| 2 - 5 files changed, 286 insertions(+), 195 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/f3676d63/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala -- diff --git a/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala b/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala index 8d556de..3b0666c 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala @@ -25,11 +25,11 @@ import org.apache.hadoop.fs.Path import org.apache.spark.SparkException import org.apache.spark.annotation.{Experimental, Since} -import org.apache.spark.broadcast.Broadcast import org.apache.spark.internal.Logging import org.apache.spark.ml.feature.Instance import org.apache.spark.ml.linalg._ -import org.apache.spark.ml.linalg.BLAS._ +import org.apache.spark.ml.optim.aggregator.HingeAggregator +import org.apache.spark.ml.optim.loss.{L2Regularization, RDDLossFunction} import org.apache.spark.ml.param._ import org.apache.spark.ml.param.shared._ import org.apache.spark.ml.util._ @@ -214,10 +214,20 @@ class LinearSVC @Since("2.2.0") ( } val featuresStd = summarizer.variance.toArray.map(math.sqrt) + val getFeaturesStd = (j: Int) => featuresStd(j) val regParamL2 = $(regParam) val bcFeaturesStd = instances.context.broadcast(featuresStd) - val costFun = new LinearSVCCostFun(instances, $(fitIntercept), -$(standardization), bcFeaturesStd, regParamL2, $(aggregationDepth)) + val regularization = if (regParamL2 != 0.0) { +val shouldApply = (idx: Int) => idx >= 0 && idx < numFeatures +Some(new L2Regularization(regParamL2, shouldApply, + if ($(standardization)) None else Some(getFeaturesStd))) + } else { +None + } + + val getAggregatorFunc = new HingeAggregator(bcFeaturesStd, $(fitIntercept))(_) + val costFun = new RDDLossFunction(instances, getAggregatorFunc, regularization, +$(aggregationDepth)) def regParamL1Fun = (index: Int) => 0D val optimizer = new BreezeOWLQN[Int, BDV[Double]]($(maxIter), 10, regParamL1Fun, $(tol)) @@ -372,189 +382,3 @@ object LinearSVCModel extends MLReadable[LinearSVCModel] { } } } - -/** - * LinearSVCCostFun implements Breeze's DiffFunction[T] for hinge loss function - */ -private class LinearSVCCostFun( -instances: RDD[Instance], -fitIntercept: Boolean, -standardization: Boolean, -bcFeaturesStd: Broadcast[Array[Double]], -regParamL2: Double, -aggregationDepth: Int) extends DiffFunction[BDV[Double]] { - - override def calculate(coefficients: BDV[Double]): (Double, BDV[Double]) = { -val coeffs = Vectors.fromBreeze(coefficients) -val bcCoeffs = instances.context.broadcast(coeffs) -val featuresStd = bcFeaturesStd.value -val numFeatures = featuresStd.length - -val svmAggregator = { - val seqOp = (c: LinearSVCAggregator, instance: Instance) => c.add(instance) - val combOp = (c1: LinearSVCAggregator, c2: LinearSVCAggregator) => c1.merge(c2) - - instances.treeAggregate( -new LinearSVCAggregator(bcCoeffs, bcFeaturesStd, fitIntercept) - )(seqOp, combOp, aggregationDepth) -} - -val totalGradientArray = svmAggregator.gradient.toArray -// regVal is the sum of coefficients squares excluding intercept for L2 regularization. -val regVal = if (regParamL2 == 0.0) { - 0.0 -} else { - var sum = 0.0 - coeffs.foreachActive { case (index, value) => -// We
spark git commit: [SPARK-21255][SQL][WIP] Fixed NPE when creating encoder for enum
Repository: spark Updated Branches: refs/heads/master f3676d639 -> 7d16776d2 [SPARK-21255][SQL][WIP] Fixed NPE when creating encoder for enum ## What changes were proposed in this pull request? Fixed NPE when creating encoder for enum. When you try to create an encoder for Enum type (or bean with enum property) via Encoders.bean(...), it fails with NullPointerException at TypeToken:495. I did a little research and it turns out, that in JavaTypeInference following code ``` def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = { val beanInfo = Introspector.getBeanInfo(beanClass) beanInfo.getPropertyDescriptors.filterNot(_.getName == "class") .filter(_.getReadMethod != null) } ``` filters out properties named "class", because we wouldn't want to serialize that. But enum types have another property of type Class named "declaringClass", which we are trying to inspect recursively. Eventually we try to inspect ClassLoader class, which has property "defaultAssertionStatus" with no read method, which leads to NPE at TypeToken:495. I added property name "declaringClass" to filtering to resolve this. ## How was this patch tested? Unit test in JavaDatasetSuite which creates an encoder for enum Author: mike Author: Mikhail Sveshnikov Closes #18488 from mike0sv/enum-support. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/7d16776d Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/7d16776d Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/7d16776d Branch: refs/heads/master Commit: 7d16776d28da5bcf656f0d8556b15ed3a5edca44 Parents: f3676d6 Author: mike Authored: Fri Aug 25 07:22:34 2017 +0100 Committer: Sean Owen Committed: Fri Aug 25 07:22:34 2017 +0100 -- .../spark/sql/catalyst/JavaTypeInference.scala | 40 ++ .../catalyst/encoders/ExpressionEncoder.scala | 14 +++- .../catalyst/expressions/objects/objects.scala | 4 +- .../org/apache/spark/sql/JavaDatasetSuite.java | 77 4 files changed, 131 insertions(+), 4 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/7d16776d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala index 21363d3..33f6ce0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala @@ -32,6 +32,7 @@ import org.apache.spark.sql.catalyst.expressions.objects._ import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, DateTimeUtils, GenericArrayData} import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.UTF8String +import org.apache.spark.util.Utils /** * Type-inference utilities for POJOs and Java collections. @@ -118,6 +119,10 @@ object JavaTypeInference { val (valueDataType, nullable) = inferDataType(valueType, seenTypeSet) (MapType(keyDataType, valueDataType, nullable), true) + case other if other.isEnum => +(StructType(Seq(StructField(typeToken.getRawType.getSimpleName, + StringType, nullable = false))), true) + case other => if (seenTypeSet.contains(other)) { throw new UnsupportedOperationException( @@ -140,6 +145,7 @@ object JavaTypeInference { def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = { val beanInfo = Introspector.getBeanInfo(beanClass) beanInfo.getPropertyDescriptors.filterNot(_.getName == "class") + .filterNot(_.getName == "declaringClass") .filter(_.getReadMethod != null) } @@ -303,6 +309,11 @@ object JavaTypeInference { keyData :: valueData :: Nil, returnNullable = false) + case other if other.isEnum => +StaticInvoke(JavaTypeInference.getClass, ObjectType(other), "deserializeEnumName", + expressions.Literal.create(other.getEnumConstants.apply(0), ObjectType(other)) +:: getPath :: Nil) + case other => val properties = getJavaBeanReadableAndWritableProperties(other) val setters = properties.map { p => @@ -345,6 +356,30 @@ object JavaTypeInference { } } + /** Returns a mapping from enum value to int for given enum type */ + def enumSerializer[T <: Enum[T]](enum: Class[T]): T => UTF8String = { +assert(enum.isEnum) +inputObject: T => + UTF8String.fromString(inputObject.name()) + } + + /** Returns value index for given enum type and value */ + def serializeEnumName[T <: Enum[T]](