Merge pull request #15 from databrickslabs/feature/fix-feedback

Fixes following feedback from nfx

Author: milos-colic

docs/source/usage/DeltaSharing.rst

@@ -2,8 +2,7 @@
 DeltaSharing
 ==================
 
-From the provider JSON we can easily instantiate our Java Connector using the DeltaSharingFactory instance.
-DeltaSharingFactory provides create API that returns and instance of DeltaSharing object based on the provider JSON.
+From the provider JSON we can easily instantiate our Java Connector using the DeltaSharing object, based on the provider JSON.
 DeltaSharing instance is used to get access ot the table reader and to interact with the delta sharing server.
 
 Examples
@@ -12,22 +11,19 @@ Examples
 .. tabs::
    .. code-tab:: java
 
-    import com.databricks.labs.delta.sharing.java.DeltaSharingFactory;
     import com.databricks.labs.delta.sharing.java.DeltaSharing;
 
-    DeltaSharing sharing = DeltaSharingFactory
-      .create(
+    DeltaSharing sharing = DeltaSharing(
          providerJSON,
          "/dedicated/persisted/cache/location/"
       );
 
 
    .. code-tab:: scala
 
-    import com.databricks.labs.delta.sharing.java.DeltaSharingFactory
+    import com.databricks.labs.delta.sharing.java.DeltaSharing
 
-    val sharing = DeltaSharingFactory
-      .create(
+    val sharing = DeltaSharing(
          providerJSON,
          "/dedicated/persisted/cache/location/"
       )

src/main/java/com/databricks/labs/delta/sharing/java/DeltaSharing.java

@@ -1,5 +1,6 @@
 package com.databricks.labs.delta.sharing.java;
 
+import com.databricks.labs.delta.sharing.java.adaptor.DeltaSharingJsonProvider;
 import com.databricks.labs.delta.sharing.java.format.parquet.TableReader;
 import io.delta.sharing.spark.DeltaSharingFileSystem;
 import io.delta.sharing.spark.DeltaSharingProfileProvider;
@@ -16,6 +17,7 @@
 import java.nio.file.OpenOption;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -31,30 +33,25 @@
 import scala.collection.Seq;
 
 /**
- * A wrapper class for {@link io.delta.sharing.spark.DeltaSharingRestClient} instance. This class
- * ensures we have access to a temp directory where parquet files will be kept during the life of
- * the JVM. The temp directory has a destroy hook register as do all the files when they land in the
- * temp directory. The class keeps a runtime map of metadata for each file stored in the temp
- * directory. If the metadata has changed we will retrieve the new state of the file. If the
- * metadata has remained the same we will use the available local copy in the temp directory.
+ * A wrapper class for {@link io.delta.sharing.spark.DeltaSharingRestClient}
+ * instance. This class ensures we have access to a temp directory where parquet
+ * files will be kept during the life of the JVM. The temp directory has a
+ * destroy hook register as do all the files when they land in the temp
+ * directory. The class keeps a runtime map of metadata for each file stored in
+ * the temp directory. If the metadata has changed we will retrieve the new
+ * state of the file. If the metadata has remained the same we will use the
+ * available local copy in the temp directory.
  * <p/>
  *
  * @author Milos Colic
  * @since 0.1.0
  */
 public class DeltaSharing {
-  DeltaSharingProfileProvider profileProvider;
-  DeltaSharingRestClient httpClient;
-  Path checkpointPath;
-  Path tempDir;
-  Map<String, DeltaTableMetadata> metadataMap;
-
-  /**
-   * Default constructor Construction of instances is delegated to a factory.
-   *
-   * @see DeltaSharingFactory
-   */
-  public DeltaSharing() {}
+  private DeltaSharingProfileProvider profileProvider;
+  private DeltaSharingRestClient httpClient;
+  private Path checkpointPath;
+  private Path tempDir;
+  private Map<String, DeltaTableMetadata> metadataMap;
 
   /**
    * Getter for {@link DeltaSharing#profileProvider}.
@@ -71,12 +68,61 @@ public DeltaSharingRestClient getHttpClient() {
   }
 
   /**
-   * Adapter method for getting a List of all tables. Scala API returns a {@link Seq} and we require
-   * a {@link List}.
+   * Getter for checkpointPath.
+   */
+  public Path getCheckpointPath() {
+    return checkpointPath;
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param profileProvider An instance of {@link DeltaSharingProfileProvider}.
+   * @param checkpointPath An path to a temporary checkpoint location.
+   * @throws IOException Transitive due to the call to
+   *         {@link Files#createTempDirectory(String, FileAttribute[])}.
+   */
+  public DeltaSharing(final DeltaSharingProfileProvider profileProvider,
+      final Path checkpointPath) throws IOException {
+
+    if (!Files.exists(checkpointPath)) {
+      Files.createDirectory(checkpointPath);
+    }
+    Path tempDir = Files.createTempDirectory(checkpointPath, "delta_sharing");
+    tempDir.toFile().deleteOnExit();
+
+    this.profileProvider = profileProvider;
+    this.httpClient =
+        new DeltaSharingRestClient(profileProvider, 120, 4, false);
+    this.checkpointPath = checkpointPath;
+    this.tempDir = tempDir;
+    this.metadataMap = new HashMap<>();
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param providerConf A valid JSON document corresponding to
+   *        {@link DeltaSharingProfileProvider}.
+   * @param checkpointLocation A string containing a path to be used as a
+   *        checkpoint location.
+   * @throws IOException Transitive due to the call to
+   *         {@link Files#createDirectories(Path, FileAttribute[])}.
+   */
+  public DeltaSharing(final String providerConf,
+      final String checkpointLocation) throws IOException {
+    this(new DeltaSharingJsonProvider(providerConf),
+        Paths.get(checkpointLocation));
+  }
+
+
+  /**
+   * Adapter method for getting a List of all tables. Scala API returns a
+   * {@link Seq} and we require a {@link List}.
    *
    * @return A list of all tables.
-   * @implNote Suppress unnecessary local variable is done to remove warnings for a decoupled Scala
-   *           to Java conversion call and a return call.
+   * @implNote Suppress unnecessary local variable is done to remove warnings
+   *           for a decoupled Scala to Java conversion call and a return call.
    */
   @SuppressWarnings("UnnecessaryLocalVariable")
   public List<Table> listAllTables() {
@@ -86,49 +132,56 @@ public List<Table> listAllTables() {
   }
 
   /**
-   * Getter for {@link io.delta.sharing.spark.DeltaSharingRestClient#getMetadata(Table)}.
+   * Getter for
+   * {@link io.delta.sharing.spark.DeltaSharingRestClient#getMetadata(Table)}.
+   *
+   * @return DeltaTableMetadata
    */
   public DeltaTableMetadata getMetadata(Table table) {
     return httpClient.getMetadata(table);
   }
 
   /**
-   * Getter for {@link io.delta.sharing.spark.DeltaSharingRestClient#getTableVersion(Table)}
+   * Getter for
+   * {@link io.delta.sharing.spark.DeltaSharingRestClient#getTableVersion(Table)}
    * (Table)}.
    */
   public long getTableVersion(Table table) {
     return httpClient.getTableVersion(table);
   }
 
   /**
-   * Adapter method for getting a List of files belonging to a {@link Table}. Scala API returns a
-   * {@link Seq} and we require a {@link List}.
+   * Adapter method for getting a List of files belonging to a {@link Table}.
+   * Scala API returns a {@link Seq} and we require a {@link List}.
    *
    * @return A list of files corresponding to a table.
-   * @implNote Suppress unnecessary local variable is done to remove warnings for a decoupled Scala
-   *           to Java conversion call and a return call.
+   * @implNote Suppress unnecessary local variable is done to remove warnings
+   *           for a decoupled Scala to Java conversion call and a return call.
    */
   @SuppressWarnings("UnnecessaryLocalVariable")
-  public List<AddFile> getFiles(Table table, List<String> predicates, Integer limit) {
-    Seq<String> predicatesSeq =
-        JavaConverters.asScalaIteratorConverter(predicates.iterator()).asScala().toSeq();
+  public List<AddFile> getFiles(Table table, List<String> predicates,
+      Integer limit) {
+    Seq<String> predicatesSeq = JavaConverters
+        .asScalaIteratorConverter(predicates.iterator()).asScala().toSeq();
     DeltaTableFiles deltaTableFiles;
     if (limit != null) {
-      deltaTableFiles = httpClient.getFiles(table, predicatesSeq, Some$.MODULE$.apply(limit));
+      deltaTableFiles =
+          httpClient.getFiles(table, predicatesSeq, Some$.MODULE$.apply(limit));
     } else {
-      deltaTableFiles = httpClient.getFiles(table, predicatesSeq, Option$.MODULE$.apply(null));
+      deltaTableFiles = httpClient.getFiles(table, predicatesSeq,
+          Option$.MODULE$.apply(null));
     }
     List<AddFile> files = JavaConverters.seqAsJavaList(deltaTableFiles.files());
     return files;
   }
 
   /**
-   * Adapter method for getting a List of files belonging to a {@link Table}. Scala API returns a
-   * {@link Seq} and we require a {@link List}.
+   * Adapter method for getting a List of files belonging to a {@link Table}.
+   * Scala API returns a {@link Seq} and we require a {@link List}.
    *
    * @return A list of files corresponding to a table.
-   * @implNote Suppress unnecessary local variable is done to remove warnings for a decoupled Scala
-   *           to Java conversion call and a return call.
+   * @implNote Suppress unnecessary local variable is done to remove warnings
+   *           for a decoupled Scala to Java conversion call and a return call.
    */
   public List<AddFile> getFiles(Table table, List<String> predicates) {
     return getFiles(table, predicates, null);
@@ -140,7 +193,8 @@ public List<AddFile> getFiles(Table table, List<String> predicates) {
    * @return A string representing coordinates of the table.
    */
   public String getCoordinates(Table table) {
-    Pattern pattern = Pattern.compile("[a-zA-Z0-9\\-_]*\\.[a-zA-Z0-9\\-_]*\\.[a-zA-Z0-9\\-_]*",
+    Pattern pattern = Pattern.compile(
+        "[a-zA-Z0-9\\-_]*\\.[a-zA-Z0-9\\-_]*\\.[a-zA-Z0-9\\-_]*",
         Pattern.CASE_INSENSITIVE);
     String coords = table.share() + "." + table.schema() + "." + table.name();
     Matcher matcher = pattern.matcher(coords);
@@ -153,7 +207,8 @@ public String getCoordinates(Table table) {
   }
 
   /**
-   * Getter for a temp file that will be stored in {@link DeltaSharing#checkpointPath}.
+   * Getter for a temp file that will be stored in
+   * {@link DeltaSharing#checkpointPath}.
    *
    * @param file File for which we are generating the checkpoint path for.
    * @return A fully qualified path for a checkpoint file copy.
@@ -185,8 +240,8 @@ private List<Path> writeCheckpointFiles(List<AddFile> files)
       throws IOException, URISyntaxException {
     List<Path> paths = new LinkedList<>();
     for (AddFile file : files) {
-      FSDataInputStream stream =
-          new FSDataInputStream(new InMemoryHttpInputStream(new URI(file.url())));
+      FSDataInputStream stream = new FSDataInputStream(
+          new InMemoryHttpInputStream(new URI(file.url())));
       Path path = getFileCheckpointPath(file);
       paths.add(path);
       Files.write(path, IOUtils.toByteArray(stream));
@@ -196,7 +251,8 @@ private List<Path> writeCheckpointFiles(List<AddFile> files)
   }
 
   /**
-   * Getter for a temp files that will be stored in {@link DeltaSharing#checkpointPath}.
+   * Getter for a temp files that will be stored in
+   * {@link DeltaSharing#checkpointPath}.
    *
    * @param files Files for which we are generating the checkpoint paths for.
    * @return A List of fully qualified paths for a checkpoint file copies.
@@ -211,13 +267,16 @@ private List<Path> getCheckpointPaths(List<AddFile> files) {
   }
 
   /**
-   * Resolves and constructs the {@link TableReader} instance associated with the table. It inspects
-   * the available {@link DeltaSharing#metadataMap} and based on the metadata it re-fetches the
-   * stale into the {@link DeltaSharing#checkpointPath} directory.
+   * Resolves and constructs the {@link TableReader} instance associated with
+   * the table. It inspects the available {@link DeltaSharing#metadataMap} and
+   * based on the metadata it re-fetches the stale into the
+   * {@link DeltaSharing#checkpointPath} directory.
    *
    * @param table Table whose reader is requested.
-   * @return An instance of {@link TableReader} that will manage the reads from the table.
-   * @throws IOException Transitive due to the call to {@link TableReader#TableReader(List)}.
+   * @return An instance of {@link TableReader} that will manage the reads from
+   *         the table.
+   * @throws IOException Transitive due to the call to
+   *         {@link TableReader#TableReader(List)}.
    */
   @SuppressWarnings("UnnecessaryLocalVariable")
   public TableReader<GenericRecord> getTableReader(Table table)
@@ -247,13 +306,16 @@ public TableReader<GenericRecord> getTableReader(Table table)
   }
 
   /**
-   * A reader method that reads all the records from all the files that belong to a table.
+   * A reader method that reads all the records from all the files that belong
+   * to a table.
    *
    * @param table An instance of {@link Table} whose records we are reading.
    * @return A list of records from the table instance.
-   * @throws IOException Transitive due to the call to {@link TableReader#read()}
+   * @throws IOException Transitive due to the call to
+   *         {@link TableReader#read()}
    */
-  public List<GenericRecord> getAllRecords(Table table) throws IOException, URISyntaxException {
+  public List<GenericRecord> getAllRecords(Table table)
+      throws IOException, URISyntaxException {
     TableReader<GenericRecord> tableReader = getTableReader(table);
     List<GenericRecord> records = new LinkedList<>();
     GenericRecord currentRecord = tableReader.read();
@@ -265,16 +327,18 @@ public List<GenericRecord> getAllRecords(Table table) throws IOException, URISyn
   }
 
   /**
-   * A reader method that reads all the records from all the files that belong to a table. This call
-   * will always create a new instance of the reader and will always return the same N records. For
-   * full reads use {@link DeltaSharing#getTableReader(Table)} to access the reader and then use
-   * {@link TableReader#readN(Integer)} to read blocks of records.
+   * A reader method that reads all the records from all the files that belong
+   * to a table. This call will always create a new instance of the reader and
+   * will always return the same N records. For full reads use
+   * {@link DeltaSharing#getTableReader(Table)} to access the reader and then
+   * use {@link TableReader#readN(Integer)} to read blocks of records.
    *
    * @param table An instance of {@link Table} whose records we are reading.
    * @param numRec Number of records to be read at most.
-   * @return A list of records from the table instance. If less records are available, only the
-   *         available records will be returned.
-   * @throws IOException Transitive due to the call to {@link TableReader#read()}
+   * @return A list of records from the table instance. If less records are
+   *         available, only the available records will be returned.
+   * @throws IOException Transitive due to the call to
+   *         {@link TableReader#read()}
    */
   public List<GenericRecord> getNumRecords(Table table, int numRec)
       throws IOException, URISyntaxException {