malinjawi commented on code in PR #12647:
URL: https://github.com/apache/gluten/pull/12647#discussion_r4350731833
##########
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:
Good catch, thanks. You're right that the cached count can't be trusted for
an invariant check, so in a007c3f94 I dropped `getNullCount()` entirely. Each
constrained column now goes through a `DecodedVector`, so nulls in the base of
a dictionary or constant wrapper are included, and the nulls are counted with
`bits::countNulls`. Row-level nulls are read straight from the RowVector's own
nulls buffer, since that one is always flat.
I also moved the scan into `VeloxColumnarBatch::firstNullColumnIndex` so it
could be unit tested. `VeloxColumnarBatchTest` now covers a stale null count of
0, a dictionary with a null count of 0 over a null base value, a dictionary
whose indices skip the null, a null constant, a null row, and a child that's
longer than the batch.
##########
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:
Done, pretty much your snippet: decode with `allRows`, then
`bits::countNulls(decoded.nulls(&allRows), 0, numRows)`.
I was curious about the neutral benchmark as well, so I re-ran it and timed
the JNI call on its own. On a 2M-row append (~490 batches) the check takes
about 2 ms, both before and after this change. In these pipelines the vectors
reaching the writer happened to take the old shortcut, so the per-row loop
wasn't what made the old numbers neutral. Either way the new version is the
correct one and costs nothing measurable. The native path now comes out about
the same as an unconstrained write, and the row CHECK fallback is 21–28%
slower. The numbers are in the description.
##########
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:
Agreed, this shouldn't depend on an invariant spread across two files. Both
writers (`src-delta33` and `src-delta4`) now
`require(nativeInvariantChecker.isEmpty, ...)` in the `writeFilesOpt.isDefined`
branch (a007c3f94), so it fails loudly instead of writing unchecked. I didn't
thread the checker through that overload, because `WriteFilesExec` runs its own
tasks and there's no iterator there to wrap.
The other half, not building a checker when the plan has a `WriteFilesExec`,
now lives in one place, `GlutenDeltaInvariantChecker.forWrite` (3da00cd46),
instead of being repeated in both transactions.
##########
backends-velox/src-delta/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:
Fixed in a007c3f94. The checker now takes `columnOrdinals: Array[Int]` and
`constraints: Array[NotNull]` as parallel arrays, so both the batch path and
the row path are plain array lookups. I also added
`GlutenDeltaInvariantCheckerSuite` for the row path; it doesn't need the native
library.
##########
backends-velox/src-delta/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:
Good point, moved. There's a single copy now, in
`backends-velox/src-delta/main/scala/org/apache/spark/sql/delta/constraints/GlutenDeltaInvariantChecker.scala`
(a007c3f94). Since #13271 the 4.x side is `src-delta4`, so this one file
covers Delta 3.3 and 4.4. It also compiles against Delta 2.4 for the Spark 3.4
build, where it isn't used.
The same drift concern applied to the code deciding whether to use the
checker, which was still duplicated in both `GlutenOptimisticTransaction`s, so
that moved into the same file as `forWrite` (3da00cd46).
--
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]
