fix(ilp): UnsupportedOperationException when writing to non-WAL table via ILP/TCP (#5742)

kafka1991 · web-flow · commit e461e7dc14f0 · 2025-08-11T14:48:20.000+01:00
diff --git a/core/src/main/java/io/questdb/cutlass/line/tcp/TableUpdateDetails.java b/core/src/main/java/io/questdb/cutlass/line/tcp/TableUpdateDetails.java
@@ -281,7 +281,7 @@ public boolean isDropped() {
     }
 
     public boolean isWal() {
-        return writerThreadId == -1;
+        return tableToken.isWal();
     }
 
     public boolean isWriterInError() {
diff --git a/core/src/test/java/io/questdb/test/AbstractCairoTest.java b/core/src/test/java/io/questdb/test/AbstractCairoTest.java
@@ -1207,11 +1207,11 @@ protected static void assertMemoryLeak(@Nullable FilesFacade ff, TestUtils.LeakP
             try {
                 code.run();
                 forEachNode(node -> releaseInactive(node.getEngine()));
-                CLOSEABLES.forEach(Misc::free);
             } catch (Throwable th) {
                 LOG.error().$("Error in test: ").$(th).$();
                 throw th;
             } finally {
+                CLOSEABLES.forEach(Misc::free);
                 forEachNode(node -> node.getEngine().clear());
                 AbstractCairoTest.ff = ffBefore;
             }
diff --git a/core/src/test/java/io/questdb/test/cairo/fuzz/AbstractFuzzTest.java b/core/src/test/java/io/questdb/test/cairo/fuzz/AbstractFuzzTest.java
@@ -194,7 +194,7 @@ protected void runFuzz(Rnd rnd) throws Exception {
                 WorkerPoolUtils.setupWriterJobs(sharedWorkerPool, engine);
                 sharedWorkerPool.start(LOG);
 
-                int size = rnd.nextInt(16 * 1024 * 1024);
+                int size = rnd.nextInt(8 * 1024 * 1024);
                 node1.setProperty(PropertyKey.DEBUG_CAIRO_O3_COLUMN_MEMORY_SIZE, size);
                 setZeroWalPurgeInterval();
                 fuzzer.runFuzz(getTestName(), rnd);
diff --git a/core/src/test/java/io/questdb/test/cutlass/line/tcp/LineTcpReceiverTest.java b/core/src/test/java/io/questdb/test/cutlass/line/tcp/LineTcpReceiverTest.java
@@ -28,6 +28,7 @@
 import io.questdb.cairo.ColumnType;
 import io.questdb.cairo.PartitionBy;
 import io.questdb.cairo.TableReader;
+import io.questdb.cairo.TableToken;
 import io.questdb.cairo.TableWriter;
 import io.questdb.cairo.TableWriterAPI;
 import io.questdb.cairo.pool.PoolListener;
@@ -260,6 +261,104 @@ public void testColumnTypeStaysTheSameWhileColumnAdded() throws Exception {
         }, false, 250);
     }
 
+    @Test
+    public void testConcurrentWriteAndTruncate() throws Exception {
+        Assume.assumeTrue(walEnabled);
+        runInContext((receiver) -> {
+            String tableName = "concurrent_test";
+            final int iterations = 500;
+            final SOCountDownLatch startLatch = new SOCountDownLatch(2);
+            final SOCountDownLatch finishLatch = new SOCountDownLatch(2);
+            final AtomicInteger errorCount = new AtomicInteger(0);
+            String initialData = tableName + ",location=init_location,symbol=test temperature=20.0 1465839830100000000\n";
+            send(initialData, tableName);
+            mayDrainWalQueue();
+
+            Thread ilpWriteThread = new Thread(() -> {
+                try {
+                    startLatch.countDown();
+                    startLatch.await();
+
+                    try (Socket socket = getSocket()) {
+                        for (int i = 0; i < iterations; i++) {
+                            try {
+                                // Use duplicate timestamp and symbols to trigger metadata operations
+                                String lineData = tableName + ",location=test_location,symbol=sym" + (i % 3) +
+                                        " temperature=" + (20.0 + i) + " " +
+                                        (1465839830102000000L + (i % 5)) + "\n";
+                                sendToSocket(socket, lineData);
+
+                                if (i % 50 == 0) {
+                                    mayDrainWalQueue();
+                                }
+
+                                if (i % 20 == 0) {
+                                    Os.sleep(1);
+                                }
+                            } catch (Throwable e) {
+                                errorCount.incrementAndGet();
+                            }
+                        }
+                    }
+                } catch (Throwable e) {
+                    errorCount.incrementAndGet();
+                } finally {
+                    Path.clearThreadLocals();
+                    finishLatch.countDown();
+                }
+            });
+
+            Thread truncateThread = new Thread(() -> {
+                try {
+                    startLatch.countDown();
+                    startLatch.await();
+                    boolean drop = false;
+
+                    for (int i = 0; i < iterations; i++) {
+                        try {
+                            // Alternate between truncate and drop operations
+                            if (i % 15 == 0) {
+                                TableToken tt = engine.getTableTokenIfExists(tableName);
+                                if (tt != null) {
+                                    try (TableWriterAPI writer = getTableWriterAPI(tableName)) {
+                                        writer.truncateSoft();
+                                    }
+                                    mayDrainWalQueue();
+                                    drop = true;
+                                }
+                            } else if (i % 25 == 0 && drop) {
+                                TableToken tt = engine.getTableTokenIfExists(tableName);
+                                if (tt != null) {
+                                    engine.dropTableOrMatView(path, tt);
+                                    drop = false;
+                                }
+                            }
+                            Os.sleep(2);
+                        } catch (Throwable e) {
+                            errorCount.incrementAndGet();
+                        }
+                    }
+                } catch (Throwable e) {
+                    errorCount.incrementAndGet();
+                } finally {
+                    Path.clearThreadLocals();
+                    finishLatch.countDown();
+                }
+            });
+
+            ilpWriteThread.start();
+            truncateThread.start();
+
+            try {
+                finishLatch.await();
+                Assert.assertEquals(0, errorCount.get());
+            } finally {
+                ilpWriteThread.interrupt();
+                truncateThread.interrupt();
+            }
+        });
+    }
+
     @Test
     public void testCreationAttemptNonPartitionedTableWithWal() throws Exception {
         Assume.assumeTrue(walEnabled);

Original file line number	Diff line number	Diff line change
`@@ -281,7 +281,7 @@ public boolean isDropped() {`
`281`	`281`	`}`
`282`	`282`
`283`	`283`	`public boolean isWal() {`
`284`		`- return writerThreadId == -1;`
	`284`	`+ return tableToken.isWal();`
`285`	`285`	`}`
`286`	`286`
`287`	`287`	`public boolean isWriterInError() {`