Add feature table deletion

common-test/src/main/java/feast/common/it/SimpleCoreClient.java

@@ -218,4 +218,12 @@ public FeatureTableProto.FeatureTable applyFeatureTable(
                 .build())
         .getTable();
   }
+
+  public void deleteFeatureTable(String projectName, String featureTableName) {
+    stub.deleteFeatureTable(
+        CoreServiceProto.DeleteFeatureTableRequest.newBuilder()
+            .setProject(projectName)
+            .setName(featureTableName)
+            .build());
+  }
 }

core/src/main/java/feast/core/grpc/CoreServiceImpl.java

@@ -491,4 +491,30 @@ public void getFeatureTable(
           Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
     }
   }
+
+  @Override
+  public void deleteFeatureTable(
+      DeleteFeatureTableRequest request,
+      StreamObserver<DeleteFeatureTableResponse> responseObserver) {
+    String projectName = request.getProject();
+    try {
+      // Check if user has authorization to delete feature table
+      authorizationService.authorizeRequest(SecurityContextHolder.getContext(), projectName);
+      specService.deleteFeatureTable(request);
+
+      responseObserver.onNext(DeleteFeatureTableResponse.getDefaultInstance());
+      responseObserver.onCompleted();
+    } catch (NoSuchElementException e) {
+      log.error(
+          String.format(
+              "DeleteFeatureTable: No such Feature Table: (project: %s, name: %s)",
+              request.getProject(), request.getName()));
+      responseObserver.onError(
+          Status.NOT_FOUND.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    } catch (Exception e) {
+      log.error("DeleteFeatureTable: Exception has occurred: ", e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    }
+  }
 }

core/src/main/java/feast/core/model/FeatureTable.java

@@ -16,6 +16,7 @@
  */
 package feast.core.model;
 
+import com.google.common.hash.Hashing;
 import com.google.protobuf.Duration;
 import com.google.protobuf.Timestamp;
 import feast.core.dao.EntityRepository;
@@ -24,12 +25,7 @@
 import feast.proto.core.FeatureProto.FeatureSpecV2;
 import feast.proto.core.FeatureTableProto;
 import feast.proto.core.FeatureTableProto.FeatureTableSpec;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 import javax.persistence.CascadeType;
 import javax.persistence.Column;
@@ -107,6 +103,9 @@ public class FeatureTable extends AbstractTimestampEntity {
   @Column(name = "revision", nullable = false)
   private int revision;
 
+  @Column(name = "is_deleted", nullable = false)
+  private boolean isDeleted;
+
   public FeatureTable() {};
 
   /**
@@ -202,6 +201,9 @@ public void updateFromProto(
       this.streamSource = null;
     }
 
+    // Set isDeleted to false
+    this.setDeleted(false);
+
     // Bump revision no.
     this.revision++;
   }
@@ -211,6 +213,7 @@ public FeatureTableProto.FeatureTable toProto() {
     // Convert field types to Protobuf compatible types
     Timestamp creationTime = TypeConversion.convertTimestamp(getCreated());
     Timestamp updatedTime = TypeConversion.convertTimestamp(getLastUpdated());
+    String metadataHashBytes = this.protoHash();
 
     List<FeatureSpecV2> featureSpecs =
         getFeatures().stream().map(FeatureV2::toProto).collect(Collectors.toList());
@@ -236,6 +239,7 @@ public FeatureTableProto.FeatureTable toProto() {
                 .setRevision(getRevision())
                 .setCreatedTimestamp(creationTime)
                 .setLastUpdatedTimestamp(updatedTime)
+                .setHash(metadataHashBytes)
                 .build())
         .setSpec(spec.build())
         .build();
@@ -279,6 +283,37 @@ public Map<String, String> getLabelsMap() {
     return TypeConversion.convertJsonStringToMap(getLabelsJSON());
   }
 
+  public void delete() {
+    this.setDeleted(true);
+    this.setRevision(0);
+  }
+
+  public String protoHash() {
+    List<String> sortedEntities =
+        this.getEntities().stream().map(entity -> entity.getName()).collect(Collectors.toList());
+    Collections.sort(sortedEntities);
+
+    List<FeatureV2> sortedFeatures = new ArrayList(this.getFeatures());
+    List<FeatureSpecV2> sortedFeatureSpecs =
+        sortedFeatures.stream().map(featureV2 -> featureV2.toProto()).collect(Collectors.toList());
+    sortedFeatures.sort(Comparator.comparing(FeatureV2::getName));
+
+    DataSourceProto.DataSource streamSource = DataSourceProto.DataSource.getDefaultInstance();
+    if (getStreamSource() != null) {
+      streamSource = getStreamSource().toProto();
+    }
+
+    FeatureTableSpec featureTableSpec =
+        FeatureTableSpec.newBuilder()
+            .addAllEntities(sortedEntities)
+            .addAllFeatures(sortedFeatureSpecs)
+            .setBatchSource(getBatchSource().toProto())
+            .setStreamSource(streamSource)
+            .setMaxAge(Duration.newBuilder().setSeconds(getMaxAgeSecs()).build())
+            .build();
+    return Hashing.murmur3_32().hashBytes(featureTableSpec.toByteArray()).toString();
+  }
+
   @Override
   public int hashCode() {
     return Objects.hash(

core/src/main/java/feast/core/service/SpecService.java

@@ -37,6 +37,7 @@
 import feast.proto.core.CoreServiceProto.ApplyFeatureSetResponse.Status;
 import feast.proto.core.CoreServiceProto.ApplyFeatureTableRequest;
 import feast.proto.core.CoreServiceProto.ApplyFeatureTableResponse;
+import feast.proto.core.CoreServiceProto.DeleteFeatureTableRequest;
 import feast.proto.core.CoreServiceProto.GetEntityRequest;
 import feast.proto.core.CoreServiceProto.GetEntityResponse;
 import feast.proto.core.CoreServiceProto.GetFeatureSetRequest;
@@ -702,6 +703,7 @@ public ListFeatureTablesResponse listFeatureTables(ListFeatureTablesRequest.Filt
       matchingTables =
           matchingTables.stream()
               .filter(table -> table.hasAllLabels(labelsFilter))
+              .filter(table -> !table.isDeleted())
               .collect(Collectors.toList());
     }
     for (FeatureTable table : matchingTables) {
@@ -735,10 +737,36 @@ public GetFeatureTableResponse getFeatureTable(GetFeatureTableRequest request) {
               "No such Feature Table: (project: %s, name: %s)", projectName, featureTableName));
     }
 
+    if (retrieveTable.get().isDeleted()) {
+      throw new NoSuchElementException(
+          String.format(
+              "Feature Table has been deleted: (project: %s, name: %s)",
+              projectName, featureTableName));
+    }
+
     // Build GetFeatureTableResponse
     GetFeatureTableResponse response =
         GetFeatureTableResponse.newBuilder().setTable(retrieveTable.get().toProto()).build();
 
     return response;
   }
+
+  @Transactional
+  public void deleteFeatureTable(DeleteFeatureTableRequest request) {
+    String projectName = resolveProjectName(request.getProject());
+    String featureTableName = request.getName();
+
+    checkValidCharacters(projectName, "project");
+    checkValidCharacters(featureTableName, "featureTable");
+
+    Optional<FeatureTable> existingTable =
+        tableRepository.findFeatureTableByNameAndProject_Name(featureTableName, projectName);
+    if (existingTable.isEmpty()) {
+      throw new NoSuchElementException(
+          String.format(
+              "No such Feature Table: (project: %s, name: %s)", projectName, featureTableName));
+    }
+
+    existingTable.get().delete();
+  }
 }

core/src/main/resources/db/migration/V3.0__Feature_Table_Deletion.sql

@@ -0,0 +1 @@
+ALTER TABLE feature_tables ADD COLUMN is_deleted boolean NOT NULL;

core/src/test/java/feast/core/service/SpecServiceIT.java

@@ -1336,4 +1336,84 @@ public void shouldErrorOnArchivedProject() {
                       .build()));
     }
   }
+
+  @Nested
+  public class DeleteFeatureTable {
+
+    @Test
+    public void shouldReturnNoTables() {
+      String projectName = "default";
+      String featureTableName = "featuretable1";
+
+      apiClient.deleteFeatureTable(projectName, featureTableName);
+
+      CoreServiceProto.ListFeatureTablesRequest.Filter filter =
+          CoreServiceProto.ListFeatureTablesRequest.Filter.newBuilder()
+              .setProject("default")
+              .build();
+      List<FeatureTableProto.FeatureTable> featureTables =
+          apiClient.simpleListFeatureTables(filter);
+
+      StatusRuntimeException exc =
+          assertThrows(
+              StatusRuntimeException.class,
+              () -> apiClient.simpleGetFeatureTable(projectName, featureTableName));
+
+      assertThat(featureTables.size(), equalTo(0));
+      assertThat(
+          exc.getMessage(),
+          equalTo(
+              String.format(
+                  "NOT_FOUND: Feature Table has been deleted: (project: %s, name: %s)",
+                  projectName, featureTableName)));
+    }
+
+    @Test
+    public void shouldUpdateDeletedTable() {
+      String projectName = "default";
+      String featureTableName = "featuretable1";
+
+      apiClient.deleteFeatureTable(projectName, featureTableName);
+
+      FeatureTableSpec featureTableSpec =
+          DataGenerator.createFeatureTableSpec(
+                  featureTableName,
+                  Arrays.asList("entity1", "entity2"),
+                  new HashMap<>() {
+                    {
+                      put("feature3", ValueProto.ValueType.Enum.INT64);
+                    }
+                  },
+                  7200,
+                  ImmutableMap.of("feat_key3", "feat_value3"))
+              .toBuilder()
+              .setBatchSource(
+                  DataGenerator.createFileDataSourceSpec("file:///path/to/file", "ts_col", ""))
+              .build();
+
+      apiClient.applyFeatureTable(projectName, featureTableSpec);
+
+      FeatureTableProto.FeatureTable featureTable =
+          apiClient.simpleGetFeatureTable(projectName, featureTableName);
+
+      assertTrue(TestUtil.compareFeatureTableSpec(featureTable.getSpec(), featureTableSpec));
+    }
+
+    @Test
+    public void shouldErrorIfTableNotExist() {
+      String projectName = "default";
+      String featureTableName = "nonexistent_table";
+      StatusRuntimeException exc =
+          assertThrows(
+              StatusRuntimeException.class,
+              () -> apiClient.deleteFeatureTable(projectName, featureTableName));
+
+      assertThat(
+          exc.getMessage(),
+          equalTo(
+              String.format(
+                  "NOT_FOUND: No such Feature Table: (project: %s, name: %s)",
+                  projectName, featureTableName)));
+    }
+  }
 }

protos/feast/core/CoreService.proto

@@ -134,6 +134,9 @@ service CoreService {
     // Returns a specific feature table
     rpc GetFeatureTable (GetFeatureTableRequest) returns (GetFeatureTableResponse);
 
+    // Delete a specific feature table
+    rpc DeleteFeatureTable (DeleteFeatureTableRequest) returns (DeleteFeatureTableResponse);
+
 }
 
 service JobControllerService {
@@ -500,3 +503,14 @@ message ListFeatureTablesResponse {
     // List of matching Feature Tables
     repeated FeatureTable tables = 1;
 }
+
+message DeleteFeatureTableRequest {
+    // Optional. Name of the Project to delete the Feature Table from.
+    // If unspecified, will delete FeatureTable from the default project.
+    string project = 1;
+
+    // Name of the FeatureTable to delete.
+    string name = 2;
+}
+
+message DeleteFeatureTableResponse {}

protos/feast/core/FeatureTable.proto

@@ -76,4 +76,8 @@ message FeatureTableMeta {
 
     // Auto incrementing revision no. of this Feature Table
     int64 revision = 3;
+
+    // Hash entities, features, batch_source and stream_source to inform JobService if
+    // jobs should be restarted should hash change
+    string hash = 4;
 }

sdk/python/feast/client.py

@@ -51,6 +51,7 @@
     ArchiveProjectResponse,
     CreateProjectRequest,
     CreateProjectResponse,
+    DeleteFeatureTableRequest,
     GetEntityRequest,
     GetEntityResponse,
     GetFeastCoreVersionRequest,
@@ -679,6 +680,26 @@ def get_feature_table(self, name: str, project: str = None) -> FeatureTable:
             raise grpc.RpcError(e.details())
         return FeatureTable.from_proto(get_feature_table_response.table)
 
+    def delete_feature_table(self, name: str, project: str = None) -> None:
+        """
+        Deletes a feature table.
+
+        Args:
+            project: Feast project that this feature table belongs to
+            name: Name of feature table
+        """
+
+        if project is None:
+            project = self.project
+
+        try:
+            self._core_service.DeleteFeatureTable(
+                DeleteFeatureTableRequest(project=project, name=name.strip()),
+                metadata=self._get_grpc_metadata(),
+            )
+        except grpc.RpcError as e:
+            raise grpc.RpcError(e.details())
+
     def ingest(
         self,
         feature_table: Union[str, FeatureTable],