DocumentSource.js 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273
  1. var DocumentSource = module.exports = (function(){
  2. // CONSTRUCTOR
  3. /**
  4. * A base class for all document sources
  5. *
  6. * @class DocumentSource
  7. * @namespace munge.pipeline.documentsource
  8. * @module munge
  9. * @constructor
  10. * @param {ExpressionContext}
  11. **/
  12. var klass = module.exports = DocumentSource = function DocumentSource(/*pCtx*/){
  13. if(arguments.length !== 0) throw new Error("zero args expected");
  14. /*
  15. Most DocumentSources have an underlying source they get their data
  16. from. This is a convenience for them.
  17. The default implementation of setSource() sets this; if you don't
  18. need a source, override that to verify(). The default is to
  19. verify() if this has already been set.
  20. */
  21. this.pSource = null;
  22. /*
  23. The zero-based user-specified pipeline step. Used for diagnostics.
  24. Will be set to -1 for artificial pipeline steps that were not part
  25. of the original user specification.
  26. */
  27. this.step = -1;
  28. //we dont need this because we are not sharding
  29. //intrusive_ptr<ExpressionContext> pExpCtx;
  30. //this.pExpCtx = pCtx;
  31. /*
  32. for explain: # of rows returned by this source
  33. This is *not* unsigned so it can be passed to JSONObjBuilder.append().
  34. */
  35. this.nRowsOut = 0;
  36. }, base = Object, proto = klass.prototype = Object.create(base.prototype, {constructor:{value:klass}});
  37. /*
  38. class DocumentSource :
  39. public IntrusiveCounterUnsigned,
  40. public StringWriter {
  41. public:
  42. virtual ~DocumentSource();
  43. // virtuals from StringWriter
  44. virtual void writeString(stringstream &ss) const;
  45. */
  46. /**
  47. * Set the step for a user-specified pipeline step.
  48. *
  49. * @method setPipelineStep
  50. * @param {Number} step number 0 to n.
  51. **/
  52. proto.setPipelineStep = function setPipelineStep(step) {
  53. this.step = step;
  54. };
  55. /**
  56. * Get the user-specified pipeline step.
  57. *
  58. * @method getPipelineStep
  59. * @returns {Number} step
  60. **/
  61. proto.getPipelineStep = function getPipelineStep() {
  62. return this.step;
  63. };
  64. /**
  65. * Is the source at EOF?
  66. *
  67. * @method eof
  68. **/
  69. proto.eof = function eof() {
  70. throw new Error("not implemented");
  71. };
  72. /**
  73. * Advance the state of the DocumentSource so that it will return the next Document.
  74. * The default implementation returns false, after checking for interrupts.
  75. * Derived classes can call the default implementation in their own implementations in order to check for interrupts.
  76. *
  77. * @method advance
  78. * @returns {Boolean} whether there is another document to fetch, i.e., whether or not getCurrent() will succeed. This default implementation always returns false.
  79. **/
  80. proto.advance = function advance() {
  81. //pExpCtx->checkForInterrupt(); // might not return
  82. return false;
  83. };
  84. /**
  85. * some implementations do the equivalent of verify(!eof()) so check eof() first
  86. *
  87. * @method getCurrent
  88. * @returns {Document} the current Document without advancing
  89. **/
  90. proto.getCurrent = function getCurrent() {
  91. throw new Error("not implemented");
  92. };
  93. /**
  94. * Inform the source that it is no longer needed and may release its resources. After
  95. * dispose() is called the source must still be able to handle iteration requests, but may
  96. * become eof().
  97. * NOTE: For proper mutex yielding, dispose() must be called on any DocumentSource that will
  98. * not be advanced until eof(), see SERVER-6123.
  99. *
  100. * @method dispose
  101. **/
  102. proto.dispose = function dispose() {
  103. if ( this.pSource ) {
  104. // This is required for the DocumentSourceCursor to release its read lock, see
  105. // SERVER-6123.
  106. this.pSource.dispose();
  107. }
  108. };
  109. /**
  110. * Get the source's name.
  111. *
  112. * @method getSourceName
  113. * @returns {String} the string name of the source as a constant string; this is static, and there's no need to worry about adopting it
  114. **/
  115. proto.getSourceName = function getSourceName() {
  116. return "[UNKNOWN]";
  117. };
  118. /**
  119. * Set the underlying source this source should use to get Documents
  120. * from.
  121. * It is an error to set the source more than once. This is to
  122. * prevent changing sources once the original source has been started;
  123. * this could break the state maintained by the DocumentSource.
  124. * This pointer is not reference counted because that has led to
  125. * some circular references. As a result, this doesn't keep
  126. * sources alive, and is only intended to be used temporarily for
  127. * the lifetime of a Pipeline::run().
  128. *
  129. * @method setSource
  130. * @param {DocumentSource} pSource the underlying source to use
  131. **/
  132. proto.setSource = function setSource(pTheSource) {
  133. if(this.pSource){
  134. throw new Error("It is an error to set the source more than once");
  135. }
  136. this.pSource = pTheSource;
  137. };
  138. /**
  139. * Attempt to coalesce this DocumentSource with its successor in the
  140. * document processing pipeline. If successful, the successor
  141. * DocumentSource should be removed from the pipeline and discarded.
  142. * If successful, this operation can be applied repeatedly, in an
  143. * attempt to coalesce several sources together.
  144. * The default implementation is to do nothing, and return false.
  145. *
  146. * @method coalesce
  147. * @param {DocumentSource} pNextSource the next source in the document processing chain.
  148. * @returns {Boolean} whether or not the attempt to coalesce was successful or not; if the attempt was not successful, nothing has been changed
  149. **/
  150. proto.coalesce = function coalesce(pNextSource) {
  151. return false;
  152. };
  153. /**
  154. * Optimize the pipeline operation, if possible. This is a local
  155. * optimization that only looks within this DocumentSource. For best
  156. * results, first coalesce compatible sources using coalesce().
  157. * This is intended for any operations that include expressions, and
  158. * provides a hook for those to optimize those operations.
  159. * The default implementation is to do nothing.
  160. *
  161. * @method optimize
  162. **/
  163. proto.optimize = function optimize(pNextSource) {
  164. };
  165. klass.GetDepsReturn = {
  166. NOT_SUPPORTED:"NOT_SUPPORTED", // This means the set should be ignored
  167. EXHAUSTIVE:"EXHAUSTIVE", // This means that everything needed should be in the set
  168. SEE_NEXT:"SEE_NEXT", // Add the next Source's deps to the set
  169. };
  170. /**
  171. * Get the fields this operation needs to do its job.
  172. * Deps should be in "a.b.c" notation
  173. *
  174. * @method getDependencies
  175. * @param {Object} deps set (unique array) of strings
  176. * @returns DocumentSource.GetDepsReturn
  177. **/
  178. proto.getDependencies = function getDependencies(deps) {
  179. return klass.GetDepsReturn.NOT_SUPPORTED;
  180. };
  181. /**
  182. * This takes dependencies from getDependencies and
  183. * returns a projection that includes all of them
  184. *
  185. * @method depsToProjection
  186. * @param {Object} deps set (unique array) of strings
  187. * @returns {Object} JSONObj
  188. **/
  189. klass.depsToProjection = function depsToProjection(deps) {
  190. var bb = {};
  191. if (deps._id === undefined)
  192. bb._id = 0;
  193. var last = "";
  194. Object.keys(deps).forEach(function(it){
  195. if (last !== "" && it.slice(0, last.length) === last){
  196. // we are including a parent of *it so we don't need to
  197. // include this field explicitly. In fact, due to
  198. // SERVER-6527 if we included this field, the parent
  199. // wouldn't be fully included.
  200. return;
  201. }
  202. last = it + ".";
  203. bb[it] = 1;
  204. });
  205. return bb;
  206. };
  207. /**
  208. * Add the DocumentSource to the array builder.
  209. * The default implementation calls sourceToJson() in order to
  210. * convert the inner part of the object which will be added to the
  211. * array being built here.
  212. *
  213. * @method addToJsonArray
  214. * @param {Array} pBuilder JSONArrayBuilder: the array builder to add the operation to.
  215. * @param {Boolean} explain create explain output
  216. * @returns {Object}
  217. **/
  218. proto.addToJsonArray = function addToJsonArray(pBuilder, explain) {
  219. pBuilder.push(this.sourceToJson({}, explain));
  220. };
  221. /**
  222. * Create an object that represents the document source. The object
  223. * will have a single field whose name is the source's name. This
  224. * will be used by the default implementation of addToJsonArray()
  225. * to add this object to a pipeline being represented in JSON.
  226. *
  227. * @method sourceToJson
  228. * @param {Object} pBuilder JSONObjBuilder: a blank object builder to write to
  229. * @param {Boolean} explain create explain output
  230. **/
  231. proto.sourceToJson = function sourceToJson(pBuilder, explain) {
  232. throw new Error("not implemented");
  233. };
  234. /**
  235. * Reset the document source so that it is ready for a new stream of data.
  236. * Note that this is a deviation from the mongo implementation.
  237. *
  238. * @method reset
  239. **/
  240. proto.reset = function reset(){
  241. throw new Error("not implemented");
  242. };
  243. return klass;
  244. })();