You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
76 lines
1.9 KiB
76 lines
1.9 KiB
package spark
|
|
|
|
import (
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
func setupSparkIssue8234Env(t *testing.T) *TestEnvironment {
|
|
t.Helper()
|
|
|
|
env := NewTestEnvironment()
|
|
if !env.dockerAvailable {
|
|
t.Skip("Docker not available, skipping Spark S3 integration test")
|
|
}
|
|
|
|
env.StartSeaweedFS(t)
|
|
t.Cleanup(func() { env.Cleanup(t) })
|
|
|
|
createObjectBucket(t, env, "test")
|
|
env.startSparkContainer(t)
|
|
return env
|
|
}
|
|
|
|
func TestSparkS3AppendIssue8234Regression(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("Skipping Spark integration test in short mode")
|
|
}
|
|
|
|
env := setupSparkIssue8234Env(t)
|
|
|
|
script := `
|
|
import sys
|
|
import pyspark.sql.types as T
|
|
|
|
target = "s3a://test/test_table/"
|
|
sc = spark.sparkContext
|
|
df = sc.parallelize([1, 2, 3]).toDF(T.IntegerType())
|
|
|
|
df.write.format("parquet").mode("append").save(target)
|
|
print("FIRST_APPEND_OK")
|
|
|
|
try:
|
|
df.write.format("parquet").mode("append").save(target)
|
|
print("SECOND_APPEND_OK")
|
|
spark.stop()
|
|
sys.exit(0)
|
|
except Exception as err:
|
|
message = str(err)
|
|
print("SECOND_APPEND_FAILED")
|
|
print(message)
|
|
if "Copy Source must mention the source bucket and key" in message:
|
|
print("ISSUE_8234_REGRESSION")
|
|
spark.stop()
|
|
sys.exit(4)
|
|
spark.stop()
|
|
sys.exit(2)
|
|
`
|
|
|
|
code, output := runSparkPyScript(t, env.sparkContainer, script, env.s3Port)
|
|
if code != 0 {
|
|
if strings.Contains(output, "ISSUE_8234_REGRESSION") {
|
|
t.Fatalf("issue #8234 regression detected; output:\n%s", output)
|
|
}
|
|
t.Fatalf("Spark script exited with code %d; output:\n%s", code, output)
|
|
}
|
|
|
|
if !strings.Contains(output, "FIRST_APPEND_OK") {
|
|
t.Fatalf("expected first append success marker in output, got:\n%s", output)
|
|
}
|
|
if !strings.Contains(output, "SECOND_APPEND_OK") {
|
|
t.Fatalf("expected second append success marker in output, got:\n%s", output)
|
|
}
|
|
if strings.Contains(output, "Copy Source must mention the source bucket and key") {
|
|
t.Fatalf("unexpected issue #8234 error in output:\n%s", output)
|
|
}
|
|
}
|