rebased

Author: Marco Scalzo

client-spark/build.gradle.kts

@@ -0,0 +1,38 @@
+plugins {
+    java
+    id("com.diffplug.spotless")
+    id("whitefox.java-conventions")
+}
+
+group = "io.whitefox"
+version = "spark-connector"
+
+repositories {
+    mavenCentral()
+}
+
+val hadoopVersion = "3.3.6"
+dependencies {
+    // DELTA
+    testImplementation(String.format("org.apache.hadoop:hadoop-common:%s", hadoopVersion))
+    testImplementation("io.delta:delta-sharing-spark_2.12:1.0.2")
+
+    //SPARK
+    testImplementation("org.apache.spark:spark-core_2.12:3.3.2")
+    testImplementation("org.apache.spark:spark-sql_2.12:3.3.2")
+    testImplementation("com.github.mrpowers:spark-fast-tests_2.12:1.3.0")
+
+    //JUNIT
+    testImplementation("org.junit.jupiter:junit-jupiter:5.8.1")
+
+}
+
+// region code formatting
+spotless {
+    java {}
+}
+// endregion
+
+tasks.getByName<Test>("test") {
+    useJUnitPlatform()
+}

client-spark/src/test/java/client/ITDeltaSharingClient.java

@@ -0,0 +1,60 @@
+package client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.github.mrpowers.spark.fast.tests.DatasetComparer;
+import java.net.URISyntaxException;
+import java.util.List;
+import models.*;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import scala.collection.GenMap;
+import utils.StorageManagerInitializer;
+
+public class ITDeltaSharingClient implements DatasetComparer {
+
+  private final String tablePath = String.format(
+      "%s#%s.%s.%s",
+      getClass().getClassLoader().getResource("MrFoxProfile.json"),
+      "s3share",
+      "s3schema",
+      "s3Table1");
+
+  private final SparkSession spark = SparkSession.builder()
+      .appName("delta sharing client test")
+      .master("local[1, 4]")
+      .getOrCreate();
+
+  @BeforeAll
+  static void initStorageManager() throws URISyntaxException, JsonProcessingException {
+    new StorageManagerInitializer().initStorageManager();
+  }
+
+  @Test
+  void showS3Table1withQueryTableApi() {
+    var ds = spark.read().format("deltaSharing").load(tablePath);
+    var expectedSchema = new StructType(new StructField[] {
+      new StructField("id", DataType.fromDDL("long"), true, new Metadata(GenMap.empty()))
+    });
+    var expectedData = spark
+        .createDataFrame(
+            List.of(
+                new MrFoxDeltaTableSchema(0),
+                new MrFoxDeltaTableSchema(3),
+                new MrFoxDeltaTableSchema(2),
+                new MrFoxDeltaTableSchema(1),
+                new MrFoxDeltaTableSchema(4)),
+            MrFoxDeltaTableSchema.class)
+        .toDF();
+
+    assertEquals(expectedSchema.json(), ds.schema().json());
+    assertEquals(5, ds.count());
+    assertSmallDatasetEquality(ds, expectedData, true, false, false, 500);
+  }
+}

client-spark/src/test/java/models/AddTableToSchemaInput.java

