Changes to move to protobufs for DDL messages rather than JSON

.github/workflows/integration-test.yml

@@ -0,0 +1,36 @@
+name: Integration Tests
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+  packages: write
+  checks: write
+
+jobs:
+  debug-build:
+    uses: ./.github/workflows/rw-build-container.yml
+    secrets: inherit
+    permissions:
+      contents: read
+      packages: write
+    with:
+      base_image_tag: ${{ vars.IMAGE_TAG_DEBUG }}
+      build_type: "debug"
+
+  integration-test:
+    uses: ./.github/workflows/rw-run-integration-test.yml
+    needs: [ debug-build ]
+    secrets: inherit
+    permissions:
+      contents: read
+      packages: write
+      checks: write
+    with:
+      image_tag: ${{ needs.debug-build.outputs.output_image_tag }}
+      config_to_run: "nightly"
+      system_settings_override: '.github/workflows/nightly.overrides.json'
+      iterations: "1"
+      build_type: "debug"
+      reuse_code: true
+      timeout_minutes: 120

.github/workflows/rw-run-integration-test.yml

@@ -210,7 +210,15 @@ jobs:
           cd python/testing
           python3 ./springtail.py -f ../../system.json.test --load-redis --build-dir ../../${CI_BUILD_TYPE}
           python3 ./springtail.py -f ../../system.json.test --install-triggers --build-dir ../../${CI_BUILD_TYPE}
-          python3 ./springtail.py --start -f ../../system.json.test -b ../../${CI_BUILD_TYPE} -s data_init.sql
+          if ! python3 ./springtail.py --start --startup-timeout 60 -f ../../system.json.test -b ../../${CI_BUILD_TYPE} -s data_init.sql; then
+            echo "*** Startup failed, dumping daemon logs"
+            output=$(python3 springtail.py --dump -f ../../system.json.test -b ../../${CI_BUILD_TYPE}) || true
+            fname=$(echo $output | cut -f2 -d':' | xargs | head -1)
+            if [ -f "$fname" ]; then
+              mv $fname springtail-startup-failure-${_LOG_SHA}.tar.gz
+            fi
+            exit 1
+          fi
           python3 ./springtail.py --start -f ../../system.json.test -b ../../${CI_BUILD_TYPE} --check
 
           # Run integration tests

docker/Dockerfile.ci

@@ -46,8 +46,9 @@ RUN cd ${CI_BUILD_TYPE} && \
     pip3 install --break-system-packages -r python/performance/requirements.txt && \
     rm -rf $CI_WORKSPACE /home/dev/${CI_BUILD_TYPE}/src /home/dev/external/vcpkg/buildtrees /home/dev/external/vcpkg/packages /home/dev/external/vcpkg/downloads
 
-# Pull back the libpg_query
-RUN cd /home/dev/external && git clone https://github.com/Springtail-inc/libpg_query.git
+# Pull back the libpg_query (cmake FetchContent already populates it via the
+# external/ symlink during the build above; only clone if it isn't there)
+RUN cd /home/dev/external && [ -d libpg_query/.git ] || git clone https://github.com/Springtail-inc/libpg_query.git
 
 # Set tini as the container's entrypoint
 ENTRYPOINT ["/tini", "--"]

docker/ansible/roles/custom-pg/defaults/main.yml

@@ -1,3 +1,3 @@
-custom_pg_package_url: "https://s3.us-east-1.amazonaws.com/public-share.springtail.io/postgres_apt.tgz"
+custom_pg_package_url: "https://s3.us-east-1.amazonaws.com/public-share.springtail.io/postgresql-16_16.9-1_93cd1a8d-78c2-4979-8204-fa73ab5521a4.tar.gz"
 superuser_name: "springtail"
 superuser_password: "springtail"

docker/ansible/roles/custom-pg/tasks/main.yml

@@ -20,14 +20,14 @@
     path: /tmp/springtail-pg.tar.gz
   register: pg_tarball
 
-- name: Download Custom PG .deb packages from URL
+- name: Download Custom PG source from URL
   ansible.builtin.get_url:
     url: "{{ custom_pg_package_url }}"
     dest: /tmp/springtail-pg.tar.gz
     mode: '0644'
   when: not pg_tarball.stat.exists
 
-- name: Install PG runtime dependencies
+- name: Install PG build dependencies
   ansible.builtin.apt:
     name:
       - libreadline-dev
@@ -40,36 +40,38 @@
       - liblz4-dev
       - libzstd-dev
       - uuid-dev
+      - bison
+      - flex
     state: present
 
-- name: Install Custom PG from .deb packages
+- name: Build and install Custom PG from source
   ansible.builtin.shell: |
     set -e
     cd /tmp
