Merge remote-tracking branch 'origin/main' into polyglot/jdbc-v2-placeholder-scan-comments-heredocs

Author: polyglotAI-bot

CHANGELOG.md

@@ -159,6 +159,14 @@
   executes fine. The placeholder scan now skips both token kinds, like the server lexer does; a `$` that does not open a
   heredoc is still treated as an ordinary character (it is a valid identifier character).
   (https://github.com/ClickHouse/clickhouse-java/issues/3009)
+- **[jdbc-v2]** Fixed `INSERT INTO [TABLE] FUNCTION f(...) VALUES (?)` failing with
+  `Code: 60 ... does not exist. (UNKNOWN_TABLE)` when the `beta.row_binary_for_simple_insert` feature was
+  enabled. Neither SQL parser reported a table-function insert target as a function, so the statement was
+  routed to the `RowBinary` writer, which looked the function name (or a placeholder such as `unknown`) up as
+  a table. Both parsers now report such a statement as using a function, so it stays on the regular SQL path;
+  additionally the JavaCC grammar no longer mis-parses `INSERT INTO TABLE FUNCTION f(...)` by consuming
+  `FUNCTION` as the table name. Inserts into a plain table are unaffected and still use the `RowBinary`
+  writer. (https://github.com/ClickHouse/clickhouse-java/issues/3015)
 - **[jdbc-v2]** Fixed `Connection#prepareStatement` throwing a `NullPointerException` for an
   `INSERT ... VALUES (...)` statement whose values list the default JavaCC parser cannot parse — most commonly one
   containing a heredoc string (`$$...$$`), which the grammar has no token for, but also any other unparsable token
@@ -309,6 +317,11 @@
   performs the handshake and runs the post-close action; a concurrent or repeated `close()` returns immediately. A
   `close()` which fails while flushing the remaining data also marks the stream closed and runs the post-close
   action, so the stream cannot stay half-closed. (https://github.com/ClickHouse/clickhouse-java/issues/3055)
+- **[data]** Fixed `NonBlockingPipedOutputStream.close()` not being idempotent under concurrency. Two threads could
+  both flush and mutate the same pending buffer before the reader consumed it, silently replacing the payload with
+  an empty buffer and running the post-close action twice. Exactly one caller now flushes the pending data, enqueues
+  the end-of-stream marker, and runs the post-close action; concurrent or repeated `close()` calls return immediately.
+  (https://github.com/ClickHouse/clickhouse-java/issues/3057)
 - **[jdbc-v2]** Fixed JDBC escape processing rewriting text inside string literals and quoted identifiers. Because
   `PreparedStatement` inlines bound parameters into the statement text, a bound value containing `{fn ` (or `{d '...'}`
   / `{ts '...'}`) was re-read as SQL syntax: the `{fn ` was removed together with the next `}` found anywhere in the

clickhouse-data/src/main/java/com/clickhouse/data/stream/NonBlockingPipedOutputStream.java

@@ -4,6 +4,7 @@
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.clickhouse.data.ClickHouseByteBuffer;
 import com.clickhouse.data.ClickHouseChecker;
@@ -34,6 +35,7 @@ public class NonBlockingPipedOutputStream extends ClickHousePipedOutputStream {
     protected final int bufferSize;
     protected final CompletableFuture<Void> future;
     protected final long timeout;
+    private final AtomicBoolean closing = new AtomicBoolean(false);
 
     protected ByteBuffer buffer;
 
@@ -104,7 +106,7 @@ public ClickHouseInputStream getInputStream(Runnable postCloseAction) {
 
     @Override
     public void close() throws IOException {
-        if (closed) {
+        if (closed || !closing.compareAndSet(false, true)) {
             return;
         }
 

clickhouse-data/src/test/java/com/clickhouse/data/stream/NonBlockingPipedOutputStreamTest.java

@@ -9,7 +9,10 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.clickhouse.data.ClickHouseByteBuffer;
@@ -208,6 +211,61 @@ public void testWriteBytes() throws IOException {
         }
     }
 
+    @Test(groups = { "unit" })
+    public void testConcurrentClose() throws Exception {
+        final long timeout = 5000L;
+        final AtomicBoolean acceptBuffer = new AtomicBoolean(false);
+        final CountDownLatch firstBufferOffer = new CountDownLatch(1);
+        final AtomicInteger closeCount = new AtomicInteger(0);
+        final CapacityPolicy policy = current -> {
+            firstBufferOffer.countDown();
+            return acceptBuffer.get();
+        };
+        final NonBlockingPipedOutputStream stream = new NonBlockingPipedOutputStream(4, 2, timeout * 4L,
+                policy, (Runnable) closeCount::incrementAndGet);
+        final byte[] expected = new byte[] { (byte) 1, (byte) 2, (byte) 3 };
+        stream.write(expected);
+
+        final ExecutorService executor = Executors.newFixedThreadPool(2);
+        try {
+            Future<?> firstClose = executor.submit(() -> {
+                stream.close();
+                return null;
+            });
+            Assert.assertTrue(firstBufferOffer.await(timeout, TimeUnit.MILLISECONDS),
+                    "First close did not try to flush the pending buffer");
+
+            Future<?> concurrentClose = executor.submit(() -> {
+                stream.close();
+                return null;
+            });
+            boolean concurrentCloseReturned = true;
+            try {
+                concurrentClose.get(timeout, TimeUnit.MILLISECONDS);
+            } catch (TimeoutException e) {
+                concurrentCloseReturned = false;
+            }
+
+            acceptBuffer.set(true);
+            firstClose.get(timeout, TimeUnit.MILLISECONDS);
+            concurrentClose.get(timeout, TimeUnit.MILLISECONDS);
+
+            Assert.assertTrue(concurrentCloseReturned, "Concurrent close should return without waiting");
+            Assert.assertEquals(closeCount.get(), 1, "Post close action should have been executed exactly once");
+            try (InputStream in = stream.getInputStream()) {
+                byte[] actual = new byte[expected.length];
+                Assert.assertEquals(in.read(actual), expected.length);
+                Assert.assertEquals(actual, expected);
+                Assert.assertEquals(in.read(), -1);
+            }
+        } finally {
+            acceptBuffer.set(true);
+            executor.shutdownNow();
+            Assert.assertTrue(executor.awaitTermination(timeout, TimeUnit.MILLISECONDS),
+                    "Concurrent close executor did not terminate");
+        }
+    }
+
     @Test(groups = { "unit" })
     public void testPipedStream() throws InterruptedException, IOException {
         final int timeout = 10000;

jdbc-v2/src/main/java/com/clickhouse/jdbc/ConnectionImpl.java

@@ -14,6 +14,7 @@
 import com.clickhouse.jdbc.internal.JdbcConfiguration;
 import com.clickhouse.jdbc.internal.ParsedPreparedStatement;
 import com.clickhouse.jdbc.internal.SqlParserFacade;
+import com.clickhouse.jdbc.internal.parser.javacc.ClickHouseSqlStatement;
 import com.clickhouse.jdbc.metadata.DatabaseMetaDataImpl;
 import com.google.common.collect.ImmutableMap;
 import org.slf4j.Logger;
@@ -460,10 +461,14 @@ public PreparedStatement prepareStatement(String sql, int resultSetType, int res
              * - INSERT INTO t VALUES (now(), ?, ?) !# there is a function in the values
              * - INSERT INTO t VALUES (now(), ?, 1), (now(), ?, 2) !# multiple values list
              * - INSERT INTO t SELECT ?, ?, ? !# insert from select
+             * - INSERT INTO [TABLE] FUNCTION f(...) VALUES (?) !# the target is a table function
+             * - the target table could not be resolved from the statement
              */
+            String table = parsedStatement.getTable();
             if (!parsedStatement.isInsertWithSelect() && parsedStatement.getAssignValuesGroups() == 1
-                    && !parsedStatement.isUseFunction()) {
-                TableSchema tableSchema = client.getTableSchema(parsedStatement.getTable(), schema);
+                    && !parsedStatement.isUseFunction()
+                    && table != null && !ClickHouseSqlStatement.DEFAULT_TABLE.equals(table)) {
+                TableSchema tableSchema = client.getTableSchema(table, schema);
                 return new WriterStatementImpl(this, sql, tableSchema, parsedStatement);
             }
         }

jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/SqlParserFacade.java

@@ -366,6 +366,12 @@ public void enterTableExprIdentifier(ClickHouseParser.TableExprIdentifierContext
 
             @Override
             public void enterInsertStmt(ClickHouseParser.InsertStmtContext ctx) {
+                if (ctx.tableFunctionExpr() != null) {
+                    // INSERT INTO [TABLE] FUNCTION f(...) has no plain table to write into, so the
+                    // parsed target must not be treated as a table name.
+                    parsedStatement.setUseFunction(true);
+                }
+
                 ClickHouseParser.TableIdentifierContext tableId = ctx.tableIdentifier();
                 if (tableId != null) {
                     extractAndSetDatabaseAndTable(tableId);

jdbc-v2/src/main/javacc/ClickHouseSqlParser.jj

@@ -265,6 +265,7 @@ TOKEN_MGR_DECLS: {
         compressLevel = null;
         format = null;
         file = null;
+        funcUsed = false;
         parameters.clear();
         positions.clear();
         settings.clear();
@@ -592,9 +593,12 @@ void grantStmt(): {} { // not interested
 void insertStmt(): {} {
     <INSERT> <INTO>
     (
-        LOOKAHEAD({ getToken(1).kind == FUNCTION
+        LOOKAHEAD({ getToken(1).kind == TABLE && getToken(2).kind == FUNCTION
+            && !tokenIn(3, VALUES, FORMAT, SETTINGS, SELECT, WITH, INFILE)
+            && getToken(4).kind == LPAREN }) <TABLE> <FUNCTION> functionExpr() { token_source.funcUsed = true; }
+        | LOOKAHEAD({ getToken(1).kind == FUNCTION
             && !tokenIn(2, VALUES, FORMAT, SETTINGS, SELECT, WITH, INFILE)
-            && getToken(3).kind == LPAREN }) <FUNCTION> functionExpr()
+            && getToken(3).kind == LPAREN }) <FUNCTION> functionExpr() { token_source.funcUsed = true; }
         | (
             LOOKAHEAD({ getToken(1).kind == TABLE
                 && !tokenIn(2, VALUES, FORMAT, SETTINGS, SELECT, WITH, LPAREN) }) <TABLE>

jdbc-v2/src/test/java/com/clickhouse/jdbc/WriterStatementImplTest.java

@@ -245,6 +245,76 @@ public void close() throws IOException {
         }
     }
 
+    @DataProvider(name = "insertTargetFunctionForms")
+    Object[][] insertTargetFunctionForms() {
+        return new Object[][]{
+                {"JAVACC", "INSERT INTO FUNCTION null('id UInt32') VALUES (?)"},
+                {"JAVACC", "INSERT INTO TABLE FUNCTION null('id UInt32') VALUES (?)"},
+                {"ANTLR4", "INSERT INTO FUNCTION null('id UInt32') VALUES (?)"},
+                {"ANTLR4", "INSERT INTO TABLE FUNCTION null('id UInt32') VALUES (?)"},
+                {"ANTLR4_PARAMS_PARSER", "INSERT INTO FUNCTION null('id UInt32') VALUES (?)"},
+                {"ANTLR4_PARAMS_PARSER", "INSERT INTO TABLE FUNCTION null('id UInt32') VALUES (?)"},
+        };
+    }
+
+    /**
+     * An INSERT whose target is a table function has no plain table to fetch a schema for, so it must stay
+     * on the regular SQL path instead of being routed to the RowBinary writer, which would look the
+     * function name up as a table and fail with UNKNOWN_TABLE.
+     */
+    @Test(groups = {"integration"}, dataProvider = "insertTargetFunctionForms")
+    public void testInsertIntoTableFunctionUsesSqlPath(String parser, String sql) throws SQLException {
+        Properties properties = new Properties();
+        properties.setProperty(DriverProperties.BETA_ROW_BINARY_WRITER.getKey(), "true");
+        properties.setProperty(DriverProperties.SQL_PARSER.getKey(), parser);
+        properties.setProperty(ASYNC_INSERT_SETTING_KEY, ServerSettings.OFF);
+        try (Connection connection = getJdbcConnection(properties);
+             PreparedStatement ps = connection.prepareStatement(sql)) {
+            Assert.assertFalse(ps instanceof WriterStatementImpl,
+                    "INSERT into a table function must not use the RowBinary writer: " + sql);
+            ps.setInt(1, 42);
+            Assert.assertEquals(ps.executeUpdate(), 1);
+        }
+    }
+
+    @DataProvider(name = "insertTargetPlainTableForms")
+    Object[][] insertTargetPlainTableForms() {
+        return new Object[][]{
+                {"JAVACC", "INSERT INTO %s VALUES (?)"},
+                {"JAVACC", "INSERT INTO TABLE %s VALUES (?)"},
+                {"ANTLR4", "INSERT INTO %s VALUES (?)"},
+                {"ANTLR4", "INSERT INTO TABLE %s VALUES (?)"},
+                {"ANTLR4_PARAMS_PARSER", "INSERT INTO %s VALUES (?)"},
+                {"ANTLR4_PARAMS_PARSER", "INSERT INTO TABLE %s VALUES (?)"},
+        };
+    }
+
+    @Test(groups = {"integration"}, dataProvider = "insertTargetPlainTableForms")
+    public void testInsertIntoPlainTableStillUsesWriter(String parser, String sqlTemplate) throws SQLException {
+        String table = "bt_writer_plain_table";
+        Properties properties = new Properties();
+        properties.setProperty(DriverProperties.BETA_ROW_BINARY_WRITER.getKey(), "true");
+        properties.setProperty(DriverProperties.SQL_PARSER.getKey(), parser);
+        properties.setProperty(ASYNC_INSERT_SETTING_KEY, ServerSettings.OFF);
+        try (Connection connection = getJdbcConnection(properties)) {
+            try (Statement stmt = connection.createStatement()) {
+                stmt.execute("DROP TABLE IF EXISTS " + table);
+                stmt.execute("CREATE TABLE " + table + " (id Int32) Engine MergeTree ORDER BY ()");
+            }
+
+            try (PreparedStatement ps = connection.prepareStatement(String.format(sqlTemplate, table))) {
+                Assert.assertTrue(ps instanceof WriterStatementImpl,
+                        "INSERT into a plain table must keep using the RowBinary writer: " + sqlTemplate);
+                ps.setInt(1, 42);
+                Assert.assertEquals(ps.executeUpdate(), 1);
+            } finally {
+                try (Statement stmt = connection.createStatement()) {
+                    stmt.execute("DROP TABLE IF EXISTS " + table);
+                }
+            }
+        }
+    }
+
     @DataProvider(name = "antlr4ParserBackends")
     Object[][] antlr4ParserBackends() {
         return new Object[][]{

jdbc-v2/src/test/java/com/clickhouse/jdbc/internal/BaseSqlParserFacadeTest.java

@@ -486,6 +486,58 @@ private void assertInsertColumns(String sql, String... expectedColumns) {
         Assert.assertEquals(actualColumns, expectedColumns, "Insert column names mismatch for: " + sql);
     }
 
+    @Test(dataProvider = "insertTargetFunctionDP")
+    public void testInsertTargetTableFunctionIsReportedAsFunction(String sql, boolean expectedUseFunction,
+                                                                 int expectedValuesGroups) {
+        ParsedPreparedStatement stmt = parser.parsePreparedStatement(sql);
+        Assert.assertFalse(stmt.isHasErrors(), "Query should parse without errors: " + sql);
+        Assert.assertTrue(stmt.isInsert(), "Should be an INSERT: " + sql);
+        Assert.assertEquals(stmt.isUseFunction(), expectedUseFunction, "useFunction mismatch for: " + sql);
+        Assert.assertEquals(stmt.getAssignValuesGroups(), expectedValuesGroups,
+                "Values group count mismatch for: " + sql);
+    }
+
+    @DataProvider
+    public static Object[][] insertTargetFunctionDP() {
+        return new Object[][] {
+                // An INSERT whose target is a table function has no plain table behind it
+                {"INSERT INTO FUNCTION null('id UInt32') VALUES (?)", true, 1},
+                {"INSERT INTO TABLE FUNCTION null('id UInt32') VALUES (?)", true, 1},
+                {"INSERT INTO FUNCTION remoteSecure('h', 'db', 't', 'u', 'p') (id, name) VALUES (?, ?)", true, 1},
+                {"INSERT INTO FUNCTION s3('url', 'key', 'secret', 'CSV') SELECT * FROM t", true, 0},
+                {"insert into function null('id UInt32') values (?)", true, 1},
+                {"INSERT INTO\n TABLE FUNCTION null('id UInt32')\n VALUES (?)", true, 1},
+                {"INSERT INTO /* target */ FUNCTION null('id UInt32') VALUES (?)", true, 1},
+                // Contrast: plain table targets, including a table literally named "function"
+                {"INSERT INTO t VALUES (?)", false, 1},
+                {"INSERT INTO TABLE t VALUES (?)", false, 1},
+                {"INSERT INTO db.t VALUES (?)", false, 1},
+                {"INSERT INTO function VALUES (?)", false, 1},
+                {"INSERT INTO TABLE function VALUES (?)", false, 1},
+                {"INSERT INTO function (id) VALUES (?)", false, 1},
+                // Contrast: a function inside the values list is already reported as a function
+                {"INSERT INTO t VALUES (now(), ?)", true, 1},
+        };
+    }
+
+    @Test(dataProvider = "insertTargetTableNameDP")
+    public void testInsertPlainTableTargetKeepsTableName(String sql, String expectedTable) {
+        ParsedPreparedStatement stmt = parser.parsePreparedStatement(sql);
+        Assert.assertFalse(stmt.isHasErrors(), "Query should parse without errors: " + sql);
+        Assert.assertEquals(stmt.getTable(), expectedTable, "Table name mismatch for: " + sql);
+    }
+
+    @DataProvider
+    public static Object[][] insertTargetTableNameDP() {
+        return new Object[][] {
+                {"INSERT INTO t VALUES (?)", "t"},
+                {"INSERT INTO TABLE t VALUES (?)", "t"},
+                {"INSERT INTO db.t VALUES (?)", "db.t"},
+                {"INSERT INTO function VALUES (?)", "function"},
+                {"INSERT INTO TABLE function VALUES (?)", "function"},
+        };
+    }
+
     @Test(dataProvider = "testCreateStmtDP")
     public void testCreateStatement(String sql) {
         ParsedPreparedStatement stmt = parser.parsePreparedStatement(sql);