You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "meethngala (via GitHub)" <gi...@apache.org> on 2023/04/11 20:08:09 UTC
[GitHub] [gobblin] meethngala opened a new pull request, #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
meethngala opened a new pull request, #3673:
URL: https://github.com/apache/gobblin/pull/3673
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-XXX
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] meethngala commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "meethngala (via GitHub)" <gi...@apache.org>.
meethngala commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1166128803
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java:
##########
@@ -198,7 +198,8 @@ protected 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) {
Review Comment:
yeah, that's the assumption. We would want to rethink later, how we want to support in cases where destination iceberg table is absent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] homatthew commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "homatthew (via GitHub)" <gi...@apache.org>.
homatthew commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1163310390
##########
gradle/scripts/dependencyDefinitions.gradle:
##########
@@ -123,7 +123,7 @@ ext.externalDependency = [
"guiceMultibindings": "com.google.inject.extensions:guice-multibindings:4.0",
"guiceServlet": "com.google.inject.extensions:guice-servlet:4.0",
"derby": "org.apache.derby:derby:10.12.1.1",
- "mockito": "org.mockito:mockito-inline:4.11.0", // upgraded to allow mocking for constructors, static and final methods; specifically for iceberg distcp
+ "mockito": "org.mockito:mockito-core:4.11.0",
Review Comment:
Add a comment here as a warning not to change this to mockito-inline. If you need features from mockito-inline, please add mockito-inline as a mockmaker plugin
Example: https://github.com/apache/gobblin/pull/3651/files#r1163182255
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] meethngala commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "meethngala (via GitHub)" <gi...@apache.org>.
meethngala commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1164612967
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -88,7 +92,7 @@ public List<IcebergDataset> findDatasets() throws IOException {
IcebergCatalog sourceIcebergCatalog = createIcebergCatalog(this.properties, CatalogLocation.SOURCE);
IcebergCatalog destinationIcebergCatalog = createIcebergCatalog(this.properties, CatalogLocation.DESTINATION);
/* Each Iceberg dataset maps to an Iceberg table */
- matchingDatasets.add(createIcebergDataset(dbName, tblName, sourceIcebergCatalog, destinationIcebergCatalog, properties, sourceFs));
+ matchingDatasets.add(createIcebergDataset(dbName, tblName, sourceIcebergCatalog, destinationIcebergCatalog, this.properties, sourceFs));
Review Comment:
I don't see why it shouldn't be specific to the instance of the class. I'll update it in my latest commit!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] meethngala commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "meethngala (via GitHub)" <gi...@apache.org>.
meethngala commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1166089729
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDataset.java:
##########
@@ -316,8 +316,8 @@ protected DatasetDescriptor getDestinationDataset(FileSystem targetFs) {
return this.destIcebergTable.getDatasetDescriptor(targetFs);
}
- private PostPublishStep createPostPublishStep(IcebergTable srcIcebergTable, IcebergTable dstIcebergTable) {
- IcebergRegisterStep icebergRegisterStep = new IcebergRegisterStep(srcIcebergTable, dstIcebergTable);
+ private PostPublishStep createPostPublishStep(String dbName, String inputTableName, Properties properties) {
Review Comment:
Added the TODO as suggested above in my latest commit!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] meethngala commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "meethngala (via GitHub)" <gi...@apache.org>.
meethngala commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1164615833
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -49,14 +52,9 @@
@RequiredArgsConstructor
public class IcebergDatasetFinder implements IterableDatasetFinder<IcebergDataset> {
public static final String ICEBERG_DATASET_PREFIX = DatasetConstants.PLATFORM_ICEBERG + ".dataset";
- public static final String ICEBERG_CLUSTER_KEY = "cluster";
- public static final String ICEBERG_SRC_CATALOG_CLASS_KEY = ICEBERG_DATASET_PREFIX + ".source.catalog.class";
public static final String DEFAULT_ICEBERG_CATALOG_CLASS = "org.apache.gobblin.data.management.copy.iceberg.IcebergHiveCatalog";
- public static final String ICEBERG_SRC_CATALOG_URI_KEY = ICEBERG_DATASET_PREFIX + ".source.catalog.uri";
- public static final String ICEBERG_SRC_CLUSTER_NAME = ICEBERG_DATASET_PREFIX + ".source.cluster.name";
- public static final String ICEBERG_DEST_CATALOG_CLASS_KEY = ICEBERG_DATASET_PREFIX + ".destination.catalog.class";
- public static final String ICEBERG_DEST_CATALOG_URI_KEY = ICEBERG_DATASET_PREFIX + ".copy.destination.catalog.uri";
- public static final String ICEBERG_DEST_CLUSTER_NAME = ICEBERG_DATASET_PREFIX + ".destination.cluster.name";
+ public static final String ICEBERG_CATALOG_CLASS = "catalog.class";
+ public static final String ICEBERG_CATALOG_URI_KEY = "catalog.uri";
Review Comment:
I agree with you this and this might pass a lot more properties not specific to the catalog itself. I have changed the prefix to include catalog and even added the javadoc outlining the usage of these properties
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] homatthew commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "homatthew (via GitHub)" <gi...@apache.org>.
homatthew commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1163322437
##########
gradle/scripts/dependencyDefinitions.gradle:
##########
@@ -123,7 +123,7 @@ ext.externalDependency = [
"guiceMultibindings": "com.google.inject.extensions:guice-multibindings:4.0",
"guiceServlet": "com.google.inject.extensions:guice-servlet:4.0",
"derby": "org.apache.derby:derby:10.12.1.1",
- "mockito": "org.mockito:mockito-inline:4.11.0", // upgraded to allow mocking for constructors, static and final methods; specifically for iceberg distcp
+ "mockito": "org.mockito:mockito-core:4.11.0",
Review Comment:
https://github.com/mockito/mockito/wiki/What%27s-new-in-Mockito-2#unmockable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] autumnust commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "autumnust (via GitHub)" <gi...@apache.org>.
autumnust commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1166116386
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java:
##########
@@ -198,7 +198,8 @@ protected 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) {
Review Comment:
ok i guess your method javadoc explains
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] phet commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1164668823
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDataset.java:
##########
@@ -316,8 +316,8 @@ protected DatasetDescriptor getDestinationDataset(FileSystem targetFs) {
return this.destIcebergTable.getDatasetDescriptor(targetFs);
}
- private PostPublishStep createPostPublishStep(IcebergTable srcIcebergTable, IcebergTable dstIcebergTable) {
- IcebergRegisterStep icebergRegisterStep = new IcebergRegisterStep(srcIcebergTable, dstIcebergTable);
+ private PostPublishStep createPostPublishStep(String dbName, String inputTableName, Properties properties) {
Review Comment:
ah yes, I see the challenge arising from the hadoop props. these are likely to have a predictable prefix too, so we could retain only specific ones... but that might take trial-an-error, so, for now, I suggest adding a TODO documenting our preference not to serialize absolutely every prop... and yet the challenge w/ the hadoop config.
as we return to this, we should also investigate whether `Configuration` may itself already be serializable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] meethngala commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "meethngala (via GitHub)" <gi...@apache.org>.
meethngala commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1164613288
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -118,30 +116,27 @@ protected IcebergDataset createIcebergDataset(String dbName, String tblName, Ice
return new IcebergDataset(dbName, tblName, srcIcebergTable, destIcebergTable, properties, fs);
}
- protected IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
- Map<String, String> catalogProperties = new HashMap<>();
+ protected static IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
+ String prefix = getConfigPrefix(location);
+ Map<String, String> catalogProperties = loadCatalogProperties(properties, prefix);
Configuration configuration = HadoopUtils.getConfFromProperties(properties);
- String catalogUri;
- String icebergCatalogClassName;
- switch (location) {
- case SOURCE:
- catalogUri = properties.getProperty(ICEBERG_SRC_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Source Catalog Table Service URI is required", ICEBERG_SRC_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_SRC_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_SRC_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- case DESTINATION:
- catalogUri = properties.getProperty(ICEBERG_DEST_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Destination Catalog Table Service URI is required", ICEBERG_DEST_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_DEST_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_DEST_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- default:
- throw new UnsupportedOperationException("Incorrect desired location: %s provided for creating Iceberg Catalog" + location);
+ String icebergCatalogClassName = catalogProperties.getOrDefault(ICEBERG_CATALOG_CLASS, DEFAULT_ICEBERG_CATALOG_CLASS);
+ return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ }
+
+ protected static Map<String, String> loadCatalogProperties(Properties properties, String configPrefix) {
+ Map<String, String> catalogProperties = new HashMap<>();
+ Config config = ConfigBuilder.create().loadProps(properties, configPrefix).build();
+ for (Map.Entry<String, ConfigValue> entry : config.entrySet()) {
+ catalogProperties.put(entry.getKey(), entry.getValue().unwrapped().toString());
}
+ String catalogUri = config.getString(ICEBERG_CATALOG_URI_KEY);
+ Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Catalog Table Service URI is required", configPrefix + ICEBERG_CATALOG_URI_KEY);
catalogProperties.put(CatalogProperties.URI, catalogUri);
- return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ return catalogProperties;
+ }
+
+ private static String getConfigPrefix(CatalogLocation location) {
+ return ICEBERG_DATASET_PREFIX + "." + location.toString().toLowerCase() + ".";
Review Comment:
gotcha! Created a method inside enum as suggested. The changes are reflected in my latest commit.
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -118,30 +116,27 @@ protected IcebergDataset createIcebergDataset(String dbName, String tblName, Ice
return new IcebergDataset(dbName, tblName, srcIcebergTable, destIcebergTable, properties, fs);
}
- protected IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
- Map<String, String> catalogProperties = new HashMap<>();
+ protected static IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
+ String prefix = getConfigPrefix(location);
+ Map<String, String> catalogProperties = loadCatalogProperties(properties, prefix);
Configuration configuration = HadoopUtils.getConfFromProperties(properties);
- String catalogUri;
- String icebergCatalogClassName;
- switch (location) {
- case SOURCE:
- catalogUri = properties.getProperty(ICEBERG_SRC_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Source Catalog Table Service URI is required", ICEBERG_SRC_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_SRC_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_SRC_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- case DESTINATION:
- catalogUri = properties.getProperty(ICEBERG_DEST_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Destination Catalog Table Service URI is required", ICEBERG_DEST_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_DEST_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_DEST_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- default:
- throw new UnsupportedOperationException("Incorrect desired location: %s provided for creating Iceberg Catalog" + location);
+ String icebergCatalogClassName = catalogProperties.getOrDefault(ICEBERG_CATALOG_CLASS, DEFAULT_ICEBERG_CATALOG_CLASS);
+ return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ }
+
+ protected static Map<String, String> loadCatalogProperties(Properties properties, String configPrefix) {
+ Map<String, String> catalogProperties = new HashMap<>();
+ Config config = ConfigBuilder.create().loadProps(properties, configPrefix).build();
+ for (Map.Entry<String, ConfigValue> entry : config.entrySet()) {
+ catalogProperties.put(entry.getKey(), entry.getValue().unwrapped().toString());
}
+ String catalogUri = config.getString(ICEBERG_CATALOG_URI_KEY);
+ Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Catalog Table Service URI is required", configPrefix + ICEBERG_CATALOG_URI_KEY);
Review Comment:
yeah... I missed it. Updated it in my latest commit!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] codecov-commenter commented on pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#issuecomment-1507702141
## [Codecov](https://codecov.io/gh/apache/gobblin/pull/3673?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#3673](https://codecov.io/gh/apache/gobblin/pull/3673?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b4fe89e) into [master](https://codecov.io/gh/apache/gobblin/commit/bb7e31026a65d31e54ac0963d0b0b6a12ec6d71c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bb7e310) will **decrease** coverage by `2.13%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #3673 +/- ##
============================================
- Coverage 46.86% 44.74% -2.13%
+ Complexity 10755 2091 -8664
============================================
Files 2138 411 -1727
Lines 83999 17716 -66283
Branches 9334 2158 -7176
============================================
- Hits 39367 7927 -31440
+ Misses 41041 8932 -32109
+ Partials 3591 857 -2734
```
[see 1730 files with indirect coverage changes](https://codecov.io/gh/apache/gobblin/pull/3673/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] homatthew commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "homatthew (via GitHub)" <gi...@apache.org>.
homatthew commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1163280357
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -88,7 +92,7 @@ public List<IcebergDataset> findDatasets() throws IOException {
IcebergCatalog sourceIcebergCatalog = createIcebergCatalog(this.properties, CatalogLocation.SOURCE);
IcebergCatalog destinationIcebergCatalog = createIcebergCatalog(this.properties, CatalogLocation.DESTINATION);
/* Each Iceberg dataset maps to an Iceberg table */
- matchingDatasets.add(createIcebergDataset(dbName, tblName, sourceIcebergCatalog, destinationIcebergCatalog, properties, sourceFs));
+ matchingDatasets.add(createIcebergDataset(dbName, tblName, sourceIcebergCatalog, destinationIcebergCatalog, this.properties, sourceFs));
Review Comment:
`this.properties` but why not `this.sourceFs`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] phet commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "phet (via GitHub)" <gi...@apache.org>.
phet commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1163489856
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -118,30 +116,27 @@ protected IcebergDataset createIcebergDataset(String dbName, String tblName, Ice
return new IcebergDataset(dbName, tblName, srcIcebergTable, destIcebergTable, properties, fs);
}
- protected IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
- Map<String, String> catalogProperties = new HashMap<>();
+ protected static IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
+ String prefix = getConfigPrefix(location);
+ Map<String, String> catalogProperties = loadCatalogProperties(properties, prefix);
Configuration configuration = HadoopUtils.getConfFromProperties(properties);
- String catalogUri;
- String icebergCatalogClassName;
- switch (location) {
- case SOURCE:
- catalogUri = properties.getProperty(ICEBERG_SRC_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Source Catalog Table Service URI is required", ICEBERG_SRC_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_SRC_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_SRC_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- case DESTINATION:
- catalogUri = properties.getProperty(ICEBERG_DEST_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Destination Catalog Table Service URI is required", ICEBERG_DEST_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_DEST_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_DEST_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- default:
- throw new UnsupportedOperationException("Incorrect desired location: %s provided for creating Iceberg Catalog" + location);
+ String icebergCatalogClassName = catalogProperties.getOrDefault(ICEBERG_CATALOG_CLASS, DEFAULT_ICEBERG_CATALOG_CLASS);
+ return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ }
+
+ protected static Map<String, String> loadCatalogProperties(Properties properties, String configPrefix) {
+ Map<String, String> catalogProperties = new HashMap<>();
+ Config config = ConfigBuilder.create().loadProps(properties, configPrefix).build();
+ for (Map.Entry<String, ConfigValue> entry : config.entrySet()) {
+ catalogProperties.put(entry.getKey(), entry.getValue().unwrapped().toString());
}
+ String catalogUri = config.getString(ICEBERG_CATALOG_URI_KEY);
+ Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Catalog Table Service URI is required", configPrefix + ICEBERG_CATALOG_URI_KEY);
catalogProperties.put(CatalogProperties.URI, catalogUri);
- return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ return catalogProperties;
+ }
+
+ private static String getConfigPrefix(CatalogLocation location) {
+ return ICEBERG_DATASET_PREFIX + "." + location.toString().toLowerCase() + ".";
Review Comment:
should be a method on the `CatalogLocation` enum
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java:
##########
@@ -198,7 +198,8 @@ protected 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) {
- this.tableOps.commit(srcMetadata, dstMetadata);
+ // commit (baseMetadata -> destination metadata, updatedMetadata -> source metadata)
Review Comment:
overly cryptic. maybe say "use current destination metadata as 'base version' and source as update"?
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -118,30 +116,27 @@ protected IcebergDataset createIcebergDataset(String dbName, String tblName, Ice
return new IcebergDataset(dbName, tblName, srcIcebergTable, destIcebergTable, properties, fs);
}
- protected IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
- Map<String, String> catalogProperties = new HashMap<>();
+ protected static IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
+ String prefix = getConfigPrefix(location);
+ Map<String, String> catalogProperties = loadCatalogProperties(properties, prefix);
Configuration configuration = HadoopUtils.getConfFromProperties(properties);
- String catalogUri;
- String icebergCatalogClassName;
- switch (location) {
- case SOURCE:
- catalogUri = properties.getProperty(ICEBERG_SRC_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Source Catalog Table Service URI is required", ICEBERG_SRC_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_SRC_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_SRC_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- case DESTINATION:
- catalogUri = properties.getProperty(ICEBERG_DEST_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Destination Catalog Table Service URI is required", ICEBERG_DEST_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_DEST_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_DEST_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- default:
- throw new UnsupportedOperationException("Incorrect desired location: %s provided for creating Iceberg Catalog" + location);
+ String icebergCatalogClassName = catalogProperties.getOrDefault(ICEBERG_CATALOG_CLASS, DEFAULT_ICEBERG_CATALOG_CLASS);
+ return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ }
+
+ protected static Map<String, String> loadCatalogProperties(Properties properties, String configPrefix) {
+ Map<String, String> catalogProperties = new HashMap<>();
+ Config config = ConfigBuilder.create().loadProps(properties, configPrefix).build();
+ for (Map.Entry<String, ConfigValue> entry : config.entrySet()) {
+ catalogProperties.put(entry.getKey(), entry.getValue().unwrapped().toString());
}
+ String catalogUri = config.getString(ICEBERG_CATALOG_URI_KEY);
+ Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Catalog Table Service URI is required", configPrefix + ICEBERG_CATALOG_URI_KEY);
Review Comment:
shouldn't there be a `"."` between `configPrefix` and the URI KEY?
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -49,14 +52,9 @@
@RequiredArgsConstructor
public class IcebergDatasetFinder implements IterableDatasetFinder<IcebergDataset> {
public static final String ICEBERG_DATASET_PREFIX = DatasetConstants.PLATFORM_ICEBERG + ".dataset";
- public static final String ICEBERG_CLUSTER_KEY = "cluster";
- public static final String ICEBERG_SRC_CATALOG_CLASS_KEY = ICEBERG_DATASET_PREFIX + ".source.catalog.class";
public static final String DEFAULT_ICEBERG_CATALOG_CLASS = "org.apache.gobblin.data.management.copy.iceberg.IcebergHiveCatalog";
- public static final String ICEBERG_SRC_CATALOG_URI_KEY = ICEBERG_DATASET_PREFIX + ".source.catalog.uri";
- public static final String ICEBERG_SRC_CLUSTER_NAME = ICEBERG_DATASET_PREFIX + ".source.cluster.name";
- public static final String ICEBERG_DEST_CATALOG_CLASS_KEY = ICEBERG_DATASET_PREFIX + ".destination.catalog.class";
- public static final String ICEBERG_DEST_CATALOG_URI_KEY = ICEBERG_DATASET_PREFIX + ".copy.destination.catalog.uri";
- public static final String ICEBERG_DEST_CLUSTER_NAME = ICEBERG_DATASET_PREFIX + ".destination.cluster.name";
+ public static final String ICEBERG_CATALOG_CLASS = "catalog.class";
+ public static final String ICEBERG_CATALOG_URI_KEY = "catalog.uri";
Review Comment:
please document how these are used... i.e.:
```ICEBERG_CATALOG_PREFIX + "." + ("source" or "destination") + "..."```
be sure to explain how this is an open-ended pattern that allows for passing arbitrary catalog-specific props through to the catalog.
also, given that configs are essentially a global namespace, I'm not convinced that the prefix in use is discriminating enough.
whereas at present arbitrary properties go after:
```
iceberg.dataset.source
```
which means
```
iceberg.dataset.destination.database
```
would get passed through--and that seems wrong.
instead, it might be more appropriate to put them after
```
iceberg.dataset.source.catalog
```
(or `iceberg.dataset.catalog.source`)
seems most appropriate, given these are *catalog-specific* props
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDataset.java:
##########
@@ -316,8 +316,8 @@ protected DatasetDescriptor getDestinationDataset(FileSystem targetFs) {
return this.destIcebergTable.getDatasetDescriptor(targetFs);
}
- private PostPublishStep createPostPublishStep(IcebergTable srcIcebergTable, IcebergTable dstIcebergTable) {
- IcebergRegisterStep icebergRegisterStep = new IcebergRegisterStep(srcIcebergTable, dstIcebergTable);
+ private PostPublishStep createPostPublishStep(String dbName, String inputTableName, Properties properties) {
Review Comment:
seems inappropriate for the general case to serialize every single global property here. can't we constrain only to those actually related to creating the iceberg catalogs, source and dest?
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -118,30 +116,27 @@ protected IcebergDataset createIcebergDataset(String dbName, String tblName, Ice
return new IcebergDataset(dbName, tblName, srcIcebergTable, destIcebergTable, properties, fs);
}
- protected IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
- Map<String, String> catalogProperties = new HashMap<>();
+ protected static IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
+ String prefix = getConfigPrefix(location);
+ Map<String, String> catalogProperties = loadCatalogProperties(properties, prefix);
Configuration configuration = HadoopUtils.getConfFromProperties(properties);
- String catalogUri;
- String icebergCatalogClassName;
- switch (location) {
- case SOURCE:
- catalogUri = properties.getProperty(ICEBERG_SRC_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Source Catalog Table Service URI is required", ICEBERG_SRC_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_SRC_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_SRC_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- case DESTINATION:
- catalogUri = properties.getProperty(ICEBERG_DEST_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Destination Catalog Table Service URI is required", ICEBERG_DEST_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_DEST_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_DEST_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- default:
- throw new UnsupportedOperationException("Incorrect desired location: %s provided for creating Iceberg Catalog" + location);
+ String icebergCatalogClassName = catalogProperties.getOrDefault(ICEBERG_CATALOG_CLASS, DEFAULT_ICEBERG_CATALOG_CLASS);
+ return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ }
+
+ protected static Map<String, String> loadCatalogProperties(Properties properties, String configPrefix) {
Review Comment:
the semantics here seem non-obvious, hence deserving of javadoc. the name seems a bit inexact as well. maybe `buildMapFromPrefixChildren`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] meethngala commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "meethngala (via GitHub)" <gi...@apache.org>.
meethngala commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1164613567
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDatasetFinder.java:
##########
@@ -118,30 +116,27 @@ protected IcebergDataset createIcebergDataset(String dbName, String tblName, Ice
return new IcebergDataset(dbName, tblName, srcIcebergTable, destIcebergTable, properties, fs);
}
- protected IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
- Map<String, String> catalogProperties = new HashMap<>();
+ protected static IcebergCatalog createIcebergCatalog(Properties properties, CatalogLocation location) throws IOException {
+ String prefix = getConfigPrefix(location);
+ Map<String, String> catalogProperties = loadCatalogProperties(properties, prefix);
Configuration configuration = HadoopUtils.getConfFromProperties(properties);
- String catalogUri;
- String icebergCatalogClassName;
- switch (location) {
- case SOURCE:
- catalogUri = properties.getProperty(ICEBERG_SRC_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Source Catalog Table Service URI is required", ICEBERG_SRC_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_SRC_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_SRC_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- case DESTINATION:
- catalogUri = properties.getProperty(ICEBERG_DEST_CATALOG_URI_KEY);
- Preconditions.checkNotNull(catalogUri, "Provide: {%s} as Destination Catalog Table Service URI is required", ICEBERG_DEST_CATALOG_URI_KEY);
- // introducing an optional property for catalogs requiring cluster specific properties
- Optional.ofNullable(properties.getProperty(ICEBERG_DEST_CLUSTER_NAME)).ifPresent(value -> catalogProperties.put(ICEBERG_CLUSTER_KEY, value));
- icebergCatalogClassName = properties.getProperty(ICEBERG_DEST_CATALOG_CLASS_KEY, DEFAULT_ICEBERG_CATALOG_CLASS);
- break;
- default:
- throw new UnsupportedOperationException("Incorrect desired location: %s provided for creating Iceberg Catalog" + location);
+ String icebergCatalogClassName = catalogProperties.getOrDefault(ICEBERG_CATALOG_CLASS, DEFAULT_ICEBERG_CATALOG_CLASS);
+ return IcebergCatalogFactory.create(icebergCatalogClassName, catalogProperties, configuration);
+ }
+
+ protected static Map<String, String> loadCatalogProperties(Properties properties, String configPrefix) {
Review Comment:
gotcha... renamed the method and added the javadoc for it as well :)
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergDataset.java:
##########
@@ -316,8 +316,8 @@ protected DatasetDescriptor getDestinationDataset(FileSystem targetFs) {
return this.destIcebergTable.getDatasetDescriptor(targetFs);
}
- private PostPublishStep createPostPublishStep(IcebergTable srcIcebergTable, IcebergTable dstIcebergTable) {
- IcebergRegisterStep icebergRegisterStep = new IcebergRegisterStep(srcIcebergTable, dstIcebergTable);
+ private PostPublishStep createPostPublishStep(String dbName, String inputTableName, Properties properties) {
Review Comment:
I understand the concern here as we shouldn't need to serialize all the properties. I explored the option of doing it, but I am skeptical because we use those `properties` for providing configuration for Hadoop too which is utilized by `CatalogUtil.loadCatalog()`. And we pass the properties as-is to the configuration and it is set for an instance of Configurable: `(Configurable)catalog).setConf(hadoopConf)`. Now, it would be exhaustive to search how these hadoop configs are used by different catalog-impl. Does that make sense?
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java:
##########
@@ -198,7 +198,8 @@ protected 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) {
- this.tableOps.commit(srcMetadata, dstMetadata);
+ // commit (baseMetadata -> destination metadata, updatedMetadata -> source metadata)
Review Comment:
gotcha. I have simplified it in my latest commit!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] meethngala commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "meethngala (via GitHub)" <gi...@apache.org>.
meethngala commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1163342765
##########
gradle/scripts/dependencyDefinitions.gradle:
##########
@@ -123,7 +123,7 @@ ext.externalDependency = [
"guiceMultibindings": "com.google.inject.extensions:guice-multibindings:4.0",
"guiceServlet": "com.google.inject.extensions:guice-servlet:4.0",
"derby": "org.apache.derby:derby:10.12.1.1",
- "mockito": "org.mockito:mockito-inline:4.11.0", // upgraded to allow mocking for constructors, static and final methods; specifically for iceberg distcp
+ "mockito": "org.mockito:mockito-core:4.11.0",
Review Comment:
@homatthew yup makes sense... I will add a comment to avoid updating the dependency to mockito-inline
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] Will-Lo merged pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo merged PR #3673:
URL: https://github.com/apache/gobblin/pull/3673
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [gobblin] autumnust commented on a diff in pull request #3673: [GOBBLIN-1811]Fix Iceberg Registration Serialization
Posted by "autumnust (via GitHub)" <gi...@apache.org>.
autumnust commented on code in PR #3673:
URL: https://github.com/apache/gobblin/pull/3673#discussion_r1166102102
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/iceberg/IcebergTable.java:
##########
@@ -198,7 +198,8 @@ protected 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) {
Review Comment:
are we assuming that `dstMetadata` could never be null ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org