Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 39 additions & 7 deletions src/openstatspec/frontends/spss/binding.py
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,41 @@ def _result_type(
return result.value.type


def _validate_recode_string_width(
result: RecodeResult,
source: VariableDefinition,
target: VariableDefinition,
span: SourceSpan,
) -> None:
if target.storage_kind != "string" or target.declared_string_width is None:
return
if result.kind == "literal":
assert result.value is not None
if result.value.type != "string":
return
assert isinstance(result.value.value, str)
if len(result.value.value.encode("utf-8")) > target.declared_string_width:
raise frontend_error(
"string_width_exceeded",
"A RECODE string literal exceeds the target's declared string width.",
span=span,
variable=target.name,
declared_string_width=target.declared_string_width,
)
return
if result.kind == "copy" and source.storage_kind == "string":
source_width = source.declared_string_width
if source_width is None or source_width > target.declared_string_width:
raise frontend_error(
"string_width_exceeded",
"A RECODE COPY result can exceed the target's declared string width.",
span=span,
source=source.name,
variable=target.name,
declared_string_width=target.declared_string_width,
)


def _bind_recode(
command: RecodeCommandSyntax, variables: list[VariableDefinition],
) -> tuple[list[RecodeOperation], list[SourceSpan]]:
Expand Down Expand Up @@ -252,13 +287,17 @@ def _bind_recode(
else_result: RecodeResult | None = None
for clause in command.clauses:
result = _result(clause.result, source)
if target_mode == "replace":
_validate_recode_string_width(result, source, source, clause.result.span)
if clause.match.kind == "else":
else_result = result
continue
rules.append(RecodeRule(_match(clause.match, source), result))
unmatched = else_result or RecodeResult(
"system_missing" if target_mode == "create" else "copy"
)
if target_mode == "replace" and else_result is None:
_validate_recode_string_width(unmatched, source, source, command.span)
result_types = {
_result_type(result, source)
for result in [*(rule.result for rule in rules), unmatched]
Expand Down Expand Up @@ -339,13 +378,6 @@ def bind_spss_syntax(
index, variable = _resolve(
variables, variable_token.text, variable_token.span,
)
if len(variables) == 1:
raise frontend_error(
"cannot_delete_last_variable",
"DELETE VARIABLES cannot remove the final dataset variable.",
span=variable_token.span,
variable=variable.name,
)
operations.append(DeleteVariableOperation(variable.name))
spans.append(command.span)
del variables[index]
Expand Down
15 changes: 15 additions & 0 deletions src/openstatspec/frontends/spss/syntax.py
Original file line number Diff line number Diff line change
Expand Up @@ -563,10 +563,24 @@ def execute(self, start: Token) -> ExecuteCommandSyntax:
end = self.expects("period", "Expected '.' after EXECUTE.")
return ExecuteCommandSyntax(_joined_span(start.span, end.span))

@staticmethod
def reject_to_range(variables: tuple[Token, ...], command: str) -> None:
range_token = next(
(variable for variable in variables if variable.text.casefold() == "to"),
None,
)
if range_token is not None:
raise frontend_error(
"unsupported_spss_feature",
f"{command} variable ranges using TO are not supported.",
span=range_token.span,
)

def string(self, start: Token) -> StringCommandSyntax:
variables = self.variable_list(
stop_kinds=frozenset({"left_paren", "period", "eof"}),
)
self.reject_to_range(variables, "STRING")
self.expects("left_paren", "Expected '(' before a STRING width.")
width_token = self.expects(
"identifier", "STRING requires a width such as A20.",
Expand Down Expand Up @@ -595,6 +609,7 @@ def string(self, start: Token) -> StringCommandSyntax:
def delete_variables(self, start: Token) -> DeleteVariablesCommandSyntax:
self.expects_keyword("VARIABLES")
variables = self.variable_list(stop_kinds=frozenset({"period", "eof"}))
self.reject_to_range(variables, "DELETE VARIABLES")
end = self.expects("period", "Expected '.' after DELETE VARIABLES.")
return DeleteVariablesCommandSyntax(
variables, _joined_span(start.span, end.span),
Expand Down
67 changes: 28 additions & 39 deletions src/openstatspec/sql/inplace_transform.py
Original file line number Diff line number Diff line change
Expand Up @@ -389,22 +389,30 @@ def _delete_variable_metadata(
))


def _compact_variable_ordinals(
def _assert_postgresql_column_slot_available(
connection: Any,
*,
core: Any,
variables: list[dict[str, Any]],
relation: Table,
target_profile: SqlProfile,
) -> None:
"""Keep normative variable source order contiguous after deletion."""
for source_ordinal, variable in enumerate(variables, start=1):
if int(variable["source_ordinal"]) == source_ordinal:
continue
connection.execute(
update(core.variable)
.where(core.variable.c.variable_id == variable["variable_id"])
.values(source_ordinal=source_ordinal)
"""Reject PostgreSQL creates when dropped columns exhaust attribute slots."""
if connection.dialect.name != "postgresql":
return
qualified_relation = connection.dialect.identifier_preparer.format_table(relation)
attribute_slots = connection.execute(text(
"SELECT COUNT(*) FROM pg_attribute "
"WHERE attrelid = to_regclass(:relation_name) AND attnum > 0"
), {"relation_name": qualified_relation}).scalar_one()
if not isinstance(attribute_slots, int) or attribute_slots < 1:
raise TransformationError(
"physical_table_missing",
"The target dataset's physical wide table does not exist.",
)
if attribute_slots >= target_profile.max_source_variables + 1:
raise TransformationError(
"source_variable_limit",
"PostgreSQL physical column slots are exhausted for this dataset table.",
)
variable["source_ordinal"] = source_ordinal


def _failure_boundary(_name: str) -> None:
Expand Down Expand Up @@ -555,6 +563,9 @@ def _apply_plan_on_connection(
)
by_name = {str(row["source_name"]).casefold(): row for row in variables}
used_physical = {str(row["physical_name"]).casefold() for row in variables}
next_source_ordinal = max(
(int(row["source_ordinal"]) for row in variables), default=0,
) + 1
quote = connection.dialect.identifier_preparer.quote
qualified_table = connection.dialect.identifier_preparer.format_table(relation)
numeric_type = (
Expand Down Expand Up @@ -603,6 +614,9 @@ def _apply_plan_on_connection(
)
)
if creates_target:
_assert_postgresql_column_slot_available(
connection, relation=relation, target_profile=target_profile,
)
target_name = (
operation.variable
if isinstance(operation, CreateVariableOperation)
Expand Down Expand Up @@ -631,7 +645,7 @@ def _apply_plan_on_connection(
created_target = {
"variable_id": str(uuid4()),
"dataset_id": dataset_id,
"source_ordinal": len(variables) + 1,
"source_ordinal": next_source_ordinal,
"source_name": target_name,
"physical_name": target_physical,
"storage_kind": storage_kind,
Expand All @@ -648,6 +662,7 @@ def _apply_plan_on_connection(
)
connection.execute(insert(core.variable).values(**created_target))
variables.append(created_target)
next_source_ordinal += 1
by_name[str(created_target["source_name"]).casefold()] = created_target
_failure_boundary("catalog")
if isinstance(operation, CreateVariableOperation):
Expand Down Expand Up @@ -725,36 +740,10 @@ def _apply_plan_on_connection(
)
_delete_variable_metadata(connection, core=core, variable=variable)
by_name.pop(operation.variable.casefold(), None)
used_physical.discard(str(variable["physical_name"]).casefold())
variables = [
row for row in variables
if row["variable_id"] != variable["variable_id"]
]
_compact_variable_ordinals(
connection, core=core, variables=variables,
)
canonical_physical = {"__case_ordinal"}
for remaining_variable in variables:
expected_physical = physical_name(
str(remaining_variable["source_name"]), canonical_physical,
)
current_physical = str(remaining_variable["physical_name"])
if current_physical == expected_physical:
continue
connection.exec_driver_sql(
f"ALTER TABLE {qualified_table} RENAME COLUMN "
f"{quote(current_physical)} TO {quote(expected_physical)}"
)
connection.execute(
update(core.variable)
.where(
core.variable.c.variable_id
== remaining_variable["variable_id"]
)
.values(physical_name=expected_physical)
)
remaining_variable["physical_name"] = expected_physical
used_physical = canonical_physical
relation = Table(
table_name, MetaData(), schema=dataset.get("physical_table_schema"),
autoload_with=connection,
Expand Down
36 changes: 28 additions & 8 deletions src/openstatspec/sql/profiles.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,7 @@ def preflight(
variables_or_count: int | Iterable[Mapping[str, Any]],
*,
rows: Iterable[Mapping[str, Any]] | None = None,
require_canonical_mapping: bool = True,
) -> None:
"""Validate strict target capabilities before any source dataset is created."""
variables = None if isinstance(variables_or_count, int) else list(variables_or_count)
Expand All @@ -143,6 +144,7 @@ def preflight(

used = {"__case_ordinal"}
source_names: set[str] = set()
physical_names = {"__case_ordinal"}
for expected_ordinal, variable in enumerate(variables, start=1):
source_name = variable.get("source_name")
if not isinstance(source_name, str) or not source_name or source_name in source_names:
Expand All @@ -151,18 +153,36 @@ def preflight(
source_name=source_name,
)
source_names.add(source_name)
expected_name = _physical_name(source_name, used)
actual_name = variable.get("physical_name")
if variable.get("ordinal") != expected_ordinal or actual_name != expected_name:
if not isinstance(actual_name, str) or not actual_name:
raise _exceeded(
"physical_identifier_mapping_invalid",
f"{source_name!r} must map deterministically to {expected_name!r} in source order.",
source_name=source_name, expected_physical_name=expected_name,
actual_physical_name=actual_name,
f"{source_name!r} has no physical variable identifier.",
source_name=source_name, actual_physical_name=actual_name,
)
if actual_name.casefold() in physical_names:
raise _exceeded(
"physical_identifier_collision",
"physical variable identifiers must be unique.",
source_name=source_name, physical_name=actual_name,
)
physical_names.add(actual_name.casefold())
if require_canonical_mapping:
expected_name = _physical_name(source_name, used)
if variable.get("ordinal") != expected_ordinal or actual_name != expected_name:
raise _exceeded(
"physical_identifier_mapping_invalid",
f"{source_name!r} must map deterministically to {expected_name!r} in source order.",
source_name=source_name, expected_physical_name=expected_name,
actual_physical_name=actual_name,
)
preflight_identifier(
profile, expected_name, role="physical variable identifier",
)
else:
preflight_identifier(
profile, actual_name, role="physical variable identifier",
)
preflight_identifier(
profile, expected_name, role="physical variable identifier",
)
if variable.get("storage_kind") == "string":
declared_width = variable.get("string_width")
if declared_width is not None and (
Expand Down
8 changes: 6 additions & 2 deletions src/openstatspec/sql/wide.py
Original file line number Diff line number Diff line change
Expand Up @@ -1239,7 +1239,9 @@ def read_wide_dataset(
)
for row in response_sets
}, ensure_ascii=False)
preflight(profile, variables, rows=rows)
preflight(
profile, variables, rows=rows, require_canonical_mapping=False,
)
return dataset, variables, rows


Expand Down Expand Up @@ -1705,7 +1707,9 @@ def validate_wide_dataset(
database_url=database_url, dataset_id=dataset_id, profile=profile,
dolt_conformance_source=dolt_conformance_source,
)
preflight(profile, variables, rows=rows)
preflight(
profile, variables, rows=rows, require_canonical_mapping=False,
)
validate_spss_catalog(
variables,
case_weight_variable=dataset.get("case_weight_variable"),
Expand Down
48 changes: 45 additions & 3 deletions src/openstatspec/transform/validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,38 @@ def _result_type(
return result.value.type


def _validate_recode_string_width(
result: RecodeResult,
source: VariableDefinition,
target: VariableDefinition,
) -> None:
if target.storage_kind != "string" or target.declared_string_width is None:
return
if result.kind == "literal":
assert result.value is not None
if result.value.type != "string":
return
assert isinstance(result.value.value, str)
if len(result.value.value.encode("utf-8")) > target.declared_string_width:
raise frontend_error(
"string_width_exceeded",
"A RECODE string literal exceeds the target's declared string width.",
variable=target.name,
declared_string_width=target.declared_string_width,
)
return
if result.kind == "copy" and source.storage_kind == "string":
source_width = source.declared_string_width
if source_width is None or source_width > target.declared_string_width:
raise frontend_error(
"string_width_exceeded",
"A RECODE COPY result can exceed the target's declared string width.",
source=source.name,
variable=target.name,
declared_string_width=target.declared_string_width,
)


def _bind_recode(
operation: RecodeOperation, variables: list[VariableDefinition]
) -> None:
Expand All @@ -122,6 +154,10 @@ def _bind_recode(
)
for rule in operation.rules:
_validate_match(rule.match, source)
if operation.target_mode == "replace":
_validate_recode_string_width(rule.result, source, source)
if operation.target_mode == "replace":
_validate_recode_string_width(operation.unmatched, source, source)
result_types = {
_result_type(result, source)
for result in [
Expand Down Expand Up @@ -282,12 +318,16 @@ def bind_transformation_plan(
if not isinstance(schema, VariableSchema):
raise TypeError("schema must be a VariableSchema.")
variables = list(schema.variables)
for operation in plan.operations:
for operation_index, operation in enumerate(plan.operations):
later_create = any(
isinstance(later_operation, CreateVariableOperation)
for later_operation in plan.operations[operation_index + 1:]
Comment on lines +322 to +324

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Count later numeric target creation before final deletes

When a plan starts from a one-variable schema and deletes that variable before creating a numeric replacement, e.g. DELETE VARIABLES only. COMPUTE replacement = 1., the final schema is non-empty and numeric create targets are otherwise supported. This lookahead only treats CreateVariableOperation as a later create, so _bind_delete rejects the delete before a later AssignOperation/RecodeOperation with target_mode="create" can add the replacement. Include those target-creating operations in the lookahead so valid sequential plans are not rejected.

Useful? React with 👍 / 👎.

)
if isinstance(operation, CreateVariableOperation):
_bind_create(operation, variables)
continue
if isinstance(operation, DeleteVariableOperation):
_bind_delete(operation, variables)
_bind_delete(operation, variables, allow_empty=later_create)
continue
if isinstance(operation, RecodeOperation):
_bind_recode(operation, variables)
Expand Down Expand Up @@ -367,9 +407,11 @@ def _bind_create(
def _bind_delete(
operation: DeleteVariableOperation,
variables: list[VariableDefinition],
*,
allow_empty: bool,
) -> None:
index, variable = _resolve(variables, operation.variable)
if len(variables) == 1:
if len(variables) == 1 and not allow_empty:
raise frontend_error(
"cannot_delete_last_variable",
"A dataset must retain at least one variable.",
Expand Down
Loading