Azure: Add FileIO that supports ADLSv2 storage

.github/labeler.yml

@@ -72,15 +72,20 @@ PIG:
   - pig/**/*
 AWS:
   - aws/**/*
+  - aws-bundle/**/*
 NESSIE:
   - nessie/**/*
 ALIYUN:
   - aliyun/**/*
 GCP:
   - gcp/**/*
+  - gcp-bundle/**/*
 DELL:
   - dell/**/*
 SNOWFLAKE:
   - snowflake/**/*
 OPENAPI:
   - open-api/**/*
+AZURE:
+  - azure/**/*
+  - azure-bundle/**/*

azure-bundle/NOTICE

@@ -0,0 +1,6 @@
+
+Apache Iceberg
+Copyright 2017-2023 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).

azure-bundle/build.gradle

@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+project(":iceberg-azure-bundle") {
+
+  apply plugin: 'com.github.johnrengelman.shadow'
+
+  tasks.jar.dependsOn tasks.shadowJar
+
+  dependencies {
+    implementation platform(libs.azuresdk.bom)
+    implementation "com.azure:azure-storage-file-datalake"
+    implementation "com.azure:azure-identity"
+  }
+
+  shadowJar {
+    archiveClassifier.set(null)
+    zip64 true
+
+    // include the LICENSE and NOTICE files for the shaded Jar
+    from(projectDir) {
+      include 'LICENSE'
+      include 'NOTICE'
+    }
+
+    // relocate Azure-specific versions
+    relocate 'io.netty', 'org.apache.iceberg.azure.shaded.io.netty'
+    relocate 'com.fasterxml.jackson', 'org.apache.iceberg.azure.shaded.com.fasterxml.jackson'
+  }
+
+  jar {
+    enabled = false
+  }
+}

azure/src/main/java/org/apache/iceberg/azure/AzureProperties.java

