feat(spark-4): add Apache Spark 4.x connector + batch-ingestion plugin (JDK 21+)

[xiangfu0]

Summary

Adds first-class Apache Spark 4.x support to Pinot, alongside the existing Spark 3 connector and batch-ingestion plugin. The two new modules live next to their Spark 3 siblings:

New module	Replaces (for Spark 4 users)	Scope
pinot-connectors/pinot-spark-4-connector	pinot-spark-3-connector	Spark DataSourceV2 read + write
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-4	pinot-batch-ingestion-spark-3	Segment generation + push runners

Both modules compile against Apache Spark 4.1.1 (the latest 4.x release) and are activated only on JDK 21+ via the pinot-spark-4-connector / pinot-batch-ingestion-spark-4 profile in their respective parent poms (<jdk>[21,)</jdk>). On JDK 11/17 the reactor still builds Spark 3 and completely skips Spark 4, so existing Spark 3 users are unaffected.

The PR also deprecates pinot-spark-3-connector / pinot-batch-ingestion-spark-3: with Pinot's master branch raising the Java baseline to JDK 21, the Spark 3 modules are now JDK 21 + Spark 3.5.5+ only and slated for removal in the next minor release. New users should adopt Spark 4.

Why a separate module?

Apache Spark 4 requires JDK 17+, is Scala 2.13 only (no more 2.12), ships Jetty 12 / Jakarta Servlet 5, and made a few small DataSourceV2 API tweaks. Running the Spark 3 connector under Spark 4 doesn't work cleanly. The modules sit side-by-side so:

• Spark 3.5.5+ users keep the existing pinot-spark-3-connector (deprecated, one release of overlap).
• Spark 4.0/4.1 users pick up pinot-spark-4-connector.
• The Maven coordinates, artifact names, and shaded-jar layout mirror the Spark 3 pair, so deployment automation (docker assembly, plugin discovery) doesn't need special cases beyond a JDK choice.

Netty

Spark 4.1.x depends on netty 4.2.x (the new io.netty.channel.kqueue.KQueueIoHandler / EpollIoHandler APIs). Pinot-wide netty stays at 4.1.x — only the pinot-batch-ingestion-spark-4 module locally overrides netty.version to 4.2.7.Final so its embedded-SparkContext test can construct NettyRpcEnv. The spark-4-connector module does not instantiate a real SparkContext in tests, so it inherits Pinot's netty without conflict.

End-to-end tutorial

The End-to-end Docker tutorial walks through:

1. Starting Pinot via docker run apachepinot/pinot:latest QuickStart
2. Building the shaded connector jar
3. Pulling apache/spark:4.1.1-scala2.13-java21-python3-ubuntu (Spark 4.1.1 + JDK 21, no custom build needed)
4. Reading baseballStats (97,889 rows) from Spark 4 via PySpark
5. Creating a target table and writing a DataFrame through the connector
6. Pushing the produced segment tars to the controller (REST or via the SparkSegment*PushJobRunner runners)
7. Reading the written data back through the connector

Every command was executed against apachepinot/pinot:latest and apache/spark:4.0.0 before being checked in (the netty bump to 4.1.1 was applied after end-to-end validation; the Docker tutorial now references the published apache/spark:4.1.1-java21 image variants).

Quick usage (read)

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("pinot-read").getOrCreate()

df = (spark.read.format("pinot")
      .option("table", "baseballStats")
      .option("tableType", "offline")
      .option("controller", "pinot-controller:9000")
      .option("broker", "pinot-broker:8000")
      .load())

df.filter("yearID >= 2000").groupBy("teamID").count().show()

Invoke with spark-submit (Spark 4.1.1 on JDK 21):

spark-submit \
  --jars pinot-spark-4-connector-<VER>-shaded.jar,commons-lang3-3.20.0.jar \
  --conf spark.driver.extraClassPath=commons-lang3-3.20.0.jar \
  --conf spark.executor.extraClassPath=commons-lang3-3.20.0.jar \
  my_read_job.py

The prepended commons-lang3-3.20.0.jar works around Spark 4 shipping an older version that lacks ObjectUtils.getIfNull(...). See the tutorial's Known gotchas for the full list.

Quick usage (write)

(df.write.format("pinot")
    .mode("append")
    .option("table", "myTable")
    .option("tableType", "OFFLINE")
    .option("segmentNameFormat", "{table}_{partitionId:03}")
    .option("invertedIndexColumns", "name,category")
    .option("timeColumnName", "ts")
    .option("timeFormat", "EPOCH|MILLISECONDS")
    .option("timeGranularity", "1:MILLISECONDS")
    .option("controller", "pinot-controller:9000")
    .save("s3a://my-bucket/pinot-segments/myTable"))

