/*
 * Copyright 2022 VMware, Inc.
 * SPDX-License-Identifier: MIT
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

package org.dbsp.sqlCompiler.compiler.sql;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.dbsp.sqlCompiler.CompilerMain;
import org.dbsp.sqlCompiler.circuit.DBSPCircuit;
import org.dbsp.sqlCompiler.circuit.operator.DBSPSinkOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPStreamDistinctOperator;
import org.dbsp.sqlCompiler.circuit.OutputPort;
import org.dbsp.sqlCompiler.circuit.operator.IInputOperator;
import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
import org.dbsp.sqlCompiler.compiler.TestUtil;
import org.dbsp.sqlCompiler.compiler.visitors.outer.LateMaterializations;
import org.dbsp.util.HashString;
import org.dbsp.sqlCompiler.compiler.backend.JsonDecoder;
import org.dbsp.sqlCompiler.compiler.backend.MerkleOuter;
import org.dbsp.sqlCompiler.compiler.backend.ToCsvVisitor;
import org.dbsp.sqlCompiler.compiler.backend.ToJsonOuterVisitor;
import org.dbsp.sqlCompiler.compiler.backend.rust.RustFileWriter;
import org.dbsp.sqlCompiler.compiler.errors.CompilerMessages;
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.SqlToRelCompiler;
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.FunctionDocumentation;
import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject;
import org.dbsp.sqlCompiler.compiler.sql.simple.EndToEndTests;
import org.dbsp.sqlCompiler.compiler.sql.tools.BaseSQLTests;
import org.dbsp.sqlCompiler.compiler.sql.tools.Change;
import org.dbsp.sqlCompiler.compiler.sql.tools.CompilerCircuitStream;
import org.dbsp.sqlCompiler.compiler.visitors.outer.Passes;
import org.dbsp.sqlCompiler.ir.DBSPFunction;
import org.dbsp.sqlCompiler.ir.expression.DBSPApplyExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPBlockExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPTupleExpression;
import org.dbsp.sqlCompiler.ir.expression.literal.DBSPI32Literal;
import org.dbsp.sqlCompiler.ir.expression.literal.DBSPStrLiteral;
import org.dbsp.sqlCompiler.ir.expression.DBSPZSetExpression;
import org.dbsp.sqlCompiler.ir.statement.DBSPLetStatement;
import org.dbsp.sqlCompiler.ir.statement.DBSPStatement;
import org.dbsp.sqlCompiler.ir.type.user.DBSPTypeZSet;
import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeInteger;
import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeVoid;
import org.dbsp.util.IWritesLogs;
import org.dbsp.util.IndentStream;
import org.dbsp.util.Linq;
import org.dbsp.util.Logger;
import org.dbsp.util.Utilities;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;

import javax.imageio.ImageIO;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileWriter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.PrintStream;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/** Miscellaneous tests that do not fit into standard categories */
public class OtherTests extends BaseSQLTests implements IWritesLogs { // interface used for testing
    public static final String ddl = """
            CREATE TABLE T (
            COL1 INT NOT NULL
            , COL2 DOUBLE NOT NULL
            , COL3 BOOLEAN NOT NULL
            , COL4 VARCHAR NOT NULL
            , COL5 INT
            , COL6 DOUBLE
            )""";

    private DBSPCompiler compileDef() {
        DBSPCompiler compiler = this.testCompiler();
        compiler.submitStatementForCompilation(ddl);
        return compiler;
    }

