fix: address third round of reviewer feedback

- BlobReference.writeString now rejects strings longer than 65535 UTF-8
  bytes instead of silently truncating the 2-byte length prefix, which
  would produce a wrong-but-decodable reference.
- Scope BlobReferenceResolver per finish() call with explicit close so
  cached Datasets release JNI handles each batch instead of leaking for
  the writer's lifetime.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: hamersaw

lance-spark-base_2.12/src/main/java/org/lance/spark/utils/BlobReference.java

@@ -133,6 +133,12 @@ public static BlobReference deserialize(byte[] bytes) {
 
   private static void writeString(DataOutputStream out, String s) throws IOException {
     byte[] bytes = s.getBytes(StandardCharsets.UTF_8);
+    // The wire format uses a 2-byte unsigned length prefix; reject strings that would
+    // overflow it rather than silently truncating to 16 bits and corrupting the payload.
+    if (bytes.length > 0xFFFF) {
+      throw new IOException(
+          "BlobReference string exceeds maximum length of " + 0xFFFF + " bytes: " + bytes.length);
+    }
     out.writeShort(bytes.length);
     out.write(bytes);
   }

lance-spark-base_2.12/src/main/scala/org/lance/spark/arrow/LanceArrowWriter.scala

@@ -317,8 +317,6 @@ private[arrow] class LargeBinaryWriter(val valueVector: LargeVarBinaryVector)
   private val pendingIndices = new java.util.ArrayList[java.lang.Integer]()
   private val pendingRefs = new java.util.ArrayList[org.lance.spark.utils.BlobReference]()
 
-  @transient private lazy val resolver = new org.lance.spark.utils.BlobReferenceResolver()
-
   override def setNull(): Unit = {}
   override def setValue(input: SpecializedGetters, ordinal: Int): Unit = {
     val bytes = input.getBinary(ordinal)
@@ -337,14 +335,20 @@ private[arrow] class LargeBinaryWriter(val valueVector: LargeVarBinaryVector)
   override def finish(): Unit = {
     super.finish()
     if (!pendingRefs.isEmpty) {
+      // Scope the resolver per-batch so the source Datasets it opens are released
+      // (via AutoCloseable) instead of leaking JNI handles for the writer's lifetime.
+      val resolver = new org.lance.spark.utils.BlobReferenceResolver()
       try {
         resolver.resolveBatch(pendingIndices, pendingRefs, valueVector)
       } catch {
         case e: java.io.IOException =>
           throw new RuntimeException("Failed to resolve blob references", e)
       } finally {
-        pendingIndices.clear()
-        pendingRefs.clear()
+        try resolver.close()
+        finally {
+          pendingIndices.clear()
+          pendingRefs.clear()
+        }
       }
     }
   }