PostgreSQL 16 compatibility (#200)

* add pg16 compatibility to columnar
* update pg_ivm to 1.7.0
* update pgsql-http to 1.6.0
* skip acceptance tests for extensions that don't work in pg 16

.github/workflows/ci.yaml

@@ -42,7 +42,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        postgres: ["13", "14", "15"]
+        postgres: ["13", "14", "15", "16"]
     runs-on: ubuntu-latest
     timeout-minutes: 15
     steps:
@@ -69,7 +69,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        postgres: ["13", "14", "15"]
+        postgres: ["13", "14", "15", "16"]
     runs-on: ubuntu-latest
     timeout-minutes: 15
     steps:
@@ -208,7 +208,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        postgres: ["13", "14", "15"]
+        postgres: ["13", "14", "15", "16"]
     runs-on: ubuntu-latest
     timeout-minutes: 15
     steps:

.golangci_lint_version

@@ -1 +1 @@
-v1.53.2
+v1.54.2

Dockerfile

@@ -25,5 +25,5 @@ ARG POSTGRES_BASE_VERSION
 ARG TIMESTAMP=1
 COPY third-party/pgxman /tmp/pgxman/
 RUN curl -sfL https://github.com/pgxman/release/releases/latest/download/install.sh | sh -s -- /tmp/pgxman/pgxman_${POSTGRES_BASE_VERSION}.yaml && \
-  pgxman install pgsql-http=1.5.0 --pg ${POSTGRES_BASE_VERSION} --yes && \
+  pgxman install pgsql-http=1.6.0 --pg ${POSTGRES_BASE_VERSION} --yes && \
   rm -rf /tmp/pgxman

acceptance/go.mod

@@ -1,6 +1,6 @@
 module github.com/hydradatabase/hydra/acceptance
 
-go 1.20
+go 1.21
 
 require (
 	github.com/google/uuid v1.3.0

acceptance/go.sum

@@ -1,5 +1,6 @@
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
@@ -20,10 +21,12 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
 golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/text v0.3.8 h1:nAL+RVCQ9uMn3vJZbV+MRnydTJFPf8qqY42YiA6MrqY=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

acceptance/shared/cases.go

@@ -11,20 +11,31 @@ import (
 	"github.com/jackc/pgx/v5"
 )
 
+const (
+	PGVersion13 PGVersion = "13"
+	PGVersion14 PGVersion = "14"
+	PGVersion15 PGVersion = "15"
+	PGVersion16 PGVersion = "16"
+)
+
+type PGVersion string
+
 // A Case describes an acceptance test case. If a Validate function is provided
 // then the test will call that function and expect it to handle test failues.
 // Otherwise the case will fail if pool.Exec fails on the SQL.
 type Case struct {
-	Name     string                          // name of the test
-	SQL      string                          // SQL to run during the test
-	Validate func(t *testing.T, row pgx.Row) // optional validation function
-	Skip     bool                            // whether this case should be skipped
+	Name             string                          // name of the test
+	SQL              string                          // SQL to run during the test
+	Validate         func(t *testing.T, row pgx.Row) // optional validation function
+	Skip             bool                            // whether this case should be skipped
+	TargetPGVersions []PGVersion                     // target PG version
 }
 
 // AcceptanceCases describe the shared acceptance criteria for any Hydra-based
 // images.
 func AcceptanceCases() []Case {
 	cases := []Case{
+		// columnar
 		{
 			Name: "columnar ext available",
 			SQL: `
@@ -125,6 +136,7 @@ SELECT columnar.alter_columnar_table_set(
     stripe_row_limit => 10000);
 			`,
 		},
+		// mysql_fdw
 		{
 			Name: "mysql_fdw available",
 			SQL: `
@@ -218,11 +230,13 @@ SELECT * FROM warehouse ORDER BY warehouse_id LIMIT 1;
 				}
 			},
 		},
+		// multicorn
 		{
 			Name: "multicorn ext available",
 			SQL: `
 SELECT count(1) FROM pg_available_extensions WHERE name = 'multicorn';
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 			Validate: func(t *testing.T, row pgx.Row) {
 				var count int
 				if err := row.Scan(&count); err != nil {
@@ -239,6 +253,7 @@ SELECT count(1) FROM pg_available_extensions WHERE name = 'multicorn';
 			SQL: `
 CREATE EXTENSION multicorn;
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "multicorn ext enabled",
@@ -255,6 +270,7 @@ SELECT count(1) FROM pg_extension WHERE extname = 'multicorn';
 					t.Errorf("columnar ext should exist")
 				}
 			},
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "create multicorn s3 ext foreign table",
@@ -274,6 +290,7 @@ create foreign table s3 (
   filename 'test.csv'
 );
 		`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "create multicorn gspreadsheet ext foreign table",
@@ -290,7 +307,9 @@ CREATE FOREIGN TABLE test_spreadsheet (
   serviceaccount '{}'
 );
 		`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
+		// parquest_s3_fdw
 		{
 			Name: "parquet_s3_fdw available",
 			SQL: `
@@ -306,12 +325,14 @@ SELECT count(1) FROM pg_available_extensions WHERE name = 'parquet_s3_fdw';
 					t.Errorf("parquet_s3_fdw ext should exist")
 				}
 			},
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "enable parquet_s3_fdw ext",
 			SQL: `
 CREATE EXTENSION parquet_s3_fdw;
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "create parquet_s3_fdw foreign table with no aws creds",
@@ -326,12 +347,14 @@ OPTIONS (
 	dirname 's3://FAKE'
 );
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "parquet_s3_fdw foreign table with no aws creds raises error",
 			SQL: `
 SELECT count(*) FROM userdata_1;
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 			Validate: func(t *testing.T, row pgx.Row) {
 				err := row.Scan()
 				if err == nil {
@@ -356,12 +379,14 @@ OPTIONS (
 	dirname 's3://FAKE'
 );
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "parquet_s3_fdw foreign table with empty aws creds raises error",
 			SQL: `
 SELECT count(*) FROM userdata_2;
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 			Validate: func(t *testing.T, row pgx.Row) {
 				err := row.Scan()
 				if err == nil {
@@ -373,6 +398,7 @@ SELECT count(*) FROM userdata_2;
 				}
 			},
 		},
+		// pgvector
 		{
 			Name: "pg_vector available",
 			SQL: `
@@ -466,13 +492,15 @@ OPTIONS (
 	dirname 's3://%s/parquet'
 );
 			`, awsRegion, awsAccessKey, awsSecretKey, awsS3Bucket),
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 		},
 		{
 			Name: "validate parquet_s3_fdw happy path",
 			Skip: shouldSkipParquetS3Tests,
 			SQL: `
 SELECT count(*) FROM userdata_3;
 			`,
+			TargetPGVersions: []PGVersion{PGVersion13, PGVersion14, PGVersion15},
 			Validate: func(t *testing.T, row pgx.Row) {
 				var count int
 				if err := row.Scan(&count); err != nil {

acceptance/shared/pg.go

@@ -0,0 +1,29 @@
+package shared
+
+import (
+	"context"
+	"regexp"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+var (
+	regexpPGVersion = regexp.MustCompile(`^PostgreSQL (\d+)`)
+)
+
+func QueryPGVersion(t *testing.T, ctx context.Context, pool *pgxpool.Pool) PGVersion {
+	row := pool.QueryRow(ctx, "SELECT VERSION();")
+
+	var version string
+	if err := row.Scan(&version); err != nil {
+		t.Fatal(err)
+	}
+
+	matches := regexpPGVersion.FindStringSubmatch(version)
+	if len(matches) == 0 {
+		t.Fatalf("failed to parse pg_config --version output: %s", version)
+	}
+
+	return PGVersion(matches[1])
+}

acceptance/shared/runner.go

@@ -2,6 +2,7 @@ package shared
 
 import (
 	"context"
+	"slices"
 	"testing"
 	"time"
 
@@ -37,15 +38,13 @@ func RunAcceptanceTests(t *testing.T, ctx context.Context, cm DockerComposeManag
 	})
 
 	pool := cm.PGPool()
+	ver := QueryPGVersion(t, ctx, pool)
 
 	cases := append(AcceptanceCases(), additionalCases...)
-
 	for _, c := range cases {
 		c := c
 		t.Run(c.Name, func(t *testing.T) {
-			if c.Skip {
-				t.Skip("Test skipped")
-			}
+			checkSkipTest(t, c, ver)
 
 			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 			defer cancel()
@@ -72,9 +71,13 @@ func RunUpgradeTests(t *testing.T, ctx context.Context, cm DockerComposeManager)
 
 		for _, c := range BeforeUpgradeCases {
 			c := c
+
 			pool := cm.PGPool()
+			ver := QueryPGVersion(t, ctx, pool)
 
 			t.Run(c.Name, func(t *testing.T) {
+				checkSkipTest(t, c, ver)
+
 				ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 				defer cancel()
 
@@ -96,9 +99,13 @@ func RunUpgradeTests(t *testing.T, ctx context.Context, cm DockerComposeManager)
 
 		for _, c := range AfterUpgradeCases {
 			c := c
+
 			pool := cm.PGPool()
+			ver := QueryPGVersion(t, ctx, pool)
 
 			t.Run(c.Name, func(t *testing.T) {
+				checkSkipTest(t, c, ver)
+
 				ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 				defer cancel()
 
@@ -113,3 +120,13 @@ func RunUpgradeTests(t *testing.T, ctx context.Context, cm DockerComposeManager)
 		}
 	})
 }
+
+func checkSkipTest(t *testing.T, c Case, ver PGVersion) {
+	if c.Skip {
+		t.Skip("Test skipped")
+	}
+
+	if len(c.TargetPGVersions) > 0 && !slices.Contains(c.TargetPGVersions, ver) {
+		t.Skip("Skipping test due to unsupported PG version")
+	}
+}

columnar/configure

@@ -2802,7 +2802,7 @@ if test -z "$version_num"; then
   as_fn_error $? "Could not detect PostgreSQL version from pg_config." "$LINENO" 5
 fi
 
-if test "$version_num" != '13' -a "$version_num" != '14' -a "$version_num" != '15'; then
+if test "$version_num" != '13' -a "$version_num" != '14' -a "$version_num" != '15' -a "$version_num" != '16'; then
    as_fn_error $? "Citus is not compatible with the detected PostgreSQL version ${version_num}." "$LINENO" 5
 else
    { printf "%s\n" "$as_me:${as_lineno-$LINENO}: building against PostgreSQL $version_num" >&5

columnar/configure.in

@@ -74,7 +74,7 @@ if test -z "$version_num"; then
   AC_MSG_ERROR([Could not detect PostgreSQL version from pg_config.])
 fi
 
-if test "$version_num" != '13' -a "$version_num" != '14' -a "$version_num" != '15'; then
+if test "$version_num" != '13' -a "$version_num" != '14' -a "$version_num" != '15' -a "$version_num" != '16'; then
    AC_MSG_ERROR([Citus is not compatible with the detected PostgreSQL version ${version_num}.])
 else
    AC_MSG_NOTICE([building against PostgreSQL $version_num])

columnar/src/backend/columnar/columnar.c

@@ -47,7 +47,7 @@ bool columnar_enable_parallel_execution = true;
 int columnar_min_parallel_processes = 8;
 bool columnar_enable_vectorization = true;
 bool columnar_enable_dml = true;
-bool columnar_enable_page_cache = true;
+bool columnar_enable_page_cache = false;
 int columnar_page_cache_size = 200U;
 
 static const struct config_enum_entry columnar_compression_options[] =
@@ -132,7 +132,7 @@ columnar_guc_init()
 							 &columnar_enable_parallel_execution,
 							 true,
 							 PGC_USERSET,
-							 GUC_NO_SHOW_ALL,
+							 GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE,
 							 NULL, 
 							 NULL, 
 							 NULL);
@@ -156,7 +156,7 @@ columnar_guc_init()
 							 &columnar_enable_vectorization,
 							 true,
 							 PGC_USERSET,
-							 GUC_NO_SHOW_ALL,
+							 GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE,
 							 NULL, 
 							 NULL, 
 							 NULL);
@@ -167,7 +167,7 @@ columnar_guc_init()
 							&columnar_enable_dml,
 							true,
 							PGC_USERSET,
-							GUC_NO_SHOW_ALL,
+							GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE,
 							NULL, 
 							NULL, 
 							NULL);

columnar/src/backend/columnar/columnar_compression.c

@@ -14,6 +14,8 @@
 #include "postgres.h"
 
 #include "citus_version.h"
+#include "pg_version_compat.h"
+
 #include "common/pg_lzcompress.h"
 #include "lib/stringinfo.h"
 
@@ -27,6 +29,9 @@
 #include <zstd.h>
 #endif
 
+#if PG_VERSION_NUM >= PG_VERSION_16
+#include "varatt.h"
+#endif
 /*
  *	The information at the start of the compressed data. This decription is taken
  *	from pg_lzcompress in pre-9.5 version of PostgreSQL.