    @Test public void toStringTest() {
        String query = "CREATE VIEW V AS SELECT T.COL3 FROM T";
        DBSPCompiler compiler = this.compileDef();
        compiler.submitStatementForCompilation(query);
        DBSPCircuit circuit = compiler.getFinalCircuit(false);
        Assert.assertNotNull(circuit);
        String str = circuit.toString();
        String expected = """
                Circuit circuit {
                    // DBSPSourceMultisetOperator s0
                    // CREATE TABLE `t` (`col1` INTEGER NOT NULL, `col2` DOUBLE NOT NULL, `col3` BOOLEAN NOT NULL, `col4` VARCHAR NOT NULL, `col5` INTEGER, `col6` DOUBLE)
                    let s0 = t();
                    // DBSPMapOperator s1
                    let s1 = s0.map((|p0: &Tup6<i32, d, b, s, i32?, d?>|
                    Tup1::new(((*p0).2), )));
                    // CREATE VIEW `v` AS
                    // SELECT `t`.`col3`
                    // FROM `schema`.`t` AS `t`
                    let s2 = s1;
                    // DBSPConstantOperator s3
                    let s3 = constant(zset!());
                    // CREATE VIEW `error_view` AS
                    // SELECT `feldera_error_table`.`table_or_view_name`, `feldera_error_table`.`message`, `feldera_error_table`.`metadata`
                    // FROM `schema`.`feldera_error_table` AS `feldera_error_table`
                    let s4 = s3;
                }
                """;
        Assert.assertEquals(expected, str);
    }

    @Test
    public void testIntCastWarning() {
        DBSPCompiler compiler = this.testCompiler();
        compiler.options.ioOptions.quiet = false;
        String query = "CREATE VIEW V AS SELECT '1_000'::INT4";
        compiler.submitStatementForCompilation(query);
        getCircuit(compiler);  // invokes optimizer
        TestUtil.assertMessagesContain(compiler, "String '1_000' cannot be interpreted as a number");
    }

    // Test the ability to redirect logging streams.
    @Test
    public void loggerTest() {
        StringBuilder builder = new StringBuilder();
        Appendable save = Logger.INSTANCE.setDebugStream(builder);
        Logger.INSTANCE.setLoggingLevel(this.getClass(), 1);
        Assert.assertEquals("OtherTests", this.getClassName());
        Logger.INSTANCE.belowLevel(this, 1)
                .append("Logging one statement")
                .newline();
        Logger.INSTANCE.setLoggingLevel(this.getClass(), 0);
        Logger.INSTANCE.belowLevel(this, 1)
                .append("This one is not logged")
                .newline();
        Logger.INSTANCE.setDebugStream(save);
        Assert.assertEquals("Logging one statement\n", builder.toString());
        Logger.INSTANCE.setLoggingLevel(this.getClass(), 0);
    }

    // Test the -T command-line parameter
    @Test
    public void loggingParameter() throws IOException, InterruptedException, SQLException {
        StringBuilder builder = new StringBuilder();
        Appendable save = Logger.INSTANCE.setDebugStream(builder);
        String sql = """
                CREATE TABLE T (COL1 INT NOT NULL, COL2 DOUBLE NOT NULL);
                CREATE VIEW V AS SELECT COL1 FROM T;""";
        File file = createInputScript(sql);
        CompilerMain.execute("-TSqlToRelCompiler=2", "-TPasses=2",
                "-o", BaseSQLTests.TEST_FILE_PATH, file.getPath());
        BaseSQLTests.compileAndCheckRust(true);
        Logger.INSTANCE.setDebugStream(save);
        String messages = builder.toString();
        Assert.assertTrue(messages.contains("After optimizer"));
        Assert.assertTrue(messages.contains("MergeSums"));
        Logger.INSTANCE.setLoggingLevel(SqlToRelCompiler.class, 0);
        Logger.INSTANCE.setLoggingLevel(Passes.class, 0);
    }

    @Test
    public void toCsvTest() {
        DBSPCompiler compiler = testCompiler();
        DBSPZSetExpression s = new DBSPZSetExpression(EndToEndTests.E0, EndToEndTests.E1);
        StringBuilder builder = new StringBuilder();
        ToCsvVisitor visitor = new ToCsvVisitor(compiler, builder, () -> "");
        visitor.apply(s);
        String[] lines = builder.toString().split("\n");
        Arrays.sort(lines);
        Assert.assertEquals(
                "10,1.0,false,\"Hi\",1,0,5,\n" +
                        "10,12.0,true,\"Hi\",,,,",
                String.join("\n", lines));
    }

