<commit_before>/**
 * Copyright (c) 2015 Basho Technologies, Inc.
 *
 * This file is provided to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file
 * except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package com.basho.riak.spark.rdd

import java.util.concurrent.ExecutionException
import java.util.{Calendar, TimeZone, GregorianCalendar}

import com.basho.riak.client.core.netty.RiakResponseException
import com.basho.riak.client.core.operations.ts.StoreOperation
import com.basho.riak.client.core.operations.FetchBucketPropsOperation
import com.basho.riak.client.core.query.{BucketProperties, Namespace}
import com.basho.riak.client.core.query.timeseries.{Cell, Row}
import org.junit.{Assume, Before, Test}
import scala.collection.JavaConversions._
import com.basho.riak.spark.toSparkContextFunctions

case class TimeSeriesData(time: Long, user_id: String, temperature_k: Double)

class TimeSeriesTest extends AbstractRDDTest {

  protected val DEFAULT_TS_NAMESPACE = new Namespace("time_series_test","time_series_test")

  var tsRangeStart: Calendar = null
  var tsRangeEnd: Calendar = null

  private def mkTimestamp(timeInMillis: Long): Calendar = {
    val c = new GregorianCalendar(TimeZone.getTimeZone("UTC"))
    c.setTimeInMillis(timeInMillis)
    c
  }

  @Before
  def setupData(): Unit = {

    val fetchProps = new FetchBucketPropsOperation.Builder(DEFAULT_TS_NAMESPACE).build()

    withRiakDo( session => {
      session.getRiakCluster.execute(fetchProps)
    })

    try {
      val props: BucketProperties = fetchProps.get().getBucketProperties
    } catch {
      case ex :ExecutionException if ex.getCause.isInstanceOf[RiakResponseException]
          && ex.getCause.getMessage.startsWith("No bucket-type named")  =>

        val msg = "Bucket type for Time Series test data is not created, Time series tests will be skipped"

        logWarning(msg + "\n\n" +
          "To create and activate TS test bucket, please use the following commands:\n" +
          "\t" + """./riak-admin bucket-type create time_series_test '{"props":{"n_val":3, "table_def": "create table time_series_test (surrogate_key sint64 not null, family varchar not null, time timestamp not null, user_id varchar not null, temperature_k double, primary key ((surrogate_key, family, quantum(time, 10, s)), surrogate_key, family, time))"}}' """  +"\n" +
          "\t./riak-admin bucket-type activate time_series_test")

        Assume.assumeTrue(msg + " (See logs for the details)", false)
    }

    // ----------  Storing test data into Riak TS

    val testData = List(
      TimeSeriesData(111111, "bryce", 305.37),<commit_msg>Avoiding unnecessary reset for non TS bucket before TS tests.<commit_after>/**
 * Copyright (c) 2015 Basho Technologies, Inc.
 *
 * This file is provided to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file
 * except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package com.basho.riak.spark.rdd

import java.util.concurrent.ExecutionException
import java.util.{Calendar, TimeZone, GregorianCalendar}

import com.basho.riak.client.core.netty.RiakResponseException
import com.basho.riak.client.core.operations.ts.StoreOperation
import com.basho.riak.client.core.operations.FetchBucketPropsOperation
import com.basho.riak.client.core.query.{BucketProperties, Namespace}
import com.basho.riak.client.core.query.timeseries.{Cell, Row}
import org.junit.{Assume, Before, Test}
import scala.collection.JavaConversions._
import com.basho.riak.spark.toSparkContextFunctions

case class TimeSeriesData(time: Long, user_id: String, temperature_k: Double)

class TimeSeriesTest extends AbstractRDDTest {

  protected val DEFAULT_TS_NAMESPACE = new Namespace("time_series_test","time_series_test")

  var tsRangeStart: Calendar = null
  var tsRangeEnd: Calendar = null

  private def mkTimestamp(timeInMillis: Long): Calendar = {
    val c = new GregorianCalendar(TimeZone.getTimeZone("UTC"))
    c.setTimeInMillis(timeInMillis)
    c
  }

  override def initialize(): Unit = {
     /* DO NOT CALL IHERITED initialize to avoid reseting non TS bucket*/
    //super.initialize()
    sc = createSparkContext(initSparkConf())
  }

  @Before
  def setupData(): Unit = {

    val fetchProps = new FetchBucketPropsOperation.Builder(DEFAULT_TS_NAMESPACE).build()

    withRiakDo( session => {
      session.getRiakCluster.execute(fetchProps)
    })

    try {
      val props: BucketProperties = fetchProps.get().getBucketProperties
    } catch {
      case ex :ExecutionException if ex.getCause.isInstanceOf[RiakResponseException]
          && ex.getCause.getMessage.startsWith("No bucket-type named")  =>

        val msg = "Bucket type for Time Series test data is not created, Time series tests will be skipped"

        logWarning(msg + "\n\n" +
          "To create and activate TS test bucket, please use the following commands:\n" +
          "\t" + """./riak-admin bucket-type create time_series_test '{"props":{"n_val":3, "table_def": "create table time_series_test (surrogate_key sint64 not null, family varchar not null, time timestamp not null, user_id varchar not null, temperature_k double, primary key ((surrogate_key, family, quantum(time, 10, s)), surrogate_key, family, time))"}}' """  +"\n" +
          "\t./riak-admin bucket-type activate time_series_test")

        Assume.assumeTrue(msg + " (See logs for the details)", false)
    }

    // ----------  Storing test data into Riak TS

    val testData = List(
      TimeSeriesData(111111, "bryce", 305.37),
{'fraction_non_alphanumeric': 0.10230299232335892, 'fraction_numerical': 0.0075199749334168885, 'mean_word_length': 3.975837879968823, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '419473', 'n_tokens_mistral': 1945, 'n_tokens_neox': 1840, 'n_words': 572}