[DAT-3394] Add support for maxcompute as source type (batch)

caraml-store-protobuf/src/main/proto/feast/core/DataSource.proto

@@ -16,6 +16,7 @@ message DataSource {
     BATCH_FILE = 1;
     BATCH_BIGQUERY = 2;
     STREAM_KAFKA = 3;
+    BATCH_MAXCOMPUTE = 4;
   }
   SourceType type = 1;
 
@@ -56,6 +57,11 @@ message DataSource {
     SparkOverride spark_override = 2;
   }
 
+  // Defines options for DataSource that sources features from a MaxCompute Query
+  message MaxComputeOptions {
+    string table_ref = 1;
+  }
+
   // Defines options for DataSource that sources features from Kafka messages.
   // Each message should be a Protobuf that can be decoded with the generated
   // Java Protobuf class at the given class path
@@ -78,5 +84,6 @@ message DataSource {
     FileOptions file_options = 11;
     BigQueryOptions bigquery_options = 12;
     KafkaOptions kafka_options = 13;
+    MaxComputeOptions maxcompute_options = 14;
   }
 }

caraml-store-registry/src/main/java/dev/caraml/store/feature/DataSource.java

@@ -10,6 +10,7 @@
 import dev.caraml.store.protobuf.core.DataSourceProto.DataSource.BigQueryOptions;
 import dev.caraml.store.protobuf.core.DataSourceProto.DataSource.FileOptions;
 import dev.caraml.store.protobuf.core.DataSourceProto.DataSource.KafkaOptions;
+import dev.caraml.store.protobuf.core.DataSourceProto.DataSource.MaxComputeOptions;
 import dev.caraml.store.protobuf.core.DataSourceProto.DataSource.SourceType;
 import dev.caraml.store.protobuf.core.SparkOverrideProto.SparkOverride;
 import java.util.HashMap;
@@ -86,6 +87,9 @@ public static DataSource fromProto(DataSourceProto.DataSource spec) {
         populateDatasourceConfigMapWithSparkOverride(
             dataSourceConfigMap, spec.getBigqueryOptions().getSparkOverride());
       }
+      case BATCH_MAXCOMPUTE -> {
+        dataSourceConfigMap.put("table_ref", spec.getMaxcomputeOptions().getTableRef());
+      }
       case STREAM_KAFKA -> {
         dataSourceConfigMap.put("bootstrap_servers", spec.getKafkaOptions().getBootstrapServers());
         dataSourceConfigMap.put(
@@ -167,6 +171,11 @@ public DataSourceProto.DataSource toProto() {
             parseDatasourceConfigMapToSparkOverride(dataSourceConfigMap));
         spec.setBigqueryOptions(bigQueryOptions.build());
       }
+      case BATCH_MAXCOMPUTE -> {
+        MaxComputeOptions.Builder maxComputeOptions = MaxComputeOptions.newBuilder();
+        maxComputeOptions.setTableRef(dataSourceConfigMap.get("table_ref"));
+        spec.setMaxcomputeOptions(maxComputeOptions.build());
+      }
       case STREAM_KAFKA -> {
         KafkaOptions.Builder kafkaOptions = KafkaOptions.newBuilder();
         kafkaOptions.setBootstrapServers(dataSourceConfigMap.get("bootstrap_servers"));

caraml-store-registry/src/main/java/dev/caraml/store/feature/DataSourceValidator.java

@@ -17,6 +17,7 @@ public static void validate(DataSource spec) {
       }
       case BATCH_BIGQUERY -> Matchers.checkValidBigQueryTableRef(
           spec.getBigqueryOptions().getTableRef(), "FeatureTable");
+      case BATCH_MAXCOMPUTE -> {} // validation for table_ref to be added
       case STREAM_KAFKA -> {
         StreamFormat.FormatCase messageFormat =
             spec.getKafkaOptions().getMessageFormat().getFormatCase();