    @Test
    public void rustCsvTest() throws IOException, InterruptedException {
        DBSPCompiler compiler = testCompiler();
        DBSPZSetExpression data = new DBSPZSetExpression(EndToEndTests.E0, EndToEndTests.E1);
        File file = File.createTempFile("test", ".csv", new File(BaseSQLTests.RUST_DIRECTORY));
        file.deleteOnExit();
        ToCsvVisitor.toCsv(compiler, file, data);
        List<DBSPStatement> list = new ArrayList<>();
        DBSPLetStatement src = new DBSPLetStatement("src",
                new DBSPApplyExpression("read_csv", data.getType(),
                        new DBSPStrLiteral(file.getAbsolutePath())));
        list.add(src);
        list.add(new DBSPApplyExpression(
                "assert_eq!", DBSPTypeVoid.INSTANCE, src.getVarReference(),
                data).toStatement());
        DBSPExpression body = new DBSPBlockExpression(list, null);
        DBSPFunction tester = new DBSPFunction(CalciteObject.EMPTY, "test", new ArrayList<>(),
                DBSPTypeVoid.INSTANCE, body, Linq.list("#[test]"));

        PrintStream stream = new PrintStream(BaseSQLTests.TEST_FILE_PATH);
        RustFileWriter writer = new RustFileWriter(new LateMaterializations(compiler)).withTest(true);
        writer.setOutputBuilder(new IndentStream(stream));
        writer.add(tester);
        writer.write(compiler);
        stream.close();
        BaseSQLTests.compileAndTestRust(false);
    }

    @Test
    public void rustCsvTest2() throws IOException, InterruptedException {
        DBSPCompiler compiler = this.testCompiler();
        DBSPZSetExpression data = new DBSPZSetExpression(
                new DBSPTupleExpression(new DBSPI32Literal(1, true)),
                new DBSPTupleExpression(new DBSPI32Literal(2, true)),
                new DBSPTupleExpression(DBSPI32Literal.none(
                        new DBSPTypeInteger(CalciteObject.EMPTY, 32, true,true)))
        );
        File file = File.createTempFile("test", ".csv", new File(BaseSQLTests.RUST_DIRECTORY));
        file.deleteOnExit();
        ToCsvVisitor.toCsv(compiler, file, data);
        List<DBSPStatement> list = new ArrayList<>();
        DBSPLetStatement src = new DBSPLetStatement("src",
                new DBSPApplyExpression("read_csv", data.getType(),
                        new DBSPStrLiteral(file.getAbsolutePath())));
        list.add(src);
        list.add(new DBSPApplyExpression(
                "assert_eq!", DBSPTypeVoid.INSTANCE, src.getVarReference(),
                data).toStatement());
        DBSPExpression body = new DBSPBlockExpression(list, null);
        DBSPFunction tester = new DBSPFunction(CalciteObject.EMPTY, "test", new ArrayList<>(),
                DBSPTypeVoid.INSTANCE, body, Linq.list("#[test]"));

        PrintStream outputStream = new PrintStream(BaseSQLTests.TEST_FILE_PATH, StandardCharsets.UTF_8);
        RustFileWriter writer = new RustFileWriter(new LateMaterializations(compiler)).withTest(true);
        writer.setOutputBuilder(new IndentStream(outputStream));
        writer.add(tester);
        writer.write(compiler);
        outputStream.close();
        BaseSQLTests.compileAndTestRust(false);
    }

    @Test
    public void testWith() throws IOException, InterruptedException, SQLException {
        String statement =
                """
                create table VENDOR (
                    id bigint not null primary key,
                    name varchar,
                    address varchar
                );
                create table PART (
                    id bigint not null primary key,
                    name varchar
                );
                create table PRICE (
                    part bigint not null,
                    vendor bigint not null,
                    price decimal
                );

                create view LOW_PRICE AS
                WITH LOW_PRICE_CTE AS (select part, MIN(price) as price from PRICE group by part)
                SELECT * FROM LOW_PRICE_CTE""";
        File file = createInputScript(statement);
        CompilerMessages messages = CompilerMain.execute("-o", BaseSQLTests.TEST_FILE_PATH, file.getPath());
        if (messages.errorCount() > 0)
            throw new RuntimeException(messages.toString());
        BaseSQLTests.compileAndCheckRust(true);
    }

