You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by "vaibhavk1992 (via GitHub)" <gi...@apache.org> on 2025/03/02 10:26:52 UTC
[PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
vaibhavk1992 opened a new pull request, #659:
URL: https://github.com/apache/incubator-xtable/pull/659
## *Important Read*
- *Please ensure the GitHub issue is mentioned at the beginning of the PR*
## What is the purpose of the pull request
* Refactor RunSync Class for Improved Modularity and Reusability *
## Brief change log
Refactored to main method into smaller methods.
Added some positive and negative testcases
## Verify this pull request
- *Added integration tests for end-to-end.*
- *Manually verified the change by running a job locally.*
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2011346585
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java:
##########
@@ -22,19 +22,62 @@
import static org.apache.xtable.model.storage.TableFormat.HUDI;
import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+import java.io.File;
import java.io.IOException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import org.apache.xtable.iceberg.IcebergCatalogConfig;
+import org.apache.xtable.conversion.CatalogConfig;
+import org.apache.xtable.utilities.RunSync.DatasetConfig;
import org.apache.xtable.utilities.RunSync.TableFormatConverters;
import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
class TestRunSync {
+ @Test
+ public void testMain() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
+ String filePath = file.getPath();
+ String[] args = new String[] {"--datasetConfig", filePath};
+ // RunSync.main(args);
+ Assertions.assertDoesNotThrow(
+ () -> RunSync.main(args), "RunSync.main() threw an unexpected exception.");
+ }
+
+ @Test
+ public void testGetDatasetConfigWithNonExistentFile() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config1.yaml");
+ String filePath = file.getPath();
+ Assertions.assertThrows(
+ IOException.class,
+ () -> {
+ RunSync.getDatasetConfig(filePath);
+ });
+ }
+
+ @Test
+ public void testGetDatasetConfigWithValidYAML() throws IOException {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
Review Comment:
This file doesn't exist.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1986364572
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java:
##########
@@ -22,19 +22,67 @@
import static org.apache.xtable.model.storage.TableFormat.HUDI;
import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.apache.xtable.iceberg.IcebergCatalogConfig;
+import org.apache.xtable.utilities.RunSync.DatasetConfig;
import org.apache.xtable.utilities.RunSync.TableFormatConverters;
import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
class TestRunSync {
+ @Test
+ public void testMain() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
+ String filePath = file.getPath();
+ String[] args = new String[] {"--datasetConfig", filePath};
+ try {
+ RunSync.main(args);
Review Comment:
There are no assertions in this test, please add some
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1977780628
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +98,110 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ List<TargetTable> targetTables =
Review Comment:
Added and tested as per our discussion.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +98,110 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ List<TargetTable> targetTables =
Review Comment:
Added and tested as per our discussion. @rangareddy
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1977780255
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +98,110 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
Review Comment:
Added and tested as per our discussion.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2002341587
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
@vinishjail97 I have done the required changes in the functions. Please review.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1997071917
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
do you mean changing the
` IcebergCatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
**to**
`CatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
and using icebergCatalogConfig of type `CatalogConfig` in `formatConvertor` method? @vinishjail97
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#issuecomment-2771500398
Closing this , Recreating PR with another feature branch
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2014085344
##########
.gitignore:
##########
@@ -41,4 +41,3 @@ target/
*.crc
demo/jars/*
demo/notebook/.ipynb_checkpoints/*
-my_config.yaml
Review Comment:
My test cases are using this file and without it the maven build is failing. how shall we deal with it then @the-other-tim-brown ?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 closed pull request #659: [528] Refactor RunSync Class for Improved Modularity and Reusability
URL: https://github.com/apache/incubator-xtable/pull/659
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#issuecomment-2692839829
Kindly review. There are some logger statements present that I will get rid of if there are no more changes required.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1977475644
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +98,110 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ List<TargetTable> targetTables =
Review Comment:
We need to add check for tableFormatList (null/empty) and table
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1998957569
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +116,116 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
- }
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
- if (cmd.hasOption(CONTINUOUS_MODE)) {
- ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
- long intervalInSeconds = Long.parseLong(cmd.getOptionValue(CONTINUOUS_MODE_INTERVAL, "5"));
- executorService.scheduleAtFixedRate(
- () -> {
- try {
- runSync(cmd);
- } catch (IOException ex) {
- log.error("Sync operation failed", ex);
- }
- },
- 0,
- intervalInSeconds,
- TimeUnit.SECONDS);
- while (!Thread.currentThread().isInterrupted()) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ex) {
- log.debug("Received interrupt signal");
- Thread.currentThread().interrupt();
- break;
- }
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
Review Comment:
This change has been made. @the-other-tim-brown
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1999223700
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
Review Comment:
@vinishjail97 I Made the changes.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1989971255
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,115 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ public static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
+
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
+ }
}
+ }
+ public static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
Review Comment:
yes, you always want to limit the scope to what is actually required or intentional. If not, other developers can start relying on your code in ways that you do not intend. Public facing methods should be maintained and functionality kept consistent across releases.
Package private is another option to consider.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2011347211
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java:
##########
@@ -22,19 +22,62 @@
import static org.apache.xtable.model.storage.TableFormat.HUDI;
import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+import java.io.File;
import java.io.IOException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import org.apache.xtable.iceberg.IcebergCatalogConfig;
+import org.apache.xtable.conversion.CatalogConfig;
+import org.apache.xtable.utilities.RunSync.DatasetConfig;
import org.apache.xtable.utilities.RunSync.TableFormatConverters;
import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
class TestRunSync {
+ @Test
+ public void testMain() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
+ String filePath = file.getPath();
+ String[] args = new String[] {"--datasetConfig", filePath};
+ // RunSync.main(args);
+ Assertions.assertDoesNotThrow(
+ () -> RunSync.main(args), "RunSync.main() threw an unexpected exception.");
+ }
+
+ @Test
+ public void testGetDatasetConfigWithNonExistentFile() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config1.yaml");
+ String filePath = file.getPath();
+ Assertions.assertThrows(
+ IOException.class,
+ () -> {
+ RunSync.getDatasetConfig(filePath);
+ });
+ }
+
+ @Test
+ public void testGetDatasetConfigWithValidYAML() throws IOException {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
Review Comment:
@vaibhavk1992 I think you missed adding the config.yaml file.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 closed pull request #659: [528] Refactor RunSync Class for Improved Modularity and Reusability
URL: https://github.com/apache/incubator-xtable/pull/659
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#issuecomment-2752221578
@vinishjail97 I have added the config.yaml file. Workflow is building fine now in my forked repo.
<img width="1149" alt="image" src="https://github.com/user-attachments/assets/d0b1c7e0-5666-4115-aea0-0231a1dce8db" />
Also squash and merged the commits.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1996429037
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
Review Comment:
You can use lombok `import lombok.NonNull;` for non null validations to simplify.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
Review Comment:
`syncTableMetdata` is a better name here as we are calling conversionController.sync ?
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
Review Comment:
Same comment as above.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
`org.apache.xtable.conversion.CatalogConfig`
Can you refactor this to ? It's better to use interfaces instead of the implementation class for common methods like these which are format agnostic.
icebergCatalogConfig -> catalogConfig
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#issuecomment-2730228694
I have made all the changes asked by the reviewers. Are we good to approve then?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1986364791
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -142,63 +221,58 @@ public static void main(String[] args) throws IOException {
ConversionSourceProvider<?> conversionSourceProvider =
ReflectionUtils.createInstanceOfClass(sourceProviderClass);
conversionSourceProvider.init(hadoopConf);
+ return conversionSourceProvider;
+ }
+ public static List<String> getTableFormatList(DatasetConfig datasetConfig) throws IOException {
+ // Retrieve table format list
List<String> tableFormatList = datasetConfig.getTargetFormats();
- ConversionController conversionController = new ConversionController(hadoopConf);
- for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
- log.info(
- "Running sync for basePath {} for following table formats {}",
- table.getTableBasePath(),
- tableFormatList);
- Properties sourceProperties = new Properties();
- if (table.getPartitionSpec() != null) {
- sourceProperties.put(
- HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
- }
- SourceTable sourceTable =
- SourceTable.builder()
- .name(table.getTableName())
- .basePath(table.getTableBasePath())
- .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
- .dataPath(table.getTableDataPath())
- .catalogConfig(icebergCatalogConfig)
- .additionalProperties(sourceProperties)
- .formatName(sourceFormat)
- .build();
- List<TargetTable> targetTables =
- tableFormatList.stream()
- .map(
- tableFormat ->
- TargetTable.builder()
- .name(table.getTableName())
- .basePath(table.getTableBasePath())
- .namespace(
- table.getNamespace() == null
- ? null
- : table.getNamespace().split("\\."))
- .catalogConfig(icebergCatalogConfig)
- .formatName(tableFormat)
- .build())
- .collect(Collectors.toList());
+ return tableFormatList;
+ }
- ConversionConfig conversionConfig =
- ConversionConfig.builder()
- .sourceTable(sourceTable)
- .targetTables(targetTables)
- .syncMode(SyncMode.INCREMENTAL)
- .build();
- try {
- conversionController.sync(conversionConfig, conversionSourceProvider);
- } catch (Exception e) {
- log.error("Error running sync for {}", table.getTableBasePath(), e);
- }
+ public static CommandLine CommandParser(String[] args) {
+ CommandLineParser parser = new DefaultParser();
+
+ CommandLine cmd;
+ try {
+ cmd = parser.parse(OPTIONS, args);
+ } catch (ParseException e) {
+ new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
+ return null;
+ }
+
+ if (cmd.hasOption(HELP_OPTION)) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("RunSync", OPTIONS);
+ return null;
}
+ return cmd;
+ }
+
+ public static String getValueFromConfig(CommandLine cmd, String configFlag) {
+ return cmd.getOptionValue(configFlag);
+ }
+
+ public static void main(String[] args) throws IOException {
+ CommandLine cmd = CommandParser(args);
+ String datasetConfigpath = getValueFromConfig(cmd, DATASET_CONFIG_OPTION);
+ String icebergCatalogConfigpath = getValueFromConfig(cmd, ICEBERG_CATALOG_CONFIG_PATH);
+ String hadoopConfigpath = getValueFromConfig(cmd, HADOOP_CONFIG_PATH);
+ String conversionProviderConfigpath = getValueFromConfig(cmd, CONVERTERS_CONFIG_PATH);
+ DatasetConfig datasetConfig = getDatasetConfig(datasetConfigpath);
+ IcebergCatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);
+ Configuration hadoopConf = gethadoopConf(hadoopConfigpath);
+ ConversionSourceProvider conversionSourceProvider =
+ getConversionSourceProvider(conversionProviderConfigpath, datasetConfig, hadoopConf);
+ List<String> tableFormatList = getTableFormatList(datasetConfig);
+ formatConvertor(
+ datasetConfig, tableFormatList, icebergCatalogConfig, hadoopConf, conversionSourceProvider);
}
- static byte[] getCustomConfigurations(CommandLine cmd, String option) throws IOException {
+ static byte[] getCustomConfigurations(String cmd) throws IOException {
Review Comment:
The variable name implies this is still the command line object, but I am guessing this is now a path?
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java:
##########
@@ -22,19 +22,67 @@
import static org.apache.xtable.model.storage.TableFormat.HUDI;
import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.apache.xtable.iceberg.IcebergCatalogConfig;
+import org.apache.xtable.utilities.RunSync.DatasetConfig;
import org.apache.xtable.utilities.RunSync.TableFormatConverters;
import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
class TestRunSync {
+ @Test
+ public void testMain() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
+ String filePath = file.getPath();
+ String[] args = new String[] {"--datasetConfig", filePath};
+ try {
+ RunSync.main(args);
+ } catch (IOException ex) {
+ throw new UncheckedIOException(ex);
+ }
+ }
+
+ @Test
+ public void testNonExistentFile() {
Review Comment:
```suggestion
public void testGetDatasetConfigWithNonExistentFile() {
```
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -142,63 +221,58 @@ public static void main(String[] args) throws IOException {
ConversionSourceProvider<?> conversionSourceProvider =
ReflectionUtils.createInstanceOfClass(sourceProviderClass);
conversionSourceProvider.init(hadoopConf);
+ return conversionSourceProvider;
+ }
+ public static List<String> getTableFormatList(DatasetConfig datasetConfig) throws IOException {
+ // Retrieve table format list
List<String> tableFormatList = datasetConfig.getTargetFormats();
- ConversionController conversionController = new ConversionController(hadoopConf);
- for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
- log.info(
- "Running sync for basePath {} for following table formats {}",
- table.getTableBasePath(),
- tableFormatList);
- Properties sourceProperties = new Properties();
- if (table.getPartitionSpec() != null) {
- sourceProperties.put(
- HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
- }
- SourceTable sourceTable =
- SourceTable.builder()
- .name(table.getTableName())
- .basePath(table.getTableBasePath())
- .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
- .dataPath(table.getTableDataPath())
- .catalogConfig(icebergCatalogConfig)
- .additionalProperties(sourceProperties)
- .formatName(sourceFormat)
- .build();
- List<TargetTable> targetTables =
- tableFormatList.stream()
- .map(
- tableFormat ->
- TargetTable.builder()
- .name(table.getTableName())
- .basePath(table.getTableBasePath())
- .namespace(
- table.getNamespace() == null
- ? null
- : table.getNamespace().split("\\."))
- .catalogConfig(icebergCatalogConfig)
- .formatName(tableFormat)
- .build())
- .collect(Collectors.toList());
+ return tableFormatList;
+ }
- ConversionConfig conversionConfig =
- ConversionConfig.builder()
- .sourceTable(sourceTable)
- .targetTables(targetTables)
- .syncMode(SyncMode.INCREMENTAL)
- .build();
- try {
- conversionController.sync(conversionConfig, conversionSourceProvider);
- } catch (Exception e) {
- log.error("Error running sync for {}", table.getTableBasePath(), e);
- }
+ public static CommandLine CommandParser(String[] args) {
Review Comment:
make sure to follow the naming convention of using a lowercase letter at the beginning of method names
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,115 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ public static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
+
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
+ }
}
+ }
+ public static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
Review Comment:
For these new methods, will they be used elsewhere? Can they be made private?
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -142,63 +221,58 @@ public static void main(String[] args) throws IOException {
ConversionSourceProvider<?> conversionSourceProvider =
ReflectionUtils.createInstanceOfClass(sourceProviderClass);
conversionSourceProvider.init(hadoopConf);
+ return conversionSourceProvider;
+ }
+ public static List<String> getTableFormatList(DatasetConfig datasetConfig) throws IOException {
+ // Retrieve table format list
List<String> tableFormatList = datasetConfig.getTargetFormats();
Review Comment:
Why do we need a method that is wrapping a getter?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1977474360
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +98,110 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
Review Comment:
Add the null check for both table and datasetConfig
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1989941203
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -142,63 +221,58 @@ public static void main(String[] args) throws IOException {
ConversionSourceProvider<?> conversionSourceProvider =
ReflectionUtils.createInstanceOfClass(sourceProviderClass);
conversionSourceProvider.init(hadoopConf);
+ return conversionSourceProvider;
+ }
+ public static List<String> getTableFormatList(DatasetConfig datasetConfig) throws IOException {
+ // Retrieve table format list
List<String> tableFormatList = datasetConfig.getTargetFormats();
- ConversionController conversionController = new ConversionController(hadoopConf);
- for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
- log.info(
- "Running sync for basePath {} for following table formats {}",
- table.getTableBasePath(),
- tableFormatList);
- Properties sourceProperties = new Properties();
- if (table.getPartitionSpec() != null) {
- sourceProperties.put(
- HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
- }
- SourceTable sourceTable =
- SourceTable.builder()
- .name(table.getTableName())
- .basePath(table.getTableBasePath())
- .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
- .dataPath(table.getTableDataPath())
- .catalogConfig(icebergCatalogConfig)
- .additionalProperties(sourceProperties)
- .formatName(sourceFormat)
- .build();
- List<TargetTable> targetTables =
- tableFormatList.stream()
- .map(
- tableFormat ->
- TargetTable.builder()
- .name(table.getTableName())
- .basePath(table.getTableBasePath())
- .namespace(
- table.getNamespace() == null
- ? null
- : table.getNamespace().split("\\."))
- .catalogConfig(icebergCatalogConfig)
- .formatName(tableFormat)
- .build())
- .collect(Collectors.toList());
+ return tableFormatList;
+ }
- ConversionConfig conversionConfig =
- ConversionConfig.builder()
- .sourceTable(sourceTable)
- .targetTables(targetTables)
- .syncMode(SyncMode.INCREMENTAL)
- .build();
- try {
- conversionController.sync(conversionConfig, conversionSourceProvider);
- } catch (Exception e) {
- log.error("Error running sync for {}", table.getTableBasePath(), e);
- }
+ public static CommandLine CommandParser(String[] args) {
+ CommandLineParser parser = new DefaultParser();
+
+ CommandLine cmd;
+ try {
+ cmd = parser.parse(OPTIONS, args);
+ } catch (ParseException e) {
+ new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
+ return null;
+ }
+
+ if (cmd.hasOption(HELP_OPTION)) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("RunSync", OPTIONS);
+ return null;
}
+ return cmd;
+ }
+
+ public static String getValueFromConfig(CommandLine cmd, String configFlag) {
+ return cmd.getOptionValue(configFlag);
+ }
+
+ public static void main(String[] args) throws IOException {
+ CommandLine cmd = CommandParser(args);
+ String datasetConfigpath = getValueFromConfig(cmd, DATASET_CONFIG_OPTION);
+ String icebergCatalogConfigpath = getValueFromConfig(cmd, ICEBERG_CATALOG_CONFIG_PATH);
+ String hadoopConfigpath = getValueFromConfig(cmd, HADOOP_CONFIG_PATH);
+ String conversionProviderConfigpath = getValueFromConfig(cmd, CONVERTERS_CONFIG_PATH);
+ DatasetConfig datasetConfig = getDatasetConfig(datasetConfigpath);
+ IcebergCatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);
+ Configuration hadoopConf = gethadoopConf(hadoopConfigpath);
+ ConversionSourceProvider conversionSourceProvider =
+ getConversionSourceProvider(conversionProviderConfigpath, datasetConfig, hadoopConf);
+ List<String> tableFormatList = getTableFormatList(datasetConfig);
+ formatConvertor(
+ datasetConfig, tableFormatList, icebergCatalogConfig, hadoopConf, conversionSourceProvider);
}
- static byte[] getCustomConfigurations(CommandLine cmd, String option) throws IOException {
+ static byte[] getCustomConfigurations(String cmd) throws IOException {
Review Comment:
Changes have been done.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -142,63 +221,58 @@ public static void main(String[] args) throws IOException {
ConversionSourceProvider<?> conversionSourceProvider =
ReflectionUtils.createInstanceOfClass(sourceProviderClass);
conversionSourceProvider.init(hadoopConf);
+ return conversionSourceProvider;
+ }
+ public static List<String> getTableFormatList(DatasetConfig datasetConfig) throws IOException {
+ // Retrieve table format list
List<String> tableFormatList = datasetConfig.getTargetFormats();
Review Comment:
Changes have been done.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2005747616
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -26,13 +26,11 @@
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
Review Comment:
@vinishjail97 Apologies, I have made the changes and added the code back.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1977488272
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +98,110 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ public static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
+
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
+ }
}
+ }
+ public static DatasetConfig getDatasetConfig(String d) throws IOException {
+ // Initialize DatasetConfig
DatasetConfig datasetConfig = new DatasetConfig();
- try (InputStream inputStream =
- Files.newInputStream(Paths.get(cmd.getOptionValue(DATASET_CONFIG_OPTION)))) {
+
+ try (InputStream inputStream = Files.newInputStream(Paths.get(d))) {
ObjectReader objectReader = YAML_MAPPER.readerForUpdating(datasetConfig);
objectReader.readValue(inputStream);
}
+ return datasetConfig;
+ }
- byte[] customConfig = getCustomConfigurations(cmd, HADOOP_CONFIG_PATH);
+ public static Configuration gethadoopConf(String cmd) throws IOException {
+ // Load configurations
+ byte[] customConfig = getCustomConfigurations(cmd);
Configuration hadoopConf = loadHadoopConf(customConfig);
- byte[] icebergCatalogConfigInput = getCustomConfigurations(cmd, ICEBERG_CATALOG_CONFIG_PATH);
+ return hadoopConf;
+ }
+
+ public static IcebergCatalogConfig getIcebergCatalogConfig(String cmd) throws IOException {
Review Comment:
Change the variable name from cmd to variablename for example icebergCatalogConfigPath
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1989941694
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java:
##########
@@ -22,19 +22,67 @@
import static org.apache.xtable.model.storage.TableFormat.HUDI;
import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.apache.xtable.iceberg.IcebergCatalogConfig;
+import org.apache.xtable.utilities.RunSync.DatasetConfig;
import org.apache.xtable.utilities.RunSync.TableFormatConverters;
import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
class TestRunSync {
+ @Test
+ public void testMain() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
+ String filePath = file.getPath();
+ String[] args = new String[] {"--datasetConfig", filePath};
+ try {
+ RunSync.main(args);
+ } catch (IOException ex) {
+ throw new UncheckedIOException(ex);
+ }
+ }
+
+ @Test
+ public void testNonExistentFile() {
Review Comment:
Changes have accepted and been done.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1997061999
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
Sorry, I couldn't quite get it what I have to change. @vinishjail97
Could you please explain more.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#issuecomment-2748443298
@vaibhavk1992 Can you squash and merge into a single 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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1999296080
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
IcebergCatalogConfig is one of the implementation for the interface, the arguments to these functions having the interface allows other implementation classes to be passed in.
```
static SourceTable sourceTableBuilder(
@NonNull DatasetConfig.Table table,
CatalogConfig catalogConfig,
@NonNull DatasetConfig datasetConfig,
Properties sourceProperties) {
```
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1997641226
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +116,116 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
- }
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
- if (cmd.hasOption(CONTINUOUS_MODE)) {
- ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
- long intervalInSeconds = Long.parseLong(cmd.getOptionValue(CONTINUOUS_MODE_INTERVAL, "5"));
- executorService.scheduleAtFixedRate(
- () -> {
- try {
- runSync(cmd);
- } catch (IOException ex) {
- log.error("Sync operation failed", ex);
- }
- },
- 0,
- intervalInSeconds,
- TimeUnit.SECONDS);
- while (!Thread.currentThread().isInterrupted()) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ex) {
- log.debug("Received interrupt signal");
- Thread.currentThread().interrupt();
- break;
- }
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
}
- executorService.shutdownNow();
- } else {
- runSync(cmd);
}
}
-
- private static void runSync(CommandLine cmd) throws IOException {
- DatasetConfig datasetConfig;
- try (InputStream inputStream =
- Files.newInputStream(Paths.get(cmd.getOptionValue(DATASET_CONFIG_OPTION)))) {
- datasetConfig = YAML_MAPPER.readValue(inputStream, DatasetConfig.class);
+
+ static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
+ // Initialize DatasetConfig
+ DatasetConfig datasetConfig = new DatasetConfig();
+
+ try (InputStream inputStream = Files.newInputStream(Paths.get(datasetConfigPath))) {
+ ObjectReader objectReader = YAML_MAPPER.readerForUpdating(datasetConfig);
+ objectReader.readValue(inputStream);
}
+ return datasetConfig;
Review Comment:
```suggestion
try (InputStream inputStream =
Files.newInputStream(Paths.get(cmd.getOptionValue(DATASET_CONFIG_OPTION)))) {
return YAML_MAPPER.readValue(inputStream, DatasetConfig.class);
}
```
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1999136032
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
Review Comment:
We use `lombok.NonNull` in other places in the project, so let's use that ?
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
Review Comment:
Replied.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1999298458
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +117,108 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ @NonNull DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ @NonNull List<String> tableFormatList) {
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
- }
+ static void syncTableMetdata(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
- if (cmd.hasOption(CONTINUOUS_MODE)) {
- ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
- long intervalInSeconds = Long.parseLong(cmd.getOptionValue(CONTINUOUS_MODE_INTERVAL, "5"));
- executorService.scheduleAtFixedRate(
- () -> {
- try {
- runSync(cmd);
- } catch (IOException ex) {
- log.error("Sync operation failed", ex);
- }
- },
- 0,
- intervalInSeconds,
- TimeUnit.SECONDS);
- while (!Thread.currentThread().isInterrupted()) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ex) {
- log.debug("Received interrupt signal");
- Thread.currentThread().interrupt();
- break;
- }
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error("Error running sync for {}", table.getTableBasePath(), e);
}
- executorService.shutdownNow();
- } else {
- runSync(cmd);
}
}
- private static void runSync(CommandLine cmd) throws IOException {
- DatasetConfig datasetConfig;
+ static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
+ // Initialize DatasetConfig
try (InputStream inputStream =
- Files.newInputStream(Paths.get(cmd.getOptionValue(DATASET_CONFIG_OPTION)))) {
- datasetConfig = YAML_MAPPER.readValue(inputStream, DatasetConfig.class);
+ Files.newInputStream(Paths.get(datasetConfigPath))) {
+ return YAML_MAPPER.readValue(inputStream, DatasetConfig.class);
}
-
- byte[] customConfig = getCustomConfigurations(cmd, HADOOP_CONFIG_PATH);
+}
+ static Configuration gethadoopConf(String hadoopConfigPath) throws IOException {
+ // Load configurations
+ byte[] customConfig = getCustomConfigurations(hadoopConfigPath);
Configuration hadoopConf = loadHadoopConf(customConfig);
- byte[] icebergCatalogConfigInput = getCustomConfigurations(cmd, ICEBERG_CATALOG_CONFIG_PATH);
+ return hadoopConf;
+ }
+
+ static IcebergCatalogConfig getIcebergCatalogConfig(String icebergCatalogConfigPath)
Review Comment:
Change function return type to `CatalogConfig`
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1999297598
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +117,108 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ @NonNull DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
`IcebergCatalogConfig icebergCatalogConfig` change this to `CatalogConfig catalogConfig`
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1989942132
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java:
##########
@@ -22,19 +22,67 @@
import static org.apache.xtable.model.storage.TableFormat.HUDI;
import static org.apache.xtable.model.storage.TableFormat.ICEBERG;
+import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.apache.xtable.iceberg.IcebergCatalogConfig;
+import org.apache.xtable.utilities.RunSync.DatasetConfig;
import org.apache.xtable.utilities.RunSync.TableFormatConverters;
import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
class TestRunSync {
+ @Test
+ public void testMain() {
+ String projectRoot = System.getProperty("user.dir");
+ // Construct the path to the file in the root directory
+ int lastSlashIndex = projectRoot.lastIndexOf('/');
+ String result = projectRoot.substring(0, lastSlashIndex);
+ File file = new File(result, "/my_config.yaml");
+ String filePath = file.getPath();
+ String[] args = new String[] {"--datasetConfig", filePath};
+ try {
+ RunSync.main(args);
Review Comment:
Changes have been done.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2019253991
##########
.gitignore:
##########
@@ -41,4 +41,3 @@ target/
*.crc
demo/jars/*
demo/notebook/.ipynb_checkpoints/*
-my_config.yaml
Review Comment:
The changes have been made and tested too. Workflow run in forked repo is also working fine. Please review and approve.
<img width="1071" alt="image" src="https://github.com/user-attachments/assets/f071c287-03a9-4a1e-908b-2898931bae44" />
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2014064016
##########
.gitignore:
##########
@@ -41,4 +41,3 @@ target/
*.crc
demo/jars/*
demo/notebook/.ipynb_checkpoints/*
-my_config.yaml
Review Comment:
Ignoring this file is intentional since people try to run the demo and create a `my_config.yaml` and try to check it in. Please leave this in the `.gitignore`
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2014086663
##########
my_config.yaml:
##########
@@ -0,0 +1,24 @@
+#
Review Comment:
you mean to move it like this https://github.com/apache/incubator-xtable/blob/main/xtable-utilities/src/test/java/org/apache/xtable/utilities/TestRunCatalogSync.java#L36?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1997071917
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
do you mean changing the
Line ` IcebergCatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
**to**
`CatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
and using icebergCatalogConfig of type `CatalogConfig` further in code? @vinishjail97
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1997071917
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
do you mean changing the
` IcebergCatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
**to**
`CatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
and using icebergCatalogConfig of type `CatalogConfig` further in code? @vinishjail97
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1997071917
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
do you mean changing the` IcebergCatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
`CatalogConfig icebergCatalogConfig = getIcebergCatalogConfig(icebergCatalogConfigpath);`
and using icebergCatalogConfig of type `CatalogConfig` further in code? @vinishjail97
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1999224224
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
Review Comment:
changes confirmed.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#issuecomment-2726011326
Added minor refactoring comments. LGTM otherwise. Can you rebase with latest main branch as well ? There are few conflicts.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2014062790
##########
my_config.yaml:
##########
@@ -0,0 +1,24 @@
+#
Review Comment:
Should this file be a test resource?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "rangareddy (via GitHub)" <gi...@apache.org>.
rangareddy commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1977484196
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +98,110 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ public static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
+
+ ConversionController conversionController = new ConversionController(hadoopConf);
Review Comment:
Move this code outside of for loop `ConversionController conversionController = new ConversionController(hadoopConf);`
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1990534872
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,115 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ public static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
+
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
+ }
}
+ }
+ public static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
Review Comment:
Ok, I got you now. I have moved them to private package like others. @the-other-tim-brown
Kindly confirm if things looks good now.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1989940250
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,115 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ public static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ public static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ public static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
+
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
+ }
}
+ }
+ public static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
Review Comment:
@the-other-tim-brown
We are not using it anywhere but just in runsync
If I even convert to private I get a lot of compilation errors. Any reason to make them private? Is it necessary to make this change?"
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1996963988
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
+ static void formatConvertor(
Review Comment:
ok, will change the name to `syncTableMetdata` then
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1996952301
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
Review Comment:
Actually @rangareddy suggested to use Objects.requireNonNull. So Please let me know if need to change. @vinishjail97
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -98,37 +99,116 @@ public class RunSync {
+ "used for any Iceberg source or target.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
Review Comment:
Same comment as above. @vinishjail97
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1997640717
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +116,116 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
- }
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
- if (cmd.hasOption(CONTINUOUS_MODE)) {
- ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
- long intervalInSeconds = Long.parseLong(cmd.getOptionValue(CONTINUOUS_MODE_INTERVAL, "5"));
- executorService.scheduleAtFixedRate(
- () -> {
- try {
- runSync(cmd);
- } catch (IOException ex) {
- log.error("Sync operation failed", ex);
- }
- },
- 0,
- intervalInSeconds,
- TimeUnit.SECONDS);
- while (!Thread.currentThread().isInterrupted()) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ex) {
- log.debug("Received interrupt signal");
- Thread.currentThread().interrupt();
- break;
- }
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
Review Comment:
Use `{}` instead of `String.format` in the logs
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r1998956130
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +116,116 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(datasetConfig, "datasetConfig cannot be null");
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ List<String> tableFormatList) {
+ Objects.requireNonNull(table, "Table cannot be null");
+ Objects.requireNonNull(tableFormatList, "tableFormatList cannot be null");
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
- }
+ static void formatConvertor(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
- if (cmd.hasOption(CONTINUOUS_MODE)) {
- ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
- long intervalInSeconds = Long.parseLong(cmd.getOptionValue(CONTINUOUS_MODE_INTERVAL, "5"));
- executorService.scheduleAtFixedRate(
- () -> {
- try {
- runSync(cmd);
- } catch (IOException ex) {
- log.error("Sync operation failed", ex);
- }
- },
- 0,
- intervalInSeconds,
- TimeUnit.SECONDS);
- while (!Thread.currentThread().isInterrupted()) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ex) {
- log.debug("Received interrupt signal");
- Thread.currentThread().interrupt();
- break;
- }
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
}
- executorService.shutdownNow();
- } else {
- runSync(cmd);
}
}
-
- private static void runSync(CommandLine cmd) throws IOException {
- DatasetConfig datasetConfig;
- try (InputStream inputStream =
- Files.newInputStream(Paths.get(cmd.getOptionValue(DATASET_CONFIG_OPTION)))) {
- datasetConfig = YAML_MAPPER.readValue(inputStream, DatasetConfig.class);
+
+ static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
+ // Initialize DatasetConfig
+ DatasetConfig datasetConfig = new DatasetConfig();
+
+ try (InputStream inputStream = Files.newInputStream(Paths.get(datasetConfigPath))) {
+ ObjectReader objectReader = YAML_MAPPER.readerForUpdating(datasetConfig);
+ objectReader.readValue(inputStream);
}
+ return datasetConfig;
Review Comment:
@the-other-tim-brown this change was requested by @rangareddy we are not passing cmd everytime to these methods. Instead we are creating config paths just once using cmd and using them later on. So I have accepted only this line for change `return YAML_MAPPER.readValue(inputStream, DatasetConfig.class);`
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vaibhavk1992 (via GitHub)" <gi...@apache.org>.
vaibhavk1992 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2002341468
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +117,108 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ @NonNull DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ @NonNull List<String> tableFormatList) {
+ List<TargetTable> targetTables =
+ tableFormatList.stream()
+ .map(
+ tableFormat ->
+ TargetTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(
+ table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .catalogConfig(icebergCatalogConfig)
+ .formatName(tableFormat)
+ .build())
+ .collect(Collectors.toList());
+ return targetTables;
+ }
- if (cmd.hasOption(HELP_OPTION)) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("RunSync", OPTIONS);
- return;
- }
+ static void syncTableMetdata(
+ DatasetConfig datasetConfig,
+ List<String> tableFormatList,
+ IcebergCatalogConfig icebergCatalogConfig,
+ Configuration hadoopConf,
+ ConversionSourceProvider conversionSourceProvider) {
+ ConversionController conversionController = new ConversionController(hadoopConf);
+ for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
+ log.info(
+ "Running sync for basePath {} for following table formats {}",
+ table.getTableBasePath(),
+ tableFormatList);
+ Properties sourceProperties = new Properties();
+ if (table.getPartitionSpec() != null) {
+ sourceProperties.put(
+ HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec());
+ }
- if (cmd.hasOption(CONTINUOUS_MODE)) {
- ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
- long intervalInSeconds = Long.parseLong(cmd.getOptionValue(CONTINUOUS_MODE_INTERVAL, "5"));
- executorService.scheduleAtFixedRate(
- () -> {
- try {
- runSync(cmd);
- } catch (IOException ex) {
- log.error("Sync operation failed", ex);
- }
- },
- 0,
- intervalInSeconds,
- TimeUnit.SECONDS);
- while (!Thread.currentThread().isInterrupted()) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException ex) {
- log.debug("Received interrupt signal");
- Thread.currentThread().interrupt();
- break;
- }
+ SourceTable sourceTable =
+ sourceTableBuilder(table, icebergCatalogConfig, datasetConfig, sourceProperties);
+ List<TargetTable> targetTables =
+ targetTableBuilder(table, icebergCatalogConfig, tableFormatList);
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+ try {
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ } catch (Exception e) {
+ log.error("Error running sync for {}", table.getTableBasePath(), e);
}
- executorService.shutdownNow();
- } else {
- runSync(cmd);
}
}
- private static void runSync(CommandLine cmd) throws IOException {
- DatasetConfig datasetConfig;
+ static DatasetConfig getDatasetConfig(String datasetConfigPath) throws IOException {
+ // Initialize DatasetConfig
try (InputStream inputStream =
- Files.newInputStream(Paths.get(cmd.getOptionValue(DATASET_CONFIG_OPTION)))) {
- datasetConfig = YAML_MAPPER.readValue(inputStream, DatasetConfig.class);
+ Files.newInputStream(Paths.get(datasetConfigPath))) {
+ return YAML_MAPPER.readValue(inputStream, DatasetConfig.class);
}
-
- byte[] customConfig = getCustomConfigurations(cmd, HADOOP_CONFIG_PATH);
+}
+ static Configuration gethadoopConf(String hadoopConfigPath) throws IOException {
+ // Load configurations
+ byte[] customConfig = getCustomConfigurations(hadoopConfigPath);
Configuration hadoopConf = loadHadoopConf(customConfig);
- byte[] icebergCatalogConfigInput = getCustomConfigurations(cmd, ICEBERG_CATALOG_CONFIG_PATH);
+ return hadoopConf;
+ }
+
+ static IcebergCatalogConfig getIcebergCatalogConfig(String icebergCatalogConfigPath)
Review Comment:
@vinishjail97 I have done the required changes in the functions. Please review.
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -115,66 +117,108 @@ public class RunSync {
"The interval in seconds to schedule the loop. Requires --continuousMode to be set. Defaults to 5 seconds.")
.addOption(HELP_OPTION, "help", false, "Displays help information to run this utility");
- public static void main(String[] args) throws IOException {
- CommandLineParser parser = new DefaultParser();
+ static SourceTable sourceTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
+ @NonNull DatasetConfig datasetConfig,
+ Properties sourceProperties) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(table.getTableName())
+ .basePath(table.getTableBasePath())
+ .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
+ .dataPath(table.getTableDataPath())
+ .catalogConfig(icebergCatalogConfig)
+ .additionalProperties(sourceProperties)
+ .formatName(datasetConfig.sourceFormat)
+ .build();
+ return sourceTable;
+ }
- CommandLine cmd;
- try {
- cmd = parser.parse(OPTIONS, args);
- } catch (ParseException e) {
- new HelpFormatter().printHelp("xtable.jar", OPTIONS, true);
- return;
- }
+ static List<TargetTable> targetTableBuilder(
+ @NonNull DatasetConfig.Table table,
+ IcebergCatalogConfig icebergCatalogConfig,
Review Comment:
@vinishjail97 I have done the required changes in the functions. Please review.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2004129501
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -26,13 +26,11 @@
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
Review Comment:
The continuous mode code has been removed in your change, can you add it back ?
https://github.com/apache/incubator-xtable/pull/593
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [528] Refactor RunSync Class for Improved Modularity and Reusability [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #659:
URL: https://github.com/apache/incubator-xtable/pull/659#discussion_r2014489862
##########
.gitignore:
##########
@@ -41,4 +41,3 @@ target/
*.crc
demo/jars/*
demo/notebook/.ipynb_checkpoints/*
-my_config.yaml
Review Comment:
You can move the file to the resources folder here: https://github.com/apache/incubator-xtable/tree/main/xtable-utilities/src/test/resources
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org