Merge pull request #1 from feldera/feldera

[feldera] first commit: add preliminary support

Author: abhizer

pom.xml

@@ -329,8 +329,13 @@
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
-      <artifactId> slf4j-simple</artifactId>
-      <version>2.0.6</version>
+      <artifactId>slf4j-api</artifactId>
+      <version>2.0.16</version>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-simple</artifactId>
+      <version>2.0.16</version>
     </dependency>
     <dependency>
       <groupId>ru.yandex.clickhouse</groupId>
@@ -378,6 +383,12 @@
       <artifactId>flight-sql-jdbc-driver</artifactId>
       <version>16.1.0</version>
     </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.13.1</version>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
   <reporting>
     <plugins>

src/sqlancer/Main.java

@@ -33,6 +33,7 @@
 import sqlancer.databend.DatabendProvider;
 import sqlancer.doris.DorisProvider;
 import sqlancer.duckdb.DuckDBProvider;
+import sqlancer.feldera.FelderaProvider;
 import sqlancer.h2.H2Provider;
 import sqlancer.hsqldb.HSQLDBProvider;
 import sqlancer.mariadb.MariaDBProvider;
@@ -734,6 +735,7 @@ private static void checkForIssue799(List<DatabaseProvider<?, ?, ?>> providers)
             providers.add(new DatabendProvider());
             providers.add(new DorisProvider());
             providers.add(new DuckDBProvider());
+            providers.add(new FelderaProvider());
             providers.add(new H2Provider());
             providers.add(new HSQLDBProvider());
             providers.add(new MariaDBProvider());

src/sqlancer/feldera/FelderaConnection.java

@@ -0,0 +1,90 @@
+package sqlancer.feldera;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import sqlancer.SQLancerDBConnection;
+import sqlancer.feldera.client.FelderaClient;
+import sqlancer.feldera.client.FelderaPipeline;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class FelderaConnection implements SQLancerDBConnection {
+    private final FelderaClient client;
+    private final String pipelineName;
+    private final List<String> inserts;
+    private boolean ready = false;
+    private String ddl;
+
+    public FelderaConnection(String url, String pipelineName) {
+        this.client = new FelderaClient(url);
+        this.pipelineName = pipelineName;
+        this.inserts = new ArrayList<>();
+        this.ddl = "";
+    }
+
+    public void prepare() throws Exception {
+        if (!ready) {
+            ObjectMapper mapper = new ObjectMapper();
+            ObjectNode node = mapper.createObjectNode();
+
+            node.put("name", this.pipelineName);
+            node.put("description", "sqlancerTest");
+            String ddlWithInserts = ddl + "--" + String.join("\n--", inserts);
+            node.put("program_code", ddlWithInserts);
+            node.putObject("runtime_config");
+            node.putObject("program_config");
+
+            this.client.createPipeline(pipelineName, node.toString());
+            this.client.start(this.pipelineName);
+
+            for (String insert : this.inserts) {
+                this.client.exec(this.pipelineName, insert);
+            }
+
+            ready = true;
+        }
+    }
+
+    public String getPipelineName() {
+        return this.pipelineName;
+    }
+
+    public FelderaClient getClient() {
+        return this.client;
+    }
+
+    @Override
+    public String getDatabaseVersion() throws Exception {
+        int x = this.client.getPipeline(this.pipelineName).getVersion();
+        return Integer.toString(x);
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (ready) {
+            this.client.shutdown(pipelineName);
+        }
+    }
+
+    public FelderaPipeline get() throws Exception {
+        return this.client.getPipeline(pipelineName);
+    }
+
+    public void buffer(String query) throws Exception {
+        if (query.startsWith("INSERT")) {
+            this.inserts.add(query);
+        } else {
+            this.ddl += query;
+        }
+    }
+
+    public Map<String, Object> execute(String query) throws Exception {
+        return this.client.exec(this.pipelineName, query);
+    }
+
+    public void shutdown() throws Exception {
+        this.client.shutdown(this.pipelineName);
+    }
+}

src/sqlancer/feldera/FelderaExpectedError.java

@@ -0,0 +1,23 @@
+package sqlancer.feldera;
+
+import sqlancer.common.query.ExpectedErrors;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FelderaExpectedError {
+    private FelderaExpectedError() {
+    }
+
+    public static List<String> getExpectedErrors() {
+        List<String> errors = new ArrayList<>();
+        errors.add("panic message:");
+        return errors;
+    }
+
+    public static ExpectedErrors expectedErrors() {
+        ExpectedErrors res = new ExpectedErrors();
+        res.addAll(getExpectedErrors());
+        return res;
+    }
+}

src/sqlancer/feldera/FelderaGlobalState.java

