Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * inv_api.c
4 : : * routines for manipulating inversion fs large objects. This file
5 : : * contains the user-level large object application interface routines.
6 : : *
7 : : *
8 : : * Note: we access pg_largeobject.data using its C struct declaration.
9 : : * This is safe because it immediately follows pageno which is an int4 field,
10 : : * and therefore the data field will always be 4-byte aligned, even if it
11 : : * is in the short 1-byte-header format. We have to detoast it since it's
12 : : * quite likely to be in compressed or short format. We also need to check
13 : : * for NULLs, since initdb will mark loid and pageno but not data as NOT NULL.
14 : : *
15 : : * Note: many of these routines leak memory in CurrentMemoryContext, as indeed
16 : : * does most of the backend code. We expect that CurrentMemoryContext will
17 : : * be a short-lived context. Data that must persist across function calls
18 : : * is kept either in CacheMemoryContext (the Relation structs) or in the
19 : : * memory context given to inv_open (for LargeObjectDesc structs).
20 : : *
21 : : *
22 : : * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
23 : : * Portions Copyright (c) 1994, Regents of the University of California
24 : : *
25 : : *
26 : : * IDENTIFICATION
27 : : * src/backend/storage/large_object/inv_api.c
28 : : *
29 : : *-------------------------------------------------------------------------
30 : : */
31 : : #include "postgres.h"
32 : :
33 : : #include <limits.h>
34 : :
35 : : #include "access/detoast.h"
36 : : #include "access/genam.h"
37 : : #include "access/htup_details.h"
38 : : #include "access/table.h"
39 : : #include "access/xact.h"
40 : : #include "catalog/dependency.h"
41 : : #include "catalog/indexing.h"
42 : : #include "catalog/objectaccess.h"
43 : : #include "catalog/pg_largeobject.h"
44 : : #include "libpq/libpq-fs.h"
45 : : #include "miscadmin.h"
46 : : #include "storage/large_object.h"
47 : : #include "utils/acl.h"
48 : : #include "utils/fmgroids.h"
49 : : #include "utils/rel.h"
50 : : #include "utils/snapmgr.h"
51 : :
52 : :
53 : : /*
54 : : * GUC: backwards-compatibility flag to suppress LO permission checks
55 : : */
56 : : bool lo_compat_privileges;
57 : :
58 : : /*
59 : : * All accesses to pg_largeobject and its index make use of a single
60 : : * Relation reference. To guarantee that the relcache entry remains
61 : : * in the cache, on the first reference inside a subtransaction, we
62 : : * execute a slightly klugy maneuver to assign ownership of the
63 : : * Relation reference to TopTransactionResourceOwner.
64 : : */
65 : : static Relation lo_heap_r = NULL;
66 : : static Relation lo_index_r = NULL;
67 : :
68 : :
69 : : /*
70 : : * Open pg_largeobject and its index, if not already done in current xact
71 : : */
72 : : static void
7812 tgl@sss.pgh.pa.us 73 :CBC 1594 : open_lo_relation(void)
74 : : {
75 : : ResourceOwner currentOwner;
76 : :
77 [ + + + - ]: 1594 : if (lo_heap_r && lo_index_r)
78 : 1407 : return; /* already open in current xact */
79 : :
80 : : /* Arrange for the top xact to own these relation references */
81 : 187 : currentOwner = CurrentResourceOwner;
2989 82 : 187 : CurrentResourceOwner = TopTransactionResourceOwner;
83 : :
84 : : /* Use RowExclusiveLock since we might either read or write */
85 [ + - ]: 187 : if (lo_heap_r == NULL)
2522 andres@anarazel.de 86 : 187 : lo_heap_r = table_open(LargeObjectRelationId, RowExclusiveLock);
2989 tgl@sss.pgh.pa.us 87 [ + - ]: 187 : if (lo_index_r == NULL)
88 : 187 : lo_index_r = index_open(LargeObjectLOidPNIndexId, RowExclusiveLock);
89 : :
7812 90 : 187 : CurrentResourceOwner = currentOwner;
91 : : }
92 : :
93 : : /*
94 : : * Clean up at main transaction end
95 : : */
96 : : void
97 : 270 : close_lo_relation(bool isCommit)
98 : : {
99 [ + + - + ]: 270 : if (lo_heap_r || lo_index_r)
100 : : {
101 : : /*
102 : : * Only bother to close if committing; else abort cleanup will handle
103 : : * it
104 : : */
105 [ + + ]: 187 : if (isCommit)
106 : : {
107 : : ResourceOwner currentOwner;
108 : :
109 : 138 : currentOwner = CurrentResourceOwner;
2989 110 : 138 : CurrentResourceOwner = TopTransactionResourceOwner;
111 : :
112 [ + - ]: 138 : if (lo_index_r)
113 : 138 : index_close(lo_index_r, NoLock);
114 [ + - ]: 138 : if (lo_heap_r)
2522 andres@anarazel.de 115 : 138 : table_close(lo_heap_r, NoLock);
116 : :
7812 tgl@sss.pgh.pa.us 117 : 138 : CurrentResourceOwner = currentOwner;
118 : : }
119 : 187 : lo_heap_r = NULL;
120 : 187 : lo_index_r = NULL;
121 : : }
122 : 270 : }
123 : :
124 : :
125 : : /*
126 : : * Extract data field from a pg_largeobject tuple, detoasting if needed
127 : : * and verifying that the length is sane. Returns data pointer (a bytea *),
128 : : * data length, and an indication of whether to pfree the data pointer.
129 : : */
130 : : static void
4213 131 : 5162 : getdatafield(Form_pg_largeobject tuple,
132 : : bytea **pdatafield,
133 : : int *plen,
134 : : bool *pfreeit)
135 : : {
136 : : bytea *datafield;
137 : : int len;
138 : : bool freeit;
139 : :
140 : 5162 : datafield = &(tuple->data); /* see note at top of file */
141 : 5162 : freeit = false;
142 [ + + ]: 5162 : if (VARATT_IS_EXTENDED(datafield))
143 : : {
144 : : datafield = (bytea *)
2266 rhaas@postgresql.org 145 : 5079 : detoast_attr((struct varlena *) datafield);
4213 tgl@sss.pgh.pa.us 146 : 5079 : freeit = true;
147 : : }
148 : 5162 : len = VARSIZE(datafield) - VARHDRSZ;
149 [ + - - + ]: 5162 : if (len < 0 || len > LOBLKSIZE)
4213 tgl@sss.pgh.pa.us 150 [ # # ]:UBC 0 : ereport(ERROR,
151 : : (errcode(ERRCODE_DATA_CORRUPTED),
152 : : errmsg("pg_largeobject entry for OID %u, page %d has invalid data field size %d",
153 : : tuple->loid, tuple->pageno, len)));
4213 tgl@sss.pgh.pa.us 154 :CBC 5162 : *pdatafield = datafield;
155 : 5162 : *plen = len;
156 : 5162 : *pfreeit = freeit;
9185 157 : 5162 : }
158 : :
159 : :
160 : : /*
161 : : * inv_create -- create a new large object
162 : : *
163 : : * Arguments:
164 : : * lobjId - OID to use for new large object, or InvalidOid to pick one
165 : : *
166 : : * Returns:
167 : : * OID of new object
168 : : *
169 : : * If lobjId is not InvalidOid, then an error occurs if the OID is already
170 : : * in use.
171 : : */
172 : : Oid
7492 173 : 84 : inv_create(Oid lobjId)
174 : : {
175 : : Oid lobjId_new;
176 : :
177 : : /*
178 : : * Create a new largeobject with empty data pages
179 : : */
5850 itagaki.takahiro@gma 180 : 84 : lobjId_new = LargeObjectCreate(lobjId);
181 : :
182 : : /*
183 : : * dependency on the owner of largeobject
184 : : *
185 : : * Note that LO dependencies are recorded using classId
186 : : * LargeObjectRelationId for backwards-compatibility reasons. Using
187 : : * LargeObjectMetadataRelationId instead would simplify matters for the
188 : : * backend, but it'd complicate pg_dump and possibly break other clients.
189 : : */
190 : 84 : recordDependencyOnOwner(LargeObjectRelationId,
191 : : lobjId_new, GetUserId());
192 : :
193 : : /* Post creation hook for new large object */
4669 rhaas@postgresql.org 194 [ - + ]: 84 : InvokeObjectPostCreateHook(LargeObjectRelationId, lobjId_new, 0);
195 : :
196 : : /*
197 : : * Advance command counter to make new tuple visible to later operations.
198 : : */
9187 bruce@momjian.us 199 : 84 : CommandCounterIncrement();
200 : :
5850 itagaki.takahiro@gma 201 : 84 : return lobjId_new;
202 : : }
203 : :
204 : : /*
205 : : * inv_open -- access an existing large object.
206 : : *
207 : : * Returns a large object descriptor, appropriately filled in.
208 : : * The descriptor and subsidiary data are allocated in the specified
209 : : * memory context, which must be suitably long-lived for the caller's
210 : : * purposes. If the returned descriptor has a snapshot associated
211 : : * with it, the caller must ensure that it also lives long enough,
212 : : * e.g. by calling RegisterSnapshotOnOwner
213 : : */
214 : : LargeObjectDesc *
7175 tgl@sss.pgh.pa.us 215 : 278 : inv_open(Oid lobjId, int flags, MemoryContext mcxt)
216 : : {
217 : : LargeObjectDesc *retval;
4461 heikki.linnakangas@i 218 : 278 : Snapshot snapshot = NULL;
219 : 278 : int descflags = 0;
220 : :
221 : : /*
222 : : * Historically, no difference is made between (INV_WRITE) and (INV_WRITE
223 : : * | INV_READ), the caller being allowed to read the large object
224 : : * descriptor in either case.
225 : : */
9036 bruce@momjian.us 226 [ + + ]: 278 : if (flags & INV_WRITE)
2960 tgl@sss.pgh.pa.us 227 : 93 : descflags |= IFS_WRLOCK | IFS_RDLOCK;
228 [ + + ]: 278 : if (flags & INV_READ)
229 : 200 : descflags |= IFS_RDLOCK;
230 : :
231 [ - + ]: 278 : if (descflags == 0)
4818 tgl@sss.pgh.pa.us 232 [ # # ]:UBC 0 : ereport(ERROR,
233 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
234 : : errmsg("invalid flags for opening a large object: %d",
235 : : flags)));
236 : :
237 : : /* Get snapshot. If write is requested, use an instantaneous snapshot. */
2960 tgl@sss.pgh.pa.us 238 [ + + ]:CBC 278 : if (descflags & IFS_WRLOCK)
239 : 93 : snapshot = NULL;
240 : : else
241 : 185 : snapshot = GetActiveSnapshot();
242 : :
243 : : /* Can't use LargeObjectExists here because we need to specify snapshot */
461 fujii@postgresql.org 244 [ + + ]: 278 : if (!LargeObjectExistsWithSnapshot(lobjId, snapshot))
7492 tgl@sss.pgh.pa.us 245 [ + - ]: 5 : ereport(ERROR,
246 : : (errcode(ERRCODE_UNDEFINED_OBJECT),
247 : : errmsg("large object %u does not exist", lobjId)));
248 : :
249 : : /* Apply permission checks, again specifying snapshot */
2960 250 [ + - ]: 273 : if ((descflags & IFS_RDLOCK) != 0)
251 : : {
252 [ + + + + ]: 537 : if (!lo_compat_privileges &&
253 : 264 : pg_largeobject_aclcheck_snapshot(lobjId,
254 : : GetUserId(),
255 : : ACL_SELECT,
256 : : snapshot) != ACLCHECK_OK)
257 [ + - ]: 21 : ereport(ERROR,
258 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
259 : : errmsg("permission denied for large object %u",
260 : : lobjId)));
261 : : }
262 [ + + ]: 252 : if ((descflags & IFS_WRLOCK) != 0)
263 : : {
264 [ + + + + ]: 156 : if (!lo_compat_privileges &&
265 : 75 : pg_largeobject_aclcheck_snapshot(lobjId,
266 : : GetUserId(),
267 : : ACL_UPDATE,
268 : : snapshot) != ACLCHECK_OK)
269 [ + - ]: 6 : ereport(ERROR,
270 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
271 : : errmsg("permission denied for large object %u",
272 : : lobjId)));
273 : : }
274 : :
275 : : /* OK to create a descriptor */
276 : 246 : retval = (LargeObjectDesc *) MemoryContextAlloc(mcxt,
277 : : sizeof(LargeObjectDesc));
278 : 246 : retval->id = lobjId;
279 : 246 : retval->offset = 0;
280 : 246 : retval->flags = descflags;
281 : :
282 : : /* caller sets if needed, not used by the functions in this file */
1505 heikki.linnakangas@i 283 : 246 : retval->subid = InvalidSubTransactionId;
284 : :
285 : : /*
286 : : * The snapshot (if any) is just the currently active snapshot. The
287 : : * caller will replace it with a longer-lived copy if needed.
288 : : */
4461 289 : 246 : retval->snapshot = snapshot;
290 : :
9969 bruce@momjian.us 291 : 246 : return retval;
292 : : }
293 : :
294 : : /*
295 : : * Closes a large object descriptor previously made by inv_open(), and
296 : : * releases the long-term memory used by it.
297 : : */
298 : : void
10327 299 : 231 : inv_close(LargeObjectDesc *obj_desc)
300 : : {
84 peter@eisentraut.org 301 [ - + ]:GNC 231 : Assert(obj_desc);
10328 bruce@momjian.us 302 :CBC 231 : pfree(obj_desc);
10753 scrappy@hub.org 303 : 231 : }
304 : :
305 : : /*
306 : : * Destroys an existing large object (not to be confused with a descriptor!)
307 : : *
308 : : * Note we expect caller to have done any required permissions check.
309 : : */
310 : : int
9504 bruce@momjian.us 311 : 44 : inv_drop(Oid lobjId)
312 : : {
313 : : ObjectAddress object;
314 : :
315 : : /*
316 : : * Delete any comments and dependencies on the large object
317 : : */
5850 itagaki.takahiro@gma 318 : 44 : object.classId = LargeObjectRelationId;
319 : 44 : object.objectId = lobjId;
320 : 44 : object.objectSubId = 0;
5074 rhaas@postgresql.org 321 : 44 : performDeletion(&object, DROP_CASCADE, 0);
322 : :
323 : : /*
324 : : * Advance command counter so that tuple removal will be seen by later
325 : : * large-object operations in this transaction.
326 : : */
9185 tgl@sss.pgh.pa.us 327 : 44 : CommandCounterIncrement();
328 : :
329 : : /* For historical reasons, we always return 1 on success. */
10328 bruce@momjian.us 330 : 44 : return 1;
331 : : }
332 : :
333 : : /*
334 : : * Determine size of a large object
335 : : *
336 : : * NOTE: LOs can contain gaps, just like Unix files. We actually return
337 : : * the offset of the last byte + 1.
338 : : */
339 : : static uint64
9185 tgl@sss.pgh.pa.us 340 : 69 : inv_getsize(LargeObjectDesc *obj_desc)
341 : : {
4819 ishii@postgresql.org 342 : 69 : uint64 lastbyte = 0;
343 : : ScanKeyData skey[1];
344 : : SysScanDesc sd;
345 : : HeapTuple tuple;
346 : :
84 peter@eisentraut.org 347 [ - + ]:GNC 69 : Assert(obj_desc);
348 : :
7812 tgl@sss.pgh.pa.us 349 :CBC 69 : open_lo_relation();
350 : :
8071 351 : 69 : ScanKeyInit(&skey[0],
352 : : Anum_pg_largeobject_loid,
353 : : BTEqualStrategyNumber, F_OIDEQ,
354 : : ObjectIdGetDatum(obj_desc->id));
355 : :
6458 356 : 69 : sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
357 : : obj_desc->snapshot, 1, skey);
358 : :
359 : : /*
360 : : * Because the pg_largeobject index is on both loid and pageno, but we
361 : : * constrain only loid, a backwards scan should visit all pages of the
362 : : * large object in reverse pageno order. So, it's sufficient to examine
363 : : * the first valid tuple (== last valid page).
364 : : */
5850 itagaki.takahiro@gma 365 : 69 : tuple = systable_getnext_ordered(sd, BackwardScanDirection);
366 [ + + ]: 69 : if (HeapTupleIsValid(tuple))
367 : : {
368 : : Form_pg_largeobject data;
369 : : bytea *datafield;
370 : : int len;
371 : : bool pfreeit;
372 : :
6607 bruce@momjian.us 373 [ - + ]: 60 : if (HeapTupleHasNulls(tuple)) /* paranoia */
6763 tgl@sss.pgh.pa.us 374 [ # # ]:UBC 0 : elog(ERROR, "null field found in pg_largeobject");
8612 tgl@sss.pgh.pa.us 375 :CBC 60 : data = (Form_pg_largeobject) GETSTRUCT(tuple);
4213 376 : 60 : getdatafield(data, &datafield, &len, &pfreeit);
377 : 60 : lastbyte = (uint64) data->pageno * LOBLKSIZE + len;
9185 378 [ + + ]: 60 : if (pfreeit)
379 : 21 : pfree(datafield);
380 : : }
381 : :
6458 382 : 69 : systable_endscan_ordered(sd);
383 : :
9185 384 : 69 : return lastbyte;
385 : : }
386 : :
387 : : int64
4819 ishii@postgresql.org 388 : 144 : inv_seek(LargeObjectDesc *obj_desc, int64 offset, int whence)
389 : : {
390 : : int64 newoffset;
391 : :
84 peter@eisentraut.org 392 [ - + ]:GNC 144 : Assert(obj_desc);
393 : :
394 : : /*
395 : : * We allow seek/tell if you have either read or write permission, so no
396 : : * need for a permission check here.
397 : : */
398 : :
399 : : /*
400 : : * Note: overflow in the additions is possible, but since we will reject
401 : : * negative results, we don't need any extra test for that.
402 : : */
9185 tgl@sss.pgh.pa.us 403 [ + + + - ]:CBC 144 : switch (whence)
404 : : {
405 : 66 : case SEEK_SET:
4818 406 : 66 : newoffset = offset;
9185 407 : 66 : break;
408 : 9 : case SEEK_CUR:
4818 409 : 9 : newoffset = obj_desc->offset + offset;
9185 410 : 9 : break;
411 : 69 : case SEEK_END:
4818 412 : 69 : newoffset = inv_getsize(obj_desc) + offset;
9185 413 : 69 : break;
9185 tgl@sss.pgh.pa.us 414 :UBC 0 : default:
4818 415 [ # # ]: 0 : ereport(ERROR,
416 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
417 : : errmsg("invalid whence setting: %d", whence)));
418 : : newoffset = 0; /* keep compiler quiet */
419 : : break;
420 : : }
421 : :
422 : : /*
423 : : * use errmsg_internal here because we don't want to expose INT64_FORMAT
424 : : * in translatable strings; doing better is not worth the trouble
425 : : */
4818 tgl@sss.pgh.pa.us 426 [ + - - + ]:CBC 144 : if (newoffset < 0 || newoffset > MAX_LARGE_OBJECT_SIZE)
4818 tgl@sss.pgh.pa.us 427 [ # # ]:UBC 0 : ereport(ERROR,
428 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
429 : : errmsg_internal("invalid large object seek target: " INT64_FORMAT,
430 : : newoffset)));
431 : :
4818 tgl@sss.pgh.pa.us 432 :CBC 144 : obj_desc->offset = newoffset;
433 : 144 : return newoffset;
434 : : }
435 : :
436 : : int64
10327 bruce@momjian.us 437 : 24 : inv_tell(LargeObjectDesc *obj_desc)
438 : : {
84 peter@eisentraut.org 439 [ - + ]:GNC 24 : Assert(obj_desc);
440 : :
441 : : /*
442 : : * We allow seek/tell if you have either read or write permission, so no
443 : : * need for a permission check here.
444 : : */
445 : :
9969 bruce@momjian.us 446 :CBC 24 : return obj_desc->offset;
447 : : }
448 : :
449 : : int
10327 450 : 721 : inv_read(LargeObjectDesc *obj_desc, char *buf, int nbytes)
451 : : {
9036 452 : 721 : int nread = 0;
453 : : int64 n;
454 : : int64 off;
455 : : int len;
456 : 721 : int32 pageno = (int32) (obj_desc->offset / LOBLKSIZE);
457 : : uint64 pageoff;
458 : : ScanKeyData skey[2];
459 : : SysScanDesc sd;
460 : : HeapTuple tuple;
461 : :
84 peter@eisentraut.org 462 [ - + ]:GNC 721 : Assert(obj_desc);
10328 bruce@momjian.us 463 [ - + ]:CBC 721 : Assert(buf != NULL);
464 : :
2960 tgl@sss.pgh.pa.us 465 [ - + ]: 721 : if ((obj_desc->flags & IFS_RDLOCK) == 0)
2960 tgl@sss.pgh.pa.us 466 [ # # ]:UBC 0 : ereport(ERROR,
467 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
468 : : errmsg("permission denied for large object %u",
469 : : obj_desc->id)));
470 : :
9185 tgl@sss.pgh.pa.us 471 [ + + ]:CBC 721 : if (nbytes <= 0)
9187 bruce@momjian.us 472 : 9 : return 0;
473 : :
7812 tgl@sss.pgh.pa.us 474 : 712 : open_lo_relation();
475 : :
8071 476 : 712 : ScanKeyInit(&skey[0],
477 : : Anum_pg_largeobject_loid,
478 : : BTEqualStrategyNumber, F_OIDEQ,
479 : : ObjectIdGetDatum(obj_desc->id));
480 : :
481 : 712 : ScanKeyInit(&skey[1],
482 : : Anum_pg_largeobject_pageno,
483 : : BTGreaterEqualStrategyNumber, F_INT4GE,
484 : : Int32GetDatum(pageno));
485 : :
6458 486 : 712 : sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
487 : : obj_desc->snapshot, 2, skey);
488 : :
489 [ + + ]: 5256 : while ((tuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
490 : : {
491 : : Form_pg_largeobject data;
492 : : bytea *datafield;
493 : : bool pfreeit;
494 : :
6607 bruce@momjian.us 495 [ - + ]: 5087 : if (HeapTupleHasNulls(tuple)) /* paranoia */
6763 tgl@sss.pgh.pa.us 496 [ # # ]:UBC 0 : elog(ERROR, "null field found in pg_largeobject");
8612 tgl@sss.pgh.pa.us 497 :CBC 5087 : data = (Form_pg_largeobject) GETSTRUCT(tuple);
498 : :
499 : : /*
500 : : * We expect the indexscan will deliver pages in order. However,
501 : : * there may be missing pages if the LO contains unwritten "holes". We
502 : : * want missing sections to read out as zeroes.
503 : : */
4819 ishii@postgresql.org 504 : 5087 : pageoff = ((uint64) data->pageno) * LOBLKSIZE;
9185 tgl@sss.pgh.pa.us 505 [ + + ]: 5087 : if (pageoff > obj_desc->offset)
506 : : {
507 : 6 : n = pageoff - obj_desc->offset;
508 : 6 : n = (n <= (nbytes - nread)) ? n : (nbytes - nread);
509 [ - + - - : 6 : MemSet(buf + nread, 0, n);
- - - - -
- ]
510 : 6 : nread += n;
511 : 6 : obj_desc->offset += n;
512 : : }
513 : :
514 [ + + ]: 5087 : if (nread < nbytes)
515 : : {
516 [ - + ]: 5084 : Assert(obj_desc->offset >= pageoff);
517 : 5084 : off = (int) (obj_desc->offset - pageoff);
518 [ + - - + ]: 5084 : Assert(off >= 0 && off < LOBLKSIZE);
519 : :
4213 520 : 5084 : getdatafield(data, &datafield, &len, &pfreeit);
9185 521 [ + + ]: 5084 : if (len > off)
522 : : {
523 : 5026 : n = len - off;
524 : 5026 : n = (n <= (nbytes - nread)) ? n : (nbytes - nread);
525 : 5026 : memcpy(buf + nread, VARDATA(datafield) + off, n);
526 : 5026 : nread += n;
527 : 5026 : obj_desc->offset += n;
528 : : }
529 [ + + ]: 5084 : if (pfreeit)
530 : 5046 : pfree(datafield);
531 : : }
532 : :
533 [ + + ]: 5087 : if (nread >= nbytes)
534 : 543 : break;
535 : : }
536 : :
6458 537 : 712 : systable_endscan_ordered(sd);
538 : :
9969 bruce@momjian.us 539 : 712 : return nread;
540 : : }
541 : :
542 : : int
7041 543 : 792 : inv_write(LargeObjectDesc *obj_desc, const char *buf, int nbytes)
544 : : {
9036 545 : 792 : int nwritten = 0;
546 : : int n;
547 : : int off;
548 : : int len;
549 : 792 : int32 pageno = (int32) (obj_desc->offset / LOBLKSIZE);
550 : : ScanKeyData skey[2];
551 : : SysScanDesc sd;
552 : : HeapTuple oldtuple;
553 : : Form_pg_largeobject olddata;
554 : : bool neednextpage;
555 : : bytea *datafield;
556 : : bool pfreeit;
557 : : union
558 : : {
559 : : alignas(int32) bytea hdr;
560 : : /* this is to make the union big enough for a LO data chunk: */
561 : : char data[LOBLKSIZE + VARHDRSZ];
134 msawada@postgresql.o 562 :GNC 792 : } workbuf = {0};
6868 tgl@sss.pgh.pa.us 563 :CBC 792 : char *workb = VARDATA(&workbuf.hdr);
564 : : HeapTuple newtup;
565 : : Datum values[Natts_pg_largeobject];
566 : : bool nulls[Natts_pg_largeobject];
567 : : bool replace[Natts_pg_largeobject];
568 : : CatalogIndexState indstate;
569 : :
84 peter@eisentraut.org 570 [ - + ]:GNC 792 : Assert(obj_desc);
10328 bruce@momjian.us 571 [ - + ]:CBC 792 : Assert(buf != NULL);
572 : :
573 : : /* enforce writability because snapshot is probably wrong otherwise */
2960 tgl@sss.pgh.pa.us 574 [ - + ]: 792 : if ((obj_desc->flags & IFS_WRLOCK) == 0)
2960 tgl@sss.pgh.pa.us 575 [ # # ]:UBC 0 : ereport(ERROR,
576 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
577 : : errmsg("permission denied for large object %u",
578 : : obj_desc->id)));
579 : :
9185 tgl@sss.pgh.pa.us 580 [ - + ]:CBC 792 : if (nbytes <= 0)
9185 tgl@sss.pgh.pa.us 581 :UBC 0 : return 0;
582 : :
583 : : /* this addition can't overflow because nbytes is only int32 */
4819 ishii@postgresql.org 584 [ - + ]:CBC 792 : if ((nbytes + obj_desc->offset) > MAX_LARGE_OBJECT_SIZE)
4818 tgl@sss.pgh.pa.us 585 [ # # ]:UBC 0 : ereport(ERROR,
586 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
587 : : errmsg("invalid large object write request size: %d",
588 : : nbytes)));
589 : :
7812 tgl@sss.pgh.pa.us 590 :CBC 792 : open_lo_relation();
591 : :
592 : 792 : indstate = CatalogOpenIndexes(lo_heap_r);
593 : :
8071 594 : 792 : ScanKeyInit(&skey[0],
595 : : Anum_pg_largeobject_loid,
596 : : BTEqualStrategyNumber, F_OIDEQ,
597 : : ObjectIdGetDatum(obj_desc->id));
598 : :
599 : 792 : ScanKeyInit(&skey[1],
600 : : Anum_pg_largeobject_pageno,
601 : : BTGreaterEqualStrategyNumber, F_INT4GE,
602 : : Int32GetDatum(pageno));
603 : :
6458 604 : 792 : sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
605 : : obj_desc->snapshot, 2, skey);
606 : :
8612 607 : 792 : oldtuple = NULL;
9185 608 : 792 : olddata = NULL;
609 : 792 : neednextpage = true;
610 : :
9187 bruce@momjian.us 611 [ + + ]: 4782 : while (nwritten < nbytes)
612 : : {
613 : : /*
614 : : * If possible, get next pre-existing page of the LO. We expect the
615 : : * indexscan will deliver these in order --- but there may be holes.
616 : : */
9185 tgl@sss.pgh.pa.us 617 [ + + ]: 3990 : if (neednextpage)
618 : : {
6458 619 [ + + ]: 795 : if ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
620 : : {
3101 621 [ - + ]: 12 : if (HeapTupleHasNulls(oldtuple)) /* paranoia */
6763 tgl@sss.pgh.pa.us 622 [ # # ]:UBC 0 : elog(ERROR, "null field found in pg_largeobject");
8612 tgl@sss.pgh.pa.us 623 :CBC 12 : olddata = (Form_pg_largeobject) GETSTRUCT(oldtuple);
624 [ - + ]: 12 : Assert(olddata->pageno >= pageno);
625 : : }
9185 626 : 795 : neednextpage = false;
627 : : }
628 : :
629 : : /*
630 : : * If we have a pre-existing page, see if it is the page we want to
631 : : * write, or a later one.
632 : : */
633 [ + + + - ]: 3990 : if (olddata != NULL && olddata->pageno == pageno)
634 : : {
635 : : /*
636 : : * Update an existing page with fresh data.
637 : : *
638 : : * First, load old data into workbuf
639 : : */
4213 640 : 12 : getdatafield(olddata, &datafield, &len, &pfreeit);
9185 641 : 12 : memcpy(workb, VARDATA(datafield), len);
642 [ + + ]: 12 : if (pfreeit)
643 : 9 : pfree(datafield);
644 : :
645 : : /*
646 : : * Fill any hole
647 : : */
648 : 12 : off = (int) (obj_desc->offset % LOBLKSIZE);
649 [ - + ]: 12 : if (off > len)
9185 tgl@sss.pgh.pa.us 650 [ # # # # :UBC 0 : MemSet(workb + len, 0, off - len);
# # # # #
# ]
651 : :
652 : : /*
653 : : * Insert appropriate portion of new data
654 : : */
9185 tgl@sss.pgh.pa.us 655 :CBC 12 : n = LOBLKSIZE - off;
656 : 12 : n = (n <= (nbytes - nwritten)) ? n : (nbytes - nwritten);
657 : 12 : memcpy(workb + off, buf + nwritten, n);
658 : 12 : nwritten += n;
659 : 12 : obj_desc->offset += n;
660 : 12 : off += n;
661 : : /* compute valid length of new page */
662 : 12 : len = (len >= off) ? len : off;
6868 663 : 12 : SET_VARSIZE(&workbuf.hdr, len + VARHDRSZ);
664 : :
665 : : /*
666 : : * Form and insert updated tuple
667 : : */
9185 668 : 12 : memset(values, 0, sizeof(values));
6254 669 : 12 : memset(nulls, false, sizeof(nulls));
670 : 12 : memset(replace, false, sizeof(replace));
9033 671 : 12 : values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
6254 672 : 12 : replace[Anum_pg_largeobject_data - 1] = true;
673 : 12 : newtup = heap_modify_tuple(oldtuple, RelationGetDescr(lo_heap_r),
674 : : values, nulls, replace);
3241 675 : 12 : CatalogTupleUpdateWithInfo(lo_heap_r, &newtup->t_self, newtup,
676 : : indstate);
9185 677 : 12 : heap_freetuple(newtup);
678 : :
679 : : /*
680 : : * We're done with this old page.
681 : : */
8612 682 : 12 : oldtuple = NULL;
9185 683 : 12 : olddata = NULL;
684 : 12 : neednextpage = true;
685 : : }
686 : : else
687 : : {
688 : : /*
689 : : * Write a brand new page.
690 : : *
691 : : * First, fill any hole
692 : : */
693 : 3978 : off = (int) (obj_desc->offset % LOBLKSIZE);
694 [ + + ]: 3978 : if (off > 0)
695 [ - + - - : 3 : MemSet(workb, 0, off);
- - - - -
- ]
696 : :
697 : : /*
698 : : * Insert appropriate portion of new data
699 : : */
700 : 3978 : n = LOBLKSIZE - off;
701 : 3978 : n = (n <= (nbytes - nwritten)) ? n : (nbytes - nwritten);
702 : 3978 : memcpy(workb + off, buf + nwritten, n);
703 : 3978 : nwritten += n;
704 : 3978 : obj_desc->offset += n;
705 : : /* compute valid length of new page */
706 : 3978 : len = off + n;
6868 707 : 3978 : SET_VARSIZE(&workbuf.hdr, len + VARHDRSZ);
708 : :
709 : : /*
710 : : * Form and insert updated tuple
711 : : */
9185 712 : 3978 : memset(values, 0, sizeof(values));
6254 713 : 3978 : memset(nulls, false, sizeof(nulls));
9185 714 : 3978 : values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(obj_desc->id);
715 : 3978 : values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(pageno);
9033 716 : 3978 : values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
6254 717 : 3978 : newtup = heap_form_tuple(lo_heap_r->rd_att, values, nulls);
3241 718 : 3978 : CatalogTupleInsertWithInfo(lo_heap_r, newtup, indstate);
9185 719 : 3978 : heap_freetuple(newtup);
720 : : }
721 : 3990 : pageno++;
722 : : }
723 : :
6458 724 : 792 : systable_endscan_ordered(sd);
725 : :
8535 726 : 792 : CatalogCloseIndexes(indstate);
727 : :
728 : : /*
729 : : * Advance command counter so that my tuple updates will be seen by later
730 : : * large-object operations in this transaction.
731 : : */
9185 732 : 792 : CommandCounterIncrement();
733 : :
9187 bruce@momjian.us 734 : 792 : return nwritten;
735 : : }
736 : :
737 : : void
4819 ishii@postgresql.org 738 : 21 : inv_truncate(LargeObjectDesc *obj_desc, int64 len)
739 : : {
6864 bruce@momjian.us 740 : 21 : int32 pageno = (int32) (len / LOBLKSIZE);
741 : : int32 off;
742 : : ScanKeyData skey[2];
743 : : SysScanDesc sd;
744 : : HeapTuple oldtuple;
745 : : Form_pg_largeobject olddata;
746 : : union
747 : : {
748 : : alignas(int32) bytea hdr;
749 : : /* this is to make the union big enough for a LO data chunk: */
750 : : char data[LOBLKSIZE + VARHDRSZ];
134 msawada@postgresql.o 751 :GNC 21 : } workbuf = {0};
6607 bruce@momjian.us 752 :CBC 21 : char *workb = VARDATA(&workbuf.hdr);
753 : : HeapTuple newtup;
754 : : Datum values[Natts_pg_largeobject];
755 : : bool nulls[Natts_pg_largeobject];
756 : : bool replace[Natts_pg_largeobject];
757 : : CatalogIndexState indstate;
758 : :
84 peter@eisentraut.org 759 [ - + ]:GNC 21 : Assert(obj_desc);
760 : :
761 : : /* enforce writability because snapshot is probably wrong otherwise */
2960 tgl@sss.pgh.pa.us 762 [ - + ]:CBC 21 : if ((obj_desc->flags & IFS_WRLOCK) == 0)
2960 tgl@sss.pgh.pa.us 763 [ # # ]:UBC 0 : ereport(ERROR,
764 : : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
765 : : errmsg("permission denied for large object %u",
766 : : obj_desc->id)));
767 : :
768 : : /*
769 : : * use errmsg_internal here because we don't want to expose INT64_FORMAT
770 : : * in translatable strings; doing better is not worth the trouble
771 : : */
4818 tgl@sss.pgh.pa.us 772 [ + - - + ]:CBC 21 : if (len < 0 || len > MAX_LARGE_OBJECT_SIZE)
4818 tgl@sss.pgh.pa.us 773 [ # # ]:UBC 0 : ereport(ERROR,
774 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
775 : : errmsg_internal("invalid large object truncation target: " INT64_FORMAT,
776 : : len)));
777 : :
6864 bruce@momjian.us 778 :CBC 21 : open_lo_relation();
779 : :
780 : 21 : indstate = CatalogOpenIndexes(lo_heap_r);
781 : :
782 : : /*
783 : : * Set up to find all pages with desired loid and pageno >= target
784 : : */
785 : 21 : ScanKeyInit(&skey[0],
786 : : Anum_pg_largeobject_loid,
787 : : BTEqualStrategyNumber, F_OIDEQ,
788 : : ObjectIdGetDatum(obj_desc->id));
789 : :
790 : 21 : ScanKeyInit(&skey[1],
791 : : Anum_pg_largeobject_pageno,
792 : : BTGreaterEqualStrategyNumber, F_INT4GE,
793 : : Int32GetDatum(pageno));
794 : :
6458 tgl@sss.pgh.pa.us 795 : 21 : sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
796 : : obj_desc->snapshot, 2, skey);
797 : :
798 : : /*
799 : : * If possible, get the page the truncation point is in. The truncation
800 : : * point may be beyond the end of the LO or in a hole.
801 : : */
6864 bruce@momjian.us 802 : 21 : olddata = NULL;
6458 tgl@sss.pgh.pa.us 803 [ + + ]: 21 : if ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
804 : : {
3101 805 [ - + ]: 12 : if (HeapTupleHasNulls(oldtuple)) /* paranoia */
6763 tgl@sss.pgh.pa.us 806 [ # # ]:UBC 0 : elog(ERROR, "null field found in pg_largeobject");
6864 bruce@momjian.us 807 :CBC 12 : olddata = (Form_pg_largeobject) GETSTRUCT(oldtuple);
808 [ - + ]: 12 : Assert(olddata->pageno >= pageno);
809 : : }
810 : :
811 : : /*
812 : : * If we found the page of the truncation point we need to truncate the
813 : : * data in it. Otherwise if we're in a hole, we need to create a page to
814 : : * mark the end of data.
815 : : */
816 [ + + + + ]: 21 : if (olddata != NULL && olddata->pageno == pageno)
817 : 6 : {
818 : : /* First, load old data into workbuf */
819 : : bytea *datafield;
820 : : int pagelen;
821 : : bool pfreeit;
822 : :
4213 tgl@sss.pgh.pa.us 823 : 6 : getdatafield(olddata, &datafield, &pagelen, &pfreeit);
6864 bruce@momjian.us 824 : 6 : memcpy(workb, VARDATA(datafield), pagelen);
825 [ + + ]: 6 : if (pfreeit)
6607 826 : 3 : pfree(datafield);
827 : :
828 : : /*
829 : : * Fill any hole
830 : : */
6864 831 : 6 : off = len % LOBLKSIZE;
832 [ + + ]: 6 : if (off > pagelen)
6607 833 [ + - - + : 3 : MemSet(workb + pagelen, 0, off - pagelen);
- - - - -
- ]
834 : :
835 : : /* compute length of new page */
6864 836 : 6 : SET_VARSIZE(&workbuf.hdr, off + VARHDRSZ);
837 : :
838 : : /*
839 : : * Form and insert updated tuple
840 : : */
841 : 6 : memset(values, 0, sizeof(values));
6254 tgl@sss.pgh.pa.us 842 : 6 : memset(nulls, false, sizeof(nulls));
843 : 6 : memset(replace, false, sizeof(replace));
6864 bruce@momjian.us 844 : 6 : values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
6254 tgl@sss.pgh.pa.us 845 : 6 : replace[Anum_pg_largeobject_data - 1] = true;
846 : 6 : newtup = heap_modify_tuple(oldtuple, RelationGetDescr(lo_heap_r),
847 : : values, nulls, replace);
3241 848 : 6 : CatalogTupleUpdateWithInfo(lo_heap_r, &newtup->t_self, newtup,
849 : : indstate);
6864 bruce@momjian.us 850 : 6 : heap_freetuple(newtup);
851 : : }
852 : : else
853 : : {
854 : : /*
855 : : * If the first page we found was after the truncation point, we're in
856 : : * a hole that we'll fill, but we need to delete the later page
857 : : * because the loop below won't visit it again.
858 : : */
5439 tgl@sss.pgh.pa.us 859 [ + + ]: 15 : if (olddata != NULL)
860 : : {
861 [ - + ]: 6 : Assert(olddata->pageno > pageno);
3241 862 : 6 : CatalogTupleDelete(lo_heap_r, &oldtuple->t_self);
863 : : }
864 : :
865 : : /*
866 : : * Write a brand new page.
867 : : *
868 : : * Fill the hole up to the truncation point
869 : : */
6864 bruce@momjian.us 870 : 15 : off = len % LOBLKSIZE;
871 [ + - ]: 15 : if (off > 0)
872 [ - + - - : 15 : MemSet(workb, 0, off);
- - - - -
- ]
873 : :
874 : : /* compute length of new page */
875 : 15 : SET_VARSIZE(&workbuf.hdr, off + VARHDRSZ);
876 : :
877 : : /*
878 : : * Form and insert new tuple
879 : : */
880 : 15 : memset(values, 0, sizeof(values));
6254 tgl@sss.pgh.pa.us 881 : 15 : memset(nulls, false, sizeof(nulls));
6864 bruce@momjian.us 882 : 15 : values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(obj_desc->id);
883 : 15 : values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(pageno);
884 : 15 : values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&workbuf);
6254 tgl@sss.pgh.pa.us 885 : 15 : newtup = heap_form_tuple(lo_heap_r->rd_att, values, nulls);
3241 886 : 15 : CatalogTupleInsertWithInfo(lo_heap_r, newtup, indstate);
6864 bruce@momjian.us 887 : 15 : heap_freetuple(newtup);
888 : : }
889 : :
890 : : /*
891 : : * Delete any pages after the truncation point. If the initial search
892 : : * didn't find a page, then of course there's nothing more to do.
893 : : */
5439 tgl@sss.pgh.pa.us 894 [ + + ]: 21 : if (olddata != NULL)
895 : : {
896 [ + + ]: 15 : while ((oldtuple = systable_getnext_ordered(sd, ForwardScanDirection)) != NULL)
897 : : {
3241 898 : 3 : CatalogTupleDelete(lo_heap_r, &oldtuple->t_self);
899 : : }
900 : : }
901 : :
6458 902 : 21 : systable_endscan_ordered(sd);
903 : :
6864 bruce@momjian.us 904 : 21 : CatalogCloseIndexes(indstate);
905 : :
906 : : /*
907 : : * Advance command counter so that tuple updates will be seen by later
908 : : * large-object operations in this transaction.
909 : : */
910 : 21 : CommandCounterIncrement();
911 : 21 : }
|