-    mkdir -p postgres_apt
-    tar xzf springtail-pg.tar.gz -C postgres_apt
-    cd postgres_apt
-
-    # Install packages in dependency order, skipping debug symbol (.ddeb) packages
-    dpkg -i libpq5_*.deb || apt-get install -f -y
-    dpkg -i libpq-dev_*.deb || apt-get install -f -y
-    dpkg -i libecpg6_*.deb libecpg-compat3_*.deb libpgtypes3_*.deb libecpg-dev_*.deb || apt-get install -f -y
-    dpkg -i postgresql-client-16_*.deb || apt-get install -f -y
-    dpkg -i postgresql-16_*.deb || apt-get install -f -y
-    dpkg -i postgresql-server-dev-16_*.deb || apt-get install -f -y
-
-    # Optional PL language packages — install if present
-    dpkg -i postgresql-plpython3-16_*.deb 2>/dev/null || true
-    dpkg -i postgresql-plperl-16_*.deb 2>/dev/null || true
-    dpkg -i postgresql-pltcl-16_*.deb 2>/dev/null || true
-    dpkg -i postgresql-doc-16_*.deb 2>/dev/null || true
-
-    # Fix any remaining dependency issues
-    apt-get install -f -y
-
-    # Clean up
+    tar xzf springtail-pg.tar.gz
+    cd postgresql-16.9
+    ./configure \
+      --prefix=/usr/lib/postgresql/16 \
+      --bindir=/usr/lib/postgresql/16/bin \
+      --datadir=/usr/share/postgresql/16 \
+      --sysconfdir=/etc/postgresql-common \
+      --libdir=/usr/lib/postgresql/16/lib \
+      --includedir=/usr/include/postgresql/16 \
+      --with-openssl \
+      --with-libxml \
+      --with-libxslt \
+      --with-icu \
+      --with-uuid=e2fs \
+      --with-lz4 \
+      --with-zstd \
+      --with-systemd \
+      --with-pgport=5432
+    make -j$(nproc) world
+    make install-world
+    # Link system zoneinfo as PG timezone data (source tarball lacks timezone source files)
+    ln -sfn /usr/share/zoneinfo /usr/share/postgresql/16/timezone
     cd /tmp
-    rm -rf postgres_apt springtail-pg.tar.gz
+    rm -rf postgresql-16.9 springtail-pg.tar.gz
 
 - name: Enable PG16
   ansible.builtin.copy:

include/common/ddl_helpers.hh

@@ -0,0 +1,81 @@
+#pragma once
+
+#include <optional>
+#include <string>
+#include <ddl.pb.h>
+
+namespace springtail {
+
+// Extract the table ID from a DDL operation, if the operation references a specific table.
+// Returns nullopt for namespace, user type, and partition operations that don't carry a tid.
+inline std::optional<uint64_t> get_table_id(const proto::DDLOperation& op) {
+    switch (op.operation_case()) {
+        case proto::DDLOperation::kCreateTable:
+            return op.create_table().tid();
+        case proto::DDLOperation::kDropTable:
+            return op.drop_table().tid();
+        case proto::DDLOperation::kRenameTable:
+            return op.rename_table().tid();
+        case proto::DDLOperation::kSetNamespace:
+            return op.set_namespace().tid();
+        case proto::DDLOperation::kSetRlsEnabled:
+            return op.set_rls_enabled().tid();
+        case proto::DDLOperation::kSetRlsForced:
+            return op.set_rls_forced().tid();
+        case proto::DDLOperation::kColumnAdd:
+            return op.column_add().tid();
+        case proto::DDLOperation::kColumnDrop:
+            return op.column_drop().tid();
+        case proto::DDLOperation::kColumnRename:
+            return op.column_rename().tid();
+        case proto::DDLOperation::kColumnNullable:
+            return op.column_nullable().tid();
+        case proto::DDLOperation::kResync:
+            return op.resync().tid();
+        case proto::DDLOperation::kNoChange:
+            return op.no_change().tid();
+        case proto::DDLOperation::kResyncPartitions:
+        case proto::DDLOperation::kNamespaceCreate:
+        case proto::DDLOperation::kNamespaceAlter:
+        case proto::DDLOperation::kNamespaceDrop:
+        case proto::DDLOperation::kUsertypeCreate:
+        case proto::DDLOperation::kUsertypeAlter:
+        case proto::DDLOperation::kUsertypeDrop:
+        case proto::DDLOperation::kAttachPartition:
+        case proto::DDLOperation::kDetachPartition:
+        case proto::DDLOperation::OPERATION_NOT_SET:
+            return std::nullopt;
+    }
+    return std::nullopt;
+}
+
+// Return a human-readable action name for logging purposes.
+inline std::string get_action_name(const proto::DDLOperation& op) {
+    switch (op.operation_case()) {
+        case proto::DDLOperation::kCreateTable:      return "create";
+        case proto::DDLOperation::kDropTable:        return "drop";
+        case proto::DDLOperation::kRenameTable:      return "rename";
+        case proto::DDLOperation::kSetNamespace:     return "set_namespace";
+        case proto::DDLOperation::kSetRlsEnabled:    return "set_rls_enabled";
+        case proto::DDLOperation::kSetRlsForced:     return "set_rls_forced";
+        case proto::DDLOperation::kColumnAdd:        return "col_add";
+        case proto::DDLOperation::kColumnDrop:       return "col_drop";
+        case proto::DDLOperation::kColumnRename:     return "col_rename";
+        case proto::DDLOperation::kColumnNullable:   return "col_nullable";
+        case proto::DDLOperation::kResync:           return "resync";
+        case proto::DDLOperation::kNoChange:         return "no_change";
+        case proto::DDLOperation::kResyncPartitions: return "resync_partitions";
+        case proto::DDLOperation::kNamespaceCreate:  return "ns_create";
+        case proto::DDLOperation::kNamespaceAlter:   return "ns_alter";
+        case proto::DDLOperation::kNamespaceDrop:    return "ns_drop";
+        case proto::DDLOperation::kUsertypeCreate:   return "ut_create";
+        case proto::DDLOperation::kUsertypeAlter:    return "ut_alter";
+        case proto::DDLOperation::kUsertypeDrop:     return "ut_drop";
+        case proto::DDLOperation::kAttachPartition:  return "attach_partition";
+        case proto::DDLOperation::kDetachPartition:  return "detach_partition";
+        case proto::DDLOperation::OPERATION_NOT_SET: return "unknown";
+    }
+    return "unknown";
+}
+
+}  // namespace springtail

