From 483a5b794e772e58a80c17e5025e71ec1b182a7e Mon Sep 17 00:00:00 2001
From: Ankit Khullar <ankit@grata.com>
Date: Fri, 29 May 2026 20:01:25 +0530
Subject: [PATCH] Fix sink schema validation: throw a proper error for
 value-less schemas
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

PulsarSinks.validateQuery is meant to reject a sink schema with no value
fields (only reserved meta fields like __key/__topic). Two bugs made the
guard ineffective:

1. The exception was constructed but never thrown — pulsarSinkInvalidSchema
   appeared as a bare statement with no `throw`, so validation fell through.
2. The factory referenced error class "PULSAR_SINK_INVALID_SCHEMA", which is
   not defined in error/pulsar-error-classes.json (the defined key is
   "PULSAR_SINK_INVALID_SCHEMA_TYPE"), so even evaluating it raised
   INTERNAL_ERROR instead of the intended message.

- Add the missing `throw` in validateQuery.
- Point the factory at the defined error class PULSAR_SINK_INVALID_SCHEMA_TYPE.
- Add PulsarSinkValidateQuerySuite covering the reject and accept paths.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../spark/sql/pulsar/PulsarExceptions.scala   |  2 +-
 .../apache/spark/sql/pulsar/PulsarSinks.scala |  2 +-
 .../pulsar/PulsarSinkValidateQuerySuite.scala | 44 +++++++++++++++++++
 3 files changed, 46 insertions(+), 2 deletions(-)
 create mode 100644 src/test/scala/org/apache/spark/sql/pulsar/PulsarSinkValidateQuerySuite.scala

diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala
index dee4445..9340c5d 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarExceptions.scala
@@ -40,7 +40,7 @@ object PulsarExceptions {
 
   def pulsarSinkInvalidSchema: PulsarIllegalArgumentException = {
     new PulsarIllegalArgumentException(
-      errorClass = "PULSAR_SINK_INVALID_SCHEMA",
+      errorClass = "PULSAR_SINK_INVALID_SCHEMA_TYPE",
       messageParameters = Map()
     )
   }
diff --git a/src/main/scala/org/apache/spark/sql/pulsar/PulsarSinks.scala b/src/main/scala/org/apache/spark/sql/pulsar/PulsarSinks.scala
index 6e91fe3..395082a 100644
--- a/src/main/scala/org/apache/spark/sql/pulsar/PulsarSinks.scala
+++ b/src/main/scala/org/apache/spark/sql/pulsar/PulsarSinks.scala
@@ -129,7 +129,7 @@ private[pulsar] object PulsarSinks extends Logging {
       schema.filter(n => !PulsarOptions.MetaFieldNames.contains(n.name))
 
     if (valuesExpression.length == 0) {
-      PulsarExceptions.pulsarSinkInvalidSchema
+      throw PulsarExceptions.pulsarSinkInvalidSchema
     }
 
     checkForUnsupportedType(valuesExpression.map(_.dataType))
diff --git a/src/test/scala/org/apache/spark/sql/pulsar/PulsarSinkValidateQuerySuite.scala b/src/test/scala/org/apache/spark/sql/pulsar/PulsarSinkValidateQuerySuite.scala
new file mode 100644
index 0000000..3387a48
--- /dev/null
+++ b/src/test/scala/org/apache/spark/sql/pulsar/PulsarSinkValidateQuerySuite.scala
@@ -0,0 +1,44 @@
+/*
+ * Licensed 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.pulsar
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.types.{BinaryType, StringType}
+
+class PulsarSinkValidateQuerySuite extends SparkFunSuite {
+
+  private val topic = Some("persistent://public/default/t")
+
+  test("validateQuery rejects a schema with no value fields") {
+    // Only a reserved meta field (__key) is present, so there is no value column
+    // to write. validateQuery must throw a PulsarIllegalArgumentException carrying
+    // PULSAR_SINK_INVALID_SCHEMA_TYPE. Previously this raised INTERNAL_ERROR (the
+    // error class did not match error/pulsar-error-classes.json) and the exception
+    // was never actually thrown (missing `throw`), so the check was a no-op.
+    val schema = Seq(AttributeReference(PulsarOptions.KeyAttributeName, BinaryType)())
+    val e = intercept[PulsarIllegalArgumentException] {
+      PulsarSinks.validateQuery(schema, topic)
+    }
+    assert(e.getCondition === "PULSAR_SINK_INVALID_SCHEMA_TYPE")
+  }
+
+  test("validateQuery accepts a schema with at least one value field") {
+    val schema = Seq(
+      AttributeReference(PulsarOptions.KeyAttributeName, BinaryType)(),
+      AttributeReference("value", StringType)())
+    // Should not throw.
+    PulsarSinks.validateQuery(schema, topic)
+  }
+}