feat: TTL support for ScyllaDB config and FeatureViews

sdk/python/feast/feature_view.py

@@ -494,3 +494,23 @@ def most_recent_end_time(self) -> Optional[datetime]:
         if len(self.materialization_intervals) == 0:
             return None
         return max([interval[1] for interval in self.materialization_intervals])
+
+    @property
+    def online_store_ttl(self) -> Optional[int]:
+        """
+        Retrieves the online store TTL from the FeatureView's tags.
+
+        Returns:
+            An integer representing the TTL in seconds, or None if not set.
+        """
+        ttl_str = self.tags.get("online_store_ttl")
+        if ttl_str:
+            try:
+                ttl_seconds = int(ttl_str)
+                return ttl_seconds
+            except ValueError:
+                raise ValueError(
+                    f"Invalid online_store_ttl value '{ttl_str}' in tags. It must be an integer representing seconds."
+                )
+        else:
+            return None

sdk/python/feast/infra/online_stores/contrib/cassandra_online_store/cassandra_online_store.py

@@ -88,7 +88,8 @@
         event_ts        TIMESTAMP,
         created_ts      TIMESTAMP,
         PRIMARY KEY ((entity_key), feature_name)
-    ) WITH CLUSTERING ORDER BY (feature_name ASC);
+    ) WITH CLUSTERING ORDER BY (feature_name ASC)
+    {table_options};
 """
 
 DROP_TABLE_CQL_TEMPLATE = "DROP TABLE IF EXISTS {fqtable};"
@@ -159,6 +160,9 @@ class CassandraOnlineStoreConfig(FeastConfigBaseModel):
     Table deletion is not currently supported in this mode.
     """
 
+    ttl: Optional[StrictInt] = None
+    """Default TTL (in seconds) to apply to all tables if not specified in FeatureView."""
+
     class CassandraLoadBalancingPolicy(FeastConfigBaseModel):
         """
         Configuration block related to the Cluster's load-balancing policy.
@@ -566,8 +570,18 @@ def _create_table(self, config: RepoConfig, project: str, table: FeatureView):
         session: Session = self._get_session(config)
         keyspace: str = self._keyspace
         fqtable = CassandraOnlineStore._fq_table_name(keyspace, project, table)
-        create_cql = self._get_cql_statement(config, "create", fqtable)
-        logger.info(f"Creating table {fqtable}.")
+
+        ttl = (
+            table.online_store_ttl
+            if table.online_store_ttl is not None
+            else config.online_store.ttl
+        )
+        table_options = f" AND default_time_to_live = {ttl}" if ttl is not None else ""
+
+        create_cql = self._get_cql_statement(
+            config, "create", fqtable, table_options=table_options
+        )
+        logger.info(f"Creating table {fqtable} with TTL {ttl}.")
         session.execute(create_cql)
 
     def _get_cql_statement(