    void compileFile(String file, boolean run) throws SQLException, IOException, InterruptedException {
        CompilerMessages messages = CompilerMain.execute(
                "-i", "--alltables", "-q", "--ignoreOrder", "-o", BaseSQLTests.TEST_FILE_PATH, file);
        messages.print();
        Assert.assertEquals(0, messages.errorCount());
        if (run)
            BaseSQLTests.compileAndCheckRust(true);
        // cleanup after ourselves
        createEmptyStubs();
    }

    @Test
    public void testUnionWarning() throws SQLException {
        final String script = "../../crates/pipeline-manager/demos/sql/02-sec-ops.sql";
        CompilerMessages messages = CompilerMain.execute(
                "-i", "--alltables", "-q", "--ignoreOrder", "-o", BaseSQLTests.TEST_FILE_PATH, script);
        for (int i = 0; i < messages.warningCount(); i++) {
            CompilerMessages.Message msg = messages.getMessage(i);
            Assert.assertNotEquals("Fields reordered", msg.errorType);
        }
    }

    @Test
    public void testPackagedDemos() throws SQLException, IOException, InterruptedException {
        final String projectsDirectory = "../../crates/pipeline-manager/demos/sql";
        final File dir = new File(projectsDirectory);
        FilenameFilter filter = (_d, name) -> !name.contains("setup") && name.endsWith(".sql");
        String[] sqlFiles = dir.list(filter);
        Assert.assertNotNull(sqlFiles);
        Arrays.sort(sqlFiles);
        for (String sqlFile: sqlFiles) {
            // System.out.println(sqlFile);
            String basename = Utilities.getBaseName(sqlFile);
            String udf = basename + ".udf.rs";
            File udfFile = new File(dir.getPath() + "/" + udf);
            this.compileFile(dir.getPath() + "/" + sqlFile, !udfFile.exists());
        }
    }

    // Test the ignoreOrderBy compiler flag
    @Test
    public void testIgnoreOrderBy() {
        DBSPCompiler compiler = this.testCompiler();
        compiler.options.languageOptions.throwOnError = false;
        compiler.options.languageOptions.ignoreOrderBy = true;
        compiler.options.ioOptions.quiet = false;
        String query = "CREATE VIEW V AS SELECT * FROM T ORDER BY T.COL2";
        compiler.submitStatementForCompilation(ddl);
        compiler.submitStatementsForCompilation(query);
        DBSPCircuit circuit = compiler.getFinalCircuit(false);
        Assert.assertNotNull(circuit);
        DBSPTypeZSet outputType = circuit.getSingleOutputType().to(DBSPTypeZSet.class);
        IInputOperator source = circuit.getInput(compiler.canonicalName("T", false));
        Assert.assertNotNull(source);
        DBSPTypeZSet inputType = source.getDataOutputType().to(DBSPTypeZSet.class);
        Assert.assertTrue(inputType.sameType(outputType));
        TestUtil.assertMessagesContain(compiler.messages, """
             ORDER BY clause is currently ignored
             (the result will contain the correct data, but the data is not ordered)""");
    }

    // Test the outputsAreSets compiler flag
    @Test
    public void testOutputsAreSets() {
        DBSPCompiler compiler = this.testCompiler();
        compiler.options.languageOptions.throwOnError = false;
        compiler.options.languageOptions.outputsAreSets = true;
        String query = "CREATE VIEW V AS SELECT T.COL1 FROM T";
        compiler.submitStatementForCompilation(ddl);
        compiler.submitStatementsForCompilation(query);
        DBSPCircuit circuit = compiler.getFinalCircuit(false);
        Assert.assertNotNull(circuit);
        DBSPSinkOperator sink = circuit.getSink(compiler.canonicalName("V", false));
        Assert.assertNotNull(sink);
        OutputPort op = sink.input();
        // There is no optimization I can imagine which will remove the distinct
        Assert.assertTrue(op.node().is(DBSPStreamDistinctOperator.class));
    }

    @Test
    public void testNoOutput() throws IOException, SQLException {
        String sql = """
                 CREATE TABLE T (COL1 INT NOT NULL COL2 DOUBLE NOT NULL);
                 CREATE VIEW V AS SELECT COL1 FROM T""";
        File file = createInputScript(sql);
        // Redirect error stream
        ByteArrayOutputStream os = new ByteArrayOutputStream();
        PrintStream ps = new PrintStream(os);
        PrintStream old = System.err;
        System.setErr(ps);
        CompilerMain.execute(file.getPath(), file.getPath());
        // Restore error stream
        System.setErr(old);
        Assert.assertTrue(os.toString().contains("Did you forget to specify"));
    }

