You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ec...@apache.org on 2016/04/08 21:32:10 UTC
[18/52] [abbrv] hbase git commit: Revert "HBASE-15572 Adding optional
timestamp semantics to HBase-Spark (Weiqing Yang)"
Revert "HBASE-15572 Adding optional timestamp semantics to HBase-Spark (Weiqing Yang)"
This reverts commit eec27ad7ef7b5078f705301bd3042991d4d4b4d9.
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/6905d272
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/6905d272
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/6905d272
Branch: refs/heads/HBASE-14850
Commit: 6905d272d34f0d674cfd73c6a8d579e3231b5a78
Parents: eec27ad
Author: Sean Busbey <bu...@apache.org>
Authored: Thu Mar 31 21:40:50 2016 -0500
Committer: Sean Busbey <bu...@apache.org>
Committed: Thu Mar 31 21:40:50 2016 -0500
----------------------------------------------------------------------
.../hadoop/hbase/spark/DefaultSource.scala | 8 +-
.../spark/datasources/HBaseSparkConf.scala | 5 -
.../spark/datasources/HBaseTableScanRDD.scala | 26 -----
.../hadoop/hbase/spark/DefaultSourceSuite.scala | 105 +++----------------
4 files changed, 14 insertions(+), 130 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/6905d272/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
----------------------------------------------------------------------
diff --git a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
index c71ee4e..7970816 100644
--- a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
+++ b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
@@ -88,12 +88,6 @@ case class HBaseRelation (
userSpecifiedSchema: Option[StructType]
)(@transient val sqlContext: SQLContext)
extends BaseRelation with PrunedFilteredScan with InsertableRelation with Logging {
-
- val timestamp = parameters.get(HBaseSparkConf.TIMESTAMP).map(_.toLong)
- val minTimeStamp = parameters.get(HBaseSparkConf.MIN_TIMESTAMP).map(_.toLong)
- val maxTimeStamp = parameters.get(HBaseSparkConf.MAX_TIMESTAMP).map(_.toLong)
- val maxVersions = parameters.get(HBaseSparkConf.MAX_VERSIONS).map(_.toInt)
-
val catalog = HBaseTableCatalog(parameters)
def tableName = catalog.name
val configResources = parameters.getOrElse(HBaseSparkConf.HBASE_CONFIG_RESOURCES_LOCATIONS, "")
@@ -210,7 +204,7 @@ case class HBaseRelation (
System.arraycopy(x, 0, rBytes, offset, x.length)
offset += x.length
}
- val put = timestamp.fold(new Put(rBytes))(new Put(rBytes, _))
+ val put = new Put(rBytes)
colsIdxedFields.foreach { case (x, y) =>
val b = Utils.toBytes(row(x), y)
http://git-wip-us.apache.org/repos/asf/hbase/blob/6905d272/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala
----------------------------------------------------------------------
diff --git a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala
index 2e4c0b3..ca44d42 100644
--- a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala
+++ b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseSparkConf.scala
@@ -36,9 +36,4 @@ object HBaseSparkConf{
val USE_HBASE_CONTEXT = "hbase.use.hbase.context"
val PUSH_DOWN_COLUMN_FILTER = "hbase.pushdown.column.filter"
val defaultPushDownColumnFilter = true
-
- val TIMESTAMP = "timestamp"
- val MIN_TIMESTAMP = "minTimestamp"
- val MAX_TIMESTAMP = "maxTimestamp"
- val MAX_VERSIONS = "maxVersions"
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/6905d272/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableScanRDD.scala
----------------------------------------------------------------------
diff --git a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableScanRDD.scala b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableScanRDD.scala
index 886114a..2e05651 100644
--- a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableScanRDD.scala
+++ b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableScanRDD.scala
@@ -105,7 +105,6 @@ class HBaseTableScanRDD(relation: HBaseRelation,
val gets = new ArrayList[Get]()
x.foreach{ y =>
val g = new Get(y)
- handleTimeSemantics(g)
columns.foreach { d =>
if (!d.isRowKey) {
g.addColumn(d.cfBytes, d.colBytes)
@@ -158,7 +157,6 @@ class HBaseTableScanRDD(relation: HBaseRelation,
case (Some(Bound(a, b)), None) => new Scan(a)
case (None, None) => new Scan()
}
- handleTimeSemantics(scan)
columns.foreach { d =>
if (!d.isRowKey) {
@@ -228,30 +226,6 @@ class HBaseTableScanRDD(relation: HBaseRelation,
} ++ gIt
rIts
}
-
- private def handleTimeSemantics(query: Query): Unit = {
- // Set timestamp related values if present
- (query, relation.timestamp, relation.minTimeStamp, relation.maxTimeStamp) match {
- case (q: Scan, Some(ts), None, None) => q.setTimeStamp(ts)
- case (q: Get, Some(ts), None, None) => q.setTimeStamp(ts)
-
- case (q:Scan, None, Some(minStamp), Some(maxStamp)) => q.setTimeRange(minStamp, maxStamp)
- case (q:Get, None, Some(minStamp), Some(maxStamp)) => q.setTimeRange(minStamp, maxStamp)
-
- case (q, None, None, None) =>
-
- case _ => throw new IllegalArgumentException(s"Invalid combination of query/timestamp/time range provided. " +
- s"timeStamp is: ${relation.timestamp.get}, minTimeStamp is: ${relation.minTimeStamp.get}, " +
- s"maxTimeStamp is: ${relation.maxTimeStamp.get}")
- }
- if (relation.maxVersions.isDefined) {
- query match {
- case q: Scan => q.setMaxVersions(relation.maxVersions.get)
- case q: Get => q.setMaxVersions(relation.maxVersions.get)
- case _ => throw new IllegalArgumentException("Invalid query provided with maxVersions")
- }
- }
- }
}
case class SerializedFilter(b: Option[Array[Byte]])
http://git-wip-us.apache.org/repos/asf/hbase/blob/6905d272/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala
----------------------------------------------------------------------
diff --git a/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala b/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala
index 4ad10e2..500967d 100644
--- a/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala
+++ b/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala
@@ -19,39 +19,32 @@ package org.apache.hadoop.hbase.spark
import org.apache.avro.Schema
import org.apache.avro.generic.GenericData
-import org.apache.hadoop.hbase.client.{ConnectionFactory, Put}
+import org.apache.hadoop.hbase.client.{Put, ConnectionFactory}
import org.apache.hadoop.hbase.spark.datasources.HBaseSparkConf
import org.apache.hadoop.hbase.util.Bytes
-import org.apache.hadoop.hbase.{HBaseTestingUtility, TableName}
+import org.apache.hadoop.hbase.{TableName, HBaseTestingUtility}
import org.apache.spark.sql.datasources.hbase.HBaseTableCatalog
-import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SQLContext}
-import org.apache.spark.{Logging, SparkConf, SparkContext}
+import org.apache.spark.{SparkConf, SparkContext, Logging}
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, FunSuite}
case class HBaseRecord(
- col0: String,
- col1: Boolean,
- col2: Double,
- col3: Float,
- col4: Int,
- col5: Long,
- col6: Short,
- col7: String,
- col8: Byte)
+ col0: String,
+ col1: String,
+ col2: Double,
+ col3: Float,
+ col4: Int,
+ col5: Long)
object HBaseRecord {
def apply(i: Int, t: String): HBaseRecord = {
val s = s"""row${"%03d".format(i)}"""
HBaseRecord(s,
- i % 2 == 0,
+ s,
i.toDouble,
i.toFloat,
i,
- i.toLong,
- i.toShort,
- s"String$i: $t",
- i.toByte)
+ i.toLong)
}
}
@@ -822,14 +815,11 @@ BeforeAndAfterEach with BeforeAndAfterAll with Logging {
|"rowkey":"key",
|"columns":{
|"col0":{"cf":"rowkey", "col":"key", "type":"string"},
- |"col1":{"cf":"cf1", "col":"col1", "type":"boolean"},
+ |"col1":{"cf":"cf1", "col":"col1", "type":"string"},
|"col2":{"cf":"cf2", "col":"col2", "type":"double"},
|"col3":{"cf":"cf3", "col":"col3", "type":"float"},
|"col4":{"cf":"cf4", "col":"col4", "type":"int"},
- |"col5":{"cf":"cf5", "col":"col5", "type":"bigint"},
- |"col6":{"cf":"cf6", "col":"col6", "type":"smallint"},
- |"col7":{"cf":"cf7", "col":"col7", "type":"string"},
- |"col8":{"cf":"cf8", "col":"col8", "type":"tinyint"}
+ |"col5":{"cf":"cf5", "col":"col5", "type":"bigint"}}
|}
|}""".stripMargin
@@ -876,75 +866,6 @@ BeforeAndAfterEach with BeforeAndAfterAll with Logging {
assert(s.count() == 6)
}
- test("Timestamp semantics") {
- val sql = sqlContext
- import sql.implicits._
-
- // There's already some data in here from recently. Let's throw something in
- // from 1993 which we can include/exclude and add some data with the implicit (now) timestamp.
- // Then we should be able to cross-section it and only get points in between, get the most recent view
- // and get an old view.
- val oldMs = 754869600000L
- val startMs = System.currentTimeMillis()
- val oldData = (0 to 100).map { i =>
- HBaseRecord(i, "old")
- }
- val newData = (200 to 255).map { i =>
- HBaseRecord(i, "new")
- }
-
- sc.parallelize(oldData).toDF.write.options(
- Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseTableCatalog.tableName -> "5",
- HBaseSparkConf.TIMESTAMP -> oldMs.toString))
- .format("org.apache.hadoop.hbase.spark")
- .save()
- sc.parallelize(newData).toDF.write.options(
- Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseTableCatalog.tableName -> "5"))
- .format("org.apache.hadoop.hbase.spark")
- .save()
-
- // Test specific timestamp -- Full scan, Timestamp
- val individualTimestamp = sqlContext.read
- .options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseSparkConf.TIMESTAMP -> oldMs.toString))
- .format("org.apache.hadoop.hbase.spark")
- .load();
- assert(individualTimestamp.count() == 101)
-
- // Test getting everything -- Full Scan, No range
- val everything = sqlContext.read
- .options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog))
- .format("org.apache.hadoop.hbase.spark")
- .load()
- assert(everything.count() == 256)
- // Test getting everything -- Pruned Scan, TimeRange
- val element50 = everything.where(col("col0") === lit("row050")).select("col7").collect()(0)(0)
- assert(element50 == "String50: extra")
- val element200 = everything.where(col("col0") === lit("row200")).select("col7").collect()(0)(0)
- assert(element200 == "String200: new")
-
- // Test Getting old stuff -- Full Scan, TimeRange
- val oldRange = sqlContext.read
- .options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseSparkConf.MIN_TIMESTAMP -> "0",
- HBaseSparkConf.MAX_TIMESTAMP -> (oldMs + 100).toString))
- .format("org.apache.hadoop.hbase.spark")
- .load()
- assert(oldRange.count() == 101)
- // Test Getting old stuff -- Pruned Scan, TimeRange
- val oldElement50 = oldRange.where(col("col0") === lit("row050")).select("col7").collect()(0)(0)
- assert(oldElement50 == "String50: old")
-
- // Test Getting middle stuff -- Full Scan, TimeRange
- val middleRange = sqlContext.read
- .options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseSparkConf.MIN_TIMESTAMP -> "0",
- HBaseSparkConf.MAX_TIMESTAMP -> (startMs + 100).toString))
- .format("org.apache.hadoop.hbase.spark")
- .load()
- assert(middleRange.count() == 256)
- // Test Getting middle stuff -- Pruned Scan, TimeRange
- val middleElement200 = middleRange.where(col("col0") === lit("row200")).select("col7").collect()(0)(0)
- assert(middleElement200 == "String200: extra")
- }
-
// catalog for insertion
def avroWriteCatalog = s"""{
|"table":{"namespace":"default", "name":"avrotable"},