Repository navigation
test: fail Comet tests when a native plan mislabels a timestamp's timezone - #6868
Conversation
…ezone Native plans must label every TimestampType value Timestamp(µs, "UTC") and every TimestampNTZType value with no timezone. The label never changes the values, so checkSparkAnswer cannot see a wrong one, and the casts at ScanExec and the shuffle writer's SchemaAlignExec repair it silently. Add the internal test-only config spark.comet.testing.strictTimestampLabels.enabled, which CometTestBase turns on. With it, CometExecIterator checks each batch a native plan returns against the plan's Spark output types, ScanExec fails instead of casting a relabelled input timestamp, and the native shuffle writer fails planning instead of aligning one.
sunchao
left a comment
There was a problem hiding this comment.
Reviewed the full 19-file diff from 0691bd39f75da61e02aa23ce188de141bd623252 to 5e758b23c1932d623e224d191fa9f51c37fb5800. The PR is not a draft. The snapshot and live discussion contain no existing reviews or comments. No introduced P1/P2 issues found within this review.
Summary
- Prior state and problem: Row comparisons could miss incorrect timestamp timezone labels, while scan casts and shuffle schema alignment could erase the evidence. Such labels can affect subsequent native expressions even when collected values look correct.
- Design approach: Add an internal
spark.comet.testing.strictTimestampLabels.enabledswitch, enabled byCometTestBase, to reject label mismatches at native output, scan-input, and shuffle-writer boundaries. - Correctness: Checked recursive timestamp matching, dictionary-value lookup, output-type propagation, config serialization, and rejected-batch cleanup. The checks preserve timestamp values and leave ordinary casts unchanged when disabled. No reproducible P1/P2 correctness issue was identified.
- Compatibility analysis: Compared Spark’s
ArrowUtilssources for 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. All distinguish microsecond timestamps with a timezone fromTimestampNTZTypewithout one. Requiring exactlyUTCenforces Comet’s existing internal convention. No version-specific shim change is needed for this check. - Key design decisions: Keep the switch false by default, serialize its resolved boolean value, inspect dictionary values rather than indices, and report nested field paths. The JVM rejection path closes the batch before throwing.
- Implementation sketch:
NativeUtilcompares imported Arrow fields with Spark output types supplied throughCometExecRDDand iterator callers. Rust sharesdescribe_timestamp_label_mismatchbetweenScanExecand shuffle-input alignment, with the flag propagated through execution settings and the planner. - Performance: Validation traverses type metadata without scanning rows or copying payload buffers. Recursive checks are disabled by default. Output-type metadata and flag plumbing still add some per-plan work, but there is no evidence of a P1/P2 performance regression. No throughput benchmark was run.
- Design: Checking boundaries before label-repairing casts directly addresses the testing blind spot and keeps production behavior stable. The boundary checks are straightforward to reason about.
- Abstraction & complexity: The shared Rust helper avoids duplicating recursive matching. A separate JVM helper is appropriate for Arrow Java fields, Spark types, and dictionary providers. No unnecessary framework or actionable complexity problem was identified.
- Behavioral changes worth calling out: Strict Comet tests now reject labels such as
Etc/UTCor+00:00whereUTCis expected, despite equivalent offsets. Comparison withbranch-1.1ate9efd9f764ee0a59b7898ff028d6985d4a7a28e1confirms this PR adds intended test-only enforcement while retaining default boundary casts. Incorrect labels consumed entirely inside one native plan remain outside these checks’ coverage. - Suggested improvements: No additional change meeting the P1/P2 reporting bar was identified. The existing nested-type, dictionary, cleanup, and strict-versus-default tests cover the behavior introduced here.
Routed skills: review-comet-pr, review-comet-ffi-pr, review-comet-shuffle-pr, and review-comet-memory-pr, with the timezone and relevant contributor guides.
Exact-head CI: All executed check runs passed, including required-check aggregates, Linux Comet suites across Spark 3.4–4.2, Rust tests, and TPC-H/TPC-DS verification. The inspected Rust and JVM logs identify the merge of this exact head into the requested base and show the new regression tests passing. See the primary run and additional Spark-profile run. Spark’s own SQL suites, external Iceberg suites, and macOS tests were skipped.
Local validation: Three Rust tests using the exact timestamp-label helper and two Scala probe groups passed, covering nested fields and dictionary lookup. These were isolated helper checks, not full integration runs. The normal Cargo test command stopped because the offline dependency cache lacked DataFusion 55.2.0. Full native/JVM integration validation therefore relies on CI. The project checkout remains unchanged.
Resolve the conflict in CometExecRDD, where this branch's reportScanInputMetrics parameter and main's outputTypes parameter (apache#6868) were both appended to the constructor and to the apply factory. Keep both, with reportScanInputMetrics last. Every caller passes them by name. Drop the second .region(Region.US_EAST_1) the merge left in CometS3TestBase's S3 client builder: main (apache#6887) added the same call.
apache#6868 passes each native plan's Spark output types to CometExecRDD, so that under spark.comet.testing.strictTimestampLabels.enabled CometExecIterator checks the labels on every batch the plan returns. CometDeltaNativeScanExec builds its own CometExecRDD when it is not fused into a larger native block, and it passed no output types, so the check skipped it. Pass them, as CometNativeScanExec and CometIcebergNativeScanExec do.
…ng) (apache#5365) * feat: add JVM-planned native Delta Lake scan contrib module Adds an optional contrib/delta-spark module that claims delta-spark DSv1 scans through CometScanContrib and runs them on Comet's shared native parquet path, including main's JVM-exact field-name folding for case-insensitive footer matching. Deletion vectors are decoded natively into per-file ParquetAccessPlans that DataFusion intersects with row-group and page-index pruning, so DV skips and page skips compose in a single scan. Scans the native path cannot serve safely (DML row-index reads, unsupported filesystem schemes, userinfo-bearing authorities, credential-provider-only auth, S3 config divergence, multi-store shapes) fall back to Spark with an explained reason. Co-authored-by: Scott Schenkein <schenksj@yahoo.com> Co-authored-by: Aditya Vaish <adivaish@microsoft.com> * fix: contain provider link failures, gate Hadoop-only S3 endpoints, and check inline DV sizes Review round: discovery skips a provider whose class cannot link, the S3 claim gate declines a scheme-less endpoint with SSL off and any assumed-role STS endpoint, inline deletion vectors are checked against their descriptor size, row-group offsets use checked arithmetic, and the rebase arms return the input batch untouched when nothing predates the cutover. Build and CI: one delta.version, the spark test-jar only under the delta profile, a plain-runner job for the MinIO suite, a tighter change filter, and docs that name every Spark line the workflow runs. * fix: keep every leaf requested when the name fold fails during rebase planning * ci: activate the delta profile for the prerequisite installs and set up the MinIO runner with sudo * perf: pass null-free modern batches through the rebase untouched with one vectorised minimum * chore: drop redundant interpolators flagged by scalafix * ci: save the Delta contrib Maven caches on main only * fix: adapt the Delta contrib to the prepared plan-data API CometExecRDD now carries the driver-computed plan fingerprint, and PlanDataInjector splits parsing the partition-invariant common from injecting a partition. The Delta scan passes the fingerprint, and DeltaPlanDataInjector parses the common DeltaSparkScan once in prepareCommon and merges only the file list in inject. DeltaPlanDataInjectorSuite pins the contract, including reuse and stale replacement through core's shared memo. * chore: drop redundant string interpolators in CometDeltaNativeScanSuite * fix: keep the rebase pass-through for nullable all-modern batches The null-count guard sent any column holding a single null through try_unary, allocating a fresh buffer per batch even when every valid value is modern. Null-free arrays keep the vectorised minimum; arrays with nulls take one validity-aware pass, and the pass-through test asserts the same Arc comes back for masked dates and timestamps under every policy. * test: let COMET_DELTA_S3_REQUIRED turn a missing MinIO into a hard failure The dedicated runner job exists only to run CometDeltaS3Suite, so a cancelled run and a real one looked the same. The job sets the switch, and the suite then fails instead of cancelling when Docker is unavailable or MinIO does not start. * docs: guard every Delta scan config row and drop the generated-table markers Nothing fills the CONFIG_TABLE markers on the Delta page, so the table is hand-maintained. The guard now renders every DeltaScanConf entry with its default, flags stale rows, and fails rather than cancels when the page cannot be found. * ci: run the Delta contrib suites for one Spark profile in the queue The single Delta job ran three profiles on every queued change that touched native or spark sources. Spark 3.5 now runs in the queue and 4.0 and 4.1 run nightly, with run-delta-tests opting a pull request into all three, the same split the Iceberg tiers use. The reusable workflow takes the profile as an input, and the MinIO, feature-off and dev script jobs run once, on the 3.5 call. * ci: use setup-java v6 in the Delta MinIO job like the other workflows * fix: share one memory consumer across a partition's deletion vectors and drain the listener bus in the test helpers Every DV'd file registered its own memory consumer and the reservation lived in the file's extensions until the task ended, so a partition with many DV'd files lowered the fair pool's limit for every other operator in the task. attach_access_plans now registers one consumer per call and hands each file an empty reservation from it. A test pool that copies the fair limit check pins it. The plan capture helpers unregistered their listener as soon as the body returned, so a late callback was missed. They now drain the listener bus, and collectTaskInputMetrics does the same instead of polling with a floor. DV'd files are split like any other claimed file. The planning helper comment says so and what each split does today, and a test reads one small-row-group file as many splits. contrib/delta-spark reads its own delta.spark.version property so the in-tree contrib's delta.version, which spark/pom.xml overrides per Spark profile, no longer diverges from it. * style: drop a redundant string interpolator in the Delta scan suite * docs: list every check the Delta scan reuses from the core native scan gate * fix: choose the timestamp rebase policy from the requested type The native Delta scan picked a timestamp leaf's rebase policy from its physical Arrow type, while Spark's ParquetVectorUpdaterFactory keys on the Spark type the column is read as. Two cases differed. An INT96 or adjusted INT64 column read as TIMESTAMP_NTZ was checked or rebased, though Spark never rebases NTZ reads, so an ancient value failed under EXCEPTION. A timezone-free INT64 column read as TIMESTAMP was never rebased, though Spark applies the datetime rebase mode whatever isAdjustedToUTC says, so EXCEPTION returned the row and LEGACY returned it unshifted. The walk that already pairs each physical leaf with its requested type now records the leaves read as TIMESTAMP_NTZ and the timezone-free leaves read as TIMESTAMP, and the policy follows that record. Dates keep the date policy, including a DATE column read as TIMESTAMP_NTZ, which Spark 4.x rebases. A metadata-free timezone-free file read as TIMESTAMP under LEGACY is now refused, as adjusted files without a recorded writer time zone already are, because the rebase needs the JVM's time zone tables. * perf: limit Delta deletion-vector selectors to each file split * docs: describe per-split row selectors in the Delta scan planning comment Each split builds selectors only for the row groups that start inside it and shrinks its reservation to match, so the comment no longer says the access plan and reservation cover the whole file. * chore: tighten Delta scan comments Shorten several comment blocks in the Delta scan and rebase code and describe each test case directly. Rename the 1500-01-01 day rebase test after what it checks. * test: check Delta S3 key coverage in the native tests The Delta suite requires its S3 config comparator to cover every fs.s3a.* key the native S3 client reads, but it only runs in the merge queue. A core change that adds a key passed the pull request checks and then failed in the queue. A Rust test in s3.rs now reads S3ConfigKeyConsumers from DeltaScanSupport.scala and fails when a native key is missing, so the pull request run reports it. A Delta suite test checks that the same text parse matches the compiled key list, so a reformat of that block fails in the Delta change's own run instead of breaking the Rust test later. * fix: skip calendar rebase checks when Spark reads a date or timestamp as a number Spark picks a numeric updater before any temporal handling when the requested type is numeric, so a date read as an integer type or a timestamp read as BIGINT comes back as the stored value with no calendar check. The native Delta scan still applied the column's temporal rebase policy and failed ancient values under EXCEPTION mode, for example a metadata-free TIMESTAMP(NANOS) file read as BIGINT with spark.sql.legacy.parquet.nanosAsLong after CONVERT TO DELTA. Leaves requested as one of Spark's numeric pairings now get no rebase policy, while temporal requests keep theirs. * fix: decline split-dependent metadata and resolve SSE-C like Hadoop in the Delta scan Two gaps in the native Delta scan's admission checks: - `_metadata.file_block_start` and `file_block_length` came back with the wrong split's values on a split file, because DataFusion assigns a row group to the split holding its first page while Spark uses its midpoint. Core already falls back for these columns; the Delta scan claimed the read before core's guard ran. It now declines when either column is used above the scan, so deletion-vector reads, which carry the whole `_metadata` struct, stay native. - The SSE-C check resolved the encryption algorithm on the raw Hadoop configuration. S3AFileSystem resolves it on the bucket-propagated view, so a bucket override reached through a `${...}` reference could select SSE-C while the check saw AES256. The algorithm is now resolved on both views and the scan declines unless both are on the allowlist. A credential provider path is read only when every entry is a local keystore or the user provider, since Hadoop drops providers backed by other filesystems. Decline reasons no longer print a resolved configuration value: the algorithm reason names only a known algorithm, and the provider-class reason names the key and bucket. * fix: rebase INT32 values Spark reads as DATE in the Delta scan Spark's Parquet reader rebases an INT32 column requested as DATE whether or not the file annotates it as a date. The native Delta scan treated an unannotated INT32 leaf as already proleptic, so a legacy or metadata-free file converted with CONVERT TO DELTA returned 1500-01-10 where Spark returns 1500-01-01 under LEGACY, and returned the row where Spark refuses it under EXCEPTION. An INT32 leaf requested as DATE now takes the file's date policy, and the stored days are rebased before the schema adapter casts them. * fix: restore the Delta scan with its live pruning filters on transition revert When transition reversion hands a stage back to Spark, it restores each Comet node's original plan. The Delta native scan inherited the default, which returned the original FileSourceScanExec still holding the SubqueryAdaptiveBroadcastExec placeholder that AQE had already rewritten on the live scan, so execution failed with "SubqueryAdaptiveBroadcastExec does not support the execute() code path". The scan now restores the original plan with its current partition filters and logical link, as CometNativeScanExec does. Data filters keep the original scan's list, because the live list also carries subquery predicates taken from the parent filter. * test: fix strict-warnings errors in the Delta contrib suites Literal Hadoop ${...} references now sit in s-strings with $$ so the strings are unchanged, spark.range bounds use Long literals, an unused helper is removed, and an inferred existential type is spelled out. * ci: run the Delta contrib suites nightly and compile the contrib in the PR tier The Spark 3.5 Delta suites leave the merge queue and run nightly with 4.0 and 4.1, still on demand through run-delta-tests. The strict warnings job now compiles with -Pdelta, and changes under contrib/delta-spark route to the Linux PR build, so a change that breaks the contrib build on Spark 3.5 fails on its own pull request instead of in the queue. * docs: say the Delta contrib jar is built from source and mark the guide experimental No release publishes comet-contrib-delta-spark yet, so the setup step points at the contrib README's build steps, and the sidebar entry carries the same (experimental) marker as the page heading. * test: check the timestamp labels of a Delta scan that runs alone apache#6868 passes each native plan's Spark output types to CometExecRDD, so that under spark.comet.testing.strictTimestampLabels.enabled CometExecIterator checks the labels on every batch the plan returns. CometDeltaNativeScanExec builds its own CometExecRDD when it is not fused into a larger native block, and it passed no output types, so the check skipped it. Pass them, as CometNativeScanExec and CometIcebergNativeScanExec do. --------- Co-authored-by: Scott Schenkein <schenksj@yahoo.com> Co-authored-by: Aditya Vaish <adivaish@microsoft.com> Co-authored-by: Andy Grove <agrove@apache.org> Co-authored-by: Chao Sun <chao@openai.com>
Which issue does this PR close?
There is no issue for this. It follows up the timezone audit (#6335), whose label bugs, such as #6328 and #6330, shipped with passing tests.
Rationale for this change
Native plans label every
TimestampTypevalueTimestamp(µs, "UTC")and everyTimestampNTZTypevalue with no timezone. The label never changes the values, socheckSparkAnswerand friends cannot see a wrong one: they compare collected rows,export_batchhands each array to the JVM with whatever type the plan produced, andCometPlainVector.getLongreads the raw microseconds. The casts atScanExecand in the shuffle writer'sSchemaAlignExecrepair a wrong label as well.Inside a native plan the label does matter. Another zone breaks comparisons and
CASE/coalescecoercion against"UTC"arrays, and a missing one makes kernels such ashourand casts to string read the value as wall-clock time. So a test of an expression's direct output passes even when the expression breaks a query that composes it, which is how #6328 and #6330 shipped with passing tests.What changes are included in this PR?
An internal, test-only config,
spark.comet.testing.strictTimestampLabels.enabled, whichCometTestBaseturns on for every Comet suite. It defaults to false, so nothing changes outside the test suites. With it on:CometExecIteratorchecks each batch a native plan returns against the plan's Spark output types, including nested and dictionary-encoded columns (NativeUtil.checkTimestampLabels). The callers that know their output types pass them in a newoutputTypesparameter:CometNativeExec,CometNativeScanExec,CometIcebergNativeScanExec, the native limit, andCometTakeOrderedAndProjectExec.ScanExecfails, instead of casting, on an input timestamp whose label differs from the declared type.SchemaAlignExec, when the child labels a timestamp differently from the declared schema. The error includes the native plan, so it points at the expression.The native checks share
describe_timestamp_label_mismatchindatafusion-comet-common.These are boundary checks. A wrong label that is consumed entirely inside one native plan, such as
hour(f(ts))wherefdrops the label, never reaches a boundary, and still shows up only as wrong values in a non-UTC session.How are these changes tested?
describe_timestamp_label_mismatch, the strictScanExec, and the strict shuffle-writer alignment.NativeUtilSuitetests for the JVM check, covering nested types, maps, dictionary values imported over FFI, and closing a batch it rejects.CometExecSuite's config serde test covers the new natively read key.timestamp_truncchanged back to labelling its result with the session timezone (its behavior before fix: keep the input's timezone label on native date_trunc results #6345),SELECT c0, date_trunc('HOUR', c0) FROM tbland aGROUP BYover the truncated value both passedcheckSparkAnswerwith the check off. With it on, they failed at the plan output and at the shuffle writer. WithCometSparkToColumnarExeclabelling its timestampsEtc/UTC, theScanExeccheck failed the same way. On unmodified code the same probes raised nothing.mainhas no label drift on those paths. They cover the SQL file suite, casts, Parquet reads, expressions, joins, aggregates, windows, shuffle, the in-memory cache, fuzz, the Python runner, and the Iceberg read and write suites.