Repository: OpenRefine/OpenRefine
Pull request #4394 (2021-12-29 22:50:33)

Remove hadoop dependency in local runner
Although Spark was only a dependency of the Spark runner (and therefore OpenRefine could run without it), we still depended on Hadoop for its filesystem, even with the local runner.
Concretely, we used it for its ability to read text files in a partitioned way (which lets us parallelize computations run on a single file easily).

This was not really ideal for a few reasons:
- this is a big dependency, which pulls in a lot of other things we do not use. This also depends on some libraries we already used, and therefore forces us to stick to the versions of those libraries that Hadoop depends itself on (for instance, Jetty).
- the interface that Hadoop offers to control the partitioning strategy was not really ideal for us, leading to some inelegant code to run the partitioning heuristics multiple times until we get the desired result
-  the feature of Hadoop that we use (partitioning text files) can actually be implemented fairly easily without it.
- and Hadoop is a pain to run on Windows (https://issues.apache.org/jira/browse/HADOOP-13223).

Therefore this removes the dependency to Hadoop in the local runner. Installing Hadoop's winutils on Windows is still a requirement to use the Spark runner, which we cannot avoid because that is a requirement in Spark itself anyway.

wetneb: Unfortunately our UI tests do not support re-running inside the CI apparently, so I am just going to merge this and look into any integration test failures later if they appear to persist.

Patch:
diff --git a/appveyor.yml b/appveyor.yml
index 39b2354a2..22c03ea7a 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -26,7 +26,7 @@ build: off
 before_test:
 - cmd: echo Running scripts before build...
 - cmd: |-
-    PATH=C:\projects\openrefine\server\lib-local\native\windows\hadoop\bin;C:\Program Files\PostgreSQL\9.6\bin\;C:\Program Files\MySQL\MySQL Server 5.7\bin\;%PATH%
+    PATH=C:\projects\openrefine\refine-spark-runner\hadoop\bin;C:\Program Files\PostgreSQL\9.6\bin\;C:\Program Files\MySQL\MySQL Server 5.7\bin\;%PATH%
     SET MYSQL_PWD=Password12!
     mysql -u root --password=Password12! -e "create database test_db;"
     mysql -u root test_db --password=Password12! < extensions\database\tests\conf\test-mysql.sql
@@ -42,7 +42,7 @@ before_test:
     copy extensions\database\tests\conf\appveyor_tests.xml extensions\database\tests\conf\tests.xml
     copy packaging\test_pom.xml packaging\pom.xml
 
-    SET HADOOP_HOME=C:\projects\openrefine\server\lib-local\native\windows\hadoop
+    SET HADOOP_HOME=C:\projects\openrefine\refine-spark-runner\hadoop
 - cmd: |-
     mvn process-resources
     mvn package -DskipTests=true -Dmaven.javadoc.skip=true -B -V
diff --git a/extensions/database/pom.xml b/extensions/database/pom.xml
index 9a1f68f4c..dda38fd21 100644
--- a/extensions/database/pom.xml
+++ b/extensions/database/pom.xml
@@ -158,7 +158,7 @@
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
-      <artifactId>or-testing</artifactId>
+      <artifactId>refine-testing</artifactId>
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
diff --git a/extensions/jython/pom.xml b/extensions/jython/pom.xml
index 6a3bab8d0..8546f473f 100644
--- a/extensions/jython/pom.xml
+++ b/extensions/jython/pom.xml
@@ -59,7 +59,7 @@
   <dependencies>
     <dependency>
       <groupId>${project.groupId}</groupId>
-      <artifactId>or-model</artifactId>
+      <artifactId>refine-model</artifactId>
       <version>${project.version}</version>
       <scope>provided</scope>
     </dependency>
diff --git a/extensions/wikidata/pom.xml b/extensions/wikidata/pom.xml
index e4fa40821..f0d992b24 100644
--- a/extensions/wikidata/pom.xml
+++ b/extensions/wikidata/pom.xml
@@ -66,7 +66,7 @@
   <dependencies>
     <dependency>
       <groupId>${project.groupId}</groupId>
-      <artifactId>or-model</artifactId>
+      <artifactId>refine-model</artifactId>
       <version>${project.version}</version>
       <scope>provided</scope>
     </dependency>
@@ -119,7 +119,7 @@
 
     <dependency>
       <groupId>${project.groupId}</groupId>
-      <artifactId>or-testing</artifactId>
+      <artifactId>refine-testing</artifactId>
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
diff --git a/main/src/main/java/org/openrefine/importing/ImportingUtilities.java b/main/src/main/java/org/openrefine/importing/ImportingUtilities.java
index 5de920bf2..3f1924b42 100644
--- a/main/src/main/java/org/openrefine/importing/ImportingUtilities.java
+++ b/main/src/main/java/org/openrefine/importing/ImportingUtilities.java
@@ -67,7 +67,6 @@ import org.apache.commons.fileupload.ProgressListener;
 import org.apache.commons.fileupload.disk.DiskFileItemFactory;
 import org.apache.commons.fileupload.servlet.ServletFileUpload;
 import org.apache.commons.fileupload.util.Streams;
-import org.apache.hadoop.fs.Path;
 import org.apache.hc.client5.http.ClientProtocolException;
 import org.apache.hc.core5.http.ClassicHttpResponse;
 import org.apache.hc.core5.http.ContentType;
@@ -425,7 +424,7 @@ public class ImportingUtilities {
      * @return null if it does not succeed, or if there is no final name in the path
      */
     protected static String extractFilenameFromSparkURI(String uri) {
-		Path url = new Path(uri);
+		File url = new File(uri);
 		if (url.getName().length() > 0) {
 			return url.getName();
 		} else {
diff --git a/packaging/pom.xml b/packaging/pom.xml
index 7a596c401..f9278cf86 100644
--- a/packaging/pom.xml
+++ b/packaging/pom.xml
@@ -217,7 +217,6 @@
                 <initialHeapSize>512</initialHeapSize>
                 <maxHeapSize>2048</maxHeapSize>
                 <opts>
-                  <opt>-Djava.library.path=server/lib-local/native/windows/hadoop/bin</opt>
                 </opts>
               </jre>
               <versionInfo>
diff --git a/packaging/windows.xml b/packaging/windows.xml
index 82d3f37c8..17135b975 100644
--- a/packaging/windows.xml
+++ b/packaging/windows.xml
@@ -20,10 +20,6 @@
 	<include>*.jar</include>
       </includes>
     </fileSet>
-   <fileSet>
-      <directory>${rootdir}/server/lib-local/native/windows/hadoop/bin</directory>
-      <outputDirectory>server/lib-local/native/windows/hadoop/bin</outputDirectory>
-    </fileSet>
     <fileSet>
       <directory>${rootdir}/main/webapp</directory>
       <outputDirectory>webapp</outputDirectory>
diff --git a/refine-local-runner/pom.xml b/refine-local-runner/pom.xml
index 9d4cc22ed..7c1b8808a 100644
--- a/refine-local-runner/pom.xml
+++ b/refine-local-runner/pom.xml
@@ -65,16 +65,6 @@
       <artifactId>refine-util</artifactId>
       <version>${project.version}</version>
     </dependency>
-    <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-common</artifactId>
-      <version>${hadoop.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-mapreduce-client-common</artifactId>
-      <version>${hadoop.version}</version>
-    </dependency>
 
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
diff --git a/refine-local-runner/src/main/java/org/openrefine/model/LocalDatamodelRunner.java b/refine-local-runner/src/main/java/org/openrefine/model/LocalDatamodelRunner.java
index 820ddde1c..76165540a 100644
--- a/refine-local-runner/src/main/java/org/openrefine/model/LocalDatamodelRunner.java
+++ b/refine-local-runner/src/main/java/org/openrefine/model/LocalDatamodelRunner.java
@@ -10,11 +10,7 @@ import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocalFileSystem;
 import org.openrefine.importers.MultiFileReadingProgress;
-import org.openrefine.io.OrderedLocalFileSystem;
 import org.openrefine.model.GridState.Metadata;
 import org.openrefine.model.changes.ChangeData;
 import org.openrefine.model.changes.ChangeDataSerializer;
@@ -51,30 +47,10 @@ public class LocalDatamodelRunner implements DatamodelRunner {
         defaultParallelism = configuration.getIntParameter("defaultParallelism", 4);
         minSplitSize = configuration.getLongParameter("minSplitSize", 4096L);
         maxSplitSize = configuration.getLongParameter("maxSplitSize", 16777216L);
-        String hadoopDir = configuration.getParameter("hadoopHomeDir", "server/lib-local/native/windows/hadoop");
         
-        Configuration fsConf = new Configuration();
-        // set up Hadoop on Windows
-        String os = System.getProperty("os.name").toLowerCase();
-        if (os.contains("windows")) {
-            try {
-                System.setProperty("hadoop.home.dir", new File(hadoopDir).getCanonicalPath());
-            } catch (IOException e) {
-                logger.warn("unable to locate Windows Hadoop binaries, this will leave temporary files behind");
-            }
-        }
-        
-        fsConf.set("fs.file.impl", OrderedLocalFileSystem.class.getName());
-        fsConf.set("mapreduce.input.fileinputformat.split.minsize", Long.toString(minSplitSize));
-        fsConf.set("mapreduce.input.fileinputformat.split.maxsize", Long.toString(maxSplitSize));
-        try {
-            pllContext = new PLLContext(MoreExecutors.listeningDecorator(
-                    Executors.newCachedThreadPool()),
-                    LocalFileSystem.get(fsConf),
-                    defaultParallelism);
-        } catch (IOException e) {
-            throw new UncheckedIOException(e);
-        }
+        pllContext = new PLLContext(MoreExecutors.listeningDecorator(
+                Executors.newCachedThreadPool()),
+                defaultParallelism, minSplitSize, maxSplitSize);
     }
     
     public LocalDatamodelRunner() {
@@ -107,11 +83,6 @@ public class LocalDatamodelRunner implements DatamodelRunner {
         }
         return new Tuple2<Long, Row>(id.getIndex(), id.getRow());
     }
-    
-    @Override
-    public FileSystem getFileSystem() throws IOException {
-        return pllContext.getFileSystem();
-    }
 
     @Override
     public GridState create(ColumnModel columnModel, List<Row> rows, Map<String, OverlayModel> overlayModels) {
diff --git a/refine-local-runner/src/main/java/org/openrefine/model/local/PLLContext.java b/refine-local-runner/src/main/java/org/openrefine/model/local/PLLContext.java
index 0b57238c7..9662e5174 100644
--- a/refine-local-runner/src/main/java/org/openrefine/model/local/PLLContext.java
+++ b/refine-local-runner/src/main/java/org/openrefine/model/local/PLLContext.java
@@ -3,8 +3,6 @@ package org.openrefine.model.local;
 import java.io.IOException;
 import java.util.List;
 
-import org.apache.hadoop.fs.FileSystem;
-
 import com.google.common.util.concurrent.ListeningExecutorService;
 
 /**
@@ -16,20 +14,19 @@ import com.google.common.util.concurrent.ListeningExecutorService;
 public class PLLContext {
     
     private final ListeningExecutorService executorService;
-    private final FileSystem fileSystem;
     private final int defaultParallelism;
     private final long minSplitSize;
     private final long maxSplitSize;
     
     public PLLContext(
             ListeningExecutorService executorService,
-            FileSystem fileSystem,
-            int defaultParallelism) {
+            int defaultParallelism,
+            long minSplitSize,
+            long maxSplitSize) {
         this.executorService = executorService;
-        this.fileSystem = fileSystem;
         this.defaultParallelism = defaultParallelism;
-        this.minSplitSize = fileSystem.getConf().getLong("mapreduce.input.fileinputformat.split.minsize", 1048576);
-        this.maxSplitSize = fileSystem.getConf().getLong("mapreduce.input.fileinputformat.split.maxsize", Long.MAX_VALUE);
+        this.minSplitSize = minSplitSize;
+        this.maxSplitSize = maxSplitSize;
     }
     
     /**
@@ -39,13 +36,6 @@ public class PLLContext {
         return executorService;
     }
     
-    /**
-     * Returns the Hadoop filesystem used in this context
-     */
-    public FileSystem getFileSystem() {
-        return fileSystem;
-    }
-    
     /**
      * Loads a text file as a PLL.
      * 
@@ -63,7 +53,6 @@ public class PLLContext {
      */
     public void shutdown() throws IOException {
         executorService.shutdown();
-        fileSystem.close();
     }
 
     /**
diff --git a/refine-local-runner/src/main/java/org/openrefine/model/local/TextFilePLL.java b/refine-local-runner/src/main/java/org/openrefine/model/local/TextFilePLL.java
index 4f7d78535..ae01ec97b 100644
--- a/refine-local-runner/src/main/java/org/openrefine/model/local/TextFilePLL.java
+++ b/refine-local-runner/src/main/java/org/openrefine/model/local/TextFilePLL.java
@@ -1,39 +1,34 @@
 package org.openrefine.model.local;
 
+import java.io.BufferedInputStream;
 import java.io.File;
+import java.io.FileInputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.nio.channels.FileChannel;
+import java.nio.charset.Charset;
+import java.nio.file.Files;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.stream.Stream;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.JobContext;
-import org.apache.hadoop.mapred.JobContextImpl;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobID;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
-import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
+import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
 import org.openrefine.importers.MultiFileReadingProgress;
+import org.openrefine.model.local.util.LineReader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.google.common.collect.Streams;
+import com.google.common.io.CountingInputStream;
 
 /**
  * A PLL whose contents are read from a set of text files.
- * The text files are partitioned using Hadoop, using new lines as boundaries.
+ * The text files are partitioned using a method similar to that of Hadoop, using new lines as boundaries.
  * 
  * This class aims at producing a certain number of partitions determined by the
  * default parallelism of the PLL context.
@@ -44,62 +39,54 @@ import com.google.common.collect.Streams;
 public class TextFilePLL extends PLL<String> {
     
     private final static Logger logger = LoggerFactory.getLogger(TextFilePLL.class);
-    
-    private final List<HadoopPartition> partitions;
+    private final List<TextFilePartition> partitions;
     private final String path;
-    private final PLLContext context;
-    private final InputFormat<LongWritable, Text> inputFormat = new TextInputFormat();
     private ReadingProgressReporter progress;
 
     public TextFilePLL(PLLContext context, String path) throws IOException {
         super(context);
         this.path = path;
-        this.context = context;
         this.progress = null;
         
-        FileSystem fs = context.getFileSystem();
-        
-        // Setup the job to compute the splits
-        Configuration conf = new Configuration(fs.getConf());
-        Job job = Job.getInstance(conf);
-        FileInputFormat.setInputPaths(job, path);
-        JobID jobId = new JobID();
-        JobContext jobContext = new JobContextImpl((JobConf)job.getConfiguration(), jobId);
-
-        List<InputSplit> splits;
+        File file = new File(path);
         partitions = new ArrayList<>();
-        try {
-            // First attempt to get splits using the default parameters
-            splits = inputFormat.getSplits(jobContext);
+        if (file.isDirectory()) {
+            List<File> files = Arrays.asList(file.listFiles());
+            files.sort(new Comparator<File>() {
+                @Override
+                public int compare(File arg0, File arg1) {
+                    return arg0.getPath().compareTo(arg1.getPath());
+                } 
+            });
             
-            // If there are too few splits compared to the default parallelism,
-            // and at least one split is large enough to be split again, then
-            // we split again with lower maximum split size.
-            if (splits.size() < context.getDefaultParallelism()) {
-                long maxSplitSize = 0;
-                for (InputSplit split : splits) {
-                    maxSplitSize = Math.max(split.getLength(), maxSplitSize);
-                }
-                if (maxSplitSize > context.getMinSplitSize() * context.getDefaultParallelism()) {
-                    // re-split with lower maximum split size
-                    long newMaxSplitSize = maxSplitSize / context.getDefaultParallelism();
-                    conf.set("mapreduce.input.fileinputformat.split.maxsize", Long.toString(newMaxSplitSize));
-                    job.close();
-                    job = Job.getInstance(conf);
-                    FileInputFormat.setInputPaths(job, path);
-                    jobContext = new JobContextImpl((JobConf)job.getConfiguration(), jobId);
-                    splits = inputFormat.getSplits(jobContext);
-                }
+            for(File subFile : files) {
+                addPartitionsForFile(subFile);
             }
-            
-            for (int i = 0; i != splits.size(); i++) {
-                partitions.add(new HadoopPartition(i, splits.get(i)));
+        } else {
+            addPartitionsForFile(file);
+        }
+    }
+    
+    private static boolean isGzipped(File file) {
+        return file.getName().endsWith(".gz");
+    }
+    
+    private void addPartitionsForFile(File file) throws IOException {
+        long size = Files.size(file.toPath());
+        if (size < context.getMinSplitSize() * context.getDefaultParallelism() || isGzipped(file)) {
+            // a single split
+            partitions.add(new TextFilePartition(file, partitions.size(), 0L, size));
+        } else {
+            // defaultParallelism many splits, unless that makes splits too big
+            long splitSize = Math.min((size / context.getDefaultParallelism()) + 1, context.getMaxSplitSize());
+            int numSplits = (int) (size / splitSize);
+            if (numSplits * splitSize < size) {
+                numSplits++;
+            }
+            for (int i = 0; i != numSplits; i++) {
+                partitions.add(new TextFilePartition(file, partitions.size(), splitSize*i, Math.min(splitSize*(i+1), size)));
             }
-        } catch (InterruptedException e) {
-            partitions.clear();
-            e.printStackTrace();
         }
-        
     }
     
     public void setProgressHandler(MultiFileReadingProgress progress) {
@@ -116,36 +103,56 @@ public class TextFilePLL extends PLL<String> {
 
     @Override
     protected Stream<String> compute(Partition partition) {
-        HadoopPartition hadoopPartition = (HadoopPartition)partition;
-        TaskAttemptID attemptId = new TaskAttemptID();
-        TaskAttemptContext taskAttemptContext = new TaskAttemptContextImpl(context.getFileSystem().getConf(), attemptId);
+        TextFilePartition textPartition = (TextFilePartition) partition;
+       
         int reportBatchSize = 64;
         try {
-            RecordReader<LongWritable, Text> reader = inputFormat.createRecordReader(hadoopPartition.getSplit(), taskAttemptContext);
-            reader.initialize(hadoopPartition.getSplit(), taskAttemptContext);
+            FileInputStream stream = new FileInputStream(textPartition.getPath());
+            FileChannel channel = stream.getChannel();
+            if (textPartition.getStart() > 0L) {
+                channel.position(textPartition.start);
+            }
+            InputStream bufferedIs = new BufferedInputStream(stream);
+            if (isGzipped(textPartition.getPath())) {
+                bufferedIs = new GzipCompressorInputStream(bufferedIs);
+            }
+            CountingInputStream countingIs = new CountingInputStream(bufferedIs);
+            LineReader lineReader = new LineReader(countingIs, Charset.forName("UTF-8")); // TODO make charset configurable
+            
+            // if we are not reading from the first partition of the given file,
+            // we need to ignore the first "line" because it might be incomplete
+            // (it might have started before the split).
+            // The reading of the previous partition will take care of reading that line fully
+            // (so it can go beyond the planned end of the split).
+            if (textPartition.getStart() > 0) {
+                lineReader.readLine();
+            }
+            
             Iterator<String> iterator = new Iterator<String>() {
                 
-                boolean finished = false;
-                boolean havePair = false;
+                boolean nextLineAttempted = false;
+                String nextLine = null;
                 long lastOffsetReported = -1;
                 long lastOffsetSeen = -1;
                 int lastReport = 0;
 
                 @Override
                 public boolean hasNext() {
-                    if (!finished && !havePair) {
-                        try {
-                            finished = !reader.nextKeyValue();
-                        } catch (IOException | InterruptedException e) {
-                            finished = true;
-                            e.printStackTrace();
+                    long currentPosition;
+                    try {
+                        currentPosition = textPartition.start + countingIs.getCount();
+                        if (!nextLineAttempted && currentPosition <= textPartition.getEnd()) {
+                            nextLine = lineReader.readLine();
+                            nextLineAttempted = true;
+                            lastOffsetSeen = currentPosition;
                         }
-                        havePair = !finished;
-                    }
-                    if (finished && lastOffsetSeen > lastOffsetReported) {
-                        reportProgress();
+                        if (nextLine == null && lastOffsetSeen > lastOffsetReported) {
+                            reportProgress();
+                        }
+                    } catch (IOException e) {
+                        throw new UncheckedIOException(e);
                     }
-                    return !finished;
+                    return nextLine != null;
                 }
 
                 @Override
@@ -153,22 +160,16 @@ public class TextFilePLL extends PLL<String> {
                     if (!hasNext()) {
                         throw new NoSuchElementException("End of stream");
                     }
-                    String line = null;
-                    try {
-                        line = reader.getCurrentValue().toString();
-                        lastOffsetSeen = reader.getCurrentKey().get();
-                        if (lastReport >= reportBatchSize) {
-                            reportProgress();
-                        }
-                        lastReport++;
-                        if (lastOffsetReported == -1) {
-                            lastOffsetReported = lastOffsetSeen;
-                        }
-                    } catch (IOException | InterruptedException e) {
-                        finished = true;
-                        e.printStackTrace();
+                    if (lastReport >= reportBatchSize) {
+                        reportProgress();
+                    }
+                    lastReport++;
+                    if (lastOffsetReported == -1) {
+                        lastOffsetReported = lastOffsetSeen;
                     }
-                    havePair = false;
+                    String line = nextLine;
+                    nextLine = null;
+                    nextLineAttempted = false;
                     return line;
                 }
 
@@ -181,16 +182,16 @@ public class TextFilePLL extends PLL<String> {
                 }
                 
             };
-            Stream<String> stream = Streams.stream(iterator)
+            Stream<String> lineStream = Streams.stream(iterator)
                     .onClose(() -> {
                         try {
-                            reader.close();
+                            lineReader.close();
                         } catch (IOException e) {
                             e.printStackTrace();
                         }
                     });
-            return stream;
-        } catch (IOException | InterruptedException e) {
+            return lineStream;
+        } catch (IOException e) {
             e.printStackTrace();
             return Stream.empty();
         }
@@ -201,14 +202,33 @@ public class TextFilePLL extends PLL<String> {
         return partitions;
     }
     
-    protected static class HadoopPartition implements Partition {
+    protected static class TextFilePartition implements Partition {
         
+        private final File path;
         private final int index;
-        private final InputSplit split;
+        private final long start;
+        private final long end;
 
-        protected HadoopPartition(int index, InputSplit split) {
+        /**
+         * Represents a split in an uncompressed text file.
+         * 
+         * @param path  the path to the file being read
+         * @param index position of the split in the file
+         * @param start starting byte where to read from in the file
+         * @param end   first byte not to be read after the end of the file
+         */
+        protected TextFilePartition(File path, int index, long start, long end) {
+            this.path = path;
             this.index = index;
-            this.split = split;
+            this.start = start;
+            this.end = end;
+            if (isGzipped(path) && start > 0) {
+                throw new IllegalArgumentException("Unable to split a gzip file");
+            }
+        }
+        
+        public File getPath() {
+            return path;
         }
 
         @Override
@@ -221,8 +241,12 @@ public class TextFilePLL extends PLL<String> {
             return null;
         }
         
-        protected InputSplit getSplit() {
-            return split;
+        public long getStart() {
+            return start;
+        }
+        
+        public long getEnd() {
+            return end;
         }
         
     }
diff --git a/refine-local-runner/src/main/java/org/openrefine/model/local/util/LineReader.java b/refine-local-runner/src/main/java/org/openrefine/model/local/util/LineReader.java
new file mode 100644
index 000000000..d5ffc63ed
--- /dev/null
+++ b/refine-local-runner/src/main/java/org/openrefine/model/local/util/LineReader.java
@@ -0,0 +1,82 @@
+package org.openrefine.model.local.util;
+
+import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+
+/**
+ * A class which reads line by line from an {@link java.io.InputStream},
+ * making sure it only reads the bytes it needs to complete the current line.
+ * 
+ * This makes it possible to track at which byte offset each line starts,
+ * when used in combination with a {@link com.google.common.io.CountingInputStream}
+ * (see {@link org.openrefine.model.local.TextFilePLL}).
+ * 
+ * This cannot be achieved with {@link java.io.InputStreamReader} and {@link java.io.LineNumberReader},
+ * because {@link java.io.InputStreamReader} can read more bytes than it needs to output characters.
+ * 
+ * Like in {@link java.io.LineNumberReader}, a line is considered to be terminated by any one
+ * of a line feed ('\n'), a carriage return ('\r'), or a carriage return followed immediately by a line feed.
+ * 
+ * @author Antonin Delpeuch
+ */
+public class LineReader implements Closeable {
+    
+    protected final InputStream inputStream;
+    protected final Charset     charset;
+
+    /**
+     * Constructs a reader with the supplied input stream and charset.
+     * 
+     * @param inputStream
+     * @param charset
+     */
+    public LineReader(InputStream inputStream, Charset charset) {
+        this.inputStream = inputStream;
+        this.charset = charset;
+    }
+    
+    /**
+     * Reads a single line from the input stream, not reading more
+     * characters beyond the line being read.
+     * 
+     * @return the line read or null if the input stream was exhausted
+     * @throws IOException
+     */
+    public String readLine() throws IOException {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        int currentByte = inputStream.read();
+        while (currentByte >= 0) {
+            if (currentByte == '\n') {
+                currentByte = -2;
+            } else if (currentByte == '\r') {
+                // Swallow the next '\n' character if it immediately follows this one
+                inputStream.mark(2);
+                int nextByte = inputStream.read();
+                if (nextByte != '\n') {
+                    // if the next character was not '\n', put the stream back where it was
+                    inputStream.reset();
+                }
+                currentByte = -2;
+            } else {
+                baos.write(currentByte);
+                currentByte = inputStream.read();
+            }
+        }
+        if (baos.size() > 0 || currentByte == -2) {
+            return baos.toString(charset);
+        } else {
+            return null;
+        }
+    }
+
+    /**
+     * Closes the underlying input stream.
+     */
+    @Override
+    public void close() throws IOException {
+        inputStream.close();
+    }
+}
diff --git a/refine-model/pom.xml b/refine-model/pom.xml
index 73453fe37..45be4d175 100644
--- a/refine-model/pom.xml
+++ b/refine-model/pom.xml
@@ -81,19 +81,6 @@
       <artifactId>jackson-core</artifactId>
       <version>${jackson.version}</version>
     </dependency>
-    <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-common</artifactId>
-      <version>${hadoop.version}</version>
-    </dependency>
-    <dependency>
-      <!-- added to override the version of this dependency through hadoop-common. -->
-      <!-- as of hadoop-common:3.3.1, it depends on paranamer 2.3, which fails at runtime. -->
-      <!-- therefore upgrading to paranamer 2.8 -->
-      <groupId>com.thoughtworks.paranamer</groupId>
-      <artifactId>paranamer</artifactId>
-      <version>2.8</version>
-    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
@@ -129,6 +116,11 @@
       <artifactId>commons-io</artifactId>
       <version>${commons-io.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-compress</artifactId>
+      <version>${commons-compress.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.ant</groupId>
       <artifactId>ant</artifactId> <!-- for org.apache.tools.bzip2 -->
@@ -144,8 +136,11 @@
       <artifactId>jersey-guava</artifactId>
       <version>2.22.2</version>
     </dependency>
-
-
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+      <version>${guava.version}</version>
+    </dependency>
 
 
     <!-- test dependencies -->
diff --git a/refine-model/src/main/java/org/openrefine/importers/ImporterUtilities.java b/refine-model/src/main/java/org/openrefine/importers/ImporterUtilities.java
index 77cbf21af..973f5273e 100644
--- a/refine-model/src/main/java/org/openrefine/importers/ImporterUtilities.java
+++ b/refine-model/src/main/java/org/openrefine/importers/ImporterUtilities.java
@@ -53,7 +53,6 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Collectors;
 
-import org.apache.hadoop.fs.FileSystem;
 import org.openrefine.importing.ImportingFileRecord;
 import org.openrefine.importing.ImportingJob;
 import org.openrefine.model.Cell;
@@ -164,11 +163,10 @@ public class ImporterUtilities {
     
     static public MultiFileReadingProgress createMultiFileReadingProgress(
             final ImportingJob job,
-            List<ImportingFileRecord> fileRecords,
-            FileSystem hdfs) {
+            List<ImportingFileRecord> fileRecords) {
         long totalSize = 0;
         for (ImportingFileRecord fileRecord : fileRecords) {
-			totalSize += fileRecord.getSize(job.getRawDataDir(), hdfs);
+			totalSize += fileRecord.getSize(job.getRawDataDir());
         }
         
         final long totalSize2 = totalSize;
diff --git a/refine-model/src/main/java/org/openrefine/importers/ImportingParserBase.java b/refine-model/src/main/java/org/openrefine/importers/ImportingParserBase.java
index f234a62d3..b6c0d8cd1 100644
--- a/refine-model/src/main/java/org/openrefine/importers/ImportingParserBase.java
+++ b/refine-model/src/main/java/org/openrefine/importers/ImportingParserBase.java
@@ -39,7 +39,6 @@ import java.io.Reader;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.apache.hadoop.fs.FileSystem;
 import org.openrefine.ProjectMetadata;
 import org.openrefine.importing.ImportingFileRecord;
 import org.openrefine.importing.ImportingJob;
@@ -79,8 +78,7 @@ abstract public class ImportingParserBase implements ImportingParser {
     public GridState parse(ProjectMetadata metadata,
             final ImportingJob job, List<ImportingFileRecord> fileRecords, String format,
             long limit, ObjectNode options) throws Exception {
-    	FileSystem hdfs = runner.getFileSystem();
-        MultiFileReadingProgress progress = ImporterUtilities.createMultiFileReadingProgress(job, fileRecords, hdfs);
+        MultiFileReadingProgress progress = ImporterUtilities.createMultiFileReadingProgress(job, fileRecords);
         List<GridState> gridStates = new ArrayList<>(fileRecords.size());
         
         if (fileRecords.isEmpty()) {
diff --git a/refine-model/src/main/java/org/openrefine/importing/ImportingFileRecord.java b/refine-model/src/main/java/org/openrefine/importing/ImportingFileRecord.java
index a631e3d79..329e15776 100644
--- a/refine-model/src/main/java/org/openrefine/importing/ImportingFileRecord.java
+++ b/refine-model/src/main/java/org/openrefine/importing/ImportingFileRecord.java
@@ -2,10 +2,8 @@ package org.openrefine.importing;
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.file.Files;
 
-import org.apache.hadoop.fs.ContentSummary;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -195,7 +193,7 @@ public class ImportingFileRecord {
 	 *     the Hadoop file system, to read Spark URIs
 	 * @return the length of the file in bytes
 	 */
-	public long getSize(File rawDataDir, FileSystem hdfs) {
+	public long getSize(File rawDataDir) {
 		if (_size > 0) {
 			return _size;
 		}
@@ -203,13 +201,11 @@ public class ImportingFileRecord {
 			File localFile = getFile(rawDataDir);
 			_size = localFile.length();
 		} else {
-			Path path = new Path(getDerivedSparkURI(rawDataDir));
-			try {
-				ContentSummary summary = hdfs.getContentSummary(path);
-				_size = summary.getLength();
-			} catch(IOException e) {
-				_size = 0;
-			}
+		    try {
+                _size = Files.size(new File(_sparkURI).toPath());
+            } catch (IOException e) {
+                _size = 0;
+            }
 		}
 		return _size;
 	}
diff --git a/refine-model/src/main/java/org/openrefine/model/DatamodelRunner.java b/refine-model/src/main/java/org/openrefine/model/DatamodelRunner.java
index 4cf538515..f2b001884 100644
--- a/refine-model/src/main/java/org/openrefine/model/DatamodelRunner.java
+++ b/refine-model/src/main/java/org/openrefine/model/DatamodelRunner.java
@@ -5,7 +5,6 @@ import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.hadoop.fs.FileSystem;
 import org.openrefine.importers.MultiFileReadingProgress;
 import org.openrefine.model.changes.ChangeData;
 import org.openrefine.model.changes.ChangeDataSerializer;
@@ -43,13 +42,6 @@ public interface DatamodelRunner {
      */
     public <T> ChangeData<T> loadChangeData(File path, ChangeDataSerializer<T> serializer) throws IOException;
     
-    /**
-     * Returns a file system used by the implementation to read
-     * files to import.
-     * @throws IOException 
-     */
-    public FileSystem getFileSystem() throws IOException;
-    
     /**
      * Creates a {@link GridState} from an in-memory list of rows,
      * which will be numbered from 0 to length-1.
diff --git a/server/lib-local/native/windows/hadoop/bin/hadoop.dll b/refine-spark-runner/hadoop/bin/hadoop.dll
similarity index 100%
rename from server/lib-local/native/windows/hadoop/bin/hadoop.dll
rename to refine-spark-runner/hadoop/bin/hadoop.dll
diff --git a/server/lib-local/native/windows/hadoop/bin/hadoop.exp b/refine-spark-runner/hadoop/bin/hadoop.exp
similarity index 100%
rename from server/lib-local/native/windows/hadoop/bin/hadoop.exp
rename to refine-spark-runner/hadoop/bin/hadoop.exp
diff --git a/server/lib-local/native/windows/hadoop/bin/winutils.exe b/refine-spark-runner/hadoop/bin/winutils.exe
similarity index 100%
rename from server/lib-local/native/windows/hadoop/bin/winutils.exe
rename to refine-spark-runner/hadoop/bin/winutils.exe
diff --git a/refine-model/src/main/java/org/openrefine/io/OrderedLocalFileSystem.java b/refine-spark-runner/src/main/java/org/openrefine/io/OrderedLocalFileSystem.java
similarity index 100%
rename from refine-model/src/main/java/org/openrefine/io/OrderedLocalFileSystem.java
rename to refine-spark-runner/src/main/java/org/openrefine/io/OrderedLocalFileSystem.java
diff --git a/refine-spark-runner/src/main/java/org/openrefine/model/SparkDatamodelRunner.java b/refine-spark-runner/src/main/java/org/openrefine/model/SparkDatamodelRunner.java
index 12d21d5fc..2a132f6fa 100644
--- a/refine-spark-runner/src/main/java/org/openrefine/model/SparkDatamodelRunner.java
+++ b/refine-spark-runner/src/main/java/org/openrefine/model/SparkDatamodelRunner.java
@@ -117,7 +117,6 @@ public class SparkDatamodelRunner implements DatamodelRunner {
  
     }
 
-    @Override
     public FileSystem getFileSystem() throws IOException {
         return FileSystem.get(context.hadoopConfiguration());
     }
diff --git a/refine-util/pom.xml b/refine-util/pom.xml
index a88556e6f..d20355f79 100644
--- a/refine-util/pom.xml
+++ b/refine-util/pom.xml
@@ -56,10 +56,11 @@
       <version>${jackson.version}</version>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-common</artifactId>
-      <version>${hadoop.version}</version>
+      <groupId>log4j</groupId>
+      <artifactId>log4j</artifactId>
+      <version>1.2.17</version>
     </dependency>
+    
 
     <!-- test dependencies -->
 
diff --git a/refine-workflow/src/main/java/org/openrefine/importers/tree/TreeImportingParserBase.java b/refine-workflow/src/main/java/org/openrefine/importers/tree/TreeImportingParserBase.java
index 12e50fa57..4b1e6cc2c 100644
--- a/refine-workflow/src/main/java/org/openrefine/importers/tree/TreeImportingParserBase.java
+++ b/refine-workflow/src/main/java/org/openrefine/importers/tree/TreeImportingParserBase.java
@@ -86,7 +86,7 @@ abstract public class TreeImportingParserBase extends ImportingParserBase {
     public GridState parse(ProjectMetadata metadata,
             final ImportingJob job, List<ImportingFileRecord> fileRecords, String format,
             long limit, ObjectNode options) throws Exception {
-        MultiFileReadingProgress progress = ImporterUtilities.createMultiFileReadingProgress(job, fileRecords, runner.getFileSystem());
+        MultiFileReadingProgress progress = ImporterUtilities.createMultiFileReadingProgress(job, fileRecords);
         ImportColumnGroup rootColumnGroup = new ImportColumnGroup();
         
         List<Row> rows = new ArrayList<>();
diff --git a/refine-workflow/src/main/java/org/openrefine/importers/tree/XmlImportUtilities.java b/refine-workflow/src/main/java/org/openrefine/importers/tree/XmlImportUtilities.java
index c45f1126c..392f68184 100644
--- a/refine-workflow/src/main/java/org/openrefine/importers/tree/XmlImportUtilities.java
+++ b/refine-workflow/src/main/java/org/openrefine/importers/tree/XmlImportUtilities.java
@@ -42,8 +42,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 
-import javax.servlet.ServletException;
-
 import org.openrefine.importers.tree.TreeReader.Token;
 import org.openrefine.model.Cell;
 import org.openrefine.model.Row;
diff --git a/refine.bat b/refine.bat
index 07e801fab..5f215e04c 100644
--- a/refine.bat
+++ b/refine.bat
@@ -255,7 +255,7 @@ echo -----------------------
 ) > support.log
 
 set CLASSPATH="%REFINE_CLASSES_DIR%;%REFINE_LIB_DIR%\*"
-"%JAVA_HOME%\bin\java.exe" -cp %CLASSPATH% %OPTS% -Djava.library.path=server/lib-local/native/windows/hadoop/bin org.openrefine.Refine
+"%JAVA_HOME%\bin\java.exe" -cp %CLASSPATH% %OPTS% org.openrefine.Refine
 goto end
 
 :doMvn


Test patch:
diff --git a/main/src/test/java/org/openrefine/commands/project/SetProjectMetadataCommandTests.java b/main/src/test/java/org/openrefine/commands/project/SetProjectMetadataCommandTests.java
index 0b8d39389..c049ee689 100644
--- a/main/src/test/java/org/openrefine/commands/project/SetProjectMetadataCommandTests.java
+++ b/main/src/test/java/org/openrefine/commands/project/SetProjectMetadataCommandTests.java
@@ -46,7 +46,6 @@ import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
-import org.codehaus.jettison.json.JSONException;
 import org.openrefine.ProjectManager;
 import org.openrefine.ProjectMetadata;
 import org.openrefine.commands.Command;
@@ -152,7 +151,6 @@ public class SetProjectMetadataCommandTests {
     
     /**
      *  set a user defined metadata field
-     * @throws JSONException 
      */
     @Test
     public void setUserMetadataFieldTest() {
diff --git a/refine-local-runner/src/test/java/org/openrefine/model/LocalDatamodelRunnerTests.java b/refine-local-runner/src/test/java/org/openrefine/model/LocalDatamodelRunnerTests.java
index 572109702..e4e4dcf02 100644
--- a/refine-local-runner/src/test/java/org/openrefine/model/LocalDatamodelRunnerTests.java
+++ b/refine-local-runner/src/test/java/org/openrefine/model/LocalDatamodelRunnerTests.java
@@ -4,10 +4,6 @@ import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
-import org.openrefine.io.OrderedLocalFileSystem;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
 /**
  * Tests for this datamodel implementation are taken from the standard
  * test suite, in {@link DatamodelRunnerTestBase}.
@@ -25,15 +21,8 @@ public class LocalDatamodelRunnerTests extends DatamodelRunnerTestBase {
     	map.put("minSplitSize", "128");
     	map.put("maxSplitSize", "1024");
     	
-    	map.put("hadoopHomeDir", "../server/lib-local/native/windows/hadoop");
-    	
     	RunnerConfiguration runnerConf = new RunnerConfigurationImpl(map);
         return new LocalDatamodelRunner(runnerConf);
     }
-    
-    @Test
-    public void testOrderedFileSystem() throws IOException {
-        Assert.assertTrue(SUT.getFileSystem() instanceof OrderedLocalFileSystem);
-    }
 
 }
diff --git a/refine-local-runner/src/test/java/org/openrefine/model/local/PLLTestsBase.java b/refine-local-runner/src/test/java/org/openrefine/model/local/PLLTestsBase.java
index 4e0fc2141..f698764a4 100644
--- a/refine-local-runner/src/test/java/org/openrefine/model/local/PLLTestsBase.java
+++ b/refine-local-runner/src/test/java/org/openrefine/model/local/PLLTestsBase.java
@@ -1,14 +1,9 @@
 package org.openrefine.model.local;
 
-import java.io.File;
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.Executors;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocalFileSystem;
-import org.openrefine.io.OrderedLocalFileSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.AfterClass;
@@ -25,30 +20,12 @@ public class PLLTestsBase {
     @BeforeClass
     public void setUpPool() throws IOException {
         if (context == null) {
-            Configuration conf = new Configuration();
-            
-            // set up Hadoop on Windows
-            String os = System.getProperty("os.name").toLowerCase();
-            if (os.contains("windows")) {
-                try {
-                    System.setProperty("hadoop.home.dir", new File("../server/lib-local/native/windows/hadoop").getCanonicalPath());
-                } catch (IOException e) {
-                    logger.warn("unable to locate Windows Hadoop binaries, this will leave temporary files behind");
-                }
-            }
-            
-            // these values are purposely very low for testing purposes,
+            // the split sizes are purposely very low for testing purposes,
             // so that we can check the partitioning strategy without using large files
-            conf.set("fs.file.impl", OrderedLocalFileSystem.class.getName());
-            conf.setLong("mapreduce.input.fileinputformat.split.minsize", 128L);
-            conf.setLong("mapreduce.input.fileinputformat.split.maxsize", 1024L);
-            FileSystem fileSystem = LocalFileSystem.get(conf);
-
 			context = new PLLContext(
                     MoreExecutors.listeningDecorator(
                     Executors.newCachedThreadPool()),
-                    fileSystem,
-                    4);
+                    4, 128L, 1024L);
         }
     }
     
diff --git a/refine-local-runner/src/test/java/org/openrefine/model/local/TextFilePLLTests.java b/refine-local-runner/src/test/java/org/openrefine/model/local/TextFilePLLTests.java
index cdfb358e9..00574aa4c 100644
--- a/refine-local-runner/src/test/java/org/openrefine/model/local/TextFilePLLTests.java
+++ b/refine-local-runner/src/test/java/org/openrefine/model/local/TextFilePLLTests.java
@@ -5,6 +5,7 @@ import java.io.FileWriter;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.Optional;
 
 import org.apache.commons.io.FileUtils;
@@ -54,19 +55,22 @@ public class TextFilePLLTests extends PLLTestsBase {
         // this text file is too small to be split
         Assert.assertEquals(pll.getPartitions().size(), 1);
         
-        Assert.assertEquals(pll.collect(), Arrays.asList("foo", "bar", "baz"));
+        List<String> elements = pll.collect();
+        Assert.assertEquals(elements, Arrays.asList("foo", "bar", "baz"));
         // Iterate a second time
-        Assert.assertEquals(pll.collect(), Arrays.asList("foo", "bar", "baz"));
+        Assert.assertEquals(elements, Arrays.asList("foo", "bar", "baz"));
     }
     
     @Test
     public void testMorePartitions() throws IOException {
         PLL<String> pll = new TextFilePLL(context, longerTextFile.getAbsolutePath());
         Assert.assertEquals(pll.getPartitions().size(), context.getDefaultParallelism());
+        Assert.assertEquals(pll.count(), 64L);
         
         pll = new TextFilePLL(context, veryLongTextFile.getAbsolutePath());
         int nbPartitions = pll.getPartitions().size();
         Assert.assertTrue(nbPartitions > context.getDefaultParallelism());
+        Assert.assertEquals(pll.count(), 2048L);
     }
     
     @Test
diff --git a/refine-local-runner/src/test/java/org/openrefine/model/local/util/LineReaderTests.java b/refine-local-runner/src/test/java/org/openrefine/model/local/util/LineReaderTests.java
new file mode 100644
index 000000000..2cecfd355
--- /dev/null
+++ b/refine-local-runner/src/test/java/org/openrefine/model/local/util/LineReaderTests.java
@@ -0,0 +1,124 @@
+package org.openrefine.model.local.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.Charset;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.google.common.io.CountingInputStream;
+
+public class LineReaderTests {
+    
+    Charset UTF8 = Charset.forName("UTF-8");
+    
+    protected InputStream asInputStream(String input) {
+        return new ByteArrayInputStream(input.getBytes(UTF8));
+    }
+    
+    @Test
+    public void testNotBuffering() throws IOException {
+        InputStream bais = asInputStream("hello\nworld");
+        CountingInputStream countingInputStream = new CountingInputStream(bais);
+        
+        try (LineReader reader = new LineReader(countingInputStream, UTF8)) {
+            String result = reader.readLine();
+            Assert.assertEquals(result, "hello");
+            Assert.assertEquals(countingInputStream.getCount(), 6);
+        }
+    }
+    
+    @Test
+    public void testCarriageReturn() throws IOException {
+        InputStream bais = asInputStream("hello\rworld");
+        CountingInputStream countingInputStream = new CountingInputStream(bais);
+        
+        try (LineReader reader = new LineReader(countingInputStream, UTF8)) {
+            String result = reader.readLine();
+            Assert.assertEquals(result, "hello");
+            Assert.assertEquals(countingInputStream.getCount(), 6);
+        }
+    }
+    
+    @Test
+    public void testCarriageReturnAndNewline() throws IOException {
+        InputStream bais = asInputStream("hello\r\nworld");
+        CountingInputStream countingInputStream = new CountingInputStream(bais);
+        
+        try (LineReader reader = new LineReader(countingInputStream, UTF8)) {
+            String result = reader.readLine();
+            Assert.assertEquals(result, "hello");
+            Assert.assertEquals(countingInputStream.getCount(), 7);
+        }
+    }
+    
+    @Test
+    public void testReadAllLines() throws IOException {
+        InputStream bais = asInputStream("hello\n\nworld");
+        
+        try (LineReader reader = new LineReader(bais, UTF8)) {
+            String result = reader.readLine();
+            Assert.assertEquals(result, "hello");
+            result = reader.readLine();
+            Assert.assertEquals(result, "");
+            result = reader.readLine();
+            Assert.assertEquals(result, "world");
+            // here we detect the end of the file
+            result = reader.readLine();
+            Assert.assertNull(result);
+            // and if we try to read more, we still get null
+            result = reader.readLine();
+            Assert.assertNull(result);
+        }
+    }
+    
+    @Test
+    public void testTrailingNewLine() throws IOException {
+        InputStream bais = asInputStream("hello\nworld\n");
+        
+        try (LineReader reader = new LineReader(bais, UTF8)) {
+            String result = reader.readLine();
+            Assert.assertEquals(result, "hello");
+            result = reader.readLine();
+            Assert.assertEquals(result, "world");
+            result = reader.readLine();
+            Assert.assertNull(result);
+        }
+    }
+    
+    @Test
+    public void testUnicode() throws IOException {
+        // the unicode '€' character is encoded on three bytes in UTF-8
+        
+        InputStream bais = asInputStream("h€llo\nworld");
+        CountingInputStream countingInputStream = new CountingInputStream(bais);
+        
+        try (LineReader reader = new LineReader(countingInputStream, UTF8)) {
+            String result = reader.readLine();
+            Assert.assertEquals(result, "h€llo");
+            Assert.assertEquals(countingInputStream.getCount(), 8);
+        }
+    }
+    
+    @Test
+    public void testIgnoresIncompleteUnicode() throws IOException {
+        // the unicode '€' character is encoded on three bytes in UTF-8
+        
+        InputStream bais = asInputStream("h€llo\nworld");
+        // Read the first two bytes, which means the cursor is in the middle of the encoding of the € sign
+        bais.read();
+        bais.read();
+        
+        CountingInputStream countingInputStream = new CountingInputStream(bais);
+        
+        try (LineReader reader = new LineReader(countingInputStream, UTF8)) {
+            String result = reader.readLine();
+            Assert.assertTrue(result.endsWith("llo"));
+            Assert.assertEquals(countingInputStream.getCount(), 6);
+        }
+    }
+
+
+}
diff --git a/refine-testing/src/main/java/org/openrefine/model/TestingDatamodelRunner.java b/refine-testing/src/main/java/org/openrefine/model/TestingDatamodelRunner.java
index 83b5d3bac..966d17f5c 100644
--- a/refine-testing/src/main/java/org/openrefine/model/TestingDatamodelRunner.java
+++ b/refine-testing/src/main/java/org/openrefine/model/TestingDatamodelRunner.java
@@ -9,7 +9,6 @@ import java.io.InputStreamReader;
 import java.io.LineNumberReader;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
-import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -21,8 +20,6 @@ import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.zip.GZIPInputStream;
 
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocalFileSystem;
 import org.openrefine.importers.MultiFileReadingProgress;
 import org.openrefine.model.GridState.Metadata;
 import org.openrefine.model.changes.ChangeData;
@@ -198,11 +195,6 @@ public class TestingDatamodelRunner implements DatamodelRunner {
         return new TestingGridState(columnModel, rows, overlayModels);
     }
 
-    @Override
-    public FileSystem getFileSystem() throws IOException {
-        return new LocalFileSystem();
-    }
-
     @Override
     public <T> ChangeData<T> create(List<IndexedData<T>> changeData) {
         return new TestingChangeData<T>(changeData.stream()