GitHub user rajesh7738 opened a pull request:
https://github.com/apache/spark/pull/21763
Branch 2.1
## What changes were proposed in this pull request?
(Please fill in changes proposed in this fix)
## How was this patch tested?
(Please explain how this patch was tested. E.g. unit tests, integration
tests, manual tests)
(If this patch involves UI changes, please attach a screenshot; otherwise,
remove this)
Please review http://spark.apache.org/contributing.html before opening a
pull request.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/apache/spark branch-2.1
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/21763.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #21763
----
commit 664c9795c94d3536ff9fe54af06e0fb6c0012862
Author: Shixiong Zhu <shixiong@...>
Date: 2017-03-04T03:00:35Z
[SPARK-19816][SQL][TESTS] Fix an issue that DataFrameCallbackSuite doesn't
recover the log level
## What changes were proposed in this pull request?
"DataFrameCallbackSuite.execute callback functions when a DataFrame action
failed" sets the log level to "fatal" but doesn't recover it. Hence, tests
running after it won't output any logs except fatal logs.
This PR uses `testQuietly` instead to avoid changing the log level.
## How was this patch tested?
Jenkins
Author: Shixiong Zhu <shixi...@databricks.com>
Closes #17156 from zsxwing/SPARK-19816.
(cherry picked from commit fbc4058037cf5b0be9f14a7dd28105f7f8151bed)
Signed-off-by: Yin Huai <yh...@databricks.com>
commit ca7a7e8a893a30d85e4315a4fa1ca1b1c56a703c
Author: uncleGen <hustyugm@...>
Date: 2017-03-06T02:17:30Z
[SPARK-19822][TEST] CheckpointSuite.testCheckpointedOperation: should not
filter checkpointFilesOfLatestTime with the PATH string.
## What changes were proposed in this pull request?
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/73800/testReport/
```
sbt.ForkMain$ForkError:
org.scalatest.exceptions.TestFailedDueToTimeoutException: The code
passed to eventually never returned normally. Attempted 617 times over
10.003740484 seconds.
Last failure message: 8 did not equal 2.
at
org.scalatest.concurrent.Eventually$class.tryTryAgain$1(Eventually.scala:420)
at
org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:438)
at org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
at
org.scalatest.concurrent.Eventually$class.eventually(Eventually.scala:336)
at org.scalatest.concurrent.Eventually$.eventually(Eventually.scala:478)
at
org.apache.spark.streaming.DStreamCheckpointTester$class.generateOutput(CheckpointSuite
.scala:172)
at
org.apache.spark.streaming.CheckpointSuite.generateOutput(CheckpointSuite.scala:211)
```
the check condition is:
```
val checkpointFilesOfLatestTime =
Checkpoint.getCheckpointFiles(checkpointDir).filter {
_.toString.contains(clock.getTimeMillis.toString)
}
// Checkpoint files are written twice for every batch interval. So assert
that both
// are written to make sure that both of them have been written.
assert(checkpointFilesOfLatestTime.size === 2)
```
the path string may contain the `clock.getTimeMillis.toString`, like `3500`
:
```
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-500
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-1000
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-1500
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-2000
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-2500
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-3000
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-3500.bk
file:/root/dev/spark/assembly/CheckpointSuite/spark-20035007-9891-4fb6-91c1-cc15b7ccaf15/checkpoint-3500
â²â²â²â²
```
so we should only check the filename, but not the whole path.
## How was this patch tested?
Jenkins.
Author: uncleGen <husty...@gmail.com>
Closes #17167 from uncleGen/flaky-CheckpointSuite.
(cherry picked from commit 207067ead6db6dc87b0d144a658e2564e3280a89)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit fd6c6d5c363008a229759bf628edc0f6c5e00ade
Author: Tyson Condie <tcondie@...>
Date: 2017-03-07T00:39:05Z
[SPARK-19719][SS] Kafka writer for both structured streaming and batch
queires
## What changes were proposed in this pull request?
Add a new Kafka Sink and Kafka Relation for writing streaming and batch
queries, respectively, to Apache Kafka.
### Streaming Kafka Sink
- When addBatch is called
-- If batchId is great than the last written batch
--- Write batch to Kafka
---- Topic will be taken from the record, if present, or from a topic
option, which overrides topic in record.
-- Else ignore
### Batch Kafka Sink
- KafkaSourceProvider will implement CreatableRelationProvider
- CreatableRelationProvider#createRelation will write the passed in
Dataframe to a Kafka
- Topic will be taken from the record, if present, or from topic option,
which overrides topic in record.
- Save modes Append and ErrorIfExist supported under identical semantics.
Other save modes result in an AnalysisException
tdas zsxwing
## How was this patch tested?
### The following unit tests will be included
- write to stream with topic field: valid stream write with data that
includes an existing topic in the schema
- write structured streaming aggregation w/o topic field, with default
topic: valid stream write with data that does not include a topic field, but
the configuration includes a default topic
- write data with bad schema: various cases of writing data that does not
conform to a proper schema e.g., 1. no topic field or default topic, and 2. no
value field
- write data with valid schema but wrong types: data with a complete schema
but wrong types e.g., key and value types are integers.
- write to non-existing topic: write a stream to a topic that does not
exist in Kafka, which has been configured to not auto-create topics.
- write batch to kafka: simple write batch to Kafka, which goes through the
same code path as streaming scenario, so validity checks will not be redone
here.
### Examples
```scala
// Structured Streaming
val writer = inputStringStream.map(s =>
s.get(0).toString.getBytes()).toDF("value")
.selectExpr("value as key", "value as value")
.writeStream
.format("kafka")
.option("checkpointLocation", checkpointDir)
.outputMode(OutputMode.Append)
.option("kafka.bootstrap.servers", brokerAddress)
.option("topic", topic)
.queryName("kafkaStream")
.start()
// Batch
val df = spark
.sparkContext
.parallelize(Seq("1", "2", "3", "4", "5"))
.map(v => (topic, v))
.toDF("topic", "value")
df.write
.format("kafka")
.option("kafka.bootstrap.servers",brokerAddress)
.option("topic", topic)
.save()
```
Please review http://spark.apache.org/contributing.html before opening a
pull request.
Author: Tyson Condie <tcon...@gmail.com>
Closes #17043 from tcondie/kafka-writer.
commit 711addd46e98e42deca97c5b9c0e55fddebaa458
Author: Jason White <jason.white@...>
Date: 2017-03-07T21:14:37Z
[SPARK-19561] [PYTHON] cast TimestampType.toInternal output to long
## What changes were proposed in this pull request?
Cast the output of `TimestampType.toInternal` to long to allow for proper
Timestamp creation in DataFrames near the epoch.
## How was this patch tested?
Added a new test that fails without the change.
dongjoon-hyun davies Mind taking a look?
The contribution is my original work and I license the work to the project
under the projectâs open source license.
Author: Jason White <jason.wh...@shopify.com>
Closes #16896 from JasonMWhite/SPARK-19561.
(cherry picked from commit 6f4684622a951806bebe7652a14f7d1ce03e24c7)
Signed-off-by: Davies Liu <davies....@gmail.com>
commit 551b7bdbe00b9ee803baa18e6b4690c478af9161
Author: Marcelo Vanzin <vanzin@...>
Date: 2017-03-08T00:21:18Z
[SPARK-19857][YARN] Correctly calculate next credential update time.
Add parentheses so that both lines form a single statement; also add
a log message so that the issue becomes more explicit if it shows up
again.
Tested manually with integration test that exercises the feature.
Author: Marcelo Vanzin <van...@cloudera.com>
Closes #17198 from vanzin/SPARK-19857.
(cherry picked from commit 8e41c2eed873e215b13215844ba5ba73a8906c5b)
Signed-off-by: Marcelo Vanzin <van...@cloudera.com>
commit cbc37007aa07991135a3da13ad566be76a0ef577
Author: Wenchen Fan <wenchen@...>
Date: 2017-03-08T01:15:39Z
Revert "[SPARK-19561] [PYTHON] cast TimestampType.toInternal output to long"
This reverts commit 6f4684622a951806bebe7652a14f7d1ce03e24c7.
commit 3b648a62626850470f8cceea3f0ec5dfd46e4e33
Author: Shixiong Zhu <shixiong@...>
Date: 2017-03-08T04:34:55Z
[SPARK-19859][SS] The new watermark should override the old one
## What changes were proposed in this pull request?
The new watermark should override the old one. Otherwise, we just pick up
the first column which has a watermark, it may be unexpected.
## How was this patch tested?
The new test.
Author: Shixiong Zhu <shixi...@databricks.com>
Closes #17199 from zsxwing/SPARK-19859.
(cherry picked from commit d8830c5039d9c7c5ef03631904c32873ab558e22)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit 0ba9ecbea88533b2562f2f6045eafeab99d8f0c6
Author: Bryan Cutler <cutlerb@...>
Date: 2017-03-08T04:44:30Z
[SPARK-19348][PYTHON] PySpark keyword_only decorator is not thread-safe
## What changes were proposed in this pull request?
The `keyword_only` decorator in PySpark is not thread-safe. It writes
kwargs to a static class variable in the decorator, which is then retrieved
later in the class method as `_input_kwargs`. If multiple threads are
constructing the same class with different kwargs, it becomes a race condition
to read from the static class variable before it's overwritten. See
[SPARK-19348](https://issues.apache.org/jira/browse/SPARK-19348) for
reproduction code.
This change will write the kwargs to a member variable so that multiple
threads can operate on separate instances without the race condition. It does
not protect against multiple threads operating on a single instance, but that
is better left to the user to synchronize.
## How was this patch tested?
Added new unit tests for using the keyword_only decorator and a regression
test that verifies `_input_kwargs` can be overwritten from different class
instances.
Author: Bryan Cutler <cutl...@gmail.com>
Closes #17193 from
BryanCutler/pyspark-keyword_only-threadsafe-SPARK-19348-2_1.
commit 320eff14b0bb634eba2cdcae2303ba38fd0eb282
Author: Michael Armbrust <michael@...>
Date: 2017-03-08T09:32:42Z
[SPARK-18055][SQL] Use correct mirror in ExpresionEncoder
Previously, we were using the mirror of passed in `TypeTag` when reflecting
to build an encoder. This fails when the outer class is built in (i.e. `Seq`'s
default mirror is based on root classloader) but inner classes (i.e. `A` in
`Seq[A]`) are defined in the REPL or a library.
This patch changes us to always reflect based on a mirror created using the
context classloader.
Author: Michael Armbrust <mich...@databricks.com>
Closes #17201 from marmbrus/replSeqEncoder.
(cherry picked from commit 314e48a3584bad4b486b046bbf0159d64ba857bc)
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
commit f6c1ad2eb6d0706899aabbdd39e558b3488e2ef3
Author: Burak Yavuz <brkyvz@...>
Date: 2017-03-08T22:35:07Z
[SPARK-19813] maxFilesPerTrigger combo latestFirst may miss old files in
combination with maxFileAge in FileStreamSource
## What changes were proposed in this pull request?
**The Problem**
There is a file stream source option called maxFileAge which limits how old
the files can be, relative the latest file that has been seen. This is used to
limit the files that need to be remembered as "processed". Files older than the
latest processed files are ignored. This values is by default 7 days.
This causes a problem when both
latestFirst = true
maxFilesPerTrigger > total files to be processed.
Here is what happens in all combinations
1) latestFirst = false - Since files are processed in order, there wont be
any unprocessed file older than the latest processed file. All files will be
processed.
2) latestFirst = true AND maxFilesPerTrigger is not set - The maxFileAge
thresholding mechanism takes one batch initialize. If maxFilesPerTrigger is
not, then all old files get processed in the first batch, and so no file is
left behind.
3) latestFirst = true AND maxFilesPerTrigger is set to X - The first batch
process the latest X files. That sets the threshold latest file - maxFileAge,
so files older than this threshold will never be considered for processing.
The bug is with case 3.
**The Solution**
Ignore `maxFileAge` when both `maxFilesPerTrigger` and `latestFirst` are
set.
## How was this patch tested?
Regression test in `FileStreamSourceSuite`
Author: Burak Yavuz <brk...@gmail.com>
Closes #17153 from brkyvz/maxFileAge.
(cherry picked from commit a3648b5d4f99ff9461d02f53e9ec71787a3abf51)
Signed-off-by: Burak Yavuz <brk...@gmail.com>
commit 3457c32297e0150a4fbc80a30f84b9c62ca7c372
Author: Shixiong Zhu <shixiong@...>
Date: 2017-03-08T22:30:54Z
Revert "[SPARK-19413][SS] MapGroupsWithState for arbitrary stateful
operations for branch-2.1"
This reverts commit 502c927b8c8a99ef2adf4e6e1d7a6d9232d45ef5.
commit 78cc5721f07af5c561e89d1bbc72975bb67abb74
Author: Dilip Biswal <dbiswal@...>
Date: 2017-03-09T01:33:49Z
[MINOR][SQL] The analyzer rules are fired twice for cases when
AnalysisException is raised from analyzer.
## What changes were proposed in this pull request?
In general we have a checkAnalysis phase which validates the logical plan
and throws AnalysisException on semantic errors. However we also can throw
AnalysisException from a few analyzer rules like ResolveSubquery.
I found that we fire up the analyzer rules twice for the queries that throw
AnalysisException from one of the analyzer rules. This is a very minor fix. We
don't have to strictly fix it. I just got confused seeing the rule getting
fired two times when i was not expecting it.
## How was this patch tested?
Tested manually.
Author: Dilip Biswal <dbis...@us.ibm.com>
Closes #17214 from dilipbiswal/analyis_twice.
(cherry picked from commit d809ceed9762d5bbb04170e45f38751713112dd8)
Signed-off-by: Xiao Li <gatorsm...@gmail.com>
commit 00859e148fd1002fa314542953fee61a5d0fb9d9
Author: Shixiong Zhu <shixiong@...>
Date: 2017-03-09T07:15:52Z
[SPARK-19874][BUILD] Hide API docs for org.apache.spark.sql.internal
## What changes were proposed in this pull request?
The API docs should not include the "org.apache.spark.sql.internal" package
because they are internal private APIs.
## How was this patch tested?
Jenkins
Author: Shixiong Zhu <shixi...@databricks.com>
Closes #17217 from zsxwing/SPARK-19874.
(cherry picked from commit 029e40b412e332c9f0fff283d604e203066c78c0)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit 0c140c1682262bc27df94952bda6ad8e3229fda4
Author: uncleGen <hustyugm@...>
Date: 2017-03-09T07:23:10Z
[SPARK-19859][SS][FOLLOW-UP] The new watermark should override the old one.
## What changes were proposed in this pull request?
A follow up to SPARK-19859:
- extract the calculation of `delayMs` and reuse it.
- update EventTimeWatermarkExec
- use the correct `delayMs` in EventTimeWatermark
## How was this patch tested?
Jenkins.
Author: uncleGen <husty...@gmail.com>
Closes #17221 from uncleGen/SPARK-19859.
(cherry picked from commit eeb1d6db878641d9eac62d0869a90fe80c1f4461)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit 2a76e2420f6976bd2ef2fcf7b8c8db1f0d37c1ad
Author: Jason White <jason.white@...>
Date: 2017-03-09T18:34:54Z
[SPARK-19561][SQL] add int case handling for TimestampType
## What changes were proposed in this pull request?
Add handling of input of type `Int` for dataType `TimestampType` to
`EvaluatePython.scala`. Py4J serializes ints smaller than MIN_INT or larger
than MAX_INT to Long, which are handled correctly already, but values between
MIN_INT and MAX_INT are serialized to Int.
These range limits correspond to roughly half an hour on either side of the
epoch. As a result, PySpark doesn't allow TimestampType values to be created in
this range.
Alternatives attempted: patching the `TimestampType.toInternal` function to
cast return values to `long`, so Py4J would always serialize them to Scala
Long. Python3 does not have a `long` type, so this approach failed on Python3.
## How was this patch tested?
Added a new PySpark-side test that fails without the change.
The contribution is my original work and I license the work to the project
under the projectâs open source license.
Resubmission of https://github.com/apache/spark/pull/16896. The original PR
didn't go through Jenkins and broke the build. davies dongjoon-hyun
cloud-fan Could you kick off a Jenkins run for me? It passed everything for
me locally, but it's possible something has changed in the last few weeks.
Author: Jason White <jason.wh...@shopify.com>
Closes #17200 from JasonMWhite/SPARK-19561.
(cherry picked from commit 206030bd12405623c00c1ff334663984b9250adb)
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
commit ffe65b06511f3143cb2549073bfbe145663ad561
Author: uncleGen <hustyugm@...>
Date: 2017-03-09T19:07:31Z
[SPARK-19861][SS] watermark should not be a negative time.
## What changes were proposed in this pull request?
`watermark` should not be negative. This behavior is invalid, check it
before real run.
## How was this patch tested?
add new unit test.
Author: uncleGen <husty...@gmail.com>
Author: dylon <husty...@gmail.com>
Closes #17202 from uncleGen/SPARK-19861.
(cherry picked from commit 30b18e69361746b4d656474374d8b486bb48a19e)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit a59cc369f57cfd4fc8f2a7177c9519731c71c63a
Author: Burak Yavuz <brkyvz@...>
Date: 2017-03-10T01:42:10Z
[SPARK-19886] Fix reportDataLoss if statement in SS KafkaSource
## What changes were proposed in this pull request?
Fix the `throw new IllegalStateException` if statement part.
## How is this patch tested
Regression test
Author: Burak Yavuz <brk...@gmail.com>
Closes #17228 from brkyvz/kafka-cause-fix.
(cherry picked from commit 82138e09b9ad8d9609d5c64d6c11244b8f230be7)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit f0d50fd547c247df06470d68cd5245e3027e89a2
Author: Tyson Condie <tcondie@...>
Date: 2017-03-10T07:02:13Z
[SPARK-19891][SS] Await Batch Lock notified on stream execution exit
## What changes were proposed in this pull request?
We need to notify the await batch lock when the stream exits early e.g.,
when an exception has been thrown.
## How was this patch tested?
Current tests that throw exceptions at runtime will finish faster as a
result of this update.
zsxwing
Please review http://spark.apache.org/contributing.html before opening a
pull request.
Author: Tyson Condie <tcon...@gmail.com>
Closes #17231 from tcondie/kafka-writer.
(cherry picked from commit 501b7111997bc74754663348967104181b43319b)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit 5a2ad4312dd00a450eac49ce53d70d9541e9e4cb
Author: Wenchen Fan <wenchen@...>
Date: 2017-03-11T00:14:22Z
[SPARK-19893][SQL] should not run DataFrame set oprations with map type
In spark SQL, map type can't be used in equality test/comparison, and
`Intersect`/`Except`/`Distinct` do need equality test for all columns, we
should not allow map type in `Intersect`/`Except`/`Distinct`.
new regression test
Author: Wenchen Fan <wenc...@databricks.com>
Closes #17236 from cloud-fan/map.
(cherry picked from commit fb9beda54622e0c3190c6504fc468fa4e50eeb45)
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
commit e481a73819213e4a7919e14e979b79a65098224f
Author: Budde <budde@...>
Date: 2017-03-11T00:38:16Z
[SPARK-19611][SQL] Introduce configurable table schema inference
Add a new configuration option that allows Spark SQL to infer a
case-sensitive schema from a Hive Metastore table's data files when a
case-sensitive schema can't be read from the table properties.
- Add spark.sql.hive.caseSensitiveInferenceMode param to SQLConf
- Add schemaPreservesCase field to CatalogTable (set to false when schema
can't
successfully be read from Hive table props)
- Perform schema inference in HiveMetastoreCatalog if schemaPreservesCase is
false, depending on spark.sql.hive.caseSensitiveInferenceMode
- Add alterTableSchema() method to the ExternalCatalog interface
- Add HiveSchemaInferenceSuite tests
- Refactor and move ParquetFileForamt.meregeMetastoreParquetSchema() as
HiveMetastoreCatalog.mergeWithMetastoreSchema
- Move schema merging tests from ParquetSchemaSuite to
HiveSchemaInferenceSuite
[JIRA for this change](https://issues.apache.org/jira/browse/SPARK-19611)
The tests in ```HiveSchemaInferenceSuite``` should verify that schema
inference is working as expected. ```ExternalCatalogSuite``` has also been
extended to cover the new ```alterTableSchema()``` API.
Author: Budde <bu...@amazon.com>
Closes #17229 from budde/SPARK-19611-2.1.
commit f9833c66a2f11414357854dae00e9e2448869254
Author: uncleGen <hustyugm@...>
Date: 2017-03-12T08:29:37Z
[DOCS][SS] fix structured streaming python example
## What changes were proposed in this pull request?
- SS python example: `TypeError: 'xxx' object is not callable`
- some other doc issue.
## How was this patch tested?
Jenkins.
Author: uncleGen <husty...@gmail.com>
Closes #17257 from uncleGen/docs-ss-python.
(cherry picked from commit e29a74d5b1fa3f9356b7af5dd7e3fce49bc8eb7d)
Signed-off-by: Sean Owen <so...@cloudera.com>
commit 8c460804698742b0405d6c7e8a1880472f436f9e
Author: uncleGen <hustyugm@...>
Date: 2017-03-13T00:46:31Z
[SPARK-19853][SS] uppercase kafka topics fail when startingOffsets are
SpecificOffsets
When using the KafkaSource with Structured Streaming, consumer assignments
are not what the user expects if startingOffsets is set to an explicit set of
topics/partitions in JSON where the topic(s) happen to have uppercase
characters. When StartingOffsets is constructed, the original string value from
options is transformed toLowerCase to make matching on "earliest" and "latest"
case insensitive. However, the toLowerCase JSON is passed to SpecificOffsets
for the terminal condition, so topic names may not be what the user intended by
the time assignments are made with the underlying KafkaConsumer.
KafkaSourceProvider.scala:
```
val startingOffsets =
caseInsensitiveParams.get(STARTING_OFFSETS_OPTION_KEY).map(_.trim.toLowerCase)
match {
case Some("latest") => LatestOffsets
case Some("earliest") => EarliestOffsets
case Some(json) => SpecificOffsets(JsonUtils.partitionOffsets(json))
case None => LatestOffsets
}
```
Thank cbowden for reporting.
Jenkins
Author: uncleGen <husty...@gmail.com>
Closes #17209 from uncleGen/SPARK-19853.
(cherry picked from commit 0a4d06a7c3db9fec2b6f050a631e8b59b0e9376e)
Signed-off-by: Shixiong Zhu <shixi...@databricks.com>
commit 454578257181b0ae8768f9d34fb64964b32530ce
Author: Herman van Hovell <hvanhovell@...>
Date: 2017-03-14T17:52:16Z
[SPARK-19933][SQL] Do not change output of a subquery
## What changes were proposed in this pull request?
The `RemoveRedundantAlias` rule can change the output attributes (the
expression id's to be precise) of a query by eliminating the redundant alias
producing them. This is no problem for a regular query, but can cause problems
for correlated subqueries: The attributes produced by the subquery are used in
the parent plan; changing them will break the parent plan.
This PR fixes this by wrapping a subquery in a `Subquery` top level node
when it gets optimized. The `RemoveRedundantAlias` rule now recognizes
`Subquery` and makes sure that the output attributes of the `Subquery` node are
retained.
## How was this patch tested?
Added a test case to `RemoveRedundantAliasAndProjectSuite` and added a
regression test to `SubquerySuite`.
Author: Herman van Hovell <hvanhov...@databricks.com>
Closes #17278 from hvanhovell/SPARK-19933.
(cherry picked from commit e04c05cf41a125b0526f59f9b9e7fdf0b78b8b21)
Signed-off-by: Herman van Hovell <hvanhov...@databricks.com>
commit a0ce845d9ad87753f676785301ab7ccd8ddd6368
Author: Wenchen Fan <wenchen@...>
Date: 2017-03-15T00:24:41Z
[SPARK-19887][SQL] dynamic partition keys can be null or empty string
When dynamic partition value is null or empty string, we should write the
data to a directory like `a=__HIVE_DEFAULT_PARTITION__`, when we read the data
back, we should respect this special directory name and treat it as null.
This is the same behavior of impala, see
https://issues.apache.org/jira/browse/IMPALA-252
new regression test
Author: Wenchen Fan <wenc...@databricks.com>
Closes #17277 from cloud-fan/partition.
(cherry picked from commit dacc382f0c918f1ca808228484305ce0e21c705e)
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
commit 80ebca62cbdb7d5c8606e95a944164ab1a943694
Author: Reynold Xin <rxin@...>
Date: 2017-03-15T12:07:20Z
[SPARK-19944][SQL] Move SQLConf from sql/core to sql/catalyst (branch-2.1)
## What changes were proposed in this pull request?
This patch moves SQLConf from sql/core to sql/catalyst. To minimize the
changes, the patch used type alias to still keep CatalystConf (as a type alias)
and SimpleCatalystConf (as a concrete class that extends SQLConf).
Motivation for the change is that it is pretty weird to have SQLConf only
in sql/core and then we have to duplicate config options that impact
optimizer/analyzer in sql/catalyst using CatalystConf.
This is a backport into branch-2.1 to minimize merge conflicts.
## How was this patch tested?
N/A
Author: Reynold Xin <r...@databricks.com>
Closes #17301 from rxin/branch-2.1-conf.
commit 062254635a98da0b08f69dc7e8907079cfdce035
Author: hyukjinkwon <gurwls223@...>
Date: 2017-03-15T17:17:18Z
[SPARK-19872] [PYTHON] Use the correct deserializer for RDD construction
for coalesce/repartition
## What changes were proposed in this pull request?
This PR proposes to use the correct deserializer, `BatchedSerializer` for
RDD construction for coalesce/repartition when the shuffle is enabled.
Currently, it is passing `UTF8Deserializer` as is not `BatchedSerializer` from
the copied one.
with the file, `text.txt` below:
```
a
b
d
e
f
g
h
i
j
k
l
```
- Before
```python
>>> sc.textFile('text.txt').repartition(1).collect()
```
```
UTF8Deserializer(True)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File ".../spark/python/pyspark/rdd.py", line 811, in collect
return list(_load_from_socket(port, self._jrdd_deserializer))
File ".../spark/python/pyspark/serializers.py", line 549, in load_stream
yield self.loads(stream)
File ".../spark/python/pyspark/serializers.py", line 544, in loads
return s.decode("utf-8") if self.use_unicode else s
File
"/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/encodings/utf_8.py",
line 16, in decode
return codecs.utf_8_decode(input, errors, True)
UnicodeDecodeError: 'utf8' codec can't decode byte 0x80 in position 0:
invalid start byte
```
- After
```python
>>> sc.textFile('text.txt').repartition(1).collect()
```
```
[u'a', u'b', u'', u'd', u'e', u'f', u'g', u'h', u'i', u'j', u'k', u'l', u'']
```
## How was this patch tested?
Unit test in `python/pyspark/tests.py`.
Author: hyukjinkwon <gurwls...@gmail.com>
Closes #17282 from HyukjinKwon/SPARK-19872.
(cherry picked from commit 7387126f83dc0489eb1df734bfeba705709b7861)
Signed-off-by: Davies Liu <davies....@gmail.com>
commit 9d032d02c8988d221a6e4cb27e6ee31627ed8a8e
Author: windpiger <songjun@...>
Date: 2017-03-16T17:30:39Z
[SPARK-19329][SQL][BRANCH-2.1] Reading from or writing to a datasource
table with a non pre-existing location should succeed
## What changes were proposed in this pull request?
This is a backport pr of https://github.com/apache/spark/pull/16672 into
branch-2.1.
## How was this patch tested?
Existing tests.
Author: windpiger <song...@outlook.com>
Closes #17317 from windpiger/backport-insertnotexists.
commit 4b977ff041681e73529e745c63a3f7c2b185df2b
Author: Xiao Li <gatorsmile@...>
Date: 2017-03-17T02:57:53Z
[SPARK-19765][SPARK-18549][SPARK-19093][SPARK-19736][BACKPORT-2.1][SQL]
Backport Three Cache-related PRs to Spark 2.1
### What changes were proposed in this pull request?
Backport a few cache related PRs:
---
[[SPARK-19093][SQL] Cached tables are not used in
SubqueryExpression](https://github.com/apache/spark/pull/16493)
Consider the plans inside subquery expressions while looking up cache
manager to make
use of cached data. Currently CacheManager.useCachedData does not consider
the
subquery expressions in the plan.
---
[[SPARK-19736][SQL] refreshByPath should clear all cached plans with the
specified path](https://github.com/apache/spark/pull/17064)
Catalog.refreshByPath can refresh the cache entry and the associated
metadata for all dataframes (if any), that contain the given data source path.
However, CacheManager.invalidateCachedPath doesn't clear all cached plans
with the specified path. It causes some strange behaviors reported in
SPARK-15678.
---
[[SPARK-19765][SPARK-18549][SQL] UNCACHE TABLE should un-cache all cached
plans that refer to this table](https://github.com/apache/spark/pull/17097)
When un-cache a table, we should not only remove the cache entry for this
table, but also un-cache any other cached plans that refer to this table. The
following commands trigger the table uncache: `DropTableCommand`,
`TruncateTableCommand`, `AlterTableRenameCommand`, `UncacheTableCommand`,
`RefreshTable` and `InsertIntoHiveTable`
This PR also includes some refactors:
- use java.util.LinkedList to store the cache entries, so that it's safer
to remove elements while iterating
- rename invalidateCache to recacheByPlan, which is more obvious about what
it does.
### How was this patch tested?
N/A
Author: Xiao Li <gatorsm...@gmail.com>
Closes #17319 from gatorsmile/backport-17097.
commit 710b5554e8a8a502f8a4fe9ce4b865b074646157
Author: Liwei Lin <lwlin7@...>
Date: 2017-03-17T17:41:17Z
[SPARK-19721][SS][BRANCH-2.1] Good error message for version mismatch in
log files
## Problem
There are several places where we write out version identifiers in various
logs for structured streaming (usually `v1`). However, in the places where we
check for this, we throw a confusing error message.
## What changes were proposed in this pull request?
This patch made two major changes:
1. added a `parseVersion(...)` method, and based on this method, fixed the
following places the way they did version checking (no other place needed to do
this checking):
```
HDFSMetadataLog
- CompactibleFileStreamLog ------------> fixed with this patch
- FileStreamSourceLog ---------------> inherited the fix of
`CompactibleFileStreamLog`
- FileStreamSinkLog -----------------> inherited the fix of
`CompactibleFileStreamLog`
- OffsetSeqLog ------------------------> fixed with this patch
- anonymous subclass in KafkaSource ---> fixed with this patch
```
2. changed the type of `FileStreamSinkLog.VERSION`,
`FileStreamSourceLog.VERSION` etc. from `String` to `Int`, so that we can
identify newer versions via `version > 1` instead of `version != "v1"`
- note this didn't break any backwards compatibility -- we are still
writing out `"v1"` and reading back `"v1"`
## Exception message with this patch
```
java.lang.IllegalStateException: Failed to read log file
/private/var/folders/nn/82rmvkk568sd8p3p8tb33trw0000gn/T/spark-86867b65-0069-4ef1-b0eb-d8bd258ff5b8/0.
UnsupportedLogVersion: maximum supported log version is v1, but encountered
v99. The log file was produced by a newer version of Spark and cannot be read
by this version. Please upgrade.
at
org.apache.spark.sql.execution.streaming.HDFSMetadataLog.get(HDFSMetadataLog.scala:202)
at
org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite$$anonfun$3$$anonfun$apply$mcV$sp$2.apply(OffsetSeqLogSuite.scala:78)
at
org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite$$anonfun$3$$anonfun$apply$mcV$sp$2.apply(OffsetSeqLogSuite.scala:75)
at
org.apache.spark.sql.test.SQLTestUtils$class.withTempDir(SQLTestUtils.scala:133)
at
org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite.withTempDir(OffsetSeqLogSuite.scala:26)
at
org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite$$anonfun$3.apply$mcV$sp(OffsetSeqLogSuite.scala:75)
at
org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite$$anonfun$3.apply(OffsetSeqLogSuite.scala:75)
at
org.apache.spark.sql.execution.streaming.OffsetSeqLogSuite$$anonfun$3.apply(OffsetSeqLogSuite.scala:75)
at
org.scalatest.Transformer$$anonfun$apply$1.apply$mcV$sp(Transformer.scala:22)
at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
```
## How was this patch tested?
unit tests
Author: Liwei Lin <lwl...@gmail.com>
Closes #17327 from lw-lin/good-msg-2.1.
commit 5fb70831bd3acf7e1d9933986ccce12c3872432b
Author: Shixiong Zhu <shixiong@...>
Date: 2017-03-17T18:12:23Z
[SPARK-19986][TESTS] Make pyspark.streaming.tests.CheckpointTests more
stable
## What changes were proposed in this pull request?
Sometimes, CheckpointTests will hang on a busy machine because the
streaming jobs are too slow and cannot catch up. I observed the scheduled delay
was keeping increasing for dozens of seconds locally.
This PR increases the batch interval from 0.5 seconds to 2 seconds to
generate less Spark jobs. It should make
`pyspark.streaming.tests.CheckpointTests` more stable. I also replaced `sleep`
with `awaitTerminationOrTimeout` so that if the streaming job fails, it will
also fail the test.
## How was this patch tested?
Jenkins
Author: Shixiong Zhu <shixi...@databricks.com>
Closes #17323 from zsxwing/SPARK-19986.
(cherry picked from commit 376d782164437573880f0ad58cecae1cb5f212f2)
Signed-off-by: Tathagata Das <tathagata.das1...@gmail.com>
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
