[Draft] Refactor config classes

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();
+ }
+ }
+}

api/src/main/java/io/onetable/client/HudiSourceConfig.java → 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);
+ }
 }

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;
+ }
+}

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);
 }

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());
+ }
+}

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());
+ }
+}

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();
  }