Refactor policy validators to inherit an interface for future policies (

#277)

## Summary

<!--- HINT: Replace #nnn with corresponding Issue number, if you are
fixing an existing issue -->

[Issue](https://github.com/linkedin/openhouse/issues/#nnn)] Briefly
discuss the summary of the changes made in this
pull request in 2-3 lines.

As Openhouse begins to support more policies, it becomes beneficial to
have all the validators follow a common interface, defined by
`PolicySpecValidator`. It takes in the entire request sent for the
Create/Update table request since some policies may be dependent on
other properties of the table.

This PR performs a small refactor to have these classes share the same
base class, and future policy definitions to follow this pattern.

## Changes

- [ ] Client-facing API Changes
- [ ] Internal API Changes
- [ ] Bug Fixes
- [ ] New Features
- [ ] Performance Improvements
- [ ] Code Style
- [x] Refactoring
- [ ] Documentation
- [ ] Tests

For all the boxes checked, please include additional details of the
changes made in this pull request.

## Testing Done
<!--- Check any relevant boxes with "x" -->

- [ ] Manually Tested on local docker setup. Please include commands
ran, and their output.
- [ ] Added new tests for the changes made.
- [ ] Updated existing tests to reflect the changes made.
- [ ] No tests added or updated. Please explain why. If unsure, please
feel free to ask for help.
- [ ] Some other form of testing like staging or soak time in
production. Please explain.

For all the boxes checked, include a detailed description of the testing
done for the changes made in this pull request.

# Additional Information

- [ ] Breaking Changes
- [ ] Deprecations
- [ ] Large PR broken into smaller PRs, and PR plan linked in the
description.

For all the boxes checked, include additional details of the changes
made in this pull request.

services/tables/src/main/java/com/linkedin/openhouse/tables/api/validator/impl/HistoryPolicySpecValidator.java

@@ -1,19 +1,19 @@
 package com.linkedin.openhouse.tables.api.validator.impl;
 
 import com.linkedin.openhouse.common.api.spec.TableUri;
+import com.linkedin.openhouse.tables.api.spec.v0.request.CreateUpdateTableRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.request.components.History;
 import com.linkedin.openhouse.tables.api.spec.v0.request.components.TimePartitionSpec;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 @Slf4j
 @Component
-public class HistoryPolicySpecValidator {
+public class HistoryPolicySpecValidator extends PolicySpecValidator {
 
-  private String failureMessage = "";
-  private String errorField = "";
-
-  protected boolean validate(History history, TableUri tableUri) {
+  public boolean validate(
+      CreateUpdateTableRequestBody createUpdateTableRequestBody, TableUri tableUri) {
+    History history = createUpdateTableRequestBody.getPolicies().getHistory();
     if (history != null) {
       if (history.getMaxAge() <= 0 && history.getVersions() <= 0) {
         failureMessage =
@@ -86,12 +86,4 @@ protected boolean validateHistoryConfigVersionsWithinBounds(History history) {
     int versions = history.getVersions();
     return versions >= 2 && versions <= 100;
   }
-
-  public String getMessage() {
-    return failureMessage;
-  }
-
-  public String getField() {
-    return errorField;
-  }
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/api/validator/impl/OpenHouseTablesApiValidator.java

@@ -128,46 +128,24 @@ private List<String> validateUUIDForReplicaTable(
   }
 
   private void validatePolicies(CreateUpdateTableRequestBody createUpdateTableRequestBody) {
+    if (createUpdateTableRequestBody.getPolicies() == null) {
+      return;
+    }
+
     TableUri tableUri =
         TableUri.builder()
             .tableId(createUpdateTableRequestBody.getTableId())
             .clusterId(createUpdateTableRequestBody.getClusterId())
             .databaseId(createUpdateTableRequestBody.getDatabaseId())
             .build();
-    if (!retentionPolicySpecValidator.validate(
-        createUpdateTableRequestBody.getPolicies(),
-        createUpdateTableRequestBody.getTimePartitioning(),
-        tableUri,
-        createUpdateTableRequestBody.getSchema())) {
-      throw new RequestValidationFailureException(
-          Arrays.asList(
-              String.format(
-                  "%s : %s",
-                  retentionPolicySpecValidator.getField(),
-                  retentionPolicySpecValidator.getMessage())));
-    }
-    if (createUpdateTableRequestBody.getPolicies() != null
-        && createUpdateTableRequestBody.getPolicies().getReplication() != null) {
-      if (!replicationConfigValidator.validate(
-          createUpdateTableRequestBody.getPolicies().getReplication(), tableUri)) {
-        throw new RequestValidationFailureException(
-            Arrays.asList(
-                String.format(
-                    "%s : %s",
-                    replicationConfigValidator.getField(),
-                    replicationConfigValidator.getMessage())));
-      }
-    }
-    if (createUpdateTableRequestBody.getPolicies() != null
-        && createUpdateTableRequestBody.getPolicies().getHistory() != null) {
-      if (!historyPolicySpecValidator.validate(
-          createUpdateTableRequestBody.getPolicies().getHistory(), tableUri)) {
+
+    List<PolicySpecValidator> validators =
+        Arrays.asList(
+            retentionPolicySpecValidator, replicationConfigValidator, historyPolicySpecValidator);
+    for (PolicySpecValidator validator : validators) {
+      if (!validator.validate(createUpdateTableRequestBody, tableUri)) {
         throw new RequestValidationFailureException(
-            Arrays.asList(
-                String.format(
-                    "%s : %s",
-                    historyPolicySpecValidator.getField(),
-                    historyPolicySpecValidator.getMessage())));
+            Arrays.asList(String.format("%s : %s", validator.getField(), validator.getMessage())));
       }
     }
   }

services/tables/src/main/java/com/linkedin/openhouse/tables/api/validator/impl/PolicySpecValidator.java

@@ -0,0 +1,22 @@
+package com.linkedin.openhouse.tables.api.validator.impl;
+
+import com.linkedin.openhouse.common.api.spec.TableUri;
+import com.linkedin.openhouse.tables.api.spec.v0.request.CreateUpdateTableRequestBody;
+
+public abstract class PolicySpecValidator {
+
+  protected String failureMessage = "";
+
+  protected String errorField = "";
+
+  public abstract boolean validate(
+      CreateUpdateTableRequestBody createUpdateTableRequestBody, TableUri tableUri);
+
+  public String getField() {
+    return errorField;
+  }
+
+  public String getMessage() {
+    return failureMessage;
+  }
+}

services/tables/src/main/java/com/linkedin/openhouse/tables/api/validator/impl/ReplicationConfigValidator.java

@@ -1,6 +1,7 @@
 package com.linkedin.openhouse.tables.api.validator.impl;
 
 import com.linkedin.openhouse.common.api.spec.TableUri;
+import com.linkedin.openhouse.tables.api.spec.v0.request.CreateUpdateTableRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.request.components.Replication;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -11,22 +12,13 @@
  */
 @Slf4j
 @Component
-public class ReplicationConfigValidator {
-  private String failureMessage = "";
-  private String errorField = "";
-
-  public boolean validate(Replication replication, TableUri tableUri) {
+public class ReplicationConfigValidator extends PolicySpecValidator {
+  public boolean validate(
+      CreateUpdateTableRequestBody createUpdateTableRequestBody, TableUri tableUri) {
+    Replication replication = createUpdateTableRequestBody.getPolicies().getReplication();
     if (replication != null) {
       log.info(String.format("Table: %s replication: %s\n", tableUri, replication));
     }
     return true;
   }
-
-  public String getMessage() {
-    return failureMessage;
-  }
-
-  public String getField() {
-    return errorField;
-  }
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/api/validator/impl/RetentionPolicySpecValidator.java

@@ -3,7 +3,7 @@
 import static com.linkedin.openhouse.common.schema.IcebergSchemaHelper.*;
 
 import com.linkedin.openhouse.common.api.spec.TableUri;
-import com.linkedin.openhouse.tables.api.spec.v0.request.components.Policies;
+import com.linkedin.openhouse.tables.api.spec.v0.request.CreateUpdateTableRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.request.components.Retention;
 import com.linkedin.openhouse.tables.api.spec.v0.request.components.TimePartitionSpec;
 import com.linkedin.openhouse.tables.common.DefaultColumnPattern;
@@ -18,39 +18,37 @@
  */
 @Component
 @Slf4j
-public class RetentionPolicySpecValidator {
+public class RetentionPolicySpecValidator extends PolicySpecValidator {
 
-  private String failureMessage = "";
-
-  private String errorField = "";
   /**
    * Invalid cases for retention object 0. retention column not found in the schema object. 1.
    * retention column pattern containing invalid characters. 2. missing retention column pattern in
    * provided retention object when table is not time-partitioned. (Otherwise such retention isn't
    * useful) 3(1). when table is time-partitioned: providing retention column type is invalid. 3(2)
    * when table is time-partitioned: Granularity mismatch with retention column is invalid.
    *
-   * @param policies {@link Policies} Policies object that needs to be validated and set. null
-   *     policy object is accepted.
-   * @param timePartitioning {@link TimePartitionSpec} TimePartitionSpec containing the granularity
-   *     against which policies.retention.granularity is validated
-   * @param schema {@link String} Raw schema representation deserialized from wire.
+   * @param createUpdateTableRequestBody {@link CreateUpdateTableRequestBody} API request body for
+   *     creation and updating tables
    * @return Boolean validity of constraint
    */
+  @Override
   public boolean validate(
-      Policies policies, TimePartitionSpec timePartitioning, TableUri tableUri, String schema) {
+      CreateUpdateTableRequestBody createUpdateTableRequestBody, TableUri tableUri) {
+    Retention retention = createUpdateTableRequestBody.getPolicies().getRetention();
+    TimePartitionSpec timePartitioning = createUpdateTableRequestBody.getTimePartitioning();
+    String schema = createUpdateTableRequestBody.getSchema();
 
-    if (policies != null && policies.getRetention() != null) {
+    if (retention != null) {
       // Two invalid case for timePartitioned table
       if (timePartitioning != null) {
-        if (policies.getRetention().getColumnPattern() != null) {
+        if (retention.getColumnPattern() != null) {
           failureMessage =
               String.format(
                   "You can only specify retention column pattern on non-timestampPartitioned table (table[%s] is time-partitioned by[%s])",
                   tableUri, timePartitioning.getColumnName());
           return false;
         }
-        if (!policies.getRetention().getGranularity().equals(timePartitioning.getGranularity())) {
+        if (!retention.getGranularity().equals(timePartitioning.getGranularity())) {
           failureMessage =
               String.format(
                   "invalid policies retention granularity format for table %s. Policies granularity must be equal to or lesser than"
@@ -62,23 +60,22 @@ public boolean validate(
       }
 
       // invalid cases regarding the integrity of retention object.
-      if (!validateGranularityWithPattern(policies.getRetention())) {
+      if (!validateGranularityWithPattern(retention)) {
         failureMessage =
             String.format(
                 "Provided Retention Granularity[%s] is not supported with default pattern. "
                     + "Please define pattern in retention config or use one of supported granularity: %s",
-                policies.getRetention().getGranularity().name(),
-                Arrays.toString(DefaultColumnPattern.values()));
+                retention.getGranularity().name(), Arrays.toString(DefaultColumnPattern.values()));
         return false;
       }
-      if (!validatePatternIfPresent(policies.getRetention(), tableUri, schema)) {
+      if (!validatePatternIfPresent(retention, tableUri, schema)) {
         failureMessage =
             String.format(
                 "Provided pattern[%s] is not recognizable by OpenHouse for the table[%s]; Also please make sure the declared column is part of table schema.",
-                policies.getRetention().getColumnPattern(), tableUri);
+                retention.getColumnPattern(), tableUri);
         return false;
       }
-      if (timePartitioning == null && policies.getRetention().getColumnPattern() == null) {
+      if (timePartitioning == null && retention.getColumnPattern() == null) {
         failureMessage =
             String.format(
                 "For non timestamp-partitioned table %s, column pattern in retention policy is mandatory",
@@ -135,12 +132,4 @@ protected boolean validateGranularityWithPattern(Retention retention) {
 
     return true;
   }
-
-  public String getField() {
-    return errorField;
-  }
-
-  public String getMessage() {
-    return failureMessage;
-  }
 }