[Hackathon] feat: AI-augmented macro operators

amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionCoordinator.scala

@@ -573,7 +573,14 @@ class RegionExecutionCoordinator(
         val schemaOptional =
           region.getOperator(outputPortId.opId).outputPorts(outputPortId.portId)._3
         val schema =
-          schemaOptional.getOrElse(throw new IllegalStateException("Schema is missing"))
+          schemaOptional.getOrElse(
+            throw new IllegalStateException(
+              s"Schema is missing for output port: opId=${outputPortId.opId.logicalOpId.id} " +
+                s"layer=${outputPortId.opId.layerName} " +
+                s"portId=${outputPortId.portId} " +
+                s"isInternal=${outputPortId.portId.internal}"
+            )
+          )
         DocumentFactory.createDocument(storageUriToAdd, schema)
         if (!isRestart) {
           WorkflowExecutionsResource.insertOperatorPortResultUri(

amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/WorkflowExecutionCoordinator.scala

@@ -67,7 +67,14 @@ class WorkflowExecutionCoordinator(
       regionExecutionCoordinators.values.filter(!_.isCompleted).toSeq
 
     // Trigger sync for each unfinished region.
-    unfinishedRegionCoordinators.foreach(_.syncStatusAndTransitionRegionExecutionPhase())
+    // IMPORTANT: capture the sync futures so any exception thrown during phase
+    // transition (e.g. "Schema is missing" in createOutputPortStorageObjects)
+    // propagates out as a Future.exception. Previously `.foreach(...)` swallowed
+    // the returned Future, which meant phase-transition failures were
+    // discarded and the region appeared to hang silently instead of failing
+    // with a FatalError visible in the client UI.
+    val syncFutures =
+      unfinishedRegionCoordinators.map(_.syncStatusAndTransitionRegionExecutionPhase())
 
     // Wait only for region termination futures (kill path), then re-run coordination.
     val terminationFutures = unfinishedRegionCoordinators.flatMap(_.getTerminationFutureOpt)
@@ -80,7 +87,12 @@ class WorkflowExecutionCoordinator(
 
     if (regionExecutionCoordinators.values.exists(!_.isCompleted)) {
       // Some regions are still not completed yet. Cannot start the new regions.
-      return Future.Unit
+      // But before returning success, wait on the syncFutures so any
+      // transition-phase failure (e.g. "Schema is missing") makes it out
+      // of this method as a Future.exception — PortCompletedHandler's
+      // .onFailure handler will then turn it into a FatalError on the
+      // client. Without this, the failure was being swallowed in foreach.
+      return Future.collect(syncFutures).unit
     }
 
     // All existing regions are completed. Start the next region (if any).

amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala

@@ -47,6 +47,7 @@ import org.apache.texera.web.resource.dashboard.user.project.{
 }
 import org.apache.texera.web.resource.dashboard.user.quota.UserQuotaResource
 import org.apache.texera.web.resource.dashboard.user.workflow.{
+  MacroResource,
   WorkflowAccessResource,
   WorkflowExecutionsResource,
   WorkflowResource,
@@ -148,6 +149,7 @@ class TexeraWebApplication
     environment.jersey.register(classOf[PublicProjectResource])
     environment.jersey.register(classOf[WorkflowAccessResource])
     environment.jersey.register(classOf[WorkflowResource])
+    environment.jersey.register(classOf[MacroResource])
     environment.jersey.register(classOf[HubResource])
     environment.jersey.register(classOf[UserResource])
     environment.jersey.register(classOf[WorkflowVersionResource])

amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala

@@ -50,6 +50,7 @@ import org.apache.texera.dao.SqlServer
 import org.apache.texera.dao.jooq.generated.Tables.OPERATOR_EXECUTIONS
 import org.apache.texera.web.model.websocket.request.{LogicalPlanPojo, WorkflowExecuteRequest}
 import org.apache.texera.workflow.{LogicalLink, WorkflowCompiler}
+import org.apache.texera.workflow.macroOp.DbMacroRegistry
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowExecutionsResource
 import org.apache.texera.web.service.{ExecutionResultService, WorkflowService}
 import org.apache.texera.web.storage.ExecutionStateStore.updateWorkflowState
@@ -894,7 +895,7 @@ class SyncExecutionResource extends LazyLogging {
   ): Map[String, String] = {
     try {
       val tempContext = new WorkflowContext(WorkflowIdentity(workflowId))
-      val compiler = new WorkflowCompiler(tempContext)
+      val compiler = new WorkflowCompiler(tempContext, new DbMacroRegistry())
       compiler.compile(logicalPlan)
       Map.empty
     } catch {

amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala

@@ -262,7 +262,7 @@ object HubResource {
     }
 
     val records = baseWorkflowSelect()
-      .where(WORKFLOW.WID.in(wids: _*))
+      .and(WORKFLOW.WID.in(wids: _*))
       .groupBy(
         WORKFLOW.WID,
         WORKFLOW.NAME,