    @Test
    public void testRustCompiler() throws IOException, InterruptedException, SQLException {
        String sql = """
                CREATE TABLE T (COL1 INT NOT NULL, COL2 DOUBLE NOT NULL);
                CREATE VIEW V AS SELECT COL1 FROM T;""";
        File file = createInputScript(sql);
        CompilerMessages messages = CompilerMain.execute("-q", "-o", BaseSQLTests.TEST_FILE_PATH, file.getPath());
        messages.print();
        Assert.assertEquals(0, messages.exitCode);
        BaseSQLTests.compileAndCheckRust(true);
    }

    @Test
    public void testCompilerExample() throws IOException, InterruptedException, SQLException {
        // The example in sql-to-dbsp-compiler/using.md
        String sql = """
                -- define Person table
                CREATE TABLE Person
                (
                    name    VARCHAR NOT NULL,
                    age     INT,
                    present BOOLEAN
                );
                CREATE VIEW Adult AS SELECT Person.name FROM Person WHERE Person.age > 18;
                """;
        String rustHandlesTest = """
                #[test]
                pub fn test() {
                    let (mut circuit, (person, errors, adult) ) = circuit(CircuitConfig::with_workers(2)).unwrap();
                    // Feed two input records to the circuit.
                    // First input has a count of "1"
                    person.push( (SqlString::from_ref("Bob"), Some(12), Some(true)).into(), 1 );
                    // Second input has a count of "2"
                    person.push( (SqlString::from_ref("Tom"), Some(20), Some(false)).into(), 2 );
                    // Execute the circuit on these inputs
                    circuit.transaction().unwrap();
                    // Read the produced output
                    let out = adult.concat().consolidate();
                    // Print the produced output
                    println!("{:?}", out);
                }
                """;
        String rustCatalogTest = """
                #[test]
                pub fn test() {
                    use dbsp_adapters::{CircuitCatalog, RecordFormat};
                    use feldera_types::format::csv::CsvParserConfig;

                    let (mut circuit, catalog) = circuit(CircuitConfig::with_workers(2))
                        .expect("Failed to build circuit");
                    let persons = catalog
                        .input_collection_handle(&SqlIdentifier::from("PERSON"))
                        .expect("Failed to get input collection handle");
                    let mut persons_stream = persons
                        .handle
                        .configure_deserializer(RecordFormat::Csv(Default::default()))
                        .expect("Failed to configure deserializer");
                    persons_stream
                        .insert(b"Bob,12,true", &None)
                        .expect("Failed to insert data");
                    persons_stream
                        .insert(b"Tom,20,false", &None)
                        .expect("Failed to insert data");
                    persons_stream
                        .insert(b"Tom,20,false", &None)
                        .expect("Failed to insert data");  // Insert twice
                    persons_stream.flush();
                    // Execute the circuit on these inputs
                    circuit
                        .transaction()
                        .unwrap();

                    let adult = &catalog
                        .output_handles(&SqlIdentifier::from("ADULT"))
                        .expect("Failed to get output collection handles")
                        .delta_handle;

                    // Read the produced output
                    let reader = adult.concat();
                    let mut cursor = reader
                        .cursor(RecordFormat::Csv(CsvParserConfig::default()))
                        .unwrap();
                    while cursor.key_valid() {
                        let mut w = cursor.weight();
                        println!("{}: {}", cursor.key_to_json().unwrap(), w);
                        cursor.step_key();
                    }
                }""";
        File file = createInputScript(sql);
        CompilerMessages message = CompilerMain.execute(
                "--handles", "-o", BaseSQLTests.TEST_FILE_PATH, file.getPath());
        Assert.assertEquals(0, message.exitCode);
        Assert.assertTrue(file.exists());

        File rust = new File(BaseSQLTests.TEST_FILE_PATH);
        try (FileWriter fr = new FileWriter(rust, true)) { // append
            fr.write(rustHandlesTest);
        }
        if (!BaseSQLTests.skipRust)
            BaseSQLTests.compileAndTestRust(true);

        // Second test
        message = CompilerMain.execute(
                "-i", "-o", BaseSQLTests.TEST_FILE_PATH, file.getPath());
        Assert.assertEquals(0, message.exitCode);
        Assert.assertTrue(file.exists());

        rust = new File(BaseSQLTests.TEST_FILE_PATH);
        try (FileWriter fr = new FileWriter(rust, true)) { // append
            fr.write(rustCatalogTest);
        }
        if (!BaseSQLTests.skipRust)
            BaseSQLTests.compileAndTestRust(true);
    }

