update get_snowflake_path

feathr_project/feathr/client.py

@@ -236,9 +236,9 @@ def get_snowflake_path(self, database: str, schema: str, dbtable: str = None, qu
         if dbtable is None and query is None:
             raise RuntimeError("One of dbtable or query must be specified..")
         if dbtable:
-            return f"snowflake://snowflake_account/?sfDatabase={database}&fSchema={schema}&dbtable={dbtable}"
+            return f"snowflake://snowflake_account/?sfDatabase={database}&sfSchema={schema}&dbtable={dbtable}"
         else:
-            return f"snowflake://snowflake_account/?sfDatabase={database}&fSchema={schema}&query={query}"
+            return f"snowflake://snowflake_account/?sfDatabase={database}&sfSchema={schema}&query={query}"
 
     def list_registered_features(self, project_name: str = None) -> List[str]:
         """List all the already registered features under the given project.

feathr_project/feathr/registry/_feathr_registry_client_hack.py → feathr_project/feathr/registry/_feathr_registry_client_aws.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Dict
 import json
 import logging
 import requests
@@ -12,7 +12,7 @@ def check(r):
     return r
 
 
-class _FeatureRegistryHack(_FeatureRegistry):
+class _FeatureRegistryAWS(_FeatureRegistry):
     def __init__(self, project_name: str, endpoint: str, project_tags: Dict[str, str] = None, credential=None,
                  config_path=None):
         self.project_name = project_name

feathr_project/feathr/registry/feature_registry.py

@@ -55,13 +55,13 @@ def save_to_feature_config_from_context(self, anchor_list, derived_feature_list,
 def default_registry_client(project_name: str, config_path:str = "./feathr_config.yaml", project_registry_tag: Dict[str, str]=None, credential = None) -> FeathrRegistry:
     from feathr.registry._feathr_registry_client import _FeatureRegistry
     from feathr.registry._feature_registry_purview import _PurviewRegistry
-    from feathr.registry._feathr_registry_client_hack import _FeatureRegistryHack
+    from feathr.registry._feathr_registry_client_aws import _FeatureRegistryAWS
     from aws_requests_auth.aws_auth import AWSRequestsAuth
 
     envutils = _EnvVaraibleUtil(config_path)
     registry_endpoint = envutils.get_environment_variable_with_default("feature_registry", "api_endpoint")
     if registry_endpoint and isinstance(credential, AWSRequestsAuth):
-        return _FeatureRegistryHack(project_name, endpoint=registry_endpoint, project_tags=project_registry_tag, credential=credential)
+        return _FeatureRegistryAWS(project_name, endpoint=registry_endpoint, project_tags=project_registry_tag, credential=credential)
     elif registry_endpoint:
         return _FeatureRegistry(project_name, endpoint=registry_endpoint, project_tags=project_registry_tag, credential=credential)
     else:

src/main/scala/com/linkedin/feathr/offline/config/location/DataLocation.scala

@@ -25,6 +25,7 @@ import scala.collection.JavaConverters._
     new JsonSubTypes.Type(value = classOf[PathList], name = "pathlist"),
     new JsonSubTypes.Type(value = classOf[Jdbc], name = "jdbc"),
     new JsonSubTypes.Type(value = classOf[GenericLocation], name = "generic"),
+    new JsonSubTypes.Type(value = classOf[Snowflake], name = "snowflake"),
   ))
 trait DataLocation {
   /**

src/main/scala/com/linkedin/feathr/offline/source/dataloader/jdbc/JdbcConnectorChooser.scala

@@ -15,21 +15,18 @@ sealed trait JdbcConnectorChooser
 object JdbcConnectorChooser {
   case object SqlServer extends JdbcConnectorChooser
   case object Postgres extends JdbcConnectorChooser
-  case object SnowflakeSql extends JdbcConnectorChooser
   case object DefaultJDBC extends JdbcConnectorChooser
 
   def getType (url: String): JdbcConnectorChooser = url match {
     case url if url.startsWith("jdbc:sqlserver") => SqlServer
     case url if url.startsWith("jdbc:postgresql:") => Postgres
-    case url if url.startsWith("jdbc:snowflake:") => SnowflakeSql
     case _ => DefaultJDBC
   }
 
   def getJdbcConnector(ss: SparkSession, url: String): JdbcConnector = {
     val sqlDbType = getType(url)
     val dataLoader = sqlDbType match {
       case SqlServer => new SqlServerDataLoader(ss)
-      case SnowflakeSql => new SnowflakeSqlDataLoader(ss)
       case _ => new SqlServerDataLoader(ss) //default jdbc data loader place holder
     }
     dataLoader

src/main/scala/com/linkedin/feathr/offline/util/SourceUtils.scala

@@ -15,7 +15,7 @@ import com.linkedin.feathr.offline.source.SourceFormatType
 import com.linkedin.feathr.offline.source.SourceFormatType.SourceFormatType
 import com.linkedin.feathr.offline.source.dataloader.DataLoaderHandler
 import com.linkedin.feathr.offline.source.dataloader.hdfs.FileFormat
-import com.linkedin.feathr.offline.source.dataloader.jdbc.JdbcUtils
+import com.linkedin.feathr.offline.source.dataloader.jdbc.{JdbcUtils, SnowflakeUtils}
 import com.linkedin.feathr.offline.source.pathutil.{PathChecker, TimeBasedHdfsPathAnalyzer, TimeBasedHdfsPathGenerator}
 import com.linkedin.feathr.offline.util.AclCheckUtils.getLatestPath
 import com.linkedin.feathr.offline.util.datetime.OfflineDateTimeUtils
@@ -648,6 +648,9 @@ private[offline] object SourceUtils {
       case FileFormat.JDBC => {
         JdbcUtils.loadDataFrame(ss, inputData.inputPath)
       }
+      case FileFormat.SNOWFLAKE => {
+        SnowflakeUtils.loadDataFrame(ss, inputData.inputPath)
+      }
       case FileFormat.CSV => {
         ss.read.format("csv").option("header", "true").option("delimiter", csvDelimiterOption).load(inputData.inputPath)
       }