Add logging function to HoptimatorConnection so its clients can retrieve logs by registering hooks (#154)

nguyent8 · web-flow · commit 427d2903c3c2 · 2025-08-08T09:57:45.000-07:00
Clients can retrieve logs at HoptimatorDdlExecutor level by registering hooks.
diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorConnection.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorConnection.java
@@ -3,7 +3,9 @@
 import com.linkedin.hoptimator.Database;
 import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.Source;
+import com.linkedin.hoptimator.avro.AvroConverter;
 import com.linkedin.hoptimator.util.ConnectionService;
+import com.linkedin.hoptimator.util.DelegatingConnection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Statement;
@@ -13,17 +15,14 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Properties;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
-
 import org.apache.avro.Schema;
 import org.apache.calcite.jdbc.CalciteConnection;
 import org.apache.calcite.jdbc.CalcitePrepare;
 import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.plan.RelOptMaterialization;
 import org.apache.calcite.rel.RelNode;
-
-import com.linkedin.hoptimator.avro.AvroConverter;
-import com.linkedin.hoptimator.util.DelegatingConnection;
 import org.apache.calcite.util.Util;
 
 
@@ -33,6 +32,8 @@ public class HoptimatorConnection extends DelegatingConnection {
   private final Properties connectionProperties;
   private final List<RelOptMaterialization> materializations = new ArrayList<>();
 
+  private final List<Consumer<String>> logHooks = new ArrayList<>();
+
   public HoptimatorConnection(CalciteConnection connection, Properties connectionProperties) {
     super(connection);
     this.connection = connection;
@@ -95,6 +96,21 @@ public void registerMaterialization(List<String> viewPath, String querySql) {
     registerMaterialization(viewPath, tableRel, queryRel);
   }
 
+  /**
+   * Returns a logger for a client of this connection. The logger logs to both SLF4J and hooks.
+   */
+  HoptimatorConnectionDualLogger getLogger(Class<?> clazz) {
+    return new HoptimatorConnectionDualLogger(clazz, logHooks);
+  }
+
+  /**
+   * Adds a log hook to all loggers. The hook will be called for all log messages for all statements executed regardless of thread ownership.
+   * TODO: Revise to allow hooks to be added per statement.
+   */
+  public void addLogHook(Consumer<String> hook) {
+    logHooks.add(hook);
+  }
+
   private void registerMaterialization(List<String> viewPath, RelNode tableRel, RelNode queryRel) {
     materializations.add(new RelOptMaterialization(tableRel, queryRel, null, viewPath));
   }
@@ -114,4 +130,29 @@ private static String databaseName(CalcitePrepare.Context context, List<String>
     }
     return ((Database) schema.schema).databaseName();
   }
+
+  /**
+   * A logger that logs to both SLF4J logger and registered hooks.
+   */
+  static class HoptimatorConnectionDualLogger {
+    private final String className;
+    private final org.slf4j.Logger slf4jLogger;
+    private final List<Consumer<String>> hooks;
+
+    HoptimatorConnectionDualLogger(Class<?> clazz, List<Consumer<String>> hooks) {
+      this.className = clazz.getSimpleName();
+      this.slf4jLogger = org.slf4j.LoggerFactory.getLogger(clazz);
+      this.hooks = hooks;
+    }
+
+    /**
+     * Log a message with slf4j format at the INFO level.
+     */
+    public void info(String format, Object... arguments) {
+      slf4jLogger.info(format, arguments);
+      String msg = org.slf4j.helpers.MessageFormatter.arrayFormat(format, arguments).getMessage();
+      String msgWithClassName = String.format("[%s] %s", className, msg);
+      hooks.forEach(hook -> hook.accept(msgWithClassName));
+    }
+  }
 }
diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java
@@ -74,9 +74,11 @@
 public final class HoptimatorDdlExecutor extends ServerDdlExecutor {
 
   private final HoptimatorConnection connection;
+  private final HoptimatorConnection.HoptimatorConnectionDualLogger logger;
 
   public HoptimatorDdlExecutor(HoptimatorConnection connection) {
     this.connection = connection;
+    this.logger = connection.getLogger(HoptimatorDdlExecutor.class);
   }
 
   public static final SqlParserImplFactory PARSER_FACTORY = new SqlParserImplFactory() {
@@ -98,6 +100,7 @@ public DdlExecutor getDdlExecutor() {
   /** Executes a {@code CREATE VIEW} command. */
   @Override
   public void execute(SqlCreateView create, CalcitePrepare.Context context) {
+    logger.info("Validating statement: {}", create);
     try {
       ValidationService.validateOrThrow(create);
     } catch (SQLException e) {
@@ -133,18 +136,27 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
     RelProtoDataType protoType = RelDataTypeImpl.proto(analyzed.rowType);
     ViewTable viewTable = new ViewTable(Object.class, protoType, sql, schemaPath, viewPath);
     View view = new View(viewPath, sql);
+    logger.info("Validated sql statement. The view is named {} and has path {}",
+        viewName, viewPath);
 
     Collection<Deployer> deployers = null;
     try {
+      logger.info("Validating view {} with deployers", viewName);
       ValidationService.validateOrThrow(viewTable);
       deployers = DeploymentService.deployers(view, connection);
       ValidationService.validateOrThrow(deployers);
+      logger.info("Validated view {}", viewName);
       if (create.getReplace()) {
+        logger.info("Deploying update view {}", viewName);
         DeploymentService.update(deployers);
       } else {
+        logger.info("Deploying create view {}", viewName);
         DeploymentService.create(deployers);
       }
+      logger.info("Deployed view {}", viewName);
       schemaPlus.add(viewName, viewTable);
+      logger.info("Added view {} to schema {}", viewName, schemaPlus.getName());
+      logger.info("CREATE VIEW {} completed", viewName);
     } catch (Exception e) {
       if (deployers != null) {
         DeploymentService.restore(deployers);
@@ -158,6 +170,7 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
   /** Executes a {@code CREATE MATERIALIZED VIEW} command. */
   @Override
   public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context context) {
+    logger.info("Validating statement: {}", create);
     try {
       ValidationService.validateOrThrow(create);
     } catch (SQLException e) {
@@ -214,6 +227,9 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
       MaterializedViewTable materializedViewTable = new MaterializedViewTable(viewTable);
       RelDataType viewRowType = materializedViewTable.getRowType(typeFactory);
 
+      logger.info("Validated sql statement. The view is named {} and has path {}",
+          viewName, viewPath);
+
       // Support "partial views", i.e. CREATE VIEW FOO$BAR, where the view name
       // is "foo-bar" and the sink is just FOO.
       String[] viewParts = viewName.split("\\$", 2);
@@ -222,6 +238,7 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
       if (viewParts.length > 1) {
         pipelineName = pipelineName + "-" + viewParts[1];
       }
+      logger.info("Pipeline name for view {} is {}", viewName, pipelineName);
       Properties connectionProperties = connection.connectionProperties();
       connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, pipelineName);
       List<String> sinkPath = new ArrayList<>(schemaPath);
@@ -244,17 +261,23 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
 
       // Need to add the view table to the connection so that the ConnectorService can find it when resolving options.
       schemaPlus.add(viewName, materializedViewTable);
+      logger.info("Added view {} to schema {}", viewName, schemaPlus.getName());
       Pipeline pipeline = plan.pipeline(viewName, connection);
       MaterializedView hook = new MaterializedView(database, viewPath, sql, pipeline.job().sql(), pipeline);
       // TODO support CREATE ... WITH (options...)
       ValidationService.validateOrThrow(hook);
       deployers = DeploymentService.deployers(hook, connection);
+      logger.info("Validating view {} with deployers", viewName);
       ValidationService.validateOrThrow(deployers);
+      logger.info("Validated view {}", viewName);
       if (create.getReplace()) {
+        logger.info("Deploying update view {}", viewName);
         DeploymentService.update(deployers);
       } else {
+        logger.info("Deploying create view {}", viewName);
         DeploymentService.create(deployers);
       }
+      logger.info("Deployed view {}", viewName);
     } catch (SQLException | RuntimeException e) {
       if (deployers != null) {
         DeploymentService.restore(deployers);
@@ -266,6 +289,7 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
       }
       throw new DdlException(create, e.getMessage(), e);
     }
+    logger.info("CREATE MATERIALIZED VIEW {} completed", viewName);
   }
 
   // N.B. largely copy-pasted from Apache Calcite
diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/JdbcTestBase.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/JdbcTestBase.java
@@ -34,6 +34,15 @@ protected void sql(String sql) throws SQLException {
     }
   }
 
+  protected List<String> sqlReturnsLogs(String sql) throws SQLException {
+    var logs = new ArrayList<String>();
+    ((HoptimatorConnection) conn).addLogHook(logs::add);
+    try (Statement stmt = conn.createStatement()) {
+      stmt.executeUpdate(sql);
+    }
+    return logs;
+  }
+
   protected void assertQueriesEqual(String q1, String q2) throws SQLException {
     assertResultSetsEqual(query(q1), query(q2));
   }
diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TestBasicSql.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TestBasicSql.java
@@ -2,8 +2,11 @@
 
 import java.util.Arrays;
 
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
+
 
 public class TestBasicSql extends JdbcTestBase {
 
@@ -28,4 +31,26 @@ public void insertIntoSelectFrom() throws Exception {
     sql("DROP TABLE T1");
     sql("DROP TABLE T2");
   }
+
+  @Test
+  public void createView() throws Exception {
+    sql("CREATE TABLE T (X VARCHAR, Y VARCHAR)");
+    sql("INSERT INTO T VALUES ('one', 'two')");
+    assertQueriesEqual("SELECT * FROM T", "VALUES ('one', 'two')");
+    var logs = sqlReturnsLogs("CREATE VIEW V AS SELECT * FROM T");
+    assertResultSetsEqual(query("SELECT * FROM V"), query("SELECT * FROM T"));
+    sql("DROP VIEW V");
+    sql("DROP TABLE T");
+
+    var expectedLogs = List.of(
+        "[HoptimatorDdlExecutor] Validating statement: CREATE VIEW `V` AS\nSELECT *\nFROM `T`",
+        "[HoptimatorDdlExecutor] Validated sql statement. The view is named V and has path [DEFAULT, V]",
+        "[HoptimatorDdlExecutor] Validating view V with deployers",
+        "[HoptimatorDdlExecutor] Validated view V",
+        "[HoptimatorDdlExecutor] Deploying create view V",
+        "[HoptimatorDdlExecutor] Deployed view V",
+        "[HoptimatorDdlExecutor] Added view V to schema DEFAULT",
+        "[HoptimatorDdlExecutor] CREATE VIEW V completed");
+    Assertions.assertEquals(expectedLogs, logs);
+  }
 }

Original file line number	Diff line number	Diff line change
`@@ -34,6 +34,15 @@ protected void sql(String sql) throws SQLException {`
`34`	`34`	`}`
`35`	`35`	`}`
`36`	`36`
	`37`	`+ protected List<String> sqlReturnsLogs(String sql) throws SQLException {`
	`38`	`+ var logs = new ArrayList<String>();`
	`39`	`+ ((HoptimatorConnection) conn).addLogHook(logs::add);`
	`40`	`+ try (Statement stmt = conn.createStatement()) {`
	`41`	`+ stmt.executeUpdate(sql);`
	`42`	`+ }`
	`43`	`+ return logs;`
	`44`	`+ }`
	`45`	`+`
`37`	`46`	`protected void assertQueriesEqual(String q1, String q2) throws SQLException {`
`38`	`47`	`assertResultSetsEqual(query(q1), query(q2));`
`39`	`48`	`}`