    @Test
    public void issue4895() throws IOException, InterruptedException, SQLException {
        String sql = """
                create table t (x int lateness 0) with
                 ('materialized' = 'true');
                create materialized view V as SELECT * FROM t;""";
        String rustCatalogTest = """
                #[test]
                pub fn test0() {
                    use dbsp_adapters::{CircuitCatalog, RecordFormat};
                    use feldera_types::format::json::JsonFlavor;
                
                    let mut circuitAndStreams = circuit(CircuitConfig::with_workers(2usize)).unwrap();
                    let streams: Catalog = circuitAndStreams.1;
                    let t = &SqlIdentifier::new("t", false);
                    let input = streams.input_collection_handle(t).unwrap();
                    let mut writer = input
                        .handle
                        .configure_deserializer(RecordFormat::Csv(Default::default()))
                        .unwrap();
                    writer.insert(b"1", &None).unwrap();
                    writer.flush();
                    writer.insert(b"2", &None).unwrap();
                    writer.flush();
                    writer.insert(b"3", &None).unwrap();
                    writer.flush();
                    circuitAndStreams.0.transaction().unwrap();
                    // late value
                    writer.insert(b"0", &None).unwrap();
                    writer.flush();
                    circuitAndStreams.0.transaction().unwrap();
                
                    let output = streams
                        .output_handles(t)
                        .unwrap()
                        .integrate_handle
                        .clone()
                        .unwrap()
                        .clone();
                    let reader = output.concat();
                    let mut cursor = reader
                        .cursor(RecordFormat::Json(JsonFlavor::default()))
                        .unwrap();
                    let mut rows: u32 = 0;
                    while (cursor.key_valid()) {
                        let mut w = cursor.weight();
                        // println!("{}: {}", cursor.key_to_json().unwrap(), w);
                        rows = rows + 1;
                        cursor.step_key();
                    }
                    // There should be only 3 rows; the last inserted one should be missing
                    assert!(3 == rows, "Expected 3 rows, got {}", rows);
                }""";
        File file = createInputScript(sql);
        CompilerMessages message = CompilerMain.execute(
                "-i", "-o", BaseSQLTests.TEST_FILE_PATH, file.getPath());
        Assert.assertEquals(0, message.exitCode);
        Assert.assertTrue(file.exists());

        File rust = new File(BaseSQLTests.TEST_FILE_PATH);
        try (FileWriter fr = new FileWriter(rust, true)) { // append
            fr.write(rustCatalogTest);
        }
        if (!BaseSQLTests.skipRust)
            BaseSQLTests.compileAndTestRust(true);
    }

    @Test
    public void testCompilerToPng() throws IOException, SQLException {
        if (!Utilities.isDotInstalled())
            return;
        String sql ="""
                 CREATE TABLE T (COL1 INT NOT NULL, COL2 DOUBLE NOT NULL);
                 CREATE VIEW V AS SELECT COL1 FROM T WHERE COL1 > 5;""";
        File file = createInputScript(sql);
        File png = File.createTempFile("out", ".png", new File("."));
        png.deleteOnExit();
        CompilerMessages message = CompilerMain.execute("--png", "-o", png.getPath(), file.getPath());
        Assert.assertEquals(0, message.exitCode);
        Assert.assertTrue(file.exists());
        ImageIO.read(new File(png.getPath()));
    }

