[GOBBLIN-2231] Implement IcebergSource to enable copying Iceberg data files to any dest (#4146)

* Implement IcebergSource to enable copying Iceberg data files to any dest

* Introduce partition-aware discovery & lookback period based copy for IcebergSource

* Added unit tests for IcebergSource

* Added unit tests for IcebergFileStreamHelper

* Add FilePathWithPartition class in IcebergTable

* Address review comments

* Fix indentation

* Handling for yyyy-MM-dd-00 partition format

* Remove unused import

* Address review comment

* Fix core tests error

* Fix failing test

* Fix failing test

Author: khandelwal-prateek

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergFileStreamExtractor.java

@@ -0,0 +1,58 @@
+/*
+ * 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.gobblin.data.management.copy.iceberg;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.commons.lang3.NotImplementedException;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.WorkUnitState;
+import org.apache.gobblin.data.management.copy.FileAwareInputStream;
+import org.apache.gobblin.source.extractor.filebased.FileBasedExtractor;
+
+/**
+ * Extractor for file streaming mode that creates FileAwareInputStream for each file.
+ * 
+ * This extractor is used when {@code iceberg.record.processing.enabled=false} to stream
+ * Iceberg table files as binary data to destinations like Azure, HDFS
+ * 
+ * Each "record" is a {@link FileAwareInputStream} representing one file from
+ * the Iceberg table. The downstream writer handles streaming the file content.
+ */
+@Slf4j
+public class IcebergFileStreamExtractor extends FileBasedExtractor<String, FileAwareInputStream> {
+
+  public IcebergFileStreamExtractor(WorkUnitState workUnitState) throws IOException {
+    super(workUnitState, new IcebergFileStreamHelper(workUnitState));
+  }
+
+  @Override
+  public String getSchema() {
+    // For file streaming, schema is not used by IdentityConverter; returning a constant
+    return "FileAwareInputStream";
+  }
+
+  @Override
+  public Iterator<FileAwareInputStream> downloadFile(String filePath) throws IOException {
+    throw new NotImplementedException("Not yet implemented");
+  }
+
+}

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergFileStreamHelper.java

@@ -0,0 +1,140 @@
+/*
+ * 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.gobblin.data.management.copy.iceberg;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.source.extractor.filebased.FileBasedHelperException;
+import org.apache.gobblin.source.extractor.filebased.TimestampAwareFileBasedHelper;
+
+/**
+ * File-based helper for Iceberg file streaming operations.
+ *
+ * This helper supports file streaming mode where Iceberg table files
+ * are streamed as binary data without record-level processing.
+ */
+@Slf4j
+public class IcebergFileStreamHelper implements TimestampAwareFileBasedHelper {
+
+  private final State state;
+  private final Configuration configuration;
+  private FileSystem fileSystem;
+
+  public IcebergFileStreamHelper(State state) {
+    this.state = state;
+    this.configuration = new Configuration();
+
+    // Add any Hadoop configuration from job properties
+    for (String key : state.getPropertyNames()) {
+      if (key.startsWith("fs.") || key.startsWith("hadoop.")) {
+        configuration.set(key, state.getProp(key));
+      }
+    }
+  }
+
+  @Override
+  public void connect() throws FileBasedHelperException {
+    try {
+      this.fileSystem = FileSystem.get(configuration);
+      log.info("Connected to Iceberg file stream helper with FileSystem: {}", fileSystem.getClass().getSimpleName());
+    } catch (IOException e) {
+      throw new FileBasedHelperException("Failed to initialize FileSystem for Iceberg file streaming", e);
+    }
+  }
+
+  @Override
+  public List<String> ls(String path) throws FileBasedHelperException {
+    try {
+      // For Iceberg, file discovery is handled by IcebergSource
+      // This method returns files from work unit configuration
+      List<String> filesToPull = state.getPropAsList(ConfigurationKeys.SOURCE_FILEBASED_FILES_TO_PULL, "");
+      log.debug("Returning {} files for processing", filesToPull.size());
+      return filesToPull;
+    } catch (Exception e) {
+      throw new FileBasedHelperException("Failed to list files", e);
+    }
+  }
+
+  @Override
+  public InputStream getFileStream(String filePath) throws FileBasedHelperException {
+    try {
+      Path path = new Path(filePath);
+      FileSystem fs = getFileSystemForPath(path);
+      return fs.open(path);
+    } catch (IOException e) {
+      throw new FileBasedHelperException("Failed to get file stream for: " + filePath, e);
+    }
+  }
+
+  @Override
+  public long getFileSize(String filePath) throws FileBasedHelperException {
+    try {
+      Path path = new Path(filePath);
+      FileSystem fs = getFileSystemForPath(path);
+      return fs.getFileStatus(path).getLen();
+    } catch (IOException e) {
+      throw new FileBasedHelperException("Failed to get file size for: " + filePath, e);
+    }
+  }
+
+  @Override
+  public long getFileMTime(String filePath) throws FileBasedHelperException {
+    try {
+      Path path = new Path(filePath);
+      FileSystem fs = getFileSystemForPath(path);
+      return fs.getFileStatus(path).getModificationTime();
+    } catch (IOException e) {
+      throw new FileBasedHelperException("Failed to get file modification time for: " + filePath, e);
+    }
+  }
+
+  private FileSystem getFileSystemForPath(Path path) throws IOException {
+    // If path has a different scheme than the default FileSystem, get scheme-specific FS
+    if (path.toUri().getScheme() != null &&
+      !path.toUri().getScheme().equals(fileSystem.getUri().getScheme())) {
+      return path.getFileSystem(configuration);
+    }
+    return fileSystem;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (fileSystem != null) {
+      try {
+        fileSystem.close();
+        log.info("Closed Iceberg file stream helper and FileSystem connection");
+      } catch (IOException e) {
+        log.warn("Error closing FileSystem connection", e);
+        throw e;
+      }
+    } else {
+      log.debug("Closing Iceberg file stream helper - no FileSystem to close");
+    }
+  }
+
+}