feat(ilp): QWiP store-and-forward client buffer (#17)

Author: bluestreak01

.claude/skills/review-pr/SKILL.md

@@ -21,6 +21,7 @@ core/questdb/client/bin-local
 core/cmake-build-debug
 core/cmake-build-debug-coverage
 core/cmake-build-release
+core/build_native
 core/CMakeCache.txt
 **/.project
 **/.settings

.gitignore

@@ -84,7 +84,16 @@ stages:
                 maven | "$(Agent.OS)"
               path: $(HOME)/.m2/repository
             displayName: "Cache Maven repository"
-          - script: git clone --depth 1 https://github.com/questdb/questdb.git ./questdb
+          - bash: |
+              BRANCH="${SYSTEM_PULLREQUEST_SOURCEBRANCH:-$BUILD_SOURCEBRANCHNAME}"
+              BRANCH="${BRANCH#refs/heads/}"
+              if git ls-remote --exit-code --heads https://github.com/questdb/questdb.git "$BRANCH" >/dev/null 2>&1; then
+                echo "Cloning matching questdb branch: $BRANCH"
+                git clone --depth 1 --branch "$BRANCH" https://github.com/questdb/questdb.git ./questdb
+              else
+                echo "No matching questdb branch '$BRANCH', falling back to master"
+                git clone --depth 1 https://github.com/questdb/questdb.git ./questdb
+              fi
             displayName: git clone questdb
           - task: Maven@3
             displayName: "Update client version"

ci/run_tests_pipeline.yaml

@@ -53,6 +53,7 @@ set(
         src/main/c/share/cpprt_overrides.cpp
         src/main/c/share/byte_sink.cpp
         src/main/c/share/byte_sink.h
+        src/main/c/share/crc32c.c
 )
 
 # libzstd is included via a git submodule at src/main/c/share/zstd (pinned to

core/CMakeLists.txt

@@ -36,6 +36,7 @@
         <argLine>-ea -Dfile.encoding=UTF-8 -XX:+UseParallelGC -Dslf4j.provider=ch.qos.logback.classic.spi.LogbackServiceProvider</argLine>
         <test.exclude>None</test.exclude>
         <test.include>%regex[.*[^o].class]</test.include><!-- exclude module-info.class-->
+        <jmh.version>1.37</jmh.version>
     </properties>
 
     <version>1.2.1-SNAPSHOT</version>
@@ -88,6 +89,13 @@
                     <excludes>
                         <exclude>${excludeTestPattern1}</exclude>
                     </excludes>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>org.openjdk.jmh</groupId>
+                            <artifactId>jmh-generator-annprocess</artifactId>
+                            <version>${jmh.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
                 </configuration>
             </plugin>
             <plugin>
@@ -434,5 +442,17 @@
             <version>1.5.25</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-core</artifactId>
+            <version>${jmh.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-generator-annprocess</artifactId>
+            <version>${jmh.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

core/pom.xml

@@ -22,10 +22,296 @@
  *
  ******************************************************************************/
 
+#define _GNU_SOURCE
+
 #include <unistd.h>
+#include <sys/stat.h>
+#include <sys/file.h>
+#include <sys/mman.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <dirent.h>
+#include <stdint.h>
+
 #include "files.h"
 
+/* Mirror of io.questdb.client.std.Files.MAP_RO / MAP_RW. Hard-coded rather
+ * than #include'd from a javah-generated header because this file does not
+ * pull in any generated symbols (the rest of the file works the same way). */
+#define QDB_MAP_RO 1
+#define QDB_MAP_RW 2
+
+#define RESTARTABLE(_expr_, _rc_) \
+    do { _rc_ = (_expr_); } while ((_rc_) == -1 && errno == EINTR)
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_close0
         (JNIEnv *e, jclass cl, jint fd) {
     return close((int) fd);
 }
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRO0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_RDONLY), fd);
+    return (jint) fd;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openRW0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | O_RDWR, 0644), fd);
+    return (jint) fd;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openAppend0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | O_WRONLY | O_APPEND, 0644), fd);
+    return (jint) fd;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_openCleanRW0
+        (JNIEnv *e, jclass cl, jlong lpszName, jlong size) {
+    int fd;
+    RESTARTABLE(open((const char *) (uintptr_t) lpszName, O_CREAT | O_TRUNC | O_RDWR, 0644), fd);
+    if (fd < 0) {
+        return -1;
+    }
+    if (size > 0) {
+        int rc;
+        RESTARTABLE(ftruncate(fd, (off_t) size), rc);
+        if (rc != 0) {
+            int saved = errno;
+            close(fd);
+            errno = saved;
+            return -1;
+        }
+    }
+    return (jint) fd;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_read
+        (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len, jlong offset) {
+    // Reject negative len explicitly: jlong is signed but pread takes a
+    // size_t. Without this guard the cast wraps a small negative value
+    // into an enormous unsigned read length and the kernel may either
+    // SEGV on the address space or scribble far past the caller's buffer.
+    // The Win32 path already does this; matching here.
+    if (len < 0) {
+        errno = EINVAL;
+        return -1;
+    }
+    ssize_t res;
+    RESTARTABLE(pread((int) fd, (void *) (uintptr_t) addr, (size_t) len, (off_t) offset), res);
+    return (jlong) res;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_write
+        (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len, jlong offset) {
+    if (len < 0) {
+        errno = EINVAL;
+        return -1;
+    }
+    ssize_t res;
+    RESTARTABLE(pwrite((int) fd, (const void *) (uintptr_t) addr, (size_t) len, (off_t) offset), res);
+    return (jlong) res;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_append
+        (JNIEnv *e, jclass cl, jint fd, jlong addr, jlong len) {
+    if (len < 0) {
+        errno = EINVAL;
+        return -1;
+    }
+    ssize_t res;
+    RESTARTABLE(write((int) fd, (const void *) (uintptr_t) addr, (size_t) len), res);
+    return (jlong) res;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_fsync
+        (JNIEnv *e, jclass cl, jint fd) {
+    int res;
+    RESTARTABLE(fsync((int) fd), res);
+    return res;
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_truncate
+        (JNIEnv *e, jclass cl, jint fd, jlong size) {
+    int res;
+    RESTARTABLE(ftruncate((int) fd, (off_t) size), res);
+    return res == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_allocate
+        (JNIEnv *e, jclass cl, jint fd, jlong size) {
+#if defined(__linux__)
+    int res = posix_fallocate((int) fd, 0, (off_t) size);
+    if (res == 0) {
+        return JNI_TRUE;
+    }
+    if (res != EINVAL && res != EOPNOTSUPP) {
+        errno = res;
+        return JNI_FALSE;
+    }
+    /* fall through to ftruncate */
+#elif defined(__APPLE__)
+    fstore_t fst;
+    fst.fst_flags = F_ALLOCATECONTIG | F_ALLOCATEALL;
+    fst.fst_posmode = F_PEOFPOSMODE;
+    fst.fst_offset = 0;
+    fst.fst_length = (off_t) size;
+    fst.fst_bytesalloc = 0;
+    if (fcntl((int) fd, F_PREALLOCATE, &fst) == -1) {
+        fst.fst_flags = F_ALLOCATEALL;
+        (void) fcntl((int) fd, F_PREALLOCATE, &fst);
+        /* if F_PREALLOCATE fails we still try ftruncate to set logical size */
+    }
+#endif
+    int res2;
+    RESTARTABLE(ftruncate((int) fd, (off_t) size), res2);
+    return res2 == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_length
+        (JNIEnv *e, jclass cl, jint fd) {
+    struct stat st;
+    if (fstat((int) fd, &st) != 0) {
+        return -1;
+    }
+    return (jlong) st.st_size;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_length0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    struct stat st;
+    if (stat((const char *) (uintptr_t) lpszName, &st) != 0) {
+        return -1;
+    }
+    return (jlong) st.st_size;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_lock
+        (JNIEnv *e, jclass cl, jint fd) {
+    return flock((int) fd, LOCK_EX | LOCK_NB);
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_mkdir0
+        (JNIEnv *e, jclass cl, jlong lpszPath, jint mode) {
+    return mkdir((const char *) (uintptr_t) lpszPath, (mode_t) mode);
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_exists0
+        (JNIEnv *e, jclass cl, jlong lpszPath) {
+    return access((const char *) (uintptr_t) lpszPath, F_OK) == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jboolean JNICALL Java_io_questdb_client_std_Files_remove0
+        (JNIEnv *e, jclass cl, jlong lpszPath) {
+    return remove((const char *) (uintptr_t) lpszPath) == 0 ? JNI_TRUE : JNI_FALSE;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_rename0
+        (JNIEnv *e, jclass cl, jlong lpszOld, jlong lpszNew) {
+    return rename((const char *) (uintptr_t) lpszOld, (const char *) (uintptr_t) lpszNew);
+}
+
+typedef struct {
+    DIR *dir;
+    struct dirent *entry;
+} qdb_find_t;
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_findFirst0
+        (JNIEnv *e, jclass cl, jlong lpszName) {
+    DIR *dir = opendir((const char *) (uintptr_t) lpszName);
+    if (!dir) {
+        return 0;
+    }
+    qdb_find_t *find = (qdb_find_t *) malloc(sizeof(qdb_find_t));
+    if (!find) {
+        closedir(dir);
+        return 0;
+    }
+    find->dir = dir;
+    errno = 0;
+    find->entry = readdir(dir);
+    if (!find->entry) {
+        int saved = errno;
+        closedir(dir);
+        free(find);
+        errno = saved;
+        return 0;
+    }
+    return (jlong) (uintptr_t) find;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_findNext
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (!find) {
+        return -1;
+    }
+    errno = 0;
+    find->entry = readdir(find->dir);
+    if (find->entry) {
+        return 1;
+    }
+    return errno == 0 ? 0 : -1;
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_findName
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (!find || !find->entry) {
+        return 0;
+    }
+    return (jlong) (uintptr_t) find->entry->d_name;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_findType
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (!find || !find->entry) {
+        return 0;
+    }
+    return (jint) find->entry->d_type;
+}
+
+JNIEXPORT void JNICALL Java_io_questdb_client_std_Files_findClose
+        (JNIEnv *e, jclass cl, jlong findPtr) {
+    qdb_find_t *find = (qdb_find_t *) (uintptr_t) findPtr;
+    if (find) {
+        closedir(find->dir);
+        free(find);
+    }
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_getPageSize0
+        (JNIEnv *e, jclass cl) {
+    long sz = sysconf(_SC_PAGESIZE);
+    return (jlong) (sz > 0 ? sz : 4096);
+}
+
+JNIEXPORT jlong JNICALL Java_io_questdb_client_std_Files_mmap0
+        (JNIEnv *e, jclass cl, jint fd, jlong len, jlong offset, jint flags, jlong baseAddress) {
+    int prot = 0;
+    if (flags == QDB_MAP_RO) {
+        prot = PROT_READ;
+    } else if (flags == QDB_MAP_RW) {
+        prot = PROT_READ | PROT_WRITE;
+    }
+    void *addr = mmap((void *) (uintptr_t) baseAddress, (size_t) len, prot, MAP_SHARED, (int) fd, (off_t) offset);
+    /* MAP_FAILED is (void *) -1; cast to jlong gives -1 sentinel matching FAILED_MMAP_ADDRESS. */
+    return (jlong) (intptr_t) addr;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_munmap0
+        (JNIEnv *e, jclass cl, jlong address, jlong len) {
+    return munmap((void *) (uintptr_t) address, (size_t) len);
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_std_Files_msync
+        (JNIEnv *e, jclass cl, jlong addr, jlong len, jboolean async) {
+    return msync((void *) (uintptr_t) addr, (size_t) len, async ? MS_ASYNC : MS_SYNC);
+}