jamesnetherton commented on code in PR #9006: URL: https://github.com/apache/camel-quarkus/pull/9006#discussion_r3783664119
########## extensions-support/langchain4j/runtime/src/main/java/org/apache/camel/quarkus/component/support/langchain4j/tracker/jdbc/JdbcIngestionTracker.java: ########## @@ -0,0 +1,352 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.support.langchain4j.tracker.jdbc; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import javax.sql.DataSource; + +import org.apache.camel.quarkus.component.support.langchain4j.tracker.IngestionTracker; + +/** + * JDBC-backed {@link IngestionTracker}. Deliberately dialect-free SQL (works on PostgreSQL and H2): + * upserts are update-then-insert rather than vendor MERGE/ON CONFLICT. + * + * <p> + * <strong>Concurrency assumption: at most one writer per {@code (pipeline, documentId)} at a + * time.</strong> The update-then-insert is not atomic (each statement auto-commits on its own, + * there is no transaction), so two concurrent writers for the same row can both see the + * {@code UPDATE} affect zero rows and then both attempt the {@code INSERT}, and one fails on the + * {@code (pipeline, doc_id)} primary key. Wrapping the two statements in a transaction would not + * remove this race without either {@code SERIALIZABLE} isolation or a dialect-specific upsert, + * both at odds with staying dialect-free; callers (the sync pass runner processes one document + * at a time per pipeline) are relied upon to hold this invariant instead. + * + * <p> + * <strong>Status: Experimental.</strong> See {@link IngestionTracker}. + */ +public class JdbcIngestionTracker implements IngestionTracker { + + /** + * Concatenated into every statement: this must never become caller-supplied without strict + * identifier validation, or the class turns into an SQL injection vector. + */ + static final String TABLE = "camel_quarkus_ingestion_tracker"; + + private final DataSource dataSource; + private final boolean createTableIfNotExists; + + public JdbcIngestionTracker(DataSource dataSource) { + this(dataSource, true); + } + + /** + * @param createTableIfNotExists when {@code false}, {@link #ensureSchema()} only probes that + * a pre-provisioned table exists, so the runtime DB user needs + * no DDL privilege (mirrors camel-sql's + * {@code JdbcMessageIdRepository}). When this surfaces as a + * configuration option it must be {@code ConfigPhase.RUN_TIME}. + */ + public JdbcIngestionTracker(DataSource dataSource, boolean createTableIfNotExists) { + this.dataSource = dataSource; + this.createTableIfNotExists = createTableIfNotExists; + } + + @Override + public void ensureSchema() { + if (!createTableIfNotExists) { + if (!tableExists()) { + throw new IllegalStateException( + "The ingestion tracker table '" + TABLE + "' does not exist and table creation is " + + "disabled. Pre-provision the table or enable createTableIfNotExists."); + } + return; + } + String ddl = "CREATE TABLE IF NOT EXISTS " + TABLE + " (" + + "pipeline VARCHAR(128) NOT NULL, " + // 1024: the S3 object-key maximum; file paths and URLs used as ids fit as well + + "doc_id VARCHAR(1024) NOT NULL, " + + "fingerprint VARCHAR(256), " + + "content_hash VARCHAR(64), " + + "segment_count INT DEFAULT 0 NOT NULL, " + + "intended_count INT DEFAULT 0 NOT NULL, " + + "status VARCHAR(16) NOT NULL, " + + "origin VARCHAR(16) DEFAULT 'source' NOT NULL, " + + "tombstone BOOLEAN DEFAULT FALSE NOT NULL, " + + "pinned BOOLEAN DEFAULT FALSE NOT NULL, " + + "updated_at TIMESTAMP, " + + "PRIMARY KEY (pipeline, doc_id))"; + try (Connection connection = dataSource.getConnection(); + java.sql.Statement statement = connection.createStatement()) { Review Comment: `java.sql.Statement` as an inline FQN, while every other `java.sql` type in the file is imported. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
