gengliangwang commented on code in PR #55637:
URL: https://github.com/apache/spark/pull/55637#discussion_r3211964996
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Changelog.java:
##########
@@ -61,15 +65,16 @@
* snapshots) that derive {@code _commit_timestamp} from wall-clock time
at
* commit time naturally satisfy both requirements.
* {@code _commit_timestamp} must be non-{@code NULL} on every row of a
streaming
- * read engaging post-processing. The row-level rewrite raises
- * {@code CHANGELOG_CONTRACT_VIOLATION.NULL_COMMIT_TIMESTAMP} on any row
that
- * violates this; without the guard a NULL group key would never satisfy
the
- * watermark eviction predicate and the row would sit in state
indefinitely</li>
+ * read engaging post-processing; both the row-level Aggregate path and
the
+ * netChanges {@code transformWithState} path raise
+ * {@code CHANGELOG_CONTRACT_VIOLATION.NULL_COMMIT_TIMESTAMP} on a
violation</li>
* </ul>
* <p>
- * Streaming reads support carry-over removal and update detection but not net
change
- * computation. The latter requires reasoning over the entire requested range
and is
- * batch-only.
+ * Streaming reads support carry-over removal, update detection, and net change
+ * computation. Net change collapses are kept in the state store keyed by row
identity;
+ * row identities only touched in the latest observed commit are held back
until either a
+ * later commit (with strictly greater `_commit_timestamp`) advances the
global watermark
+ * past them, or the source terminates.
Review Comment:
Created https://github.com/apache/spark/pull/55776 for this.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CdcNetChangesStatefulProcessor.scala:
##########
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.{Encoder, Row}
+import org.apache.spark.sql.catalyst.CatalystTypeConverters
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+import org.apache.spark.sql.catalyst.util.TypeUtils
+import org.apache.spark.sql.connector.catalog.Changelog
+import org.apache.spark.sql.streaming._
+import org.apache.spark.sql.types.StructType
+
+/**
+ * StatefulProcessor that incrementalises CDC net-change computation for
streaming reads.
+ *
+ * The batch path (`ResolveChangelogTable.injectNetChangeComputation`) uses a
Catalyst
+ * `Window` partitioned by `rowId` and ordered by `(_commit_version,
change_type_rank)` to
+ * extract the first and last events per row identity, then applies the SPIP
collapse
+ * matrix on `(existedBefore, existsAfter)`. That `Window` is rejected on
streaming
+ * queries (`NON_TIME_WINDOW_NOT_SUPPORTED_IN_STREAMING`).
+ *
+ * This processor reuses the same SPIP collapse matrix with
`transformWithState`, applied
+ * per watermark window rather than over the full requested version range.
Per-row-identity
+ * state stores the first event ever observed and the most-recent event
observed; an event
+ * time timer keyed on `_commit_timestamp` advances with each batch and fires
once the
+ * global watermark passes the latest event time observed for the key, at
which point the
+ * SPIP matrix is evaluated and the net result is emitted. See the paragraph
below for how
+ * the per-window collapse differs from batch netChanges' range-scoped
collapse.
+ *
+ * Output schema: identical to the connector's changelog schema.
+ *
+ * Streaming netChanges is incremental: per-row-identity state is cleared once
its current
+ * net result is emitted (timer fire or end-of-stream flush). Subsequent
commits on the same
Review Comment:
Created https://github.com/apache/spark/pull/55776 for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
