diff --git a/java/datatypes/java/src/main/proto/tensorflow_metadata b/java/datatypes/java/src/main/proto/tensorflow_metadata deleted file mode 120000 index 225345970c..0000000000 --- a/java/datatypes/java/src/main/proto/tensorflow_metadata +++ /dev/null @@ -1 +0,0 @@ -../../../../../../protos/tensorflow_metadata \ No newline at end of file diff --git a/java/infra/docker/feature-server/Dockerfile b/java/infra/docker/feature-server/Dockerfile index 1ce8c7b16c..5f32a7ba6f 100644 --- a/java/infra/docker/feature-server/Dockerfile +++ b/java/infra/docker/feature-server/Dockerfile @@ -25,7 +25,6 @@ RUN mvn dependency:go-offline -DexcludeGroupIds:dev.feast 2>/dev/null || true COPY java/ . COPY protos/feast datatypes/java/src/main/proto/feast -COPY protos/tensorflow_metadata datatypes/java/src/main/proto/tensorflow_metadata ARG VERSION=dev RUN mvn --also-make --projects serving -Drevision=$VERSION \ diff --git a/java/sdk/java/src/main/java/com/gojek/feast/FeastClient.java b/java/sdk/java/src/main/java/com/gojek/feast/FeastClient.java index 94836d88aa..0c0b279be6 100644 --- a/java/sdk/java/src/main/java/com/gojek/feast/FeastClient.java +++ b/java/sdk/java/src/main/java/com/gojek/feast/FeastClient.java @@ -64,7 +64,6 @@ public static FeastClient create(String host, int port) { /** * Create a authenticated client that can access Feast serving with authentication enabled. - * Supports the {@link CallCredentials} in the {@link feast.common.auth.credentials} package. * * @param host hostname or ip address of Feast serving GRPC server * @param port port number of Feast serving GRPC server diff --git a/java/serving/pom.xml b/java/serving/pom.xml index 8fdc090b37..449599ed36 100644 --- a/java/serving/pom.xml +++ b/java/serving/pom.xml @@ -84,19 +84,6 @@ ${project.version} - - dev.feast - feast-storage-connector-bigtable - ${project.version} - - - - dev.feast - feast-storage-connector-cassandra - ${project.version} - - - org.slf4j diff --git a/java/storage/connectors/bigtable/pom.xml b/java/storage/connectors/bigtable/pom.xml deleted file mode 100644 index 81cd450745..0000000000 --- a/java/storage/connectors/bigtable/pom.xml +++ /dev/null @@ -1,45 +0,0 @@ - - - - dev.feast - feast-storage-connectors - ${revision} - - - 4.0.0 - feast-storage-connector-bigtable - - - 11 - 11 - - - - - com.google.cloud - google-cloud-bigtable - 1.21.2 - - - - - org.apache.avro - avro - 1.10.2 - - - - dev.feast - feast-storage-connector-sstable - ${project.version} - - - - com.google.guava - guava - - - - \ No newline at end of file diff --git a/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableOnlineRetriever.java b/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableOnlineRetriever.java deleted file mode 100644 index 6e67782b2a..0000000000 --- a/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableOnlineRetriever.java +++ /dev/null @@ -1,211 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2021 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.storage.connectors.bigtable.retriever; - -import com.google.cloud.bigtable.data.v2.BigtableDataClient; -import com.google.cloud.bigtable.data.v2.models.Filters; -import com.google.cloud.bigtable.data.v2.models.Query; -import com.google.cloud.bigtable.data.v2.models.Row; -import com.google.cloud.bigtable.data.v2.models.RowCell; -import com.google.protobuf.ByteString; -import com.google.protobuf.Timestamp; -import feast.proto.serving.ServingAPIProto.FeatureReferenceV2; -import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow; -import feast.storage.api.retriever.AvroFeature; -import feast.storage.api.retriever.Feature; -import feast.storage.connectors.sstable.retriever.SSTableOnlineRetriever; -import java.io.IOException; -import java.util.*; -import java.util.function.Function; -import java.util.stream.Collectors; -import java.util.stream.StreamSupport; -import org.apache.avro.AvroRuntimeException; -import org.apache.avro.generic.GenericDatumReader; -import org.apache.avro.generic.GenericRecord; -import org.apache.avro.io.*; - -public class BigTableOnlineRetriever implements SSTableOnlineRetriever { - - private BigtableDataClient client; - private BigTableSchemaRegistry schemaRegistry; - - public BigTableOnlineRetriever(BigtableDataClient client) { - this.client = client; - this.schemaRegistry = new BigTableSchemaRegistry(client); - } - - /** - * Generate BigTable key in the form of entity values joined by #. - * - * @param entityRow Single EntityRow representation in feature retrieval call - * @param entityNames List of entities related to feature references in retrieval call - * @return BigTable key for retrieval - */ - @Override - public ByteString convertEntityValueToKey(EntityRow entityRow, List entityNames) { - return ByteString.copyFrom( - entityNames.stream() - .map(entity -> entityRow.getFieldsMap().get(entity)) - .map(this::valueToString) - .collect(Collectors.joining("#")) - .getBytes()); - } - - /** - * Converts rowCell feature value into @NativeFeature type. - * - * @param tableName Name of BigTable table - * @param rowKeys List of keys of rows to retrieve - * @param rows Map of rowKey to Row related to it - * @param featureReferences List of feature references - * @return List of List of Features associated with respective rowKey - */ - @Override - public List> convertRowToFeature( - String tableName, - List rowKeys, - Map rows, - List featureReferences) { - - BinaryDecoder reusedDecoder = DecoderFactory.get().binaryDecoder(new byte[0], null); - - return rowKeys.stream() - .map( - rowKey -> { - if (!rows.containsKey(rowKey)) { - return Collections.emptyList(); - } else { - Row row = rows.get(rowKey); - return featureReferences.stream() - .map(FeatureReferenceV2::getFeatureTable) - .distinct() - .map(cf -> row.getCells(cf, "")) - .filter(ls -> !ls.isEmpty()) - .flatMap( - rowCells -> { - RowCell rowCell = rowCells.get(0); // Latest cell - String family = rowCell.getFamily(); - ByteString value = rowCell.getValue(); - - List features; - List localFeatureReferences = - featureReferences.stream() - .filter( - featureReference -> - featureReference.getFeatureTable().equals(family)) - .collect(Collectors.toList()); - - try { - features = - decodeFeatures( - tableName, - value, - localFeatureReferences, - reusedDecoder, - rowCell.getTimestamp()); - } catch (IOException e) { - throw new RuntimeException("Failed to decode features from BigTable"); - } - - return features.stream(); - }) - .collect(Collectors.toList()); - } - }) - .collect(Collectors.toList()); - } - - /** - * Retrieve rows for each row entity key by generating BigTable rowQuery with filters based on - * column families. - * - * @param tableName Name of BigTable table - * @param rowKeys List of keys of rows to retrieve - * @param columnFamilies List of FeatureTable names - * @return Map of retrieved features for each rowKey - */ - @Override - public Map getFeaturesFromSSTable( - String tableName, List rowKeys, List columnFamilies) { - Query rowQuery = Query.create(tableName); - Filters.InterleaveFilter familyFilter = Filters.FILTERS.interleave(); - columnFamilies.forEach(cf -> familyFilter.filter(Filters.FILTERS.family().exactMatch(cf))); - - for (ByteString rowKey : rowKeys) { - rowQuery.rowKey(rowKey); - } - - return StreamSupport.stream(client.readRows(rowQuery).spliterator(), false) - .collect(Collectors.toMap(Row::getKey, Function.identity())); - } - - /** - * AvroRuntimeException is thrown if feature name does not exist in avro schema. Empty Object is - * returned when null is retrieved from BigTable RowCell. - * - * @param tableName Name of BigTable table - * @param value Value of BigTable cell where first 4 bytes represent the schema reference and - * remaining bytes represent avro-serialized features - * @param featureReferences List of feature references - * @param reusedDecoder Decoder for decoding feature values - * @param timestamp Timestamp of rowCell - * @return @NativeFeature with retrieved value stored in BigTable RowCell - * @throws IOException - */ - private List decodeFeatures( - String tableName, - ByteString value, - List featureReferences, - BinaryDecoder reusedDecoder, - long timestamp) - throws IOException { - ByteString schemaReferenceBytes = value.substring(0, 4); - byte[] featureValueBytes = value.substring(4).toByteArray(); - - BigTableSchemaRegistry.SchemaReference schemaReference = - new BigTableSchemaRegistry.SchemaReference(tableName, schemaReferenceBytes); - - GenericDatumReader reader = schemaRegistry.getReader(schemaReference); - - reusedDecoder = DecoderFactory.get().binaryDecoder(featureValueBytes, reusedDecoder); - GenericRecord record = reader.read(null, reusedDecoder); - - return featureReferences.stream() - .map( - featureReference -> { - Object featureValue; - try { - featureValue = record.get(featureReference.getName()); - } catch (AvroRuntimeException e) { - // Feature is not found in schema - return null; - } - if (featureValue != null) { - return new AvroFeature( - featureReference, - Timestamp.newBuilder().setSeconds(timestamp / 1000).build(), - featureValue); - } - return new AvroFeature( - featureReference, - Timestamp.newBuilder().setSeconds(timestamp / 1000).build(), - new Object()); - }) - .filter(Objects::nonNull) - .collect(Collectors.toList()); - } -} diff --git a/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableSchemaRegistry.java b/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableSchemaRegistry.java deleted file mode 100644 index 64989f0f8d..0000000000 --- a/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableSchemaRegistry.java +++ /dev/null @@ -1,107 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2021 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.storage.connectors.bigtable.retriever; - -import com.google.cloud.bigtable.data.v2.BigtableDataClient; -import com.google.cloud.bigtable.data.v2.models.Filters; -import com.google.cloud.bigtable.data.v2.models.Row; -import com.google.cloud.bigtable.data.v2.models.RowCell; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import com.google.common.collect.Iterables; -import com.google.protobuf.ByteString; -import java.util.concurrent.ExecutionException; -import org.apache.avro.Schema; -import org.apache.avro.generic.GenericDatumReader; -import org.apache.avro.generic.GenericRecord; - -public class BigTableSchemaRegistry { - private final BigtableDataClient client; - private final LoadingCache> cache; - - private static String COLUMN_FAMILY = "metadata"; - private static String QUALIFIER = "avro"; - private static String KEY_PREFIX = "schema#"; - - public static class SchemaReference { - private final String tableName; - private final ByteString schemaHash; - - public SchemaReference(String tableName, ByteString schemaHash) { - this.tableName = tableName; - this.schemaHash = schemaHash; - } - - public String getTableName() { - return tableName; - } - - public ByteString getSchemaHash() { - return schemaHash; - } - - @Override - public int hashCode() { - int result = tableName.hashCode(); - result = 31 * result + schemaHash.hashCode(); - return result; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - SchemaReference that = (SchemaReference) o; - - if (!tableName.equals(that.tableName)) return false; - return schemaHash.equals(that.schemaHash); - } - } - - public BigTableSchemaRegistry(BigtableDataClient client) { - this.client = client; - - CacheLoader> schemaCacheLoader = - CacheLoader.from(this::loadReader); - - cache = CacheBuilder.newBuilder().build(schemaCacheLoader); - } - - public GenericDatumReader getReader(SchemaReference reference) { - GenericDatumReader reader; - try { - reader = this.cache.get(reference); - } catch (ExecutionException | CacheLoader.InvalidCacheLoadException e) { - throw new RuntimeException(String.format("Unable to find Schema"), e); - } - return reader; - } - - private GenericDatumReader loadReader(SchemaReference reference) { - Row row = - client.readRow( - reference.getTableName(), - ByteString.copyFrom(KEY_PREFIX.getBytes()).concat(reference.getSchemaHash()), - Filters.FILTERS.family().exactMatch(COLUMN_FAMILY)); - RowCell last = Iterables.getLast(row.getCells(COLUMN_FAMILY, QUALIFIER)); - - Schema schema = new Schema.Parser().parse(last.getValue().toStringUtf8()); - return new GenericDatumReader<>(schema); - } -} diff --git a/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableStoreConfig.java b/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableStoreConfig.java deleted file mode 100644 index 11a0445006..0000000000 --- a/java/storage/connectors/bigtable/src/main/java/feast/storage/connectors/bigtable/retriever/BigTableStoreConfig.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2021 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.storage.connectors.bigtable.retriever; - -public class BigTableStoreConfig { - private final String projectId; - private final String instanceId; - private final String appProfileId; - - public BigTableStoreConfig(String projectId, String instanceId, String appProfileId) { - this.projectId = projectId; - this.instanceId = instanceId; - this.appProfileId = appProfileId; - } - - public String getProjectId() { - return this.projectId; - } - - public String getInstanceId() { - return this.instanceId; - } - - public String getAppProfileId() { - return this.appProfileId; - } -} diff --git a/java/storage/connectors/cassandra/pom.xml b/java/storage/connectors/cassandra/pom.xml deleted file mode 100644 index 32e4d73bca..0000000000 --- a/java/storage/connectors/cassandra/pom.xml +++ /dev/null @@ -1,45 +0,0 @@ - - - - feast-storage-connectors - dev.feast - ${revision} - - - 4.0.0 - feast-storage-connector-cassandra - - - 11 - 11 - - - - - org.apache.avro - avro - 1.10.2 - - - - dev.feast - feast-storage-connector-sstable - ${project.version} - - - - com.datastax.oss - java-driver-core - 4.11.0 - - - - com.datastax.oss - java-driver-query-builder - 4.11.0 - - - - \ No newline at end of file diff --git a/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraOnlineRetriever.java b/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraOnlineRetriever.java deleted file mode 100644 index 9b9de7bdb9..0000000000 --- a/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraOnlineRetriever.java +++ /dev/null @@ -1,271 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2021 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.storage.connectors.cassandra.retriever; - -import com.datastax.oss.driver.api.core.AsyncPagingIterable; -import com.datastax.oss.driver.api.core.CqlSession; -import com.datastax.oss.driver.api.core.cql.AsyncResultSet; -import com.datastax.oss.driver.api.core.cql.PreparedStatement; -import com.datastax.oss.driver.api.core.cql.Row; -import com.datastax.oss.driver.api.querybuilder.QueryBuilder; -import com.datastax.oss.driver.api.querybuilder.select.Select; -import com.google.protobuf.Timestamp; -import feast.proto.serving.ServingAPIProto.FeatureReferenceV2; -import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow; -import feast.storage.api.retriever.AvroFeature; -import feast.storage.api.retriever.Feature; -import feast.storage.connectors.sstable.retriever.SSTableOnlineRetriever; -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.*; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.ExecutionException; -import java.util.function.Function; -import java.util.stream.Collectors; -import org.apache.avro.AvroRuntimeException; -import org.apache.avro.generic.GenericDatumReader; -import org.apache.avro.generic.GenericRecord; -import org.apache.avro.io.BinaryDecoder; -import org.apache.avro.io.DecoderFactory; - -public class CassandraOnlineRetriever implements SSTableOnlineRetriever { - - private final CqlSession session; - private final CassandraSchemaRegistry schemaRegistry; - - private static final String ENTITY_KEY = "key"; - private static final String SCHEMA_REF_SUFFIX = "__schema_ref"; - private static final String EVENT_TIMESTAMP_SUFFIX = "__event_timestamp"; - private static final int MAX_TABLE_NAME_LENGTH = 48; - - public CassandraOnlineRetriever(CqlSession session) { - this.session = session; - this.schemaRegistry = new CassandraSchemaRegistry(session); - } - - /** - * Generate Cassandra key in the form of entity values joined by #. - * - * @param entityRow Single EntityRow representation in feature retrieval call - * @param entityNames List of entities related to feature references in retrieval call - * @return Cassandra key for retrieval - */ - @Override - public ByteBuffer convertEntityValueToKey(EntityRow entityRow, List entityNames) { - return ByteBuffer.wrap( - entityNames.stream() - .map(entity -> entityRow.getFieldsMap().get(entity)) - .map(this::valueToString) - .collect(Collectors.joining("#")) - .getBytes()); - } - - /** - * Generate Cassandra table name, with limit of 48 characters. - * - * @param project Name of Feast project - * @param entityNames List of entities used in retrieval call - * @return Cassandra table name for retrieval - */ - @Override - public String getSSTable(String project, List entityNames) { - String tableName = String.format("%s__%s", project, String.join("__", entityNames)); - return trimAndHash(tableName, MAX_TABLE_NAME_LENGTH); - } - - /** - * Converts Cassandra rows into @NativeFeature type. - * - * @param tableName Name of Cassandra table - * @param rowKeys List of keys of rows to retrieve - * @param rows Map of rowKey to Row related to it - * @param featureReferences List of feature references - * @return List of List of Features associated with respective rowKey - */ - @Override - public List> convertRowToFeature( - String tableName, - List rowKeys, - Map rows, - List featureReferences) { - - BinaryDecoder reusedDecoder = DecoderFactory.get().binaryDecoder(new byte[0], null); - - return rowKeys.stream() - .map( - rowKey -> { - if (!rows.containsKey(rowKey)) { - return Collections.emptyList(); - } else { - Row row = rows.get(rowKey); - return featureReferences.stream() - .map(FeatureReferenceV2::getFeatureTable) - .distinct() - .flatMap( - featureTableColumn -> { - ByteBuffer featureValues = row.getByteBuffer(featureTableColumn); - ByteBuffer schemaRefKey = - row.getByteBuffer(featureTableColumn + SCHEMA_REF_SUFFIX); - - // Prevent retrieval of features from incorrect FeatureTable - List localFeatureReferences = - featureReferences.stream() - .filter( - featureReference -> - featureReference - .getFeatureTable() - .equals(featureTableColumn)) - .collect(Collectors.toList()); - - List features; - try { - features = - decodeFeatures( - schemaRefKey, - featureValues, - localFeatureReferences, - reusedDecoder, - row.getLong(featureTableColumn + EVENT_TIMESTAMP_SUFFIX)); - } catch (IOException e) { - throw new RuntimeException("Failed to decode features from Cassandra"); - } - - return features.stream(); - }) - .collect(Collectors.toList()); - } - }) - .collect(Collectors.toList()); - } - - /** - * Retrieve rows for each row entity key by generating Cassandra Query with filters based on - * columns. - * - * @param tableName Name of Cassandra table - * @param rowKeys List of keys of rows to retrieve - * @param columnFamilies List of FeatureTable names - * @return Map of retrieved features for each rowKey - */ - @Override - public Map getFeaturesFromSSTable( - String tableName, List rowKeys, List columnFamilies) { - List schemaRefColumns = - columnFamilies.stream().map(c -> c + SCHEMA_REF_SUFFIX).collect(Collectors.toList()); - Select query = - QueryBuilder.selectFrom(tableName) - .columns(columnFamilies) - .columns(schemaRefColumns) - .column(ENTITY_KEY); - for (String columnFamily : columnFamilies) { - query = query.writeTime(columnFamily).as(columnFamily + EVENT_TIMESTAMP_SUFFIX); - } - query = query.whereColumn(ENTITY_KEY).isEqualTo(QueryBuilder.bindMarker()); - - PreparedStatement preparedStatement = session.prepare(query.build()); - - List> completableAsyncResultSets = - rowKeys.stream() - .map(preparedStatement::bind) - .map(session::executeAsync) - .map(CompletionStage::toCompletableFuture) - .collect(Collectors.toList()); - - CompletableFuture allResultComputed = - CompletableFuture.allOf(completableAsyncResultSets.toArray(new CompletableFuture[0])); - - Map resultMap; - try { - resultMap = - allResultComputed - .thenApply( - v -> - completableAsyncResultSets.stream() - .map(CompletableFuture::join) - .filter(result -> result.remaining() != 0) - .map(AsyncPagingIterable::one) - .filter(Objects::nonNull) - .collect( - Collectors.toMap( - (Row row) -> row.getByteBuffer(ENTITY_KEY), Function.identity()))) - .get(); - } catch (InterruptedException | ExecutionException e) { - throw new RuntimeException(e.getMessage()); - } - - return resultMap; - } - - /** - * AvroRuntimeException is thrown if feature name does not exist in avro schema. - * - * @param schemaRefKey Schema reference key - * @param value Value of Cassandra cell where bytes represent avro-serialized features - * @param featureReferences List of feature references - * @param reusedDecoder Decoder for decoding feature values - * @param timestamp Timestamp of rowCell - * @return @NativeFeature with retrieved value stored in Cassandra cell - * @throws IOException - */ - private List decodeFeatures( - ByteBuffer schemaRefKey, - ByteBuffer value, - List featureReferences, - BinaryDecoder reusedDecoder, - long timestamp) - throws IOException { - - if (value == null || schemaRefKey == null) { - return Collections.emptyList(); - } - - CassandraSchemaRegistry.SchemaReference schemaReference = - new CassandraSchemaRegistry.SchemaReference(schemaRefKey); - - // Convert ByteBuffer to ByteArray - byte[] bytesArray = new byte[value.remaining()]; - value.get(bytesArray, 0, bytesArray.length); - GenericDatumReader reader = schemaRegistry.getReader(schemaReference); - reusedDecoder = DecoderFactory.get().binaryDecoder(bytesArray, reusedDecoder); - GenericRecord record = reader.read(null, reusedDecoder); - - return featureReferences.stream() - .map( - featureReference -> { - Object featureValue; - try { - featureValue = record.get(featureReference.getName()); - } catch (AvroRuntimeException e) { - // Feature is not found in schema - return null; - } - if (featureValue != null) { - return new AvroFeature( - featureReference, - Timestamp.newBuilder().setSeconds(timestamp / 1000).build(), - featureValue); - } - return new AvroFeature( - featureReference, - Timestamp.newBuilder().setSeconds(timestamp / 1000).build(), - new Object()); - }) - .filter(Objects::nonNull) - .collect(Collectors.toList()); - } -} diff --git a/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraSchemaRegistry.java b/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraSchemaRegistry.java deleted file mode 100644 index 7915b370a4..0000000000 --- a/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraSchemaRegistry.java +++ /dev/null @@ -1,106 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2021 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.storage.connectors.cassandra.retriever; - -import com.datastax.oss.driver.api.core.CqlSession; -import com.datastax.oss.driver.api.core.cql.BoundStatement; -import com.datastax.oss.driver.api.core.cql.PreparedStatement; -import com.datastax.oss.driver.api.core.cql.Row; -import com.datastax.oss.driver.api.querybuilder.QueryBuilder; -import com.datastax.oss.driver.api.querybuilder.select.Select; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; -import java.util.Objects; -import java.util.concurrent.ExecutionException; -import org.apache.avro.Schema; -import org.apache.avro.generic.GenericDatumReader; -import org.apache.avro.generic.GenericRecord; - -public class CassandraSchemaRegistry { - private final CqlSession session; - private final PreparedStatement preparedStatement; - private final LoadingCache> cache; - - private static String SCHEMA_REF_TABLE = "feast_schema_reference"; - private static String SCHEMA_REF_COLUMN = "schema_ref"; - private static String SCHEMA_COLUMN = "avro_schema"; - - public static class SchemaReference { - private final ByteBuffer schemaHash; - - public SchemaReference(ByteBuffer schemaHash) { - this.schemaHash = schemaHash; - } - - public ByteBuffer getSchemaHash() { - return schemaHash; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - SchemaReference that = (SchemaReference) o; - return Objects.equals(schemaHash, that.schemaHash); - } - - @Override - public int hashCode() { - return Objects.hash(schemaHash); - } - } - - public CassandraSchemaRegistry(CqlSession session) { - this.session = session; - String tableName = String.format("\"%s\"", SCHEMA_REF_TABLE); - Select query = - QueryBuilder.selectFrom(tableName) - .column(SCHEMA_COLUMN) - .whereColumn(SCHEMA_REF_COLUMN) - .isEqualTo(QueryBuilder.bindMarker()); - this.preparedStatement = session.prepare(query.build()); - - CacheLoader> schemaCacheLoader = - CacheLoader.from(this::loadReader); - - cache = CacheBuilder.newBuilder().build(schemaCacheLoader); - } - - public GenericDatumReader getReader(SchemaReference reference) { - GenericDatumReader reader; - try { - reader = this.cache.get(reference); - } catch (ExecutionException | CacheLoader.InvalidCacheLoadException e) { - throw new RuntimeException("Unable to find Schema"); - } - return reader; - } - - private GenericDatumReader loadReader(SchemaReference reference) { - BoundStatement statement = preparedStatement.bind(reference.getSchemaHash()); - - Row row = session.execute(statement).one(); - - Schema schema = - new Schema.Parser() - .parse(StandardCharsets.UTF_8.decode(row.getByteBuffer(SCHEMA_COLUMN)).toString()); - return new GenericDatumReader<>(schema); - } -} diff --git a/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraStoreConfig.java b/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraStoreConfig.java deleted file mode 100644 index 3ee25dfe8a..0000000000 --- a/java/storage/connectors/cassandra/src/main/java/feast/storage/connectors/cassandra/retriever/CassandraStoreConfig.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2021 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.storage.connectors.cassandra.retriever; - -public class CassandraStoreConfig { - - private final String connectionString; - private final String dataCenter; - private final String keySpace; - - public CassandraStoreConfig(String connectionString, String dataCenter, String keySpace) { - this.connectionString = connectionString; - this.dataCenter = dataCenter; - this.keySpace = keySpace; - } - - public String getConnectionString() { - return this.connectionString; - } - - public String getDataCenter() { - return this.dataCenter; - } - - public String getKeySpace() { - return this.keySpace; - } -} diff --git a/java/storage/connectors/pom.xml b/java/storage/connectors/pom.xml index 1c4b75afa9..4969364b62 100644 --- a/java/storage/connectors/pom.xml +++ b/java/storage/connectors/pom.xml @@ -16,9 +16,6 @@ redis - bigtable - cassandra - sstable diff --git a/java/storage/connectors/sstable/pom.xml b/java/storage/connectors/sstable/pom.xml deleted file mode 100644 index 8c7a271aac..0000000000 --- a/java/storage/connectors/sstable/pom.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - feast-storage-connectors - dev.feast - ${revision} - - - 4.0.0 - feast-storage-connector-sstable - - - 11 - 11 - - - \ No newline at end of file diff --git a/java/storage/connectors/sstable/src/main/java/feast/storage/connectors/sstable/retriever/SSTableOnlineRetriever.java b/java/storage/connectors/sstable/src/main/java/feast/storage/connectors/sstable/retriever/SSTableOnlineRetriever.java deleted file mode 100644 index c86923a2ec..0000000000 --- a/java/storage/connectors/sstable/src/main/java/feast/storage/connectors/sstable/retriever/SSTableOnlineRetriever.java +++ /dev/null @@ -1,163 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2021 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.storage.connectors.sstable.retriever; - -import com.google.common.hash.Hashing; -import feast.proto.serving.ServingAPIProto.FeatureReferenceV2; -import feast.proto.serving.ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow; -import feast.proto.types.ValueProto; -import feast.storage.api.retriever.Feature; -import feast.storage.api.retriever.OnlineRetrieverV2; -import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; - -/** - * @param Decoded value type of the partition key - * @param Type of the SSTable row - */ -public interface SSTableOnlineRetriever extends OnlineRetrieverV2 { - - int MAX_TABLE_NAME_LENGTH = 50; - - @Override - default List> getOnlineFeatures( - String project, - List entityRows, - List featureReferences, - List entityNames) { - - List columnFamilies = getSSTableColumns(featureReferences); - String tableName = getSSTable(project, entityNames); - - List rowKeys = - entityRows.stream() - .map(row -> convertEntityValueToKey(row, entityNames)) - .collect(Collectors.toList()); - - Map rowsFromSSTable = getFeaturesFromSSTable(tableName, rowKeys, columnFamilies); - - return convertRowToFeature(tableName, rowKeys, rowsFromSSTable, featureReferences); - } - - /** - * Generate SSTable key. - * - * @param entityRow Single EntityRow representation in feature retrieval call - * @param entityNames List of entities related to feature references in retrieval call - * @return SSTable key for retrieval - */ - K convertEntityValueToKey(EntityRow entityRow, List entityNames); - - /** - * Converts SSTable rows into @NativeFeature type. - * - * @param tableName Name of SSTable - * @param rowKeys List of keys of rows to retrieve - * @param rows Map of rowKey to Row related to it - * @param featureReferences List of feature references - * @return List of List of Features associated with respective rowKey - */ - List> convertRowToFeature( - String tableName, - List rowKeys, - Map rows, - List featureReferences); - - /** - * Retrieve rows for each row entity key. - * - * @param tableName Name of SSTable - * @param rowKeys List of keys of rows to retrieve - * @param columnFamilies List of column names - * @return Map of retrieved features for each rowKey - */ - Map getFeaturesFromSSTable(String tableName, List rowKeys, List columnFamilies); - - /** - * Retrieve name of SSTable corresponding to entities in retrieval call - * - * @param project Name of Feast project - * @param entityNames List of entities used in retrieval call - * @return Name of Cassandra table - */ - default String getSSTable(String project, List entityNames) { - return trimAndHash( - String.format("%s__%s", project, String.join("__", entityNames)), MAX_TABLE_NAME_LENGTH); - } - - /** - * Convert Entity value from Feast valueType to String type. Currently only supports STRING_VAL, - * INT64_VAL, INT32_VAL and BYTES_VAL. - * - * @param v Entity value of Feast valueType - * @return String representation of Entity value - */ - default String valueToString(ValueProto.Value v) { - String stringRepr; - switch (v.getValCase()) { - case STRING_VAL: - stringRepr = v.getStringVal(); - break; - case INT64_VAL: - stringRepr = String.valueOf(v.getInt64Val()); - break; - case INT32_VAL: - stringRepr = String.valueOf(v.getInt32Val()); - break; - case BYTES_VAL: - stringRepr = v.getBytesVal().toString(); - break; - default: - throw new RuntimeException("Type is not supported to be entity"); - } - - return stringRepr; - } - - /** - * Retrieve SSTable columns based on Feature references. - * - * @param featureReferences List of feature references in retrieval call - * @return List of String of column names - */ - default List getSSTableColumns(List featureReferences) { - return featureReferences.stream() - .map(FeatureReferenceV2::getFeatureTable) - .distinct() - .collect(Collectors.toList()); - } - - /** - * Trims long SSTable table names and appends hash suffix for uniqueness. - * - * @param expr Original SSTable table name - * @param maxLength Maximum length allowed for SSTable - * @return Hashed suffix SSTable table name - */ - default String trimAndHash(String expr, int maxLength) { - // Length 8 as derived from murmurhash_32 implementation - int maxPrefixLength = maxLength - 8; - String finalName = expr; - if (expr.length() > maxLength) { - String hashSuffix = - Hashing.murmur3_32().hashBytes(expr.substring(maxPrefixLength).getBytes()).toString(); - finalName = expr.substring(0, Math.min(expr.length(), maxPrefixLength)).concat(hashSuffix); - } - return finalName; - } -} diff --git a/protos/feast/core/CoreService.proto b/protos/feast/core/CoreService.proto deleted file mode 100644 index e9b7b4c43b..0000000000 --- a/protos/feast/core/CoreService.proto +++ /dev/null @@ -1,306 +0,0 @@ -// -// Copyright 2018 The Feast Authors -// -// Licensed 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 -// -// https://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. -// - -syntax = "proto3"; -package feast.core; - -option go_package = "github.com/feast-dev/feast/sdk/go/protos/feast/core"; -option java_outer_classname = "CoreServiceProto"; -option java_package = "feast.proto.core"; - -import "google/protobuf/timestamp.proto"; -import "tensorflow_metadata/proto/v0/statistics.proto"; -import "feast/core/Entity.proto"; -import "feast/core/Feature.proto"; -import "feast/core/FeatureTable.proto"; -import "feast/core/Store.proto"; - -service CoreService { - // Retrieve version information about this Feast deployment - rpc GetFeastCoreVersion (GetFeastCoreVersionRequest) returns (GetFeastCoreVersionResponse); - - // Returns a specific entity - rpc GetEntity (GetEntityRequest) returns (GetEntityResponse); - - // Returns all feature references and respective features matching that filter. If none are found - // an empty map will be returned - // If no filter is provided in the request, the response will contain all the features - // currently stored in the default project. - rpc ListFeatures (ListFeaturesRequest) returns (ListFeaturesResponse); - - // Retrieve store details given a filter. - // - // Returns all stores matching that filter. If none are found, an empty list will be returned. - // If no filter is provided in the request, the response will contain all the stores currently - // stored in the registry. - rpc ListStores (ListStoresRequest) returns (ListStoresResponse); - - // Create or update and existing entity. - // - // This function is idempotent - it will not create a new entity if schema does not change. - // Schema changes will update the entity if the changes are valid. - // Following changes are not valid: - // - Changes to name - // - Changes to type - rpc ApplyEntity (ApplyEntityRequest) returns (ApplyEntityResponse); - - // Returns all entity references and respective entities matching that filter. If none are found - // an empty map will be returned - // If no filter is provided in the request, the response will contain all the entities - // currently stored in the default project. - rpc ListEntities (ListEntitiesRequest) returns (ListEntitiesResponse); - - // Updates core with the configuration of the store. - // - // If the changes are valid, core will return the given store configuration in response, and - // start or update the necessary feature population jobs for the updated store. - rpc UpdateStore (UpdateStoreRequest) returns (UpdateStoreResponse); - - // Creates a project. Projects serve as namespaces within which resources like features will be - // created. Feature table names as must be unique within a project while field (Feature/Entity) names - // must be unique within a Feature Table. Project names themselves must be globally unique. - rpc CreateProject (CreateProjectRequest) returns (CreateProjectResponse); - - // Archives a project. Archived projects will continue to exist and function, but won't be visible - // through the Core API. Any existing ingestion or serving requests will continue to function, - // but will result in warning messages being logged. It is not possible to unarchive a project - // through the Core API - rpc ArchiveProject (ArchiveProjectRequest) returns (ArchiveProjectResponse); - - // Lists all projects active projects. - rpc ListProjects (ListProjectsRequest) returns (ListProjectsResponse); - - /* Feature Tables */ - // Create or update an existing feature table. - // This function is idempotent - it will not create a new feature table if the schema does not change. - // Schema changes will update the feature table if the changes are valid. - // All changes except the following are valid: - // - Changes to feature table name. - // - Changes to entities - // - Changes to feature name and type - rpc ApplyFeatureTable (ApplyFeatureTableRequest) returns (ApplyFeatureTableResponse); - - // List feature tables that match a given filter. - // Returns the references of the Feature Tables matching that filter. If none are found, - // an empty list will be returned. - // If no filter is provided in the request, the response will match all the feature - // tables currently stored in the registry. - rpc ListFeatureTables (ListFeatureTablesRequest) returns (ListFeatureTablesResponse); - - // Returns a specific feature table - rpc GetFeatureTable (GetFeatureTableRequest) returns (GetFeatureTableResponse); - - // Delete a specific feature table - rpc DeleteFeatureTable (DeleteFeatureTableRequest) returns (DeleteFeatureTableResponse); - -} - -// Request for a single entity -message GetEntityRequest { - // Name of entity (required). - string name = 1; - - // Name of project the entity belongs to. If omitted will default to 'default' project. - string project = 2; -} - -// Response containing a single entity -message GetEntityResponse { - feast.core.Entity entity = 1; -} - -// Retrieves details for all versions of a specific entity -message ListEntitiesRequest { - Filter filter = 1; - - message Filter { - // Optional. Specifies the name of the project to list Entities in. - // It is NOT possible to provide an asterisk with a string in order to do pattern matching. - // If unspecified, this field will default to the default project 'default'. - string project = 3; - - // Optional. User defined metadata for entity. - // Entities with all matching labels will be returned. - map labels = 4; - } -} - -message ListEntitiesResponse { - repeated feast.core.Entity entities = 1; -} - -message ListFeaturesRequest { - message Filter { - // User defined metadata for feature. - // Features with all matching labels will be returned. - map labels = 1; - - // List of entities contained within the featureSet that the feature belongs to. - // Only feature tables with these entities will be searched for features. - repeated string entities = 2; - - // Name of project that the feature tables belongs to. Filtering on projects is disabled. - // It is NOT possible to provide an asterisk with a string in order to do pattern matching. - // If unspecified this field will default to the default project 'default'. - string project = 3; - } - - Filter filter = 1; -} - -message ListFeaturesResponse { - reserved 1; - - map features = 2; -} - -message ListStoresRequest { - message Filter { - // Name of desired store. Regex is not supported in this query. - string name = 1; - } - - Filter filter = 1; -} - -message ListStoresResponse { - repeated feast.core.Store store = 1; -} - -message ApplyEntityRequest { - // If project is unspecified, will default to 'default' project. - // If project specified does not exist, the project would be automatically created. - feast.core.EntitySpecV2 spec = 1; - - // Name of project that this entity belongs to. - string project = 2; -} - -message ApplyEntityResponse { - feast.core.Entity entity = 1; -} - -message GetFeastCoreVersionRequest { -} - -message GetFeastCoreVersionResponse { - string version = 1; -} - -message UpdateStoreRequest { - feast.core.Store store = 1; -} - -message UpdateStoreResponse { - enum Status { - // Existing store config matching the given store id is identical to the given store config. - NO_CHANGE = 0; - - // New store created or existing config updated. - UPDATED = 1; - } - feast.core.Store store = 1; - Status status = 2; -} - -// Request to create a project -message CreateProjectRequest { - // Name of project (required) - string name = 1; -} - -// Response for creation of a project -message CreateProjectResponse { -} - -// Request for the archival of a project -message ArchiveProjectRequest { - // Name of project to be archived - string name = 1; -} - -// Response for archival of a project -message ArchiveProjectResponse { -} - -// Request for listing of projects -message ListProjectsRequest { -} - -// Response for listing of projects -message ListProjectsResponse { - // List of project names (archived projects are filtered out) - repeated string projects = 1; -} - -message UpdateFeatureSetStatusResponse {} - -message ApplyFeatureTableRequest { - // Optional. Name of the Project to apply the Feature Table to. - // If unspecified, will apply FeatureTable to the default project. - string project = 1; - // Feature Table specification to apply - FeatureTableSpec table_spec = 2; -} - -message ApplyFeatureTableResponse { - FeatureTable table = 1; -} - -message GetFeatureTableRequest { - // Optional. Name of the Project to retrieve the Feature Table from. - // If unspecified, will apply FeatureTable to the default project. - string project = 1; - - // Name of the FeatureTable to retrieve. - string name = 2; -} - -message GetFeatureTableResponse { - // The Feature Table retrieved. - FeatureTable table = 1; -} - -message ListFeatureTablesRequest { - message Filter { - // Optional. Specifies the name of the project to list Feature Tables in. - // If unspecified would list Feature Tables in the default project. - string project = 1; - - // Optional. Feature Tables with all matching labels will be returned. - // If unspecified would list Feature Tables without filtering by labels. - map labels = 3; - } - - // Filter used when listing Feature Tables - Filter filter = 1; -} - -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 {} diff --git a/protos/feast/serving/ServingService.proto b/protos/feast/serving/ServingService.proto index 577b4cf9ce..e37ecbbdde 100644 --- a/protos/feast/serving/ServingService.proto +++ b/protos/feast/serving/ServingService.proto @@ -20,7 +20,6 @@ package feast.serving; import "google/protobuf/timestamp.proto"; import "feast/types/Value.proto"; -import "tensorflow_metadata/proto/v0/statistics.proto"; option java_package = "feast.proto.serving"; option java_outer_classname = "ServingAPIProto";