You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by ti...@apache.org on 2024/04/01 04:56:10 UTC
(incubator-xtable) branch 406-client-rename created (now 5df53d8c)
This is an automated email from the ASF dual-hosted git repository.
timbrown pushed a change to branch 406-client-rename
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
at 5df53d8c rename and replace 'client'
This branch includes the following new commits:
new 5df53d8c rename and replace 'client'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(incubator-xtable) 01/01: rename and replace 'client'
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
timbrown pushed a commit to branch 406-client-rename
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
commit 5df53d8ca14e42fd73a252b023239b64e6a086aa
Author: Timothy Brown <ti...@onehouse.ai>
AuthorDate: Sun Mar 31 23:56:01 2024 -0500
rename and replace 'client'
---
README.md | 26 +--
.../{client => conversion}/CatalogConfig.java | 2 +-
.../{client => conversion}/HudiSourceConfig.java | 2 +-
.../{client => conversion}/PerTableConfig.java | 2 +-
.../{SourceClient.java => ConversionSource.java} | 2 +-
.../xtable/spi/extractor/ExtractFromSource.java | 8 +-
.../{TargetClient.java => ConversionTarget.java} | 4 +-
.../apache/xtable/spi/sync/TableFormatSync.java | 32 ++--
.../spi/extractor/TestExtractFromSource.java | 14 +-
.../xtable/spi/sync/TestTableFormatSync.java | 114 ++++++------
.../ConversionController.java} | 76 ++++----
.../ConversionSourceProvider.java} | 33 ++--
.../ConversionTargetFactory.java} | 37 ++--
.../{client => conversion}/PerTableConfigImpl.java | 2 +-
...ourceClient.java => DeltaConversionSource.java} | 6 +-
...der.java => DeltaConversionSourceProvider.java} | 28 ++-
...DeltaClient.java => DeltaConversionTarget.java} | 14 +-
...aClientUtils.java => DeltaConversionUtils.java} | 6 +-
.../xtable/delta/DeltaIncrementalChangesState.java | 4 +-
.../{HudiClient.java => HudiConversionSource.java} | 6 +-
...ider.java => HudiConversionSourceProvider.java} | 12 +-
...TargetClient.java => HudiConversionTarget.java} | 12 +-
.../apache/xtable/hudi/HudiSourceConfigImpl.java | 2 +-
.../xtable/iceberg/IcebergCatalogConfig.java | 2 +-
...rceClient.java => IcebergConversionSource.java} | 6 +-
...r.java => IcebergConversionSourceProvider.java} | 12 +-
...ergClient.java => IcebergConversionTarget.java} | 12 +-
...=> org.apache.xtable.spi.sync.ConversionTarget} | 6 +-
...ableClient.java => ITConversionController.java} | 153 ++++++++-------
.../TestConversionController.java} | 205 +++++++++++----------
.../TestConversionTargetFactory.java} | 34 ++--
.../{client => conversion}/TestPerTableConfig.java | 2 +-
...ent.java => ITDeltaConversionTargetSource.java} | 93 +++++-----
.../org/apache/xtable/delta/TestDeltaSync.java | 23 ++-
...ient.java => ITHudiConversionSourceSource.java} | 22 +--
...ient.java => ITHudiConversionSourceTarget.java} | 26 +--
...nt.java => TestHudiConversionSourceTarget.java} | 54 +++---
...t.java => ITIcebergConversionTargetSource.java} | 58 +++---
...java => TestIcebergConversionTargetSource.java} | 47 ++---
.../org/apache/xtable/iceberg/TestIcebergSync.java | 48 +++--
.../java/org/apache/xtable/loadtest/LoadTest.java | 26 +--
demo/notebook/demo.ipynb | 32 ++--
hudi-support/extensions/README.md | 2 +-
...eTableSyncConfig.java => XTableSyncConfig.java} | 4 +-
.../{OneTableSyncTool.java => XTableSyncTool.java} | 28 +--
...eTableSyncTool.java => TestXTableSyncTool.java} | 8 +-
.../java/org/apache/xtable/utilities/RunSync.java | 80 ++++----
...ults.yaml => onetable-conversion-defaults.yaml} | 30 +--
.../org/apache/xtable/utilities/TestRunSync.java | 62 ++++---
49 files changed, 793 insertions(+), 726 deletions(-)
diff --git a/README.md b/README.md
index 0f78f750..08bb33a5 100644
--- a/README.md
+++ b/README.md
@@ -66,23 +66,23 @@ datasets:
- `DAY`: same as `YEAR` but with day granularity
- `HOUR`: same as `YEAR` but with hour granularity
- `format`: if your partition type is `YEAR`, `MONTH`, `DAY`, or `HOUR` specify the format for the date string as it appears in your file paths
-3. The default implementations of table format clients can be replaced with custom implementations by specifying a client configs yaml file in the format below:
+3. The default implementations of table format converters can be replaced with custom implementations by specifying a converter configs yaml file in the format below:
```yaml
-# sourceClientProviderClass: The class name of a table format's client factory, where the client is
+# conversionSourceProviderClass: The class name of a table format's converter factory, where the converter is
# used for reading from a table of this format. All user configurations, including hadoop config
-# and client specific configuration, will be available to the factory for instantiation of the
-# client.
-# targetClientProviderClass: The class name of a table format's client factory, where the client is
+# and converter specific configuration, will be available to the factory for instantiation of the
+# converter.
+# conversionTargetProviderClass: The class name of a table format's converter factory, where the converter is
# used for writing to a table of this format.
-# configuration: A map of configuration values specific to this client.
-tableFormatsClients:
+# configuration: A map of configuration values specific to this converter.
+tableFormatConverters:
HUDI:
- sourceClientProviderClass: org.apache.xtable.hudi.HudiSourceClientProvider
+ conversionSourceProviderClass: org.apache.xtable.hudi.HudiConversionSourceProvider
DELTA:
- targetClientProviderClass: org.apache.xtable.delta.DeltaClient
+ conversionTargetProviderClass: org.apache.xtable.delta.DeltaConversionTarget
configuration:
spark.master: local[2]
- spark.app.name: onetableclient
+ spark.app.name: xtable
```
4. A catalog can be used when reading and updating Iceberg tables. The catalog can be specified in a yaml file and passed in with the `--icebergCatalogConfig` option. The format of the catalog config file is:
```yaml
@@ -92,8 +92,8 @@ catalogOptions: # all other options are passed through in a map
key1: value1
key2: value2
```
-5. run with `java -jar utilities/target/utilities-0.1.0-SNAPSHOT-bundled.jar --datasetConfig my_config.yaml [--hadoopConfig hdfs-site.xml] [--clientsConfig clients.yaml] [--icebergCatalogConfig catalog.yaml]`
-The bundled jar includes hadoop dependencies for AWS, Azure, and GCP. Sample hadoop configurations for configuring the clients
+5. run with `java -jar utilities/target/utilities-0.1.0-SNAPSHOT-bundled.jar --datasetConfig my_config.yaml [--hadoopConfig hdfs-site.xml] [--convertersConfig converters.yaml] [--icebergCatalogConfig catalog.yaml]`
+The bundled jar includes hadoop dependencies for AWS, Azure, and GCP. Sample hadoop configurations for configuring the converters
can be found in the [onetable-hadoop-defaults.xml](https://github.com/apache/incubator-xtable/blob/main/utilities/src/main/resources/onetable-hadoop-defaults.xml) file.
The custom hadoop configurations can be passed in with the `--hadoopConfig [custom-hadoop-config-file]` option.
The config in custom hadoop config file will override the default hadoop configurations. For an example
@@ -107,7 +107,7 @@ For setting up the repo on IntelliJ, open the project and change the java versio
You have found a bug, or have a cool idea you that want to contribute to the project ? Please file a GitHub issue [here](https://github.com/apache/incubator-xtable/issues)
## Adding a new target format
-Adding a new target format requires a developer implement [TargetClient](./api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java). Once you have implemented that interface, you can integrate it into the [OneTableClient](./core/src/main/java/org/apache/xtable/client/OneTableClient.java). If you think others may find that target useful, please raise a Pull Request to add it to the project.
+Adding a new target format requires a developer implement [ConversionTarget](./api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java). Once you have implemented that interface, you can integrate it into the [OneTableClient](./core/src/main/java/org/apache/xtable/client/OneTableClient.java). If you think others may find that target useful, please raise a Pull Request to add it to the project.
## Overview of the sync process
diff --git a/api/src/main/java/org/apache/xtable/client/CatalogConfig.java b/api/src/main/java/org/apache/xtable/conversion/CatalogConfig.java
similarity index 97%
rename from api/src/main/java/org/apache/xtable/client/CatalogConfig.java
rename to api/src/main/java/org/apache/xtable/conversion/CatalogConfig.java
index 4184b4c8..a432b749 100644
--- a/api/src/main/java/org/apache/xtable/client/CatalogConfig.java
+++ b/api/src/main/java/org/apache/xtable/conversion/CatalogConfig.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import java.util.Map;
diff --git a/api/src/main/java/org/apache/xtable/client/HudiSourceConfig.java b/api/src/main/java/org/apache/xtable/conversion/HudiSourceConfig.java
similarity index 96%
rename from api/src/main/java/org/apache/xtable/client/HudiSourceConfig.java
rename to api/src/main/java/org/apache/xtable/conversion/HudiSourceConfig.java
index 113fcb75..1d59466a 100644
--- a/api/src/main/java/org/apache/xtable/client/HudiSourceConfig.java
+++ b/api/src/main/java/org/apache/xtable/conversion/HudiSourceConfig.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import org.apache.xtable.spi.extractor.SourcePartitionSpecExtractor;
diff --git a/api/src/main/java/org/apache/xtable/client/PerTableConfig.java b/api/src/main/java/org/apache/xtable/conversion/PerTableConfig.java
similarity index 97%
rename from api/src/main/java/org/apache/xtable/client/PerTableConfig.java
rename to api/src/main/java/org/apache/xtable/conversion/PerTableConfig.java
index f9179439..612fc9b3 100644
--- a/api/src/main/java/org/apache/xtable/client/PerTableConfig.java
+++ b/api/src/main/java/org/apache/xtable/conversion/PerTableConfig.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import java.util.List;
diff --git a/api/src/main/java/org/apache/xtable/spi/extractor/SourceClient.java b/api/src/main/java/org/apache/xtable/spi/extractor/ConversionSource.java
similarity index 98%
rename from api/src/main/java/org/apache/xtable/spi/extractor/SourceClient.java
rename to api/src/main/java/org/apache/xtable/spi/extractor/ConversionSource.java
index 40bd0f0d..ed1486b9 100644
--- a/api/src/main/java/org/apache/xtable/spi/extractor/SourceClient.java
+++ b/api/src/main/java/org/apache/xtable/spi/extractor/ConversionSource.java
@@ -33,7 +33,7 @@ import org.apache.xtable.model.schema.SchemaCatalog;
* source system. The client uses {@link Instant} to represent the point in time a commit was made
* to be as generic as possible across source table formats.
*/
-public interface SourceClient<COMMIT> extends Closeable {
+public interface ConversionSource<COMMIT> extends Closeable {
/**
* Extracts the {@link InternalTable} definition as of the provided commit.
*
diff --git a/api/src/main/java/org/apache/xtable/spi/extractor/ExtractFromSource.java b/api/src/main/java/org/apache/xtable/spi/extractor/ExtractFromSource.java
index f09462b9..e1604435 100644
--- a/api/src/main/java/org/apache/xtable/spi/extractor/ExtractFromSource.java
+++ b/api/src/main/java/org/apache/xtable/spi/extractor/ExtractFromSource.java
@@ -32,20 +32,20 @@ import org.apache.xtable.model.TableChange;
@AllArgsConstructor(staticName = "of")
@Getter
public class ExtractFromSource<COMMIT> {
- private final SourceClient<COMMIT> sourceClient;
+ private final ConversionSource<COMMIT> conversionSource;
public InternalSnapshot extractSnapshot() {
- return sourceClient.getCurrentSnapshot();
+ return conversionSource.getCurrentSnapshot();
}
public IncrementalTableChanges extractTableChanges(
InstantsForIncrementalSync instantsForIncrementalSync) {
CommitsBacklog<COMMIT> commitsBacklog =
- sourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
// No overlap between updatedPendingCommits and commitList, process separately.
Iterator<TableChange> tableChangeIterator =
commitsBacklog.getCommitsToProcess().stream()
- .map(sourceClient::getTableChangeForCommit)
+ .map(conversionSource::getTableChangeForCommit)
.iterator();
return IncrementalTableChanges.builder()
.tableChanges(tableChangeIterator)
diff --git a/api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java b/api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
similarity index 97%
rename from api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java
rename to api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
index d8132182..37eb25f1 100644
--- a/api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java
+++ b/api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
@@ -23,7 +23,7 @@ import java.util.Optional;
import org.apache.hadoop.conf.Configuration;
-import org.apache.xtable.client.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfig;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.TableSyncMetadata;
import org.apache.xtable.model.schema.InternalPartitionField;
@@ -32,7 +32,7 @@ import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.PartitionFileGroup;
/** A client that provides the major functionality for syncing changes to a target system. */
-public interface TargetClient {
+public interface ConversionTarget {
/**
* Syncs the current schema state.
diff --git a/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java b/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java
index 7285ea28..cd4de0ff 100644
--- a/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java
+++ b/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java
@@ -54,21 +54,21 @@ public class TableFormatSync {
/**
* Syncs the provided snapshot to the target table formats.
*
- * @param targetClients the targets to sync with the snapshot
+ * @param conversionTargets the targets to sync with the snapshot
* @param snapshot the snapshot to sync
* @return the result of the sync process
*/
public Map<String, SyncResult> syncSnapshot(
- Collection<TargetClient> targetClients, InternalSnapshot snapshot) {
+ Collection<ConversionTarget> conversionTargets, InternalSnapshot snapshot) {
Instant startTime = Instant.now();
Map<String, SyncResult> results = new HashMap<>();
- for (TargetClient targetClient : targetClients) {
+ for (ConversionTarget conversionTarget : conversionTargets) {
try {
InternalTable internalTable = snapshot.getTable();
results.put(
- targetClient.getTableFormat(),
+ conversionTarget.getTableFormat(),
getSyncResult(
- targetClient,
+ conversionTarget,
SyncMode.FULL,
internalTable,
client -> client.syncFilesForSnapshot(snapshot.getPartitionedDataFiles()),
@@ -77,7 +77,7 @@ public class TableFormatSync {
} catch (Exception e) {
log.error("Failed to sync snapshot", e);
results.put(
- targetClient.getTableFormat(), buildResultForError(SyncMode.FULL, startTime, e));
+ conversionTarget.getTableFormat(), buildResultForError(SyncMode.FULL, startTime, e));
}
}
return results;
@@ -91,13 +91,13 @@ public class TableFormatSync {
* @return the results of trying to sync each change
*/
public Map<String, List<SyncResult>> syncChanges(
- Map<TargetClient, TableSyncMetadata> targetClientWithMetadata,
+ Map<ConversionTarget, TableSyncMetadata> targetClientWithMetadata,
IncrementalTableChanges changes) {
Map<String, List<SyncResult>> results = new HashMap<>();
- Set<TargetClient> clientsWithFailures = new HashSet<>();
+ Set<ConversionTarget> clientsWithFailures = new HashSet<>();
while (changes.getTableChanges().hasNext()) {
TableChange change = changes.getTableChanges().next();
- Collection<TargetClient> clientsToSync =
+ Collection<ConversionTarget> clientsToSync =
targetClientWithMetadata.entrySet().stream()
.filter(
entry -> {
@@ -106,17 +106,17 @@ public class TableFormatSync {
})
.map(Map.Entry::getKey)
.collect(Collectors.toList());
- for (TargetClient targetClient : clientsToSync) {
- if (clientsWithFailures.contains(targetClient)) {
+ for (ConversionTarget conversionTarget : clientsToSync) {
+ if (clientsWithFailures.contains(conversionTarget)) {
continue;
}
Instant startTime = Instant.now();
List<SyncResult> resultsForFormat =
- results.computeIfAbsent(targetClient.getTableFormat(), key -> new ArrayList<>());
+ results.computeIfAbsent(conversionTarget.getTableFormat(), key -> new ArrayList<>());
try {
resultsForFormat.add(
getSyncResult(
- targetClient,
+ conversionTarget,
SyncMode.INCREMENTAL,
change.getTableAsOfChange(),
client -> client.syncFilesForDiff(change.getFilesDiff()),
@@ -125,7 +125,7 @@ public class TableFormatSync {
} catch (Exception e) {
log.error("Failed to sync table changes", e);
resultsForFormat.add(buildResultForError(SyncMode.INCREMENTAL, startTime, e));
- clientsWithFailures.add(targetClient);
+ clientsWithFailures.add(conversionTarget);
}
}
}
@@ -144,7 +144,7 @@ public class TableFormatSync {
}
private SyncResult getSyncResult(
- TargetClient client,
+ ConversionTarget client,
SyncMode mode,
InternalTable tableState,
SyncFiles fileSyncMethod,
@@ -175,7 +175,7 @@ public class TableFormatSync {
@FunctionalInterface
private interface SyncFiles {
- void sync(TargetClient client);
+ void sync(ConversionTarget client);
}
private SyncResult buildResultForError(SyncMode mode, Instant startTime, Exception e) {
diff --git a/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java b/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java
index de77fad7..82b61609 100644
--- a/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java
+++ b/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java
@@ -47,7 +47,7 @@ import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.PartitionFileGroup;
public class TestExtractFromSource {
- private final SourceClient<TestCommit> mockSourceClient = mock(SourceClient.class);
+ private final ConversionSource<TestCommit> mockConversionSource = mock(ConversionSource.class);
@Test
public void extractSnapshot() {
@@ -60,8 +60,8 @@ public class TestExtractFromSource {
.table(table)
.partitionedDataFiles(dataFiles)
.build();
- when(mockSourceClient.getCurrentSnapshot()).thenReturn(internalSnapshot);
- assertEquals(internalSnapshot, ExtractFromSource.of(mockSourceClient).extractSnapshot());
+ when(mockConversionSource.getCurrentSnapshot()).thenReturn(internalSnapshot);
+ assertEquals(internalSnapshot, ExtractFromSource.of(mockConversionSource).extractSnapshot());
}
@Test
@@ -80,7 +80,7 @@ public class TestExtractFromSource {
.commitsToProcess(Arrays.asList(firstCommitToSync, secondCommitToSync))
.inFlightInstants(Collections.singletonList(inflightInstant))
.build();
- when(mockSourceClient.getCommitsBacklog(instantsForIncrementalSync))
+ when(mockConversionSource.getCommitsBacklog(instantsForIncrementalSync))
.thenReturn(commitsBacklogToReturn);
// drop a file and add a file
@@ -93,7 +93,7 @@ public class TestExtractFromSource {
.filesDiff(
DataFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
.build();
- when(mockSourceClient.getTableChangeForCommit(firstCommitToSync))
+ when(mockConversionSource.getTableChangeForCommit(firstCommitToSync))
.thenReturn(tableChangeToReturnAtFirstInstant);
TableChange expectedFirstTableChange =
TableChange.builder()
@@ -117,7 +117,7 @@ public class TestExtractFromSource {
.filesRemoved(Arrays.asList(initialFile3, newFile1))
.build())
.build();
- when(mockSourceClient.getTableChangeForCommit(secondCommitToSync))
+ when(mockConversionSource.getTableChangeForCommit(secondCommitToSync))
.thenReturn(tableChangeToReturnAtSecondInstant);
TableChange expectedSecondTableChange =
TableChange.builder()
@@ -130,7 +130,7 @@ public class TestExtractFromSource {
.build();
IncrementalTableChanges actual =
- ExtractFromSource.of(mockSourceClient).extractTableChanges(instantsForIncrementalSync);
+ ExtractFromSource.of(mockConversionSource).extractTableChanges(instantsForIncrementalSync);
assertEquals(Collections.singletonList(inflightInstant), actual.getPendingCommits());
List<TableChange> actualTableChanges = new ArrayList<>();
actual.getTableChanges().forEachRemaining(actualTableChanges::add);
diff --git a/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java b/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
index fbbacfba..4160bc17 100644
--- a/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
+++ b/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
@@ -56,8 +56,8 @@ import org.apache.xtable.model.sync.SyncMode;
import org.apache.xtable.model.sync.SyncResult;
public class TestTableFormatSync {
- private final TargetClient mockTargetClient1 = mock(TargetClient.class);
- private final TargetClient mockTargetClient2 = mock(TargetClient.class);
+ private final ConversionTarget mockConversionTarget1 = mock(ConversionTarget.class);
+ private final ConversionTarget mockConversionTarget2 = mock(ConversionTarget.class);
@Test
void syncSnapshotWithFailureForOneFormat() {
@@ -77,12 +77,14 @@ public class TestTableFormatSync {
.partitionedDataFiles(fileGroups)
.pendingCommits(pendingCommitInstants)
.build();
- when(mockTargetClient1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
- when(mockTargetClient2.getTableFormat()).thenReturn(TableFormat.DELTA);
- doThrow(new RuntimeException("Failure")).when(mockTargetClient1).beginSync(startingTableState);
+ when(mockConversionTarget1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
+ when(mockConversionTarget2.getTableFormat()).thenReturn(TableFormat.DELTA);
+ doThrow(new RuntimeException("Failure"))
+ .when(mockConversionTarget1)
+ .beginSync(startingTableState);
Map<String, SyncResult> result =
TableFormatSync.getInstance()
- .syncSnapshot(Arrays.asList(mockTargetClient1, mockTargetClient2), snapshot);
+ .syncSnapshot(Arrays.asList(mockConversionTarget1, mockConversionTarget2), snapshot);
assertEquals(2, result.size());
SyncResult successResult = result.get(TableFormat.DELTA);
@@ -103,10 +105,10 @@ public class TestTableFormatSync {
.build(),
failureResult.getStatus());
- verifyBaseClientCalls(mockTargetClient2, startingTableState, pendingCommitInstants);
- verify(mockTargetClient2).syncFilesForSnapshot(fileGroups);
- verify(mockTargetClient2).completeSync();
- verify(mockTargetClient1, never()).completeSync();
+ verifyBaseClientCalls(mockConversionTarget2, startingTableState, pendingCommitInstants);
+ verify(mockConversionTarget2).syncFilesForSnapshot(fileGroups);
+ verify(mockConversionTarget2).completeSync();
+ verify(mockConversionTarget1, never()).completeSync();
}
private static void assertSyncResultTimes(SyncResult syncResult, Instant start) {
@@ -132,11 +134,11 @@ public class TestTableFormatSync {
TableChange.builder().tableAsOfChange(tableState3).filesDiff(dataFilesDiff3).build();
List<Instant> pendingCommitInstants = Collections.singletonList(Instant.now());
- when(mockTargetClient1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
- when(mockTargetClient2.getTableFormat()).thenReturn(TableFormat.DELTA);
+ when(mockConversionTarget1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
+ when(mockConversionTarget2.getTableFormat()).thenReturn(TableFormat.DELTA);
// throw exception on second change and show that first change is still returned for this format
// and other client is not affected
- doThrow(new RuntimeException("Failure")).when(mockTargetClient1).beginSync(tableState2);
+ doThrow(new RuntimeException("Failure")).when(mockConversionTarget1).beginSync(tableState2);
List<TableChange> tableChanges = Arrays.asList(tableChange1, tableChange2, tableChange3);
IncrementalTableChanges incrementalTableChanges =
@@ -145,12 +147,12 @@ public class TestTableFormatSync {
.tableChanges(tableChanges.iterator())
.build();
- Map<TargetClient, TableSyncMetadata> clientWithMetadata = new HashMap<>();
+ Map<ConversionTarget, TableSyncMetadata> clientWithMetadata = new HashMap<>();
clientWithMetadata.put(
- mockTargetClient1,
+ mockConversionTarget1,
TableSyncMetadata.of(Instant.now().minus(1, ChronoUnit.HOURS), Collections.emptyList()));
clientWithMetadata.put(
- mockTargetClient2,
+ mockConversionTarget2,
TableSyncMetadata.of(Instant.now().minus(1, ChronoUnit.HOURS), Collections.emptyList()));
Map<String, List<SyncResult>> result =
@@ -190,16 +192,16 @@ public class TestTableFormatSync {
assertSyncResultTimes(successResults.get(i), start);
}
- verifyBaseClientCalls(mockTargetClient1, tableState1, pendingCommitInstants);
- verify(mockTargetClient1).syncFilesForDiff(dataFilesDiff1);
- verifyBaseClientCalls(mockTargetClient2, tableState1, pendingCommitInstants);
- verify(mockTargetClient2).syncFilesForDiff(dataFilesDiff1);
- verifyBaseClientCalls(mockTargetClient2, tableState2, pendingCommitInstants);
- verify(mockTargetClient2).syncFilesForDiff(dataFilesDiff2);
- verifyBaseClientCalls(mockTargetClient2, tableState3, pendingCommitInstants);
- verify(mockTargetClient2).syncFilesForDiff(dataFilesDiff3);
- verify(mockTargetClient1, times(1)).completeSync();
- verify(mockTargetClient2, times(3)).completeSync();
+ verifyBaseClientCalls(mockConversionTarget1, tableState1, pendingCommitInstants);
+ verify(mockConversionTarget1).syncFilesForDiff(dataFilesDiff1);
+ verifyBaseClientCalls(mockConversionTarget2, tableState1, pendingCommitInstants);
+ verify(mockConversionTarget2).syncFilesForDiff(dataFilesDiff1);
+ verifyBaseClientCalls(mockConversionTarget2, tableState2, pendingCommitInstants);
+ verify(mockConversionTarget2).syncFilesForDiff(dataFilesDiff2);
+ verifyBaseClientCalls(mockConversionTarget2, tableState3, pendingCommitInstants);
+ verify(mockConversionTarget2).syncFilesForDiff(dataFilesDiff3);
+ verify(mockConversionTarget1, times(1)).completeSync();
+ verify(mockConversionTarget2, times(3)).completeSync();
}
@Test
@@ -219,8 +221,8 @@ public class TestTableFormatSync {
TableChange.builder().tableAsOfChange(tableState3).filesDiff(dataFilesDiff3).build();
List<Instant> pendingCommitInstants = Collections.singletonList(Instant.now());
- when(mockTargetClient1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
- when(mockTargetClient2.getTableFormat()).thenReturn(TableFormat.DELTA);
+ when(mockConversionTarget1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
+ when(mockConversionTarget2.getTableFormat()).thenReturn(TableFormat.DELTA);
List<TableChange> tableChanges = Arrays.asList(tableChange1, tableChange2, tableChange3);
IncrementalTableChanges incrementalTableChanges =
@@ -229,17 +231,17 @@ public class TestTableFormatSync {
.tableChanges(tableChanges.iterator())
.build();
- Map<TargetClient, TableSyncMetadata> clientWithMetadata = new HashMap<>();
- // mockTargetClient1 will have the first table change as a previously pending instant and
+ Map<ConversionTarget, TableSyncMetadata> clientWithMetadata = new HashMap<>();
+ // mockConversionTarget1 will have the first table change as a previously pending instant and
// otherwise be synced up to the 2nd change
clientWithMetadata.put(
- mockTargetClient1,
+ mockConversionTarget1,
TableSyncMetadata.of(
tableChange2.getTableAsOfChange().getLatestCommitTime(),
Collections.singletonList(tableChange1.getTableAsOfChange().getLatestCommitTime())));
- // mockTargetClient2 will have synced the first table change previously
+ // mockConversionTarget2 will have synced the first table change previously
clientWithMetadata.put(
- mockTargetClient2,
+ mockConversionTarget2,
TableSyncMetadata.of(
tableChange1.getTableAsOfChange().getLatestCommitTime(), Collections.emptyList()));
@@ -275,17 +277,17 @@ public class TestTableFormatSync {
}
// client1 syncs table changes 1 and 3
- verifyBaseClientCalls(mockTargetClient1, tableState1, pendingCommitInstants);
- verify(mockTargetClient1).syncFilesForDiff(dataFilesDiff1);
- verifyBaseClientCalls(mockTargetClient1, tableState3, pendingCommitInstants);
- verify(mockTargetClient1).syncFilesForDiff(dataFilesDiff3);
- verify(mockTargetClient1, times(2)).completeSync();
+ verifyBaseClientCalls(mockConversionTarget1, tableState1, pendingCommitInstants);
+ verify(mockConversionTarget1).syncFilesForDiff(dataFilesDiff1);
+ verifyBaseClientCalls(mockConversionTarget1, tableState3, pendingCommitInstants);
+ verify(mockConversionTarget1).syncFilesForDiff(dataFilesDiff3);
+ verify(mockConversionTarget1, times(2)).completeSync();
// client2 syncs table changes 2 and 3
- verifyBaseClientCalls(mockTargetClient2, tableState2, pendingCommitInstants);
- verify(mockTargetClient2).syncFilesForDiff(dataFilesDiff2);
- verifyBaseClientCalls(mockTargetClient2, tableState3, pendingCommitInstants);
- verify(mockTargetClient2).syncFilesForDiff(dataFilesDiff3);
- verify(mockTargetClient2, times(2)).completeSync();
+ verifyBaseClientCalls(mockConversionTarget2, tableState2, pendingCommitInstants);
+ verify(mockConversionTarget2).syncFilesForDiff(dataFilesDiff2);
+ verifyBaseClientCalls(mockConversionTarget2, tableState3, pendingCommitInstants);
+ verify(mockConversionTarget2).syncFilesForDiff(dataFilesDiff3);
+ verify(mockConversionTarget2, times(2)).completeSync();
}
@Test
@@ -297,8 +299,8 @@ public class TestTableFormatSync {
TableChange.builder().tableAsOfChange(tableState1).filesDiff(dataFilesDiff1).build();
List<Instant> pendingCommitInstants = Collections.emptyList();
- when(mockTargetClient1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
- when(mockTargetClient2.getTableFormat()).thenReturn(TableFormat.DELTA);
+ when(mockConversionTarget1.getTableFormat()).thenReturn(TableFormat.ICEBERG);
+ when(mockConversionTarget2.getTableFormat()).thenReturn(TableFormat.DELTA);
List<TableChange> tableChanges = Collections.singletonList(tableChange1);
IncrementalTableChanges incrementalTableChanges =
@@ -307,13 +309,13 @@ public class TestTableFormatSync {
.tableChanges(tableChanges.iterator())
.build();
- Map<TargetClient, TableSyncMetadata> clientWithMetadata = new HashMap<>();
- // mockTargetClient1 will have nothing to sync
+ Map<ConversionTarget, TableSyncMetadata> clientWithMetadata = new HashMap<>();
+ // mockConversionTarget1 will have nothing to sync
clientWithMetadata.put(
- mockTargetClient1, TableSyncMetadata.of(Instant.now(), Collections.emptyList()));
- // mockTargetClient2 will have synced the first table change previously
+ mockConversionTarget1, TableSyncMetadata.of(Instant.now(), Collections.emptyList()));
+ // mockConversionTarget2 will have synced the first table change previously
clientWithMetadata.put(
- mockTargetClient2,
+ mockConversionTarget2,
TableSyncMetadata.of(Instant.now().minus(1, ChronoUnit.HOURS), Collections.emptyList()));
Map<String, List<SyncResult>> result =
@@ -328,12 +330,12 @@ public class TestTableFormatSync {
assertSyncResultTimes(syncResult, start);
});
- verify(mockTargetClient1, never()).beginSync(any());
- verify(mockTargetClient1, never()).syncFilesForDiff(any());
- verify(mockTargetClient1, never()).completeSync();
+ verify(mockConversionTarget1, never()).beginSync(any());
+ verify(mockConversionTarget1, never()).syncFilesForDiff(any());
+ verify(mockConversionTarget1, never()).completeSync();
- verifyBaseClientCalls(mockTargetClient2, tableState1, pendingCommitInstants);
- verify(mockTargetClient2).syncFilesForDiff(dataFilesDiff1);
+ verifyBaseClientCalls(mockConversionTarget2, tableState1, pendingCommitInstants);
+ verify(mockConversionTarget2).syncFilesForDiff(dataFilesDiff1);
}
/**
@@ -371,7 +373,7 @@ public class TestTableFormatSync {
}
private void verifyBaseClientCalls(
- TargetClient mockClient,
+ ConversionTarget mockClient,
InternalTable startingTableState,
List<Instant> pendingCommitInstants) {
verify(mockClient).beginSync(startingTableState);
diff --git a/core/src/main/java/org/apache/xtable/client/OneTableClient.java b/core/src/main/java/org/apache/xtable/conversion/ConversionController.java
similarity index 79%
rename from core/src/main/java/org/apache/xtable/client/OneTableClient.java
rename to core/src/main/java/org/apache/xtable/conversion/ConversionController.java
index 280bf6b4..1d356e6e 100644
--- a/core/src/main/java/org/apache/xtable/client/OneTableClient.java
+++ b/core/src/main/java/org/apache/xtable/conversion/ConversionController.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import java.io.IOException;
import java.time.Instant;
@@ -45,10 +45,10 @@ import org.apache.xtable.model.InternalSnapshot;
import org.apache.xtable.model.TableSyncMetadata;
import org.apache.xtable.model.sync.SyncMode;
import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.spi.extractor.ConversionSource;
import org.apache.xtable.spi.extractor.ExtractFromSource;
-import org.apache.xtable.spi.extractor.SourceClient;
+import org.apache.xtable.spi.sync.ConversionTarget;
import org.apache.xtable.spi.sync.TableFormatSync;
-import org.apache.xtable.spi.sync.TargetClient;
/**
* Responsible for completing the entire lifecycle of the sync process given {@link
@@ -62,13 +62,13 @@ import org.apache.xtable.spi.sync.TargetClient;
*/
@Log4j2
@RequiredArgsConstructor(access = AccessLevel.PACKAGE)
-public class OneTableClient {
+public class ConversionController {
private final Configuration conf;
- private final TableFormatClientFactory tableFormatClientFactory;
+ private final ConversionTargetFactory conversionTargetFactory;
private final TableFormatSync tableFormatSync;
- public OneTableClient(Configuration conf) {
- this(conf, TableFormatClientFactory.getInstance(), TableFormatSync.getInstance());
+ public ConversionController(Configuration conf) {
+ this(conf, ConversionTargetFactory.getInstance(), TableFormatSync.getInstance());
}
/**
@@ -76,38 +76,43 @@ public class OneTableClient {
*
* @param config A per table level config containing tableBasePath, partitionFieldSpecConfig,
* targetTableFormats and syncMode
- * @param sourceClientProvider A provider for the source client instance, {@link
- * SourceClientProvider#init(Configuration, Map)} must be called before calling this method.
+ * @param conversionSourceProvider A provider for the {@link ConversionSource} instance, {@link
+ * ConversionSourceProvider#init(Configuration, Map)} must be called before calling this
+ * method.
* @return Returns a map containing the table format, and it's sync result. Run sync for a table
* with the provided per table level configuration.
*/
public <COMMIT> Map<String, SyncResult> sync(
- PerTableConfig config, SourceClientProvider<COMMIT> sourceClientProvider) {
+ PerTableConfig config, ConversionSourceProvider<COMMIT> conversionSourceProvider) {
if (config.getTargetTableFormats() == null || config.getTargetTableFormats().isEmpty()) {
throw new IllegalArgumentException("Please provide at-least one format to sync");
}
- try (SourceClient<COMMIT> sourceClient = sourceClientProvider.getSourceClientInstance(config)) {
- ExtractFromSource<COMMIT> source = ExtractFromSource.of(sourceClient);
+ try (ConversionSource<COMMIT> conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(config)) {
+ ExtractFromSource<COMMIT> source = ExtractFromSource.of(conversionSource);
- Map<String, TargetClient> syncClientByFormat =
+ Map<String, ConversionTarget> conversionTargetByFormat =
config.getTargetTableFormats().stream()
.collect(
Collectors.toMap(
Function.identity(),
tableFormat ->
- tableFormatClientFactory.createForFormat(tableFormat, config, conf)));
+ conversionTargetFactory.createForFormat(tableFormat, config, conf)));
// State for each TableFormat
Map<String, Optional<TableSyncMetadata>> lastSyncMetadataByFormat =
- syncClientByFormat.entrySet().stream()
+ conversionTargetByFormat.entrySet().stream()
.collect(
Collectors.toMap(
Map.Entry::getKey, entry -> entry.getValue().getTableMetadata()));
- Map<String, TargetClient> formatsToSyncIncrementally =
+ Map<String, ConversionTarget> formatsToSyncIncrementally =
getFormatsToSyncIncrementally(
- config, syncClientByFormat, lastSyncMetadataByFormat, source.getSourceClient());
- Map<String, TargetClient> formatsToSyncBySnapshot =
- syncClientByFormat.entrySet().stream()
+ config,
+ conversionTargetByFormat,
+ lastSyncMetadataByFormat,
+ source.getConversionSource());
+ Map<String, ConversionTarget> formatsToSyncBySnapshot =
+ conversionTargetByFormat.entrySet().stream()
.filter(entry -> !formatsToSyncIncrementally.containsKey(entry.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
SyncResultForTableFormats syncResultForSnapshotSync =
@@ -134,7 +139,7 @@ public class OneTableClient {
}
return syncResultsMerged;
} catch (IOException ioException) {
- throw new OneIOException("Failed to close source client", ioException);
+ throw new OneIOException("Failed to close source converter", ioException);
}
}
@@ -146,16 +151,16 @@ public class OneTableClient {
.collect(Collectors.joining(","));
}
- private <COMMIT> Map<String, TargetClient> getFormatsToSyncIncrementally(
+ private <COMMIT> Map<String, ConversionTarget> getFormatsToSyncIncrementally(
PerTableConfig perTableConfig,
- Map<String, TargetClient> syncClientByFormat,
+ Map<String, ConversionTarget> conversionTargetByFormat,
Map<String, Optional<TableSyncMetadata>> lastSyncMetadataByFormat,
- SourceClient<COMMIT> sourceClient) {
+ ConversionSource<COMMIT> conversionSource) {
if (perTableConfig.getSyncMode() == SyncMode.FULL) {
// Full sync requested by config, hence no incremental sync.
return Collections.emptyMap();
}
- return syncClientByFormat.entrySet().stream()
+ return conversionTargetByFormat.entrySet().stream()
.filter(
entry -> {
Optional<Instant> lastSyncInstant =
@@ -167,30 +172,31 @@ public class OneTableClient {
.get(entry.getKey())
.map(TableSyncMetadata::getInstantsToConsiderForNextSync)
.orElse(Collections.emptyList());
- return isIncrementalSyncSufficient(sourceClient, lastSyncInstant, pendingInstants);
+ return isIncrementalSyncSufficient(
+ conversionSource, lastSyncInstant, pendingInstants);
})
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
private <COMMIT> SyncResultForTableFormats syncSnapshot(
- Map<String, TargetClient> syncClientByFormat, ExtractFromSource<COMMIT> source) {
+ Map<String, ConversionTarget> conversionTargetByFormat, ExtractFromSource<COMMIT> source) {
InternalSnapshot snapshot = source.extractSnapshot();
Map<String, SyncResult> syncResultsByFormat =
- tableFormatSync.syncSnapshot(syncClientByFormat.values(), snapshot);
+ tableFormatSync.syncSnapshot(conversionTargetByFormat.values(), snapshot);
return SyncResultForTableFormats.builder().lastSyncResult(syncResultsByFormat).build();
}
private <COMMIT> SyncResultForTableFormats syncIncrementalChanges(
- Map<String, TargetClient> syncClientByFormat,
+ Map<String, ConversionTarget> conversionTargetByFormat,
Map<String, Optional<TableSyncMetadata>> lastSyncMetadataByFormat,
ExtractFromSource<COMMIT> source) {
Map<String, SyncResult> syncResultsByFormat = Collections.emptyMap();
- Map<TargetClient, TableSyncMetadata> filteredSyncMetadataByFormat =
+ Map<ConversionTarget, TableSyncMetadata> filteredSyncMetadataByFormat =
lastSyncMetadataByFormat.entrySet().stream()
- .filter(entry -> syncClientByFormat.containsKey(entry.getKey()))
+ .filter(entry -> conversionTargetByFormat.containsKey(entry.getKey()))
.collect(
Collectors.toMap(
- entry -> syncClientByFormat.get(entry.getKey()),
+ entry -> conversionTargetByFormat.get(entry.getKey()),
entry -> entry.getValue().get()));
InstantsForIncrementalSync instantsForIncrementalSync =
@@ -211,14 +217,14 @@ public class OneTableClient {
/**
* Checks if incremental sync is sufficient for a target table format.
*
- * @param sourceClient {@link SourceClient}
+ * @param conversionSource {@link ConversionSource}
* @param lastSyncInstant the last instant at which the target table format was synced
* @param pendingInstants the list of pending instants for the target table format to consider for
* next sync
* @return true if incremental sync is sufficient, false otherwise.
*/
private <COMMIT> boolean isIncrementalSyncSufficient(
- SourceClient<COMMIT> sourceClient,
+ ConversionSource<COMMIT> conversionSource,
Optional<Instant> lastSyncInstant,
List<Instant> pendingInstants) {
Stream<Instant> pendingInstantsStream =
@@ -235,7 +241,7 @@ public class OneTableClient {
return false;
}
boolean isIncrementalSafeFromInstant =
- sourceClient.isIncrementalSyncSafeFrom(earliestInstant.get());
+ conversionSource.isIncrementalSyncSafeFrom(earliestInstant.get());
if (!isIncrementalSafeFromInstant) {
log.info(
"Incremental sync is not safe from instant {}. Falling back to snapshot sync.",
@@ -246,7 +252,7 @@ public class OneTableClient {
}
private InstantsForIncrementalSync getMostOutOfSyncCommitAndPendingCommits(
- Map<TargetClient, TableSyncMetadata> lastSyncMetadataByFormat) {
+ Map<ConversionTarget, TableSyncMetadata> lastSyncMetadataByFormat) {
Optional<Instant> mostOutOfSyncCommit =
lastSyncMetadataByFormat.values().stream()
.map(TableSyncMetadata::getLastInstantSynced)
diff --git a/core/src/main/java/org/apache/xtable/client/SourceClientProvider.java b/core/src/main/java/org/apache/xtable/conversion/ConversionSourceProvider.java
similarity index 58%
rename from core/src/main/java/org/apache/xtable/client/SourceClientProvider.java
rename to core/src/main/java/org/apache/xtable/conversion/ConversionSourceProvider.java
index 9c4c91bb..09cad6db 100644
--- a/core/src/main/java/org/apache/xtable/client/SourceClientProvider.java
+++ b/core/src/main/java/org/apache/xtable/conversion/ConversionSourceProvider.java
@@ -16,42 +16,43 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
-import org.apache.xtable.spi.extractor.SourceClient;
+import org.apache.xtable.spi.extractor.ConversionSource;
/**
- * Provides the functionality to provide an instance of the {@link SourceClient} for a given table
- * format. The provider can create a new instance of the client for each table format or reuse the
- * same instance.
+ * Provides the functionality to provide an instance of the {@link ConversionSource} for a given
+ * table format. The provider can create a new instance of the conversion source for each table
+ * format or reuse the same instance.
*/
-public abstract class SourceClientProvider<COMMIT> {
+public abstract class ConversionSourceProvider<COMMIT> {
/** The Hadoop configuration to use when reading from the source table. */
protected Configuration hadoopConf;
- /** The configuration for the client. */
- protected Map<String, String> clientConf;
+ /** The configuration for the source. */
+ protected Map<String, String> sourceConf;
/** The configuration for the table to read from. */
protected PerTableConfig sourceTableConfig;
- /** Initializes the provider various client specific configurations. */
- public void init(Configuration hadoopConf, Map<String, String> clientConf) {
+ /** Initializes the provider various source specific configurations. */
+ public void init(Configuration hadoopConf, Map<String, String> sourceConf) {
this.hadoopConf = hadoopConf;
- this.clientConf = clientConf;
+ this.sourceConf = sourceConf;
}
/**
- * Returns an instance of the {@link SourceClient} for the given table format. The {@link
- * SourceClient} is source table aware. A source table may require creation of a new instance of
- * the client. In some cases reusing the same instance across table configurations may be
+ * Returns an instance of the {@link ConversionSource} for the given table format. The {@link
+ * ConversionSource} is source table aware. A source table may require creation of a new instance
+ * of the source. In some cases reusing the same instance across table configurations may be
* efficient. The provider can decide how to manage this.
*
- * @return the source client
+ * @return the conversion source
*/
- public abstract SourceClient<COMMIT> getSourceClientInstance(PerTableConfig sourceTableConfig);
+ public abstract ConversionSource<COMMIT> getConversionSourceInstance(
+ PerTableConfig sourceTableConfig);
}
diff --git a/core/src/main/java/org/apache/xtable/client/TableFormatClientFactory.java b/core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java
similarity index 55%
rename from core/src/main/java/org/apache/xtable/client/TableFormatClientFactory.java
rename to core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java
index 6b6a8796..68e85b4d 100644
--- a/core/src/main/java/org/apache/xtable/client/TableFormatClientFactory.java
+++ b/core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import java.util.ServiceLoader;
@@ -26,45 +26,46 @@ import lombok.NoArgsConstructor;
import org.apache.hadoop.conf.Configuration;
import org.apache.xtable.exception.NotSupportedException;
-import org.apache.xtable.spi.sync.TargetClient;
+import org.apache.xtable.spi.sync.ConversionTarget;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class TableFormatClientFactory {
- private static final TableFormatClientFactory INSTANCE = new TableFormatClientFactory();
+public class ConversionTargetFactory {
+ private static final ConversionTargetFactory INSTANCE = new ConversionTargetFactory();
- public static TableFormatClientFactory getInstance() {
+ public static ConversionTargetFactory getInstance() {
return INSTANCE;
}
/**
- * Create a fully initialized instance of the TargetClient represented by the given Table Format
- * name. Initialization is done with the config provideed through PerTableConfig and Conifuration
- * params.
+ * Create a fully initialized instance of the ConversionTarget represented by the given Table
+ * Format name. Initialization is done with the config provided through PerTableConfig and
+ * Configuration params.
*
* @param tableFormat
* @param perTableConfig
* @param configuration
* @return
*/
- public TargetClient createForFormat(
+ public ConversionTarget createForFormat(
String tableFormat, PerTableConfig perTableConfig, Configuration configuration) {
- TargetClient targetClient = createTargetClientForName(tableFormat);
+ ConversionTarget conversionTarget = createConversionTargetForName(tableFormat);
- targetClient.init(perTableConfig, configuration);
- return targetClient;
+ conversionTarget.init(perTableConfig, configuration);
+ return conversionTarget;
}
/**
- * Create an instance of the TargetClient via the default no-arg constructor. Expectation is that
- * target client specific settings may be provided prior to the calling of TargetClient.init()
- * which must be called prior to actual use of the target client returned by this factory method.
+ * Create an instance of the ConversionTarget via the default no-arg constructor. Expectation is
+ * that target conversion specific settings may be provided prior to the calling of
+ * ConversionTarget.init() which must be called prior to actual use of the conversion target
+ * returned by this factory method.
*
* @param tableFormatName
* @return
*/
- public TargetClient createTargetClientForName(String tableFormatName) {
- ServiceLoader<TargetClient> loader = ServiceLoader.load(TargetClient.class);
- for (TargetClient target : loader) {
+ public ConversionTarget createConversionTargetForName(String tableFormatName) {
+ ServiceLoader<ConversionTarget> loader = ServiceLoader.load(ConversionTarget.class);
+ for (ConversionTarget target : loader) {
if (target.getTableFormat().equalsIgnoreCase(tableFormatName)) {
return target;
}
diff --git a/core/src/main/java/org/apache/xtable/client/PerTableConfigImpl.java b/core/src/main/java/org/apache/xtable/conversion/PerTableConfigImpl.java
similarity index 99%
rename from core/src/main/java/org/apache/xtable/client/PerTableConfigImpl.java
rename to core/src/main/java/org/apache/xtable/conversion/PerTableConfigImpl.java
index 37bacd4d..f34402b1 100644
--- a/core/src/main/java/org/apache/xtable/client/PerTableConfigImpl.java
+++ b/core/src/main/java/org/apache/xtable/conversion/PerTableConfigImpl.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import java.util.List;
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaSourceClient.java b/core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
similarity index 97%
rename from core/src/main/java/org/apache/xtable/delta/DeltaSourceClient.java
rename to core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
index c33db77d..ae572ca1 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaSourceClient.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
@@ -57,12 +57,12 @@ import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.FileFormat;
import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.PartitionFileGroup;
+import org.apache.xtable.spi.extractor.ConversionSource;
import org.apache.xtable.spi.extractor.DataFileIterator;
-import org.apache.xtable.spi.extractor.SourceClient;
@Log4j2
@Builder
-public class DeltaSourceClient implements SourceClient<Long> {
+public class DeltaConversionSource implements ConversionSource<Long> {
@Builder.Default
private final DeltaDataFileExtractor dataFileExtractor = DeltaDataFileExtractor.builder().build();
@@ -109,7 +109,7 @@ public class DeltaSourceClient implements SourceClient<Long> {
@Override
public TableChange getTableChangeForCommit(Long versionNumber) {
InternalTable tableAtVersion = tableExtractor.table(deltaLog, tableName, versionNumber);
- // Client to call getCommitsBacklog and call this method.
+ // Source to call getCommitsBacklog and call this method.
List<Action> actionsForVersion = getChangesState().getActionsForVersion(versionNumber);
Snapshot snapshotAtVersion = deltaLog.getSnapshotAt(versionNumber, Option.empty());
FileFormat fileFormat =
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaSourceClientProvider.java b/core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java
similarity index 56%
rename from core/src/main/java/org/apache/xtable/delta/DeltaSourceClientProvider.java
rename to core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java
index c885efba..f42425db 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaSourceClientProvider.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaConversionSourceProvider.java
@@ -22,23 +22,21 @@ import org.apache.spark.sql.SparkSession;
import io.delta.tables.DeltaTable;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.SourceClientProvider;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.PerTableConfig;
-/** A concrete implementation of {@link SourceClientProvider} for Delta Lake table format. */
-public class DeltaSourceClientProvider extends SourceClientProvider<Long> {
+/** A concrete implementation of {@link ConversionSourceProvider} for Delta Lake table format. */
+public class DeltaConversionSourceProvider extends ConversionSourceProvider<Long> {
@Override
- public DeltaSourceClient getSourceClientInstance(PerTableConfig perTableConfig) {
- SparkSession sparkSession = DeltaClientUtils.buildSparkSession(hadoopConf);
+ public DeltaConversionSource getConversionSourceInstance(PerTableConfig perTableConfig) {
+ SparkSession sparkSession = DeltaConversionUtils.buildSparkSession(hadoopConf);
DeltaTable deltaTable = DeltaTable.forPath(sparkSession, perTableConfig.getTableBasePath());
- DeltaSourceClient deltaSourceClient =
- DeltaSourceClient.builder()
- .sparkSession(sparkSession)
- .tableName(perTableConfig.getTableName())
- .basePath(perTableConfig.getTableBasePath())
- .deltaTable(deltaTable)
- .deltaLog(deltaTable.deltaLog())
- .build();
- return deltaSourceClient;
+ return DeltaConversionSource.builder()
+ .sparkSession(sparkSession)
+ .tableName(perTableConfig.getTableName())
+ .basePath(perTableConfig.getTableBasePath())
+ .deltaTable(deltaTable)
+ .deltaLog(deltaTable.deltaLog())
+ .build();
}
}
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaClient.java b/core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java
similarity index 96%
rename from core/src/main/java/org/apache/xtable/delta/DeltaClient.java
rename to core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java
index 9d54e2e7..798c3907 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaClient.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java
@@ -54,7 +54,7 @@ import scala.collection.Seq;
import com.google.common.annotations.VisibleForTesting;
-import org.apache.xtable.client.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfig;
import org.apache.xtable.exception.NotSupportedException;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.TableSyncMetadata;
@@ -63,9 +63,9 @@ import org.apache.xtable.model.schema.InternalSchema;
import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.PartitionFileGroup;
import org.apache.xtable.model.storage.TableFormat;
-import org.apache.xtable.spi.sync.TargetClient;
+import org.apache.xtable.spi.sync.ConversionTarget;
-public class DeltaClient implements TargetClient {
+public class DeltaConversionTarget implements ConversionTarget {
private static final String MIN_READER_VERSION = String.valueOf(1);
// gets access to generated columns.
private static final String MIN_WRITER_VERSION = String.valueOf(4);
@@ -79,9 +79,9 @@ public class DeltaClient implements TargetClient {
private int logRetentionInHours;
private TransactionState transactionState;
- public DeltaClient() {}
+ public DeltaConversionTarget() {}
- public DeltaClient(PerTableConfig perTableConfig, SparkSession sparkSession) {
+ public DeltaConversionTarget(PerTableConfig perTableConfig, SparkSession sparkSession) {
this(
perTableConfig.getTableDataPath(),
perTableConfig.getTableName(),
@@ -93,7 +93,7 @@ public class DeltaClient implements TargetClient {
}
@VisibleForTesting
- DeltaClient(
+ DeltaConversionTarget(
String tableDataPath,
String tableName,
int logRetentionInHours,
@@ -135,7 +135,7 @@ public class DeltaClient implements TargetClient {
@Override
public void init(PerTableConfig perTableConfig, Configuration configuration) {
- SparkSession sparkSession = DeltaClientUtils.buildSparkSession(configuration);
+ SparkSession sparkSession = DeltaConversionUtils.buildSparkSession(configuration);
_init(
perTableConfig.getTableDataPath(),
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaClientUtils.java b/core/src/main/java/org/apache/xtable/delta/DeltaConversionUtils.java
similarity index 94%
rename from core/src/main/java/org/apache/xtable/delta/DeltaClientUtils.java
rename to core/src/main/java/org/apache/xtable/delta/DeltaConversionUtils.java
index fb5817bb..3401df84 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaClientUtils.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaConversionUtils.java
@@ -26,14 +26,14 @@ import org.apache.spark.SparkConf;
import org.apache.spark.serializer.KryoSerializer;
import org.apache.spark.sql.SparkSession;
-/** A utility class for Delta client. */
+/** A utility class for Delta conversion. */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class DeltaClientUtils {
+public class DeltaConversionUtils {
static SparkSession buildSparkSession(Configuration conf) {
SparkConf sparkConf =
new SparkConf()
- .setAppName("onetableclient")
+ .setAppName("xtable")
.set("spark.serializer", KryoSerializer.class.getName())
.set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.set(
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java b/core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java
index 70b4aaaa..4642e818 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaIncrementalChangesState.java
@@ -67,8 +67,8 @@ public class DeltaIncrementalChangesState {
/**
* Returns the versions in sorted order. The start version is the next one after the last sync
- * version to the target client. The end version is the latest version in the Delta table at the
- * time of initialization.
+ * version to the target. The end version is the latest version in the Delta table at the time of
+ * initialization.
*
* @return
*/
diff --git a/core/src/main/java/org/apache/xtable/hudi/HudiClient.java b/core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java
similarity index 98%
rename from core/src/main/java/org/apache/xtable/hudi/HudiClient.java
rename to core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java
index 2cdc4015..9e0a8d88 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiClient.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java
@@ -50,15 +50,15 @@ import org.apache.xtable.model.InternalSnapshot;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.TableChange;
import org.apache.xtable.model.schema.SchemaCatalog;
-import org.apache.xtable.spi.extractor.SourceClient;
+import org.apache.xtable.spi.extractor.ConversionSource;
-public class HudiClient implements SourceClient<HoodieInstant> {
+public class HudiConversionSource implements ConversionSource<HoodieInstant> {
private final HoodieTableMetaClient metaClient;
private final HudiTableExtractor tableExtractor;
private final HudiDataFileExtractor dataFileExtractor;
- public HudiClient(
+ public HudiConversionSource(
HoodieTableMetaClient metaClient,
HudiSourcePartitionSpecExtractor sourcePartitionSpecExtractor) {
this.metaClient = metaClient;
diff --git a/core/src/main/java/org/apache/xtable/hudi/HudiSourceClientProvider.java b/core/src/main/java/org/apache/xtable/hudi/HudiConversionSourceProvider.java
similarity index 78%
rename from core/src/main/java/org/apache/xtable/hudi/HudiSourceClientProvider.java
rename to core/src/main/java/org/apache/xtable/hudi/HudiConversionSourceProvider.java
index 2c49f60e..f6e5e28d 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiSourceClientProvider.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiConversionSourceProvider.java
@@ -24,15 +24,15 @@ import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.SourceClientProvider;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.PerTableConfig;
-/** A concrete implementation of {@link SourceClientProvider} for Hudi table format. */
+/** A concrete implementation of {@link ConversionSourceProvider} for Hudi table format. */
@Log4j2
-public class HudiSourceClientProvider extends SourceClientProvider<HoodieInstant> {
+public class HudiConversionSourceProvider extends ConversionSourceProvider<HoodieInstant> {
@Override
- public HudiClient getSourceClientInstance(PerTableConfig sourceTableConfig) {
+ public HudiConversionSource getConversionSourceInstance(PerTableConfig sourceTableConfig) {
this.sourceTableConfig = sourceTableConfig;
HoodieTableMetaClient metaClient =
HoodieTableMetaClient.builder()
@@ -48,6 +48,6 @@ public class HudiSourceClientProvider extends SourceClientProvider<HoodieInstant
(HudiSourcePartitionSpecExtractor)
sourceTableConfig.getHudiSourceConfig().loadSourcePartitionSpecExtractor();
- return new HudiClient(metaClient, sourcePartitionSpecExtractor);
+ return new HudiConversionSource(metaClient, sourcePartitionSpecExtractor);
}
}
diff --git a/core/src/main/java/org/apache/xtable/hudi/HudiTargetClient.java b/core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java
similarity index 99%
rename from core/src/main/java/org/apache/xtable/hudi/HudiTargetClient.java
rename to core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java
index 0168ae5b..424d8228 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiTargetClient.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java
@@ -79,7 +79,7 @@ import org.apache.hudi.table.action.clean.CleanPlanner;
import com.google.common.annotations.VisibleForTesting;
import org.apache.xtable.avro.AvroSchemaConverter;
-import org.apache.xtable.client.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfig;
import org.apache.xtable.exception.NotSupportedException;
import org.apache.xtable.exception.OneIOException;
import org.apache.xtable.model.InternalTable;
@@ -90,10 +90,10 @@ import org.apache.xtable.model.schema.InternalSchema;
import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.PartitionFileGroup;
import org.apache.xtable.model.storage.TableFormat;
-import org.apache.xtable.spi.sync.TargetClient;
+import org.apache.xtable.spi.sync.ConversionTarget;
@Log4j2
-public class HudiTargetClient implements TargetClient {
+public class HudiConversionTarget implements ConversionTarget {
private BaseFileUpdatesExtractor baseFileUpdatesExtractor;
private AvroSchemaConverter avroSchemaConverter;
private HudiTableManager hudiTableManager;
@@ -104,10 +104,10 @@ public class HudiTargetClient implements TargetClient {
private Optional<HoodieTableMetaClient> metaClient;
private CommitState commitState;
- public HudiTargetClient() {}
+ public HudiConversionTarget() {}
@VisibleForTesting
- HudiTargetClient(
+ HudiConversionTarget(
PerTableConfig perTableConfig,
Configuration configuration,
int maxNumDeltaCommitsBeforeCompaction) {
@@ -124,7 +124,7 @@ public class HudiTargetClient implements TargetClient {
}
@VisibleForTesting
- HudiTargetClient(
+ HudiConversionTarget(
String tableDataPath,
int timelineRetentionInHours,
int maxNumDeltaCommitsBeforeCompaction,
diff --git a/core/src/main/java/org/apache/xtable/hudi/HudiSourceConfigImpl.java b/core/src/main/java/org/apache/xtable/hudi/HudiSourceConfigImpl.java
index e5fcf820..0fa8b529 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiSourceConfigImpl.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiSourceConfigImpl.java
@@ -28,7 +28,7 @@ import lombok.Value;
import com.google.common.base.Preconditions;
-import org.apache.xtable.client.HudiSourceConfig;
+import org.apache.xtable.conversion.HudiSourceConfig;
import org.apache.xtable.model.schema.PartitionTransformType;
import org.apache.xtable.reflection.ReflectionUtils;
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergCatalogConfig.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergCatalogConfig.java
index 317ddc0e..d5d7a3c5 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergCatalogConfig.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergCatalogConfig.java
@@ -25,7 +25,7 @@ import lombok.Builder;
import lombok.NonNull;
import lombok.Value;
-import org.apache.xtable.client.CatalogConfig;
+import org.apache.xtable.conversion.CatalogConfig;
@Value
@Builder
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClient.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSource.java
similarity index 98%
rename from core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClient.java
rename to core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSource.java
index 415356c7..76775b49 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClient.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSource.java
@@ -36,7 +36,7 @@ import org.apache.iceberg.hadoop.HadoopFileIO;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
-import org.apache.xtable.client.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfig;
import org.apache.xtable.exception.OneIOException;
import org.apache.xtable.model.CommitsBacklog;
import org.apache.xtable.model.InstantsForIncrementalSync;
@@ -50,11 +50,11 @@ import org.apache.xtable.model.schema.SchemaVersion;
import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.storage.*;
import org.apache.xtable.model.storage.InternalDataFile;
-import org.apache.xtable.spi.extractor.SourceClient;
+import org.apache.xtable.spi.extractor.ConversionSource;
@Log4j2
@Builder
-public class IcebergSourceClient implements SourceClient<Snapshot> {
+public class IcebergConversionSource implements ConversionSource<Snapshot> {
@NonNull private final Configuration hadoopConf;
@NonNull private final PerTableConfig sourceTableConfig;
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClientProvider.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSourceProvider.java
similarity index 69%
rename from core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClientProvider.java
rename to core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSourceProvider.java
index d3abf238..9a2bb840 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClientProvider.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionSourceProvider.java
@@ -20,14 +20,14 @@ package org.apache.xtable.iceberg;
import org.apache.iceberg.Snapshot;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.SourceClientProvider;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.PerTableConfig;
-/** A concrete implementation of {@link SourceClientProvider} for Hudi table format. */
-public class IcebergSourceClientProvider extends SourceClientProvider<Snapshot> {
+/** A concrete implementation of {@link ConversionSourceProvider} for Hudi table format. */
+public class IcebergConversionSourceProvider extends ConversionSourceProvider<Snapshot> {
@Override
- public IcebergSourceClient getSourceClientInstance(PerTableConfig sourceTableConfig) {
- return IcebergSourceClient.builder()
+ public IcebergConversionSource getConversionSourceInstance(PerTableConfig sourceTableConfig) {
+ return IcebergConversionSource.builder()
.sourceTableConfig(sourceTableConfig)
.hadoopConf(hadoopConf)
.build();
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergClient.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java
similarity index 97%
rename from core/src/main/java/org/apache/xtable/iceberg/IcebergClient.java
rename to core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java
index 83957d44..f04b3419 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergClient.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java
@@ -43,7 +43,7 @@ import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NotFoundException;
-import org.apache.xtable.client.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfig;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.TableSyncMetadata;
import org.apache.xtable.model.schema.InternalPartitionField;
@@ -51,10 +51,10 @@ import org.apache.xtable.model.schema.InternalSchema;
import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.PartitionFileGroup;
import org.apache.xtable.model.storage.TableFormat;
-import org.apache.xtable.spi.sync.TargetClient;
+import org.apache.xtable.spi.sync.ConversionTarget;
@Log4j2
-public class IcebergClient implements TargetClient {
+public class IcebergConversionTarget implements ConversionTarget {
private static final String METADATA_DIR_PATH = "/metadata/";
private IcebergSchemaExtractor schemaExtractor;
private IcebergSchemaSync schemaSync;
@@ -71,9 +71,9 @@ public class IcebergClient implements TargetClient {
private Table table;
private InternalTable internalTableState;
- public IcebergClient() {}
+ public IcebergConversionTarget() {}
- IcebergClient(
+ IcebergConversionTarget(
PerTableConfig perTableConfig,
Configuration configuration,
IcebergSchemaExtractor schemaExtractor,
@@ -123,7 +123,7 @@ public class IcebergClient implements TargetClient {
// Load the table state if it already exists
this.table = tableManager.getTable(catalogConfig, tableIdentifier, basePath);
}
- // Clear any corrupted state before using the target client
+ // Clear any corrupted state before using the conversion target
rollbackCorruptCommits();
}
diff --git a/core/src/main/resources/META-INF/services/org.apache.xtable.spi.sync.TargetClient b/core/src/main/resources/META-INF/services/org.apache.xtable.spi.sync.ConversionTarget
similarity index 86%
rename from core/src/main/resources/META-INF/services/org.apache.xtable.spi.sync.TargetClient
rename to core/src/main/resources/META-INF/services/org.apache.xtable.spi.sync.ConversionTarget
index 07eda44e..a00b41e0 100644
--- a/core/src/main/resources/META-INF/services/org.apache.xtable.spi.sync.TargetClient
+++ b/core/src/main/resources/META-INF/services/org.apache.xtable.spi.sync.ConversionTarget
@@ -16,6 +16,6 @@
# limitations under the License.
##########################################################################
-org.apache.xtable.hudi.HudiTargetClient
-org.apache.xtable.delta.DeltaClient
-org.apache.xtable.iceberg.IcebergClient
+org.apache.xtable.hudi.HudiConversionTarget
+org.apache.xtable.delta.DeltaConversionTarget
+org.apache.xtable.iceberg.IcebergConversionTarget
diff --git a/core/src/test/java/org/apache/xtable/ITOneTableClient.java b/core/src/test/java/org/apache/xtable/ITConversionController.java
similarity index 84%
rename from core/src/test/java/org/apache/xtable/ITOneTableClient.java
rename to core/src/test/java/org/apache/xtable/ITConversionController.java
index 6f59e676..fda52d9c 100644
--- a/core/src/test/java/org/apache/xtable/ITOneTableClient.java
+++ b/core/src/test/java/org/apache/xtable/ITConversionController.java
@@ -81,19 +81,19 @@ import org.apache.spark.sql.delta.DeltaLog;
import com.google.common.collect.ImmutableList;
-import org.apache.xtable.client.OneTableClient;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.PerTableConfigImpl;
-import org.apache.xtable.client.SourceClientProvider;
-import org.apache.xtable.delta.DeltaSourceClientProvider;
-import org.apache.xtable.hudi.HudiSourceClientProvider;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfigImpl;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
import org.apache.xtable.hudi.HudiSourceConfigImpl;
import org.apache.xtable.hudi.HudiTestUtil;
-import org.apache.xtable.iceberg.IcebergSourceClientProvider;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
-public class ITOneTableClient {
+public class ITConversionController {
@TempDir public static Path tempDir;
private static final DateTimeFormatter DATE_FORMAT =
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS").withZone(ZoneId.of("UTC"));
@@ -143,20 +143,22 @@ public class ITOneTableClient {
return Stream.of(Arguments.of(SyncMode.INCREMENTAL), Arguments.of(SyncMode.FULL));
}
- private SourceClientProvider<?> getSourceClientProvider(String sourceTableFormat) {
+ private ConversionSourceProvider<?> getConversionSourceProvider(String sourceTableFormat) {
if (sourceTableFormat.equalsIgnoreCase(HUDI)) {
- SourceClientProvider<HoodieInstant> hudiSourceClientProvider = new HudiSourceClientProvider();
- hudiSourceClientProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
- return hudiSourceClientProvider;
+ ConversionSourceProvider<HoodieInstant> hudiConversionSourceProvider =
+ new HudiConversionSourceProvider();
+ hudiConversionSourceProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
+ return hudiConversionSourceProvider;
} else if (sourceTableFormat.equalsIgnoreCase(DELTA)) {
- SourceClientProvider<Long> deltaSourceClientProvider = new DeltaSourceClientProvider();
- deltaSourceClientProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
- return deltaSourceClientProvider;
+ ConversionSourceProvider<Long> deltaConversionSourceProvider =
+ new DeltaConversionSourceProvider();
+ deltaConversionSourceProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
+ return deltaConversionSourceProvider;
} else if (sourceTableFormat.equalsIgnoreCase(ICEBERG)) {
- SourceClientProvider<Snapshot> icebergSourceClientProvider =
- new IcebergSourceClientProvider();
- icebergSourceClientProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
- return icebergSourceClientProvider;
+ ConversionSourceProvider<Snapshot> icebergConversionSourceProvider =
+ new IcebergConversionSourceProvider();
+ icebergConversionSourceProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
+ return icebergConversionSourceProvider;
} else {
throw new IllegalArgumentException("Unsupported source format: " + sourceTableFormat);
}
@@ -177,13 +179,14 @@ public class ITOneTableClient {
public void testVariousOperations(
String sourceTableFormat, SyncMode syncMode, boolean isPartitioned) {
String tableName = getTableName();
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
+ ConversionController conversionController = new ConversionController(jsc.hadoopConfiguration());
List<String> targetTableFormats = getOtherFormats(sourceTableFormat);
String oneTablePartitionConfig = null;
if (isPartitioned) {
oneTablePartitionConfig = "level:VALUE";
}
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(sourceTableFormat);
+ ConversionSourceProvider<?> conversionSourceProvider =
+ getConversionSourceProvider(sourceTableFormat);
List<?> insertRecords;
try (GenericTable table =
GenericTable.getInstance(
@@ -202,17 +205,17 @@ public class ITOneTableClient {
.build())
.syncMode(syncMode)
.build();
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 100);
// make multiple commits and then sync
table.insertRows(100);
table.upsertRows(insertRecords.subList(0, 20));
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 200);
table.deleteRows(insertRecords.subList(30, 50));
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 180);
checkDatasetEquivalenceWithFilter(
sourceTableFormat, table, targetTableFormats, table.getFilterQuery());
@@ -235,27 +238,27 @@ public class ITOneTableClient {
.build();
List<Row> insertsAfterSchemaUpdate = tableWithUpdatedSchema.insertRows(100);
tableWithUpdatedSchema.reload();
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 280);
tableWithUpdatedSchema.deleteRows(insertsAfterSchemaUpdate.subList(60, 90));
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 250);
if (isPartitioned) {
// Adds new partition.
tableWithUpdatedSchema.insertRecordsForSpecialPartition(50);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 300);
// Drops partition.
tableWithUpdatedSchema.deleteSpecialPartition();
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 250);
// Insert records to the dropped partition again.
tableWithUpdatedSchema.insertRecordsForSpecialPartition(50);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 300);
}
}
@@ -266,7 +269,7 @@ public class ITOneTableClient {
public void testConcurrentInsertWritesInSource(
SyncMode syncMode, PartitionConfig partitionConfig) {
String tableName = getTableName();
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(HUDI);
+ ConversionSourceProvider<?> conversionSourceProvider = getConversionSourceProvider(HUDI);
List<String> targetTableFormats = getOtherFormats(HUDI);
try (TestJavaHudiTable table =
TestJavaHudiTable.forStandardSchema(
@@ -291,12 +294,13 @@ public class ITOneTableClient {
.build())
.syncMode(syncMode)
.build();
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, targetTableFormats, 50);
table.insertRecordsWithCommitAlreadyStarted(insertsForCommit1, commitInstant1, true);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, targetTableFormats, 100);
}
}
@@ -306,7 +310,7 @@ public class ITOneTableClient {
public void testConcurrentInsertsAndTableServiceWrites(
SyncMode syncMode, PartitionConfig partitionConfig) {
HoodieTableType tableType = HoodieTableType.MERGE_ON_READ;
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(HUDI);
+ ConversionSourceProvider<?> conversionSourceProvider = getConversionSourceProvider(HUDI);
List<String> targetTableFormats = getOtherFormats(HUDI);
String tableName = getTableName();
try (TestSparkHudiTable table =
@@ -325,8 +329,9 @@ public class ITOneTableClient {
.build())
.syncMode(syncMode)
.build();
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, targetTableFormats, 50);
table.deleteRecords(insertedRecords1.subList(0, 20), true);
@@ -334,7 +339,7 @@ public class ITOneTableClient {
String scheduledCompactionInstant = table.onlyScheduleCompaction();
table.insertRecords(50, true);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
Map<String, String> sourceHudiOptions =
Collections.singletonMap("hoodie.datasource.query.type", "read_optimized");
// Because compaction is not completed yet and read optimized query, there are 100 records.
@@ -342,13 +347,13 @@ public class ITOneTableClient {
HUDI, table, sourceHudiOptions, targetTableFormats, Collections.emptyMap(), 100);
table.insertRecords(50, true);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
// Because compaction is not completed yet and read optimized query, there are 150 records.
checkDatasetEquivalence(
HUDI, table, sourceHudiOptions, targetTableFormats, Collections.emptyMap(), 150);
table.completeScheduledCompaction(scheduledCompactionInstant);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, targetTableFormats, 130);
}
}
@@ -369,21 +374,23 @@ public class ITOneTableClient {
.tableDataPath(table.getDataPath())
.syncMode(SyncMode.INCREMENTAL)
.build();
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(sourceTableFormat);
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ ConversionSourceProvider<?> conversionSourceProvider =
+ getConversionSourceProvider(sourceTableFormat);
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
+ conversionController.sync(perTableConfig, conversionSourceProvider);
Instant instantAfterFirstSync = Instant.now();
// sleep before starting the next commit to avoid any rounding issues
Thread.sleep(1000);
table.insertRows(50);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
Instant instantAfterSecondSync = Instant.now();
// sleep before starting the next commit to avoid any rounding issues
Thread.sleep(1000);
table.insertRows(50);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(
sourceTableFormat,
@@ -470,7 +477,8 @@ public class ITOneTableClient {
Optional<String> hudiPartitionConfig = tableFormatPartitionDataHolder.getHudiSourceConfig();
String oneTablePartitionConfig = tableFormatPartitionDataHolder.getOneTablePartitionConfig();
String filter = tableFormatPartitionDataHolder.getFilter();
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(sourceTableFormat);
+ ConversionSourceProvider<?> conversionSourceProvider =
+ getConversionSourceProvider(sourceTableFormat);
GenericTable table;
if (hudiPartitionConfig.isPresent()) {
table =
@@ -494,11 +502,12 @@ public class ITOneTableClient {
.syncMode(SyncMode.INCREMENTAL)
.build();
tableToClose.insertRows(100);
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
+ conversionController.sync(perTableConfig, conversionSourceProvider);
// Do a second sync to force the test to read back the metadata it wrote earlier
tableToClose.insertRows(100);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalenceWithFilter(
sourceTableFormat, tableToClose, targetTableFormats, filter);
@@ -509,7 +518,7 @@ public class ITOneTableClient {
@EnumSource(value = SyncMode.class)
public void testSyncWithSingleFormat(SyncMode syncMode) {
String tableName = getTableName();
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(HUDI);
+ ConversionSourceProvider<?> conversionSourceProvider = getConversionSourceProvider(HUDI);
try (TestJavaHudiTable table =
TestJavaHudiTable.forStandardSchema(
tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
@@ -531,16 +540,17 @@ public class ITOneTableClient {
.syncMode(syncMode)
.build();
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
- oneTableClient.sync(perTableConfigIceberg, sourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
+ conversionController.sync(perTableConfigIceberg, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 100);
- oneTableClient.sync(perTableConfigDelta, sourceClientProvider);
+ conversionController.sync(perTableConfigDelta, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Collections.singletonList(DELTA), 100);
table.insertRecords(100, true);
- oneTableClient.sync(perTableConfigIceberg, sourceClientProvider);
+ conversionController.sync(perTableConfigIceberg, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 200);
- oneTableClient.sync(perTableConfigDelta, sourceClientProvider);
+ conversionController.sync(perTableConfigDelta, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Collections.singletonList(DELTA), 200);
}
}
@@ -548,7 +558,7 @@ public class ITOneTableClient {
@Test
public void testOutOfSyncIncrementalSyncs() {
String tableName = getTableName();
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(HUDI);
+ ConversionSourceProvider<?> conversionSourceProvider = getConversionSourceProvider(HUDI);
try (TestJavaHudiTable table =
TestJavaHudiTable.forStandardSchema(
tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
@@ -569,14 +579,15 @@ public class ITOneTableClient {
.build();
table.insertRecords(50, true);
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
// sync iceberg only
- oneTableClient.sync(singleTableConfig, sourceClientProvider);
+ conversionController.sync(singleTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 50);
// insert more records
table.insertRecords(50, true);
// iceberg will be an incremental sync and delta will need to bootstrap with snapshot sync
- oneTableClient.sync(dualTableConfig, sourceClientProvider);
+ conversionController.sync(dualTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Arrays.asList(ICEBERG, DELTA), 100);
// insert more records
@@ -584,13 +595,13 @@ public class ITOneTableClient {
// insert more records
table.insertRecords(50, true);
// incremental sync for two commits for iceberg only
- oneTableClient.sync(singleTableConfig, sourceClientProvider);
+ conversionController.sync(singleTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 200);
// insert more records
table.insertRecords(50, true);
// incremental sync for one commit for iceberg and three commits for delta
- oneTableClient.sync(dualTableConfig, sourceClientProvider);
+ conversionController.sync(dualTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Arrays.asList(ICEBERG, DELTA), 250);
}
}
@@ -598,12 +609,13 @@ public class ITOneTableClient {
@Test
public void testIcebergCorruptedSnapshotRecovery() throws Exception {
String tableName = getTableName();
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(HUDI);
+ ConversionSourceProvider<?> conversionSourceProvider = getConversionSourceProvider(HUDI);
try (TestJavaHudiTable table =
TestJavaHudiTable.forStandardSchema(
tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
table.insertRows(20);
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
PerTableConfig perTableConfig =
PerTableConfigImpl.builder()
.tableName(tableName)
@@ -611,11 +623,11 @@ public class ITOneTableClient {
.tableBasePath(table.getBasePath())
.syncMode(SyncMode.INCREMENTAL)
.build();
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
table.insertRows(10);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
table.insertRows(10);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
// corrupt last two snapshots
Table icebergTable = new HadoopTables(jsc.hadoopConfiguration()).load(table.getBasePath());
long currentSnapshotId = icebergTable.currentSnapshot().snapshotId();
@@ -625,7 +637,7 @@ public class ITOneTableClient {
Files.delete(
Paths.get(URI.create(icebergTable.snapshot(previousSnapshotId).manifestListLocation())));
table.insertRows(10);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 50);
}
}
@@ -633,7 +645,7 @@ public class ITOneTableClient {
@Test
public void testMetadataRetention() throws Exception {
String tableName = getTableName();
- SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(HUDI);
+ ConversionSourceProvider<?> conversionSourceProvider = getConversionSourceProvider(HUDI);
try (TestJavaHudiTable table =
TestJavaHudiTable.forStandardSchema(
tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
@@ -645,9 +657,10 @@ public class ITOneTableClient {
.syncMode(SyncMode.INCREMENTAL)
.targetMetadataRetentionInHours(0) // force cleanup
.build();
- OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
table.insertRecords(10, true);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
// later we will ensure we can still read the source table at this instant to ensure that
// neither target cleaned up the underlying parquet files in the table
Instant instantAfterFirstCommit = Instant.now();
@@ -658,7 +671,7 @@ public class ITOneTableClient {
.forEach(
unused -> {
table.insertRecords(10, true);
- oneTableClient.sync(perTableConfig, sourceClientProvider);
+ conversionController.sync(perTableConfig, conversionSourceProvider);
});
// ensure that hudi rows can still be read and underlying files were not removed
List<Row> rows =
diff --git a/core/src/test/java/org/apache/xtable/client/TestOneTableClient.java b/core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
similarity index 67%
rename from core/src/test/java/org/apache/xtable/client/TestOneTableClient.java
rename to core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
index 263013c7..c30d0878 100644
--- a/core/src/test/java/org/apache/xtable/client/TestOneTableClient.java
+++ b/core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import static org.apache.xtable.GenericTable.getTableName;
import static org.junit.jupiter.api.Assertions.*;
@@ -53,21 +53,21 @@ import org.apache.xtable.model.TableSyncMetadata;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
import org.apache.xtable.model.sync.SyncResult;
-import org.apache.xtable.spi.extractor.SourceClient;
+import org.apache.xtable.spi.extractor.ConversionSource;
+import org.apache.xtable.spi.sync.ConversionTarget;
import org.apache.xtable.spi.sync.TableFormatSync;
-import org.apache.xtable.spi.sync.TargetClient;
-public class TestOneTableClient {
+public class TestConversionController {
private final Configuration mockConf = mock(Configuration.class);
- private final SourceClientProvider<Instant> mockSourceClientProvider =
- mock(SourceClientProvider.class);
- private final SourceClient<Instant> mockSourceClient = mock(SourceClient.class);
- private final TableFormatClientFactory mockTableFormatClientFactory =
- mock(TableFormatClientFactory.class);
+ private final ConversionSourceProvider<Instant> mockConversionSourceProvider =
+ mock(ConversionSourceProvider.class);
+ private final ConversionSource<Instant> mockConversionSource = mock(ConversionSource.class);
+ private final ConversionTargetFactory mockConversionTargetFactory =
+ mock(ConversionTargetFactory.class);
private final TableFormatSync tableFormatSync = mock(TableFormatSync.class);
- private final TargetClient mockTargetClient1 = mock(TargetClient.class);
- private final TargetClient mockTargetClient2 = mock(TargetClient.class);
+ private final ConversionTarget mockConversionTarget1 = mock(ConversionTarget.class);
+ private final ConversionTarget mockConversionTarget2 = mock(ConversionTarget.class);
@Test
void testAllSnapshotSyncAsPerConfig() {
@@ -81,21 +81,21 @@ public class TestOneTableClient {
perTableResults.put(TableFormat.DELTA, syncResult);
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
- when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
- .thenReturn(mockSourceClient);
- when(mockTableFormatClientFactory.createForFormat(
- TableFormat.ICEBERG, perTableConfig, mockConf))
- .thenReturn(mockTargetClient1);
- when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
- .thenReturn(mockTargetClient2);
- when(mockSourceClient.getCurrentSnapshot()).thenReturn(internalSnapshot);
+ when(mockConversionSourceProvider.getConversionSourceInstance(perTableConfig))
+ .thenReturn(mockConversionSource);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.ICEBERG, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget1);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget2);
+ when(mockConversionSource.getCurrentSnapshot()).thenReturn(internalSnapshot);
when(tableFormatSync.syncSnapshot(
- argThat(containsAll(Arrays.asList(mockTargetClient1, mockTargetClient2))),
+ argThat(containsAll(Arrays.asList(mockConversionTarget1, mockConversionTarget2))),
eq(internalSnapshot)))
.thenReturn(perTableResults);
- OneTableClient oneTableClient =
- new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
- Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(mockConf, mockConversionTargetFactory, tableFormatSync);
+ Map<String, SyncResult> result =
+ conversionController.sync(perTableConfig, mockConversionSourceProvider);
assertEquals(perTableResults, result);
}
@@ -104,13 +104,12 @@ public class TestOneTableClient {
SyncMode syncMode = SyncMode.INCREMENTAL;
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
- when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
- .thenReturn(mockSourceClient);
- when(mockTableFormatClientFactory.createForFormat(
- TableFormat.ICEBERG, perTableConfig, mockConf))
- .thenReturn(mockTargetClient1);
- when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
- .thenReturn(mockTargetClient2);
+ when(mockConversionSourceProvider.getConversionSourceInstance(perTableConfig))
+ .thenReturn(mockConversionSource);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.ICEBERG, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget1);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget2);
Instant instantAsOfNow = Instant.now();
Instant instantAt15 = getInstantAtLastNMinutes(instantAsOfNow, 15);
@@ -119,8 +118,8 @@ public class TestOneTableClient {
Instant instantAt8 = getInstantAtLastNMinutes(instantAsOfNow, 8);
Instant instantAt5 = getInstantAtLastNMinutes(instantAsOfNow, 5);
Instant instantAt2 = getInstantAtLastNMinutes(instantAsOfNow, 2);
- when(mockSourceClient.isIncrementalSyncSafeFrom(eq(instantAt15))).thenReturn(true);
- when(mockSourceClient.isIncrementalSyncSafeFrom(eq(instantAt8))).thenReturn(true);
+ when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt15))).thenReturn(true);
+ when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt8))).thenReturn(true);
// Iceberg last synced at instantAt10 and has pending instants at instantAt15.
Instant icebergLastSyncInstant = instantAt10;
@@ -140,18 +139,19 @@ public class TestOneTableClient {
Arrays.asList(instantAt15, instantAt14, instantAt10, instantAt8, instantAt5, instantAt2);
CommitsBacklog<Instant> commitsBacklog =
CommitsBacklog.<Instant>builder().commitsToProcess(instantsToProcess).build();
- Optional<TableSyncMetadata> targetClient1Metadata =
+ Optional<TableSyncMetadata> conversionTarget1Metadata =
Optional.of(TableSyncMetadata.of(icebergLastSyncInstant, pendingInstantsForIceberg));
- when(mockTargetClient1.getTableMetadata()).thenReturn(targetClient1Metadata);
- Optional<TableSyncMetadata> targetClient2Metadata =
+ when(mockConversionTarget1.getTableMetadata()).thenReturn(conversionTarget1Metadata);
+ Optional<TableSyncMetadata> conversionTarget2Metadata =
Optional.of(TableSyncMetadata.of(deltaLastSyncInstant, pendingInstantsForDelta));
- when(mockTargetClient2.getTableMetadata()).thenReturn(targetClient2Metadata);
- when(mockSourceClient.getCommitsBacklog(instantsForIncrementalSync)).thenReturn(commitsBacklog);
+ when(mockConversionTarget2.getTableMetadata()).thenReturn(conversionTarget2Metadata);
+ when(mockConversionSource.getCommitsBacklog(instantsForIncrementalSync))
+ .thenReturn(commitsBacklog);
List<TableChange> tableChanges = new ArrayList<>();
for (Instant instant : instantsToProcess) {
TableChange tableChange = getTableChange(instant);
tableChanges.add(tableChange);
- when(mockSourceClient.getTableChangeForCommit(instant)).thenReturn(tableChange);
+ when(mockConversionSource.getTableChangeForCommit(instant)).thenReturn(tableChange);
}
// Iceberg needs to sync last pending instant at instantAt15 and instants after last sync
// instant
@@ -166,18 +166,19 @@ public class TestOneTableClient {
Map<String, List<SyncResult>> allResults = new HashMap<>();
allResults.put(TableFormat.ICEBERG, icebergSyncResults);
allResults.put(TableFormat.DELTA, deltaSyncResults);
- Map<TargetClient, TableSyncMetadata> clientToMetadata = new HashMap<>();
- clientToMetadata.put(mockTargetClient1, targetClient1Metadata.get());
- clientToMetadata.put(mockTargetClient2, targetClient2Metadata.get());
+ Map<ConversionTarget, TableSyncMetadata> conversionTargetToMetadata = new HashMap<>();
+ conversionTargetToMetadata.put(mockConversionTarget1, conversionTarget1Metadata.get());
+ conversionTargetToMetadata.put(mockConversionTarget2, conversionTarget2Metadata.get());
when(tableFormatSync.syncChanges(
- eq(clientToMetadata), argThat(matches(incrementalTableChanges))))
+ eq(conversionTargetToMetadata), argThat(matches(incrementalTableChanges))))
.thenReturn(allResults);
Map<String, SyncResult> expectedSyncResult = new HashMap<>();
expectedSyncResult.put(TableFormat.ICEBERG, getLastSyncResult(icebergSyncResults));
expectedSyncResult.put(TableFormat.DELTA, getLastSyncResult(deltaSyncResults));
- OneTableClient oneTableClient =
- new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
- Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(mockConf, mockConversionTargetFactory, tableFormatSync);
+ Map<String, SyncResult> result =
+ conversionController.sync(perTableConfig, mockConversionSourceProvider);
assertEquals(expectedSyncResult, result);
}
@@ -193,32 +194,32 @@ public class TestOneTableClient {
syncResults.put(TableFormat.DELTA, syncResult);
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
- when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
- .thenReturn(mockSourceClient);
- when(mockTableFormatClientFactory.createForFormat(
- TableFormat.ICEBERG, perTableConfig, mockConf))
- .thenReturn(mockTargetClient1);
- when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
- .thenReturn(mockTargetClient2);
+ when(mockConversionSourceProvider.getConversionSourceInstance(perTableConfig))
+ .thenReturn(mockConversionSource);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.ICEBERG, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget1);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget2);
Instant instantAsOfNow = Instant.now();
Instant instantAt5 = getInstantAtLastNMinutes(instantAsOfNow, 5);
- when(mockSourceClient.isIncrementalSyncSafeFrom(eq(instantAt5))).thenReturn(false);
+ when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt5))).thenReturn(false);
// Both Iceberg and Delta last synced at instantAt5 and have no pending instants.
- when(mockTargetClient1.getTableMetadata())
+ when(mockConversionTarget1.getTableMetadata())
.thenReturn(Optional.of(TableSyncMetadata.of(instantAt5, Collections.emptyList())));
- when(mockTargetClient2.getTableMetadata())
+ when(mockConversionTarget2.getTableMetadata())
.thenReturn(Optional.of(TableSyncMetadata.of(instantAt5, Collections.emptyList())));
- when(mockSourceClient.getCurrentSnapshot()).thenReturn(internalSnapshot);
+ when(mockConversionSource.getCurrentSnapshot()).thenReturn(internalSnapshot);
when(tableFormatSync.syncSnapshot(
- argThat(containsAll(Arrays.asList(mockTargetClient1, mockTargetClient2))),
+ argThat(containsAll(Arrays.asList(mockConversionTarget1, mockConversionTarget2))),
eq(internalSnapshot)))
.thenReturn(syncResults);
- OneTableClient oneTableClient =
- new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
- Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(mockConf, mockConversionTargetFactory, tableFormatSync);
+ Map<String, SyncResult> result =
+ conversionController.sync(perTableConfig, mockConversionSourceProvider);
assertEquals(syncResults, result);
}
@@ -227,13 +228,12 @@ public class TestOneTableClient {
SyncMode syncMode = SyncMode.INCREMENTAL;
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
- when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
- .thenReturn(mockSourceClient);
- when(mockTableFormatClientFactory.createForFormat(
- TableFormat.ICEBERG, perTableConfig, mockConf))
- .thenReturn(mockTargetClient1);
- when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
- .thenReturn(mockTargetClient2);
+ when(mockConversionSourceProvider.getConversionSourceInstance(perTableConfig))
+ .thenReturn(mockConversionSource);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.ICEBERG, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget1);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget2);
Instant instantAsOfNow = Instant.now();
Instant instantAt15 = getInstantAtLastNMinutes(instantAsOfNow, 15);
@@ -242,8 +242,8 @@ public class TestOneTableClient {
Instant instantAt5 = getInstantAtLastNMinutes(instantAsOfNow, 5);
Instant instantAt2 = getInstantAtLastNMinutes(instantAsOfNow, 2);
- when(mockSourceClient.isIncrementalSyncSafeFrom(eq(instantAt8))).thenReturn(true);
- when(mockSourceClient.isIncrementalSyncSafeFrom(eq(instantAt15))).thenReturn(false);
+ when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt8))).thenReturn(true);
+ when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt15))).thenReturn(false);
// Iceberg last synced at instantAt10 and has pending instants at instantAt15.
Instant icebergLastSyncInstant = instantAt10;
@@ -259,18 +259,19 @@ public class TestOneTableClient {
List<Instant> instantsToProcess = Arrays.asList(instantAt8, instantAt2);
CommitsBacklog<Instant> commitsBacklog =
CommitsBacklog.<Instant>builder().commitsToProcess(instantsToProcess).build();
- when(mockTargetClient1.getTableMetadata())
+ when(mockConversionTarget1.getTableMetadata())
.thenReturn(
Optional.of(TableSyncMetadata.of(icebergLastSyncInstant, pendingInstantsForIceberg)));
- Optional<TableSyncMetadata> targetClient2Metadata =
+ Optional<TableSyncMetadata> conversionTarget2Metadata =
Optional.of(TableSyncMetadata.of(deltaLastSyncInstant, pendingInstantsForDelta));
- when(mockTargetClient2.getTableMetadata()).thenReturn(targetClient2Metadata);
- when(mockSourceClient.getCommitsBacklog(instantsForIncrementalSync)).thenReturn(commitsBacklog);
+ when(mockConversionTarget2.getTableMetadata()).thenReturn(conversionTarget2Metadata);
+ when(mockConversionSource.getCommitsBacklog(instantsForIncrementalSync))
+ .thenReturn(commitsBacklog);
List<TableChange> tableChanges = new ArrayList<>();
for (Instant instant : instantsToProcess) {
TableChange tableChange = getTableChange(instant);
tableChanges.add(tableChange);
- when(mockSourceClient.getTableChangeForCommit(instant)).thenReturn(tableChange);
+ when(mockConversionSource.getTableChangeForCommit(instant)).thenReturn(tableChange);
}
// Iceberg needs to sync by snapshot since instant15 is affected by table clean-up.
InternalTable internalTable = getOneTable();
@@ -279,9 +280,9 @@ public class TestOneTableClient {
SyncResult syncResult = buildSyncResult(syncMode, instantBeforeHour);
Map<String, SyncResult> snapshotResult =
Collections.singletonMap(TableFormat.ICEBERG, syncResult);
- when(mockSourceClient.getCurrentSnapshot()).thenReturn(internalSnapshot);
+ when(mockConversionSource.getCurrentSnapshot()).thenReturn(internalSnapshot);
when(tableFormatSync.syncSnapshot(
- argThat(containsAll(Collections.singletonList(mockTargetClient1))),
+ argThat(containsAll(Collections.singletonList(mockConversionTarget1))),
eq(internalSnapshot)))
.thenReturn(snapshotResult);
// Delta needs to sync last pending instant at instantAt8 and instants after last sync instant
@@ -290,15 +291,16 @@ public class TestOneTableClient {
IncrementalTableChanges incrementalTableChanges =
IncrementalTableChanges.builder().tableChanges(tableChanges.iterator()).build();
when(tableFormatSync.syncChanges(
- eq(Collections.singletonMap(mockTargetClient2, targetClient2Metadata.get())),
+ eq(Collections.singletonMap(mockConversionTarget2, conversionTarget2Metadata.get())),
argThat(matches(incrementalTableChanges))))
.thenReturn(Collections.singletonMap(TableFormat.DELTA, deltaSyncResults));
Map<String, SyncResult> expectedSyncResult = new HashMap<>();
expectedSyncResult.put(TableFormat.ICEBERG, syncResult);
expectedSyncResult.put(TableFormat.DELTA, getLastSyncResult(deltaSyncResults));
- OneTableClient oneTableClient =
- new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
- Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(mockConf, mockConversionTargetFactory, tableFormatSync);
+ Map<String, SyncResult> result =
+ conversionController.sync(perTableConfig, mockConversionSourceProvider);
assertEquals(expectedSyncResult, result);
}
@@ -307,19 +309,18 @@ public class TestOneTableClient {
SyncMode syncMode = SyncMode.INCREMENTAL;
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
- when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
- .thenReturn(mockSourceClient);
- when(mockTableFormatClientFactory.createForFormat(
- TableFormat.ICEBERG, perTableConfig, mockConf))
- .thenReturn(mockTargetClient1);
- when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
- .thenReturn(mockTargetClient2);
+ when(mockConversionSourceProvider.getConversionSourceInstance(perTableConfig))
+ .thenReturn(mockConversionSource);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.ICEBERG, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget1);
+ when(mockConversionTargetFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+ .thenReturn(mockConversionTarget2);
Instant instantAsOfNow = Instant.now();
Instant instantAt5 = getInstantAtLastNMinutes(instantAsOfNow, 5);
- when(mockSourceClient.isIncrementalSyncSafeFrom(eq(instantAt5))).thenReturn(true);
- when(mockSourceClient.isIncrementalSyncSafeFrom(eq(instantAt5))).thenReturn(true);
+ when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt5))).thenReturn(true);
+ when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt5))).thenReturn(true);
// Iceberg last synced at instantAt5, the last instant in the source
Instant icebergLastSyncInstant = instantAt5;
@@ -330,18 +331,19 @@ public class TestOneTableClient {
List<Instant> instantsToProcess = Collections.emptyList();
CommitsBacklog<Instant> commitsBacklog =
CommitsBacklog.<Instant>builder().commitsToProcess(instantsToProcess).build();
- Optional<TableSyncMetadata> targetClient1Metadata =
+ Optional<TableSyncMetadata> conversionTarget1Metadata =
Optional.of(TableSyncMetadata.of(icebergLastSyncInstant, Collections.emptyList()));
- when(mockTargetClient1.getTableMetadata()).thenReturn(targetClient1Metadata);
- Optional<TableSyncMetadata> targetClient2Metadata =
+ when(mockConversionTarget1.getTableMetadata()).thenReturn(conversionTarget1Metadata);
+ Optional<TableSyncMetadata> conversionTarget2Metadata =
Optional.of(TableSyncMetadata.of(deltaLastSyncInstant, Collections.emptyList()));
- when(mockTargetClient2.getTableMetadata()).thenReturn(targetClient2Metadata);
- when(mockSourceClient.getCommitsBacklog(instantsForIncrementalSync)).thenReturn(commitsBacklog);
- Map<TargetClient, TableSyncMetadata> clientsWithMetadata = new HashMap<>();
- clientsWithMetadata.put(mockTargetClient1, targetClient1Metadata.get());
- clientsWithMetadata.put(mockTargetClient2, targetClient2Metadata.get());
+ when(mockConversionTarget2.getTableMetadata()).thenReturn(conversionTarget2Metadata);
+ when(mockConversionSource.getCommitsBacklog(instantsForIncrementalSync))
+ .thenReturn(commitsBacklog);
+ Map<ConversionTarget, TableSyncMetadata> conversionTargetWithMetadata = new HashMap<>();
+ conversionTargetWithMetadata.put(mockConversionTarget1, conversionTarget1Metadata.get());
+ conversionTargetWithMetadata.put(mockConversionTarget2, conversionTarget2Metadata.get());
when(tableFormatSync.syncChanges(
- eq(clientsWithMetadata),
+ eq(conversionTargetWithMetadata),
argThat(
matches(
IncrementalTableChanges.builder()
@@ -350,9 +352,10 @@ public class TestOneTableClient {
.thenReturn(Collections.emptyMap());
// Iceberg and Delta have no commits to sync
Map<String, SyncResult> expectedSyncResult = Collections.emptyMap();
- OneTableClient oneTableClient =
- new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
- Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+ ConversionController conversionController =
+ new ConversionController(mockConf, mockConversionTargetFactory, tableFormatSync);
+ Map<String, SyncResult> result =
+ conversionController.sync(perTableConfig, mockConversionSourceProvider);
assertEquals(expectedSyncResult, result);
}
diff --git a/core/src/test/java/org/apache/xtable/client/TestTableFormatClientFactory.java b/core/src/test/java/org/apache/xtable/conversion/TestConversionTargetFactory.java
similarity index 77%
rename from core/src/test/java/org/apache/xtable/client/TestTableFormatClientFactory.java
rename to core/src/test/java/org/apache/xtable/conversion/TestConversionTargetFactory.java
index 9a7e4175..34d3cbef 100644
--- a/core/src/test/java/org/apache/xtable/client/TestTableFormatClientFactory.java
+++ b/core/src/test/java/org/apache/xtable/conversion/TestConversionTargetFactory.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import static org.apache.xtable.GenericTable.getTableName;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -33,14 +33,14 @@ import org.junit.jupiter.api.Test;
import org.apache.xtable.exception.NotSupportedException;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
-import org.apache.xtable.spi.sync.TargetClient;
+import org.apache.xtable.spi.sync.ConversionTarget;
-public class TestTableFormatClientFactory {
+public class TestConversionTargetFactory {
@Test
- public void testTableClientFromNameForDELTA() {
- TargetClient tc =
- TableFormatClientFactory.getInstance().createTargetClientForName(TableFormat.DELTA);
+ public void testConversionTargetFromNameForDELTA() {
+ ConversionTarget tc =
+ ConversionTargetFactory.getInstance().createConversionTargetForName(TableFormat.DELTA);
assertNotNull(tc);
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.DELTA), SyncMode.INCREMENTAL);
@@ -51,9 +51,9 @@ public class TestTableFormatClientFactory {
}
@Test
- public void testTableClientFromNameForHUDI() {
- TargetClient tc =
- TableFormatClientFactory.getInstance().createTargetClientForName(TableFormat.HUDI);
+ public void testConversionTargetFromNameForHUDI() {
+ ConversionTarget tc =
+ ConversionTargetFactory.getInstance().createConversionTargetForName(TableFormat.HUDI);
assertNotNull(tc);
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.HUDI), SyncMode.INCREMENTAL);
@@ -64,9 +64,9 @@ public class TestTableFormatClientFactory {
}
@Test
- public void testTableClientFromNameForICEBERG() {
- TargetClient tc =
- TableFormatClientFactory.getInstance().createTargetClientForName(TableFormat.ICEBERG);
+ public void testConversionTargetFromNameForICEBERG() {
+ ConversionTarget tc =
+ ConversionTargetFactory.getInstance().createConversionTargetForName(TableFormat.ICEBERG);
assertNotNull(tc);
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.ICEBERG), SyncMode.INCREMENTAL);
@@ -77,23 +77,23 @@ public class TestTableFormatClientFactory {
}
@Test
- public void testTableClientFromNameForUNKOWN() {
+ public void testConversionTargetFromNameForUNKOWN() {
NotSupportedException thrown =
assertThrows(
NotSupportedException.class,
- () -> TableFormatClientFactory.getInstance().createTargetClientForName("UNKNOWN"),
+ () -> ConversionTargetFactory.getInstance().createConversionTargetForName("UNKNOWN"),
"NotSupportedException expected and operation succeeded inappropriately.");
assertTrue(thrown.getMessage().contains("UNKNOWN"));
}
@Test
- public void testTableClientFromFormatType() {
+ public void testConversionTargetFromFormatType() {
PerTableConfig perTableConfig =
getPerTableConfig(Arrays.asList(TableFormat.DELTA), SyncMode.INCREMENTAL);
Configuration conf = new Configuration();
conf.setStrings("spark.master", "local");
- TargetClient tc =
- TableFormatClientFactory.getInstance()
+ ConversionTarget tc =
+ ConversionTargetFactory.getInstance()
.createForFormat(TableFormat.DELTA, perTableConfig, conf);
assertEquals(tc.getTableFormat(), TableFormat.DELTA);
}
diff --git a/core/src/test/java/org/apache/xtable/client/TestPerTableConfig.java b/core/src/test/java/org/apache/xtable/conversion/TestPerTableConfig.java
similarity index 99%
rename from core/src/test/java/org/apache/xtable/client/TestPerTableConfig.java
rename to core/src/test/java/org/apache/xtable/conversion/TestPerTableConfig.java
index 1687cd1c..4f057385 100644
--- a/core/src/test/java/org/apache/xtable/client/TestPerTableConfig.java
+++ b/core/src/test/java/org/apache/xtable/conversion/TestPerTableConfig.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.xtable.client;
+package org.apache.xtable.conversion;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
diff --git a/core/src/test/java/org/apache/xtable/delta/ITDeltaSourceClient.java b/core/src/test/java/org/apache/xtable/delta/ITDeltaConversionTargetSource.java
similarity index 89%
rename from core/src/test/java/org/apache/xtable/delta/ITDeltaSourceClient.java
rename to core/src/test/java/org/apache/xtable/delta/ITDeltaConversionTargetSource.java
index 3a0f589a..6a29bd18 100644
--- a/core/src/test/java/org/apache/xtable/delta/ITDeltaSourceClient.java
+++ b/core/src/test/java/org/apache/xtable/delta/ITDeltaConversionTargetSource.java
@@ -53,8 +53,8 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.apache.xtable.GenericTable;
import org.apache.xtable.TestSparkDeltaTable;
import org.apache.xtable.ValidationTestHelper;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.PerTableConfigImpl;
+import org.apache.xtable.conversion.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfigImpl;
import org.apache.xtable.model.CommitsBacklog;
import org.apache.xtable.model.InstantsForIncrementalSync;
import org.apache.xtable.model.InternalSnapshot;
@@ -73,7 +73,7 @@ import org.apache.xtable.model.stat.Range;
import org.apache.xtable.model.storage.*;
import org.apache.xtable.model.storage.InternalDataFile;
-public class ITDeltaSourceClient {
+public class ITDeltaConversionTargetSource {
private static final InternalField COL1_INT_FIELD =
InternalField.builder()
@@ -118,7 +118,7 @@ public class ITDeltaSourceClient {
@TempDir private static Path tempDir;
private static SparkSession sparkSession;
- private DeltaSourceClientProvider clientProvider;
+ private DeltaConversionSourceProvider conversionSourceProvider;
@BeforeAll
public static void setupOnce() {
@@ -150,8 +150,8 @@ public class ITDeltaSourceClient {
Configuration hadoopConf = new Configuration();
hadoopConf.set("fs.defaultFS", "file:///");
- clientProvider = new DeltaSourceClientProvider();
- clientProvider.init(hadoopConf, null);
+ conversionSourceProvider = new DeltaConversionSourceProvider();
+ conversionSourceProvider.init(hadoopConf, null);
}
@Test
@@ -166,16 +166,17 @@ public class ITDeltaSourceClient {
+ "` USING DELTA LOCATION '"
+ basePath
+ "' AS SELECT * FROM VALUES (1, 2)");
- // Create Delta source client
+ // Create Delta source
PerTableConfig tableConfig =
PerTableConfigImpl.builder()
.tableName(tableName)
.tableBasePath(basePath.toString())
.targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
.build();
- DeltaSourceClient client = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
// Get current snapshot
- InternalSnapshot snapshot = client.getCurrentSnapshot();
+ InternalSnapshot snapshot = conversionSource.getCurrentSnapshot();
// Validate table
List<InternalField> fields = Arrays.asList(COL1_INT_FIELD, COL2_INT_FIELD);
validateTable(
@@ -228,16 +229,17 @@ public class ITDeltaSourceClient {
+ "LOCATION '"
+ basePath
+ "' AS SELECT 'SingleValue' AS part_col, 1 AS col1, 2 AS col2");
- // Create Delta source client
+ // Create Delta source
PerTableConfig tableConfig =
PerTableConfigImpl.builder()
.tableName(tableName)
.tableBasePath(basePath.toString())
.targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
.build();
- DeltaSourceClient client = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
// Get current snapshot
- InternalSnapshot snapshot = client.getCurrentSnapshot();
+ InternalSnapshot snapshot = conversionSource.getCurrentSnapshot();
// Validate table
InternalField partCol =
InternalField.builder()
@@ -320,16 +322,17 @@ public class ITDeltaSourceClient {
"INSERT INTO TABLE `"
+ tableName
+ "` VALUES(1, CAST('2012-02-12 00:12:34' AS TIMESTAMP))");
- // Create Delta source client
+ // Create Delta source
PerTableConfig tableConfig =
PerTableConfigImpl.builder()
.tableName(tableName)
.tableBasePath(basePath.toString())
.targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
.build();
- DeltaSourceClient client = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
// Get current snapshot
- InternalSnapshot snapshot = client.getCurrentSnapshot();
+ InternalSnapshot snapshot = conversionSource.getCurrentSnapshot();
}
@ParameterizedTest
@@ -365,9 +368,10 @@ public class ITDeltaSourceClient {
.tableBasePath(testSparkDeltaTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
.build();
- DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(180L, testSparkDeltaTable.getNumRows());
- InternalSnapshot internalSnapshot = deltaSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
if (isPartitioned) {
validateDeltaPartitioning(internalSnapshot);
@@ -380,9 +384,9 @@ public class ITDeltaSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Long> commitsBacklog =
- deltaSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Long version : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = deltaSourceClient.getTableChangeForCommit(version);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(version);
allTableChanges.add(tableChange);
}
ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
@@ -404,8 +408,9 @@ public class ITDeltaSourceClient {
.tableBasePath(testSparkDeltaTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
.build();
- DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
- InternalSnapshot snapshotAfterCommit1 = deltaSourceClient.getCurrentSnapshot();
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
+ InternalSnapshot snapshotAfterCommit1 = conversionSource.getCurrentSnapshot();
List<String> allActivePaths = ValidationTestHelper.getAllFilePaths(snapshotAfterCommit1);
assertEquals(1, allActivePaths.size());
String activePathAfterCommit1 = allActivePaths.get(0);
@@ -423,19 +428,19 @@ public class ITDeltaSourceClient {
InstantsForIncrementalSync.builder()
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
- deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ conversionSource = conversionSourceProvider.getConversionSourceInstance(tableConfig);
CommitsBacklog<Long> instantCurrentCommitState =
- deltaSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
boolean areFilesRemoved = false;
for (Long version : instantCurrentCommitState.getCommitsToProcess()) {
- TableChange tableChange = deltaSourceClient.getTableChangeForCommit(version);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(version);
areFilesRemoved = areFilesRemoved | checkIfFileIsRemoved(activePathAfterCommit1, tableChange);
}
assertTrue(areFilesRemoved);
- assertTrue(deltaSourceClient.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)));
+ assertTrue(conversionSource.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)));
// Table doesn't have instant of this older commit, hence it is not safe.
Instant instantAsOfHourAgo = Instant.now().minus(1, ChronoUnit.HOURS);
- assertFalse(deltaSourceClient.isIncrementalSyncSafeFrom(instantAsOfHourAgo));
+ assertFalse(conversionSource.isIncrementalSyncSafeFrom(instantAsOfHourAgo));
}
@ParameterizedTest
@@ -471,9 +476,10 @@ public class ITDeltaSourceClient {
.tableBasePath(testSparkDeltaTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
.build();
- DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(130L, testSparkDeltaTable.getNumRows());
- InternalSnapshot internalSnapshot = deltaSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
if (isPartitioned) {
validateDeltaPartitioning(internalSnapshot);
}
@@ -485,9 +491,9 @@ public class ITDeltaSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Long> commitsBacklog =
- deltaSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Long version : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = deltaSourceClient.getTableChangeForCommit(version);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(version);
allTableChanges.add(tableChange);
}
ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
@@ -518,9 +524,10 @@ public class ITDeltaSourceClient {
.tableBasePath(testSparkDeltaTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
.build();
- DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(150L, testSparkDeltaTable.getNumRows());
- InternalSnapshot internalSnapshot = deltaSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
if (isPartitioned) {
validateDeltaPartitioning(internalSnapshot);
}
@@ -532,9 +539,9 @@ public class ITDeltaSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Long> commitsBacklog =
- deltaSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Long version : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = deltaSourceClient.getTableChangeForCommit(version);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(version);
allTableChanges.add(tableChange);
}
ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
@@ -574,10 +581,11 @@ public class ITDeltaSourceClient {
.tableBasePath(testSparkDeltaTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
.build();
- DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(
120 - rowsByPartition.get(partitionValueToDelete).size(), testSparkDeltaTable.getNumRows());
- InternalSnapshot internalSnapshot = deltaSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
validateDeltaPartitioning(internalSnapshot);
ValidationTestHelper.validateSnapshot(
@@ -588,9 +596,9 @@ public class ITDeltaSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Long> commitsBacklog =
- deltaSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Long version : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = deltaSourceClient.getTableChangeForCommit(version);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(version);
allTableChanges.add(tableChange);
}
ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
@@ -633,9 +641,10 @@ public class ITDeltaSourceClient {
.tableBasePath(testSparkDeltaTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
.build();
- DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ DeltaConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(250L, testSparkDeltaTable.getNumRows());
- InternalSnapshot internalSnapshot = deltaSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
if (isPartitioned) {
validateDeltaPartitioning(internalSnapshot);
}
@@ -647,9 +656,9 @@ public class ITDeltaSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Long> commitsBacklog =
- deltaSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Long version : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = deltaSourceClient.getTableChangeForCommit(version);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(version);
allTableChanges.add(tableChange);
}
ValidationTestHelper.validateTableChanges(allActiveFiles, allTableChanges);
diff --git a/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java b/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java
index b248a630..514cc394 100644
--- a/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java
+++ b/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java
@@ -76,7 +76,7 @@ import io.delta.standalone.expressions.Literal;
import io.delta.standalone.types.IntegerType;
import io.delta.standalone.types.StringType;
-import org.apache.xtable.client.PerTableConfigImpl;
+import org.apache.xtable.conversion.PerTableConfigImpl;
import org.apache.xtable.model.InternalSnapshot;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.schema.InternalField;
@@ -105,7 +105,7 @@ public class TestDeltaSync {
private static SparkSession sparkSession;
@TempDir public static java.nio.file.Path tempDir;
- private DeltaClient deltaClient;
+ private DeltaConversionTarget conversionTarget;
private Path basePath;
private String tableName;
@@ -124,8 +124,8 @@ public class TestDeltaSync {
tableName = "test-" + UUID.randomUUID();
basePath = tempDir.resolve(tableName);
Files.createDirectories(basePath);
- deltaClient =
- new DeltaClient(
+ conversionTarget =
+ new DeltaConversionTarget(
PerTableConfigImpl.builder()
.tableName(tableName)
.tableBasePath(basePath.toString())
@@ -160,10 +160,12 @@ public class TestDeltaSync {
InternalSnapshot snapshot1 = buildSnapshot(table1, dataFile1, dataFile2);
InternalSnapshot snapshot2 = buildSnapshot(table2, dataFile2, dataFile3);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(deltaClient), snapshot1);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile1, dataFile2)), null);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(deltaClient), snapshot2);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot2);
validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile2, dataFile3)), null);
}
@@ -212,7 +214,8 @@ public class TestDeltaSync {
InternalSnapshot snapshot1 = buildSnapshot(table, dataFile1, dataFile2, dataFile3);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(deltaClient), snapshot1);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile3)), equalToExpr);
}
@@ -290,7 +293,8 @@ public class TestDeltaSync {
And CombinedExpr = new And(equalToExpr1, equalToExpr2);
InternalSnapshot snapshot1 = buildSnapshot(table, dataFile1, dataFile2, dataFile3);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(deltaClient), snapshot1);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
validateDeltaTable(basePath, new HashSet<>(Arrays.asList(dataFile2)), CombinedExpr);
}
@@ -333,7 +337,8 @@ public class TestDeltaSync {
InternalSnapshot snapshot1 = buildSnapshot(table, dataFile1, dataFile2, dataFile3);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(deltaClient), snapshot1);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
Dataset<Row> dataset = sparkSession.read().format("delta").load(basePath.toString());
org.apache.spark.sql.catalyst.expressions.Expression expression =
diff --git a/core/src/test/java/org/apache/xtable/hudi/ITHudiSourceClient.java b/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceSource.java
similarity index 98%
rename from core/src/test/java/org/apache/xtable/hudi/ITHudiSourceClient.java
rename to core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceSource.java
index 2f782f8b..241a33a7 100644
--- a/core/src/test/java/org/apache/xtable/hudi/ITHudiSourceClient.java
+++ b/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceSource.java
@@ -70,7 +70,7 @@ import org.apache.xtable.model.TableChange;
/**
* A suite of functional tests that the extraction from Hudi to Intermediate representation works.
*/
-public class ITHudiSourceClient {
+public class ITHudiConversionSourceSource {
@TempDir public static Path tempDir;
private static JavaSparkContext jsc;
private static SparkSession sparkSession;
@@ -133,7 +133,7 @@ public class ITHudiSourceClient {
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
}
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(
CONFIGURATION, table.getBasePath(), partitionConfig.getOneTableConfig());
// Get the current snapshot
@@ -181,7 +181,7 @@ public class ITHudiSourceClient {
table.deleteRecords(insertsForCommit1.subList(15, 30), true);
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(
CONFIGURATION, table.getBasePath(), partitionConfig.getOneTableConfig());
// Get the current snapshot
@@ -223,7 +223,7 @@ public class ITHudiSourceClient {
table.clean(); // cleans up file groups from commitInstant1
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(
CONFIGURATION, table.getBasePath(), partitionConfig.getOneTableConfig());
// commitInstant1 is not safe for incremental sync as cleaner has run after and touched
@@ -270,7 +270,7 @@ public class ITHudiSourceClient {
table.insertRecords(20, partitionToDelete, true);
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(CONFIGURATION, table.getBasePath(), "level:VALUE");
// Get the current snapshot
InternalSnapshot internalSnapshot = hudiClient.getCurrentSnapshot();
@@ -322,7 +322,7 @@ public class ITHudiSourceClient {
table.insertRecords(20, selectedPartition, true);
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(CONFIGURATION, table.getBasePath(), "level:VALUE");
// Get the current snapshot
InternalSnapshot internalSnapshot = hudiClient.getCurrentSnapshot();
@@ -388,7 +388,7 @@ public class ITHudiSourceClient {
table.insertRecords(100, true);
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(
CONFIGURATION, table.getBasePath(), partitionConfig.getOneTableConfig());
// Get the current snapshot
@@ -443,7 +443,7 @@ public class ITHudiSourceClient {
table.insertRecords(50, true);
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(
CONFIGURATION, table.getBasePath(), partitionConfig.getOneTableConfig());
// Get the current snapshot
@@ -509,7 +509,7 @@ public class ITHudiSourceClient {
table.insertRecordsWithCommitAlreadyStarted(insertsForCommit4, commitInstant4, true);
List<String> baseFilesAfterCommit4 = table.getAllLatestBaseFilePaths();
- HudiClient hudiClient =
+ HudiConversionSource hudiClient =
getHudiSourceClient(
CONFIGURATION, table.getBasePath(), partitionConfig.getOneTableConfig());
// Get the current snapshot
@@ -559,7 +559,7 @@ public class ITHudiSourceClient {
tableType -> partitionConfigs.stream().map(config -> Arguments.of(tableType, config)));
}
- private HudiClient getHudiSourceClient(
+ private HudiConversionSource getHudiSourceClient(
Configuration conf, String basePath, String onetablePartitionConfig) {
HoodieTableMetaClient hoodieTableMetaClient =
HoodieTableMetaClient.builder()
@@ -572,7 +572,7 @@ public class ITHudiSourceClient {
HudiSourceConfigImpl.builder()
.partitionFieldSpecConfig(onetablePartitionConfig)
.build());
- return new HudiClient(hoodieTableMetaClient, partitionSpecExtractor);
+ return new HudiConversionSource(hoodieTableMetaClient, partitionSpecExtractor);
}
private boolean checkIfNewFileGroupIsAdded(String activePath, TableChange tableChange) {
diff --git a/core/src/test/java/org/apache/xtable/hudi/ITHudiTargetClient.java b/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceTarget.java
similarity index 97%
rename from core/src/test/java/org/apache/xtable/hudi/ITHudiTargetClient.java
rename to core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceTarget.java
index ebe0d6bc..a0d51b7a 100644
--- a/core/src/test/java/org/apache/xtable/hudi/ITHudiTargetClient.java
+++ b/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceTarget.java
@@ -70,7 +70,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.metadata.HoodieBackedTableMetadata;
import org.apache.hudi.metadata.HoodieMetadataFileSystemView;
-import org.apache.xtable.client.PerTableConfigImpl;
+import org.apache.xtable.conversion.PerTableConfigImpl;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.TableSyncMetadata;
import org.apache.xtable.model.schema.InternalField;
@@ -87,13 +87,13 @@ import org.apache.xtable.model.storage.FileFormat;
import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.PartitionFileGroup;
import org.apache.xtable.model.storage.TableFormat;
-import org.apache.xtable.spi.sync.TargetClient;
+import org.apache.xtable.spi.sync.ConversionTarget;
/**
* A suite of functional tests that assert that the metadata for the hudi table is properly written
* to disk.
*/
-public class ITHudiTargetClient {
+public class ITHudiConversionSourceTarget {
@TempDir public static Path tempDir;
private static final Configuration CONFIGURATION = new Configuration();
private static final HoodieEngineContext CONTEXT = new HoodieJavaEngineContext(CONFIGURATION);
@@ -199,7 +199,7 @@ public class ITHudiTargetClient {
.fileRemoved(fileToRemove)
.build();
// perform sync
- HudiTargetClient targetClient = getTargetClient();
+ HudiConversionTarget targetClient = getTargetClient();
InternalTable initialState = getState(Instant.now());
targetClient.beginSync(initialState);
targetClient.syncFilesForDiff(dataFilesDiff);
@@ -240,7 +240,7 @@ public class ITHudiTargetClient {
.build());
// sync snapshot and metadata
InternalTable initialState = getState(Instant.now());
- HudiTargetClient targetClient = getTargetClient();
+ HudiConversionTarget targetClient = getTargetClient();
targetClient.beginSync(initialState);
targetClient.syncFilesForSnapshot(snapshot);
TableSyncMetadata latestState =
@@ -285,7 +285,7 @@ public class ITHudiTargetClient {
.build());
// sync snapshot and metadata
InternalTable initialState = getState(Instant.now().minus(24, ChronoUnit.HOURS));
- HudiTargetClient targetClient = getTargetClient();
+ HudiConversionTarget targetClient = getTargetClient();
targetClient.beginSync(initialState);
targetClient.syncFilesForSnapshot(snapshot);
TableSyncMetadata latestState =
@@ -374,19 +374,19 @@ public class ITHudiTargetClient {
}
private TableSyncMetadata incrementalSync(
- TargetClient targetClient,
+ ConversionTarget conversionTarget,
List<InternalDataFile> filesToAdd,
List<InternalDataFile> filesToRemove,
Instant commitStart) {
DataFilesDiff dataFilesDiff2 =
DataFilesDiff.builder().filesAdded(filesToAdd).filesRemoved(filesToRemove).build();
InternalTable state3 = getState(commitStart);
- targetClient.beginSync(state3);
- targetClient.syncFilesForDiff(dataFilesDiff2);
+ conversionTarget.beginSync(state3);
+ conversionTarget.syncFilesForDiff(dataFilesDiff2);
TableSyncMetadata latestState =
TableSyncMetadata.of(state3.getLatestCommitTime(), Collections.emptyList());
- targetClient.syncMetadata(latestState);
- targetClient.completeSync();
+ conversionTarget.syncMetadata(latestState);
+ conversionTarget.completeSync();
return latestState;
}
@@ -583,8 +583,8 @@ public class ITHudiTargetClient {
.build();
}
- private HudiTargetClient getTargetClient() {
- return new HudiTargetClient(
+ private HudiConversionTarget getTargetClient() {
+ return new HudiConversionTarget(
PerTableConfigImpl.builder()
.tableBasePath(tableBasePath)
.targetTableFormats(Collections.singletonList(TableFormat.HUDI))
diff --git a/core/src/test/java/org/apache/xtable/hudi/TestHudiTargetClient.java b/core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSourceTarget.java
similarity index 84%
rename from core/src/test/java/org/apache/xtable/hudi/TestHudiTargetClient.java
rename to core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSourceTarget.java
index 775307c0..8f2ccd94 100644
--- a/core/src/test/java/org/apache/xtable/hudi/TestHudiTargetClient.java
+++ b/core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSourceTarget.java
@@ -53,10 +53,10 @@ import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.PartitionFileGroup;
/**
- * Unit tests that focus on the basic control flow of the HudiTargetClient. Functional tests are in
- * {@link ITHudiTargetClient}.
+ * Unit tests that focus on the basic control flow of the HudiConversionTarget. Functional tests are
+ * in {@link ITHudiConversionSourceTarget}.
*/
-public class TestHudiTargetClient {
+public class TestHudiConversionSourceTarget {
private static final int RETENTION_IN_HOURS = 1;
private static final int MAX_DELTA_COMMITS = 2;
private static final Instant COMMIT_TIME = Instant.ofEpochMilli(1598644800000L);
@@ -73,13 +73,13 @@ public class TestHudiTargetClient {
mock(BaseFileUpdatesExtractor.class);
private final AvroSchemaConverter mockAvroSchemaConverter = mock(AvroSchemaConverter.class);
private final HudiTableManager mockHudiTableManager = mock(HudiTableManager.class);
- private final HudiTargetClient.CommitStateCreator mockCommitStateCreator =
- mock(HudiTargetClient.CommitStateCreator.class);
+ private final HudiConversionTarget.CommitStateCreator mockCommitStateCreator =
+ mock(HudiConversionTarget.CommitStateCreator.class);
- private HudiTargetClient getTargetClient(HoodieTableMetaClient mockMetaClient) {
+ private HudiConversionTarget getTargetClient(HoodieTableMetaClient mockMetaClient) {
when(mockHudiTableManager.loadTableMetaClientIfExists(BASE_PATH))
.thenReturn(Optional.ofNullable(mockMetaClient));
- return new HudiTargetClient(
+ return new HudiConversionTarget(
BASE_PATH,
RETENTION_IN_HOURS,
MAX_DELTA_COMMITS,
@@ -96,8 +96,9 @@ public class TestHudiTargetClient {
@Test
void syncSchema() {
- HudiTargetClient targetClient = getTargetClient(null);
- HudiTargetClient.CommitState mockCommitState = initMocksForBeginSync(targetClient).getLeft();
+ HudiConversionTarget targetClient = getTargetClient(null);
+ HudiConversionTarget.CommitState mockCommitState =
+ initMocksForBeginSync(targetClient).getLeft();
InternalSchema input =
InternalSchema.builder()
.name("schema")
@@ -114,7 +115,7 @@ public class TestHudiTargetClient {
@Test
void syncSchemaWithRecordKeyChanged() {
- HudiTargetClient targetClient = getTargetClient(null);
+ HudiConversionTarget targetClient = getTargetClient(null);
initMocksForBeginSync(targetClient);
List<InternalField> recordKeyFields =
Arrays.asList(
@@ -137,8 +138,9 @@ public class TestHudiTargetClient {
@Test
void syncMetadata() {
- HudiTargetClient targetClient = getTargetClient(null);
- HudiTargetClient.CommitState mockCommitState = initMocksForBeginSync(targetClient).getLeft();
+ HudiConversionTarget targetClient = getTargetClient(null);
+ HudiConversionTarget.CommitState mockCommitState =
+ initMocksForBeginSync(targetClient).getLeft();
TableSyncMetadata metadata = TableSyncMetadata.of(COMMIT_TIME, Collections.emptyList());
targetClient.syncMetadata(metadata);
// validate that metadata is set in commitState
@@ -147,8 +149,9 @@ public class TestHudiTargetClient {
@Test
void completeSync() {
- HudiTargetClient targetClient = getTargetClient(null);
- HudiTargetClient.CommitState mockCommitState = initMocksForBeginSync(targetClient).getLeft();
+ HudiConversionTarget targetClient = getTargetClient(null);
+ HudiConversionTarget.CommitState mockCommitState =
+ initMocksForBeginSync(targetClient).getLeft();
targetClient.completeSync();
// validate that commit is called
verify(mockCommitState).commit();
@@ -156,7 +159,7 @@ public class TestHudiTargetClient {
@Test
void syncPartitionSpecSucceedsWithoutChanges() {
- HudiTargetClient targetClient = getTargetClient(null);
+ HudiConversionTarget targetClient = getTargetClient(null);
HoodieTableMetaClient mockMetaClient = initMocksForBeginSync(targetClient).getRight();
String field1 = "field1";
String field2 = "field2";
@@ -179,7 +182,7 @@ public class TestHudiTargetClient {
@ParameterizedTest
@ValueSource(strings = {"", "field1", "field2,field1"})
void syncPartitionSpecFailsWithChanges(String partitionFields) {
- HudiTargetClient targetClient = getTargetClient(null);
+ HudiConversionTarget targetClient = getTargetClient(null);
HoodieTableMetaClient mockMetaClient = initMocksForBeginSync(targetClient).getRight();
String field1 = "field1";
String field2 = "field2";
@@ -204,8 +207,9 @@ public class TestHudiTargetClient {
@Test
void syncFilesForDiff() {
- HudiTargetClient targetClient = getTargetClient(null);
- HudiTargetClient.CommitState mockCommitState = initMocksForBeginSync(targetClient).getLeft();
+ HudiConversionTarget targetClient = getTargetClient(null);
+ HudiConversionTarget.CommitState mockCommitState =
+ initMocksForBeginSync(targetClient).getLeft();
String instant = "commit";
DataFilesDiff input = DataFilesDiff.builder().build();
BaseFileUpdatesExtractor.ReplaceMetadata output =
@@ -221,10 +225,10 @@ public class TestHudiTargetClient {
@Test
void syncFilesForSnapshot() {
- HudiTargetClient targetClient = getTargetClient(null);
- Pair<HudiTargetClient.CommitState, HoodieTableMetaClient> mocks =
+ HudiConversionTarget targetClient = getTargetClient(null);
+ Pair<HudiConversionTarget.CommitState, HoodieTableMetaClient> mocks =
initMocksForBeginSync(targetClient);
- HudiTargetClient.CommitState mockCommitState = mocks.getLeft();
+ HudiConversionTarget.CommitState mockCommitState = mocks.getLeft();
HoodieTableMetaClient mockMetaClient = mocks.getRight();
String instant = "commit";
List<PartitionFileGroup> input = Collections.emptyList();
@@ -243,7 +247,7 @@ public class TestHudiTargetClient {
@Test
void beginSyncForExistingTable() {
HoodieTableMetaClient mockMetaClient = mock(HoodieTableMetaClient.class);
- HudiTargetClient targetClient = getTargetClient(mockMetaClient);
+ HudiConversionTarget targetClient = getTargetClient(mockMetaClient);
targetClient.beginSync(TABLE);
// verify meta client timeline refreshed
@@ -253,15 +257,15 @@ public class TestHudiTargetClient {
.create(mockMetaClient, COMMIT, RETENTION_IN_HOURS, MAX_DELTA_COMMITS);
}
- private Pair<HudiTargetClient.CommitState, HoodieTableMetaClient> initMocksForBeginSync(
- HudiTargetClient targetClient) {
+ private Pair<HudiConversionTarget.CommitState, HoodieTableMetaClient> initMocksForBeginSync(
+ HudiConversionTarget targetClient) {
HoodieTableMetaClient mockMetaClient = mock(HoodieTableMetaClient.class);
HoodieTableConfig mockTableConfig = mock(HoodieTableConfig.class);
when(mockMetaClient.getTableConfig()).thenReturn(mockTableConfig);
when(mockTableConfig.getRecordKeyFields())
.thenReturn(Option.of(new String[] {"record_key_field"}));
when(mockHudiTableManager.initializeHudiTable(BASE_PATH, TABLE)).thenReturn(mockMetaClient);
- HudiTargetClient.CommitState mockCommitState = mock(HudiTargetClient.CommitState.class);
+ HudiConversionTarget.CommitState mockCommitState = mock(HudiConversionTarget.CommitState.class);
when(mockCommitStateCreator.create(
mockMetaClient, COMMIT, RETENTION_IN_HOURS, MAX_DELTA_COMMITS))
.thenReturn(mockCommitState);
diff --git a/core/src/test/java/org/apache/xtable/iceberg/ITIcebergSourceClient.java b/core/src/test/java/org/apache/xtable/iceberg/ITIcebergConversionTargetSource.java
similarity index 86%
rename from core/src/test/java/org/apache/xtable/iceberg/ITIcebergSourceClient.java
rename to core/src/test/java/org/apache/xtable/iceberg/ITIcebergConversionTargetSource.java
index a1b16ff7..997ab184 100644
--- a/core/src/test/java/org/apache/xtable/iceberg/ITIcebergSourceClient.java
+++ b/core/src/test/java/org/apache/xtable/iceberg/ITIcebergConversionTargetSource.java
@@ -46,8 +46,8 @@ import org.apache.iceberg.Snapshot;
import org.apache.iceberg.data.Record;
import org.apache.xtable.TestIcebergTable;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.PerTableConfigImpl;
+import org.apache.xtable.conversion.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfigImpl;
import org.apache.xtable.model.CommitsBacklog;
import org.apache.xtable.model.InstantsForIncrementalSync;
import org.apache.xtable.model.InternalSnapshot;
@@ -56,16 +56,16 @@ import org.apache.xtable.model.schema.InternalPartitionField;
import org.apache.xtable.model.schema.PartitionTransformType;
import org.apache.xtable.model.storage.TableFormat;
-public class ITIcebergSourceClient {
+public class ITIcebergConversionTargetSource {
private static final Configuration hadoopConf = new Configuration();
@TempDir public static Path tempDir;
- private IcebergSourceClientProvider clientProvider;
+ private IcebergConversionSourceProvider sourceProvider;
@BeforeEach
void setup() {
- clientProvider = new IcebergSourceClientProvider();
- clientProvider.init(hadoopConf, null);
+ sourceProvider = new IcebergConversionSourceProvider();
+ sourceProvider.init(hadoopConf, null);
}
@ParameterizedTest
@@ -103,9 +103,10 @@ public class ITIcebergSourceClient {
.tableBasePath(testIcebergTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
.build();
- IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ IcebergConversionSource conversionSource =
+ sourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(180L, testIcebergTable.getNumRows());
- InternalSnapshot internalSnapshot = icebergSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
if (isPartitioned) {
validateIcebergPartitioning(internalSnapshot);
@@ -117,9 +118,9 @@ public class ITIcebergSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Snapshot> commitsBacklog =
- icebergSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Snapshot snapshot : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = icebergSourceClient.getTableChangeForCommit(snapshot);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(snapshot);
allTableChanges.add(tableChange);
}
validateTableChanges(allActiveFiles, allTableChanges);
@@ -162,11 +163,12 @@ public class ITIcebergSourceClient {
.tableBasePath(testIcebergTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
.build();
- IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ IcebergConversionSource conversionSource =
+ sourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(
120 - recordsByPartition.get(partitionValueToDelete).size(),
testIcebergTable.getNumRows());
- InternalSnapshot internalSnapshot = icebergSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
validateIcebergPartitioning(internalSnapshot);
validateSnapshot(internalSnapshot, allActiveFiles.get(allActiveFiles.size() - 1));
@@ -176,9 +178,9 @@ public class ITIcebergSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Snapshot> commitsBacklog =
- icebergSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Snapshot snapshot : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = icebergSourceClient.getTableChangeForCommit(snapshot);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(snapshot);
allTableChanges.add(tableChange);
}
validateTableChanges(allActiveFiles, allTableChanges);
@@ -221,11 +223,12 @@ public class ITIcebergSourceClient {
.tableBasePath(testIcebergTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
.build();
- IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ IcebergConversionSource conversionSource =
+ sourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(
120 - recordsByPartition.get(partitionValueToDelete).size(),
testIcebergTable.getNumRows());
- InternalSnapshot internalSnapshot = icebergSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
validateIcebergPartitioning(internalSnapshot);
validateSnapshot(internalSnapshot, allActiveFiles.get(allActiveFiles.size() - 1));
@@ -235,9 +238,9 @@ public class ITIcebergSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Snapshot> commitsBacklog =
- icebergSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Snapshot snapshot : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = icebergSourceClient.getTableChangeForCommit(snapshot);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(snapshot);
allTableChanges.add(tableChange);
}
validateTableChanges(allActiveFiles, allTableChanges);
@@ -280,9 +283,10 @@ public class ITIcebergSourceClient {
.tableBasePath(testIcebergTable.getBasePath())
.targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
.build();
- IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ IcebergConversionSource conversionSource =
+ sourceProvider.getConversionSourceInstance(tableConfig);
assertEquals(200L, testIcebergTable.getNumRows());
- InternalSnapshot internalSnapshot = icebergSourceClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = conversionSource.getCurrentSnapshot();
if (isPartitioned) {
validateIcebergPartitioning(internalSnapshot);
@@ -294,9 +298,9 @@ public class ITIcebergSourceClient {
.lastSyncInstant(Instant.ofEpochMilli(timestamp1))
.build();
CommitsBacklog<Snapshot> commitsBacklog =
- icebergSourceClient.getCommitsBacklog(instantsForIncrementalSync);
+ conversionSource.getCommitsBacklog(instantsForIncrementalSync);
for (Snapshot snapshot : commitsBacklog.getCommitsToProcess()) {
- TableChange tableChange = icebergSourceClient.getTableChangeForCommit(snapshot);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(snapshot);
allTableChanges.add(tableChange);
}
validateTableChanges(allActiveFiles, allTableChanges);
@@ -333,16 +337,16 @@ public class ITIcebergSourceClient {
// Expire snapshotAfterCommit2.
testIcebergTable.expireSnapshot(snapshotIdAfterCommit2);
}
- IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+ IcebergConversionSource conversionSource =
+ sourceProvider.getConversionSourceInstance(tableConfig);
if (shouldExpireSnapshots) {
- assertFalse(
- icebergSourceClient.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)));
+ assertFalse(conversionSource.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)));
} else {
- assertTrue(icebergSourceClient.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)));
+ assertTrue(conversionSource.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)));
}
// Table doesn't have instant of this older commit, hence it is not safe.
Instant instantAsOfHourAgo = Instant.now().minus(1, ChronoUnit.HOURS);
- assertFalse(icebergSourceClient.isIncrementalSyncSafeFrom(instantAsOfHourAgo));
+ assertFalse(conversionSource.isIncrementalSyncSafeFrom(instantAsOfHourAgo));
}
}
diff --git a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSourceClient.java b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergConversionTargetSource.java
similarity index 91%
rename from core/src/test/java/org/apache/xtable/iceberg/TestIcebergSourceClient.java
rename to core/src/test/java/org/apache/xtable/iceberg/TestIcebergConversionTargetSource.java
index bca730a8..010a524b 100644
--- a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSourceClient.java
+++ b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergConversionTargetSource.java
@@ -49,8 +49,8 @@ import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.types.Types;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.PerTableConfigImpl;
+import org.apache.xtable.conversion.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfigImpl;
import org.apache.xtable.model.CommitsBacklog;
import org.apache.xtable.model.InstantsForIncrementalSync;
import org.apache.xtable.model.InternalSnapshot;
@@ -66,12 +66,12 @@ import org.apache.xtable.model.storage.*;
import org.apache.xtable.model.storage.FileFormat;
import org.apache.xtable.model.storage.InternalDataFile;
-class TestIcebergSourceClient {
+class TestIcebergConversionTargetSource {
private IcebergTableManager tableManager;
private Schema csSchema;
private PartitionSpec csPartitionSpec;
- private IcebergSourceClientProvider clientProvider;
+ private IcebergConversionSourceProvider sourceProvider;
private Configuration hadoopConf;
@BeforeEach
@@ -79,8 +79,8 @@ class TestIcebergSourceClient {
hadoopConf = new Configuration();
hadoopConf.set("fs.defaultFS", "file:///");
- clientProvider = new IcebergSourceClientProvider();
- clientProvider.init(hadoopConf, null);
+ sourceProvider = new IcebergConversionSourceProvider();
+ sourceProvider.init(hadoopConf, null);
tableManager = IcebergTableManager.of(hadoopConf);
@@ -96,10 +96,11 @@ class TestIcebergSourceClient {
Table catalogSales = createTestTableWithData(workingDir.toString());
PerTableConfig sourceTableConfig = getPerTableConfig(catalogSales);
- IcebergSourceClient client = clientProvider.getSourceClientInstance(sourceTableConfig);
+ IcebergConversionSource conversionSource =
+ sourceProvider.getConversionSourceInstance(sourceTableConfig);
Snapshot snapshot = catalogSales.currentSnapshot();
- InternalTable internalTable = client.getTable(snapshot);
+ InternalTable internalTable = conversionSource.getTable(snapshot);
Assertions.assertNotNull(internalTable);
assertEquals(TableFormat.ICEBERG, internalTable.getTableFormat());
Assertions.assertTrue(internalTable.getName().endsWith("catalog_sales"));
@@ -132,10 +133,11 @@ class TestIcebergSourceClient {
PerTableConfig sourceTableConfig = getPerTableConfig(catalogSales);
- IcebergSourceClient client = clientProvider.getSourceClientInstance(sourceTableConfig);
- IcebergSourceClient spyClient = Mockito.spy(client);
+ IcebergConversionSource conversionSource =
+ sourceProvider.getConversionSourceInstance(sourceTableConfig);
+ IcebergConversionSource spyConversionSource = Mockito.spy(conversionSource);
- SchemaCatalog schemaCatalog = spyClient.getSchemaCatalog(null, iceCurrentSnapshot);
+ SchemaCatalog schemaCatalog = spyConversionSource.getSchemaCatalog(null, iceCurrentSnapshot);
Assertions.assertNotNull(schemaCatalog);
Map<SchemaVersion, InternalSchema> schemas = schemaCatalog.getSchemas();
assertEquals(1, schemas.size());
@@ -156,21 +158,22 @@ class TestIcebergSourceClient {
IcebergPartitionValueConverter spyPartitionConverter =
spy(IcebergPartitionValueConverter.getInstance());
- IcebergSourceClient spyClient =
+ IcebergConversionSource spyConversionSource =
spy(
- IcebergSourceClient.builder()
+ IcebergConversionSource.builder()
.hadoopConf(hadoopConf)
.sourceTableConfig(sourceTableConfig)
.dataFileExtractor(spyDataFileExtractor)
.partitionConverter(spyPartitionConverter)
.build());
- InternalSnapshot internalSnapshot = spyClient.getCurrentSnapshot();
+ InternalSnapshot internalSnapshot = spyConversionSource.getCurrentSnapshot();
Assertions.assertNotNull(internalSnapshot);
assertEquals(String.valueOf(iceCurrentSnapshot.snapshotId()), internalSnapshot.getVersion());
Assertions.assertNotNull(internalSnapshot.getTable());
- verify(spyClient, times(1)).getTable(iceCurrentSnapshot);
- verify(spyClient, times(1)).getSchemaCatalog(internalSnapshot.getTable(), iceCurrentSnapshot);
+ verify(spyConversionSource, times(1)).getTable(iceCurrentSnapshot);
+ verify(spyConversionSource, times(1))
+ .getSchemaCatalog(internalSnapshot.getTable(), iceCurrentSnapshot);
verify(spyPartitionConverter, times(5)).toOneTable(any(), any(), any());
verify(spyDataFileExtractor, times(5)).fromIceberg(any(), any(), any());
@@ -328,8 +331,8 @@ class TestIcebergSourceClient {
InstantsForIncrementalSync.builder()
.lastSyncInstant(Instant.ofEpochMilli(lastSync.timestampMillis()))
.build();
- IcebergSourceClient sourceClient = getIcebergSourceClient(table);
- CommitsBacklog<Snapshot> commitsBacklog = sourceClient.getCommitsBacklog(instant);
+ IcebergConversionSource conversionSource = getIcebergConversionSource(table);
+ CommitsBacklog<Snapshot> commitsBacklog = conversionSource.getCommitsBacklog(instant);
assertEquals(0, commitsBacklog.getInFlightInstants().size());
Assertions.assertArrayEquals(snapshots, commitsBacklog.getCommitsToProcess().toArray());
}
@@ -342,8 +345,8 @@ class TestIcebergSourceClient {
private void validateTableChangeDiffSize(
Table table, Snapshot snapshot, int addedFiles, int removedFiles) {
- IcebergSourceClient sourceClient = getIcebergSourceClient(table);
- TableChange tableChange = sourceClient.getTableChangeForCommit(snapshot);
+ IcebergConversionSource conversionSource = getIcebergConversionSource(table);
+ TableChange tableChange = conversionSource.getTableChangeForCommit(snapshot);
assertEquals(addedFiles, tableChange.getFilesDiff().getFilesAdded().size());
assertEquals(removedFiles, tableChange.getFilesDiff().getFilesRemoved().size());
}
@@ -411,10 +414,10 @@ class TestIcebergSourceClient {
return tableManager.getOrCreateTable(null, null, csPath, csSchema, csPartitionSpec);
}
- private IcebergSourceClient getIcebergSourceClient(Table catalogSales) {
+ private IcebergConversionSource getIcebergConversionSource(Table catalogSales) {
PerTableConfig tableConfig = getPerTableConfig(catalogSales);
- return IcebergSourceClient.builder()
+ return IcebergConversionSource.builder()
.hadoopConf(hadoopConf)
.sourceTableConfig(tableConfig)
.build();
diff --git a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
index df24495d..f676f8a9 100644
--- a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
+++ b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
@@ -81,8 +81,8 @@ import org.apache.iceberg.types.Types;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
-import org.apache.xtable.ITOneTableClient;
-import org.apache.xtable.client.PerTableConfigImpl;
+import org.apache.xtable.ITConversionController;
+import org.apache.xtable.conversion.PerTableConfigImpl;
import org.apache.xtable.model.InternalSnapshot;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.TableSyncMetadata;
@@ -104,8 +104,8 @@ import org.apache.xtable.schema.SchemaFieldFinder;
import org.apache.xtable.spi.sync.TableFormatSync;
/**
- * Validates that the metadata for the table is properly created/updated. {@link ITOneTableClient}
- * validates that the table and its data can be properly read.
+ * Validates that the metadata for the table is properly created/updated. {@link
+ * ITConversionController} validates that the table and its data can be properly read.
*/
public class TestIcebergSync {
private static final Random RANDOM = new Random();
@@ -123,7 +123,7 @@ public class TestIcebergSync {
Mockito.mock(IcebergPartitionSpecSync.class);
private final IcebergColumnStatsConverter mockColumnStatsConverter =
Mockito.mock(IcebergColumnStatsConverter.class);
- private IcebergClient icebergClient;
+ private IcebergConversionTarget conversionTarget;
private final InternalSchema internalSchema =
InternalSchema.builder()
@@ -187,11 +187,11 @@ public class TestIcebergSync {
tableName = "test-" + UUID.randomUUID();
basePath = tempDir.resolve(tableName);
Files.createDirectories(basePath);
- icebergClient = getIcebergClient();
+ conversionTarget = getConversionTarget();
}
- private IcebergClient getIcebergClient() {
- return new IcebergClient(
+ private IcebergConversionTarget getConversionTarget() {
+ return new IcebergConversionTarget(
PerTableConfigImpl.builder()
.tableBasePath(basePath.toString())
.tableName(tableName)
@@ -247,10 +247,12 @@ public class TestIcebergSync {
mockColStatsForFile(dataFile2, 2);
mockColStatsForFile(dataFile3, 1);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot1);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
validateIcebergTable(tableName, table1, Sets.newHashSet(dataFile1, dataFile2), null);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot2);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot2);
validateIcebergTable(tableName, table2, Sets.newHashSet(dataFile2, dataFile3), null);
ArgumentCaptor<Transaction> transactionArgumentCaptor =
@@ -346,9 +348,11 @@ public class TestIcebergSync {
mockColStatsForFile(dataFile3, 2);
mockColStatsForFile(dataFile4, 1);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot1);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot1);
long snapshotIdBeforeCorruption = getTable(basePath).currentSnapshot().snapshotId();
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot2);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot2);
String manifestFile =
new HadoopTables(CONFIGURATION)
.load(basePath.toString())
@@ -357,11 +361,12 @@ public class TestIcebergSync {
Files.delete(Paths.get(URI.create(manifestFile)));
Optional<Instant> actual =
- getIcebergClient().getTableMetadata().map(TableSyncMetadata::getLastInstantSynced);
+ getConversionTarget().getTableMetadata().map(TableSyncMetadata::getLastInstantSynced);
// assert that the last commit is rolled back and the metadata is removed
assertFalse(actual.isPresent());
// get a new iceberg sync to make sure table is re-read from disk and no metadata is cached
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot3);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot3);
validateIcebergTable(tableName, table2, Sets.newHashSet(dataFile3, dataFile4), null);
// Validate Iceberg table state
Table table = getTable(basePath);
@@ -421,7 +426,8 @@ public class TestIcebergSync {
mockColStatsForFile(dataFile1, 1);
mockColStatsForFile(dataFile2, 1);
mockColStatsForFile(dataFile3, 1);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot);
assertTrue(schemaArgumentCaptor.getValue().sameSchema(icebergSchema));
validateIcebergTable(
@@ -484,7 +490,8 @@ public class TestIcebergSync {
mockColStatsForFile(dataFile1, 1);
mockColStatsForFile(dataFile2, 1);
mockColStatsForFile(dataFile3, 1);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot);
assertTrue(schemaArgumentCaptor.getValue().sameSchema(icebergSchema));
validateIcebergTable(
@@ -540,7 +547,8 @@ public class TestIcebergSync {
mockColStatsForFile(dataFile1, 1);
mockColStatsForFile(dataFile2, 1);
mockColStatsForFile(dataFile3, 1);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot);
assertTrue(schemaArgumentCaptor.getValue().sameSchema(icebergSchema));
validateIcebergTable(
@@ -618,7 +626,8 @@ public class TestIcebergSync {
mockColStatsForFile(dataFile1, 1);
mockColStatsForFile(dataFile2, 1);
mockColStatsForFile(dataFile3, 1);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot);
assertTrue(schemaArgumentCaptor.getValue().sameSchema(icebergSchema));
validateIcebergTable(
@@ -685,7 +694,8 @@ public class TestIcebergSync {
mockColStatsForFile(dataFile1, 1);
mockColStatsForFile(dataFile2, 1);
mockColStatsForFile(dataFile3, 1);
- TableFormatSync.getInstance().syncSnapshot(Collections.singletonList(icebergClient), snapshot);
+ TableFormatSync.getInstance()
+ .syncSnapshot(Collections.singletonList(conversionTarget), snapshot);
assertTrue(schemaArgumentCaptor.getValue().sameSchema(icebergSchema));
validateIcebergTable(
diff --git a/core/src/test/java/org/apache/xtable/loadtest/LoadTest.java b/core/src/test/java/org/apache/xtable/loadtest/LoadTest.java
index a5020fff..8b1a24e0 100644
--- a/core/src/test/java/org/apache/xtable/loadtest/LoadTest.java
+++ b/core/src/test/java/org/apache/xtable/loadtest/LoadTest.java
@@ -35,11 +35,11 @@ import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.config.HoodieArchivalConfig;
import org.apache.xtable.TestJavaHudiTable;
-import org.apache.xtable.client.OneTableClient;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.PerTableConfigImpl;
-import org.apache.xtable.client.SourceClientProvider;
-import org.apache.xtable.hudi.HudiSourceClientProvider;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfigImpl;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
@@ -51,12 +51,12 @@ import org.apache.xtable.model.sync.SyncMode;
public class LoadTest {
@TempDir public static Path tempDir;
private static final Configuration CONFIGURATION = new Configuration();
- private SourceClientProvider<HoodieInstant> hudiSourceClientProvider;
+ private ConversionSourceProvider<HoodieInstant> hudiConversionSourceProvider;
@BeforeEach
public void setup() {
- hudiSourceClientProvider = new HudiSourceClientProvider();
- hudiSourceClientProvider.init(CONFIGURATION, Collections.emptyMap());
+ hudiConversionSourceProvider = new HudiConversionSourceProvider();
+ hudiConversionSourceProvider.init(CONFIGURATION, Collections.emptyMap());
}
@Test
@@ -82,9 +82,9 @@ public class LoadTest {
.tableBasePath(table.getBasePath())
.syncMode(SyncMode.FULL)
.build();
- OneTableClient oneTableClient = new OneTableClient(CONFIGURATION);
+ ConversionController conversionController = new ConversionController(CONFIGURATION);
long start = System.currentTimeMillis();
- oneTableClient.sync(perTableConfig, hudiSourceClientProvider);
+ conversionController.sync(perTableConfig, hudiConversionSourceProvider);
long end = System.currentTimeMillis();
System.out.println("Full sync took " + (end - start) + "ms");
}
@@ -112,8 +112,8 @@ public class LoadTest {
.syncMode(SyncMode.INCREMENTAL)
.build();
// sync once to establish first commit
- OneTableClient oneTableClient = new OneTableClient(CONFIGURATION);
- oneTableClient.sync(perTableConfig, hudiSourceClientProvider);
+ ConversionController conversionController = new ConversionController(CONFIGURATION);
+ conversionController.sync(perTableConfig, hudiConversionSourceProvider);
for (int i = 0; i < numCommits; i++) {
table.insertRecords(
1,
@@ -124,7 +124,7 @@ public class LoadTest {
}
long start = System.currentTimeMillis();
- oneTableClient.sync(perTableConfig, hudiSourceClientProvider);
+ conversionController.sync(perTableConfig, hudiConversionSourceProvider);
long end = System.currentTimeMillis();
System.out.println("Incremental sync took " + (end - start) + "ms");
}
diff --git a/demo/notebook/demo.ipynb b/demo/notebook/demo.ipynb
index f7379365..244c52b4 100644
--- a/demo/notebook/demo.ipynb
+++ b/demo/notebook/demo.ipynb
@@ -33,7 +33,7 @@
"import $ivy.`org.apache.iceberg:iceberg-hive-runtime:1.3.1`\n",
"import $ivy.`io.trino:trino-jdbc:431`\n",
"import java.util._\n",
- "import org.apache.xtable.client._\n",
+ "import org.apache.xtable.conversion._\n",
"import org.apache.xtable.model.sync._\n",
"import org.apache.xtable.model.storage._\n",
"import org.apache.xtable.hudi._\n",
@@ -128,10 +128,10 @@
" .icebergCatalogConfig(icebergCatalogConfig)\n",
" .syncMode(SyncMode.INCREMENTAL)\n",
" .build()\n",
- "val hudiSourceClientProvider = new HudiSourceClientProvider()\n",
- "hudiSourceClientProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
- "val oneTableClient = new OneTableClient(spark.sparkContext.hadoopConfiguration)\n",
- "oneTableClient.sync(perTableConfig, hudiSourceClientProvider)"
+ "val hudiConversionSourceProvider = new HudiConversionSourceProvider()\n",
+ "hudiConversionSourceProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
+ "val conversionController = new ConversionController(spark.sparkContext.hadoopConfiguration)\n",
+ "conversionController.sync(perTableConfig, hudiConversionSourceProvider)"
]
},
{
@@ -162,7 +162,7 @@
" .options(hudiWriteOptions)\n",
" .mode(\"append\")\n",
" .save(hudiBasePath)\n",
- "oneTableClient.sync(perTableConfig, hudiSourceClientProvider)"
+ "conversionController.sync(perTableConfig, hudiConversionSourceProvider)"
]
},
{
@@ -207,8 +207,8 @@
"metadata": {},
"outputs": [],
"source": [
- "val icebergSourceClientProvider = new IcebergSourceClientProvider()\n",
- "icebergSourceClientProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
+ "val icebergConversionSourceProvider = new IcebergConversionSourceProvider()\n",
+ "icebergConversionSourceProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
"val icebergSourcePerTableConfig = PerTableConfigImpl.builder()\n",
" .tableName(hudiTableName)\n",
" .namespace(namespaceArray)\n",
@@ -217,7 +217,7 @@
" .icebergCatalogConfig(icebergCatalogConfig)\n",
" .syncMode(SyncMode.INCREMENTAL)\n",
" .build()\n",
- "oneTableClient.sync(icebergSourcePerTableConfig, icebergSourceClientProvider)"
+ "conversionController.sync(icebergSourcePerTableConfig, icebergConversionSourceProvider)"
]
},
{
@@ -240,15 +240,15 @@
"source": [
"// Sync from a Delta Lake table to Hudi\n",
"import org.apache.xtable.delta._\n",
- "val deltaSourceClientProvider = new DeltaSourceClientProvider()\n",
- "deltaSourceClientProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
+ "val deltaConversionSourceProvider = new DeltaConversionSourceProvider()\n",
+ "deltaConversionSourceProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
"val deltaSourcePerTableConfig = PerTableConfigImpl.builder()\n",
" .tableName(deltaTableName)\n",
" .targetTableFormats(Arrays.asList(TableFormat.HUDI))\n",
" .tableBasePath(deltaBasePath)\n",
" .syncMode(SyncMode.INCREMENTAL)\n",
" .build()\n",
- "oneTableClient.sync(deltaSourcePerTableConfig, deltaSourceClientProvider)"
+ "conversionController.sync(deltaSourcePerTableConfig, deltaConversionSourceProvider)"
]
},
{
@@ -302,10 +302,10 @@
" .icebergCatalogConfig(icebergCatalogConfig)\n",
" .syncMode(SyncMode.INCREMENTAL)\n",
" .build()\n",
- "val deltaSourceClientProvider = new DeltaSourceClientProvider()\n",
- "deltaSourceClientProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
- "val oneTableClient = new OneTableClient(spark.sparkContext.hadoopConfiguration)\n",
- "oneTableClient.sync(perTableConfig, deltaSourceClientProvider)"
+ "val deltaConversionSourceProvider = new DeltaConversionSourceProvider()\n",
+ "deltaConversionSourceProvider.init(spark.sparkContext.hadoopConfiguration, Collections.emptyMap())\n",
+ "val conversionController = new ConversionController(spark.sparkContext.hadoopConfiguration)\n",
+ "conversionController.sync(perTableConfig, deltaConversionSourceProvider)"
]
},
{
diff --git a/hudi-support/extensions/README.md b/hudi-support/extensions/README.md
index 51ea57b6..26782ed1 100644
--- a/hudi-support/extensions/README.md
+++ b/hudi-support/extensions/README.md
@@ -16,7 +16,7 @@ For example, if you're using the Hudi [quick-start guide](https://hudi.apache.or
If you want to use OneTable with Hudi [streaming ingestion](https://hudi.apache.org/docs/hoodie_streaming_ingestion) to sync each commit into other table formats.
### How do you use them?
1. Add the extensions jar (`hudi-extensions-0.1.0-SNAPSHOT-bundled.jar`) to your class path.
-2. Add `org.apache.xtable.hudi.sync.OneTableSyncTool` to your list of sync classes
+2. Add `org.apache.xtable.hudi.sync.XTableSyncTool` to your list of sync classes
3. Set the following configurations based on your preferences:
`hoodie.onetable.formats: "ICEBERG,DELTA"` (or simply use one format)
`hoodie.onetable.target.metadata.retention.hr: 168` (default retention for target format metadata is 168 hours)
\ No newline at end of file
diff --git a/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/OneTableSyncConfig.java b/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/XTableSyncConfig.java
similarity index 92%
rename from hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/OneTableSyncConfig.java
rename to hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/XTableSyncConfig.java
index fe716ad4..c4e6a600 100644
--- a/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/OneTableSyncConfig.java
+++ b/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/XTableSyncConfig.java
@@ -24,7 +24,7 @@ import java.util.Properties;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.sync.common.HoodieSyncConfig;
-public class OneTableSyncConfig extends HoodieSyncConfig implements Serializable {
+public class XTableSyncConfig extends HoodieSyncConfig implements Serializable {
public static final ConfigProperty<String> ONE_TABLE_FORMATS =
ConfigProperty.key("hoodie.onetable.formats.to.sync")
@@ -36,7 +36,7 @@ public class OneTableSyncConfig extends HoodieSyncConfig implements Serializable
.defaultValue(24 * 7)
.withDocumentation("Retention in hours for metadata in target table.");
- public OneTableSyncConfig(Properties props) {
+ public XTableSyncConfig(Properties props) {
super(props);
}
}
diff --git a/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/OneTableSyncTool.java b/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/XTableSyncTool.java
similarity index 83%
rename from hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/OneTableSyncTool.java
rename to hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/XTableSyncTool.java
index b84171b1..49a7a743 100644
--- a/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/OneTableSyncTool.java
+++ b/hudi-support/extensions/src/main/java/org/apache/xtable/hudi/sync/XTableSyncTool.java
@@ -34,10 +34,10 @@ import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.sync.common.HoodieSyncConfig;
import org.apache.hudi.sync.common.HoodieSyncTool;
-import org.apache.xtable.client.OneTableClient;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.PerTableConfigImpl;
-import org.apache.xtable.hudi.HudiSourceClientProvider;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfigImpl;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
import org.apache.xtable.hudi.HudiSourceConfigImpl;
import org.apache.xtable.model.schema.PartitionTransformType;
import org.apache.xtable.model.sync.SyncMode;
@@ -47,21 +47,21 @@ import org.apache.xtable.model.sync.SyncResult;
* A HoodieSyncTool for syncing a Hudi table to other formats (Delta and Iceberg) with
* InternalTable.
*/
-public class OneTableSyncTool extends HoodieSyncTool {
- private final OneTableSyncConfig config;
- private final HudiSourceClientProvider hudiSourceClientProvider;
+public class XTableSyncTool extends HoodieSyncTool {
+ private final XTableSyncConfig config;
+ private final HudiConversionSourceProvider hudiConversionSourceProvider;
- public OneTableSyncTool(Properties props, Configuration hadoopConf) {
+ public XTableSyncTool(Properties props, Configuration hadoopConf) {
super(props, hadoopConf);
- this.config = new OneTableSyncConfig(props);
- this.hudiSourceClientProvider = new HudiSourceClientProvider();
- hudiSourceClientProvider.init(hadoopConf, Collections.emptyMap());
+ this.config = new XTableSyncConfig(props);
+ this.hudiConversionSourceProvider = new HudiConversionSourceProvider();
+ hudiConversionSourceProvider.init(hadoopConf, Collections.emptyMap());
}
@Override
public void syncHoodieTable() {
List<String> formatsToSync =
- Arrays.stream(config.getString(OneTableSyncConfig.ONE_TABLE_FORMATS).split(","))
+ Arrays.stream(config.getString(XTableSyncConfig.ONE_TABLE_FORMATS).split(","))
.map(format -> format.toUpperCase())
.collect(Collectors.toList());
String basePath = config.getString(HoodieSyncConfig.META_SYNC_BASE_PATH);
@@ -77,10 +77,10 @@ public class OneTableSyncTool extends HoodieSyncTool {
.build())
.syncMode(SyncMode.INCREMENTAL)
.targetMetadataRetentionInHours(
- config.getInt(OneTableSyncConfig.ONE_TABLE_TARGET_METADATA_RETENTION_HOURS))
+ config.getInt(XTableSyncConfig.ONE_TABLE_TARGET_METADATA_RETENTION_HOURS))
.build();
Map<String, SyncResult> results =
- new OneTableClient(hadoopConf).sync(perTableConfig, hudiSourceClientProvider);
+ new ConversionController(hadoopConf).sync(perTableConfig, hudiConversionSourceProvider);
String failingFormats =
results.entrySet().stream()
.filter(
diff --git a/hudi-support/extensions/src/test/java/org/apache/xtable/hudi/sync/TestOneTableSyncTool.java b/hudi-support/extensions/src/test/java/org/apache/xtable/hudi/sync/TestXTableSyncTool.java
similarity index 95%
rename from hudi-support/extensions/src/test/java/org/apache/xtable/hudi/sync/TestOneTableSyncTool.java
rename to hudi-support/extensions/src/test/java/org/apache/xtable/hudi/sync/TestXTableSyncTool.java
index 7c4fac61..d44465c8 100644
--- a/hudi-support/extensions/src/test/java/org/apache/xtable/hudi/sync/TestOneTableSyncTool.java
+++ b/hudi-support/extensions/src/test/java/org/apache/xtable/hudi/sync/TestXTableSyncTool.java
@@ -52,7 +52,7 @@ import org.apache.hudi.DataSourceWriteOptions;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import org.apache.hudi.sync.common.HoodieSyncConfig;
-public class TestOneTableSyncTool {
+public class TestXTableSyncTool {
private static SparkSession spark;
@TempDir static Path tempDir;
@@ -68,7 +68,7 @@ public class TestOneTableSyncTool {
public static void initSpark() {
SparkConf sparkConf =
new SparkConf()
- .setAppName("onetableclient")
+ .setAppName("xtable")
.set("spark.serializer", KryoSerializer.class.getName())
.set("spark.databricks.delta.constraints.allowUnenforcedNotNull.enabled", "true")
.set("spark.sql.shuffle.partitions", "1")
@@ -98,12 +98,12 @@ public class TestOneTableSyncTool {
writeBasicHudiTable(path, options);
Properties properties = new Properties();
- properties.put(OneTableSyncConfig.ONE_TABLE_FORMATS.key(), "iceberg,DELTA");
+ properties.put(XTableSyncConfig.ONE_TABLE_FORMATS.key(), "iceberg,DELTA");
properties.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), partitionPath);
properties.put(HoodieSyncConfig.META_SYNC_BASE_PATH.key(), path);
properties.putAll(options);
- new OneTableSyncTool(properties, new Configuration()).syncHoodieTable();
+ new XTableSyncTool(properties, new Configuration()).syncHoodieTable();
// lightweight check to make sure metadata dirs are made - assumes that InternalTable sync is
// correct if it succeeds
assertTrue(Files.exists(Paths.get(URI.create(path + "/_delta_log"))));
diff --git a/utilities/src/main/java/org/apache/xtable/utilities/RunSync.java b/utilities/src/main/java/org/apache/xtable/utilities/RunSync.java
index 9741a0b4..09ce2c2b 100644
--- a/utilities/src/main/java/org/apache/xtable/utilities/RunSync.java
+++ b/utilities/src/main/java/org/apache/xtable/utilities/RunSync.java
@@ -44,17 +44,17 @@ import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import com.google.common.annotations.VisibleForTesting;
-import org.apache.xtable.client.OneTableClient;
-import org.apache.xtable.client.PerTableConfig;
-import org.apache.xtable.client.PerTableConfigImpl;
-import org.apache.xtable.client.SourceClientProvider;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.PerTableConfig;
+import org.apache.xtable.conversion.PerTableConfigImpl;
import org.apache.xtable.hudi.ConfigurationBasedPartitionSpecExtractor;
import org.apache.xtable.hudi.HudiSourceConfigImpl;
import org.apache.xtable.iceberg.IcebergCatalogConfig;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
import org.apache.xtable.reflection.ReflectionUtils;
-import org.apache.xtable.utilities.RunSync.TableFormatClients.ClientConfig;
+import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
/**
* Provides a standalone runner for the sync process. See README.md for more details on how to run
@@ -66,7 +66,7 @@ public class RunSync {
public static final ObjectMapper YAML_MAPPER = new ObjectMapper(new YAMLFactory());
private static final String DATASET_CONFIG_OPTION = "d";
private static final String HADOOP_CONFIG_PATH = "p";
- private static final String CLIENTS_CONFIG_PATH = "c";
+ private static final String CONVERTERS_CONFIG_PATH = "c";
private static final String ICEBERG_CATALOG_CONFIG_PATH = "i";
private static final String HELP_OPTION = "h";
@@ -84,10 +84,10 @@ public class RunSync {
"Hadoop config xml file path containing configs necessary to access the "
+ "file system. These configs will override the default configs.")
.addOption(
- CLIENTS_CONFIG_PATH,
- "clientsConfig",
+ CONVERTERS_CONFIG_PATH,
+ "convertersConfig",
true,
- "The path to a yaml file containing InternalTable client configurations. "
+ "The path to a yaml file containing InternalTable converter configurations. "
+ "These configs will override the default")
.addOption(
ICEBERG_CATALOG_CONFIG_PATH,
@@ -127,22 +127,23 @@ public class RunSync {
IcebergCatalogConfig icebergCatalogConfig = loadIcebergCatalogConfig(icebergCatalogConfigInput);
String sourceFormat = datasetConfig.sourceFormat;
- customConfig = getCustomConfigurations(cmd, CLIENTS_CONFIG_PATH);
- TableFormatClients tableFormatClients = loadTableFormatClientConfigs(customConfig);
- ClientConfig sourceClientConfig = tableFormatClients.getTableFormatsClients().get(sourceFormat);
- if (sourceClientConfig == null) {
+ customConfig = getCustomConfigurations(cmd, CONVERTERS_CONFIG_PATH);
+ TableFormatConverters tableFormatConverters = loadTableFormatConversionConfigs(customConfig);
+ ConversionConfig sourceConversionConfig =
+ tableFormatConverters.getTableFormatConverters().get(sourceFormat);
+ if (sourceConversionConfig == null) {
throw new IllegalArgumentException(
String.format(
"Source format %s is not supported. Known source and target formats are %s",
- sourceFormat, tableFormatClients.getTableFormatsClients().keySet()));
+ sourceFormat, tableFormatConverters.getTableFormatConverters().keySet()));
}
- String sourceProviderClass = sourceClientConfig.sourceClientProviderClass;
- SourceClientProvider<?> sourceClientProvider =
+ String sourceProviderClass = sourceConversionConfig.conversionSourceProviderClass;
+ ConversionSourceProvider<?> conversionSourceProvider =
ReflectionUtils.createInstanceOfClass(sourceProviderClass);
- sourceClientProvider.init(hadoopConf, sourceClientConfig.configuration);
+ conversionSourceProvider.init(hadoopConf, sourceConversionConfig.configuration);
List<String> tableFormatList = datasetConfig.getTargetFormats();
- OneTableClient client = new OneTableClient(hadoopConf);
+ ConversionController conversionController = new ConversionController(hadoopConf);
for (DatasetConfig.Table table : datasetConfig.getDatasets()) {
log.info(
"Running sync for basePath {} for following table formats {}",
@@ -165,7 +166,7 @@ public class RunSync {
.syncMode(SyncMode.INCREMENTAL)
.build();
try {
- client.sync(config, sourceClientProvider);
+ conversionController.sync(config, conversionSourceProvider);
} catch (Exception e) {
log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
}
@@ -191,22 +192,24 @@ public class RunSync {
}
/**
- * Loads the client configs. The method first loads the default configs and then merges any custom
- * configs provided by the user.
+ * Loads the conversion configs. The method first loads the default configs and then merges any
+ * custom configs provided by the user.
*
* @param customConfigs the custom configs provided by the user
- * @return available tableFormatsClients and their configs
+ * @return available tableFormatConverters and their configs
*/
@VisibleForTesting
- static TableFormatClients loadTableFormatClientConfigs(byte[] customConfigs) throws IOException {
- // get resource stream from default client config yaml file
+ static TableFormatConverters loadTableFormatConversionConfigs(byte[] customConfigs)
+ throws IOException {
+ // get resource stream from default converter config yaml file
try (InputStream inputStream =
- RunSync.class.getClassLoader().getResourceAsStream("onetable-client-defaults.yaml")) {
- TableFormatClients clients = YAML_MAPPER.readValue(inputStream, TableFormatClients.class);
+ RunSync.class.getClassLoader().getResourceAsStream("onetable-conversion-defaults.yaml")) {
+ TableFormatConverters converters =
+ YAML_MAPPER.readValue(inputStream, TableFormatConverters.class);
if (customConfigs != null) {
- YAML_MAPPER.readerForUpdating(clients).readValue(customConfigs);
+ YAML_MAPPER.readerForUpdating(converters).readValue(customConfigs);
}
- return clients;
+ return converters;
}
}
@@ -250,19 +253,22 @@ public class RunSync {
}
@Data
- public static class TableFormatClients {
- /** Map of table format name to the client configs. */
- @JsonProperty("tableFormatsClients")
+ public static class TableFormatConverters {
+ /** Map of table format name to the conversion configs. */
+ @JsonProperty("tableFormatConverters")
@JsonMerge
- Map<String, ClientConfig> tableFormatsClients;
+ Map<String, ConversionConfig> tableFormatConverters;
@Data
- public static class ClientConfig {
- /** The class name of the source client which reads the table metadata. */
- String sourceClientProviderClass;
+ public static class ConversionConfig {
+ /**
+ * The class name of the {@link ConversionSourceProvider} that will generate the {@link
+ * org.apache.xtable.spi.extractor.ConversionSource}.
+ */
+ String conversionSourceProviderClass;
- /** The class name of the target client which writes the table metadata. */
- String targetClientProviderClass;
+ /** The class name of the target converter which writes the table metadata. */
+ String conversionTargetProviderClass;
/** the configuration specific to the table format. */
@JsonMerge Map<String, String> configuration;
diff --git a/utilities/src/main/resources/onetable-client-defaults.yaml b/utilities/src/main/resources/onetable-conversion-defaults.yaml
similarity index 52%
rename from utilities/src/main/resources/onetable-client-defaults.yaml
rename to utilities/src/main/resources/onetable-conversion-defaults.yaml
index 93e5cad3..c80c939b 100644
--- a/utilities/src/main/resources/onetable-client-defaults.yaml
+++ b/utilities/src/main/resources/onetable-conversion-defaults.yaml
@@ -14,27 +14,27 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-## This file contains the default configuration of known InternalTable tableFormatsClients for both source and target
-## table formats. The config maps a table format name to a client class, and contains the default configuration
-## values needed by the client.
+## This file contains the default configuration of known InternalTable tableFormatConverters for both source and target
+## table formats. The config maps a table format name to a converter class, and contains the default configuration
+## values needed by the converter.
-## sourceClientProviderClass: The class name of a table format's client factory, where the client is
+## conversionSourceProviderClass: The class name of a table format's converter factory, where the converter is
## used for reading from a table of this format. All user configurations, including hadoop config
-## and client specific configuration, will be available to the factory for instantiation of the
-## client.
-## targetClientProviderClass: The class name of a table format's client factory, where the client is
+## and converter specific configuration, will be available to the factory for instantiation of the
+## converter.
+## conversionTargetProviderClass: The class name of a table format's converter factory, where the converter is
## used for writing to a table of this format.
-## configuration: A map of configuration values specific to this client.
+## configuration: A map of configuration values specific to this converter.
-tableFormatsClients:
+tableFormatConverters:
HUDI:
- sourceClientProviderClass: org.apache.xtable.hudi.HudiSourceClientProvider
+ conversionSourceProviderClass: org.apache.xtable.hudi.HudiConversionSourceProvider
DELTA:
- sourceClientProviderClass: org.apache.xtable.delta.DeltaSourceClientProvider
- targetClientProviderClass: org.apache.xtable.delta.DeltaClient
+ conversionSourceProviderClass: org.apache.xtable.delta.DeltaConversionSourceProvider
+ conversionTargetProviderClass: org.apache.xtable.delta.DeltaConversionTarget
configuration:
spark.master: local[2]
- spark.app.name: onetableclient
+ spark.app.name: xtable
ICEBERG:
- sourceClientProviderClass: org.apache.xtable.iceberg.IcebergSourceClientProvider
- targetClientProviderClass: org.apache.xtable.iceberg.IcebergTargetClient
\ No newline at end of file
+ conversionSourceProviderClass: org.apache.xtable.iceberg.IcebergConversionSourceProvider
+ conversionTargetProviderClass: org.apache.xtable.iceberg.IcebergConversionTarget
\ No newline at end of file
diff --git a/utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java b/utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java
index 1249f736..a61d948e 100644
--- a/utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java
+++ b/utilities/src/test/java/org/apache/xtable/utilities/TestRunSync.java
@@ -30,8 +30,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.apache.xtable.iceberg.IcebergCatalogConfig;
-import org.apache.xtable.utilities.RunSync.TableFormatClients;
-import org.apache.xtable.utilities.RunSync.TableFormatClients.ClientConfig;
+import org.apache.xtable.utilities.RunSync.TableFormatConverters;
+import org.apache.xtable.utilities.RunSync.TableFormatConverters.ConversionConfig;
class TestRunSync {
@@ -77,50 +77,52 @@ class TestRunSync {
}
@Test
- public void testTableFormatClientConfigDefault() throws IOException {
- TableFormatClients clients = RunSync.loadTableFormatClientConfigs(null);
- Map<String, ClientConfig> tfClients = clients.getTableFormatsClients();
- Assertions.assertEquals(3, tfClients.size());
- Assertions.assertNotNull(tfClients.get(DELTA));
- Assertions.assertNotNull(tfClients.get(HUDI));
- Assertions.assertNotNull(tfClients.get(ICEBERG));
+ public void testTableFormatConverterConfigDefault() throws IOException {
+ TableFormatConverters converters = RunSync.loadTableFormatConversionConfigs(null);
+ Map<String, ConversionConfig> tfConverters = converters.getTableFormatConverters();
+ Assertions.assertEquals(3, tfConverters.size());
+ Assertions.assertNotNull(tfConverters.get(DELTA));
+ Assertions.assertNotNull(tfConverters.get(HUDI));
+ Assertions.assertNotNull(tfConverters.get(ICEBERG));
Assertions.assertEquals(
- "org.apache.xtable.hudi.HudiSourceClientProvider",
- tfClients.get(HUDI).getSourceClientProviderClass());
+ "org.apache.xtable.hudi.HudiConversionSourceProvider",
+ tfConverters.get(HUDI).getConversionSourceProviderClass());
Assertions.assertEquals(
- "org.apache.xtable.iceberg.IcebergTargetClient",
- tfClients.get(ICEBERG).getTargetClientProviderClass());
+ "org.apache.xtable.iceberg.IcebergConversionTarget",
+ tfConverters.get(ICEBERG).getConversionTargetProviderClass());
Assertions.assertEquals(
- "org.apache.xtable.iceberg.IcebergSourceClientProvider",
- tfClients.get(ICEBERG).getSourceClientProviderClass());
+ "org.apache.xtable.iceberg.IcebergConversionSourceProvider",
+ tfConverters.get(ICEBERG).getConversionSourceProviderClass());
}
@Test
- public void testTableFormatClientCustom() throws IOException {
- String customClients =
- "tableFormatsClients:\n"
+ public void testTableFormatConverterCustom() throws IOException {
+ String customConverters =
+ "tableFormatConverters:\n"
+ " HUDI:\n"
- + " sourceClientProviderClass: foo\n"
+ + " conversionSourceProviderClass: foo\n"
+ " DELTA:\n"
+ " configuration:\n"
+ " spark.master: local[4]\n"
+ " foo: bar\n"
+ " NEW_FORMAT:\n"
- + " sourceClientProviderClass: bar\n";
- TableFormatClients clients = RunSync.loadTableFormatClientConfigs(customClients.getBytes());
- Map<String, ClientConfig> tfClients = clients.getTableFormatsClients();
- Assertions.assertEquals(4, tfClients.size());
+ + " conversionSourceProviderClass: bar\n";
+ TableFormatConverters converters =
+ RunSync.loadTableFormatConversionConfigs(customConverters.getBytes());
+ Map<String, ConversionConfig> tfConverters = converters.getTableFormatConverters();
+ Assertions.assertEquals(4, tfConverters.size());
- Assertions.assertNotNull(tfClients.get("NEW_FORMAT"));
- Assertions.assertEquals("bar", tfClients.get("NEW_FORMAT").getSourceClientProviderClass());
+ Assertions.assertNotNull(tfConverters.get("NEW_FORMAT"));
+ Assertions.assertEquals(
+ "bar", tfConverters.get("NEW_FORMAT").getConversionSourceProviderClass());
- Assertions.assertEquals("foo", tfClients.get(HUDI).getSourceClientProviderClass());
+ Assertions.assertEquals("foo", tfConverters.get(HUDI).getConversionSourceProviderClass());
- Map<String, String> deltaClientConfigs = tfClients.get(DELTA).getConfiguration();
- Assertions.assertEquals(3, deltaClientConfigs.size());
- Assertions.assertEquals("local[4]", deltaClientConfigs.get("spark.master"));
- Assertions.assertEquals("bar", deltaClientConfigs.get("foo"));
+ Map<String, String> deltaConfigs = tfConverters.get(DELTA).getConfiguration();
+ Assertions.assertEquals(3, deltaConfigs.size());
+ Assertions.assertEquals("local[4]", deltaConfigs.get("spark.master"));
+ Assertions.assertEquals("bar", deltaConfigs.get("foo"));
}
@Test