felipepessoto commented on code in PR #12647:
URL: https://github.com/apache/gluten/pull/12647#discussion_r3669964820
##########
cpp/velox/jni/VeloxJniWrapper.cc:
##########
@@ -717,6 +717,72 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_columnarbatch_VeloxColumnarBatchJ
JNI_METHOD_END(kInvalidObjectHandle)
}
+JNIEXPORT jint JNICALL
Java_org_apache_gluten_columnarbatch_VeloxColumnarBatchJniWrapper_firstNullColumnIndex(
// NOLINT
+ JNIEnv* env,
+ jobject wrapper,
+ jlong veloxBatchHandle,
+ jintArray columnOrdinals) {
+ JNI_METHOD_START
+ auto ctx = getRuntime(env, wrapper);
+ auto runtime = dynamic_cast<VeloxRuntime*>(ctx);
+ auto batch = ObjectStore::retrieve<ColumnarBatch>(veloxBatchHandle);
+ auto safeColumnOrdinals = getIntArrayElementsSafe(env, columnOrdinals);
+
+ const auto numColumnsToCheck = safeColumnOrdinals.length();
+ if (numColumnsToCheck == 0 || batch->numRows() == 0) {
+ return -1;
+ }
+
+ auto veloxBatch =
VeloxColumnarBatch::from(runtime->memoryManager()->getLeafMemoryPool().get(),
batch);
+ auto rowVector = veloxBatch->getRowVector();
+ const auto rowCount = rowVector->size();
+ const auto columnCount = static_cast<int32_t>(rowVector->childrenSize());
+
+ for (int i = 0; i < numColumnsToCheck; ++i) {
+ const auto ordinal = safeColumnOrdinals.elems()[i];
+ GLUTEN_CHECK(ordinal >= 0 && ordinal < columnCount, "Column ordinal
overflow when checking Delta invariant");
+ }
+
+ if (rowVector->mayHaveNulls()) {
+ const auto nullCount = rowVector->getNullCount();
+ if (nullCount.has_value()) {
+ if (*nullCount > 0) {
+ return 0;
+ }
+ } else {
+ for (auto row = 0; row < rowCount; ++row) {
+ if (rowVector->isNullAt(row)) {
+ return 0;
+ }
+ }
+ }
+ }
+
+ for (int i = 0; i < numColumnsToCheck; ++i) {
+ const auto ordinal = safeColumnOrdinals.elems()[i];
+ const auto child = rowVector->childAt(ordinal);
+ VELOX_CHECK_NOT_NULL(child, "Column vector is null when checking Delta
invariant.");
+ if (!child->mayHaveNulls()) {
+ continue;
+ }
+ const auto nullCount = child->getNullCount();
+ if (nullCount.has_value()) {
+ if (*nullCount > 0) {
+ return i;
+ }
+ continue;
+ }
+ for (auto row = 0; row < rowCount; ++row) {
+ if (child->isNullAt(row)) {
+ return i;
+ }
+ }
Review Comment:
**The advertised fast path rarely fires, and this fallback is the expensive
one.**
`nullCount_` is `std::nullopt` for essentially everything a Velox pipeline
hands us: `wrapInDictionary()` leaves it unset, and `ensureWritable()` /
`prepareForReuse()` reset it to `nullopt`. So in practice this per-row loop
*is* the common path, not the "near-free metadata read" described in the PR
body — a virtual `isNullAt()` per row per constrained column, plus dictionary
indirection for dictionary-encoded columns (which is exactly what you get
reading Parquet/Delta sources). That probably explains why the benchmark came
out neutral.
Decoding once and counting bits is both sound (fixes the previous comment)
and considerably faster:
```cpp
DecodedVector decoded(*child, allRows);
if (decoded.mayHaveNulls() &&
bits::countNulls(decoded.nulls(&allRows), 0, rowCount) > 0) {
return i;
}
```
(`DecodedVector::nulls(const SelectivityVector*)` —
`velox/vector/DecodedVector.h:146`; `bits::countNulls` —
`velox/common/base/Nulls.h:59`.) Or, if you'd rather not build a
`DecodedVector`, `bits::countNulls(child->rawNulls(), 0, rowCount)` for the
flat case. Either way it's a word-at-a-time scan instead of per-row virtual
dispatch, and it removes the need to trust `getNullCount()` at all.
##########
cpp/velox/jni/VeloxJniWrapper.cc:
##########
@@ -717,6 +717,72 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_columnarbatch_VeloxColumnarBatchJ
JNI_METHOD_END(kInvalidObjectHandle)
}
+JNIEXPORT jint JNICALL
Java_org_apache_gluten_columnarbatch_VeloxColumnarBatchJniWrapper_firstNullColumnIndex(
// NOLINT
+ JNIEnv* env,
+ jobject wrapper,
+ jlong veloxBatchHandle,
+ jintArray columnOrdinals) {
+ JNI_METHOD_START
+ auto ctx = getRuntime(env, wrapper);
+ auto runtime = dynamic_cast<VeloxRuntime*>(ctx);
+ auto batch = ObjectStore::retrieve<ColumnarBatch>(veloxBatchHandle);
+ auto safeColumnOrdinals = getIntArrayElementsSafe(env, columnOrdinals);
+
+ const auto numColumnsToCheck = safeColumnOrdinals.length();
+ if (numColumnsToCheck == 0 || batch->numRows() == 0) {
+ return -1;
+ }
+
+ auto veloxBatch =
VeloxColumnarBatch::from(runtime->memoryManager()->getLeafMemoryPool().get(),
batch);
+ auto rowVector = veloxBatch->getRowVector();
+ const auto rowCount = rowVector->size();
+ const auto columnCount = static_cast<int32_t>(rowVector->childrenSize());
+
+ for (int i = 0; i < numColumnsToCheck; ++i) {
+ const auto ordinal = safeColumnOrdinals.elems()[i];
+ GLUTEN_CHECK(ordinal >= 0 && ordinal < columnCount, "Column ordinal
overflow when checking Delta invariant");
+ }
+
+ if (rowVector->mayHaveNulls()) {
+ const auto nullCount = rowVector->getNullCount();
+ if (nullCount.has_value()) {
+ if (*nullCount > 0) {
+ return 0;
+ }
+ } else {
+ for (auto row = 0; row < rowCount; ++row) {
+ if (rowVector->isNullAt(row)) {
+ return 0;
+ }
+ }
+ }
+ }
+
+ for (int i = 0; i < numColumnsToCheck; ++i) {
+ const auto ordinal = safeColumnOrdinals.elems()[i];
+ const auto child = rowVector->childAt(ordinal);
+ VELOX_CHECK_NOT_NULL(child, "Column vector is null when checking Delta
invariant.");
+ if (!child->mayHaveNulls()) {
+ continue;
+ }
+ const auto nullCount = child->getNullCount();
+ if (nullCount.has_value()) {
+ if (*nullCount > 0) {
+ return i;
+ }
+ continue;
+ }
Review Comment:
**Correctness risk: `getNullCount()` isn't a sound gate for a NOT NULL
check.**
Both early-outs in this function (the `rowVector` one here, the `child` one
below) treat `getNullCount() == 0` as "no nulls" and skip the scan. Two
problems, checked against the Velox revision we vendor:
1. **Wrapped vectors.** `DictionaryVector::mayHaveNulls()` is overridden to
recurse into the base values (`velox/vector/DictionaryVector.h:72-81`), but
`getNullCount()` is *not* — it returns `BaseVector::nullCount_`, i.e. only the
wrapper's own nulls buffer (`velox/vector/BaseVector.h:198`). A dictionary
whose wrapper nulls were reset (`setNulls(nullptr)` / `clearNulls()` set
`nullCount_ = 0`, see `velox/vector/BaseVector.cpp:620,642,660`) but whose base
values contain nulls at referenced indices passes `mayHaveNulls()` and then
takes the `continue` branch.
2. **Staleness.** `nullCount_` is a cached data-dependent statistic, not a
maintained invariant. `setNull()` does not invalidate it
(`BaseVector.h:416-423`); only `resetDataDependentFlags()` does, and that is
called just from `ensureWritable()` / `prepareForReuse()`
(`BaseVector.h:980-993`).
A wrapper-local or stale `0` means we `continue` and **silently write a NULL
into a NOT NULL column** — no exception, table invariant broken, nothing
downstream catches it. The opposite (a stale `> 0`) fails a valid write with a
spurious `DeltaInvariantViolationException`. Velox itself only uses the
`mayHaveNulls() || getNullCount() == 0` shortcut on flat vectors it owns
(`velox/vector/arrow/Bridge.cpp:510`), where it is an optimization rather than
a correctness gate.
Since this is enforcing a table constraint, I don't think we can trust the
cached count. Either gate the early-out on `child->isFlatEncoding()`, or drop
it entirely and always scan — per the next comment, the scan can be made
cheaper than it is today anyway.
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/constraints/GlutenDeltaInvariantChecker.scala:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.delta.constraints
Review Comment:
This file is byte-identical to
`backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/constraints/GlutenDeltaInvariantChecker.scala`
— I diffed the two copies and there isn't a single character of difference.
`backends-velox/src-delta/main/scala` is already registered as a shared
source root across Delta versions (`pom.xml:1699`), and everything this file
touches (`Constraints.NotNull`, `SchemaUtils.DELTA_COL_RESOLVER`,
`DeltaInvariantViolationException`) is identical on 3.3 and 4.0 — otherwise the
two copies couldn't be identical. Could this live in `src-delta` once instead
of twice?
Worth doing while the file is new: the delta33/delta40 copies of
`GlutenOptimisticTransaction` have already drifted
(`DeltaInvariantCheckerExec(spark, ...)` vs `DeltaInvariantCheckerExec(...)`),
and a duplicated checker will drift the same way — except here a divergence
means one Delta version enforces constraints differently from the other.
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala:
##########
@@ -225,7 +233,8 @@ object GlutenDeltaFileFormatWriter extends LoggingShims {
partitionColumns,
sortColumns,
orderingMatched,
- isNativeWritable
+ isNativeWritable,
+ nativeInvariantChecker
Review Comment:
**The checker is silently dropped in the sibling write branch.**
This call is the `else` of `if (writeFilesOpt.isDefined)` (L204). The
`WriteFilesExec` branch calls the other `executeWrite(sparkSession, plan,
writeSpec, job)` overload, which has no `nativeInvariantChecker` parameter — so
if we ever reach it holding a checker, **no** NOT NULL constraint is enforced
anywhere: the plan no longer carries `DeltaInvariantCheckerExec` either, and
the write silently succeeds with invalid data.
Today that's unreachable — `GlutenOptimisticTransaction` only builds the
checker when `V1WritesUtils.getWriteFilesOpt(empty2NullPlan).isEmpty`, and
nothing applied between there and here (`Transitions.toBatchPlan`,
`GlutenDeltaOptimizedWriterExec`) can introduce a `WriteFilesExec`. But that's
an invariant spread across two files with nothing enforcing it, and the failure
mode is silent data corruption rather than an error.
Could we make it fail loudly — e.g. `require(nativeInvariantChecker.isEmpty,
"...")` in the `writeFilesOpt.isDefined` branch, or thread the checker through
that overload as well? Everything else in this PR fails safe; this is the one
path that doesn't.
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/constraints/GlutenDeltaInvariantChecker.scala:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.delta.constraints
+
+import org.apache.gluten.columnarbatch.VeloxColumnarBatches
+import org.apache.gluten.execution.{PlaceholderRow, TerminalRow}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.delta.constraints.Constraints.NotNull
+import org.apache.spark.sql.delta.schema.{DeltaInvariantViolationException,
SchemaUtils}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+/**
+ * Native write-time invariant checker for constraints that can be validated
without converting
+ * Velox batches back to Spark rows.
+ *
+ * Violations are detected natively from batch null metadata (no per-row
expression evaluation, no
+ * extra operator in the write plan) but thrown on the JVM side, preserving
Delta's typed
+ * [[DeltaInvariantViolationException]] error contract. Offloading Delta's
CheckDeltaInvariant
+ * expression to Velox instead would lose that contract: exceptions raised
inside Velox surface only
+ * as GlutenException with a message string. CHECK constraints and nested NOT
NULL constraints fall
+ * back to Delta's row-based DeltaInvariantCheckerExec.
+ */
+private[delta] case class GlutenDeltaInvariantChecker private (
+ notNullConstraints: Seq[(Int, NotNull)])
+ extends Serializable {
+
+ @transient private lazy val columnOrdinals: Array[Int] =
+ notNullConstraints.map(_._1).toArray
+
+ def wrap(rows: Iterator[InternalRow]): Iterator[InternalRow] = {
+ rows.map {
+ row =>
+ check(row)
+ row
+ }
+ }
+
+ private def check(row: InternalRow): Unit = row match {
+ case _: PlaceholderRow =>
+ case terminal: TerminalRow => check(terminal.batch())
+ case other => checkRow(other)
+ }
+
+ private def check(batch: ColumnarBatch): Unit = {
+ val failedConstraintIndex =
VeloxColumnarBatches.firstNullColumnIndex(batch, columnOrdinals)
+ if (failedConstraintIndex >= 0) {
+ throw
DeltaInvariantViolationException(notNullConstraints(failedConstraintIndex)._2)
+ }
+ }
+
+ private def checkRow(row: InternalRow): Unit = {
+ var i = 0
+ while (i < notNullConstraints.size) {
+ val (ordinal, constraint) = notNullConstraints(i)
+ if (row.isNullAt(ordinal)) {
+ throw DeltaInvariantViolationException(constraint)
+ }
+ i += 1
+ }
+ }
Review Comment:
`notNullConstraints` is a `Seq` — a `List` in practice, since `create`
builds it with `collect`/`map` — so both `.size` and `apply(i)` are O(n), and
`.size` is re-evaluated on every iteration, making this O(n^2) per row. Small
`n`, but this runs for every row of every non-offloaded write, so it seems
worth precomputing.
`columnOrdinals` is already materialized as an `Array` above; a parallel
array of constraints would make this allocation- and traversal-free:
```scala
@transient private lazy val constraints: Array[NotNull] =
notNullConstraints.map(_._2).toArray
private def checkRow(row: InternalRow): Unit = {
var i = 0
while (i < columnOrdinals.length) {
if (row.isNullAt(columnOrdinals(i))) {
throw DeltaInvariantViolationException(constraints(i))
}
i += 1
}
}
```
That would also let the batch path at L63 use
`constraints(failedConstraintIndex)` instead of
`notNullConstraints(failedConstraintIndex)._2`, which pays the same O(n)
`apply` cost.
--
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]