@@ -0,0 +1,38 @@
+package models;
+
+public class AddTableToSchemaInput {
+  private final String name;
+  private final TableReference reference;
+
+  public AddTableToSchemaInput(String name, TableReference reference) {
+    this.name = name;
+    this.reference = reference;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public TableReference getReference() {
+    return reference;
+  }
+
+  public static class TableReference {
+
+    private final String providerName;
+    private final String name;
+
+    public TableReference(String providerName, String name) {
+      this.providerName = providerName;
+      this.name = name;
+    }
+
+    public String getProviderName() {
+      return providerName;
+    }
+
+    public String getName() {
+      return name;
+    }
+  }
+}

client-spark/src/test/java/models/CreateShareInput.java

@@ -0,0 +1,35 @@
+package models;
+
+import java.util.List;
+
+public class CreateShareInput {
+
+  private final String name;
+  private final String comment;
+  private final List<String> recipients;
+  private final List<String> schemas;
+
+  public CreateShareInput(
+      String name, String comment, List<String> recipients, List<String> schemas) {
+    this.name = name;
+    this.comment = comment;
+    this.recipients = recipients;
+    this.schemas = schemas;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public String getComment() {
+    return comment;
+  }
+
+  public List<String> getRecipients() {
+    return recipients;
+  }
+
+  public List<String> getSchemas() {
+    return schemas;
+  }
+}

client-spark/src/test/java/models/CreateStorage.java

@@ -0,0 +1,39 @@
+package models;
+
+public class CreateStorage {
+
+  private final String name;
+  private final String comment;
+  private final String type;
+  private final S3Properties properties;
+  private final boolean skipValidation;
+
+  public CreateStorage(
+      String name, String comment, String type, S3Properties properties, boolean skipValidation) {
+    this.name = name;
+    this.comment = comment;
+    this.type = type;
+    this.properties = properties;
+    this.skipValidation = skipValidation;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public String getComment() {
+    return comment;
+  }
+
+  public String getType() {
+    return type;
+  }
+
+  public S3Properties getProperties() {
+    return properties;
+  }
+
+  public boolean isSkipValidation() {
+    return skipValidation;
+  }
+}

client-spark/src/test/java/models/CreateTableInput.java

@@ -0,0 +1,56 @@
+package models;
+
+public class CreateTableInput {
+  private final String name;
+  private final String comment;
+  private final boolean skipValidation;
+  private final DeltaTableProperties properties;
+
+  public CreateTableInput(
+      String name, String comment, boolean skipValidation, DeltaTableProperties properties) {
+    this.name = name;
+    this.comment = comment;
+    this.skipValidation = skipValidation;
+    this.properties = properties;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public String getComment() {
+    return comment;
+  }
+
+  public boolean isSkipValidation() {
+    return skipValidation;
+  }
+
+  public DeltaTableProperties getProperties() {
+    return properties;
+  }
+
+  public static class DeltaTableProperties {
+    private final String type;
+    private final String location;
+    private final String additionalProperties;
+
+    public DeltaTableProperties(String type, String location, String additionalProperties) {
+      this.type = type;
+      this.location = location;
+      this.additionalProperties = additionalProperties;
+    }
+
+    public String getType() {
+      return type;
+    }
+
+    public String getLocation() {
+      return location;
+    }
+
+    public String getAdditionalProperties() {
+      return additionalProperties;
+    }
+  }
+}

client-spark/src/test/java/models/MrFoxDeltaTableSchema.java

@@ -0,0 +1,13 @@
+package models;
+
+public class MrFoxDeltaTableSchema {
+  private final long id;
+
+  public MrFoxDeltaTableSchema(long id) {
+    this.id = id;
+  }
+
+  public long getId() {
+    return id;
+  }
+}

client-spark/src/test/java/models/ProviderInput.java

@@ -0,0 +1,26 @@
+package models;
+
+public class ProviderInput {
+
+  private final String name;
+  private final String storageName;
+  private final String metastoreName;
+
+  public ProviderInput(String name, String storageName, String metastoreName) {
+    this.name = name;
+    this.storageName = storageName;
+    this.metastoreName = metastoreName;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public String getStorageName() {
+    return storageName;
+  }
+
+  public String getMetastoreName() {
+    return metastoreName;
+  }
+}

client-spark/src/test/java/models/S3Properties.java

@@ -0,0 +1,37 @@
+package models;
+
+public class S3Properties {
+  private final AwsCredentials credentials;
+
+  public S3Properties(AwsCredentials credentials) {
+    this.credentials = credentials;
+  }
+
+  public AwsCredentials getCredentials() {
+    return credentials;
+  }
+
+  public static class AwsCredentials {
+    private final String awsAccessKeyId;
+    private final String awsSecretAccessKey;
+    private final String region;
+
+    public AwsCredentials(String awsAccessKeyId, String awsSecretAccessKey, String region) {
+      this.awsAccessKeyId = awsAccessKeyId;
+      this.awsSecretAccessKey = awsSecretAccessKey;
+      this.region = region;
+    }
+
+    public String getAwsAccessKeyId() {
+      return awsAccessKeyId;
+    }
+
+    public String getAwsSecretAccessKey() {
+      return awsSecretAccessKey;
+    }
+
+    public String getRegion() {
+      return region;
+    }
+  }
+}

client-spark/src/test/java/models/S3TestConfig.java

@@ -0,0 +1,26 @@
+package models;
+
+public class S3TestConfig {
+
+  private final String region;
+  private final String accessKey;
+  private final String secretKey;
+
+  public String getRegion() {
+    return region;
+  }
+
+  public String getAccessKey() {
+    return accessKey;
+  }
+
+  public String getSecretKey() {
+    return secretKey;
+  }
+
+  public S3TestConfig(String region, String accessKey, String secretKey) {
+    this.region = region;
+    this.accessKey = accessKey;
+    this.secretKey = secretKey;
+  }
+}