<commit_before>"use strict";

(function() {
  var castor = require('../castor-client');
  var Q = require('q');
  var Query = require('./Query');
  var Field = require('./Field');
  var Schema = require('./Schema');
  
  /**
   * Get class
   * 
   * Usage example:
   * 
   * castor.get('table')
   *   .fields(['field', 'field'])
   *   .filter('field', 'uuid', '...')
   *   .orderBy('field', 'asc')
   *   .limit(10)
   *   .cache(60)
   *   .consistency(castor.CONSISTENCY_QUORUM)
   * .then(...
   */
  var Get = function(transport, schema, keyspace, table) {
    this._transport = transport;
    this._schema = schema.get(table);
    this._schemaFull = schema;
    this._keyspace = keyspace;
    this._table = table;
    this._fields = [];
    this._filter = [];
    this._orderBy = null;
    this._orderByDirection = 'asc';
    this._limit = null;
    this._allowFiltering = false;
    this._consistency = 0x0004;
    this._joins = [];
  };
  
  Get.prototype.fields = function(fields) {
    this._fields = fields;
    return this;
  };
  
  Get.prototype.filter = function(name, value, operator) {
    if (typeof operator === 'undefined') {
      operator = '=';
    }
    if (operator != '=' && operator != '<' && operator != '>' && operator != '<=' && operator != '>=') {
      throw Error('Unknown operator: ' + operator);
    }
    this._filter.push({name: name, value: value, operator: operator});
    return this;
  };
  
  Get.prototype.orderBy = function(name, direction) {
    this._orderBy = name;
    if (direction !== undefined) {
      this._orderByDirection = direction;
    }
    return this;
  };
  
  Get.prototype.limit = function(limit) {
    this._limit = limit;
    return this;
  };
  
  Get.prototype.allowFiltering = function(allow) {
    if (typeof allow === 'undefined') {
      allow = true;
    }
    this._allowFiltering = allow;
    return this;
  };
  
  Get.prototype.consistency = function(consistency) {
    this._consistency = consistency;
    return this;
  };
  
  Get.prototype.join = function(leftField, rightField, fields) {
    // Right field is in format "table" or "table.field". The fieldname is
    // identical to the left field if not specified.
    var parts = rightField.split('.');
    var rightTable = parts[0];
    rightField = parts.length > 1 ? parts[1] : leftField;
    this._joins.push({
      leftField: leftField,
      rightTable: rightTable,
      rightField: rightField,
      fields: fields
    });
    return this;
  };
  
  Get.prototype.execute = function() {
    var defer = Q.defer();
    var self = this;
    
    Q.when(this._schema).then(function(schema) {
      var cql = 'SELECT ';
      if (self._fields.length) {
        cql = cql + self._fields.join(', ');
      }
      else {
        cql = cql + '*';
      }
      cql = cql + ' FROM ' + self._keyspace + '.' + self._table;
      if (self._filter.length) {
        for (var i = 0; i < self._filter.length; ++i) {
          var type = schema.columns[self._filter[i].name];
          if (typeof type === 'undefined') {
            throw Error('Unknown field ' + self._filter[i].name + ' in filter');
          }
          var filter = new Field(self._filter[i].name, type, self._filter[i].value, self._filter[i].operator);
          cql = cql + (i == 0 ? ' WHERE ' : ' AND ');
          cql = cql + filter.toString();
        }
      }
      if (self._orderBy !== null) {
        cql = cql + ' ORDER BY ' + self._orderBy + ' ' + self._orderByDirection.toUpperCase();
      }
      if (self._limit !== null) {
        cql = cql + ' LIMIT ' + self._limit;
      }
      if (self._allowFiltering) {
        cql = cql + ' ALLOW FILTERING';
      }
      if (self._joins.length) {
        self._executeWithJoins(cql, defer);
      }
      else {
        new Query(self._transport, cql, self._consistency).execute(defer);
      }
    }).fail(function(error) {
      defer.reject(error);
    });
    
    return defer.promise;
  };
  
  Get.prototype.then = function(callback) {
    return this.execute().then(callback);
  };
  
  Get.prototype._executeWithJoins = function(cql, defer) {
    var self = this;
    new Query(self._transport, cql, self._consistency).then(function(rows) {
      var joins = [];
      self._joins.forEach(function(join) {
        joins.push(self._executeJoin(rows, join));
      });
      Q.all(joins).then(function(results) {
        results.forEach(function(result) {
          Object.keys(result.columns).forEach(function(column) {
            rows.addColumn(result.columns[column], result.values[column]);
          });
        });
        defer.resolve(rows);
      }).fail(function(error) {
        defer.reject(error);
      }).done();
    });
  };
  
  Get.prototype._executeJoin = function(rows, join) {
    var self = this;
    var defer = Q.defer();
    var columns = {};
    var values = {};
    var next = function() {
      if (!rows.valid()) {
        defer.resolve({columns: columns, values: values});
        return;
      }
      var row = rows.current();
      if (typeof row[join.leftField] !== 'undefined') {
        new Get(self._transport, self._schemaFull, self._keyspace, join.rightTable)
          .fields(join.fields)
          .filter(join.rightField, row[join.leftField])
          .limit(1)
        .then(function(joinedRows) {
          joinedRows.getColumns().forEach(function(column) {
            var name = column.getName();
            if (typeof columns[name] === 'undefined') {
              columns[name] = column;
              values[name] = [];
            }
          });
          if (joinedRows.valid()) {
            var joinedRow = joinedRows.current();
            Object.keys(joinedRow).forEach(function(key) {
              values[key].push(joinedRow[key]);
            });
          }
          else {
            // No valid row found.
            join.fields.forEach(function(field) {
              values[field].push(null);
            });
          }<commit_msg>Rewind rows iterator after reading data for joins<commit_after>"use strict";

(function() {
  var castor = require('../castor-client');
  var Q = require('q');
  var Query = require('./Query');
  var Field = require('./Field');
  var Schema = require('./Schema');
  
  /**
   * Get class
   * 
   * Usage example:
   * 
   * castor.get('table')
   *   .fields(['field', 'field'])
   *   .filter('field', 'uuid', '...')
   *   .orderBy('field', 'asc')
   *   .limit(10)
   *   .cache(60)
   *   .consistency(castor.CONSISTENCY_QUORUM)
   * .then(...
   */
  var Get = function(transport, schema, keyspace, table) {
    this._transport = transport;
    this._schema = schema.get(table);
    this._schemaFull = schema;
    this._keyspace = keyspace;
    this._table = table;
    this._fields = [];
    this._filter = [];
    this._orderBy = null;
    this._orderByDirection = 'asc';
    this._limit = null;
    this._allowFiltering = false;
    this._consistency = 0x0004;
    this._joins = [];
  };
  
  Get.prototype.fields = function(fields) {
    this._fields = fields;
    return this;
  };
  
  Get.prototype.filter = function(name, value, operator) {
    if (typeof operator === 'undefined') {
      operator = '=';
    }
    if (operator != '=' && operator != '<' && operator != '>' && operator != '<=' && operator != '>=') {
      throw Error('Unknown operator: ' + operator);
    }
    this._filter.push({name: name, value: value, operator: operator});
    return this;
  };
  
  Get.prototype.orderBy = function(name, direction) {
    this._orderBy = name;
    if (direction !== undefined) {
      this._orderByDirection = direction;
    }
    return this;
  };
  
  Get.prototype.limit = function(limit) {
    this._limit = limit;
    return this;
  };
  
  Get.prototype.allowFiltering = function(allow) {
    if (typeof allow === 'undefined') {
      allow = true;
    }
    this._allowFiltering = allow;
    return this;
  };
  
  Get.prototype.consistency = function(consistency) {
    this._consistency = consistency;
    return this;
  };
  
  Get.prototype.join = function(leftField, rightField, fields) {
    // Right field is in format "table" or "table.field". The fieldname is
    // identical to the left field if not specified.
    var parts = rightField.split('.');
    var rightTable = parts[0];
    rightField = parts.length > 1 ? parts[1] : leftField;
    this._joins.push({
      leftField: leftField,
      rightTable: rightTable,
      rightField: rightField,
      fields: fields
    });
    return this;
  };
  
  Get.prototype.execute = function() {
    var defer = Q.defer();
    var self = this;
    
    Q.when(this._schema).then(function(schema) {
      var cql = 'SELECT ';
      if (self._fields.length) {
        cql = cql + self._fields.join(', ');
      }
      else {
        cql = cql + '*';
      }
      cql = cql + ' FROM ' + self._keyspace + '.' + self._table;
      if (self._filter.length) {
        for (var i = 0; i < self._filter.length; ++i) {
          var type = schema.columns[self._filter[i].name];
          if (typeof type === 'undefined') {
            throw Error('Unknown field ' + self._filter[i].name + ' in filter');
          }
          var filter = new Field(self._filter[i].name, type, self._filter[i].value, self._filter[i].operator);
          cql = cql + (i == 0 ? ' WHERE ' : ' AND ');
          cql = cql + filter.toString();
        }
      }
      if (self._orderBy !== null) {
        cql = cql + ' ORDER BY ' + self._orderBy + ' ' + self._orderByDirection.toUpperCase();
      }
      if (self._limit !== null) {
        cql = cql + ' LIMIT ' + self._limit;
      }
      if (self._allowFiltering) {
        cql = cql + ' ALLOW FILTERING';
      }
      if (self._joins.length) {
        self._executeWithJoins(cql, defer);
      }
      else {
        new Query(self._transport, cql, self._consistency).execute(defer);
      }
    }).fail(function(error) {
      defer.reject(error);
    });
    
    return defer.promise;
  };
  
  Get.prototype.then = function(callback) {
    return this.execute().then(callback);
  };
  
  Get.prototype._executeWithJoins = function(cql, defer) {
    var self = this;
    new Query(self._transport, cql, self._consistency).then(function(rows) {
      var joins = [];
      self._joins.forEach(function(join) {
        joins.push(self._executeJoin(rows, join));
      });
      Q.all(joins).then(function(results) {
        results.forEach(function(result) {
          Object.keys(result.columns).forEach(function(column) {
            rows.addColumn(result.columns[column], result.values[column]);
          });
        });
        defer.resolve(rows);
      }).fail(function(error) {
        defer.reject(error);
      }).done();
    });
  };
  
  Get.prototype._executeJoin = function(rows, join) {
    var self = this;
    var defer = Q.defer();
    var columns = {};
    var values = {};
    var next = function() {
      if (!rows.valid()) {
        rows.rewind();
        defer.resolve({columns: columns, values: values});
        return;
      }
      var row = rows.current();
      if (typeof row[join.leftField] !== 'undefined') {
        new Get(self._transport, self._schemaFull, self._keyspace, join.rightTable)
          .fields(join.fields)
          .filter(join.rightField, row[join.leftField])
          .limit(1)
        .then(function(joinedRows) {
          joinedRows.getColumns().forEach(function(column) {
            var name = column.getName();
            if (typeof columns[name] === 'undefined') {
              columns[name] = column;
              values[name] = [];
            }
          });
          if (joinedRows.valid()) {
            var joinedRow = joinedRows.current();
            Object.keys(joinedRow).forEach(function(key) {
              values[key].push(joinedRow[key]);
            });
          }
          else {
            // No valid row found.
            join.fields.forEach(function(field) {
              values[field].push(null);
            });
          }
{'fraction_non_alphanumeric': 0.15643481882548937, 'fraction_numerical': 0.0024989587671803417, 'mean_word_length': 2.1561514195583595, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3656524', 'n_tokens_mistral': 3760, 'n_tokens_neox': 3698, 'n_words': 742}