[GOBBLIN-1961] Allow IcebergDatasetFinder to use separate names for source vs. destination-side DB and table
#3835
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -63,10 +63,8 @@ | |
| @Slf4j | ||
| @Getter | ||
| public class IcebergDataset implements PrioritizedCopyableDataset { | ||
| private final IcebergTable srcIcebergTable; | ||
| /* CAUTION: *hopefully* `destIcebergTable` exists... although that's not necessarily been verified yet */ | ||
| private final IcebergTable destIcebergTable; | ||
| protected final Properties properties; | ||
| protected final FileSystem sourceFs; | ||
|
|
@@ -75,9 +73,7 @@ public class IcebergDataset implements PrioritizedCopyableDataset { | |
| /** Destination database name */ | ||
| public static final String DESTINATION_DATABASE_KEY = IcebergDatasetFinder.ICEBERG_DATASET_PREFIX + ".destination.database"; | ||
|
|
||
| public IcebergDataset(IcebergTable srcIcebergTable, IcebergTable destIcebergTable, Properties properties, FileSystem sourceFs) { | ||
| this.srcIcebergTable = srcIcebergTable; | ||
| this.destIcebergTable = destIcebergTable; | ||
| this.properties = properties; | ||
|
|
@@ -117,17 +113,17 @@ public Iterator<FileSet<CopyEntity>> getFileSetIterator(FileSystem targetFs, Cop | |
| return createFileSets(targetFs, configuration); | ||
| } | ||
|
|
||
| /** @return unique ID for dataset (based on the source-side table), usable as a {@link CopyEntity#getFileSet}, for atomic publication grouping */ | ||
| protected String getFileSetId() { | ||
|
Contributor
There was a problem hiding this comment. Nit: I feel like this is better encapsulated as getSourceTableId because fileSetId() makes me think about some subset of the table, not just the src table.
Contributor
Author
There was a problem hiding this comment. the name originally arose because it's the value provided to the does it make more sense in context, or do you still recommend to change?
Contributor
There was a problem hiding this comment. I see, I guess we can keep it the same here for consistency, just that in this usecase it really is referring to the table when this is a table distcp |
||
| return this.srcIcebergTable.getTableId().toString(); | ||
| } | ||
|
|
||
| /** | ||
| * Generates {@link FileSet}s, being themselves able to generate {@link CopyEntity}s for all files, data and metadata, | ||
| * comprising the iceberg/table, so as to fully specify remaining table replication. | ||
| */ | ||
| protected Iterator<FileSet<CopyEntity>> createFileSets(FileSystem targetFs, CopyConfiguration configuration) { | ||
| FileSet<CopyEntity> fileSet = new IcebergTableFileSet(this.getFileSetId(), this, targetFs, configuration); | ||
| return Iterators.singletonIterator(fileSet); | ||
| } | ||
|
|
||
|
|
@@ -140,7 +136,7 @@ Collection<CopyEntity> generateCopyEntities(FileSystem targetFs, CopyConfigurati | |
| String fileSet = this.getFileSetId(); | ||
| List<CopyEntity> copyEntities = Lists.newArrayList(); | ||
| Map<Path, FileStatus> pathToFileStatus = getFilePathsToFileStatus(targetFs, copyConfig); | ||
| log.info("~{}~ found {} candidate source paths", fileSet, pathToFileStatus.size()); | ||
|
|
||
| Configuration defaultHadoopConfiguration = new Configuration(); | ||
| for (Map.Entry<Path, FileStatus> entry : pathToFileStatus.entrySet()) { | ||
|
|
@@ -165,8 +161,8 @@ Collection<CopyEntity> generateCopyEntities(FileSystem targetFs, CopyConfigurati | |
| copyEntities.add(fileEntity); | ||
| } | ||
| // TODO: Filter properties specific to iceberg registration and avoid serializing every global property | ||
| copyEntities.add(createPostPublishStep()); | ||
| log.info("~{}~ generated {} copy entities", fileSet, copyEntities.size()); | ||
| return copyEntities; | ||
| } | ||
|
|
||
|
|
@@ -187,8 +183,8 @@ protected Map<Path, FileStatus> getFilePathsToFileStatus(FileSystem targetFs, Co | |
| IcebergSnapshotInfo currentSnapshotOverview = icebergTable.getCurrentSnapshotInfoOverviewOnly(); | ||
| if (currentSnapshotOverview.getMetadataPath().map(isPresentOnTarget).orElse(false) && | ||
| isPresentOnTarget.apply(currentSnapshotOverview.getManifestListPath())) { | ||
| log.info("~{}~ skipping entire iceberg, since snapshot '{}' at '{}' and metadata '{}' both present on target", | ||
| this.getFileSetId(), currentSnapshotOverview.getSnapshotId(), | ||
| currentSnapshotOverview.getManifestListPath(), | ||
| currentSnapshotOverview.getMetadataPath().orElse("<<ERROR: MISSING!>>")); | ||
| return Maps.newHashMap(); | ||
|
|
@@ -198,7 +194,7 @@ protected Map<Path, FileStatus> getFilePathsToFileStatus(FileSystem targetFs, Co | |
| Iterators.transform(icebergIncrementalSnapshotInfos, snapshotInfo -> { | ||
| // log each snapshot, for context, in case of `FileNotFoundException` during `FileSystem.getFileStatus()` | ||
| String manListPath = snapshotInfo.getManifestListPath(); | ||
| log.info("~{}~ loaded snapshot '{}' at '{}' from metadata path: '{}'", this.getFileSetId(), | ||
| snapshotInfo.getSnapshotId(), manListPath, snapshotInfo.getMetadataPath().orElse("<<inherited>>")); | ||
| // ALGO: an iceberg's files form a tree of four levels: metadata.json -> manifest-list -> manifest -> data; | ||
| // most critically, all are presumed immutable and uniquely named, although any may be replaced. we depend | ||
|
|
@@ -224,18 +220,17 @@ protected Map<Path, FileStatus> getFilePathsToFileStatus(FileSystem targetFs, Co | |
| missingPaths.addAll(mfi.getListedFilePaths()); | ||
| } | ||
| } | ||
| log.info("~{}~ snapshot '{}': collected {} additional source paths", | ||
| this.getFileSetId(), snapshotInfo.getSnapshotId(), missingPaths.size()); | ||
| return missingPaths.iterator(); | ||
| } else { | ||
| log.info("~{}~ snapshot '{}' already present on target... skipping (including contents)", | ||
| this.getFileSetId(), snapshotInfo.getSnapshotId()); | ||
| // IMPORTANT: separately consider metadata path, to handle case of 'metadata-only' snapshot reusing mf-list | ||
| Optional<String> metadataPath = snapshotInfo.getMetadataPath(); | ||
| Optional<String> nonReplicatedMetadataPath = metadataPath.filter(p -> !isPresentOnTarget.apply(p)); | ||
| metadataPath.ifPresent(ignore -> | ||
| log.info("~{}~ metadata IS {} already present on target", this.getFileSetId(), nonReplicatedMetadataPath.isPresent() ? "NOT" : "ALSO") | ||
| ); | ||
| return nonReplicatedMetadataPath.map(p -> Lists.newArrayList(p).iterator()).orElse(Collections.emptyIterator()); | ||
| } | ||
|
|
@@ -255,7 +250,7 @@ protected Map<Path, FileStatus> getFilePathsToFileStatus(FileSystem targetFs, Co | |
| try { | ||
| results.put(path, this.sourceFs.getFileStatus(path)); | ||
| if (growthTracker.isAnotherMilestone(results.size())) { | ||
| log.info("~{}~ collected file status on '{}' source paths", this.getFileSetId(), results.size()); | ||
| } | ||
| } catch (FileNotFoundException fnfe) { | ||
| if (!shouldTolerateMissingSourceFiles) { | ||
|
|
@@ -265,7 +260,7 @@ protected Map<Path, FileStatus> getFilePathsToFileStatus(FileSystem targetFs, Co | |
| String total = ++numSourceFilesNotFound + " total"; | ||
| String speculation = "either premature deletion broke time-travel or metadata read interleaved among delete"; | ||
| errorConsolidator.prepLogMsg(path).ifPresent(msg -> | ||
| log.warn("~{}~ source {} ({}... {})", this.getFileSetId(), msg, speculation, total) | ||
| ); | ||
| } | ||
| } | ||
|
|
@@ -326,8 +321,8 @@ protected DatasetDescriptor getDestinationDataset(FileSystem targetFs) { | |
| return this.destIcebergTable.getDatasetDescriptor(targetFs); | ||
| } | ||
|
|
||
| private PostPublishStep createPostPublishStep() { | ||
| IcebergRegisterStep icebergRegisterStep = new IcebergRegisterStep(this.srcIcebergTable.getTableId(), this.destIcebergTable.getTableId(), this.properties); | ||
| return new PostPublishStep(getFileSetId(), Maps.newHashMap(), icebergRegisterStep, 0); | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Might be out of scope for this PR, but shouldn't our default behavior not have that assumption and create the table if missing? That's the contract for Hive distcp
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
for some catalogs, like openhouse, that's not actually possible: the tables must have been already mated in a way that we are unable to accomplish ourselves (i.e. from outside the catalog itself)