Revert "Merge pull request #166 from marklogic/feature/12332-read-triples"

This reverts commit c7f0fa8.

Author: rjrudin

build.gradle

@@ -13,10 +13,8 @@ group 'com.marklogic'
 version '2.2-SNAPSHOT'
 
 java {
-  // To support reading RDF files, Apache Jena is used - but that requires Java 11. If we want to do a 2.2.0 release
-  // without requiring Java 11, we'll remove the support for reading RDF files along with the Jena dependency.
-  sourceCompatibility = 11
-  targetCompatibility = 11
+  sourceCompatibility = 1.8
+  targetCompatibility = 1.8
 }
 
 repositories {
@@ -42,8 +40,6 @@ dependencies {
     exclude module: "scala-library"
   }
 
-  implementation "org.apache.jena:jena-arq:4.10.0"
-
   testImplementation 'org.apache.spark:spark-sql_2.12:' + sparkVersion
 
   // The exclusions in these two modules ensure that we use the Jackson libraries from spark-sql when running the tests.
@@ -60,7 +56,7 @@ dependencies {
     exclude module: 'jackson-dataformat-csv'
   }
 
-  testImplementation "ch.qos.logback:logback-classic:1.3.14"
+  testImplementation "ch.qos.logback:logback-classic:1.3.5"
   testImplementation "org.slf4j:jcl-over-slf4j:1.7.36"
   testImplementation "org.skyscreamer:jsonassert:1.5.1"
 }

src/main/java/com/marklogic/spark/DefaultSource.java

@@ -21,7 +21,6 @@
 import com.marklogic.spark.reader.document.DocumentRowSchema;
 import com.marklogic.spark.reader.document.DocumentTable;
 import com.marklogic.spark.reader.file.FileRowSchema;
-import com.marklogic.spark.reader.file.TripleRowSchema;
 import com.marklogic.spark.reader.optic.SchemaInferrer;
 import com.marklogic.spark.writer.WriteContext;
 import org.apache.spark.sql.SparkSession;
@@ -64,7 +63,7 @@ public String shortName() {
     public StructType inferSchema(CaseInsensitiveStringMap options) {
         final Map<String, String> properties = options.asCaseSensitiveMap();
         if (isFileOperation(properties)) {
-            return "rdf".equals(properties.get(Options.READ_FILES_TYPE)) ? TripleRowSchema.SCHEMA : FileRowSchema.SCHEMA;
+            return FileRowSchema.SCHEMA;
         }
         if (isReadDocumentsOperation(properties)) {
             return DocumentRowSchema.SCHEMA;
@@ -86,7 +85,8 @@ public Table getTable(StructType schema, Transform[] partitioning, Map<String, S
 
         if (isReadDocumentsOperation(properties)) {
             return new DocumentTable();
-        } else if (isReadOperation(properties)) {
+        }
+        else if (isReadOperation(properties)) {
             if (logger.isDebugEnabled()) {
                 logger.debug("Creating new table for reading");
             }

src/main/java/com/marklogic/spark/Options.java

@@ -55,7 +55,6 @@ public abstract class Options {
     public static final String READ_DOCUMENTS_TRANSFORM_PARAMS_DELIMITER = "spark.marklogic.read.documents.transformParamsDelimiter";
     public static final String READ_DOCUMENTS_PARTITIONS_PER_FOREST = "spark.marklogic.read.documents.partitionsPerForest";
 
-    public static final String READ_FILES_TYPE = "spark.marklogic.read.files.type";
     public static final String READ_FILES_COMPRESSION = "spark.marklogic.read.files.compression";
 
     // "Aggregate" = an XML document containing N child elements, each of which should become a row / document.

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

@@ -6,7 +6,7 @@
 public abstract class DocumentRowSchema {
 
     public static final StructType SCHEMA = new StructType()
-        .add("URI", DataTypes.StringType, false)
+        .add("URI", DataTypes.StringType)
         .add("content", DataTypes.BinaryType)
         .add("format", DataTypes.StringType)
         .add("collections", DataTypes.createArrayType(DataTypes.StringType))

src/main/java/com/marklogic/spark/reader/file/FilePartitionReaderFactory.java

@@ -25,15 +25,10 @@ class FilePartitionReaderFactory implements PartitionReaderFactory {
     @Override
     public PartitionReader<InternalRow> createReader(InputPartition partition) {
         FilePartition filePartition = (FilePartition) partition;
-
         String compression = this.properties.get(Options.READ_FILES_COMPRESSION);
         final boolean isZip = "zip".equalsIgnoreCase(compression);
         final boolean isGzip = "gzip".equalsIgnoreCase(compression);
 
-        if ("rdf".equalsIgnoreCase(this.properties.get(Options.READ_FILES_TYPE))) {
-            return new RdfFileReader(filePartition, hadoopConfiguration);
-        }
-
         String aggregateXmlElement = this.properties.get(Options.READ_AGGREGATES_XML_ELEMENT);
         if (aggregateXmlElement != null && !aggregateXmlElement.trim().isEmpty()) {
             if (isZip) {

src/main/java/com/marklogic/spark/reader/file/FileRowSchema.java

@@ -8,10 +8,10 @@ public abstract class FileRowSchema {
     // Same as Spark's binaryType.
     // See https://spark.apache.org/docs/latest/sql-data-sources-binaryFile.html .
     public static final StructType SCHEMA = new StructType()
-        .add("path", DataTypes.StringType, false)
+        .add("path", DataTypes.StringType)
         .add("modificationTime", DataTypes.TimestampType)
         .add("length", DataTypes.LongType)
-        .add("content", DataTypes.BinaryType, false);
+        .add("content", DataTypes.BinaryType);
 
     private FileRowSchema() {
     }