Pre-delete dependency guard for DROP TABLE

hoptimator-api/build.gradle

@@ -4,7 +4,8 @@ plugins {
 }
 
 dependencies {
-  // plz keep it this way
+  // This package should have minimal dependencies
+  compileOnly 'com.google.code.findbugs:jsr305:3.0.2'
 }
 
 publishing {

hoptimator-api/src/main/java/com/linkedin/hoptimator/PendingDelete.java

@@ -0,0 +1,55 @@
+package com.linkedin.hoptimator;
+
+import java.util.Objects;
+import javax.annotation.Nullable;
+
+
+/**
+ * A type-tagged wrapper signaling that {@code target} is about to be deleted. Pre-delete
+ * validators (e.g. dependency guards that block DROP TABLE when a pipeline still references
+ * the resource) key off this wrapper rather than the raw target type — so an unrelated future
+ * caller of {@code ValidationService.validateOrThrow(source, connection)} doesn't accidentally
+ * trigger delete-intent checks.
+ *
+ * <p>An optional {@code (selfOwnerKind, selfOwnerName)} lets the caller declare an "umbrella"
+ * K8s resource whose owned objects should be excluded from the dependent set — e.g. a
+ * LogicalTable CRD, so its child Pipeline CRDs (which reference tier sources by SQL) don't
+ * self-block the drop.
+ */
+public final class PendingDelete<T> {
+
+  private final T target;
+  private final String selfOwnerKind;
+  private final String selfOwnerName;
+
+  public PendingDelete(T target) {
+    this(target, null, null);
+  }
+
+  public PendingDelete(T target, @Nullable String selfOwnerKind, @Nullable String selfOwnerName) {
+    this.target = Objects.requireNonNull(target, "target");
+    this.selfOwnerKind = selfOwnerKind;
+    this.selfOwnerName = selfOwnerName;
+  }
+
+  public T target() {
+    return target;
+  }
+
+  /** Kind of the K8s resource whose owned objects should be excluded from the dependent set. */
+  public @Nullable String selfOwnerKind() {
+    return selfOwnerKind;
+  }
+
+  /** Name of the K8s resource whose owned objects should be excluded from the dependent set. */
+  public @Nullable String selfOwnerName() {
+    return selfOwnerName;
+  }
+
+  @Override
+  public String toString() {
+    String self = (selfOwnerKind != null && selfOwnerName != null)
+        ? ", self=" + selfOwnerKind + "/" + selfOwnerName : "";
+    return "PendingDelete[" + target + self + "]";
+  }
+}

hoptimator-api/src/main/java/com/linkedin/hoptimator/Validated.java

@@ -1,6 +1,14 @@
 package com.linkedin.hoptimator;
 
+import java.sql.Connection;
+
+
 public interface Validated {
 
-  void validate(Validator.Issues issues);
+  /**
+   * Validates {@code this}, recording any problems in {@code issues}. The connection is always
+   * supplied so validators can run lookups against external systems (e.g. pre-delete dependency
+   * checks).
+   */
+  void validate(Validator.Issues issues, Connection connection);
 }

hoptimator-api/src/main/java/com/linkedin/hoptimator/Validator.java

@@ -1,5 +1,6 @@
 package com.linkedin.hoptimator;
 
+import java.sql.Connection;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -52,8 +53,8 @@ public DefaultValidator(T t) {
     }
 
     @Override
-    public void validate(Issues issues) {
-      t.validate(issues.child(t.getClass().getSimpleName()));
+    public void validate(Issues issues, Connection connection) {
+      t.validate(issues.child(t.getClass().getSimpleName()), connection);
     }
   }
 

hoptimator-api/src/main/java/com/linkedin/hoptimator/ValidatorProvider.java

@@ -1,9 +1,13 @@
 package com.linkedin.hoptimator;
 
+import java.sql.Connection;
 import java.util.Collection;
 
 
 public interface ValidatorProvider {
 
-  <T> Collection<Validator> validators(T obj);
+  /**
+   * Returns validators that should be applied to {@code obj}.
+   */
+  <T> Collection<Validator> validators(T obj, Connection connection);
 }

hoptimator-api/src/test/java/com/linkedin/hoptimator/PendingDeleteTest.java

@@ -0,0 +1,85 @@
+package com.linkedin.hoptimator;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+class PendingDeleteTest {
+
+  @Test
+  void singleArgConstructorLeavesSelfOwnerNull() {
+    Object target = new Object();
+    PendingDelete<Object> pd = new PendingDelete<>(target);
+
+    assertSame(target, pd.target());
+    assertNull(pd.selfOwnerKind());
+    assertNull(pd.selfOwnerName());
+  }
+
+  @Test
+  void threeArgConstructorStoresSelfOwner() {
+    Object target = new Object();
+    PendingDelete<Object> pd = new PendingDelete<>(target, "LogicalTable", "my-table");
+
+    assertSame(target, pd.target());
+    assertEquals("LogicalTable", pd.selfOwnerKind());
+    assertEquals("my-table", pd.selfOwnerName());
+  }
+
+  @Test
+  void threeArgConstructorAcceptsNullSelfOwner() {
+    Object target = new Object();
+    PendingDelete<Object> pd = new PendingDelete<>(target, null, null);
+
+    assertNull(pd.selfOwnerKind());
+    assertNull(pd.selfOwnerName());
+  }
+
+  @Test
+  void nullTargetThrows() {
+    assertThrows(NullPointerException.class, () -> new PendingDelete<>(null));
+    assertThrows(NullPointerException.class,
+        () -> new PendingDelete<>(null, "LogicalTable", "my-table"));
+  }
+
+  @Test
+  void toStringIncludesTargetAndSelfOwnerWhenPresent() {
+    PendingDelete<String> pd = new PendingDelete<>("the-target", "LogicalTable", "my-table");
+    String s = pd.toString();
+    assertTrue(s.contains("the-target"), "toString should include target: " + s);
+    assertTrue(s.contains("LogicalTable/my-table"), "toString should include kind/name: " + s);
+  }
+
+  @Test
+  void toStringOmitsSelfOwnerWhenNull() {
+    PendingDelete<String> pd = new PendingDelete<>("the-target");
+    String s = pd.toString();
+    assertTrue(s.contains("the-target"));
+    assertFalse(s.contains("self="), "toString should not include self= when not set: " + s);
+  }
+
+  @Test
+  void toStringOmitsSelfOwnerWhenOnlyOneFieldSet() {
+    // The toString contract says self= appears only when both kind and name are non-null.
+    // (The K8sPipelineDependencyChecker.isSelfOwned guard also requires both.)
+    PendingDelete<String> kindOnly = new PendingDelete<>("t", "LogicalTable", null);
+    assertFalse(kindOnly.toString().contains("self="));
+
+    PendingDelete<String> nameOnly = new PendingDelete<>("t", null, "my-table");
+    assertFalse(nameOnly.toString().contains("self="));
+  }
+
+  @Test
+  void targetGenericTypeIsPreserved() {
+    Source source = new Source("db", java.util.List.of("schema", "tbl"), java.util.Map.of());
+    PendingDelete<Source> pd = new PendingDelete<>(source);
+    Source unwrapped = pd.target();
+    assertEquals("tbl", unwrapped.table());
+  }
+}

hoptimator-api/src/test/java/com/linkedin/hoptimator/ValidatorTest.java

@@ -184,10 +184,10 @@ void testCheckClosedMessageContainsPath() {
 
   @Test
   void testDefaultValidatorDelegatesToTarget() {
-    Validated target = issues -> issues.error("target error");
+    Validated target = (issues, conn) -> issues.error("target error");
     Validator.DefaultValidator<Validated> validator = new Validator.DefaultValidator<>(target);
     Validator.Issues issues = new Validator.Issues("root");
-    validator.validate(issues);
+    validator.validate(issues, null);
     assertFalse(issues.valid());
   }
 

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroTableValidator.java

@@ -17,6 +17,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.sql.Connection;
 
 
 /** Validates that tables follow Avro schema evolution rules.  */
@@ -29,7 +30,7 @@ class AvroTableValidator implements Validator {
   }
 
   @Override
-  public void validate(Issues issues) {
+  public void validate(Issues issues, Connection connection) {
     try {
       CalciteSchema originalSchema = schema.unwrap(CalciteSchema.class);
       if (originalSchema == null || originalSchema.schema == null) {

hoptimator-avro/src/main/java/com/linkedin/hoptimator/avro/AvroValidatorProvider.java

@@ -4,6 +4,7 @@
 import com.linkedin.hoptimator.ValidatorProvider;
 import org.apache.calcite.schema.SchemaPlus;
 
+import java.sql.Connection;
 import java.util.Collection;
 import java.util.Collections;
 
@@ -12,7 +13,7 @@
 public class AvroValidatorProvider implements ValidatorProvider {
 
   @Override
-  public <T> Collection<Validator> validators(T obj) {
+  public <T> Collection<Validator> validators(T obj, Connection connection) {
     if (obj instanceof SchemaPlus) {
       return Collections.singletonList(new AvroTableValidator((SchemaPlus) obj));
     } else {

hoptimator-avro/src/test/java/com/linkedin/hoptimator/avro/AvroTableValidatorTest.java

@@ -41,7 +41,7 @@ void testValidateCatchesClassCastExceptionSilently() {
 
     AvroTableValidator validator = new AvroTableValidator(schema);
     Validator.Issues issues = new Validator.Issues("test");
-    validator.validate(issues);
+    validator.validate(issues, null);
 
     assertTrue(issues.valid(), "ClassCastException should be silently caught");
   }
@@ -53,7 +53,7 @@ void testValidateThrowsForNullOriginalSchema() {
     AvroTableValidator validator = new AvroTableValidator(schema);
     Validator.Issues issues = new Validator.Issues("test");
 
-    assertThrows(IllegalArgumentException.class, () -> validator.validate(issues));
+    assertThrows(IllegalArgumentException.class, () -> validator.validate(issues, null));
   }
 
   @Test
@@ -99,7 +99,7 @@ protected Map<String, Table> getTableMap() {
 
     AvroTableValidator validator = new AvroTableValidator(schema);
     Validator.Issues issues = new Validator.Issues("root");
-    validator.validate(issues);
+    validator.validate(issues, null);
 
     assertFalse(issues.valid(),
         "Incompatible schema (INT→VARCHAR) should produce validation errors");
@@ -137,7 +137,7 @@ protected Map<String, Table> getTableMap() {
 
     AvroTableValidator validator = new AvroTableValidator(schema);
     Validator.Issues issues = new Validator.Issues("root");
-    validator.validate(issues);
+    validator.validate(issues, null);
 
     assertTrue(issues.valid(), "Compatible schemas should pass validation without errors");
   }

hoptimator-avro/src/test/java/com/linkedin/hoptimator/avro/AvroValidatorProviderTest.java

@@ -24,7 +24,7 @@ class AvroValidatorProviderTest {
   void testValidatorsReturnsAvroTableValidatorForSchemaPlus() {
     AvroValidatorProvider provider = new AvroValidatorProvider();
 
-    Collection<Validator> validators = provider.validators(schemaPlus);
+    Collection<Validator> validators = provider.validators(schemaPlus, null);
 
     assertEquals(1, validators.size());
     assertInstanceOf(AvroTableValidator.class, validators.iterator().next());
@@ -34,7 +34,7 @@ void testValidatorsReturnsAvroTableValidatorForSchemaPlus() {
   void testValidatorsReturnsEmptyForNonSchemaPlus() {
     AvroValidatorProvider provider = new AvroValidatorProvider();
 
-    Collection<Validator> validators = provider.validators("not-a-schema");
+    Collection<Validator> validators = provider.validators("not-a-schema", null);
 
     assertTrue(validators.isEmpty());
   }
@@ -43,7 +43,7 @@ void testValidatorsReturnsEmptyForNonSchemaPlus() {
   void testValidatorsReturnsEmptyForNull() {
     AvroValidatorProvider provider = new AvroValidatorProvider();
 
-    Collection<Validator> validators = provider.validators(null);
+    Collection<Validator> validators = provider.validators(null, null);
 
     assertTrue(validators.isEmpty());
   }

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/CompatibilityValidatorBase.java

@@ -6,6 +6,8 @@
 import org.apache.calcite.schema.Table;
 import org.apache.calcite.schema.lookup.LikePattern;
 
+import java.sql.Connection;
+
 
 /** Base class for shared schema evolution validators.  */
 abstract class CompatibilityValidatorBase implements Validator {
@@ -17,7 +19,7 @@ abstract class CompatibilityValidatorBase implements Validator {
   }
 
   @Override
-  public void validate(Issues issues) {
+  public void validate(Issues issues, Connection connection) {
     try {
       CalciteSchema originalSchema = schema.unwrap(CalciteSchema.class);
       if (originalSchema == null || originalSchema.schema == null) {

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/CompatibilityValidatorProvider.java

@@ -4,6 +4,7 @@
 import com.linkedin.hoptimator.ValidatorProvider;
 import org.apache.calcite.schema.SchemaPlus;
 
+import java.sql.Connection;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -13,7 +14,7 @@
 public class CompatibilityValidatorProvider implements ValidatorProvider {
 
   @Override
-  public <T> Collection<Validator> validators(T obj) {
+  public <T> Collection<Validator> validators(T obj, Connection connection) {
     if (obj instanceof SchemaPlus) {
       return Arrays.asList(new Validator[]{new BackwardCompatibilityValidator((SchemaPlus) obj),
           new ForwardCompatibilityValidator((SchemaPlus) obj)});

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DefaultValidatorProvider.java

@@ -4,6 +4,7 @@
 import com.linkedin.hoptimator.Validator;
 import com.linkedin.hoptimator.ValidatorProvider;
 
+import java.sql.Connection;
 import java.util.Collection;
 import java.util.Collections;
 
@@ -12,7 +13,7 @@
 public class DefaultValidatorProvider implements ValidatorProvider {
 
   @Override
-  public <T> Collection<Validator> validators(T obj) {
+  public <T> Collection<Validator> validators(T obj, Connection connection) {
     if (obj instanceof Validated) {
       return Collections.singletonList(new Validator.DefaultValidator<>((Validated) obj));
     } else {