<commit_before>package org.apache.zeppelin.notebook.repo;


import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;

import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.NoteInfo;
import org.apache.zeppelin.notebook.Paragraph;
import org.apache.zeppelin.search.SearchService;
import org.apache.zeppelin.user.AuthenticationInfo;
import org.apache.zeppelin.util.GsonUtil;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.HasParentQueryBuilder;
import org.elasticsearch.index.query.MultiMatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHitField;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * ElasticSearch-backed NoteRepo and Service implementation
 */
public class ElasticSearchRepo implements NotebookRepo, SearchService {

  private static final Logger LOG = LoggerFactory.getLogger(ElasticSearchRepo.class);

  private static final String SEARCH_FIELD_TITLE = "title";
  private static final String SEARCH_FIELD_TEXT = "text";

  private static final String PARAGRAPH = "paragraph";
  private static final String ID_FIELD = "_id";

  private static final String AGGREGATION_FILED_TAGS = "tags";
  private static final String AGGREGATION_NAME_TAGS = "tags_aggs";

  private static final String AGGREGATION_FILED_AUTHOR = "createdBy";
  private static final String AGGREGATION_NAME_AUTHOR = "createdBy_agg";

  private static final String AGGREGATION_FILED_LAST_UPDATED = "lastUpdated";
  private static final String AGGREGATION_NAME_LAST_UPDATED = "lastUpdated_agg";

  private static final String DATE_RANGE_FORMAT = "yyyy-MM";

  private Client client;
  private String indexName;
  private String noteTypeName;
  private String paragraphTypeName;//note-to-pargraph modeled as parent-to-child relation in ES
  private int defaultPageSize;
  private int defaultTermsAggSize;


  public ElasticSearchRepo(ZeppelinConfiguration conf) throws IOException {
    String esHost = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_HOST);
    int esPort = conf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_PORT);

    this.indexName = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_INDEX_NAME);
    this.noteTypeName = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_NOTE_TYPE_NAME);
    this.paragraphTypeName = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_PARAGRAPH_TYPE_NAME);
    this.defaultPageSize = conf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_SEARCH_PAGE_SIZE);
    this.defaultTermsAggSize = conf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_TERMS_AGGREGATION_SIZE);

//    Settings settings = Settings.settingsBuilder()
//            .put("cluster.name", "zeppelin").build();

    TransportClient transportClient = TransportClient.builder().build()
            .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(esHost), esPort));
    client = transportClient;

    if (transportClient.connectedNodes() == null || transportClient.connectedNodes().size() == 0) {
      throw new IOException(String.format("There are no active nodes available for the transport@%s:%s", esHost, esPort));
    }
  }


  @Override
  public List<NoteInfo> list(AuthenticationInfo subject) throws IOException {
    SearchResponse response = client.prepareSearch(indexName)
            .setTypes(noteTypeName)
            .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
            .addAggregation(AggregationBuilders.terms(AGGREGATION_NAME_TAGS).field(AGGREGATION_FILED_TAGS).size(defaultTermsAggSize))
            .addAggregation(AggregationBuilders.terms(AGGREGATION_NAME_AUTHOR).field(AGGREGATION_FILED_AUTHOR).size(defaultTermsAggSize))
            .addAggregation(AggregationBuilders.dateHistogram(AGGREGATION_NAME_LAST_UPDATED).field(AGGREGATION_FILED_LAST_UPDATED).interval(DateHistogramInterval.MONTH).format(DATE_RANGE_FORMAT))
            .execute()
            .actionGet();

    List<NoteInfo> results = new LinkedList<NoteInfo>();
    SearchHits hits = response.getHits();
    long count = hits.getTotalHits();
    if (count > 0) {
      results = new ArrayList<NoteInfo>((int) count);
      for (SearchHit hit : hits.getHits()) {
        Note noteParsed = GsonUtil.fromJson(hit.getSourceAsString(), Note.class);
        results.add(new NoteInfo(noteParsed));
      }
    }

    //TODO: search with aggs,fields tags/topic?(qy)
    //aggregation parse
    Aggregations aggregations = response.getAggregations();
    termsAggregationParse(aggregations, AGGREGATION_NAME_TAGS);
    termsAggregationParse(aggregations, AGGREGATION_NAME_AUTHOR);
    dateHistogramAggregationParse(aggregations, AGGREGATION_NAME_LAST_UPDATED);

    return results;
  }

  /**
   * parse date histogram aggregation
   */
  private void dateHistogramAggregationParse(Aggregations aggregations, String aggName) {
    if (aggregations != null) {
      Histogram agg = aggregations.get(aggName);
      // For each entry
      for (Histogram.Bucket entry : agg.getBuckets()) {
        DateTime key = (DateTime) entry.getKey();    // Key
        String keyAsString = entry.getKeyAsString(); // Key as String
        long docCount = entry.getDocCount();         // Doc count

        LOG.info("key [{}], date [{}], doc_count [{}]", keyAsString, key.getYear(), docCount);
      }

    }
  }<commit_msg>when list on ES,use scroll to aviod deep pagination, and filter by term AuthenticationInfo.getUser() when parameter not null<commit_after>package org.apache.zeppelin.notebook.repo;


import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;

import org.apache.lucene.queryparser.xml.builders.FilteredQueryBuilder;
import org.apache.lucene.queryparser.xml.builders.TermQueryBuilder;
import org.apache.lucene.search.TermQuery;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.NoteInfo;
import org.apache.zeppelin.notebook.Paragraph;
import org.apache.zeppelin.search.SearchService;
import org.apache.zeppelin.user.AuthenticationInfo;
import org.apache.zeppelin.util.GsonUtil;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.HasParentQueryBuilder;
import org.elasticsearch.index.query.MultiMatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHitField;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.sort.SortParseElement;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * ElasticSearch-backed NoteRepo and Service implementation
 */
