mattcasters commented on code in PR #8428: URL: https://github.com/apache/hop/pull/8428#discussion_r4029621227
########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/upsert/PgVectorUpsertDialog.java: ########## @@ -0,0 +1,396 @@ +/* + * 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.hop.pgvector.transforms.upsert; + +import java.util.ArrayList; +import java.util.List; +import org.apache.hop.core.Const; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.pgvector.util.PgVectorColumnMapping; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.ui.core.PropsUi; +import org.apache.hop.ui.core.dialog.BaseDialog; +import org.apache.hop.ui.core.dialog.ErrorDialog; +import org.apache.hop.ui.core.gui.GuiResource; +import org.apache.hop.ui.core.widget.ColumnInfo; +import org.apache.hop.ui.core.widget.ComboVar; +import org.apache.hop.ui.core.widget.LabelTextVar; +import org.apache.hop.ui.core.widget.MetaSelectionLine; +import org.apache.hop.ui.core.widget.TableView; +import org.apache.hop.ui.pipeline.transform.BaseTransformDialog; +import org.eclipse.swt.SWT; +import org.eclipse.swt.custom.CCombo; +import org.eclipse.swt.custom.CTabFolder; +import org.eclipse.swt.custom.CTabItem; +import org.eclipse.swt.events.FocusAdapter; +import org.eclipse.swt.events.FocusEvent; +import org.eclipse.swt.layout.FormAttachment; +import org.eclipse.swt.layout.FormData; +import org.eclipse.swt.layout.FormLayout; +import org.eclipse.swt.widgets.Button; +import org.eclipse.swt.widgets.Composite; +import org.eclipse.swt.widgets.Control; +import org.eclipse.swt.widgets.Label; +import org.eclipse.swt.widgets.Shell; +import org.eclipse.swt.widgets.TableItem; + +public class PgVectorUpsertDialog extends BaseTransformDialog { + + private static final Class<?> PKG = PgVectorUpsertMeta.class; + + private final PgVectorUpsertMeta input; + + private MetaSelectionLine<org.apache.hop.core.database.DatabaseMeta> wConnection; + private LabelTextVar wSchemaName; + private LabelTextVar wTableName; + private ComboVar wIdField; + private ComboVar wDocumentIdField; + private ComboVar wChunkIndexField; + private ComboVar wContentField; + private ComboVar wEmbeddingField; + private LabelTextVar wEmbeddingDimensions; + private Button wCreateTableIfMissing; + private Button wCreateHnswIndex; + private Button wDeleteDocumentBeforeUpsert; + private CCombo wIndexMetric; + private LabelTextVar wCommitSize; + private TableView wMappings; + + public PgVectorUpsertDialog( + Shell parent, + IVariables variables, + PgVectorUpsertMeta transformMeta, + PipelineMeta pipelineMeta) { + super(parent, variables, transformMeta, pipelineMeta); + input = transformMeta; + } + + @Override + public String open() { + Control lastControl = + createShell(BaseMessages.getString(PKG, "PgVectorUpsertDialog.Shell.Title")); + buildButtonBar().ok(e -> ok()).cancel(e -> cancel()).build(); + + CTabFolder tabFolder = new CTabFolder(shell, SWT.BORDER); Review Comment: **[bug]** Same layout violation as search: hand-built tabs and FormAttachment rows for ~14 main-tab controls, no scrolled composite, mappings table not registered as an extra group. This is the crowded dialog; lower options (index metric, commit size) are the ones that disappear first. `Const.toInt` / `Const.toDouble` on `LabelTextVar` values also silently replace `${TOP_K}`-style input with defaults at OK time. **Suggestion:** Follow the same GuiCompositeWidgets TABS recipe as `ParquetOutputDialog` / `DatabaseValueValidationDialog`. Keep commit size, embedding dimensions, and similar numerics as resolvable strings if variables should work at runtime. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/search/PgVectorSearchDialog.java: ########## @@ -0,0 +1,386 @@ +/* + * 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.hop.pgvector.transforms.search; + +import java.util.ArrayList; +import java.util.List; +import org.apache.hop.core.Const; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.pgvector.util.PgVectorSearchFilter; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.ui.core.PropsUi; +import org.apache.hop.ui.core.dialog.BaseDialog; +import org.apache.hop.ui.core.dialog.ErrorDialog; +import org.apache.hop.ui.core.gui.GuiResource; +import org.apache.hop.ui.core.widget.ColumnInfo; +import org.apache.hop.ui.core.widget.ComboVar; +import org.apache.hop.ui.core.widget.LabelTextVar; +import org.apache.hop.ui.core.widget.MetaSelectionLine; +import org.apache.hop.ui.core.widget.TableView; +import org.apache.hop.ui.pipeline.transform.BaseTransformDialog; +import org.eclipse.swt.SWT; +import org.eclipse.swt.custom.CCombo; +import org.eclipse.swt.custom.CTabFolder; +import org.eclipse.swt.custom.CTabItem; +import org.eclipse.swt.events.FocusAdapter; +import org.eclipse.swt.events.FocusEvent; +import org.eclipse.swt.layout.FormAttachment; +import org.eclipse.swt.layout.FormData; +import org.eclipse.swt.layout.FormLayout; +import org.eclipse.swt.widgets.Button; +import org.eclipse.swt.widgets.Composite; +import org.eclipse.swt.widgets.Control; +import org.eclipse.swt.widgets.Label; +import org.eclipse.swt.widgets.Shell; +import org.eclipse.swt.widgets.TableItem; + +public class PgVectorSearchDialog extends BaseTransformDialog { + + private static final Class<?> PKG = PgVectorSearchMeta.class; + + private final PgVectorSearchMeta input; + + private MetaSelectionLine<org.apache.hop.core.database.DatabaseMeta> wConnection; + private LabelTextVar wSchemaName; + private LabelTextVar wTableName; + private ComboVar wEmbeddingField; + private LabelTextVar wTopK; + private LabelTextVar wMinScore; + private Button wEatRowOnNoMatch; + private CCombo wDistanceMetric; + private LabelTextVar wResultIdField; + private LabelTextVar wResultDocumentIdField; + private LabelTextVar wResultChunkIndexField; + private LabelTextVar wResultContentField; + private LabelTextVar wResultScoreField; + private TableView wFilters; + + public PgVectorSearchDialog( + Shell parent, + IVariables variables, + PgVectorSearchMeta transformMeta, + PipelineMeta pipelineMeta) { + super(parent, variables, transformMeta, pipelineMeta); + input = transformMeta; + } + + @Override + public String open() { + Control lastControl = + createShell(BaseMessages.getString(PKG, "PgVectorSearchDialog.Shell.Title")); + buildButtonBar().ok(e -> ok()).cancel(e -> cancel()).build(); + + CTabFolder tabFolder = new CTabFolder(shell, SWT.BORDER); Review Comment: **[bug]** New search dialog hand-lays a `CTabFolder` with FormAttachment rows instead of grouped `@GuiWidgetElement` + `GuiCompositeWidgets.addScrolledComposite`. The main tab is not a scrolled composite, so the stacked connection/fields/metric/result widgets clip on short windows while OK/Cancel stay pinned. Filters `TableView` is a free-floating extra group, which the Hop dialog recipe requires via `registerExtraGroup`. Field combos are `SWT.READ_ONLY`, so names and variables cannot be typed. **Suggestion:** Annotate `PgVectorSearchMeta` with `@GuiPlugin` and `@GuiWidgetElement` (`GuiWidgetGroupType.TABS`, Main + Filters). Build the dialog with `createShell` / `buildButtonBar` / `GuiCompositeWidgets.addScrolledComposite`, and register the filters `TableView` as an extra group. Use `GuiElementType.METADATA` for the connection and non-read-only field widgets. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/search/PgVectorSearchMeta.java: ########## @@ -0,0 +1,244 @@ +/* + * 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.hop.pgvector.transforms.search; + +import java.util.ArrayList; +import java.util.List; +import lombok.Getter; +import lombok.Setter; +import org.apache.hop.core.CheckResult; +import org.apache.hop.core.ICheckResult; +import org.apache.hop.core.annotations.Transform; +import org.apache.hop.core.exception.HopPluginException; +import org.apache.hop.core.exception.HopTransformException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.value.ValueMetaInteger; +import org.apache.hop.core.row.value.ValueMetaNumber; +import org.apache.hop.core.row.value.ValueMetaString; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.metadata.api.HopMetadataProperty; +import org.apache.hop.metadata.api.HopMetadataPropertyType; +import org.apache.hop.metadata.api.IHopMetadataProvider; +import org.apache.hop.pgvector.util.PgVectorSearchFilter; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransformMeta; +import org.apache.hop.pipeline.transform.TransformMeta; + +@Getter +@Setter +@Transform( + id = "PgVectorSearch", + image = "vector-store.svg", + name = "i18n::PgVectorSearch.Name", + description = "i18n::PgVectorSearch.Description", + categoryDescription = "i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Lookup", + keywords = "pgvector,vector,embedding,ai,rag,postgres,postgresql,similarity,search", + documentationUrl = "/pipeline/transforms/pgvector-search.html") +public class PgVectorSearchMeta extends BaseTransformMeta<PgVectorSearch, PgVectorSearchData> { Review Comment: **[bug]** `PgVectorSearch` calls `putError` when `isDoingErrorHandling()` is true, but `BaseTransformMeta.supportsErrorHandling()` defaults to false and this class does not override it. Hop GUI will not offer an error hop, so the error-handling path is unreachable from the UI. **Suggestion:** Override `supportsErrorHandling()` to return true, matching `TableOutputMeta` and other lookup/output transforms that implement `putError`. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/upsert/PgVectorUpsert.java: ########## @@ -0,0 +1,429 @@ +/* + * 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.hop.pgvector.transforms.upsert; + +import java.sql.BatchUpdateException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.util.Utils; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.pgvector.util.EmbeddingJsonParser; +import org.apache.hop.pgvector.util.PgVectorColumnMapping; +import org.apache.hop.pgvector.util.PgVectorColumnType; +import org.apache.hop.pgvector.util.PgVectorDatabase; +import org.apache.hop.pgvector.util.PgVectorSchemaBuilder; +import org.apache.hop.pgvector.util.PgVectorSqlBuilder; +import org.apache.hop.pgvector.util.PgVectorTableColumn; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.Pipeline; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransform; +import org.apache.hop.pipeline.transform.TransformMeta; + +/** Upserts chunk rows with embeddings into a PostgreSQL pgvector table. */ +public class PgVectorUpsert extends BaseTransform<PgVectorUpsertMeta, PgVectorUpsertData> { + + private static final Class<?> PKG = PgVectorUpsertMeta.class; + + /** + * Upper bound on the per-document delete cache. Without a cap this set grows for the lifetime of + * the run, which matters when a corpus has millions of distinct documents. + */ + private static final int MAX_DELETED_DOCUMENT_CACHE = 100_000; + + public PgVectorUpsert( + TransformMeta transformMeta, + PgVectorUpsertMeta meta, + PgVectorUpsertData data, + int copyNr, + PipelineMeta pipelineMeta, + Pipeline pipeline) { + super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline); + } + + @Override + public boolean init() { + if (Utils.isEmpty(meta.getConnection())) { + logError(BaseMessages.getString(PKG, "PgVectorUpsert.Validation.ConnectionRequired")); + return false; + } + return super.init(); + } + + @Override + public boolean processRow() throws HopException { + Object[] row = getRow(); + if (row == null) { + commitBatch(); + setOutputDone(); + return false; + } + + if (first) { + first = false; + data.inputRowMeta = getInputRowMeta(); + openDatabase(); + resolveFieldIndices(); + } + + try { + addRowToBatch(row); + } catch (HopException e) { + if (getTransformMeta().isDoingErrorHandling()) { + putError(data.inputRowMeta, row, 1, e.getMessage(), null, "PGVECTORUPSERT001"); + } else { + throw e; + } + } + + return true; + } + + private void addRowToBatch(Object[] row) throws HopException { + String documentId = getFieldAsString(row, data.documentIdFieldIndex); + String content = getFieldAsString(row, data.contentFieldIndex); + // Taken as the raw value: a field of the Vector value type is already a float[], and + // rendering it to text here only to parse it back would be wasted work on every row. + Object embedding = row[data.embeddingFieldIndex]; + + if (Utils.isEmpty(content) || EmbeddingJsonParser.isEmpty(embedding)) { + // Nothing to store for this row, but it still belongs in the output stream. + putRow(data.inputRowMeta, row); + return; + } + + // The delete has to be flushed together with the inserts it makes room for. Committing the + // pending batch first keeps "delete document, then re-insert its chunks" from being split + // across a failure boundary, which would otherwise leave the document deleted and not + // replaced. + if (shouldDeleteDocument(documentId)) { + commitBatch(); + deleteDocument(documentId); + } + + String id = resolveId(row); + String chunkIndex = getFieldAsString(row, data.chunkIndexFieldIndex); + + try { + bindInsertRow(row, id, documentId, chunkIndex, content, embedding); + data.insertStatement.addBatch(); + data.batchRows.add(row); + } catch (HopException e) { + throw e; + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.BindingRow", String.valueOf(id)), e); + } + + if (meta.getCommitSize() > 0 && data.batchRows.size() >= meta.getCommitSize()) { + commitBatch(); + } + } + + private boolean shouldDeleteDocument(String documentId) { + return meta.isDeleteDocumentBeforeUpsert() + && data.deleteStatement != null + && !Utils.isEmpty(documentId) + && !data.deletedDocuments.contains(documentId); + } + + private void deleteDocument(String documentId) throws HopException { + try { + data.deleteStatement.setString(1, documentId); + data.deleteStatement.executeUpdate(); + if (data.deletedDocuments.size() >= MAX_DELETED_DOCUMENT_CACHE) { + Iterator<String> oldest = data.deletedDocuments.iterator(); + oldest.next(); + oldest.remove(); + } + data.deletedDocuments.add(documentId); + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.DeletingDocument", documentId), e); + } + } + + private void bindInsertRow( + Object[] row, + String id, + String documentId, + String chunkIndex, + String content, + Object embedding) + throws Exception { + String vectorLiteral = EmbeddingJsonParser.toPgVectorLiteral(embedding); + for (PgVectorUpsertData.MappingBinding binding : data.mappingBindings) { + Object value = + switch (binding.column.name()) { + case "id" -> id; + case "document_id" -> documentId; + case "chunk_index" -> parseChunkIndex(chunkIndex); + case "content" -> content; + case "embedding" -> vectorLiteral; + default -> getFieldAsString(row, binding.streamFieldIndex); + }; + if (binding.column.type() == PgVectorColumnType.INTEGER) { + data.insertStatement.setInt(binding.statementIndex, (Integer) value); + } else { + data.insertStatement.setString( + binding.statementIndex, value == null ? null : value.toString()); + } + } + } + + private void openDatabase() throws HopException { + try { + data.database = + PgVectorDatabase.connect(this, this, getMetadataProvider(), meta.getConnection()); + } catch (HopException e) { + throw e; + } + + try { + data.database.setCommit(meta.getCommitSize()); + VectorDistanceMetric indexMetric = + meta.getIndexMetric() != null ? meta.getIndexMetric() : VectorDistanceMetric.COSINE; + String schemaName = resolve(meta.getSchemaName()); + String tableName = resolve(meta.getTableName()); + if (meta.isCreateTableIfMissing()) { + PgVectorDatabase.ensureSchema(data.database, meta, schemaName, tableName, indexMetric); + } + String qualifiedTable = PgVectorSqlBuilder.qualifiedTable(schemaName, tableName); + data.tableColumns = PgVectorSchemaBuilder.tableColumns(meta); + data.insertStatement = + data.database + .getConnection() + .prepareStatement(PgVectorSqlBuilder.upsertSql(qualifiedTable, data.tableColumns)); + if (meta.isDeleteDocumentBeforeUpsert()) { + data.deleteStatement = + data.database + .getConnection() + .prepareStatement(PgVectorSqlBuilder.deleteByDocumentIdSql(qualifiedTable)); + } + } catch (Exception e) { + // Do not leak the connection when the statements cannot be prepared. + closeDatabase(); + throw new HopException(BaseMessages.getString(PKG, "PgVectorUpsert.Error.Initializing"), e); + } + } + + private void resolveFieldIndices() throws HopException { + data.contentFieldIndex = requireFieldIndex(meta.getContentField()); + data.embeddingFieldIndex = requireFieldIndex(meta.getEmbeddingField()); + data.documentIdFieldIndex = optionalFieldIndex(meta.getDocumentIdField()); + data.chunkIndexFieldIndex = optionalFieldIndex(meta.getChunkIndexField()); + if (!Utils.isEmpty(meta.getIdField())) { + data.idFieldIndex = requireFieldIndex(meta.getIdField()); + } + + data.mappingBindings = new ArrayList<>(); + int parameterIndex = 1; + for (PgVectorTableColumn column : data.tableColumns) { + int streamIndex = resolveStreamIndex(column.name()); + data.mappingBindings.add( + new PgVectorUpsertData.MappingBinding(streamIndex, parameterIndex++, column)); + } + } + + private int resolveStreamIndex(String columnName) throws HopException { + return switch (columnName) { + case "id" -> data.idFieldIndex; + case "document_id" -> data.documentIdFieldIndex; + case "chunk_index" -> data.chunkIndexFieldIndex; + case "content" -> data.contentFieldIndex; + case "embedding" -> data.embeddingFieldIndex; + default -> requireMappedFieldIndex(columnName); + }; + } + + private int requireMappedFieldIndex(String columnName) throws HopException { + if (meta.getColumnMappings() != null) { + for (PgVectorColumnMapping mapping : meta.getColumnMappings()) { + if (mapping != null + && columnName.equals( + PgVectorSchemaBuilder.normalizeColumnName(mapping.getColumnName()))) { + return requireFieldIndex(mapping.getStreamField()); + } + } + } + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.NoMappingForColumn", columnName)); + } + + private int requireFieldIndex(String fieldName) throws HopException { + int index = data.inputRowMeta.indexOfValue(fieldName); + if (index < 0) { + throw new HopException( + BaseMessages.getString( + PKG, "PgVectorUpsert.Error.FieldNotFound", String.valueOf(fieldName))); + } + return index; + } + + private int optionalFieldIndex(String fieldName) { + if (Utils.isEmpty(fieldName)) { + return -1; + } + return data.inputRowMeta.indexOfValue(fieldName); + } + + private String resolveId(Object[] row) throws HopException { + if (data.idFieldIndex >= 0) { + return getFieldAsString(row, data.idFieldIndex); + } + String documentId = getFieldAsString(row, data.documentIdFieldIndex); + String chunkIndex = getFieldAsString(row, data.chunkIndexFieldIndex); + if (Utils.isEmpty(documentId) && Utils.isEmpty(chunkIndex)) { + throw new HopException(BaseMessages.getString(PKG, "PgVectorUpsert.Error.NoIdAvailable")); + } + return documentId + "_" + chunkIndex; + } + + private String getFieldAsString(Object[] row, int index) throws HopException { + if (index < 0) { + return null; + } + return data.inputRowMeta.getString(row, index); + } + + private int parseChunkIndex(String chunkIndex) throws HopException { + if (Utils.isEmpty(chunkIndex)) { + return 0; + } + try { + return Integer.parseInt(chunkIndex.trim()); + } catch (NumberFormatException e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.InvalidChunkIndex", chunkIndex), e); + } + } + + /** + * Executes and commits the pending batch, then releases the buffered rows downstream. When error + * handling is on, {@link BatchUpdateException#getUpdateCounts()} is used to divert only the rows + * the database actually rejected. + */ + private void commitBatch() throws HopException { + if (data.insertStatement == null || data.batchRows.isEmpty()) { + return; + } + + List<Object[]> rows = new ArrayList<>(data.batchRows); + data.batchRows.clear(); + + boolean[] failed = new boolean[rows.size()]; + HopException failure = null; + + try { + data.insertStatement.executeBatch(); + } catch (BatchUpdateException e) { + failure = new HopException(BaseMessages.getString(PKG, "PgVectorUpsert.Error.Committing"), e); + markFailedRows(e, failed); + } catch (Exception e) { + failure = new HopException(BaseMessages.getString(PKG, "PgVectorUpsert.Error.Committing"), e); + java.util.Arrays.fill(failed, true); + } + + if (failure != null && !getTransformMeta().isDoingErrorHandling()) { + throw failure; + } + + try { + if (meta.getCommitSize() > 0) { Review Comment: **[bug]** After `executeBatch()` throws, the code still calls `database.commit()` when error handling is on. PostgreSQL does not support error recovery on batch updates (`PostgreSqlDatabaseMeta.IsSupportsErrorHandlingOnBatchUpdates()` is false; it uses savepoints instead). A failed statement aborts the transaction, so this commit throws "current transaction is aborted", `batchRows` was already cleared, and remaining rows are neither putRow'd nor putError'd. `TableOutput` disables batch mode on PostgreSQL when error handling is enabled and uses savepoints. **Suggestion:** If error handling is on, do not use JDBC batch on PostgreSQL (or use savepoints per row). On batch failure, `clearBatch()`, rollback or finish the aborted transaction, then divert rows. Do not commit a transaction that just failed `executeBatch()`. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/upsert/PgVectorUpsertMeta.java: ########## @@ -0,0 +1,249 @@ +/* + * 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.hop.pgvector.transforms.upsert; + +import java.util.ArrayList; +import java.util.List; +import lombok.Getter; +import lombok.Setter; +import org.apache.hop.core.CheckResult; +import org.apache.hop.core.ICheckResult; +import org.apache.hop.core.annotations.Transform; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.metadata.api.HopMetadataProperty; +import org.apache.hop.metadata.api.HopMetadataPropertyType; +import org.apache.hop.metadata.api.IHopMetadataProvider; +import org.apache.hop.pgvector.util.PgVectorColumnMapping; +import org.apache.hop.pgvector.util.PgVectorSchemaBuilder; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransformMeta; +import org.apache.hop.pipeline.transform.TransformMeta; + +@Getter +@Setter +@Transform( + id = "PgVectorUpsert", + image = "vector-store.svg", + name = "i18n::PgVectorUpsert.Name", + description = "i18n::PgVectorUpsert.Description", + categoryDescription = "i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Output", + keywords = "pgvector,vector,embedding,ai,rag,postgres,postgresql,upsert", + documentationUrl = "/pipeline/transforms/pgvector-upsert.html") +public class PgVectorUpsertMeta extends BaseTransformMeta<PgVectorUpsert, PgVectorUpsertData> { Review Comment: **[bug]** Same gap as search: upsert implements `putError` for bind failures and batch rejects, but does not override `supportsErrorHandling()`, so users cannot attach an error hop. **Suggestion:** Override `supportsErrorHandling()` to return true. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/upsert/PgVectorUpsert.java: ########## @@ -0,0 +1,429 @@ +/* + * 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.hop.pgvector.transforms.upsert; + +import java.sql.BatchUpdateException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.util.Utils; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.pgvector.util.EmbeddingJsonParser; +import org.apache.hop.pgvector.util.PgVectorColumnMapping; +import org.apache.hop.pgvector.util.PgVectorColumnType; +import org.apache.hop.pgvector.util.PgVectorDatabase; +import org.apache.hop.pgvector.util.PgVectorSchemaBuilder; +import org.apache.hop.pgvector.util.PgVectorSqlBuilder; +import org.apache.hop.pgvector.util.PgVectorTableColumn; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.Pipeline; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransform; +import org.apache.hop.pipeline.transform.TransformMeta; + +/** Upserts chunk rows with embeddings into a PostgreSQL pgvector table. */ +public class PgVectorUpsert extends BaseTransform<PgVectorUpsertMeta, PgVectorUpsertData> { + + private static final Class<?> PKG = PgVectorUpsertMeta.class; + + /** + * Upper bound on the per-document delete cache. Without a cap this set grows for the lifetime of + * the run, which matters when a corpus has millions of distinct documents. + */ + private static final int MAX_DELETED_DOCUMENT_CACHE = 100_000; + + public PgVectorUpsert( + TransformMeta transformMeta, + PgVectorUpsertMeta meta, + PgVectorUpsertData data, + int copyNr, + PipelineMeta pipelineMeta, + Pipeline pipeline) { + super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline); + } + + @Override + public boolean init() { + if (Utils.isEmpty(meta.getConnection())) { + logError(BaseMessages.getString(PKG, "PgVectorUpsert.Validation.ConnectionRequired")); + return false; + } + return super.init(); + } + + @Override + public boolean processRow() throws HopException { + Object[] row = getRow(); + if (row == null) { + commitBatch(); + setOutputDone(); + return false; + } + + if (first) { + first = false; + data.inputRowMeta = getInputRowMeta(); + openDatabase(); + resolveFieldIndices(); + } + + try { + addRowToBatch(row); + } catch (HopException e) { + if (getTransformMeta().isDoingErrorHandling()) { + putError(data.inputRowMeta, row, 1, e.getMessage(), null, "PGVECTORUPSERT001"); + } else { + throw e; + } + } + + return true; + } + + private void addRowToBatch(Object[] row) throws HopException { + String documentId = getFieldAsString(row, data.documentIdFieldIndex); + String content = getFieldAsString(row, data.contentFieldIndex); + // Taken as the raw value: a field of the Vector value type is already a float[], and + // rendering it to text here only to parse it back would be wasted work on every row. + Object embedding = row[data.embeddingFieldIndex]; + + if (Utils.isEmpty(content) || EmbeddingJsonParser.isEmpty(embedding)) { + // Nothing to store for this row, but it still belongs in the output stream. + putRow(data.inputRowMeta, row); + return; + } + + // The delete has to be flushed together with the inserts it makes room for. Committing the + // pending batch first keeps "delete document, then re-insert its chunks" from being split + // across a failure boundary, which would otherwise leave the document deleted and not + // replaced. + if (shouldDeleteDocument(documentId)) { + commitBatch(); + deleteDocument(documentId); + } + + String id = resolveId(row); + String chunkIndex = getFieldAsString(row, data.chunkIndexFieldIndex); + + try { + bindInsertRow(row, id, documentId, chunkIndex, content, embedding); + data.insertStatement.addBatch(); + data.batchRows.add(row); + } catch (HopException e) { + throw e; + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.BindingRow", String.valueOf(id)), e); + } + + if (meta.getCommitSize() > 0 && data.batchRows.size() >= meta.getCommitSize()) { + commitBatch(); + } + } + + private boolean shouldDeleteDocument(String documentId) { + return meta.isDeleteDocumentBeforeUpsert() + && data.deleteStatement != null + && !Utils.isEmpty(documentId) + && !data.deletedDocuments.contains(documentId); + } + + private void deleteDocument(String documentId) throws HopException { + try { + data.deleteStatement.setString(1, documentId); + data.deleteStatement.executeUpdate(); + if (data.deletedDocuments.size() >= MAX_DELETED_DOCUMENT_CACHE) { + Iterator<String> oldest = data.deletedDocuments.iterator(); + oldest.next(); + oldest.remove(); + } + data.deletedDocuments.add(documentId); + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.DeletingDocument", documentId), e); + } + } + + private void bindInsertRow( + Object[] row, + String id, + String documentId, + String chunkIndex, + String content, + Object embedding) + throws Exception { + String vectorLiteral = EmbeddingJsonParser.toPgVectorLiteral(embedding); + for (PgVectorUpsertData.MappingBinding binding : data.mappingBindings) { + Object value = + switch (binding.column.name()) { + case "id" -> id; + case "document_id" -> documentId; + case "chunk_index" -> parseChunkIndex(chunkIndex); + case "content" -> content; + case "embedding" -> vectorLiteral; + default -> getFieldAsString(row, binding.streamFieldIndex); + }; + if (binding.column.type() == PgVectorColumnType.INTEGER) { + data.insertStatement.setInt(binding.statementIndex, (Integer) value); + } else { + data.insertStatement.setString( + binding.statementIndex, value == null ? null : value.toString()); + } + } + } + + private void openDatabase() throws HopException { + try { + data.database = + PgVectorDatabase.connect(this, this, getMetadataProvider(), meta.getConnection()); + } catch (HopException e) { + throw e; + } + + try { + data.database.setCommit(meta.getCommitSize()); Review Comment: **[bug]** `setCommit(commitSize)` with 0 turns JDBC autocommit on, but `addRowToBatch` only flushes when `commitSize > 0`, so rows accumulate until end-of-stream. With "Delete document before upsert", `deleteDocument()` then autocommits immediately while the replacement inserts stay uncommitted in the batch — the failure boundary the comments say they are avoiding. The tooltip "Use 0 for autocommit" does not match this behavior (`TableOutput` treats 0 as one transaction for the whole run). **Suggestion:** Pick one meaning and implement it. Either treat 0 like Table Output (`Integer.MAX_VALUE`, autocommit off, commit at EOF) or flush/commit each row when autocommit is requested. Deletes and the inserts they make room for must stay in the same transaction. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/util/VectorDistanceMetric.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.hop.pgvector.util; + +/** + * pgvector distance operators, together with the expression that turns each one into a comparable + * similarity score. + * + * <p>Every metric yields a <em>higher is better</em> score, so the minimum-score filter means the + * same thing whichever metric is chosen. Without this normalisation cosine returns a similarity, L2 + * returns a raw distance where lower is better, and pgvector's {@code <#>} returns a negated inner + * product — three different scales behind one output field. + */ +public enum VectorDistanceMetric { + + /** Cosine similarity in [0, 2]: {@code 1 - cosine_distance}. */ Review Comment: **[nit]** Comment says cosine similarity is in `[0, 2]`. pgvector cosine distance is `[0, 2]`; `1 - distance` is cosine similarity in `[-1, 1]`. Default `minScore` 0.0 therefore drops obtuse matches, which may be intended but does not match the comment. **Suggestion:** Document the actual range `[-1, 1]` (and briefly the L2 `(0, 1]` / unbounded inner-product scales) in the enum and user docs. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/upsert/PgVectorUpsertMeta.java: ########## @@ -0,0 +1,249 @@ +/* + * 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.hop.pgvector.transforms.upsert; + +import java.util.ArrayList; +import java.util.List; +import lombok.Getter; +import lombok.Setter; +import org.apache.hop.core.CheckResult; +import org.apache.hop.core.ICheckResult; +import org.apache.hop.core.annotations.Transform; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.metadata.api.HopMetadataProperty; +import org.apache.hop.metadata.api.HopMetadataPropertyType; +import org.apache.hop.metadata.api.IHopMetadataProvider; +import org.apache.hop.pgvector.util.PgVectorColumnMapping; +import org.apache.hop.pgvector.util.PgVectorSchemaBuilder; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransformMeta; +import org.apache.hop.pipeline.transform.TransformMeta; + +@Getter +@Setter +@Transform( + id = "PgVectorUpsert", + image = "vector-store.svg", + name = "i18n::PgVectorUpsert.Name", + description = "i18n::PgVectorUpsert.Description", + categoryDescription = "i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Output", + keywords = "pgvector,vector,embedding,ai,rag,postgres,postgresql,upsert", + documentationUrl = "/pipeline/transforms/pgvector-upsert.html") +public class PgVectorUpsertMeta extends BaseTransformMeta<PgVectorUpsert, PgVectorUpsertData> { + + private static final Class<?> PKG = PgVectorUpsertMeta.class; + + @HopMetadataProperty( + key = "connection", + hopMetadataPropertyType = HopMetadataPropertyType.RDBMS_CONNECTION) + private String connection; + + @HopMetadataProperty(key = "schemaName") + private String schemaName = "public"; + + @HopMetadataProperty(key = "tableName") + private String tableName = "hop_rag_chunks"; + + @HopMetadataProperty(key = "idField") + private String idField; + + @HopMetadataProperty(key = "documentIdField") + private String documentIdField = "document_id"; + + @HopMetadataProperty(key = "chunkIndexField") + private String chunkIndexField = "chunk_index"; + + @HopMetadataProperty(key = "contentField") + private String contentField = "chunk_text"; + + @HopMetadataProperty(key = "embeddingField") + private String embeddingField = "embedding"; + + @HopMetadataProperty(key = "embeddingDimensions") + private int embeddingDimensions = 768; + + @HopMetadataProperty(key = "createTableIfMissing") + private boolean createTableIfMissing = true; + + @HopMetadataProperty(key = "createHnswIndex") + private boolean createHnswIndex = true; + + @HopMetadataProperty(key = "deleteDocumentBeforeUpsert") + private boolean deleteDocumentBeforeUpsert = false; + + @HopMetadataProperty(key = "commitSize") + private int commitSize = 100; + + @HopMetadataProperty(key = "indexMetric") + private VectorDistanceMetric indexMetric = VectorDistanceMetric.COSINE; + + @HopMetadataProperty( + key = "mapping", + injectionGroupKey = "MAPPINGS", + injectionKeyDescription = "PgVectorUpsertMeta.Injection.MAPPING") + private List<PgVectorColumnMapping> columnMappings = new ArrayList<>(); + + public PgVectorUpsertMeta() { + columnMappings = new ArrayList<>(); + } + + @Override + public Object clone() { + PgVectorUpsertMeta copy = (PgVectorUpsertMeta) super.clone(); + copy.columnMappings = new ArrayList<>(); + if (columnMappings != null) { + for (PgVectorColumnMapping mapping : columnMappings) { + copy.columnMappings.add( + new PgVectorColumnMapping(mapping.getColumnName(), mapping.getStreamField())); + } + } + return copy; + } + + @Override + public void setDefault() { + connection = ""; + schemaName = "public"; + tableName = "hop_rag_chunks"; + documentIdField = "document_id"; + chunkIndexField = "chunk_index"; + contentField = "chunk_text"; + embeddingField = "embedding"; + embeddingDimensions = 768; + createTableIfMissing = true; + createHnswIndex = true; + deleteDocumentBeforeUpsert = false; + commitSize = 100; + indexMetric = VectorDistanceMetric.COSINE; + columnMappings = new ArrayList<>(); + } + + @Override + public void getFields( + IRowMeta row, + String origin, + IRowMeta[] info, + TransformMeta nextTransform, + IVariables variables, + IHopMetadataProvider metadataProvider) { + // Pass-through transform: output row layout matches input. + } + + @Override + public void check( + List<ICheckResult> remarks, + PipelineMeta pipelineMeta, + TransformMeta transformMeta, + IRowMeta prev, + String[] input, + String[] output, + IRowMeta info, + IVariables variables, + IHopMetadataProvider metadataProvider) { + + if (Utils.isEmpty(connection)) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.ConnectionRequired"); + } + if (Utils.isEmpty(tableName)) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.TableRequired"); + } + + if (Utils.isEmpty(contentField)) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.ContentFieldRequired"); + } else if (prev != null && prev.indexOfValue(contentField) < 0) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.ContentFieldNotFound", contentField); + } + + if (Utils.isEmpty(embeddingField)) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.EmbeddingFieldRequired"); + } else if (prev != null && prev.indexOfValue(embeddingField) < 0) { + error( + remarks, + transformMeta, + "PgVectorUpsert.Validation.EmbeddingFieldNotFound", + embeddingField); + } + + if (Utils.isEmpty(idField)) { + if (Utils.isEmpty(documentIdField) || Utils.isEmpty(chunkIndexField)) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.NoIdSource"); + } + } else if (prev != null && prev.indexOfValue(idField) < 0) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.IdFieldNotFound", idField); + } + + if (embeddingDimensions <= 0) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.DimensionsPositive"); + } + if (commitSize < 0) { + error(remarks, transformMeta, "PgVectorUpsert.Validation.CommitSizeNonNegative"); + } + + if (columnMappings != null) { + for (PgVectorColumnMapping mapping : columnMappings) { + if (mapping == null || Utils.isEmpty(mapping.getColumnName())) { + continue; + } + if (PgVectorSchemaBuilder.isReservedColumn(mapping.getColumnName())) { + error( + remarks, + transformMeta, + "PgVectorUpsert.Validation.MappingColumnReserved", + mapping.getColumnName()); + } + if (!Utils.isEmpty(mapping.getStreamField()) + && prev != null + && prev.indexOfValue(mapping.getStreamField()) < 0) { + error( + remarks, + transformMeta, + "PgVectorUpsert.Validation.MappingStreamFieldNotFound", + mapping.getStreamField()); + } + } + } + + if (createTableIfMissing) { + warning(remarks, transformMeta, "PgVectorUpsert.Validation.CreateExtensionPrivileges"); + if (transformMeta != null && transformMeta.getCopies(variables) > 1) { Review Comment: **[suggestion]** `check()` errors on copies > 1 only when DDL is enabled. "Delete document before upsert" with multiple copies can let copy 2 delete rows copy 1 just inserted for the same document ID (`deletedDocuments` is per-copy). **Suggestion:** Also reject or warn on copies > 1 when `deleteDocumentBeforeUpsert` is enabled. ########## docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/pgvector-search.adoc: ########## @@ -0,0 +1,81 @@ +//// +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. +//// +:documentationPath: /pipeline/transforms/ +:language: en_US +:description: Run a top-k similarity search against a PostgreSQL table that uses the pgvector extension. + += image:transforms/icons/vector-store.svg[pgvector search transform Icon, role="image-doc-icon"] pgvector search + +[%noheader,cols="3a,1a", role="table-no-borders" ] +|=== +| +== Description + +The pgvector search transform looks up the nearest stored chunks for a query embedding in a PostgreSQL table that uses the https://github.com/pgvector/pgvector[pgvector] extension. It is the read half of a retrieval pipeline: embed the question, search, then feed the matches to a language model. + +Each incoming row carries one query embedding, and the transform emits one output row per match, with the original row's fields repeated on each. + +It uses a regular xref:database/databases/postgresql.adoc[PostgreSQL] relational database connection. + + +| +== Supported Engines +[%noheader,cols="2,1a",frame=none, role="table-supported-engines"] +!=== +!Hop Engine! image:check_mark.svg[Supported, 24] +!Single Threaded! image:check_mark.svg[Supported, 24] +!Native Spark! image:question_mark.svg[Maybe Supported, 24] +!Beam Spark! image:question_mark.svg[Maybe Supported, 24] +!Beam Flink! image:question_mark.svg[Maybe Supported, 24] +!Beam Dataflow! image:question_mark.svg[Maybe Supported, 24] +!=== +|=== + +== Embedding fields + +As in pgvector upsert, the query embedding field may be either a field of the **Vector** value type or a String field holding a JSON array. + +A row with an empty query embedding is treated as a row with no matches rather than as an error, so a stream that happens to carry a few unembedded rows does not stop the pipeline. + +== Options + +[options="header"] +|=== +|Option |Description + +|Transform name|Name of the transform, unique within the pipeline. +|Connection|The PostgreSQL database connection to search. +|Schema|Database schema holding the table. +|Table|Table that stores the chunks and embeddings. +|Query embedding field|Field holding the query embedding, either a Vector field or a String holding a JSON array. +|Top K|Maximum number of matches to return per input row. +|Minimum score|Only return matches whose similarity score is at least this value. +|Distance metric|Similarity metric used by pgvector. Use the same metric the index was built for. +|Result ID field|Output field for the matched row ID. +|Result document ID field|Output field for the matched document ID. +|Result chunk index field|Output field for the matched chunk index. +|Result content field|Output field for the matched chunk text. +|Result score field|Output field for the similarity score. Review Comment: **[suggestion]** The search dialog exposes "Do not pass the row if the search finds nothing", but the Options table does not document it. Empty-embedding behavior is described; the eat-row flag is not. **Suggestion:** Add the option to the Options table (and note that empty embeddings follow the same path). ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/upsert/PgVectorUpsert.java: ########## @@ -0,0 +1,429 @@ +/* + * 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.hop.pgvector.transforms.upsert; + +import java.sql.BatchUpdateException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.util.Utils; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.pgvector.util.EmbeddingJsonParser; +import org.apache.hop.pgvector.util.PgVectorColumnMapping; +import org.apache.hop.pgvector.util.PgVectorColumnType; +import org.apache.hop.pgvector.util.PgVectorDatabase; +import org.apache.hop.pgvector.util.PgVectorSchemaBuilder; +import org.apache.hop.pgvector.util.PgVectorSqlBuilder; +import org.apache.hop.pgvector.util.PgVectorTableColumn; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.Pipeline; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransform; +import org.apache.hop.pipeline.transform.TransformMeta; + +/** Upserts chunk rows with embeddings into a PostgreSQL pgvector table. */ +public class PgVectorUpsert extends BaseTransform<PgVectorUpsertMeta, PgVectorUpsertData> { + + private static final Class<?> PKG = PgVectorUpsertMeta.class; + + /** + * Upper bound on the per-document delete cache. Without a cap this set grows for the lifetime of + * the run, which matters when a corpus has millions of distinct documents. + */ + private static final int MAX_DELETED_DOCUMENT_CACHE = 100_000; + + public PgVectorUpsert( + TransformMeta transformMeta, + PgVectorUpsertMeta meta, + PgVectorUpsertData data, + int copyNr, + PipelineMeta pipelineMeta, + Pipeline pipeline) { + super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline); + } + + @Override + public boolean init() { + if (Utils.isEmpty(meta.getConnection())) { + logError(BaseMessages.getString(PKG, "PgVectorUpsert.Validation.ConnectionRequired")); + return false; + } + return super.init(); + } + + @Override + public boolean processRow() throws HopException { + Object[] row = getRow(); + if (row == null) { + commitBatch(); + setOutputDone(); + return false; + } + + if (first) { + first = false; + data.inputRowMeta = getInputRowMeta(); + openDatabase(); + resolveFieldIndices(); + } + + try { + addRowToBatch(row); + } catch (HopException e) { + if (getTransformMeta().isDoingErrorHandling()) { + putError(data.inputRowMeta, row, 1, e.getMessage(), null, "PGVECTORUPSERT001"); + } else { + throw e; + } + } + + return true; + } + + private void addRowToBatch(Object[] row) throws HopException { + String documentId = getFieldAsString(row, data.documentIdFieldIndex); + String content = getFieldAsString(row, data.contentFieldIndex); + // Taken as the raw value: a field of the Vector value type is already a float[], and + // rendering it to text here only to parse it back would be wasted work on every row. + Object embedding = row[data.embeddingFieldIndex]; + + if (Utils.isEmpty(content) || EmbeddingJsonParser.isEmpty(embedding)) { + // Nothing to store for this row, but it still belongs in the output stream. + putRow(data.inputRowMeta, row); + return; + } + + // The delete has to be flushed together with the inserts it makes room for. Committing the + // pending batch first keeps "delete document, then re-insert its chunks" from being split + // across a failure boundary, which would otherwise leave the document deleted and not + // replaced. + if (shouldDeleteDocument(documentId)) { + commitBatch(); + deleteDocument(documentId); + } + + String id = resolveId(row); + String chunkIndex = getFieldAsString(row, data.chunkIndexFieldIndex); + + try { + bindInsertRow(row, id, documentId, chunkIndex, content, embedding); + data.insertStatement.addBatch(); + data.batchRows.add(row); + } catch (HopException e) { + throw e; + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.BindingRow", String.valueOf(id)), e); + } + + if (meta.getCommitSize() > 0 && data.batchRows.size() >= meta.getCommitSize()) { + commitBatch(); + } + } + + private boolean shouldDeleteDocument(String documentId) { + return meta.isDeleteDocumentBeforeUpsert() + && data.deleteStatement != null + && !Utils.isEmpty(documentId) + && !data.deletedDocuments.contains(documentId); + } + + private void deleteDocument(String documentId) throws HopException { + try { + data.deleteStatement.setString(1, documentId); + data.deleteStatement.executeUpdate(); + if (data.deletedDocuments.size() >= MAX_DELETED_DOCUMENT_CACHE) { + Iterator<String> oldest = data.deletedDocuments.iterator(); + oldest.next(); + oldest.remove(); + } + data.deletedDocuments.add(documentId); + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.DeletingDocument", documentId), e); + } + } + + private void bindInsertRow( + Object[] row, + String id, + String documentId, + String chunkIndex, + String content, + Object embedding) + throws Exception { + String vectorLiteral = EmbeddingJsonParser.toPgVectorLiteral(embedding); + for (PgVectorUpsertData.MappingBinding binding : data.mappingBindings) { + Object value = + switch (binding.column.name()) { + case "id" -> id; + case "document_id" -> documentId; + case "chunk_index" -> parseChunkIndex(chunkIndex); + case "content" -> content; + case "embedding" -> vectorLiteral; + default -> getFieldAsString(row, binding.streamFieldIndex); + }; + if (binding.column.type() == PgVectorColumnType.INTEGER) { + data.insertStatement.setInt(binding.statementIndex, (Integer) value); + } else { + data.insertStatement.setString( + binding.statementIndex, value == null ? null : value.toString()); + } + } + } + + private void openDatabase() throws HopException { + try { + data.database = + PgVectorDatabase.connect(this, this, getMetadataProvider(), meta.getConnection()); + } catch (HopException e) { + throw e; + } + + try { + data.database.setCommit(meta.getCommitSize()); + VectorDistanceMetric indexMetric = + meta.getIndexMetric() != null ? meta.getIndexMetric() : VectorDistanceMetric.COSINE; + String schemaName = resolve(meta.getSchemaName()); + String tableName = resolve(meta.getTableName()); + if (meta.isCreateTableIfMissing()) { Review Comment: **[suggestion]** HNSW index creation and `ensureMappedColumns` live inside `PgVectorDatabase.ensureSchema()`, which is only called when "Create table if missing" is true. The HNSW checkbox and column-mapping ALTER therefore do nothing on an existing table unless that flag is also on, which is not obvious from the independent checkboxes. **Suggestion:** Run index/column ensure independently of table creation, or disable/hide those controls when create-table is off. ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/upsert/PgVectorUpsert.java: ########## @@ -0,0 +1,429 @@ +/* + * 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.hop.pgvector.transforms.upsert; + +import java.sql.BatchUpdateException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import org.apache.hop.core.exception.HopException; +import org.apache.hop.core.util.Utils; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.pgvector.util.EmbeddingJsonParser; +import org.apache.hop.pgvector.util.PgVectorColumnMapping; +import org.apache.hop.pgvector.util.PgVectorColumnType; +import org.apache.hop.pgvector.util.PgVectorDatabase; +import org.apache.hop.pgvector.util.PgVectorSchemaBuilder; +import org.apache.hop.pgvector.util.PgVectorSqlBuilder; +import org.apache.hop.pgvector.util.PgVectorTableColumn; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.Pipeline; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransform; +import org.apache.hop.pipeline.transform.TransformMeta; + +/** Upserts chunk rows with embeddings into a PostgreSQL pgvector table. */ +public class PgVectorUpsert extends BaseTransform<PgVectorUpsertMeta, PgVectorUpsertData> { + + private static final Class<?> PKG = PgVectorUpsertMeta.class; + + /** + * Upper bound on the per-document delete cache. Without a cap this set grows for the lifetime of + * the run, which matters when a corpus has millions of distinct documents. + */ + private static final int MAX_DELETED_DOCUMENT_CACHE = 100_000; + + public PgVectorUpsert( + TransformMeta transformMeta, + PgVectorUpsertMeta meta, + PgVectorUpsertData data, + int copyNr, + PipelineMeta pipelineMeta, + Pipeline pipeline) { + super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline); + } + + @Override + public boolean init() { + if (Utils.isEmpty(meta.getConnection())) { + logError(BaseMessages.getString(PKG, "PgVectorUpsert.Validation.ConnectionRequired")); + return false; + } + return super.init(); + } + + @Override + public boolean processRow() throws HopException { + Object[] row = getRow(); + if (row == null) { + commitBatch(); + setOutputDone(); + return false; + } + + if (first) { + first = false; + data.inputRowMeta = getInputRowMeta(); + openDatabase(); + resolveFieldIndices(); + } + + try { + addRowToBatch(row); + } catch (HopException e) { + if (getTransformMeta().isDoingErrorHandling()) { + putError(data.inputRowMeta, row, 1, e.getMessage(), null, "PGVECTORUPSERT001"); + } else { + throw e; + } + } + + return true; + } + + private void addRowToBatch(Object[] row) throws HopException { + String documentId = getFieldAsString(row, data.documentIdFieldIndex); + String content = getFieldAsString(row, data.contentFieldIndex); + // Taken as the raw value: a field of the Vector value type is already a float[], and + // rendering it to text here only to parse it back would be wasted work on every row. + Object embedding = row[data.embeddingFieldIndex]; + + if (Utils.isEmpty(content) || EmbeddingJsonParser.isEmpty(embedding)) { + // Nothing to store for this row, but it still belongs in the output stream. + putRow(data.inputRowMeta, row); + return; + } + + // The delete has to be flushed together with the inserts it makes room for. Committing the + // pending batch first keeps "delete document, then re-insert its chunks" from being split + // across a failure boundary, which would otherwise leave the document deleted and not + // replaced. + if (shouldDeleteDocument(documentId)) { + commitBatch(); + deleteDocument(documentId); + } + + String id = resolveId(row); + String chunkIndex = getFieldAsString(row, data.chunkIndexFieldIndex); + + try { + bindInsertRow(row, id, documentId, chunkIndex, content, embedding); + data.insertStatement.addBatch(); + data.batchRows.add(row); + } catch (HopException e) { + throw e; + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.BindingRow", String.valueOf(id)), e); + } + + if (meta.getCommitSize() > 0 && data.batchRows.size() >= meta.getCommitSize()) { + commitBatch(); + } + } + + private boolean shouldDeleteDocument(String documentId) { + return meta.isDeleteDocumentBeforeUpsert() + && data.deleteStatement != null + && !Utils.isEmpty(documentId) + && !data.deletedDocuments.contains(documentId); + } + + private void deleteDocument(String documentId) throws HopException { + try { + data.deleteStatement.setString(1, documentId); + data.deleteStatement.executeUpdate(); + if (data.deletedDocuments.size() >= MAX_DELETED_DOCUMENT_CACHE) { + Iterator<String> oldest = data.deletedDocuments.iterator(); + oldest.next(); + oldest.remove(); + } + data.deletedDocuments.add(documentId); + } catch (Exception e) { + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.DeletingDocument", documentId), e); + } + } + + private void bindInsertRow( + Object[] row, + String id, + String documentId, + String chunkIndex, + String content, + Object embedding) + throws Exception { + String vectorLiteral = EmbeddingJsonParser.toPgVectorLiteral(embedding); + for (PgVectorUpsertData.MappingBinding binding : data.mappingBindings) { + Object value = + switch (binding.column.name()) { + case "id" -> id; + case "document_id" -> documentId; + case "chunk_index" -> parseChunkIndex(chunkIndex); + case "content" -> content; + case "embedding" -> vectorLiteral; + default -> getFieldAsString(row, binding.streamFieldIndex); + }; + if (binding.column.type() == PgVectorColumnType.INTEGER) { + data.insertStatement.setInt(binding.statementIndex, (Integer) value); + } else { + data.insertStatement.setString( + binding.statementIndex, value == null ? null : value.toString()); + } + } + } + + private void openDatabase() throws HopException { + try { + data.database = + PgVectorDatabase.connect(this, this, getMetadataProvider(), meta.getConnection()); + } catch (HopException e) { + throw e; + } + + try { + data.database.setCommit(meta.getCommitSize()); + VectorDistanceMetric indexMetric = + meta.getIndexMetric() != null ? meta.getIndexMetric() : VectorDistanceMetric.COSINE; + String schemaName = resolve(meta.getSchemaName()); + String tableName = resolve(meta.getTableName()); + if (meta.isCreateTableIfMissing()) { + PgVectorDatabase.ensureSchema(data.database, meta, schemaName, tableName, indexMetric); + } + String qualifiedTable = PgVectorSqlBuilder.qualifiedTable(schemaName, tableName); + data.tableColumns = PgVectorSchemaBuilder.tableColumns(meta); + data.insertStatement = + data.database + .getConnection() + .prepareStatement(PgVectorSqlBuilder.upsertSql(qualifiedTable, data.tableColumns)); + if (meta.isDeleteDocumentBeforeUpsert()) { + data.deleteStatement = + data.database + .getConnection() + .prepareStatement(PgVectorSqlBuilder.deleteByDocumentIdSql(qualifiedTable)); + } + } catch (Exception e) { + // Do not leak the connection when the statements cannot be prepared. + closeDatabase(); + throw new HopException(BaseMessages.getString(PKG, "PgVectorUpsert.Error.Initializing"), e); + } + } + + private void resolveFieldIndices() throws HopException { + data.contentFieldIndex = requireFieldIndex(meta.getContentField()); + data.embeddingFieldIndex = requireFieldIndex(meta.getEmbeddingField()); + data.documentIdFieldIndex = optionalFieldIndex(meta.getDocumentIdField()); + data.chunkIndexFieldIndex = optionalFieldIndex(meta.getChunkIndexField()); + if (!Utils.isEmpty(meta.getIdField())) { + data.idFieldIndex = requireFieldIndex(meta.getIdField()); + } + + data.mappingBindings = new ArrayList<>(); + int parameterIndex = 1; + for (PgVectorTableColumn column : data.tableColumns) { + int streamIndex = resolveStreamIndex(column.name()); + data.mappingBindings.add( + new PgVectorUpsertData.MappingBinding(streamIndex, parameterIndex++, column)); + } + } + + private int resolveStreamIndex(String columnName) throws HopException { + return switch (columnName) { + case "id" -> data.idFieldIndex; + case "document_id" -> data.documentIdFieldIndex; + case "chunk_index" -> data.chunkIndexFieldIndex; + case "content" -> data.contentFieldIndex; + case "embedding" -> data.embeddingFieldIndex; + default -> requireMappedFieldIndex(columnName); + }; + } + + private int requireMappedFieldIndex(String columnName) throws HopException { + if (meta.getColumnMappings() != null) { + for (PgVectorColumnMapping mapping : meta.getColumnMappings()) { + if (mapping != null + && columnName.equals( + PgVectorSchemaBuilder.normalizeColumnName(mapping.getColumnName()))) { + return requireFieldIndex(mapping.getStreamField()); + } + } + } + throw new HopException( + BaseMessages.getString(PKG, "PgVectorUpsert.Error.NoMappingForColumn", columnName)); + } + + private int requireFieldIndex(String fieldName) throws HopException { + int index = data.inputRowMeta.indexOfValue(fieldName); + if (index < 0) { + throw new HopException( + BaseMessages.getString( + PKG, "PgVectorUpsert.Error.FieldNotFound", String.valueOf(fieldName))); + } + return index; + } + + private int optionalFieldIndex(String fieldName) { + if (Utils.isEmpty(fieldName)) { + return -1; + } + return data.inputRowMeta.indexOfValue(fieldName); + } + + private String resolveId(Object[] row) throws HopException { + if (data.idFieldIndex >= 0) { + return getFieldAsString(row, data.idFieldIndex); + } + String documentId = getFieldAsString(row, data.documentIdFieldIndex); + String chunkIndex = getFieldAsString(row, data.chunkIndexFieldIndex); + if (Utils.isEmpty(documentId) && Utils.isEmpty(chunkIndex)) { + throw new HopException(BaseMessages.getString(PKG, "PgVectorUpsert.Error.NoIdAvailable")); + } + return documentId + "_" + chunkIndex; Review Comment: **[bug]** When no explicit ID field is set, a missing/null chunk index does not fail the "both empty" check if document ID is present. Java concatenates `"doc" + "_" + null` to `"doc_null"`, while `parseChunkIndex(null)` stores `0` in `chunk_index`. Rows with a null chunk index therefore get a primary key that does not match the table columns. **Suggestion:** Treat a null chunk index like 0 (or reject the row). Build the synthetic id from the same values that are bound (`documentId + "_" + parsedChunkIndex`). ########## plugins/tech/pgvector/src/main/java/org/apache/hop/pgvector/transforms/search/PgVectorSearchMeta.java: ########## @@ -0,0 +1,244 @@ +/* + * 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.hop.pgvector.transforms.search; + +import java.util.ArrayList; +import java.util.List; +import lombok.Getter; +import lombok.Setter; +import org.apache.hop.core.CheckResult; +import org.apache.hop.core.ICheckResult; +import org.apache.hop.core.annotations.Transform; +import org.apache.hop.core.exception.HopPluginException; +import org.apache.hop.core.exception.HopTransformException; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.IValueMeta; +import org.apache.hop.core.row.value.ValueMetaInteger; +import org.apache.hop.core.row.value.ValueMetaNumber; +import org.apache.hop.core.row.value.ValueMetaString; +import org.apache.hop.core.util.Utils; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.metadata.api.HopMetadataProperty; +import org.apache.hop.metadata.api.HopMetadataPropertyType; +import org.apache.hop.metadata.api.IHopMetadataProvider; +import org.apache.hop.pgvector.util.PgVectorSearchFilter; +import org.apache.hop.pgvector.util.VectorDistanceMetric; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.transform.BaseTransformMeta; +import org.apache.hop.pipeline.transform.TransformMeta; + +@Getter +@Setter +@Transform( + id = "PgVectorSearch", + image = "vector-store.svg", + name = "i18n::PgVectorSearch.Name", + description = "i18n::PgVectorSearch.Description", + categoryDescription = "i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Lookup", + keywords = "pgvector,vector,embedding,ai,rag,postgres,postgresql,similarity,search", + documentationUrl = "/pipeline/transforms/pgvector-search.html") +public class PgVectorSearchMeta extends BaseTransformMeta<PgVectorSearch, PgVectorSearchData> { + + private static final Class<?> PKG = PgVectorSearchMeta.class; + + @HopMetadataProperty( + key = "connection", + hopMetadataPropertyType = HopMetadataPropertyType.RDBMS_CONNECTION) + private String connection; + + @HopMetadataProperty(key = "schemaName") + private String schemaName = "public"; + + @HopMetadataProperty(key = "tableName") + private String tableName = "hop_rag_chunks"; + + @HopMetadataProperty(key = "embeddingField") + private String embeddingField = "embedding"; + + @HopMetadataProperty(key = "topK") + private int topK = 5; + + @HopMetadataProperty(key = "distanceMetric") + private VectorDistanceMetric distanceMetric = VectorDistanceMetric.COSINE; + + @HopMetadataProperty(key = "minScore") + private double minScore = 0.0; + + /** + * Have the search eat the incoming row when the query returns nothing. Mirrors the option on + * Hop's Database Lookup: off by default, so a row that finds no match still reaches the output + * with empty match fields rather than disappearing. + */ + @HopMetadataProperty( + key = "eat_row_on_no_match", + injectionKey = "EAT_ROW_ON_NO_MATCH", + injectionKeyDescription = "PgVectorSearchMeta.Injection.EAT_ROW_ON_NO_MATCH") + private boolean eatingRowOnNoMatch; + + @HopMetadataProperty(key = "resultIdField") + private String resultIdField = "match_id"; + + @HopMetadataProperty(key = "resultDocumentIdField") + private String resultDocumentIdField = "match_document_id"; + + @HopMetadataProperty(key = "resultChunkIndexField") + private String resultChunkIndexField = "match_chunk_index"; + + @HopMetadataProperty(key = "resultContentField") + private String resultContentField = "match_content"; + + @HopMetadataProperty(key = "resultScoreField") + private String resultScoreField = "match_score"; + + @HopMetadataProperty( + key = "filter", + injectionGroupKey = "FILTERS", + injectionKeyDescription = "PgVectorSearchMeta.Injection.FILTER") Review Comment: **[suggestion]** `injectionKeyDescription` values `PgVectorSearchMeta.Injection.FILTER` and `PgVectorUpsertMeta.Injection.MAPPING` are not defined in the English message bundles, so metadata injection UI will show the raw keys. **Suggestion:** Add the missing keys next to the existing `EAT_ROW_ON_NO_MATCH` injection string. -- 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]