    @Test
    public void testUDT() {
        this.getCCS("""
                CREATE TYPE address_typ AS (
                   street          VARCHAR(30),
                   city            VARCHAR(20),
                   state           CHAR(2),
                   postal_code     VARCHAR(6));
                CREATE TABLE T(street VARCHAR, city VARCHAR, year INT);
                CREATE VIEW V AS SELECT address_typ(T.street, city, 'CA', 94087) as address, T.year as year FROM T;""");
    }

    @Test
    public void rawCalciteTest() throws SQLException {
        Connection connection = DriverManager.getConnection("jdbc:calcite:");
        String query = "SELECT timestampdiff(MONTH, TIMESTAMP'2021-02-28 12:00:00', TIMESTAMP'2021-03-28 11:59:59')";
        try (PreparedStatement ps = connection.prepareStatement(query)) {
            ps.execute();
            try (ResultSet resultSet = ps.getResultSet()) {
                while (resultSet.next()) {
                    int result = resultSet.getInt(1);
                    Assert.assertEquals(0, result);
                }
            }
        }
    }

    @Test
    public void testLocalView() {
        String sql = """
                    CREATE TABLE T(I INTEGER, S VARCHAR);
                    CREATE LOCAL VIEW V AS SELECT * FROM T;""";
        DBSPCompiler compiler = this.testCompiler();
        compiler.submitStatementsForCompilation(sql);
        DBSPCircuit circuit = getCircuit(compiler);
        Assert.assertEquals(1, circuit.getOutputCount());

        String oneMore = "CREATE VIEW W AS SELECT * FROM V;";
        compiler = this.testCompiler();
        compiler.submitStatementsForCompilation(sql + oneMore);
        circuit = getCircuit(compiler);
        Assert.assertEquals(2, circuit.getOutputCount());
    }

    @Test
    public void testRemove() {
        DBSPCompiler compiler = this.testCompiler();
        compiler.submitStatementForCompilation("CREATE TABLE T(I INTEGER, S VARCHAR)");
        CompilerCircuitStream ccs = this.getCCS(compiler);
        Change change = ccs.toChange("""
                INSERT INTO T VALUES(1, 'x');
                REMOVE FROM T VALUES(2, 'Y');
                REMOVE FROM T VALUES(3, 'Z');""").simplify(compiler);
        String set = change.getSet(0).toString();
        Assert.assertEquals("""
                TableData[name=t, data=zset!(
                    Tup2::new(Some(1), Some("x"), ) => 1i64,
                    Tup2::new(Some(2), Some("Y"), ) => -1i64,
                    Tup2::new(Some(3), Some("Z"), ) => -1i64,
                ), primaryKeys=[]]""", set);
    }

    @Test
    public void indexTest() {
        String sql = """
                CREATE TABLE T(id int, v VARCHAR, z INT ARRAY);
                CREATE VIEW V AS SELECT * FROM T;
                CREATE INDEX IX ON V(id, v);""";
        this.compileRustTestCase(sql);
    }

    @Test
    public void merkleTest() {
        // Compile two programs that have some query in common and check that the circuits produced
        // have nodes in common.
        var cc0 = this.getCC("""
                CREATE TABLE tab0(pk INTEGER, col0 INTEGER, col1 REAL, col2 TEXT, col3 INTEGER);
                CREATE VIEW V AS SELECT pk FROM tab0
                WHERE (col3 < 73 AND col3 IN (SELECT col0 FROM tab0 WHERE col0 = 3)) OR col1 > 8.64""");
        MerkleOuter visitor0 = new MerkleOuter(cc0.compiler, false);
        visitor0.apply(cc0.getCircuit());

        var cc1 = this.getCC("""
                CREATE TABLE tab0(pk INTEGER, col0 INTEGER, col1 REAL, col2 TEXT, col3 INTEGER);
                CREATE VIEW W AS SELECT pk + col0 FROM tab0;
                CREATE VIEW V AS SELECT pk FROM tab0
                WHERE (col3 < 73 AND col3 IN (SELECT col0 FROM tab0 WHERE col0 = 3)) OR col1 > 8.64""");
        MerkleOuter visitor1 = new MerkleOuter(cc1.compiler, false);
        visitor1.apply(cc1.getCircuit());
        Set<HashString> c0 = new HashSet<>(visitor0.operatorHash.values());
        Set<HashString> c1 = new HashSet<>(visitor1.operatorHash.values());
        Set<HashString> common = new HashSet<>(c0);
        // All nodes in circuit0 must be in circuit1
        common.retainAll(c1);
        Assert.assertEquals(common.size(), c0.size());
    }