public class ElasticSearchRepo implements NotebookRepo, SearchService {

  private static final Logger LOG = LoggerFactory.getLogger(ElasticSearchRepo.class);

  private static final String SEARCH_FIELD_TITLE = "title";
  private static final String SEARCH_FIELD_TEXT = "text";

  private static final String PARAGRAPH = "paragraph";
  private static final String ID_FIELD = "_id";

  private static final String AGGREGATION_FILED_TAGS = "tags";
  private static final String AGGREGATION_NAME_TAGS = "tags_aggs";

  private static final String AGGREGATION_FILED_AUTHOR = "createdBy";
  private static final String AGGREGATION_NAME_AUTHOR = "createdBy_agg";

  private static final String AGGREGATION_FILED_LAST_UPDATED = "lastUpdated";
  private static final String AGGREGATION_NAME_LAST_UPDATED = "lastUpdated_agg";

  private static final String DATE_RANGE_FORMAT = "yyyy-MM";

  private Client client;
  private String indexName;
  private String noteTypeName;
  private String paragraphTypeName;//note-to-pargraph modeled as parent-to-child relation in ES
  private int defaultPageSize;
  private int defaultTermsAggSize;


  public ElasticSearchRepo(ZeppelinConfiguration conf) throws IOException {
    String esHost = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_HOST);
    int esPort = conf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_PORT);

    this.indexName = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_INDEX_NAME);
    this.noteTypeName = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_NOTE_TYPE_NAME);
    this.paragraphTypeName = conf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_PARAGRAPH_TYPE_NAME);
    this.defaultPageSize = conf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_SEARCH_PAGE_SIZE);
    this.defaultTermsAggSize = conf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTE_REPO_ES_TERMS_AGGREGATION_SIZE);

//    Settings settings = Settings.settingsBuilder()
//            .put("cluster.name", "zeppelin").build();

    TransportClient transportClient = TransportClient.builder().build()
            .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(esHost), esPort));
    client = transportClient;

    if (transportClient.connectedNodes() == null || transportClient.connectedNodes().size() == 0) {
      throw new IOException(String.format("There are no active nodes available for the transport@%s:%s", esHost, esPort));
    }
  }


  /**
   * use ES scroll api to handler large dataset
   *
   * @param subject contains user information.
   * @return notes from all shards within scroll time range
   */
  @Override
  public List<NoteInfo> list(AuthenticationInfo subject) throws IOException {
    SearchResponse scrollResp = client.prepareSearch(indexName).setTypes(noteTypeName)
            .addSort(AGGREGATION_FILED_LAST_UPDATED, SortOrder.DESC)
            .setScroll(new TimeValue(60000))
            .setQuery(subject == null ? QueryBuilders.matchAllQuery() : QueryBuilders.boolQuery()
                    .must(QueryBuilders.termQuery(AGGREGATION_FILED_AUTHOR, subject.getUser())))
            .addAggregation(AggregationBuilders.terms(AGGREGATION_NAME_TAGS).field(AGGREGATION_FILED_TAGS).size(defaultTermsAggSize))
            .addAggregation(AggregationBuilders.terms(AGGREGATION_NAME_AUTHOR).field(AGGREGATION_FILED_AUTHOR).size(defaultTermsAggSize))
            .addAggregation(AggregationBuilders.dateHistogram(AGGREGATION_NAME_LAST_UPDATED).field(AGGREGATION_FILED_LAST_UPDATED).interval(DateHistogramInterval.MONTH).format(DATE_RANGE_FORMAT))
            .setSize(100).execute().actionGet(); //100 hits per shard will be returned for each scroll

    List<NoteInfo> results = new LinkedList<NoteInfo>();
    //Scroll until no hits are returned
    while (true) {
      for (SearchHit hit : scrollResp.getHits().getHits()) {
        //handle hitted note
        Note noteParsed = GsonUtil.fromJson(hit.getSourceAsString(), Note.class);
        results.add(new NoteInfo(noteParsed));

        //aggregation parse
        //TODO: search with aggs,fields tags/topic?(qy)
        Aggregations aggregations = scrollResp.getAggregations();
        termsAggregationParse(aggregations, AGGREGATION_NAME_TAGS);
        termsAggregationParse(aggregations, AGGREGATION_NAME_AUTHOR);
        dateHistogramAggregationParse(aggregations, AGGREGATION_NAME_LAST_UPDATED);
      }
      scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(60000)).execute().actionGet();
      //Break condition: No hits are returned
      if (scrollResp.getHits().getHits().length == 0) {
        break;
      }
    }

    return results;
  }

  /**
   * parse date histogram aggregation
   */
  private void dateHistogramAggregationParse(Aggregations aggregations, String aggName) {
    if (aggregations != null) {
      Histogram agg = aggregations.get(aggName);
      // For each entry
      for (Histogram.Bucket entry : agg.getBuckets()) {
        DateTime key = (DateTime) entry.getKey();    // Key
        String keyAsString = entry.getKeyAsString(); // Key as String
        long docCount = entry.getDocCount();         // Doc count

        LOG.info("key [{}], date [{}], doc_count [{}]", keyAsString, key.getYear(), docCount);
      }

    }
  }
{'fraction_non_alphanumeric': 0.0938562091503268, 'fraction_numerical': 0.001568627450980392, 'mean_word_length': 5.152392440691596, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '4674298', 'n_tokens_mistral': 4555, 'n_tokens_neox': 4488, 'n_words': 805}