feat: native Spark add_months scalar function

CMakeLists.txt

@@ -78,7 +78,8 @@ set(EXTENSION_SOURCES
     ${LPTS_DIR}/src/lpts_ast_renderer.cpp
     ${LPTS_DIR}/src/lpts_ast_builder.cpp
     ${LPTS_DIR}/src/lpts_ast_flattener.cpp
-    ${LPTS_DIR}/src/dialect_function_map.cpp)
+    ${LPTS_DIR}/src/dialect_function_map.cpp
+    ${LPTS_DIR}/src/spark_scalar_functions.cpp)
 
 build_static_extension(${TARGET_NAME} ${EXTENSION_SOURCES})
 build_loadable_extension(${TARGET_NAME} " " ${EXTENSION_SOURCES})

src/delta/operators/join.cpp

@@ -834,20 +834,61 @@ void AppendMultiplicityToAncestorProjectionMaps(unique_ptr<LogicalOperator> &ter
 				if (mul_idx == DConstants::INVALID_INDEX) {
 					continue;
 				}
+				// Appending to this join's own left_projection_map grows its LEFT
+				// contribution width, which shifts the absolute position where its RIGHT
+				// contribution starts within its own combined GetColumnBindings(). A
+				// grandparent ancestor may already have a projection map entry referencing
+				// (by that now-stale absolute position) a column from this join's right
+				// side; left as-is, that entry would silently start pointing at the
+				// newly-inserted column instead, dropping the real column it used to
+				// select. Appending to right_projection_map never has this effect: right
+				// contributions are always placed last, so a new entry there only ever
+				// extends the combined output with a brand-new highest index.
+				idx_t old_width = proj_map.size();
+				auto shift_stale_parent_indexes = [&](idx_t added) {
+					if (child_side != 0 || added == 0 || depth == 0) {
+						return;
+					}
+					size_t parent_side = leaf_path[depth - 1];
+					auto *parent_join = dynamic_cast<LogicalJoin *>(ancestors[depth - 1]);
+					if (!parent_join || parent_side >= parent_join->children.size()) {
+						return;
+					}
+					auto &parent_map =
+					    (parent_side == 0) ? parent_join->left_projection_map : parent_join->right_projection_map;
+					for (auto &parent_idx : parent_map) {
+						if (parent_idx >= old_width) {
+							parent_idx += added;
+						}
+					}
+				};
 				if (preserve_full_child) {
 					idx_t projectable_count = MinValue<idx_t>(mul_idx + 1, child_bindings.size());
+					idx_t added = 0;
 					for (idx_t binding_idx = 0; binding_idx < projectable_count; binding_idx++) {
 						if (std::find(proj_map.begin(), proj_map.end(), binding_idx) != proj_map.end()) {
 							continue;
 						}
 						proj_map.push_back(binding_idx);
+						added++;
 						OPENIVM_DEBUG_PRINT("[%s] Preserved child col %lu in immediate %s proj_map\n", context_label,
 						                    (unsigned long)binding_idx, child_side == 0 ? "left" : "right");
 					}
-				} else if (std::find(proj_map.begin(), proj_map.end(), mul_idx) == proj_map.end()) {
-					proj_map.push_back(mul_idx);
-					OPENIVM_DEBUG_PRINT("[%s] Added mul col %lu to ancestor %s proj_map\n", context_label,
-					                    (unsigned long)mul_idx, child_side == 0 ? "left" : "right");
+					shift_stale_parent_indexes(added);
+				} else {
+					// proj_map entries are positions into the child's *current* combined
+					// GetColumnBindings(). Testing raw index membership of mul_idx against
+					// proj_map can alias onto an unrelated pre-existing entry that now shares
+					// the same numeric position after a deeper level's own map grew. Compare
+					// by column identity against what this ancestor currently exposes instead
+					// of trusting the raw index.
+					auto exposed = join->GetColumnBindings();
+					if (std::find(exposed.begin(), exposed.end(), mul_binding) == exposed.end()) {
+						proj_map.push_back(mul_idx);
+						shift_stale_parent_indexes(1);
+						OPENIVM_DEBUG_PRINT("[%s] Added mul col %lu to ancestor %s proj_map\n", context_label,
+						                    (unsigned long)mul_idx, child_side == 0 ? "left" : "right");
+					}
 				}
 				join->ResolveOperatorTypes();
 			}
