You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/20 19:01:15 UTC
(spark) branch master updated: [SPARK-60380][SQL] Fix Parquet filter pushdown on nanosecond timestamp columns read with a lower precision
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 4ba8bfe54cbe [SPARK-60380][SQL] Fix Parquet filter pushdown on nanosecond timestamp columns read with a lower precision
4ba8bfe54cbe is described below
commit 4ba8bfe54cbe7e68b944971b2f3b7af330447361
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Oct 20 12:01:02 2026 -0700
[SPARK-60380][SQL] Fix Parquet filter pushdown on nanosecond timestamp columns read with a lower precision
### What changes were proposed in this pull request?
This PR aims to fix the Parquet filter pushdown on `TIMESTAMP_LTZ(p)` and `TIMESTAMP_NTZ(p)` columns with `p < 9`.
The Parquet reader truncates the stored nanoseconds to the read precision, while Parquet evaluates a pushed down predicate against the stored values. Like SPARK-60358 for `TIME`, the `TIMESTAMP(NANOS)` filter ops now override `ParquetFilterOps.acceptsReadType`, so only `IsNull` and `IsNotNull` are pushed down for such a column.
- #59606
### Why are the changes needed?
The query results are wrong. For example, `00:00:00.123456789` is read as `00:00:00.1234567` with `TIMESTAMP_NTZ(7)`, but the pushed down `=` compares the literal with the stored value.
```
$ bin/spark-shell -c spark.sql.timestampNanosTypes.enabled=true
scala> sql("SELECT CAST('1970-01-01 00:00:00.123456789' AS TIMESTAMP_NTZ(9)) AS ts UNION ALL SELECT CAST('1970-01-01 00:00:01' AS TIMESTAMP_NTZ(9))").coalesce(1).write.parquet("/tmp/nanos")
scala> val df = spark.read.schema("ts TIMESTAMP_NTZ(7)").parquet("/tmp/nanos")
val df: org.apache.spark.sql.DataFrame = [ts: timestamp_ntz(7)]
scala> df.collect()
val res1: Array[org.apache.spark.sql.Row] = Array([1970-01-01T00:00:00.123456700], [1970-01-01T00:00:01])
scala> df.where("ts = CAST('1970-01-01 00:00:00.1234567' AS TIMESTAMP_NTZ(7))").collect()
val res2: Array[org.apache.spark.sql.Row] = Array()
```
`<=>`, `<=` and `IN` return no rows in the same way. The expected result is `Array([1970-01-01T00:00:00.123456700])`, which is the result with `spark.sql.parquet.filterPushdown=false`.
### Does this PR introduce _any_ user-facing change?
Yes, this fixes the wrong query results in the unreleased branches (`master`, `branch-4.x` and `branch-4.3`), including [v4.3.0-rc1](https://github.com/apache/spark/releases/tag/v4.3.0-rc1) (2026-08-31) and [v4.3.0-rc2](https://github.com/apache/spark/releases/tag/v4.3.0-rc2) (2026-10-12). The predicates with a value on the columns of precision 7 and 8 are not pushed down anymore, even for the files written by Spark. The nanosecond timestamp types are a preview feature, which is disab [...]
### How was this patch tested?
Pass the CIs with the newly added test cases. The existing test cases in `ParquetFilterSuite` and `ParquetTimestampNanosSuite` are updated for the new behavior, including the one with `TODO(SPARK-60380)`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59641 from dongjoon-hyun/parquet-nanos-narrow-read-pushdown.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../types/ops/TimestampNanosParquetOps.scala | 14 ++
.../datasources/parquet/ParquetFilterSuite.scala | 174 +++++++++++++++------
.../parquet/ParquetTimestampNanosSuite.scala | 22 +--
3 files changed, 155 insertions(+), 55 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
index 701751e95950..2ed1fe1eca2f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/TimestampNanosParquetOps.scala
@@ -320,6 +320,16 @@ private[ops] object TimestampNanosParquetOps {
private def epochNanosInRange(encode: => JLong): Boolean =
try { encode; true } catch { case _: ArithmeticException => false }
+ // Below nanosecond precision, the reader truncates the stored epoch-nanoseconds to the read
+ // precision (see nanosConverter and TimestampNanosVectorUpdater), while a pushed down predicate
+ // compares the stored epoch-nanoseconds. E.g. a stored 00:00:00.123456789 is read as
+ // 00:00:00.1234567 for TIMESTAMP_NTZ(7), so a pushed down `ts = 00:00:00.1234567` would skip it.
+ private def acceptsNanosReadType(readType: DataType): Boolean = readType match {
+ case t: TimestampLTZNanosType => t.precision == TimestampLTZNanosType.NANOS_PRECISION
+ case t: TimestampNTZNanosType => t.precision == TimestampNTZNanosType.NANOS_PRECISION
+ case _ => true
+ }
+
/**
* Parquet filter-pushdown ops for the nanosecond timestamp types, registered in
* [[ParquetTypeOps.filterOpsList]]. Filter dispatch is keyed on the file's on-disk encoding, so
@@ -339,6 +349,8 @@ private[ops] object TimestampNanosParquetOps {
case _ => false
}
+ override def acceptsReadType(readType: DataType): Boolean = acceptsNanosReadType(readType)
+
override protected def toLong(value: Any): JLong =
instantToEpochNanos(value.asInstanceOf[Instant])
}
@@ -352,6 +364,8 @@ private[ops] object TimestampNanosParquetOps {
case _ => false
}
+ override def acceptsReadType(readType: DataType): Boolean = acceptsNanosReadType(readType)
+
override protected def toLong(value: Any): JLong =
localDateTimeToEpochNanos(value.asInstanceOf[LocalDateTime])
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
index 4daaa5eecdff..342fc230ad71 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
@@ -2416,52 +2416,50 @@ abstract class ParquetFilterSuite extends ParquetTest with SharedSparkSession {
withSQLConf(
SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> "1") {
- Seq(7, 8, 9).foreach { p =>
- // NTZ column: filter values are LocalDateTime. Use a sub-microsecond fraction so the
- // epoch-nanos encoding (not truncated micros) is exercised.
- val ntzReadSchema = new StructType().add("c", TimestampNTZNanosType(p))
- val ntzSchema = new SparkToParquetSchemaConverter(conf).convert(ntzReadSchema)
- val ntzFilters = createParquetFilters(ntzSchema, readSchema = Some(ntzReadSchema))
- val ntz = LocalDateTime.parse("2020-01-01T12:34:56.000000123")
- val ntzHi = LocalDateTime.parse("2020-01-01T12:34:57.000000123")
- assert(ntzFilters.createFilter(sources.IsNull("c")).exists(_.isInstanceOf[Eq[_]]))
- assert(ntzFilters.createFilter(sources.IsNotNull("c")).exists(_.isInstanceOf[NotEq[_]]))
- assert(ntzFilters.createFilter(sources.EqualTo("c", ntz)).exists(_.isInstanceOf[Eq[_]]))
- assert(ntzFilters.createFilter(sources.EqualNullSafe("c", ntz))
- .exists(_.isInstanceOf[Eq[_]]))
- assert(ntzFilters.createFilter(sources.Not(sources.EqualTo("c", ntz)))
- .exists(_.isInstanceOf[NotEq[_]]))
- assert(ntzFilters.createFilter(sources.LessThan("c", ntz)).exists(_.isInstanceOf[Lt[_]]))
- assert(ntzFilters.createFilter(sources.LessThanOrEqual("c", ntz))
- .exists(_.isInstanceOf[LtEq[_]]))
- assert(ntzFilters.createFilter(sources.GreaterThan("c", ntz)).exists(_.isInstanceOf[Gt[_]]))
- assert(ntzFilters.createFilter(sources.GreaterThanOrEqual("c", ntz))
- .exists(_.isInstanceOf[GtEq[_]]))
- assert(ntzFilters.createFilter(sources.In("c", Array[Any](ntz, ntzHi)))
- .exists(_.isInstanceOf[FilterIn[_]]))
-
- // LTZ column: filter values are Instant.
- val ltzReadSchema = new StructType().add("c", TimestampLTZNanosType(p))
- val ltzSchema = new SparkToParquetSchemaConverter(conf).convert(ltzReadSchema)
- val ltzFilters = createParquetFilters(ltzSchema, readSchema = Some(ltzReadSchema))
- val ltz = Instant.parse("2020-01-01T12:34:56.000000123Z")
- val ltzHi = Instant.parse("2020-01-01T12:34:57.000000123Z")
- assert(ltzFilters.createFilter(sources.IsNull("c")).exists(_.isInstanceOf[Eq[_]]))
- assert(ltzFilters.createFilter(sources.IsNotNull("c")).exists(_.isInstanceOf[NotEq[_]]))
- assert(ltzFilters.createFilter(sources.EqualTo("c", ltz)).exists(_.isInstanceOf[Eq[_]]))
- assert(ltzFilters.createFilter(sources.EqualNullSafe("c", ltz))
- .exists(_.isInstanceOf[Eq[_]]))
- assert(ltzFilters.createFilter(sources.Not(sources.EqualTo("c", ltz)))
- .exists(_.isInstanceOf[NotEq[_]]))
- assert(ltzFilters.createFilter(sources.LessThan("c", ltz)).exists(_.isInstanceOf[Lt[_]]))
- assert(ltzFilters.createFilter(sources.LessThanOrEqual("c", ltz))
- .exists(_.isInstanceOf[LtEq[_]]))
- assert(ltzFilters.createFilter(sources.GreaterThan("c", ltz)).exists(_.isInstanceOf[Gt[_]]))
- assert(ltzFilters.createFilter(sources.GreaterThanOrEqual("c", ltz))
- .exists(_.isInstanceOf[GtEq[_]]))
- assert(ltzFilters.createFilter(sources.In("c", Array[Any](ltz, ltzHi)))
- .exists(_.isInstanceOf[FilterIn[_]]))
- }
+ // NTZ column: filter values are LocalDateTime. Use a sub-microsecond fraction so the
+ // epoch-nanos encoding (not truncated micros) is exercised.
+ val ntzReadSchema = new StructType().add("c", TimestampNTZNanosType(9))
+ val ntzSchema = new SparkToParquetSchemaConverter(conf).convert(ntzReadSchema)
+ val ntzFilters = createParquetFilters(ntzSchema, readSchema = Some(ntzReadSchema))
+ val ntz = LocalDateTime.parse("2020-01-01T12:34:56.000000123")
+ val ntzHi = LocalDateTime.parse("2020-01-01T12:34:57.000000123")
+ assert(ntzFilters.createFilter(sources.IsNull("c")).exists(_.isInstanceOf[Eq[_]]))
+ assert(ntzFilters.createFilter(sources.IsNotNull("c")).exists(_.isInstanceOf[NotEq[_]]))
+ assert(ntzFilters.createFilter(sources.EqualTo("c", ntz)).exists(_.isInstanceOf[Eq[_]]))
+ assert(ntzFilters.createFilter(sources.EqualNullSafe("c", ntz))
+ .exists(_.isInstanceOf[Eq[_]]))
+ assert(ntzFilters.createFilter(sources.Not(sources.EqualTo("c", ntz)))
+ .exists(_.isInstanceOf[NotEq[_]]))
+ assert(ntzFilters.createFilter(sources.LessThan("c", ntz)).exists(_.isInstanceOf[Lt[_]]))
+ assert(ntzFilters.createFilter(sources.LessThanOrEqual("c", ntz))
+ .exists(_.isInstanceOf[LtEq[_]]))
+ assert(ntzFilters.createFilter(sources.GreaterThan("c", ntz)).exists(_.isInstanceOf[Gt[_]]))
+ assert(ntzFilters.createFilter(sources.GreaterThanOrEqual("c", ntz))
+ .exists(_.isInstanceOf[GtEq[_]]))
+ assert(ntzFilters.createFilter(sources.In("c", Array[Any](ntz, ntzHi)))
+ .exists(_.isInstanceOf[FilterIn[_]]))
+
+ // LTZ column: filter values are Instant.
+ val ltzReadSchema = new StructType().add("c", TimestampLTZNanosType(9))
+ val ltzSchema = new SparkToParquetSchemaConverter(conf).convert(ltzReadSchema)
+ val ltzFilters = createParquetFilters(ltzSchema, readSchema = Some(ltzReadSchema))
+ val ltz = Instant.parse("2020-01-01T12:34:56.000000123Z")
+ val ltzHi = Instant.parse("2020-01-01T12:34:57.000000123Z")
+ assert(ltzFilters.createFilter(sources.IsNull("c")).exists(_.isInstanceOf[Eq[_]]))
+ assert(ltzFilters.createFilter(sources.IsNotNull("c")).exists(_.isInstanceOf[NotEq[_]]))
+ assert(ltzFilters.createFilter(sources.EqualTo("c", ltz)).exists(_.isInstanceOf[Eq[_]]))
+ assert(ltzFilters.createFilter(sources.EqualNullSafe("c", ltz))
+ .exists(_.isInstanceOf[Eq[_]]))
+ assert(ltzFilters.createFilter(sources.Not(sources.EqualTo("c", ltz)))
+ .exists(_.isInstanceOf[NotEq[_]]))
+ assert(ltzFilters.createFilter(sources.LessThan("c", ltz)).exists(_.isInstanceOf[Lt[_]]))
+ assert(ltzFilters.createFilter(sources.LessThanOrEqual("c", ltz))
+ .exists(_.isInstanceOf[LtEq[_]]))
+ assert(ltzFilters.createFilter(sources.GreaterThan("c", ltz)).exists(_.isInstanceOf[Gt[_]]))
+ assert(ltzFilters.createFilter(sources.GreaterThanOrEqual("c", ltz))
+ .exists(_.isInstanceOf[GtEq[_]]))
+ assert(ltzFilters.createFilter(sources.In("c", Array[Any](ltz, ltzHi)))
+ .exists(_.isInstanceOf[FilterIn[_]]))
}
}
@@ -2557,6 +2555,90 @@ abstract class ParquetFilterSuite extends ParquetTest with SharedSparkSession {
}
}
+ test("SPARK-60380: filter pushdown - nanosecond timestamp read with a lower precision") {
+ // A TIMESTAMP(9) value 1970-01-01 00:00:00.123456789 read as TIMESTAMP(7) is truncated to
+ // 00:00:00.1234567, while Parquet evaluates a pushed down predicate against the stored value.
+ // For example, a pushed down `ts = 1970-01-01 00:00:00.1234567` would skip the row group that
+ // holds the matching row.
+ withSQLConf(
+ SQLConf.TIMESTAMP_NANOS_TYPES_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ Seq(
+ "TIMESTAMP_LTZ" -> ((s: String) => Instant.parse(s"1970-01-01T${s}Z")),
+ "TIMESTAMP_NTZ" -> ((s: String) => LocalDateTime.parse(s"1970-01-01T$s"))
+ ).foreach { case (tsType, toValue) =>
+ withTempPath { dir =>
+ sql(s"SELECT CAST(c AS $tsType(9)) AS ts FROM VALUES " +
+ "('1970-01-01 00:00:00.123456789'), ('1970-01-01 00:00:01'), (NULL) AS t(c)")
+ .coalesce(1).write.parquet(dir.getCanonicalPath)
+ Seq(7, 8).foreach { p =>
+ val readDF = spark.read.schema(s"ts $tsType($p)").parquet(dir.getCanonicalPath)
+ // The stored 00:00:00.123456789 is read as `v`, e.g. 00:00:00.1234567 for p = 7.
+ val v = "00:00:00." + "123456789".take(p)
+ def lit(s: String): String = s"CAST('1970-01-01 $s' AS $tsType($p))"
+ def check(condition: String, expected: Seq[String]): Unit = {
+ checkAnswer(readDF.where(condition), expected.map(s => Row(toValue(s))))
+ }
+ withAllParquetReaders {
+ check(s"ts = ${lit(v)}", Seq(v))
+ check(s"ts <=> ${lit(v)}", Seq(v))
+ check(s"ts != ${lit(v)}", Seq("00:00:01"))
+ check(s"ts < ${lit("00:00:01")}", Seq(v))
+ check(s"ts <= ${lit(v)}", Seq(v))
+ check(s"ts > ${lit(v)}", Seq("00:00:01"))
+ check(s"ts >= ${lit(v)}", Seq(v, "00:00:01"))
+ Seq(1, 10).foreach { threshold =>
+ withSQLConf(
+ SQLConf.PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD.key -> s"$threshold") {
+ check(s"ts IN (${lit(v)}, ${lit("00:00:02")})", Seq(v))
+ }
+ }
+ checkAnswer(readDF.where("ts IS NULL"), Row(null))
+ check("ts IS NOT NULL", Seq(v, "00:00:01"))
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("SPARK-60380: TIMESTAMP(NANOS) predicates with a value are not pushed down below " +
+ "precision 9") {
+ // The reader truncates TIMESTAMP(NANOS) values to a read precision below 9, so only IsNull and
+ // IsNotNull are pushed down then. Both LTZ (isAdjustedToUTC=true) and NTZ are covered.
+ val schema = MessageTypeParser.parseMessageType(
+ """message root {
+ | optional int64 ltz(TIMESTAMP(NANOS,true));
+ | optional int64 ntz(TIMESTAMP(NANOS,false));
+ |}""".stripMargin)
+ val ltz = Instant.parse("1970-01-01T00:00:00.1234567Z")
+ val ntz = LocalDateTime.parse("1970-01-01T00:00:00.1234567")
+ Seq(7, 8, 9).foreach { p =>
+ val readSchema = new StructType()
+ .add("ltz", TimestampLTZNanosType(p))
+ .add("ntz", TimestampNTZNanosType(p))
+ val parquetFilters = createParquetFilters(schema, readSchema = Some(readSchema))
+ Seq[(String, Any)]("ltz" -> ltz, "ntz" -> ntz).foreach { case (name, value) =>
+ Seq(sources.IsNull(name), sources.IsNotNull(name)).foreach { filter =>
+ assert(parquetFilters.createFilter(filter).isDefined, s"$filter, precision $p")
+ }
+ Seq(
+ sources.EqualTo(name, value),
+ sources.EqualNullSafe(name, value),
+ sources.Not(sources.EqualTo(name, value)),
+ sources.LessThan(name, value),
+ sources.LessThanOrEqual(name, value),
+ sources.GreaterThan(name, value),
+ sources.GreaterThanOrEqual(name, value),
+ sources.In(name, Array[Any](value))
+ ).foreach { filter =>
+ assert(parquetFilters.createFilter(filter).isDefined === (p == 9),
+ s"$filter, precision $p")
+ }
+ }
+ }
+ }
+
// ----------------------------------------------------------------------------------------------
// Shredded-variant filter pushdown (SPARK-55817).
//
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
index b167da3fe7d6..44523253525d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTimestampNanosSuite.scala
@@ -433,7 +433,7 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
// scan reads more than the matching rows but fewer than all rows) and (b) the full query result
// equals reading the same predicate with pushdown disabled. Record-level filtering is disabled
// so ONLY row-group-level skipping is exercised; the non-vectorized reader is used so
- // `stripSparkFilter` reflects exactly what Parquet returned.
+ // `stripSparkFilter` reflects exactly what Parquet returned. (a) holds only for precision 9.
withNanosEnabled {
Seq(7, 8, 9).foreach { p =>
val frac = "000000123".take(p)
@@ -459,15 +459,20 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
// Row-group-level skipping: with record filtering off, a stripped scan returns whole
// surviving row groups. Fewer than all rows (skipping happened) but more than the one
- // matching row (record filtering is off).
+ // matching row (record filtering is off). Below precision 9, the filter is not pushed
+ // down because the reader truncates the stored values to the read precision.
withSQLConf(
SQLConf.PARQUET_RECORD_FILTER_ENABLED.key -> "false",
SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key -> "true",
SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
val df = spark.read.parquet(path).filter(predicate)
val actual = stripSparkFilter(df).collect().length
- assert(actual > 1 && actual < 1024,
- s"Expected row-group skipping (p=$p, $kind) but scanned $actual of 1024 rows.")
+ if (p == 9) {
+ assert(actual > 1 && actual < 1024,
+ s"Expected row-group skipping (p=$p, $kind) but scanned $actual of 1024 rows.")
+ } else {
+ assert(actual === 1024, s"Expected no row-group skipping (p=$p, $kind).")
+ }
}
// Results must be identical whether or not the filter is pushed to Parquet.
@@ -671,9 +676,7 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
}
}
- // TODO(SPARK-60380): make pushdown respect the read precision, then expect the pushdown and
- // no-pushdown results to match for every predicate.
- test("filter pushdown under a lower-precision read schema compares raw nanoseconds") {
+ test("filter pushdown under a lower-precision read schema matches the no-pushdown results") {
withNanosEnabled {
withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
withAllParquetReaders {
@@ -690,11 +693,12 @@ class ParquetTimestampNanosSuite extends QueryTest with ParquetTest with SharedS
spark.read.schema(s"ts $typ(7)").parquet(file.getCanonicalPath)
.filter(predicate).count()
}
- // The pushed bound is compared with the raw value, so the row group is skipped.
+ // A predicate with a value is not pushed down below precision 9, so the row group
+ // that holds the matching row is not skipped.
Seq(s"ts = $lit", s"ts <=> $lit", s"ts IN ($lit, $other)", s"ts <= $lit")
.foreach { predicate =>
assert(count(predicate, pushdown = false) === 1, predicate)
- assert(count(predicate, pushdown = true) === 0, predicate)
+ assert(count(predicate, pushdown = true) === 1, predicate)
}
Seq(s"ts < $lit", s"ts >= $lit", s"ts > $lit").foreach { predicate =>
assert(count(predicate, pushdown = true) ===
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org