coordinates,
+ Number scale, Color fillColor, Color strokeColor, Color outsideColor) {
+ this.filename = filename;
+ this.label = label;
+ this.coordinates = coordinates;
+ this.scale = scale;
+ this.fillColor = fillColor;
+ this.strokeColor = strokeColor;
this.outsideColor = outsideColor;
}
}
-}
+
+ /**
+ * Describes a directed edge between two vertices identified by their
+ * zero-based index in the {@code entries} list.
+ *
+ * All fields except {@code fromIndex} and {@code toIndex} are optional.
+ * If no edges are provided in the source, the edge list stays empty
+ * and the edge DB table is never created, keeping full backward compatibility.
+ */
+ private static class Edge {
+ /** Source vertex — zero-based index into the {@code entries} list. */
+ private final int fromIndex;
+ /** Target vertex — zero-based index into the {@code entries} list. */
+ private final int toIndex;
+ /** Optional numeric weight / distance / cost associated with the edge. */
+ private final Number number;
+ /** Optional human-readable label shown alongside the edge. */
+ private final String label;
+ /** Display color of the edge line; defaults to {@link Color#GRAY}. */
+ private final Color color;
+ /** Source filename — kept for traceability in the DB. */
+ private final String filename;
+
+ private Edge(String filename, int fromIndex, int toIndex,
+ Number number, String label, Color color) {
+ this.filename = filename;
+ this.fromIndex = fromIndex;
+ this.toIndex = toIndex;
+ this.number = number;
+ this.label = label;
+ this.color = color;
+ }
+ }
+
+ private record PointData(double x, double y) {}
+}
\ No newline at end of file
diff --git a/src/main/java/org/texttechnologylab/udav/generators/TextFormatting.java b/src/main/java/org/texttechnologylab/udav/generators/TextFormatting.java
index 3de9f870..f2b892ff 100644
--- a/src/main/java/org/texttechnologylab/udav/generators/TextFormatting.java
+++ b/src/main/java/org/texttechnologylab/udav/generators/TextFormatting.java
@@ -73,7 +73,7 @@ public void setup_step1() throws SQLException {
SourceUIMA sourceUIMA = (SourceUIMA) source;
this.UIMAsofaID = settings.getStringSettingOrDefault("sofaID", null);
String sofaFile = settings.getStringSettingOrDefault("sofaFile", null);
- if (sofaFile == null) {
+ if (sofaFile == null || sofaFile.isBlank()) {
Set allFiles = sourceUIMA.determineAllSourceFiles();
settings.defineFilterListUniversalSetString("files", allFiles);
FilterList filterListSourceFiles = settings.generateStringFilterList("files");
@@ -181,6 +181,12 @@ public void writeToDB() throws SQLException {
Field END_SEGS = DSL.field(DSL.name(schema, DBConstants.TABLENAME_GENERATORDATA_TYPESEGMENTS, DBConstants.TABLEATTR_GENERATORDATA_END), Integer.class);
Field CAT_SEGS = DSL.field(DSL.name(schema, DBConstants.TABLENAME_GENERATORDATA_TYPESEGMENTS, DBConstants.TABLEATTR_GENERATORDATA_CATEGORY), String.class);
+ // Replace previous rows for this generator to keep saves idempotent.
+ dsl.deleteFrom(T_SEGS).where(GID_SEGS.eq(id)).execute();
+ dsl.deleteFrom(T_COLOR).where(GID_COLOR.eq(id)).execute();
+ dsl.deleteFrom(T_STYLE).where(GID_STYLE.eq(id)).execute();
+ dsl.deleteFrom(T_TEXT).where(GID_TEXT.eq(id)).execute();
+
// ---------- Insert text ----------
dsl.insertInto(T_TEXT)
.columns(GID_TEXT, TXT_TEXT)
diff --git a/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJson.java b/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJson.java
index db8968bc..db5100b3 100644
--- a/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJson.java
+++ b/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJson.java
@@ -4,11 +4,18 @@
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Getter;
+import org.jooq.DSLContext;
+import org.jooq.Field;
+import org.jooq.Table;
+import org.jooq.impl.DSL;
+import org.texttechnologylab.udav.db.SchemaObjectNames;
import org.texttechnologylab.udav.generators.settings.GeneratorSettings;
import org.texttechnologylab.udav.pipeline.JSONView;
+import org.texttechnologylab.udav.sources.DBAccess;
import java.io.IOException;
-import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -23,9 +30,9 @@ public class SourceJson extends Source {
protected final JSONView singleFileJSONView;
protected final Map filenameToJsonView; // TODO: Use or remove
- public SourceJson(String filepath) throws IOException {
+ public SourceJson(String filepath, DBAccess dbAccess) throws IOException, SQLException {
this.singleFileName = SOURCE_FILES_PATH + "/" + filepath.trim();
- this.singleFileJSONView = readJsonViewFromFile(singleFileName);
+ this.singleFileJSONView = readJsonViewFromDB(singleFileName, dbAccess);
this.filenameToJsonView = null;
}
public SourceJson(String normalizedFilepath, JSONView jsonView) {
@@ -49,6 +56,25 @@ public List> generateKeysMap(GeneratorSettings settings) {
return keysMap;
}
+ /**
+ * Same settings grammar as {@link #generateKeysMap(GeneratorSettings)}, but interpreted row-wise.
+ * Useful for list-root JSON where each list item is one logical row (e.g. edge pairs).
+ */
+ public List> generateKeysMapRowWise(GeneratorSettings settings) {
+
+ Map map_keysMap = (Map) settings.getMapSettingOrDefault("keysMap", null);
+ List> keysMap;
+ if (map_keysMap == null) {
+ map_keysMap = (Map) settings.getMapSettingOrDefault("keys", null);
+ keysMap = generateFlatKeysRowWise(map_keysMap);
+ } else {
+ keysMap = generateFlatKeysMapRowWise(map_keysMap);
+ }
+ addFixedKeysToKeysMap(keysMap, (Map) settings.getMapSettingOrDefault("fixedKeys", null));
+
+ return keysMap;
+ }
+
private void addFixedKeysToKeysMap(List> keysMap, Map fixedKeys) {
if (fixedKeys == null) return;
for (Map m : keysMap) {
@@ -93,6 +119,50 @@ private List> generateFlatKeysMap(Map keysMa
generateFlatKeysMapRecursive(keysMapRoot, singleFileJSONView, flatKeysMap);
return flatKeysMap;
}
+
+ private List> generateFlatKeysRowWise(Map keysRoot) {
+ ArrayList> flatKeysMap = new ArrayList<>();
+ if (keysRoot == null) return flatKeysMap;
+ if (!singleFileJSONView.isList()) return flatKeysMap;
+
+ for (Object rowObj : singleFileJSONView.asList()) {
+ JSONView row = new JSONView(rowObj);
+ HashMap currentMap = new HashMap<>();
+ for (Map.Entry entry : keysRoot.entrySet()) {
+ String key = entry.getKey();
+ if (entry.getValue() instanceof List valueList) {
+ int i = 0;
+ HashMap innerMap = new HashMap<>();
+ for (Object e : valueList) {
+ String value = (String) e;
+ innerMap.put(Integer.toString(i), getChildNode(row, value));
+ i++;
+ }
+ currentMap.put(key, innerMap);
+ } else {
+ String value = (String) entry.getValue();
+ currentMap.put(key, getChildNode(row, value));
+ }
+ }
+ flatKeysMap.add(currentMap);
+ }
+
+ return flatKeysMap;
+ }
+
+ private List> generateFlatKeysMapRowWise(Map keysMapRoot) {
+ ArrayList> flatKeysMap = new ArrayList<>();
+ if (keysMapRoot == null) return flatKeysMap;
+ if (!singleFileJSONView.isList()) return flatKeysMap;
+
+ for (Object rowObj : singleFileJSONView.asList()) {
+ HashMap currentMap = new HashMap<>();
+ generateFlatKeysMapRecursiveRowWise(keysMapRoot, new JSONView(rowObj), currentMap);
+ flatKeysMap.add(currentMap);
+ }
+
+ return flatKeysMap;
+ }
private void generateFlatKeysMapRecursive(Map keysMapCurrentPosition, JSONView currentPosition, ArrayList> flatKeysMap) {
for (Map.Entry entry : keysMapCurrentPosition.entrySet()) {
String key = entry.getKey();
@@ -175,25 +245,116 @@ private void generateFlatKeysMapRecursive(Map keysMapCurrentPosi
}
}
- private static JSONView readJsonViewFromFile(String path) throws IOException {
- try (InputStream in = SourceJson.class.getClassLoader().getResourceAsStream(path)) {
- if (in == null) {
- throw new IllegalArgumentException("File not found: " + path);
+ private void generateFlatKeysMapRecursiveRowWise(Map keysMapCurrentPosition, JSONView currentPosition, Map currentMap) {
+ for (Map.Entry entry : keysMapCurrentPosition.entrySet()) {
+ String key = entry.getKey();
+ Object value = entry.getValue();
+
+ JSONView valuePosition = getChildView(currentPosition, key);
+ if (valuePosition == null) {
+ continue;
+ }
+
+ if (value instanceof String stringValue) {
+ putMappedValue(currentMap, stringValue.trim(), valuePosition.getNode());
+ } else if (value instanceof List listValue) {
+ Object rawNode = valuePosition.getNode();
+ if (!(rawNode instanceof List valueNode)) {
+ continue;
+ }
+ int listSize = listValue.size();
+ for (int i = 0; i < listSize; i++) {
+ String stringValue = (String) listValue.get(i);
+ Object mappedNode = i < valueNode.size() ? valueNode.get(i) : null;
+ putMappedValue(currentMap, stringValue, mappedNode);
+ }
+ } else if (value instanceof Map subMap) {
+ generateFlatKeysMapRecursiveRowWise((Map) subMap, valuePosition, currentMap);
+ }
+ }
+ }
+
+ private static void putMappedValue(Map destination, String target, Object value) {
+ String trimmed = target.trim();
+ if (trimmed.contains("@")) {
+ String[] split = trimmed.split("@", 2);
+ Map innerValueMap = (Map) destination.get(split[0]);
+ if (innerValueMap == null) {
+ innerValueMap = new HashMap<>();
+ destination.put(split[0], innerValueMap);
}
+ innerValueMap.put(split[1], value);
+ } else {
+ destination.put(trimmed, value);
+ }
+ }
- ObjectMapper mapper = new ObjectMapper();
- JsonNode rootNode = mapper.readTree(in);
-
- Object rootValue;
- if (rootNode.isObject()) {
- rootValue = mapper.convertValue(rootNode, new TypeReference>() {});
- } else if (rootNode.isArray()) {
- rootValue = mapper.convertValue(rootNode, new TypeReference>() {});
- } else {
- // For primitive root nodes (string, number, boolean, null)
- rootValue = mapper.convertValue(rootNode, Object.class);
+ private static JSONView getChildView(JSONView currentPosition, String key) {
+ try {
+ if (currentPosition.isMap()) {
+ return currentPosition.get(key);
}
- return new JSONView(rootValue);
+ if (currentPosition.isList()) {
+ int idx = Integer.parseInt(key);
+ return currentPosition.get(idx);
+ }
+ } catch (Exception ignored) {
+ return null;
+ }
+ return null;
+ }
+
+ private static Object getChildNode(JSONView currentPosition, String key) {
+ JSONView child = getChildView(currentPosition, key);
+ return child != null ? child.getNode() : null;
+ }
+
+ private static JSONView readJsonViewFromDB(String path, DBAccess dbAccess) throws IOException, SQLException {
+ if (dbAccess == null) {
+ throw new IllegalArgumentException("dbAccess must not be null");
+ }
+
+ String sourceFileName = path.trim();
+ int slashIdx = sourceFileName.lastIndexOf('/');
+ if (slashIdx >= 0 && slashIdx < sourceFileName.length() - 1) {
+ sourceFileName = sourceFileName.substring(slashIdx + 1);
+ }
+
+ String json = readJsonString(dbAccess, sourceFileName);
+ if (json == null && !"public".equalsIgnoreCase(dbAccess.getSchema())) {
+ json = readJsonString(dbAccess, "public", sourceFileName);
+ }
+ if (json == null) {
+ throw new IllegalArgumentException("JSON source not found in DB table \"" + SchemaObjectNames.TABLE_JSON_DATA + "\": " + sourceFileName);
+ }
+
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode rootNode = mapper.readTree(json);
+
+ Object rootValue;
+ if (rootNode.isObject()) {
+ rootValue = mapper.convertValue(rootNode, new TypeReference>() {});
+ } else if (rootNode.isArray()) {
+ rootValue = mapper.convertValue(rootNode, new TypeReference>() {});
+ } else {
+ rootValue = mapper.convertValue(rootNode, Object.class);
+ }
+ return new JSONView(rootValue);
+ }
+
+ private static String readJsonString(DBAccess dbAccess, String sourceFileName) throws SQLException {
+ return readJsonString(dbAccess, dbAccess.getSchema(), sourceFileName);
+ }
+
+ private static String readJsonString(DBAccess dbAccess, String schema, String sourceFileName) throws SQLException {
+ try (Connection connection = dbAccess.getDataSource().getConnection()) {
+ DSLContext dsl = DSL.using(connection);
+ Table table = DSL.table(DSL.name(schema, SchemaObjectNames.TABLE_JSON_DATA));
+ Field fName = DSL.field(DSL.name(schema, SchemaObjectNames.TABLE_JSON_DATA, SchemaObjectNames.COL_JSON_DATA_SOURCEFILE_NAME), String.class);
+ Field fJson = DSL.field(DSL.name(schema, SchemaObjectNames.TABLE_JSON_DATA, SchemaObjectNames.COL_JSON_DATA_JSON), String.class);
+ return dsl.select(fJson).from(table).where(fName.eq(sourceFileName)).fetchOne(fJson);
+ } catch (org.jooq.exception.DataAccessException e) {
+ return null;
}
}
}
diff --git a/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJsonN.java b/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJsonN.java
index 29259d33..e1d45322 100644
--- a/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJsonN.java
+++ b/src/main/java/org/texttechnologylab/udav/generators/sources/SourceJsonN.java
@@ -1,16 +1,19 @@
package org.texttechnologylab.udav.generators.sources;
+import org.texttechnologylab.udav.sources.DBAccess;
+
import java.io.IOException;
-import java.util.HashMap;
+import java.sql.SQLException;
+import java.util.LinkedHashMap;
import java.util.Map;
public class SourceJsonN extends SourceJson implements SourceN {
private final Map subSources;
- public SourceJsonN(String filepath) throws IOException {
- super(filepath);
- this.subSources = new HashMap<>();
+ public SourceJsonN(String filepath, DBAccess dbAccess) throws IOException, SQLException {
+ super(filepath, dbAccess);
+ this.subSources = new LinkedHashMap<>();
Map map = singleFileJSONView.asMap();
for (String key : map.keySet()) subSources.put(key, new SourceJson(singleFileName, singleFileJSONView.get(key)));
}
diff --git a/src/main/java/org/texttechnologylab/udav/generators/sources/SourceUIMA.java b/src/main/java/org/texttechnologylab/udav/generators/sources/SourceUIMA.java
index f4cd874a..03dda4de 100644
--- a/src/main/java/org/texttechnologylab/udav/generators/sources/SourceUIMA.java
+++ b/src/main/java/org/texttechnologylab/udav/generators/sources/SourceUIMA.java
@@ -25,7 +25,7 @@ public SourceUIMA(String uri, DBAccess dbAccess) throws SQLException {
this.uri = uri.trim();
this.dbAccess = dbAccess;
try (Connection connection = dbAccess.getDataSource().getConnection()) {
- DSLContext dsl = DSL.using(dbAccess.getDataSource().getConnection());
+ DSLContext dsl = DSL.using(connection);
TypeTableResolver resolver = new TypeTableResolver(dsl, DB_SCHEMA_UIMA);
this.tableHash = resolver.tableForType(uri);
}
diff --git a/src/main/java/org/texttechnologylab/udav/importer/DUUIImporter.java b/src/main/java/org/texttechnologylab/udav/importer/DUUIImporter.java
index 0a967478..3e1fa778 100644
--- a/src/main/java/org/texttechnologylab/udav/importer/DUUIImporter.java
+++ b/src/main/java/org/texttechnologylab/udav/importer/DUUIImporter.java
@@ -6,7 +6,6 @@
import org.apache.uima.resource.metadata.TypeSystemDescription;
import org.apache.uima.util.XMLInputSource;
import org.dkpro.core.io.xmi.XmiWriter;
-import org.junit.jupiter.api.DisplayName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
@@ -27,7 +26,10 @@
import java.io.File;
import java.io.IOException;
+import java.lang.reflect.Method;
import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
import static org.apache.uima.fit.factory.AnalysisEngineFactory.createEngineDescription;
@@ -38,9 +40,12 @@
public class DUUIImporter implements ApplicationRunner {
private static final Logger LOGGER = LoggerFactory.getLogger(DUUIImporter.class);
+
private final DbProps db;
private final DUUIImporterProps duuiProps;
private final PostImportRowCounter postImportRowCounter;
+ private final PostImportIndexBuilder postImportIndexBuilder;
+
private DUUIComposer composer;
private TypeSystemDescription externalTypeSystem;
@@ -48,7 +53,7 @@ private void init() throws IOException, URISyntaxException, UIMAException, SAXEx
DUUILuaContext ctx = new DUUILuaContext().withJsonLibrary();
composer = new DUUIComposer()
- .withSkipVerification(true)
+ .withSkipVerification(resolveBooleanProp("skipVerification", "DUUI_IMPORTER_SKIP_VERIFICATION", false))
.withLuaContext(ctx)
.withWorkers(duuiProps.workers());
@@ -58,46 +63,58 @@ private void init() throws IOException, URISyntaxException, UIMAException, SAXEx
composer.addDriver(uimaDriver, dockerDriver);
String tsPath = duuiProps.typeSystemPath();
+
if (tsPath != null && !tsPath.isBlank()) {
File tsFile = new File(tsPath);
- if (tsFile.isFile()) {
- LOGGER.info("Loading external type system from: {}", tsPath);
- TypeSystemDescription tsd = UIMAFramework.getXMLParser()
- .parseTypeSystemDescription(new XMLInputSource(tsFile));
- tsd.resolveImports();
-
- composer.setInstantiatedTypeSystem(tsd);
- externalTypeSystem = tsd;
- } else {
- LOGGER.warn("DUUI_IMPORTER_TYPE_SYSTEM_PATH is set but file not found: {}", tsPath);
+
+ if (!tsFile.isFile()) {
+ throw new IOException("Configured DUUI type system file not found: " + tsPath);
}
+
+ LOGGER.info("Loading external type system from: {}", tsPath);
+
+ TypeSystemDescription tsd = UIMAFramework.getXMLParser()
+ .parseTypeSystemDescription(new XMLInputSource(tsFile));
+
+ tsd.resolveImports();
+
+ composer.setInstantiatedTypeSystem(tsd);
+ externalTypeSystem = tsd;
}
}
- @DisplayName("NLP")
public void execute() throws Exception {
+ boolean storeCoveredText = resolveBooleanProp(
+ "storeCoveredText",
+ "DUUI_IMPORTER_STORE_COVERED_TEXT",
+ false
+ );
+
+ boolean prepareDbSchema = resolveBooleanProp(
+ "prepareDbSchema",
+ "DUUI_IMPORTER_PREPARE_DB_SCHEMA",
+ true
+ );
+
+ int dbScale = resolveIntProp("dbWorkers", "DUUI_IMPORTER_DB_WORKERS", 1);
+
DUUIFileReaderLazy corpusReader =
- new DUUIFileReaderLazy(duuiProps.inputPath(), duuiProps.inputFileEnding(), 10);
+ new DUUIFileReaderLazy(
+ duuiProps.inputPath(),
+ duuiProps.inputFileEnding(),
+ resolveIntProp("readerBatchSize", "DUUI_IMPORTER_READER_BATCH_SIZE", 10)
+ );
DUUIAsynchronousProcessor processor = new DUUIAsynchronousProcessor(corpusReader);
- // Docker NLP
-// composer.add(new DUUIDockerDriver.Component("docker.texttechnologylab.org/textimager-duui-spacy-single-de_core_news_sm:0.1.4")
-// .withScale(duuiProps.workers())
-// .withImageFetching()
-// .build());
-
- // Cleanup
composer.add(new DUUIUIMADriver.Component(
createEngineDescription(RemoveMetaInformation.class, externalTypeSystem))
.withScale(duuiProps.workers())
.build());
- // Debug XMI is a major bottleneck; keep it OFF by default.
- // Enable by setting env DUUI_IMPORTER_DEBUG_XMI=true
- boolean debugXmi = Boolean.parseBoolean(System.getenv().getOrDefault("DUUI_IMPORTER_DEBUG_XMI", "false"));
- if (debugXmi) {
- String target = System.getenv().getOrDefault("DUUI_IMPORTER_DEBUG_XMI_PATH", "/tmp/export");
+ if (resolveBooleanProp("debugXmi", "DUUI_IMPORTER_DEBUG_XMI", false)) {
+ String target = resolveStringProp("debugXmiPath", "DUUI_IMPORTER_DEBUG_XMI_PATH", "/tmp/export");
+
composer.add(new DUUIUIMADriver.Component(
createEngineDescription(
XmiWriter.class,
@@ -108,11 +125,40 @@ public void execute() throws Exception {
XmiWriter.PARAM_VERSION, "1.1",
XmiWriter.PARAM_COMPRESSION, "GZIP"
))
- .withScale(1) // do not scale disk writers
+ .withScale(1)
+ .build());
+ }
+
+ if (prepareDbSchema) {
+ LOGGER.info("Adding DB schema-preparation stage. DDL will run with scale=1 before parallel DB COPY writers.");
+
+ composer.add(new DUUIUIMADriver.Component(
+ createEngineDescription(
+ JooqDatabaseWriter.class,
+ externalTypeSystem,
+ JooqDatabaseWriter.PARAM_JDBC_URL, db.getUrl(),
+ JooqDatabaseWriter.PARAM_DB_USER, db.getUser(),
+ JooqDatabaseWriter.PARAM_DB_PASS, db.getPass(),
+ JooqDatabaseWriter.PARAM_SCHEMA, db.getSchema(),
+ JooqDatabaseWriter.PARAM_BATCH_SIZE, db.getBatchSize(),
+ JooqDatabaseWriter.PARAM_MAX_IDENT, db.getMaxIdent(),
+ JooqDatabaseWriter.PARAM_SQL_DIALECT, db.getDialect(),
+ JooqDatabaseWriter.PARAM_PIPELINE_HASH, buildPipelineHash(),
+ JooqDatabaseWriter.PARAM_STORE_COVERED_TEXT, storeCoveredText,
+ JooqDatabaseWriter.PARAM_ALLOW_DDL, true,
+ JooqDatabaseWriter.PARAM_PREPARE_SCHEMA_ONLY, true
+ ))
+ .withScale(1)
.build());
}
- int dbScale = Math.max(1, Math.min(duuiProps.workers(), 4)); // start conservative; raise if DB can handle it
+ LOGGER.info(
+ "Adding DB COPY writer stage. dbWorkers={} storeCoveredText={} allowDdl={}",
+ dbScale,
+ storeCoveredText,
+ !prepareDbSchema
+ );
+
composer.add(new DUUIUIMADriver.Component(
createEngineDescription(
JooqDatabaseWriter.class,
@@ -123,19 +169,105 @@ public void execute() throws Exception {
JooqDatabaseWriter.PARAM_SCHEMA, db.getSchema(),
JooqDatabaseWriter.PARAM_BATCH_SIZE, db.getBatchSize(),
JooqDatabaseWriter.PARAM_MAX_IDENT, db.getMaxIdent(),
- JooqDatabaseWriter.PARAM_SQL_DIALECT, db.getDialect()
+ JooqDatabaseWriter.PARAM_SQL_DIALECT, db.getDialect(),
+ JooqDatabaseWriter.PARAM_PIPELINE_HASH, buildPipelineHash(),
+ JooqDatabaseWriter.PARAM_STORE_COVERED_TEXT, storeCoveredText,
+ JooqDatabaseWriter.PARAM_ALLOW_DDL, !prepareDbSchema,
+ JooqDatabaseWriter.PARAM_PREPARE_SCHEMA_ONLY, false
))
.withScale(dbScale)
.build());
- composer.run(processor, "Importer");
- composer.shutdown();
+ try {
+ composer.run(processor, "Importer");
+ } finally {
+ if (composer != null) {
+ composer.shutdown();
+ }
+ }
}
@Override
public void run(ApplicationArguments args) throws Exception {
init();
execute();
+ postImportIndexBuilder.buildIndexes();
postImportRowCounter.updateRowCounts();
}
+
+ private String buildPipelineHash() {
+ List parts = new ArrayList<>();
+
+ parts.add("writerSchemaVersion=6-no-rowhash-no-pk");
+ parts.add("inputEnding=" + duuiProps.inputFileEnding());
+ parts.add("typeSystemPath=" + duuiProps.typeSystemPath());
+ parts.add("removeMetaInformation=true");
+ parts.add("debugXmi=" + resolveBooleanProp("debugXmi", "DUUI_IMPORTER_DEBUG_XMI", false));
+ parts.add("storeCoveredText=" + resolveBooleanProp("storeCoveredText", "DUUI_IMPORTER_STORE_COVERED_TEXT", false));
+
+ String explicit = System.getenv("DUUI_IMPORTER_PIPELINE_HASH_EXTRA");
+
+ if (explicit != null && !explicit.isBlank()) {
+ parts.add("extra=" + explicit);
+ }
+
+ return org.apache.commons.codec.digest.DigestUtils.sha256Hex(String.join("\n", parts));
+ }
+
+ private boolean resolveBooleanProp(String methodName, String envName, boolean defaultValue) {
+ Object reflected = tryCallNoArg(methodName);
+
+ if (reflected instanceof Boolean b) return b;
+ if (reflected instanceof String s && !s.isBlank()) return Boolean.parseBoolean(s);
+
+ String env = System.getenv(envName);
+
+ if (env == null || env.isBlank()) return defaultValue;
+
+ return Boolean.parseBoolean(env);
+ }
+
+ private int resolveIntProp(String methodName, String envName, int defaultValue) {
+ Object reflected = tryCallNoArg(methodName);
+
+ if (reflected instanceof Number n) return n.intValue();
+
+ if (reflected instanceof String s && !s.isBlank()) {
+ try {
+ return Integer.parseInt(s.trim());
+ } catch (NumberFormatException ignored) {
+ }
+ }
+
+ String env = System.getenv(envName);
+
+ if (env == null || env.isBlank()) return defaultValue;
+
+ try {
+ return Integer.parseInt(env.trim());
+ } catch (NumberFormatException ignored) {
+ return defaultValue;
+ }
+ }
+
+ private String resolveStringProp(String methodName, String envName, String defaultValue) {
+ Object reflected = tryCallNoArg(methodName);
+
+ if (reflected instanceof String s && !s.isBlank()) return s;
+
+ String env = System.getenv(envName);
+
+ if (env == null || env.isBlank()) return defaultValue;
+
+ return env;
+ }
+
+ private Object tryCallNoArg(String methodName) {
+ try {
+ Method m = duuiProps.getClass().getMethod(methodName);
+ return m.invoke(duuiProps);
+ } catch (ReflectiveOperationException ignored) {
+ return null;
+ }
+ }
}
diff --git a/src/main/java/org/texttechnologylab/udav/importer/DatabaseGenerator.java b/src/main/java/org/texttechnologylab/udav/importer/DatabaseGenerator.java
deleted file mode 100644
index 9ab314c7..00000000
--- a/src/main/java/org/texttechnologylab/udav/importer/DatabaseGenerator.java
+++ /dev/null
@@ -1,126 +0,0 @@
-// src/main/java/uni/textimager/sandbox/importer/DatabaseGenerator.java
-package org.texttechnologylab.udav.importer;
-
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.core.Ordered;
-import org.springframework.core.annotation.Order;
-import org.springframework.jdbc.core.JdbcTemplate;
-import org.texttechnologylab.udav.importer.dialect.SqlDialect;
-import org.texttechnologylab.udav.importer.service.DataInserterService;
-import org.texttechnologylab.udav.importer.service.NameSanitizer;
-import org.texttechnologylab.udav.importer.service.SchemaGeneratorService;
-import org.texttechnologylab.udav.importer.service.XmiParserService;
-
-import java.nio.file.DirectoryStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.*;
-
-/**
- * Application runner that generates database schema and populates tables from XMI files.
- *
- * - Reads .xmi files from configured input directory.
- * - Parses each file into EntityRecord objects.
- * - Computes maximum attribute lengths per table for schema generation.
- * - Creates or updates tables based on computed lengths.
- * - Inserts all parsed records into respective tables.
- * - Creates metadata table TableNames and records each table name.
- *
- */
-//@Component
-@Order(Ordered.HIGHEST_PRECEDENCE)
-@ConditionalOnProperty(name = "app.database-generator.enabled", havingValue = "true", matchIfMissing = true)
-@Deprecated
-public class DatabaseGenerator implements ApplicationRunner {
- private final JdbcTemplate jdbc;
- private final SqlDialect dialect;
- private final XmiParserService parser;
- private final NameSanitizer sanitizer;
- private final SchemaGeneratorService schemaGen;
- private final DataInserterService dataInserter;
-
- @Value("${app.input-dir}")
- private String inputDir;
-
- /**
- * Construct DatabaseGenerator with required services and JDBC components.
- *
- * @param jdbc JdbcTemplate for executing SQL statements
- * @param dialect database-specific SQL dialect implementation
- * @param parser service to parse XMI files into EntityRecord objects
- * @param sanitizer utility to sanitize tag and attribute names
- * @param schemaGen service to generate database schema based on attribute lengths
- * @param dataInserter service to batch-insert EntityRecord instances into tables
- */
- public DatabaseGenerator(
- JdbcTemplate jdbc,
- SqlDialect dialect,
- XmiParserService parser,
- NameSanitizer sanitizer,
- SchemaGeneratorService schemaGen,
- DataInserterService dataInserter) {
- this.jdbc = jdbc;
- this.dialect = dialect;
- this.parser = parser;
- this.sanitizer = sanitizer;
- this.schemaGen = schemaGen;
- this.dataInserter = dataInserter;
- }
-
- /**
- * Execute on application startup:
- *
- * - Scan input directory for .xmi files.
- * - Parse each file and collect EntityRecord objects.
- * - Track maximum string lengths per table column for schema sizing.
- * - Generate or update tables via SchemaGeneratorService.
- * - Insert parsed records via DataInserterService.
- * - Create metadata table "TableNames" and insert table names.
- *
- *
- * @param args application arguments (ignored)
- * @throws Exception on file, I/O or parsing errors
- */
- @Override
- public void run(ApplicationArguments args) throws Exception {
- Map> maxLengths = new LinkedHashMap<>();
- List allRecords = new ArrayList<>();
-
- try (DirectoryStream directoryStream = Files.newDirectoryStream(Path.of(inputDir), "*.xmi")) {
- for (Path file : directoryStream) {
- String fileName = file.getFileName().toString();
- for (EntityRecord entityRecord : parser.parse(file)) {
- String table = sanitizer.toClassName(entityRecord.tag());
- maxLengths.computeIfAbsent(table, k -> new HashMap<>())
- .merge("filename", fileName.length(), Math::max);
- entityRecord.attributes().forEach((raw, val) ->
- maxLengths.get(table)
- .merge(sanitizer.sanitize(raw), val.length(), Math::max)
- );
- allRecords.add(entityRecord);
- }
- }
- }
-
- schemaGen.generateSchema(maxLengths);
- dataInserter.insertRecords(allRecords);
-
- // Insert metadata about tables
- String metaTable = "TableNames";
- String metaPk = "tablenames_id";
- String ddlMeta = "CREATE TABLE IF NOT EXISTS " + metaTable + " ("
- + dialect.autoIncrementPrimaryKey(metaPk) + ", "
- + "table_name " + dialect.varcharType(255) + ", "
- + "PRIMARY KEY(" + metaPk + "))";
- jdbc.execute(ddlMeta);
-
- String insertMeta = "INSERT INTO " + metaTable + "(table_name) VALUES(?)";
- List batchArgs = maxLengths.keySet().stream()
- .map(name -> new Object[]{name})
- .toList();
- jdbc.batchUpdate(insertMeta, batchArgs);
- }
-}
diff --git a/src/main/java/org/texttechnologylab/udav/importer/JooqDatabaseWriter.java b/src/main/java/org/texttechnologylab/udav/importer/JooqDatabaseWriter.java
index 1c1bb69c..5675f0a3 100644
--- a/src/main/java/org/texttechnologylab/udav/importer/JooqDatabaseWriter.java
+++ b/src/main/java/org/texttechnologylab/udav/importer/JooqDatabaseWriter.java
@@ -19,12 +19,18 @@
import org.jooq.exception.DataAccessException;
import org.jooq.impl.DSL;
import org.jooq.impl.SQLDataType;
+import org.postgresql.PGConnection;
+import org.postgresql.copy.CopyManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.io.StringReader;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+import java.sql.SQLException;
import java.util.*;
import java.util.Comparator;
import java.util.concurrent.ConcurrentHashMap;
@@ -34,7 +40,6 @@
import static org.jooq.impl.DSL.*;
public class JooqDatabaseWriter extends JCasAnnotator_ImplBase {
-
public static final String PARAM_JDBC_URL = "jdbcUrl";
public static final String PARAM_DB_USER = "dbUser";
public static final String PARAM_DB_PASS = "dbPass";
@@ -42,45 +47,64 @@ public class JooqDatabaseWriter extends JCasAnnotator_ImplBase {
public static final String PARAM_BATCH_SIZE = "batchSize";
public static final String PARAM_MAX_IDENT = "maxIdentifierLength";
public static final String PARAM_SQL_DIALECT = "sqlDialect";
-
+ public static final String PARAM_PIPELINE_HASH = "pipelineHash";
+ public static final String PARAM_STORE_COVERED_TEXT = "storeCoveredText";
+ public static final String PARAM_ALLOW_DDL = "allowDdl";
+ public static final String PARAM_PREPARE_SCHEMA_ONLY = "prepareSchemaOnly";
private static final Logger LOGGER = LoggerFactory.getLogger(JooqDatabaseWriter.class);
-
- private static final Map> UIMA_PRIMITIVE_TO_SQL = Map.of(
- "uima.cas.String", SQLDataType.CLOB,
- "uima.cas.Integer", SQLDataType.INTEGER,
- "uima.cas.Float", SQLDataType.REAL,
- "uima.cas.Double", SQLDataType.DOUBLE,
- "uima.cas.Boolean", SQLDataType.BOOLEAN,
- "uima.cas.Long", SQLDataType.BIGINT,
- "uima.cas.Short", SQLDataType.SMALLINT,
- "uima.cas.Byte", SQLDataType.SMALLINT
- );
-
- private static final int TABLE_HASH_LEN = 16;
-
- private static final Set seenTsFingerprints = ConcurrentHashMap.newKeySet();
- private static final Object DDL_LOCK = new Object();
- private static final AtomicBoolean REGISTRY_READY = new AtomicBoolean(false);
-
+ private static final Map> UIMA_PRIMITIVE_TO_SQL = Map.of("uima.cas.String", SQLDataType.CLOB, "uima.cas.Integer", SQLDataType.INTEGER, "uima.cas.Float", SQLDataType.REAL, "uima.cas.Double", SQLDataType.DOUBLE, "uima.cas.Boolean", SQLDataType.BOOLEAN, "uima.cas.Long", SQLDataType.BIGINT, "uima.cas.Short", SQLDataType.SMALLINT, "uima.cas.Byte", SQLDataType.SMALLINT);
+ private static final int TABLE_HASH_LEN = 8;
+ private static final int COPY_FLUSH_CHARS = 16 * 1024 * 1024;
+ private static final Map REGISTRY_READY = new ConcurrentHashMap<>();
+ private static final Map DDL_LOCKS = new ConcurrentHashMap<>();
+ private static final Map> SEEN_TS_FINGERPRINTS = new ConcurrentHashMap<>();
+ private static final ThreadLocal SHA256 = ThreadLocal.withInitial(() -> {
+ try {
+ return MessageDigest.getInstance("SHA-256");
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 unavailable", e);
+ }
+ });
+ private static final char[] HEX = "0123456789abcdef".toCharArray();
private final Map typeToTable = new ConcurrentHashMap<>();
private final Set createdTables = ConcurrentHashMap.newKeySet();
-
+ @ConfigurationParameter(name = PARAM_JDBC_URL, mandatory = true)
+ private String jdbcUrl;
+ @ConfigurationParameter(name = PARAM_DB_USER, mandatory = false)
+ private String dbUser;
+ @ConfigurationParameter(name = PARAM_DB_PASS, mandatory = false)
+ private String dbPass;
+ @ConfigurationParameter(name = PARAM_SQL_DIALECT, mandatory = false)
+ private String sqlDialectName;
@ConfigurationParameter(name = PARAM_SCHEMA, mandatory = false, defaultValue = "public")
- private String schema;
-
- @ConfigurationParameter(name = PARAM_BATCH_SIZE, mandatory = false, defaultValue = "1000")
- private int batchSize;
-
+ private String schema = "public";
+ @ConfigurationParameter(name = PARAM_BATCH_SIZE, mandatory = false, defaultValue = "10000")
+ private int batchSize = 10000;
@ConfigurationParameter(name = PARAM_MAX_IDENT, mandatory = false, defaultValue = "63")
- private int maxIdentifierLength;
-
+ private int maxIdentifierLength = 63;
+ @ConfigurationParameter(name = PARAM_PIPELINE_HASH, mandatory = false, defaultValue = "unknown")
+ private String pipelineHash = "unknown";
+ @ConfigurationParameter(name = PARAM_STORE_COVERED_TEXT, mandatory = false, defaultValue = "false")
+ private boolean storeCoveredText = false;
+ @ConfigurationParameter(name = PARAM_ALLOW_DDL, mandatory = false, defaultValue = "true")
+ private boolean allowDdl = true;
+ @ConfigurationParameter(name = PARAM_PREPARE_SCHEMA_ONLY, mandatory = false, defaultValue = "false")
+ private boolean prepareSchemaOnly = false;
private DSLContext dsl;
private HikariDataSource dataSource;
+ private TypeSystem cachedTs;
+ private TsCache tsCache;
+
+ private static void updateHash(MessageDigest md, String key, String value) {
+ md.update(key.getBytes(StandardCharsets.UTF_8));
+ md.update((byte) '=');
+ md.update((value == null ? "" : value).getBytes(StandardCharsets.UTF_8));
+ md.update((byte) 0);
+ }
- private static Iterable iterable(Iterator it) {
- List out = new ArrayList<>();
- while (it.hasNext()) out.add(it.next());
- return out;
+ private static String featSortName(Feature f) {
+ String s = f.getShortName();
+ return s != null ? s : f.getName();
}
private static DocumentMetaData getOrCreateDocumentMeta(JCas jCas) {
@@ -88,7 +112,7 @@ private static DocumentMetaData getOrCreateDocumentMeta(JCas jCas) {
return DocumentMetaData.get(jCas);
} catch (IllegalArgumentException e) {
DocumentMetaData md = new DocumentMetaData(jCas);
- md.setDocumentId(UUID.randomUUID().toString());
+ md.setDocumentId(deterministicTextId(jCas));
md.setDocumentTitle("unknown");
md.setDocumentUri(null);
md.addToIndexes();
@@ -101,11 +125,46 @@ private static T safe(T v, Supplier... fallbacks) {
if (v != null && !(v instanceof String s && s.isBlank())) return v;
for (Supplier fb : fallbacks) {
T t = fb.get();
- if (t != null && (!(t instanceof String) || !((String) t).isBlank())) return t;
+ if (t != null && (!(t instanceof String) || !((String) t).isBlank())) {
+ return t;
+ }
}
return null;
}
+ private static String deterministicDocumentId(JCas jCas, DocumentMetaData md) {
+ if (md != null) {
+ if (!isBlank(md.getDocumentId())) return stripXmiSuffix(md.getDocumentId());
+ if (!isBlank(md.getDocumentUri())) return DigestUtils.sha256Hex("uri:" + md.getDocumentUri());
+ if (!isBlank(md.getDocumentTitle())) return DigestUtils.sha256Hex("title:" + md.getDocumentTitle());
+ }
+ return deterministicTextId(jCas);
+ }
+
+ // ".xmi" is a serialization-format suffix, not part of a logical document identity.
+ // Stripping it here keeps doc_id consistent across docs whose upstream readers happen
+ // to set DocumentMetaData.documentId with or without the extension.
+ private static String stripXmiSuffix(String id) {
+ String s = id.trim();
+ if (s.length() > 4 && s.regionMatches(true, s.length() - 4, ".xmi", 0, 4)) {
+ return s.substring(0, s.length() - 4);
+ }
+ return s;
+ }
+
+ private static String deterministicTextId(JCas jCas) {
+ String text = null;
+ try {
+ text = jCas.getDocumentText();
+ } catch (Throwable ignored) {
+ }
+ return DigestUtils.sha256Hex("text:" + (text == null ? "" : text));
+ }
+
+ private static String emptyToNull(String s) {
+ return (s == null || s.isEmpty()) ? null : s;
+ }
+
private static boolean isBlank(String s) {
return s == null || s.trim().isEmpty();
}
@@ -131,590 +190,633 @@ private static SQLDialect resolveDialect(String explicit, String url) throws Res
return SQLDialect.DEFAULT;
}
- private static String emptyToNull(String s) {
- return (s == null || s.isEmpty()) ? null : s;
+ private static int hardIdentifierLimitForDialect(SQLDialect dialect) {
+ return switch (dialect.family()) {
+ case POSTGRES -> 63;
+ case MYSQL, MARIADB -> 64;
+ default -> 63;
+ };
}
- @Override
- public void initialize(UimaContext context) throws ResourceInitializationException {
- super.initialize(context);
+ private static int normalizeIdentifierLimit(Integer configured, SQLDialect dialect) {
+ int hardLimit = hardIdentifierLimitForDialect(dialect);
+ int requested = configured == null || configured <= 0 ? hardLimit : configured;
+ return Math.max(16, Math.min(requested, hardLimit));
+ }
- String jdbcUrl = (String) context.getConfigParameterValue(PARAM_JDBC_URL);
- String dbUser = (String) context.getConfigParameterValue(PARAM_DB_USER);
- String dbPass = (String) context.getConfigParameterValue(PARAM_DB_PASS);
+ private static boolean isPgTypeCreateRace(DataAccessException e) {
+ String msg = e.getMessage();
+ return msg != null && msg.contains("pg_type_typname_nsp_index");
+ }
- this.schema = (String) context.getConfigParameterValue(PARAM_SCHEMA);
- this.batchSize = (Integer) context.getConfigParameterValue(PARAM_BATCH_SIZE);
- this.maxIdentifierLength = (Integer) context.getConfigParameterValue(PARAM_MAX_IDENT);
- String sqlDialectName = (String) context.getConfigParameterValue(PARAM_SQL_DIALECT);
+ private static String rootMsg(Throwable t) {
+ Throwable c = t;
+ while (c.getCause() != null && c.getCause() != c) {
+ c = c.getCause();
+ }
+ return c.getMessage();
+ }
- if (isBlank(jdbcUrl)) {
- throw new ResourceInitializationException(
- new IllegalArgumentException("JooqDatabaseWriter: jdbcUrl missing."));
+ private static String bytesToHex(byte[] bytes) {
+ char[] out = new char[bytes.length * 2];
+ for (int i = 0, j = 0; i < bytes.length; i++) {
+ int b = bytes[i] & 0xff;
+ out[j++] = HEX[b >>> 4];
+ out[j++] = HEX[b & 0x0f];
}
+ return new String(out);
+ }
- SQLDialect dialect = resolveDialect(sqlDialectName, jdbcUrl);
- this.schema = normalizeSchemaForDialect(this.schema, dialect);
+ private static long advisoryLockKey(String key) {
+ MessageDigest md = SHA256.get();
+ md.reset();
+ byte[] digest = md.digest(key.getBytes(StandardCharsets.UTF_8));
+ return ByteBuffer.wrap(digest).getLong();
+ }
- HikariConfig cfg = new HikariConfig();
- cfg.setJdbcUrl(jdbcUrl);
- cfg.setUsername(dbUser);
- cfg.setPassword(dbPass);
+ private static Set newIdentitySet() {
+ return Collections.newSetFromMap(new IdentityHashMap<>());
+ }
- // scale writers -> need enough DB connections for batching + DDL + metadata
- cfg.setMaximumPoolSize(16);
- cfg.setMinimumIdle(1);
+ private static String stringParam(UimaContext context, String name, String fallback) {
+ Object v = context.getConfigParameterValue(name);
+ if (v == null) return fallback;
+ String s = String.valueOf(v);
+ return s.isBlank() ? fallback : s;
+ }
- // one transaction per document, no auto-commit
- cfg.setAutoCommit(false);
+ private static int intParam(UimaContext context, String name, int fallback) {
+ Object v = context.getConfigParameterValue(name);
+ if (v == null) return fallback;
+ if (v instanceof Number n) return n.intValue();
+ return Integer.parseInt(String.valueOf(v));
+ }
- cfg.setPoolName("JooqWriterPool");
+ private static boolean booleanParam(UimaContext context, String name, boolean fallback) {
+ Object v = context.getConfigParameterValue(name);
+ if (v == null) return fallback;
+ if (v instanceof Boolean b) return b;
+ return Boolean.parseBoolean(String.valueOf(v));
+ }
+ @Override
+ public void initialize(UimaContext context) throws ResourceInitializationException {
+ super.initialize(context);
+ this.jdbcUrl = stringParam(context, PARAM_JDBC_URL, this.jdbcUrl);
+ this.dbUser = stringParam(context, PARAM_DB_USER, this.dbUser);
+ this.dbPass = stringParam(context, PARAM_DB_PASS, this.dbPass);
+ this.schema = stringParam(context, PARAM_SCHEMA, this.schema);
+ this.sqlDialectName = stringParam(context, PARAM_SQL_DIALECT, this.sqlDialectName);
+ this.pipelineHash = stringParam(context, PARAM_PIPELINE_HASH, this.pipelineHash);
+ this.batchSize = intParam(context, PARAM_BATCH_SIZE, this.batchSize);
+ this.maxIdentifierLength = intParam(context, PARAM_MAX_IDENT, this.maxIdentifierLength);
+ this.storeCoveredText = booleanParam(context, PARAM_STORE_COVERED_TEXT, this.storeCoveredText);
+ this.allowDdl = booleanParam(context, PARAM_ALLOW_DDL, this.allowDdl);
+ this.prepareSchemaOnly = booleanParam(context, PARAM_PREPARE_SCHEMA_ONLY, this.prepareSchemaOnly);
+ if (isBlank(jdbcUrl)) {
+ throw new ResourceInitializationException(new IllegalArgumentException("JooqDatabaseWriter: jdbcUrl missing."));
+ }
+ if (batchSize <= 0) {
+ throw new ResourceInitializationException(new IllegalArgumentException("batchSize must be > 0."));
+ }
+ if (prepareSchemaOnly && !allowDdl) {
+ throw new ResourceInitializationException(new IllegalArgumentException("prepareSchemaOnly=true requires allowDdl=true."));
+ }
+ SQLDialect dialect = resolveDialect(sqlDialectName, jdbcUrl);
+ if (dialect.family() != SQLDialect.POSTGRES) {
+ throw new ResourceInitializationException(new IllegalArgumentException("JooqDatabaseWriter currently supports PostgreSQL only. Detected: " + dialect));
+ }
+ this.maxIdentifierLength = normalizeIdentifierLimit(this.maxIdentifierLength, dialect);
+ this.schema = normalizeSchemaForDialect(this.schema, dialect);
+ HikariConfig cfg = new HikariConfig();
+ cfg.setJdbcUrl(jdbcUrl);
+ if (!isBlank(dbUser)) {
+ cfg.setUsername(dbUser);
+ }
+ if (dbPass != null) {
+ cfg.setPassword(dbPass);
+ }
+ cfg.setMaximumPoolSize(1);
+ cfg.setMinimumIdle(0);
+ cfg.setAutoCommit(false);
+ cfg.setPoolName("JooqWriterPool-" + Integer.toHexString(System.identityHashCode(this)));
+ cfg.addDataSourceProperty("reWriteBatchedInserts", "true");
+ cfg.addDataSourceProperty("ApplicationName", "udav-duui-importer");
try {
this.dataSource = new HikariDataSource(cfg);
} catch (IllegalArgumentException e) {
throw new ResourceInitializationException(e);
}
-
- // Reduce quoting overhead; still safe if you keep identifiers sane.
Settings settings = new Settings().withRenderQuotedNames(RenderQuotedNames.EXPLICIT_DEFAULT_QUOTED);
this.dsl = DSL.using(this.dataSource, dialect, settings);
-
- ensureRegistryTablesOnce();
- }
-
- private void ensureRegistryTablesOnce() {
- if (REGISTRY_READY.get()) return;
- synchronized (DDL_LOCK) {
- if (REGISTRY_READY.get()) return;
-
- // Run all DDL on a single dedicated connection with autoCommit=true so that
- // each statement is immediately visible to the next one on the same connection.
- // This avoids the "relation does not exist" error that occurs when DDL and the
- // subsequent CREATE INDEX run on different pool connections with autoCommit=false.
- dsl.connection(conn -> {
- boolean prevAutoCommit = conn.getAutoCommit();
- conn.setAutoCommit(true);
- try {
- DSLContext ddl = DSL.using(conn, dsl.dialect(), dsl.settings());
-
- ddl.createSchemaIfNotExists(DSL.name(schema)).execute();
-
- ddl.createTableIfNotExists(DSL.name(schema, "uima_type_registry"))
- .column("id", SQLDataType.BIGINT.identity(true))
- .column("uima_type_uri", SQLDataType.CLOB.nullable(false))
- .column("table_name", SQLDataType.CLOB.nullable(false))
- .column("row_count", SQLDataType.BIGINT.defaultValue(0L))
- .column("created_at", SQLDataType.TIMESTAMPWITHTIMEZONE.defaultValue(DSL.currentOffsetDateTime()))
- .constraints(
- DSL.constraint(DSL.name(cut("pk_uima_type_registry"))).primaryKey("id"),
- DSL.constraint(DSL.name(cut("uq_type_uri"))).unique("uima_type_uri"),
- DSL.constraint(DSL.name(cut("uq_table_name"))).unique("table_name")
- )
- .execute();
-
- ddl.createTableIfNotExists(DSL.name(schema, "documents"))
- .column("doc_id", SQLDataType.CLOB.nullable(false))
- .column("uri", SQLDataType.CLOB.nullable(true))
- .column("language", SQLDataType.CLOB.nullable(true))
- .column("content_hash", SQLDataType.VARCHAR(64).nullable(true))
- .column("ts_hash", SQLDataType.VARCHAR(64).nullable(true))
- .constraints(DSL.constraint(DSL.name(cut("pk_documents"))).primaryKey("doc_id"))
- .execute();
-
- ddl.createTableIfNotExists(DSL.name(schema, "type_system_fingerprints"))
- .column("ts_hash", SQLDataType.CLOB.nullable(false))
- .column("created_at", SQLDataType.TIMESTAMPWITHTIMEZONE.defaultValue(DSL.currentOffsetDateTime()))
- .constraints(DSL.constraint(DSL.name(cut("pk_ts_fingerprint"))).primaryKey("ts_hash"))
- .execute();
-
- ddl.createTableIfNotExists(DSL.name(schema, "sofas"))
- .column("doc_id", SQLDataType.CLOB.nullable(false))
- .column("sofa_id", SQLDataType.VARCHAR(128).nullable(false))
- .column("sofa_num", SQLDataType.INTEGER.nullable(true))
- .column("mime_type", SQLDataType.CLOB.nullable(true))
- .column("sofa_uri", SQLDataType.CLOB.nullable(true))
- .column("sofa_string", SQLDataType.CLOB.nullable(true))
- .column("sofa_hash", SQLDataType.VARCHAR(64).nullable(true))
- .column("created_at", SQLDataType.TIMESTAMPWITHTIMEZONE.defaultValue(DSL.currentOffsetDateTime()))
- .constraints(
- DSL.constraint(DSL.name(cut("pk_sofas"))).primaryKey("doc_id", "sofa_id")
- )
- .execute();
-
- ddl.createIndexIfNotExists(DSL.name(cut("idx_sofas_doc_id")))
- .on(DSL.table(DSL.name(schema, "sofas")), DSL.field(DSL.name("doc_id")))
- .execute();
-
- ensureRowCountColumn(ddl);
- } finally {
- conn.setAutoCommit(prevAutoCommit);
- }
- });
-
- REGISTRY_READY.set(true);
- }
- }
-
- private void ensureRowCountColumn(DSLContext ctx) {
- try {
- ctx.select(field(name("row_count")))
- .from(table(name(schema, "uima_type_registry")))
- .limit(1)
- .fetch();
- } catch (DataAccessException e) {
- if (e.getMessage() != null && e.getMessage().contains("does not exist")) {
- try {
- String alterSql = String.format(
- "ALTER TABLE \"%s\".\"uima_type_registry\" ADD COLUMN \"row_count\" BIGINT DEFAULT 0",
- schema);
- ctx.execute(alterSql);
- } catch (DataAccessException ignore) {
- }
- }
+ if (allowDdl) {
+ ensureRegistryTablesOnce();
}
}
@Override
public void process(JCas jCas) {
- // one transaction per document
+ long t0 = System.nanoTime();
+ TypeSystem ts = jCas.getTypeSystem();
+ TsCache cache = getOrBuildTsCache(ts);
+ ensureTablesForTypeSystem(cache);
+ long t1 = System.nanoTime();
+ if (prepareSchemaOnly) {
+ LOGGER.debug("[schema-prep] TypeSystem prepared. tsHash={} ddlMs={}", cache.tsHash, (t1 - t0) / 1_000_000);
+ return;
+ }
+ DocumentMetaData md = getOrCreateDocumentMeta(jCas);
+ final String docId = deterministicDocumentId(jCas, md);
+ final String uri = safe(md.getDocumentUri(), md::getDocumentTitle, () -> docId);
+ final String lang = jCas.getDocumentLanguage();
+ final Map sofas = collectSofas(jCas);
+ final String contentHash = computeContentHashFromSofas(sofas);
+ long t2 = System.nanoTime();
+ DocumentState documentState = getDocumentState(dsl, docId, cache.tsHash, contentHash, pipelineHash);
+ if (documentState.upToDate()) {
+ LOGGER.info("[skip] Document '{}' is up-to-date, skipping.", docId);
+ return;
+ }
+ long t3 = System.nanoTime();
+ LOGGER.info("[process] Document '{}' needs (re)import. ddlMs={} extractMs={} skipCheckMs={}", docId, (t1 - t0) / 1_000_000, (t2 - t1) / 1_000_000, (t3 - t2) / 1_000_000);
+ final Map sofasBySofaId = sofas;
+ final List typeMetas = cache.types;
dsl.transaction(conf -> {
DSLContext tx = DSL.using(conf);
-
- TypeSystem ts = jCas.getTypeSystem();
- ensureTablesForTypeSystem(tx, ts);
-
- DocumentMetaData md = getOrCreateDocumentMeta(jCas);
- String docId = safe(md.getDocumentId(), md::getDocumentUri, md::getDocumentTitle, () -> UUID.randomUUID().toString());
- String uri = safe(md.getDocumentUri(), md::getDocumentTitle, () -> docId);
- String lang = jCas.getDocumentLanguage();
-
- String tsHash = computeTypeSystemHash(ts);
-
- // collect sofa data from the CAS in-memory — no DB writes yet.
- Map sofas = collectSofas(jCas);
- String contentHash = computeContentHashFromSofas(tsHash, sofaHashMap(sofas));
-
- LOGGER.debug("[skip-check] docId='{}' tsHash='{}' contentHash='{}'", docId, tsHash, contentHash);
- if (isDocumentUpToDate(tx, docId, tsHash, contentHash)) {
- LOGGER.info("[skip] Document '{}' is up-to-date, skipping.", docId);
+ acquireDocumentLock(tx, docId);
+ DocumentState lockedState = getDocumentState(tx, docId, cache.tsHash, contentHash, pipelineHash);
+ if (lockedState.upToDate()) {
+ LOGGER.info("[skip] Document '{}' became up-to-date after lock, skipping.", docId);
return;
}
- LOGGER.info("[process] Document '{}' needs (re)import.", docId);
-
- // only write to the DB if the document actually needs (re)importing.
+ if (lockedState.exists()) {
+ deleteExistingRowsForDocument(tx, docId);
+ }
upsertSofas(tx, docId, sofas);
- upsertDocument(tx, docId, uri, lang, tsHash, contentHash);
-
- for (Type t : iterable(ts.getTypeIterator())) {
- if (isSkippableType(t)) continue;
-
- String tableNameHash = typeToTable.get(t.getName());
- if (tableNameHash == null) continue;
-
- String colRowHash = sysColName(tableNameHash, "row_hash");
- String colDocId = sysColName(tableNameHash, "doc_id");
- String colSofaId = sysColName(tableNameHash, "sofa_id");
- String colBegin = sysColName(tableNameHash, "fs_begin");
- String colEnd = sysColName(tableNameHash, "fs_end");
- String colText = sysColName(tableNameHash, "covered_text");
- String colFsJson = sysColName(tableNameHash, "fs_json");
-
- boolean isAnno = ts.subsumes(ts.getType("uima.tcas.Annotation"), t);
-
- if (isAnno) {
- var idx = jCas.getCas().getAnnotationIndex(t);
- if (idx == null || idx.isEmpty()) continue;
-
- String docText = jCas.getDocumentText();
- int docLength = docText != null ? docText.length() : 0;
-
- List batch = new ArrayList<>(Math.min(idx.size(), batchSize));
- for (AnnotationFS fs : idx) {
- String sofaId = sofaIdForFs(fs);
-
- Map, Object> values = new LinkedHashMap<>();
- values.put(field(name(colRowHash)), computeRowHash(ts, t, docId, tableNameHash, fs));
- values.put(field(name(colDocId)), docId);
- values.put(field(name(colSofaId)), sofaId);
- values.put(field(name(colBegin)), fs.getBegin());
- values.put(field(name(colEnd)), fs.getEnd());
- values.put(field(name(colText)), safeCoveredText(docText, docLength, fs.getBegin(), fs.getEnd()));
-
- for (Feature f : t.getFeatures()) {
- if (!isPrimitive(f)) continue;
- values.putIfAbsent(field(name(featColName(tableNameHash, f))),
- FeatureJsonSerializer.readPrimitive(fs, f));
+ upsertDocument(tx, docId, uri, lang, cache.tsHash, contentHash, pipelineHash);
+ org.apache.uima.cas.CAS base = jCas.getCas();
+ List views = new ArrayList<>();
+ for (Iterator vit = base.getViewIterator(); vit.hasNext(); ) {
+ views.add(vit.next());
+ }
+ for (TypeMeta meta : typeMetas) {
+ if (meta.tableNameHash == null) continue;
+ long typeStart = System.nanoTime();
+ int rowsForType = 0;
+ CopyBatch copy = new CopyBatch(tx, meta.tableNameHash, copyColumnNames(meta.tableNameHash, meta.isAnno, meta.primFeats));
+ Object[] row = new Object[meta.bindCount];
+ Object[] featValues = new Object[meta.primFeats.size()];
+ Set seenAnnoFs = meta.isAnno ? newIdentitySet() : null;
+ Set seenGenericFs = meta.isAnno ? null : newIdentitySet();
+ for (org.apache.uima.cas.CAS view : views) {
+ if (meta.isAnno) {
+ var idx = view.getAnnotationIndex(meta.type);
+ if (idx == null || idx.size() == 0) continue;
+ SofaData sd = sofaDataForView(sofasBySofaId, view);
+ String docText = sd != null ? sd.text() : view.getDocumentText();
+ int docLength = docText != null ? docText.length() : 0;
+ for (AnnotationFS fs : idx) {
+ if (!fs.getType().getName().equals(meta.typeName)) continue;
+ if (!seenAnnoFs.add(fs)) continue;
+ for (int i = 0; i < meta.primFeats.size(); i++) {
+ featValues[i] = FeatureJsonSerializer.readPrimitive(fs, meta.primFeats.get(i));
+ }
+ String fsViewName = sofaIdForFs(fs);
+ int begin = fs.getBegin();
+ int end = fs.getEnd();
+ row[0] = docId;
+ row[1] = fsViewName;
+ row[2] = begin;
+ row[3] = end;
+ int offset;
+ if (storeCoveredText) {
+ row[4] = safeCoveredText(docText, docLength, begin, end);
+ offset = 5;
+ } else {
+ offset = 4;
+ }
+ for (int i = 0; i < meta.primFeats.size(); i++) {
+ row[offset + i] = featValues[i];
+ }
+ copy.add(row);
+ rowsForType++;
}
-
- values.put(field(name(colFsJson)), FeatureJsonSerializer.toJsonPrimitivesOnly(fs));
-
- batch.add(insertIgnore(tx, tableNameHash, values, colRowHash));
- if (batch.size() >= batchSize) {
- tx.batch(batch).execute();
- batch.clear();
+ } else {
+ var it = view.getIndexRepository().getAllIndexedFS(meta.type);
+ if (it == null) continue;
+ while (it.hasNext()) {
+ org.apache.uima.cas.FeatureStructure fs = it.next();
+ if (!fs.getType().getName().equals(meta.typeName)) continue;
+ if (!seenGenericFs.add(fs)) continue;
+ for (int i = 0; i < meta.primFeats.size(); i++) {
+ featValues[i] = FeatureJsonSerializer.readPrimitive(fs, meta.primFeats.get(i));
+ }
+ String fsViewName = sofaIdForFs(fs);
+ row[0] = docId;
+ row[1] = fsViewName;
+ for (int i = 0; i < meta.primFeats.size(); i++) {
+ row[2 + i] = featValues[i];
+ }
+ copy.add(row);
+ rowsForType++;
}
}
- if (!batch.isEmpty()) tx.batch(batch).execute();
-
- } else {
- var it = jCas.getCas().getIndexRepository().getAllIndexedFS(t);
- if (it == null) continue;
-
- List batch = new ArrayList<>(batchSize);
- it.forEachRemaining(fs -> {
- String sofaId = sofaIdForFs(fs);
-
- Map, Object> values = new LinkedHashMap<>();
- values.put(field(name(colRowHash)), computeRowHash(ts, t, docId, tableNameHash, fs));
- values.put(field(name(colDocId)), docId);
- values.put(field(name(colSofaId)), sofaId);
-
- for (Feature f : t.getFeatures()) {
- if (!isPrimitive(f)) continue;
- values.put(field(name(featColName(tableNameHash, f))),
- FeatureJsonSerializer.readPrimitive(fs, f));
- }
- values.put(field(name(colFsJson)), FeatureJsonSerializer.toJsonPrimitivesOnly(fs));
-
- batch.add(insertIgnore(tx, tableNameHash, values, colRowHash));
- if (batch.size() >= batchSize) {
- tx.batch(batch).execute();
- batch.clear();
- }
- });
- if (!batch.isEmpty()) tx.batch(batch).execute();
+ }
+ copy.flush();
+ if (rowsForType > 0) {
+ LOGGER.debug("[type-import-copy] doc={} type={} table={} rows={} ms={}", docId, meta.typeName, meta.tableNameHash, rowsForType, (System.nanoTime() - typeStart) / 1_000_000);
}
}
});
+ LOGGER.info("[done] Document '{}' imported in {}ms.", docId, (System.nanoTime() - t0) / 1_000_000);
}
- private void ensureTablesForTypeSystem(DSLContext ctx, TypeSystem ts) {
- String tsHash = computeTypeSystemHash(ts);
- if (seenTsFingerprints.contains(tsHash)) return;
-
- synchronized (DDL_LOCK) {
- if (seenTsFingerprints.contains(tsHash)) return;
-
- if (fingerprintExists(ctx, tsHash)) {
- preloadTypeToTableFromRegistry(ctx, ts);
- seenTsFingerprints.add(tsHash);
- return;
- }
-
- for (Type t : iterable(ts.getTypeIterator())) {
- if (isSkippableType(t)) continue;
-
- String uimaType = t.getName();
- String tableNameHash = typeToTable.computeIfAbsent(uimaType, this::toSafeTableName);
- if (createdTables.contains(tableNameHash)) continue;
-
- String pkCol = pkColName(tableNameHash);
- String colRowH = sysColName(tableNameHash, "row_hash");
- String colDoc = sysColName(tableNameHash, "doc_id");
- String colSofa = sysColName(tableNameHash, "sofa_id");
- String colBegin = sysColName(tableNameHash, "fs_begin");
- String colEnd = sysColName(tableNameHash, "fs_end");
- String colText = sysColName(tableNameHash, "covered_text");
- String colFsJs = sysColName(tableNameHash, "fs_json");
-
- List> cols = new ArrayList<>();
- cols.add(field(name(pkCol), SQLDataType.BIGINT.identity(true)));
- cols.add(field(name(colRowH), SQLDataType.VARCHAR(64).nullable(false)));
- cols.add(field(name(colDoc), SQLDataType.CLOB.nullable(false)));
- cols.add(field(name(colSofa), SQLDataType.VARCHAR(128).nullable(false)));
-
- boolean isAnno = ts.subsumes(ts.getType("uima.tcas.Annotation"), t);
- if (isAnno) {
- cols.add(field(name(colBegin), SQLDataType.INTEGER.nullable(false)));
- cols.add(field(name(colEnd), SQLDataType.INTEGER.nullable(false)));
- cols.add(field(name(colText), SQLDataType.CLOB.nullable(true)));
+ private void appendCopyTextValue(StringBuilder sb, Object value) {
+ if (value == null) {
+ sb.append("\\N");
+ return;
+ }
+ String s;
+ if (value instanceof Boolean b) {
+ s = b ? "true" : "false";
+ } else {
+ s = String.valueOf(value);
+ }
+ for (int i = 0; i < s.length(); i++) {
+ char c = s.charAt(i);
+ switch (c) {
+ case '\\' -> sb.append("\\\\");
+ case '\t' -> sb.append("\\t");
+ case '\n' -> sb.append("\\n");
+ case '\r' -> sb.append("\\r");
+ case '\0' -> {
}
+ default -> sb.append(c);
+ }
+ }
+ }
- for (Feature f : t.getFeatures()) {
- if (!isPrimitive(f)) continue;
+ private List copyColumnNames(String tableNameHash, boolean isAnno, List primFeats) {
+ List cols = new ArrayList<>();
+ cols.add(sysColName(tableNameHash, "doc_id"));
+ cols.add(sysColName(tableNameHash, "sofa_id"));
+ if (isAnno) {
+ cols.add(sysColName(tableNameHash, "fs_begin"));
+ cols.add(sysColName(tableNameHash, "fs_end"));
+ if (storeCoveredText) {
+ cols.add(sysColName(tableNameHash, "covered_text"));
+ }
+ }
+ for (Feature f : primFeats) {
+ cols.add(featColName(tableNameHash, f));
+ }
+ return cols;
+ }
- DataType dt = mapPrimitiveType(f.getRange().getName()).nullable(true);
- cols.add(field(name(featColName(tableNameHash, f)), dt));
- }
+ private void acquireDocumentLock(DSLContext tx, String docId) {
+ tx.execute("SELECT pg_advisory_xact_lock(?)", advisoryLockKey("uima-doc:" + docId));
+ }
- cols.add(field(name(colFsJs), SQLDataType.JSON.nullable(true)));
-
- ctx.createTableIfNotExists(name(schema, tableNameHash))
- .columns(cols)
- .constraints(
- constraint(name(cut("pk_" + tableNameHash))).primaryKey(field(name(pkCol))),
- constraint(name(cut("uq_" + tableNameHash + "_rowhash"))).unique(field(name(colRowH)))
- )
- .execute();
-
- if (isAnno) {
- ctx.createIndexIfNotExists(name(cut("idx_" + tableNameHash + "_doc_sofa_begin")))
- .on(table(name(schema, tableNameHash)),
- field(name(colDoc)), field(name(colSofa)), field(name(colBegin)))
- .execute();
- } else {
- ctx.createIndexIfNotExists(name(cut("idx_" + tableNameHash + "_doc_sofa")))
- .on(table(name(schema, tableNameHash)),
- field(name(colDoc)), field(name(colSofa)))
- .execute();
- }
+ private void deleteExistingRowsForDocument(DSLContext tx, String docId) {
+ for (String tableName : existingRegisteredTypeTables(tx)) {
+ Field