    @Test
    public void serializationTest() throws JsonProcessingException {
        var cc = this.getCC("""
                CREATE TABLE tab0(pk INTEGER, col0 INTEGER, col1 REAL, col2 TEXT, col3 INTEGER);
                CREATE VIEW V AS SELECT pk FROM tab0
                WHERE (col3 < 73 AND col3 IN (SELECT col0 FROM tab0 WHERE col0 = 3)) OR col1 > 8.64""");
        ToJsonOuterVisitor visitor = ToJsonOuterVisitor.create(cc.compiler, 1);
        visitor.apply(cc.getCircuit());
        String str = visitor.getJsonString();
        ObjectMapper mapper = Utilities.deterministicObjectMapper();
        JsonNode parsed = mapper.readTree(str);
        JsonDecoder decoder = new JsonDecoder(cc.compiler.sqlToRelCompiler.typeFactory);
        DBSPCircuit decoded = decoder.decodeOuter(parsed, DBSPCircuit.class);
        Assert.assertNotNull(decoded);
    }

    @Test
    public void compileAggregationTests() throws URISyntaxException, IOException {
        List<String> tests = TestUtil.enumerateResourceFiles();
        for (String sqlFile: tests) {
            if (!sqlFile.endsWith("sql")) continue;
            if (!sqlFile.startsWith("aggregation_tests")) continue;
            String sql = TestUtil.readStringFromResourceFile(sqlFile);
            this.getCC(sql);
        }
    }

    @Test
    public void argMinDate() {
        this.getCCS("""
                CREATE TABLE date_tbl(
                id INT,
                c1 DATE NOT NULL,
                c2 DATE);
                
                CREATE VIEW V0 AS SELECT
                id, ARG_MIN(c1, c2) AS c1, ARG_MIN(c2, c1) AS c2
                FROM date_tbl
                GROUP BY id;
                
                CREATE VIEW V1 AS SELECT
                ARG_MIN(c1, c2) AS c1, ARG_MIN(c2, c1) AS c2
                FROM date_tbl;""");
    }

    @Test
    public void argMinString() {
        this.getCCS("""
                CREATE TABLE varchar_tbl(
                id INT,
                c1 VARCHAR,
                c2 VARCHAR NULL);
                
                CREATE VIEW atbl_charn AS SELECT
                id,
                CAST(c1 AS CHAR(7)) AS f_c1,
                CAST(c2 AS CHAR(7)) AS f_c2
                FROM varchar_tbl;
                
                CREATE VIEW V2 AS SELECT id, ARG_MIN(f_c1, f_c2) AS c1, ARG_MIN(f_c2, f_c1) AS c2
                FROM atbl_charn
                GROUP BY id;""");
    }

    @Test @Ignore("To be invoked manually every time a new function is added")
    public void generateFunctionIndex() throws IOException, InterruptedException {
        // When invoked it generates documentation for the supported functions and operators
        // in the specified file.
        String file = "../../docs.feldera.com/docs/sql/function-index.md";
        FunctionDocumentation.generateIndex(file);
        Utilities.runProcess(BaseSQLTests.PROJECT_DIRECTORY + "/../docs.feldera.com",
                "yarn");
        Utilities.runProcess(BaseSQLTests.PROJECT_DIRECTORY + "/../docs.feldera.com",
                "yarn", "build");
    }

    @Test
    public void rustFmt() throws IOException, InterruptedException {
        // Don't run this in CI
        if (Utilities.inCI())
            return;
        // Check that the rust library is properly formatted
        Utilities.runProcess(BaseSQLTests.PROJECT_DIRECTORY + "/../crates/sqllib",
                "cargo", "fmt", "--all", "--", "--check");
        Utilities.runProcess(BaseSQLTests.PROJECT_DIRECTORY + "/../crates/sqllib",
                "cargo", "clippy", "--all-targets", "--tests", "--", "-D", "warnings");
    }
}
