Integ tests

Author: nargokul

sagemaker-mlops/src/sagemaker/mlops/feature_store/dataset_builder.py

@@ -143,7 +143,7 @@ def construct_feature_group_to_be_merged(
         raise RuntimeError(f"No metastore configured for FeatureGroup {fg.feature_group_name}.")
 
     catalog_config = fg.offline_store_config.data_catalog_config
-    disable_glue = catalog_config.disable_glue_table_creation or False
+    disable_glue = getattr(catalog_config, "disable_glue_table_creation", False) or False
 
     features = [fd.feature_name for fd in fg.feature_definitions]
     record_id = fg.record_identifier_feature_name

sagemaker-mlops/src/sagemaker/mlops/feature_store/feature_utils.py

@@ -11,8 +11,8 @@
 import boto3
 import pandas
 import pandas as pd
-from pandas import DataFrame, Series, read_csv\
-
+from pandas import DataFrame, Series, read_csv
+from sagemaker.core.utils.utils import Unassigned
 from sagemaker.mlops.feature_store import FeatureGroup as CoreFeatureGroup, FeatureGroup
 from sagemaker.core.helper.session_helper import Session
 from sagemaker.core.s3.client import S3Uploader, S3Downloader
@@ -369,7 +369,7 @@ def create_athena_query(feature_group_name: str, session: Session):
         raise RuntimeError("No metastore is configured with this feature group.")
 
     catalog_config = fg.offline_store_config.data_catalog_config
-    disable_glue = catalog_config.disable_glue_table_creation or False
+    disable_glue = getattr(catalog_config, "disable_glue_table_creation", False) or False
 
     return AthenaQuery(
         catalog=catalog_config.catalog if disable_glue else "AwsDataCatalog",
@@ -451,7 +451,16 @@ def ingest_dataframe(
         raise ValueError("max_workers must be greater than 0.")
 
     fg = CoreFeatureGroup.get(feature_group_name=feature_group_name)
-    feature_definitions = {fd.feature_name: fd.feature_type for fd in fg.feature_definitions}
+    feature_definitions = {}
+    for fd in fg.feature_definitions:
+        collection_type = getattr(fd, "collection_type", None)
+        # Handle Unassigned, empty string, or None as None
+        if isinstance(collection_type, Unassigned) or collection_type == "" or collection_type is None:
+            collection_type = None
+        feature_definitions[fd.feature_name] = {
+            "FeatureType": fd.feature_type,
+            "CollectionType": collection_type,
+        }
 
     manager = IngestionManagerPandas(
         feature_group_name=feature_group_name,

sagemaker-mlops/src/sagemaker/mlops/feature_store/ingestion_manager_pandas.py

@@ -15,6 +15,7 @@
 
 from sagemaker.core.resources import FeatureGroup as CoreFeatureGroup
 from sagemaker.core.shapes import FeatureValue
+from sagemaker.core.utils.utils import Unassigned
 
 logger = logging.getLogger(__name__)
 
@@ -293,7 +294,10 @@ def _is_feature_collection_type(
         """Check if the feature is a collection type."""
         feature_def = feature_definitions.get(feature_name)
         if feature_def:
-            return feature_def.get("CollectionType") is not None
+            collection_type = feature_def.get("CollectionType")
+            if isinstance(collection_type, Unassigned) or collection_type is None or collection_type == "":
+                return False
+            return True
         return False
 
     @staticmethod