You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by tcondie <gi...@git.apache.org> on 2016/11/11 18:38:14 UTC
[GitHub] spark pull request #15852: Spark 18187
GitHub user tcondie opened a pull request:
https://github.com/apache/spark/pull/15852
Spark 18187
## What changes were proposed in this pull request?
CompactibleFileStreamLog relys on "compactInterval" to detect a compaction batch. If the "compactInterval" is reset by user, CompactibleFileStreamLog will return wrong answer, resulting data loss. This PR procides a way to check the validity of 'compactInterval', and calculate an appropriate value.
## How was this patch tested?
When restart a stream, we change the 'spark.sql.streaming.fileSource.log.compactInterval' different with the former one.
The primary solution to this issue was given by @uncleGen
Added extensions include an additional metadata field in OffsetSeq and CompactibleFileStreamLog APIs. @zsxwing
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tcondie/spark spark-18187
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/15852.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 #15852
----
commit 65395dddb505f6084db471430da1486d75a77e2a
Author: genmao.ygm <ge...@genmaoygmdemacbook-air.local>
Date: 2016-11-09T08:21:09Z
SPARK-18187: CompactibleFileStreamLog should not rely on "compactInterval" to detect a compaction batch
commit d556933e0f039d661989e07f381aff185c9fac1b
Author: genmao.ygm <ge...@genmaoygmdemacbook-air.local>
Date: 2016-11-09T08:24:53Z
comment update
commit 8b56f70b2dffd69dbc37007e923f3d5a56fce039
Author: genmao.ygm <ge...@genmaoygmdemacbook-air.local>
Date: 2016-11-09T08:34:11Z
revert
commit 4a7e28c4e372caa3b16b979273577bd6aa2c11f3
Author: genmao.ygm <ge...@genmaoygmdemacbook-air.local>
Date: 2016-11-09T08:35:13Z
unit test - compacat metadata log
change compactInterval from 4 to 5
commit 23e1baf454bde511ed1963a27f6492100823d496
Author: genmao.ygm <ge...@genmaoygmdemacbook-air.local>
Date: 2016-11-09T09:34:15Z
bug fix: /zero
commit 7d37e08026eaa1364e8a4fb10fb7cfb93cb51229
Author: Tyson Condie <tc...@gmail.com>
Date: 2016-11-11T00:50:02Z
Merge branch 'SPARK-18187' of https://github.com/uncleGen/spark into spark-18187
commit d3f7bbf32d0debba24853a38eb48bfcdcdb517be
Author: Tyson Condie <tc...@gmail.com>
Date: 2016-11-11T00:52:24Z
Merge branch 'master' of https://github.com/apache/spark into spark-18187
commit 6901eacdddf235db4ba91a0903ce8826978d778a
Author: Tyson Condie <tc...@gmail.com>
Date: 2016-11-11T18:16:41Z
extend offset seq to include metadata
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68816 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68816/consoleFull)** for PR 15852 at commit [`dbd8b67`](https://github.com/apache/spark/commit/dbd8b6788aae2a563ad282931323f0b0c04d4d48).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88352568
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
--- End diff --
Right. Please update the comment accordingly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68808 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68808/consoleFull)** for PR 15852 at commit [`6211faa`](https://github.com/apache/spark/commit/6211faad1ddb1f9e9c0eced2eecd1a0773888aa7).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88357129
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
+ // 3. If there is only one '.compact' file, then we must find a compact interval
+ // that is compatible with (i.e., a divisor of) the previous compact file, and that
+ // faithfully tries to represent the revised default compact interval i.e., is at least
+ // is large if possible.
+ // e.g., if defaultCompactInterval is 5 (and previous compact interval could have
+ // been any 2,3,4,6,12), then a log could be: 11.compact, 12, 13, in which case
+ // will ensure that the new compactInterval = 6 > 5 and (11 + 1) % 6 == 0
+ val compactibleBatchIds = fileManager.list(metadataPath, batchFilesFilter)
+ .filter(f => f.getPath.toString.endsWith(CompactibleFileStreamLog.COMPACT_FILE_SUFFIX))
+ .map(f => pathToBatchId(f.getPath))
+ .sorted
+ .reverse
+
+ // Case 1
+ var interval = defaultCompactInterval
+ if (compactibleBatchIds.length >= 2) {
+ // Case 2
+ val latestCompactBatchId = compactibleBatchIds(0)
+ val previousCompactBatchId = compactibleBatchIds(1)
+ interval = (latestCompactBatchId - previousCompactBatchId).toInt
+ logInfo(s"Compact interval case 2 = $interval")
+ } else if (compactibleBatchIds.length == 1) {
+ // Case 3
+ val latestCompactBatchId = compactibleBatchIds(0).toInt
+ if (latestCompactBatchId + 1 <= defaultCompactInterval) {
+ interval = latestCompactBatchId + 1
+ } else if (defaultCompactInterval < (latestCompactBatchId + 1) / 2) {
+ // Find the first divisor >= default compact interval
+ def properDivisors(n: Int, min: Int) =
+ (min to n/2).filter(i => n % i == 0) :+ n
+
--- End diff --
I would use the following codes to avoid generating the real number sequence:
```Scala
def properDivisors(n: Int, min: Int) = (min to n/2).view.filter(n % _ == 0) :+ n
interval = properDivisors(latestCompactBatchId + 1, defaultCompactInterval).head
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68530 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68530/consoleFull)** for PR 15852 at commit [`6901eac`](https://github.com/apache/spark/commit/6901eacdddf235db4ba91a0903ce8826978d778a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class OffsetSeq(offsets: Seq[Option[Offset]], metadata: Option[String] = None) `
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68639/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68807 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68807/consoleFull)** for PR 15852 at commit [`50207b5`](https://github.com/apache/spark/commit/50207b53eb0afdebac12a7b61ca429b4bf6ed253).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68816/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by tcondie <gi...@git.apache.org>.
Github user tcondie commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88362276
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
--- End diff --
FileStreamSourceLog uses compactInterval in multiple places. Please advise?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68807 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68807/consoleFull)** for PR 15852 at commit [`50207b5`](https://github.com/apache/spark/commit/50207b53eb0afdebac12a7b61ca429b4bf6ed253).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68639 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68639/consoleFull)** for PR 15852 at commit [`efa7022`](https://github.com/apache/spark/commit/efa7022bcc2e8b169c7dd109d878439ac9f058a9).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68541 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68541/consoleFull)** for PR 15852 at commit [`7a71be8`](https://github.com/apache/spark/commit/7a71be8843baf31d501d126b37f33069ac5c2529).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68541 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68541/consoleFull)** for PR 15852 at commit [`7a71be8`](https://github.com/apache/spark/commit/7a71be8843baf31d501d126b37f33069ac5c2529).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68644/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68809 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68809/consoleFull)** for PR 15852 at commit [`13abd7d`](https://github.com/apache/spark/commit/13abd7d0ebd736f000f23e7ca1ede31eb12b41c5).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the issue:
https://github.com/apache/spark/pull/15852
@uncleGen @tcondie thanks for working on this.
My major concern is this approach might disallow changing the `compactInterval` once there were at least two compact files. Should we disallow it? Or as a alternative, what do you think of the approach taken in https://github.com/apache/spark/pull/15828?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68541/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88580974
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala ---
@@ -330,34 +331,51 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
startedTest.foreach { action =>
logInfo(s"Processing test stream action: $action")
action match {
- case StartStream(trigger, triggerClock) =>
+ case StartStream(trigger, triggerClock, additionalConfs) =>
verify(currentStream == null, "stream already running")
verify(triggerClock.isInstanceOf[SystemClock]
|| triggerClock.isInstanceOf[StreamManualClock],
"Use either SystemClock or StreamManualClock to start the stream")
if (triggerClock.isInstanceOf[StreamManualClock]) {
manualClockExpectedTime = triggerClock.asInstanceOf[StreamManualClock].getTimeMillis()
}
- lastStream = currentStream
- currentStream =
- spark
- .streams
- .startQuery(
- None,
- Some(metadataRoot),
- stream,
- sink,
- outputMode,
- trigger = trigger,
- triggerClock = triggerClock)
- .asInstanceOf[StreamExecution]
- currentStream.microBatchThread.setUncaughtExceptionHandler(
- new UncaughtExceptionHandler {
- override def uncaughtException(t: Thread, e: Throwable): Unit = {
- streamDeathCause = e
- }
+
+ val resetConfValues = mutable.Map[String, Option[String]]()
+ try {
+ additionalConfs.foreach(pair => {
+ val value =
+ if (spark.conf.contains(pair._1)) Some(spark.conf.get(pair._1)) else None
+ resetConfValues(pair._1) = value
+ spark.conf.set(pair._1, pair._2)
})
+ lastStream = currentStream
+ currentStream =
+ spark
+ .streams
+ .startQuery(
+ None,
+ Some(metadataRoot),
+ stream,
+ sink,
+ outputMode,
+ trigger = trigger,
+ triggerClock = triggerClock)
+ .asInstanceOf[StreamExecution]
+ currentStream.microBatchThread.setUncaughtExceptionHandler(
+ new UncaughtExceptionHandler {
+ override def uncaughtException(t: Thread, e: Throwable): Unit = {
+ streamDeathCause = e
+ }
+ })
+ } finally {
+ // Rollback previous configuration values
+ resetConfValues.foreach {
--- End diff --
The reset logic should be at the end of the method. Otherwise, it will change confs during a query is running.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88353911
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala ---
@@ -330,14 +331,17 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
startedTest.foreach { action =>
logInfo(s"Processing test stream action: $action")
action match {
- case StartStream(trigger, triggerClock) =>
+ case StartStream(trigger, triggerClock, pairs) =>
verify(currentStream == null, "stream already running")
verify(triggerClock.isInstanceOf[SystemClock]
|| triggerClock.isInstanceOf[StreamManualClock],
"Use either SystemClock or StreamManualClock to start the stream")
if (triggerClock.isInstanceOf[StreamManualClock]) {
manualClockExpectedTime = triggerClock.asInstanceOf[StreamManualClock].getTimeMillis()
}
+
+ pairs.foreach(pair => spark.conf.set(pair._1, pair._2))
--- End diff --
You need to also change the confs back at the end of this method to avoid affecting other tests sharing the same SparkSession.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68809/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68808 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68808/consoleFull)** for PR 15852 at commit [`6211faa`](https://github.com/apache/spark/commit/6211faad1ddb1f9e9c0eced2eecd1a0773888aa7).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68816 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68816/consoleFull)** for PR 15852 at commit [`dbd8b67`](https://github.com/apache/spark/commit/dbd8b6788aae2a563ad282931323f0b0c04d4d48).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: [Spark-18187] [SQL] CompactibleFileStreamLog should not ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/15852
LGTM. Thanks! Merging to master and 2.1.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68827/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68808/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68533/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/15852
(Update the title please; see others for format)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88135983
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -38,8 +38,9 @@ class FileStreamSourceLog(
import CompactibleFileStreamLog._
// Configurations about metadata compaction
- protected override val compactInterval =
+ protected override def defaultCompactInterval: Int =
--- End diff --
nit: def => val
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on the issue:
https://github.com/apache/spark/pull/15852
LGTM overall. If this is accepted, then i will close #15827
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88363586
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
--- End diff --
> FileStreamSourceLog uses compactInterval in multiple places. Please advise?
Sorry. My bad. Didn't notice that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r87936909
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
--- End diff --
The log garbage operation is controlled by 'spark.sql.streaming.fileSource.log.deletion'. When it is 'false', there may be two or more '.compact' files
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88353713
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala ---
@@ -161,7 +161,8 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
/** Starts the stream, resuming if data has already been processed. It must not be running. */
case class StartStream(
trigger: Trigger = ProcessingTime(0),
- triggerClock: Clock = new SystemClock)
+ triggerClock: Clock = new SystemClock,
+ pairs: mutable.Map[String, String] = mutable.Map.empty)
--- End diff --
nit: it's better to rename `pairs` to `additionalConfs` for readability.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68539 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68539/consoleFull)** for PR 15852 at commit [`4f8ff16`](https://github.com/apache/spark/commit/4f8ff1649cf785e79ae4c8a57f78ef82c72f83cf).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68795/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88534136
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -245,4 +284,25 @@ object CompactibleFileStreamLog {
def nextCompactionBatchId(batchId: Long, compactInterval: Long): Long = {
(batchId + compactInterval + 1) / compactInterval * compactInterval - 1
}
+
+ /**
+ * Derives a compact interval from the latest compact batch id and
+ * a default compact interval.
+ */
+ def deriveCompactInterval(defaultInterval: Int, latestCompactBatchId: Int) : Int = {
+ if (latestCompactBatchId + 1 <= defaultInterval) {
+ latestCompactBatchId + 1
+ } else if (defaultInterval < (latestCompactBatchId + 1) / 2) {
+ // Find the first divisor >= default compact interval
+ def properDivisors(min: Int, n: Int) =
+ (min to n/2).filter(i => n % i == 0) :+ n
--- End diff --
nit: Use `(min to n/2).view.filter(i => n % i == 0) :+ n` so that it will stop when finding the first element.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68639 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68639/consoleFull)** for PR 15852 at commit [`efa7022`](https://github.com/apache/spark/commit/efa7022bcc2e8b169c7dd109d878439ac9f058a9).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68539/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68539 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68539/consoleFull)** for PR 15852 at commit [`4f8ff16`](https://github.com/apache/spark/commit/4f8ff1649cf785e79ae4c8a57f78ef82c72f83cf).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88354869
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
+ // 3. If there is only one '.compact' file, then we must find a compact interval
+ // that is compatible with (i.e., a divisor of) the previous compact file, and that
+ // faithfully tries to represent the revised default compact interval i.e., is at least
+ // is large if possible.
+ // e.g., if defaultCompactInterval is 5 (and previous compact interval could have
+ // been any 2,3,4,6,12), then a log could be: 11.compact, 12, 13, in which case
+ // will ensure that the new compactInterval = 6 > 5 and (11 + 1) % 6 == 0
+ val compactibleBatchIds = fileManager.list(metadataPath, batchFilesFilter)
+ .filter(f => f.getPath.toString.endsWith(CompactibleFileStreamLog.COMPACT_FILE_SUFFIX))
+ .map(f => pathToBatchId(f.getPath))
+ .sorted
+ .reverse
+
+ // Case 1
+ var interval = defaultCompactInterval
+ if (compactibleBatchIds.length >= 2) {
+ // Case 2
+ val latestCompactBatchId = compactibleBatchIds(0)
+ val previousCompactBatchId = compactibleBatchIds(1)
+ interval = (latestCompactBatchId - previousCompactBatchId).toInt
+ logInfo(s"Compact interval case 2 = $interval")
+ } else if (compactibleBatchIds.length == 1) {
+ // Case 3
+ val latestCompactBatchId = compactibleBatchIds(0).toInt
+ if (latestCompactBatchId + 1 <= defaultCompactInterval) {
+ interval = latestCompactBatchId + 1
+ } else if (defaultCompactInterval < (latestCompactBatchId + 1) / 2) {
+ // Find the first divisor >= default compact interval
+ def properDivisors(n: Int, min: Int) =
+ (min to n/2).filter(i => n % i == 0) :+ n
+
+ interval = properDivisors(latestCompactBatchId + 1, defaultCompactInterval)(0)
+ }
+ else {
+ // default compact interval > than any divisor other than latest compact id
+ interval = latestCompactBatchId + 1
+ }
+ logInfo(s"Compact interval case 3 = $interval")
--- End diff --
nit: it's better to include all infos in the log, like
```
Set the compact interval to XXX [latestCompactBatchId: XXX, defaultCompactInterval: XXX]
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68538 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68538/consoleFull)** for PR 15852 at commit [`d216acf`](https://github.com/apache/spark/commit/d216acfe8a263fb1983e78183366b7969c040a28).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68644 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68644/consoleFull)** for PR 15852 at commit [`24e3617`](https://github.com/apache/spark/commit/24e36177e1eb24e7b250cb5356b47c0507e96d68).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88353628
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala ---
@@ -161,7 +161,8 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
/** Starts the stream, resuming if data has already been processed. It must not be running. */
case class StartStream(
trigger: Trigger = ProcessingTime(0),
- triggerClock: Clock = new SystemClock)
+ triggerClock: Clock = new SystemClock,
+ pairs: mutable.Map[String, String] = mutable.Map.empty)
--- End diff --
nit: use `Map[String, String]` instead since it won't be changed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68807/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68827 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68827/consoleFull)** for PR 15852 at commit [`6537e7c`](https://github.com/apache/spark/commit/6537e7cf6186f6fcd4a5d21832572560cbd72895).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88354703
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
+ // 3. If there is only one '.compact' file, then we must find a compact interval
+ // that is compatible with (i.e., a divisor of) the previous compact file, and that
+ // faithfully tries to represent the revised default compact interval i.e., is at least
+ // is large if possible.
+ // e.g., if defaultCompactInterval is 5 (and previous compact interval could have
+ // been any 2,3,4,6,12), then a log could be: 11.compact, 12, 13, in which case
+ // will ensure that the new compactInterval = 6 > 5 and (11 + 1) % 6 == 0
+ val compactibleBatchIds = fileManager.list(metadataPath, batchFilesFilter)
+ .filter(f => f.getPath.toString.endsWith(CompactibleFileStreamLog.COMPACT_FILE_SUFFIX))
+ .map(f => pathToBatchId(f.getPath))
+ .sorted
+ .reverse
+
+ // Case 1
+ var interval = defaultCompactInterval
+ if (compactibleBatchIds.length >= 2) {
+ // Case 2
+ val latestCompactBatchId = compactibleBatchIds(0)
+ val previousCompactBatchId = compactibleBatchIds(1)
+ interval = (latestCompactBatchId - previousCompactBatchId).toInt
+ logInfo(s"Compact interval case 2 = $interval")
+ } else if (compactibleBatchIds.length == 1) {
+ // Case 3
+ val latestCompactBatchId = compactibleBatchIds(0).toInt
--- End diff --
Could you pull this branch into a method in `object CompactibleFileStreamLog` so that it's easy to write tests for this complicated logic? And please add tests as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68795 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68795/consoleFull)** for PR 15852 at commit [`82adb39`](https://github.com/apache/spark/commit/82adb39b28638985160a538acc7bf5fd8420a038).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88581282
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLogSuite.scala ---
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.test.SharedSQLContext
+
+class CompactibleFileStreamLogSuite extends SparkFunSuite with SharedSQLContext {
--- End diff --
nit: SharedSQLContext is not needed. Please remove it to avoid creating SQLContext.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68533 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68533/consoleFull)** for PR 15852 at commit [`96d2dfe`](https://github.com/apache/spark/commit/96d2dfe97b51df3d31923791b10bb54778399be2).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68533 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68533/consoleFull)** for PR 15852 at commit [`96d2dfe`](https://github.com/apache/spark/commit/96d2dfe97b51df3d31923791b10bb54778399be2).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r87941243
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
+ // 3. If there is only one '.compact' file, then we must find a compact interval
+ // that is compatible with (i.e., a divisor of) the previous compact file, and that
+ // faithfully tries to represent the revised default compact interval i.e., is at least
+ // is large if possible.
+ // e.g., if defaultCompactInterval is 5 (and previous compact interval could have
+ // been any 2,3,4,6,12), then a log could be: 11.compact, 12, 13, in which case
+ // will ensure that the new compactInterval = 6 > 5 and (11 + 1) % 6 == 0
+ val compactibleBatchIds = fileManager.list(metadataPath, batchFilesFilter)
+ .filter(f => f.getPath.toString.endsWith(CompactibleFileStreamLog.COMPACT_FILE_SUFFIX))
+ .map(f => pathToBatchId(f.getPath))
+ .sorted
+ .reverse
+
+ // Case 1
+ var interval = defaultCompactInterval
+ if (compactibleBatchIds.length >= 2) {
+ // Case 2
+ val latestCompactBatchId = compactibleBatchIds(0)
+ val previousCompactBatchId = compactibleBatchIds(1)
+ interval = (latestCompactBatchId - previousCompactBatchId).toInt
+ logInfo(s"Compact interval case 2 = $interval")
+ } else if (compactibleBatchIds.length == 1) {
+ // Case 3
+ val latestCompactBatchId = compactibleBatchIds(0).toInt
+ if (latestCompactBatchId + 1 <= defaultCompactInterval) {
+ interval = latestCompactBatchId + 1
+ } else if (defaultCompactInterval < (latestCompactBatchId + 1) / 2) {
+ // Find the first divisor >= default compact interval
+ def properDivisors(n: Int, min: Int) =
+ (min to n/2).filter(i => n % i == 0) :+ n
+
--- End diff --
to => until ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68644 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68644/consoleFull)** for PR 15852 at commit [`24e3617`](https://github.com/apache/spark/commit/24e36177e1eb24e7b250cb5356b47c0507e96d68).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68795 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68795/consoleFull)** for PR 15852 at commit [`82adb39`](https://github.com/apache/spark/commit/82adb39b28638985160a538acc7bf5fd8420a038).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68530 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68530/consoleFull)** for PR 15852 at commit [`6901eac`](https://github.com/apache/spark/commit/6901eacdddf235db4ba91a0903ce8826978d778a).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68538 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68538/consoleFull)** for PR 15852 at commit [`d216acf`](https://github.com/apache/spark/commit/d216acfe8a263fb1983e78183366b7969c040a28).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88353490
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala ---
@@ -896,32 +898,35 @@ class FileStreamSourceSuite extends FileStreamSourceTest {
}
}
- test("compacat metadata log") {
+ test("compact interval metadata log") {
val _sources = PrivateMethod[Seq[Source]]('sources)
val _metadataLog = PrivateMethod[FileStreamSourceLog]('metadataLog)
- def verify(execution: StreamExecution)
- (batchId: Long, expectedBatches: Int): Boolean = {
+ def verify(execution: StreamExecution, batchId: Long,
+ expectedBatches: Int, expectedCompactInterval: Int): Boolean = {
--- End diff --
nit: the correct style should be
```Scala
def verify(
execution: StreamExecution,
batchId: Long,
expectedBatches: Int,
expectedCompactInterval: Int): Boolean = {
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark 18187
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68530/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68538/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88114440
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
--- End diff --
nit: please change `protected` to `private` since this should not be used by subclassed now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68809 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68809/consoleFull)** for PR 15852 at commit [`13abd7d`](https://github.com/apache/spark/commit/13abd7d0ebd736f000f23e7ca1ede31eb12b41c5).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88581369
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -245,4 +284,25 @@ object CompactibleFileStreamLog {
def nextCompactionBatchId(batchId: Long, compactInterval: Long): Long = {
(batchId + compactInterval + 1) / compactInterval * compactInterval - 1
}
+
+ /**
+ * Derives a compact interval from the latest compact batch id and
+ * a default compact interval.
+ */
+ def deriveCompactInterval(defaultInterval: Int, latestCompactBatchId: Int) : Int = {
+ if (latestCompactBatchId + 1 <= defaultInterval) {
+ latestCompactBatchId + 1
+ } else if (defaultInterval < (latestCompactBatchId + 1) / 2) {
+ // Find the first divisor >= default compact interval
+ def properDivisors(min: Int, n: Int) =
+ (min to n/2).view.filter(i => n % i == 0) :+ n
+
+ properDivisors(defaultInterval, latestCompactBatchId + 1).head
+ }
+ else {
--- End diff --
nit: move `else {` to the above line.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: [Spark-18187] [SQL] CompactibleFileStreamLog shou...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/15852
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15852
**[Test build #68827 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68827/consoleFull)** for PR 15852 at commit [`6537e7c`](https://github.com/apache/spark/commit/6537e7cf6186f6fcd4a5d21832572560cbd72895).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88352982
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
+ // 3. If there is only one '.compact' file, then we must find a compact interval
+ // that is compatible with (i.e., a divisor of) the previous compact file, and that
+ // faithfully tries to represent the revised default compact interval i.e., is at least
+ // is large if possible.
+ // e.g., if defaultCompactInterval is 5 (and previous compact interval could have
+ // been any 2,3,4,6,12), then a log could be: 11.compact, 12, 13, in which case
+ // will ensure that the new compactInterval = 6 > 5 and (11 + 1) % 6 == 0
+ val compactibleBatchIds = fileManager.list(metadataPath, batchFilesFilter)
+ .filter(f => f.getPath.toString.endsWith(CompactibleFileStreamLog.COMPACT_FILE_SUFFIX))
+ .map(f => pathToBatchId(f.getPath))
+ .sorted
+ .reverse
+
+ // Case 1
+ var interval = defaultCompactInterval
+ if (compactibleBatchIds.length >= 2) {
+ // Case 2
+ val latestCompactBatchId = compactibleBatchIds(0)
+ val previousCompactBatchId = compactibleBatchIds(1)
+ interval = (latestCompactBatchId - previousCompactBatchId).toInt
+ logInfo(s"Compact interval case 2 = $interval")
--- End diff --
nit: Please use a better message, like
```
Set the compact interval to XXX [the previous two batch Ids: XXX, XXX]
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15852: Spark-18187 [SQL] CompactibleFileStreamLog should...
Posted by tcondie <gi...@git.apache.org>.
Github user tcondie commented on a diff in the pull request:
https://github.com/apache/spark/pull/15852#discussion_r88359775
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -63,7 +63,60 @@ abstract class CompactibleFileStreamLog[T <: AnyRef : ClassTag](
protected def isDeletingExpiredLog: Boolean
- protected def compactInterval: Int
+ protected def defaultCompactInterval: Int
+
+ protected final lazy val compactInterval: Int = {
+ // SPARK-18187: "compactInterval" can be set by user via defaultCompactInterval.
+ // If there are existing log entries, then we should ensure a compatible compactInterval
+ // is used, irrespective of the defaultCompactInterval. There are three cases:
+ //
+ // 1. If there is no '.compact' file, we can use the default setting directly.
+ // 2. If there are two or more '.compact' files, we use the interval of patch id suffix with
+ // '.compact' as compactInterval. It is unclear whether this case will ever happen in the
+ // current code, since only the latest '.compact' file is retained i.e., other are garbage
+ // collected.
+ // 3. If there is only one '.compact' file, then we must find a compact interval
+ // that is compatible with (i.e., a divisor of) the previous compact file, and that
+ // faithfully tries to represent the revised default compact interval i.e., is at least
+ // is large if possible.
+ // e.g., if defaultCompactInterval is 5 (and previous compact interval could have
+ // been any 2,3,4,6,12), then a log could be: 11.compact, 12, 13, in which case
+ // will ensure that the new compactInterval = 6 > 5 and (11 + 1) % 6 == 0
+ val compactibleBatchIds = fileManager.list(metadataPath, batchFilesFilter)
+ .filter(f => f.getPath.toString.endsWith(CompactibleFileStreamLog.COMPACT_FILE_SUFFIX))
+ .map(f => pathToBatchId(f.getPath))
+ .sorted
+ .reverse
+
+ // Case 1
+ var interval = defaultCompactInterval
+ if (compactibleBatchIds.length >= 2) {
+ // Case 2
+ val latestCompactBatchId = compactibleBatchIds(0)
+ val previousCompactBatchId = compactibleBatchIds(1)
+ interval = (latestCompactBatchId - previousCompactBatchId).toInt
+ logInfo(s"Compact interval case 2 = $interval")
--- End diff --
This was debugging info that I was going to remove.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15852: Spark-18187 [SQL] CompactibleFileStreamLog should not us...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15852
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org