test: add PostgreSQL TPC-H integration tests

.github/workflows/pr.yml

@@ -63,9 +63,31 @@ jobs:
           git submodule foreach 'git fetch --unshallow || true'
 
           ./gradlew build --rerun-tasks
+  integration:
+    name: Integration tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          submodules: recursive
+      - name: Set up JDK 17
+        uses: actions/setup-java@v5
+        with:
+          java-version: '17'
+          distribution: 'temurin'
+      - name: Setup Gradle
+        uses: gradle/actions/setup-gradle@v6
+      - name: Build with Gradle
+        run: |
+          # fetch submodule tags since actions/checkout@v6 does not
+          git submodule foreach 'git fetch --unshallow || true'
+
+          ./gradlew integrationTest
   isthmus-native-image-mac-linux:
     name: Build Isthmus Native Image
-    needs: java
+    needs:
+      - java
+      - integration
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:

gradle.properties

@@ -1,4 +1,4 @@
-org.gradle.jvmargs=-XX:+UseG1GC -Xmx1g -XX:MaxMetaspaceSize=512m --add-exports jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED \
+org.gradle.jvmargs=-XX:+UseG1GC -Xmx2g -XX:MaxMetaspaceSize=512m --add-exports jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED \
   --add-exports jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED \
   --add-exports jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED \
   --add-exports jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED \

gradle/libs.versions.toml

@@ -13,6 +13,7 @@ jspecify = "1.0.0"
 junit = "6.1.0"
 nmcp = "1.5.0"
 picocli = "4.7.7"
+postgresql = "42.7.11"
 protobuf-plugin = "0.10.0"
 protobuf = "3.25.9"
 scala-2-12 = "2.12.20"
@@ -27,6 +28,7 @@ spark-3-4 = "3.4.4"
 spark-3-5 = "3.5.4"
 spark-4-0 = "4.0.2"
 spotless = "8.6.0"
+testcontainers = "2.0.5"
 validator = "3.0.3"
 
 [libraries]
