Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[GOBBLIN-2157] Copy table properties in iceberg distcp #4056

Merged
merged 5 commits into from
Sep 19, 2024
Merged

[GOBBLIN-2157] Copy table properties in iceberg distcp #4056

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8c59157
Replicate table properties when registering an iceberg table with src…
Will-Lo Sep 13, 2024
8b566f4
Cleanup
Will-Lo Sep 13, 2024
aa4e35e
Add comment to describe reasoning around overwriting conflicting tabl…
Will-Lo Sep 13, 2024
6a27404
Use source table properties as source of truth
Will-Lo Sep 16, 2024
e4331fe
Address review
Will-Lo Sep 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions ...anagement/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,8 +212,9 @@ public DatasetDescriptor getDatasetDescriptor(FileSystem fs) {
* @param dstMetadata is null if destination {@link IcebergTable} is absent, in which case registration is skipped */
protected void registerIcebergTable(TableMetadata srcMetadata, TableMetadata dstMetadata) {
if (dstMetadata != null) {
// use current destination metadata as 'base metadata' and source as 'updated metadata' while committing
this.tableOps.commit(dstMetadata, srcMetadata.replaceProperties(dstMetadata.properties()));
// Use current destination metadata as 'base metadata' and but update to source metadata when committing
// If any of the source table properties are deleted, they will be reflected in the destination table
Will-Lo marked this conversation as resolved.
Show resolved Hide resolved
this.tableOps.commit(dstMetadata, srcMetadata);
}
}
}
30 changes: 30 additions & 0 deletions ...ement/src/test/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Predicate;
import java.util.stream.Collectors;
Expand All @@ -35,6 +36,7 @@
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.avro.AvroSchemaUtil;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
Expand All @@ -47,6 +49,7 @@
import org.testng.annotations.Test;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;


Expand Down Expand Up @@ -181,6 +184,33 @@ public void testGetIncrementalSnapshotInfosIteratorRepeatedFiles() throws IOExce
}
}

/** Verify that registerIcebergTable will update existing table properties */
@Test
public void testNewTablePropertiesAreRegistered() throws Exception {
Map<String, String> srcTableProperties = Maps.newHashMap();
Map<String, String> destTableProperties = Maps.newHashMap();

srcTableProperties.put("newKey", "newValue");
// Expect the old value to be overwritten by the new value
srcTableProperties.put("testKey", "testValueNew");
destTableProperties.put("testKey", "testValueOld");
// Expect existing property values to be maintained if it does not exist on the source
Will-Lo marked this conversation as resolved.
Show resolved Hide resolved
destTableProperties.put("deletedTableProperty", "deletedTablePropertyValue");

TableIdentifier destTableId = TableIdentifier.of(dbName, "destTable");
catalog.createTable(destTableId, icebergSchema, null, destTableProperties);

IcebergTable destIcebergTable = new IcebergTable(destTableId, catalog.newTableOps(destTableId), catalogUri);
// Mock a source table with the same table UUID copying new properties
TableMetadata newSourceTableProperties = destIcebergTable.accessTableMetadata().replaceProperties(srcTableProperties);

destIcebergTable.registerIcebergTable(newSourceTableProperties, destIcebergTable.accessTableMetadata());
Assert.assertEquals(destIcebergTable.accessTableMetadata().properties().size(), 2);
Assert.assertEquals(destIcebergTable.accessTableMetadata().properties().get("newKey"), "newValue");
Assert.assertEquals(destIcebergTable.accessTableMetadata().properties().get("testKey"), "testValueNew");
Assert.assertNull(destIcebergTable.accessTableMetadata().properties().get("deletedTableProperty"));
}

/** full validation for a particular {@link IcebergSnapshotInfo} */
protected void verifySnapshotInfo(IcebergSnapshotInfo snapshotInfo, List<List<String>> perSnapshotFilesets, int overallNumSnapshots) {
// verify metadata file
Expand Down
Loading