include/grpc/grpc_server_manager.hh

@@ -23,6 +23,7 @@ public:
     void shutdown();
 
 private:
+    std::unique_ptr<grpc::ServerBuilder> _create_builder(const std::string& address);
     static std::string read_file_contents(const std::string& path);
 
     int _worker_thread_count = 0;

include/pg_ext/string.hh

@@ -62,13 +62,20 @@ extern "C" PGEXT_API char *upperstr_with_len(const char *str, int len);
 extern "C" PGEXT_API int pg_snprintf(char *str, size_t count, const char *fmt, ...);
 extern "C" PGEXT_API int pg_database_encoding_max_length(void);
 extern "C" PGEXT_API int pg_mblen(const char *mbstr);
+extern "C" PGEXT_API int pg_mblen_cstr(const char *mbstr);
+extern "C" PGEXT_API int pg_mblen_unbounded(const char *mbstr);
+extern "C" PGEXT_API int pg_mblen_range(const char *mbstr, const char *end);
 extern "C" PGEXT_API int pg_mbstrlen_with_len(const char *mbstr, int len);
 extern "C" PGEXT_API int pg_mb2wchar_with_len(const char *mbstr, wchar_t *wstr, int len);
 extern "C" PGEXT_API int pg_wchar2mb_with_len(const wchar_t *wstr, char *mbstr, int len);
 extern "C" PGEXT_API int t_isalnum(const char *ptr);
 extern "C" PGEXT_API int t_isspace(const char *p);
 extern "C" PGEXT_API int t_isdigit(const char *p);
 extern "C" PGEXT_API int t_isalpha(const char *p);
+extern "C" PGEXT_API int t_isalnum_with_len(const char *ptr, int len);
+extern "C" PGEXT_API int t_isspace_with_len(const char *p, int len);
+extern "C" PGEXT_API int t_isdigit_with_len(const char *p, int len);
+extern "C" PGEXT_API int t_isalpha_with_len(const char *p, int len);
 extern "C" PGEXT_API char *str_tolower(const char *buff, size_t nbytes, Oid collid);
 extern "C" PGEXT_API char *str_toupper(const char *buff, size_t nbytes, Oid collid);
 extern "C" PGEXT_API char *pstrdup(const char *in);

include/pg_fdw/pg_ddl_mgr.hh

@@ -6,6 +6,7 @@
 #include <common/object_cache.hh>
 #include <common/libpq_connection.hh>
 
