chore: Improved Input Validation (#44)

* chore: Misc. code improv

Signed-off-by: Anush008 <anushshetty90@gmail.com>

* chore: Bump version

Signed-off-by: Anush008 <anushshetty90@gmail.com>

---------

Signed-off-by: Anush008 <anushshetty90@gmail.com>

Author: Anush008

pom.xml

@@ -6,7 +6,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>io.qdrant</groupId>
 	<artifactId>spark</artifactId>
-	<version>2.3.4</version>
+	<version>2.3.5</version>
 	<name>qdrant-spark</name>
 	<url>https://github.com/qdrant/qdrant-spark</url>
 	<description>An Apache Spark connector for the Qdrant vector database</description>

src/main/java/io/qdrant/spark/QdrantDataWriter.java

@@ -51,25 +51,34 @@ private void writeBatch(int retries) {
 
     try {
       doWriteBatch();
-      pointsBuffer.clear();
     } catch (Exception e) {
-      LOG.error("Exception while uploading batch to Qdrant: {}", e.getMessage());
+      LOG.error("Exception while uploading batch to Qdrant", e);
       if (retries > 0) {
-        LOG.info("Retrying upload batch to Qdrant");
+        LOG.info("Retrying upload batch to Qdrant (retries remaining: {})", retries);
         writeBatch(retries - 1);
+        return;
       } else {
-        throw new RuntimeException(e);
+        pointsBuffer.clear();
+        throw new RuntimeException(
+            "Failed to write batch after " + (options.retries + 1) + " attempts", e);
       }
     }
+    pointsBuffer.clear();
   }
 
   private void doWriteBatch() throws Exception {
     LOG.info("Uploading batch of {} points to Qdrant", pointsBuffer.size());
 
     // Instantiate QdrantGrpc client for each batch to maintain serializability
-    QdrantGrpc qdrant = new QdrantGrpc(new URL(options.qdrantUrl), options.apiKey);
-    qdrant.upsert(options.collectionName, pointsBuffer, options.shardKeySelector, options.wait);
-    qdrant.close();
+    QdrantGrpc qdrant = null;
+    try {
+      qdrant = new QdrantGrpc(new URL(options.qdrantUrl), options.apiKey);
+      qdrant.upsert(options.collectionName, pointsBuffer, options.shardKeySelector, options.wait);
+    } finally {
+      if (qdrant != null) {
+        qdrant.close();
+      }
+    }
   }
 
   @Override

src/main/java/io/qdrant/spark/QdrantGrpc.java

@@ -6,7 +6,6 @@
 import io.qdrant.client.grpc.Points.ShardKeySelector;
 import io.qdrant.client.grpc.Points.UpsertPoints;
 import java.io.Serializable;
-import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
@@ -16,8 +15,14 @@ public class QdrantGrpc implements Serializable {
 
   private final QdrantClient client;
 
-  public QdrantGrpc(URL url, String apiKey) throws MalformedURLException {
+  public QdrantGrpc(URL url, String apiKey) {
+    if (url == null) {
+      throw new IllegalArgumentException("URL cannot be null");
+    }
     String host = url.getHost();
+    if (host == null || host.isEmpty()) {
+      throw new IllegalArgumentException("Invalid URL: host is missing. Provided URL: " + url);
+    }
     int port = url.getPort() == -1 ? 6334 : url.getPort();
     boolean useTls = url.getProtocol().equalsIgnoreCase("https");
     client =

src/main/java/io/qdrant/spark/QdrantOptions.java

@@ -40,9 +40,25 @@ public QdrantOptions(Map<String, String> options) {
     Objects.requireNonNull(options);
 
     qdrantUrl = options.get("qdrant_url");
+    if (qdrantUrl == null || qdrantUrl.isEmpty()) {
+      throw new IllegalArgumentException("qdrant_url option is required and cannot be empty");
+    }
+
     collectionName = options.get("collection_name");
+    if (collectionName == null || collectionName.isEmpty()) {
+      throw new IllegalArgumentException("collection_name option is required and cannot be empty");
+    }
+
     batchSize = getIntOption(options, "batch_size", DEFAULT_BATCH_SIZE);
+    if (batchSize <= 0) {
+      throw new IllegalArgumentException("batch_size must be positive, got: " + batchSize);
+    }
+
     retries = getIntOption(options, "retries", DEFAULT_RETRIES);
+    if (retries < 0) {
+      throw new IllegalArgumentException("retries cannot be negative, got: " + retries);
+    }
+
     idField = options.getOrDefault("id_field", "");
     apiKey = options.getOrDefault("api_key", "");
     embeddingField = options.getOrDefault("embedding_field", "");
@@ -61,22 +77,33 @@ public QdrantOptions(Map<String, String> options) {
 
     validateSparseVectorFields();
     validateVectorFields();
+    validateMultiVectorFields();
 
     payloadFieldsToSkip = buildPayloadFieldsToSkip();
   }
 
   private int getIntOption(Map<String, String> options, String key, int defaultValue) {
-    return Integer.parseInt(options.getOrDefault(key, String.valueOf(defaultValue)));
+    String value = options.getOrDefault(key, String.valueOf(defaultValue));
+    try {
+      return Integer.parseInt(value);
+    } catch (NumberFormatException e) {
+      throw new IllegalArgumentException(
+          "Invalid value for option '" + key + "': '" + value + "'. Expected an integer.", e);
+    }
   }
 
   private boolean getBooleanOption(Map<String, String> options, String key, boolean defaultValue) {
     return Boolean.parseBoolean(options.getOrDefault(key, String.valueOf(defaultValue)));
   }
 
   private String[] parseArray(String input) {
-    return input == null
-        ? new String[0]
-        : Arrays.stream(input.split(",")).map(String::trim).toArray(String[]::new);
+    if (input == null || input.trim().isEmpty()) {
+      return new String[0];
+    }
+    return Arrays.stream(input.split(","))
+        .map(String::trim)
+        .filter(s -> !s.isEmpty())
+        .toArray(String[]::new);
   }
 
   private void validateSparseVectorFields() {
@@ -93,6 +120,13 @@ private void validateVectorFields() {
     }
   }
 
+  private void validateMultiVectorFields() {
+    if (multiVectorFields.length != multiVectorNames.length) {
+      throw new IllegalArgumentException(
+          "Multi vector fields and names should have the same length");
+    }
+  }
+
   private ShardKeySelector parseShardKeys(@Nullable String shardKeys) {
     if (shardKeys == null) {
       return null;

src/main/java/io/qdrant/spark/QdrantPointIdHandler.java

@@ -16,18 +16,57 @@ static PointId preparePointId(InternalRow record, StructType schema, QdrantOptio
       return id(UUID.randomUUID());
     }
 
-    int idFieldIndex = schema.fieldIndex(idField);
+    int idFieldIndex;
+    try {
+      idFieldIndex = schema.fieldIndex(idField);
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException(
+          "Field '"
+              + idField
+              + "' specified in 'id_field' does not exist in the schema. "
+              + "Available fields: "
+              + String.join(", ", schema.fieldNames()),
+          e);
+    }
+
     DataType idFieldType = schema.fields()[idFieldIndex].dataType();
     switch (idFieldType.typeName()) {
       case "string":
-        return id(UUID.fromString(record.getString(idFieldIndex)));
+        String idString = record.getString(idFieldIndex);
+        if (idString == null) {
+          throw new IllegalArgumentException(
+              "The 'id_field' contains a null value. IDs cannot be null. "
+                  + "Either provide valid ID values or remove 'id_field' option to use"
+                  + " auto-generated UUIDs.");
+        }
+        try {
+          return id(UUID.fromString(idString));
+        } catch (IllegalArgumentException e) {
+          throw new IllegalArgumentException(
+              "The 'id_field' value '"
+                  + idString
+                  + "' is not a valid UUID. "
+                  + "String IDs must be in UUID format (e.g.,"
+                  + " '550e8400-e29b-41d4-a716-446655440000'). "
+                  + "For non-UUID string IDs, consider using integer IDs or hashing your strings to"
+                  + " UUIDs.",
+              e);
+        }
 
       case "integer":
-      case "long":
         return id(record.getInt(idFieldIndex));
 
+      case "long":
+        return id(record.getLong(idFieldIndex));
+
       default:
-        throw new IllegalArgumentException("Point ID should be of type string or integer");
+        throw new IllegalArgumentException(
+            "Point ID field '"
+                + idField
+                + "' has unsupported type '"
+                + idFieldType.typeName()
+                + "'. "
+                + "Supported types: string (UUID format), integer, long");
     }
   }
 }