| 1193 | |
| 1194 | |
| 1195 | static bool |
| 1196 | PersistentDestReceiveCore(pgbson *resultBson, |
| 1197 | PersistentTupleDestReceiver *receiver) |
| 1198 | { |
| 1199 | BsonStoreTupleDestReceiverBase *base = &receiver->base; |
| 1200 | uint32_t datumSize = VARSIZE_ANY_EXHDR(resultBson); |
| 1201 | |
| 1202 | /* if the new total size is > Max Bson Size */ |
| 1203 | if (datumSize > BSON_MAX_ALLOWED_SIZE) |
| 1204 | { |
| 1205 | ereport(ERROR, (errcode(ERRCODE_DOCUMENTDB_BSONOBJECTTOOLARGE), |
| 1206 | errmsg("Size %u is larger than MaxDocumentSize %u", |
| 1207 | datumSize, BSON_MAX_ALLOWED_SIZE))); |
| 1208 | } |
| 1209 | |
| 1210 | int64_t totalSize = base->currentAccumulatedSize + datumSize + |
| 1211 | PER_DOC_OVERHEAD; |
| 1212 | |
| 1213 | /* we need to allow at least 1 tuple per response. */ |
| 1214 | bool sizeLimitReached = (totalSize >= BSON_MAX_ALLOWED_SIZE && |
| 1215 | base->numRowsFetched > 0); |
| 1216 | |
| 1217 | if (sizeLimitReached || |
| 1218 | (base->numRowsFetched >= (uint32_t) base->batchSize)) |
| 1219 | { |
| 1220 | /* We exhausted the current batch. We need to either persist or move on */ |
| 1221 | if (receiver->closeCursor) |
| 1222 | { |
| 1223 | /* We need to close the cursor stop - no point enumerating any further */ |
| 1224 | return false; |
| 1225 | } |
| 1226 | else if (UseFileBasedPersistedCursors) |
| 1227 | { |
| 1228 | if (receiver->cursorFileState == NULL) |
| 1229 | { |
| 1230 | MemoryContext oldContext = MemoryContextSwitchTo( |
| 1231 | base->writerContext); |
| 1232 | receiver->cursorFileState = CreateCursorFile( |
| 1233 | receiver->cursorName); |
| 1234 | MemoryContextSwitchTo(oldContext); |
| 1235 | } |
| 1236 | |
| 1237 | /* Dump the tuple into the cursor state */ |
| 1238 | WriteToCursorFile(receiver->cursorFileState, resultBson); |
| 1239 | } |
| 1240 | else |
| 1241 | { |
| 1242 | ereport(ERROR, (errcode(ERRCODE_DOCUMENTDB_INTERNALERROR), |
| 1243 | errmsg( |
| 1244 | "Cursor based paging with DestReceiver is not supported yet - this codepath should not be hit"))); |
| 1245 | } |
| 1246 | } |
| 1247 | else |
| 1248 | { |
| 1249 | /* We need to create a persistent hold store and dump the tuple there. */ |
| 1250 | MemoryContext oldContext = MemoryContextSwitchTo(base->writerContext); |
| 1251 | PgbsonArrayWriterWriteDocument(base->writer, resultBson); |
| 1252 | MemoryContextSwitchTo(oldContext); |
no test coverage detected