Another batch of Polaris fixes

Author: rjrudin

marklogic-spark-api/src/main/java/com/marklogic/spark/core/extraction/TikaTextExtractor.java

@@ -26,6 +26,7 @@ public class TikaTextExtractor implements TextExtractor {
 
     @Override
     public Optional<ExtractionResult> extractText(DocumentInputs inputs) {
+        Objects.requireNonNull(inputs);
         if (inputs.getContent() == null) {
             return Optional.empty();
         }

marklogic-spark-connector/src/main/java/com/marklogic/spark/reader/document/DocumentRowSchema.java

@@ -93,8 +93,9 @@ private static void addPermissionsToMetadata(InternalRow row, DocumentMetadataHa
                 ArrayData caps = capabilities.getArray(i);
                 DocumentMetadataHandle.Capability[] capArray = new DocumentMetadataHandle.Capability[caps.numElements()];
                 for (int j = 0; j < caps.numElements(); j++) {
-                    String value = caps.get(j, DataTypes.StringType).toString();
-                    capArray[j] = DocumentMetadataHandle.Capability.valueOf(value.toUpperCase());
+                    Object value = caps.get(j, DataTypes.StringType);
+                    Objects.requireNonNull(value);
+                    capArray[j] = DocumentMetadataHandle.Capability.valueOf(value.toString().toUpperCase());
                 }
                 metadata.getPermissions().add(role, capArray);
             }
@@ -118,9 +119,11 @@ private static void addMetadataValuesToMetadata(InternalRow row, DocumentMetadat
             ArrayData keys = properties.keyArray();
             ArrayData values = properties.valueArray();
             for (int i = 0; i < keys.numElements(); i++) {
-                String key = keys.get(i, DataTypes.StringType).toString();
-                String value = values.get(i, DataTypes.StringType).toString();
-                metadata.getMetadataValues().put(key, value);
+                Object key = keys.get(i, DataTypes.StringType);
+                Object value = values.get(i, DataTypes.StringType);
+                if (key != null && value != null) {
+                    metadata.getMetadataValues().put(key.toString(), value.toString());
+                }
             }
         }
     }

marklogic-spark-connector/src/main/java/com/marklogic/spark/reader/document/OpticTriplesReader.java

@@ -112,7 +112,9 @@ private void readNextBatchOfTriples(List<String> uris) {
             .where(op.cts.documentQuery(op.xs.stringSeq(uris.toArray(new String[0]))));
 
         if (documentContext.hasOption(Options.READ_TRIPLES_GRAPHS)) {
-            String[] graphs = documentContext.getStringOption(Options.READ_TRIPLES_GRAPHS).split(",");
+            String value = documentContext.getStringOption(Options.READ_TRIPLES_GRAPHS);
+            Objects.requireNonNull(value);
+            String[] graphs = value.split(",");
             plan = plan.where(op.in(op.col(GRAPH_COLUMN), op.xs.stringSeq(graphs)));
         }
 

marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/WriteContext.java

@@ -166,6 +166,7 @@ private void forEachOptionStartingWith(final String prefix, BiConsumer<String, S
     public Format getDocumentFormat() {
         if (hasOption(Options.WRITE_DOCUMENT_TYPE)) {
             String value = getStringOption(Options.WRITE_DOCUMENT_TYPE);
+            Objects.requireNonNull(value);
             try {
                 return Format.valueOf(value.toUpperCase());
             } catch (IllegalArgumentException e) {

marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/file/ContentWriter.java

@@ -67,6 +67,7 @@ class ContentWriter {
     }
 
     void writeContent(InternalRow row, OutputStream outputStream) throws IOException {
+        Objects.requireNonNull(outputStream);
         if (this.isStreamingFiles) {
             streamDocumentToFile(row, outputStream);
         } else if (this.prettyPrint) {
@@ -146,10 +147,10 @@ private void prettyPrintContent(InternalRow row, OutputStream outputStream) thro
         } else if ("XML".equalsIgnoreCase(format)) {
             prettyPrintXml(content, outputStream);
         } else {
+            Objects.requireNonNull(content);
             if (this.encoding != null) {
                 outputStream.write(new String(content).getBytes(this.encoding));
             } else {
-                Objects.requireNonNull(content);
                 outputStream.write(content);
             }
         }