From 30c4d71f87c922acf9f55fa4af67ab515ab543b8 Mon Sep 17 00:00:00 2001
From: Timothy Brown <tim@onehouse.ai>
Date: Wed, 17 Jan 2024 12:16:07 -0700
Subject: [PATCH 1/3] Refactor config classes

---
 .../io/onetable/client/ExternalTable.java     |  67 ++++
 ...HudiSourceConfig.java => SourceTable.java} |  19 +-
 .../java/io/onetable/client/TargetTable.java  |  43 +++
 .../io/onetable/spi/sync/TargetClient.java    |   4 +-
 .../io/onetable/client/TestSourceTable.java   |  58 ++++
 .../io/onetable/client/TestTargetTable.java   |  73 ++++
 .../io/onetable/client/OneTableClient.java    |  24 +-
 .../onetable/client/PerTableConfigImpl.java   | 142 --------
 .../onetable/client/SourceClientProvider.java |  14 +-
 .../client/TableFormatClientFactory.java      |   9 +-
 .../io/onetable/client/TableSyncConfig.java   |  30 +-
 .../java/io/onetable/delta/DeltaClient.java   |  18 +-
 .../delta/DeltaSourceClientProvider.java      |  13 +-
 .../hudi/HudiSourceClientProvider.java        |  13 +-
 .../onetable/hudi/HudiSourceConfigImpl.java   |  32 +-
 .../io/onetable/hudi/HudiTargetClient.java    |  20 +-
 .../io/onetable/iceberg/IcebergClient.java    |  22 +-
 .../onetable/iceberg/IcebergSourceClient.java |  10 +-
 .../iceberg/IcebergSourceClientProvider.java  |   7 +-
 .../java/io/onetable/ITOneTableClient.java    | 311 ++++++++++--------
 .../onetable/client/TestOneTableClient.java   | 100 ++++--
 .../onetable/client/TestPerTableConfig.java   | 118 -------
 .../client/TestTableFormatClientFactory.java  |  35 +-
 .../onetable/delta/ITDeltaSourceClient.java   | 124 +++----
 .../java/io/onetable/delta/TestDeltaSync.java |  12 +-
 .../io/onetable/hudi/ITHudiSourceClient.java  |   4 +-
 .../io/onetable/hudi/ITHudiTargetClient.java  |  12 +-
 .../iceberg/ITIcebergSourceClient.java        |  74 +++--
 .../iceberg/TestIcebergSourceClient.java      |  29 +-
 .../io/onetable/iceberg/TestIcebergSync.java  |  13 +-
 .../java/io/onetable/loadtest/LoadTest.java   |  72 ++--
 .../onetable/hudi/sync/OneTableSyncTool.java  |  51 ++-
 .../java/io/onetable/utilities/RunSync.java   |  58 +++-
 33 files changed, 872 insertions(+), 759 deletions(-)
 create mode 100644 api/src/main/java/io/onetable/client/ExternalTable.java
 rename api/src/main/java/io/onetable/client/{HudiSourceConfig.java => SourceTable.java} (66%)
 create mode 100644 api/src/main/java/io/onetable/client/TargetTable.java
 create mode 100644 api/src/test/java/io/onetable/client/TestSourceTable.java
 create mode 100644 api/src/test/java/io/onetable/client/TestTargetTable.java
 delete mode 100644 core/src/main/java/io/onetable/client/PerTableConfigImpl.java
 rename api/src/main/java/io/onetable/client/PerTableConfig.java => core/src/main/java/io/onetable/client/TableSyncConfig.java (72%)
 delete mode 100644 core/src/test/java/io/onetable/client/TestPerTableConfig.java

diff --git a/api/src/main/java/io/onetable/client/ExternalTable.java b/api/src/main/java/io/onetable/client/ExternalTable.java
new file mode 100644
index 000000000..beedee58a
--- /dev/null
+++ b/api/src/main/java/io/onetable/client/ExternalTable.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package io.onetable.client;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NonNull;
+
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Preconditions;
+
+@Getter
+@EqualsAndHashCode
+public class ExternalTable {
+  @NonNull String name;
+  @NonNull String formatName;
+  @NonNull String basePath;
+  @NonNull String dataPath;
+  String[] namespace;
+  CatalogConfig catalogConfig;
+
+  ExternalTable(
+      @NonNull String name,
+      @NonNull String formatName,
+      @NonNull String basePath,
+      String dataPath,
+      String[] namespace,
+      CatalogConfig catalogConfig) {
+    this.name = name;
+    this.formatName = formatName;
+    this.basePath = sanitizeBasePath(basePath);
+    this.dataPath = dataPath == null ? this.basePath : sanitizeBasePath(dataPath);
+    this.namespace = namespace;
+    this.catalogConfig = catalogConfig;
+  }
+
+  private String sanitizeBasePath(String tableBasePath) {
+    Path path = new Path(tableBasePath);
+    Preconditions.checkArgument(path.isAbsolute(), "Table base path must be absolute");
+    if (path.isAbsoluteAndSchemeAuthorityNull()) {
+      // assume this is local file system and append scheme
+      return "file://" + path;
+    } else if (path.toUri().getScheme().equals("file")) {
+      // add extra slashes
+      return "file://" + path.toUri().getPath();
+    } else {
+      return path.toString();
+    }
+  }
+}
diff --git a/api/src/main/java/io/onetable/client/HudiSourceConfig.java b/api/src/main/java/io/onetable/client/SourceTable.java
similarity index 66%
rename from api/src/main/java/io/onetable/client/HudiSourceConfig.java
rename to api/src/main/java/io/onetable/client/SourceTable.java
index 78d082314..61d7c45b1 100644
--- a/api/src/main/java/io/onetable/client/HudiSourceConfig.java
+++ b/api/src/main/java/io/onetable/client/SourceTable.java
@@ -18,10 +18,19 @@
  
 package io.onetable.client;
 
-import io.onetable.spi.extractor.SourcePartitionSpecExtractor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
 