@@ -0,0 +1,57 @@
+package sqlancer.feldera;
+
+import sqlancer.ExecutionTimer;
+import sqlancer.GlobalState;
+import sqlancer.common.query.Query;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FelderaGlobalState extends GlobalState<FelderaOptions, FelderaSchema, FelderaConnection> {
+    List<String> views = new ArrayList<>();
+
+    @Override
+    protected void executeEpilogue(Query<?> q, boolean success, ExecutionTimer timer) throws Exception {
+        boolean logExecutionTime = getOptions().logExecutionTime();
+        if (success && getOptions().printSucceedingStatements()) {
+            System.out.println(q.getQueryString());
+        }
+        if (logExecutionTime) {
+            getLogger().writeCurrent(" -- " + timer.end().asString());
+        }
+        if (q.couldAffectSchema()) {
+            updateSchema();
+        }
+    }
+
+    @Override
+    public void updateSchema() {
+        ; // do nothing
+    }
+
+    @Override
+    public FelderaSchema getSchema() {
+        return super.getSchema();
+    }
+
+    @Override
+    protected FelderaSchema readSchema() throws Exception {
+        return FelderaSchema.fromConnection(getConnection());
+    }
+
+    public void addTable(FelderaSchema.FelderaTable table) {
+        FelderaSchema sch = getSchema();
+        if (sch == null) {
+            sch = new FelderaSchema(getConnection().getPipelineName());
+        }
+        setSchema(sch.addTable(table));
+    }
+
+    public void addView(String view) {
+        this.views.add(view);
+    }
+
+    public List<String> getViews() {
+        return this.views;
+    }
+}

src/sqlancer/feldera/FelderaLoggableFactory.java

@@ -0,0 +1,54 @@
+package sqlancer.feldera;
+
+import sqlancer.common.log.Loggable;
+import sqlancer.common.log.LoggableFactory;
+import sqlancer.common.log.LoggedString;
+import sqlancer.common.query.ExpectedErrors;
+import sqlancer.common.query.Query;
+import sqlancer.feldera.query.FelderaOtherQuery;
+import sqlancer.feldera.query.FelderaQueryAdapter;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+public class FelderaLoggableFactory extends LoggableFactory {
+    @Override
+    protected Loggable createLoggable(String input, String suffix) {
+        String completeString = input.trim();
+        if (!input.endsWith(";")) {
+            completeString += ";";
+        }
+        if (suffix != null && !suffix.isEmpty()) {
+            completeString += suffix;
+        }
+        return new LoggedString(completeString);
+    }
+
+    @Override
+    public FelderaQueryAdapter getQueryForStateToReproduce(String queryString) {
+        return new FelderaOtherQuery(queryString, FelderaExpectedError.expectedErrors());
+    }
+
+    @Override
+    public FelderaQueryAdapter commentOutQuery(Query<?> query) {
+        String queryString = query.getQueryString();
+        String newQueryString = "-- " + queryString;
+        ExpectedErrors errors = new ExpectedErrors();
+        return new FelderaOtherQuery(newQueryString, errors);
+    }
+
+    @Override
+    protected Loggable infoToLoggable(String time, String pipelineName, String databaseVersion, long seedValue) {
+        String sb = "-- Time: " + time + "\n" + "-- Pipeline: " + pipelineName + "\n " + "-- Pipeline version: "
+                + databaseVersion + "\n" + "-- seed value: " + seedValue + "\n";
+        return new LoggedString(sb);
+    }
+
+    @Override
+    public Loggable convertStacktraceToLoggable(Throwable throwable) {
+        StringWriter sw = new StringWriter();
+        PrintWriter pw = new PrintWriter(sw);
+        throwable.printStackTrace(pw);
+        return new LoggedString("--" + sw.toString().replace("\n", "\n--"));
+    }
+}

src/sqlancer/feldera/FelderaOptions.java

@@ -0,0 +1,26 @@
+package sqlancer.feldera;
+
+import java.util.List;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.Parameters;
+
+import sqlancer.DBMSSpecificOptions;
+
+@Parameters(separators = "=", commandDescription = "Feldera (default " + FelderaOptions.DEFAULT_URL + ")")
+public class FelderaOptions implements DBMSSpecificOptions<FelderaOracleFactory> {
+
+    public static final String DEFAULT_URL = "http://127.0.0.1:8080";
+
+    @Parameter(names = "--oracle", description = "Specifies which test oracle should be used for Feldera")
+    public List<FelderaOracleFactory> oracle = List.of(FelderaOracleFactory.NOREC);
+
+    @Parameter(names = "--connection-url", description = "Specifies the URL for connecting to the Feldera", arity = 1)
+    public String connection_url = DEFAULT_URL;
+
+    @Override
+    public List<FelderaOracleFactory> getTestOracleFactory() {
+        return oracle;
+    }
+
+}

src/sqlancer/feldera/FelderaOracleFactory.java

@@ -0,0 +1,14 @@
+package sqlancer.feldera;
+
+import sqlancer.OracleFactory;
+import sqlancer.feldera.oracle.FelderaNoRECOracle;
+import sqlancer.common.oracle.TestOracle;
+
+public enum FelderaOracleFactory implements OracleFactory<FelderaGlobalState> {
+    NOREC {
+        @Override
+        public TestOracle<FelderaGlobalState> create(FelderaGlobalState globalState) {
+            return new FelderaNoRECOracle(globalState);
+        }
+    },
+}