@@ -57,6 +59,7 @@ junit-platform-launcher = { module = "org.junit.platform:junit-platform-launcher
 junit-platform-engine = { module = "org.junit.platform:junit-platform-engine" }
 picocli = { module = "info.picocli:picocli", version.ref = "picocli" }
 picocli-codegen = { module = "info.picocli:picocli-codegen", version.ref = "picocli" }
+postgresql-jdbc = { module = "org.postgresql:postgresql", version.ref = "postgresql" }
 protobuf-java = { module = "com.google.protobuf:protobuf-java", version.ref = "protobuf" }
 protobuf-java-util = { module = "com.google.protobuf:protobuf-java-util", version.ref = "protobuf" }
 protoc = { module = "com.google.protobuf:protoc", version.ref = "protobuf" }
@@ -80,9 +83,13 @@ spark-catalyst-4-0-2-13 = { module = "org.apache.spark:spark-catalyst_2.13", ver
 spark-core-4-0-2-13 = { module = "org.apache.spark:spark-core_2.13", version.ref = "spark-4-0" }
 spark-hive-4-0-2-13 = { module = "org.apache.spark:spark-hive_2.13", version.ref = "spark-4-0" }
 spark-sql-4-0-2-13 = { module = "org.apache.spark:spark-sql_2.13", version.ref = "spark-4-0" }
+testcontainers = { module = "org.testcontainers:testcontainers", version.ref = "testcontainers" }
+testcontainers-junit5 = { module = "org.testcontainers:testcontainers-junit-jupiter", version.ref = "testcontainers" }
+testcontainers-postgres = { module = "org.testcontainers:testcontainers-postgresql", version.ref = "testcontainers" }
 
 [bundles]
 jackson = [ "jackson-databind", "jackson-annotations", "jackson-datatype-jdk8", "jackson-dataformat-yaml" ]
+testcontainers = [ "testcontainers", "testcontainers-junit5", "testcontainers-postgres" ]
 
 [plugins]
 graal = { id = "org.graalvm.buildtools.native", version.ref = "graal-plugin" }

isthmus/build.gradle.kts

@@ -118,6 +118,12 @@ dependencies {
   }
   testImplementation(libs.protobuf.java)
   api(libs.jspecify)
+
+  testImplementation(libs.bundles.testcontainers)
+
+  testImplementation(libs.postgresql.jdbc)
+
+  testImplementation(libs.slf4j.jdk14)
 }
 
 tasks {
@@ -139,6 +145,23 @@ tasks {
 
   // Only set the compile release since JUnit 6 requires Java 17 to run tests.
   compileJava { options.release = 11 }
+
+  test {
+    // Exclude integration tests by default
+    useJUnitPlatform { excludeTags("integration") }
+  }
+}
+
+// Register a separate task to run integration tests
+val test by testing.suites.existing(JvmTestSuite::class)
+
+tasks.register<Test>("integrationTest") {
+  description = "Run integration tests"
+  group = "verification"
+  testClassesDirs = files(test.map { it.sources.output.classesDirs })
+  classpath = files(test.map { it.sources.runtimeClasspath })
+  useJUnitPlatform { includeTags("integration") }
+  shouldRunAfter(tasks.test)
 }
 
 sourceSets { test { proto.srcDirs("src/test/resources/extensions") } }

isthmus/src/test/java/io/substrait/isthmus/integration/PostgreSqlIntegrationTest.java

@@ -0,0 +1,138 @@
+package io.substrait.isthmus.integration;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.substrait.isthmus.ConverterProvider;
+import io.substrait.isthmus.PlanTestBase;
+import io.substrait.isthmus.SqlToSubstrait;
+import io.substrait.isthmus.SubstraitToSql;
+import io.substrait.plan.Plan;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.calcite.sql.dialect.PostgresqlSqlDialect;
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.BindMode;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.JdbcDatabaseContainer.NoDriverFoundException;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.postgresql.PostgreSQLContainer;
+import org.testcontainers.utility.DockerImageName;
+
+@Tag("integration")
+@Testcontainers
+class PostgreSqlIntegrationTest extends PlanTestBase {
+  private static final Logger LOG = LoggerFactory.getLogger(PostgreSqlIntegrationTest.class);
+
+  // TODO: These queries produce different results when generated from Substrait
+  private static final List<Integer> EXCLUDED_QUERIES = List.of(14);
+
+  private static final DockerImageName UV_IMAGE =
+      DockerImageName.parse("ghcr.io/astral-sh/uv:python3.14-trixie-slim");
+  private static final DockerImageName POSTGRES_IMAGE = DockerImageName.parse("postgres:18");
+
+  private static final String TPCH_DATA_HOST_PATH = "tpch/data";
+  private static final String TPCH_DATA_CONTAINER_PATH = "/tmp/tpc-h";
+  private static final String TPCH_INIT_HOST_PATH = "tpch/postgresql/tpch_init.sql";
+
+  private static final List<String> TPCHGEN_ARGS =
+      List.of(
+          "--scale-factor", "0.001", "--format", "csv", "--output-dir", TPCH_DATA_CONTAINER_PATH);
+
+  private static final List<String> TPCHGEN_CMD =
+      Stream.concat(
+              Stream.of("uvx", "--from", "tpchgen-cli == 2.*", "tpchgen-cli"),
+              TPCHGEN_ARGS.stream())
+          .collect(Collectors.toList());
+
+  /** Create TPC-H test data. */
+  @Container
+  @SuppressWarnings("resource")
+  private static final GenericContainer<?> tpchgen =
+      new GenericContainer<>(UV_IMAGE)
+          .withClasspathResourceMapping(
+              TPCH_DATA_HOST_PATH, TPCH_DATA_CONTAINER_PATH, BindMode.READ_WRITE)
+          .withCommand(TPCHGEN_CMD.toArray(new String[0]))
+          .withStartupCheckStrategy(new SuccessfulExitCheckStrategy());
+
+  /** PostgreSQL instance shared across all test methods in this class. */
+  @Container
+  @SuppressWarnings("resource")
+  private static final PostgreSQLContainer postgres =
+      new PostgreSQLContainer(POSTGRES_IMAGE)
+          .dependsOn(tpchgen)
+          .withClasspathResourceMapping(
+              TPCH_DATA_HOST_PATH, TPCH_DATA_CONTAINER_PATH, BindMode.READ_ONLY)
+          .withClasspathResourceMapping(
+              TPCH_INIT_HOST_PATH, "/docker-entrypoint-initdb.d/tpch_init.sql", BindMode.READ_ONLY);
+
+  private static final String COMPARE_RESULTS_SQL_TEMPLATE =
+      """
+      WITH expected AS (%s),
+          actual AS (%s)
+      SELECT count(*) FROM (
+        SELECT * FROM
+          (SELECT * FROM expected EXCEPT SELECT * FROM actual)
+          UNION (SELECT * FROM actual EXCEPT SELECT * FROM expected)
+      )
+      """;
+
+  static IntStream tpcHTestCases() {
+    return IntStream.rangeClosed(1, 22).filter(i -> !EXCLUDED_QUERIES.contains(i));
+  }
+
+  @ParameterizedTest
+  @MethodSource("tpcHTestCases")
+  void testTpcH(final int queryNo)
+      throws NoDriverFoundException, SQLException, IOException, SqlParseException {
+
+    final String inputSql = asString(String.format("tpch/queries/%02d.sql", queryNo));
+    final SqlToSubstrait sqlToSubstrait = new SqlToSubstrait();
+    final Plan plan = sqlToSubstrait.convert(inputSql, TPCH_CATALOG);
+
+    final ConverterProvider provider = new ConverterProvider(extensions);
+    final SubstraitToSql substraitToSql = new SubstraitToSql(provider);
+
+    final String generatedSql = substraitToSql.convert(plan, PostgresqlSqlDialect.DEFAULT).get(0);
+
+    final String referenceSql = asString(String.format("tpch/postgresql/%02d.sql", queryNo));
+
+    final String compareSql =
+        String.format(COMPARE_RESULTS_SQL_TEMPLATE, referenceSql, generatedSql);
+
+    LOG.atDebug().log(compareSql);
+
+    try (Connection conn = postgres.createConnection("");
+        Statement stmt = conn.createStatement();
+        ResultSet result = stmt.executeQuery(compareSql); ) {
+      // we expect exactly one row
+      assertTrue(result.next());
+
+      // the count should be zero if both the reference and generated SQL produce the same results
+      int differenceCount = result.getInt(1);
+      assertEquals(
+          0,
+          differenceCount,
+          String.format(
+              "Reference and generated SQL produce %d different results.\n\nReference SQL:\n%s\n\nGenerated SQL:\n%s",
+              differenceCount, referenceSql, generatedSql));
+
+      // we expect exactly one row
+      assertFalse(result.next());
+    }
+  }
+}

isthmus/src/test/java/io/substrait/isthmus/integration/SuccessfulExitCheckStrategy.java

@@ -0,0 +1,28 @@
+package io.substrait.isthmus.integration;
+
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import org.testcontainers.containers.startupcheck.StartupCheckStrategy;
+import org.testcontainers.utility.DockerStatus;
+
+/**
+ * A {@link StartupCheckStrategy} that checks if the container has exited with a successful exit
+ * code. This allows use of a container that is launched purely to execute a command prior to
+ * startup of dependent containers.
+ */
+public class SuccessfulExitCheckStrategy extends StartupCheckStrategy {
+  @Override
+  public StartupStatus checkStartupState(DockerClient dockerClient, String containerId) {
+    InspectContainerResponse.ContainerState state = getCurrentState(dockerClient, containerId);
+
+    if (!DockerStatus.isContainerStopped(state)) {
+      return StartupStatus.NOT_YET_KNOWN;
+    }
+
+    if (!DockerStatus.isContainerExitCodeSuccess(state)) {
+      return StartupStatus.FAILED;
+    }
+
+    return StartupStatus.SUCCESSFUL;
+  }
+}

isthmus/src/test/resources/logging.properties

@@ -0,0 +1,2 @@
+# Root logger level - set to INFO for test visibility
+.level=INFO

isthmus/src/test/resources/tpch/data/.gitignore

@@ -0,0 +1,5 @@
+# TPC-H test data is generated to this directory.
+
+*.tbl
+*.csv
+*.parquet

isthmus/src/test/resources/tpch/postgresql/01.sql

@@ -0,0 +1,22 @@
+select
+  "L_RETURNFLAG",
+  "L_LINESTATUS",
+  sum("L_QUANTITY") as "SUM_QTY",
+  sum("L_EXTENDEDPRICE") as "SUM_BASE_PRICE",
+  sum("L_EXTENDEDPRICE" * (1 - "L_DISCOUNT")) as "SUM_DISC_PRICE",
+  sum("L_EXTENDEDPRICE" * (1 - "L_DISCOUNT") * (1 + "L_TAX")) as "SUM_CHARGE",
+  avg("L_QUANTITY") as "AVG_QTY",
+  avg("L_EXTENDEDPRICE") as "AVG_PRICE",
+  avg("L_DISCOUNT") as "AVG_DISC",
+  count(*) as "COUNT_ORDER"
+from
+  "LINEITEM"
+where
+  "L_SHIPDATE" <= date '1998-12-01' - interval '120 days'
+group by
+  "L_RETURNFLAG",
+  "L_LINESTATUS"
+
+order by
+  "L_RETURNFLAG",
+  "L_LINESTATUS"

isthmus/src/test/resources/tpch/postgresql/02.sql

@@ -0,0 +1,47 @@
+select
+  "S"."S_ACCTBAL",
+  "S"."S_NAME",
+  "N"."N_NAME",
+  "P"."P_PARTKEY",
+  "P"."P_MFGR",
+  "S"."S_ADDRESS",
+  "S"."S_PHONE",
+  "S"."S_COMMENT"
+from
+  "PART" "P",
+  "SUPPLIER" "S",
+  "PARTSUPP" "PS",
+  "NATION" "N",
+  "REGION" "R"
+where
+  "P"."P_PARTKEY" = "PS"."PS_PARTKEY"
+  and "S"."S_SUPPKEY" = "PS"."PS_SUPPKEY"
+  and "P"."P_SIZE" = 41
+  and "P"."P_TYPE" like '%NICKEL'
+  and "S"."S_NATIONKEY" = "N"."N_NATIONKEY"
+  and "N"."N_REGIONKEY" = "R"."R_REGIONKEY"
+  and "R"."R_NAME" = 'EUROPE'
+  and "PS"."PS_SUPPLYCOST" = (
+
+    select
+      min("PS"."PS_SUPPLYCOST")
+
+    from
+      "PARTSUPP" "PS",
+      "SUPPLIER" "S",
+      "NATION" "N",
+      "REGION" "R"
+    where
+      "P"."P_PARTKEY" = "PS"."PS_PARTKEY"
+      and "S"."S_SUPPKEY" = "PS"."PS_SUPPKEY"
+      and "S"."S_NATIONKEY" = "N"."N_NATIONKEY"
+      and "N"."N_REGIONKEY" = "R"."R_REGIONKEY"
+      and "R"."R_NAME" = 'EUROPE'
+  )
+
+order by
+  "S"."S_ACCTBAL" desc,
+  "N"."N_NAME",
+  "S"."S_NAME",
+  "P"."P_PARTKEY"
+limit 100

isthmus/src/test/resources/tpch/postgresql/03.sql

@@ -0,0 +1,26 @@
+select
+  "L"."L_ORDERKEY",
+  sum("L"."L_EXTENDEDPRICE" * (1 - "L"."L_DISCOUNT")) as "REVENUE",
+  "O"."O_ORDERDATE",
+  "O"."O_SHIPPRIORITY"
+
+from
+  "CUSTOMER" "C",
+  "ORDERS" "O",
+  "LINEITEM" "L"
+
+where
+  "C"."C_MKTSEGMENT" = 'HOUSEHOLD'
+  and "C"."C_CUSTKEY" = "O"."O_CUSTKEY"
+  and "L"."L_ORDERKEY" = "O"."O_ORDERKEY"
+  and "O"."O_ORDERDATE" < date '1995-03-25'
+  and "L"."L_SHIPDATE" > date '1995-03-25'
+
+group by
+  "L"."L_ORDERKEY",
+  "O"."O_ORDERDATE",
+  "O"."O_SHIPPRIORITY"
+order by
+  "REVENUE" desc,
+  "O"."O_ORDERDATE"
+limit 10