@@ -0,0 +1,59 @@
+/*
+ * 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.azure;
+
+import com.azure.identity.DefaultAzureCredential;
+import com.azure.identity.DefaultAzureCredentialBuilder;
+import com.azure.storage.file.datalake.DataLakePathClientBuilder;
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Optional;
+
+public class AzureProperties implements Serializable {
+  public static final String ADLSV2_READ_BLOCK_SIZE = "adlsv2.read.block-size-bytes";
+  public static final String ADLSV2_WRITE_BLOCK_SIZE = "adlsv2.write.block-size-bytes";
+  private static final DefaultAzureCredential DEFAULT_CREDENTIAL =
+      new DefaultAzureCredentialBuilder().build();
+
+  private Integer adlsv2ReadBlockSize;
+  private Long adlsv2WriteBlockSize;
+
+  public AzureProperties() {}
+
+  public AzureProperties(Map<String, String> properties) {
+    if (properties.containsKey(ADLSV2_READ_BLOCK_SIZE)) {
+      adlsv2ReadBlockSize = Integer.parseInt(properties.get(ADLSV2_READ_BLOCK_SIZE));
+    }
+    if (properties.containsKey(ADLSV2_WRITE_BLOCK_SIZE)) {
+      adlsv2WriteBlockSize = Long.parseLong(properties.get(ADLSV2_WRITE_BLOCK_SIZE));
+    }
+  }
+
+  public Optional<Integer> adlsv2ReadBlockSize() {
+    return Optional.ofNullable(adlsv2ReadBlockSize);
+  }
+
+  public Optional<Long> adlsv2WriteBlockSize() {
+    return Optional.ofNullable(adlsv2WriteBlockSize);
+  }
+
+  public <T extends DataLakePathClientBuilder> void applyCredentialConfiguration(T builder) {
+    builder.credential(DEFAULT_CREDENTIAL);
+  }
+}

azure/src/main/java/org/apache/iceberg/azure/adlsv2/ADLSv2FileIO.java

@@ -0,0 +1,132 @@
+/*
+ * 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.azure.adlsv2;
+
+import com.azure.core.http.HttpClient;
+import com.azure.storage.file.datalake.DataLakeFileClient;
+import com.azure.storage.file.datalake.DataLakePathClientBuilder;
+import java.util.Map;
+import org.apache.iceberg.azure.AzureProperties;
+import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.metrics.MetricsContext;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.util.SerializableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * FileIO implementation backed by Azure Data Lake Storage v2 (ADLSv2)
+ *
+ * <p>Locations follow the conventions used by Hadoop's Azure support, i.e.
+ *
+ * <pre>{@code abfs[s]://<container>@<storage account>.dfs.core.windows.net/<blob_path>}</pre>
+ *
+ * <p>See <a href="https://hadoop.apache.org/docs/stable/hadoop-azure/abfs.html">Hadoop Azure
+ * Support</a>
+ */
+public class ADLSv2FileIO implements FileIO {
+  private static final Logger LOG = LoggerFactory.getLogger(ADLSv2FileIO.class);
+  private static final String DEFAULT_METRICS_IMPL =
+      "org.apache.iceberg.hadoop.HadoopMetricsContext";
+
+  private static final HttpClient HTTP = HttpClient.createDefault();
+
+  private AzureProperties azureProperties;
+  private MetricsContext metrics = MetricsContext.nullMetrics();
+  private SerializableMap<String, String> properties = null;
+
+  /**
+   * No-arg constructor to load the FileIO dynamically.
+   *
+   * <p>All fields are initialized by calling {@link ADLSv2FileIO#initialize(Map)} later.
+   */
+  public ADLSv2FileIO() {}
+
+  @Override
+  public InputFile newInputFile(String path) {
+    return ADLSv2InputFile.of(path, client(path), azureProperties, metrics);
+  }
+
+  @Override
+  public InputFile newInputFile(String path, long length) {
+    return ADLSv2InputFile.of(path, length, client(path), azureProperties, metrics);
+  }
+
+  @Override
+  public OutputFile newOutputFile(String path) {
+    return ADLSv2OutputFile.of(path, client(path), azureProperties, metrics);
+  }
+
+  @Override
+  public void deleteFile(String path) {
+    // There is no specific contract about whether delete should fail
+    // and other FileIO providers ignore failure.  Log the failure for
+    // now as it is not a required operation for Iceberg.
+    try {
+      client(path).delete();
+    } catch (Exception e) {
+      LOG.warn("Failed to delete path: {}", path, e);
+    }
+  }
+
+  @Override
+  public Map<String, String> properties() {
+    return properties.immutableMap();
+  }
+
+  @VisibleForTesting
+  DataLakeFileClient client(String path) {
+    ADLSv2Location location = new ADLSv2Location(path);
+    DataLakePathClientBuilder clientBuilder =
+        new DataLakePathClientBuilder()
+            .httpClient(HTTP)
+            .endpoint(location.storageAccountUrl())
+            .fileSystemName(location.container())
+            .pathName(location.path());
+
+    azureProperties.applyCredentialConfiguration(clientBuilder);
+
+    return clientBuilder.buildFileClient();
+  }
+
+  @Override
+  public void initialize(Map<String, String> props) {
+    this.properties = SerializableMap.copyOf(props);
+    this.azureProperties = new AzureProperties(properties);
+
+    // Report Hadoop metrics if Hadoop is available
+    try {
+      DynConstructors.Ctor<MetricsContext> ctor =
+          DynConstructors.builder(MetricsContext.class)
+              .hiddenImpl(DEFAULT_METRICS_IMPL, String.class)
+              .buildChecked();
+      MetricsContext context = ctor.newInstance("abfs");
+      context.initialize(properties);
+      this.metrics = context;
+    } catch (NoClassDefFoundError | NoSuchMethodException | ClassCastException e) {
+      LOG.warn(
+          "Unable to load metrics class: '{}', falling back to null metrics",
+          DEFAULT_METRICS_IMPL,
+          e);
+    }
+  }
+}

azure/src/main/java/org/apache/iceberg/azure/adlsv2/ADLSv2InputFile.java

@@ -0,0 +1,70 @@
+/*
+ * 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.azure.adlsv2;
+
+import com.azure.storage.file.datalake.DataLakeFileClient;
+import org.apache.iceberg.azure.AzureProperties;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.SeekableInputStream;
+import org.apache.iceberg.metrics.MetricsContext;
+
+class ADLSv2InputFile extends BaseADLSv2File implements InputFile {
+  private Long fileSize;
+
+  static ADLSv2InputFile of(
+      String location,
+      DataLakeFileClient fileClient,
+      AzureProperties azureProperties,
+      MetricsContext metrics) {
+    return new ADLSv2InputFile(location, null, fileClient, azureProperties, metrics);
+  }
+
+  static ADLSv2InputFile of(
+      String location,
+      long length,
+      DataLakeFileClient fileClient,
+      AzureProperties azureProperties,
+      MetricsContext metrics) {
+    return new ADLSv2InputFile(
+        location, length > 0 ? length : null, fileClient, azureProperties, metrics);
+  }
+
+  ADLSv2InputFile(
+      String location,
+      Long fileSize,
+      DataLakeFileClient fileClient,
+      AzureProperties azureProperties,
+      MetricsContext metrics) {
+    super(location, fileClient, azureProperties, metrics);
+    this.fileSize = fileSize;
+  }
+
+  @Override
+  public long getLength() {
+    if (fileSize == null) {
+      this.fileSize = fileClient().getProperties().getFileSize();
+    }
+    return fileSize;
+  }
+
+  @Override
+  public SeekableInputStream newStream() {
+    return new ADLSv2InputStream(fileClient(), fileSize, azureProperties(), metrics());
+  }
+}