text:
How much memory should we use for broker and namesrv when using cluster mode?

First, 3Q for your work, after research, we decided to use rocketmq, but we met some problem when deploying rocketmq cluster.
RT：
How much memory should we use for broker and namesrv when using cluster mode?
I noticed the default memory is set up to 4G of namesrv, 8G of broker.
And direct memory is set up to 15G like this:
MaxDirectMemorySize=15g
Is that mean I should use 30G for every namesrv, broker? that looks unreasonable.
And, is some advice about memory? 3Q so much.

Hello,the machine configuration needs to depend on your messages.
there is some configuration we are using now:
CPU:8C
Memory:16G
DISK:1T
ASYNC_FLUSH and SYNC_MASTER is adviced.
and you can extend the server to meet your needs.

Hello,the machine configuration needs to depend on your messages.
there is some configuration we are using now:
CPU:8C

Memory:16G
How much should we set about the jvm param MaxDirectMemorySize? default value is 15G.

DISK:1T
ASYNC_FLUSH and SYNC_MASTER is adviced.
and you can extend the server to meet your needs.

I think this is the config  and machine-resource for the broker node. How about advices about config and machine-resource for namesrv machine?

Hello,the machine configuration needs to depend on your messages.
there is some configuration we are using now:
CPU:8C
Memory:16G
DISK:1T
ASYNC_FLUSH and SYNC_MASTER is adviced.
and you can extend the server to meet your needs.

And how about the benchmark when using these resource and config?

Hello,the machine configuration needs to depend on your messages.
there is some configuration we are using now:
CPU:8C
Memory:16G
DISK:1T
ASYNC_FLUSH and SYNC_MASTER is adviced.
and you can extend the server to meet your needs.

I still have some questions, 3Q~
1.And how about the benchmark when using these resource and config?
2.Is disk ssd?
This is our play, is this ok?

namesrv manage broker info and topic info and store all info in the memory,so i think 16g mem is more
suitable,but if your topic and queue are relatively few,8g is OK。for 3 broker, i think 2 namesrv is OK.
and only master may lost messages.
so 2M - 2S is adviced。
There are some configuration we are using now
2 namesrv: 8cpu and 16g
4 cluster,per cluster have 2 master and 2slave.
broker : 2M-2S.
3E messages a day.

3Q so much~~

namesrv manage broker info and topic info and store all info in the memory,so i think 16g mem is more
suitable,but if your topic and queue are relatively few,8g is OK。for 3 broker, i think 2 namesrv is OK.
and only master may lost messages.
I have some confused, the namesrv just save broker and topic info, I think they cost memory little, is 16G is too large? I think even 8G is also too large, is this right?

About lost message when only using master, is this happen only when master disk is wrong? I mean if we don't consider the disk broken, the rocketmq won't lose message, is this right?
If we use slave feature, is it will have some influences about the qps? And How big is the impact?

so 2M - 2S is adviced。
There are some configuration we are using now
2 namesrv: 8cpu and 16g
4 cluster,per cluster have 2 master and 2slave.
broker : 2M-2S.
3E messages a day.
And what about the qps?

3Q~

namesrv manage broker info and topic info and store all info in the memory,so i think 16g mem is more
suitable,but if your topic and queue are relatively few,8g is OK。for 3 broker, i think 2 namesrv is OK.
and only master may lost messages.
so 2M - 2S is adviced。
There are some configuration we are using now
2 namesrv: 8cpu and 16g
4 cluster,per cluster have 2 master and 2slave.
broker : 2M-2S.
3E messages a day.

3Q for your answer, but I still have some confused.
1.the namesrv just save broker and topic info, I think they cost memory little, is 16G is too large? I think even 8G is also too large, is this right?
2.About lost message when only using master, is this happen only when master disk is wrong? I mean if we don't consider the disk broken, the rocketmq won't lose message, is this right?
3.If we use slave feature, is it will have some influences about the qps? And How big is the impact?
4.And what about the qps when using the config as you advice?

@hepyu
1.the namesrv just save broker and topic info, I think they cost memory little, is 16G is too large? I think even 8G is also too large, is this right?
it depends on your topics, 8g is OK.but 2 namesrv is enough.
2.About lost message when only using master, is this happen only when master disk is wrong? I mean if we don't consider the disk broken, the rocketmq won't lose message, is this right?
if you use async_flush,when broker crash and message hasn't flush to the disk,then the message lost.
3.If we use slave feature, is it will have some influences about the qps? And How big is the impact?
yes ,but it can gurantee the message reliability
4.And what about the qps when using the config as you advice?
1.pub message: message size:2K  and tps is 40000
2.consume message:messages size 2k and tps is 60000.
message size affects  TPS.

@hepyu
1.the namesrv just save broker and topic info, I think they cost memory little, is 16G is too large? I think even 8G is also too large, is this right?
it depends on your topics, 8g is OK.but 2 namesrv is enough.

How could I understand "it depends on your topics", is this mean if topics/queues/brokers are too many, the infos will cost much more memory? But will they cost 8g?
In other words, I mean , 8g memory of namesrv could hold how many topics/queues/brokers?
Just to control the money if biz accept...

2.About lost message when only using master, is this happen only when master disk is wrong? I mean if we don't consider the disk broken, the rocketmq won't lose message, is this right?
if you use async_flush,when broker crash and message hasn't flush to the disk,then the message lost.

yeah, it is, but I think the broker would run stablely at most of the time, if it is and we could accept the little crase case about broker, is this make some sence?
And how about the broker crase rate?
And What about the disk type? namesrv and broker are all ssd?

3.If we use slave feature, is it will have some influences about the qps? And How big is the impact?
yes ,but it can gurantee the message reliability

Yes, I agree. Just to control the money if biz accept...

4.And what about the qps when using the config as you advice?
1.pub message: message size:2K and tps is 40000
2.consume message:messages size 2k and tps is 60000.
message size affects TPS.

Is this tps running under these resources as you adviced before?
4 cluster,per cluster have 2 master and 2slave.
broker : 2M-2S.
namesrv-machine: 8cpu and 8g
broker-machine: 8cpu and 16g

@hepyu
SSD will improve broker performance significantly and we  use HDD in production。
1.rocketmq support 10W+ topics,8g is enough in pre-application.
2.you  must run benchmark to test it.'

authors: Cicizz hepyu