-public interface HudiSourceConfig {
-  public String getPartitionSpecExtractorClass();
-
-  SourcePartitionSpecExtractor loadSourcePartitionSpecExtractor();
+@EqualsAndHashCode(callSuper = true)
+public class SourceTable extends ExternalTable {
+  @Builder(toBuilder = true)
+  public SourceTable(
+      String name,
+      String formatName,
+      String basePath,
+      String dataPath,
+      String[] namespace,
+      CatalogConfig catalogConfig) {
+    super(name, formatName, basePath, dataPath, namespace, catalogConfig);
+  }
 }
diff --git a/api/src/main/java/io/onetable/client/TargetTable.java b/api/src/main/java/io/onetable/client/TargetTable.java
new file mode 100644
index 000000000..526451872
--- /dev/null
+++ b/api/src/main/java/io/onetable/client/TargetTable.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package io.onetable.client;
+
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+
+@Getter
+@EqualsAndHashCode(callSuper = true)
+public class TargetTable extends ExternalTable {
+  int metadataRetentionInHours;
+
+  @Builder(toBuilder = true)
+  public TargetTable(
+      String name,
+      String formatName,
+      String basePath,
+      String dataPath,
+      String[] namespace,
+      CatalogConfig catalogConfig,
+      Integer metadataRetentionInHours) {
+    super(name, formatName, basePath, dataPath, namespace, catalogConfig);
+    this.metadataRetentionInHours =
+        metadataRetentionInHours == null ? 24 * 7 : metadataRetentionInHours;
+  }
+}
diff --git a/api/src/main/java/io/onetable/spi/sync/TargetClient.java b/api/src/main/java/io/onetable/spi/sync/TargetClient.java
index bd9b4962b..3b9e0beb6 100644
--- a/api/src/main/java/io/onetable/spi/sync/TargetClient.java
+++ b/api/src/main/java/io/onetable/spi/sync/TargetClient.java
@@ -23,7 +23,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 
-import io.onetable.client.PerTableConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.model.OneTable;
 import io.onetable.model.OneTableMetadata;
 import io.onetable.model.schema.OnePartitionField;
@@ -89,5 +89,5 @@ public interface TargetClient {
   String getTableFormat();
 
   /** Initializes the client with provided configuration */
-  void init(PerTableConfig perTableConfig, Configuration configuration);
+  void init(TargetTable targetTable, Configuration configuration);
 }
diff --git a/api/src/test/java/io/onetable/client/TestSourceTable.java b/api/src/test/java/io/onetable/client/TestSourceTable.java
new file mode 100644
index 000000000..fb1daf7f7
--- /dev/null
+++ b/api/src/test/java/io/onetable/client/TestSourceTable.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package io.onetable.client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.Test;
+
+public class TestSourceTable {
+  @Test
+  void sanitizePath() {
+    SourceTable tooManySlashes =
+        SourceTable.builder().basePath("s3://bucket//path").name("name").formatName("hudi").build();
+    assertEquals("s3://bucket/path", tooManySlashes.getBasePath());
+
+    SourceTable localFilePath =
+        SourceTable.builder().basePath("/local/data//path").name("name").formatName("hudi").build();
+    assertEquals("file:///local/data/path", localFilePath.getBasePath());
+
+    SourceTable properLocalFilePath =
+        SourceTable.builder()
+            .basePath("file:///local/data//path")
+            .name("name")
+            .formatName("hudi")
+            .build();
+    assertEquals("file:///local/data/path", properLocalFilePath.getBasePath());
+  }
+
+  @Test
+  void errorIfRequiredArgsNotSet() {
+    assertThrows(NullPointerException.class, () -> SourceTable.builder().name("name").build());
+
+    assertThrows(
+        NullPointerException.class,
+        () -> SourceTable.builder().basePath("file://bucket/path").build());
+
+    assertThrows(
+        NullPointerException.class,
+        () -> SourceTable.builder().basePath("file://bucket/path").name("name").build());
+  }
+}
diff --git a/api/src/test/java/io/onetable/client/TestTargetTable.java b/api/src/test/java/io/onetable/client/TestTargetTable.java
new file mode 100644
index 000000000..a39157684
--- /dev/null
+++ b/api/src/test/java/io/onetable/client/TestTargetTable.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package io.onetable.client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.Test;
+
+public class TestTargetTable {
+  @Test
+  void sanitizePath() {
+    TargetTable tooManySlashes =
+        TargetTable.builder().basePath("s3://bucket//path").name("name").formatName("hudi").build();
+    assertEquals("s3://bucket/path", tooManySlashes.getBasePath());
+
+    TargetTable localFilePath =
+        TargetTable.builder().basePath("/local/data//path").name("name").formatName("hudi").build();
+    assertEquals("file:///local/data/path", localFilePath.getBasePath());
+
+    TargetTable properLocalFilePath =
+        TargetTable.builder()
+            .basePath("file:///local/data//path")
+            .name("name")
+            .formatName("hudi")
+            .build();
+    assertEquals("file:///local/data/path", properLocalFilePath.getBasePath());
+  }
+
+  @Test
+  void defaultValueSet() {
+    TargetTable table =
+        TargetTable.builder()
+            .basePath("file://bucket/path")
+            .name("name")
+            .formatName("hudi")
+            .build();
+
+    assertEquals(24 * 7, table.getMetadataRetentionInHours());
+    assertNull(table.getNamespace());
+    assertNull(table.getCatalogConfig());
+  }
+
+  @Test
+  void errorIfRequiredArgsNotSet() {
+    assertThrows(NullPointerException.class, () -> SourceTable.builder().name("name").build());
+
+    assertThrows(
+        NullPointerException.class,
+        () -> SourceTable.builder().basePath("file://bucket/path").build());
+
+    assertThrows(
+        NullPointerException.class,
+        () -> SourceTable.builder().basePath("file://bucket/path").name("name").build());
+  }
+}
diff --git a/core/src/main/java/io/onetable/client/OneTableClient.java b/core/src/main/java/io/onetable/client/OneTableClient.java
index 0881546f8..b0f8a0681 100644
--- a/core/src/main/java/io/onetable/client/OneTableClient.java
+++ b/core/src/main/java/io/onetable/client/OneTableClient.java
@@ -26,7 +26,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -52,7 +51,7 @@
 
 /**
  * Responsible for completing the entire lifecycle of the sync process given {@link
- * PerTableConfigImpl}. This is done in three steps,
+ * TableSyncConfig}. This is done in three steps,
  *
  * <ul>
  *   <li>1. Extracting snapshot {@link OneSnapshot} from the source table format.
@@ -77,26 +76,27 @@ public OneTableClient(Configuration conf) {
    * @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.
+   *     SourceClientProvider#init(Configuration)} 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) {
-    if (config.getTargetTableFormats() == null || config.getTargetTableFormats().isEmpty()) {
+      TableSyncConfig config, SourceClientProvider<COMMIT> sourceClientProvider) {
+    if (config.getTargetTables() == null || config.getTargetTables().isEmpty()) {
       throw new IllegalArgumentException("Please provide at-least one format to sync");
     }
 
-    try (SourceClient<COMMIT> sourceClient = sourceClientProvider.getSourceClientInstance(config)) {
+    try (SourceClient<COMMIT> sourceClient =
+        sourceClientProvider.getSourceClientInstance(
+            config.getSourceTable(), config.getProperties())) {
       ExtractFromSource<COMMIT> source = ExtractFromSource.of(sourceClient);
 
       Map<String, TargetClient> syncClientByFormat =
-          config.getTargetTableFormats().stream()
+          config.getTargetTables().stream()
               .collect(
                   Collectors.toMap(
-                      Function.identity(),
-                      tableFormat ->
-                          tableFormatClientFactory.createForFormat(tableFormat, config, conf)));
+                      TargetTable::getFormatName,
+                      targetTable -> tableFormatClientFactory.createForFormat(targetTable, conf)));
       // State for each TableFormat
       Map<String, Optional<OneTableMetadata>> lastSyncMetadataByFormat =
           syncClientByFormat.entrySet().stream()
@@ -147,11 +147,11 @@ private static String getFormatsWithStatusCode(
   }
 
   private <COMMIT> Map<String, TargetClient> getFormatsToSyncIncrementally(
-      PerTableConfig perTableConfig,
+      TableSyncConfig syncConfig,
       Map<String, TargetClient> syncClientByFormat,
       Map<String, Optional<OneTableMetadata>> lastSyncMetadataByFormat,
       SourceClient<COMMIT> sourceClient) {
-    if (perTableConfig.getSyncMode() == SyncMode.FULL) {
+    if (syncConfig.getSyncMode() == SyncMode.FULL) {
       // Full sync requested by config, hence no incremental sync.
       return Collections.emptyMap();
     }
diff --git a/core/src/main/java/io/onetable/client/PerTableConfigImpl.java b/core/src/main/java/io/onetable/client/PerTableConfigImpl.java
deleted file mode 100644
index a17aeae7b..000000000
--- a/core/src/main/java/io/onetable/client/PerTableConfigImpl.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
- 
-package io.onetable.client;
-
-import java.util.List;
-
-import javax.annotation.Nonnull;
-
-import lombok.Builder;
-import lombok.NonNull;
-import lombok.Value;
-
-import org.apache.hadoop.fs.Path;
-
-import com.google.common.base.Preconditions;
-
-import io.onetable.hudi.HudiSourceConfigImpl;
-import io.onetable.iceberg.IcebergCatalogConfig;
-import io.onetable.model.sync.SyncMode;
-
-/** Represents input configuration to the sync process. */
-@Value
-public class PerTableConfigImpl implements PerTableConfig {
-  /** table base path in local file system or HDFS or object stores like S3, GCS etc. */
-  @Nonnull String tableBasePath;
-  /** the base path for the data folder, defaults to the tableBasePath if not specified */
-  @Nonnull String tableDataPath;
-
-  /** The name of the table */
-  @Nonnull String tableName;
-
-  /** The namespace of the table (optional) */
-  String[] namespace;
-
-  /**
-   * HudiSourceConfig is a config that allows us to infer partition values for hoodie source tables.
-   * If the table is not partitioned, leave it blank. If it is partitioned, you can specify a spec
-   * with a comma separated list with format path:type:format.
-   *
-   * <p><ui>
-   * <li>partitionSpecExtractorClass: class to extract partition fields from the given
-   *     spec.ConfigurationBasedPartitionSpecExtractor is the default class
-   * <li>partitionFieldSpecConfig: path:type:format spec to infer partition values </ui>
-   *
-   *     <ul>
-   *       <li>path: is a dot separated path to the partition field
-   *       <li>type: describes how the partition value was generated from the column value
-   *           <ul>
-   *             <li>VALUE: an identity transform of field value to partition value
-   *             <li>YEAR: data is partitioned by a field representing a date and year granularity
-   *                 is used
-   *             <li>MONTH: same as YEAR but with month granularity
-   *             <li>DAY: same as YEAR but with day granularity
-   *             <li>HOUR: same as YEAR but with hour granularity
-   *           </ul>
-   *       <li>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
-   *     </ul>
-   */
-  @Nonnull HudiSourceConfigImpl hudiSourceConfig;
-
-  /** List of table formats to sync. */
-  @Nonnull List<String> targetTableFormats;
-
-  /** Configuration options for integrating with an existing Iceberg Catalog (optional) */
-  IcebergCatalogConfig icebergCatalogConfig;
-
-  /**
-   * Mode of a sync. FULL is only supported right now.
-   *
-   * <ul>
-   *   <li>FULL: Full sync will create a checkpoint of ALL the files relevant at a certain point in
-   *       time
-   *   <li>INCREMENTAL: Incremental will sync differential structures to bring the table state from
-   *       and to points in the timeline
-   * </ul>
-   */
-  @Nonnull SyncMode syncMode;
-
-  /**
-   * The retention for metadata or versions of the table in the target systems to bound the size of
-   * any metadata tracked in the target system. Specified in hours.
-   */
-  int targetMetadataRetentionInHours;
-
-  @Builder
-  PerTableConfigImpl(
-      @NonNull String tableBasePath,
-      String tableDataPath,
-      @NonNull String tableName,
-      String[] namespace,
-      HudiSourceConfigImpl hudiSourceConfig,
-      @NonNull List<String> targetTableFormats,
-      IcebergCatalogConfig icebergCatalogConfig,
-      SyncMode syncMode,
-      Integer targetMetadataRetentionInHours) {
-    // sanitize source path
-    this.tableBasePath = sanitizeBasePath(tableBasePath);
-    this.tableDataPath = tableDataPath == null ? tableBasePath : sanitizeBasePath(tableDataPath);
-    this.tableName = tableName;
-    this.namespace = namespace;
-    this.hudiSourceConfig =
-        hudiSourceConfig == null ? HudiSourceConfigImpl.builder().build() : hudiSourceConfig;
-    Preconditions.checkArgument(
-        targetTableFormats.size() > 0, "Please provide at-least one format to sync");
-    this.targetTableFormats = targetTableFormats;
-    this.icebergCatalogConfig = icebergCatalogConfig;
-    this.syncMode = syncMode == null ? SyncMode.INCREMENTAL : syncMode;
-    this.targetMetadataRetentionInHours =
-        targetMetadataRetentionInHours == null ? 24 * 7 : targetMetadataRetentionInHours;
-  }
-
-  private String sanitizeBasePath(String tableBasePath) {
-    Path path = new Path(tableBasePath);
-    Preconditions.checkArgument(path.isAbsolute(), "Table base path must be absolute");
-    if (path.isAbsoluteAndSchemeAuthorityNull()) {
-      // assume this is local file system and append scheme
-      return "file://" + path;
-    } else if (path.toUri().getScheme().equals("file")) {
-      // add extra slashes
-      return "file://" + path.toUri().getPath();
-    } else {
-      return path.toString();
-    }
-  }
-}
diff --git a/core/src/main/java/io/onetable/client/SourceClientProvider.java b/core/src/main/java/io/onetable/client/SourceClientProvider.java
index 098ae145d..d5dfbfd3e 100644
--- a/core/src/main/java/io/onetable/client/SourceClientProvider.java
+++ b/core/src/main/java/io/onetable/client/SourceClientProvider.java
@@ -33,16 +33,9 @@ public abstract class SourceClientProvider<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 table to read from. */
-  protected PerTableConfig sourceTableConfig;
-
   /** Initializes the provider various client specific configurations. */
-  public void init(Configuration hadoopConf, Map<String, String> clientConf) {
-    this.hadoopConf = hadoopConf;
-    this.clientConf = clientConf;
+  public void init(Configuration hadoopConf) {
+    this.hadoopConf = hadoopConf; // todo can we move this into some other config?
   }
 
   /**
@@ -53,5 +46,6 @@ public void init(Configuration hadoopConf, Map<String, String> clientConf) {
    *
    * @return the source client
    */
-  public abstract SourceClient<COMMIT> getSourceClientInstance(PerTableConfig sourceTableConfig);
+  public abstract SourceClient<COMMIT> getSourceClientInstance(
+      SourceTable sourceTableConfig, Map<String, String> clientConf);
 }
diff --git a/core/src/main/java/io/onetable/client/TableFormatClientFactory.java b/core/src/main/java/io/onetable/client/TableFormatClientFactory.java
index 9d085d526..cd6491087 100644
--- a/core/src/main/java/io/onetable/client/TableFormatClientFactory.java
+++ b/core/src/main/java/io/onetable/client/TableFormatClientFactory.java
@@ -42,15 +42,14 @@ public static TableFormatClientFactory getInstance() {
    * params.
    *
    * @param tableFormat
-   * @param perTableConfig
+   * @param targetTable
    * @param configuration
    * @return
    */
-  public TargetClient createForFormat(
-      String tableFormat, PerTableConfig perTableConfig, Configuration configuration) {
-    TargetClient targetClient = createTargetClientForName(tableFormat);
+  public TargetClient createForFormat(TargetTable targetTable, Configuration configuration) {
+    TargetClient targetClient = createTargetClientForName(targetTable.getFormatName());
 
-    targetClient.init(perTableConfig, configuration);
+    targetClient.init(targetTable, configuration);
     return targetClient;
   }
 
diff --git a/api/src/main/java/io/onetable/client/PerTableConfig.java b/core/src/main/java/io/onetable/client/TableSyncConfig.java
similarity index 72%
rename from api/src/main/java/io/onetable/client/PerTableConfig.java
rename to core/src/main/java/io/onetable/client/TableSyncConfig.java
index 53efd36bb..263064215 100644
--- a/api/src/main/java/io/onetable/client/PerTableConfig.java
+++ b/core/src/main/java/io/onetable/client/TableSyncConfig.java
@@ -19,25 +19,19 @@
 package io.onetable.client;
 
 import java.util.List;
+import java.util.Map;
 
-import io.onetable.model.sync.SyncMode;
-
-public interface PerTableConfig {
-  int getTargetMetadataRetentionInHours();
-
-  String getTableBasePath();
-
-  String getTableDataPath();
-
-  String getTableName();
+import lombok.Builder;
+import lombok.Value;
 
-  HudiSourceConfig getHudiSourceConfig();
-
-  List<String> getTargetTableFormats();
-
-  SyncMode getSyncMode();
-
-  String[] getNamespace();
+import io.onetable.model.sync.SyncMode;
 
-  CatalogConfig getIcebergCatalogConfig();
+@Value
+@Builder(toBuilder = true)
+public class TableSyncConfig {
+  // TODO add docs for all fields
+  SourceTable sourceTable;
+  List<TargetTable> targetTables;
+  SyncMode syncMode;
+  Map<String, String> properties;
 }
diff --git a/core/src/main/java/io/onetable/delta/DeltaClient.java b/core/src/main/java/io/onetable/delta/DeltaClient.java
index 8b2ffdbbe..69012a9f9 100644
--- a/core/src/main/java/io/onetable/delta/DeltaClient.java
+++ b/core/src/main/java/io/onetable/delta/DeltaClient.java
@@ -53,7 +53,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 
-import io.onetable.client.PerTableConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.exception.NotSupportedException;
 import io.onetable.model.OneTable;
 import io.onetable.model.OneTableMetadata;
@@ -80,11 +80,11 @@ public class DeltaClient implements TargetClient {
 
   public DeltaClient() {}
 
-  public DeltaClient(PerTableConfig perTableConfig, SparkSession sparkSession) {
+  public DeltaClient(TargetTable targetTable, SparkSession sparkSession) {
     this(
-        perTableConfig.getTableDataPath(),
-        perTableConfig.getTableName(),
-        perTableConfig.getTargetMetadataRetentionInHours(),
+        targetTable.getDataPath(),
+        targetTable.getName(),
+        targetTable.getMetadataRetentionInHours(),
         sparkSession,
         DeltaSchemaExtractor.getInstance(),
         DeltaPartitionExtractor.getInstance(),
@@ -133,13 +133,13 @@ private void _init(
   }
 
   @Override
-  public void init(PerTableConfig perTableConfig, Configuration configuration) {
+  public void init(TargetTable targetTable, Configuration configuration) {
     SparkSession sparkSession = DeltaClientUtils.buildSparkSession(configuration);
 
     _init(
-        perTableConfig.getTableDataPath(),
-        perTableConfig.getTableName(),
-        perTableConfig.getTargetMetadataRetentionInHours(),
+        targetTable.getDataPath(),
+        targetTable.getName(),
+        targetTable.getMetadataRetentionInHours(),
         sparkSession,
         DeltaSchemaExtractor.getInstance(),
         DeltaPartitionExtractor.getInstance(),
diff --git a/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java b/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java
index 78098d06c..53efcc33c 100644
--- a/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java
+++ b/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java
@@ -18,24 +18,27 @@
  
 package io.onetable.delta;
 
+import java.util.Map;
+
 import org.apache.spark.sql.SparkSession;
 
 import io.delta.tables.DeltaTable;
 
-import io.onetable.client.PerTableConfig;
 import io.onetable.client.SourceClientProvider;
+import io.onetable.client.SourceTable;
 
 /** A concrete implementation of {@link SourceClientProvider} for Delta Lake table format. */
 public class DeltaSourceClientProvider extends SourceClientProvider<Long> {
   @Override
-  public DeltaSourceClient getSourceClientInstance(PerTableConfig perTableConfig) {
+  public DeltaSourceClient getSourceClientInstance(
+      SourceTable sourceTable, Map<String, String> clientConfigs) {
     SparkSession sparkSession = DeltaClientUtils.buildSparkSession(hadoopConf);
-    DeltaTable deltaTable = DeltaTable.forPath(sparkSession, perTableConfig.getTableBasePath());
+    DeltaTable deltaTable = DeltaTable.forPath(sparkSession, sourceTable.getBasePath());
     DeltaSourceClient deltaSourceClient =
         DeltaSourceClient.builder()
             .sparkSession(sparkSession)
-            .tableName(perTableConfig.getTableName())
-            .basePath(perTableConfig.getTableBasePath())
+            .tableName(sourceTable.getName())
+            .basePath(sourceTable.getBasePath())
             .deltaTable(deltaTable)
             .deltaLog(deltaTable.deltaLog())
             .build();
diff --git a/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java b/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
index 5bd20ebe2..fa8fd9ce8 100644
--- a/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
+++ b/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
@@ -18,26 +18,28 @@
  
 package io.onetable.hudi;
 
+import java.util.Map;
+
 import lombok.extern.log4j.Log4j2;
 
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 
-import io.onetable.client.PerTableConfig;
 import io.onetable.client.SourceClientProvider;
+import io.onetable.client.SourceTable;
 
 /** A concrete implementation of {@link SourceClientProvider} for Hudi table format. */
 @Log4j2
 public class HudiSourceClientProvider extends SourceClientProvider<HoodieInstant> {
 
   @Override
-  public HudiClient getSourceClientInstance(PerTableConfig sourceTableConfig) {
-    this.sourceTableConfig = sourceTableConfig;
+  public HudiClient getSourceClientInstance(
+      SourceTable sourceTableConfig, Map<String, String> clientConfigs) {
     HoodieTableMetaClient metaClient =
         HoodieTableMetaClient.builder()
             .setConf(hadoopConf)
-            .setBasePath(this.sourceTableConfig.getTableBasePath())
+            .setBasePath(sourceTableConfig.getBasePath())
             .setLoadActiveTimelineOnLoad(true)
             .build();
     if (!metaClient.getTableConfig().getTableType().equals(HoodieTableType.COPY_ON_WRITE)) {
@@ -45,8 +47,7 @@ public HudiClient getSourceClientInstance(PerTableConfig sourceTableConfig) {
     }
 
     final HudiSourcePartitionSpecExtractor sourcePartitionSpecExtractor =
-        (HudiSourcePartitionSpecExtractor)
-            sourceTableConfig.getHudiSourceConfig().loadSourcePartitionSpecExtractor();
+        HudiSourceConfigImpl.fromProperties(clientConfigs).loadSourcePartitionSpecExtractor();
 
     return new HudiClient(metaClient, sourcePartitionSpecExtractor);
   }
diff --git a/core/src/main/java/io/onetable/hudi/HudiSourceConfigImpl.java b/core/src/main/java/io/onetable/hudi/HudiSourceConfigImpl.java
index ee5b802b3..bb86ff5cd 100644
--- a/core/src/main/java/io/onetable/hudi/HudiSourceConfigImpl.java
+++ b/core/src/main/java/io/onetable/hudi/HudiSourceConfigImpl.java
@@ -22,29 +22,41 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 
-import lombok.Builder;
 import lombok.Value;
 
 import com.google.common.base.Preconditions;
 
-import io.onetable.client.HudiSourceConfig;
 import io.onetable.model.schema.PartitionTransformType;
 import io.onetable.reflection.ReflectionUtils;
 
 /** Configuration of Hudi source format for the sync process. */
 @Value
-public class HudiSourceConfigImpl implements HudiSourceConfig {
+public class HudiSourceConfigImpl {
+  public static final String PARTITION_SPEC_EXTRACTOR_CLASS =
+      "onetable.hudi.source.partition_spec_extractor_class";
+  public static final String PARTITION_FIELD_SPEC_CONFIG =
+      "onetable.hudi.source.partition_field_spec_config";
+
   String partitionSpecExtractorClass;
   List<PartitionFieldSpec> partitionFieldSpecs;
 
-  @Builder
-  public HudiSourceConfigImpl(String partitionSpecExtractorClass, String partitionFieldSpecConfig) {
-    this.partitionSpecExtractorClass =
-        partitionSpecExtractorClass == null
-            ? ConfigurationBasedPartitionSpecExtractor.class.getName()
-            : partitionSpecExtractorClass;
-    this.partitionFieldSpecs = parsePartitionFieldSpecs(partitionFieldSpecConfig);
+  public static HudiSourceConfigImpl fromPartitionFieldSpecConfig(String partitionFieldSpecConfig) {
+    return new HudiSourceConfigImpl(
+        ConfigurationBasedPartitionSpecExtractor.class.getName(),
+        parsePartitionFieldSpecs(partitionFieldSpecConfig));
+  }
+
+  public static HudiSourceConfigImpl fromProperties(Map<String, String> properties) {
+    String partitionSpecExtractorClass =
+        properties.getOrDefault(
+            PARTITION_SPEC_EXTRACTOR_CLASS,
+            ConfigurationBasedPartitionSpecExtractor.class.getName());
+    String partitionFieldSpecString = properties.get(PARTITION_FIELD_SPEC_CONFIG);
+    List<PartitionFieldSpec> partitionFieldSpecs =
+        parsePartitionFieldSpecs(partitionFieldSpecString);
+    return new HudiSourceConfigImpl(partitionSpecExtractorClass, partitionFieldSpecs);
   }
 
   @Value
diff --git a/core/src/main/java/io/onetable/hudi/HudiTargetClient.java b/core/src/main/java/io/onetable/hudi/HudiTargetClient.java
index 3c34b91a0..639d6002f 100644
--- a/core/src/main/java/io/onetable/hudi/HudiTargetClient.java
+++ b/core/src/main/java/io/onetable/hudi/HudiTargetClient.java
@@ -81,7 +81,7 @@
 import com.google.common.annotations.VisibleForTesting;
 
 import io.onetable.avro.AvroSchemaConverter;
-import io.onetable.client.PerTableConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.exception.NotSupportedException;
 import io.onetable.exception.OneIOException;
 import io.onetable.model.OneTable;
@@ -110,16 +110,15 @@ public HudiTargetClient() {}
 
   @VisibleForTesting
   HudiTargetClient(
-      PerTableConfig perTableConfig,
+      TargetTable targetTable,
       Configuration configuration,
       int maxNumDeltaCommitsBeforeCompaction) {
     this(
-        perTableConfig.getTableDataPath(),
-        perTableConfig.getTargetMetadataRetentionInHours(),
+        targetTable.getDataPath(),
+        targetTable.getMetadataRetentionInHours(),
         maxNumDeltaCommitsBeforeCompaction,
         BaseFileUpdatesExtractor.of(
-            new HoodieJavaEngineContext(configuration),
-            new CachingPath(perTableConfig.getTableDataPath())),
+            new HoodieJavaEngineContext(configuration), new CachingPath(targetTable.getDataPath())),
         AvroSchemaConverter.getInstance(),
         HudiTableManager.of(configuration),
         CommitState::new);
@@ -165,14 +164,13 @@ private void _init(
   }
 
   @Override
-  public void init(PerTableConfig perTableConfig, Configuration configuration) {
+  public void init(TargetTable targetTable, Configuration configuration) {
     _init(
-        perTableConfig.getTableDataPath(),
-        perTableConfig.getTargetMetadataRetentionInHours(),
+        targetTable.getDataPath(),
+        targetTable.getMetadataRetentionInHours(),
         HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.defaultValue(),
         BaseFileUpdatesExtractor.of(
-            new HoodieJavaEngineContext(configuration),
-            new CachingPath(perTableConfig.getTableDataPath())),
+            new HoodieJavaEngineContext(configuration), new CachingPath(targetTable.getDataPath())),
         AvroSchemaConverter.getInstance(),
         HudiTableManager.of(configuration),
         CommitState::new);
diff --git a/core/src/main/java/io/onetable/iceberg/IcebergClient.java b/core/src/main/java/io/onetable/iceberg/IcebergClient.java
index 36f48a07d..30b9a02e7 100644
--- a/core/src/main/java/io/onetable/iceberg/IcebergClient.java
+++ b/core/src/main/java/io/onetable/iceberg/IcebergClient.java
@@ -43,7 +43,7 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NotFoundException;
 
-import io.onetable.client.PerTableConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.model.OneTable;
 import io.onetable.model.OneTableMetadata;
 import io.onetable.model.schema.OnePartitionField;
@@ -74,7 +74,7 @@ public class IcebergClient implements TargetClient {
   public IcebergClient() {}
 
   IcebergClient(
-      PerTableConfig perTableConfig,
+      TargetTable targetTable,
       Configuration configuration,
       IcebergSchemaExtractor schemaExtractor,
       IcebergSchemaSync schemaSync,
@@ -83,7 +83,7 @@ public IcebergClient() {}
       IcebergDataFileUpdatesSync dataFileUpdatesExtractor,
       IcebergTableManager tableManager) {
     _init(
-        perTableConfig,
+        targetTable,
         configuration,
         schemaExtractor,
         schemaSync,
@@ -94,7 +94,7 @@ public IcebergClient() {}
   }
 
   private void _init(
-      PerTableConfig perTableConfig,
+      TargetTable targetTable,
       Configuration configuration,
       IcebergSchemaExtractor schemaExtractor,
       IcebergSchemaSync schemaSync,
@@ -107,17 +107,17 @@ private void _init(
     this.partitionSpecExtractor = partitionSpecExtractor;
     this.partitionSpecSync = partitionSpecSync;
     this.dataFileUpdatesExtractor = dataFileUpdatesExtractor;
-    String tableName = perTableConfig.getTableName();
-    this.basePath = perTableConfig.getTableBasePath();
+    String tableName = targetTable.getName();
+    this.basePath = targetTable.getBasePath();
     this.configuration = configuration;
-    this.snapshotRetentionInHours = perTableConfig.getTargetMetadataRetentionInHours();
-    String[] namespace = perTableConfig.getNamespace();
+    this.snapshotRetentionInHours = targetTable.getMetadataRetentionInHours();
+    String[] namespace = targetTable.getNamespace();
     this.tableIdentifier =
         namespace == null
             ? TableIdentifier.of(tableName)
             : TableIdentifier.of(Namespace.of(namespace), tableName);
     this.tableManager = tableManager;
-    this.catalogConfig = (IcebergCatalogConfig) perTableConfig.getIcebergCatalogConfig();
+    this.catalogConfig = (IcebergCatalogConfig) targetTable.getCatalogConfig();
 
     if (tableManager.tableExists(catalogConfig, tableIdentifier, basePath)) {
       // Load the table state if it already exists
@@ -128,9 +128,9 @@ private void _init(
   }
 
   @Override
-  public void init(PerTableConfig perTableConfig, Configuration configuration) {
+  public void init(TargetTable targetTable, Configuration configuration) {
     _init(
-        perTableConfig,
+        targetTable,
         configuration,
         IcebergSchemaExtractor.getInstance(),
         IcebergSchemaSync.getInstance(),
diff --git a/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java b/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java
index 1ad53c1f6..c83d5c3df 100644
--- a/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java
+++ b/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java
@@ -36,7 +36,7 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
 
-import io.onetable.client.PerTableConfig;
+import io.onetable.client.SourceTable;
 import io.onetable.exception.OneIOException;
 import io.onetable.model.*;
 import io.onetable.model.schema.OnePartitionField;
@@ -55,7 +55,7 @@
 @Builder
 public class IcebergSourceClient implements SourceClient<Snapshot> {
   @NonNull private final Configuration hadoopConf;
-  @NonNull private final PerTableConfig sourceTableConfig;
+  @NonNull private final SourceTable sourceTableConfig;
 
   @Getter(lazy = true, value = AccessLevel.PACKAGE)
   private final Table sourceTable = initSourceTable();
@@ -74,15 +74,15 @@ public class IcebergSourceClient implements SourceClient<Snapshot> {
   private Table initSourceTable() {
     IcebergTableManager tableManager = IcebergTableManager.of(hadoopConf);
     String[] namespace = sourceTableConfig.getNamespace();
-    String tableName = sourceTableConfig.getTableName();
+    String tableName = sourceTableConfig.getName();
     TableIdentifier tableIdentifier =
         namespace == null
             ? TableIdentifier.of(tableName)
             : TableIdentifier.of(Namespace.of(namespace), tableName);
     return tableManager.getTable(
-        (IcebergCatalogConfig) sourceTableConfig.getIcebergCatalogConfig(),
+        (IcebergCatalogConfig) sourceTableConfig.getCatalogConfig(),
         tableIdentifier,
-        sourceTableConfig.getTableBasePath());
+        sourceTableConfig.getBasePath());
   }
 
   private FileIO initTableOps() {
diff --git a/core/src/main/java/io/onetable/iceberg/IcebergSourceClientProvider.java b/core/src/main/java/io/onetable/iceberg/IcebergSourceClientProvider.java
index ac67f2186..8299c6de2 100644
--- a/core/src/main/java/io/onetable/iceberg/IcebergSourceClientProvider.java
+++ b/core/src/main/java/io/onetable/iceberg/IcebergSourceClientProvider.java
@@ -18,15 +18,18 @@
  
 package io.onetable.iceberg;
 
+import java.util.Map;
+
 import org.apache.iceberg.Snapshot;
 
-import io.onetable.client.PerTableConfig;
 import io.onetable.client.SourceClientProvider;
+import io.onetable.client.SourceTable;
 
 /** A concrete implementation of {@link SourceClientProvider} for Hudi table format. */
 public class IcebergSourceClientProvider extends SourceClientProvider<Snapshot> {
   @Override
-  public IcebergSourceClient getSourceClientInstance(PerTableConfig sourceTableConfig) {
+  public IcebergSourceClient getSourceClientInstance(
+      SourceTable sourceTableConfig, Map<String, String> clientConf) {
     return IcebergSourceClient.builder()
         .sourceTableConfig(sourceTableConfig)
         .hadoopConf(hadoopConf)
diff --git a/core/src/test/java/io/onetable/ITOneTableClient.java b/core/src/test/java/io/onetable/ITOneTableClient.java
index 9e4b9d0c8..a67de91e0 100644
--- a/core/src/test/java/io/onetable/ITOneTableClient.java
+++ b/core/src/test/java/io/onetable/ITOneTableClient.java
@@ -19,6 +19,7 @@
 package io.onetable;
 
 import static io.onetable.GenericTable.getTableName;
+import static io.onetable.hudi.HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG;
 import static io.onetable.hudi.HudiTestUtil.PartitionConfig;
 import static io.onetable.model.storage.TableFormat.DELTA;
 import static io.onetable.model.storage.TableFormat.HUDI;
@@ -82,12 +83,12 @@
 import com.google.common.collect.ImmutableList;
 
 import io.onetable.client.OneTableClient;
-import io.onetable.client.PerTableConfig;
-import io.onetable.client.PerTableConfigImpl;
 import io.onetable.client.SourceClientProvider;
+import io.onetable.client.SourceTable;
+import io.onetable.client.TableSyncConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.delta.DeltaSourceClientProvider;
 import io.onetable.hudi.HudiSourceClientProvider;
-import io.onetable.hudi.HudiSourceConfigImpl;
 import io.onetable.hudi.HudiTestUtil;
 import io.onetable.iceberg.IcebergSourceClientProvider;
 import io.onetable.model.storage.TableFormat;
@@ -146,16 +147,16 @@ private static Stream<Arguments> testCasesWithSyncModes() {
   private SourceClientProvider<?> getSourceClientProvider(String sourceTableFormat) {
     if (sourceTableFormat.equalsIgnoreCase(HUDI)) {
       SourceClientProvider<HoodieInstant> hudiSourceClientProvider = new HudiSourceClientProvider();
-      hudiSourceClientProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
+      hudiSourceClientProvider.init(jsc.hadoopConfiguration());
       return hudiSourceClientProvider;
     } else if (sourceTableFormat.equalsIgnoreCase(DELTA)) {
       SourceClientProvider<Long> deltaSourceClientProvider = new DeltaSourceClientProvider();
-      deltaSourceClientProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
+      deltaSourceClientProvider.init(jsc.hadoopConfiguration());
       return deltaSourceClientProvider;
     } else if (sourceTableFormat.equalsIgnoreCase(ICEBERG)) {
       SourceClientProvider<Snapshot> icebergSourceClientProvider =
           new IcebergSourceClientProvider();
-      icebergSourceClientProvider.init(jsc.hadoopConfiguration(), Collections.emptyMap());
+      icebergSourceClientProvider.init(jsc.hadoopConfiguration());
       return icebergSourceClientProvider;
     } else {
       throw new IllegalArgumentException("Unsupported source format: " + sourceTableFormat);
@@ -190,29 +191,26 @@ public void testVariousOperations(
             tableName, tempDir, sparkSession, jsc, sourceTableFormat, isPartitioned)) {
       insertRecords = table.insertRows(100);
 
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(targetTableFormats)
-              .tableBasePath(table.getBasePath())
-              .tableDataPath(table.getDataPath())
-              .hudiSourceConfig(
-                  HudiSourceConfigImpl.builder()
-                      .partitionFieldSpecConfig(oneTablePartitionConfig)
-                      .build())
-              .syncMode(syncMode)
-              .build();
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              sourceTableFormat,
+              syncMode,
+              tableName,
+              table,
+              targetTableFormats,
+              oneTablePartitionConfig,
+              null);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 100);
 
       // make multiple commits and then sync
       table.insertRows(100);
       table.upsertRows(insertRecords.subList(0, 20));
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 200);
 
       table.deleteRows(insertRecords.subList(30, 50));
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 180);
       checkDatasetEquivalenceWithFilter(
           sourceTableFormat, table, targetTableFormats, table.getFilterQuery());
@@ -221,41 +219,38 @@ public void testVariousOperations(
     try (GenericTable tableWithUpdatedSchema =
         GenericTable.getInstanceWithAdditionalColumns(
             tableName, tempDir, sparkSession, jsc, sourceTableFormat, isPartitioned)) {
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(targetTableFormats)
-              .tableBasePath(tableWithUpdatedSchema.getBasePath())
-              .tableDataPath(tableWithUpdatedSchema.getDataPath())
-              .hudiSourceConfig(
-                  HudiSourceConfigImpl.builder()
-                      .partitionFieldSpecConfig(oneTablePartitionConfig)
-                      .build())
-              .syncMode(syncMode)
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              sourceTableFormat,
+              syncMode,
+              tableName,
+              tableWithUpdatedSchema,
+              targetTableFormats,
+              oneTablePartitionConfig,
+              null);
       List<Row> insertsAfterSchemaUpdate = tableWithUpdatedSchema.insertRows(100);
       tableWithUpdatedSchema.reload();
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 280);
 
       tableWithUpdatedSchema.deleteRows(insertsAfterSchemaUpdate.subList(60, 90));
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 250);
 
       if (isPartitioned) {
         // Adds new partition.
         tableWithUpdatedSchema.insertRecordsForSpecialPartition(50);
-        oneTableClient.sync(perTableConfig, sourceClientProvider);
+        oneTableClient.sync(tableSyncConfig, sourceClientProvider);
         checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 300);
 
         // Drops partition.
         tableWithUpdatedSchema.deleteSpecialPartition();
-        oneTableClient.sync(perTableConfig, sourceClientProvider);
+        oneTableClient.sync(tableSyncConfig, sourceClientProvider);
         checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 250);
 
         // Insert records to the dropped partition again.
         tableWithUpdatedSchema.insertRecordsForSpecialPartition(50);
-        oneTableClient.sync(perTableConfig, sourceClientProvider);
+        oneTableClient.sync(tableSyncConfig, sourceClientProvider);
         checkDatasetEquivalence(sourceTableFormat, tableWithUpdatedSchema, targetTableFormats, 300);
       }
     }
@@ -280,23 +275,21 @@ public void testConcurrentInsertWritesInSource(
       String commitInstant2 = table.startCommit();
       table.insertRecordsWithCommitAlreadyStarted(insertsForCommit2, commitInstant2, true);
 
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(targetTableFormats)
-              .tableBasePath(table.getBasePath())
-              .hudiSourceConfig(
-                  HudiSourceConfigImpl.builder()
-                      .partitionFieldSpecConfig(partitionConfig.getOneTableConfig())
-                      .build())
-              .syncMode(syncMode)
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              HUDI,
+              syncMode,
+              tableName,
+              table,
+              targetTableFormats,
+              partitionConfig.getOneTableConfig(),
+              null);
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
 
       checkDatasetEquivalence(HUDI, table, targetTableFormats, 50);
       table.insertRecordsWithCommitAlreadyStarted(insertsForCommit1, commitInstant1, true);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, targetTableFormats, 100);
     }
   }
@@ -314,19 +307,17 @@ public void testConcurrentInsertsAndTableServiceWrites(
             tableName, tempDir, jsc, partitionConfig.getHudiConfig(), tableType)) {
       List<HoodieRecord<HoodieAvroPayload>> insertedRecords1 = table.insertRecords(50, true);
 
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(targetTableFormats)
-              .tableBasePath(table.getBasePath())
-              .hudiSourceConfig(
-                  HudiSourceConfigImpl.builder()
-                      .partitionFieldSpecConfig(partitionConfig.getOneTableConfig())
-                      .build())
-              .syncMode(syncMode)
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              HUDI,
+              syncMode,
+              tableName,
+              table,
+              targetTableFormats,
+              partitionConfig.getOneTableConfig(),
+              null);
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, targetTableFormats, 50);
 
       table.deleteRecords(insertedRecords1.subList(0, 20), true);
@@ -334,7 +325,7 @@ public void testConcurrentInsertsAndTableServiceWrites(
       String scheduledCompactionInstant = table.onlyScheduleCompaction();
 
       table.insertRecords(50, true);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       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 +333,13 @@ public void testConcurrentInsertsAndTableServiceWrites(
           HUDI, table, sourceHudiOptions, targetTableFormats, Collections.emptyMap(), 100);
 
       table.insertRecords(50, true);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       // 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);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, targetTableFormats, 130);
     }
   }
@@ -361,29 +352,31 @@ public void testTimeTravelQueries(String sourceTableFormat) throws Exception {
         GenericTable.getInstance(tableName, tempDir, sparkSession, jsc, sourceTableFormat, false)) {
       table.insertRows(50);
       List<String> targetTableFormats = getOtherFormats(sourceTableFormat);
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(targetTableFormats)
-              .tableBasePath(table.getBasePath())
-              .tableDataPath(table.getDataPath())
-              .syncMode(SyncMode.INCREMENTAL)
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              sourceTableFormat,
+              SyncMode.INCREMENTAL,
+              tableName,
+              table,
+              targetTableFormats,
+              null,
+              null);
+
       SourceClientProvider<?> sourceClientProvider = getSourceClientProvider(sourceTableFormat);
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       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);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       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);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
 
       checkDatasetEquivalence(
           sourceTableFormat,
@@ -481,24 +474,21 @@ public void testPartitionedData(TableFormatPartitionDataHolder tableFormatPartit
           GenericTable.getInstance(tableName, tempDir, sparkSession, jsc, sourceTableFormat, true);
     }
     try (GenericTable tableToClose = table) {
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(targetTableFormats)
-              .tableBasePath(tableToClose.getBasePath())
-              .tableDataPath(tableToClose.getDataPath())
-              .hudiSourceConfig(
-                  HudiSourceConfigImpl.builder()
-                      .partitionFieldSpecConfig(oneTablePartitionConfig)
-                      .build())
-              .syncMode(SyncMode.INCREMENTAL)
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              sourceTableFormat,
+              SyncMode.INCREMENTAL,
+              tableName,
+              tableToClose,
+              targetTableFormats,
+              null,
+              null);
       tableToClose.insertRows(100);
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       // Do a second sync to force the test to read back the metadata it wrote earlier
       tableToClose.insertRows(100);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
 
       checkDatasetEquivalenceWithFilter(
           sourceTableFormat, tableToClose, targetTableFormats, filter);
@@ -515,32 +505,22 @@ public void testSyncWithSingleFormat(SyncMode syncMode) {
             tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
       table.insertRecords(100, true);
 
-      PerTableConfig perTableConfigIceberg =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(ImmutableList.of(ICEBERG))
-              .tableBasePath(table.getBasePath())
-              .syncMode(syncMode)
-              .build();
-
-      PerTableConfig perTableConfigDelta =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(ImmutableList.of(DELTA))
-              .tableBasePath(table.getBasePath())
-              .syncMode(syncMode)
-              .build();
+      TableSyncConfig tableSyncConfigIceberg =
+          getTableSyncConfig(
+              HUDI, syncMode, tableName, table, ImmutableList.of(ICEBERG), null, null);
+      TableSyncConfig tableSyncConfigDelta =
+          getTableSyncConfig(HUDI, syncMode, tableName, table, ImmutableList.of(DELTA), null, null);
 
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
-      oneTableClient.sync(perTableConfigIceberg, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfigIceberg, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 100);
-      oneTableClient.sync(perTableConfigDelta, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfigDelta, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, Collections.singletonList(DELTA), 100);
 
       table.insertRecords(100, true);
-      oneTableClient.sync(perTableConfigIceberg, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfigIceberg, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 200);
-      oneTableClient.sync(perTableConfigDelta, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfigDelta, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, Collections.singletonList(DELTA), 200);
     }
   }
@@ -552,21 +532,19 @@ public void testOutOfSyncIncrementalSyncs() {
     try (TestJavaHudiTable table =
         TestJavaHudiTable.forStandardSchema(
             tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
-      PerTableConfig singleTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(ImmutableList.of(ICEBERG))
-              .tableBasePath(table.getBasePath())
-              .syncMode(SyncMode.INCREMENTAL)
-              .build();
-
-      PerTableConfig dualTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(Arrays.asList(ICEBERG, DELTA))
-              .tableBasePath(table.getBasePath())
-              .syncMode(SyncMode.INCREMENTAL)
-              .build();
+
+      TableSyncConfig singleTableConfig =
+          getTableSyncConfig(
+              HUDI, SyncMode.INCREMENTAL, tableName, table, ImmutableList.of(ICEBERG), null, null);
+      TableSyncConfig dualTableConfig =
+          getTableSyncConfig(
+              HUDI,
+              SyncMode.INCREMENTAL,
+              tableName,
+              table,
+              Arrays.asList(ICEBERG, DELTA),
+              null,
+              null);
 
       table.insertRecords(50, true);
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
@@ -604,18 +582,21 @@ public void testIcebergCorruptedSnapshotRecovery() throws Exception {
             tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
       table.insertRows(20);
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(Collections.singletonList(ICEBERG))
-              .tableBasePath(table.getBasePath())
-              .syncMode(SyncMode.INCREMENTAL)
-              .build();
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              HUDI,
+              SyncMode.INCREMENTAL,
+              tableName,
+              table,
+              Collections.singletonList(ICEBERG),
+              null,
+              null);
+
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       table.insertRows(10);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       table.insertRows(10);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       // corrupt last two snapshots
       Table icebergTable = new HadoopTables(jsc.hadoopConfiguration()).load(table.getBasePath());
       long currentSnapshotId = icebergTable.currentSnapshot().snapshotId();
@@ -625,7 +606,7 @@ public void testIcebergCorruptedSnapshotRecovery() throws Exception {
       Files.delete(
           Paths.get(URI.create(icebergTable.snapshot(previousSnapshotId).manifestListLocation())));
       table.insertRows(10);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       checkDatasetEquivalence(HUDI, table, Collections.singletonList(ICEBERG), 50);
     }
   }
@@ -637,17 +618,18 @@ public void testMetadataRetention() throws Exception {
     try (TestJavaHudiTable table =
         TestJavaHudiTable.forStandardSchema(
             tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(Arrays.asList(ICEBERG, DELTA))
-              .tableBasePath(table.getBasePath())
-              .syncMode(SyncMode.INCREMENTAL)
-              .targetMetadataRetentionInHours(0) // force cleanup
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              HUDI,
+              SyncMode.INCREMENTAL,
+              tableName,
+              table,
+              Arrays.asList(ICEBERG, DELTA),
+              null,
+              0); // force cleanup
       OneTableClient oneTableClient = new OneTableClient(jsc.hadoopConfiguration());
       table.insertRecords(10, true);
-      oneTableClient.sync(perTableConfig, sourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, sourceClientProvider);
       // 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 +640,7 @@ public void testMetadataRetention() throws Exception {
           .forEach(
               unused -> {
                 table.insertRecords(10, true);
-                oneTableClient.sync(perTableConfig, sourceClientProvider);
+                oneTableClient.sync(tableSyncConfig, sourceClientProvider);
               });
       // ensure that hudi rows can still be read and underlying files were not removed
       List<Row> rows =
@@ -850,4 +832,41 @@ private static class TableFormatPartitionDataHolder {
     Optional<String> hudiSourceConfig;
     String filter;
   }
+
+  private static TableSyncConfig getTableSyncConfig(
+      String sourceTableFormat,
+      SyncMode syncMode,
+      String tableName,
+      GenericTable table,
+      List<String> targetTableFormats,
+      String oneTablePartitionConfig,
+      Integer metadataRetentionInHours) {
+    SourceTable sourceTable =
+        SourceTable.builder()
+            .name(tableName)
+            .formatName(sourceTableFormat)
+            .basePath(table.getBasePath())
+            .dataPath(table.getDataPath())
+            .build();
+
+    List<TargetTable> targetTables =
+        targetTableFormats.stream()
+            .map(
+                formatName ->
+                    TargetTable.builder()
+                        .name(tableName)
+                        .formatName(formatName)
+                        .basePath(table.getBasePath())
+                        .dataPath(table.getDataPath())
+                        .metadataRetentionInHours(metadataRetentionInHours)
+                        .build())
+            .collect(Collectors.toList());
+
+    return TableSyncConfig.builder()
+        .sourceTable(sourceTable)
+        .targetTables(targetTables)
+        .syncMode(syncMode)
+        .properties(Collections.singletonMap(PARTITION_FIELD_SPEC_CONFIG, oneTablePartitionConfig))
+        .build();
+  }
 }
diff --git a/core/src/test/java/io/onetable/client/TestOneTableClient.java b/core/src/test/java/io/onetable/client/TestOneTableClient.java
index fc349d3bd..2dc56f59a 100644
--- a/core/src/test/java/io/onetable/client/TestOneTableClient.java
+++ b/core/src/test/java/io/onetable/client/TestOneTableClient.java
@@ -18,7 +18,7 @@
  
 package io.onetable.client;
 
-import static io.onetable.GenericTable.getTableName;
+import static io.onetable.model.storage.TableFormat.HUDI;
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
@@ -79,14 +79,16 @@ void testAllSnapshotSyncAsPerConfig() {
     Map<String, SyncResult> perTableResults = new HashMap<>();
     perTableResults.put(TableFormat.ICEBERG, syncResult);
     perTableResults.put(TableFormat.DELTA, syncResult);
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
-    when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
+    TableSyncConfig tableSyncConfig =
+        getTableSyncConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
+    when(mockSourceClientProvider.getSourceClientInstance(
+            tableSyncConfig.getSourceTable(), tableSyncConfig.getProperties()))
         .thenReturn(mockSourceClient);
     when(mockTableFormatClientFactory.createForFormat(
-            TableFormat.ICEBERG, perTableConfig, mockConf))
+            tableSyncConfig.getTargetTables().get(0), mockConf))
         .thenReturn(mockTargetClient1);
-    when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+    when(mockTableFormatClientFactory.createForFormat(
+            tableSyncConfig.getTargetTables().get(1), mockConf))
         .thenReturn(mockTargetClient2);
     when(mockSourceClient.getCurrentSnapshot()).thenReturn(oneSnapshot);
     when(tableFormatSync.syncSnapshot(
@@ -95,21 +97,23 @@ void testAllSnapshotSyncAsPerConfig() {
         .thenReturn(perTableResults);
     OneTableClient oneTableClient =
         new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
-    Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+    Map<String, SyncResult> result = oneTableClient.sync(tableSyncConfig, mockSourceClientProvider);
     assertEquals(perTableResults, result);
   }
 
   @Test
   void testAllIncrementalSyncAsPerConfigAndNoFallbackNecessary() {
     SyncMode syncMode = SyncMode.INCREMENTAL;
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
-    when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
+    TableSyncConfig tableSyncConfig =
+        getTableSyncConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
+    when(mockSourceClientProvider.getSourceClientInstance(
+            tableSyncConfig.getSourceTable(), tableSyncConfig.getProperties()))
         .thenReturn(mockSourceClient);
     when(mockTableFormatClientFactory.createForFormat(
-            TableFormat.ICEBERG, perTableConfig, mockConf))
+            tableSyncConfig.getTargetTables().get(0), mockConf))
         .thenReturn(mockTargetClient1);
-    when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+    when(mockTableFormatClientFactory.createForFormat(
+            tableSyncConfig.getTargetTables().get(1), mockConf))
         .thenReturn(mockTargetClient2);
 
     Instant instantAsOfNow = Instant.now();
@@ -177,7 +181,7 @@ void testAllIncrementalSyncAsPerConfigAndNoFallbackNecessary() {
     expectedSyncResult.put(TableFormat.DELTA, getLastSyncResult(deltaSyncResults));
     OneTableClient oneTableClient =
         new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
-    Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+    Map<String, SyncResult> result = oneTableClient.sync(tableSyncConfig, mockSourceClientProvider);
     assertEquals(expectedSyncResult, result);
   }
 
@@ -191,14 +195,16 @@ void testIncrementalSyncFallBackToSnapshotForAllFormats() {
     Map<String, SyncResult> syncResults = new HashMap<>();
     syncResults.put(TableFormat.ICEBERG, syncResult);
     syncResults.put(TableFormat.DELTA, syncResult);
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
-    when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
+    TableSyncConfig tableSyncConfig =
+        getTableSyncConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
+    when(mockSourceClientProvider.getSourceClientInstance(
+            tableSyncConfig.getSourceTable(), tableSyncConfig.getProperties()))
         .thenReturn(mockSourceClient);
     when(mockTableFormatClientFactory.createForFormat(
-            TableFormat.ICEBERG, perTableConfig, mockConf))
+            tableSyncConfig.getTargetTables().get(0), mockConf))
         .thenReturn(mockTargetClient1);
-    when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+    when(mockTableFormatClientFactory.createForFormat(
+            tableSyncConfig.getTargetTables().get(1), mockConf))
         .thenReturn(mockTargetClient2);
 
     Instant instantAsOfNow = Instant.now();
@@ -218,21 +224,23 @@ void testIncrementalSyncFallBackToSnapshotForAllFormats() {
         .thenReturn(syncResults);
     OneTableClient oneTableClient =
         new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
-    Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+    Map<String, SyncResult> result = oneTableClient.sync(tableSyncConfig, mockSourceClientProvider);
     assertEquals(syncResults, result);
   }
 
   @Test
   void testIncrementalSyncFallbackToSnapshotForOnlySingleFormat() {
     SyncMode syncMode = SyncMode.INCREMENTAL;
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
-    when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
+    TableSyncConfig tableSyncConfig =
+        getTableSyncConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
+    when(mockSourceClientProvider.getSourceClientInstance(
+            tableSyncConfig.getSourceTable(), tableSyncConfig.getProperties()))
         .thenReturn(mockSourceClient);
     when(mockTableFormatClientFactory.createForFormat(
-            TableFormat.ICEBERG, perTableConfig, mockConf))
+            tableSyncConfig.getTargetTables().get(0), mockConf))
         .thenReturn(mockTargetClient1);
-    when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+    when(mockTableFormatClientFactory.createForFormat(
+            tableSyncConfig.getTargetTables().get(1), mockConf))
         .thenReturn(mockTargetClient2);
 
     Instant instantAsOfNow = Instant.now();
@@ -297,21 +305,23 @@ void testIncrementalSyncFallbackToSnapshotForOnlySingleFormat() {
     expectedSyncResult.put(TableFormat.DELTA, getLastSyncResult(deltaSyncResults));
     OneTableClient oneTableClient =
         new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
-    Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+    Map<String, SyncResult> result = oneTableClient.sync(tableSyncConfig, mockSourceClientProvider);
     assertEquals(expectedSyncResult, result);
   }
 
   @Test
   void incrementalSyncWithNoPendingInstantsForAllFormats() {
     SyncMode syncMode = SyncMode.INCREMENTAL;
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
-    when(mockSourceClientProvider.getSourceClientInstance(perTableConfig))
+    TableSyncConfig tableSyncConfig =
+        getTableSyncConfig(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA), syncMode);
+    when(mockSourceClientProvider.getSourceClientInstance(
+            tableSyncConfig.getSourceTable(), tableSyncConfig.getProperties()))
         .thenReturn(mockSourceClient);
     when(mockTableFormatClientFactory.createForFormat(
-            TableFormat.ICEBERG, perTableConfig, mockConf))
+            tableSyncConfig.getTargetTables().get(0), mockConf))
         .thenReturn(mockTargetClient1);
-    when(mockTableFormatClientFactory.createForFormat(TableFormat.DELTA, perTableConfig, mockConf))
+    when(mockTableFormatClientFactory.createForFormat(
+            tableSyncConfig.getTargetTables().get(1), mockConf))
         .thenReturn(mockTargetClient2);
 
     Instant instantAsOfNow = Instant.now();
@@ -351,7 +361,7 @@ void incrementalSyncWithNoPendingInstantsForAllFormats() {
     Map<String, SyncResult> expectedSyncResult = Collections.emptyMap();
     OneTableClient oneTableClient =
         new OneTableClient(mockConf, mockTableFormatClientFactory, tableFormatSync);
-    Map<String, SyncResult> result = oneTableClient.sync(perTableConfig, mockSourceClientProvider);
+    Map<String, SyncResult> result = oneTableClient.sync(tableSyncConfig, mockSourceClientProvider);
     assertEquals(expectedSyncResult, result);
   }
 
@@ -393,12 +403,30 @@ private Instant getInstantAtLastNMinutes(Instant currentInstant, int n) {
     return Instant.now().minus(Duration.ofMinutes(n));
   }
 
-  private PerTableConfig getPerTableConfig(List<String> targetTableFormats, SyncMode syncMode) {
-    return PerTableConfigImpl.builder()
-        .tableName(getTableName())
-        .tableBasePath("/tmp/doesnt/matter")
-        .targetTableFormats(targetTableFormats)
+  private TableSyncConfig getTableSyncConfig(List<String> targetTableFormats, SyncMode syncMode) {
+    SourceTable sourceTable =
+        SourceTable.builder()
+            .name("tablename")
+            .formatName(HUDI)
+            .basePath("/tmp/doesnt/matter")
+            .build();
+
+    List<TargetTable> targetTables =
+        targetTableFormats.stream()
+            .map(
+                formatName ->
+                    TargetTable.builder()
+                        .name("tablename")
+                        .formatName(formatName)
+                        .basePath("/tmp/doesnt/matter")
+                        .build())
+            .collect(Collectors.toList());
+
+    return TableSyncConfig.builder()
+        .sourceTable(sourceTable)
+        .targetTables(targetTables)
         .syncMode(syncMode)
+        .properties(Collections.singletonMap("some.property", "value"))
         .build();
   }
 
diff --git a/core/src/test/java/io/onetable/client/TestPerTableConfig.java b/core/src/test/java/io/onetable/client/TestPerTableConfig.java
deleted file mode 100644
index 2bcd8d372..000000000
--- a/core/src/test/java/io/onetable/client/TestPerTableConfig.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
- 
-package io.onetable.client;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-import java.util.Collections;
-
-import org.junit.jupiter.api.Test;
-
-import io.onetable.hudi.HudiSourceConfigImpl;
-import io.onetable.model.storage.TableFormat;
-import io.onetable.model.sync.SyncMode;
-
-class TestPerTableConfig {
-
-  @Test
-  void sanitizePath() {
-    PerTableConfig tooManySlashes =
-        PerTableConfigImpl.builder()
-            .tableBasePath("s3://bucket//path")
-            .tableName("name")
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
-            .build();
-    assertEquals("s3://bucket/path", tooManySlashes.getTableBasePath());
-
-    PerTableConfig localFilePath =
-        PerTableConfigImpl.builder()
-            .tableBasePath("/local/data//path")
-            .tableName("name")
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
-            .build();
-    assertEquals("file:///local/data/path", localFilePath.getTableBasePath());
-
-    PerTableConfig properLocalFilePath =
-        PerTableConfigImpl.builder()
-            .tableBasePath("file:///local/data//path")
-            .tableName("name")
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
-            .build();
-    assertEquals("file:///local/data/path", properLocalFilePath.getTableBasePath());
-  }
-
-  @Test
-  void defaultValueSet() {
-    PerTableConfig perTableConfig =
-        PerTableConfigImpl.builder()
-            .tableBasePath("file://bucket/path")
-            .tableName("name")
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
-            .build();
-
-    assertEquals(24 * 7, perTableConfig.getTargetMetadataRetentionInHours());
-    assertEquals(SyncMode.INCREMENTAL, perTableConfig.getSyncMode());
-    assertEquals(HudiSourceConfigImpl.builder().build(), perTableConfig.getHudiSourceConfig());
-    assertNull(perTableConfig.getNamespace());
-    assertNull(perTableConfig.getIcebergCatalogConfig());
-  }
-
-  @Test
-  void errorIfRequiredArgsNotSet() {
-    assertThrows(
-        NullPointerException.class,
-        () ->
-            PerTableConfigImpl.builder()
-                .tableName("name")
-                .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
-                .build());
-
-    assertThrows(
-        NullPointerException.class,
-        () ->
-            PerTableConfigImpl.builder()
-                .tableBasePath("file://bucket/path")
-                .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
-                .build());
-
-    assertThrows(
-        NullPointerException.class,
-        () ->
-            PerTableConfigImpl.builder()
-                .tableBasePath("file://bucket/path")
-                .tableName("name")
-                .build());
-  }
-
-  @Test
-  void errorIfNoTargetsSet() {
-    Exception thrownException =
-        assertThrows(
-            IllegalArgumentException.class,
-            () ->
-                PerTableConfigImpl.builder()
-                    .tableName("name")
-                    .tableBasePath("file://bucket/path")
-                    .targetTableFormats(Collections.emptyList())
-                    .build());
-    assertEquals("Please provide at-least one format to sync", thrownException.getMessage());
-  }
-}
diff --git a/core/src/test/java/io/onetable/client/TestTableFormatClientFactory.java b/core/src/test/java/io/onetable/client/TestTableFormatClientFactory.java
index 6897f50b8..7ba17c353 100644
--- a/core/src/test/java/io/onetable/client/TestTableFormatClientFactory.java
+++ b/core/src/test/java/io/onetable/client/TestTableFormatClientFactory.java
@@ -24,15 +24,11 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-import java.util.Arrays;
-import java.util.List;
-
 import org.apache.hadoop.conf.Configuration;
 import org.junit.jupiter.api.Test;
 
 import io.onetable.exception.NotSupportedException;
 import io.onetable.model.storage.TableFormat;
-import io.onetable.model.sync.SyncMode;
 import io.onetable.spi.sync.TargetClient;
 
 public class TestTableFormatClientFactory {
@@ -42,11 +38,10 @@ public void testTableClientFromNameForDELTA() {
     TargetClient tc =
         TableFormatClientFactory.getInstance().createTargetClientForName(TableFormat.DELTA);
     assertNotNull(tc);
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.DELTA), SyncMode.INCREMENTAL);
+    TargetTable targetTable = getPerTableConfig(TableFormat.DELTA);
     Configuration conf = new Configuration();
     conf.setStrings("spark.master", "local");
-    tc.init(perTableConfig, conf);
+    tc.init(targetTable, conf);
     assertEquals(tc.getTableFormat(), TableFormat.DELTA);
   }
 
@@ -55,11 +50,10 @@ public void testTableClientFromNameForHUDI() {
     TargetClient tc =
         TableFormatClientFactory.getInstance().createTargetClientForName(TableFormat.HUDI);
     assertNotNull(tc);
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.HUDI), SyncMode.INCREMENTAL);
+    TargetTable targetTable = getPerTableConfig(TableFormat.HUDI);
     Configuration conf = new Configuration();
     conf.setStrings("spark.master", "local");
-    tc.init(perTableConfig, conf);
+    tc.init(targetTable, conf);
     assertEquals(tc.getTableFormat(), TableFormat.HUDI);
   }
 
@@ -68,8 +62,7 @@ public void testTableClientFromNameForICEBERG() {
     TargetClient tc =
         TableFormatClientFactory.getInstance().createTargetClientForName(TableFormat.ICEBERG);
     assertNotNull(tc);
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.ICEBERG), SyncMode.INCREMENTAL);
+    TargetTable perTableConfig = getPerTableConfig(TableFormat.ICEBERG);
     Configuration conf = new Configuration();
     conf.setStrings("spark.master", "local");
     tc.init(perTableConfig, conf);
@@ -88,22 +81,18 @@ public void testTableClientFromNameForUNKOWN() {
 
   @Test
   public void testTableClientFromFormatType() {
-    PerTableConfig perTableConfig =
-        getPerTableConfig(Arrays.asList(TableFormat.DELTA), SyncMode.INCREMENTAL);
+    TargetTable perTableConfig = getPerTableConfig(TableFormat.DELTA);
     Configuration conf = new Configuration();
     conf.setStrings("spark.master", "local");
-    TargetClient tc =
-        TableFormatClientFactory.getInstance()
-            .createForFormat(TableFormat.DELTA, perTableConfig, conf);
+    TargetClient tc = TableFormatClientFactory.getInstance().createForFormat(perTableConfig, conf);
     assertEquals(tc.getTableFormat(), TableFormat.DELTA);
   }
 
-  private PerTableConfig getPerTableConfig(List<String> targetTableFormats, SyncMode syncMode) {
-    return PerTableConfigImpl.builder()
-        .tableName(getTableName())
-        .tableBasePath("/tmp/doesnt/matter")
-        .targetTableFormats(targetTableFormats)
-        .syncMode(syncMode)
+  private TargetTable getPerTableConfig(String tableFormat) {
+    return TargetTable.builder()
+        .name(getTableName())
+        .basePath("/tmp/doesnt/matter")
+        .formatName(tableFormat)
         .build();
   }
 }
diff --git a/core/src/test/java/io/onetable/delta/ITDeltaSourceClient.java b/core/src/test/java/io/onetable/delta/ITDeltaSourceClient.java
index 5cd13a2b6..7007be286 100644
--- a/core/src/test/java/io/onetable/delta/ITDeltaSourceClient.java
+++ b/core/src/test/java/io/onetable/delta/ITDeltaSourceClient.java
@@ -54,8 +54,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import io.onetable.TestSparkDeltaTable;
-import io.onetable.client.PerTableConfig;
-import io.onetable.client.PerTableConfigImpl;
+import io.onetable.client.SourceTable;
 import io.onetable.model.*;
 import io.onetable.model.CommitsBacklog;
 import io.onetable.model.schema.OneField;
@@ -144,7 +143,7 @@ void setUp() {
     hadoopConf.set("fs.defaultFS", "file:///");
 
     clientProvider = new DeltaSourceClientProvider();
-    clientProvider.init(hadoopConf, null);
+    clientProvider.init(hadoopConf);
   }
 
   @Test
@@ -160,13 +159,14 @@ void getCurrentSnapshotNonPartitionedTest() throws URISyntaxException {
             + basePath
             + "' AS SELECT * FROM VALUES (1, 2)");
     // Create Delta source client
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(tableName)
-            .tableBasePath(basePath.toString())
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(tableName)
+            .basePath(basePath.toString())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient client = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient client =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     // Get current snapshot
     OneSnapshot snapshot = client.getCurrentSnapshot();
     // Validate table
@@ -218,13 +218,14 @@ void getCurrentSnapshotPartitionedTest() throws URISyntaxException {
             + basePath
             + "' AS SELECT 'SingleValue' AS part_col, 1 AS col1, 2 AS col2");
     // Create Delta source client
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(tableName)
-            .tableBasePath(basePath.toString())
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(tableName)
+            .basePath(basePath.toString())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient client = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient client =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     // Get current snapshot
     OneSnapshot snapshot = client.getCurrentSnapshot();
     // Validate table
@@ -306,13 +307,14 @@ void getCurrentSnapshotGenColPartitionedTest() {
             + tableName
             + "` VALUES(1, CAST('2012-02-12 00:12:34' AS TIMESTAMP))");
     // Create Delta source client
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(tableName)
-            .tableBasePath(basePath.toString())
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(tableName)
+            .basePath(basePath.toString())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient client = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient client =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     // Get current snapshot
     OneSnapshot snapshot = client.getCurrentSnapshot();
     // TODO: Complete and enable test (see https://github.com/onetable-io/onetable/issues/90)
@@ -345,13 +347,14 @@ public void testInsertsUpsertsAndDeletes(boolean isPartitioned) {
 
     testSparkDeltaTable.insertRows(50);
     allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(testSparkDeltaTable.getTableName())
-            .tableBasePath(testSparkDeltaTable.getBasePath())
-            .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(testSparkDeltaTable.getTableName())
+            .basePath(testSparkDeltaTable.getBasePath())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient deltaSourceClient =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     assertEquals(180L, testSparkDeltaTable.getNumRows());
     OneSnapshot oneSnapshot = deltaSourceClient.getCurrentSnapshot();
 
@@ -383,13 +386,14 @@ public void testsShowingVacuumHasNoEffectOnIncrementalSync() {
     // Insert 50 rows to 2018 partition.
     List<Row> commit1Rows = testSparkDeltaTable.insertRowsForPartition(50, 2018);
     Long timestamp1 = testSparkDeltaTable.getLastCommitTimestamp();
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(testSparkDeltaTable.getTableName())
-            .tableBasePath(testSparkDeltaTable.getBasePath())
-            .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(testSparkDeltaTable.getTableName())
+            .basePath(testSparkDeltaTable.getBasePath())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient deltaSourceClient =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     OneSnapshot snapshotAfterCommit1 = deltaSourceClient.getCurrentSnapshot();
     List<String> allActivePaths = getAllFilePaths(snapshotAfterCommit1);
     assertEquals(1, allActivePaths.size());
@@ -408,7 +412,7 @@ public void testsShowingVacuumHasNoEffectOnIncrementalSync() {
         InstantsForIncrementalSync.builder()
             .lastSyncInstant(Instant.ofEpochMilli(timestamp1))
             .build();
-    deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+    deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     CommitsBacklog<Long> instantCurrentCommitState =
         deltaSourceClient.getCommitsBacklog(instantsForIncrementalSync);
     boolean areFilesRemoved = false;
@@ -450,13 +454,14 @@ public void testVacuum(boolean isPartitioned) {
     testSparkDeltaTable.insertRows(50);
     allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
 
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(testSparkDeltaTable.getTableName())
-            .tableBasePath(testSparkDeltaTable.getBasePath())
-            .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(testSparkDeltaTable.getTableName())
+            .basePath(testSparkDeltaTable.getBasePath())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient deltaSourceClient =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     assertEquals(130L, testSparkDeltaTable.getNumRows());
     OneSnapshot oneSnapshot = deltaSourceClient.getCurrentSnapshot();
     if (isPartitioned) {
@@ -496,13 +501,14 @@ public void testAddColumns(boolean isPartitioned) {
     testSparkDeltaTable.insertRows(50);
     allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
 
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(testSparkDeltaTable.getTableName())
-            .tableBasePath(testSparkDeltaTable.getBasePath())
-            .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(testSparkDeltaTable.getTableName())
+            .basePath(testSparkDeltaTable.getBasePath())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient deltaSourceClient =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     assertEquals(150L, testSparkDeltaTable.getNumRows());
     OneSnapshot oneSnapshot = deltaSourceClient.getCurrentSnapshot();
     if (isPartitioned) {
@@ -551,13 +557,14 @@ public void testDropPartition() {
     testSparkDeltaTable.insertRowsForPartition(20, partitionValueToDelete);
     allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
 
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(testSparkDeltaTable.getTableName())
-            .tableBasePath(testSparkDeltaTable.getBasePath())
-            .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(testSparkDeltaTable.getTableName())
+            .basePath(testSparkDeltaTable.getBasePath())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient deltaSourceClient =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     assertEquals(
         120 - rowsByPartition.get(partitionValueToDelete).size(), testSparkDeltaTable.getNumRows());
     OneSnapshot oneSnapshot = deltaSourceClient.getCurrentSnapshot();
@@ -609,13 +616,14 @@ public void testOptimizeAndClustering(boolean isPartitioned) {
     testSparkDeltaTable.insertRows(50);
     allActiveFiles.add(testSparkDeltaTable.getAllActiveFiles());
 
-    PerTableConfig tableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(testSparkDeltaTable.getTableName())
-            .tableBasePath(testSparkDeltaTable.getBasePath())
-            .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG))
+    SourceTable tableConfig =
+        SourceTable.builder()
+            .name(testSparkDeltaTable.getTableName())
+            .basePath(testSparkDeltaTable.getBasePath())
+            .formatName(TableFormat.DELTA)
             .build();
-    DeltaSourceClient deltaSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+    DeltaSourceClient deltaSourceClient =
+        clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
     assertEquals(250L, testSparkDeltaTable.getNumRows());
     OneSnapshot oneSnapshot = deltaSourceClient.getCurrentSnapshot();
     if (isPartitioned) {
diff --git a/core/src/test/java/io/onetable/delta/TestDeltaSync.java b/core/src/test/java/io/onetable/delta/TestDeltaSync.java
index 5b275f1ad..2063574e3 100644
--- a/core/src/test/java/io/onetable/delta/TestDeltaSync.java
+++ b/core/src/test/java/io/onetable/delta/TestDeltaSync.java
@@ -76,7 +76,7 @@
 import io.delta.standalone.types.IntegerType;
 import io.delta.standalone.types.StringType;
 
-import io.onetable.client.PerTableConfigImpl;
+import io.onetable.client.TargetTable;
 import io.onetable.model.OneSnapshot;
 import io.onetable.model.OneTable;
 import io.onetable.model.schema.OneField;
@@ -126,11 +126,11 @@ public void setup() throws IOException {
     Files.createDirectories(basePath);
     deltaClient =
         new DeltaClient(
-            PerTableConfigImpl.builder()
-                .tableName(tableName)
-                .tableBasePath(basePath.toString())
-                .targetMetadataRetentionInHours(1)
-                .targetTableFormats(Collections.singletonList(TableFormat.DELTA))
+            TargetTable.builder()
+                .name(tableName)
+                .basePath(basePath.toString())
+                .metadataRetentionInHours(1)
+                .formatName(TableFormat.DELTA)
                 .build(),
             sparkSession);
   }
diff --git a/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java b/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
index 0fa4a7259..725618376 100644
--- a/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
+++ b/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
@@ -534,9 +534,7 @@ private HudiClient getHudiSourceClient(
             .build();
     HudiSourcePartitionSpecExtractor partitionSpecExtractor =
         new ConfigurationBasedPartitionSpecExtractor(
-            HudiSourceConfigImpl.builder()
-                .partitionFieldSpecConfig(onetablePartitionConfig)
-                .build());
+            HudiSourceConfigImpl.fromPartitionFieldSpecConfig(onetablePartitionConfig));
     return new HudiClient(hoodieTableMetaClient, partitionSpecExtractor);
   }
 
diff --git a/core/src/test/java/io/onetable/hudi/ITHudiTargetClient.java b/core/src/test/java/io/onetable/hudi/ITHudiTargetClient.java
index e3706768c..e549e6de9 100644
--- a/core/src/test/java/io/onetable/hudi/ITHudiTargetClient.java
+++ b/core/src/test/java/io/onetable/hudi/ITHudiTargetClient.java
@@ -70,7 +70,7 @@
 import org.apache.hudi.metadata.HoodieBackedTableMetadata;
 import org.apache.hudi.metadata.HoodieMetadataFileSystemView;
 
-import io.onetable.client.PerTableConfigImpl;
+import io.onetable.client.TargetTable;
 import io.onetable.model.OneTable;
 import io.onetable.model.OneTableMetadata;
 import io.onetable.model.schema.OneField;
@@ -581,11 +581,11 @@ private OneTable getState(Instant latestCommitTime) {
 
   private HudiTargetClient getTargetClient() {
     return new HudiTargetClient(
-        PerTableConfigImpl.builder()
-            .tableBasePath(tableBasePath)
-            .targetTableFormats(Collections.singletonList(TableFormat.HUDI))
-            .tableName("test_table")
-            .targetMetadataRetentionInHours(4)
+        TargetTable.builder()
+            .basePath(tableBasePath)
+            .formatName(TableFormat.HUDI)
+            .name("test_table")
+            .metadataRetentionInHours(4)
             .build(),
         CONFIGURATION,
         3);
diff --git a/core/src/test/java/io/onetable/iceberg/ITIcebergSourceClient.java b/core/src/test/java/io/onetable/iceberg/ITIcebergSourceClient.java
index 2e6c83ff1..d98f22113 100644
--- a/core/src/test/java/io/onetable/iceberg/ITIcebergSourceClient.java
+++ b/core/src/test/java/io/onetable/iceberg/ITIcebergSourceClient.java
@@ -27,7 +27,7 @@
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -44,8 +44,7 @@
 import org.apache.iceberg.data.Record;
 
 import io.onetable.TestIcebergTable;
-import io.onetable.client.PerTableConfig;
-import io.onetable.client.PerTableConfigImpl;
+import io.onetable.client.SourceTable;
 import io.onetable.model.CommitsBacklog;
 import io.onetable.model.InstantsForIncrementalSync;
 import io.onetable.model.OneSnapshot;
@@ -63,7 +62,7 @@ public class ITIcebergSourceClient {
   @BeforeEach
   void setup() {
     clientProvider = new IcebergSourceClientProvider();
-    clientProvider.init(hadoopConf, null);
+    clientProvider.init(hadoopConf);
   }
 
   @ParameterizedTest
@@ -95,13 +94,14 @@ public void testInsertsUpsertsAndDeletes(boolean isPartitioned) {
       testIcebergTable.insertRows(50);
       allActiveFiles.add(testIcebergTable.getAllActiveFiles());
 
-      PerTableConfig tableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(testIcebergTable.getTableName())
-              .tableBasePath(testIcebergTable.getBasePath())
-              .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
+      SourceTable tableConfig =
+          SourceTable.builder()
+              .name(testIcebergTable.getTableName())
+              .basePath(testIcebergTable.getBasePath())
+              .formatName(TableFormat.ICEBERG)
               .build();
-      IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+      IcebergSourceClient icebergSourceClient =
+          clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
       assertEquals(180L, testIcebergTable.getNumRows());
       OneSnapshot oneSnapshot = icebergSourceClient.getCurrentSnapshot();
 
@@ -154,13 +154,14 @@ public void testDropPartition() {
       testIcebergTable.insertRecordsForPartition(20, partitionValueToDelete);
       allActiveFiles.add(testIcebergTable.getAllActiveFiles());
 
-      PerTableConfig tableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(testIcebergTable.getTableName())
-              .tableBasePath(testIcebergTable.getBasePath())
-              .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
+      SourceTable tableConfig =
+          SourceTable.builder()
+              .name(testIcebergTable.getTableName())
+              .basePath(testIcebergTable.getBasePath())
+              .formatName(TableFormat.ICEBERG)
               .build();
-      IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+      IcebergSourceClient icebergSourceClient =
+          clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
       assertEquals(
           120 - recordsByPartition.get(partitionValueToDelete).size(),
           testIcebergTable.getNumRows());
@@ -213,13 +214,14 @@ public void testDeleteAllRecordsInPartition() {
       testIcebergTable.insertRecordsForPartition(20, partitionValueToDelete);
       allActiveFiles.add(testIcebergTable.getAllActiveFiles());
 
-      PerTableConfig tableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(testIcebergTable.getTableName())
-              .tableBasePath(testIcebergTable.getBasePath())
-              .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
+      SourceTable tableConfig =
+          SourceTable.builder()
+              .name(testIcebergTable.getTableName())
+              .basePath(testIcebergTable.getBasePath())
+              .formatName(TableFormat.ICEBERG)
               .build();
-      IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+      IcebergSourceClient icebergSourceClient =
+          clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
       assertEquals(
           120 - recordsByPartition.get(partitionValueToDelete).size(),
           testIcebergTable.getNumRows());
@@ -272,13 +274,14 @@ public void testExpireSnapshots(boolean isPartitioned) throws InterruptedExcepti
       testIcebergTable.insertRows(50);
       allActiveFiles.add(testIcebergTable.getAllActiveFiles());
 
-      PerTableConfig tableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(testIcebergTable.getTableName())
-              .tableBasePath(testIcebergTable.getBasePath())
-              .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
+      SourceTable tableConfig =
+          SourceTable.builder()
+              .name(testIcebergTable.getTableName())
+              .basePath(testIcebergTable.getBasePath())
+              .formatName(TableFormat.ICEBERG)
               .build();
-      IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+      IcebergSourceClient icebergSourceClient =
+          clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
       assertEquals(200L, testIcebergTable.getNumRows());
       OneSnapshot oneSnapshot = icebergSourceClient.getCurrentSnapshot();
 
@@ -312,12 +315,12 @@ public void testForIncrementalSyncSafetyCheck(boolean shouldExpireSnapshots) {
       // Insert 50 rows to INFO partition.
       List<Record> commit1Rows = testIcebergTable.insertRecordsForPartition(50, "INFO");
       Long timestamp1 = testIcebergTable.getLastCommitTimestamp();
-      PerTableConfig tableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(testIcebergTable.getTableName())
-              .tableBasePath(testIcebergTable.getBasePath())
-              .tableDataPath(testIcebergTable.getDataPath())
-              .targetTableFormats(Arrays.asList(TableFormat.HUDI, TableFormat.DELTA))
+      SourceTable tableConfig =
+          SourceTable.builder()
+              .name(testIcebergTable.getTableName())
+              .basePath(testIcebergTable.getBasePath())
+              .dataPath(testIcebergTable.getDataPath())
+              .formatName(TableFormat.ICEBERG)
               .build();
 
       // Upsert all rows inserted before, so all files are replaced.
@@ -331,7 +334,8 @@ public void testForIncrementalSyncSafetyCheck(boolean shouldExpireSnapshots) {
         // Expire snapshotAfterCommit2.
         testIcebergTable.expireSnapshot(snapshotIdAfterCommit2);
       }
-      IcebergSourceClient icebergSourceClient = clientProvider.getSourceClientInstance(tableConfig);
+      IcebergSourceClient icebergSourceClient =
+          clientProvider.getSourceClientInstance(tableConfig, Collections.emptyMap());
       if (shouldExpireSnapshots) {
         assertFalse(
             icebergSourceClient.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(timestamp1)));
diff --git a/core/src/test/java/io/onetable/iceberg/TestIcebergSourceClient.java b/core/src/test/java/io/onetable/iceberg/TestIcebergSourceClient.java
index 2b7d0a98e..756744a42 100644
--- a/core/src/test/java/io/onetable/iceberg/TestIcebergSourceClient.java
+++ b/core/src/test/java/io/onetable/iceberg/TestIcebergSourceClient.java
@@ -45,8 +45,7 @@
 import org.apache.iceberg.parquet.Parquet;
 import org.apache.iceberg.types.Types;
 
-import io.onetable.client.PerTableConfig;
-import io.onetable.client.PerTableConfigImpl;
+import io.onetable.client.SourceTable;
 import io.onetable.model.*;
 import io.onetable.model.schema.*;
 import io.onetable.model.stat.PartitionValue;
@@ -70,7 +69,7 @@ void setUp() throws IOException {
     hadoopConf.set("fs.defaultFS", "file:///");
 
     clientProvider = new IcebergSourceClientProvider();
-    clientProvider.init(hadoopConf, null);
+    clientProvider.init(hadoopConf);
 
     tableManager = IcebergTableManager.of(hadoopConf);
 
@@ -84,9 +83,10 @@ void setUp() throws IOException {
   @Test
   void getTableTest(@TempDir Path workingDir) throws IOException {
     Table catalogSales = createTestTableWithData(workingDir.toString());
-    PerTableConfig sourceTableConfig = getPerTableConfig(catalogSales);
+    SourceTable sourceTableConfig = getPerTableConfig(catalogSales);
 
-    IcebergSourceClient client = clientProvider.getSourceClientInstance(sourceTableConfig);
+    IcebergSourceClient client =
+        clientProvider.getSourceClientInstance(sourceTableConfig, Collections.emptyMap());
 
     Snapshot snapshot = catalogSales.currentSnapshot();
     OneTable oneTable = client.getTable(snapshot);
@@ -119,9 +119,10 @@ public void getSchemaCatalogTest(@TempDir Path workingDir) throws IOException {
     assertEquals(2, catalogSales.schemas().size());
     assertEquals(0, iceCurrentSnapshot.schemaId());
 
-    PerTableConfig sourceTableConfig = getPerTableConfig(catalogSales);
+    SourceTable sourceTableConfig = getPerTableConfig(catalogSales);
 
-    IcebergSourceClient client = clientProvider.getSourceClientInstance(sourceTableConfig);
+    IcebergSourceClient client =
+        clientProvider.getSourceClientInstance(sourceTableConfig, Collections.emptyMap());
     IcebergSourceClient spyClient = spy(client);
 
     SchemaCatalog schemaCatalog = spyClient.getSchemaCatalog(null, iceCurrentSnapshot);
@@ -139,7 +140,7 @@ public void testGetCurrentSnapshot(@TempDir Path workingDir) throws IOException
     Table catalogSales = createTestTableWithData(workingDir.toString());
     Snapshot iceCurrentSnapshot = catalogSales.currentSnapshot();
 
-    PerTableConfig sourceTableConfig = getPerTableConfig(catalogSales);
+    SourceTable sourceTableConfig = getPerTableConfig(catalogSales);
 
     IcebergDataFileExtractor spyDataFileExtractor = spy(IcebergDataFileExtractor.builder().build());
     IcebergPartitionValueConverter spyPartitionConverter =
@@ -399,7 +400,7 @@ private Table createTestCatalogTable(String workingDir) {
   }
 
   private IcebergSourceClient getIcebergSourceClient(Table catalogSales) {
-    PerTableConfig tableConfig = getPerTableConfig(catalogSales);
+    SourceTable tableConfig = getPerTableConfig(catalogSales);
 
     return IcebergSourceClient.builder()
         .hadoopConf(hadoopConf)
@@ -407,11 +408,11 @@ private IcebergSourceClient getIcebergSourceClient(Table catalogSales) {
         .build();
   }
 
-  private static PerTableConfig getPerTableConfig(Table catalogSales) {
-    return PerTableConfigImpl.builder()
-        .tableName(catalogSales.name())
-        .tableBasePath(catalogSales.location())
-        .targetTableFormats(Collections.singletonList(TableFormat.DELTA))
+  private static SourceTable getPerTableConfig(Table catalogSales) {
+    return SourceTable.builder()
+        .name(catalogSales.name())
+        .basePath(catalogSales.location())
+        .formatName(TableFormat.ICEBERG)
         .build();
   }
 
diff --git a/core/src/test/java/io/onetable/iceberg/TestIcebergSync.java b/core/src/test/java/io/onetable/iceberg/TestIcebergSync.java
index f1467e6d3..91f3f9ddf 100644
--- a/core/src/test/java/io/onetable/iceberg/TestIcebergSync.java
+++ b/core/src/test/java/io/onetable/iceberg/TestIcebergSync.java
@@ -18,6 +18,7 @@
  
 package io.onetable.iceberg;
 
+import static io.onetable.model.storage.TableFormat.ICEBERG;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -82,7 +83,7 @@
 import com.google.common.collect.Sets;
 
 import io.onetable.ITOneTableClient;
-import io.onetable.client.PerTableConfigImpl;
+import io.onetable.client.TargetTable;
 import io.onetable.model.OneSnapshot;
 import io.onetable.model.OneTable;
 import io.onetable.model.OneTableMetadata;
@@ -187,11 +188,11 @@ public void setup() throws IOException {
 
   private IcebergClient getIcebergClient() {
     return new IcebergClient(
-        PerTableConfigImpl.builder()
-            .tableBasePath(basePath.toString())
-            .tableName(tableName)
-            .targetMetadataRetentionInHours(1)
-            .targetTableFormats(Collections.singletonList(TableFormat.ICEBERG))
+        TargetTable.builder()
+            .basePath(basePath.toString())
+            .name(tableName)
+            .formatName(ICEBERG)
+            .metadataRetentionInHours(1)
             .build(),
         CONFIGURATION,
         mockSchemaExtractor,
diff --git a/core/src/test/java/io/onetable/loadtest/LoadTest.java b/core/src/test/java/io/onetable/loadtest/LoadTest.java
index 4a392ce44..c7afc1fc6 100644
--- a/core/src/test/java/io/onetable/loadtest/LoadTest.java
+++ b/core/src/test/java/io/onetable/loadtest/LoadTest.java
@@ -18,9 +18,12 @@
  
 package io.onetable.loadtest;
 
+import static io.onetable.hudi.HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG;
+
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -34,11 +37,13 @@
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.config.HoodieArchivalConfig;
 
+import io.onetable.GenericTable;
 import io.onetable.TestJavaHudiTable;
 import io.onetable.client.OneTableClient;
-import io.onetable.client.PerTableConfig;
-import io.onetable.client.PerTableConfigImpl;
 import io.onetable.client.SourceClientProvider;
+import io.onetable.client.SourceTable;
+import io.onetable.client.TableSyncConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.hudi.HudiSourceClientProvider;
 import io.onetable.model.storage.TableFormat;
 import io.onetable.model.sync.SyncMode;
@@ -56,7 +61,7 @@ public class LoadTest {
   @BeforeEach
   public void setup() {
     hudiSourceClientProvider = new HudiSourceClientProvider();
-    hudiSourceClientProvider.init(CONFIGURATION, Collections.emptyMap());
+    hudiSourceClientProvider.init(CONFIGURATION);
   }
 
   @Test
@@ -75,16 +80,15 @@ void testFullSyncWithManyPartitions() {
                 .collect(Collectors.toList()),
             false);
       }
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA))
-              .tableBasePath(table.getBasePath())
-              .syncMode(SyncMode.FULL)
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              SyncMode.FULL,
+              tableName,
+              table,
+              Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA));
       OneTableClient oneTableClient = new OneTableClient(CONFIGURATION);
       long start = System.currentTimeMillis();
-      oneTableClient.sync(perTableConfig, hudiSourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, hudiSourceClientProvider);
       long end = System.currentTimeMillis();
       System.out.println("Full sync took " + (end - start) + "ms");
     }
@@ -104,16 +108,15 @@ void testIncrementalSyncWithManyCommits() {
         TestJavaHudiTable.forStandardSchema(
             tableName, tempDir, "level:SIMPLE", HoodieTableType.COPY_ON_WRITE, archivalConfig)) {
       table.insertRecords(1, "partition0", false);
-      PerTableConfig perTableConfig =
-          PerTableConfigImpl.builder()
-              .tableName(tableName)
-              .targetTableFormats(Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA))
-              .tableBasePath(table.getBasePath())
-              .syncMode(SyncMode.INCREMENTAL)
-              .build();
+      TableSyncConfig tableSyncConfig =
+          getTableSyncConfig(
+              SyncMode.INCREMENTAL,
+              tableName,
+              table,
+              Arrays.asList(TableFormat.ICEBERG, TableFormat.DELTA));
       // sync once to establish first commit
       OneTableClient oneTableClient = new OneTableClient(CONFIGURATION);
-      oneTableClient.sync(perTableConfig, hudiSourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, hudiSourceClientProvider);
       for (int i = 0; i < numCommits; i++) {
         table.insertRecords(
             1,
@@ -124,9 +127,38 @@ void testIncrementalSyncWithManyCommits() {
       }
 
       long start = System.currentTimeMillis();
-      oneTableClient.sync(perTableConfig, hudiSourceClientProvider);
+      oneTableClient.sync(tableSyncConfig, hudiSourceClientProvider);
       long end = System.currentTimeMillis();
       System.out.println("Incremental sync took " + (end - start) + "ms");
     }
   }
+
+  private static TableSyncConfig getTableSyncConfig(
+      SyncMode syncMode, String tableName, GenericTable table, List<String> targetTableFormats) {
+    SourceTable sourceTable =
+        SourceTable.builder()
+            .name(tableName)
+            .formatName(TableFormat.HUDI)
+            .basePath(table.getBasePath())
+            .dataPath(table.getDataPath())
+            .build();
+
+    List<TargetTable> targetTables =
+        targetTableFormats.stream()
+            .map(
+                formatName ->
+                    TargetTable.builder()
+                        .name(tableName)
+                        .formatName(formatName)
+                        .basePath(table.getBasePath())
+                        .build())
+            .collect(Collectors.toList());
+
+    return TableSyncConfig.builder()
+        .sourceTable(sourceTable)
+        .targetTables(targetTables)
+        .syncMode(syncMode)
+        .properties(Collections.singletonMap(PARTITION_FIELD_SPEC_CONFIG, "level:VALUE"))
+        .build();
+  }
 }
diff --git a/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java b/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
index 71902a5f9..89c8c7e12 100644
--- a/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
+++ b/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
@@ -35,11 +35,13 @@
 import org.apache.hudi.sync.common.HoodieSyncTool;
 
 import io.onetable.client.OneTableClient;
-import io.onetable.client.PerTableConfig;
-import io.onetable.client.PerTableConfigImpl;
+import io.onetable.client.SourceTable;
+import io.onetable.client.TableSyncConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.hudi.HudiSourceClientProvider;
 import io.onetable.hudi.HudiSourceConfigImpl;
 import io.onetable.model.schema.PartitionTransformType;
+import io.onetable.model.storage.TableFormat;
 import io.onetable.model.sync.SyncMode;
 import io.onetable.model.sync.SyncResult;
 
@@ -52,39 +54,54 @@ public OneTableSyncTool(Properties props, Configuration hadoopConf) {
     super(props, hadoopConf);
     this.config = new OneTableSyncConfig(props);
     this.hudiSourceClientProvider = new HudiSourceClientProvider();
-    hudiSourceClientProvider.init(hadoopConf, Collections.emptyMap());
+    hudiSourceClientProvider.init(hadoopConf);
   }
 
   @Override
   public void syncHoodieTable() {
     List<String> formatsToSync =
         Arrays.stream(config.getString(OneTableSyncConfig.ONE_TABLE_FORMATS).split(","))
-            .map(format -> format.toUpperCase())
+            .map(String::toUpperCase)
             .collect(Collectors.toList());
     String basePath = config.getString(HoodieSyncConfig.META_SYNC_BASE_PATH);
     String tableName = config.getString(HoodieTableConfig.HOODIE_TABLE_NAME_KEY);
-    PerTableConfig perTableConfig =
-        PerTableConfigImpl.builder()
-            .tableName(tableName)
-            .tableBasePath(basePath)
-            .targetTableFormats(formatsToSync)
-            .hudiSourceConfig(
-                HudiSourceConfigImpl.builder()
-                    .partitionFieldSpecConfig(getPartitionSpecConfig())
-                    .build())
+    SourceTable sourceTable =
+        SourceTable.builder()
+            .name(tableName)
+            .basePath(basePath)
+            .formatName(TableFormat.HUDI)
+            .build();
+    List<TargetTable> targetTables =
+        formatsToSync.stream()
+            .map(
+                format ->
+                    TargetTable.builder()
+                        .formatName(format)
+                        .name(tableName)
+                        .metadataRetentionInHours(
+                            config.getInt(
+                                OneTableSyncConfig.ONE_TABLE_TARGET_METADATA_RETENTION_HOURS))
+                        .basePath(basePath)
+                        .build())
+            .collect(Collectors.toList());
+    TableSyncConfig tableSyncConfig =
+        TableSyncConfig.builder()
+            .sourceTable(sourceTable)
+            .targetTables(targetTables)
             .syncMode(SyncMode.INCREMENTAL)
-            .targetMetadataRetentionInHours(
-                config.getInt(OneTableSyncConfig.ONE_TABLE_TARGET_METADATA_RETENTION_HOURS))
+            .properties(
+                Collections.singletonMap(
+                    HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG, getPartitionSpecConfig()))
             .build();
     Map<String, SyncResult> results =
-        new OneTableClient(hadoopConf).sync(perTableConfig, hudiSourceClientProvider);
+        new OneTableClient(hadoopConf).sync(tableSyncConfig, hudiSourceClientProvider);
     String failingFormats =
         results.entrySet().stream()
             .filter(
                 entry ->
                     entry.getValue().getStatus().getStatusCode()
                         != SyncResult.SyncStatusCode.SUCCESS)
-            .map(entry -> entry.getKey().toString())
+            .map(Map.Entry::getKey)
             .collect(Collectors.joining(","));
     if (!failingFormats.isEmpty()) {
       throw new HoodieException("Unable to sync to OneTable for formats: " + failingFormats);
diff --git a/utilities/src/main/java/io/onetable/utilities/RunSync.java b/utilities/src/main/java/io/onetable/utilities/RunSync.java
index 5ece0c29b..849f143c5 100644
--- a/utilities/src/main/java/io/onetable/utilities/RunSync.java
+++ b/utilities/src/main/java/io/onetable/utilities/RunSync.java
@@ -23,8 +23,10 @@
 import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import lombok.Data;
 import lombok.extern.log4j.Log4j2;
@@ -45,10 +47,10 @@
 import com.google.common.annotations.VisibleForTesting;
 
 import io.onetable.client.OneTableClient;
-import io.onetable.client.PerTableConfig;
-import io.onetable.client.PerTableConfigImpl;
 import io.onetable.client.SourceClientProvider;
-import io.onetable.hudi.ConfigurationBasedPartitionSpecExtractor;
+import io.onetable.client.SourceTable;
+import io.onetable.client.TableSyncConfig;
+import io.onetable.client.TargetTable;
 import io.onetable.hudi.HudiSourceConfigImpl;
 import io.onetable.iceberg.IcebergCatalogConfig;
 import io.onetable.model.storage.TableFormat;
@@ -139,7 +141,7 @@ public static void main(String[] args) throws IOException {
     String sourceProviderClass = sourceClientConfig.sourceClientProviderClass;
     SourceClientProvider<?> sourceClientProvider =
         ReflectionUtils.createInstanceOfClass(sourceProviderClass);
-    sourceClientProvider.init(hadoopConf, sourceClientConfig.configuration);
+    sourceClientProvider.init(hadoopConf);
 
     List<String> tableFormatList = datasetConfig.getTargetFormats();
     OneTableClient client = new OneTableClient(hadoopConf);
@@ -148,24 +150,44 @@ public static void main(String[] args) throws IOException {
           "Running sync for basePath {} for following table formats {}",
           table.getTableBasePath(),
           tableFormatList);
-      PerTableConfig config =
-          PerTableConfigImpl.builder()
-              .tableBasePath(table.getTableBasePath())
-              .tableName(table.getTableName())
+      SourceTable sourceTable =
+          SourceTable.builder()
+              .name(table.getTableName())
+              .basePath(table.getTableBasePath())
               .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
-              .tableDataPath(table.getTableDataPath())
-              .icebergCatalogConfig(icebergCatalogConfig)
-              .hudiSourceConfig(
-                  HudiSourceConfigImpl.builder()
-                      .partitionSpecExtractorClass(
-                          ConfigurationBasedPartitionSpecExtractor.class.getName())
-                      .partitionFieldSpecConfig(table.getPartitionSpec())
-                      .build())
-              .targetTableFormats(tableFormatList)
+              .dataPath(table.getTableDataPath())
+              .catalogConfig(icebergCatalogConfig)
+              .formatName(sourceFormat)
+              .build();
+
+      List<TargetTable> targetTables =
+          tableFormatList.stream()
+              .map(
+                  tableFormat ->
+                      TargetTable.builder()
+                          .name(table.getTableName())
+                          .basePath(table.getTableBasePath())
+                          .namespace(
+                              table.getNamespace() == null
+                                  ? null
+                                  : table.getNamespace().split("\\."))
+                          .dataPath(table.getTableDataPath())
+                          .catalogConfig(icebergCatalogConfig)
+                          .formatName(tableFormat)
+                          .build())
+              .collect(Collectors.toList());
+
+      TableSyncConfig tableSyncConfig =
+          TableSyncConfig.builder()
+              .sourceTable(sourceTable)
+              .targetTables(targetTables)
               .syncMode(SyncMode.INCREMENTAL)
+              .properties(
+                  Collections.singletonMap(
+                      HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec()))
               .build();
       try {
-        client.sync(config, sourceClientProvider);
+        client.sync(tableSyncConfig, sourceClientProvider);
       } catch (Exception e) {
         log.error(String.format("Error running sync for %s", table.getTableBasePath()), e);
       }

From ee628768d0ae0baa9b4345011b3cddcb9917d7d3 Mon Sep 17 00:00:00 2001
From: Timothy Brown <tim@onehouse.ai>
Date: Wed, 17 Jan 2024 12:26:07 -0700
Subject: [PATCH 2/3] rename HudiSourceConfigImpl to HudiSourceConfig

---
 .../hudi/ConfigurationBasedPartitionSpecExtractor.java |  4 ++--
 .../io/onetable/hudi/HudiSourceClientProvider.java     |  2 +-
 ...HudiSourceConfigImpl.java => HudiSourceConfig.java} | 10 +++++-----
 core/src/test/java/io/onetable/ITOneTableClient.java   |  2 +-
 .../test/java/io/onetable/hudi/ITHudiSourceClient.java |  2 +-
 core/src/test/java/io/onetable/loadtest/LoadTest.java  |  2 +-
 .../java/io/onetable/hudi/sync/OneTableSyncTool.java   |  4 ++--
 .../src/main/java/io/onetable/utilities/RunSync.java   |  4 ++--
 8 files changed, 15 insertions(+), 15 deletions(-)
 rename core/src/main/java/io/onetable/hudi/{HudiSourceConfigImpl.java => HudiSourceConfig.java} (90%)

diff --git a/core/src/main/java/io/onetable/hudi/ConfigurationBasedPartitionSpecExtractor.java b/core/src/main/java/io/onetable/hudi/ConfigurationBasedPartitionSpecExtractor.java
index 9cb679a05..8fa1d61a1 100644
--- a/core/src/main/java/io/onetable/hudi/ConfigurationBasedPartitionSpecExtractor.java
+++ b/core/src/main/java/io/onetable/hudi/ConfigurationBasedPartitionSpecExtractor.java
@@ -36,13 +36,13 @@
  */
 @AllArgsConstructor
 public class ConfigurationBasedPartitionSpecExtractor implements HudiSourcePartitionSpecExtractor {
-  private final HudiSourceConfigImpl config;
+  private final HudiSourceConfig config;
 
   @Override
   public List<OnePartitionField> spec(OneSchema tableSchema) {
     List<OnePartitionField> partitionFields =
         new ArrayList<>(config.getPartitionFieldSpecs().size());
-    for (HudiSourceConfigImpl.PartitionFieldSpec fieldSpec : config.getPartitionFieldSpecs()) {
+    for (HudiSourceConfig.PartitionFieldSpec fieldSpec : config.getPartitionFieldSpecs()) {
       OneField sourceField =
           SchemaFieldFinder.getInstance()
               .findFieldByPath(tableSchema, fieldSpec.getSourceFieldPath());
diff --git a/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java b/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
index fa8fd9ce8..e20163fb0 100644
--- a/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
+++ b/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
@@ -47,7 +47,7 @@ public HudiClient getSourceClientInstance(
     }
 
     final HudiSourcePartitionSpecExtractor sourcePartitionSpecExtractor =
-        HudiSourceConfigImpl.fromProperties(clientConfigs).loadSourcePartitionSpecExtractor();
+        HudiSourceConfig.fromProperties(clientConfigs).loadSourcePartitionSpecExtractor();
 
     return new HudiClient(metaClient, sourcePartitionSpecExtractor);
   }
diff --git a/core/src/main/java/io/onetable/hudi/HudiSourceConfigImpl.java b/core/src/main/java/io/onetable/hudi/HudiSourceConfig.java
similarity index 90%
rename from core/src/main/java/io/onetable/hudi/HudiSourceConfigImpl.java
rename to core/src/main/java/io/onetable/hudi/HudiSourceConfig.java
index bb86ff5cd..aa29ef227 100644
--- a/core/src/main/java/io/onetable/hudi/HudiSourceConfigImpl.java
+++ b/core/src/main/java/io/onetable/hudi/HudiSourceConfig.java
@@ -33,7 +33,7 @@
 
 /** Configuration of Hudi source format for the sync process. */
 @Value
-public class HudiSourceConfigImpl {
+public class HudiSourceConfig {
   public static final String PARTITION_SPEC_EXTRACTOR_CLASS =
       "onetable.hudi.source.partition_spec_extractor_class";
   public static final String PARTITION_FIELD_SPEC_CONFIG =
@@ -42,13 +42,13 @@ public class HudiSourceConfigImpl {
   String partitionSpecExtractorClass;
   List<PartitionFieldSpec> partitionFieldSpecs;
 
-  public static HudiSourceConfigImpl fromPartitionFieldSpecConfig(String partitionFieldSpecConfig) {
-    return new HudiSourceConfigImpl(
+  public static HudiSourceConfig fromPartitionFieldSpecConfig(String partitionFieldSpecConfig) {
+    return new HudiSourceConfig(
         ConfigurationBasedPartitionSpecExtractor.class.getName(),
         parsePartitionFieldSpecs(partitionFieldSpecConfig));
   }
 
-  public static HudiSourceConfigImpl fromProperties(Map<String, String> properties) {
+  public static HudiSourceConfig fromProperties(Map<String, String> properties) {
     String partitionSpecExtractorClass =
         properties.getOrDefault(
             PARTITION_SPEC_EXTRACTOR_CLASS,
@@ -56,7 +56,7 @@ public static HudiSourceConfigImpl fromProperties(Map<String, String> properties
     String partitionFieldSpecString = properties.get(PARTITION_FIELD_SPEC_CONFIG);
     List<PartitionFieldSpec> partitionFieldSpecs =
         parsePartitionFieldSpecs(partitionFieldSpecString);
-    return new HudiSourceConfigImpl(partitionSpecExtractorClass, partitionFieldSpecs);
+    return new HudiSourceConfig(partitionSpecExtractorClass, partitionFieldSpecs);
   }
 
   @Value
diff --git a/core/src/test/java/io/onetable/ITOneTableClient.java b/core/src/test/java/io/onetable/ITOneTableClient.java
index a67de91e0..8c35b2262 100644
--- a/core/src/test/java/io/onetable/ITOneTableClient.java
+++ b/core/src/test/java/io/onetable/ITOneTableClient.java
@@ -19,7 +19,7 @@
 package io.onetable;
 
 import static io.onetable.GenericTable.getTableName;
-import static io.onetable.hudi.HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG;
+import static io.onetable.hudi.HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG;
 import static io.onetable.hudi.HudiTestUtil.PartitionConfig;
 import static io.onetable.model.storage.TableFormat.DELTA;
 import static io.onetable.model.storage.TableFormat.HUDI;
diff --git a/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java b/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
index 725618376..48b020038 100644
--- a/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
+++ b/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
@@ -534,7 +534,7 @@ private HudiClient getHudiSourceClient(
             .build();
     HudiSourcePartitionSpecExtractor partitionSpecExtractor =
         new ConfigurationBasedPartitionSpecExtractor(
-            HudiSourceConfigImpl.fromPartitionFieldSpecConfig(onetablePartitionConfig));
+            HudiSourceConfig.fromPartitionFieldSpecConfig(onetablePartitionConfig));
     return new HudiClient(hoodieTableMetaClient, partitionSpecExtractor);
   }
 
diff --git a/core/src/test/java/io/onetable/loadtest/LoadTest.java b/core/src/test/java/io/onetable/loadtest/LoadTest.java
index c7afc1fc6..5fe5e657c 100644
--- a/core/src/test/java/io/onetable/loadtest/LoadTest.java
+++ b/core/src/test/java/io/onetable/loadtest/LoadTest.java
@@ -18,7 +18,7 @@
  
 package io.onetable.loadtest;
 
-import static io.onetable.hudi.HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG;
+import static io.onetable.hudi.HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG;
 
 import java.nio.file.Path;
 import java.util.Arrays;
diff --git a/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java b/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
index 89c8c7e12..a123f44b6 100644
--- a/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
+++ b/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
@@ -39,7 +39,7 @@
 import io.onetable.client.TableSyncConfig;
 import io.onetable.client.TargetTable;
 import io.onetable.hudi.HudiSourceClientProvider;
-import io.onetable.hudi.HudiSourceConfigImpl;
+import io.onetable.hudi.HudiSourceConfig;
 import io.onetable.model.schema.PartitionTransformType;
 import io.onetable.model.storage.TableFormat;
 import io.onetable.model.sync.SyncMode;
@@ -91,7 +91,7 @@ public void syncHoodieTable() {
             .syncMode(SyncMode.INCREMENTAL)
             .properties(
                 Collections.singletonMap(
-                    HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG, getPartitionSpecConfig()))
+                    HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, getPartitionSpecConfig()))
             .build();
     Map<String, SyncResult> results =
         new OneTableClient(hadoopConf).sync(tableSyncConfig, hudiSourceClientProvider);
diff --git a/utilities/src/main/java/io/onetable/utilities/RunSync.java b/utilities/src/main/java/io/onetable/utilities/RunSync.java
index 849f143c5..fb0e538c9 100644
--- a/utilities/src/main/java/io/onetable/utilities/RunSync.java
+++ b/utilities/src/main/java/io/onetable/utilities/RunSync.java
@@ -51,7 +51,7 @@
 import io.onetable.client.SourceTable;
 import io.onetable.client.TableSyncConfig;
 import io.onetable.client.TargetTable;
-import io.onetable.hudi.HudiSourceConfigImpl;
+import io.onetable.hudi.HudiSourceConfig;
 import io.onetable.iceberg.IcebergCatalogConfig;
 import io.onetable.model.storage.TableFormat;
 import io.onetable.model.sync.SyncMode;
@@ -184,7 +184,7 @@ public static void main(String[] args) throws IOException {
               .syncMode(SyncMode.INCREMENTAL)
               .properties(
                   Collections.singletonMap(
-                      HudiSourceConfigImpl.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec()))
+                      HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, table.getPartitionSpec()))
               .build();
       try {
         client.sync(tableSyncConfig, sourceClientProvider);

From c05eeadd3d71cae89b29699bb4c64d8b226ff584 Mon Sep 17 00:00:00 2001
From: Timothy Brown <tim@onehouse.ai>
Date: Sun, 28 Jan 2024 12:33:43 -0600
Subject: [PATCH 3/3] make data path only part of the source

---
 .../java/io/onetable/client/ExternalTable.java   |  9 +++------
 .../java/io/onetable/client/SourceTable.java     |  8 ++++++--
 .../java/io/onetable/client/TargetTable.java     | 16 +++++++++-------
 .../java/io/onetable/client/TestSourceTable.java |  6 +++---
 .../java/io/onetable/client/TestTargetTable.java |  6 +++---
 .../main/java/io/onetable/delta/DeltaClient.java |  8 ++++----
 .../delta/DeltaSourceClientProvider.java         |  4 ++--
 .../onetable/hudi/HudiSourceClientProvider.java  |  2 +-
 .../java/io/onetable/hudi/HudiTargetClient.java  | 12 ++++++------
 .../java/io/onetable/iceberg/IcebergClient.java  |  4 ++--
 .../io/onetable/iceberg/IcebergSourceClient.java |  2 +-
 .../io/onetable/hudi/sync/OneTableSyncTool.java  |  9 +++++----
 .../main/java/io/onetable/utilities/RunSync.java |  6 +++---
 13 files changed, 48 insertions(+), 44 deletions(-)

diff --git a/api/src/main/java/io/onetable/client/ExternalTable.java b/api/src/main/java/io/onetable/client/ExternalTable.java
index beedee58a..fe6100c88 100644
--- a/api/src/main/java/io/onetable/client/ExternalTable.java
+++ b/api/src/main/java/io/onetable/client/ExternalTable.java
@@ -31,8 +31,7 @@
 public class ExternalTable {
   @NonNull String name;
   @NonNull String formatName;
-  @NonNull String basePath;
-  @NonNull String dataPath;
+  @NonNull String metadataPath;
   String[] namespace;
   CatalogConfig catalogConfig;
 
@@ -40,18 +39,16 @@ public class ExternalTable {
       @NonNull String name,
       @NonNull String formatName,
       @NonNull String basePath,
-      String dataPath,
       String[] namespace,
       CatalogConfig catalogConfig) {
     this.name = name;
     this.formatName = formatName;
-    this.basePath = sanitizeBasePath(basePath);
-    this.dataPath = dataPath == null ? this.basePath : sanitizeBasePath(dataPath);
+    this.metadataPath = sanitizeBasePath(basePath);
     this.namespace = namespace;
     this.catalogConfig = catalogConfig;
   }
 
-  private String sanitizeBasePath(String tableBasePath) {
+  protected String sanitizeBasePath(String tableBasePath) {
     Path path = new Path(tableBasePath);
     Preconditions.checkArgument(path.isAbsolute(), "Table base path must be absolute");
     if (path.isAbsoluteAndSchemeAuthorityNull()) {
diff --git a/api/src/main/java/io/onetable/client/SourceTable.java b/api/src/main/java/io/onetable/client/SourceTable.java
index 61d7c45b1..56ad118de 100644
--- a/api/src/main/java/io/onetable/client/SourceTable.java
+++ b/api/src/main/java/io/onetable/client/SourceTable.java
@@ -20,17 +20,21 @@
 
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
+import lombok.NonNull;
 
 @EqualsAndHashCode(callSuper = true)
 public class SourceTable extends ExternalTable {
+  @NonNull String dataPath;
+
   @Builder(toBuilder = true)
   public SourceTable(
       String name,
       String formatName,
-      String basePath,
+      String metadataPath,
       String dataPath,
       String[] namespace,
       CatalogConfig catalogConfig) {
-    super(name, formatName, basePath, dataPath, namespace, catalogConfig);
+    super(name, formatName, metadataPath, namespace, catalogConfig);
+    this.dataPath = dataPath == null ? this.metadataPath : sanitizeBasePath(dataPath);
   }
 }
diff --git a/api/src/main/java/io/onetable/client/TargetTable.java b/api/src/main/java/io/onetable/client/TargetTable.java
index 526451872..a42ee5cbc 100644
--- a/api/src/main/java/io/onetable/client/TargetTable.java
+++ b/api/src/main/java/io/onetable/client/TargetTable.java
@@ -22,22 +22,24 @@
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+
 @Getter
 @EqualsAndHashCode(callSuper = true)
 public class TargetTable extends ExternalTable {
-  int metadataRetentionInHours;
+  Duration metadataRetention;
 
   @Builder(toBuilder = true)
   public TargetTable(
       String name,
       String formatName,
-      String basePath,
-      String dataPath,
+      String metadataPath,
       String[] namespace,
       CatalogConfig catalogConfig,
-      Integer metadataRetentionInHours) {
-    super(name, formatName, basePath, dataPath, namespace, catalogConfig);
-    this.metadataRetentionInHours =
-        metadataRetentionInHours == null ? 24 * 7 : metadataRetentionInHours;
+      Duration metadataRetention) {
+    super(name, formatName, metadataPath, namespace, catalogConfig);
+    this.metadataRetention =
+        metadataRetention == null ? Duration.of(7, ChronoUnit.DAYS) : metadataRetention;
   }
 }
diff --git a/api/src/test/java/io/onetable/client/TestSourceTable.java b/api/src/test/java/io/onetable/client/TestSourceTable.java
index fb1daf7f7..b32c75472 100644
--- a/api/src/test/java/io/onetable/client/TestSourceTable.java
+++ b/api/src/test/java/io/onetable/client/TestSourceTable.java
@@ -28,11 +28,11 @@ public class TestSourceTable {
   void sanitizePath() {
     SourceTable tooManySlashes =
         SourceTable.builder().basePath("s3://bucket//path").name("name").formatName("hudi").build();
-    assertEquals("s3://bucket/path", tooManySlashes.getBasePath());
+    assertEquals("s3://bucket/path", tooManySlashes.getMetadataPath());
 
     SourceTable localFilePath =
         SourceTable.builder().basePath("/local/data//path").name("name").formatName("hudi").build();
-    assertEquals("file:///local/data/path", localFilePath.getBasePath());
+    assertEquals("file:///local/data/path", localFilePath.getMetadataPath());
 
     SourceTable properLocalFilePath =
         SourceTable.builder()
@@ -40,7 +40,7 @@ void sanitizePath() {
             .name("name")
             .formatName("hudi")
             .build();
-    assertEquals("file:///local/data/path", properLocalFilePath.getBasePath());
+    assertEquals("file:///local/data/path", properLocalFilePath.getMetadataPath());
   }
 
   @Test
diff --git a/api/src/test/java/io/onetable/client/TestTargetTable.java b/api/src/test/java/io/onetable/client/TestTargetTable.java
index a39157684..b4ffe535b 100644
--- a/api/src/test/java/io/onetable/client/TestTargetTable.java
+++ b/api/src/test/java/io/onetable/client/TestTargetTable.java
@@ -29,11 +29,11 @@ public class TestTargetTable {
   void sanitizePath() {
     TargetTable tooManySlashes =
         TargetTable.builder().basePath("s3://bucket//path").name("name").formatName("hudi").build();
-    assertEquals("s3://bucket/path", tooManySlashes.getBasePath());
+    assertEquals("s3://bucket/path", tooManySlashes.getMetadataPath());
 
     TargetTable localFilePath =
         TargetTable.builder().basePath("/local/data//path").name("name").formatName("hudi").build();
-    assertEquals("file:///local/data/path", localFilePath.getBasePath());
+    assertEquals("file:///local/data/path", localFilePath.getMetadataPath());
 
     TargetTable properLocalFilePath =
         TargetTable.builder()
@@ -41,7 +41,7 @@ void sanitizePath() {
             .name("name")
             .formatName("hudi")
             .build();
-    assertEquals("file:///local/data/path", properLocalFilePath.getBasePath());
+    assertEquals("file:///local/data/path", properLocalFilePath.getMetadataPath());
   }
 
   @Test
diff --git a/core/src/main/java/io/onetable/delta/DeltaClient.java b/core/src/main/java/io/onetable/delta/DeltaClient.java
index 69012a9f9..50d9ea89c 100644
--- a/core/src/main/java/io/onetable/delta/DeltaClient.java
+++ b/core/src/main/java/io/onetable/delta/DeltaClient.java
@@ -82,9 +82,9 @@ public DeltaClient() {}
 
   public DeltaClient(TargetTable targetTable, SparkSession sparkSession) {
     this(
-        targetTable.getDataPath(),
+        targetTable.getMetadataPath(),
         targetTable.getName(),
-        targetTable.getMetadataRetentionInHours(),
+        (int) targetTable.getMetadataRetention().toHours(),
         sparkSession,
         DeltaSchemaExtractor.getInstance(),
         DeltaPartitionExtractor.getInstance(),
@@ -137,9 +137,9 @@ public void init(TargetTable targetTable, Configuration configuration) {
     SparkSession sparkSession = DeltaClientUtils.buildSparkSession(configuration);
 
     _init(
-        targetTable.getDataPath(),
+        targetTable.getMetadataPath(),
         targetTable.getName(),
-        targetTable.getMetadataRetentionInHours(),
+        (int) targetTable.getMetadataRetention().toHours(),
         sparkSession,
         DeltaSchemaExtractor.getInstance(),
         DeltaPartitionExtractor.getInstance(),
diff --git a/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java b/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java
index 53efcc33c..710307e5c 100644
--- a/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java
+++ b/core/src/main/java/io/onetable/delta/DeltaSourceClientProvider.java
@@ -33,12 +33,12 @@ public class DeltaSourceClientProvider extends SourceClientProvider<Long> {
   public DeltaSourceClient getSourceClientInstance(
       SourceTable sourceTable, Map<String, String> clientConfigs) {
     SparkSession sparkSession = DeltaClientUtils.buildSparkSession(hadoopConf);
-    DeltaTable deltaTable = DeltaTable.forPath(sparkSession, sourceTable.getBasePath());
+    DeltaTable deltaTable = DeltaTable.forPath(sparkSession, sourceTable.getMetadataPath());
     DeltaSourceClient deltaSourceClient =
         DeltaSourceClient.builder()
             .sparkSession(sparkSession)
             .tableName(sourceTable.getName())
-            .basePath(sourceTable.getBasePath())
+            .basePath(sourceTable.getMetadataPath())
             .deltaTable(deltaTable)
             .deltaLog(deltaTable.deltaLog())
             .build();
diff --git a/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java b/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
index e20163fb0..704e6bde8 100644
--- a/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
+++ b/core/src/main/java/io/onetable/hudi/HudiSourceClientProvider.java
@@ -39,7 +39,7 @@ public HudiClient getSourceClientInstance(
     HoodieTableMetaClient metaClient =
         HoodieTableMetaClient.builder()
             .setConf(hadoopConf)
-            .setBasePath(sourceTableConfig.getBasePath())
+            .setBasePath(sourceTableConfig.getMetadataPath())
             .setLoadActiveTimelineOnLoad(true)
             .build();
     if (!metaClient.getTableConfig().getTableType().equals(HoodieTableType.COPY_ON_WRITE)) {
diff --git a/core/src/main/java/io/onetable/hudi/HudiTargetClient.java b/core/src/main/java/io/onetable/hudi/HudiTargetClient.java
index 639d6002f..6d4e99399 100644
--- a/core/src/main/java/io/onetable/hudi/HudiTargetClient.java
+++ b/core/src/main/java/io/onetable/hudi/HudiTargetClient.java
@@ -114,11 +114,11 @@ public HudiTargetClient() {}
       Configuration configuration,
       int maxNumDeltaCommitsBeforeCompaction) {
     this(
-        targetTable.getDataPath(),
-        targetTable.getMetadataRetentionInHours(),
+        targetTable.getMetadataPath(),
+        (int) targetTable.getMetadataRetention().toHours(),
         maxNumDeltaCommitsBeforeCompaction,
         BaseFileUpdatesExtractor.of(
-            new HoodieJavaEngineContext(configuration), new CachingPath(targetTable.getDataPath())),
+            new HoodieJavaEngineContext(configuration), new CachingPath(targetTable.getMetadataPath())),
         AvroSchemaConverter.getInstance(),
         HudiTableManager.of(configuration),
         CommitState::new);
@@ -166,11 +166,11 @@ private void _init(
   @Override
   public void init(TargetTable targetTable, Configuration configuration) {
     _init(
-        targetTable.getDataPath(),
-        targetTable.getMetadataRetentionInHours(),
+        targetTable.getMetadataPath(),
+        (int) targetTable.getMetadataRetention().toHours(),
         HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.defaultValue(),
         BaseFileUpdatesExtractor.of(
-            new HoodieJavaEngineContext(configuration), new CachingPath(targetTable.getDataPath())),
+            new HoodieJavaEngineContext(configuration), new CachingPath(targetTable.getMetadataPath())),
         AvroSchemaConverter.getInstance(),
         HudiTableManager.of(configuration),
         CommitState::new);
diff --git a/core/src/main/java/io/onetable/iceberg/IcebergClient.java b/core/src/main/java/io/onetable/iceberg/IcebergClient.java
index 30b9a02e7..d2fe326b6 100644
--- a/core/src/main/java/io/onetable/iceberg/IcebergClient.java
+++ b/core/src/main/java/io/onetable/iceberg/IcebergClient.java
@@ -108,9 +108,9 @@ private void _init(
     this.partitionSpecSync = partitionSpecSync;
     this.dataFileUpdatesExtractor = dataFileUpdatesExtractor;
     String tableName = targetTable.getName();
-    this.basePath = targetTable.getBasePath();
+    this.basePath = targetTable.getMetadataPath();
     this.configuration = configuration;
-    this.snapshotRetentionInHours = targetTable.getMetadataRetentionInHours();
+    this.snapshotRetentionInHours = (int) targetTable.getMetadataRetention().toHours();
     String[] namespace = targetTable.getNamespace();
     this.tableIdentifier =
         namespace == null
diff --git a/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java b/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java
index c83d5c3df..14492ce25 100644
--- a/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java
+++ b/core/src/main/java/io/onetable/iceberg/IcebergSourceClient.java
@@ -82,7 +82,7 @@ private Table initSourceTable() {
     return tableManager.getTable(
         (IcebergCatalogConfig) sourceTableConfig.getCatalogConfig(),
         tableIdentifier,
-        sourceTableConfig.getBasePath());
+        sourceTableConfig.getMetadataPath());
   }
 
   private FileIO initTableOps() {
diff --git a/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java b/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
index a123f44b6..db91f6885 100644
--- a/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
+++ b/hudi-support/extensions/src/main/java/io/onetable/hudi/sync/OneTableSyncTool.java
@@ -18,6 +18,7 @@
  
 package io.onetable.hudi.sync;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -68,7 +69,7 @@ public void syncHoodieTable() {
     SourceTable sourceTable =
         SourceTable.builder()
             .name(tableName)
-            .basePath(basePath)
+            .metadataPath(basePath)
             .formatName(TableFormat.HUDI)
             .build();
     List<TargetTable> targetTables =
@@ -78,10 +79,10 @@ public void syncHoodieTable() {
                     TargetTable.builder()
                         .formatName(format)
                         .name(tableName)
-                        .metadataRetentionInHours(
+                        .metadataRetention(Duration.ofHours(
                             config.getInt(
-                                OneTableSyncConfig.ONE_TABLE_TARGET_METADATA_RETENTION_HOURS))
-                        .basePath(basePath)
+                                OneTableSyncConfig.ONE_TABLE_TARGET_METADATA_RETENTION_HOURS)))
+                        .metadataPath(basePath)
                         .build())
             .collect(Collectors.toList());
     TableSyncConfig tableSyncConfig =
diff --git a/utilities/src/main/java/io/onetable/utilities/RunSync.java b/utilities/src/main/java/io/onetable/utilities/RunSync.java
index fb0e538c9..22650c38f 100644
--- a/utilities/src/main/java/io/onetable/utilities/RunSync.java
+++ b/utilities/src/main/java/io/onetable/utilities/RunSync.java
@@ -153,7 +153,7 @@ public static void main(String[] args) throws IOException {
       SourceTable sourceTable =
           SourceTable.builder()
               .name(table.getTableName())
-              .basePath(table.getTableBasePath())
+              .metadataPath(table.getTableBasePath())
               .namespace(table.getNamespace() == null ? null : table.getNamespace().split("\\."))
               .dataPath(table.getTableDataPath())
               .catalogConfig(icebergCatalogConfig)
@@ -166,12 +166,12 @@ public static void main(String[] args) throws IOException {
                   tableFormat ->
                       TargetTable.builder()
                           .name(table.getTableName())
-                          .basePath(table.getTableBasePath())
+                          .metadataPath(table.getTableBasePath())
                           .namespace(
                               table.getNamespace() == null
                                   ? null
                                   : table.getNamespace().split("\\."))
-                          .dataPath(table.getTableDataPath())
+                          .metadataPath(table.getTableDataPath())
                           .catalogConfig(icebergCatalogConfig)
                           .formatName(tableFormat)
                           .build())