spark git commit: [MINOR] Add 1.3, 1.3.1 to master branch EC2 scripts
Repository: spark
Updated Branches:
refs/heads/master ba4f8ca0d -> 1a7b9ce80
[MINOR] Add 1.3, 1.3.1 to master branch EC2 scripts
cc pwendell
P.S: I can't believe this was outdated all along ?
Author: Shivaram Venkataraman
Closes #6215 from shivaram/update-ec2-map and squashes the following commits:
ae3937a [Shivaram Venkataraman] Add 1.3, 1.3.1 to master branch EC2 scripts
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/1a7b9ce8
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/1a7b9ce8
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/1a7b9ce8
Branch: refs/heads/master
Commit: 1a7b9ce80bb5649796dda48d6a6d662a2809d0ef
Parents: ba4f8ca
Author: Shivaram Venkataraman
Authored: Sun May 17 00:12:20 2015 -0700
Committer: Shivaram Venkataraman
Committed: Sun May 17 00:12:20 2015 -0700
--
ec2/spark_ec2.py | 6 +-
1 file changed, 5 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/1a7b9ce8/ec2/spark_ec2.py
--
diff --git a/ec2/spark_ec2.py b/ec2/spark_ec2.py
index ab4a96f..be92d5f 100755
--- a/ec2/spark_ec2.py
+++ b/ec2/spark_ec2.py
@@ -48,7 +48,7 @@ else:
from urllib.request import urlopen, Request
from urllib.error import HTTPError
-SPARK_EC2_VERSION = "1.2.1"
+SPARK_EC2_VERSION = "1.3.1"
SPARK_EC2_DIR = os.path.dirname(os.path.realpath(__file__))
VALID_SPARK_VERSIONS = set([
@@ -65,6 +65,8 @@ VALID_SPARK_VERSIONS = set([
"1.1.1",
"1.2.0",
"1.2.1",
+"1.3.0",
+"1.3.1",
])
SPARK_TACHYON_MAP = {
@@ -75,6 +77,8 @@ SPARK_TACHYON_MAP = {
"1.1.1": "0.5.0",
"1.2.0": "0.5.0",
"1.2.1": "0.5.0",
+"1.3.0": "0.5.0",
+"1.3.1": "0.5.0",
}
DEFAULT_SPARK_VERSION = SPARK_EC2_VERSION
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [MINOR] Add 1.3, 1.3.1 to master branch EC2 scripts
Repository: spark
Updated Branches:
refs/heads/branch-1.4 671a6bca5 -> 0ed376afa
[MINOR] Add 1.3, 1.3.1 to master branch EC2 scripts
cc pwendell
P.S: I can't believe this was outdated all along ?
Author: Shivaram Venkataraman
Closes #6215 from shivaram/update-ec2-map and squashes the following commits:
ae3937a [Shivaram Venkataraman] Add 1.3, 1.3.1 to master branch EC2 scripts
(cherry picked from commit 1a7b9ce80bb5649796dda48d6a6d662a2809d0ef)
Signed-off-by: Shivaram Venkataraman
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0ed376af
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0ed376af
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0ed376af
Branch: refs/heads/branch-1.4
Commit: 0ed376afad603b7afd86bb8eb312cad6edae2b9c
Parents: 671a6bc
Author: Shivaram Venkataraman
Authored: Sun May 17 00:12:20 2015 -0700
Committer: Shivaram Venkataraman
Committed: Sun May 17 00:12:46 2015 -0700
--
ec2/spark_ec2.py | 6 +-
1 file changed, 5 insertions(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/0ed376af/ec2/spark_ec2.py
--
diff --git a/ec2/spark_ec2.py b/ec2/spark_ec2.py
index ab4a96f..be92d5f 100755
--- a/ec2/spark_ec2.py
+++ b/ec2/spark_ec2.py
@@ -48,7 +48,7 @@ else:
from urllib.request import urlopen, Request
from urllib.error import HTTPError
-SPARK_EC2_VERSION = "1.2.1"
+SPARK_EC2_VERSION = "1.3.1"
SPARK_EC2_DIR = os.path.dirname(os.path.realpath(__file__))
VALID_SPARK_VERSIONS = set([
@@ -65,6 +65,8 @@ VALID_SPARK_VERSIONS = set([
"1.1.1",
"1.2.0",
"1.2.1",
+"1.3.0",
+"1.3.1",
])
SPARK_TACHYON_MAP = {
@@ -75,6 +77,8 @@ SPARK_TACHYON_MAP = {
"1.1.1": "0.5.0",
"1.2.0": "0.5.0",
"1.2.1": "0.5.0",
+"1.3.0": "0.5.0",
+"1.3.1": "0.5.0",
}
DEFAULT_SPARK_VERSION = SPARK_EC2_VERSION
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SQL] [MINOR] Skip unresolved expression for InConversion
Repository: spark
Updated Branches:
refs/heads/master 1a7b9ce80 -> edf09ea1b
[SQL] [MINOR] Skip unresolved expression for InConversion
Author: scwf
Closes #6145 from scwf/InConversion and squashes the following commits:
5c8ac6b [scwf] minir fix for InConversion
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/edf09ea1
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/edf09ea1
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/edf09ea1
Branch: refs/heads/master
Commit: edf09ea1bd4bf7692e0085ad9c70cb1bfc8d06d8
Parents: 1a7b9ce
Author: scwf
Authored: Sun May 17 15:17:11 2015 +0800
Committer: Cheng Lian
Committed: Sun May 17 15:17:11 2015 +0800
--
.../org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala | 3 +++
1 file changed, 3 insertions(+)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/edf09ea1/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
--
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
index fe0d3f2..b45b17d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HiveTypeCoercion.scala
@@ -296,6 +296,9 @@ trait HiveTypeCoercion {
*/
object InConversion extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
+ // Skip nodes who's children have not been resolved yet.
+ case e if !e.childrenResolved => e
+
case i @ In(a, b) if b.exists(_.dataType != a.dataType) =>
i.makeCopy(Array(a, b.map(Cast(_, a.dataType
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7447] [SQL] Don't re-merge Parquet schema when the relation is deserialized
Repository: spark
Updated Branches:
refs/heads/master edf09ea1b -> 339905578
[SPARK-7447] [SQL] Don't re-merge Parquet schema when the relation is
deserialized
JIRA: https://issues.apache.org/jira/browse/SPARK-7447
`MetadataCache` in `ParquetRelation2` is annotated as `transient`. When
`ParquetRelation2` is deserialized, we ask `MetadataCache` to refresh and
perform schema merging again. It is time-consuming especially for very many
parquet files.
With the new `FSBasedParquetRelation`, although `MetadataCache` is not
`transient` now, `MetadataCache.refresh()` still performs schema merging again
when the relation is deserialized.
Author: Liang-Chi Hsieh
Closes #6012 from viirya/without_remerge_schema and squashes the following
commits:
2663957 [Liang-Chi Hsieh] Merge remote-tracking branch 'upstream/master' into
without_remerge_schema
6ac7d93 [Liang-Chi Hsieh] Merge remote-tracking branch 'upstream/master' into
without_remerge_schema
b0fc09b [Liang-Chi Hsieh] Don't generate and merge parquetSchema multiple times.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/33990557
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/33990557
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/33990557
Branch: refs/heads/master
Commit: 339905578790fa37fcad9684b859b443313a5aa2
Parents: edf09ea
Author: Liang-Chi Hsieh
Authored: Sun May 17 15:42:21 2015 +0800
Committer: Cheng Lian
Committed: Sun May 17 15:42:21 2015 +0800
--
.../apache/spark/sql/parquet/newParquet.scala | 32 +++-
1 file changed, 18 insertions(+), 14 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/33990557/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
--
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
index 946062f..bcbdb1e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
@@ -340,7 +340,7 @@ private[sql] class ParquetRelation2(
// Schema of the actual Parquet files, without partition columns
discovered from partition
// directory paths.
-var dataSchema: StructType = _
+var dataSchema: StructType = null
// Schema of the whole table, including partition columns.
var schema: StructType = _
@@ -379,19 +379,23 @@ private[sql] class ParquetRelation2(
f -> new Footer(f.getPath, parquetMetadata)
}.seq.toMap
- dataSchema = {
-val dataSchema0 =
- maybeDataSchema
-.orElse(readSchema())
-.orElse(maybeMetastoreSchema)
-.getOrElse(sys.error("Failed to get the schema."))
-
-// If this Parquet relation is converted from a Hive Metastore table,
must reconcile case
-// case insensitivity issue and possible schema mismatch (probably
caused by schema
-// evolution).
-maybeMetastoreSchema
- .map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema0))
- .getOrElse(dataSchema0)
+ // If we already get the schema, don't need to re-compute it since the
schema merging is
+ // time-consuming.
+ if (dataSchema == null) {
+dataSchema = {
+ val dataSchema0 =
+maybeDataSchema
+ .orElse(readSchema())
+ .orElse(maybeMetastoreSchema)
+ .getOrElse(sys.error("Failed to get the schema."))
+
+ // If this Parquet relation is converted from a Hive Metastore
table, must reconcile case
+ // case insensitivity issue and possible schema mismatch (probably
caused by schema
+ // evolution).
+ maybeMetastoreSchema
+.map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema0))
+.getOrElse(dataSchema0)
+}
}
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7447] [SQL] Don't re-merge Parquet schema when the relation is deserialized
Repository: spark
Updated Branches:
refs/heads/branch-1.4 0ed376afa -> 898be6248
[SPARK-7447] [SQL] Don't re-merge Parquet schema when the relation is
deserialized
JIRA: https://issues.apache.org/jira/browse/SPARK-7447
`MetadataCache` in `ParquetRelation2` is annotated as `transient`. When
`ParquetRelation2` is deserialized, we ask `MetadataCache` to refresh and
perform schema merging again. It is time-consuming especially for very many
parquet files.
With the new `FSBasedParquetRelation`, although `MetadataCache` is not
`transient` now, `MetadataCache.refresh()` still performs schema merging again
when the relation is deserialized.
Author: Liang-Chi Hsieh
Closes #6012 from viirya/without_remerge_schema and squashes the following
commits:
2663957 [Liang-Chi Hsieh] Merge remote-tracking branch 'upstream/master' into
without_remerge_schema
6ac7d93 [Liang-Chi Hsieh] Merge remote-tracking branch 'upstream/master' into
without_remerge_schema
b0fc09b [Liang-Chi Hsieh] Don't generate and merge parquetSchema multiple times.
(cherry picked from commit 339905578790fa37fcad9684b859b443313a5aa2)
Signed-off-by: Cheng Lian
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/898be624
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/898be624
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/898be624
Branch: refs/heads/branch-1.4
Commit: 898be6248940acddab6f40af0fbc7e7abb3adb76
Parents: 0ed376a
Author: Liang-Chi Hsieh
Authored: Sun May 17 15:42:21 2015 +0800
Committer: Cheng Lian
Committed: Sun May 17 15:42:40 2015 +0800
--
.../apache/spark/sql/parquet/newParquet.scala | 32 +++-
1 file changed, 18 insertions(+), 14 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/898be624/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
--
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
index 946062f..bcbdb1e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
@@ -340,7 +340,7 @@ private[sql] class ParquetRelation2(
// Schema of the actual Parquet files, without partition columns
discovered from partition
// directory paths.
-var dataSchema: StructType = _
+var dataSchema: StructType = null
// Schema of the whole table, including partition columns.
var schema: StructType = _
@@ -379,19 +379,23 @@ private[sql] class ParquetRelation2(
f -> new Footer(f.getPath, parquetMetadata)
}.seq.toMap
- dataSchema = {
-val dataSchema0 =
- maybeDataSchema
-.orElse(readSchema())
-.orElse(maybeMetastoreSchema)
-.getOrElse(sys.error("Failed to get the schema."))
-
-// If this Parquet relation is converted from a Hive Metastore table,
must reconcile case
-// case insensitivity issue and possible schema mismatch (probably
caused by schema
-// evolution).
-maybeMetastoreSchema
- .map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema0))
- .getOrElse(dataSchema0)
+ // If we already get the schema, don't need to re-compute it since the
schema merging is
+ // time-consuming.
+ if (dataSchema == null) {
+dataSchema = {
+ val dataSchema0 =
+maybeDataSchema
+ .orElse(readSchema())
+ .orElse(maybeMetastoreSchema)
+ .getOrElse(sys.error("Failed to get the schema."))
+
+ // If this Parquet relation is converted from a Hive Metastore
table, must reconcile case
+ // case insensitivity issue and possible schema mismatch (probably
caused by schema
+ // evolution).
+ maybeMetastoreSchema
+.map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema0))
+.getOrElse(dataSchema0)
+}
}
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7669] Builds against Hadoop 2.6+ get inconsistent curator depend…
Repository: spark
Updated Branches:
refs/heads/branch-1.4 898be6248 -> 0feb3ded2
[SPARK-7669] Builds against Hadoop 2.6+ get inconsistent curator dependâ¦
This adds a new profile, `hadoop-2.6`, copying over the hadoop-2.4 properties,
updating ZK to 3.4.6 and making the curator version a configurable option. That
keeps the curator-recipes JAR in sync with that used in hadoop.
There's one more option to consider: making the full curator-client version
explicit with its own dependency version. This will pin down the version from
hadoop and hive imports
Author: Steve Loughran
Closes #6191 from steveloughran/stevel/SPARK-7669-hadoop-2.6 and squashes the
following commits:
e3e281a [Steve Loughran] SPARK-7669 declare the version of curator-client and
curator-framework JARs
2901ea9 [Steve Loughran] SPARK-7669 Builds against Hadoop 2.6+ get inconsistent
curator dependencies
(cherry picked from commit 50217667cc1239ed3b15f4d10907b727ed85d7fa)
Signed-off-by: Sean Owen
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0feb3ded
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0feb3ded
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0feb3ded
Branch: refs/heads/branch-1.4
Commit: 0feb3ded2e1681e0c282b73a1ec5af454bc78a13
Parents: 898be62
Author: Steve Loughran
Authored: Sun May 17 17:03:11 2015 +0100
Committer: Sean Owen
Committed: Sun May 17 17:03:20 2015 +0100
--
pom.xml | 26 --
1 file changed, 24 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/0feb3ded/pom.xml
--
diff --git a/pom.xml b/pom.xml
index 1b45cdb..6768a03 100644
--- a/pom.xml
+++ b/pom.xml
@@ -130,6 +130,7 @@
hbase
1.4.0
3.4.5
+2.4.0
org.spark-project.hive
0.13.1a
@@ -707,7 +708,7 @@
org.apache.curator
curator-recipes
-2.4.0
+${curator.version}
${hadoop.deps.scope}
@@ -717,6 +718,16 @@
+org.apache.curator
+curator-client
+${curator.version}
+
+
+org.apache.curator
+curator-framework
+${curator.version}
+
+
org.apache.hadoop
hadoop-client
${hadoop.version}
@@ -1680,6 +1691,17 @@
+ hadoop-2.6
+
+2.6.0
+0.9.3
+3.1.1
+3.4.6
+2.6.0
+
+
+
+
yarn
yarn
@@ -1709,7 +1731,7 @@
org.apache.curator
curator-recipes
- 2.4.0
+ ${curator.version}
org.apache.zookeeper
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7669] Builds against Hadoop 2.6+ get inconsistent curator depend…
Repository: spark
Updated Branches:
refs/heads/master 339905578 -> 50217667c
[SPARK-7669] Builds against Hadoop 2.6+ get inconsistent curator dependâ¦
This adds a new profile, `hadoop-2.6`, copying over the hadoop-2.4 properties,
updating ZK to 3.4.6 and making the curator version a configurable option. That
keeps the curator-recipes JAR in sync with that used in hadoop.
There's one more option to consider: making the full curator-client version
explicit with its own dependency version. This will pin down the version from
hadoop and hive imports
Author: Steve Loughran
Closes #6191 from steveloughran/stevel/SPARK-7669-hadoop-2.6 and squashes the
following commits:
e3e281a [Steve Loughran] SPARK-7669 declare the version of curator-client and
curator-framework JARs
2901ea9 [Steve Loughran] SPARK-7669 Builds against Hadoop 2.6+ get inconsistent
curator dependencies
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/50217667
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/50217667
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/50217667
Branch: refs/heads/master
Commit: 50217667cc1239ed3b15f4d10907b727ed85d7fa
Parents: 3399055
Author: Steve Loughran
Authored: Sun May 17 17:03:11 2015 +0100
Committer: Sean Owen
Committed: Sun May 17 17:03:11 2015 +0100
--
pom.xml | 26 --
1 file changed, 24 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/50217667/pom.xml
--
diff --git a/pom.xml b/pom.xml
index 1b45cdb..6768a03 100644
--- a/pom.xml
+++ b/pom.xml
@@ -130,6 +130,7 @@
hbase
1.4.0
3.4.5
+2.4.0
org.spark-project.hive
0.13.1a
@@ -707,7 +708,7 @@
org.apache.curator
curator-recipes
-2.4.0
+${curator.version}
${hadoop.deps.scope}
@@ -717,6 +718,16 @@
+org.apache.curator
+curator-client
+${curator.version}
+
+
+org.apache.curator
+curator-framework
+${curator.version}
+
+
org.apache.hadoop
hadoop-client
${hadoop.version}
@@ -1680,6 +1691,17 @@
+ hadoop-2.6
+
+2.6.0
+0.9.3
+3.1.1
+3.4.6
+2.6.0
+
+
+
+
yarn
yarn
@@ -1709,7 +1731,7 @@
org.apache.curator
curator-recipes
- 2.4.0
+ ${curator.version}
org.apache.zookeeper
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
Repository: spark
Updated Branches:
refs/heads/master 50217667c -> f2cc6b5bc
[SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
This patch wraps `SnappyOutputStream` to ensure that `close()` is idempotent
and to guard against write-after-`close()` bugs. This is a workaround for
https://github.com/xerial/snappy-java/issues/107, a bug where a non-idempotent
`close()` method can lead to stream corruption. We can remove this workaround
if we upgrade to a snappy-java version that contains my fix for this bug, but
in the meantime this patch offers a backportable Spark fix.
Author: Josh Rosen
Closes #6176 from JoshRosen/SPARK-7660-wrap-snappy and squashes the following
commits:
8b77aae [Josh Rosen] Wrap SnappyOutputStream to fix SPARK-7660
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/f2cc6b5b
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/f2cc6b5b
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/f2cc6b5b
Branch: refs/heads/master
Commit: f2cc6b5bccc3a70fd7d69183b1a068800831fe19
Parents: 5021766
Author: Josh Rosen
Authored: Sun May 17 09:30:49 2015 -0700
Committer: Josh Rosen
Committed: Sun May 17 09:30:49 2015 -0700
--
.../org/apache/spark/io/CompressionCodec.scala | 49 +++-
.../unsafe/UnsafeShuffleWriterSuite.java| 8
2 files changed, 47 insertions(+), 10 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/f2cc6b5b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
--
diff --git a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
index 0756cdb..0d8ac1f 100644
--- a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
+++ b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
@@ -17,7 +17,7 @@
package org.apache.spark.io
-import java.io.{InputStream, OutputStream}
+import java.io.{IOException, InputStream, OutputStream}
import com.ning.compress.lzf.{LZFInputStream, LZFOutputStream}
import net.jpountz.lz4.{LZ4BlockInputStream, LZ4BlockOutputStream}
@@ -154,8 +154,53 @@ class SnappyCompressionCodec(conf: SparkConf) extends
CompressionCodec {
override def compressedOutputStream(s: OutputStream): OutputStream = {
val blockSize =
conf.getSizeAsBytes("spark.io.compression.snappy.blockSize", "32k").toInt
-new SnappyOutputStream(s, blockSize)
+new SnappyOutputStreamWrapper(new SnappyOutputStream(s, blockSize))
}
override def compressedInputStream(s: InputStream): InputStream = new
SnappyInputStream(s)
}
+
+/**
+ * Wrapper over [[SnappyOutputStream]] which guards against write-after-close
and double-close
+ * issues. See SPARK-7660 for more details. This wrapping can be removed if we
upgrade to a version
+ * of snappy-java that contains the fix for
https://github.com/xerial/snappy-java/issues/107.
+ */
+private final class SnappyOutputStreamWrapper(os: SnappyOutputStream) extends
OutputStream {
+
+ private[this] var closed: Boolean = false
+
+ override def write(b: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte]): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte], off: Int, len: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b, off, len)
+ }
+
+ override def flush(): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.flush()
+ }
+
+ override def close(): Unit = {
+if (!closed) {
+ closed = true
+ os.close()
+}
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/f2cc6b5b/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
--
diff --git
a/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
b/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
index 78e5264..730d265 100644
---
a/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
+++
b/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
@@ -35,7 +35,6 @@ import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-import org.xerial.snappy.buffer.CachedBufferAllocator;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.lessThan;
@@ -97,13
spark git commit: [SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
Repository: spark
Updated Branches:
refs/heads/branch-1.4 0feb3ded2 -> 6df71eb8c
[SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
This patch wraps `SnappyOutputStream` to ensure that `close()` is idempotent
and to guard against write-after-`close()` bugs. This is a workaround for
https://github.com/xerial/snappy-java/issues/107, a bug where a non-idempotent
`close()` method can lead to stream corruption. We can remove this workaround
if we upgrade to a snappy-java version that contains my fix for this bug, but
in the meantime this patch offers a backportable Spark fix.
Author: Josh Rosen
Closes #6176 from JoshRosen/SPARK-7660-wrap-snappy and squashes the following
commits:
8b77aae [Josh Rosen] Wrap SnappyOutputStream to fix SPARK-7660
(cherry picked from commit f2cc6b5bccc3a70fd7d69183b1a068800831fe19)
Signed-off-by: Josh Rosen
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/6df71eb8
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/6df71eb8
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/6df71eb8
Branch: refs/heads/branch-1.4
Commit: 6df71eb8c1e05d2bcec16aab986d4cc15ac4aa8b
Parents: 0feb3de
Author: Josh Rosen
Authored: Sun May 17 09:30:49 2015 -0700
Committer: Josh Rosen
Committed: Sun May 17 09:33:49 2015 -0700
--
.../org/apache/spark/io/CompressionCodec.scala | 49 +++-
.../unsafe/UnsafeShuffleWriterSuite.java| 8
2 files changed, 47 insertions(+), 10 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/6df71eb8/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
--
diff --git a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
index 0756cdb..0d8ac1f 100644
--- a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
+++ b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
@@ -17,7 +17,7 @@
package org.apache.spark.io
-import java.io.{InputStream, OutputStream}
+import java.io.{IOException, InputStream, OutputStream}
import com.ning.compress.lzf.{LZFInputStream, LZFOutputStream}
import net.jpountz.lz4.{LZ4BlockInputStream, LZ4BlockOutputStream}
@@ -154,8 +154,53 @@ class SnappyCompressionCodec(conf: SparkConf) extends
CompressionCodec {
override def compressedOutputStream(s: OutputStream): OutputStream = {
val blockSize =
conf.getSizeAsBytes("spark.io.compression.snappy.blockSize", "32k").toInt
-new SnappyOutputStream(s, blockSize)
+new SnappyOutputStreamWrapper(new SnappyOutputStream(s, blockSize))
}
override def compressedInputStream(s: InputStream): InputStream = new
SnappyInputStream(s)
}
+
+/**
+ * Wrapper over [[SnappyOutputStream]] which guards against write-after-close
and double-close
+ * issues. See SPARK-7660 for more details. This wrapping can be removed if we
upgrade to a version
+ * of snappy-java that contains the fix for
https://github.com/xerial/snappy-java/issues/107.
+ */
+private final class SnappyOutputStreamWrapper(os: SnappyOutputStream) extends
OutputStream {
+
+ private[this] var closed: Boolean = false
+
+ override def write(b: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte]): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte], off: Int, len: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b, off, len)
+ }
+
+ override def flush(): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.flush()
+ }
+
+ override def close(): Unit = {
+if (!closed) {
+ closed = true
+ os.close()
+}
+ }
+}
http://git-wip-us.apache.org/repos/asf/spark/blob/6df71eb8/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
--
diff --git
a/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
b/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
index 78e5264..730d265 100644
---
a/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
+++
b/core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
@@ -35,7 +35,6 @@ import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-import org.xerial.snappy.buffer.CachedBufferAllocator;
import static org.hamcrest.MatcherAssert.assertThat;
i
spark git commit: [SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
Repository: spark
Updated Branches:
refs/heads/branch-1.3 91442fdfc -> 0a6310373
[SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
This patch wraps `SnappyOutputStream` to ensure that `close()` is idempotent
and to guard against write-after-`close()` bugs. This is a workaround for
https://github.com/xerial/snappy-java/issues/107, a bug where a non-idempotent
`close()` method can lead to stream corruption. We can remove this workaround
if we upgrade to a snappy-java version that contains my fix for this bug, but
in the meantime this patch offers a backportable Spark fix.
Author: Josh Rosen
Closes #6176 from JoshRosen/SPARK-7660-wrap-snappy and squashes the following
commits:
8b77aae [Josh Rosen] Wrap SnappyOutputStream to fix SPARK-7660
(cherry picked from commit f2cc6b5bccc3a70fd7d69183b1a068800831fe19)
Signed-off-by: Josh Rosen
Conflicts:
core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0a631037
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0a631037
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0a631037
Branch: refs/heads/branch-1.3
Commit: 0a63103739db029c1e871ed0f6c36aa43e326121
Parents: 91442fd
Author: Josh Rosen
Authored: Sun May 17 09:30:49 2015 -0700
Committer: Josh Rosen
Committed: Sun May 17 09:38:31 2015 -0700
--
.../org/apache/spark/io/CompressionCodec.scala | 49 +++-
1 file changed, 47 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/0a631037/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
--
diff --git a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
index 0709b6d..2fcb53d 100644
--- a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
+++ b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
@@ -17,7 +17,7 @@
package org.apache.spark.io
-import java.io.{InputStream, OutputStream}
+import java.io.{IOException, InputStream, OutputStream}
import com.ning.compress.lzf.{LZFInputStream, LZFOutputStream}
import net.jpountz.lz4.{LZ4BlockInputStream, LZ4BlockOutputStream}
@@ -154,8 +154,53 @@ class SnappyCompressionCodec(conf: SparkConf) extends
CompressionCodec {
override def compressedOutputStream(s: OutputStream): OutputStream = {
val blockSize = conf.getInt("spark.io.compression.snappy.block.size",
32768)
-new SnappyOutputStream(s, blockSize)
+new SnappyOutputStreamWrapper(new SnappyOutputStream(s, blockSize))
}
override def compressedInputStream(s: InputStream): InputStream = new
SnappyInputStream(s)
}
+
+/**
+ * Wrapper over [[SnappyOutputStream]] which guards against write-after-close
and double-close
+ * issues. See SPARK-7660 for more details. This wrapping can be removed if we
upgrade to a version
+ * of snappy-java that contains the fix for
https://github.com/xerial/snappy-java/issues/107.
+ */
+private final class SnappyOutputStreamWrapper(os: SnappyOutputStream) extends
OutputStream {
+
+ private[this] var closed: Boolean = false
+
+ override def write(b: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte]): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte], off: Int, len: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b, off, len)
+ }
+
+ override def flush(): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.flush()
+ }
+
+ override def close(): Unit = {
+if (!closed) {
+ closed = true
+ os.close()
+}
+ }
+}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
Repository: spark
Updated Branches:
refs/heads/branch-1.2 5505a0d07 -> 6c41e1cb9
[SPARK-7660] Wrap SnappyOutputStream to work around snappy-java bug
This patch wraps `SnappyOutputStream` to ensure that `close()` is idempotent
and to guard against write-after-`close()` bugs. This is a workaround for
https://github.com/xerial/snappy-java/issues/107, a bug where a non-idempotent
`close()` method can lead to stream corruption. We can remove this workaround
if we upgrade to a snappy-java version that contains my fix for this bug, but
in the meantime this patch offers a backportable Spark fix.
Author: Josh Rosen
Closes #6176 from JoshRosen/SPARK-7660-wrap-snappy and squashes the following
commits:
8b77aae [Josh Rosen] Wrap SnappyOutputStream to fix SPARK-7660
(cherry picked from commit f2cc6b5bccc3a70fd7d69183b1a068800831fe19)
Signed-off-by: Josh Rosen
Conflicts:
core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
core/src/test/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriterSuite.java
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/6c41e1cb
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/6c41e1cb
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/6c41e1cb
Branch: refs/heads/branch-1.2
Commit: 6c41e1cb9c4913d9d539cf7a5b4fe6cb2c075032
Parents: 5505a0d
Author: Josh Rosen
Authored: Sun May 17 09:30:49 2015 -0700
Committer: Josh Rosen
Committed: Sun May 17 09:41:08 2015 -0700
--
.../org/apache/spark/io/CompressionCodec.scala | 49 +++-
1 file changed, 47 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/6c41e1cb/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
--
diff --git a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
index 1ac7f4e..2343e69 100644
--- a/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
+++ b/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala
@@ -17,7 +17,7 @@
package org.apache.spark.io
-import java.io.{InputStream, OutputStream}
+import java.io.{IOException, InputStream, OutputStream}
import com.ning.compress.lzf.{LZFInputStream, LZFOutputStream}
import net.jpountz.lz4.{LZ4BlockInputStream, LZ4BlockOutputStream}
@@ -122,8 +122,53 @@ class SnappyCompressionCodec(conf: SparkConf) extends
CompressionCodec {
override def compressedOutputStream(s: OutputStream): OutputStream = {
val blockSize = conf.getInt("spark.io.compression.snappy.block.size",
32768)
-new SnappyOutputStream(s, blockSize)
+new SnappyOutputStreamWrapper(new SnappyOutputStream(s, blockSize))
}
override def compressedInputStream(s: InputStream): InputStream = new
SnappyInputStream(s)
}
+
+/**
+ * Wrapper over [[SnappyOutputStream]] which guards against write-after-close
and double-close
+ * issues. See SPARK-7660 for more details. This wrapping can be removed if we
upgrade to a version
+ * of snappy-java that contains the fix for
https://github.com/xerial/snappy-java/issues/107.
+ */
+private final class SnappyOutputStreamWrapper(os: SnappyOutputStream) extends
OutputStream {
+
+ private[this] var closed: Boolean = false
+
+ override def write(b: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte]): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b)
+ }
+
+ override def write(b: Array[Byte], off: Int, len: Int): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.write(b, off, len)
+ }
+
+ override def flush(): Unit = {
+if (closed) {
+ throw new IOException("Stream is closed")
+}
+os.flush()
+ }
+
+ override def close(): Unit = {
+if (!closed) {
+ closed = true
+ os.close()
+}
+ }
+}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7686] [SQL] DescribeCommand is assigned wrong output attributes in SparkStrategies
Repository: spark
Updated Branches:
refs/heads/master f2cc6b5bc -> 564562874
[SPARK-7686] [SQL] DescribeCommand is assigned wrong output attributes in
SparkStrategies
In `SparkStrategies`, `RunnableDescribeCommand` is called with the output
attributes of the table being described rather than the attributes for the
`describe` command's output. I discovered this issue because it caused type
conversion errors in some UnsafeRow conversion code that I'm writing.
Author: Josh Rosen
Closes #6217 from JoshRosen/SPARK-7686 and squashes the following commits:
953a344 [Josh Rosen] Fix SPARK-7686 with a simple change in SparkStrategies.
a4eec9f [Josh Rosen] Add failing regression test for SPARK-7686
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/56456287
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/56456287
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/56456287
Branch: refs/heads/master
Commit: 564562874f589c4c8bcabcd9d6eb9a6b0eada938
Parents: f2cc6b5
Author: Josh Rosen
Authored: Sun May 17 11:59:28 2015 -0700
Committer: Reynold Xin
Committed: Sun May 17 11:59:28 2015 -0700
--
.../scala/org/apache/spark/sql/execution/SparkStrategies.scala | 4 ++--
.../test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala | 6 ++
2 files changed, 8 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/56456287/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
--
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index af0029c..3f6a034 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -354,10 +354,10 @@ private[sql] abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
case c: CreateTableUsingAsSelect if !c.temporary =>
sys.error("Tables created with SQLContext must be TEMPORARY. Use a
HiveContext instead.")
- case LogicalDescribeCommand(table, isExtended) =>
+ case describe @ LogicalDescribeCommand(table, isExtended) =>
val resultPlan = self.sqlContext.executePlan(table).executedPlan
ExecutedCommand(
- RunnableDescribeCommand(resultPlan, resultPlan.output, isExtended))
:: Nil
+ RunnableDescribeCommand(resultPlan, describe.output, isExtended)) ::
Nil
case _ => Nil
}
http://git-wip-us.apache.org/repos/asf/spark/blob/56456287/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
--
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
index 6664e8d..f5106f6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
@@ -99,4 +99,10 @@ class DDLTestSuite extends DataSourceTest {
Row("arrayType", "array", ""),
Row("structType", "struct", "")
))
+
+ test("SPARK-7686 DescribeCommand should have correct physical plan output
attributes") {
+val attributes = sql("describe
ddlPeople").queryExecution.executedPlan.output
+assert(attributes.map(_.name) === Seq("col_name", "data_type", "comment"))
+assert(attributes.map(_.dataType).toSet === Set(StringType))
+ }
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7686] [SQL] DescribeCommand is assigned wrong output attributes in SparkStrategies
Repository: spark
Updated Branches:
refs/heads/branch-1.4 6df71eb8c -> 53d6ab51b
[SPARK-7686] [SQL] DescribeCommand is assigned wrong output attributes in
SparkStrategies
In `SparkStrategies`, `RunnableDescribeCommand` is called with the output
attributes of the table being described rather than the attributes for the
`describe` command's output. I discovered this issue because it caused type
conversion errors in some UnsafeRow conversion code that I'm writing.
Author: Josh Rosen
Closes #6217 from JoshRosen/SPARK-7686 and squashes the following commits:
953a344 [Josh Rosen] Fix SPARK-7686 with a simple change in SparkStrategies.
a4eec9f [Josh Rosen] Add failing regression test for SPARK-7686
(cherry picked from commit 564562874f589c4c8bcabcd9d6eb9a6b0eada938)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/53d6ab51
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/53d6ab51
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/53d6ab51
Branch: refs/heads/branch-1.4
Commit: 53d6ab51b21de9a1bfc355f8493fcd6e9159ab07
Parents: 6df71eb
Author: Josh Rosen
Authored: Sun May 17 11:59:28 2015 -0700
Committer: Reynold Xin
Committed: Sun May 17 11:59:35 2015 -0700
--
.../scala/org/apache/spark/sql/execution/SparkStrategies.scala | 4 ++--
.../test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala | 6 ++
2 files changed, 8 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/53d6ab51/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
--
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index af0029c..3f6a034 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -354,10 +354,10 @@ private[sql] abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
case c: CreateTableUsingAsSelect if !c.temporary =>
sys.error("Tables created with SQLContext must be TEMPORARY. Use a
HiveContext instead.")
- case LogicalDescribeCommand(table, isExtended) =>
+ case describe @ LogicalDescribeCommand(table, isExtended) =>
val resultPlan = self.sqlContext.executePlan(table).executedPlan
ExecutedCommand(
- RunnableDescribeCommand(resultPlan, resultPlan.output, isExtended))
:: Nil
+ RunnableDescribeCommand(resultPlan, describe.output, isExtended)) ::
Nil
case _ => Nil
}
http://git-wip-us.apache.org/repos/asf/spark/blob/53d6ab51/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
--
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
index 6664e8d..f5106f6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala
@@ -99,4 +99,10 @@ class DDLTestSuite extends DataSourceTest {
Row("arrayType", "array", ""),
Row("structType", "struct", "")
))
+
+ test("SPARK-7686 DescribeCommand should have correct physical plan output
attributes") {
+val attributes = sql("describe
ddlPeople").queryExecution.executedPlan.output
+assert(attributes.map(_.name) === Seq("col_name", "data_type", "comment"))
+assert(attributes.map(_.dataType).toSet === Set(StringType))
+ }
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7491] [SQL] Allow configuration of classloader isolation for hive
Repository: spark
Updated Branches:
refs/heads/master 564562874 -> 2ca60ace8
[SPARK-7491] [SQL] Allow configuration of classloader isolation for hive
Author: Michael Armbrust
Closes #6167 from marmbrus/configureIsolation and squashes the following
commits:
6147cbe [Michael Armbrust] filter other conf
22cc3bc7 [Michael Armbrust] Merge remote-tracking branch 'origin/master' into
configureIsolation
07476ee [Michael Armbrust] filter empty prefixes
dfdf19c [Michael Armbrust] [SPARK-6906][SQL] Allow configuration of classloader
isolation for hive
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2ca60ace
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2ca60ace
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2ca60ace
Branch: refs/heads/master
Commit: 2ca60ace8f42cf0bd4569d86c86c37a8a2b6a37c
Parents: 5645628
Author: Michael Armbrust
Authored: Sun May 17 12:43:15 2015 -0700
Committer: Michael Armbrust
Committed: Sun May 17 12:43:15 2015 -0700
--
.../org/apache/spark/sql/hive/HiveContext.scala | 33 ++--
.../sql/hive/client/IsolatedClientLoader.scala | 14 +
.../apache/spark/sql/hive/test/TestHive.scala | 9 +-
3 files changed, 46 insertions(+), 10 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/2ca60ace/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
--
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
index 9d98c36..2733ebd 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
@@ -122,6 +122,29 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc)
{
protected[hive] def hiveMetastoreJars: String =
getConf(HIVE_METASTORE_JARS, "builtin")
+ /**
+ * A comma separated list of class prefixes that should be loaded using the
classloader that
+ * is shared between Spark SQL and a specific version of Hive. An example of
classes that should
+ * be shared is JDBC drivers that are needed to talk to the metastore. Other
classes that need
+ * to be shared are those that interact with classes that are already
shared. For example,
+ * custom appenders that are used by log4j.
+ */
+ protected[hive] def hiveMetastoreSharedPrefixes: Seq[String] =
+getConf("spark.sql.hive.metastore.sharedPrefixes", jdbcPrefixes)
+ .split(",").filterNot(_ == "")
+
+ private def jdbcPrefixes = Seq(
+"com.mysql.jdbc", "org.postgresql", "com.microsoft.sqlserver",
"oracle.jdbc").mkString(",")
+
+ /**
+ * A comma separated list of class prefixes that should explicitly be
reloaded for each version
+ * of Hive that Spark SQL is communicating with. For example, Hive UDFs
that are declared in a
+ * prefix that typically would be shared (i.e. org.apache.spark.*)
+ */
+ protected[hive] def hiveMetastoreBarrierPrefixes: Seq[String] =
+getConf("spark.sql.hive.metastore.barrierPrefixes", "")
+ .split(",").filterNot(_ == "")
+
@transient
protected[sql] lazy val substitutor = new VariableSubstitution()
@@ -179,12 +202,14 @@ class HiveContext(sc: SparkContext) extends
SQLContext(sc) {
version = metaVersion,
execJars = jars.toSeq,
config = allConfig,
-isolationOn = true)
+isolationOn = true,
+barrierPrefixes = hiveMetastoreBarrierPrefixes,
+sharedPrefixes = hiveMetastoreSharedPrefixes)
} else if (hiveMetastoreJars == "maven") {
// TODO: Support for loading the jars from an already downloaded
location.
logInfo(
s"Initializing HiveMetastoreConnection version $hiveMetastoreVersion
using maven.")
- IsolatedClientLoader.forVersion(hiveMetastoreVersion, allConfig )
+ IsolatedClientLoader.forVersion(hiveMetastoreVersion, allConfig)
} else {
// Convert to files and expand any directories.
val jars =
@@ -210,7 +235,9 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
version = metaVersion,
execJars = jars.toSeq,
config = allConfig,
-isolationOn = true)
+isolationOn = true,
+barrierPrefixes = hiveMetastoreBarrierPrefixes,
+sharedPrefixes = hiveMetastoreSharedPrefixes)
}
isolatedLoader.client
}
http://git-wip-us.apache.org/repos/asf/spark/blob/2ca60ace/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala
--
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala
b/sql/hive/src/main/scal
spark git commit: [SPARK-7491] [SQL] Allow configuration of classloader isolation for hive
Repository: spark
Updated Branches:
refs/heads/branch-1.4 53d6ab51b -> a8556086d
[SPARK-7491] [SQL] Allow configuration of classloader isolation for hive
Author: Michael Armbrust
Closes #6167 from marmbrus/configureIsolation and squashes the following
commits:
6147cbe [Michael Armbrust] filter other conf
22cc3bc7 [Michael Armbrust] Merge remote-tracking branch 'origin/master' into
configureIsolation
07476ee [Michael Armbrust] filter empty prefixes
dfdf19c [Michael Armbrust] [SPARK-6906][SQL] Allow configuration of classloader
isolation for hive
(cherry picked from commit 2ca60ace8f42cf0bd4569d86c86c37a8a2b6a37c)
Signed-off-by: Michael Armbrust
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/a8556086
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/a8556086
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/a8556086
Branch: refs/heads/branch-1.4
Commit: a8556086d33cb993fab0ae2751e31455e6c664ab
Parents: 53d6ab5
Author: Michael Armbrust
Authored: Sun May 17 12:43:15 2015 -0700
Committer: Michael Armbrust
Committed: Sun May 17 12:43:26 2015 -0700
--
.../org/apache/spark/sql/hive/HiveContext.scala | 33 ++--
.../sql/hive/client/IsolatedClientLoader.scala | 14 +
.../apache/spark/sql/hive/test/TestHive.scala | 9 +-
3 files changed, 46 insertions(+), 10 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/a8556086/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
--
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
index 9d98c36..2733ebd 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala
@@ -122,6 +122,29 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc)
{
protected[hive] def hiveMetastoreJars: String =
getConf(HIVE_METASTORE_JARS, "builtin")
+ /**
+ * A comma separated list of class prefixes that should be loaded using the
classloader that
+ * is shared between Spark SQL and a specific version of Hive. An example of
classes that should
+ * be shared is JDBC drivers that are needed to talk to the metastore. Other
classes that need
+ * to be shared are those that interact with classes that are already
shared. For example,
+ * custom appenders that are used by log4j.
+ */
+ protected[hive] def hiveMetastoreSharedPrefixes: Seq[String] =
+getConf("spark.sql.hive.metastore.sharedPrefixes", jdbcPrefixes)
+ .split(",").filterNot(_ == "")
+
+ private def jdbcPrefixes = Seq(
+"com.mysql.jdbc", "org.postgresql", "com.microsoft.sqlserver",
"oracle.jdbc").mkString(",")
+
+ /**
+ * A comma separated list of class prefixes that should explicitly be
reloaded for each version
+ * of Hive that Spark SQL is communicating with. For example, Hive UDFs
that are declared in a
+ * prefix that typically would be shared (i.e. org.apache.spark.*)
+ */
+ protected[hive] def hiveMetastoreBarrierPrefixes: Seq[String] =
+getConf("spark.sql.hive.metastore.barrierPrefixes", "")
+ .split(",").filterNot(_ == "")
+
@transient
protected[sql] lazy val substitutor = new VariableSubstitution()
@@ -179,12 +202,14 @@ class HiveContext(sc: SparkContext) extends
SQLContext(sc) {
version = metaVersion,
execJars = jars.toSeq,
config = allConfig,
-isolationOn = true)
+isolationOn = true,
+barrierPrefixes = hiveMetastoreBarrierPrefixes,
+sharedPrefixes = hiveMetastoreSharedPrefixes)
} else if (hiveMetastoreJars == "maven") {
// TODO: Support for loading the jars from an already downloaded
location.
logInfo(
s"Initializing HiveMetastoreConnection version $hiveMetastoreVersion
using maven.")
- IsolatedClientLoader.forVersion(hiveMetastoreVersion, allConfig )
+ IsolatedClientLoader.forVersion(hiveMetastoreVersion, allConfig)
} else {
// Convert to files and expand any directories.
val jars =
@@ -210,7 +235,9 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
version = metaVersion,
execJars = jars.toSeq,
config = allConfig,
-isolationOn = true)
+isolationOn = true,
+barrierPrefixes = hiveMetastoreBarrierPrefixes,
+sharedPrefixes = hiveMetastoreSharedPrefixes)
}
isolatedLoader.client
}
http://git-wip-us.apache.org/repos/asf/spark/blob/a8556086/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala
--
diff --git
a
spark git commit: [SPARK-6514] [SPARK-5960] [SPARK-6656] [SPARK-7679] [STREAMING] [KINESIS] Updates to the Kinesis API
Repository: spark
Updated Branches:
refs/heads/master 2ca60ace8 -> ca4257aec
[SPARK-6514] [SPARK-5960] [SPARK-6656] [SPARK-7679] [STREAMING] [KINESIS]
Updates to the Kinesis API
SPARK-6514 - Use correct region
SPARK-5960 - Allow AWS Credentials to be directly passed
SPARK-6656 - Specify kinesis application name explicitly
SPARK-7679 - Upgrade to latest KCL and AWS SDK.
Author: Tathagata Das
Closes #6147 from tdas/kinesis-api-update and squashes the following commits:
f23ea77 [Tathagata Das] Updated versions and updated APIs
373b201 [Tathagata Das] Updated Kinesis API
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/ca4257ae
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/ca4257ae
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/ca4257ae
Branch: refs/heads/master
Commit: ca4257aec658aaa87f4f097dd7534033d5f13ddc
Parents: 2ca60ac
Author: Tathagata Das
Authored: Sun May 17 16:49:07 2015 -0700
Committer: Tathagata Das
Committed: Sun May 17 16:49:07 2015 -0700
--
.../kinesis/KinesisCheckpointState.scala| 2 +-
.../streaming/kinesis/KinesisReceiver.scala | 152 ++-
.../kinesis/KinesisRecordProcessor.scala| 32 ++-
.../spark/streaming/kinesis/KinesisUtils.scala | 263 ---
.../kinesis/KinesisReceiverSuite.scala | 15 +-
pom.xml | 4 +-
6 files changed, 348 insertions(+), 120 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/ca4257ae/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
--
diff --git
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
index 588e86a..1c9b0c2 100644
---
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
+++
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
@@ -48,7 +48,7 @@ private[kinesis] class KinesisCheckpointState(
/**
* Advance the checkpoint clock by the checkpoint interval.
*/
- def advanceCheckpoint() = {
+ def advanceCheckpoint(): Unit = {
checkpointClock.advance(checkpointInterval.milliseconds)
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/ca4257ae/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
--
diff --git
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
index a7fe447..01608fb 100644
---
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
+++
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
@@ -16,32 +16,31 @@
*/
package org.apache.spark.streaming.kinesis
-import java.net.InetAddress
import java.util.UUID
+import com.amazonaws.auth.{AWSCredentials, AWSCredentialsProvider,
BasicAWSCredentials, DefaultAWSCredentialsProviderChain}
+import
com.amazonaws.services.kinesis.clientlibrary.interfaces.{IRecordProcessor,
IRecordProcessorFactory}
+import
com.amazonaws.services.kinesis.clientlibrary.lib.worker.{InitialPositionInStream,
KinesisClientLibConfiguration, Worker}
+
import org.apache.spark.Logging
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.Duration
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.util.Utils
-import com.amazonaws.auth.AWSCredentialsProvider
-import com.amazonaws.auth.DefaultAWSCredentialsProviderChain
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor
-import
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory
-import
com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
-import
com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker
+
+private[kinesis]
+case class SerializableAWSCredentials(accessKeyId: String, secretKey: String)
+ extends BasicAWSCredentials(accessKeyId, secretKey) with Serializable
/**
* Custom AWS Kinesis-specific implementation of Spark Streaming's Receiver.
* This implementation relies on the Kinesis Client Library (KCL) Worker as
described here:
* https://github.com/awslabs/amazon-kinesis-client
- * This is a custom receiver used with
StreamingContext.receiverStream(Receiver)
- * as desc
spark git commit: [SPARK-6514] [SPARK-5960] [SPARK-6656] [SPARK-7679] [STREAMING] [KINESIS] Updates to the Kinesis API
Repository: spark
Updated Branches:
refs/heads/branch-1.4 a8556086d -> e0632ffaf
[SPARK-6514] [SPARK-5960] [SPARK-6656] [SPARK-7679] [STREAMING] [KINESIS]
Updates to the Kinesis API
SPARK-6514 - Use correct region
SPARK-5960 - Allow AWS Credentials to be directly passed
SPARK-6656 - Specify kinesis application name explicitly
SPARK-7679 - Upgrade to latest KCL and AWS SDK.
Author: Tathagata Das
Closes #6147 from tdas/kinesis-api-update and squashes the following commits:
f23ea77 [Tathagata Das] Updated versions and updated APIs
373b201 [Tathagata Das] Updated Kinesis API
(cherry picked from commit ca4257aec658aaa87f4f097dd7534033d5f13ddc)
Signed-off-by: Tathagata Das
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/e0632ffa
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/e0632ffa
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/e0632ffa
Branch: refs/heads/branch-1.4
Commit: e0632ffafd7c4bc6c87514f591fd6eb17f68cfed
Parents: a855608
Author: Tathagata Das
Authored: Sun May 17 16:49:07 2015 -0700
Committer: Tathagata Das
Committed: Sun May 17 16:49:31 2015 -0700
--
.../kinesis/KinesisCheckpointState.scala| 2 +-
.../streaming/kinesis/KinesisReceiver.scala | 152 ++-
.../kinesis/KinesisRecordProcessor.scala| 32 ++-
.../spark/streaming/kinesis/KinesisUtils.scala | 263 ---
.../kinesis/KinesisReceiverSuite.scala | 15 +-
pom.xml | 4 +-
6 files changed, 348 insertions(+), 120 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/e0632ffa/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
--
diff --git
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
index 588e86a..1c9b0c2 100644
---
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
+++
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointState.scala
@@ -48,7 +48,7 @@ private[kinesis] class KinesisCheckpointState(
/**
* Advance the checkpoint clock by the checkpoint interval.
*/
- def advanceCheckpoint() = {
+ def advanceCheckpoint(): Unit = {
checkpointClock.advance(checkpointInterval.milliseconds)
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/e0632ffa/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
--
diff --git
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
index a7fe447..01608fb 100644
---
a/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
+++
b/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala
@@ -16,32 +16,31 @@
*/
package org.apache.spark.streaming.kinesis
-import java.net.InetAddress
import java.util.UUID
+import com.amazonaws.auth.{AWSCredentials, AWSCredentialsProvider,
BasicAWSCredentials, DefaultAWSCredentialsProviderChain}
+import
com.amazonaws.services.kinesis.clientlibrary.interfaces.{IRecordProcessor,
IRecordProcessorFactory}
+import
com.amazonaws.services.kinesis.clientlibrary.lib.worker.{InitialPositionInStream,
KinesisClientLibConfiguration, Worker}
+
import org.apache.spark.Logging
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.Duration
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.util.Utils
-import com.amazonaws.auth.AWSCredentialsProvider
-import com.amazonaws.auth.DefaultAWSCredentialsProviderChain
-import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor
-import
com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory
-import
com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
-import
com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
-import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker
+
+private[kinesis]
+case class SerializableAWSCredentials(accessKeyId: String, secretKey: String)
+ extends BasicAWSCredentials(accessKeyId, secretKey) with Serializable
/**
* Custom AWS Kinesis-specific implementation of Spark Streaming's Receiver.
* This implementation relies on the Kinesis Client Library (KCL) Worker as
described here:
* https://github.com/awslabs/amazon-kine
spark git commit: [SQL] [MINOR] use catalyst type converter in ScalaUdf
Repository: spark Updated Branches: refs/heads/master ca4257aec -> 2f22424e9 [SQL] [MINOR] use catalyst type converter in ScalaUdf It's a follow-up of https://github.com/apache/spark/pull/5154, we can speed up scala udf evaluation by create type converter in advance. Author: Wenchen Fan Closes #6182 from cloud-fan/tmp and squashes the following commits: 241cfe9 [Wenchen Fan] use converter in ScalaUdf Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2f22424e Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2f22424e Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2f22424e Branch: refs/heads/master Commit: 2f22424e9f6624097b292cb70e00787b69d80718 Parents: ca4257a Author: Wenchen Fan Authored: Sun May 17 16:51:57 2015 -0700 Committer: Yin Huai Committed: Sun May 17 16:51:57 2015 -0700 -- .../org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/2f22424e/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala index 9a77ca6..d22eb10 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala @@ -956,7 +956,7 @@ case class ScalaUdf(function: AnyRef, dataType: DataType, children: Seq[Expressi } // scalastyle:on - - override def eval(input: Row): Any = CatalystTypeConverters.convertToCatalyst(f(input), dataType) + val converter = CatalystTypeConverters.createToCatalystConverter(dataType) + override def eval(input: Row): Any = converter(f(input)) } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SQL] [MINOR] use catalyst type converter in ScalaUdf
Repository: spark Updated Branches: refs/heads/branch-1.4 e0632ffaf -> be66d1924 [SQL] [MINOR] use catalyst type converter in ScalaUdf It's a follow-up of https://github.com/apache/spark/pull/5154, we can speed up scala udf evaluation by create type converter in advance. Author: Wenchen Fan Closes #6182 from cloud-fan/tmp and squashes the following commits: 241cfe9 [Wenchen Fan] use converter in ScalaUdf (cherry picked from commit 2f22424e9f6624097b292cb70e00787b69d80718) Signed-off-by: Yin Huai Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/be66d192 Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/be66d192 Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/be66d192 Branch: refs/heads/branch-1.4 Commit: be66d1924edc5c99987c80d445f34a690c3789a9 Parents: e0632ff Author: Wenchen Fan Authored: Sun May 17 16:51:57 2015 -0700 Committer: Yin Huai Committed: Sun May 17 16:52:21 2015 -0700 -- .../org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/be66d192/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala -- diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala index 9a77ca6..d22eb10 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ScalaUdf.scala @@ -956,7 +956,7 @@ case class ScalaUdf(function: AnyRef, dataType: DataType, children: Seq[Expressi } // scalastyle:on - - override def eval(input: Row): Any = CatalystTypeConverters.convertToCatalyst(f(input), dataType) + val converter = CatalystTypeConverters.createToCatalystConverter(dataType) + override def eval(input: Row): Any = converter(f(input)) } - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7693][Core] Remove "import scala.concurrent.ExecutionContext.Implicits.global"
Repository: spark
Updated Branches:
refs/heads/master 2f22424e9 -> ff71d34e0
[SPARK-7693][Core] Remove "import
scala.concurrent.ExecutionContext.Implicits.global"
Learnt a lesson from SPARK-7655: Spark should avoid to use
`scala.concurrent.ExecutionContext.Implicits.global` because the user may
submit blocking actions to `scala.concurrent.ExecutionContext.Implicits.global`
and exhaust all threads in it. This could crash Spark. So Spark should always
use its own thread pools for safety.
This PR removes all usages of
`scala.concurrent.ExecutionContext.Implicits.global` and uses proper thread
pools to replace them.
Author: zsxwing
Closes #6223 from zsxwing/SPARK-7693 and squashes the following commits:
a33ff06 [zsxwing] Decrease the max thread number from 1024 to 128
cf4b3fc [zsxwing] Remove "import
scala.concurrent.ExecutionContext.Implicits.global"
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/ff71d34e
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/ff71d34e
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/ff71d34e
Branch: refs/heads/master
Commit: ff71d34e00b64d70f671f9bf3e63aec39cd525e5
Parents: 2f22424
Author: zsxwing
Authored: Sun May 17 20:37:19 2015 -0700
Committer: Reynold Xin
Committed: Sun May 17 20:37:19 2015 -0700
--
.../executor/CoarseGrainedExecutorBackend.scala | 9 +++---
.../org/apache/spark/rdd/AsyncRDDActions.scala | 13 +++--
.../org/apache/spark/storage/BlockManager.scala | 17 +---
.../spark/storage/BlockManagerMaster.scala | 29
.../sql/execution/joins/BroadcastHashJoin.scala | 2 +-
.../streaming/receiver/ReceiverSupervisor.scala | 14 +++---
6 files changed, 58 insertions(+), 26 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/ff71d34e/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
--
diff --git
a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
index ed159de..f3a26f5 100644
---
a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
+++
b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -33,7 +33,7 @@ import org.apache.spark.deploy.worker.WorkerWatcher
import org.apache.spark.scheduler.TaskDescription
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
import org.apache.spark.serializer.SerializerInstance
-import org.apache.spark.util.{SignalLogger, Utils}
+import org.apache.spark.util.{ThreadUtils, SignalLogger, Utils}
private[spark] class CoarseGrainedExecutorBackend(
override val rpcEnv: RpcEnv,
@@ -55,18 +55,19 @@ private[spark] class CoarseGrainedExecutorBackend(
private[this] val ser: SerializerInstance =
env.closureSerializer.newInstance()
override def onStart() {
-import scala.concurrent.ExecutionContext.Implicits.global
logInfo("Connecting to driver: " + driverUrl)
rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref =>
+ // This is a very fast action so we can use "ThreadUtils.sameThread"
driver = Some(ref)
ref.ask[RegisteredExecutor.type](
RegisterExecutor(executorId, self, hostPort, cores, extractLogUrls))
-} onComplete {
+}(ThreadUtils.sameThread).onComplete {
+ // This is a very fast action so we can use "ThreadUtils.sameThread"
case Success(msg) => Utils.tryLogNonFatalError {
Option(self).foreach(_.send(msg)) // msg must be RegisteredExecutor
}
case Failure(e) => logError(s"Cannot register with driver: $driverUrl",
e)
-}
+}(ThreadUtils.sameThread)
}
def extractLogUrls: Map[String, String] = {
http://git-wip-us.apache.org/repos/asf/spark/blob/ff71d34e/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
--
diff --git a/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
b/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
index ec18534..bbf1b83 100644
--- a/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
@@ -19,8 +19,10 @@ package org.apache.spark.rdd
import java.util.concurrent.atomic.AtomicLong
+import org.apache.spark.util.ThreadUtils
+
import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.ExecutionContext
import scala.reflect.ClassTag
import org.apache.spark.{ComplexFutureAction, FutureAction, Logging}
@@ -66,6 +68,8 @@ class AsyncRDDActions[T: ClassTag](self: RDD[T])
spark git commit: [SPARK-7693][Core] Remove "import scala.concurrent.ExecutionContext.Implicits.global"
Repository: spark
Updated Branches:
refs/heads/branch-1.4 be66d1924 -> 2a42d2d8f
[SPARK-7693][Core] Remove "import
scala.concurrent.ExecutionContext.Implicits.global"
Learnt a lesson from SPARK-7655: Spark should avoid to use
`scala.concurrent.ExecutionContext.Implicits.global` because the user may
submit blocking actions to `scala.concurrent.ExecutionContext.Implicits.global`
and exhaust all threads in it. This could crash Spark. So Spark should always
use its own thread pools for safety.
This PR removes all usages of
`scala.concurrent.ExecutionContext.Implicits.global` and uses proper thread
pools to replace them.
Author: zsxwing
Closes #6223 from zsxwing/SPARK-7693 and squashes the following commits:
a33ff06 [zsxwing] Decrease the max thread number from 1024 to 128
cf4b3fc [zsxwing] Remove "import
scala.concurrent.ExecutionContext.Implicits.global"
(cherry picked from commit ff71d34e00b64d70f671f9bf3e63aec39cd525e5)
Signed-off-by: Reynold Xin
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2a42d2d8
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2a42d2d8
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2a42d2d8
Branch: refs/heads/branch-1.4
Commit: 2a42d2d8f26e3e66ab8c926e952a20a3900ca7f3
Parents: be66d19
Author: zsxwing
Authored: Sun May 17 20:37:19 2015 -0700
Committer: Reynold Xin
Committed: Sun May 17 20:37:27 2015 -0700
--
.../executor/CoarseGrainedExecutorBackend.scala | 9 +++---
.../org/apache/spark/rdd/AsyncRDDActions.scala | 13 +++--
.../org/apache/spark/storage/BlockManager.scala | 17 +---
.../spark/storage/BlockManagerMaster.scala | 29
.../sql/execution/joins/BroadcastHashJoin.scala | 2 +-
.../streaming/receiver/ReceiverSupervisor.scala | 14 +++---
6 files changed, 58 insertions(+), 26 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/2a42d2d8/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
--
diff --git
a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
index ed159de..f3a26f5 100644
---
a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
+++
b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -33,7 +33,7 @@ import org.apache.spark.deploy.worker.WorkerWatcher
import org.apache.spark.scheduler.TaskDescription
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
import org.apache.spark.serializer.SerializerInstance
-import org.apache.spark.util.{SignalLogger, Utils}
+import org.apache.spark.util.{ThreadUtils, SignalLogger, Utils}
private[spark] class CoarseGrainedExecutorBackend(
override val rpcEnv: RpcEnv,
@@ -55,18 +55,19 @@ private[spark] class CoarseGrainedExecutorBackend(
private[this] val ser: SerializerInstance =
env.closureSerializer.newInstance()
override def onStart() {
-import scala.concurrent.ExecutionContext.Implicits.global
logInfo("Connecting to driver: " + driverUrl)
rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref =>
+ // This is a very fast action so we can use "ThreadUtils.sameThread"
driver = Some(ref)
ref.ask[RegisteredExecutor.type](
RegisterExecutor(executorId, self, hostPort, cores, extractLogUrls))
-} onComplete {
+}(ThreadUtils.sameThread).onComplete {
+ // This is a very fast action so we can use "ThreadUtils.sameThread"
case Success(msg) => Utils.tryLogNonFatalError {
Option(self).foreach(_.send(msg)) // msg must be RegisteredExecutor
}
case Failure(e) => logError(s"Cannot register with driver: $driverUrl",
e)
-}
+}(ThreadUtils.sameThread)
}
def extractLogUrls: Map[String, String] = {
http://git-wip-us.apache.org/repos/asf/spark/blob/2a42d2d8/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
--
diff --git a/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
b/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
index ec18534..bbf1b83 100644
--- a/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/AsyncRDDActions.scala
@@ -19,8 +19,10 @@ package org.apache.spark.rdd
import java.util.concurrent.atomic.AtomicLong
+import org.apache.spark.util.ThreadUtils
+
import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.ExecutionContext
import scala.reflect.ClassTag
import org.apache.spark.{Comp
spark git commit: [SPARK-7694] [MLLIB] Use getOrElse for getting the threshold of LR model
Repository: spark
Updated Branches:
refs/heads/master ff71d34e0 -> 775e6f990
[SPARK-7694] [MLLIB] Use getOrElse for getting the threshold of LR model
The `toString` method of `LogisticRegressionModel` calls `get` method on an
Option (threshold) without a safeguard. In spark-shell, the following code `val
model = algorithm.run(data).clearThreshold()` in lbfgs code will fail as
`toString `method will be called right after `clearThreshold()` to show the
results in the REPL.
Author: Shuo Xiang
Closes #6224 from coderxiang/getorelse and squashes the following commits:
d5f53c9 [Shuo Xiang] use getOrElse for getting the threshold of LR model
5f109b4 [Shuo Xiang] Merge remote-tracking branch 'upstream/master'
c5c5bfe [Shuo Xiang] Merge remote-tracking branch 'upstream/master'
98804c9 [Shuo Xiang] fix bug in topBykey and update test
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/775e6f99
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/775e6f99
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/775e6f99
Branch: refs/heads/master
Commit: 775e6f9909d4495cbc11c377508b43482d782742
Parents: ff71d34
Author: Shuo Xiang
Authored: Sun May 17 21:16:52 2015 -0700
Committer: Xiangrui Meng
Committed: Sun May 17 21:16:52 2015 -0700
--
.../org/apache/spark/mllib/classification/LogisticRegression.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/775e6f99/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
--
diff --git
a/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
b/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
index bd2e907..2df4d21 100644
---
a/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
+++
b/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
@@ -163,7 +163,7 @@ class LogisticRegressionModel (
override protected def formatVersion: String = "1.0"
override def toString: String = {
-s"${super.toString}, numClasses = ${numClasses}, threshold =
${threshold.get}"
+s"${super.toString}, numClasses = ${numClasses}, threshold =
${threshold.getOrElse("None")}"
}
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
spark git commit: [SPARK-7694] [MLLIB] Use getOrElse for getting the threshold of LR model
Repository: spark
Updated Branches:
refs/heads/branch-1.4 2a42d2d8f -> 0b6bc8a23
[SPARK-7694] [MLLIB] Use getOrElse for getting the threshold of LR model
The `toString` method of `LogisticRegressionModel` calls `get` method on an
Option (threshold) without a safeguard. In spark-shell, the following code `val
model = algorithm.run(data).clearThreshold()` in lbfgs code will fail as
`toString `method will be called right after `clearThreshold()` to show the
results in the REPL.
Author: Shuo Xiang
Closes #6224 from coderxiang/getorelse and squashes the following commits:
d5f53c9 [Shuo Xiang] use getOrElse for getting the threshold of LR model
5f109b4 [Shuo Xiang] Merge remote-tracking branch 'upstream/master'
c5c5bfe [Shuo Xiang] Merge remote-tracking branch 'upstream/master'
98804c9 [Shuo Xiang] fix bug in topBykey and update test
(cherry picked from commit 775e6f9909d4495cbc11c377508b43482d782742)
Signed-off-by: Xiangrui Meng
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0b6bc8a2
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0b6bc8a2
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0b6bc8a2
Branch: refs/heads/branch-1.4
Commit: 0b6bc8a2394a26203d7be9ab53cc9dc5970d41a6
Parents: 2a42d2d
Author: Shuo Xiang
Authored: Sun May 17 21:16:52 2015 -0700
Committer: Xiangrui Meng
Committed: Sun May 17 21:16:59 2015 -0700
--
.../org/apache/spark/mllib/classification/LogisticRegression.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/0b6bc8a2/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
--
diff --git
a/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
b/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
index bd2e907..2df4d21 100644
---
a/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
+++
b/mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala
@@ -163,7 +163,7 @@ class LogisticRegressionModel (
override protected def formatVersion: String = "1.0"
override def toString: String = {
-s"${super.toString}, numClasses = ${numClasses}, threshold =
${threshold.get}"
+s"${super.toString}, numClasses = ${numClasses}, threshold =
${threshold.getOrElse("None")}"
}
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
