diff --git a/.changes/unreleased/Features-20240927-133927.yaml b/.changes/unreleased/Features-20240927-133927.yaml
new file mode 100644
index 000000000..ce04ac073
--- /dev/null
+++ b/.changes/unreleased/Features-20240927-133927.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Enable setting current value of dbt_valid_to
+time: 2024-09-27T13:39:27.268886-04:00
+custom:
+  Author: gshank
+  Issue: "1112"
diff --git a/dbt/include/spark/macros/materializations/snapshot.sql b/dbt/include/spark/macros/materializations/snapshot.sql
index 43c4750f6..b4ef6e5d4 100644
--- a/dbt/include/spark/macros/materializations/snapshot.sql
+++ b/dbt/include/spark/macros/materializations/snapshot.sql
@@ -24,7 +24,12 @@
     {% endif %}
     on DBT_INTERNAL_SOURCE.{{ columns.dbt_scd_id }} = DBT_INTERNAL_DEST.{{ columns.dbt_scd_id }}
     when matched
-     and DBT_INTERNAL_DEST.{{ columns.dbt_valid_to }} is null
+     {% if config.get("dbt_valid_to_current") %}
+       and ( DBT_INTERNAL_DEST.{{ columns.dbt_valid_to }} = {{ config.get('dbt_valid_to_current') }} or
+             DBT_INTERNAL_DEST.{{ columns.dbt_valid_to }} is null )
+     {% else %}
+       and DBT_INTERNAL_DEST.{{ columns.dbt_valid_to }} is null
+     {% endif %}
      and DBT_INTERNAL_SOURCE.dbt_change_type in ('update', 'delete')
         then update
         set {{ columns.dbt_valid_to }} = DBT_INTERNAL_SOURCE.{{ columns.dbt_valid_to }}
diff --git a/tests/functional/adapter/test_python_model.py b/tests/functional/adapter/test_python_model.py
index 50132b883..2ecce2662 100644
--- a/tests/functional/adapter/test_python_model.py
+++ b/tests/functional/adapter/test_python_model.py
@@ -67,7 +67,7 @@ def model(dbt, spark):
                 "ResourceClass": "SingleNode"
             }
         },
-        packages=['spacy', 'torch', 'pydantic>=1.10.8']
+        packages=['spacy', 'torch', 'pydantic>=1.10.8', 'numpy<2']
     )
     data = [[1,2]] * 10
     return spark.createDataFrame(data, schema=['test', 'test2'])
@@ -85,7 +85,11 @@ def model(dbt, spark):
 
 
 @pytest.mark.skip_profile(
-    "apache_spark", "spark_session", "databricks_sql_endpoint", "spark_http_odbc"
+    "apache_spark",
+    "spark_session",
+    "databricks_sql_endpoint",
+    "spark_http_odbc",
+    "databricks_http_cluster",
 )
 class TestChangingSchemaSpark:
     """