Validate query engine presence when table format engine is configured (#1491)

Author: ferenc-csaky

sqrl-cli/src/main/java/com/datasqrl/cli/AbstractCmd.java

@@ -20,6 +20,7 @@
 import com.datasqrl.error.ErrorCollector;
 import com.datasqrl.error.ErrorPrinter;
 import com.datasqrl.util.OsProcessManager;
+import com.google.inject.ProvisionException;
 import java.nio.file.Path;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.SneakyThrows;
@@ -47,10 +48,14 @@ public void run() {
     try {
       runInternal(collector);
       cli.statusHook.onSuccess(collector);
-    } catch (CollectedException e) {
-      if (e.isInternalError()) e.printStackTrace();
-      e.printStackTrace();
-      cli.statusHook.onFailure(e, collector);
+
+    } catch (ProvisionException | CollectedException e) {
+      var ce = unwrapCollectedException(e);
+      if (ce.isInternalError()) {
+        ce.printStackTrace();
+      }
+      cli.statusHook.onFailure(ce, collector);
+
     } catch (Throwable e) { // unknown exception
       collector.getCatcher().handle(e);
       e.printStackTrace();
@@ -90,4 +95,28 @@ protected Path getTargetDir() {
   public int getExitCode() {
     return exitCode.get();
   }
+
+  /**
+   * Unwraps {@link CollectedException} from Guice {@link ProvisionException} wrappers to provide
+   * clean error messages.
+   *
+   * <p>Validation errors during dependency injection (e.g., in pipeline configuration) are thrown
+   * as {@link CollectedException} with clear messages. Guice wraps them in {@link
+   * ProvisionException}, obscuring the original error with verbose DI stack traces.
+   *
+   * @param e the runtime exception that may be a CollectedException or contain one as a cause
+   * @return the unwrapped CollectedException
+   * @throws RuntimeException the original exception if it doesn't contain a CollectedException
+   */
+  private CollectedException unwrapCollectedException(RuntimeException e) {
+    if (e instanceof CollectedException ce) {
+      return ce;
+    }
+
+    if (e.getCause() instanceof CollectedException ce) {
+      return ce;
+    }
+
+    throw e;
+  }
 }

sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java

@@ -62,7 +62,7 @@ public Pair<PhysicalPlan, TestPlan> executeCompilation(Optional<Path> testsPath)
     var environment =
         new Sqrl2FlinkSQLTranslator(
             buildPath,
-            (FlinkStreamEngine) planner.getStreamStage().getEngine(),
+            (FlinkStreamEngine) planner.getStreamStage().engine(),
             config.getCompilerConfig());
     planner.planMain(mainScript, environment);
     var dagBuilder = planner.getDagBuilder();

sqrl-cli/src/main/java/com/datasqrl/packager/Packager.java

@@ -291,7 +291,7 @@ public void postprocess(Path targetDir, PhysicalPlan plan, TestPlan testPlan) {
     // We'll write a single asset for each folder in the physical plan stage, plus any deployment
     // artifacts that the plan has
     for (PhysicalStagePlan stagePlan : plan.getStagePlans()) {
-      writePlan(stagePlan.getStage().getName(), stagePlan.getPlan(), planDir);
+      writePlan(stagePlan.getStage().name(), stagePlan.getPlan(), planDir);
     }
 
     if (testPlan != null) {

sqrl-planner/src/main/java/com/datasqrl/config/PipelineFactory.java

@@ -17,7 +17,6 @@
 
 import com.datasqrl.cli.EngineIds;
 import com.datasqrl.engine.ExecutionEngine;
-import com.datasqrl.engine.IExecutionEngine;
 import com.datasqrl.engine.database.DatabaseEngine;
 import com.datasqrl.engine.database.QueryEngine;
 import com.datasqrl.engine.export.PrintEngineFactory;
@@ -35,13 +34,8 @@
 import java.util.Set;
 import lombok.Getter;
 import lombok.NonNull;
-import org.apache.commons.lang3.tuple.Pair;
 
-/**
- * Configuration for the engines
- *
- * <p>
- */
+/** Configuration for the engines */
 public class PipelineFactory {
 
   public static final List<String> defaultEngines = List.of(PrintEngineFactory.NAME);
@@ -59,18 +53,28 @@ public PipelineFactory(
     this.engineConfig = engineConfig;
   }
 
+  public ExecutionPipeline createPipeline() {
+    var errorCollector = injector.getInstance(ErrorCollector.class);
+    return SimplePipeline.of(getEngines(), errorCollector);
+  }
+
+  public Map<String, ExecutionEngine> getEngines() {
+    return getEngines(Optional.empty());
+  }
+
   private Map<String, ExecutionEngine> getEngines(Optional<EngineType> engineType) {
     Map<String, ExecutionEngine> engines = new HashMap<>();
     Set<String> allEngines = new HashSet<>(enabledEngines);
     allEngines.addAll(defaultEngines);
+
     for (String engineId : allEngines) {
       if (engineId.equalsIgnoreCase(EngineIds.TEST)) {
         continue;
       }
       var engineFactory =
           ServiceLoaderDiscovery.get(EngineFactory.class, EngineFactory::getEngineName, engineId);
 
-      IExecutionEngine engine = injector.getInstance(engineFactory.getFactoryClass());
+      var engine = injector.getInstance(engineFactory.getFactoryClass());
       if (engineType.map(type -> engine.getType() == type).orElse(true)) {
         engines.put(engineId, (ExecutionEngine) engine);
       }
@@ -86,25 +90,4 @@ private Map<String, ExecutionEngine> getEngines(Optional<EngineType> engineType)
                     .forEach(databaseEngine::addQueryEngine));
     return engines;
   }
-
-  public Map<String, ExecutionEngine> getEngines() {
-    return getEngines(Optional.empty());
-  }
-
-  public Pair<String, ExecutionEngine> getEngine(EngineType type) {
-    var engines = getEngines(Optional.of(type));
-    // Todo: error collector
-    var errors = ErrorCollector.root();
-    errors.checkFatal(
-        !engines.isEmpty(), "Need to configure a %s engine", type.name().toLowerCase());
-    errors.checkFatal(
-        engines.size() == 1,
-        "Currently support only a single %s engine",
-        type.name().toLowerCase());
-    return Pair.of(engines.entrySet().iterator().next());
-  }
-
-  public ExecutionPipeline createPipeline() {
-    return SimplePipeline.of(getEngines(), /*todo*/ ErrorCollector.root());
-  }
 }

sqrl-planner/src/main/java/com/datasqrl/engine/pipeline/EngineStage.java

@@ -17,22 +17,11 @@
 
 import com.datasqrl.engine.EngineFeature;
 import com.datasqrl.engine.ExecutionEngine;
-import lombok.Value;
 
-@Value
-public class EngineStage implements ExecutionStage {
-
-  String name;
-  ExecutionEngine engine;
+public record EngineStage(String name, ExecutionEngine engine) implements ExecutionStage {
 
   @Override
   public boolean supportsFeature(EngineFeature capability) {
     return engine.supports(capability);
   }
-
-  //  @Override
-  //  public boolean supportsFunction(FunctionDefinition function) {
-  //    return engine.supports(function);
-  //  }
-
 }

sqrl-planner/src/main/java/com/datasqrl/engine/pipeline/ExecutionPipeline.java

@@ -26,10 +26,10 @@
 
 public interface ExecutionPipeline {
 
-  List<ExecutionStage> getStages();
+  List<ExecutionStage> stages();
 
   default List<ExecutionStage> getReadStages() {
-    return getStages().stream().filter(ExecutionStage::isRead).collect(Collectors.toList());
+    return stages().stream().filter(ExecutionStage::isRead).collect(Collectors.toList());
   }
 
   /**
@@ -40,12 +40,12 @@ default List<ExecutionStage> getReadStages() {
   default Optional<ServerEngine> getServerEngine() {
     return StreamUtil.getOnlyElement(
         getStagesByType(EngineType.SERVER).stream()
-            .map(stage -> (ServerEngine) stage.getEngine())
+            .map(stage -> (ServerEngine) stage.engine())
             .distinct());
   }
 
   default boolean hasReadStages() {
-    return getStages().stream().anyMatch(ExecutionStage::isRead);
+    return stages().stream().anyMatch(ExecutionStage::isRead);
   }
 
   Set<ExecutionStage> getUpStreamFrom(ExecutionStage stage);
@@ -54,18 +54,18 @@ default boolean hasReadStages() {
 
   default Optional<ExecutionStage> getStage(String name) {
     return StreamUtil.getOnlyElement(
-        getStages().stream().filter(s -> s.getName().equalsIgnoreCase(name)));
+        stages().stream().filter(s -> s.name().equalsIgnoreCase(name)));
   }
 
   default Optional<ExecutionStage> getMutationStage() {
     return StreamUtil.getOnlyElement(
         getStagesByType(EngineType.LOG).stream()
-            .filter(stage -> stage.getEngine().supports(EngineFeature.MUTATIONS)));
+            .filter(stage -> stage.engine().supports(EngineFeature.MUTATIONS)));
   }
 
   default Optional<ExecutionStage> getStageByType(EngineType type) {
     return StreamUtil.getOnlyElement(
-        getStages().stream().filter(s -> s.getEngine().getType().equals(type)));
+        stages().stream().filter(s -> s.engine().getType().equals(type)));
   }
 
   /**
@@ -77,8 +77,8 @@ default Optional<ExecutionStage> getStageByType(EngineType type) {
    * @return the stage for a given {@link EngineType}.
    */
   default List<ExecutionStage> getStagesByType(EngineType type) {
-    return getStages().stream()
-        .filter(s -> s.getEngine().getType().equals(type))
+    return stages().stream()
+        .filter(s -> s.engine().getType().equals(type))
         .collect(Collectors.toList());
   }
 }

sqrl-planner/src/main/java/com/datasqrl/engine/pipeline/ExecutionStage.java

@@ -23,33 +23,31 @@
 
 public interface ExecutionStage {
 
-  String getName();
+  String name();
 
   default boolean supportsAllFeatures(Collection<EngineFeature> capabilities) {
     return capabilities.stream().allMatch(this::supportsFeature);
   }
 
   default EngineType getType() {
-    return getEngine().getType();
+    return engine().getType();
   }
 
   boolean supportsFeature(EngineFeature capability);
 
-  //  boolean supportsFunction(FunctionDefinition function);
-
   default boolean isRead() {
-    return getEngine().getType().isRead();
+    return engine().getType().isRead();
   }
 
   default boolean isWrite() {
-    return getEngine().getType().isWrite();
+    return engine().getType().isWrite();
   }
 
   default boolean isCompute() {
-    return getEngine().getType().isCompute();
+    return engine().getType().isCompute();
   }
 
-  ExecutionEngine getEngine();
+  ExecutionEngine engine();
 
   default boolean supportsFunction(SqlOperator operator) {
     return true;