+#include <ddl.pb.h>
 #include <redis/db_state_change.hh>
 
 namespace springtail::pg_fdw {
@@ -88,7 +89,7 @@ namespace springtail::pg_fdw {
 
         struct DBData {
             std::mutex pending_ddls_mutex;                      ///< mutex for applying changes to pending_ddls
-            std::map<uint64_t, nlohmann::json> pending_ddls;    ///< map of xid to pending ddl
+            std::map<uint64_t, std::vector<proto::DDLOperation>> pending_ddls;    ///< map of xid to pending ddl operations
             std::shared_mutex db_mutex;                         ///< mutex for changes to this data structure
             std::string db_name{};                              ///< database name
             std::atomic<redis::db_state_change::DBState> state{redis::db_state_change::DB_STATE_UNKNOWN};
@@ -201,12 +202,12 @@ namespace springtail::pg_fdw {
         /**
          * @brief Helper to apply outstanding DDL changes to the FDW tables.
          * @param db_id The database ID to apply the changes to.
-         * @param schema_xid The XID at which the DDL changes were applied.
-         * @param ddls A JSON array of DDL statements to apply.
+         * @param db_name The database name.
+         * @param xid_map Map of XIDs to DDL operations to apply.
          * @return Status of the operation. True if successful, false otherwise.
          */
         bool _update_schemas(uint64_t db_id, const std::string &db_name,
-                             const std::map<uint64_t, nlohmann::json> &xid_map);
+                             const std::map<uint64_t, std::vector<proto::DDLOperation>> &xid_map);
 
         /** Helper to execute ddl statements for this db */
         /**
@@ -237,15 +238,15 @@ namespace springtail::pg_fdw {
         _get_usertypes(uint64_t db_id, uint64_t xid);
 
         /**
-         * @brief Helper to generate sql statement from json.  Decodes the ddl json.
+         * @brief Helper to generate sql statement from a protobuf DDL operation.
          * @param conn LibPqConnectionPtr connection
          * @param server_name fdw server name
-         * @param ddl json object containing ddl
+         * @param op protobuf DDLOperation
          * @return std::string sql statement
          */
-        std::string _gen_sql_from_json(LibPqConnectionPtr conn,
-                                       const std::string &server_name,
-                                       const nlohmann::json &ddl);
+        std::string _gen_sql_from_ddl(LibPqConnectionPtr conn,
+                                      const std::string &server_name,
+                                      const proto::DDLOperation &op);
 
         /**
          * @brief Function for creating a replicated database
@@ -404,7 +405,7 @@ namespace springtail::pg_fdw {
          * @param db_id - database id
          * @param xid_map - map of xids to ddl
          */
-        void _queue_request(uint64_t db_id, const std::map<uint64_t, nlohmann::json> &xid_map);
+        void _queue_request(uint64_t db_id, const std::map<uint64_t, std::vector<proto::DDLOperation>> &xid_map);
     };
 
 } // springtail::pg_fdw

include/pg_log_mgr/committer.hh

@@ -5,6 +5,7 @@
 #include <pg_log_mgr/index_requests_manager.hh>
 #include <pg_log_mgr/table_copy_tracker.hh>
 #include <pg_log_mgr/xid_ready.hh>
+#include <ddl.pb.h>
 #include <redis/redis_ddl.hh>
 #include <sys_tbl_mgr/table.hh>
 #include <sys_tbl_mgr/mutable_table.hh>
@@ -88,7 +89,7 @@ namespace springtail::committer {
         /**
          * Clear the SysTblMgr::Client cache for any tables with DDL mutations.
          */
-        void _invalidate_systbl_cache(uint64_t db, const nlohmann::json &completed_ddls);
+        void _invalidate_systbl_cache(uint64_t db, const std::vector<proto::DDLOperation> &completed_ddls);
 
         /**
          * @brief Expire dropped table dirs
@@ -97,7 +98,7 @@ namespace springtail::committer {
          * @param completed_ddls DDLs processed
          * @param committed_xid  XID at which ddls were processed
          */
-        void _expire_table_drops(uint64_t db_id, const nlohmann::json &completed_ddls, uint64_t committed_xid);
+        void _expire_table_drops(uint64_t db_id, const std::vector<proto::DDLOperation> &completed_ddls, uint64_t committed_xid);
 
         /**
          * @brief Expire dropped index paths

include/pg_log_mgr/xid_ready.hh

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <common/common.hh>
+#include <ddl.pb.h>
 #include <fmt/ranges.h>
 #include <proto/pg_copy_table.pb.h>
 
@@ -41,7 +42,7 @@ namespace springtail::committer {
          */
         class SwapMsg {
         public:
-            SwapMsg(uint64_t xid, nlohmann::json &&ddls, std::vector<uint32_t>&& tids)
+            SwapMsg(uint64_t xid, std::vector<proto::DDLOperation> &&ddls, std::vector<uint32_t>&& tids)
                 : _xid(xid),
                   _ddls(std::move(ddls)),
                   _tids(std::move(tids))
@@ -51,7 +52,7 @@ namespace springtail::committer {
                 return _xid;
             }
 
-            const nlohmann::json &ddls() const {
+            const std::vector<proto::DDLOperation> &ddls() const {
                 return _ddls;
             }
 
@@ -61,7 +62,7 @@ namespace springtail::committer {
 
         private:
             uint64_t _xid;
-            nlohmann::json _ddls;
+            std::vector<proto::DDLOperation> _ddls;
             std::vector<uint32_t> _tids;
         };