Fix atomic catalog writes for invalid SQL identifiers

CHANGELOG.md

@@ -7,6 +7,8 @@ Write the date in place of the "Unreleased" in the case a new version is release
 
 ### Fixed
 
+- Roll back catalog node creation when SQL identifier validation fails, so a
+  corrected retry is not blocked by a partially-created node.
 - Extend to zarr routes the previous fix for reads of array data whose
   on-disk shape has diverged from the shape recorded in the catalog
   structure, which can happen while an array is being extended

tests/test_writing.py

@@ -285,6 +285,36 @@ def test_write_table_full(tree):
         assert result.specs == specs
 
 
+@pytest.mark.parametrize(
+    "column_name, table_name",
+    [
+        (
+            "column_name_that_exceeds_the_sixty_three_byte_sql_identifier_limit",
+            None,
+        ),
+        ("1invalid_start", None),
+        ('invalid"name', None),
+        ("valid_column", "select"),
+    ],
+)
+def test_invalid_sql_identifier_does_not_create_catalog_node(
+    tree, column_name, table_name
+):
+    with Context.from_app(build_app(tree)) as context:
+        client = from_context(context)
+        key = "invalid_sql_identifier"
+        schema = pyarrow.schema([pyarrow.field(column_name, pyarrow.int64())])
+
+        with fail_with_status_code(HTTP_422_UNPROCESSABLE_CONTENT):
+            client.create_appendable_table(schema, key=key, table_name=table_name)
+
+        assert key not in client
+        client.create_appendable_table(
+            pyarrow.schema([pyarrow.field("valid_column", pyarrow.int64())]), key=key
+        )
+        assert key in client
+
+
 def test_write_table_partitioned(tree):
     with Context.from_app(
         build_app(tree, validation_registry=validation_registry)
@@ -940,38 +970,26 @@ def test_append_partition(
 
 
 @pytest.mark.parametrize(
-    "table_name, expected",
+    "table_name, valid",
     [
-        (None, None),
-        ("valid_table_name", None),
-        (
-            "_invalid_table_name",
-            pytest.raises(ValueError, match=r"Malformed SQL identifier.+"),
-        ),
-        (
-            "invalid-table-name",
-            pytest.raises(ValueError, match=r"Malformed SQL identifier.+"),
-        ),
-        (
-            "UPPERCASE_TABLE_NAME",
-            pytest.raises(ValueError, match=r"Malformed SQL identifier.+"),
-        ),
-        (
-            "",
-            pytest.raises(ValueError, match=r"Malformed SQL identifier.+"),
-        ),
+        (None, True),
+        ("valid_table_name", True),
+        ("_invalid_table_name", False),
+        ("invalid-table-name", False),
+        ("UPPERCASE_TABLE_NAME", False),
+        ("", False),
     ],
 )
 def test_create_table_with_custom_name(
     tree: CatalogContainerAdapter,
     table_name: str,
-    expected: str,
+    valid: bool,
 ):
     table = pyarrow.Table.from_arrays([[1, 2, 3]], ["column_name"])
     with Context.from_app(build_app(tree)) as context:
         client = from_context(context, include_data_sources=True)
-        if isinstance(expected, type(pytest.raises(ValueError))):
-            with expected:
+        if not valid:
+            with fail_with_status_code(HTTP_422_UNPROCESSABLE_CONTENT):
                 client.create_appendable_table(table.schema, table_name=table_name)
         else:
             x = client.create_appendable_table(table.schema, table_name=table_name)

tiled/adapters/sql.py

@@ -1032,15 +1032,15 @@ def is_safe_identifier(
         )
 
     if not allow_reserved_words and identifier.lower() in RESERVED_WORDS:
-        raise ValueError(
+        raise UnsafeIdentifier(
             f'Reserved SQL keywords are not allowed in identifiers, "{identifier}"'
         )
 
     if pattern.match(identifier) is None:
-        raise ValueError(f'Malformed SQL identifier "{identifier}"')
+        raise UnsafeIdentifier(f'Malformed SQL identifier "{identifier}"')
 
     if match := FORBIDDEN_CHARACTERS.search(identifier):
-        raise ValueError(
+        raise UnsafeIdentifier(
             f'Invalid SQL identifier "{identifier}" '
             f"contains forbidden character(s): {', '.join(match.groups())}"
         )

tiled/catalog/adapter.py

@@ -924,6 +924,7 @@ async def create_node(
         access_blob = access_blob or {}
         key = key or self.context.key_maker()
         data_sources = data_sources or []
+        child_path = list(await self.path_segments()) + [key]
 
         node = orm.Node(
             key=key,
@@ -934,17 +935,17 @@ async def create_node(
             access_blob=access_blob,
         )
         async with self.context.session() as db:
-            # TODO Consider using nested transitions to ensure that
-            # both the node is created (name not already taken)
-            # and the directory/file is created---or neither are.
+            # Flush the node so collisions surface before storage initialization,
+            # but do not commit it until its data sources are ready. If storage
+            # initialization fails, closing the session rolls the node back.
             try:
                 db.add(node)
-                await db.commit()
+                await db.flush()
             except IntegrityError as exc:
                 UNIQUE_CONSTRAINT_FAILED = "gkpj"
                 if exc.code == UNIQUE_CONSTRAINT_FAILED:
                     await db.rollback()
-                    raise Collision(f"/{'/'.join(await self.path_segments() + [key])}")
+                    raise Collision(f"/{'/'.join(child_path)}")
                 raise
             await db.refresh(node)
             for data_source in data_sources:
@@ -985,7 +986,7 @@ async def create_node(
                         adapter_cls.init_storage,
                         storage,
                         data_source,
-                        await self.path_segments() + [key],
+                        child_path,
                     )
                 else:
                     if data_source.mimetype not in self.context.adapters_by_mimetype:
@@ -1074,8 +1075,6 @@ async def create_node(
                 # a notification about it.
                 await self.context.streaming_cache.set(self.node.id, sequence, metadata)
             if self.context.webhook_dispatcher:
-                segments = list(await self.path_segments())
-                child_path = segments + [key]
                 await self.context.webhook_dispatcher.dispatch(
                     ContainerChildCreatedEvent(
                         timestamp=datetime.now(tz=timezone.utc),