Dell: Add Dell EMC EcsFileIO.

build.gradle

@@ -566,6 +566,23 @@ project(':iceberg-nessie') {
   }
 }
 
+project(':iceberg-dell') {
+  dependencies {
+    implementation project(':iceberg-core')
+    implementation project(':iceberg-common')
+    implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')
+    implementation 'com.emc.ecs:object-client-bundle'
+
+    testImplementation("org.apache.hadoop:hadoop-common") {
+      exclude group: 'org.apache.avro', module: 'avro'
+      exclude group: 'org.slf4j', module: 'slf4j-log4j12'
+    }
+    testImplementation "javax.xml.bind:jaxb-api"
+    testImplementation "javax.activation:activation"
+    testImplementation "org.glassfish.jaxb:jaxb-runtime"
+  }
+}
+
 @Memoized
 boolean isVersionFileExists() {
   return file('version.txt').exists()

dell/src/main/java/org/apache/iceberg/dell/BaseEcsFile.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.dell;
+
+import com.emc.object.s3.S3Client;
+import com.emc.object.s3.S3Exception;
+
+public class BaseEcsFile {
+
+  protected final S3Client client;
+  protected final String location;
+  protected final EcsURI uri;
+
+  public BaseEcsFile(S3Client client, String location) {
+    this.client = client;
+    this.location = location;
+    this.uri = EcsURI.create(location);
+  }
+
+  public String location() {
+    return location;
+  }
+
+  /**
+   * Check whether data file exists.
+   */
+  public boolean exists() {
+    try {
+      client.getObjectMetadata(uri.getBucket(), uri.getName());
+      return true;
+    } catch (S3Exception e) {
+      if (e.getHttpCode() == 404) {
+        return false;
+      } else {
+        throw e;
+      }
+    }
+  }
+}

dell/src/main/java/org/apache/iceberg/dell/EcsAppendOutputStream.java

@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.dell;
+
+import com.emc.object.s3.S3Client;
+import com.emc.object.s3.request.PutObjectRequest;
+import java.io.ByteArrayInputStream;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.io.PositionOutputStream;
+
+/**
+ * Use ECS append API to write data.
+ */
+class EcsAppendOutputStream extends PositionOutputStream {
+
+  private final S3Client client;
+
+  private final EcsURI key;
+
+  /**
+   * Local bytes cache that avoid too many requests
+   * <p>
+   * Use {@link ByteBuffer} to maintain offset.
+   */
+  private final ByteBuffer localCache;
+
+  /**
+   * A marker for data file to put first part instead of append first part.
+   */
+  private boolean firstPart = true;
+
+  /**
+   * Pos for {@link PositionOutputStream}
+   */
+  private long pos;
+
+  private EcsAppendOutputStream(S3Client client, EcsURI key, byte[] localCache) {
+    this.client = client;
+    this.key = key;
+    this.localCache = ByteBuffer.wrap(localCache);
+  }
+
+  /**
+   * Use built-in 1 KiB byte buffer
+   */
+  static EcsAppendOutputStream create(S3Client client, EcsURI uri) {
+    return createWithBufferSize(client, uri, 1024);
+  }
+
+  /**
+   * Create {@link PositionOutputStream} with specific buffer size.
+   */
+  static EcsAppendOutputStream createWithBufferSize(S3Client client, EcsURI uri, int size) {
+    return new EcsAppendOutputStream(client, uri, new byte[size]);
+  }
+
+  /**
+   * Write a byte. If buffer is full, upload the buffer.
+   */
+  @Override
+  public void write(int b) {
+    if (!checkBuffer(1)) {
+      flush();
+    }
+
+    localCache.put((byte) b);
+    pos += 1;
+  }
+
+  /**
+   * Write a byte.
+   * If buffer is full, upload the buffer.
+   * If buffer size &lt; input bytes, upload input bytes.
+   */
+  @Override
+  public void write(byte[] b, int off, int len) {
+    if (!checkBuffer(len)) {
+      flush();
+    }
+
+    if (checkBuffer(len)) {
+      localCache.put(b, off, len);
+    } else {
+      // if content > cache, directly flush itself.
+      flushBuffer(b, off, len);
+    }
+
+    pos += len;
+  }
+
+  private boolean checkBuffer(int nextWrite) {
+    return localCache.remaining() >= nextWrite;
+  }
+
+  private void flushBuffer(byte[] buffer, int offset, int length) {
+    if (firstPart) {
+      client.putObject(new PutObjectRequest(key.getBucket(), key.getName(),
+          new ByteArrayInputStream(buffer, offset, length)));
+      firstPart = false;
+    } else {
+      client.appendObject(key.getBucket(), key.getName(), new ByteArrayInputStream(buffer, offset, length));
+    }
+  }
+
+  /**
+   * Pos of the file
+   */
+  @Override
+  public long getPos() {
+    return pos;
+  }
+
+  /**
+   * Write cached bytes if present.
+   */
+  @Override
+  public void flush() {
+    if (localCache.remaining() < localCache.capacity()) {
+      localCache.flip();
+      flushBuffer(localCache.array(), localCache.arrayOffset(), localCache.remaining());
+      localCache.clear();
+    }
+  }
+
+  /**
+   * Trigger flush() when closing stream.
+   */
+  @Override
+  public void close() {
+    flush();
+  }
+}

dell/src/main/java/org/apache/iceberg/dell/EcsClientFactory.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.dell;
+
+import com.emc.object.s3.S3Client;
+import com.emc.object.s3.S3Config;
+import com.emc.object.s3.jersey.S3JerseyClient;
+import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public interface EcsClientFactory {
+
+  /**
+   * Create the ECS S3 Client from properties
+   */
+  static S3Client create(Map<String, String> properties) {
+    return createWithFactory(properties).orElseGet(() -> createDefault(properties));
+  }
+
+  /**
+   * Try to create the ECS S3 client from factory method.
+   */
+  static Optional<S3Client> createWithFactory(Map<String, String> properties) {
+    String factory = properties.get(EcsClientProperties.ECS_CLIENT_FACTORY);
+    if (factory == null || factory.isEmpty()) {
+      return Optional.empty();
+    }
+
+    DynConstructors.Ctor<EcsClientFactory> ctor;
+    try {
+      ctor = DynConstructors.builder(EcsClientFactory.class).impl(factory).buildChecked();
+    } catch (NoSuchMethodException e) {
+      throw new IllegalArgumentException(String.format(
+          "Cannot find EcsClientFactory implementation %s: %s", factory, e.getMessage()), e);
+    }
+
+    EcsClientFactory clientFactory;
+    try {
+      clientFactory = ctor.newInstance();
+    } catch (ClassCastException e) {
+      throw new IllegalArgumentException(
+          String.format("Cannot initialize Catalog, %s does not implement EcsClientFactory.", factory), e);
+    }
+
+    S3Client client = clientFactory.createS3Client(properties);
+
+    if (client == null) {
+      throw new IllegalArgumentException(String.format(
+          "Invalid EcsClientFactory %s that return null client",
+          factory));
+    }
+
+    return Optional.of(client);
+  }
+
+  /**
+   * Get built-in ECS S3 client.
+   */
+  static S3Client createDefault(Map<String, String> properties) {
+    Preconditions.checkNotNull(properties.get(EcsClientProperties.ENDPOINT),
+        "Endpoint(%s) cannot be null", EcsClientProperties.ENDPOINT);
+    Preconditions.checkNotNull(properties.get(EcsClientProperties.ACCESS_KEY_ID),
+        "Access key(%s) cannot be null", EcsClientProperties.ACCESS_KEY_ID);
+    Preconditions.checkNotNull(properties.get(EcsClientProperties.SECRET_ACCESS_KEY),
+        "Secret key(%s) cannot be null", EcsClientProperties.SECRET_ACCESS_KEY);
+
+    S3Config config = new S3Config(URI.create(properties.get(EcsClientProperties.ENDPOINT)));
+
+    config.withIdentity(properties.get(EcsClientProperties.ACCESS_KEY_ID))
+        .withSecretKey(properties.get(EcsClientProperties.SECRET_ACCESS_KEY));
+
+    return new S3JerseyClient(config);
+  }
+
+  S3Client createS3Client(Map<String, String> properties);
+}

dell/src/main/java/org/apache/iceberg/dell/EcsClientProperties.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.dell;
+
+/**
+ * Property constants of catalog
+ */
+public interface EcsClientProperties {
+
+  /**
+   * Access key id
+   */
+  String ACCESS_KEY_ID = "ecs.s3.access.key.id";
+
+  /**
+   * Secret access key
+   */
+  String SECRET_ACCESS_KEY = "ecs.s3.secret.access.key";
+
+  /**
+   * S3 endpoint
+   */
+  String ENDPOINT = "ecs.s3.endpoint";
+
+  /**
+   * Factory class of {@link EcsClientFactory}.
+   * <p>
+   * The config is optional. If properties above aren't enough, use this.
+   */
+  String ECS_CLIENT_FACTORY = "ecs.client.factory";
+}