@@ -1497,9 +1538,20 @@ BuildInclusionExclusionTerms(DeltaOperatorInput input, ClientContext &context, B
 		for (size_t i = 0; i < N; i++) {
 			if (mask & (1ULL << i)) {
 				if (leaves[i].get) {
-					DeltaGetResult delta_i = CreateDeltaGetNode(context, binder, leaves[i].get, input.context.view);
+					// leaves[] was collected once on the ORIGINAL input.plan, before this
+					// mask's own renumber_and_rebind_subtree pass, so leaves[i].get is a
+					// stale pointer carrying the pre-renumbering table_index. The rest of
+					// `term` (join conditions, transitioning-key guards, etc.) was rebound
+					// to the FRESH per-term index, so the replacement delta node -- which
+					// reuses old_get->table_index verbatim -- must be built from term's own,
+					// already-renumbered GET at this leaf's (renumbering-invariant) path,
+					// not from leaves[i].get, or every reference elsewhere in `term` to this
+					// leaf's fresh index is left dangling.
+					auto &leaf_node_ref = GetNodeAtPath(term, leaves[i].path);
+					auto &term_local_get = leaf_node_ref->Cast<LogicalGet>();
+					DeltaGetResult delta_i = CreateDeltaGetNode(context, binder, &term_local_get, input.context.view);
 					mul_bindings.push_back(delta_i.mul_binding);
-					GetNodeAtPath(term, leaves[i].path) = std::move(delta_i.node);
+					leaf_node_ref = std::move(delta_i.node);
 					UpdateParentProjectionMap(term, leaves[i], delta_i.mul_binding);
 				} else {
 					auto &subtree_ref = GetNodeAtPath(term, leaves[i].path);
@@ -1616,6 +1668,22 @@ static bool HasOnlyInnerJoins(LogicalOperator *node) {
 	return true;
 }
 
+static bool HasOnlyInnerOrLeftJoins(LogicalOperator *node) {
+	if (node->type == LogicalOperatorType::LOGICAL_COMPARISON_JOIN ||
+	    node->type == LogicalOperatorType::LOGICAL_ANY_JOIN) {
+		auto *join = dynamic_cast<LogicalJoin *>(node);
+		if (!join || (join->join_type != JoinType::INNER && join->join_type != JoinType::LEFT)) {
+			return false;
+		}
+	}
+	for (auto &child : node->children) {
+		if (!HasOnlyInnerOrLeftJoins(child.get())) {
+			return false;
+		}
+	}
+	return true;
+}
+
 static bool SupportsRegularNtermLeaf(const JoinLeafInfo &leaf) {
 	if (leaf.get) {
 		return leaf.get->GetTable().get() != nullptr;
@@ -1705,7 +1773,7 @@ static DeltaPlanFragment CompileRegularLeafDelta(const DeltaOperatorInput &input
 
 static vector<unique_ptr<LogicalOperator>> BuildRegularJoinTerms(DeltaOperatorInput input, ClientContext &context,
                                                                  Binder &binder, const vector<JoinLeafInfo> &leaves,
-                                                                 uint64_t unchanged_mask) {
+                                                                 uint64_t unchanged_mask, bool has_left_join) {
 	vector<unique_ptr<LogicalOperator>> terms;
 	// Base scans see post-DML state. Term i uses current state before i, delta i, and reconstructs old state after i as
 	// current - delta. These disjoint telescoping terms cover every non-empty delta combination exactly once.
@@ -1735,6 +1803,16 @@ static vector<unique_ptr<LogicalOperator>> BuildRegularJoinTerms(DeltaOperatorIn
 		LogicalOperator *term_root = term.get();
 		CollectJoinLeaves(term.get(), {}, term_leaves);
 		D_ASSERT(term_leaves.size() == leaves.size());
+
+		// LEFT-JOIN telescoping: demote only the outer join(s) whose NULL-supplying
+		// subtree contains this term's single delta leaf, mirroring the DuckLake
+		// N-term path (DemoteLeftJoinsForMask). Preserved outer joins elsewhere keep
+		// their NULL-padded rows; the upsert layer's key-based partial recompute
+		// (BuildLeftJoinProjectionRefresh) fixes NULL<->match transition rows.
+		if (has_left_join) {
+			DemoteLeftJoinsForMask(term.get(), term_leaves, (1ULL << delta_leaf));
+		}
+
 		vector<ColumnBinding> mul_bindings;
 
 		for (size_t leaf = 0; leaf < term_leaves.size(); leaf++) {
@@ -1880,11 +1958,22 @@ DeltaPlanFragment CompileJoinDelta(DeltaOperatorInput input) {
 	}
 	auto compile_facts = openivm::CompileFactsContextSlot::Get(context);
 	auto unchanged_mask = ComputeFactsUnchangedMask(compile_facts, leaves);
-	bool regular_nterm = !all_ducklake && compile_facts.compile_only && !has_left_join &&
-	                     input.context.model.type == RefreshType::SIMPLE_PROJECTION &&
-	                     HasOnlyInnerJoins(input.plan.get()) &&
-	                     RegularNtermPreservesFKPruning(context, compile_facts, leaves, input.plan.get()) &&
-	                     SqlUtils::GetBoolSetting(context, "openivm_regular_nterm", true);
+	bool regular_nterm_base = !all_ducklake && compile_facts.compile_only &&
+	                          input.context.model.type == RefreshType::SIMPLE_PROJECTION &&
+	                          SqlUtils::GetBoolSetting(context, "openivm_regular_nterm", true);
+	bool regular_nterm;
+	if (has_left_join) {
+		// LEFT-JOIN telescoping: the regular N-term delta extends to LEFT joins via
+		// per-term demotion of only the outer join whose NULL-supplying side carries
+		// that term's delta (see BuildRegularJoinTerms). FULL OUTER / RIGHT shapes and
+		// the inclusion-exclusion FK-pruning path are out of scope; NULL-padded row
+		// correctness is completed by BuildLeftJoinProjectionRefresh in the upsert layer.
+		regular_nterm = regular_nterm_base && HasOnlyInnerOrLeftJoins(input.plan.get()) &&
+		                SqlUtils::GetBoolSetting(context, "openivm_regular_nterm_left", true);
+	} else {
+		regular_nterm = regular_nterm_base && HasOnlyInnerJoins(input.plan.get()) &&
+		                RegularNtermPreservesFKPruning(context, compile_facts, leaves, input.plan.get());
+	}
 	if (regular_nterm) {
 		for (auto &leaf : leaves) {
 			if (!SupportsRegularNtermLeaf(leaf)) {
@@ -1902,7 +1991,7 @@ DeltaPlanFragment CompileJoinDelta(DeltaOperatorInput input) {
 	if (all_ducklake) {
 		terms = BuildDuckLakeJoinTerms(input, context, binder, leaves, has_left_join, flattened_ducklake);
 	} else if (regular_nterm) {
-		terms = BuildRegularJoinTerms(input, context, binder, leaves, unchanged_mask);
+		terms = BuildRegularJoinTerms(input, context, binder, leaves, unchanged_mask, has_left_join);
 	} else {
 		terms = BuildInclusionExclusionTerms(input, context, binder, leaves, has_left_join, transition_ctes);
 	}

src/include/upsert/refresh_compiler.hpp

@@ -58,6 +58,28 @@ string CompileWindowRecompute(const string &view_name, const string &view_query_
                               const vector<string> &column_names = {}, bool running_window_incremental = false);
 string CompileFullRecompute(const string &view_name, const string &view_query_sql, const string &catalog_prefix = "");
 
+/// Full recompute of `openivm_data_<view>` that ALSO emits the exact signed
+/// multiset view-delta into `openivm_delta_<view>`.
+///
+/// The partial-recompute paths (`WINDOW_PARTITION`, `GROUP_RECOMPUTE`) degrade to
+/// a full recompute whenever the affected partition/group key set cannot be
+/// scoped from the source deltas — an unpartitioned surrogate-key
+/// `ROW_NUMBER() OVER (ORDER BY ...)`, a partition key that is a computed
+/// expression absent from every source delta table, or incomplete multi-source
+/// lineage. The view keeps its `WINDOW_PARTITION` / `GROUP_RECOMPUTE`
+/// classification, so a caller that asked for a cascade delta
+/// (`CompileFacts::force_view_delta_cascade`) would otherwise receive a program
+/// that writes no `openivm_delta_<view>` rows at all, and every downstream MV
+/// would have to be demoted to a full refresh.
+///
+/// Retracting the whole pre-refresh content at multiplicity -1 and adding the
+/// whole post-refresh content at +1 is the exact Z-set delta of the view
+/// (`new_bag - old_bag`): unchanged rows contribute cancelling -1/+1 pairs, so
+/// bag semantics are preserved exactly. Statement shapes mirror the
+/// `CompileWindowRecompute` / `CompileGroupRecompute` cascade branches.
+string CompileFullRecomputeWithCascadeDelta(const string &view_name, const string &view_query_sql,
+                                            const string &catalog_prefix = "");
+
 /// Group-level partial recompute, used by `RefreshType::GROUP_RECOMPUTE`
 /// (inner-DISTINCT under aggregate). For each base table T_i with a non-empty
 /// delta, builds a "view query with T_i restricted to its delta" variant by

src/openivm_extension.cpp

@@ -2,6 +2,7 @@
 
 #include "core/openivm_extension.hpp"
 #include "compile_facts.hpp"
+#include "spark_scalar_functions.hpp"
 #include "core/openivm_constants.hpp"
 #include "core/refresh_metadata.hpp"
 #include "core/refresh_daemon.hpp"
@@ -166,6 +167,8 @@ static void LoadInternal(ExtensionLoader &loader) {
 	// statement OpenIVM does not recognize with DuckDB's native parser.
 	db_config.SetOption(AllowParserOverrideExtensionSetting::SettingIndex, Value("fallback"));
 
+	RegisterSparkScalarFunctions(loader);
+
 	db_config.AddExtensionOption("openivm_files_path", "path for compiled SQL reference files", LogicalType::VARCHAR);
 	db_config.AddExtensionOption("openivm_refresh_mode", "refresh strategy: incremental, full, or auto",
 	                             LogicalType::VARCHAR, Value("incremental"));
@@ -191,6 +194,9 @@ static void LoadInternal(ExtensionLoader &loader) {
 	                             LogicalType::BOOLEAN, Value::BOOLEAN(true));
 	db_config.AddExtensionOption("openivm_regular_nterm", "use N-term telescoping for compile-only regular inner joins",
 	                             LogicalType::BOOLEAN, Value::BOOLEAN(true));
+	db_config.AddExtensionOption("openivm_regular_nterm_left",
+	                             "extend compile-only N-term telescoping to LEFT-join projection views",
+	                             LogicalType::BOOLEAN, Value::BOOLEAN(true));
 	db_config.AddExtensionOption("openivm_fk_pruning", "prune inclusion-exclusion join terms using FK constraints",
 	                             LogicalType::BOOLEAN, Value::BOOLEAN(true));
 	db_config.AddExtensionOption("openivm_scd2_range_join_accel",

src/upsert/refresh_compiler.cpp

@@ -1455,15 +1455,40 @@ string CompileFullRecompute(const string &view_name, const string &view_query_sq
 	return SqlUtils::BuildFullRecomputeSQL(data_table, view_query_sql);
 }
 
+string CompileFullRecomputeWithCascadeDelta(const string &view_name, const string &view_query_sql,
+                                            const string &catalog_prefix) {
+	string data_table = catalog_prefix + SqlUtils::QuoteIdentifier(IncrementalTableNames::DataTableName(view_name));
+	string delta_table = catalog_prefix + SqlUtils::QuoteIdentifier(SqlUtils::DeltaName(view_name));
+	string old_temp_table = SqlUtils::QuoteIdentifier(string(openivm::TEMP_TABLE_PREFIX) + view_name);
+	string new_temp_table = SqlUtils::QuoteIdentifier(string("openivm_new_") + view_name);
+
+	string sql;
+	sql += "CREATE OR REPLACE TEMP TABLE " + old_temp_table + " AS\nSELECT * FROM " + data_table + " openivm_old;\n\n";
+	sql += "CREATE OR REPLACE TEMP TABLE " + new_temp_table + " AS\nSELECT * FROM (" + view_query_sql +
+	       ") openivm_recompute;\n\n";
+	sql += "DELETE FROM " + data_table + ";\n";
+	sql += "INSERT INTO " + data_table + "\nSELECT * FROM " + new_temp_table + ";\n";
+	sql += "\n" + BuildSignedMultisetDeltaInsertSQL(delta_table, old_temp_table, new_temp_table);
+	sql += "DROP TABLE IF EXISTS " + old_temp_table + ";\n";
+	sql += "DROP TABLE IF EXISTS " + new_temp_table + ";\n";
+	OPENIVM_DEBUG_PRINT("[CompileFullRecomputeWithCascadeDelta] unscopable recompute for '%s' — emitting signed "
+	                    "whole-view cascade delta\n",
+	                    view_name.c_str());
+	return sql;
+}
+
 string CompileGroupRecompute(const string &view_name, const string &view_query_sql, const vector<string> &group_columns,
                              const vector<GroupRecomputeDeltaSpec> &delta_table_specs, const string &catalog_prefix,
                              const string &lpts_table_prefix, bool emit_cascade_delta,
                              GroupRecomputeAffectedMode affected_mode) {
 	string data_table = catalog_prefix + SqlUtils::QuoteIdentifier(IncrementalTableNames::DataTableName(view_name));
 
 	// No GROUP BY columns or no source deltas registered → can't scope; fall back to full.
+	// A cascade delta was still requested, so emit the whole-view signed delta rather than
+	// silently producing a program with no `openivm_delta_<view>` rows.
 	if (group_columns.empty() || delta_table_specs.empty()) {
-		return CompileFullRecompute(view_name, view_query_sql, catalog_prefix);
+		return emit_cascade_delta ? CompileFullRecomputeWithCascadeDelta(view_name, view_query_sql, catalog_prefix)
+		                          : CompileFullRecompute(view_name, view_query_sql, catalog_prefix);
 	}
 
 	string group_csv = SqlUtils::JoinQuotedColumns(group_columns);

src/upsert/refresh_compiler_aux.cpp

@@ -1600,7 +1600,11 @@ string CompileWindowRecompute(const string &view_name, const string &view_query_
                               bool running_window_incremental) {
 	bool have_affected_keys = !affected_keys_sql.empty();
 	if (!have_affected_keys && (partition_columns.empty() || partition_delta_specs.empty())) {
-		return CompileFullRecompute(view_name, view_query_sql, catalog_prefix);
+		// No PARTITION BY (global surrogate-key window) or no partition key resolvable in any
+		// source delta table → nothing to scope the recompute to. Keep the cascade delta the
+		// caller asked for so downstream MVs stay incremental.
+		return emit_cascade_delta ? CompileFullRecomputeWithCascadeDelta(view_name, view_query_sql, catalog_prefix)
+		                          : CompileFullRecompute(view_name, view_query_sql, catalog_prefix);
 	}
 	if (running_window_incremental) {
 		auto suffix_sql = BuildRunningWindowSuffixRefreshSQL(view_name, view_query_sql, delta_ts_filter, catalog_prefix,

src/upsert/refresh_sql.cpp

@@ -532,9 +532,9 @@
 	// session-scoped planning settings still need to be mirrored onto the fresh
 	// planning connection.
 	static const char *PLANNING_SETTINGS[] = {
-	    "openivm_adaptive_refresh", "openivm_cost_decay",     "openivm_skip_empty_deltas",
-	    "openivm_fk_pruning",       "openivm_ducklake_nterm", "openivm_scd2_range_join_accel",
-	    "openivm_regular_nterm",
+	    "openivm_adaptive_refresh", "openivm_cost_decay",         "openivm_skip_empty_deltas",
+	    "openivm_fk_pruning",       "openivm_ducklake_nterm",     "openivm_scd2_range_join_accel",
+	    "openivm_regular_nterm",    "openivm_regular_nterm_left",
 	};
 	for (auto setting_name : PLANNING_SETTINGS) {
 		CopyOpenIvmSetting(from, to, setting_name);
@@ -1356,7 +1356,7 @@
 		OPENIVM_DEBUG_PRINT("[UPSERT] DISTINCT_INCREMENTAL view has no aux meta — "
 		                    "falling through to "
 		                    "GROUP_RECOMPUTE\n");
 		[[fallthrough]];
 	}
 	case RefreshType::SEMI_ANTI_RECOMPUTE: {
 		RefreshMetadata::SemiAntiAuxMeta aux_meta;
@@ -1390,7 +1390,7 @@
 		OPENIVM_DEBUG_PRINT("[UPSERT] SEMI_ANTI_RECOMPUTE view has no aux meta — "
 		                    "falling through to "
 		                    "GROUP_RECOMPUTE\n");
 		[[fallthrough]];
 	}
 	case RefreshType::GROUP_RECOMPUTE: {
 		auto group_columns = metadata.GetGroupColumns(view_name);
@@ -1442,7 +1442,7 @@
 		break;
 	}
 	case RefreshType::TOP_K:
 		[[fallthrough]];
 	case RefreshType::FULL_REFRESH: {
 		string full_recompute_query = view_query_sql;
 		if (active_facts.target_dialect == SqlDialect::SPARK) {