This is an automated email from the ASF dual-hosted git repository.
wlo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 18ec55bcd [GOBBLIN-1709] Create Iceberg Datasets Finder, Iceberg
Dataset and FileSet to generate Copy Entities to support Distcp for Iceberg
(#3560)
18ec55bcd is described below
commit 18ec55bcd7f78ba0c977c7236856b3d39f5ed7a6
Author: meethngala <[email protected]>
AuthorDate: Thu Sep 22 12:03:49 2022 -0500
[GOBBLIN-1709] Create Iceberg Datasets Finder, Iceberg Dataset and FileSet
to generate Copy Entities to support Distcp for Iceberg (#3560)
* initial commit for iceberg distcp.
* adding copy entity helper and icerbeg distcp template and test case.
* Adding unit tests and refactoring method definitions for an Iceberg
dataset.
* resolve conflicts after cleaning history
* update iceberg dataset and finder to include javadoc
* addressed comments on PR and aligned code check style
* renamed vars, added logging and updated javadoc
* update dataset descriptor with ternary operation and rename fs to sourceFs
* added source and target fs and update iceberg dataset finder constructor
* Update source and dest dataset methods as protected and add req args
constructor
* change the order of attributes for iceberg dataset finder ctor
* update iceberg dataset methods with correct source and target fs
Co-authored-by: Meeth Gala <[email protected]>
---
.../apache/gobblin/dataset/DatasetConstants.java | 1 +
.../management/copy/iceberg/IcebergDataset.java | 212 +++++++++++++++++++++
.../copy/iceberg/IcebergDatasetFinder.java | 97 ++++++++++
.../copy/iceberg/IcebergTableFileSet.java | 50 +++++
.../copy/iceberg/IcebergDatasetTest.java | 192 +++++++++++++++++++
.../embedded/EmbeddedGobblinDistcpTest.java | 2 +-
6 files changed, 553 insertions(+), 1 deletion(-)
diff --git
a/gobblin-api/src/main/java/org/apache/gobblin/dataset/DatasetConstants.java
b/gobblin-api/src/main/java/org/apache/gobblin/dataset/DatasetConstants.java
index d7045258a..dde22c0fa 100644
--- a/gobblin-api/src/main/java/org/apache/gobblin/dataset/DatasetConstants.java
+++ b/gobblin-api/src/main/java/org/apache/gobblin/dataset/DatasetConstants.java
@@ -25,6 +25,7 @@ public class DatasetConstants {
public static final String PLATFORM_HIVE = "hive";
public static final String PLATFORM_SALESFORCE = "salesforce";
public static final String PLATFORM_MYSQL = "mysql";
+ public static final String PLATFORM_ICEBERG = "iceberg";
/** Common metadata */
public static final String BRANCH = "branch";
diff --git
a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDataset.java
b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDataset.java
new file mode 100644
index 000000000..8d905c8e6
--- /dev/null
+++
b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDataset.java
@@ -0,0 +1,212 @@
+/*
+ * 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.net.URI;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.jetbrains.annotations.NotNull;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import lombok.Data;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.data.management.copy.CopyConfiguration;
+import org.apache.gobblin.data.management.copy.CopyEntity;
+import org.apache.gobblin.data.management.copy.CopyableDataset;
+import org.apache.gobblin.data.management.copy.CopyableFile;
+import
org.apache.gobblin.data.management.copy.prioritization.PrioritizedCopyableDataset;
+import org.apache.gobblin.data.management.partition.FileSet;
+import org.apache.gobblin.dataset.DatasetConstants;
+import org.apache.gobblin.dataset.DatasetDescriptor;
+import org.apache.gobblin.util.request_allocation.PushDownRequestor;
+
+/**
+ * Iceberg dataset implementing {@link CopyableDataset}.
+ */
+@Slf4j
+@Getter
+public class IcebergDataset implements PrioritizedCopyableDataset {
+ private final String dbName;
+ private final String inputTableName;
+ private final IcebergTable icebergTable;
+ protected final Properties properties;
+ protected final FileSystem sourceFs;
+
+ private final Optional<String> sourceMetastoreURI;
+ private final Optional<String> targetMetastoreURI;
+
+ /** Target metastore URI */
+ public static final String TARGET_METASTORE_URI_KEY =
+ IcebergDatasetFinder.ICEBERG_DATASET_PREFIX +
".copy.target.metastore.uri";
+ /** Target database name */
+ public static final String TARGET_DATABASE_KEY =
IcebergDatasetFinder.ICEBERG_DATASET_PREFIX + ".copy.target.database";
+
+ public IcebergDataset(String db, String table, IcebergTable icebergTbl,
Properties properties, FileSystem sourceFs) {
+ this.dbName = db;
+ this.inputTableName = table;
+ this.icebergTable = icebergTbl;
+ this.properties = properties;
+ this.sourceFs = sourceFs;
+ this.sourceMetastoreURI =
+
Optional.fromNullable(this.properties.getProperty(IcebergDatasetFinder.ICEBERG_HIVE_CATALOG_METASTORE_URI_KEY));
+ this.targetMetastoreURI =
+
Optional.fromNullable(this.properties.getProperty(TARGET_METASTORE_URI_KEY));
+ }
+
+ /**
+ * Represents a source {@link FileStatus} and a {@link Path} destination.
+ */
+ @Data
+ private static class SourceAndDestination {
+ private final FileStatus source;
+ private final Path destination;
+ }
+
+ @Override
+ public String datasetURN() {
+ // TODO: verify!
+ return this.dbName + "." + this.inputTableName;
+ }
+
+ /**
+ * Finds all files read by the table and generates CopyableFiles.
+ * For the specific semantics see {@link #getCopyEntities}.
+ */
+ @Override
+ public Iterator<FileSet<CopyEntity>> getFileSetIterator(FileSystem targetFs,
CopyConfiguration configuration) {
+ return getCopyEntities(targetFs, configuration);
+ }
+ /**
+ * Finds all files read by the table and generates CopyableFiles.
+ * For the specific semantics see {@link #getCopyEntities}.
+ */
+ @Override
+ public Iterator<FileSet<CopyEntity>> getFileSetIterator(FileSystem targetFs,
CopyConfiguration configuration,
+ Comparator<FileSet<CopyEntity>> prioritizer,
PushDownRequestor<FileSet<CopyEntity>> requestor) {
+ // TODO: Implement PushDownRequestor and priority based copy entity
iteration
+ return getCopyEntities(targetFs, configuration);
+ }
+
+ /**
+ * Finds all files read by the table and generates {@link CopyEntity}s for
duplicating the table.
+ */
+ Iterator<FileSet<CopyEntity>> getCopyEntities(FileSystem targetFs,
CopyConfiguration configuration) {
+ FileSet<CopyEntity> fileSet = new
IcebergTableFileSet(this.getInputTableName(), this, targetFs, configuration);
+ return Iterators.singletonIterator(fileSet); }
+
+ /**
+ * Finds all files read by the table file set and generates {@link
CopyEntity}s for duplicating the table.
+ */
+ @VisibleForTesting
+ Collection<CopyEntity> generateCopyEntities(FileSystem targetFs,
CopyConfiguration configuration) throws IOException {
+ String fileSet = this.getInputTableName();
+ List<CopyEntity> copyEntities = Lists.newArrayList();
+ Map<Path, FileStatus> pathToFileStatus = getFilePathsToFileStatus();
+ log.info("{}.{} - found {} candidate source paths", dbName,
inputTableName, pathToFileStatus.size());
+
+ for (CopyableFile.Builder builder :
getCopyableFilesFromPaths(pathToFileStatus, configuration, targetFs)) {
+ CopyableFile fileEntity =
+
builder.fileSet(fileSet).datasetOutputPath(targetFs.getUri().getPath()).build();
+ fileEntity.setSourceData(getSourceDataset(this.sourceFs));
+ fileEntity.setDestinationData(getDestinationDataset(targetFs));
+ copyEntities.add(fileEntity);
+ }
+ log.info("{}.{} - generated {} copy entities", dbName, inputTableName,
copyEntities.size());
+ return copyEntities;
+ }
+
+ /**
+ * Get builders for a {@link CopyableFile} for each file path
+ */
+ protected List<CopyableFile.Builder> getCopyableFilesFromPaths(Map<Path,
FileStatus> pathToFileStatus, CopyConfiguration configuration, FileSystem
targetFs) throws IOException {
+
+ List<CopyableFile.Builder> builders = Lists.newArrayList();
+ List<SourceAndDestination> dataFiles = Lists.newArrayList();
+ Configuration defaultHadoopConfiguration = new Configuration();
+ FileSystem actualSourceFs;
+
+ for (Map.Entry<Path, FileStatus> entry : pathToFileStatus.entrySet()) {
+ dataFiles.add(new SourceAndDestination(entry.getValue(),
targetFs.makeQualified(entry.getKey())));
+ }
+
+ for (SourceAndDestination sourceAndDestination : dataFiles) {
+ actualSourceFs =
sourceAndDestination.getSource().getPath().getFileSystem(defaultHadoopConfiguration);
+
+ // TODO: Add ancestor owner and permissions in future releases
+ builders.add(CopyableFile.fromOriginAndDestination(actualSourceFs,
sourceAndDestination.getSource(),
+ sourceAndDestination.getDestination(), configuration));
+ }
+ return builders;
+ }
+ /**
+ * Finds all files of the Iceberg's current snapshot
+ * Returns a map of path, file status for each file that needs to be copied
+ */
+ protected Map<Path, FileStatus> getFilePathsToFileStatus() throws
IOException {
+ Map<Path, FileStatus> result = Maps.newHashMap();
+ IcebergTable icebergTable = this.getIcebergTable();
+ IcebergSnapshotInfo icebergSnapshotInfo =
icebergTable.getCurrentSnapshotInfo();
+
+ log.info("{}.{} - loaded snapshot '{}' from metadata path: '{}'", dbName,
inputTableName,
+ icebergSnapshotInfo.getSnapshotId(),
icebergSnapshotInfo.getMetadataPath());
+ List<String> pathsToCopy = icebergSnapshotInfo.getAllPaths();
+
+ for (String pathString : pathsToCopy) {
+ Path path = new Path(pathString);
+ result.put(path, this.sourceFs.getFileStatus(path));
+ }
+ return result;
+ }
+
+ protected DatasetDescriptor getSourceDataset(FileSystem sourceFs) {
+ return getDatasetDescriptor(sourceMetastoreURI, sourceFs);
+ }
+
+ protected DatasetDescriptor getDestinationDataset(FileSystem targetFs) {
+ return getDatasetDescriptor(targetMetastoreURI, targetFs);
+ }
+
+ @NotNull
+ private DatasetDescriptor getDatasetDescriptor(Optional<String>
stringMetastoreURI, FileSystem fs) {
+ String currentTable = this.getDbName() + "." + this.getInputTableName();
+
+ URI hiveMetastoreURI = stringMetastoreURI.isPresent() ?
URI.create(stringMetastoreURI.get()) : null;
+
+ DatasetDescriptor currentDataset =
+ new DatasetDescriptor(DatasetConstants.PLATFORM_ICEBERG,
hiveMetastoreURI, currentTable);
+ currentDataset.addMetadata(DatasetConstants.FS_URI,
fs.getUri().toString());
+ return currentDataset;
+ }
+}
diff --git
a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java
b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java
new file mode 100644
index 000000000..4eb77980b
--- /dev/null
+++
b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java
@@ -0,0 +1,97 @@
+/*
+ * 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.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.dataset.DatasetConstants;
+import org.apache.gobblin.dataset.IterableDatasetFinder;
+import org.apache.gobblin.util.HadoopUtils;
+
+/**
+ * Finds {@link IcebergDataset}s. Will look for tables in a database using a
{@link IcebergCatalog},
+ * and creates a {@link IcebergDataset} for each one.
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class IcebergDatasetFinder implements
IterableDatasetFinder<IcebergDataset> {
+
+ public static final String ICEBERG_DATASET_PREFIX =
DatasetConstants.PLATFORM_ICEBERG + ".dataset";
+ public static final String ICEBERG_HIVE_CATALOG_METASTORE_URI_KEY =
ICEBERG_DATASET_PREFIX + ".hive.metastore.uri";
+ public static final String ICEBERG_DB_NAME = ICEBERG_DATASET_PREFIX +
".database.name";
+ public static final String ICEBERG_TABLE_NAME = ICEBERG_DATASET_PREFIX +
".table.name";
+
+ protected final FileSystem sourceFs;
+ private final Properties properties;
+
+ /**
+ * Finds all {@link IcebergDataset}s in the file system using the Iceberg
Catalog.
+ * Both Iceberg database name and table name are mandatory based on current
implementation.
+ * Later we may explore supporting datasets similar to Hive
+ * @return List of {@link IcebergDataset}s in the file system.
+ * @throws IOException
+ */
+ @Override
+ public List<IcebergDataset> findDatasets() throws IOException {
+ List<IcebergDataset> matchingDatasets = new ArrayList<>();
+ if (StringUtils.isBlank(properties.getProperty(ICEBERG_DB_NAME)) ||
StringUtils.isBlank(properties.getProperty(ICEBERG_TABLE_NAME))) {
+ throw new IllegalArgumentException(String.format("Iceberg database name:
{%s} or Iceberg table name: {%s} is missing",
+ ICEBERG_DB_NAME, ICEBERG_TABLE_NAME));
+ }
+ String dbName = properties.getProperty(ICEBERG_DB_NAME);
+ String tblName = properties.getProperty(ICEBERG_TABLE_NAME);
+
+ Configuration configuration =
HadoopUtils.getConfFromProperties(properties);
+
+ IcebergCatalog icebergCatalog =
IcebergCatalogFactory.create(configuration);
+ /* Each Iceberg dataset maps to an Iceberg table
+ * TODO: The user provided database and table names needs to be
pre-checked and verified against the existence of a valid Iceberg table
+ */
+ matchingDatasets.add(createIcebergDataset(dbName, tblName, icebergCatalog,
properties, sourceFs));
+ log.info("Found {} matching datasets: {} for the database name: {} and
table name: {}", matchingDatasets.size(), matchingDatasets, dbName, tblName);
+
+ return matchingDatasets;
+ }
+
+ @Override
+ public Path commonDatasetRoot() {
+ return new Path("/");
+ }
+
+ @Override
+ public Iterator<IcebergDataset> getDatasetsIterator() throws IOException {
+ return findDatasets().iterator();
+ }
+
+ protected IcebergDataset createIcebergDataset(String dbName, String tblName,
IcebergCatalog icebergCatalog, Properties properties, FileSystem fs) {
+ IcebergTable icebergTable = icebergCatalog.openTable(dbName, tblName);
+ return new IcebergDataset(dbName, tblName, icebergTable, properties, fs);
+ }
+}
diff --git
a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTableFileSet.java
b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTableFileSet.java
new file mode 100644
index 000000000..1e5abfbee
--- /dev/null
+++
b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTableFileSet.java
@@ -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.
+ */
+
+package org.apache.gobblin.data.management.copy.iceberg;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import org.apache.hadoop.fs.FileSystem;
+
+import org.apache.gobblin.data.management.copy.CopyConfiguration;
+import org.apache.gobblin.data.management.copy.CopyEntity;
+import org.apache.gobblin.data.management.partition.FileSet;
+
+
+/**
+ * A {@link FileSet} for Iceberg datasets containing information associated
with an Iceberg table and generates {@link CopyEntity}
+ */
+public class IcebergTableFileSet extends FileSet<CopyEntity> {
+
+ private final CopyConfiguration copyConfiguration;
+ private final FileSystem targetFs;
+ private final IcebergDataset icebergDataset;
+
+ public IcebergTableFileSet(String name, IcebergDataset icebergDataset,
FileSystem targetFs, CopyConfiguration configuration) {
+ super(name, icebergDataset);
+ this.copyConfiguration = configuration;
+ this.targetFs = targetFs;
+ this.icebergDataset = icebergDataset;
+ }
+
+ @Override
+ protected Collection<CopyEntity> generateCopyEntities() throws IOException {
+ return this.icebergDataset.generateCopyEntities(this.targetFs,
this.copyConfiguration);
+ }
+}
diff --git
a/gobblin-data-management/src/test/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetTest.java
b/gobblin-data-management/src/test/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetTest.java
new file mode 100644
index 000000000..f409ee67a
--- /dev/null
+++
b/gobblin-data-management/src/test/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetTest.java
@@ -0,0 +1,192 @@
+/*
+ * 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.net.URI;
+import java.net.URISyntaxException;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.google.api.client.util.Maps;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+
+import org.apache.gobblin.data.management.copy.CopyConfiguration;
+import org.apache.gobblin.data.management.copy.CopyContext;
+import org.apache.gobblin.data.management.copy.CopyEntity;
+import org.apache.gobblin.data.management.copy.PreserveAttributes;
+
+
+public class
+IcebergDatasetTest {
+
+ static final String METADATA_PATH = "/root/iceberg/test/metadata";
+ static final String MANIFEST_PATH =
"/root/iceberg/test/metadata/test_manifest";
+ static final String MANIFEST_LIST_PATH =
"/root/iceberg/test/metadata/test_manifest/data";
+ static final String MANIFEST_FILE_PATH1 =
"/root/iceberg/test/metadata/test_manifest/data/a";
+ static final String MANIFEST_FILE_PATH2 =
"/root/iceberg/test/metadata/test_manifest/data/b";
+
+ @Test
+ public void testGetFilePaths() throws IOException {
+
+ List<String> pathsToCopy = new ArrayList<>();
+ pathsToCopy.add(MANIFEST_FILE_PATH1);
+ pathsToCopy.add(MANIFEST_FILE_PATH2);
+ Map<Path, FileStatus> expected = Maps.newHashMap();
+ expected.put(new Path(MANIFEST_FILE_PATH1), null);
+ expected.put(new Path(MANIFEST_FILE_PATH2), null);
+
+ IcebergTable icebergTable = Mockito.mock(IcebergTable.class);
+ FileSystem fs = Mockito.mock(FileSystem.class);
+ IcebergSnapshotInfo icebergSnapshotInfo =
Mockito.mock(IcebergSnapshotInfo.class);
+
+
Mockito.when(icebergTable.getCurrentSnapshotInfo()).thenReturn(icebergSnapshotInfo);
+ Mockito.when(icebergSnapshotInfo.getAllPaths()).thenReturn(pathsToCopy);
+ IcebergDataset icebergDataset = new IcebergDataset("test_db_name",
"test_tbl_name", icebergTable, new Properties(), fs);
+
+ Map<Path, FileStatus> actual = icebergDataset.getFilePathsToFileStatus();
+ Assert.assertEquals(actual, expected);
+ }
+
+ /**
+ * Test case to generate copy entities for all the file paths for a mocked
iceberg table.
+ * The assumption here is that we create copy entities for all the matching
file paths,
+ * without calculating any difference between the source and destination
+ */
+ @Test
+ public void testGenerateCopyEntitiesWhenDestEmpty() throws IOException,
URISyntaxException {
+
+ FileSystem fs = Mockito.mock(FileSystem.class);
+ String test_db_name = "test_db_name";
+ String test_table_name = "test_tbl_name";
+ String test_qualified_path =
"/root/iceberg/test/destination/sub_path_destination";
+ String test_uri_path = "/root/iceberg/test/output";
+ Properties properties = new Properties();
+ properties.setProperty("data.publisher.final.dir", "/test");
+ List<String> expected = new ArrayList<>(Arrays.asList(METADATA_PATH,
MANIFEST_PATH, MANIFEST_LIST_PATH, MANIFEST_FILE_PATH1, MANIFEST_FILE_PATH2));
+
+ CopyConfiguration copyConfiguration = CopyConfiguration.builder(null,
properties)
+ .preserve(PreserveAttributes.fromMnemonicString(""))
+ .copyContext(new CopyContext())
+ .build();
+
+ List<String> listedManifestFilePaths = Arrays.asList(MANIFEST_FILE_PATH1,
MANIFEST_FILE_PATH2);
+ IcebergSnapshotInfo.ManifestFileInfo manifestFileInfo = new
IcebergSnapshotInfo.ManifestFileInfo(MANIFEST_LIST_PATH,
listedManifestFilePaths);
+ List<IcebergSnapshotInfo.ManifestFileInfo> manifestFiles =
Arrays.asList(manifestFileInfo);
+ IcebergTable icebergTable = new MockedIcebergTable(METADATA_PATH,
MANIFEST_PATH, manifestFiles);
+ IcebergDataset icebergDataset = new IcebergDataset(test_db_name,
test_table_name, icebergTable, new Properties(), fs);
+ DestinationFileSystem destinationFileSystem = new DestinationFileSystem();
+ destinationFileSystem.addPath(METADATA_PATH);
+ destinationFileSystem.addPath(MANIFEST_PATH);
+ destinationFileSystem.addPath(MANIFEST_LIST_PATH);
+ destinationFileSystem.addPath(MANIFEST_FILE_PATH1);
+ destinationFileSystem.addPath(MANIFEST_FILE_PATH2);
+
+ mockFileSystemMethodCalls(fs, destinationFileSystem.pathToFileStatus,
test_qualified_path, test_uri_path);
+
+ Collection<CopyEntity> copyEntities =
icebergDataset.generateCopyEntities(fs, copyConfiguration);
+ verifyCopyEntities(copyEntities, expected);
+
+ }
+
+ private void verifyCopyEntities(Collection<CopyEntity> copyEntities,
List<String> expected) {
+ List<String> actual = new ArrayList<>();
+ for (CopyEntity copyEntity : copyEntities) {
+ String json = copyEntity.toString();
+ JsonObject jsonObject = new Gson().fromJson(json, JsonObject.class);
+ JsonObject objectData =
+
jsonObject.getAsJsonObject("object-data").getAsJsonObject("origin").getAsJsonObject("object-data");
+ JsonObject pathObject =
objectData.getAsJsonObject("path").getAsJsonObject("object-data").getAsJsonObject("uri");
+ String filepath =
pathObject.getAsJsonPrimitive("object-data").getAsString();
+ actual.add(filepath);
+ }
+ Assert.assertEquals(actual.size(), expected.size());
+ Assert.assertEqualsNoOrder(actual.toArray(), expected.toArray());
+ }
+
+ private void mockFileSystemMethodCalls(FileSystem fs, Map<Path, FileStatus>
pathToFileStatus, String qualifiedPath, String uriPath)
+ throws URISyntaxException, IOException {
+
+ Mockito.when(fs.getUri()).thenReturn(new URI(null, null, uriPath, null));
+ for (Map.Entry<Path, FileStatus> entry : pathToFileStatus.entrySet()) {
+ Path path = entry.getKey();
+ FileStatus fileStatus = entry.getValue();
+ Mockito.when(fs.getFileStatus(path)).thenReturn(fileStatus);
+ Mockito.when(fs.makeQualified(path)).thenReturn(new Path(qualifiedPath));
+ }
+ }
+
+ private static class MockedIcebergTable extends IcebergTable {
+
+ String metadataPath;
+ String manifestListPath;
+ List<IcebergSnapshotInfo.ManifestFileInfo> manifestFiles;
+
+ public MockedIcebergTable(String metadataPath, String manifestListPath,
List<IcebergSnapshotInfo.ManifestFileInfo> manifestFiles) {
+ super(null);
+ this.metadataPath = metadataPath;
+ this.manifestListPath = manifestListPath;
+ this.manifestFiles = manifestFiles;
+ }
+
+ @Override
+ public IcebergSnapshotInfo getCurrentSnapshotInfo() {
+ Long snapshotId = 0L;
+ Instant timestamp = Instant.ofEpochMilli(0L);
+ return new IcebergSnapshotInfo(snapshotId, timestamp, metadataPath,
manifestListPath, manifestFiles);
+ }
+ }
+
+ private static class DestinationFileSystem {
+ Map<Path, FileStatus> pathToFileStatus;
+
+ public DestinationFileSystem() {
+ this.pathToFileStatus = Maps.newHashMap();
+ }
+
+ public void addPath(String pathString) {
+ if (StringUtils.isBlank(pathString)) {
+ throw new IllegalArgumentException("Missing path value for the file
system");
+ }
+ Path path = new Path(pathString);
+ FileStatus fileStatus = new FileStatus();
+ fileStatus.setPath(path);
+ this.pathToFileStatus.put(path, fileStatus);
+ }
+
+ public void addPath(String pathString, FileStatus fileStatus) {
+ Path path = new Path(pathString);
+ this.pathToFileStatus.put(path, fileStatus);
+ }
+ }
+}
+
diff --git
a/gobblin-data-management/src/test/java/org/apache/gobblin/runtime/embedded/EmbeddedGobblinDistcpTest.java
b/gobblin-data-management/src/test/java/org/apache/gobblin/runtime/embedded/EmbeddedGobblinDistcpTest.java
index c042ac1b2..2fefded1f 100644
---
a/gobblin-data-management/src/test/java/org/apache/gobblin/runtime/embedded/EmbeddedGobblinDistcpTest.java
+++
b/gobblin-data-management/src/test/java/org/apache/gobblin/runtime/embedded/EmbeddedGobblinDistcpTest.java
@@ -386,4 +386,4 @@ public class EmbeddedGobblinDistcpTest {
}
}
-}
+}
\ No newline at end of file
