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:11 UTC
(incubator-xtable) 01/01: rename and replace 'client'
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