This produces segment tar files at the save(...) path. Then push them to the controller — either REST for one-offs:

for seg in /path/to/*.tar.gz; do
  curl -X POST -H 'UPLOAD_TYPE: SEGMENT' -F "file=@${seg}" \
    "http://pinot-controller:9000/v2/segments?tableName=myTable&tableType=OFFLINE"
done

…or in production use the runners from pinot-batch-ingestion-spark-4 via a SegmentGenerationJobSpec YAML (see the module's README):

• SparkSegmentGenerationJobRunner — build segments from Avro/CSV/Parquet input
• SparkSegmentTarPushJobRunner — upload tars to the controller
• SparkSegmentUriPushJobRunner — register tars already staged in deep storage (S3/GCS/HDFS)
• SparkSegmentMetadataPushJobRunner — metadata-only push for large segments (avoid streaming the whole tar through the controller)

⚠️ Overwrite semantics

df.write.mode("overwrite").save(...), df.writeTo(...).overwrite(...), and SQL INSERT OVERWRITE all fail fast with UnsupportedOperationException. Pinot's write path can only append new segments — it cannot atomically drop or replace rows matching a Spark predicate. To replace data, drop the table first (controller REST) or use a push runner with REFRESH / consistent-push enabled. Prior to the fixes in this PR, overwrite calls silently appended without replacement, leaving stale data queryable.

Compatibility matrix

Module	Spark	Scala	JDK (build & runtime)
pinot-spark-3-connector / pinot-batch-ingestion-spark-3 (deprecated)	3.5.5+	2.12 / 2.13	21 only (master baseline)
pinot-spark-4-connector / pinot-batch-ingestion-spark-4	4.1.x (compiled against 4.1.1)	2.13 only	21 only

Both connector jars are now compiled with --release 21 (class file major version 65). Deployments stuck on JDK 17 should pin to Pinot 1.5.x until they can upgrade.

Correctness fixes (in addition to the Spark 4 port)

Multi-round review surfaced numerous correctness issues — many pre-existing in the Spark 3 code that was being ported. All fixes are applied symmetrically to both modules with regression tests on each side.

Filter pushdown

1. Compound filter soundness. Or / And / Not were marked as unconditionally supported even if a child was a filter type the connector does not recognize. Spark would remove the compound from its residual filter list, while compileFilter silently returned None, so Pinot received no filter at all and returned unfiltered rows. Fix: isFilterSupported is now recursive — compounds are accepted only when every child is supported.

2. Null-leaf gating across all comparison operators. EqualNullSafe(_, null) and In(_, [..., null, ...]) originally rendered attr <op> null via compileValue's fallback branch; comparison operators (EqualTo / LessThan / LessThanOrEqual / GreaterThan / GreaterThanOrEqual) had the same gap. Fix: every leaf with a null literal is rejected and falls through to Spark post-scan; an In with a null value array (NPE risk) is also rejected. Compound gating ensures the rejection propagates through enclosing And/Or/Not.

3. Collection-shaped literal values. EqualTo("attr", Seq(1, 2)) rendered as attr = List(1, 2) via value.toString — malformed SQL. Fix: new isPushableValue whitelist allows only String / Number / Boolean / Timestamp / Date; everything else (Seq / Map / Vector / Set / case classes) routes to post-scan.

4. LIKE pushdown escaping + cross-module invariant. StringStartsWith / StringEndsWith / StringContains interpolated raw user-supplied strings into LIKE '...' without escaping %, _, \, or '. Values containing these changed the predicate meaning (or produced invalid SQL). Fix: escapeLikeLiteral escapes LIKE wildcards, backslashes, and single-quotes; each LIKE predicate carries an ESCAPE '\' clause; values containing a literal \ are rejected upstream because Pinot's RegexpPatternConverterUtils#likeToRegexpLike does not round-trip \\ correctly. A new regression test runs the emitted SQL fragment back through RegexpPatternConverterUtils.likeToRegexpLike and verifies the resulting regex (via Matcher#find, mirroring RegexpLikeConstFunctions) matches the original literal — pinning the cross-module \ escape contract.

5. Identifier escaping. escapeAttr now wraps any column name in double-quotes and doubles inner " so SQL stays well-formed even for column names containing stray quotes.

Type translator

6. Throw on unknown types instead of returning null. Previously translateType returned null for unsupported Spark types, with the caller checking and rethrowing — a future caller forgetting the null-check would silently skip the field. Fix: translator throws UnsupportedOperationException directly with the offending type in the message.

7. Reject TimestampType / DateType. Spark stores TimestampType as microseconds-since-epoch, while Pinot's broker convention is milliseconds-since-epoch (TimestampUtils#toMillisSinceEpoch). The prior translator silently produced values 1000× too large. Fix: translator rejects with a hint to cast upstream (col("ts").cast("long") / 1000); same for DateType. Documented in the spark-3 README as a breaking change.

8. Reject ArrayType(BinaryType) / ArrayType(DecimalType). Pinot has no MV BYTES / MV BIG_DECIMAL writer support. Fix: rejected at translator with a clear error rather than failing deep in the segment driver.

Writer (PinotDataWriter)

9. SupportsOverwrite[V2] silently dropped the overwrite intent. The writer advertised SupportsOverwrite (V1) / SupportsOverwriteV2 (V2) but ignored filters/predicates and returned an append-only writer. df.write.mode("overwrite") and df.writeTo(...).overwrite(...) callers believed matching rows would be replaced. Fix: overwrite(...), truncate(), and canOverwrite(...) all now reject. truncate() is load-bearing: SupportsOverwrite[V2] extends SupportsTruncate, and Spark's V2Writes rule dispatches df.write.mode("overwrite") → truncate() (which defaulted to returning this → silent append) rather than overwrite([AlwaysTrue]).

10. Per-partition temp-dir leak. Files.createTempDirectory in generateSegment() was never deleted, so long-running executors accumulated uncompressed segment trees under the JVM tmpdir. Fix: track the temp dir on a module-private field; wrap commit() in try/finally; also delete from abort() and close() (idempotent via null-out). closeOnce() wraps the reader close in try/finally so cleanup runs even if bufferedRecordReader.close() throws.

11. PinotWrite.commit/abort used println. Fix: switched to SLF4J. Driver-side abort(messages) now best-effort deletes leftover segment tars at savePath for every SuccessWriterCommitMessage, tolerating null entries, unknown subclasses, and malformed/empty savePath URIs.

12. ArrayData accessor correctness. record.getArray(idx).array.map(_.asInstanceOf[T]) only works on GenericArrayData (test scaffolding) and throws on UnsafeArrayData (real workloads); for StringType arrays it would ClassCastException on every row (UTF8String → String). Fix: use type-specific ArrayData.toIntArray() / toLongArray() / … accessors and explicit getUTF8String → toString conversion for StringType.

13. Null-aware writer + null-safe time-column tracking. Spark's primitive accessors silently return 0 / false for null cells in UnsafeRow, and getString / getDecimal NPE on null. Fix: the per-field branch in internalRowToGenericRow checks record.isNullAt(idx) first and marks the field via addNullValueField instead of synthesizing zeros. The time-column-tracking branch in write() also honors isNullAt(timeColumnIndex) (one null cell would have collapsed startTime to 0, corrupting {startTime} / {endTime} in segmentNameFormat) and dispatches on the column's actual Spark type — record.getLong(idx) on an IntegerType slot in UnsafeRow reads 8 bytes from a 4-byte field and returns garbage.

14. segmentNameFormat validation. Width spec on a non-numeric variable ({table:N}) previously crashed with ClassCastException at commit time; an unknown placeholder ({instanceId}) crashed with NoSuchElementException. Fix: both now fail fast at job submission with a clear IllegalArgumentException listing the supported placeholder names and noting that width spec is numeric-only.

15. Add scalar DecimalType writer support. Schema translator already mapped to BIG_DECIMAL; writer now correctly reads via record.getDecimal(idx, precision, scale).toJavaBigDecimal.

Other

16. PinotDataSource classpath-collision guard (Spark 4 only). Both v3 and v4 register the "pinot" short name; if both jars end up on a classpath, Spark resolves non-deterministically. The Spark 4 connector now probes for the v3 class on every constructor call (Class.forName is JVM-cached), checks both the module classloader and the thread context classloader, swallows LinkageError, and throws with a clear message instructing the user to remove one jar. Escape hatch: -Dpinot.spark.connector.skip-conflict-guard=true.

17. PinotBufferedRecordReader.next(reuse). A defensive .copy() is preserved (an earlier round of review tried removing it for performance; the more recent review correctly noted that the underlying invariant — that no TransformPipeline transformer mutates a value object in place — is undocumented at the SPI level, so the deep copy is restored to defend against silent pass-2 corruption).

Regression tests for every numbered fix above are added to both pinot-spark-3-connector and pinot-spark-4-connector.

Test plan

• ./mvnw -pl pinot-connectors/pinot-spark-4-connector,pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-4 -am test on JDK 21 with Spark 4.1.1 — 56 spark-4-connector scalatest cases plus 3 batch-ingestion-spark-4 testng cases, all green.
• ./mvnw -pl pinot-connectors/pinot-spark-3-connector -am test on JDK 21 — 54 scalatest cases, all green.
• ./mvnw -pl pinot-integration-tests -am test-compile — integration tests still compile unchanged.
• spotless:apply checkstyle:check license:check apache-rat:check clean on all modified modules.
• Live end-to-end validation against apachepinot/pinot:latest + apache/spark:4.0.0 (+JDK 21): read 97,889-row baseballStats table, write a 10-row DataFrame to a fresh OFFLINE table, push via controller REST, verify roundtrip through both the Pinot SQL broker and the connector's own read path.

Known coverage gaps (tracked for follow-up)

• End-to-end Spark 4 integration test — A SparkSegmentMetadataPushIntegrationTest equivalent for Spark 4 was added then dropped (commit f18309c95e) because the Spark 4 runtime (Jakarta Servlet 5) conflicts with pinot-integration-tests' Spark 3 + javax stack in a shared Maven module. Restoring end-to-end integration coverage requires either a dedicated pinot-spark-4-integration-tests module or failsafe with a separate classpath realm. Noted in the pinot-batch-ingestion-spark-4 README.
• Push-runner unit tests — SparkSegmentMetadataPushJobRunner / SparkSegmentTarPushJobRunner / SparkSegmentUriPushJobRunner have no unit tests in either the Spark 3 or the Spark 4 batch-ingestion module today; only SparkSegmentGenerationJobRunnerTest exercises the in-process driver path. Adding coverage is orthogonal to this PR and would benefit both modules.
• PinotWrite.abort(...) partial-failure leftover tars — driver-side cleanup runs only for tasks that produced a SuccessWriterCommitMessage; tasks that copied a tar to savePath and then died before commit message return leave the tar behind. The contract-level guard against silent overwrite is provided by the overwrite/truncate rejection (Add a Gitter chat badge to README.md #9 above); this runtime-level cleanup is best-effort.

Notes for reviewers

• Commons-lang3 shading — currently users must prepend commons-lang3:3.20+ on spark.{driver,executor}.extraClassPath. A follow-up PR can relocate org.apache.commons.lang3.* in the connector's shade config to eliminate this step.
• Scala 2.12 — Apache Spark 4 is Scala 2.13 only. The modules have requireProperty scala.compat.version=2.13 enforcer rules so -Pscala-2.12 builds fail fast with a clear message.
• -Ppinot-fastdev — activation is JDK-based, so -Ppinot-fastdev on JDK 21 also builds the Spark 4 modules. The Spark 4 shade is disabled under -Ppinot-fastdev via each module's own profile. Explicit opt-out: -P!pinot-spark-4-connector / -P!pinot-batch-ingestion-spark-4.
• Local netty override — pinot-batch-ingestion-spark-4/pom.xml pins netty.version=4.2.7.Final so the embedded SparkContext in SparkSegmentGenerationJobRunnerTest can construct its NettyRpcEnv (Spark 4.1+ requires netty 4.2 APIs). The override is module-scoped; the rest of Pinot stays on netty 4.1.x.

🤖 Generated with Claude Code

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 63.41%. Comparing base (4499bf5) to head (f0cd1ad).
⚠️ Report is 6 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18261      +/-   ##
============================================
+ Coverage     63.38%   63.41%   +0.03%     
+ Complexity     1668     1644      -24     
============================================
  Files          3252     3253       +1     
  Lines        198661   198764     +103     
  Branches      30770    30791      +21     
============================================
+ Hits         125925   126051     +126     
+ Misses        62666    62642      -24     
- Partials      10070    10071       +1     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-21	63.41% <ø> (+0.03%)	⬆️
temurin	63.41% <ø> (+0.03%)	⬆️
unittests	63.41% <ø> (+0.03%)	⬆️
unittests1	55.38% <ø> (+0.04%)	⬆️
unittests2	34.93% <ø> (+0.03%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[xiangfu0]

Found a few high-signal correctness issues; see inline comments.

[xiangfu0]

Marking compound filters as supported without recursively checking their children can silently drop predicates. For example, an Or/And/Not containing an unsupported child is removed from Spark residual filtering, then compileFilter returns None, so Pinot can return rows that do not satisfy the original Spark filter. Recursively accept compounds only when every child compiles; otherwise return the original filter as residual.

[xiangfu0]

The LIKE pushdowns interpolate the raw Spark value while equality predicates escape string literals. Values containing quotes, percent, or underscore change the Pinot SQL predicate semantics, which can produce silent wrong results. Escape SQL literals and LIKE wildcards for StringStartsWith/StringEndsWith/StringContains before pushing these filters.

[xiangfu0]

This advertises SupportsOverwriteV2 but drops the overwrite predicates and builds an unconditional writer. Spark overwrite-by-filter callers will believe Pinot replaced the matching data, while this path only writes new segment files, which can leave old matching segments in place and produce duplicate or stale query results. Either implement predicate-aware replacement semantics or reject overwrite predicates explicitly.