text:
[ISSUE #5738] simplify the implementation of updateFaultItem

Make sure set the target branch to develop
What is the purpose of the change
fix #5738
Brief changelog
XX
Verifying this change
XXXX
Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

[x] Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
[x] Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
[x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
[x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
[x] Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test  to make sure integration-test pass.
[ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.

Codecov Report

Merging #5743 (59f2b61) into develop (e77d067) will increase coverage by 0.06%.
The diff coverage is 100.00%.

@@              Coverage Diff              @@
##             develop    #5743      +/-   ##
=============================================
+ Coverage      42.42%   42.49%   +0.06%
- Complexity      8204     8211       +7
=============================================
  Files           1060     1060
  Lines          73176    73168       -8
  Branches        9606     9604       -2
=============================================
+ Hits           31047    31094      +47
+ Misses         38210    38159      -51
+ Partials        3919     3915       -4

Impacted Files
Coverage Δ

...etmq/client/latency/LatencyFaultToleranceImpl.java
52.17% <100.00%> (+2.82%)
:arrow_up:

...ache/rocketmq/common/stats/MomentStatsItemSet.java
39.13% <0.00%> (-4.35%)
:arrow_down:

...tmq/remoting/protocol/body/ConsumerConnection.java
95.83% <0.00%> (-4.17%)
:arrow_down:

.../apache/rocketmq/controller/ControllerManager.java
73.11% <0.00%> (-2.16%)
:arrow_down:

...mq/client/impl/consumer/RebalanceLitePullImpl.java
69.86% <0.00%> (-1.37%)
:arrow_down:

...cketmq/broker/schedule/ScheduleMessageService.java
56.91% <0.00%> (-1.14%)
:arrow_down:

...che/rocketmq/acl/plain/PlainPermissionManager.java
78.22% <0.00%> (-0.54%)
:arrow_down:

...q/controller/impl/manager/ReplicasInfoManager.java
61.23% <0.00%> (-0.37%)
:arrow_down:

...nt/impl/consumer/ConsumeMessageOrderlyService.java
49.82% <0.00%> (-0.36%)
:arrow_down:

...rocketmq/client/impl/factory/MQClientInstance.java
46.67% <0.00%> (-0.15%)
:arrow_down:

... and 16 more

:mega: We’re building smart automated test selection to slash your CI/CD build times. Learn more

source: gharchive/pull-request

authors: codecov-commenter xiaoyifang

repo: apache/rocketmq

license_type: permissive

license_source: github-api


text:
[Improve] When starting, check whether the environment variable "SEATUNNEL_HOME" exists or not.

Purpose of this pull request
When starting, check whether the environment variable "SEATUNNEL_HOME" exists or not.

Check list

[ ] Code changed are covered with tests, or it does not need tests for reason:
[ ] If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
[ ] If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs

This is test result.

repo: apache/seatunnel-web


text:
Where is the new operator's manual?

Please answer these questions before submitting your issue.

Why do you submit this issue?
[ ] Question or discussion
[ ] Bug
[ ] Requirement
[ ] Feature or performance improvement

Question

What do you want to know?

Bug

Which version of SkyWalking, OS, and JRE?

Which company or project?

What happened?
If possible, provide a way to reproduce the error. e.g. demo application, component version.

Requirement or improvement

Please describe your requirements or improvement suggestions.

Please read the website page before you submit this kind of issue.
https://github.com/apache/skywalking-swck
https://skywalking.apache.org/docs/

Please read the website page before you submit this kind of issue.
https://github.com/apache/skywalking-swck
https://skywalking.apache.org/docs/

提交此类问题之前，请阅读网站页面。
https://github.com/apache/skywalking-swck
https://skywalking.apache.org/docs/

There used to be principle pictures, but now it's just a brief introduction. Are there no configuration items in the project?
Is this the so-called new operator's manual? It is recommended to refer to Seata's rookie Handbook.

提交此类问题之前，请阅读网站页面。
https://github.com/apache/skywalking-swck
https://skywalking.apache.org/docs/

There used to be principle pictures, but now it's just a brief introduction. Are there no configuration items in the project?
Is this the so-called new operator's manual? It is recommended to refer to Seata's rookie Handbook.

You are asking questions from wrong perspective. We trust our PMC and committer team. They think it is enough, then, it is.
If you want to contribute more details, we are welcome.

You are asking questions from wrong perspective. We trust our PMC and committer team. They think it is enough, then, it is.
If you want to contribute more details, we are welcome.

I am a new user. If I want to use a service, I will first describe how to operate in the specific configuration items of the server and project side, rather than the general description. Therefore, there is no wrong point of view in my question.
If the founder of the service thinks that it is enough, then as a user, I have nothing to say.

I am a new user. If I want to use a service, I will first describe how to operate in the specific configuration items of the server and project side, rather than the general description. Therefore, there is no wrong point of view in my question.
If the founder of the service thinks that it is enough, then as a user, I have nothing to say.

source: gharchive/issue

repo: apache/skywalking

license_source: bigquery


text:
[Feature] set instance properties in json format: agent.instance_properties_json

Search before asking

[X] I had searched in the issues and found no similar feature requirement.

Description

I find some configurations are in configurations.md , but not in the agent.config , such as agent.trace_segment_ref_limit_per_span 、agent.instance_properties[key]=value . Otherwise , the configuration plugin.dubbo.consumer_provider_length_threshold  in the agent.config is not the same as plugin.dubbo.provider_arguments_length_threshold in the configurations.md.

@dashanji can you open a pull request to fix this? For agent.instance_properties[xxx], I think we also need to migrate to the format of agent.instance_properties_json={}.  plugin.dubbo.consumer_provider_length_threshold should be typo of plugin.dubbo.consumer_length_threshold
Originally posted by @kezhenxu94 in https://github.com/apache/skywalking/issues/7470#issuecomment-922471256
Use case
For containerized scenarios, we should be able to set the agent.instance_properties easily via environment variables, instead of modifying the config file.
Related issues
https://github.com/apache/skywalking/pull/7470#issuecomment-922471256
Are you willing to submit a PR?

[X] Yes I am willing to submit a PR!

Code of Conduct

[X] I agree to follow this project's Code of Conduct

Anyone who wants to take this, please reply and I can assign.

@kezhenxu94 Hi, I would like to take this up, can you please assign it to me? Since I am new to the project, can you also please give a bit more context about the task, and the files which I should start looking at?

@kezhenxu94 Hi, I would like to take this up, can you please assign it to me? Since I am new to the project, can you also please give a bit more context about the task, and the files which I should start looking at?

Hi @deepto98 , thanks for the interest, the rationale behind this task is that, for containerized scenarios, the style of agent.instance_properties[xxx]=xxx is hard (if not impossible) to modify, using json style like agent.instance_properties_json={xxx:yyy}, the users can modify via environment variables.

https://github.com/apache/skywalking-java/pull/67 supported


text:
MGet and Get method didn't support alias index.

Use the index name instead of the alias name when using get method in Elasticsearch client.
Use Noop get instead of MGet because of the ids maybe in different indexes.
#2860

Coverage decreased (-0.01%) to 17.229% when pulling dfb7fec18500376c9c18d24f3283a7524df72630 on peng-yongsheng:fixed/2860 into dbb8bd8a7fcd6efb079bd530628bbb8a62266d52 on apache:master.

Coverage decreased (-0.01%) to 17.229% when pulling dfb7fec18500376c9c18d24f3283a7524df72630 on peng-yongsheng:fixed/2860 into dbb8bd8a7fcd6efb079bd530628bbb8a62266d52 on apache:master.

authors: coveralls peng-yongsheng


text:
Fix lib names in analysis-extras module README.md

Not creating a JIRA issue, since this is something minor that wouldn't make it into release notes.
Description
It seems like the lucene-analyzers .jar files were renamed at somepoint, and the README docs are out-of-date.
solr@2bbdbb1f5b80:/opt/solr/modules/analysis-extras/lib$ ls
icu4j-70.1.jar                        lucene-analysis-smartcn-9.4.2.jar  morfologik-stemming-2.1.9.jar
lucene-analysis-icu-9.4.2.jar         lucene-analysis-stempel-9.4.2.jar  morfologik-ukrainian-search-4.9.1.jar
lucene-analysis-morfologik-9.4.2.jar  morfologik-fsa-2.1.9.jar           opennlp-tools-1.9.4.jar
lucene-analysis-opennlp-9.4.2.jar     morfologik-polish-2.1.9.jar        solr-analysis-extras-9.2.1.jar

They used to be called lucene-analyzers-*, but are now lucene-analysis-*.
Solution
I updated the names in the README.
Tests
Please describe the tests you've developed or run to confirm this patch implements the feature or solves the problem.
Checklist
Please review the following and check all that apply:

[x] I have reviewed the guidelines for How to Contribute and my code conforms to the standards described there to the best of my ability.
[x] I have created a Jira issue and added the issue ID to my pull request title.

Skipped since minor

[x] I have given Solr maintainers access to contribute to my PR branch. (optional but recommended)
[x] I have developed this patch against the main branch.
[NA] I have run ./gradlew check.
[NA] I have added tests for my changes.
[x] I have added documentation for the Reference Guide

Nice catch.   Are there any other changes you want to make on this PR?

Thanks! I think that's all; this is where the docs link to, so fixing these should resolve my problem!

authors: cdrini epugh

repo: apache/solr


text:
SOLR-17597: Fix CaffeineCache.put() ramBytes decrement

https://issues.apache.org/jira/browse/SOLR-17597

This probably does not warrant a CHANGES.txt entry, since it doesn't impact any existing/default implementation or invocation of CaffeineCache.
NOTE: the second (V oldValue) arg to the private method CaffeineCache.recordRamBytes() is always null as now invoked, and could be removed. I will remove this unless anyone objects.

This is a trivial change with no practical impact on the vast majority of users. I plan to merge this tomorrow, pending any feedback.


text:
[SPARK-49305] Revise Spark Operator docker image

What changes were proposed in this pull request?
This PR aims to revise Spark Operator docker image.
Why are the changes needed?

Environment Variables.

Remove APP_VERSION because image tag and metadata contain the version information already.
Rename OPERATOR_JAR to SPARK_OPERATOR_JAR like SPARK_OPERATOR_HOME and SPARK_OPERATOR_WORK_DIR.

Remove version string from jar name to simplify.

BEFORE
$ docker run -it --rm spark-kubernetes-operator:0.1.0-SNAPSHOT ls -al | tail -n1
-rw-r--r-- 1 spark spark 154955452 Aug 19 19:29 spark-kubernetes-operator-0.1.0-SNAPSHOT-all.jar

AFTER
$ docker run -it --rm spark-kubernetes-operator:0.1.0-SNAPSHOT ls -al | tail -n1
-rw-r--r-- 1 spark spark 154955452 Aug 19 19:18 spark-kubernetes-operator.jar

Does this PR introduce any user-facing change?
No.
How was this patch tested?
Pass the CIs.
Was this patch authored or co-authored using generative AI tooling?
No.

Thank you, @viirya . Merged to main.

repo: apache/spark-kubernetes-operator


text:
[SPARK-13304] [SQL] Fix worst case of broadcast join of two ints

If the two join columns have the same value, the hash code of them will be (a ^ b), which is 0, then the HashMap will be very very slow.
This PR will rotate the second int to avoid this case. In theory, it's still have the possibility that has lots of collisions, the pattern will be (1, 131072), (2, 131073) ... (n, n + 131072).
This PR also added some micro benchmark, and updated the results for broadcast hash joins.

Test build #51201 has started for PR 11188 at commit 1c0ee96.

Test build #51201 has finished for PR 11188 at commit 1c0ee96.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test FAILed.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51201/
Test FAILed.

Test build #51571 has started for PR 11188 at commit f6416a6.

Test build #51571 has finished for PR 11188 at commit f6416a6.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test FAILed.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51571/
Test FAILed.

authors: AmplabJenkins SparkQA davies

repo: apache/spark


text:
[SPARK-19709][SQL] Read empty file with CSV data source

What changes were proposed in this pull request?
Bugfix for reading empty file with CSV data source. Instead of throwing NoSuchElementException, an empty data frame is returned.
How was this patch tested?
Added new unit test in org.apache.spark.sql.execution.datasources.csv.CSVSuite

Can one of the admins verify this patch?

@HyukjinKwon can somebody allow to test this PR as there are no more comments?

Hi @wojtek-szymanski, these are all from me. Let me cc @cloud-fan as my PRs related with this were reviewed by him and I guess I can't trigger the test.

ok to test

Test build #73950 has started for PR 17068 at commit b9e9a84.

Test build #73950 has finished for PR 17068 at commit b9e9a84.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test FAILed.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/73950/
Test FAILed.

Test build #73970 has started for PR 17068 at commit e7faa80.

Merged build finished. Test FAILed.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/73970/
Test FAILed.

retest this please

Test build #73983 has started for PR 17068 at commit e7faa80.

Test build #73983 has finished for PR 17068 at commit e7faa80.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/73983/
Test PASSed.

thanks, merging to master!

authors: AmplabJenkins HyukjinKwon SparkQA cloud-fan wojtek-szymanski


text:
[SPARK-21132] [SQL] DISTINCT modifier of function arguments should not be silently ignored

What changes were proposed in this pull request?
We should not silently ignore DISTINCT when they are not supported in the function arguments. This PR is to block these cases and issue the error messages.
How was this patch tested?
Added test cases for both regular functions and window functions

Test build #78220 has started for PR 18340 at commit 350852d.

Test build #78220 has finished for PR 18340 at commit 350852d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/78220/
Test PASSed.

+1, LGTM.

cc @cloud-fan

LGTM, merging to master/2.2

authors: AmplabJenkins SparkQA cloud-fan dongjoon-hyun gatorsmile


text:
[SPARK-21287][SQL] Ability to use Integer.MIN_VALUE as a fetchSize

What changes were proposed in this pull request?
FIX for https://issues.apache.org/jira/browse/SPARK-21287
How was this patch tested?
Existing automated tests + manual tests.

Test build #79100 has started for PR 18515 at commit 97b6d7b.

See JIRA; I don't think we should allow this.

Test build #79100 has finished for PR 18515 at commit 97b6d7b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/79100/
Test PASSed.

@maver1ck  Any update?

Test build #81467 has started for PR 18515 at commit 97b6d7b.

Test build #81467 has finished for PR 18515 at commit 97b6d7b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/81467/
Test PASSed.

authors: AmplabJenkins SparkQA gatorsmile maver1ck srowen


text:
[SPARK-23217][ML] Add cosine distance measure to ClusteringEvaluator

What changes were proposed in this pull request?
The PR provided an implementation of ClusteringEvaluator using the cosine distance measure.
This allows to evaluate clustering results created using the cosine distance, introduced in SPARK-22119.
In the corresponding JIRA, there is a design document for the algorithm implemented here.
How was this patch tested?
Added UT which compares the result to the one provided by python sklearn.

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution/666/
Test PASSed.

Test build #87161 has started for PR 20396 at commit 39d3695.

Test build #87161 has finished for PR 20396 at commit 39d3695.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/87161/
Test PASSed.

kindly ping @jkbradley @MLnick @viirya

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/testing-k8s-prb-make-spark-distribution/856/
Test PASSed.

Test build #87394 has finished for PR 20396 at commit ab34243.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Merged build finished. Test PASSed.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/87394/
Test PASSed.

Merged to master

@srowen sorry, this needs also to add the parameter to the python API; given our discussion on JIRA, what should I do? Create a follow-up PR or a new ticket on JIRA?
Thanks.

I think you can just add a follow-up

authors: AmplabJenkins SparkQA mgaido91 srowen


text:
[SPARK-34094][SQL] Extends StringTranslate to support unicode characters whose code point >= U+10000

What changes were proposed in this pull request?

This PR extends StringTranslate to support unicode characters whose code point >= U+10000.
Why are the changes needed?

To make it work with wide variety of characters.
Does this PR introduce any user-facing change?

Yes. Users can use StringTranslate with unicode characters whose code point >= U+10000.
How was this patch tested?

New assertion added to the existing test.

Test build #133994 has started for PR 31164 at commit 045c141.

Test build #133994 has started for PR 31164 at commit 045c141.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38582/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38582/

Test build #133994 has finished for PR 31164 at commit 045c141.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

Test build #133994 has finished for PR 31164 at commit 045c141.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133994/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133994/

Test build #134002 has started for PR 31164 at commit 9fb231e.

Test build #134002 has started for PR 31164 at commit 9fb231e.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38590/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38590/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38590/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38590/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38590/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38590/

Test build #134002 has finished for PR 31164 at commit 9fb231e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test build #134002 has finished for PR 31164 at commit 9fb231e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134002/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134002/

cc @srowen FYI

cc @srowen FYI

Test build #134031 has started for PR 31164 at commit 0bc8e24.

Test build #134031 has started for PR 31164 at commit 0bc8e24.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38618/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/38618/

Test build #134031 has finished for PR 31164 at commit 0bc8e24.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test build #134031 has finished for PR 31164 at commit 0bc8e24.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134031/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/134031/

OK so it doesn't show up in the tests, but these code points would be more than one char? is there any easy way to add a test for this case then? that seems to be the purpose of this change.

OK so it doesn't show up in the tests, but these code points would be more than one char? is there any easy way to add a test for this case then? that seems to be the purpose of this change.

OK so it doesn't show up in the tests, but these code points would be more than one char?

@srowen
It's already added to StringExpressionSuite. For example,  \uD840\uDC0B here is one Unicode character which is needed to be represented as two char values.

OK so it doesn't show up in the tests, but these code points would be more than one char?

@srowen
It's already added to StringExpressionSuite. For example,  \uD840\uDC0B here is one Unicode character which is needed to be represented as two char values.

Merged to master

Merged to master

authors: AmplabJenkins HyukjinKwon SparkQA sarutak srowen


text:
[SPARK-34549][BUILD] Upgrade aws kinesis to 1.14.0 and java sdk 1.11.844

What changes were proposed in this pull request?

This patch tries to upgrade aws kinesis and java sdk version.
Why are the changes needed?

Upgrade aws kinesis and java sdk to catch up minimum requirement for new feature like IAM role for service accounts: https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts-minimum-sdk.html
Does this PR introduce any user-facing change?

No
How was this patch tested?

Existing tests.

Test build #135494 has started for PR 31658 at commit a0b3e42.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40075/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/40075/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/40075/

Test build #135494 has finished for PR 31658 at commit a0b3e42.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/135494/

Merged to master.

Thank you! @HyukjinKwon

authors: AmplabJenkins HyukjinKwon SparkQA viirya


text:
[SPARK-35465][PYTHON] Set up the mypy configuration to enable disallow_untyped_defs check for pandas APIs on Spark module.

What changes were proposed in this pull request?
Sets up the mypy configuration to enable disallow_untyped_defs check for pandas APIs on Spark module.
Why are the changes needed?
Currently many functions in the main codes in pandas APIs on Spark module are still missing type annotations and disabled mypy check disallow_untyped_defs.
We should add more type annotations and enable the mypy check.
Does this PR introduce any user-facing change?
Yes.
This PR adds more type annotations in pandas APIs on Spark module, which can impact interaction with development tools for users.
How was this patch tested?
The mypy check with a new configuration and existing tests should pass.

Test build #138771 has started for PR 32614 at commit 14f1327.

Test build #138771 has finished for PR 32614 at commit 14f1327.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class CachedAccessor(Generic[T]):

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138771/

Test build #138772 has started for PR 32614 at commit 9f870fe.

Test build #138772 has finished for PR 32614 at commit 9f870fe.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138772/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43294/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43295/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43294/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43294/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43295/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43295/

Thanks! merging to master.

authors: AmplabJenkins SparkQA ueshin


text:
[SPARK-35586][K8S][TESTS] Set a default value for spark.kubernetes.test.sparkTgz in pom.xml for Kubernetes integration tests

What changes were proposed in this pull request?

This PR set a default value for spark.kubernetes.test.sparkTgz in kubernetes/integration-tests/pom.xml for Kubernetes integration tests.
Why are the changes needed?

In the current master, running the integration tests with the following command will fail because there is no default value set for the property.
build/mvn -Dspark.kubernetes.test.namespace=default -Pkubernetes -Pkubernetes-integration-tests -Psparkr  -pl resource-managers/kubernetes/integration-tests integration-test

+ mkdir -p /home/kou/work/oss/spark/resource-managers/kubernetes/integration-tests/target/spark-dist-unpacked
+ tar -xzvf --test-exclude-tags --strip-components=1 -C /home/kou/work/oss/spark/resource-managers/kubernetes/integration-tests/target/spark-dist-unpacked
tar (child): --test-exclude-tags: Cannot open: No such file or directory
tar (child): Error is not recoverable: exiting now
tar: Child returned status 2
tar: Error is not recoverable: exiting now
[ERROR] Command execution failed.

According to setup-integration-test-env.sh, N/A is intended as the default value so this PR choose it.
SPARK_TGZ="N/A"
MVN="$TEST_ROOT_DIR/build/mvn"
EXCLUDE_TAGS=""

Does this PR introduce any user-facing change?

No.
How was this patch tested?

Build and tests successfully finish with the command shown above.

Test build #139132 has started for PR 32722 at commit f2d9f30.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43652/

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43652/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43652/

Test build #139132 has finished for PR 32722 at commit f2d9f30.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/139132/

authors: AmplabJenkins SparkQA sarutak


text:
[SPARK-36402][PYTHON] Implement Series.combine

What changes were proposed in this pull request?
Implement Series.combine
Why are the changes needed?
Increase pandas API coverage in PySpark
Does this PR introduce any user-facing change?
User can use
>>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
>>> s1
falcon    330.0
eagle     160.0
dtype: float64

>>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
>>> s2
falcon    345.0
eagle     200.0
duck       30.0
dtype: float64

>>> s1.combine(s2, max)
duck        NaN
eagle     200.0
falcon    345.0
dtype: float64

How was this patch tested?
unit tests and docstest

FYI @ueshin @HyukjinKwon

ok to test

Test build #143007 has started for PR 33858 at commit b90d214.

Test build #143007 has finished for PR 33858 at commit b90d214.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47509/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47509/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143007/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47509/

@itholic can you review this one please

Test build #143046 has started for PR 33858 at commit db0d855.

Test build #143046 has finished for PR 33858 at commit db0d855.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47549/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143046/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47549/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47549/

Test build #143053 has started for PR 33858 at commit 25a09d0.

Test build #143053 has finished for PR 33858 at commit 25a09d0.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47556/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143053/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47556/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47556/

Test build #143083 has started for PR 33858 at commit cbe25d2.

Test build #143083 has finished for PR 33858 at commit cbe25d2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47585/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143083/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47585/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47585/

@ueshin @itholic @xinrong-databricks Could you please take another look ? thank you

@itholic how do you like this?

Test build #143261 has started for PR 33858 at commit 2e415cf.

Test build #143261 has finished for PR 33858 at commit 2e415cf.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47764/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47764/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143261/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47764/

Test build #143362 has started for PR 33858 at commit 83e2948.

@ueshin Could you take a look when you have time ? :D
Thank you!

Test build #143362 has finished for PR 33858 at commit 83e2948.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143362/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47869/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47869/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47869/

Test build #143418 has started for PR 33858 at commit 433eb64.

Test build #143418 has finished for PR 33858 at commit 433eb64.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47925/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47925/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143418/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47925/

Test build #143428 has started for PR 33858 at commit 6d71e11.

Test build #143428 has finished for PR 33858 at commit 6d71e11.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143428/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47935/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47935/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47935/

Test build #143435 has started for PR 33858 at commit d5f894e.

Test build #143435 has finished for PR 33858 at commit d5f894e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143435/

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47943/

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47943/

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/47943/

Could you please take another look :D ? @ueshin

Gently ping @HyukjinKwon @itholic @ueshin  , Should I add more work on this ? Does it look promising?

@dchvn Sure, go ahead !!
Which work do you want to add? Or wouldn't you just leave the comment ??

Thank you, I just want ask your thought about this one.
I will recreate it in my personal account as discuss with @HyukjinKwon . Sorry about the inconvenience!

@dchvn Got it, no worries!!
IMO, working with the personal account from the next PR should be fine.
But let me leave the decision to @ueshin and @HyukjinKwon who have committer rights :-)

authors: AmplabJenkins HyukjinKwon SparkQA dchvn dgd-contributor itholic xinrong-databricks


text:
[SPARK-38030][SQL] Canonicalization of cast should remove nullability of target dataType

What changes were proposed in this pull request?
Canonicalization of cast should remove nullability information of target dataType so that it matches the canonicalized version of its child expression.

Why are the changes needed?
SPARK-38030 lists an issue where canonicalization of cast resulted in an unresolved expression, thus causing query failure. Although the exact repro listed in SPARK-38030 no longer works in master due to an unrelated change (details in the JIRA), some other codepaths which depend on canonicalized representations can trigger the same issue.

Does this PR introduce any user-facing change?
No

How was this patch tested?
Added unit test to ensure that canonicalization result in a resolved cast with nullability of target dataType removed

cc: @cloud-fan @viirya @HyukjinKwon

Can one of the admins verify this patch?

@cloud-fan @viirya @xkrogen Thanks for the reviews! I have updated the PR with the new approach of preserving nullability during canonicalization of AttributeReference.

LGTM, also cc @sigmod

thanks, merging to master!

authors: AmplabJenkins cloud-fan shardulm94


text:
[SPARK-38124][SQL][SS] Introduce StatefulOpClusteredDistribution and apply to stream-stream join

What changes were proposed in this pull request?
This PR revives HashClusteredDistribution and renames to StatefulOpClusteredDistribution so that the rationalization of the distribution is clear from the name. Renaming is safe because this class no longer needs to be general one - in SPARK-35703 we moved out the usages of HashClusteredDistribution to ClusteredDistribution; stateful operators are exceptions.
Only HashPartitioning with same expressions and number of partitions can satisfy StatefulOpClusteredDistribution. That said, we cannot modify HashPartitioning unless we clone HashPartitioning and assign the clone to StatefulOpClusteredDistribution.
This PR documents the expectation of stateful operator on partitioning in the classdoc of StatefulOpClusteredDistribution.
This PR also changes stream-stream join to use StatefulOpClusteredDistribution instead of ClusteredDistribution. This effectively reverts a part of SPARK-35703 which hasn't been shipped to any releases. This PR doesn't deal with other stateful operators since it has been long standing issue (probably Spark 2.2.0+) and we need a plan for dealing with existing state.
Why are the changes needed?
Spark does not guarantee stable physical partitioning for stateful operators across query lifetime, and due to the relaxed distribution requirement it is hard to expect what would be the current physical partitioning of the state.
(We expect hash partitioning with grouping keys, but ClusteredDistribution does not "guarantee" the partitioning. It is much more relaxed.)
This PR will enforce the physical partitioning of stream-stream join operators to be hash partition with grouping keys, which is our general expectation of state store partitioning.
Does this PR introduce any user-facing change?
No, since SPARK-35703 hasn't been shipped to any release yet.
How was this patch tested?
Existing tests.

TODO:
I'll fix the build and make sure tests are passing.
I'll also try to see whether I can add tests for stateful operators to see whether the physical partitioning is consistent with several pre-partitioning.

Implementation-wise, the PR is ready to review.
cc. @cloud-fan @sunchao @viirya @HyukjinKwon @c21 who are listed as reviewers of #32875.
cc. @tdas @zsxwing @brkyvz @jose-torres @xuanyuanking as reviewers on streaming area.

I see the concern of fixing stateful operators due to the state on existing queries. stream-stream join has been using HashClusteredDistribution and it shouldn't suffer this long standing problem.
So if we want to split down the problem with stream-stream join and others, I'll move out the changes on other stateful operators, and in there we have to also come up with the plan how to deal with state on existing queries.
Would it work for all?

(I'm reading through #19080, and just realized there has been NO such strict requirement before we introduce HashClusteredDistribution in #19080 if I understand correctly... If then, it doesn't even start from 2.3+. It starts from 2.2+ where we introduced stateful operators in SS.)

So if we want to split down the problem with stream-stream join and others, I'll move out the changes on other stateful operators, and in there we have to also come up with the plan how to deal with state on existing queries.

+1, SGTM.

Just updated JIRA ticket, PR description, PR code diff to only contain the change of stream-stream join. I'll file another JIRA ticket for dealing with other stateful operators.

In the meanwhile I'll think of how to deal with other stateful operators with existing state. One rough idea is having the functionality of validation on state partitioning against child output partitioning, although it is not "future-proof" validation.

@viirya Could you please take a look again? I've reflected your review comments via 753a5b6

Thanks @HeartSaVioR . LGTM

Thanks! Merging to master.

authors: HeartSaVioR c21 viirya


text:
[SPARK-38124][SS][FOLLOWUP] Add test to harden assumption of SS partitioning requirement

What changes were proposed in this pull request?

This is a followup of https://github.com/apache/spark/pull/35419#discussion_r801354342, to add unit test to harden the assumption of SS partitioning and distribution requirement:

Check the HashPartitioning.partitionIdExpression to be exactly expected format
Check all different kinds of Partitioning against StatefulOpClusteredDistribution.

Also add a minor comment for StatefulOpClusteredDistribution, as SinglePartition can also satisfy the distribution.
Why are the changes needed?

Document our assumption of SS in code as unit test.
So next time when we introduce intrusive code change, the unit test can save us by failing loudly.
Does this PR introduce any user-facing change?

No.
How was this patch tested?

The added unit test itself.

cc @cloud-fan, @sunchao, @viirya, @HyukjinKwon and @HeartSaVioR, thanks.

Thanks! Merging to master.

Thank you all for review!


text:
[SPARK-38300][SQL] Use ByteStreams.toByteArray to simplify fileToString and resourceToBytes in catalyst.util

What changes were proposed in this pull request?
This pr replace the manually InputStream to ByteArray codes with com.google.common.io.ByteStreams.toByteArray to simplify the methods fileToString and resourceToBytes in catalyst.util.  At the same time, the new method has better efficiency.
Why are the changes needed?
Simplify code and use more efficient method.
Does this PR introduce any user-facing change?
No
How was this patch tested?

Pass GA
Manual test:

we can prove that the ByteStreams.toByteArray is more efficient than old code through the following test:
def testToByteArray(fileSie: Int): Unit = {
    // Prepare data
    val bytes = RandomUtils.nextBytes(fileSie)
    val file = File.createTempFile(s"$fileSie-${UUID.randomUUID()}", ".dat")
    val fileForGuavaApi = File.createTempFile(s"$fileSie-${UUID.randomUUID()}", ".dat")
    file.deleteOnExit()
    fileForGuavaApi.deleteOnExit()
    FileUtils.writeByteArrayToFile(file, bytes)
    FileUtils.writeByteArrayToFile(fileForGuavaApi, bytes)

    val benchmark = new Benchmark(s"ToByteArray with fileSize $fileSie bytes ", 1, output = output)

    benchmark.addCase("toByteArray: byte by byte copy") { _: Int =>
      toByteArray(file)
    }

    benchmark.addCase("toByteArray: use Guava api") { _: Int =>
      toByteArrayUseGuava(fileForGuavaApi)
    }
    benchmark.run()
  }

  private def toByteArrayUseGuava(file: File): Array[Byte] = {
    val inStream = new FileInputStream(file)
    try {
      ByteStreams.toByteArray(inStream)
    } finally {
      inStream.close()
    }
  }

  private def toByteArray(file: File): Array[Byte] = {
    val inStream = new FileInputStream(file)
    val outStream = new ByteArrayOutputStream
    try {
      var reading = true
      while (reading) {
        inStream.read() match {
          case -1 => reading = false
          case c => outStream.write(c)
        }
      }
      outStream.flush()
    } finally {
      inStream.close()
    }
    outStream.toByteArray
  }

  override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
    // Test 1K, 1M, 10M, 100M fileSize
    Seq(1024, 1024 * 1024, 1024 * 1024 * 10, 1024 * 1024 * 100).foreach { fileSize =>
      testToByteArray(fileSize)
    }
  }

The results of the above  test code using GA are as follows:
OpenJDK 64-Bit Server VM 1.8.0_322-b06 on Linux 5.11.0-1028-azure
Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
ToByteArray with fileSize 1024 bytes :    Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
toByteArray: byte by byte copy                        1              1           0          0.0      781310.0       1.0X
toByteArray: use Guava api                            0              0           0          0.2        6100.0     128.1X

OpenJDK 64-Bit Server VM 1.8.0_322-b06 on Linux 5.11.0-1028-azure
Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
ToByteArray with fileSize 1048576 bytes :  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
toByteArray: byte by byte copy                      812            817           4          0.0   812133973.0       1.0X
toByteArray: use Guava api                            1              1           0          0.0     1002514.0     810.1X

OpenJDK 64-Bit Server VM 1.8.0_322-b06 on Linux 5.11.0-1028-azure
Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
ToByteArray with fileSize 10485760 bytes :  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-------------------------------------------------------------------------------------------------------------------------
toByteArray: byte by byte copy                      8203           8206           4          0.0  8203332962.0       1.0X
toByteArray: use Guava api                            18             20           2          0.0    18325138.0     447.7X

OpenJDK 64-Bit Server VM 1.8.0_322-b06 on Linux 5.11.0-1028-azure
Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
ToByteArray with fileSize 104857600 bytes :  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
--------------------------------------------------------------------------------------------------------------------------
toByteArray: byte by byte copy                      82543          82671         181          0.0 82542615028.0       1.0X
toByteArray: use Guava api                            182            195          24          0.0   182350535.0     452.7X

I don't believe it was benchmarked, but copying chunks of bytes at a time rather than one at a time has to be faster - a tiny bit in the case the input stream is buffered anyway, a lot in the case that it isn't

@dongjoon-hyun I remove At the same time, the new method has better efficiency. from pr description first and I'll investigate and give a conclusion.

def testToByteArray(fileSie: Int): Unit = {

    val bytes = RandomUtils.nextBytes(fileSie)
    val file0 = File.createTempFile(s"$fileSie-${UUID.randomUUID()}", ".dat")
    val file1 = File.createTempFile(s"$fileSie-${UUID.randomUUID()}", ".dat")
    file0.deleteOnExit()
    file1.deleteOnExit()
    FileUtils.writeByteArrayToFile(file0, bytes)
    FileUtils.writeByteArrayToFile(file1, bytes)

    val benchmark = new Benchmark(
      s"ToByteArray with $fileSie ",
      1,
      output = output)

    benchmark.addCase("toByteArray") { _: Int =>
      toByteArray(file0)
    }

    benchmark.addCase("toByteArray Use Guava") { _: Int =>
      toByteArrayUseGuava(file1)
    }
    benchmark.run()
  }

  private def toByteArrayUseGuava(file: File): Array[Byte] = {
    val inStream = new FileInputStream(file)
    try {
      ByteStreams.toByteArray(inStream)
    } finally {
      inStream.close()
    }
  }

  private def toByteArray(file: File): Array[Byte] = {
    val inStream = new FileInputStream(file)
    val outStream = new ByteArrayOutputStream
    try {
      var reading = true
      while (reading) {
        inStream.read() match {
          case -1 => reading = false
          case c => outStream.write(c)
        }
      }
      outStream.flush()
    } finally {
      inStream.close()
    }
    outStream.toByteArray
  }

  override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
    Seq(1024, 1024 * 1024, 1024 * 1024 * 10, 1024 * 1024 * 100).foreach { fileSize =>
      testToByteArray(fileSize)
    }
  }

I write a micro bench as above to compare the old method and  ByteStreams.toByteArray

The result as follows:
Running benchmark: ToByteArray with 1024
  Running case: toByteArray
  Stopped after 4491 iterations, 2000 ms
  Running case: toByteArray Use Guava
  Stopped after 148518 iterations, 2000 ms

OpenJDK 64-Bit Server VM 1.8.0_312-b07 on Mac OS X 11.4
Apple M1
ToByteArray with 1024 :                   Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
toByteArray                                           0              0           0          0.0      424250.0       1.0X
toByteArray Use Guava                                 0              0           0          0.1       11416.0      37.2X

Running benchmark: ToByteArray with 1048576
  Running case: toByteArray
  Stopped after 5 iterations, 2182 ms
  Running case: toByteArray Use Guava
  Stopped after 6650 iterations, 2000 ms

OpenJDK 64-Bit Server VM 1.8.0_312-b07 on Mac OS X 11.4
Apple M1
ToByteArray with 1048576 :                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
toByteArray                                         430            436          10          0.0   429517208.0       1.0X
toByteArray Use Guava                                 0              0           0          0.0      273167.0    1572.4X

Running benchmark: ToByteArray with 10485760
  Running case: toByteArray
  Stopped after 2 iterations, 8609 ms
  Running case: toByteArray Use Guava
  Stopped after 542 iterations, 2001 ms

OpenJDK 64-Bit Server VM 1.8.0_312-b07 on Mac OS X 11.4
Apple M1
ToByteArray with 10485760 :               Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
toByteArray                                        4300           4305           7          0.0  4299691500.0       1.0X
toByteArray Use Guava                                 3              4           1          0.0     3324875.0    1293.2X

Running benchmark: ToByteArray with 104857600
  Running case: toByteArray
  Stopped after 2 iterations, 86419 ms
  Running case: toByteArray Use Guava
  Stopped after 53 iterations, 2007 ms

OpenJDK 64-Bit Server VM 1.8.0_312-b07 on Mac OS X 11.4
Apple M1
ToByteArray with 104857600 :              Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
toByteArray                                       43197          43210          18          0.0 43197043042.0       1.0X
toByteArray Use Guava                                33             38           6          0.0    33091375.0    1305.4X

It seems that ByteStreams.toByteArray is faster @dongjoon-hyun @srowen

Thank you, @LuciferYang . Please put your result and benchmark code into the PR description in order to make them as a commit log.

Thank you, @LuciferYang . Please put your benchmark code and result into the PR description in order to make them as a commit log.

OK

Waiting for GA to output the benchmark results and upload

Thank you, @LuciferYang . Please put your benchmark code and result into the PR description in order to make them as a commit log.

@dongjoon-hyun done, the bench result is produced using GA

Seems fine, I don't think a benchmark is important in cases like this but we have it

Merged to master. Thank you, @LuciferYang and @srowen .

thanks all

authors: LuciferYang dongjoon-hyun srowen


text:
[SPARK-39301][SQL][PYTHON] Leverage LocalRelation and respect Arrow batch size in createDataFrame with Arrow optimization

What changes were proposed in this pull request?
This PR proposes to use LocalRelation instead of LogicalRDD when creating a (small) DataFrame with Arrow optimization, which passes the data as a local data in the driver side (which is consistent with Scala code path).
Namely:
import pandas as pd
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", True)
spark.createDataFrame(pd.DataFrame({'a': [1, 2, 3, 4]})).explain(True)

Before
== Parsed Logical Plan ==
LogicalRDD [a#0L], false

== Analyzed Logical Plan ==
a: bigint
LogicalRDD [a#0L], false

== Optimized Logical Plan ==
LogicalRDD [a#0L], false

== Physical Plan ==
*(1) Scan ExistingRDD arrow[a#0L]

After
== Parsed Logical Plan ==
LocalRelation [a#0L]

== Analyzed Logical Plan ==
a: bigint
LocalRelation [a#0L]

== Optimized Logical Plan ==
LocalRelation [a#0L]

== Physical Plan ==
LocalTableScan [a#0L]

This is controlled by a new configuration spark.sql.execution.arrow.localRelationThreshold defaulting to 48MB. This default was picked by benchmark I ran below.
In addition, this PR also fixes createDataFrame to respect spark.sql.execution.arrow.maxRecordsPerBatch configuration when creating Arrow bathes. Previously, we divided the input pandas DataFrame by the default partition number which forced users to set spark.rpc.message.maxSize when the input pandas DataFrame is too large. See the benchmark performed below.
Why are the changes needed?
We have some nice optimization for LocalRelation (e.g., ConvertToLocalRelation). For example, the stats are fully known when you use LocalRelation. With LogicalRDD, many optimizations cannot be applied. Even in some cases (e.g., executeCollect), we can avoid creating RDDs too.
For respecting spark.sql.execution.arrow.maxRecordsPerBatch, 1. we can avoid forcing users to set spark.rpc.message.maxSize, and 2. I believe the configuration is supposed to be respected for all code path that creates Arrow batches if possible.
Does this PR introduce any user-facing change?
No, it is an optimization. The number of partitions can be different, but that should be internal.
How was this patch tested?

Manually tested.
Added a unittest.
I did two benchmark tests with 1 Driver & 4 Workers (i3.xlarge), see below.

Benchmark 1 (best cases)
import time
import random
import string

import pandas as pd

names = [random.choice(list(string.ascii_lowercase)) for i in range(1000)]
ages = [random.randint(0, 100) for i in range(1000)]
l = list(zip(names, ages))
d = [{'name': a_name, 'age': an_age} for a_name, an_age in l]
pdf = pd.DataFrame({'name': names, 'age': ages})
spark.range(1).count()  # heat up

start = time.time()
for _ in range(100):
    _ = spark.createDataFrame(pdf)

end = time.time()
print(end - start)

Before
10.250491698582968
After
6.004616181055705
Benchmark 2 (worst cases)
curl -O https://eforexcel.com/wp/wp-content/uploads/2020/09/HR2m.zip
unzip HR2m.zip

import pandas as pd
pdf = pd.read_csv("HR2m.csv")
pdf23 = pdf.iloc[:int(len(pdf)/32)]
pdf45 = pdf.iloc[:int(len(pdf)/16)]
pdf90 = pdf.iloc[:int(len(pdf)/8)]
pdf175 = pdf.iloc[:int(len(pdf)/4)]
pdf350 = pdf.iloc[:int(len(pdf)/2)]
pdf700 = pdf.iloc[:int(len(pdf))]
pdf2gb = pd.concat([pdf, pdf, pdf])
pdf5gb = pd.concat([pdf2gb, pdf2gb])

spark.createDataFrame(pdf23)._jdf.rdd().count()  # explicitly create RDD.
...

Before
23MB: 1.02 seconds
45MB: 1.69 seconds
90MB: 2.38 seconds
175MB: 3.19 seconds
350MB: 6.10 seconds
2GB: 43.21 seconds
5GB: X (threw an exception that says to set 'spark.rpc.message.size' higher)
After
23MB: 1.31 seconds (local collection is used)
45MB: 2.47 seconds (local collection is used)
90MB: 1.79 seconds
175MB: 3.22 seconds
350MB: 6.41 seconds
2GB: 47.12 seconds
5GB: 1.29 minutes
NOTE that the performance varies depending on network stability, and the numbers above are from second run (it's not the average).

cc @ueshin @viirya @BryanCutler FYI

Wow, thanks for reviews guys. Let me think a bit more and push some changes soon.

This PR is ready for a look now. TL;DR:

I added one new configuration spark.sql.execution.arrow.localRelationThreshold as a threshold.
I fixed the code path to respect spark.sql.execution.arrow.maxRecordsPerBatch when creating Arrow batches to avoid forcing users to set a larger spark.rpc.message.maxSize.

Gentle ping for a review :-). I know it has some trade-off but I believe this addresses more common cases and benefit more users.

cc @mengxr and @WeichenXu123 in case you guys have some comments.

Let me merge this in few days ... assuming that we're all good. Hopefully my benchmark is good enough.

Rebased

Let me get this in. It's the early stage of Spark 3.4 so should be good timing to merge such stuff.
Merged to master.


text:
[SPARK-41256][CONNECT] Implement DataFrame.withColumn(s)

What changes were proposed in this pull request?

Implement DataFrame.withColumn(s).
Why are the changes needed?

API coverage
Does this PR introduce any user-facing change?

NO
How was this patch tested?

UT

@zhengruifeng @grundprinzip @cloud-fan @HyukjinKwon

I ended up with

Change WithColumns proto to use a list of name/columns.
I didn't choose to use Alias which is not that ideal for this case. I just use a small message to wrap the name/expression pair.
I follow current API to just throw exception when there are duplicated names.

@amaliujia why is using Alias not ideal for this case? It seems to fit the bill rather nicely.

@hvanhovell
Well the simple answer is WithColumn API's parameters are not named Alias.
But my longer thought is:
Alias has its defined semantics (including those fields). That semantics does not match what this PR looks for. Alias happens to have two fields that can help this PR (and Alias has one more field which is metadata for the its semantics).
I think it's a good to avoid using a message only because the fields type fit but the semantics does not match.  In principle,   The message semantics can evolve so over-using it will cause trouble. Just a bad example: one day we will decide that Alias should only accept an alias which type is float. Then those places that truly deal with Alias which makes sense. But our WithColumns will not be compatible because we happen to use Alias due to same field type but not match with its semantics.
So we re-use a message because of the matching semantics. If the semantics match but even there is a lack of capability, we can even extend the message by adding more fields etc. to help.  But it does not work vice verse.

@amaliujia the semantics of Alias and the name + expression are 100% the same. Both are named expressions, and at the end of the day withColumns will produce a Project using exactly those Alias expressions. The replacement/appending semantics of withColumn does not change anything here. Come to thing of it, having metadata here would actually make sense.
There is no way we are ever going to change Alias, this has been in databases for the last 50+ years. However, let's entertain the idea of us changing Alias, you can solve those problems with proper versioning and a breaking change process (we are going to need both at some point in the not so far away future). I am assuming you are changing Alias in the connect protocol, in that case such a breaking change probably means you will introduce a new expression.

@hvanhovell
I see you are thinking better now.
When I say different semantics, what I mean is at the DataFrame API surface (which is WithColumn). But if you say that at the end, in implementation those parameters are converted to Alias, which is right and at that moment the semantics match.
I don't have strong opinion here. Probably I will just change to use Alias.

I pushed a commit to use Alias but now I know why I have been having a sense of the different semantics:
  message Alias {
    Expression expr = 1;
    repeated string name = 2;
    // Alias metadata expressed as a JSON map.
    optional string metadata = 3;
  }

Alias has a list of name parts while WithColumn assumes one name part (a single string). Right now I just assume there is one name part passed from the Alias messages.

What is the reason for alias to have a multi part identifier? Does this actually make sense? Maybe we need to fix this as well.

hmm well @grundprinzip maybe you should tell us why it is repeated names :) https://github.com/apache/spark/commit/0f7eaeee6445aaf05310229bdec22f6953113f2e

By reading the commit https://github.com/apache/spark/commit/0f7eaeee6445aaf05310229bdec22f6953113f2e honestly I am not sure if Python side has a valid usage for Alias or not.
Maybe @cloud-fan do you recall we support multiple name parts for alias?

Good point! The reason for making it repeated was that if you have a function that you alias the function might produce multiple columns.
So I think the repeated is fine here and we should probably update the documentation of the proto to leave a trace so that we don't have to do a history search all the time.
The key part is that for scalar columns just one value is present.

My two cents:
In general, I agree with @amaliujia that we should not reuse a proto message with a mismatched semantic. But we should define the semantic by the proto message itself, not the DataFrame API. For a WithColumns operator, I think using Alias matches its semantic (add/replace columns with the provided expressions according to the provided names). The problem I see here is Alias can contain a list of alias names. We should either document it clearly that WithColumns can only use Alias with a single alias name, or introduce MultiAlias like what catalyst does.

I don't think anyone is disputing that reusing messages in different contexts just because the fields are identical is a bad idea.
One thing I'm wondering and I'm sorry that I didn't bring this up earlier is if we shouldn't implement WithColum using a Projection? The main difference I see is that to make projection work we have to resolve the schema to be able to append columns.
Maybe we can do Project[*, add columns]?

Maybe we can do Project[*, add columns]?

The semantic is different. df.withColumns can replace columns if the given name(s) conflict with the existing columns. Project[*, added columns] can't do it.

@cloud-fan thanks. I will stay on the Alias probably try to document it clearly for the WithColumn

Maybe we can do Project[*, add columns]?

The semantic is different. df.withColumns can replace columns if the given name(s) conflict with the existing columns. Project[*, added columns] can't do it.

Yes Project[*, added columns] cant do that, but you're able to replace the columns on the client side by resolving the schema first. What I'm trying to figure out is where we draw the line between adding another input relation type to avoid become a different Py4j.

Hmmmm it's a trade off between proto number and implementation complexity. I personally choose proto because there are more than 1 client. The implementation complexity will need to times N.

@grundprinzip IMO we should go for a server side implementation of withColumns for the following reasons:

Connect in its current form is lazy. Going the project route would violate that, because we would need to execute a RPC to get the schema. This in itself is not a problem, however when the schema of the table changes you might end up with missing columns or errors.
Code reuse. As @amaliujia mentioned, we are likely to have withColumns in multiple clients (scala will definitely have this as well). I want to drive home the point here that withColumns has some non-trivial name resolution logic (with name parsing and a bunch of other things), I think it is better to do it once, and have a consistent user experience.
Customers love to stack withColumns on top of each other (e.g. do analysis of all columns in a dataframe in a loop). I would love to move this into the analyzer at some point. That should make it quite a bit faster.

Taking a step back I think we need some decision framework on where to implement what. In this case the logic is non-trivial and I think it should belong on the sever.

Rebased and tried to address existing comments

Thanks for the good discussion and the clear commit message.

@amaliujia  It would be great if you can update the commit message with a bit more context on what proto fields were added and why it's implemented the way it is as a summary of our discussion here. So when someone goes through the history of the commits they can see the decision making process.

@grundprinzip the commit message eventually is the PR description so I updated there.
Maybe in the future we need a developer doc to explain some of our current proto design decisions.

thanks, merging to master!

authors: amaliujia cloud-fan grundprinzip hvanhovell


text:
[SPARK-43516] [ML] Base interfaces of sparkML for spark3.5: estimator/transformer/model/evaluator

What changes were proposed in this pull request?

Defines basic interfaces of Evaluator / Transformer / Model / Evaluator, these interfaces are designed to support both spark connect and legacy spark mode, and are designed to support train / transform / evaluate over either spark dataframe or pandas dataframe
Implement a feature transformer MaxAbsScaler, ScalerScaler
Implement a regressor evaluator RegressorEvaluator that supports MSE and R2 metric evaluation
Implement a summarizer via pure python code that can summarize array type columns on spark dataframe.
Some utility methods.

Why are the changes needed?
Project: Distributed ML <> spark connect
Does this PR introduce any user-facing change?
Yes.
How was this patch tested?
Unit tests.

I think you may need to add torchmetrics to https://github.com/apache/spark/blob/f55fdca10b1d9df2d8126cde07a26f89a75ae1d2/dev/infra/Dockerfile#L74


text:
[SPARK-5336][YARN]spark.executor.cores must not be less than spark.task.cpus

https://issues.apache.org/jira/browse/SPARK-5336

Test build #25826 has started for   PR 4123 at commit 3112af9.

This patch merges cleanly.

Test build #25826 has finished for   PR 4123 at commit 3112af9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25826/
Test PASSed.

We should warn about this in standalone mode and mesos coarse grained mode as well, right?

@sryza
In standalone and mesos mode it is a bit different. JIRA is here:https://issues.apache.org/jira/browse/SPARK-5337
Seems like @CodingCat has created a PR: https://github.com/apache/spark/pull/4129, we can discuss the details there.

Test build #25883 has started for   PR 4123 at commit d03d6fa.

This patch merges cleanly.

Test build #25887 has started for   PR 4123 at commit 9632d3a.

This patch merges cleanly.

Actually when cluster has not enough cores, the driver will also hang up. I made a trivial change here and hope it would not make much noise for this pr.

Test build #25883 has finished for   PR 4123 at commit d03d6fa.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25883/
Test PASSed.

Test build #25887 has finished for   PR 4123 at commit 9632d3a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25887/
Test PASSed.

Test build #25890 has started for   PR 4123 at commit 6c9676a.

This patch merges cleanly.

@tgravescs Updated as you said. Perhaps we should do the same change in ExecutorAllocationManager, but I am not familiar with that future, so let's leave it with guys who is in charge of that.

I agree, the text was that way in ExecutorAllocationManager so lets leave it for now.
This looks good.  Thanks @WangTaoTheTonic

Test build #25890 has finished for   PR 4123 at commit 6c9676a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25890/
Test PASSed.

authors: AmplabJenkins SparkQA WangTaoTheTonic sryza tgravescs


text:
[SPARK-5337][Mesos][Standalone] respect spark.task.cpus when scheduling Applications

https://issues.apache.org/jira/browse/SPARK-5337
Currently, we didn't consider spark.task.cpus when scheduling the applications in Master, so that we may fall into one of the following cases

the executor gets N cores but we need M cores to run a single task, where N < M

the executor gets N cores, we need M cores to run a single task, where N % M != 0 && N > M; so that we waste some cores in the executor

Patch for YARN is in submitted by @WangTaoTheTonic : https://github.com/apache/spark/pull/4123

Test build #25862 has started for   PR 4129 at commit d383b44.

This patch merges cleanly.

Test build #25862 has finished for   PR 4129 at commit d383b44.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25862/
Test FAILed.

Test build #25863 has started for   PR 4129 at commit e28e19f.

This patch merges cleanly.

Test build #25864 has started for   PR 4129 at commit 523b3b7.

This patch merges cleanly.

Test build #25863 has finished for   PR 4129 at commit e28e19f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25863/
Test PASSed.

Test build #25864 has finished for   PR 4129 at commit 523b3b7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25864/
Test PASSed.

What if app.coresLeft < app.desc.coreNumPerTask?
And should we fix the same issue in mesos mode in this PR?

Hi， @WangTaoTheTonic , if app.coresLeft < app.desc.coreNumPerTask, we should not assign more cores to the executor...as those cores will just be wasted since we cannot run a task with that...
in the PR, it's just
 while (toAssign >= app.desc.coreNumPerTask) {

and
if (coresToUse >= app.desc.coreNumPerTask) {

just upload the fix for Mesos...

Test build #25888 has started for   PR 4129 at commit 619c8b9.

This patch merges cleanly.

Test build #25888 has finished for   PR 4129 at commit 619c8b9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25888/
Test PASSed.

Hi, @WangTaoTheTonic, I just addressed your comments
any other comments?

Test build #25892 has started for   PR 4129 at commit d54c5f5.

This patch merges cleanly.

Test build #25893 has started for   PR 4129 at commit 9f2a8be.

This patch merges cleanly.

Test build #25892 has finished for   PR 4129 at commit d54c5f5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25892/
Test PASSed.

Test build #25893 has finished for   PR 4129 at commit 9f2a8be.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25893/
Test PASSed.

Besides, the # of cores needed by an application is set via spark.cores.max.
What will happen if we set spark.cores.max < spark.task.cpus ? I think the appcalition submitted will hang up infinitely.

Test build #25974 has started for   PR 4129 at commit bca1080.

This patch merges cleanly.

Test build #25975 has started for   PR 4129 at commit 8b088b2.

This patch merges cleanly.

Test build #25974 has finished for   PR 4129 at commit bca1080.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25974/
Test PASSed.

Test build #25975 has finished for   PR 4129 at commit 8b088b2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/25975/
Test PASSed.

Test build #27373 has started for   PR 4129 at commit 43852cb.

This patch merges cleanly.

Test build #27373 has finished for   PR 4129 at commit 43852cb.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/27373/
Test PASSed.

@WangTaoTheTonic any other comments?

Test build #28528 has started for   PR 4129 at commit 6d76f12.

This patch merges cleanly.

Test build #28528 has finished for   PR 4129 at commit 6d76f12.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/28528/
Test PASSed.

Looks good logically. But we need to do some tests based on this commit especially in mesos mode(fine/coarse grained).
@pwendell @JoshRosen Any comments?

Test build #28560 has started for   PR 4129 at commit 744e91b.

This patch merges cleanly.

Test build #28560 has finished for   PR 4129 at commit 744e91b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/28560/
Test PASSed.

I don't see fine grained mode (MesosSchedulerBackened) being changed at all?

If I read the code correctly , I don't need to change that, as it starts mesos task with the correct cpu setup https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala#L142

@CodingCat sorry you're right, I didn't realize CPUS_PER_TASK was configured to that flag. LGTM

@CodingCat good to see this being fixed. However, as it stands the existing solution does not seem to be sufficient. I pointed out a scenario where we will fall into an infinite scheduling loop if the conditions are not met. Also, once you have the time would you mind bringing this up to date? I believe your other patch I just merged in today conflicts quite significantly with your changes here.

sure, will work on it soon

Test build #30601 has started for   PR 4129 at commit da8d446.

Test build #30601 has finished for   PR 4129 at commit da8d446.

This patch fails Scala style tests.

This patch merges cleanly.

This patch adds the following public classes (experimental):

public class MapConfigProvider extends ConfigProvider

This patch does not change any dependencies.

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/30601/
Test FAILed.

Test build #30602 has started for   PR 4129 at commit 55d9143.

Test build #30603 has started for   PR 4129 at commit c10f980.

Test build #30602 has finished for   PR 4129 at commit 55d9143.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/30602/
Test PASSed.

Test build #30603 has finished for   PR 4129 at commit c10f980.

This patch passes all tests.

This patch merges cleanly.

This patch adds the following public classes (experimental):

public class MapConfigProvider extends ConfigProvider

This patch does not change any dependencies.

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/30603/
Test PASSed.

@andrewor14 , I updated the patch, how about the current version?

@CodingCat there have been large changes to the standalone scheduler in 1.5. I don't think this patch in its current state can be easily merged anymore. If you have time, would you mind closing this and reopening a new patch against the latest master?

sure, let me do it in the weekend

authors: AmplabJenkins CodingCat SparkQA WangTaoTheTonic andrewor14 tnachen


text:
[SPARK-44878][SS] Disable strict limit for RocksDB write manager to avoid insertion exception on cache full

What changes were proposed in this pull request?
Disable strict limit for RocksDB write manager to avoid insertion exception on cache full
Why are the changes needed?
In some cases, if the memory limit is reached, on insert/get, we are seeing the following exception
org.apache.spark.SparkException: Job aborted due to stage failure: Task 42 in stage 9.0 failed 4 times, most recent failure: Lost task 42.3 in stage 9.0 (TID 2950) (96.104.176.55 executor 0): org.rocksdb.RocksDBException: Insert failed due to LRU cache being full.
	at org.rocksdb.RocksDB.get(Native Method)
	at org.rocksdb.RocksDB.get(RocksDB.java:2053)
	at org.apache.spark.sql.execution.streaming.state.RocksDB.get(RocksDB.scala:299)
	at org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider$RocksDBStateStore.get(RocksDBStateStoreProvider.scala:55)

It seems this is being thrown with strict memory limit within RocksDB here - https://github.com/facebook/rocksdb/blob/0fa0c97d3e9ac5dfc2e7ae94834b0850cdef5df7/cache/lru_cache.cc#L394
It seems this issue can only happen with the strict mode as described here - https://github.com/facebook/rocksdb/issues/5048#issuecomment-470788792
Seems like there is a pending issue for RocksDB around this as well - https://github.com/facebook/rocksdb/issues/8670
There is probably a relevant fix, but not sure whether this addresses the issue completely - https://github.com/facebook/rocksdb/pull/6619
(cc - @siying )
Does this PR introduce any user-facing change?
No
How was this patch tested?
Existing tests

cc - @HeartSaVioR , @siying - PTAL, thx

When block cache in strict mode is full, we try to insert more data, some exception will be thrown one way of another. I don't think SS has a way to handle the error, and the consequence will be task failure. If that is not a behavior we can afford, we should disable strict mode.
It is a trade-off here: either RocksDB reject queries and background tasks that require more memory than needed, or we use more memory than configured. RocksDB might not have a better way to handle it. It is a less used feature, and is mainly used in those applications that would rather to fail than using more configured memory. An example, sometimes people don't feel comfortable for an administrative demon to use more memory and crash a host that might serves online queries.
I think it is a decision Spark SS needs to make. When users misconfigured memory cap that turns out to be not enough, are we going to fail their queries, or are we going to allow RocksDB to use more memory than needed and potentially cause an OOM. I can see both pros and cons on it.

@siying - how does the cache eviction mechanism work though for RocksDB ? since we are accounting everything towards block cache, is it not possible to spill write buffer/block cache blocks to disk if we hit this cache full case ? Maybe we won't be able to evict pinned blocks, but what about other blocks ?
Also, if we do decide to allow the tasks to fail, will some memory be reclaimed eventually ? i.e. will such future task attempts ever succeed ? or do we need to restart the process ?

@anishshri-db the blocks in use cannot be spilled to disk. Those are objects that the higher level code fetched to use. They are supposed to access the contents using pointers to the memory. After it is used, they are returned to block cache. Those returned to block cache are evicted based on LRU. The insertion is rejected with an error only happens when all objects are being used already take all allocated capacity.
Whatever is returned to block cache can be evicted using LRU.

@siying - Thanks for the explanation. If thats the case, then I think we might have to set this setting to false, atleast for now. Even if we fail this task, there is no guarantee that other blocks could be returned in time for eviction before other task attempts are made. Effectively this would mean that the query would fail (after 4 consecutive attempts) and we would have to restart and reinstantiate the state (which might be quite expensive). We might be better off with a soft limit to avoid this kind of scenario.
In the long term, we probably might need to address this by auto-scaling the cluster and moving partitions to other nodes such that the individual limit of a node is not reached for the allocated/assigned memory.
cc - @HeartSaVioR - thoughts ?

I'd still like to understand how this is different from not capping the memory at all. Does capping the memory avoid RocksDB using the memory excessively? Or is there no difference between capping with soft limit vs no capping at all?
Also, there is another aspect to think of - OOM kills the executor which could affect all stateful, stateless, batch queries. This error will only affect stateful queries. If people intends to set the limit on RocksDB memory usage considering this fact, soft limiting would break the intention, although they may still need to restart the cluster or at least executor to apply the new setting of memory limit on RocksDB. Looks to be very tricky to adjust from users' point of view when the error happens...
Ideally we will need to rebalance the state if the memory hit happens, but maybe not happening in the short term.

Does capping the memory avoid RocksDB using the memory excessively? Or is there no difference between capping with soft limit vs no capping at all?

I believe there is a difference right. Without the write buffer manager, memory is not accounted towards a single block cache but towards as many DB instances that are present on the node along with memory required for write buffers/other pinned blocks. With the soft limit, RocksDB can allocate more memory temporarily to serve the gets if the existing blocks are not returned to the cache. But once they are returned, some blocks should be evicted from the LRU cache.

Looks to be very tricky to adjust from users' point of view when the error happens
Not sure what you mean here. They might still need to restart the cluster to allow for updated memory limit values to take effect, correct ?

Ideally we will need to rebalance the state if the memory hit happens, but maybe not happening in the short term.
Yup

Thanks for clarifying. So the capped memory usage could be temporarily exceeded but RocksDB will try hard to respect the requirement as soon as it can. That explains the enough difference on rationalization of the capped memory.
I'm OK with soft limit. I'd say hard limit is very useful to restrict the blast radius and the streaming query should just fail if it cannot live with hard limit despite of proper rebalancing of state. But given that we don't have a proper rebalancing of state, hard limiting may play as random failures depending on scheduling of stateful partitions.
We should probably document the behavior though, so that some users would be able to plan for some margin. Could you please update the doc?

They might still need to restart the cluster to allow for updated memory limit values to take effect, correct ?

Yes, that's what I meant.

(Maybe conservative one is to have another conf to determine whether it should be soft or strict, but I know we have too many confs already...)

Thanks! Merging to master.

authors: HeartSaVioR anishshri-db siying


text:
[SPARK-46765][PYTHON][CONNECT] Make shuffle specify the datatype of seed

What changes were proposed in this pull request?
Make shuffle specify the datatype of seed
Why are the changes needed?
shuffle function may fail with an extreme low possibility (~ 2e-10) :
shuffle requires a Long type seed, in an unregistered function, and this Long value is extracted in Planner.
in Scala client the SparkClassUtils.random.nextLong make sure the type;
while in Python, lit(random.randint(0, sys.maxsize)) may return a Literal Integer instead of Literal Long.
In [26]: from pyspark.sql import functions as sf

In [27]: df = spark.createDataFrame([([1, 20, 3, 5],)], ['data'])

In [28]: df.select(sf.shuffle(df.data)).show()
+-------------+
|shuffle(data)|
+-------------+
|[1, 3, 5, 20]|
+-------------+

In [29]: df.select(sf.call_udf("shuffle", df.data, sf.lit(123456789000000))).show()
+-------------+
|shuffle(data)|
+-------------+
|[20, 1, 5, 3]|
+-------------+

In [30]: df.select(sf.call_udf("shuffle", df.data, sf.lit(12345))).show()
...
SparkConnectGrpcException: (org.apache.spark.sql.connect.common.InvalidPlanInput) seed should be a literal long, but got 12345

Another case is uuid, but it is not supported in Python due to namespace conflicts.
I don't find other similar cases.
Does this PR introduce any user-facing change?
no
How was this patch tested?
manually check
Was this patch authored or co-authored using generative AI tooling?
no

ci: https://github.com/zhengruifeng/spark/actions/runs/7578671965/job/20641673985

thanks, merged to master


text:
[SPARK-48477][SQL][TESTS] Use withSQLConf in tests: Refactor CollationSuite, CoalesceShufflePartitionsSuite, SQLExecutionSuite

What changes were proposed in this pull request?

Use withSQLConf in tests when it is appropriate.
Why are the changes needed?

Enforce good practice for setting config in test cases.
Does this PR introduce any user-facing change?

NO
How was this patch tested?

Existing UT
Was this patch authored or co-authored using generative AI tooling?

NO

@cloud-fan

thanks, merging to master!

authors: amaliujia cloud-fan