From f106b418c7812a513126f7dcf0bbe9d25175b830 Mon Sep 17 00:00:00 2001 From: jian he Date: Wed, 30 Sep 2026 19:03:02 +0800 Subject: [PATCH v15 2/2] COPY FROM (on_error table, error_table errtbl) The COPY FROM syntax is: COPY tbl FROM ... WITH (ON_ERROR table, ERROR_TABLE 'errsave'); ERROR_TABLE can only be used when ON_ERROR is set to "table", and ON_ERROR table requires ERROR_TABLE; both are COPY FROM only. The errsave table must be a typed table, based on internal composite type pg_catalog.copy_error_saving, it's opened at the start of the COPY: the current user must have INSERT privilege on it, and a preliminary lock check ensures that inserting error rows later cannot block or deadlock in the middle of the COPY. When an error occurs, we record the error metadata and insert it into the errsave table, then proceed to the next row. Although the errsave table may not capture information for every invalid column in each row, but it retains the raw_field_value, which can be used for further investigation. The build-in type: pg_catalog.copy_error_saving definition is CREATE TYPE copy_error_saving AS ( starttime timestamptz, userid oid, copy_tbl oid, filename text COLLATE "C", lineno bigint, line text COLLATE "C", colname text COLLATE "C", raw_field_value text COLLATE "C", err_message text COLLATE "C", err_detail text COLLATE "C", errorcode text COLLATE "C"); It's declared in src/backend/catalog/system_functions.sql. reference: https://postgr.es/m/CACJufxHi53OpGYPAe6SdCb4m=-+H8L+7LDbUWvTiJp=V4YYEqA@mail.gmail.com reference: https://postgr.es/m/752672.1699474336%40sss.pgh.pa.us discussion: https://postgr.es/m/CACJufxH_OJpVra=0c4ow8fbxHj7heMcVaTNEPa5vAurSeNA-6Q@mail.gmail.com commitfest entry: https://commitfest.postgresql.org/patch/4817 --- doc/src/sgml/datatype.sgml | 115 ++++++++ doc/src/sgml/monitoring.sgml | 2 +- doc/src/sgml/ref/copy.sgml | 51 +++- src/backend/catalog/system_functions.sql | 14 + src/backend/commands/copy.c | 29 ++ src/backend/commands/copyfrom.c | 333 ++++++++++++++++++++++- src/backend/commands/copyfromparse.c | 142 ++++++++++ src/backend/parser/gram.y | 1 + src/bin/psql/tab-complete.in.c | 23 +- src/include/commands/copy.h | 2 + src/include/commands/copyfrom_internal.h | 9 + src/test/regress/expected/copy.out | 6 + src/test/regress/expected/copy2.out | 249 +++++++++++++++++ src/test/regress/sql/copy.sql | 9 + src/test/regress/sql/copy2.sql | 207 ++++++++++++++ 15 files changed, 1178 insertions(+), 14 deletions(-) diff --git a/doc/src/sgml/datatype.sgml b/doc/src/sgml/datatype.sgml index 89985ab7b16..022a7734d05 100644 --- a/doc/src/sgml/datatype.sgml +++ b/doc/src/sgml/datatype.sgml @@ -5121,6 +5121,121 @@ WHERE ... + + <type>copy_error_saving</type> Type + + copy_error_saving + + + + The built-in composite type copy_error_saving is used by the + COPY FROM command. + It contains the following fields, which are used to store information when COPY FROM + encounters an error converting a column's input value to its data type. + + + + + + + + Column name + Data type + Description + + + + + + starttime + timestamp with time zone + Start time of the COPY FROM command that saved + the row. All rows saved by one command share the same value, which can + be used to tell the rows of different commands apart. + + + + + userid + oid + The COPY FROM operation user + (references pg_authid.oid). + Since there is no dependency on pg_authid, + dropping the role does not affect the saved row; the value then no longer + identifies any existing role. + + + + + copy_tbl + oid + The COPY FROM operation destination table + (references pg_class.oid). + Since there is no dependency on pg_class, + dropping the table does not affect the saved row; the value then no + longer identifies any existing table. + + + + + filename + text + The path name of the COPY FROM input + + + + lineno + bigint + Line number where the error occurred, counting from 1 + + + + line + text + Raw content of the error occurred line + + + + colname + text + Field where the error occurred + + + + raw_field_value + text + Raw content of the error occurred field + + + + err_message + text + The primary error message, + see ereport + + + + err_detail + text + The optionally detailed error message, + see ereport + + + + errorcode + text + The SQLSTATE error identifier code for the error condition, + see Error Codes + + + + + + + + + + <type>pg_lsn</type> Type diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 0d038de1a23..3dec5d40c1b 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -7095,7 +7095,7 @@ FROM pg_stat_get_backend_idset() AS backendid; Number of tuples skipped because they contain malformed data. This counter only advances when - ignore is specified to the ON_ERROR + ignore or table is specified to the ON_ERROR option. diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml index b23433b2c41..8bfb05485a8 100644 --- a/doc/src/sgml/ref/copy.sgml +++ b/doc/src/sgml/ref/copy.sgml @@ -46,6 +46,7 @@ COPY { table_name [ ( column_name [, ...] ) | * } ON_ERROR error_action REJECT_LIMIT maxerror + ERROR_TABLE error_saving_table ENCODING 'encoding_name' LOG_VERBOSITY verbosity @@ -450,11 +451,13 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) stop means fail the command, while ignore means discard the input row and continue with the next one, and set_null means replace the field containing the invalid - input value with a null value and continue to the next field. + input value with a null value and continue to the next field, + and table means save error information to + error_saving_table and continue with the next one. The default is stop. - The ignore and set_null + The ignore, set_null and table options are applicable only for COPY FROM when the FORMAT is text or csv. @@ -463,19 +466,34 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) set_null, a NOTICE message is emitted at the end of the COPY FROM command containing the count of rows that were ignored or changed, if at least one row was affected. + If ON_ERROR option is set to table, + a NOTICE message containing the row count inserted to + error_saving_table is + emitted at the end of the COPY FROM. When LOG_VERBOSITY option is set to verbose, - for ignore option, a NOTICE message + for ignore, a NOTICE message containing the line of the input file and the column name whose input conversion has failed is emitted for each discarded row; for set_null option, a NOTICE message containing the line of the input file and the column name where value was replaced with NULL for each input conversion - failure. + failure; + for table option, a NOTICE message + with the same information is emitted for each row saved to + error_saving_table. When it is set to silent, no message is emitted regarding input conversion failed rows. + + + Triggers on error_saving_table + will be fired. Therefore, the NOTICE message regarding + rows inserted into error_saving_table + may differ from what is actually being inserted. + + @@ -497,6 +515,27 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) + + ERROR_TABLE + + + Specifies the table in which the details of input conversion errors are + saved, one row for each input row whose conversion failed. + This option is allowed only in COPY FROM and + ON_ERROR is specified as TABLE. + The error_saving_table must + be a typed table based on the system composite type + copy_error_saving, + and the user running COPY FROM requires + INSERT privileges on it. During the error records + insertion, NOT NULL and CHECK + constraints are enforced, and both row-level and statement-level + triggers will be fired. + + + + + ENCODING @@ -522,8 +561,8 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) This is currently used in COPY FROM command when - ON_ERROR option is set to ignore - or set_null. + ON_ERROR option is set to ignore, + set_null or table. diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql index c3c0a6e84ed..55808286f61 100644 --- a/src/backend/catalog/system_functions.sql +++ b/src/backend/catalog/system_functions.sql @@ -366,3 +366,17 @@ CREATE OR REPLACE FUNCTION ts_debug(document text, BEGIN ATOMIC SELECT * FROM ts_debug(get_current_ts_config(), $1); END; + +CREATE TYPE copy_error_saving AS( + starttime timestamptz, + userid oid, + copy_tbl oid, + filename text COLLATE "C", + lineno bigint, + line text COLLATE "C", + colname text COLLATE "C", + raw_field_value text COLLATE "C", + err_message text COLLATE "C", + err_detail text COLLATE "C", + errorcode text COLLATE "C" +); diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 003b70852bb..74ffc20c0c9 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -494,6 +494,8 @@ defGetCopyOnErrorChoice(DefElem *def, ParseState *pstate, bool is_from) return COPY_ON_ERROR_IGNORE; if (pg_strcasecmp(sval, "set_null") == 0) return COPY_ON_ERROR_SET_NULL; + if (pg_strcasecmp(sval, "table") == 0) + return COPY_ON_ERROR_TABLE; ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), @@ -587,6 +589,7 @@ ProcessCopyOptions(ParseState *pstate, bool freeze_specified = false; bool header_specified = false; bool on_error_specified = false; + bool error_rel_specified = false; bool log_verbosity_specified = false; bool reject_limit_specified = false; bool force_array_specified = false; @@ -774,6 +777,19 @@ ProcessCopyOptions(ParseState *pstate, reject_limit_specified = true; opts_out->reject_limit = defGetCopyRejectLimitOption(defel); } + else if (strcmp(defel->defname, "error_table") == 0) + { + if (!is_from) + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("COPY %s cannot be used with %s", "ERROR_TABLE", "COPY TO"), + parser_errposition(pstate, defel->location)); + + if (error_rel_specified) + errorConflictingDefElem(defel, pstate); + error_rel_specified = true; + opts_out->error_table = defGetString(defel); + } else ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -782,6 +798,19 @@ ProcessCopyOptions(ParseState *pstate, parser_errposition(pstate, defel->location))); } + if (opts_out->on_error == COPY_ON_ERROR_TABLE) + { + if (opts_out->error_table == NULL) + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot set option %s to \"%s\" when \"%s\" is not specified", "ON_ERROR", "TABLE", "ERROR_TABLE"), + errhint("\"%s\" option is required", "ERROR_TABLE")); + } + else if (opts_out->error_table != NULL) + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("COPY %s can only be used when option %s is set to \"%s\"", "ERROR_TABLE", "ON_ERROR", "TABLE")); + /* * Check for incompatible options (must do these three before inserting * defaults) diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index 746b4fdf6c3..1937359dc67 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -29,6 +29,7 @@ #include "access/tupconvert.h" #include "access/xact.h" #include "catalog/namespace.h" +#include "catalog/pg_namespace.h" #include "commands/copyapi.h" #include "commands/copyfrom_internal.h" #include "commands/progress.h" @@ -42,16 +43,20 @@ #include "miscadmin.h" #include "nodes/miscnodes.h" #include "optimizer/optimizer.h" +#include "parser/parse_relation.h" #include "pgstat.h" #include "rewrite/rewriteHandler.h" #include "storage/fd.h" #include "tcop/tcopprot.h" +#include "utils/builtins.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/portal.h" #include "utils/rel.h" +#include "utils/regproc.h" #include "utils/snapmgr.h" #include "utils/typcache.h" +#include "utils/syscache.h" /* * No more than this many tuples per CopyMultiInsertBuffer @@ -120,6 +125,9 @@ static void CopyFromBinaryInFunc(CopyFromState cstate, Oid atttypid, FmgrInfo *finfo, Oid *typioparam); static void CopyFromBinaryStart(CopyFromState cstate, TupleDesc tupDesc); static void CopyFromBinaryEnd(CopyFromState cstate); +static void RangeVarCallbackForCopyErrorTable(const RangeVar *rv, Oid relid, Oid oldrelid, + void *arg); +static void CopyFromErrorTableInit(CopyFromState cstate); /* @@ -986,6 +994,11 @@ CopyFrom(CopyFromState cstate) cstate->qualexpr = ExecInitQual(castNode(List, cstate->whereClause), &mtstate->ps); + if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + CopyFromErrorTableInit(cstate); + else + cstate->mtcontext = NULL; + /* * It's generally more efficient to prepare a bunch of tuples for * insertion, and insert them in one @@ -1052,6 +1065,23 @@ CopyFrom(CopyFromState cstate) */ insertMethod = CIM_SINGLE; } + else if (cstate->error_rel != NULL && + cstate->error_rel->trigdesc != NULL && + (cstate->error_rel->trigdesc->trig_insert_before_row || + cstate->error_rel->trigdesc->trig_insert_after_row || + cstate->error_rel->trigdesc->trig_insert_before_statement || + cstate->error_rel->trigdesc->trig_insert_after_statement)) + { + /* + * Likewise, can't support multi-inserts when the error-saving table + * has any INSERT triggers (including those implementing foreign + * keys). Unlike the target table's AFTER triggers, they are fired + * immediately for each saved error row, while tuples already read for + * the target table may still be waiting in the multi-insert buffer; + * such triggers would then not see those tuples. + */ + insertMethod = CIM_SINGLE; + } else { /* @@ -1151,14 +1181,13 @@ CopyFrom(CopyFromState cstate) if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull)) break; - if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE && + if ((cstate->opts.on_error == COPY_ON_ERROR_IGNORE || + cstate->opts.on_error == COPY_ON_ERROR_TABLE) && cstate->escontext->error_occurred) { /* * Soft error occurred, skip this tuple and just make - * ErrorSaveContext ready for the next NextCopyFrom. Since we - * don't set details_wanted and error_data is not to be filled, - * just resetting error_occurred is enough. + * ErrorSaveContext ready for the next NextCopyFrom. */ cstate->escontext->error_occurred = false; @@ -1166,7 +1195,8 @@ CopyFrom(CopyFromState cstate) pgstat_progress_update_param(PROGRESS_COPY_TUPLES_SKIPPED, cstate->num_errors); - if (cstate->opts.reject_limit > 0 && + if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE && + cstate->opts.reject_limit > 0 && cstate->num_errors > cstate->opts.reject_limit) ereport(ERROR, (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), @@ -1480,6 +1510,13 @@ CopyFrom(CopyFromState cstate) "in %" PRIu64 " rows, columns were set to null due to data type incompatibility", cstate->num_errors, cstate->num_errors)); + else if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + ereport(NOTICE, + errmsg_plural("%" PRIu64 " row was saved to table \"%s\" due to data type incompatibility", + "%" PRIu64 " rows were saved to table \"%s\" due to data type incompatibility", + cstate->num_errors, + cstate->num_errors, + RelationGetRelationName(cstate->error_rel))); } if (bistate != NULL) @@ -1515,6 +1552,40 @@ CopyFrom(CopyFromState cstate) FreeExecutorState(estate); + /* + * This should be aligned with the resource release/destruction performed + * by ExecutorFinish and ExecutorEnd on the EState. + */ + if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + { + MemoryContext tmpcontext; + ModifyTableState *error_mtstate; + + tmpcontext = + MemoryContextSwitchTo(cstate->mtcontext->estate->es_query_cxt); + + error_mtstate = cstate->mtcontext->mtstate; + error_mtstate->mt_done = true; + cstate->mtcontext->estate->es_finished = true; + + /* Release resources associated with error_table */ + ExecResetTupleTable(cstate->mtcontext->estate->es_tupleTable, false); + ExecCloseResultRelations(cstate->mtcontext->estate); + ExecCloseRangeTableRelations(cstate->mtcontext->estate); + + /* Do away with our snapshots */ + UnregisterSnapshot(cstate->mtcontext->estate->es_snapshot); + UnregisterSnapshot(cstate->mtcontext->estate->es_crosscheck_snapshot); + + /* + * Must switch out of context before destroying it + */ + MemoryContextSwitchTo(tmpcontext); + + FreeExecutorState(cstate->mtcontext->estate); + cstate->mtcontext = NULL; + } + return processed; } @@ -1630,6 +1701,15 @@ BeginCopyFrom(ParseState *pstate, if (cstate->opts.on_error == COPY_ON_ERROR_IGNORE || cstate->opts.on_error == COPY_ON_ERROR_SET_NULL) cstate->escontext->details_wanted = false; + else if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + { + /* + * For ON_ERROR = TABLE, we must set details_wanted to true. This + * ensures that ErrorData is populated when the next error occurs, + * allowing us to capture error metadata. + */ + cstate->escontext->details_wanted = true; + } } else cstate->escontext = NULL; @@ -1653,6 +1733,90 @@ BeginCopyFrom(ParseState *pstate, cstate->domain_with_constraint[attno - 1] = DomainHasConstraints(att->atttypid, NULL); } } + else if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + { + /* Set up COPY FROM (ON_ERROR TABLE) */ + HeapTuple tp; + Oid errRelOid, + typoid; + Oid reloftype = InvalidOid; + + List *relname_list = stringToQualifiedNameList(cstate->opts.error_table, + NULL); + + RangeVar *relvar = makeRangeVarFromNameList(relname_list); + + /* + * We may insert tuples into error_table later. To avoid a deadlock or + * long hang during COPY FROM, verify that the table is not already + * locked; otherwise, report a lock conflict error. + */ + errRelOid = RangeVarGetRelidExtended(relvar, + RowExclusiveLock, + RVR_NOWAIT, + RangeVarCallbackForCopyErrorTable, + NULL); + + if (RelationGetRelid(cstate->rel) == errRelOid) + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot use relation \"%s\" for COPY error saving while copying data to it", + RelationGetRelationName(cstate->rel))); + + cstate->error_rel = table_open(errRelOid, NoLock); + + if (XactReadOnly && !cstate->error_rel->rd_islocaltemp) + PreventCommandIfReadOnly("COPY FROM"); + + /* + * The error-saving table must be a plain table. It cannot have + * rewrite rules or any enabled row security policies. + */ + if (cstate->error_rel->rd_rel->relrowsecurity || cstate->error_rel->rd_rules) + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot use relation \"%s\" for COPY error saving", + RelationGetRelationName(cstate->error_rel)), + cstate->error_rel->rd_rel->relrowsecurity + ? errdetail("The error saving table cannot have row-level security policies.") + : errdetail("The error saving table cannot have rules.")); + + if (RELATION_IS_OTHER_TEMP(cstate->error_rel)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot use temporary tables of other sessions for COPY error saving"))); + + typoid = GetSysCacheOid2(TYPENAMENSP, Anum_pg_type_oid, + CStringGetDatum("copy_error_saving"), + ObjectIdGetDatum(PG_CATALOG_NAMESPACE)); + if (!OidIsValid(typoid)) + elog(ERROR, "cache lookup failed for catalog type %s", "copy_error_saving"); + + tp = SearchSysCache1(RELOID, ObjectIdGetDatum(errRelOid)); + if (!HeapTupleIsValid(tp)) + elog(ERROR, "cache lookup failed for relation %u", errRelOid); + else + { + Form_pg_class reltup = (Form_pg_class) GETSTRUCT(tp); + + reloftype = reltup->reloftype; + + if (reloftype != typoid) + ereport(ERROR, + errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("cannot use relation \"%s\" for COPY error saving", + RelationGetRelationName(cstate->error_rel)), + OidIsValid(reloftype) + ? errdetail("Relation \"%s\" is a typed table based on type \"%s\".", + RelationGetRelationName(cstate->error_rel), + format_type_be_qualified(reloftype)) + : 0, + errhint("The COPY error saving table must be a typed table based on type \"%s\".", + format_type_be_qualified(typoid))); + + ReleaseSysCache(tp); + } + } /* Convert FORCE_NULL name list to per-column flags, check validity */ cstate->opts.force_null_flags = palloc0_array(bool, num_phys_attrs); @@ -1956,6 +2120,9 @@ EndCopyFrom(CopyFromState cstate) pgstat_progress_end_command(); + if (cstate->error_rel) + table_close(cstate->error_rel, NoLock); + MemoryContextDelete(cstate->copycontext); pfree(cstate); } @@ -1994,3 +2161,159 @@ ClosePipeFromProgram(CopyFromState cstate) errdetail_internal("%s", wait_result_to_str(pclose_rc)))); } } + +/* + * Perform permission and other checks for error_table, and initialize + * cstate->mtcontext. + */ +static void +CopyFromErrorTableInit(CopyFromState cstate) +{ + ModifyTableState *mtstate; + ModifyTable *node; + MemoryContext tmpcontext; + ParseState *pstate = make_parsestate(NULL); + EState *estate = CreateExecutorState(); + RTEPermissionInfo *perminfo; + ParseNamespaceItem *nsitem; + TupleDesc tupDesc = RelationGetDescr(cstate->error_rel); + + cstate->mtcontext = palloc0_object(ModifyTableContext); + + Assert(cstate->opts.on_error == COPY_ON_ERROR_TABLE); + + tmpcontext = MemoryContextSwitchTo(estate->es_query_cxt); + + estate->es_output_cid = GetCurrentCommandId(true); + estate->es_snapshot = RegisterSnapshot(GetActiveSnapshot()); + estate->es_crosscheck_snapshot = RegisterSnapshot(InvalidSnapshot); + + nsitem = addRangeTableEntryForRelation(pstate, + cstate->error_rel, + RowExclusiveLock, + NULL, + false, + false); + + perminfo = nsitem->p_perminfo; + perminfo->requiredPerms = ACL_INSERT; + + /* We insert into every (non-dropped) column of the error table */ + for (int i = 0; i < tupDesc->natts; i++) + { + Form_pg_attribute att = TupleDescAttr(tupDesc, i); + + if (!att->attisdropped) + perminfo->insertedCols = + bms_add_member(perminfo->insertedCols, + att->attnum - FirstLowInvalidHeapAttributeNumber); + } + + ExecCheckPermissions(pstate->p_rtable, list_make1(perminfo), true); + + ExecInitRangeTable(estate, pstate->p_rtable, pstate->p_rteperminfos, + bms_make_singleton(1)); + + node = makeNode(ModifyTable); + node->operation = CMD_INSERT; + node->canSetTag = false; + node->rootRelation = 0; + node->resultRelations = list_make1_int(1); + node->onConflictAction = ONCONFLICT_NONE; + + /* + * Populate ModifyTableState for inserting record to error saving table. + */ + mtstate = makeNode(ModifyTableState); + mtstate->ps.plan = (Plan *) node; + mtstate->ps.state = estate; + + mtstate->operation = CMD_INSERT; + mtstate->canSetTag = false; + mtstate->mt_done = false; + + mtstate->mt_nrels = 1; + mtstate->resultRelInfo = palloc_array(ResultRelInfo, 1); + + mtstate->rootResultRelInfo = mtstate->resultRelInfo; + ExecInitResultRelation(estate, mtstate->resultRelInfo, + linitial_int(node->resultRelations)); + + /* Verify the named relation is a valid target for INSERT */ + CheckValidResultRel(mtstate->resultRelInfo, node->operation, + node->onConflictAction, NIL); + + /* + * Open the table's indexes, if we have not done so already, so that we + * can add new index entries for the inserted tuple. + */ + if (cstate->error_rel->rd_rel->relhasindex && + mtstate->resultRelInfo->ri_IndexRelationDescs == NULL) + ExecOpenIndices(mtstate->resultRelInfo, false); + + MemoryContextSwitchTo(tmpcontext); + + cstate->mtcontext->mtstate = mtstate; + cstate->mtcontext->estate = estate; + free_parsestate(pstate); +} + +/* + * Callback to RangeVarGetRelidExtended(). + * + * Checks the following: + * - the relation specified is a table. + * - the table is not a system table. + * + * If any of these checks fails then an error is raised. + */ +static void +RangeVarCallbackForCopyErrorTable(const RangeVar *rv, Oid relid, Oid oldrelid, + void *arg) +{ + AclResult aclresult; + HeapTuple tuple; + Form_pg_class classform; + char relkind; + + tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid)); + if (!HeapTupleIsValid(tuple)) + return; + + classform = (Form_pg_class) GETSTRUCT(tuple); + relkind = classform->relkind; + + if (!allowSystemTableMods && IsSystemClass(relid, classform)) + ereport(ERROR, + errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("permission denied: \"%s\" is a system catalog", + rv->relname)); + + /* + * Currently, the error-saving table must be a regular relation. + * + * TODO: Allow cstate->error_rel to be a partitioned table. This should + * not be difficult, but requires proper handling of constraints and + * triggers on the partitioned table. + */ + if (relkind != RELKIND_RELATION) + ereport(ERROR, + errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("cannot use relation \"%s\" for COPY error saving", + rv->relname), + errdetail_relkind_not_supported(relkind)); + + /* + * COPY (ON_ERROR TABLE) inserts COPY FROM error details to error_table, + * therefore, the current user must have INSERT privileges on error_table. + */ + aclresult = pg_class_aclcheck(classform->oid, + GetUserId(), + ACL_INSERT); + if (aclresult != ACLCHECK_OK) + aclcheck_error(aclresult, + get_relkind_objtype(relkind), + get_rel_name(classform->oid)); + + ReleaseSysCache(tuple); +} diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c index 06450535779..cebf71848b3 100644 --- a/src/backend/commands/copyfromparse.c +++ b/src/backend/commands/copyfromparse.c @@ -62,6 +62,7 @@ #include #include +#include "access/xact.h" #include "commands/copyapi.h" #include "commands/copyfrom_internal.h" #include "commands/progress.h" @@ -76,6 +77,7 @@ #include "port/simd.h" #include "utils/builtins.h" #include "utils/rel.h" +#include "utils/timestamp.h" #include "utils/wait_event.h" #define ISOCTAL(c) (((c) >= '0') && ((c) <= '7')) @@ -1126,6 +1128,131 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, cstate->num_errors++; } } + else if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + { + char *err_code; + Datum *newvalues; + bool *newnulls; + MemoryContext tmpcontext; + TupleTableSlot *myslot; + ModifyTableState *mtstate; + EState *estate; + + if (cstate->mtcontext == NULL) + elog(ERROR, "ON_ERROR table is only supported by the COPY command"); + + mtstate = cstate->mtcontext->mtstate; + estate = mtstate->ps.state; + + /* + * Reset the per-tuple exprcontext. We do this after every + * tuple, to clean-up after expression evaluations etc. + */ + ResetPerTupleExprContext(estate); + + tmpcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); + + /* + * Use the ResultRelInfo's RETURNING slot to build the error + * row. COPY FROM has no RETURNING clause (nor ON CONFLICT or + * partition routing on the error table), so this slot is + * otherwise unused here and free to borrow as scratch space. + */ + myslot = ExecGetReturningSlot(estate, + mtstate->resultRelInfo); + + ExecClearTuple(myslot); + + newvalues = myslot->tts_values; + newnulls = myslot->tts_isnull; + + Assert(RelationGetDescr(cstate->error_rel)->natts == 11); + + for (int i = 0; i < RelationGetDescr(cstate->error_rel)->natts; i++) + newnulls[i] = false; + + /* + * Record when the COPY command started, so that the rows + * saved by one command can be told apart from those of other + * commands. + */ + newvalues[0] = + TimestampTzGetDatum(GetCurrentStatementStartTimestamp()); + newvalues[1] = ObjectIdGetDatum(GetUserId()); + newvalues[2] = ObjectIdGetDatum(cstate->rel->rd_rel->oid); + newvalues[3] = CStringGetTextDatum(cstate->filename ? cstate->filename : "STDIN"); + newvalues[4] = Int64GetDatum((int64) cstate->cur_lineno); + newvalues[5] = CStringGetTextDatum(cstate->line_buf.data); + newvalues[6] = CStringGetTextDatum(cstate->cur_attname); + + if (string) + newvalues[7] = CStringGetTextDatum(string); + else + { + newvalues[7] = (Datum) 0; + newnulls[7] = true; + } + + if (cstate->escontext->error_data->message) + newvalues[8] = + CStringGetTextDatum(cstate->escontext->error_data->message); + else + { + newvalues[8] = (Datum) 0; + newnulls[8] = true; + } + + if (cstate->escontext->error_data->detail) + newvalues[9] = + CStringGetTextDatum(cstate->escontext->error_data->detail); + else + { + newvalues[9] = (Datum) 0; + newnulls[9] = true; + } + + err_code = + unpack_sql_state(cstate->escontext->error_data->sqlerrcode); + newvalues[10] = CStringGetTextDatum(err_code); + + /* Build the virtual tuple. */ + ExecStoreVirtualTuple(myslot); + + /* + * For each row inserted into error_table, the BEFORE + * STATEMENT, BEFORE ROW, AFTER ROW, and AFTER STATEMENT + * triggers are fired. + */ + AfterTriggerBeginQuery(); + + /* + * Cannot reuse the previous TransitionCaptureState: + * AfterTriggerEndQuery frees it below. + */ + mtstate->mt_transition_capture = + MakeTransitionCaptureState(cstate->error_rel->trigdesc, + RelationGetRelid(cstate->error_rel), + CMD_INSERT); + + ExecBSInsertTriggers(estate, mtstate->rootResultRelInfo); + + ExecInsert(cstate->mtcontext, + mtstate->resultRelInfo, + myslot, + false, + NULL, + NULL); + + ExecASInsertTriggers(estate, + mtstate->rootResultRelInfo, + mtstate->mt_transition_capture); + + AfterTriggerEndQuery(estate); + + MemoryContextSwitchTo(tmpcontext); + + cstate->num_errors++; + } if (cstate->opts.log_verbosity == COPY_LOG_VERBOSITY_VERBOSE) { @@ -1155,6 +1282,13 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, cstate->cur_lineno, cstate->cur_attname, attval)); + else if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + ereport(NOTICE, + errmsg("saving error information to table \"%s\" due to data type incompatibility at line %" PRIu64 " for column \"%s\": \"%s\"", + RelationGetRelationName(cstate->error_rel), + cstate->cur_lineno, + cstate->cur_attname, + attval)); pfree(attval); } else @@ -1164,6 +1298,12 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, errmsg("skipping row due to data type incompatibility at line %" PRIu64 " for column \"%s\": null input", cstate->cur_lineno, cstate->cur_attname)); + else if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + ereport(NOTICE, + errmsg("saving error information to table \"%s\" due to data type incompatibility at line %" PRIu64 " for column \"%s\": null input", + RelationGetRelationName(cstate->error_rel), + cstate->cur_lineno, + cstate->cur_attname)); } /* reset relname_only */ cstate->relname_only = false; @@ -1173,6 +1313,8 @@ CopyFromTextLikeOneRow(CopyFromState cstate, ExprContext *econtext, return true; else if (cstate->opts.on_error == COPY_ON_ERROR_SET_NULL) continue; + else if (cstate->opts.on_error == COPY_ON_ERROR_TABLE) + return true; } cstate->cur_attname = NULL; diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 0563453fe24..170c66ca1bf 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -3672,6 +3672,7 @@ copy_generic_opt_arg: | NumericOnly { $$ = (Node *) $1; } | '*' { $$ = (Node *) makeNode(A_Star); } | DEFAULT { $$ = (Node *) makeString("default"); } + | TABLE { $$ = (Node *) makeString("table"); } | '(' copy_generic_opt_arg_list ')' { $$ = (Node *) $2; } | /* EMPTY */ { $$ = NULL; } ; diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index b3bfe050b18..0de46168db4 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -1241,7 +1241,7 @@ Alter_routine_options, "CALLED ON NULL INPUT", "RETURNS NULL ON NULL INPUT", \ /* COPY FROM options */ #define Copy_from_options \ Copy_common_options, "DEFAULT", "FORCE_NOT_NULL", "FORCE_NULL", "FREEZE", \ -"LOG_VERBOSITY", "ON_ERROR", "REJECT_LIMIT" +"LOG_VERBOSITY", "ON_ERROR", "REJECT_LIMIT", "ERROR_TABLE" /* COPY TO options */ #define Copy_to_options \ @@ -3435,7 +3435,26 @@ match_previous_words(int pattern_id, /* Complete COPY FROM filename WITH (ON_ERROR */ else if (TailMatches("ON_ERROR")) - COMPLETE_WITH("stop", "ignore", "set_null"); + COMPLETE_WITH("stop", "ignore", "set_null", "table"); + + /* + * Complete COPY FROM filename WITH (ON_ERROR with + * nothing, lest the words_after_create fallback offer table names + * after the "table" value. + */ + else if (TailMatches("ON_ERROR", MatchAny)) + { + COMPLETE_WITH_CONST(true, ""); + /* Also, prevent Readline from appending stuff to the non-match */ + rl_completion_append_character = '\0'; +#ifdef HAVE_RL_COMPLETION_SUPPRESS_QUOTE + rl_completion_suppress_quote = 1; +#endif + } + + /* Complete COPY FROM filename WITH (ERROR_TABLE */ + else if (TailMatches("ERROR_TABLE")) + COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables); /* Complete COPY FROM filename WITH (LOG_VERBOSITY */ else if (TailMatches("LOG_VERBOSITY")) diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h index abecfe51098..3745342d0f9 100644 --- a/src/include/commands/copy.h +++ b/src/include/commands/copy.h @@ -36,6 +36,7 @@ typedef enum CopyOnErrorChoice COPY_ON_ERROR_STOP = 0, /* immediately throw errors, default */ COPY_ON_ERROR_IGNORE, /* ignore errors */ COPY_ON_ERROR_SET_NULL, /* set error field to null */ + COPY_ON_ERROR_TABLE, /* save input conversion errors info to table */ } CopyOnErrorChoice; /* @@ -96,6 +97,7 @@ typedef struct CopyFormatOptions CopyOnErrorChoice on_error; /* what to do when error happened */ CopyLogVerbosityChoice log_verbosity; /* verbosity of logged messages */ int64 reject_limit; /* maximum tolerable number of errors */ + char *error_table; /* on error, save error info to the table, table name */ List *convert_select; /* list of column names (can be NIL) */ } CopyFormatOptions; diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h index 9d3e244ee55..64d0b204090 100644 --- a/src/include/commands/copyfrom_internal.h +++ b/src/include/commands/copyfrom_internal.h @@ -16,6 +16,7 @@ #include "commands/copy.h" #include "commands/trigger.h" +#include "executor/nodeModifyTable.h" #include "nodes/miscnodes.h" /* @@ -73,6 +74,7 @@ typedef struct CopyFromStateData /* parameters from the COPY command */ Relation rel; /* relation to copy from */ + Relation error_rel; /* relation for copy from error saving */ List *attnumlist; /* integer list of attnums to copy */ char *filename; /* filename, or NULL for STDIN */ bool is_program; /* is 'filename' a program to popen? */ @@ -189,6 +191,13 @@ typedef struct CopyFromStateData #define RAW_BUF_BYTES(cstate) ((cstate)->raw_buf_len - (cstate)->raw_buf_index) uint64 bytes_processed; /* number of bytes processed so far */ + + /* + * INSERT operation context for inserting COPY FROM input conversion + * failure error information to error_table. Populated only when ON_ERROR + * is specified as 'TABLE'. + */ + ModifyTableContext *mtcontext; } CopyFromStateData; extern void ReceiveCopyBegin(CopyFromState cstate); diff --git a/src/test/regress/expected/copy.out b/src/test/regress/expected/copy.out index 0af0b646921..fa86af92501 100644 --- a/src/test/regress/expected/copy.out +++ b/src/test/regress/expected/copy.out @@ -442,6 +442,12 @@ INFO: progress: {"type": "FILE", "command": "COPY FROM", "relname": "tab_progre copy tab_progress_reporting from stdin(on_error ignore); NOTICE: 2 rows were skipped due to data type incompatibility INFO: progress: {"type": "PIPE", "command": "COPY FROM", "relname": "tab_progress_reporting", "tuples_skipped": 2, "has_bytes_total": false, "tuples_excluded": 0, "tuples_processed": 1, "has_bytes_processed": true} +begin; +create temp table err_tbl OF copy_error_saving; +copy tab_progress_reporting from stdin(on_error table, error_table err_tbl); +NOTICE: 2 rows were saved to table "err_tbl" due to data type incompatibility +INFO: progress: {"type": "PIPE", "command": "COPY FROM", "relname": "tab_progress_reporting", "tuples_skipped": 2, "has_bytes_total": false, "tuples_excluded": 0, "tuples_processed": 1, "has_bytes_processed": true} +rollback; drop trigger check_after_tab_progress_reporting on tab_progress_reporting; drop function notice_after_tab_progress_reporting(); drop table tab_progress_reporting; diff --git a/src/test/regress/expected/copy2.out b/src/test/regress/expected/copy2.out index 919eabd5f78..efb3c9c7b79 100644 --- a/src/test/regress/expected/copy2.out +++ b/src/test/regress/expected/copy2.out @@ -888,7 +888,256 @@ ERROR: skipped more than REJECT_LIMIT (3) rows due to data type incompatibility CONTEXT: COPY check_ign_err, line 5, column n: "" COPY check_ign_err FROM STDIN WITH (on_error ignore, reject_limit 4); NOTICE: 4 rows were skipped due to data type incompatibility +-- Tests for on_error table, save COPY FROM input data type conversion error +-- information to a user-defined table +-- Direct modification of system catalog composite type copy_error_saving is +-- not allowed +ALTER TYPE copy_error_saving ADD ATTRIBUTE b text; +ERROR: permission denied: "copy_error_saving" is a system catalog +ALTER TYPE copy_error_saving DROP ATTRIBUTE userid; +ERROR: permission denied: "copy_error_saving" is a system catalog +ALTER TYPE copy_error_saving RENAME ATTRIBUTE userid to userid1; +ERROR: permission denied: "copy_error_saving" is a system catalog +ALTER TYPE copy_error_saving ALTER ATTRIBUTE userid SET DATA TYPE OID8; +ERROR: permission denied: "copy_error_saving" is a system catalog +CREATE TABLE t_on_error_table(a jsonb, b int, c int, d dcheck_ign_err2); +CREATE TABLE err_tbl OF copy_error_saving; +CREATE TABLE err_tbl1 OF copy_error_saving; +CREATE TABLE err_tbl2 OF copy_error_saving PARTITION BY RANGE (lineno); +CREATE UNIQUE INDEX err_tbl_idx ON err_tbl(colname); +CREATE TYPE t_copy_typ AS (a int, b int, c int); +CREATE TABLE t_copy_tbl1 OF t_copy_typ; +CREATE TEMP VIEW t_copy_v1 AS SELECT * FROM t_on_error_table; +CREATE TEMP TABLE ro_tmp (a int); +-- all of the following should fail +COPY t_on_error_table FROM STDIN WITH (error_table err_tbl); +ERROR: COPY ERROR_TABLE can only be used when option ON_ERROR is set to "TABLE" +COPY t_on_error_table FROM STDIN WITH (on_error table); +ERROR: cannot set option ON_ERROR to "TABLE" when "ERROR_TABLE" is not specified +HINT: "ERROR_TABLE" option is required +COPY t_on_error_table TO STDIN WITH (on_error table, error_table err_tbl); +ERROR: COPY ON_ERROR cannot be used with COPY TO +LINE 1: COPY t_on_error_table TO STDIN WITH (on_error table, error_t... + ^ +COPY t_on_error_table FROM STDIN WITH (format binary, on_error table, error_table err_tbl); +ERROR: only ON_ERROR STOP is allowed in BINARY mode +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table t_on_error_table); +ERROR: cannot use relation "t_on_error_table" for COPY error saving while copying data to it +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table t_copy_tbl1); +ERROR: cannot use relation "t_copy_tbl1" for COPY error saving +DETAIL: Relation "t_copy_tbl1" is a typed table based on type "public.t_copy_typ". +HINT: The COPY error saving table must be a typed table based on type "pg_catalog.copy_error_saving". +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table t_copy_v1); +ERROR: cannot use relation "t_copy_v1" for COPY error saving +DETAIL: This operation is not supported for views. +COPY t_on_error_table FROM STDIN WITH (on_error table, reject_limit 10, error_table err_tbl); +ERROR: COPY REJECT_LIMIT requires ON_ERROR to be set to IGNORE +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table not_existsx); +ERROR: relation "not_existsx" does not exist +START TRANSACTION READ ONLY; +COPY ro_tmp FROM STDIN WITH (on_error table, error_table err_tbl); +ERROR: cannot execute COPY FROM in a read-only transaction +ROLLBACK; +-- all of the above should fail +-- error, error_table cannot be partitioned table +COPY t_on_error_table(a,b) FROM STDIN WITH (on_error table, error_table err_tbl2); +ERROR: cannot use relation "err_tbl2" for COPY error saving +DETAIL: This operation is not supported for partitioned tables. +CREATE RULE regtest_test_rule AS ON INSERT TO err_tbl1 DO ALSO NOTHING; +-- error, error_table cannot be have rules +COPY t_on_error_table(a,b) FROM STDIN WITH (on_error table, error_table err_tbl1); +ERROR: cannot use relation "err_tbl1" for COPY error saving +DETAIL: The error saving table cannot have rules. +DROP RULE regtest_test_rule ON err_tbl1; +CREATE POLICY p1 ON err_tbl1 FOR SELECT USING (true); +ALTER TABLE err_tbl1 ENABLE ROW LEVEL SECURITY; +ALTER TABLE err_tbl1 FORCE ROW LEVEL SECURITY; +-- error, error_table cannot be have RLS policies +COPY t_on_error_table(a,b) FROM STDIN WITH (on_error table, error_table err_tbl1); +ERROR: cannot use relation "err_tbl1" for COPY error saving +DETAIL: The error saving table cannot have row-level security policies. +DROP POLICY IF EXISTS p1 ON err_tbl1; +ALTER TABLE err_tbl1 DISABLE ROW LEVEL SECURITY; +ALTER TABLE err_tbl ADD CONSTRAINT cc2 CHECK(lineno > 0); +ALTER TABLE err_tbl ADD CONSTRAINT cc3 NOT NULL userid; +-- ok, constraints on table err_tbl1 will be verified +COPY t_on_error_table(b, a) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +NOTICE: 1 row was saved to table "err_tbl" due to data type incompatibility +ALTER TABLE err_tbl DROP CONSTRAINT cc2; +ALTER TABLE err_tbl DROP CONSTRAINT cc3; +TRUNCATE err_tbl; +-- foreign key on error_table is supported +BEGIN; +CREATE TABLE pktest(lineno int8 primary key); +INSERT INTO pktest VALUES (2); +ALTER TABLE err_tbl ADD CONSTRAINT cc4 FOREIGN KEY (lineno) REFERENCES pktest; +COPY t_on_error_table(b, a) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); -- ok +NOTICE: 1 row was saved to table "err_tbl" due to data type incompatibility +COPY t_on_error_table(b, c) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); -- error +ERROR: insert or update on table "err_tbl" violates foreign key constraint "cc4" +DETAIL: Key (lineno)=(1) is not present in table "pktest". +CONTEXT: COPY t_on_error_table, line 1, column c: "c" +ROLLBACK; +-- fail, copied data have extra columns +COPY t_on_error_table(a,b) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +ERROR: extra data after last expected column +CONTEXT: COPY t_on_error_table, line 1: "1,2,3,4,5" +-- fail, copied data have less columns +COPY t_on_error_table(a,b) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +ERROR: extra data after last expected column +CONTEXT: COPY t_on_error_table, line 1: "1,2,3" +-- permission check +BEGIN; +CREATE USER regress_user30; +GRANT INSERT(userid, copy_tbl, filename, lineno, line, colname, raw_field_value, err_message, err_detail) + ON TABLE err_tbl TO regress_user30; +GRANT INSERT ON TABLE t_on_error_table TO regress_user30; +GRANT SELECT ON TABLE err_tbl TO regress_user30; +SAVEPOINT s1; +SET ROLE regress_user30; +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table err_tbl); -- error, insufficient privilege +ERROR: permission denied for table err_tbl +ROLLBACK TO SAVEPOINT s1; +RESET ROLE; +GRANT INSERT on TABLE err_tbl to regress_user30; +SET ROLE regress_user30; +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl, log_verbosity verbose); -- ok +NOTICE: saving error information to table "err_tbl" due to data type incompatibility at line 1 for column "a": "a" +NOTICE: 1 row was saved to table "err_tbl" due to data type incompatibility +SELECT copy_tbl::regclass, filename, lineno, line, colname, raw_field_value, err_message, err_detail, errorcode +FROM err_tbl; + copy_tbl | filename | lineno | line | colname | raw_field_value | err_message | err_detail | errorcode +------------------+----------+--------+---------+---------+-----------------+------------------------------------+-----------------------+----------- + t_on_error_table | STDIN | 1 | a,b,3,4 | a | a | invalid input syntax for type json | Token "a" is invalid. | 22P02 +(1 row) + +-- error, unique constraint violation +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +ERROR: duplicate key value violates unique constraint "err_tbl_idx" +DETAIL: Key (colname)=(a) already exists. +CONTEXT: COPY t_on_error_table, line 1, column a: "a" +ROLLBACK; +DROP INDEX err_tbl_idx; +BEGIN; +TRUNCATE err_tbl; +TRUNCATE t_on_error_table; +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl, FREEZE); -- ok +NOTICE: 2 rows were saved to table "err_tbl" due to data type incompatibility +ROLLBACK; +CREATE FUNCTION trig_copy_error_saving_insert() +RETURNS TRIGGER LANGUAGE plpgsql AS +$$ +BEGIN + if (TG_LEVEL = 'STATEMENT' and TG_WHEN = 'AFTER') then + RAISE NOTICE E'trigger name: %, % % FOR EACH %\n', TG_NAME, TG_WHEN, TG_OP, TG_LEVEL; + else + RAISE NOTICE 'trigger name: %, % % FOR EACH %', TG_NAME, TG_WHEN, TG_OP, TG_LEVEL; + end if; + if TG_LEVEL = 'ROW' then + RAISE NOTICE 'NEW raw_field_value: %, err_message: %', NEW.raw_field_value, NEW.err_message; + end if; + RETURN NEW; +END; +$$; +CREATE TRIGGER err_tbl_after_ins_row + AFTER INSERT ON err_tbl + FOR EACH ROW EXECUTE PROCEDURE trig_copy_error_saving_insert(); +CREATE TRIGGER err_tbl_before_ins_row + BEFORE INSERT ON err_tbl + FOR EACH ROW EXECUTE PROCEDURE trig_copy_error_saving_insert(); +CREATE TRIGGER err_tbl_bfore_ins_stmt + BEFORE INSERT ON err_tbl + FOR EACH STATEMENT EXECUTE PROCEDURE trig_copy_error_saving_insert(); +CREATE TRIGGER err_tbl_after_ins_stmt + AFTER INSERT ON err_tbl + REFERENCING NEW TABLE AS new_rows + FOR EACH STATEMENT EXECUTE PROCEDURE trig_copy_error_saving_insert(); +-- Each insert on the error_table invokes both the statement-level and row-level triggers. +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +NOTICE: trigger name: err_tbl_bfore_ins_stmt, BEFORE INSERT FOR EACH STATEMENT +NOTICE: trigger name: err_tbl_before_ins_row, BEFORE INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: a, err_message: invalid input syntax for type integer: "a" +NOTICE: trigger name: err_tbl_after_ins_row, AFTER INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: a, err_message: invalid input syntax for type integer: "a" +NOTICE: trigger name: err_tbl_after_ins_stmt, AFTER INSERT FOR EACH STATEMENT + +NOTICE: trigger name: err_tbl_bfore_ins_stmt, BEFORE INSERT FOR EACH STATEMENT +NOTICE: trigger name: err_tbl_before_ins_row, BEFORE INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: , err_message: domain dcheck_ign_err2 does not allow null values +NOTICE: trigger name: err_tbl_after_ins_row, AFTER INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: , err_message: domain dcheck_ign_err2 does not allow null values +NOTICE: trigger name: err_tbl_after_ins_stmt, AFTER INSERT FOR EACH STATEMENT + +NOTICE: trigger name: err_tbl_bfore_ins_stmt, BEFORE INSERT FOR EACH STATEMENT +NOTICE: trigger name: err_tbl_before_ins_row, BEFORE INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: _junk, err_message: invalid input syntax for type integer: "_junk" +NOTICE: trigger name: err_tbl_after_ins_row, AFTER INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: _junk, err_message: invalid input syntax for type integer: "_junk" +NOTICE: trigger name: err_tbl_after_ins_stmt, AFTER INSERT FOR EACH STATEMENT + +NOTICE: trigger name: err_tbl_bfore_ins_stmt, BEFORE INSERT FOR EACH STATEMENT +NOTICE: trigger name: err_tbl_before_ins_row, BEFORE INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: cola, err_message: invalid input syntax for type json +NOTICE: trigger name: err_tbl_after_ins_row, AFTER INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: cola, err_message: invalid input syntax for type json +NOTICE: trigger name: err_tbl_after_ins_stmt, AFTER INSERT FOR EACH STATEMENT + +NOTICE: trigger name: err_tbl_bfore_ins_stmt, BEFORE INSERT FOR EACH STATEMENT +NOTICE: trigger name: err_tbl_before_ins_row, BEFORE INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: 4238679732489879879, err_message: value "4238679732489879879" is out of range for type integer +NOTICE: trigger name: err_tbl_after_ins_row, AFTER INSERT FOR EACH ROW +NOTICE: NEW raw_field_value: 4238679732489879879, err_message: value "4238679732489879879" is out of range for type integer +NOTICE: trigger name: err_tbl_after_ins_stmt, AFTER INSERT FOR EACH STATEMENT + +NOTICE: 5 rows were saved to table "err_tbl" due to data type incompatibility +SELECT copy_tbl::regclass, filename, lineno, line, colname ,raw_field_value, err_message, err_detail, errorcode +FROM err_tbl; + copy_tbl | filename | lineno | line | colname | raw_field_value | err_message | err_detail | errorcode +------------------+----------+--------+----------------------------+---------+---------------------+--------------------------------------------------------------+--------------------------+----------- + t_on_error_table | STDIN | 1 | 1,2,a,1 | c | a | invalid input syntax for type integer: "a" | | 22P02 + t_on_error_table | STDIN | 2 | 1,2,3,\N | d | | domain dcheck_ign_err2 does not allow null values | | 23502 + t_on_error_table | STDIN | 3 | 1,_junk,test,11 | b | _junk | invalid input syntax for type integer: "_junk" | | 22P02 + t_on_error_table | STDIN | 4 | cola,colb,colc,12 | a | cola | invalid input syntax for type json | Token "cola" is invalid. | 22P02 + t_on_error_table | STDIN | 6 | 1,11,4238679732489879879,2 | c | 4238679732489879879 | value "4238679732489879879" is out of range for type integer | | 22003 +(5 rows) + +-- all rows saved by one COPY share its statement start time +SELECT count(DISTINCT starttime) AS distinct_starttime, + bool_and(starttime <= now()) AS not_in_future +FROM err_tbl; + distinct_starttime | not_in_future +--------------------+--------------- + 1 | t +(1 row) + +CREATE FUNCTION triga_test() +RETURNS TRIGGER LANGUAGE plpgsql AS +$$ +BEGIN + ALTER TABLE err_tbl ADD COLUMN x INT DEFAULT 11; + RETURN NEW; +END; +$$; +CREATE TRIGGER err_tblatest + BEFORE INSERT ON err_tbl + FOR EACH STATEMENT + EXECUTE PROCEDURE triga_test(); +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); -- error +NOTICE: trigger name: err_tbl_bfore_ins_stmt, BEFORE INSERT FOR EACH STATEMENT +ERROR: cannot ALTER TABLE "err_tbl" because it is being used by active queries in this session +CONTEXT: SQL statement "ALTER TABLE err_tbl ADD COLUMN x INT DEFAULT 11" +PL/pgSQL function triga_test() line 3 at SQL statement +COPY t_on_error_table, line 1, column b: "_junk" -- clean up +DROP TABLE err_tbl; +DROP TABLE err_tbl1; +DROP TABLE err_tbl2; +DROP VIEW t_copy_v1; +DROP TABLE t_on_error_table; +DROP TABLE t_copy_tbl1; +DROP TYPE t_copy_typ; +DROP FUNCTION trig_copy_error_saving_insert(); +DROP FUNCTION triga_test(); DROP TABLE forcetest; DROP TABLE vistest; DROP FUNCTION truncate_in_subxact(); diff --git a/src/test/regress/sql/copy.sql b/src/test/regress/sql/copy.sql index 14da3c5ec37..7c77a38cb9c 100644 --- a/src/test/regress/sql/copy.sql +++ b/src/test/regress/sql/copy.sql @@ -386,6 +386,15 @@ sharon 25 (15,12) 1000 sam sharon y (15,12) x sam \. +begin; +create temp table err_tbl OF copy_error_saving; +copy tab_progress_reporting from stdin(on_error table, error_table err_tbl); +sharon x (15,12) x sam +sharon 25 (15,12) 1000 sam +sharon y (15,12) x sam +\. +rollback; + drop trigger check_after_tab_progress_reporting on tab_progress_reporting; drop function notice_after_tab_progress_reporting(); drop table tab_progress_reporting; diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql index 696ceeeaf82..7b9bbee1e2a 100644 --- a/src/test/regress/sql/copy2.sql +++ b/src/test/regress/sql/copy2.sql @@ -687,7 +687,214 @@ a {7} 7 10 {10} 10 \. +-- Tests for on_error table, save COPY FROM input data type conversion error +-- information to a user-defined table + +-- Direct modification of system catalog composite type copy_error_saving is +-- not allowed +ALTER TYPE copy_error_saving ADD ATTRIBUTE b text; +ALTER TYPE copy_error_saving DROP ATTRIBUTE userid; +ALTER TYPE copy_error_saving RENAME ATTRIBUTE userid to userid1; +ALTER TYPE copy_error_saving ALTER ATTRIBUTE userid SET DATA TYPE OID8; + +CREATE TABLE t_on_error_table(a jsonb, b int, c int, d dcheck_ign_err2); +CREATE TABLE err_tbl OF copy_error_saving; +CREATE TABLE err_tbl1 OF copy_error_saving; +CREATE TABLE err_tbl2 OF copy_error_saving PARTITION BY RANGE (lineno); +CREATE UNIQUE INDEX err_tbl_idx ON err_tbl(colname); +CREATE TYPE t_copy_typ AS (a int, b int, c int); +CREATE TABLE t_copy_tbl1 OF t_copy_typ; +CREATE TEMP VIEW t_copy_v1 AS SELECT * FROM t_on_error_table; +CREATE TEMP TABLE ro_tmp (a int); + +-- all of the following should fail +COPY t_on_error_table FROM STDIN WITH (error_table err_tbl); +\. +COPY t_on_error_table FROM STDIN WITH (on_error table); +\. +COPY t_on_error_table TO STDIN WITH (on_error table, error_table err_tbl); +COPY t_on_error_table FROM STDIN WITH (format binary, on_error table, error_table err_tbl); +\. +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table t_on_error_table); +\. +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table t_copy_tbl1); +\. +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table t_copy_v1); +\. +COPY t_on_error_table FROM STDIN WITH (on_error table, reject_limit 10, error_table err_tbl); +\. +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table not_existsx); +\. +START TRANSACTION READ ONLY; +COPY ro_tmp FROM STDIN WITH (on_error table, error_table err_tbl); +\. +ROLLBACK; +-- all of the above should fail + +-- error, error_table cannot be partitioned table +COPY t_on_error_table(a,b) FROM STDIN WITH (on_error table, error_table err_tbl2); +\. +CREATE RULE regtest_test_rule AS ON INSERT TO err_tbl1 DO ALSO NOTHING; +-- error, error_table cannot be have rules +COPY t_on_error_table(a,b) FROM STDIN WITH (on_error table, error_table err_tbl1); +\. +DROP RULE regtest_test_rule ON err_tbl1; + +CREATE POLICY p1 ON err_tbl1 FOR SELECT USING (true); +ALTER TABLE err_tbl1 ENABLE ROW LEVEL SECURITY; +ALTER TABLE err_tbl1 FORCE ROW LEVEL SECURITY; +-- error, error_table cannot be have RLS policies +COPY t_on_error_table(a,b) FROM STDIN WITH (on_error table, error_table err_tbl1); +\. +DROP POLICY IF EXISTS p1 ON err_tbl1; +ALTER TABLE err_tbl1 DISABLE ROW LEVEL SECURITY; + +ALTER TABLE err_tbl ADD CONSTRAINT cc2 CHECK(lineno > 0); +ALTER TABLE err_tbl ADD CONSTRAINT cc3 NOT NULL userid; +-- ok, constraints on table err_tbl1 will be verified +COPY t_on_error_table(b, a) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +a,b +\. +ALTER TABLE err_tbl DROP CONSTRAINT cc2; +ALTER TABLE err_tbl DROP CONSTRAINT cc3; +TRUNCATE err_tbl; + +-- foreign key on error_table is supported +BEGIN; +CREATE TABLE pktest(lineno int8 primary key); +INSERT INTO pktest VALUES (2); +ALTER TABLE err_tbl ADD CONSTRAINT cc4 FOREIGN KEY (lineno) REFERENCES pktest; +COPY t_on_error_table(b, a) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); -- ok +1,null +a,b +\. +COPY t_on_error_table(b, c) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); -- error +1,c +\. +ROLLBACK; + +-- fail, copied data have extra columns +COPY t_on_error_table(a,b) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +1,2,3,4,5 +\. + +-- fail, copied data have less columns +COPY t_on_error_table(a,b) FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +1,2,3 +\. + +-- permission check +BEGIN; +CREATE USER regress_user30; +GRANT INSERT(userid, copy_tbl, filename, lineno, line, colname, raw_field_value, err_message, err_detail) + ON TABLE err_tbl TO regress_user30; +GRANT INSERT ON TABLE t_on_error_table TO regress_user30; +GRANT SELECT ON TABLE err_tbl TO regress_user30; +SAVEPOINT s1; + +SET ROLE regress_user30; +COPY t_on_error_table FROM STDIN WITH (on_error table, error_table err_tbl); -- error, insufficient privilege +\. +ROLLBACK TO SAVEPOINT s1; + +RESET ROLE; +GRANT INSERT on TABLE err_tbl to regress_user30; +SET ROLE regress_user30; +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl, log_verbosity verbose); -- ok +a,b,3,4 +\. + +SELECT copy_tbl::regclass, filename, lineno, line, colname, raw_field_value, err_message, err_detail, errorcode +FROM err_tbl; +-- error, unique constraint violation +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +a,b,3,4 +\. +ROLLBACK; +DROP INDEX err_tbl_idx; + +BEGIN; +TRUNCATE err_tbl; +TRUNCATE t_on_error_table; +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl, FREEZE); -- ok +1,2,a,1 +1,2,3,\N +\. +ROLLBACK; + +CREATE FUNCTION trig_copy_error_saving_insert() +RETURNS TRIGGER LANGUAGE plpgsql AS +$$ +BEGIN + if (TG_LEVEL = 'STATEMENT' and TG_WHEN = 'AFTER') then + RAISE NOTICE E'trigger name: %, % % FOR EACH %\n', TG_NAME, TG_WHEN, TG_OP, TG_LEVEL; + else + RAISE NOTICE 'trigger name: %, % % FOR EACH %', TG_NAME, TG_WHEN, TG_OP, TG_LEVEL; + end if; + if TG_LEVEL = 'ROW' then + RAISE NOTICE 'NEW raw_field_value: %, err_message: %', NEW.raw_field_value, NEW.err_message; + end if; + RETURN NEW; +END; +$$; + +CREATE TRIGGER err_tbl_after_ins_row + AFTER INSERT ON err_tbl + FOR EACH ROW EXECUTE PROCEDURE trig_copy_error_saving_insert(); +CREATE TRIGGER err_tbl_before_ins_row + BEFORE INSERT ON err_tbl + FOR EACH ROW EXECUTE PROCEDURE trig_copy_error_saving_insert(); +CREATE TRIGGER err_tbl_bfore_ins_stmt + BEFORE INSERT ON err_tbl + FOR EACH STATEMENT EXECUTE PROCEDURE trig_copy_error_saving_insert(); +CREATE TRIGGER err_tbl_after_ins_stmt + AFTER INSERT ON err_tbl + REFERENCING NEW TABLE AS new_rows + FOR EACH STATEMENT EXECUTE PROCEDURE trig_copy_error_saving_insert(); + +-- Each insert on the error_table invokes both the statement-level and row-level triggers. +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); +1,2,a,1 +1,2,3,\N +1,_junk,test,11 +cola,colb,colc,12 +4,5,6,1111 +1,11,4238679732489879879,2 +\. +SELECT copy_tbl::regclass, filename, lineno, line, colname ,raw_field_value, err_message, err_detail, errorcode +FROM err_tbl; +-- all rows saved by one COPY share its statement start time +SELECT count(DISTINCT starttime) AS distinct_starttime, + bool_and(starttime <= now()) AS not_in_future +FROM err_tbl; + +CREATE FUNCTION triga_test() +RETURNS TRIGGER LANGUAGE plpgsql AS +$$ +BEGIN + ALTER TABLE err_tbl ADD COLUMN x INT DEFAULT 11; + RETURN NEW; +END; +$$; + +CREATE TRIGGER err_tblatest + BEFORE INSERT ON err_tbl + FOR EACH STATEMENT + EXECUTE PROCEDURE triga_test(); +COPY t_on_error_table FROM STDIN WITH (DELIMITER ',', on_error table, error_table err_tbl); -- error +1,_junk,test,11 +\. + -- clean up +DROP TABLE err_tbl; +DROP TABLE err_tbl1; +DROP TABLE err_tbl2; +DROP VIEW t_copy_v1; +DROP TABLE t_on_error_table; +DROP TABLE t_copy_tbl1; +DROP TYPE t_copy_typ; +DROP FUNCTION trig_copy_error_saving_insert(); +DROP FUNCTION triga_test(); DROP TABLE forcetest; DROP TABLE vistest; DROP FUNCTION truncate_in_subxact(); -- 2.34.1